From f94d1da0cab960da9a536c4c0b1926b8bee8b119 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 5 Nov 2026 03:21:02 +0000
Subject: [PATCH 1/9] fix(service-queue): reclaim a running job whose lease ran
 out, and renew a live handler's lease

Claude-Session: https://claude.ai/code/session_01Utp6CaNicTDZwYQgDRwD7a
Co-authored-by: Claude <noreply@anthropic.com>
---
 .../service-queue/src/db-queue-adapter.ts     | 210 ++++++++++++++++-
 .../src/db-queue-idle-backoff.test.ts         | 215 +++++++++++++++++-
 2 files changed, 406 insertions(+), 19 deletions(-)

diff --git a/packages/services/service-queue/src/db-queue-adapter.ts b/packages/services/service-queue/src/db-queue-adapter.ts
index 67be673ab50..12b18edfbae 100644
--- a/packages/services/service-queue/src/db-queue-adapter.ts
+++ b/packages/services/service-queue/src/db-queue-adapter.ts
@@ -155,7 +155,23 @@ export interface DbQueueAdapterOptions {
   maxIdleIntervalMs?: number;
   /** Max messages claimed per poll tick (default 10) */
   batchSize?: number;
-  /** Lease duration before another worker may reclaim (ms, default 30000) */
+  /**
+   * Lease duration before another worker may reclaim (ms, default 30000).
+   *
+   * [#30228] A claimed message is `running` under this worker until
+   * `locked_until`. A LIVE handler never loses it however long it runs: the
+   * worker renews the lease every third of this window while the handler is
+   * inside it. A lease that runs out
+   * therefore means the worker is gone — a crash, an OOM kill, a redeploy — or
+   * stalled its event loop for most of a window; any worker reclaims such a
+   * row on its next poll tick. The lost run
+   * counts as an attempt: the row returns to `pending` on its backoff, or to
+   * the dead-letter surface once `maxAttempts` is spent, so a message that
+   * kills its worker every time cannot loop forever.
+   *
+   * Set it above the longest synchronous stretch a handler can block the
+   * event loop for: renewal is a timer, and a blocked loop renews nothing.
+   */
   leaseMs?: number;
   /**
    * Idempotency window — how long the same key blocks re-publish (ms, default 24h).
@@ -186,7 +202,9 @@ interface RegisteredHandler {
  * Persists every message to `sys_job_queue`. A polling worker leases
  * pending messages (CAS update status pending→running with a lease),
  * invokes registered subscribers, and retries with backoff on failure.
- * Messages that exceed `max_attempts` land in `status='dlq'`.
+ * Messages that exceed `max_attempts` land in `status='dlq'`. A `running`
+ * row whose lease ran out — its worker died mid-handler — is reclaimed onto
+ * that same retry path (#30228, {@link DbQueueAdapterOptions.leaseMs}).
  *
  * Idempotency: publish suppresses duplicates within a configurable
  * window when `(queue, idempotencyKey)` is non-null.
@@ -211,6 +229,13 @@ export class DbQueueAdapter implements IQueueService {
 
   private readonly handlers = new Map<string, RegisteredHandler[]>();
   private loop?: DispatchLoop;
+  /**
+   * [#30228] Adapter-clock instant (ms) before which no `running` row of a
+   * subscribed queue can have an expired lease, so {@link reclaimExpired}
+   * need not read. `0` = read on the next tick. See that method for why the
+   * bound is sound.
+   */
+  private reclaimNotBeforeMs = 0;
 
   constructor(args: {
     engine: JobEngine;
@@ -502,12 +527,16 @@ export class DbQueueAdapter implements IQueueService {
     this.loop?.wake();
   }
 
-  /** Test-friendly synchronous poll. */
+  /**
+   * Test-friendly synchronous poll. Answers how many rows the tick moved:
+   * messages dispatched, plus expired leases reclaimed (#30228) — a reclaimed
+   * row is work the next tick claims, so the loop must not back off over it.
+   */
   async pollOnce(): Promise<number> {
     const queues = [...this.handlers.keys()];
     if (queues.length === 0) return 0;
 
-    let processed = 0;
+    let processed = await this.reclaimExpired(queues);
     for (const queue of queues) {
       const claimed = await this.claimBatch(queue, this.opts.batchSize);
       for (const row of claimed) {
@@ -520,6 +549,160 @@ export class DbQueueAdapter implements IQueueService {
 
   // ── Internals ────────────────────────────────────────────────────
 
+  /**
+   * [#30228] Return every `running` row of a subscribed queue whose lease ran
+   * out to the retry path, and answer how many were reclaimed.
+   *
+   * ## Why a lease that ran out means a dead worker, not a slow one
+   *
+   * {@link dispatch} renews its lease every third of `leaseMs` for as long as
+   * the handler runs ({@link renewLease}). So `locked_until` passes only when
+   * the holder stopped renewing — the process died, or its event loop was
+   * blocked for most of a window. Without the renewal, any handler slower than
+   * `leaseMs` would be reclaimed while still running and run twice.
+   *
+   * ## Why the lost run counts as an attempt
+   *
+   * The handler's attempt is written only after it settles, so a worker that
+   * died left the row's `attempts` one short. Reclaiming without counting would
+   * retry forever a message that kills its worker every time (an OOM on a
+   * large payload). So the reclaim writes that attempt down and takes the same
+   * branch a thrown handler takes: `pending` on its backoff, or `dlq` once
+   * `max_attempts` is spent.
+   *
+   * ## Why the write is a compare-and-set
+   *
+   * The write re-states the expiry in its predicate (`status`, `locked_by` and
+   * an expired `locked_until`), so a lease renewed between this read and this
+   * write — a holder that was alive after all — matches nothing and keeps its
+   * run. It is a `multi: true` write on purpose: the by-id path binds only
+   * the primary key and would drop the rest of the predicate. Two workers
+   * reclaiming the same row write the same values from the same read; neither
+   * runs it, so that race costs nothing.
+   *
+   * ## Why it does not read on every tick
+   *
+   * One read answers every subscribed queue, and it also returns the
+   * earliest lease still running. Until the earlier of that lease and
+   * `now + leaseMs` no lease can run out: every row already `running` was
+   * in the read, and a row claimed after it carries a lease of at least
+   * `leaseMs` from its claim — every node of one deployment runs the same
+   * `QueueServicePlugin` options — and renewal only ever moves a lease later. So
+   * a busy 1 s poll pays one read per lease window, an idle one at most one
+   * per tick, and an expired lease is still reclaimed on the first tick after
+   * it runs out.
+   */
+  private async reclaimExpired(queues: string[]): Promise<number> {
+    const now = this.now();
+    const nowMs = now.getTime();
+    if (nowMs < this.reclaimNotBeforeMs) return 0;
+    const max = this.opts.batchSize;
+    const rows = await this.engine.find(QUEUE_TABLE, {
+      where: { queue: { $in: queues }, status: 'running' },
+      fields: [
+        'id', 'locked_by', 'locked_until', 'attempts', 'max_attempts',
+        'backoff_type', 'backoff_delay_ms', 'backoff_max_delay_ms',
+      ],
+      orderBy: [{ field: 'locked_until', order: 'asc' }],
+      limit: max,
+      context: SYSTEM_CTX,
+    });
+
+    let nextExpiryMs = nowMs + this.opts.leaseMs;
+    let reclaimed = 0;
+    for (const row of rows ?? []) {
+      // Instants, not strings: `locked_until` is a declared `Field.datetime`
+      // and the dialects materialize it differently (see claimBatch). A
+      // `running` row with no lease at all was never validly claimed, so it is
+      // as orphaned as an expired one.
+      const leaseMs = row.locked_until ? new Date(row.locked_until).getTime() : NaN;
+      if (Number.isFinite(leaseMs) && leaseMs > nowMs) {
+        nextExpiryMs = Math.min(nextExpiryMs, leaseMs);
+        continue;
+      }
+      if (await this.reclaimRow(row, now)) reclaimed++;
+    }
+    // A full page may hide more expired rows behind it: read again next tick.
+    this.reclaimNotBeforeMs = (rows?.length ?? 0) >= max ? 0 : nextExpiryMs;
+    return reclaimed;
+  }
+
+  /** [#30228] The compare-and-set write of {@link reclaimExpired}, one row. */
+  private async reclaimRow(row: any, now: Date): Promise<boolean> {
+    const nowIso = now.toISOString();
+    const attempts = (row.attempts ?? 0) + 1;
+    const maxAttempts = row.max_attempts ?? this.opts.defaultMaxAttempts;
+    const lastError = `lease expired: worker ${row.locked_by ?? '(none)'} did not finish within its lease `
+      + '(process crash, OOM kill or redeploy); reclaimed as a failed attempt';
+    const patch = attempts >= maxAttempts
+      ? { status: 'dlq', completed_at: nowIso }
+      : { status: 'pending', scheduled_for: new Date(now.getTime() + this.computeBackoff(row, attempts)).toISOString() };
+    try {
+      const res = await this.engine.update(QUEUE_TABLE, {
+        ...patch,
+        attempts,
+        last_error: lastError,
+        locked_by: null,
+        locked_until: null,
+        updated_at: nowIso,
+      }, {
+        where: {
+          id: row.id,
+          status: 'running',
+          locked_by: row.locked_by ?? null,
+          $or: [{ locked_until: null }, { locked_until: { $lte: nowIso } }],
+        },
+        multi: true,
+        context: SYSTEM_CTX,
+      });
+      const landed = typeof res === 'number' ? res > 0 : true;
+      if (landed) {
+        this.logger?.warn?.('DbQueueAdapter: reclaimed a message whose lease expired', {
+          id: row.id, lockedBy: row.locked_by, attempts, maxAttempts, status: patch.status,
+        } as any);
+      }
+      return landed;
+    } catch (err) {
+      this.logger?.warn?.('DbQueueAdapter: reclaim CAS failed', err as any);
+      return false;
+    }
+  }
+
+  /**
+   * [#30228] Keep this worker's lease on a row while its handler runs: every
+   * third of `leaseMs`, move `locked_until` a full window ahead. The write is
+   * a compare-and-set on `locked_by` — `multi: true` for the reason
+   * {@link reclaimExpired} gives — so a lease that was reclaimed underneath
+   * (this worker stalled past it) is not silently taken back. Returns the
+   * stop function {@link dispatch} calls once the handler settles.
+   */
+  private renewLease(id: string): () => void {
+    let inFlight = false;
+    const timer = setInterval(() => {
+      if (inFlight) return;
+      inFlight = true;
+      const now = this.now();
+      void Promise.resolve()
+        .then(() => this.engine.update(QUEUE_TABLE, {
+          locked_until: new Date(now.getTime() + this.opts.leaseMs).toISOString(),
+          updated_at: now.toISOString(),
+        }, {
+          where: { id, status: 'running', locked_by: this.opts.workerId },
+          multi: true,
+          context: SYSTEM_CTX,
+        }))
+        .then((res) => {
+          if (typeof res === 'number' && res === 0) {
+            this.logger?.warn?.('DbQueueAdapter: lease lost while the handler was still running', { id } as any);
+          }
+        })
+        .catch((err) => { this.logger?.warn?.('DbQueueAdapter: lease renewal failed', err as any); })
+        .finally(() => { inFlight = false; });
+    }, Math.max(1, Math.floor(this.opts.leaseMs / 3)));
+    (timer as { unref?: () => void }).unref?.();
+    return () => clearInterval(timer);
+  }
+
   private async claimBatch(queue: string, max: number): Promise<any[]> {
     const now = this.now();
     const candidates = await this.engine.find(QUEUE_TABLE, {
@@ -604,14 +787,19 @@ export class DbQueueAdapter implements IQueueService {
 
     let success = true;
     let lastError: string | undefined;
-    for (const h of handlers) {
-      try { await h.fn(msg); }
-      catch (err) {
-        success = false;
-        lastError = err instanceof Error ? err.message : String(err);
-        this.logger?.warn?.(`DbQueueAdapter: handler failed on ${row.queue}`, err as any);
-        break;
+    const stopRenewing = this.renewLease(String(row.id));
+    try {
+      for (const h of handlers) {
+        try { await h.fn(msg); }
+        catch (err) {
+          success = false;
+          lastError = err instanceof Error ? err.message : String(err);
+          this.logger?.warn?.(`DbQueueAdapter: handler failed on ${row.queue}`, err as any);
+          break;
+        }
       }
+    } finally {
+      stopRenewing();
     }
 
     const now = this.now();
diff --git a/packages/services/service-queue/src/db-queue-idle-backoff.test.ts b/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
index cf95f5f818a..6df9474d084 100644
--- a/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
+++ b/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
@@ -113,8 +113,16 @@ function makeCountingEngine() {
             // the same reason `delete` below is: a predicate update rewrites
             // every matching row, so a double looser than the engine here hides
             // exactly the writes it was introduced to observe.
-            assertEngineUpdateDispatch(patch, options);
+            const dispatch = assertEngineUpdateDispatch(patch, options);
             if (table === QUEUE_TABLE) calls.update++;
+            // [#30228] A predicate write rewrites the rows its `where` matches
+            // and reports how many — the compare-and-set the lease reclaim and
+            // renewal ride on, which a by-id double would turn unconditional.
+            if (dispatch.kind === 'multi') {
+                const hit = (tables.get(table) ?? []).filter((x) => matches(x, options?.where ?? {}));
+                for (const x of hit) Object.assign(x, patch);
+                return hit.length;
+            }
             const r = (tables.get(table) ?? []).find((x) => x.id === patch.id);
             if (!r) throw new Error(`row ${patch.id} not found in ${table}`);
             Object.assign(r, patch);
@@ -162,11 +170,12 @@ describe('#17612 DbQueueAdapter — idle poll cost', () => {
         await vi.advanceTimersByTimeAsync(HOUR);
         await adapter.stop();
 
-        // 3600 s / 1 s, plus the tick `start()` runs immediately.
-        expect(engine.calls.find).toBe(3601);
+        // 3600 s / 1 s, plus the tick `start()` runs immediately — and [#30228]
+        // one lease-reclaim read per 30 s lease window (t = 0, 30 s, … 3600 s).
+        expect(engine.calls.find).toBe(3601 + 121);
     });
 
-    it('backs off while idle: the same hour, same queue, is 124 reads instead of 3600', async () => {
+    it('backs off while idle: the same hour, same queue, is 124 claim reads instead of 3600', async () => {
         vi.useFakeTimers();
         const engine = makeCountingEngine();
         const adapter = adapterOn(engine);
@@ -176,19 +185,26 @@ describe('#17612 DbQueueAdapter — idle poll cost', () => {
         await adapter.stop();
 
         // 1s,2s,4s,8s,16s then the 30 s ceiling — pinned exactly, because a
-        // bound alone is met by a loop that stopped.
-        expect(engine.calls.find).toBe(124);
+        // bound alone is met by a loop that stopped. [#30228] Plus the
+        // lease-reclaim read, at most once per 30 s lease window: on the first
+        // tick and on every ceiling tick (the 1/3/7/15 s ramp ticks fall inside
+        // the first window), 124 - 4 = 120 of them.
+        expect(engine.calls.find).toBe(124 + 120);
         // Nothing was claimed, so nothing was written.
         expect(engine.calls.update).toBe(0);
     });
 
-    it('costs exactly one read per registered queue per tick', async () => {
+    it('costs exactly one read per registered queue per tick, plus one lease read per lease window', async () => {
         for (const n of [1, 3]) {
             const engine = makeCountingEngine();
             const adapter = adapterOn(engine);
             for (let i = 0; i < n; i++) await adapter.subscribe(`q${i}`, async () => {});
             expect(await adapter.pollOnce()).toBe(0);
-            expect(engine.calls.find).toBe(n);
+            // [#30228] One reclaim read answers every queue at once…
+            expect(engine.calls.find).toBe(n + 1);
+            // …and a tick inside the same lease window skips it.
+            expect(await adapter.pollOnce()).toBe(0);
+            expect(engine.calls.find).toBe(2 * n + 1);
         }
     });
 
@@ -288,3 +304,186 @@ describe('#17612 DbQueueAdapter — idle poll cost', () => {
         expect(engine.calls.find).toBe(findsAtStop);
     });
 });
+
+/**
+ * [#30228] A worker that dies mid-handler — a crash, an OOM kill, a redeploy —
+ * leaves its row `running`. `leaseMs` documents that another worker reclaims
+ * it once the lease runs out; before #30228 nothing did, and the row sat
+ * `running` forever: never retried, never dead-lettered.
+ *
+ * The CRASHED-WORKER FIXTURE is that row written straight into the table: the
+ * state a dead process leaves behind, with no adapter alive to have written
+ * it. Measured before the fix (this file's legs run against
+ * `origin/main`'s adapter): the fixture stays `running` with `attempts: 0`
+ * through an hour of ticks and its handler never runs.
+ *
+ * The risk the fix itself introduces is the opposite one — reclaiming a row
+ * whose handler is still running, and running it twice. The live-handler legs
+ * pin that against a handler that runs three leases long.
+ */
+describe('#30228 DbQueueAdapter — a running row whose lease ran out is reclaimed', () => {
+    const LEASE = 3_000;
+
+    function crashedRow(over: Row = {}): Row {
+        const nowMs = Date.now();
+        return {
+            id: 'msg_crashed',
+            queue: 'q1',
+            payload_json: JSON.stringify({ n: 1 }),
+            status: 'running',
+            priority: 100,
+            attempts: 0,
+            max_attempts: 3,
+            backoff_type: 'exponential',
+            backoff_delay_ms: 1000,
+            backoff_max_delay_ms: null,
+            scheduled_for: new Date(nowMs - 60_000).toISOString(),
+            locked_by: 'worker_dead',
+            locked_until: new Date(nowMs - 1).toISOString(),
+            created_at: new Date(nowMs - 60_000).toISOString(),
+            ...over,
+        };
+    }
+
+    it('CRASHED-WORKER FIXTURE: an expired lease goes back to the retry path with the lost run counted, then runs', async () => {
+        vi.useFakeTimers();
+        vi.setSystemTime(1_000_000);
+        const engine = makeCountingEngine();
+        engine.tables.set(QUEUE_TABLE, [crashedRow()]);
+        const adapter = adapterOn(engine, { leaseMs: LEASE });
+        const seen: Array<{ data: unknown; attempts: number }> = [];
+        await adapter.subscribe('q1', async (msg: any) => { seen.push({ data: msg.data, attempts: msg.attempts }); });
+
+        // One tick reclaims it — and reports the row as moved, so the loop does
+        // not back off over work it just made claimable.
+        expect(await adapter.pollOnce()).toBe(1);
+        const [row] = engine.tables.get(QUEUE_TABLE) ?? [];
+        expect(row).toMatchObject({
+            status: 'pending',
+            attempts: 1,
+            locked_by: null,
+            locked_until: null,
+            // exponential, base 1 s, attempt 1 → 1 s
+            scheduled_for: new Date(1_000_000 + 1000).toISOString(),
+        });
+        expect(row!.last_error).toMatch(/^lease expired: worker worker_dead /);
+        expect(seen).toEqual([]);
+
+        // On its backoff the ordinary claim path runs it, as attempt 2.
+        vi.setSystemTime(1_000_000 + 1000);
+        expect(await adapter.pollOnce()).toBe(1);
+        expect(seen).toEqual([{ data: { n: 1 }, attempts: 2 }]);
+        expect(row).toMatchObject({ status: 'completed', attempts: 2 });
+    });
+
+    it('a lease that runs out while the loop is deep in its idle backoff is reclaimed within one idle tick', async () => {
+        vi.useFakeTimers();
+        vi.setSystemTime(0);
+        const engine = makeCountingEngine();
+        const adapter = adapterOn(engine, { leaseMs: LEASE });
+        const at: number[] = [];
+        await adapter.subscribe('q1', async () => { at.push(Date.now()); });
+        adapter.start();
+        await vi.advanceTimersByTimeAsync(5 * 60_000);
+
+        // Another node claimed a row and died: its lease runs out 10 s from now,
+        // and nothing tells this process about it.
+        const expiresAt = Date.now() + 10_000;
+        engine.tables.set(QUEUE_TABLE, [crashedRow({ locked_until: new Date(expiresAt).toISOString() })]);
+        await vi.advanceTimersByTimeAsync(10_000 + DEFAULT_MAX_IDLE_INTERVAL_MS + 2 * BASE);
+        await adapter.stop();
+
+        const [row] = engine.tables.get(QUEUE_TABLE) ?? [];
+        expect(row).toMatchObject({ status: 'completed', attempts: 2 });
+        // Reclaimed on the first tick after expiry (≤ one 30 s idle interval),
+        // then claimed on the base interval the reclaim snapped the loop to.
+        expect(at).toHaveLength(1);
+        expect(at[0]! - expiresAt).toBeLessThanOrEqual(DEFAULT_MAX_IDLE_INTERVAL_MS + 2 * BASE);
+    });
+
+    it('the lost run counts: a row whose last attempt died with its worker is dead-lettered, not retried forever', async () => {
+        vi.useFakeTimers();
+        vi.setSystemTime(1_000_000);
+        const engine = makeCountingEngine();
+        engine.tables.set(QUEUE_TABLE, [crashedRow({ attempts: 2, max_attempts: 3 })]);
+        const adapter = adapterOn(engine, { leaseMs: LEASE });
+        let ran = 0;
+        await adapter.subscribe('q1', async () => { ran++; });
+
+        expect(await adapter.pollOnce()).toBe(1);
+        const [row] = engine.tables.get(QUEUE_TABLE) ?? [];
+        expect(row).toMatchObject({ status: 'dlq', attempts: 3, locked_by: null, locked_until: null });
+        expect(row!.last_error).toMatch(/^lease expired: /);
+        expect(await adapter.listFailed('q1')).toHaveLength(1);
+        await adapter.pollOnce();
+        expect(ran).toBe(0);
+    });
+
+    it('a row INSIDE its lease is never reclaimed', async () => {
+        vi.useFakeTimers();
+        vi.setSystemTime(1_000_000);
+        const engine = makeCountingEngine();
+        const live = crashedRow({ locked_by: 'worker_live', locked_until: new Date(1_000_000 + 10_000).toISOString() });
+        engine.tables.set(QUEUE_TABLE, [{ ...live }]);
+        const adapter = adapterOn(engine, { leaseMs: LEASE });
+        let ran = 0;
+        await adapter.subscribe('q1', async () => { ran++; });
+
+        expect(await adapter.pollOnce()).toBe(0);
+        expect((engine.tables.get(QUEUE_TABLE) ?? [])[0]).toEqual(live);
+        expect(ran).toBe(0);
+        expect(engine.calls.update).toBe(0);
+    });
+
+    it('the reclaim is a compare-and-set: a lease renewed between its read and its write keeps its run', async () => {
+        vi.useFakeTimers();
+        vi.setSystemTime(1_000_000);
+        const engine = makeCountingEngine();
+        engine.tables.set(QUEUE_TABLE, [crashedRow({ locked_by: 'worker_slow' })]);
+        // The holder renews right after the reclaim read saw the lease expired.
+        const find = engine.find.bind(engine);
+        engine.find = async (table: string, opts: any = {}) => {
+            const out = await find(table, opts);
+            if (opts?.where?.status === 'running') {
+                const [held] = engine.tables.get(QUEUE_TABLE) ?? [];
+                held!.locked_until = new Date(1_000_000 + LEASE).toISOString();
+            }
+            return out;
+        };
+        const adapter = adapterOn(engine, { leaseMs: LEASE });
+        await adapter.subscribe('q1', async () => {});
+
+        expect(await adapter.pollOnce()).toBe(0);
+        expect((engine.tables.get(QUEUE_TABLE) ?? [])[0]).toMatchObject({
+            status: 'running', locked_by: 'worker_slow', attempts: 0,
+        });
+    });
+
+    it('a LIVE handler three leases long is never double-claimed: its worker renews the lease while it runs', async () => {
+        vi.useFakeTimers();
+        vi.setSystemTime(0);
+        const engine = makeCountingEngine();
+        const runs: string[] = [];
+        const slow = (who: string) => async () => {
+            runs.push(who);
+            await new Promise((r) => setTimeout(r, 3 * LEASE));
+        };
+        const a = adapterOn(engine, { leaseMs: LEASE, workerId: 'worker_a' });
+        const b = adapterOn(engine, { leaseMs: LEASE, workerId: 'worker_b' });
+        await a.subscribe('q1', slow('a'));
+        await b.subscribe('q1', slow('b'));
+        await a.publish('q1', { n: 1 });
+        a.start();
+        // A is inside the handler before B starts polling the same table.
+        await vi.advanceTimersByTimeAsync(10);
+        expect(runs).toEqual(['a']);
+        b.start();
+
+        await vi.advanceTimersByTimeAsync(3 * LEASE + 5 * BASE);
+        await a.stop();
+        await b.stop();
+
+        expect(runs).toEqual(['a']);
+        expect((engine.tables.get(QUEUE_TABLE) ?? [])[0]).toMatchObject({ status: 'completed', attempts: 1 });
+    });
+});

From 198be8f5bfe1514f760d51cab16818b3bb70d49b Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 5 Nov 2026 03:22:02 +0000
Subject: [PATCH 2/9] test(service-queue): idle-cost pin counts the lease read
 per window

Claude-Session: https://claude.ai/code/session_01Utp6CaNicTDZwYQgDRwD7a
Co-authored-by: Claude <noreply@anthropic.com>
---
 .../service-queue/src/db-queue-idle-backoff.test.ts         | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/packages/services/service-queue/src/db-queue-idle-backoff.test.ts b/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
index 6df9474d084..cfcd4f9fa06 100644
--- a/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
+++ b/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
@@ -187,9 +187,9 @@ describe('#17612 DbQueueAdapter — idle poll cost', () => {
         // 1s,2s,4s,8s,16s then the 30 s ceiling — pinned exactly, because a
         // bound alone is met by a loop that stopped. [#30228] Plus the
         // lease-reclaim read, at most once per 30 s lease window: on the first
-        // tick and on every ceiling tick (the 1/3/7/15 s ramp ticks fall inside
-        // the first window), 124 - 4 = 120 of them.
-        expect(engine.calls.find).toBe(124 + 120);
+        // tick and on every ceiling tick (the ramp ticks at 2, 6 and 14 s fall
+        // inside the first window), 124 - 3 = 121 of them.
+        expect(engine.calls.find).toBe(124 + 121);
         // Nothing was claimed, so nothing was written.
         expect(engine.calls.update).toBe(0);
     });

From fc4de71ac3a4490bfcab9e0045680fa7b52e5f5b Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 5 Nov 2026 03:22:39 +0000
Subject: [PATCH 3/9] test(service-queue): live-handler leg outlasts a second
 run before stopping

Claude-Session: https://claude.ai/code/session_01Utp6CaNicTDZwYQgDRwD7a
Co-authored-by: Claude <noreply@anthropic.com>
---
 .../services/service-queue/src/db-queue-idle-backoff.test.ts  | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/packages/services/service-queue/src/db-queue-idle-backoff.test.ts b/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
index cfcd4f9fa06..71df735d46d 100644
--- a/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
+++ b/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
@@ -479,7 +479,9 @@ describe('#30228 DbQueueAdapter — a running row whose lease ran out is reclaim
         expect(runs).toEqual(['a']);
         b.start();
 
-        await vi.advanceTimersByTimeAsync(3 * LEASE + 5 * BASE);
+        // Past A's run and past any second run a reclaim would have started, so
+        // `stop()` — which waits out an in-flight tick — has nothing left to wait for.
+        await vi.advanceTimersByTimeAsync(7 * LEASE + 5 * BASE);
         await a.stop();
         await b.stop();
 

From 15565f9c9c440c608cfb3b0dd5f942a490b9985b Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 5 Nov 2026 03:23:38 +0000
Subject: [PATCH 4/9] test(service-queue): the CAS leg reads a snapshot, as a
 driver does

Claude-Session: https://claude.ai/code/session_01Utp6CaNicTDZwYQgDRwD7a
Co-authored-by: Claude <noreply@anthropic.com>
---
 .../services/service-queue/src/db-queue-idle-backoff.test.ts  | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/packages/services/service-queue/src/db-queue-idle-backoff.test.ts b/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
index 71df735d46d..ac8f8d77e5e 100644
--- a/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
+++ b/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
@@ -443,7 +443,9 @@ describe('#30228 DbQueueAdapter — a running row whose lease ran out is reclaim
         // The holder renews right after the reclaim read saw the lease expired.
         const find = engine.find.bind(engine);
         engine.find = async (table: string, opts: any = {}) => {
-            const out = await find(table, opts);
+            // Copies, as a driver hands out: the read is a snapshot, so the
+            // renewal below changes the table and not what the reclaim saw.
+            const out = (await find(table, opts)).map((r) => ({ ...r }));
             if (opts?.where?.status === 'running') {
                 const [held] = engine.tables.get(QUEUE_TABLE) ?? [];
                 held!.locked_until = new Date(1_000_000 + LEASE).toISOString();

From e40e04854f6203835d2a7d67e6fd1f3f0b02d8aa Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 5 Nov 2026 03:24:26 +0000
Subject: [PATCH 5/9] chore(service-queue): changeset for the expired-lease
 reclaim

Claude-Session: https://claude.ai/code/session_01Utp6CaNicTDZwYQgDRwD7a
Co-authored-by: Claude <noreply@anthropic.com>
---
 .changeset/30228-queue-reclaim-expired-lease.md    | 14 ++++++++++++++
 .../src/db-queue-idle-backoff.test.ts              |  7 ++++---
 2 files changed, 18 insertions(+), 3 deletions(-)
 create mode 100644 .changeset/30228-queue-reclaim-expired-lease.md

diff --git a/.changeset/30228-queue-reclaim-expired-lease.md b/.changeset/30228-queue-reclaim-expired-lease.md
new file mode 100644
index 00000000000..2361c708226
--- /dev/null
+++ b/.changeset/30228-queue-reclaim-expired-lease.md
@@ -0,0 +1,14 @@
+---
+'@objectstack/service-queue': patch
+---
+
+fix(service-queue): a job whose worker died mid-handler is reclaimed once its lease runs out, instead of staying `running` forever
+
+Clause-②: no — restores the documented `leaseMs` reclaim; no accept set or export widens
+Metadata-protocol: none
+
+`DbQueueAdapterOptions.leaseMs` documented a lease "before another worker may reclaim", but the claim read only `pending` rows, so a worker that crashed, was OOM-killed or was redeployed mid-handler left its `sys_job_queue` row `running` for good: never retried, never dead-lettered.
+
+- **Reclaim.** Each poll tick reads the `running` rows of its subscribed queues whose `locked_until` has passed (one read for all queues, at most once per lease window) and returns them to the retry path. The lost run counts as an attempt: the row goes back to `pending` on its backoff, or to `dlq` once `maxAttempts` is spent, with `last_error` saying the lease expired. The write is a compare-and-set on the expired lease, so a holder that renewed in the meantime keeps its run.
+- **Renewal.** While a handler runs, its worker moves its lease a full window ahead every third of `leaseMs`, so a slow but live handler is never reclaimed and run twice. A handler must not block the event loop for most of a lease window.
+- **Cost.** An idle queue pays one more indexed read per lease window (default 30 s): an idle hour at the defaults goes from 124 to 245 reads.
diff --git a/packages/services/service-queue/src/db-queue-idle-backoff.test.ts b/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
index ac8f8d77e5e..9f3ce5b4c41 100644
--- a/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
+++ b/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
@@ -313,9 +313,10 @@ describe('#17612 DbQueueAdapter — idle poll cost', () => {
  *
  * The CRASHED-WORKER FIXTURE is that row written straight into the table: the
  * state a dead process leaves behind, with no adapter alive to have written
- * it. Measured before the fix (this file's legs run against
- * `origin/main`'s adapter): the fixture stays `running` with `attempts: 0`
- * through an hour of ticks and its handler never runs.
+ * it. Measured before the fix (these legs against the pre-fix adapter): the
+ * three reclaim legs fail — the fixture stays `running` with `attempts: 0`,
+ * a tick reports nothing moved, and its handler never runs. The other three
+ * pass there too, trivially, because nothing reclaimed anything.
  *
  * The risk the fix itself introduces is the opposite one — reclaiming a row
  * whose handler is still running, and running it twice. The live-handler legs

From b95311b2fb032049f7d2f76a6942efdd7588d980 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 5 Nov 2026 03:40:11 +0000
Subject: [PATCH 6/9] docs(permissions): tenant-audit census counts the two
 lease write sites

Claude-Session: https://claude.ai/code/session_01Utp6CaNicTDZwYQgDRwD7a
Co-authored-by: Claude <noreply@anthropic.com>
---
 .../docs/permissions/tenant-audit-census.mdx  | 38 +++++++++----------
 ...08-tenant-audit-write-call-sites.counts.md | 22 +++++------
 2 files changed, 30 insertions(+), 30 deletions(-)

diff --git a/content/docs/permissions/tenant-audit-census.mdx b/content/docs/permissions/tenant-audit-census.mdx
index 8b7b5d30ff4..ff22ac4cded 100644
--- a/content/docs/permissions/tenant-audit-census.mdx
+++ b/content/docs/permissions/tenant-audit-census.mdx
@@ -132,7 +132,7 @@ are reported as `undecidable` rather than assumed either way.
 
 The same holds twice over for the context. An options argument spelled as a
 literal can be read; one spelled `options`, `{ ...opts }`, or handed through a
-forwarding shim cannot, and **55 of the 259 sites are spelled that way**. A
+forwarding shim cannot, and **55 of the 261 sites are spelled that way**. A
 context resolved from an inline literal or a local `const` can be tested for
 `isSystem`; one arriving from a helper call cannot.
 
@@ -197,10 +197,10 @@ reproduce them. Where it disagrees, it disagrees on the page:
 
 | carried figure | where it survives | this census |
 | :--- | :--- | ---: |
-| 175 write call sites | quoted in the merged changeset | **259** |
+| 175 write call sites | quoted in the merged changeset | **261** |
 | 24 carrying no tenant context | quoted in the merged changeset | **2** provable and tenancy-enabled; **29** more whose options argument is unreadable |
-| 127 of 175 statically decidable, 48 runtime-parameter-name sites | restated on the `isSystem`-scoping card | **182 of 259** decidable, **77** undecidable |
-| 135 (77%) silenced by the `isSystem` guard before the posture gate | the lost issue body — **no surviving corroboration** | **not reproduced**: 144 decidably elevated, 0 decidably not, 107 undecidable |
+| 127 of 175 statically decidable, 48 runtime-parameter-name sites | restated on the `isSystem`-scoping card | **184 of 261** decidable, **77** undecidable |
+| 135 (77%) silenced by the `isSystem` guard before the posture gate | the lost issue body — **no surviving corroboration** | **not reproduced**: 144 decidably elevated, 0 decidably not, 109 undecidable |
 | 141 and 132, two independent re-derivations | the card that filed this work | — |
 
 **The differences are not reconciled, and deliberately so.** The old census's
@@ -211,17 +211,17 @@ at any commit.
 
 Two structural facts do plausibly widen this reading against any hand or regex
 one, and both are counted in the generated tables below: the 62 sites reached
-through an erased (`any`) receiver, and the 66 that name their object through a
+through an erased (`any`) receiver, and the 68 that name their object through a
 `const` rather than inline. An instrument that read either the way a person does
 would report a smaller number and would not say so.
 
 The fourth row is the one worth flagging to anyone citing it. **The 135 / 77%
 figure has no surviving corroboration anywhere in the tree.** This census reads
-144 of 259 (56%) as decidably elevated, with 107 more whose elevation is a
+144 of 261 (55%) as decidably elevated, with 109 more whose elevation is a
 run-time fact — so the claim is neither confirmed nor refuted, and the honest
 answer is that a static reading cannot settle it.
 
-⇒ **Cite `2 / 259`, and say what it is**: the sites whose options argument was
+⇒ **Cite `2 / 261`, and say what it is**: the sites whose options argument was
 READ and holds no tenant context, against a decidably tenancy-enabled object.
 That is the control's provable yield surface. ⛔ Do not cite it as "the sites
 without tenant context" — **29 further sites** have an options argument this
@@ -233,29 +233,29 @@ cannot read, and they are neither in nor out.
 
 | what | count |
 | :--- | ---: |
-| write call sites on the application surface | **259** |
-| …whose object name is statically decidable | 182 |
+| write call sites on the application surface | **261** |
+| …whose object name is statically decidable | 184 |
 | …whose object name is chosen at run time | 77 |
-| …against an object with tenancy ENABLED | 176 |
+| …against an object with tenancy ENABLED | 178 |
 | …against an object that declares tenancy off | 6 |
-| threading a tenant context | 196 |
+| threading a tenant context | 198 |
 | PROVABLY carrying none (options read, no context key) | **8** |
 | …of those, against a decidably tenancy-enabled object | **2** |
 | options argument UNREADABLE — may or may not carry one | 55 |
 | …of those, against a decidably tenancy-enabled object | 29 |
 | threading a decidably ELEVATED (`isSystem`) context | 144 |
 | threading a context that is decidably NOT elevated | 0 |
-| threading a context whose elevation is a run-time fact | 107 |
+| threading a context whose elevation is a run-time fact | 109 |
 
 | how the instrument reached the site | count |
 | :--- | ---: |
-| receiver carried a readable engine type | 197 |
+| receiver carried a readable engine type | 199 |
 | receiver erased, placed by the object NAME | 52 |
 | receiver erased, placed by an `object: string` PARAMETER | 5 |
 | receiver erased, placed by an `UNTYPED_RECEIVERS` row | 5 |
 
 | object name spelled inline | 116 |
-| object name spelled through a `const` | 66 |
+| object name spelled through a `const` | 68 |
 | object name is an `object: string` parameter | 7 |
 | object name is some other run-time expression | 70 |
 
@@ -307,13 +307,13 @@ holds still. They are required to be HERE and to say WHEN they were true;
 their values are not compared. The reasoning, and the measurement behind it,
 are in `scripts/check-tenant-audit-census.mjs`.
 
-Measured on 2026-11-03 at `1c86b45c5`.
+Measured on 2026-11-05 at `e40e04854`.
 
 | corpus scale (not enforced) | count |
 | :--- | ---: |
-| tracked non-test sources scanned | 856 |
-| engine-shaped types recognised | 103 |
-| declared objects in the registry | 126 |
-| same-named calls subtracted as non-engine | 226 |
+| tracked non-test sources scanned | 871 |
+| engine-shaped types recognised | 105 |
+| declared objects in the registry | 127 |
+| same-named calls subtracted as non-engine | 229 |
 
 {/* END GENERATED: tenant-audit-census */}
diff --git a/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md b/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md
index 83252cab991..d05ca1acbbb 100644
--- a/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md
+++ b/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md
@@ -33,19 +33,19 @@ silent, and `node scripts/tenant-audit-census.mjs --write` is the resolution.
 
 | Measure | Value |
 |---|---:|
-| Write call sites | 259 |
-| Object name statically decidable | 182 |
+| Write call sites | 261 |
+| Object name statically decidable | 184 |
 | Object name chosen at run time | 77 |
-| Against a tenancy-enabled object | 176 |
+| Against a tenancy-enabled object | 178 |
 | Against an object declaring tenancy off | 6 |
-| Threading a tenant context | 196 |
+| Threading a tenant context | 198 |
 | Provably carrying none | 8 |
 | …and decidably tenancy-enabled | 2 |
 | Options argument unreadable | 55 |
 | …and decidably tenancy-enabled | 29 |
 | Threading a decidably elevated context | 144 |
 | Threading a decidably non-elevated context | 0 |
-| Threading a context of undecidable elevation | 107 |
+| Threading a context of undecidable elevation | 109 |
 
 ## Subtractions the census could NOT defend — enforced
 
@@ -90,14 +90,14 @@ holds still. They are required to be HERE and to say WHEN they were true;
 their values are not compared. The reasoning, and the measurement behind it,
 are in `scripts/check-tenant-audit-census.mjs`.
 
-Measured on 2026-11-03 at `1c86b45c5`.
+Measured on 2026-11-05 at `e40e04854`.
 
 | corpus scale (not enforced) | count |
 | :--- | ---: |
-| tracked non-test sources scanned | 856 |
-| engine-shaped types recognised | 103 |
-| declared objects in the registry | 126 |
-| same-named calls subtracted as non-engine | 226 |
+| tracked non-test sources scanned | 871 |
+| engine-shaped types recognised | 105 |
+| declared objects in the registry | 127 |
+| same-named calls subtracted as non-engine | 229 |
 
 ## Every site
 
@@ -248,7 +248,7 @@ Measured on 2026-11-03 at `1c86b45c5`.
 | `packages/services/service-messaging/src/sql-outbox.ts` | `update` | `this.objectName` | undecidable | options unreadable | 1 |
 | `packages/services/service-queue/src/db-queue-adapter.ts` | `delete` | `sys_job_queue` | enabled | context, elevation undecidable | 2 |
 | `packages/services/service-queue/src/db-queue-adapter.ts` | `insert` | `sys_job_queue` | enabled | context, elevation undecidable | 1 |
-| `packages/services/service-queue/src/db-queue-adapter.ts` | `update` | `sys_job_queue` | enabled | context, elevation undecidable | 6 |
+| `packages/services/service-queue/src/db-queue-adapter.ts` | `update` | `sys_job_queue` | enabled | context, elevation undecidable | 8 |
 | `packages/services/service-settings/src/config-change-audit.ts` | `insert` | `sys_audit_log` | enabled | context, elevation undecidable | 1 |
 | `packages/services/service-settings/src/settings-service-plugin.ts` | `insert` | `objectName` | undecidable | options unreadable | 1 |
 | `packages/services/service-settings/src/settings-service-plugin.ts` | `update` | `objectName` | undecidable | options unreadable | 2 |

From 9a2d7712570aee741fb99c54b35322d6ae9f6d4b Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 5 Nov 2026 04:05:31 +0000
Subject: [PATCH 7/9] chore(service-queue): changeset names the due-time leg
 and the moved costs

Claude-Session: https://claude.ai/code/session_01Utp6CaNicTDZwYQgDRwD7a
Co-authored-by: Claude <noreply@anthropic.com>
---
 .changeset/30228-queue-reclaim-expired-lease.md | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/.changeset/30228-queue-reclaim-expired-lease.md b/.changeset/30228-queue-reclaim-expired-lease.md
index 2361c708226..47ee267fa56 100644
--- a/.changeset/30228-queue-reclaim-expired-lease.md
+++ b/.changeset/30228-queue-reclaim-expired-lease.md
@@ -11,4 +11,5 @@ Metadata-protocol: none
 
 - **Reclaim.** Each poll tick reads the `running` rows of its subscribed queues whose `locked_until` has passed (one read for all queues, at most once per lease window) and returns them to the retry path. The lost run counts as an attempt: the row goes back to `pending` on its backoff, or to `dlq` once `maxAttempts` is spent, with `last_error` saying the lease expired. The write is a compare-and-set on the expired lease, so a holder that renewed in the meantime keeps its run.
 - **Renewal.** While a handler runs, its worker moves its lease a full window ahead every third of `leaseMs`, so a slow but live handler is never reclaimed and run twice. A handler must not block the event loop for most of a lease window.
-- **Cost.** An idle queue pays one more indexed read per lease window (default 30 s): an idle hour at the defaults goes from 124 to 245 reads.
+- **Due time.** `DbQueueAdapter`'s idle due-time answer (`nextDueAt`) also counts the earliest live lease its last reclaim read saw, at no read of its own, so a `soleWriter` loop wakes when a row its own previous process left `running` runs out of lease instead of at the idle ceiling.
+- **Cost.** An idle queue pays one more indexed read per lease window (default 30 s): an idle hour at the defaults goes from 125 to 246 reads; a fresh sole-writer kernel's first minute from 2 to 3.

From 87fd3a2efac706e180129c18b2edce98cc939f52 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 5 Nov 2026 04:44:01 +0000
Subject: [PATCH 8/9] fix(service-queue): a sole writer reads leases once per
 boot, and the reclaim read uses no $in

Claude-Session: https://claude.ai/code/session_01Utp6CaNicTDZwYQgDRwD7a
Co-authored-by: Claude <noreply@anthropic.com>
---
 .../service-queue/src/db-queue-adapter.ts     | 89 +++++++++++++++----
 .../src/db-queue-idle-backoff.test.ts         | 61 ++++++++++++-
 2 files changed, 133 insertions(+), 17 deletions(-)

diff --git a/packages/services/service-queue/src/db-queue-adapter.ts b/packages/services/service-queue/src/db-queue-adapter.ts
index f785346de5b..006919eb9b3 100644
--- a/packages/services/service-queue/src/db-queue-adapter.ts
+++ b/packages/services/service-queue/src/db-queue-adapter.ts
@@ -239,8 +239,8 @@ export class DbQueueAdapter implements IQueueService {
   private readonly handlers = new Map<string, RegisteredHandler[]>();
   private loop?: DispatchLoop;
   /**
-   * [#30228] Adapter-clock instant (ms) before which no `running` row of a
-   * subscribed queue can have an expired lease, so {@link reclaimExpired}
+   * [#30228] Adapter-clock instant (ms) before which no `running` row can
+   * have an expired lease, so {@link reclaimExpired}
    * need not read. `0` = read on the next tick. See that method for why the
    * bound is sound.
    */
@@ -251,6 +251,16 @@ export class DbQueueAdapter implements IQueueService {
    * {@link nextDueAt} reports it, so an idle loop wakes when it runs out.
    */
   private earliestLiveLeaseMs: number | null = null;
+  /**
+   * [#30228] {@link DbQueueAdapterOptions.soleWriter} only: the lease
+   * (adapter-clock ms) of each row this process claimed and has not yet
+   * settled — completed, retried, dead-lettered or released. A sole writer's
+   * only other source of `running` rows is its previous process, which the
+   * first read after `start` covers; so instead of re-reading every lease
+   * window, it reads again only when one of these runs out — a settling write
+   * that failed and left its row `running`.
+   */
+  private readonly unsettledLeases = new Map<string, number>();
 
   constructor(args: {
     engine: JobEngine;
@@ -557,7 +567,7 @@ export class DbQueueAdapter implements IQueueService {
     const queues = [...this.handlers.keys()];
     if (queues.length === 0) return 0;
 
-    let processed = await this.reclaimExpired(queues);
+    let processed = await this.reclaimExpired();
     for (const queue of queues) {
       const claimed = await this.claimBatch(queue, this.opts.batchSize);
       for (const row of claimed) {
@@ -582,8 +592,10 @@ export class DbQueueAdapter implements IQueueService {
    * {@link reclaimExpired} may take it back. The earliest live lease that
    * method's last read saw is folded in at no read of its own — that read ran
    * on the empty tick this answer follows, or the bound it set has not passed
-   * yet. So a sole-writer loop, which otherwise sleeps to the ceiling, wakes
-   * for a row its own previous process left `running` across a restart.
+   * yet — and so are a sole writer's own unsettled claims. So a sole-writer
+   * loop, which otherwise sleeps to the ceiling, wakes for a row its own
+   * previous process left `running` across a restart, or one a failed
+   * settling write of this process left behind.
    */
   private async nextDueAt(): Promise<number | null> {
     const queues = [...this.handlers.keys()];
@@ -604,10 +616,14 @@ export class DbQueueAdapter implements IQueueService {
     // Instants, not strings — the dialects materialize `scheduled_for`
     // differently (ISO text on SQLite/turso, a `Date` on Postgres/MySQL).
     const scheduledMs = cell ? new Date(cell).getTime() : NaN;
-    const leaseMs = this.earliestLiveLeaseMs ?? NaN;
-    const dueMs = Number.isFinite(scheduledMs) && Number.isFinite(leaseMs)
-      ? Math.min(scheduledMs, leaseMs)
-      : Number.isFinite(scheduledMs) ? scheduledMs : leaseMs;
+    const leaseMs = Math.min(
+      this.earliestLiveLeaseMs ?? Infinity,
+      this.earliestUnsettledLeaseMs() ?? Infinity,
+    );
+    const dueMs = Math.min(
+      Number.isFinite(scheduledMs) ? scheduledMs : Infinity,
+      leaseMs,
+    );
     if (!Number.isFinite(dueMs)) return null;
     // Re-based onto the loop's clock, so an injected `clock` keeps its meaning.
     return Date.now() + (dueMs - now.getTime());
@@ -616,8 +632,14 @@ export class DbQueueAdapter implements IQueueService {
   // ── Internals ────────────────────────────────────────────────────
 
   /**
-   * [#30228] Return every `running` row of a subscribed queue whose lease ran
-   * out to the retry path, and answer how many were reclaimed.
+   * [#30228] Return every `running` row whose lease ran out to the retry path,
+   * and answer how many were reclaimed.
+   *
+   * Every queue's, not only the subscribed ones': reclaiming runs nothing (the
+   * row goes back to `pending` for whichever worker subscribes to it), and the
+   * predicate is then the claim read's own plain equality on `status`, served
+   * by the declared `status` index — no operator a stand-in engine might not
+   * answer.
    *
    * ## Why a lease that ran out means a dead worker, not a slow one
    *
@@ -648,7 +670,7 @@ export class DbQueueAdapter implements IQueueService {
    *
    * ## Why it does not read on every tick
    *
-   * One read answers every subscribed queue, and it also returns the
+   * One read answers every queue, and it also returns the
    * earliest lease still running. Until the earlier of that lease and
    * `now + leaseMs` no lease can run out: every row already `running` was
    * in the read, and a row claimed after it carries a lease of at least
@@ -657,14 +679,20 @@ export class DbQueueAdapter implements IQueueService {
    * a busy 1 s poll pays one read per lease window, an idle one at most one
    * per tick, and an expired lease is still reclaimed on the first tick after
    * it runs out.
+   *
+   * A {@link DbQueueAdapterOptions.soleWriter} drops the `now + leaseMs`
+   * term: no other process claims rows, so after the first read (which finds
+   * what a crashed previous process left behind) the only leases that can run
+   * out are the ones that read saw and this process's own unsettled claims
+   * ({@link unsettledLeases}). An idle sole writer then reads once per boot.
    */
-  private async reclaimExpired(queues: string[]): Promise<number> {
+  private async reclaimExpired(): Promise<number> {
     const now = this.now();
     const nowMs = now.getTime();
-    if (nowMs < this.reclaimNotBeforeMs) return 0;
+    if (nowMs < this.reclaimNotBeforeMs && !this.anyUnsettledLeaseExpired(nowMs)) return 0;
     const max = this.opts.batchSize;
     const rows = await this.engine.find(QUEUE_TABLE, {
-      where: { queue: { $in: queues }, status: 'running' },
+      where: { status: 'running' },
       fields: [
         'id', 'locked_by', 'locked_until', 'attempts', 'max_attempts',
         'backoff_type', 'backoff_delay_ms', 'backoff_max_delay_ms',
@@ -689,13 +717,35 @@ export class DbQueueAdapter implements IQueueService {
       if (await this.reclaimRow(row, now)) reclaimed++;
     }
     this.earliestLiveLeaseMs = earliestLive;
+    // The read is the truth about every `running` row, this process's own
+    // included: no dispatch is in flight while a tick reclaims.
+    this.unsettledLeases.clear();
+    const windowMs = this.opts.soleWriter ? Infinity : nowMs + this.opts.leaseMs;
     // A full page may hide more expired rows behind it: read again next tick.
     this.reclaimNotBeforeMs = (rows?.length ?? 0) >= max
       ? 0
-      : Math.min(nowMs + this.opts.leaseMs, earliestLive ?? Infinity);
+      : Math.min(windowMs, earliestLive ?? Infinity);
     return reclaimed;
   }
 
+  /** [#30228] Whether one of this sole writer's unsettled claims ran out of lease. */
+  private anyUnsettledLeaseExpired(nowMs: number): boolean {
+    for (const until of this.unsettledLeases.values()) if (until <= nowMs) return true;
+    return false;
+  }
+
+  /** [#30228] Earliest lease among this sole writer's unsettled claims, or null. */
+  private earliestUnsettledLeaseMs(): number | null {
+    let min: number | null = null;
+    for (const until of this.unsettledLeases.values()) min = min === null ? until : Math.min(min, until);
+    return min;
+  }
+
+  /** [#30228] A claimed row reached a settling write: its lease is no longer this process's concern. */
+  private settled(id: unknown): void {
+    this.unsettledLeases.delete(String(id));
+  }
+
   /** [#30228] The compare-and-set write of {@link reclaimExpired}, one row. */
   private async reclaimRow(row: any, now: Date): Promise<boolean> {
     const nowIso = now.toISOString();
@@ -763,6 +813,8 @@ export class DbQueueAdapter implements IQueueService {
         .then((res) => {
           if (typeof res === 'number' && res === 0) {
             this.logger?.warn?.('DbQueueAdapter: lease lost while the handler was still running', { id } as any);
+          } else if (this.unsettledLeases.has(id)) {
+            this.unsettledLeases.set(id, now.getTime() + this.opts.leaseMs);
           }
         })
         .catch((err) => { this.logger?.warn?.('DbQueueAdapter: lease renewal failed', err as any); })
@@ -832,6 +884,7 @@ export class DbQueueAdapter implements IQueueService {
           updated_at: now.toISOString(),
         }, { context: SYSTEM_CTX });
         out.push({ ...row, status: 'running' });
+        if (this.opts.soleWriter) this.unsettledLeases.set(String(row.id), now.getTime() + this.opts.leaseMs);
       } catch (err) {
         this.logger?.warn?.('DbQueueAdapter: claim CAS failed', err as any);
       }
@@ -882,6 +935,7 @@ export class DbQueueAdapter implements IQueueService {
         locked_until: null,
         updated_at: now.toISOString(),
       }, { context: SYSTEM_CTX });
+      this.settled(row.id);
       return;
     }
 
@@ -898,6 +952,7 @@ export class DbQueueAdapter implements IQueueService {
         locked_until: null,
         updated_at: now.toISOString(),
       }, { context: SYSTEM_CTX });
+      this.settled(row.id);
       return;
     }
 
@@ -912,6 +967,7 @@ export class DbQueueAdapter implements IQueueService {
       locked_until: null,
       updated_at: now.toISOString(),
     }, { context: SYSTEM_CTX });
+    this.settled(row.id);
   }
 
   private computeBackoff(row: any, attempt: number): number {
@@ -933,6 +989,7 @@ export class DbQueueAdapter implements IQueueService {
         scheduled_for: new Date(now.getTime() + this.opts.pollIntervalMs * 5).toISOString(),
         updated_at: now.toISOString(),
       }, { context: SYSTEM_CTX });
+      this.settled(id);
     } catch (err) {
       this.logger?.warn?.('DbQueueAdapter: release failed', err as any);
     }
diff --git a/packages/services/service-queue/src/db-queue-idle-backoff.test.ts b/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
index 436c799101d..7a362a3c0f5 100644
--- a/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
+++ b/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
@@ -48,7 +48,8 @@ type Row = Record<string, any>;
  */
 function makeCountingEngine() {
     const tables = new Map<string, Row[]>();
-    const calls = { find: 0, insert: 0, update: 0, delete: 0 };
+    // [#30228] `reclaim` counts the lease-reclaim read alone (a `find` too).
+    const calls = { find: 0, insert: 0, update: 0, delete: 0, reclaim: 0 };
     function compare(cell: any, v: any): boolean {
         if (v !== null && typeof v === 'object' && !Array.isArray(v)) {
             for (const [op, target] of Object.entries(v)) {
@@ -82,6 +83,7 @@ function makeCountingEngine() {
         calls,
         async find(table: string, opts: any = {}) {
             if (table === QUEUE_TABLE) calls.find++;
+            if (table === QUEUE_TABLE && opts?.where?.status === 'running') calls.reclaim++;
             const t = tables.get(table) ?? [];
             let out = opts.where ? t.filter((r) => matches(r, opts.where)) : [...t];
             if (opts.orderBy) {
@@ -587,6 +589,63 @@ describe('#29919 DbQueueAdapter — wake at the next due time, no ramp for a sol
         expect(at[1]! - at[0]!).toBe(15_000);
     });
 
+    it('[#30228] an idle sole writer reads for expired leases ONCE an hour, at boot — even after work; the control reads 120', async () => {
+        const hourOf = async (soleWriter: boolean) => {
+            vi.useFakeTimers();
+            const engine = makeCountingEngine();
+            const adapter = adapterOn(engine, { soleWriter });
+            let ran = 0;
+            await adapter.subscribe('q1', async () => { ran++; });
+            adapter.start();
+            await vi.advanceTimersByTimeAsync(10 * 60_000);
+            // Work in between: its claim is settled, so it costs no later read.
+            await adapter.publish('q1', { n: 1 });
+            await vi.advanceTimersByTimeAsync(HOUR - 10 * 60_000);
+            await adapter.stop();
+            vi.useRealTimers();
+            expect(ran).toBe(1);
+            return engine.calls.reclaim;
+        };
+        // No other process claims rows: the boot read covers what a crashed
+        // previous process left, and nothing else can run out of lease.
+        expect(await hourOf(true)).toBe(1);
+        // CONTROL: other nodes may claim at any time, so one read per 30 s
+        // lease window that a tick lands in — at boot, then on the ceiling
+        // ticks (the publish at 10 min restarts the ramp, and its 2/6/14 s
+        // ticks fall inside one window, so one boundary fewer than idle: 120).
+        expect(await hourOf(false)).toBe(120);
+    });
+
+    it('[#30228] a sole writer\'s own claim whose settling write failed is reclaimed when its lease runs out', async () => {
+        vi.useFakeTimers();
+        const engine = makeCountingEngine();
+        // The `completed` write fails once (a dropped connection, say): the
+        // handler ran, but the row is still `running` under this worker.
+        const update = engine.update.bind(engine);
+        let failOnce = true;
+        engine.update = async (table: string, patch: Row, options?: any) => {
+            if (failOnce && patch.status === 'completed') { failOnce = false; throw new Error('connection reset'); }
+            return update(table, patch, options);
+        };
+        const adapter = adapterOn(engine, { soleWriter: true, leaseMs: 3_000 });
+        const at: number[] = [];
+        await adapter.subscribe('q1', async () => { at.push(Date.now()); });
+        adapter.start();
+        await vi.advanceTimersByTimeAsync(10);
+        const t0 = Date.now();
+        await adapter.publish('q1', { n: 1 });
+        await vi.advanceTimersByTimeAsync(60_000);
+        await adapter.stop();
+
+        // Run, settling write lost, lease out at +3 s, reclaimed (the lost
+        // settle counts as an attempt: delivery is at-least-once), run again.
+        expect(at).toHaveLength(2);
+        expect(at[1]! - t0).toBeLessThanOrEqual(3_000 + DEFAULT_MAX_IDLE_INTERVAL_MS + 2 * BASE);
+        expect((engine.tables.get(QUEUE_TABLE) ?? [])[0]).toMatchObject({ status: 'completed', attempts: 2 });
+        // The boot read, and the one its expired lease asked for.
+        expect(engine.calls.reclaim).toBe(2);
+    });
+
     it('[#30228] a row its own previous process left running is reclaimed when the lease runs out, not at the ceiling', async () => {
         vi.useFakeTimers();
         const engine = makeCountingEngine();

From 8d430e0cdaf1b7a8032c1b54e7497d601e645262 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 5 Nov 2026 04:48:55 +0000
Subject: [PATCH 9/9] chore(service-queue): changeset states both modes'
 reclaim cost

Claude-Session: https://claude.ai/code/session_01Utp6CaNicTDZwYQgDRwD7a
Co-authored-by: Claude <noreply@anthropic.com>
---
 .changeset/30228-queue-reclaim-expired-lease.md | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/.changeset/30228-queue-reclaim-expired-lease.md b/.changeset/30228-queue-reclaim-expired-lease.md
index 47ee267fa56..e9de72d922a 100644
--- a/.changeset/30228-queue-reclaim-expired-lease.md
+++ b/.changeset/30228-queue-reclaim-expired-lease.md
@@ -9,7 +9,7 @@ Metadata-protocol: none
 
 `DbQueueAdapterOptions.leaseMs` documented a lease "before another worker may reclaim", but the claim read only `pending` rows, so a worker that crashed, was OOM-killed or was redeployed mid-handler left its `sys_job_queue` row `running` for good: never retried, never dead-lettered.
 
-- **Reclaim.** Each poll tick reads the `running` rows of its subscribed queues whose `locked_until` has passed (one read for all queues, at most once per lease window) and returns them to the retry path. The lost run counts as an attempt: the row goes back to `pending` on its backoff, or to `dlq` once `maxAttempts` is spent, with `last_error` saying the lease expired. The write is a compare-and-set on the expired lease, so a holder that renewed in the meantime keeps its run.
+- **Reclaim.** A poll tick reads the table's `running` rows (one read for every queue, at most once per lease window) and returns each whose `locked_until` has passed to the retry path. The lost run counts as an attempt: the row goes back to `pending` on its backoff, or to `dlq` once `maxAttempts` is spent, with `last_error` saying the lease expired. The write is a compare-and-set on the expired lease, so a holder that renewed in the meantime keeps its run.
 - **Renewal.** While a handler runs, its worker moves its lease a full window ahead every third of `leaseMs`, so a slow but live handler is never reclaimed and run twice. A handler must not block the event loop for most of a lease window.
-- **Due time.** `DbQueueAdapter`'s idle due-time answer (`nextDueAt`) also counts the earliest live lease its last reclaim read saw, at no read of its own, so a `soleWriter` loop wakes when a row its own previous process left `running` runs out of lease instead of at the idle ceiling.
-- **Cost.** An idle queue pays one more indexed read per lease window (default 30 s): an idle hour at the defaults goes from 125 to 246 reads; a fresh sole-writer kernel's first minute from 2 to 3.
+- **Sole writer.** With `soleWriter` no other process claims rows, so the reclaim reads once per boot (finding what a crashed previous process left behind) and again only when one of this process's own claims was left unsettled by a failed write and its lease runs out. `nextDueAt` folds in those leases and the earliest live lease the last read saw, at no read of its own, so the loop wakes for them instead of at the idle ceiling.
+- **Cost.** Without `soleWriter`, an idle queue pays one more indexed read per lease window (default 30 s): an idle hour at the defaults goes from 125 to 246 reads. With `soleWriter`, one more read per boot: a fresh kernel's first minute goes from 2 to 3 reads, and an idle hour carries no other.