diff --git a/doc/native-workspace-finalization-recovery.md b/doc/native-workspace-finalization-recovery.md index 33208b6d02a..1628394d481 100644 --- a/doc/native-workspace-finalization-recovery.md +++ b/doc/native-workspace-finalization-recovery.md @@ -306,6 +306,12 @@ extracted. No host path or link target is copied into the informational run even ## Repairing a failed workspace export without rerunning the agent +The background workspace worker runs up to four copy-backs at once, one per +task, so a slow sandbox export does not hold every other task's retry. Due work +runs oldest first. Work without a saved retry time, such as a copy-back that a +restart interrupted, is ordered by when it became pending; it does not go ahead +of older retries that are already due. + Transient export failures retry three times, then produce `native_workspace_sync_out_retry_exhausted`. The accepted result stays saved, and the exact sandbox is stopped and retained for export-only recovery. The diff --git a/server/src/__tests__/native-workspace-finalization-work.test.ts b/server/src/__tests__/native-workspace-finalization-work.test.ts index 0328f6bf1ee..db3ed355af0 100644 --- a/server/src/__tests__/native-workspace-finalization-work.test.ts +++ b/server/src/__tests__/native-workspace-finalization-work.test.ts @@ -13,6 +13,7 @@ import { agents, companies, completionContracts, createDb, heartbeatRuns, issues import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; import { issueService } from "../services/issues.js"; import { nativeWorkspaceFinalizationWork } from "../services/native-runtime/native-workspace-finalization-work.js"; +import { attemptNativeWorkspaceFinalization } from "../services/native-runtime/native-workspace-finalizer.js"; import { getRunExecutionTraceContext, measureRunPhase } from "../services/execution-preparation-timing.js"; import { withNativeWorkspaceFinalizationOwnership } from "../services/native-runtime/native-workspace-finalization-ownership.js"; import { resumeNativeWorkspaceAfterOwnerStop } from "../services/native-runtime/native-workspace-owner-recovery.js"; @@ -239,6 +240,48 @@ describe("workspace finalization scheduled lifecycle", () => { await vi.advanceTimersByTimeAsync(30_000); await flush(); expect(await successful(binding.runId)).toHaveLength(1); await quiet(); }); + it("retries another task while an older copy-back is still running", async () => { + const slow = await seed(); await complete(slow); + const fast = await seed(); await complete(fast); + // Both retries are overdue; the slow task is older, so it is selected first. + for (const [binding, ageMs] of [[slow, 120_000], [fast, 60_000]] as const) { + await db.update(nativeRunFinalizations).set({ phase: "retryable_failure", nextAttemptAt: new Date(Date.now() - ageMs) }) + .where(eq(nativeRunFinalizations.runId, binding.runId)); + } + let release!: () => void; + const blocked = new Promise(resolve => { release = resolve; }); + const attempt = vi.fn(async (input: Parameters[0]) => { + if (input.runId === slow.runId) await blocked; + return attemptNativeWorkspaceFinalization(input); + }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted, + onError: error => errors.push(error) }); + worker = nativeWorkspaceFinalizationWork(db, { attempt, onWorkspaceSettled: settled }).register(coordinator, () => admitted); + try { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await successful(fast.runId)).toHaveLength(1); + }, { timeout: 15_000 }); + expect(await successful(slow.runId)).toHaveLength(0); + } finally { release(); } + await worker.ready; await flush(); + expect(await successful(slow.runId)).toHaveLength(1); await quiet(); + }); + it("orders work without a saved retry behind older overdue retries", async () => { + const overdue = await seed(); await complete(overdue); + const failedAt = new Date(Date.now() - 60 * 60_000); + await db.update(nativeRunFinalizations).set({ phase: "retryable_failure", + nextAttemptAt: new Date(failedAt.getTime() + 30_000), updatedAt: failedAt }) + .where(eq(nativeRunFinalizations.runId, overdue.runId)); + // A copy-back interrupted by a later restart has no saved retry deadline. + const interrupted = await seed(); await complete(interrupted); + const entries = await nativeWorkspaceFinalizationWork(db, { onWorkspaceSettled: settled }).load({ companyId }); + const pending = entries.find(item => item.id === interrupted.runId)!; + expect(pending.dueAt).toBeGreaterThan(failedAt.getTime()); + expect(pending.dueAt).toBeLessThanOrEqual(Date.now()); + expect([...entries].sort((a, b) => a.dueAt - b.dueAt).map(item => item.id)) + .toEqual([overdue.runId, interrupted.runId]); + }); it("waits quietly for the actual local owner to release", async () => { const binding = await seed(); await complete(binding); await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: binding.runId }, async () => { diff --git a/server/src/services/native-runtime/native-workspace-finalization-work.ts b/server/src/services/native-runtime/native-workspace-finalization-work.ts index cc73e11b46f..c9d2413a8a7 100644 --- a/server/src/services/native-runtime/native-workspace-finalization-work.ts +++ b/server/src/services/native-runtime/native-workspace-finalization-work.ts @@ -24,6 +24,7 @@ export function nativeWorkspaceFinalizationWork(db: Db, options: { return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, issueId: nativeRunFinalizations.issueId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt, leaseOwner: nativeRunFinalizations.leaseOwner, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, + updatedAt: nativeRunFinalizations.updatedAt, }).from(nativeRunFinalizations) .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId))) @@ -45,8 +46,13 @@ export function nativeWorkspaceFinalizationWork(db: Db, options: { )).orderBy(asc(nativeRunFinalizations.runId)).limit(100); } function entry(row: Awaited>[number]): Entry { + // Work without a saved retry is due now. Order it by when it became pending, + // so copy-backs interrupted by a later restart cannot starve older retries. + // The database clock may lead this process slightly; never defer such work. + const pendingSince = Math.min(row.updatedAt.getTime(), Date.now()); return { id: row.id, companyId: row.companyId, issueId: row.issueId, agentId: row.agentId, - dueAt: Math.max(row.nextAttemptAt?.getTime() ?? 0, row.leaseOwner ? row.leaseExpiresAt?.getTime() ?? 0 : 0) }; + dueAt: Math.max(row.nextAttemptAt?.getTime() ?? pendingSince, + row.leaseOwner ? row.leaseExpiresAt?.getTime() ?? 0 : 0) }; } async function load(scope?: NativeWorkspaceFinalizationScope, signal?: AbortSignal, canRun = () => true): Promise { const entries: Entry[] = []; @@ -90,7 +96,10 @@ export function nativeWorkspaceFinalizationWork(db: Db, options: { function register(coordinator: ReturnType, canRun: () => boolean, canRecover: () => Promise = async () => true) { const worker = registerScopedDeadlineWork(coordinator, { - queue: DELIVERY_QUEUES.nativeWorkspaceFinalization, batchSize: 1, + queue: DELIVERY_QUEUES.nativeWorkspaceFinalization, + // A slow sandbox export can take many minutes. Independent tasks proceed + // in parallel; runs that share a task keep their copy-backs serialized. + concurrentChecks: { limit: 4, key: (item: Entry) => `issue:${item.companyId}:${item.issueId}` }, decodeScope: (hint: string): NativeWorkspaceFinalizationScope | undefined => hint === "*" ? undefined : JSON.parse(hint), matches: (item: Entry, scope?: NativeWorkspaceFinalizationScope) => !scope || item.companyId === scope.companyId && (!scope.issueId || item.issueId === scope.issueId) && (!scope.runId || item.id === scope.runId),