feat(selfhost): add automatic dead-letter job retry for self-host queues

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{" "}
         <code>Dead jobs stay at zero</code> routine check below is watching for.
       </p>
+      <p>
+        Dead-lettered jobs also get one automatic revival attempt every 30 minutes (
+        <code>QUEUE_DEAD_LETTER_REVIVE_INTERVAL_MS</code>), as long as the job hasn't already been
+        revived more than a small, bounded number of extra times (
+        <code>QUEUE_DEAD_LETTER_AUTO_RETRY_MAX_EXTRA_ATTEMPTS</code>, 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.
+      </p>
 
       <h2>Docker resource hygiene</h2>
       <p>

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<number>;
   stats(): Promise<Record<string, number>>;
   snapshot(): Promise<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(): Promise<number>;
 }
 
 interface JobRow {
@@ -115,6 +121,7 @@ export function createPgQueue(
   let activeBackground = 0;
   const activeJobIds = new Set<string>();
   let timer: ReturnType<typeof setTimeout> | null = null;
+  let deadLetterReviveTimer: ReturnType<typeof setInterval> | null = null;
 
   async function init(): Promise<void> {
     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<number> {
+    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<number> {
+    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<number> {
     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<number> {

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;

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<string, number>;
   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<number>();
   let timer: ReturnType<typeof setTimeout> | null = null;
+  let deadLetterReviveTimer: ReturnType<typeof setInterval> | 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<?`,
+    [ceiling],
+  );
+  let revived = 0;
+  const now = Date.now();
+  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);
+    // 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(

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<QueryResult>[] = [];
   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<ceiling
+      const q = createPgQueue(m.pool, async () => 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<ceiling -- a stale snapshot of both rows
+      // Row "1" lost the race (another reviver/claim already moved it out of 'dead' -- UPDATE affects 0 rows);
+      // row "2" is still genuinely dead and gets revived.
+      m.setReviveUpdateRowCounts([0, 1]);
+      const q = createPgQueue(m.pool, async () => 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);