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).

Author: JSONbored

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;
   }

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;
 }

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;
     },
@@ -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<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 () => {

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();