From 18f9735396f7310c7f7d2c10570c861c0411004b Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Thu, 2 Jul 2026 02:54:33 -0700 Subject: [PATCH 1/2] feat(selfhost): add automatic dead-letter job retry for self-host queues Neither self-host queue backend (SQLite or Postgres) has a way to move a job out of status='dead' once it lands there. The Cloudflare Workers Queues DLQ redrive path (src/queue/dlq.ts) is wired exclusively to that platform's native queue consumer and is not reachable from the self-host tables. In practice this means: a job that dies from a transient bug, or a bug that's since been fixed and redeployed, requires a manual database UPDATE/DELETE to ever run again -- indefinitely. - Add reviveDeadLetterJobs() to both createSqliteQueue and createPgQueue: requeues dead jobs whose lifetime `attempts` is still under a ceiling (maxRetries + a small configurable extra-attempts budget, default 3), clearing last_error. `attempts` itself is left untouched, so a revived job gets exactly ONE more attempt before either succeeding or dying again -- not a fresh full retry budget -- bounding how many times a permanently- broken job can cycle before it stops being revived and requires manual intervention. - Runs on a separate, much slower timer (default 30 minutes, QUEUE_DEAD_LETTER_REVIVE_INTERVAL_MS) than the normal poll tick -- that interval IS the cooldown between auto-retry rounds, so no separate timestamp bookkeeping was needed. Reuses the existing attempts/job_key columns rather than a parallel table. - New gittensory_jobs_dead_letter_revived_total metric, distinct from the existing crash-recovery gittensory_jobs_recovered_total counter. - reviveDeadLetterJobs() is exposed on the public queue interface (not just wired to the internal timer) so it's directly testable without waiting on real intervals, and available for a future operator-triggered repair path. Validation: full local gate green; every new/changed line and branch across sqlite-queue.ts, pg-queue.ts, and queue-common.ts is covered per coverage/lcov.info (cross-checked against the diff hunks directly). New tests cover: revival under the ceiling with last_error cleared, the ceiling actually stopping further revival, no-op with nothing dead, the periodic timer firing a real revival end-to-end (sqlite, fake timers), the equivalent mocked-pool behavior for Postgres, and the two new env-var config knobs. --- .../routes/docs.self-hosting-operations.tsx | 10 ++ src/selfhost/pg-queue.ts | 50 ++++++++++ src/selfhost/queue-common.ts | 17 ++++ src/selfhost/sqlite-queue.ts | 49 ++++++++++ test/unit/selfhost-pg-queue.test.ts | 47 ++++++++++ test/unit/selfhost-queue-common.test.ts | 25 +++++ test/unit/selfhost-sqlite-queue.test.ts | 94 +++++++++++++++++++ 7 files changed, 292 insertions(+) diff --git a/apps/gittensory-ui/src/routes/docs.self-hosting-operations.tsx b/apps/gittensory-ui/src/routes/docs.self-hosting-operations.tsx index e62f18ed74..6b36b12475 100644 --- a/apps/gittensory-ui/src/routes/docs.self-hosting-operations.tsx +++ b/apps/gittensory-ui/src/routes/docs.self-hosting-operations.tsx @@ -120,6 +120,16 @@ docker compose --profile postgres --profile observability --profile backup up -d check if you haven't wired up push notifications yet — it's exactly what the{" "} Dead jobs stay at zero routine check below is watching for.

+

+ Dead-lettered jobs also get one automatic revival attempt every 30 minutes ( + QUEUE_DEAD_LETTER_REVIVE_INTERVAL_MS), as long as the job hasn't already been + revived more than a small, bounded number of extra times ( + QUEUE_DEAD_LETTER_AUTO_RETRY_MAX_EXTRA_ATTEMPTS, default 3) — so a job that + died from a bug that's since been fixed and redeployed recovers on its own within the next + cycle, without needing direct database access. A job that keeps failing the same way + eventually exhausts this budget and stays dead, which is exactly what the alert above is + watching for. +

Docker resource hygiene

diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index 5dc1773823..18b6848ce3 100644 --- a/src/selfhost/pg-queue.ts +++ b/src/selfhost/pg-queue.ts @@ -24,6 +24,8 @@ import { jobPriority, parsePositiveIntEnv, queueBackgroundConcurrency, + queueDeadLetterAutoRetryMaxExtraAttempts, + queueDeadLetterReviveIntervalMs, queueProcessingTimeoutMs, queueRecoveryJitterMs, queueStartupJitterMinJobs, @@ -68,6 +70,10 @@ export interface PgDurableQueue { deadCount(): Promise; stats(): Promise>; snapshot(): Promise; + /** Requeues dead-lettered jobs still under the auto-retry attempts ceiling. Called on a timer while + * running (see start()), and exposed directly so tests and an operator-triggered repair path don't have + * to wait for the real interval. Returns the number of jobs revived. */ + reviveDeadLetterJobs(): Promise; } interface JobRow { @@ -115,6 +121,7 @@ export function createPgQueue( let activeBackground = 0; const activeJobIds = new Set(); let timer: ReturnType | null = null; + let deadLetterReviveTimer: ReturnType | null = null; async function init(): Promise { await pool.query(DDL); @@ -204,6 +211,43 @@ export function createPgQueue( return changed; } + // Dead-letter auto-retry (#audit-rate-headroom): a job dies once `attempts >= maxRetries` (see the + // max-retries branch in processOne below). Reviving it here only clears `status`/`run_after`/`last_error` + // -- `attempts` is left untouched, so it already satisfies `attempts >= maxRetries` and will die again + // after exactly ONE more failed attempt, not a fresh full retry budget. The `attempts < ceiling` filter + // (ceiling = maxRetries + the configured extra-attempts budget) is what actually bounds how many times a + // permanently-broken job can be revived before it stops being a candidate here and requires manual + // intervention. + async function reviveEligibleDeadJobs(): Promise { + const ceiling = maxRetries + queueDeadLetterAutoRetryMaxExtraAttempts(); + const res = await pool.query( + `SELECT id, payload, job_key FROM ${TABLE} WHERE status='dead' AND attempts<$1`, + [ceiling], + ); + let revived = 0; + const now = Date.now(); + const maxJitter = queueRecoveryJitterMs(); + for (const row of res.rows as Array<{ id: string; payload: string; job_key?: string | null }>) { + const runAfter = now + deterministicJitterMs(`revive:${row.job_key ?? ""}:${row.id}:${row.payload}`, maxJitter); + await pool.query(`UPDATE ${TABLE} SET status='pending', run_after=$1, last_error=NULL WHERE id=$2`, [ + runAfter, + row.id, + ]); + revived += 1; + } + return revived; + } + + async function reviveDeadLetterJobs(): Promise { + const revived = await reviveEligibleDeadJobs(); + if (revived) { + await recordQueueMetric("gittensory_jobs_dead_letter_revived_total", revived); + console.log(JSON.stringify({ event: "selfhost_queue_dead_letter_revived", count: revived })); + kickAll(); + } + return revived; + } + async function spreadDueJobsOnStartup(): Promise { const now = Date.now(); const res = await pool.query( @@ -614,10 +658,15 @@ export function createPgQueue( timer = setTimeout(tick, pollIntervalMs); }; tick(); + // Separate, much slower interval than the poll tick above -- reviving a dead job every second would + // recreate the retry storm this feature exists to bound. The interval itself is the cooldown between + // auto-retry rounds for any one job. + deadLetterReviveTimer = setInterval(() => void reviveDeadLetterJobs(), queueDeadLetterReviveIntervalMs()); }, async stop() { running = false; if (timer) clearTimeout(timer); + if (deadLetterReviveTimer) clearInterval(deadLetterReviveTimer); while (active > 0) await new Promise((r) => setTimeout(r, 10)); }, async drain() { @@ -646,6 +695,7 @@ export function createPgQueue( return readQueueStats(); }, snapshot: binding.snapshot, + reviveDeadLetterJobs, }; async function reclaimExpiredProcessingJobs(): Promise { diff --git a/src/selfhost/queue-common.ts b/src/selfhost/queue-common.ts index d3fc47f809..a5b56029b2 100644 --- a/src/selfhost/queue-common.ts +++ b/src/selfhost/queue-common.ts @@ -22,6 +22,12 @@ const DEFAULT_SCHEDULED_ENQUEUE_JITTER_MS = 5 * 60_000; const DEFAULT_STARTUP_JITTER_MIN_JOBS = 8; const DEFAULT_PROCESSING_TIMEOUT_MS = 30 * 60_000; const DEFAULT_BACKGROUND_CONCURRENCY = 1; +// Dead-letter auto-retry (#audit-rate-headroom): a job that exhausted its normal retry budget and landed in +// `dead` gets ONE more attempt every revive interval, as long as its lifetime attempts stay under +// maxRetries + this extra ceiling — bounded so a permanently-broken job cannot cycle dead→pending→dead +// forever. The revive interval itself IS the cooldown; no separate timestamp bookkeeping is needed. +const DEFAULT_DEAD_LETTER_REVIVE_INTERVAL_MS = 30 * 60_000; +const DEFAULT_DEAD_LETTER_AUTO_RETRY_MAX_EXTRA_ATTEMPTS = 3; export const FOREGROUND_QUEUE_PRIORITY_FLOOR = 8; export type SelfHostQueueJobStatus = "pending" | "processing" | "dead"; @@ -579,6 +585,17 @@ export function resolvePostgresPoolMax(): number { return parsePositiveIntEnv("PGPOOL_MAX", { min: 1, fallback: 10 }); } +export function queueDeadLetterReviveIntervalMs(): number { + return envDurationMs("QUEUE_DEAD_LETTER_REVIVE_INTERVAL_MS", DEFAULT_DEAD_LETTER_REVIVE_INTERVAL_MS); +} + +export function queueDeadLetterAutoRetryMaxExtraAttempts(): number { + return parsePositiveIntEnv("QUEUE_DEAD_LETTER_AUTO_RETRY_MAX_EXTRA_ATTEMPTS", { + min: 0, + fallback: DEFAULT_DEAD_LETTER_AUTO_RETRY_MAX_EXTRA_ATTEMPTS, + }); +} + export function deterministicJitterMs(seed: string, maxJitterMs: number): number { if (!Number.isFinite(maxJitterMs) || maxJitterMs <= 0) return 0; let h = 2166136261; diff --git a/src/selfhost/sqlite-queue.ts b/src/selfhost/sqlite-queue.ts index 923c8ad8b2..e49da97bea 100644 --- a/src/selfhost/sqlite-queue.ts +++ b/src/selfhost/sqlite-queue.ts @@ -25,6 +25,8 @@ import { jobPriority, parsePositiveIntEnv, queueBackgroundConcurrency, + queueDeadLetterAutoRetryMaxExtraAttempts, + queueDeadLetterReviveIntervalMs, queueProcessingTimeoutMs, queueRecoveryJitterMs, queueStartupJitterMinJobs, @@ -70,6 +72,10 @@ export interface DurableQueue { deadCount(): number; stats(): Record; snapshot(): SelfHostQueueSnapshot; + /** Requeues dead-lettered jobs still under the auto-retry attempts ceiling. Called on a timer while + * running (see start()), and exposed directly so tests and an operator-triggered repair path don't have + * to wait for the real interval. Returns the number of jobs revived. */ + reviveDeadLetterJobs(): number; } interface JobRow { @@ -169,6 +175,17 @@ export function createSqliteQueue( let activeBackground = 0; const activeJobIds = new Set(); let timer: ReturnType | null = null; + let deadLetterReviveTimer: ReturnType | null = null; + + function reviveDeadLetterJobs(): number { + const revived = reviveEligibleDeadJobs(driver, maxRetries); + if (revived) { + recordQueueMetric(driver, "gittensory_jobs_dead_letter_revived_total", revived); + console.log(JSON.stringify({ event: "selfhost_queue_dead_letter_revived", count: revived })); + kickAll(); + } + return revived; + } function enqueue(message: JobMessage, delaySeconds: number): void { const now = Date.now(); @@ -556,10 +573,15 @@ export function createSqliteQueue( timer = setTimeout(tick, pollIntervalMs); }; tick(); + // Separate, much slower interval than the poll tick above -- reviving a dead job every second would + // recreate the retry storm this feature exists to bound. The interval itself is the cooldown between + // auto-retry rounds for any one job. + deadLetterReviveTimer = setInterval(reviveDeadLetterJobs, queueDeadLetterReviveIntervalMs()); }, async stop() { running = false; if (timer) clearTimeout(timer); + if (deadLetterReviveTimer) clearInterval(deadLetterReviveTimer); while (active > 0) await new Promise((r) => setTimeout(r, 10)); // let in-flight pumps finish }, async drain() { @@ -591,6 +613,7 @@ export function createSqliteQueue( return readQueueStats(driver); }, snapshot: binding.snapshot, + reviveDeadLetterJobs, }; } @@ -646,6 +669,32 @@ function recoverProcessingJobs(driver: SqliteDriver): number { return changed; } +// Dead-letter auto-retry (#audit-rate-headroom): a job dies once `attempts >= maxRetries` (see the +// max-retries branch in processOne below). Reviving it here only clears `status`/`run_after`/`last_error` — +// `attempts` is left untouched, so it already satisfies `attempts >= maxRetries` and will die again after +// exactly ONE more failed attempt, not a fresh full retry budget. The `attempts < ceiling` filter (ceiling = +// maxRetries + the configured extra-attempts budget) is what actually bounds how many times a permanently- +// broken job can be revived before it stops being a candidate here and requires manual intervention. +function reviveEligibleDeadJobs(driver: SqliteDriver, maxRetries: number): number { + const ceiling = maxRetries + queueDeadLetterAutoRetryMaxExtraAttempts(); + const { rows } = driver.query( + `SELECT id, payload, job_key FROM ${TABLE} WHERE status='dead' AND attempts) { + const runAfter = now + deterministicJitterMs(`revive:${row.job_key ?? ""}:${row.id}:${row.payload}`, maxJitter); + driver.query( + `UPDATE ${TABLE} SET status='pending', run_after=?, last_error=NULL WHERE id=?`, + [runAfter, row.id], + ); + revived += 1; + } + return revived; +} + function spreadDueJobsOnStartup(driver: SqliteDriver): number { const now = Date.now(); const { rows } = driver.query( diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts index 8bd245821f..3446bf5cb5 100644 --- a/test/unit/selfhost-pg-queue.test.ts +++ b/test/unit/selfhost-pg-queue.test.ts @@ -914,6 +914,53 @@ describe("createPgQueue (durable #977)", () => { expect(calls).toBe(2); }); + describe("reviveDeadLetterJobs (#audit-rate-headroom)", () => { + afterEach(() => { + delete process.env.QUEUE_DEAD_LETTER_AUTO_RETRY_MAX_EXTRA_ATTEMPTS; + }); + + it("requeues dead jobs still under the auto-retry ceiling, clearing last_error, and records the metric", async () => { + process.env.QUEUE_DEAD_LETTER_AUTO_RETRY_MAX_EXTRA_ATTEMPTS = "2"; + const m = makePool(); + m.fn.mockResolvedValueOnce({ + rows: [ + { id: "1", payload: JSON.stringify({ type: "t" }), job_key: null }, + { id: "2", payload: JSON.stringify({ type: "t" }), job_key: "k" }, + ], + rowCount: 2, + }); // SELECT status='dead' AND attempts undefined, { maxRetries: 1 }); + + const revived = await q.reviveDeadLetterJobs(); + + expect(revived).toBe(2); + // The SELECT was bound to the ceiling (maxRetries=1 + extra=2 = 3), not a raw maxRetries. + expect(m.fn).toHaveBeenCalledWith(expect.stringContaining("status='dead' AND attempts<$1"), [3]); + // Each eligible row is revived to pending with last_error cleared -- not a fresh retry budget (attempts + // is never touched here). + expect(m.fn).toHaveBeenCalledWith( + expect.stringContaining("SET status='pending', run_after=$1, last_error=NULL"), + expect.arrayContaining(["1"]), + ); + expect(m.fn).toHaveBeenCalledWith( + expect.stringContaining("SET status='pending', run_after=$1, last_error=NULL"), + expect.arrayContaining(["2"]), + ); + expect(await renderMetrics()).toContain("gittensory_jobs_dead_letter_revived_total 2"); + }); + + it("is a no-op (and records nothing) when no dead job is under the ceiling", async () => { + const m = makePool(); + m.fn.mockResolvedValueOnce({ rows: [], rowCount: 0 }); + const q = createPgQueue(m.pool, async () => undefined, { maxRetries: 1 }); + + const revived = await q.reviveDeadLetterJobs(); + + expect(revived).toBe(0); + expect(await renderMetrics()).not.toContain("gittensory_jobs_dead_letter_revived_total"); + }); + }); + it("reschedules GitHub rate-limit failures without consuming the dead-letter budget", async () => { const m = makePool(); m.enqueueJob("1", { type: "github-webhook" }, 4); diff --git a/test/unit/selfhost-queue-common.test.ts b/test/unit/selfhost-queue-common.test.ts index 0780eba652..4506b5bebb 100644 --- a/test/unit/selfhost-queue-common.test.ts +++ b/test/unit/selfhost-queue-common.test.ts @@ -24,6 +24,8 @@ import { nonConsumingRetryDelayMs, parsePositiveIntEnv, queueBackgroundConcurrency, + queueDeadLetterAutoRetryMaxExtraAttempts, + queueDeadLetterReviveIntervalMs, queueProcessingTimeoutMs, queueRecoveryJitterMs, queueSnapshotBacklog, @@ -1193,3 +1195,26 @@ describe("resolvePostgresPoolMax (#audit-rate-headroom)", () => { expect(warn).toHaveBeenCalledOnce(); }); }); + +describe("dead-letter auto-retry config (#audit-rate-headroom)", () => { + afterEach(() => { + delete process.env.QUEUE_DEAD_LETTER_REVIVE_INTERVAL_MS; + delete process.env.QUEUE_DEAD_LETTER_AUTO_RETRY_MAX_EXTRA_ATTEMPTS; + }); + + it("queueDeadLetterReviveIntervalMs defaults to 30 minutes and honors its env override", () => { + delete process.env.QUEUE_DEAD_LETTER_REVIVE_INTERVAL_MS; + expect(queueDeadLetterReviveIntervalMs()).toBe(30 * 60_000); + + process.env.QUEUE_DEAD_LETTER_REVIVE_INTERVAL_MS = "60000"; + expect(queueDeadLetterReviveIntervalMs()).toBe(60_000); + }); + + it("queueDeadLetterAutoRetryMaxExtraAttempts defaults to 3 and honors its env override", () => { + delete process.env.QUEUE_DEAD_LETTER_AUTO_RETRY_MAX_EXTRA_ATTEMPTS; + expect(queueDeadLetterAutoRetryMaxExtraAttempts()).toBe(3); + + process.env.QUEUE_DEAD_LETTER_AUTO_RETRY_MAX_EXTRA_ATTEMPTS = "0"; + expect(queueDeadLetterAutoRetryMaxExtraAttempts()).toBe(0); + }); +}); diff --git a/test/unit/selfhost-sqlite-queue.test.ts b/test/unit/selfhost-sqlite-queue.test.ts index eb33d0ad2b..cb39bc3e98 100644 --- a/test/unit/selfhost-sqlite-queue.test.ts +++ b/test/unit/selfhost-sqlite-queue.test.ts @@ -1190,6 +1190,100 @@ describe("createSqliteQueue (durable #980)", () => { expect(q.size()).toBe(0); }); + describe("reviveDeadLetterJobs (#audit-rate-headroom)", () => { + beforeEach(() => { + // Deterministic zero jitter so a revived job's run_after is exactly "now" -- otherwise + // deterministicJitterMs's (up to 60s default) spread makes it not-yet-due for drain()/the poll + // tick in these tests. Mirrors the existing recoverProcessingJobs test convention in this file. + process.env.QUEUE_RECOVERY_JITTER_MS = "0"; + }); + afterEach(() => { + delete process.env.QUEUE_RECOVERY_JITTER_MS; + delete process.env.QUEUE_DEAD_LETTER_AUTO_RETRY_MAX_EXTRA_ATTEMPTS; + delete process.env.QUEUE_DEAD_LETTER_REVIVE_INTERVAL_MS; + }); + + it("requeues a dead job under the auto-retry ceiling and clears its last_error", async () => { + process.env.QUEUE_DEAD_LETTER_AUTO_RETRY_MAX_EXTRA_ATTEMPTS = "2"; + const driver = makeDriver(); + let calls = 0; + const q = createSqliteQueue( + driver, + async () => { + calls += 1; + throw new Error("boom"); + }, + { maxRetries: 1, backoffMs: () => 0 }, + ); + await q.binding.send(msg("x")); + await q.drain(); // dies at attempts=1 (maxRetries=1) + expect(q.deadCount()).toBe(1); + calls = 0; + + const revived = q.reviveDeadLetterJobs(); + + expect(revived).toBe(1); + const { rows } = driver.query("SELECT status, attempts, last_error FROM _selfhost_jobs", []); + const row = rows[0] as { status: string; attempts: number; last_error: string | null }; + // With zero jitter the revived job is immediately due, so kickAll()'s synchronous claim step may have + // already advanced it past 'pending' to 'processing' by the time this assertion runs -- either is + // correct proof of revival; only 'dead' would indicate the revival didn't happen. + expect(row.status).not.toBe("dead"); + expect(row.attempts).toBe(1); // untouched -- one more failure re-dead-letters it, not a fresh budget + expect(row.last_error).toBeNull(); + + await q.drain(); // the one extra attempt the revival granted + expect(calls).toBe(1); + expect(q.deadCount()).toBe(1); // failed again -- back to dead, attempts now 2 + }); + + it("stops reviving a job once it reaches the auto-retry ceiling (maxRetries + extra attempts)", async () => { + process.env.QUEUE_DEAD_LETTER_AUTO_RETRY_MAX_EXTRA_ATTEMPTS = "1"; + const driver = makeDriver(); + const q = createSqliteQueue(driver, async () => { throw new Error("boom"); }, { maxRetries: 1, backoffMs: () => 0 }); + await q.binding.send(msg("x")); + await q.drain(); // attempts=1, dead (ceiling = maxRetries(1) + extra(1) = 2) + + expect(q.reviveDeadLetterJobs()).toBe(1); // attempts(1) < ceiling(2) -- eligible + await q.drain(); // fails again -- attempts=2, dead again + + expect(q.reviveDeadLetterJobs()).toBe(0); // attempts(2) is NOT < ceiling(2) -- exhausted, stays dead + expect(q.deadCount()).toBe(1); + }); + + it("is a no-op when there are no dead jobs", () => { + const driver = makeDriver(); + const q = createSqliteQueue(driver, async () => undefined); + expect(q.reviveDeadLetterJobs()).toBe(0); + }); + + it("runs automatically on the configured revive interval while the queue is running", async () => { + process.env.QUEUE_DEAD_LETTER_REVIVE_INTERVAL_MS = "1000"; + vi.useFakeTimers(); + const driver = makeDriver(); + let calls = 0; + const q = createSqliteQueue( + driver, + async () => { + calls += 1; + throw new Error("boom"); + }, + { maxRetries: 1, backoffMs: () => 0, pollIntervalMs: 50 }, + ); + await q.binding.send(msg("x")); + await vi.advanceTimersByTimeAsync(200); // dies at attempts=1 + expect(q.deadCount()).toBe(1); + calls = 0; + + q.start(); + await vi.advanceTimersByTimeAsync(1000); // the revive interval fires once + await vi.advanceTimersByTimeAsync(200); // the poll tick picks up the revived job + + expect(calls).toBe(1); // the auto-revived job was actually re-attempted + await q.stop(); + }); + }); + it("reschedules GitHub rate-limit failures without consuming the dead-letter budget", async () => { const driver = makeDriver(); let calls = 0; From ae2472b331d8677188d1e3854f14101ce920995e Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Thu, 2 Jul 2026 04:43:45 -0700 Subject: [PATCH 2/2] fix(selfhost): re-check dead status on dead-letter revive UPDATE The AI review on this PR flagged a real defect: reviveEligibleDeadJobs selects candidate dead jobs, then updates each one back to 'pending' by id alone. That SELECT is a stale snapshot -- an overlapping reviver (a second self-host instance sharing the same Postgres database, or a slow prior revive tick still running when the next one fires) can move a row out of 'dead' between the SELECT and this row's own UPDATE. With no re-check, the UPDATE blindly flips whatever the row's CURRENT status is back to 'pending', including a row already claimed into 'processing' -- letting the job run a second time concurrently. Both backends now add "AND status='dead'" to the revive UPDATE, exactly mirroring the same defensive predicate already used elsewhere in both files for the same class of stale-snapshot sweep (reclaimExpiredProcessingJobs / deferPendingJobsForRateLimit). The revived count now reflects rows the UPDATE actually matched (pg's rowCount / sqlite's changes), not the raw SELECT count, so a row another reviver already claimed is correctly excluded rather than double-counted. Added a regression test per backend: Postgres via a mock pool with per-call configurable UPDATE rowCounts (proving a 0-rowCount UPDATE isn't counted as revived); SQLite via a driver.query spy that injects the competing status change at the exact point the real UPDATE would otherwise race against it (proving the row is left untouched, not reverted to pending). --- src/selfhost/pg-queue.ts | 15 +++++--- src/selfhost/sqlite-queue.ts | 11 ++++-- test/unit/selfhost-pg-queue.test.ts | 47 +++++++++++++++++++++++++ test/unit/selfhost-sqlite-queue.test.ts | 28 +++++++++++++++ 4 files changed, 93 insertions(+), 8 deletions(-) diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index 18b6848ce3..e02990a1be 100644 --- a/src/selfhost/pg-queue.ts +++ b/src/selfhost/pg-queue.ts @@ -229,11 +229,16 @@ export function createPgQueue( const maxJitter = queueRecoveryJitterMs(); for (const row of res.rows as Array<{ id: string; payload: string; job_key?: string | null }>) { const runAfter = now + deterministicJitterMs(`revive:${row.job_key ?? ""}:${row.id}:${row.payload}`, maxJitter); - await pool.query(`UPDATE ${TABLE} SET status='pending', run_after=$1, last_error=NULL WHERE id=$2`, [ - runAfter, - row.id, - ]); - revived += 1; + // AND status='dead' re-checks the row is STILL dead at UPDATE time (mirrors reclaimExpiredProcessingJobs / + // deferPendingJobsForRateLimit above) — the SELECT above is a stale snapshot, and without this predicate an + // overlapping reviver (another self-host instance, or a slow prior revive tick still running when the next + // one fires) could flip a row that's already been claimed into 'processing' back to 'pending', letting it + // run a second time concurrently. rowCount is 0 (not counted as revived) when another reviver won the race. + const update = await pool.query( + `UPDATE ${TABLE} SET status='pending', run_after=$1, last_error=NULL WHERE id=$2 AND status='dead'`, + [runAfter, row.id], + ); + revived += update.rowCount ?? 0; } return revived; } diff --git a/src/selfhost/sqlite-queue.ts b/src/selfhost/sqlite-queue.ts index e49da97bea..53d272fe4e 100644 --- a/src/selfhost/sqlite-queue.ts +++ b/src/selfhost/sqlite-queue.ts @@ -686,11 +686,16 @@ function reviveEligibleDeadJobs(driver: SqliteDriver, maxRetries: number): numbe const maxJitter = queueRecoveryJitterMs(); for (const row of rows as Array<{ id: number; payload: string; job_key?: string | null }>) { const runAfter = now + deterministicJitterMs(`revive:${row.job_key ?? ""}:${row.id}:${row.payload}`, maxJitter); - driver.query( - `UPDATE ${TABLE} SET status='pending', run_after=?, last_error=NULL WHERE id=?`, + // AND status='dead' re-checks the row is STILL dead at UPDATE time (mirrors deferPendingJobsForRateLimit / + // the processing-lease reclaim below) — the SELECT above is a stale snapshot, and without this predicate an + // overlapping revive (a slow prior revive tick still running when the next one fires) could flip a row + // that's already been claimed into 'processing' back to 'pending', letting it run a second time concurrently. + // `changes` is 0 (not counted as revived) when the row already moved out of 'dead'. + const { changes } = driver.query( + `UPDATE ${TABLE} SET status='pending', run_after=?, last_error=NULL WHERE id=? AND status='dead'`, [runAfter, row.id], ); - revived += 1; + revived += changes; } return revived; } diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts index 3446bf5cb5..8f52ca9b56 100644 --- a/test/unit/selfhost-pg-queue.test.ts +++ b/test/unit/selfhost-pg-queue.test.ts @@ -58,12 +58,17 @@ interface MockPool { /** Pre-load a job to be returned by the next RETURNING claim query. */ enqueueJob(id: string, payload: object, attempts?: number, jobKey?: string | null): void; setDeferUpdateRowCount(rowCount: number): void; + /** Queues per-call rowCounts for the "AND status='dead'" revive UPDATE, one entry consumed per call in order + * (default 1 when the queue is empty) — lets a test simulate an overlapping reviver already winning the race + * on a specific row (rowCount 0) while another succeeds (rowCount 1). */ + setReviveUpdateRowCounts(rowCounts: number[]): void; setRateLimitRows(rows: Array<{ admission_key?: string | null; repo_full_name?: string | null; remaining: number | string | null; reset_at: string | null; observed_at?: string | null }>): void; } function makePool(): MockPool { const results: Partial[] = []; let deferUpdateRowCount = 1; + const reviveUpdateRowCounts: number[] = []; let rateLimitRows: Array<{ admission_key?: string | null; repo_full_name?: string | null; remaining: number | string | null; reset_at: string | null; observed_at?: string | null }> = []; const fn = vi.fn().mockImplementation(async (sql: unknown, params?: unknown[]) => { const q = String(sql); @@ -84,6 +89,10 @@ function makePool(): MockPool { if (q.includes("SET status='pending', run_after=GREATEST")) { return { rows: [], rowCount: deferUpdateRowCount }; } + if (q.includes("SET status='pending', run_after=$1, last_error=NULL")) { + const rowCount = reviveUpdateRowCounts.length > 0 ? (reviveUpdateRowCounts.shift() ?? 1) : 1; + return { rows: [], rowCount }; + } // Claim queries use RETURNING — pop from queue; fall through to empty default otherwise. if (q.includes("RETURNING")) { const next = results.shift(); @@ -105,6 +114,10 @@ function makePool(): MockPool { setDeferUpdateRowCount(rowCount) { deferUpdateRowCount = rowCount; }, + setReviveUpdateRowCounts(rowCounts) { + reviveUpdateRowCounts.length = 0; + reviveUpdateRowCounts.push(...rowCounts); + }, setRateLimitRows(rows) { rateLimitRows = rows; }, @@ -959,6 +972,40 @@ describe("createPgQueue (durable #977)", () => { expect(revived).toBe(0); expect(await renderMetrics()).not.toContain("gittensory_jobs_dead_letter_revived_total"); }); + + // REGRESSION (#2581 review defect): the SELECT is a stale snapshot. Without an "AND status='dead'" re-check on + // the UPDATE, an overlapping reviver (another self-host instance, or a slow prior tick still running when the + // next one fires) that already moved a row out of 'dead' -- e.g. into 'processing' via a normal claim -- would + // get silently flipped back to 'pending' by this stale UPDATE, letting the job run a second time concurrently. + it("does NOT count a row as revived when another reviver already moved it out of 'dead' (rowCount 0) -- only the row that actually changed status counts", async () => { + const m = makePool(); + m.fn.mockResolvedValueOnce({ + rows: [ + { id: "1", payload: JSON.stringify({ type: "t" }), job_key: null }, + { id: "2", payload: JSON.stringify({ type: "t" }), job_key: "k" }, + ], + rowCount: 2, + }); // SELECT status='dead' AND attempts undefined, { maxRetries: 1 }); + + const revived = await q.reviveDeadLetterJobs(); + + // Only the ONE row whose UPDATE actually matched (still 'dead' at UPDATE time) counts -- not the raw SELECT + // count of 2, which would have double-counted the row another reviver already claimed. + expect(revived).toBe(1); + expect(m.fn).toHaveBeenCalledWith( + expect.stringContaining("SET status='pending', run_after=$1, last_error=NULL WHERE id=$2 AND status='dead'"), + expect.arrayContaining(["1"]), + ); + expect(m.fn).toHaveBeenCalledWith( + expect.stringContaining("SET status='pending', run_after=$1, last_error=NULL WHERE id=$2 AND status='dead'"), + expect.arrayContaining(["2"]), + ); + expect(await renderMetrics()).toContain("gittensory_jobs_dead_letter_revived_total 1"); + }); }); it("reschedules GitHub rate-limit failures without consuming the dead-letter budget", async () => { diff --git a/test/unit/selfhost-sqlite-queue.test.ts b/test/unit/selfhost-sqlite-queue.test.ts index cb39bc3e98..b84f16119e 100644 --- a/test/unit/selfhost-sqlite-queue.test.ts +++ b/test/unit/selfhost-sqlite-queue.test.ts @@ -1257,6 +1257,34 @@ describe("createSqliteQueue (durable #980)", () => { expect(q.reviveDeadLetterJobs()).toBe(0); }); + // REGRESSION (#2581 review defect, parity with the same fix in pg-queue.ts): the SELECT that finds eligible + // dead jobs is a stale snapshot. Without an "AND status='dead'" re-check on the UPDATE, a row that stops + // being 'dead' between the SELECT and this row's own UPDATE (e.g. claimed by an overlapping revive/process) + // would get silently flipped back to 'pending' regardless of its CURRENT status, letting it run a second + // time concurrently. Engineered here via a driver.query spy that injects the status change at the exact + // point the real revive UPDATE would otherwise race against it. + it("does not flip a row back to pending if it stops being 'dead' between the SELECT and its own UPDATE", async () => { + const driver = makeDriver(); + const realQuery = driver.query.bind(driver); + const q = createSqliteQueue(driver, async () => { throw new Error("boom"); }, { maxRetries: 1, backoffMs: () => 0 }); + await q.binding.send(msg("x")); + await q.drain(); // dies at attempts=1 (maxRetries=1) + expect(q.deadCount()).toBe(1); + + vi.spyOn(driver, "query").mockImplementation((sql: string, params: unknown[]) => { + if (sql.includes("SET status='pending', run_after=?, last_error=NULL")) { + realQuery(`UPDATE _selfhost_jobs SET status='processing' WHERE id=?`, [params[1] as number]); + } + return realQuery(sql, params); + }); + + const revived = q.reviveDeadLetterJobs(); + + expect(revived).toBe(0); // the UPDATE's "AND status='dead'" matched zero rows -- not counted as revived + const { rows } = driver.query("SELECT status FROM _selfhost_jobs", []); + expect((rows[0] as { status: string }).status).toBe("processing"); // untouched, NOT reverted to pending + }); + it("runs automatically on the configured revive interval while the queue is running", async () => { process.env.QUEUE_DEAD_LETTER_REVIVE_INTERVAL_MS = "1000"; vi.useFakeTimers();