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..e02990a1be 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,48 @@ 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); + // 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; + } + + 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 +663,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 +700,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..53d272fe4e 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,37 @@ 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); + // 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 += changes; + } + 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..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; }, @@ -914,6 +927,87 @@ 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"); + }); + + // 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 () => { 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..b84f16119e 100644 --- a/test/unit/selfhost-sqlite-queue.test.ts +++ b/test/unit/selfhost-sqlite-queue.test.ts @@ -1190,6 +1190,128 @@ 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); + }); + + // 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(); + 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;