From 996e07ba17b0b65b2454939691899fb682c6091b Mon Sep 17 00:00:00 2001 From: selezenart Date: Sun, 13 Sep 2026 16:14:25 +0200 Subject: [PATCH] fix(jobs): recover a delivery stranded in PENDING A committed job whose delivery reached PENDING could be left with nothing able to move it. The worker no-ops a payload whose delivery_attempt no longer matches and then marks the outbox row DELIVERED, and resumeDelivery re-queued only NOT_REQUESTED or RETRIEVAL_FAILED, so the job kept a committed payment, an unretrieved result, and no path forward. PENDING alone is still not treated as resumable. The deciding evidence is whether a fulfill_supplier_order row is still queued for that job: a row a worker currently holds is status PENDING and so counts as queued, which means an in-flight retrieval is never duplicated. Only a job with no such row left is re-queued, under a fresh delivery_attempt that fences the old retrieval. The claim is a compare-and-set against the state read under the job's FOR UPDATE lock, so two concurrent resumes cannot both take one delivery. No payment work is created on this path; the committed settlement is untouched. --- ...260913T170000Z-pending-delivery-refresh.md | 29 +++++++--- packages/storage-postgres/src/jobs.ts | 28 +++++++++- packages/storage-postgres/test/jobs.test.ts | 56 +++++++++++++++++-- 3 files changed, 98 insertions(+), 15 deletions(-) diff --git a/.agent/context/20260913T170000Z-pending-delivery-refresh.md b/.agent/context/20260913T170000Z-pending-delivery-refresh.md index 7d76869..85da9c4 100644 --- a/.agent/context/20260913T170000Z-pending-delivery-refresh.md +++ b/.agent/context/20260913T170000Z-pending-delivery-refresh.md @@ -44,8 +44,15 @@ Follow-up: skip Gate A and Gate B, open the pull request as a draft. after a report of unwanted traffic. The bound and the pending-only condition are what keep both reports satisfied; the resume control it removed stays removed. -- A delivery stranded in `PENDING` server-side is out of scope here and is - recorded below as an unresolved risk. +- A delivery stranded in `PENDING` is now recoverable. `PENDING` alone is not + treated as resumable: the deciding evidence is whether a `fulfill_supplier_order` + row is still queued (`status = 'PENDING'`) for that job. A row a worker is + currently holding is still queued, so an in-flight retrieval is never + duplicated; only a job with nothing left to move it is re-queued. +- Selected `.agent/TEST_MATRIX.md` cases: downstream failure after payment (the + committed payment is preserved and no payment work is created on the recovery + path) and parallel/duplicate claim (the compare-and-set against the locked + row's state means two concurrent resumes cannot both claim one delivery). ## Files/components touched @@ -55,13 +62,19 @@ Follow-up: skip Gate A and Gate B, open the pull request as a draft. stops once nothing is pending, and gives up on a delivery that stays pending. - `apps/web/browser/p5.spec.ts` - the read count is now a lower bound, because an exact count would flake once a pending delivery is re-read on a timer. +- `packages/storage-postgres/src/jobs.ts` - `resumeDelivery` now also recovers a + delivery stranded in `PENDING` with no queued fulfilment row, claiming it + against the state read under the job's row lock. +- `packages/storage-postgres/test/jobs.test.ts` - a queued or in-flight + fulfilment row is still left alone; a stranded one is re-queued under a fresh + `delivery_attempt`, with no payment work created. ## Commands/checks - `pnpm format:check` - PASS - `pnpm lint` - PASS - `pnpm typecheck` - PASS -- `pnpm test` - PASS, 81 files / 1057 tests, includes build. The root vitest +- `pnpm test` - PASS, 81 files / 1059 tests, includes build. The root vitest config covers no `.tsx` file, so the command below is the one that exercises these components. - `pnpm --filter @oneshot/web test` - PASS, 17 files / 97 tests @@ -75,11 +88,11 @@ Follow-up: skip Gate A and Gate B, open the pull request as a draft. ## Unresolved questions -- A delivery that is stranded in `PENDING` cannot be recovered: `resumeDelivery` - re-queues only `NOT_REQUESTED` or `RETRIEVAL_FAILED`, and the web resume - control was removed, so a `PENDING` job whose outbox row was already consumed - has no path forward. The bounded re-reads give up on such a job rather than - fixing it. This needs a separate backend change. +- Nothing in the web UI calls `POST /v1/jobs/:jobId/resume`, so recovering a + stranded delivery still needs an API call. The control that used to do it was + removed deliberately; re-adding one is a product decision, not a defect fix. +- Whether any currently stranded job exists in the user's environment is + unverified here: it is inferred from the state machine, not from their data. ## Git and PR state diff --git a/packages/storage-postgres/src/jobs.ts b/packages/storage-postgres/src/jobs.ts index 7e7fa55..7751e68 100644 --- a/packages/storage-postgres/src/jobs.ts +++ b/packages/storage-postgres/src/jobs.ts @@ -424,17 +424,39 @@ export class JobLedger { await client.query('COMMIT'); return undefined; } + // A delivery is only PENDING legitimately while its fulfilment row is + // still queued or being worked. If no such row is left, nothing will ever + // move this job again: the worker no-ops a payload whose delivery_attempt + // no longer matches, marks the row DELIVERED, and PENDING was not + // resumable, so the job was stranded with the payment already committed. + // The queued-row check runs under the job's FOR UPDATE lock, so a + // concurrent resume cannot also claim it, and a row a worker currently + // holds is still status PENDING and so still counts as queued. + const queuedDelivery = await client.query( + `SELECT 1 FROM outbox_jobs + WHERE task_identifier = 'fulfill_supplier_order' + AND status = 'PENDING' + AND payload->>'job_id' = $1 + LIMIT 1`, + [jobId], + ); + const strandedDelivery = job.delivery_state === 'PENDING' && queuedDelivery.rowCount === 0; if ( job.payment_state === 'COMMITTED' && - (job.delivery_state === 'NOT_REQUESTED' || job.delivery_state === 'RETRIEVAL_FAILED') + (job.delivery_state === 'NOT_REQUESTED' || + job.delivery_state === 'RETRIEVAL_FAILED' || + strandedDelivery) ) { const now = this.#dependencies.now(); + // Compare and set against the state just read under the row lock, so a + // stranded PENDING is claimed exactly once and the retrieval that a new + // delivery_attempt fences off can never complete under the old one. const resumed = await client.query<{ delivery_attempt: number }>( `UPDATE resumable_jobs SET delivery_state = 'PENDING', delivery_attempt = delivery_attempt + 1, updated_at = $1 - WHERE job_id = $2 AND delivery_state IN ('NOT_REQUESTED', 'RETRIEVAL_FAILED') + WHERE job_id = $2 AND delivery_state = $3 RETURNING delivery_attempt`, - [now, jobId], + [now, jobId, job.delivery_state], ); const deliveryAttempt = resumed.rows[0]?.delivery_attempt; if (deliveryAttempt === undefined) { diff --git a/packages/storage-postgres/test/jobs.test.ts b/packages/storage-postgres/test/jobs.test.ts index 089746f..b47ba74 100644 --- a/packages/storage-postgres/test/jobs.test.ts +++ b/packages/storage-postgres/test/jobs.test.ts @@ -248,7 +248,7 @@ describe('JobLedger delivery recovery', () => { const resumed = await ledger.resumeDelivery('workspace-unit', failedJob.job_id); expect(resumed).toMatchObject({ delivery_state: 'PENDING', payment_state: 'COMMITTED' }); - const outbox = calls.find((call) => call.sql.includes("'fulfill_supplier_order'")); + const outbox = calls.find((call) => call.sql.includes('INSERT INTO outbox_jobs')); expect(outbox?.values?.[1]).toBe(`fulfill:${failedJob.job_id}:team_report_order_unit:2`); expect(outbox?.values?.[2]).toBe( JSON.stringify({ job_id: failedJob.job_id, delivery_attempt: 2 }), @@ -257,7 +257,7 @@ describe('JobLedger delivery recovery', () => { expect(calls.some((call) => call.sql.includes('attempts'))).toBe(false); }); - it('does not enqueue a duplicate delivery while an attempt is already pending', async () => { + it('does not enqueue a duplicate delivery while an attempt is already queued', async () => { const calls: string[] = []; const pendingJob = { ...failedJob, delivery_state: 'PENDING' as const }; const client = { @@ -266,7 +266,10 @@ describe('JobLedger delivery recovery', () => { if (sql.includes('FOR UPDATE OF j') || sql.includes('WHERE j.workspace_id')) { return { rows: [pendingJob] }; } - return { rows: [] }; + // The fulfilment row is still queued, or a worker is holding it: either + // way something will still move this delivery. + if (sql.includes('FROM outbox_jobs')) return { rows: [{ '?column?': 1 }], rowCount: 1 }; + return { rows: [], rowCount: 0 }; }, release() {}, }; @@ -278,6 +281,51 @@ describe('JobLedger delivery recovery', () => { await ledger.resumeDelivery('workspace-unit', pendingJob.job_id); expect(calls.some((sql) => sql.includes('RETURNING delivery_attempt'))).toBe(false); - expect(calls.some((sql) => sql.includes("'fulfill_supplier_order'"))).toBe(false); + expect(calls.some((sql) => sql.includes('INSERT INTO outbox_jobs'))).toBe(false); + }); + + /** + * The payment is committed and the result was never retrieved, but nothing is + * queued to retrieve it: the worker no-ops a payload whose delivery_attempt no + * longer matches and then marks the row DELIVERED. PENDING was not resumable, + * so such a job could never move again. + */ + it('re-queues a pending delivery that has no fulfilment work left', async () => { + const calls: Array<{ sql: string; values?: readonly unknown[] }> = []; + const strandedJob = { ...failedJob, delivery_state: 'PENDING' as const }; + const client = { + async query(sql: string, values?: readonly unknown[]) { + calls.push({ sql, values }); + if (sql.includes('FOR UPDATE OF j')) return { rows: [strandedJob], rowCount: 1 }; + if (sql.includes('RETURNING delivery_attempt')) { + return { rows: [{ delivery_attempt: 2 }], rowCount: 1 }; + } + if (sql.includes('WHERE j.workspace_id')) { + return { rows: [{ ...strandedJob, delivery_attempt: 2 }], rowCount: 1 }; + } + // Nothing queued and nothing in flight. + return { rows: [], rowCount: 0 }; + }, + release() {}, + }; + const ledger = new JobLedger({ connect: async () => client } as never, { + now: () => new Date('2026-09-07T12:02:00.000Z'), + nextAttemptId: () => 'unused', + }); + + await ledger.resumeDelivery('workspace-unit', strandedJob.job_id); + + // Claimed against the state read under the row lock, so two concurrent + // resumes cannot both take it. + const claim = calls.find((call) => call.sql.includes('RETURNING delivery_attempt')); + expect(claim?.values?.[2]).toBe('PENDING'); + // A fresh attempt fences the retrieval, and no payment work is created. + const outbox = calls.find((call) => call.sql.includes('INSERT INTO outbox_jobs')); + expect(outbox?.values?.[1]).toBe(`fulfill:${strandedJob.job_id}:team_report_order_unit:2`); + expect(outbox?.values?.[2]).toBe( + JSON.stringify({ job_id: strandedJob.job_id, delivery_attempt: 2 }), + ); + expect(calls.some((call) => call.sql.includes('submit_settlement'))).toBe(false); + expect(calls.some((call) => call.sql.includes('attempts'))).toBe(false); }); });