From dff6c00468671a088dc307f18132334e855f5d59 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 19:50:27 -0500 Subject: [PATCH] Join receipt acceptance passes before advancing saved deadlines Co-Authored-By: Paperclip --- .../2026-10-30-deferred-execution-sleep.md | 4 +++ .../deferred-execution-receipt-work.test.ts | 29 +++++++++++++++---- 2 files changed, 28 insertions(+), 5 deletions(-) diff --git a/doc/plans/2026-10-30-deferred-execution-sleep.md b/doc/plans/2026-10-30-deferred-execution-sleep.md index 0478fe79300..56e3611d939 100644 --- a/doc/plans/2026-10-30-deferred-execution-sleep.md +++ b/doc/plans/2026-10-30-deferred-execution-sleep.md @@ -17,3 +17,7 @@ Three scoped owners reconstruct only their actual durable receipt predicates, wi The generic run-dispatch notifier and broad issue review-state notifier deliberately do not fan out deferred eligibility. Queue bookkeeping may write them while processing its own saved input. Separate eligibility commits and settlement-only chat notifications avoid a self-restaging loop. C2 remains separate and mandatory: active output silence deadlines, remaining orphan/process/controller/remote authority and resource debts, legacy shutdown settlements, stranded agent status, and final aggregate chain removal. Retained workspace and migrated native finalization workers remain their existing owners. + +Review controls cover queue edits that move an already loaded grace deadline. Each due check re-reads current stored grace or claim time before admission. Subtree hold release, subtree restore, and member archival publish eligibility inside their own transactions. Member archival preserves the cancelled Retry tombstone created by an earlier user reassignment. + +A resume that first records a new execution-wait reason arms its existing 30-second grace once. Writing the same closed-gate reason again does not publish another hint. A clock obligation, an eligibility change, and an infrastructure failure therefore retain separate exits: stored deadline, owning committed producer, and bounded failure backoff. Refreshing a diagnostic timestamp alone does not create permanent polling behind a deliberate pause. diff --git a/server/src/__tests__/deferred-execution-receipt-work.test.ts b/server/src/__tests__/deferred-execution-receipt-work.test.ts index eb60e2ed449..914f33fe609 100644 --- a/server/src/__tests__/deferred-execution-receipt-work.test.ts +++ b/server/src/__tests__/deferred-execution-receipt-work.test.ts @@ -1,3 +1,4 @@ +import { setImmediate as immediate } from "node:timers/promises"; import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js"; import { accessService } from "../services/access.js"; import { writeDeferredExecutionWait } from "../services/deferred-execution-wait.js"; @@ -27,6 +28,7 @@ describe("deferred receipt durable dispatch", () => { let coordinator: ReturnType; let drain = false; const errors: unknown[] = []; + const passes = new Set>(); beforeAll(async () => { const external = resolveExternalTestDatabaseUrl("PAPERCLIP_DEFERRED_RECEIPT_TEST_DATABASE_URL"); if (!external) temporary = await startEmbeddedPostgresTestDatabase("deferred-receipts-"); @@ -38,6 +40,11 @@ describe("deferred receipt durable dispatch", () => { heartbeat = heartbeatService(db); coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !drain, canReconcile: () => true, onError: error => errors.push(error) }); + const register = coordinator.register.bind(coordinator); + coordinator.register = (queue, task) => register(queue, { ...task, run: signal => { + const pass = Promise.resolve(task.run(signal)); passes.add(pass); + void pass.then(() => passes.delete(pass), () => passes.delete(pass)); return pass; + } }); }); afterEach(async () => { await coordinator.stop(); await heartbeat.drainActiveRunExecutions(); @@ -70,12 +77,24 @@ describe("deferred receipt durable dispatch", () => { } async function finish(issueId: string) { await issueService(db).update(issueId, { status: "done" }); - await heartbeat.drainActiveRunExecutions(); + await vi.waitFor(async () => { + await Promise.allSettled([...passes]); await heartbeat.drainActiveRunExecutions(); + expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0); + }, { timeout: 15_000 }); const saved = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.status, "scheduled_retry")); - expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0); - vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); - if (saved.length) await vi.advanceTimersByTimeAsync(Math.max(0, Math.max(...saved.map(run => run.scheduledRetryAt!.getTime())) - Date.now()) + 1); - await vi.waitFor(async () => expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.status, "scheduled_retry"))).toHaveLength(0)); + if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + if (saved.length) { + const due = Math.max(...saved.map(run => run.scheduledRetryAt!.getTime())); + while (Date.now() <= due) { + await vi.advanceTimersByTimeAsync(Math.min(500, due - Date.now() + 1)); + await Promise.allSettled([...passes]); await immediate(); + } + } + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(500); await Promise.allSettled([...passes]); + await heartbeat.drainActiveRunExecutions(); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.status, "scheduled_retry"))).toHaveLength(0); + }); } it("is quiet after empty reconstruction and terminal receipt history", async () => { const f = await fixture();