mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
fix(workspaces): back off failed terminal cleanup and retire abandoned reservations through remove (#20020)
This commit is contained in:
1 parent
5de2842620
commit
17ad23aafe
9 files changed
+434
-16
No files matched your search
@@ -218,6 +218,26 @@ Legacy intents without a recovery hint retain their deterministic ID. A hinted
|
||||
intent without a recorded ID is ambiguous and fails closed. None of these
|
||||
receipts replaces the provider's checks.
|
||||
|
||||
A reservation is a workspace whose `prepare` never returned a binding. When its
|
||||
source task is done or cancelled, no task points at it, and no run for the task
|
||||
is queued or running, Core archives it and sends `remove` with `binding: null`
|
||||
and `expectedHeadSha: null` instead of reconciling it with an agent-free
|
||||
`prepare`. The provider must resolve its own record for that execution
|
||||
workspace, discard anything the unfinished preparation created, and return
|
||||
`{ kind: "removed" }`, including when nothing was created. The removal is
|
||||
persisted and replayed with the same operation ID like any other removal. A
|
||||
provider that refuses it fails the removal, which backs off like any other
|
||||
cleanup failure. A reservation that a task still points at, or whose task is
|
||||
open, keeps the `prepare` reconciliation.
|
||||
|
||||
When terminal cleanup inspection or removal fails, Core records
|
||||
`metadata.terminalCleanupRetry` on the workspace with the failure count, the
|
||||
next attempt time, and the failure `kind`, `reason`, and Core's message. Retries
|
||||
back off from 5 minutes, doubling up to 4 hours. A successful inspection or
|
||||
removal clears the record, and a reopen, which raises the lifecycle generation,
|
||||
discards it. Other wakes, including provider upgrades, wait for the next attempt
|
||||
time. Core keeps the record through workspace metadata updates.
|
||||
|
||||
The optional `prepared.destinationRef` records the resolved delivery branch as
|
||||
`refs/heads/<branch>`, up to 266 characters including the prefix. Providers should
|
||||
return null when preparation cannot establish an unambiguous branch destination
|
||||
|
||||
@@ -65,7 +65,7 @@ export const workspaceProviderOperationSchema = z.discriminatedUnion("kind", [
|
||||
z
|
||||
.object({
|
||||
kind: z.literal("remove"),
|
||||
binding: workspaceProviderBindingSchema,
|
||||
binding: workspaceProviderBindingSchema.nullable(),
|
||||
expectedHeadSha: sha.nullable(),
|
||||
recoveryPullRequest: workspaceRecoveryPullRequestSchema.optional(),
|
||||
})
|
||||
|
||||
@@ -61,6 +61,11 @@ describe("workspace provider SDK", () => {
|
||||
}
|
||||
}
|
||||
});
|
||||
it("accepts an unbound reservation removal but requires a binding elsewhere", () => {
|
||||
expect(workspaceProviderOperationSchema.parse({ kind: "remove", binding: null, expectedHeadSha: null }))
|
||||
.toEqual({ kind: "remove", binding: null, expectedHeadSha: null });
|
||||
expect(() => workspaceProviderOperationSchema.parse({ kind: "inspect", binding: null })).toThrow();
|
||||
});
|
||||
it("accepts only canonical optional preparation destinations", () => {
|
||||
for (const destinationRef of [undefined, null, "refs/heads/main", "refs/heads/release/one"]) {
|
||||
expect(parseWorkspaceProviderResult(params, { ...result, ...(destinationRef !== undefined ? { destinationRef } : {}) }))
|
||||
|
||||
@@ -0,0 +1,250 @@
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { createDb, executionWorkspaces, heartbeatRuns, issues } from "@paperclipai/db";
|
||||
import { JsonRpcCallError, PLUGIN_RPC_ERROR_CODES } from "@paperclipai/plugin-sdk";
|
||||
import { createPluginWorkspaceFixture } from "./helpers/plugin-workspaces.js";
|
||||
import { inspectPluginWorkspace, preparePluginWorkspace, removePluginWorkspace, type WorkspaceWorkers } from "../services/plugin-workspaces.js";
|
||||
import { executionWorkspaceService } from "../services/execution-workspaces.js";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
const MINUTE = 60_000;
|
||||
|
||||
function providerRefusal() {
|
||||
return new JsonRpcCallError({ code: PLUGIN_RPC_ERROR_CODES.WORKER_ERROR, message: "private provider text",
|
||||
data: { schema: "paperclip/workspace-operation-error/v1", reason: "unsupported_operation" } });
|
||||
}
|
||||
|
||||
(support.supported ? describe : describe.skip)("plugin workspace terminal cleanup retry", () => {
|
||||
let db: ReturnType<typeof createDb>, stopDb: (() => Promise<void>) | undefined;
|
||||
beforeAll(async () => {
|
||||
const started = await startEmbeddedPostgresTestDatabase("plugin-workspace-terminal-retry");
|
||||
db = createDb(started.connectionString, { maxConnections: 1 });
|
||||
stopDb = started.cleanup;
|
||||
});
|
||||
afterAll(async () => {
|
||||
await db?.$client.end();
|
||||
await stopDb?.();
|
||||
});
|
||||
|
||||
async function setup() {
|
||||
const f = await createPluginWorkspaceFixture(db);
|
||||
const failures = { inspect: false, remove: false, unboundRemove: false, reopenFirst: false };
|
||||
const workers = { ...f.workers, call: async (...args: Parameters<WorkspaceWorkers["call"]>) => {
|
||||
const params = args[2] as { agentId: string | null; operation: { kind: string; binding?: unknown } };
|
||||
if (params.operation.kind === "prepare" && params.agentId === null) {
|
||||
f.calls.push(params);
|
||||
throw providerRefusal();
|
||||
}
|
||||
if (params.operation.kind === "remove" && params.operation.binding === null) {
|
||||
f.calls.push(params);
|
||||
if (failures.unboundRemove) throw providerRefusal();
|
||||
if (failures.remove) throw new Error("lost removal");
|
||||
return { kind: "removed" };
|
||||
}
|
||||
if ((params.operation.kind === "inspect" && failures.inspect) || (params.operation.kind === "remove" && failures.remove)) {
|
||||
f.calls.push(params);
|
||||
if (failures.reopenFirst) await db.update(executionWorkspaces).set({ metadata: sql`jsonb_set(${executionWorkspaces.metadata},
|
||||
'{lifecycleGeneration}', to_jsonb(coalesce((${executionWorkspaces.metadata}->>'lifecycleGeneration')::int, 0) + 1))` })
|
||||
.where(eq(executionWorkspaces.id, (params as unknown as { executionWorkspaceId: string }).executionWorkspaceId));
|
||||
throw new Error("provider outage");
|
||||
}
|
||||
return f.workers.call(...args);
|
||||
} } as WorkspaceWorkers;
|
||||
const read = async (id: string) => (await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, id)))[0]!;
|
||||
const finishTask = async () => {
|
||||
await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issueId));
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId));
|
||||
};
|
||||
const abandonReservation = async () => {
|
||||
f.lose();
|
||||
await expect(preparePluginWorkspace(db, workers, f.input)).rejects.toThrow("incomplete");
|
||||
const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId));
|
||||
const reservation = await read(task.executionWorkspaceId!);
|
||||
expect(reservation.providerRef).toBeNull();
|
||||
await db.update(issues).set({ executionWorkspaceId: null }).where(eq(issues.id, f.issueId));
|
||||
const replacement = await preparePluginWorkspace(db, workers, f.input);
|
||||
expect(replacement!.id).not.toBe(reservation.id);
|
||||
return reservation;
|
||||
};
|
||||
const callsFor = (id: string, kind: string) => f.calls.filter(call => call.executionWorkspaceId === id && call.operation.kind === kind);
|
||||
return { f, workers, failures, read, finishTask, abandonReservation, callsFor };
|
||||
}
|
||||
|
||||
it("retires an unbound reservation of a finished task through an agent-free remove and archives it", async () => {
|
||||
const s = await setup();
|
||||
const reservation = await s.abandonReservation();
|
||||
await s.finishTask();
|
||||
s.f.dirty();
|
||||
await executionWorkspaceService(db, { pluginWorkerManager: s.workers, workspaceReaperCooldownDays: 0 }).sweepTerminalWorkspaces(100);
|
||||
expect(s.callsFor(reservation.id, "prepare")).toHaveLength(1);
|
||||
expect(s.callsFor(reservation.id, "remove")).toEqual([expect.objectContaining({ agentId: null, runId: null,
|
||||
operation: { kind: "remove", binding: null, expectedHeadSha: null } })]);
|
||||
const retired = await s.read(reservation.id);
|
||||
expect(retired).toMatchObject({ status: "archived", cleanupReason: "issue_terminal",
|
||||
metadata: { workspaceProviderRemoved: true, workspaceProviderRemoval: null } });
|
||||
expect(retired.closedAt).not.toBeNull();
|
||||
expect(retired.metadata?.lifecycleGeneration).toBe(((reservation.metadata?.lifecycleGeneration as number) ?? 0) + 1);
|
||||
});
|
||||
|
||||
it("replays a failed reservation removal with the same identity and backs it off", async () => {
|
||||
const s = await setup();
|
||||
const reservation = await s.abandonReservation();
|
||||
await s.finishTask();
|
||||
s.f.dirty();
|
||||
s.failures.remove = true;
|
||||
let clock = Date.now();
|
||||
const svc = executionWorkspaceService(db, { pluginWorkerManager: s.workers, workspaceReaperCooldownDays: 0, now: () => new Date(clock) });
|
||||
await svc.sweepTerminalWorkspaces(100);
|
||||
const pending = await s.read(reservation.id);
|
||||
expect(pending).toMatchObject({ status: "archived",
|
||||
metadata: { workspaceProviderRemoval: { operation: { kind: "remove", binding: null } } } });
|
||||
expect(pending.metadata?.terminalCleanupRetry).toMatchObject({ failures: 1, generation: pending.metadata?.lifecycleGeneration,
|
||||
retryAt: new Date(clock + 5 * MINUTE).toISOString(), failure: { kind: "transport" } });
|
||||
const retryAt = Date.parse((pending.metadata?.terminalCleanupRetry as { retryAt: string }).retryAt);
|
||||
clock = Date.now() + 1;
|
||||
await svc.sweepTerminalWorkspaces(100);
|
||||
expect(s.callsFor(reservation.id, "remove")).toHaveLength(1);
|
||||
s.failures.remove = false;
|
||||
clock = retryAt;
|
||||
await svc.sweepTerminalWorkspaces(100);
|
||||
const removes = s.callsFor(reservation.id, "remove");
|
||||
expect(removes).toHaveLength(2);
|
||||
expect(new Set(removes.map(call => call.operationId)).size).toBe(1);
|
||||
expect(s.callsFor(reservation.id, "prepare")).toHaveLength(1);
|
||||
const retired = await s.read(reservation.id);
|
||||
expect(retired).toMatchObject({ status: "archived", metadata: { workspaceProviderRemoved: true } });
|
||||
expect(retired.metadata).not.toHaveProperty("terminalCleanupRetry");
|
||||
});
|
||||
|
||||
it("keeps the reservation a task still points at and records why its reconciliation failed", async () => {
|
||||
const s = await setup();
|
||||
s.f.lose();
|
||||
await expect(preparePluginWorkspace(db, s.workers, s.f.input)).rejects.toThrow("incomplete");
|
||||
const [task] = await db.select().from(issues).where(eq(issues.id, s.f.issueId));
|
||||
await s.finishTask();
|
||||
const clock = Date.now();
|
||||
await executionWorkspaceService(db, { pluginWorkerManager: s.workers, workspaceReaperCooldownDays: 0, now: () => new Date(clock) })
|
||||
.sweepTerminalWorkspaces(100);
|
||||
expect(s.callsFor(task.executionWorkspaceId!, "remove")).toHaveLength(0);
|
||||
expect(s.callsFor(task.executionWorkspaceId!, "prepare")).toHaveLength(2);
|
||||
const kept = await s.read(task.executionWorkspaceId!);
|
||||
expect(kept.status).not.toBe("archived");
|
||||
expect(kept.providerRef).toBeNull();
|
||||
expect(kept.metadata?.terminalCleanupRetry).toMatchObject({ failures: 1, retryAt: new Date(clock + 5 * MINUTE).toISOString(),
|
||||
failure: { kind: "provider", reason: "unsupported_operation", message: expect.not.stringContaining("private") } });
|
||||
});
|
||||
|
||||
it("does not retire a reservation whose task is still open", async () => {
|
||||
const s = await setup();
|
||||
const reservation = await s.abandonReservation();
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, s.f.runId));
|
||||
await expect(inspectPluginWorkspace(db, s.workers, reservation)).rejects.toThrow();
|
||||
expect(s.callsFor(reservation.id, "remove")).toHaveLength(0);
|
||||
expect(s.callsFor(reservation.id, "prepare")).toHaveLength(2);
|
||||
const kept = await s.read(reservation.id);
|
||||
expect(kept.status).toBe(reservation.status);
|
||||
expect(kept.metadata?.workspaceProviderRemoval).toBeUndefined();
|
||||
});
|
||||
|
||||
it("backs off and keeps the reason when the provider rejects an unbound removal", async () => {
|
||||
const s = await setup();
|
||||
const reservation = await s.abandonReservation();
|
||||
await s.finishTask();
|
||||
s.f.dirty();
|
||||
s.failures.unboundRemove = true;
|
||||
let clock = Date.now();
|
||||
const svc = executionWorkspaceService(db, { pluginWorkerManager: s.workers, workspaceReaperCooldownDays: 0, now: () => new Date(clock) });
|
||||
await svc.sweepTerminalWorkspaces(100);
|
||||
const refused = await s.read(reservation.id);
|
||||
expect(refused).toMatchObject({ status: "archived", metadata: { workspaceProviderRemoval: { operation: { binding: null } } } });
|
||||
expect(refused.metadata?.workspaceProviderRemoved).not.toBe(true);
|
||||
expect(refused.metadata?.terminalCleanupRetry).toMatchObject({ failures: 1, retryAt: new Date(clock + 5 * MINUTE).toISOString(),
|
||||
generation: refused.metadata?.lifecycleGeneration,
|
||||
failure: { kind: "provider", reason: "unsupported_operation", message: expect.not.stringContaining("private") } });
|
||||
clock = Date.now() + 1;
|
||||
await svc.sweepTerminalWorkspaces(100);
|
||||
expect(s.callsFor(reservation.id, "remove")).toHaveLength(1);
|
||||
expect(s.callsFor(reservation.id, "prepare")).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("backs off repeated inspection failures, records the reason, and resets after a success", async () => {
|
||||
const s = await setup();
|
||||
const workspace = (await preparePluginWorkspace(db, s.workers, s.f.input))!;
|
||||
await s.finishTask();
|
||||
s.f.dirty();
|
||||
s.failures.inspect = true;
|
||||
let clock = Date.now();
|
||||
const svc = executionWorkspaceService(db, { pluginWorkerManager: s.workers, workspaceReaperCooldownDays: 0, now: () => new Date(clock) });
|
||||
const retryAfter = async () => {
|
||||
const retry = (await s.read(workspace.id)).metadata?.terminalCleanupRetry as { failures: number; retryAt: string } | undefined;
|
||||
return retry && { failures: retry.failures, delay: Date.parse(retry.retryAt) - clock };
|
||||
};
|
||||
await svc.sweepTerminalWorkspaces(100);
|
||||
expect(await retryAfter()).toEqual({ failures: 1, delay: 5 * MINUTE });
|
||||
expect((await s.read(workspace.id)).metadata?.terminalCleanupRetry).toMatchObject({
|
||||
failure: { kind: "transport", reason: null, message: "Workspace storage operation is incomplete; retry the same workspace" } });
|
||||
expect((await svc.sweepTerminalWorkspaces(100)).skippedBackoff).toBeGreaterThanOrEqual(1);
|
||||
expect(s.callsFor(workspace.id, "inspect")).toHaveLength(1);
|
||||
clock += 5 * MINUTE;
|
||||
await svc.sweepTerminalWorkspaces(100);
|
||||
expect(await retryAfter()).toEqual({ failures: 2, delay: 10 * MINUTE });
|
||||
clock += 10 * MINUTE;
|
||||
await svc.sweepTerminalWorkspaces(100);
|
||||
expect(await retryAfter()).toEqual({ failures: 3, delay: 20 * MINUTE });
|
||||
svc.terminalWorkspaceWorkSettled();
|
||||
await svc.sweepTerminalWorkspaces(100);
|
||||
expect(s.callsFor(workspace.id, "inspect")).toHaveLength(3);
|
||||
const edited = await executionWorkspaceService(db).update(workspace.id, { metadata: { note: "operator edit" } });
|
||||
expect(edited!.metadata).toMatchObject({ note: "operator edit", terminalCleanupRetry: { failures: 3 } });
|
||||
clock += 20 * MINUTE;
|
||||
s.failures.inspect = false;
|
||||
await svc.sweepTerminalWorkspaces(100);
|
||||
expect(s.callsFor(workspace.id, "inspect")).toHaveLength(4);
|
||||
expect(await retryAfter()).toBeUndefined();
|
||||
s.failures.inspect = true;
|
||||
clock += MINUTE;
|
||||
await svc.sweepTerminalWorkspaces(100);
|
||||
expect(await retryAfter()).toEqual({ failures: 1, delay: 5 * MINUTE });
|
||||
});
|
||||
|
||||
it("drops a cleanup failure when the workspace moves to a new lifecycle before it is recorded", async () => {
|
||||
const s = await setup();
|
||||
const workspace = (await preparePluginWorkspace(db, s.workers, s.f.input))!;
|
||||
await s.finishTask();
|
||||
s.f.dirty();
|
||||
s.failures.inspect = true;
|
||||
s.failures.reopenFirst = true;
|
||||
await executionWorkspaceService(db, { pluginWorkerManager: s.workers, workspaceReaperCooldownDays: 0 }).sweepTerminalWorkspaces(100);
|
||||
expect(s.callsFor(workspace.id, "inspect")).toHaveLength(1);
|
||||
const moved = await s.read(workspace.id);
|
||||
expect(moved.metadata?.lifecycleGeneration).toBe(((workspace.metadata?.lifecycleGeneration as number) ?? 0) + 1);
|
||||
expect(moved.metadata).not.toHaveProperty("terminalCleanupRetry");
|
||||
});
|
||||
|
||||
it("backs off a pending storage removal that keeps failing", async () => {
|
||||
const s = await setup();
|
||||
const workspace = (await preparePluginWorkspace(db, s.workers, s.f.input))!;
|
||||
await s.finishTask();
|
||||
s.failures.remove = true;
|
||||
await expect(removePluginWorkspace(db, s.workers, workspace, "a".repeat(40))).rejects.toThrow();
|
||||
await db.update(executionWorkspaces).set({ status: "cleanup_failed", closedAt: new Date() }).where(eq(executionWorkspaces.id, workspace.id));
|
||||
let clock = Date.now();
|
||||
const svc = executionWorkspaceService(db, { pluginWorkerManager: s.workers, workspaceReaperCooldownDays: 0, now: () => new Date(clock) });
|
||||
const delays: number[] = [];
|
||||
for (let attempt = 0; attempt < 3; attempt++) {
|
||||
await svc.sweepTerminalWorkspaces(100);
|
||||
const retry = (await s.read(workspace.id)).metadata?.terminalCleanupRetry as { failures: number; retryAt: string };
|
||||
expect(retry.failures).toBe(attempt + 1);
|
||||
delays.push(Date.parse(retry.retryAt) - clock);
|
||||
clock = Date.parse(retry.retryAt);
|
||||
}
|
||||
expect(delays).toEqual([5 * MINUTE, 10 * MINUTE, 20 * MINUTE]);
|
||||
expect(s.callsFor(workspace.id, "remove")).toHaveLength(4);
|
||||
s.failures.remove = false;
|
||||
await svc.sweepTerminalWorkspaces(100);
|
||||
const removed = await s.read(workspace.id);
|
||||
expect(removed).toMatchObject({ status: "archived", metadata: { workspaceProviderRemoved: true } });
|
||||
expect(removed.metadata).not.toHaveProperty("terminalCleanupRetry");
|
||||
});
|
||||
});
|
||||
@@ -439,9 +439,11 @@ const support = await getEmbeddedPostgresTestSupport();
|
||||
return call(...args);
|
||||
},
|
||||
} as unknown as WorkspaceWorkers;
|
||||
let clock = Date.now();
|
||||
const svc = executionWorkspaceService(db, {
|
||||
pluginWorkerManager: workers,
|
||||
workspaceReaperCooldownDays: 0,
|
||||
now: () => new Date(clock),
|
||||
});
|
||||
const sweep = () => svc.sweepTerminalWorkspaces(100);
|
||||
expect(await sweep()).toMatchObject({ cleanupFailed: 1 });
|
||||
@@ -451,7 +453,10 @@ const support = await getEmbeddedPostgresTestSupport();
|
||||
.where(eq(executionWorkspaces.id, workspace!.id));
|
||||
expect(failed.status).toBe("cleanup_failed");
|
||||
expect(failed.metadata?.workspaceProviderRemoval).toBeUndefined();
|
||||
expect(await svc.nextTerminalWorkspaceSweepAt()).not.toBeNull();
|
||||
expect(failed.metadata?.terminalCleanupRetry).toMatchObject({ failures: 1, failure: { kind: "transport" },
|
||||
retryAt: new Date(clock + 300_000).toISOString() });
|
||||
expect(await sweep()).toMatchObject({ skippedBackoff: 1 });
|
||||
clock += 300_000;
|
||||
await sweep();
|
||||
const [removed] = await db
|
||||
.select()
|
||||
@@ -459,6 +464,7 @@ const support = await getEmbeddedPostgresTestSupport();
|
||||
.where(eq(executionWorkspaces.id, workspace!.id));
|
||||
expect(removed.metadata?.workspaceProviderRemoved).toBe(true);
|
||||
expect(removed.status).toBe("archived");
|
||||
expect(removed.metadata).not.toHaveProperty("terminalCleanupRetry");
|
||||
});
|
||||
it("prepares a durable reference without a local checkout or agent allocation", async () => {
|
||||
const f = await fixture();
|
||||
|
||||
@@ -39,7 +39,11 @@ import { authorizationService, executionWorkspaceReadSqlCondition, projectReadSq
|
||||
import type { SQL } from "drizzle-orm";
|
||||
import { nativeSessionProviderEvidence } from "./native-runtime/native-session-resume.js";
|
||||
import { hasWorkspaceRestoreFailure } from "@paperclipai/shared";
|
||||
import { notifyTerminalWorkspaceWork, TERMINAL_WORKSPACE_RETRY_MS } from "./terminal-workspace-work.js";
|
||||
import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js";
|
||||
import {
|
||||
clearTerminalWorkspaceRetry, readTerminalWorkspaceRetry, recordTerminalWorkspaceFailure, terminalWorkspaceFailure,
|
||||
TERMINAL_WORKSPACE_RETRY_KEY, type TerminalWorkspaceFailure,
|
||||
} from "./terminal-workspace-retry.js";
|
||||
import {
|
||||
deleteWorkspaceCompactionTrash,
|
||||
findRegenerableWorkspaceDirs,
|
||||
@@ -296,10 +300,10 @@ function preserveWorkspaceMetadataForPatch(
|
||||
}
|
||||
}
|
||||
}
|
||||
const freshIntentKeys = ["workspaceProviderFresh", "runnerTaskFreshWorkspace", "runnerTaskWorkspaceOwner"] as const;
|
||||
if (patch.metadata !== undefined && (patch.metadata !== null || freshIntentKeys.some(key => current.metadata?.[key] !== undefined))) {
|
||||
const coreOwnedKeys = ["workspaceProviderFresh", "runnerTaskFreshWorkspace", "runnerTaskWorkspaceOwner", TERMINAL_WORKSPACE_RETRY_KEY] as const;
|
||||
if (patch.metadata !== undefined && (patch.metadata !== null || coreOwnedKeys.some(key => current.metadata?.[key] !== undefined))) {
|
||||
nextPatch.metadata = { ...(nextPatch.metadata ?? patch.metadata) };
|
||||
for (const key of freshIntentKeys) {
|
||||
for (const key of coreOwnedKeys) {
|
||||
if (current.metadata?.[key] !== undefined) nextPatch.metadata[key] = current.metadata[key];
|
||||
else delete nextPatch.metadata[key];
|
||||
}
|
||||
@@ -1516,7 +1520,10 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
|
||||
if (!row) throw notFound("Workspace is unavailable");
|
||||
let state;
|
||||
try { state = await inspectPluginWorkspace(db, opts.pluginWorkerManager, row, recoveryPullRequest); }
|
||||
catch { return { statusInspectionSucceeded: false, warnings: ["Workspace storage inspection is unavailable; data was retained."], git: null }; }
|
||||
catch (error) {
|
||||
const failure = terminalWorkspaceFailure(error);
|
||||
return { statusInspectionSucceeded: false, warnings: [`Workspace storage inspection is unavailable; data was retained. ${failure.message}`], git: null, failure };
|
||||
}
|
||||
return { statusInspectionSucceeded: true, warnings: [], git: {
|
||||
repoRoot: `/workspaces/${workspace.id}`, workspacePath: `/workspaces/${workspace.id}`,
|
||||
branchName: state.branchName, baseRef: workspace.baseRef,
|
||||
@@ -3614,7 +3621,21 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
|
||||
compacted: 0,
|
||||
compactionFailed: 0,
|
||||
};
|
||||
const readinessByWorkingTree = new Map<string, ReturnType<typeof inspectGitCloseReadiness>>();
|
||||
const readinessByWorkingTree = new Map<string, ReturnType<typeof inspectWorkspace>>();
|
||||
const backOffTerminalWorkspace = async (snapshot: ExecutionWorkspaceRow, failure: TerminalWorkspaceFailure) => {
|
||||
const [target] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, snapshot.id));
|
||||
if (!target) return;
|
||||
const generation = readExecutionWorkspaceLifecycleGeneration(target.metadata);
|
||||
const removalFence = (target.metadata?.workspaceProviderRemoval as { generation?: unknown } | null | undefined)?.generation === generation;
|
||||
if (generation !== readExecutionWorkspaceLifecycleGeneration(snapshot.metadata) && !removalFence) return;
|
||||
const retry = await recordTerminalWorkspaceFailure(db, target, generation, failure, now());
|
||||
terminalWorkspaceDeadlines.set(target.id, Date.parse(retry.retryAt));
|
||||
logger.warn({ event: "execution_workspace.terminal_cleanup_retry", executionWorkspaceId: target.id, sourceIssueId: target.sourceIssueId,
|
||||
failures: retry.failures, retryAt: retry.retryAt, failure: retry.failure }, "terminal workspace cleanup failed; a later sweep retries");
|
||||
};
|
||||
const settleTerminalWorkspaceRetry = async (target: ExecutionWorkspaceRow) => {
|
||||
if (target.metadata?.[TERMINAL_WORKSPACE_RETRY_KEY] !== undefined) await clearTerminalWorkspaceRetry(db, target.id);
|
||||
};
|
||||
let taskDirectoryBudget = opts.taskDirectoryRetention?.perSweep ?? 8;
|
||||
const expireTaskDirectory = async (workspace: ExecutionWorkspaceRow, notBefore: Date) => {
|
||||
const settings = opts.taskDirectoryRetention!;
|
||||
@@ -3845,6 +3866,12 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
|
||||
result.skippedCooldown += 1;
|
||||
continue;
|
||||
}
|
||||
const terminalRetry = readTerminalWorkspaceRetry(workspace.metadata, readExecutionWorkspaceLifecycleGeneration(workspace.metadata));
|
||||
if (terminalRetry && Date.parse(terminalRetry.retryAt) > now().getTime()) {
|
||||
terminalWorkspaceDeadlines.set(workspace.id, Date.parse(terminalRetry.retryAt));
|
||||
result.skippedBackoff += 1;
|
||||
continue;
|
||||
}
|
||||
if (await workspaceHasActiveRun(workspace)) {
|
||||
if (reopenPending) result.skippedReopened += 1;
|
||||
else result.skippedActiveRun += 1;
|
||||
@@ -3856,10 +3883,10 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
|
||||
const cleanup = await cleanupTerminalWorkspace(workspace, null, generation);
|
||||
if (cleanup.skippedReopened) result.skippedReopened += 1;
|
||||
else if (!cleanup.cleaned) result.cleanupFailed += 1;
|
||||
else { result.archived += 1; terminalWorkspaceDeadlines.delete(workspace.id); }
|
||||
else { result.archived += 1; terminalWorkspaceDeadlines.delete(workspace.id); await settleTerminalWorkspaceRetry(workspace); }
|
||||
} catch (error) {
|
||||
result.cleanupFailed += 1;
|
||||
terminalWorkspaceDeadlines.set(workspace.id, now().getTime() + TERMINAL_WORKSPACE_RETRY_MS);
|
||||
await backOffTerminalWorkspace(workspace, terminalWorkspaceFailure(error));
|
||||
await markTerminalCleanupFailedFenced({ workspaceId: workspace.id, capturedGeneration: generation,
|
||||
cleanupReason: `${ISSUE_TERMINAL_WORKSPACE_CLEANUP_REASON} | ${error instanceof Error ? error.message : String(error)}` });
|
||||
}
|
||||
@@ -3966,12 +3993,20 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
|
||||
inspection = inspectWorkspace(toExecutionWorkspace(workspace), recoveryPullRequest);
|
||||
readinessByWorkingTree.set(readinessKey, inspection);
|
||||
}
|
||||
const { git, statusInspectionSucceeded } = await inspection;
|
||||
const inspected = await inspection;
|
||||
const { git, statusInspectionSucceeded } = inspected;
|
||||
if (!statusInspectionSucceeded) {
|
||||
if (remoteWorkspace) terminalWorkspaceDeadlines.delete(workspace.id);
|
||||
if ("failure" in inspected) await backOffTerminalWorkspace(workspace, inspected.failure);
|
||||
result.skippedUndelivered += 1;
|
||||
continue;
|
||||
}
|
||||
if (workspace.metadata?.[TERMINAL_WORKSPACE_RETRY_KEY] !== undefined) {
|
||||
await settleTerminalWorkspaceRetry(workspace);
|
||||
const metadata = { ...workspace.metadata };
|
||||
delete metadata[TERMINAL_WORKSPACE_RETRY_KEY];
|
||||
workspace = { ...workspace, metadata };
|
||||
}
|
||||
if (deliveryCandidate && (git?.headSha !== deliveryCandidate.headSha || git.branchName !== workspace.branchName
|
||||
|| (await workspaceDeliveryCandidate(workspace))?.operationId !== deliveryCandidate.operationId)) {
|
||||
result.skippedUndelivered += 1;
|
||||
@@ -4089,10 +4124,10 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
|
||||
const cleanup = await cleanupTerminalWorkspace(workspace, assessment.workspaceHeadSha, generation, recoveryPullRequest, verifyDeliveryBeforeRemove);
|
||||
if (cleanup.skippedReopened) result.skippedReopened += 1;
|
||||
else if (!cleanup.cleaned) result.cleanupFailed += 1;
|
||||
else { result.archived += 1; terminalWorkspaceDeadlines.delete(workspace.id); }
|
||||
else { result.archived += 1; terminalWorkspaceDeadlines.delete(workspace.id); await settleTerminalWorkspaceRetry(workspace); }
|
||||
} catch (error) {
|
||||
result.cleanupFailed += 1;
|
||||
terminalWorkspaceDeadlines.set(workspace.id, now().getTime() + TERMINAL_WORKSPACE_RETRY_MS);
|
||||
await backOffTerminalWorkspace(workspace, terminalWorkspaceFailure(error));
|
||||
await markTerminalCleanupFailedFenced({ workspaceId: workspace.id, capturedGeneration: generation,
|
||||
cleanupReason: `${ISSUE_TERMINAL_WORKSPACE_CLEANUP_REASON} | ${error instanceof Error ? error.message : String(error)}` });
|
||||
logger.warn({ err: error, executionWorkspaceId: workspace.id }, "workspace storage cleanup remains pending");
|
||||
@@ -4225,10 +4260,10 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
|
||||
const cleanup = await cleanupTerminalWorkspace(archived, assessment.workspaceHeadSha, capturedGeneration, recoveryPullRequest, verifyDeliveryBeforeRemove);
|
||||
if (cleanup.skippedReopened) result.skippedReopened += 1;
|
||||
else if (!cleanup.cleaned) result.cleanupFailed += 1;
|
||||
else result.archived += 1;
|
||||
else { result.archived += 1; await settleTerminalWorkspaceRetry(archived); }
|
||||
} catch (error) {
|
||||
result.cleanupFailed += 1;
|
||||
if (archived.providerType === "plugin" || remoteWorkspace) terminalWorkspaceDeadlines.set(archived.id, now().getTime() + TERMINAL_WORKSPACE_RETRY_MS);
|
||||
if (archived.providerType === "plugin" || remoteWorkspace) await backOffTerminalWorkspace(archived, terminalWorkspaceFailure(error));
|
||||
const failure = error instanceof Error ? error.message : String(error);
|
||||
// Mark cleanup_failed only while the row is still closed at the
|
||||
// generation this sweep captured. A reopen that raced after the failure
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { and, eq, isNull, sql } from "drizzle-orm";
|
||||
import { executionWorkspaces, heartbeatRuns, issues, type Db } from "@paperclipai/db";
|
||||
import { workspaceProviderOperationSchema } from "@paperclipai/plugin-sdk";
|
||||
import { logger } from "../middleware/logger.js";
|
||||
import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js";
|
||||
|
||||
type Workspace = typeof executionWorkspaces.$inferSelect;
|
||||
|
||||
export async function retireAbandonedPluginWorkspaceReservation(db: Db, workspace: Workspace): Promise<Workspace | null> {
|
||||
if (!workspace.sourceIssueId) return null;
|
||||
const previous = typeof workspace.metadata?.lifecycleGeneration === "number" ? workspace.metadata.lifecycleGeneration : 0;
|
||||
const generation = previous + 1;
|
||||
const operation = workspaceProviderOperationSchema.parse({ kind: "remove", binding: null, expectedHeadSha: null });
|
||||
const operationId = createHash("sha256").update(JSON.stringify([workspace.id, generation, operation])).digest("hex");
|
||||
const abandoned = and(
|
||||
eq(executionWorkspaces.id, workspace.id),
|
||||
eq(executionWorkspaces.companyId, workspace.companyId),
|
||||
isNull(executionWorkspaces.providerRef),
|
||||
sql`coalesce((${executionWorkspaces.metadata}->>'lifecycleGeneration')::int, 0) = ${previous}`,
|
||||
sql`EXISTS (SELECT 1 FROM ${issues} WHERE ${issues.companyId} = ${workspace.companyId}
|
||||
AND ${issues.id} = ${workspace.sourceIssueId} AND ${issues.status} IN ('done', 'cancelled'))`,
|
||||
sql`NOT EXISTS (SELECT 1 FROM ${issues} WHERE ${issues.companyId} = ${workspace.companyId}
|
||||
AND ${issues.executionWorkspaceId} = ${workspace.id})`,
|
||||
sql`NOT EXISTS (SELECT 1 FROM ${heartbeatRuns} WHERE ${heartbeatRuns.companyId} = ${workspace.companyId}
|
||||
AND ${heartbeatRuns.issueId} = ${workspace.sourceIssueId} AND ${heartbeatRuns.status} IN ('queued', 'running'))`,
|
||||
);
|
||||
const [candidate] = await db.select({ id: executionWorkspaces.id }).from(executionWorkspaces).where(abandoned);
|
||||
if (!candidate) return null;
|
||||
const closedAt = new Date();
|
||||
return db.transaction(async (tx) => {
|
||||
await notifyTerminalWorkspaceWork(tx, `workspace:${workspace.id}`);
|
||||
const [retired] = await tx.update(executionWorkspaces).set({
|
||||
status: "archived",
|
||||
closedAt: workspace.closedAt ?? closedAt,
|
||||
cleanupEligibleAt: workspace.cleanupEligibleAt ?? closedAt,
|
||||
cleanupReason: "issue_terminal",
|
||||
updatedAt: closedAt,
|
||||
metadata: { ...workspace.metadata, lifecycleGeneration: generation, workspaceProviderRemoval: { generation, operation, operationId } },
|
||||
}).where(abandoned).returning();
|
||||
if (!retired) return null;
|
||||
logger.info({ event: "execution_workspace.reservation_retired", executionWorkspaceId: workspace.id, sourceIssueId: workspace.sourceIssueId },
|
||||
"Unused workspace storage reservation is being removed");
|
||||
return retired;
|
||||
});
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
|
||||
import { effectiveProjectWorkspacePolicy, isWorkspaceProvider } from "./project-workspace-provider.js";
|
||||
import { releaseProviderRecoveryTransfers, stageProviderFreshWorkspace } from "./provider-fresh-workspace.js";
|
||||
import { retireAbandonedPluginWorkspaceReservation } from "./plugin-workspace-reservations.js";
|
||||
import { randomUUID, createHash } from "node:crypto";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import {
|
||||
@@ -557,8 +558,11 @@ async function reconcilePluginWorkspaceReservation(
|
||||
),
|
||||
);
|
||||
if (!workspace) throw conflict("Workspace is unavailable");
|
||||
if (readWorkspaceProviderBinding(workspace.metadata)) return workspace;
|
||||
if (readWorkspaceProviderBinding(workspace.metadata) || workspace.metadata?.workspaceProviderRemoval ||
|
||||
workspace.metadata?.workspaceProviderRemoved === true) return workspace;
|
||||
const owner = workspace.metadata?.workspaceProviderOwner;
|
||||
const retired = typeof owner === "string" ? await retireAbandonedPluginWorkspaceReservation(db, workspace) : null;
|
||||
if (retired) return retired;
|
||||
if (typeof owner !== "string" || !workspace.metadata?.workspaceProviderSource)
|
||||
throw conflict("Workspace storage reservation is unavailable");
|
||||
const operation = workspaceProviderOperationSchema.parse({
|
||||
@@ -765,8 +769,10 @@ async function finishPendingRemoval(
|
||||
const operationId = createHash("sha256").update(JSON.stringify([workspace.id, pending.generation, parsed.data])).digest("hex");
|
||||
if (pending.operationId !== undefined && pending.operationId !== operationId)
|
||||
throw conflict("Workspace removal identity does not match its recorded payload");
|
||||
const owner = workspace.metadata?.workspaceProviderOwner;
|
||||
const result = await callWorkspaceProvider(db, workers, {
|
||||
workspace,
|
||||
pluginId: typeof owner === "string" ? owner : undefined,
|
||||
operation: parsed.data,
|
||||
removalOperationId: operationId,
|
||||
});
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { executionWorkspaces, type Db } from "@paperclipai/db";
|
||||
import { HttpError } from "../errors.js";
|
||||
import { findWorkspaceProviderFailure } from "./workspace-provider-failure.js";
|
||||
import { TERMINAL_WORKSPACE_RETRY_MS } from "./terminal-workspace-work.js";
|
||||
|
||||
export const TERMINAL_WORKSPACE_RETRY_KEY = "terminalCleanupRetry";
|
||||
export const TERMINAL_WORKSPACE_RETRY_MAX_MS = 4 * 60 * 60 * 1000;
|
||||
|
||||
export type TerminalWorkspaceFailure = { kind: string | null; reason: string | null; message: string };
|
||||
export type TerminalWorkspaceRetry = {
|
||||
generation: number;
|
||||
failures: number;
|
||||
failedAt: string;
|
||||
retryAt: string;
|
||||
failure: TerminalWorkspaceFailure;
|
||||
};
|
||||
|
||||
export function terminalWorkspaceFailure(error: unknown): TerminalWorkspaceFailure {
|
||||
const provider = findWorkspaceProviderFailure(error);
|
||||
const message = error instanceof HttpError ? error.message : "Workspace storage operation failed";
|
||||
return { kind: provider?.kind ?? null, reason: provider?.reason ?? null, message: message.slice(0, 500) };
|
||||
}
|
||||
|
||||
export function terminalWorkspaceRetryDelayMs(failures: number) {
|
||||
return Math.min(TERMINAL_WORKSPACE_RETRY_MS * 2 ** Math.min(Math.max(failures - 1, 0), 16), TERMINAL_WORKSPACE_RETRY_MAX_MS);
|
||||
}
|
||||
|
||||
export function readTerminalWorkspaceRetry(metadata: Record<string, unknown> | null | undefined, generation: number) {
|
||||
const retry = metadata?.[TERMINAL_WORKSPACE_RETRY_KEY] as Partial<TerminalWorkspaceRetry> | undefined;
|
||||
return retry && typeof retry === "object" && retry.generation === generation && Number.isInteger(retry.failures)
|
||||
&& typeof retry.failedAt === "string" && typeof retry.retryAt === "string" ? retry as TerminalWorkspaceRetry : null;
|
||||
}
|
||||
|
||||
export async function recordTerminalWorkspaceFailure(db: Db, workspace: { id: string; metadata: Record<string, unknown> | null },
|
||||
generation: number, failure: TerminalWorkspaceFailure, now: Date): Promise<TerminalWorkspaceRetry> {
|
||||
const failures = (readTerminalWorkspaceRetry(workspace.metadata, generation)?.failures ?? 0) + 1;
|
||||
const retry = { generation, failures, failedAt: now.toISOString(),
|
||||
retryAt: new Date(now.getTime() + terminalWorkspaceRetryDelayMs(failures)).toISOString(), failure };
|
||||
await db.update(executionWorkspaces)
|
||||
.set({ metadata: sql`coalesce(${executionWorkspaces.metadata}, '{}'::jsonb) || jsonb_build_object(${TERMINAL_WORKSPACE_RETRY_KEY}::text, ${JSON.stringify(retry)}::jsonb)` })
|
||||
.where(eq(executionWorkspaces.id, workspace.id));
|
||||
return retry;
|
||||
}
|
||||
|
||||
export async function clearTerminalWorkspaceRetry(db: Db, workspaceId: string) {
|
||||
await db.update(executionWorkspaces)
|
||||
.set({ metadata: sql`${executionWorkspaces.metadata} - ${TERMINAL_WORKSPACE_RETRY_KEY}::text` })
|
||||
.where(and(eq(executionWorkspaces.id, workspaceId), sql`${executionWorkspaces.metadata} -> ${TERMINAL_WORKSPACE_RETRY_KEY}::text IS NOT NULL`));
|
||||
}
|
||||
Reference in new issue
Block a user