From 482ec0aea9ab6f2684baaa1487f2c7a2eedf009e Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 4 Nov 2026 19:26:02 -0600 Subject: [PATCH] fix(native): run saved copy-back retries on schedule behind a slow export (#19145) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Native runs copy their sandbox workspace back after the provider finishes. A failed copy-back saves a retry 30 seconds later, with a bound of 3 attempts. > - One shared delivery queue runs these retries. A pass of that queue ends only when its slowest check ends. > - A legitimate sandbox export can take 10+ minutes. While it runs, all other due retries wait. > - On staging, a run active across a deploy showed "Reconnecting" with an overdue retry for about 24 minutes. Its retries ran at +24 min and +3.5 min instead of +30 s. It settled only after about 32 minutes. > - This pull request lets idle check slots admit entries that come due during a pass. It does not overlap a running check's resource and does not start work after abort or drain. > - The benefit is that the existing 3-attempt bound settles a failing copy-back in about 1 minute, not after the slowest unrelated export. ## Linked Issues or Issue Description **What happened?** A native run whose result is durable but whose workspace copy-back failed stays in `running` with the label "Reconnecting" for a long time. Its saved retry time is in the past. The retry does not run until an unrelated slow copy-back in the same queue pass finishes. **Steps to reproduce** 1. Start two native runs on different tasks with accepted results. 2. Let the copy-back of run A take many minutes. 3. Let the copy-back of run B fail. It saves a retry 30 seconds later. 4. Watch run B. Its retry does not run until run A's copy-back ends. **Expected behavior** Run B retries on its saved schedule. After 3 failed attempts it settles with `native_workspace_sync_out_retry_exhausted` and a recovery action, while run A continues. **Paperclip version or commit** master at a0c0851766. Observed on a staging deployment of 69dea878. **Agent adapter(s) involved** Not adapter-specific (core bug). Observed with the Paperclip runner. **Relevant logs or output** In-run copy-back failed at 23:44:57 (retry saved for 23:45:28). The deploy restarted the server at 23:49:45. One queue check exported another task's workspace from 23:53:54 to 00:06:15. The overdue retry ran at 00:09:06, then 00:12:39, and the run settled at 00:12:40 with `native_workspace_sync_out_retry_exhausted`. About 20 other runs followed the same timeline. Their failures at 00:12–00:13 are this same retry exhaustion. They do not come from a reattach sweep. ## What Changed - `registerScopedDeadlineWork` has a new opt-in `concurrentChecks.admitDueWhileActive` option. While a pass still has a running check, idle slots admit entries that are due (saved deadline, failure backoff, and minimum spacing). They wait for the next deadline or the next check completion. - Admission skips entries owned by the pass and entries whose resource keys a running check holds. It stops on abort or when `canRun()` is false. The pass still ends only when all running checks end. - The native workspace finalization queue opts in. Its `check` already rereads the durable coordinator row before it acts. - Regression tests: a real-database test in `native-workspace-finalization-work.test.ts`, plus two helper tests (resource-key safety and shutdown). ## Verification - `cd server && npx vitest run src/__tests__/native-workspace-finalization-work.test.ts src/__tests__/scoped-deadline-work.test.ts src/__tests__/delivery-work-coordinator.test.ts` passes (68 tests). - Red before the fix: "keeps another task's saved retries on schedule and settles it while a slow copy-back is running" failed because `onWorkspaceSettled` was never called for the failing run. "admits saved deadlines that come due behind a slow check without overlapping its resource" failed too. - Green after the fix: the failing run makes exactly 3 attempts, settles as `terminal_failure` / `native_workspace_sync_out_retry_exhausted`, and the slow copy-back is admitted once and then succeeds. - `tsc --noEmit` shows no errors in the changed files. - Not done: a live staging check. This change is not deployed. ## Risks - Low to medium. Only the workspace finalization queue opts in. Other queues keep the current behavior. - An admitted entry is not reconstructed first. The finalization `check` rereads durable state, so a stale in-memory entry cannot start a second export. The advisory lock and owner receipt still prevent two copy-backs for one run. - This change does not fix the cause of the copy-back failures on staging (`native_session_lease_lost` in-run, then `workspace_sync_out_failed` with no detail). That needs a separate change. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code execution in Claude Code. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- ...native-workspace-finalization-work.test.ts | 38 ++++ .../__tests__/scoped-deadline-work.test.ts | 163 ++++++++++++++++++ .../native-workspace-finalization-work.ts | 4 +- server/src/services/scoped-deadline-work.ts | 124 +++++++++++-- 4 files changed, 314 insertions(+), 15 deletions(-) diff --git a/server/src/__tests__/native-workspace-finalization-work.test.ts b/server/src/__tests__/native-workspace-finalization-work.test.ts index db3ed355af0..724a2bce425 100644 --- a/server/src/__tests__/native-workspace-finalization-work.test.ts +++ b/server/src/__tests__/native-workspace-finalization-work.test.ts @@ -267,6 +267,44 @@ describe("workspace finalization scheduled lifecycle", () => { await worker.ready; await flush(); expect(await successful(slow.runId)).toHaveLength(1); await quiet(); }); + it("keeps another task's saved retries on schedule and settles it while a slow copy-back is running", async () => { + const slow = await seed(); await complete(slow); + const failing = await seed(); + await db.update(workspaceOperations).set({ cwd: join(root, "absent") }).where(eq(workspaceOperations.heartbeatRunId, failing.runId)); + await complete(failing); + let release!: () => void; + const blocked = new Promise(resolve => { release = resolve; }); + const attempts: string[] = []; + const attempt = vi.fn(async (input: Parameters[0]) => { + attempts.push(input.runId); + 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 { + // Each failed export saves a 30-second retry. A sandbox export of another + // task can legitimately take many minutes; it must not hold those retries + // until it finishes, or the failing run stays retryable with an overdue deadline. + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(5000); + expect(settled).toHaveBeenCalledWith(expect.objectContaining({ runId: failing.runId, succeeded: false })); + }, { timeout: 10_000 }); + const [exhausted] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, failing.runId)); + expect(exhausted.phase).toBe("terminal_failure"); + expect(exhausted.failureCode).toBe("native_workspace_sync_out_retry_exhausted"); + expect(attempts.filter(id => id === failing.runId)).toHaveLength(3); + // The slow copy-back was admitted once and is still the only owner of its task. + expect(attempts.filter(id => id === slow.runId)).toHaveLength(1); + expect(await successful(slow.runId)).toHaveLength(0); + } finally { release(); } + await worker.ready; await flush(); + expect(await successful(slow.runId)).toHaveLength(1); + expect(attempts.filter(id => id === slow.runId)).toHaveLength(1); + expect(attempts.filter(id => id === failing.runId)).toHaveLength(3); + await quiet(); + }, 30_000); 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); diff --git a/server/src/__tests__/scoped-deadline-work.test.ts b/server/src/__tests__/scoped-deadline-work.test.ts index 94e29b05da3..8352ea3bedf 100644 --- a/server/src/__tests__/scoped-deadline-work.test.ts +++ b/server/src/__tests__/scoped-deadline-work.test.ts @@ -415,3 +415,166 @@ it("coalesces broad reconstruction without delaying an exact object hint or save await vi.advanceTimersByTimeAsync(120_000); expect(load).toHaveBeenCalledTimes(4); expect(check).toHaveBeenCalledOnce(); }); + +it("admits saved deadlines that come due behind a slow check without overlapping its resource", async () => { + vi.useFakeTimers(); + const onError = vi.fn(); + const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []), transaction: vi.fn(async () => undefined) }); + coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError }); + const now = Date.now(); + const started: string[] = []; + let releaseSlow!: () => void; + const slow = new Promise(resolve => { releaseSlow = resolve; }); + const retries = new Map(); + const check = vi.fn(async (item: Entry, signal?: AbortSignal) => { + started.push(item.id); + if (item.id === "slow") { await slow; return null; } + // A failed attempt saves a retry 30 seconds later, three attempts in all. + const count = (retries.get(item.id) ?? 0) + 1; + retries.set(item.id, count); + expect(signal?.aborted).toBe(false); + return count < 3 ? { ...item, dueAt: Date.now() + 30_000 } : null; + }); + const worker = registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.nativeResponsePresentation, + concurrentChecks: { limit: 4, key: (item: Entry) => item.issueId, admitDueWhileActive: true }, + decodeScope: () => undefined, + matches: () => true, + load: async () => [ + { id: "slow", issueId: "shared", companyId: "company", dueAt: now }, + { id: "retrying", issueId: "other", companyId: "company", dueAt: now }, + // Same resource as the slow check: it must stay queued behind it. + { id: "sibling", issueId: "shared", companyId: "company", dueAt: now + 10_000 }, + ], + check, + }, () => true); + try { + await vi.advanceTimersByTimeAsync(1); + expect(new Set(started)).toEqual(new Set(["slow", "retrying"])); + await vi.advanceTimersByTimeAsync(30_000); + expect(retries.get("retrying")).toBe(2); + await vi.advanceTimersByTimeAsync(30_000); + expect(retries.get("retrying")).toBe(3); + await vi.advanceTimersByTimeAsync(120_000); + expect(started).not.toContain("sibling"); + releaseSlow(); + await vi.advanceTimersByTimeAsync(1); + expect(started.filter(id => id === "sibling")).toHaveLength(1); + } finally { releaseSlow(); await worker.ready; } + expect(started.filter(id => id === "slow")).toHaveLength(1); + expect(retries.get("retrying")).toBe(3); + expect(onError).not.toHaveBeenCalled(); +}); + +it("admits a later deadline when the pass starts with only one slow resource due", async () => { + vi.useFakeTimers(); + const onError = vi.fn(); + const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []), transaction: vi.fn(async () => undefined) }); + coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError }); + const now = Date.now(); + const started: string[] = []; + let releaseSlow!: () => void; + const slow = new Promise(resolve => { releaseSlow = resolve; }); + const worker = registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.nativeResponsePresentation, + concurrentChecks: { limit: 4, key: (item: Entry) => item.issueId, admitDueWhileActive: true }, + decodeScope: () => undefined, + matches: () => true, + load: async () => [ + { id: "slow", issueId: "a", companyId: "company", dueAt: now }, + { id: "retry", issueId: "b", companyId: "company", dueAt: now + 30_000 }, + ], + check: async (item: Entry) => { started.push(item.id); if (item.id === "slow") await slow; return null; }, + }, () => true); + try { + await vi.advanceTimersByTimeAsync(1); + expect(started).toEqual(["slow"]); + await vi.advanceTimersByTimeAsync(30_000); + expect(started).toEqual(["slow", "retry"]); + } finally { releaseSlow(); await worker.ready; } + expect(onError).not.toHaveBeenCalled(); +}); + +it("never admits a later entry between queued siblings that share its resource", async () => { + vi.useFakeTimers(); + const onError = vi.fn(); + const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []), transaction: vi.fn(async () => undefined) }); + coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError }); + const now = Date.now(); + const started: string[] = []; + const active = new Set(); + const overlaps: string[] = []; + const releases = new Map void>(); + const worker = registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.nativeResponsePresentation, + concurrentChecks: { limit: 4, key: (item: Entry) => item.issueId, admitDueWhileActive: true }, + decodeScope: () => undefined, + matches: () => true, + load: async () => [ + { id: "a", issueId: "x", companyId: "company", dueAt: now }, + { id: "b", issueId: "x", companyId: "company", dueAt: now }, + { id: "c", issueId: "x", companyId: "company", dueAt: now + 10_000 }, + // Keeps the pass open so an idle slot waits to admit "c". + { id: "hold", issueId: "y", companyId: "company", dueAt: now }, + // Finishes at once, leaving its slot idle and waiting for "c". + { id: "quick", issueId: "z", companyId: "company", dueAt: now }, + ], + check: async (item: Entry) => { + if (item.id === "quick") return null; + started.push(item.id); + for (const other of active) if (other !== "hold" && item.id !== "hold") overlaps.push(`${other}+${item.id}`); + active.add(item.id); + await new Promise(resolve => releases.set(item.id, resolve)); + active.delete(item.id); + return null; + }, + }, () => true); + try { + await vi.advanceTimersByTimeAsync(1); + expect(new Set(started)).toEqual(new Set(["a", "hold"])); + await vi.advanceTimersByTimeAsync(10_000); + releases.get("a")!(); + await vi.advanceTimersByTimeAsync(1); + expect(started.filter(id => id !== "hold")).toEqual(["a", "b"]); + releases.get("b")!(); + await vi.advanceTimersByTimeAsync(1); + expect(started.filter(id => id !== "hold")).toEqual(["a", "b", "c"]); + releases.get("c")!(); + } finally { for (const resolve of releases.values()) resolve(); await vi.advanceTimersByTimeAsync(1); await worker.ready; } + expect(overlaps).toEqual([]); + expect(onError).not.toHaveBeenCalled(); +}); + +it("wakes an idle admission slot during shutdown and starts nothing new", async () => { + vi.useFakeTimers(); + const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []), transaction: vi.fn(async () => undefined) }); + coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError: vi.fn() }); + const now = Date.now(); + const started: string[] = []; + let releaseSlow!: () => void; + const slow = new Promise(resolve => { releaseSlow = resolve; }); + const worker = registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.nativeResponsePresentation, + concurrentChecks: { limit: 4, key: (item: Entry) => item.issueId, admitDueWhileActive: true }, + decodeScope: () => undefined, + matches: () => true, + load: async () => [ + { id: "slow", issueId: "a", companyId: "company", dueAt: now }, + { id: "later", issueId: "b", companyId: "company", dueAt: now + 60_000 }, + ], + check: async (item: Entry) => { started.push(item.id); if (item.id === "slow") await slow; return null; }, + }, () => true); + await vi.advanceTimersByTimeAsync(1); + expect(started).toEqual(["slow"]); + let stopped = false; + const stopping = coordinator.stop().then(() => { stopped = true; }); + await vi.advanceTimersByTimeAsync(120_000); + expect(stopped).toBe(false); + expect(started).toEqual(["slow"]); + releaseSlow(); + await vi.advanceTimersByTimeAsync(1); + await stopping; + expect(stopped).toBe(true); + expect(started).toEqual(["slow"]); + await worker.ready; +}); 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 c9d2413a8a7..0152995b6b2 100644 --- a/server/src/services/native-runtime/native-workspace-finalization-work.ts +++ b/server/src/services/native-runtime/native-workspace-finalization-work.ts @@ -99,7 +99,9 @@ export function nativeWorkspaceFinalizationWork(db: Db, options: { 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}` }, + // Retries saved while another copy-back is still exporting are admitted + // on schedule instead of waiting for that export to finish. + concurrentChecks: { limit: 4, key: (item: Entry) => `issue:${item.companyId}:${item.issueId}`, admitDueWhileActive: true }, 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), diff --git a/server/src/services/scoped-deadline-work.ts b/server/src/services/scoped-deadline-work.ts index 6ce95086ad0..07be4b96469 100644 --- a/server/src/services/scoped-deadline-work.ts +++ b/server/src/services/scoped-deadline-work.ts @@ -12,7 +12,17 @@ export function registerScopedDeadlineWork( queue: DeliveryQueue; batchSize?: number; minimumPassIntervalMs?: number; - concurrentChecks?: { limit: number; key(entry: Entry): string | readonly string[] }; + concurrentChecks?: { + limit: number; + key(entry: Entry): string | readonly string[]; + // A pass otherwise ends only after its slowest check. Owners whose checks + // can legitimately run for minutes may let idle slots keep admitting + // entries that become due meanwhile, so one slow resource cannot hold + // every other resource's saved deadline until it finishes. check() must + // revalidate each entry against durable state, because an entry admitted + // this way is not reconstructed first. + admitDueWhileActive?: boolean; + }; reconstructionIntervalMs?: number | ((scope: Scope | undefined) => number); beforePass?(): Promise; decodeScope(hint: string): Scope | undefined; @@ -145,21 +155,107 @@ export function registerScopedDeadlineWork( } const pending = [...groups.values()]; let nextGroup = 0; - const checkGroups = async () => { - while (!signal.aborted && canRun()) { - const group = pending[nextGroup++]; - if (!group) return; - for (const entry of group) { - if (signal.aborted || !canRun()) return; - try { - const current = await options.check(entry, signal); - if (current) entries.set(entry.id, current); else entries.delete(entry.id); - failures.delete(entry.id); - } catch (error) { failed(failures, entry.id); errors.push(error); } - } + const admitDueWhileActive = options.concurrentChecks?.admitDueWhileActive === true; + const keysOf = (entry: Entry) => { + const configured = options.concurrentChecks?.key(entry) ?? ""; + return typeof configured === "string" ? [configured] : [...configured]; + }; + // Entries owned by this pass (queued in a group or running) and the + // resource keys they hold. Every batch group reserves its keys until the + // whole group finishes, so admission can neither overlap a queued + // sibling nor jump ahead of it. Admission never overlaps either set. + const owned = new Set(batch.entries.map(entry => entry.id)); + const busyKeys = new Map(); + const reserve = (keys: readonly string[]) => { + for (const key of keys) busyKeys.set(key, (busyKeys.get(key) ?? 0) + 1); + }; + const release = (keys: readonly string[]) => { + for (const key of keys) { + const count = (busyKeys.get(key) ?? 1) - 1; + if (count > 0) busyKeys.set(key, count); else busyKeys.delete(key); } }; - const checks = await Promise.allSettled(Array.from({ length: Math.min(options.concurrentChecks?.limit ?? 1, pending.length) }, checkGroups)); + const groupKeys = pending.map(group => [...new Set(group.flatMap(keysOf))]); + for (const keys of groupKeys) reserve(keys); + const lastChecked = new Map(); + const spacing = options.minimumPassIntervalMs ?? 1000; + let running = 0; + let changed: (() => void) | null = null; + const notify = () => { const resolve = changed; changed = null; resolve?.(); }; + const checkEntry = async (entry: Entry) => { + running += 1; + try { + const current = await options.check(entry, signal); + if (current) entries.set(entry.id, current); else entries.delete(entry.id); + failures.delete(entry.id); + } catch (error) { failed(failures, entry.id); errors.push(error); } + finally { + running -= 1; + lastChecked.set(entry.id, Date.now()); + notify(); + } + }; + const nextAdmission = () => { + // Mid-pass settlements mark scopes dirty, but reconstruction waits for + // the next pass. Admission therefore relies on check() revalidating + // the entry against durable state, as the option requires. + const now = Date.now(); + let wakeAt: number | null = null; + let candidate: Entry | null = null; + for (const item of entries.values()) { + if (owned.has(item.id)) continue; + if (keysOf(item).some(key => busyKeys.has(key))) continue; + const at = Math.max(item.dueAt, failures.get(item.id)?.retryAt ?? 0, (lastChecked.get(item.id) ?? -Infinity) + spacing); + if (at > now) { wakeAt = wakeAt === null ? at : Math.min(wakeAt, at); continue; } + if (!candidate || item.dueAt < candidate.dueAt || (item.dueAt === candidate.dueAt && item.id < candidate.id)) candidate = item; + } + return { candidate, wakeAt }; + }; + const waitForChange = (wakeAt: number | null) => new Promise(resolve => { + const timer = wakeAt === null ? undefined : setTimeout(finish, Math.max(0, wakeAt - Date.now())); + function finish() { + if (timer !== undefined) clearTimeout(timer); + signal.removeEventListener("abort", finish); + if (changed === finish) changed = null; + resolve(); + } + const previous = changed; + changed = previous ? () => { previous(); finish(); } : finish; + signal.addEventListener("abort", finish, { once: true }); + }); + const checkGroups = async () => { + while (!signal.aborted && canRun()) { + const index = nextGroup++; + const group = pending[index]; + if (group) { + try { + for (const entry of group) { + if (signal.aborted || !canRun()) return; + try { await checkEntry(entry); } finally { owned.delete(entry.id); } + } + } finally { release(groupKeys[index]); notify(); } + continue; + } + if (!admitDueWhileActive) return; + const { candidate, wakeAt } = nextAdmission(); + if (candidate) { + const keys = keysOf(candidate); + owned.add(candidate.id); reserve(keys); + try { await checkEntry(candidate); } finally { owned.delete(candidate.id); release(keys); notify(); } + continue; + } + // Nothing is admissible now. Stay available only while another + // check still holds this pass open; otherwise the pass ends and the + // coordinator schedules the next saved deadline as usual. + if (running === 0) return; + await waitForChange(wakeAt); + } + }; + // With admission enabled, spare slots start too: a pass that begins with + // one slow group must still admit other resources that come due later. + const workers = admitDueWhileActive ? options.concurrentChecks!.limit + : Math.min(options.concurrentChecks?.limit ?? 1, pending.length); + const checks = await Promise.allSettled(Array.from({ length: workers }, checkGroups)); for (const result of checks) if (result.status === "rejected") errors.push(result.reason); if (errors.length) throw new AggregateError(errors, `${options.queue} could not settle`); },