mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-08 22:13:59 +01:00
fix(native): run saved copy-back retries on schedule behind a slow export (#19145)
## 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 ata0c0851766. Observed on a staging deployment of69dea878. **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 <noreply@anthropic.com>
This commit is contained in:
1 parent
c2cc69d1e4
commit
482ec0aea9
4 files changed
+314
-15
No files matched your search
@@ -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<void>(resolve => { release = resolve; });
|
||||
const attempts: string[] = [];
|
||||
const attempt = vi.fn(async (input: Parameters<typeof attemptNativeWorkspaceFinalization>[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);
|
||||
|
||||
@@ -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<void>(resolve => { releaseSlow = resolve; });
|
||||
const retries = new Map<string, number>();
|
||||
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<void>(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<string>();
|
||||
const overlaps: string[] = [];
|
||||
const releases = new Map<string, () => 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<void>(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<void>(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;
|
||||
});
|
||||
Reference in new issue
Block a user