perf(queue): build one recovery index per listing instead of a json_each scan per row (#2588)

A queue listing asks findQueueRecovery once per row, and each ask ran json_each over every recovery-tagged row. withRecoveryIndex(db, fn) now opens a scope for one synchronous listing; inside it, the first lookup builds an index with the same json_each, the same recovery tag filter and order and text values only, and later lookups read it, while a compact listing that never asks builds nothing. Outside the scope every lookup is the original per-row query, and nested calls share the scope. QueueRepository.list wraps its existing body. Limits: the index assumes recovery data doesn't change during the listing; nothing a listing calls changes recovery membership, state or its ordering time, although the activity reader it calls can write other events, and a callee made async would simply fall back to the per-row query. Equivalence was traced in source and checked against an offline copy of a real queue; it wasn't measured on a live daemon, and malformed Unicode tag values weren't compared.

packages/daemon/src/domain/queue-recovery.ts

@@ -26,13 +26,54 @@ export function recoveryId(db: Database.Database, qitemId: string, overdueDeadli
  * planner read that small index instead of scanning and parsing every row. */
 const RECOVERY_TAG_ROWS = `(tags LIKE '%"recovery-for:%' OR tags LIKE '%\\%')`;
 const RECOVERY_ORDER = "ORDER BY CASE WHEN state IN ('pending','in-progress','blocked') THEN 0 ELSE 1 END, ts_updated DESC, qitem_id DESC LIMIT 1";
+/** A listing's recovery index, built by SQLite's own json_each (the same values, and the same truthy excludeTag
+ * test, as the per-row query): for each string tag value, the rows that carry it in RECOVERY_ORDER, and each row's
+ * string values. Built lazily, on the first lookup inside `withRecoveryIndex`, so a listing that never asks (a
+ * compact one) pays nothing; scoped to that synchronous span, so no write can land between the read and its use.
+ * Outside it every lookup is the original query. */
+type RecoveryRow = { qitem_id: string; state: string; ts_updated: string; values: Set<string> };
+type RecoveryScope = { db: Database.Database; byTag: Map<string, RecoveryRow[]> | null };
+let scope: RecoveryScope | null = null;
+/** Statement text of the index read; exported for tests that count how often a listing builds it. */
+export const RECOVERY_INDEX_SQL = `SELECT q.qitem_id, q.state, q.ts_updated, j.value AS value
+    FROM queue_items q, json_each(q.tags) j
+    WHERE ${RECOVERY_TAG_ROWS.replace(/\btags\b/g, "q.tags")} AND json_valid(q.tags) AND typeof(j.value) = 'text'
+    ${RECOVERY_ORDER.replace(/ LIMIT 1$/, "").replace(/\b(state|ts_updated|qitem_id)\b/g, "q.$1")}`;
+function buildRecoveryIndex(db: Database.Database): Map<string, RecoveryRow[]> {
+  const rows = db.prepare(RECOVERY_INDEX_SQL).all() as Array<{ qitem_id: string; state: string; ts_updated: string; value: string }>;
+  const byRow = new Map<string, RecoveryRow>();
+  const byTag = new Map<string, RecoveryRow[]>();
+  for (const r of rows) {
+    let row = byRow.get(r.qitem_id);
+    if (!row) byRow.set(r.qitem_id, row = { qitem_id: r.qitem_id, state: r.state, ts_updated: r.ts_updated, values: new Set() });
+    if (row.values.has(r.value)) continue;
+    row.values.add(r.value);
+    const list = byTag.get(r.value);
+    if (list) list.push(row); else byTag.set(r.value, [row]);
+  }
+  return byTag;
+}
+/** Runs `fn` (synchronous) with one lazily built recovery index for `db`: a listing asks findQueueRecovery once
+ * per row, and each ask used to json_each every recovery-tagged row. Re-entrant: a nested call shares the scope. */
+export function withRecoveryIndex<T>(db: Database.Database, fn: () => T): T {
+  if (scope?.db === db) return fn();
+  const previous = scope;
+  scope = { db, byTag: null };
+  try { return fn(); } finally { scope = previous; }
+}
 export function findQueueRecovery(db: Database.Database, qitemId: string, excludeTag?: string): { qitemId: string; state: string } | null {
   const tag = recoveryTag(qitemId);
-  const row = db.prepare(`SELECT qitem_id, state, ts_updated FROM queue_items
+  let row: { qitem_id: string; state: string; ts_updated: string } | undefined;
+  if (scope?.db === db) {
+    const byTag = scope.byTag ??= buildRecoveryIndex(db);
+    row = (byTag.get(tag) ?? []).find((r) => !excludeTag || !r.values.has(excludeTag));
+  } else {
+    row = db.prepare(`SELECT qitem_id, state, ts_updated FROM queue_items
     WHERE ${RECOVERY_TAG_ROWS} AND json_valid(tags) AND EXISTS (SELECT 1 FROM json_each(tags) WHERE value = ?)
     ${excludeTag ? "AND NOT EXISTS (SELECT 1 FROM json_each(tags) WHERE value = ?)" : ""}
     ${RECOVERY_ORDER}`)
-    .get(...(excludeTag ? [tag, excludeTag] : [tag])) as { qitem_id: string; state: string; ts_updated: string } | undefined;
+      .get(...(excludeTag ? [tag, excludeTag] : [tag])) as { qitem_id: string; state: string; ts_updated: string } | undefined;
+  }
   if (!row) return null;
   if (!["pending", "in-progress", "blocked"].includes(row.state)) {
     const changed = lastMeaningfulTransition(db, qitemId);

packages/daemon/src/domain/queue-repository.ts

@@ -12,6 +12,7 @@ import { QueueTransitionLog, type OwnerNotificationLevel, type QueueTransition,
 import { WAKE_INTENT_PREFIX, type OutboxHandler, type OutboxEntry } from "./outbox-handler.js";
 import type { QueueWakeFacts } from "./draft-aware-delivery.js";
 import { derivePickup, type PickupReceipt } from "./queue-pickup.js";
+import { withRecoveryIndex } from "./queue-recovery.js";
 import { lastMeaningfulTransition, readWaitingView, type WaitingView, type WaitingActivityReader } from "./queue-waiting.js";
 import { isOpenRigSender, OPENRIG_SENDER, wrapPaneEnvelope } from "../lib/pane-envelope.js";
 import { isBoundTerminalSeat } from "./terminal-seat.js";
@@ -4204,6 +4205,10 @@ export class QueueRepository {
   list(opts?: QueueListOptions & { compact?: false }): QueueItem[];
   list(opts?: QueueListOptions): Array<QueueItem | CompactQueueItem>;
   list(opts?: QueueListOptions): Array<QueueItem | CompactQueueItem> {
+    // one recovery index for the whole listing (built on first use) instead of one json_each scan per row
+    return withRecoveryIndex(this.db, () => this.listRows(opts));
+  }
+  private listRows(opts?: QueueListOptions): Array<QueueItem | CompactQueueItem> {
     const limit = opts?.limit ?? 100;
     const conditions: string[] = [];
     const params: unknown[] = [];

packages/daemon/test/queue-recovery-listing-index.test.ts

@@ -0,0 +1,127 @@
+// A queue listing builds ONE recovery index (withRecoveryIndex) instead of one json_each scan of every
+// recovery-tagged row per listed row, and findQueueRecovery still chooses exactly the row the query chooses.
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import Database from "better-sqlite3";
+import { migrate } from "../src/db/migrate.js";
+import { ALL_MIGRATIONS } from "../src/db/all-migrations.js";
+import { EventBus } from "../src/domain/event-bus.js";
+import { QueueRepository } from "../src/domain/queue-repository.js";
+import { RECOVERY_INDEX_SQL, findQueueRecovery, recoveryTag, withRecoveryIndex } from "../src/domain/queue-recovery.js";
+
+describe("withRecoveryIndex: one recovery index per listing", () => {
+  let db: Database.Database, repo: QueueRepository;
+  const at = "2026-10-09T10:00:00.000Z";
+  function row(id: string, state = "done", tags: string | null = "[]", updated = at, destination = "worker@r") {
+    db.prepare(`INSERT INTO queue_items(qitem_id,ts_created,ts_updated,source_session,destination_session,state,tags,body)
+      VALUES (?, ?, ?, 'sender@r', ?, ?, ?, 'body')`).run(id, at, updated, destination, state, tags);
+  }
+  const indexBuilds = (run: () => unknown): number => {
+    const prepare = db.prepare.bind(db); let n = 0;
+    const spy = vi.spyOn(db, "prepare").mockImplementation((sql: string) => { if (sql === RECOVERY_INDEX_SQL) n++; return prepare(sql); });
+    try { run(); } finally { spy.mockRestore(); }
+    return n;
+  };
+  const recoveryLookups = (run: () => unknown): number => {
+    const prepare = db.prepare.bind(db); let n = 0;
+    const spy = vi.spyOn(db, "prepare").mockImplementation((sql: string) => {
+      // the per-row recovery lookup (findQueueRecovery's query), not the cheap per-qitem tag checks
+      if (sql.includes("json_each(tags) WHERE value = ?") && sql.includes("qitem_id DESC LIMIT 1")) n++;
+      return prepare(sql);
+    });
+    try { run(); } finally { spy.mockRestore(); }
+    return n;
+  };
+
+  beforeEach(() => {
+    db = new Database(":memory:"); migrate(db, ALL_MIGRATIONS);
+    repo = new QueueRepository(db, new EventBus(db), { validateRig: () => true });
+  });
+  afterEach(() => { vi.restoreAllMocks(); db.close(); });
+
+  it("chooses the same recovery row as the query, for every ordering and excludeTag case", () => {
+    row("qitem-a", "pending"); row("qitem-b", "pending"); row("qitem-c", "pending"); row("qitem-d", "pending");
+    // a: an active recovery beats a newer closed one; b: the newest of two closed; c: excludeTag skips one;
+    // d: escaped JSON text (the LIKE '%\%' arm) and an object-shaped tags value; e: no recovery at all
+    row("rec-a1", "done", JSON.stringify([recoveryTag("qitem-a")]), "2026-10-09T12:00:00.000Z");
+    row("rec-a2", "in-progress", JSON.stringify([recoveryTag("qitem-a")]), "2026-10-09T11:00:00.000Z");
+    row("rec-b1", "done", JSON.stringify([recoveryTag("qitem-b")]), "2026-10-09T11:00:00.000Z");
+    row("rec-b2", "done", JSON.stringify([recoveryTag("qitem-b")]), "2026-10-09T13:00:00.000Z");
+    row("rec-c1", "blocked", JSON.stringify([recoveryTag("qitem-c"), "prompt-alert"]), "2026-10-09T14:00:00.000Z");
+    row("rec-c2", "blocked", JSON.stringify([recoveryTag("qitem-c")]), "2026-10-09T12:00:00.000Z");
+    row("rec-d1", "pending", `["recovery-for:qitem-\\u0064"]`, "2026-10-09T12:00:00.000Z");
+    row("rec-d2", "pending", JSON.stringify({ k: recoveryTag("qitem-d") }), "2026-10-09T13:00:00.000Z");
+    row("rec-x", "pending", "not json", "2026-10-09T15:00:00.000Z");
+    const ids = ["qitem-a", "qitem-b", "qitem-c", "qitem-d", "qitem-e"];
+    const asks: Array<string | undefined> = [undefined, "prompt-alert", "slack-undeliverable"];
+    const before = ids.flatMap((id) => asks.map((ex) => findQueueRecovery(db, id, ex)));
+    const inside = withRecoveryIndex(db, () => ids.flatMap((id) => asks.map((ex) => findQueueRecovery(db, id, ex))));
+    expect(inside).toEqual(before);
+    expect(before.map((r) => r?.qitemId ?? null).filter((_, i) => i % 3 === 0)).toEqual(["rec-a2", "rec-b2", "rec-c1", "rec-d2", null]);
+    expect(before[7]?.qitemId).toBe("rec-c2"); // qitem-c with excludeTag prompt-alert
+  });
+
+  it("matches json_each on a duplicated object key and keeps the query's truthy excludeTag ('')", () => {
+    row("qitem-k", "pending"); row("qitem-z", "pending");
+    // json_each sees both values of a duplicated key (JSON.parse would keep only the last one)
+    row("rec-k", "pending", `{"k":"${recoveryTag("qitem-k")}","k":"other"}`, "2026-10-09T12:00:00.000Z");
+    row("rec-z", "pending", JSON.stringify([recoveryTag("qitem-z"), ""]), "2026-10-09T12:00:00.000Z");
+    const asks: Array<[string, string | undefined]> = [["qitem-k", undefined], ["qitem-k", "other"], ["qitem-z", ""], ["qitem-z", undefined]];
+    const before = asks.map(([id, ex]) => findQueueRecovery(db, id, ex));
+    expect(before.map((r) => r?.qitemId ?? null)).toEqual(["rec-k", null, "rec-z", "rec-z"]);
+    expect(withRecoveryIndex(db, () => asks.map(([id, ex]) => findQueueRecovery(db, id, ex)))).toEqual(before);
+  });
+
+  it("compares nested arrays and objects as their JSON text, like the query's value = ?", () => {
+    row("qitem-n1", "pending"); row("qitem-n2", "pending");
+    row("rec-n1", "pending", JSON.stringify([recoveryTag("qitem-n1"), ["nested"]]), "2026-10-09T12:00:00.000Z");
+    row("rec-n2", "pending", JSON.stringify([recoveryTag("qitem-n2"), { k: "v" }]), "2026-10-09T12:00:00.000Z");
+    const asks: Array<[string, string | undefined]> = [["qitem-n1", '["nested"]'], ["qitem-n2", '{"k":"v"}'], ["qitem-n1", undefined]];
+    const before = asks.map(([id, ex]) => findQueueRecovery(db, id, ex));
+    expect(before.map((r) => r?.qitemId ?? null)).toEqual([null, null, "rec-n1"]);
+    expect(withRecoveryIndex(db, () => asks.map(([id, ex]) => findQueueRecovery(db, id, ex)))).toEqual(before);
+  });
+
+  it("builds the index lazily: a compact listing never builds it, a full one once, nested calls share it", () => {
+    for (let i = 0; i < 5; i++) {
+      row(`qitem-${i}`, "pending", "[]", at, "orq@r");
+      row(`rec-${i}`, "done", JSON.stringify([recoveryTag(`qitem-${i}`)]));
+    }
+    expect(indexBuilds(() => repo.list({ destinationSession: "orq@r", state: ["pending"], limit: 1000, compact: true }))).toBe(0);
+    expect(indexBuilds(() => repo.list({ destinationSession: "orq@r", state: ["pending"], limit: 1000 }))).toBe(1);
+    expect(indexBuilds(() => withRecoveryIndex(db, () => {
+      findQueueRecovery(db, "qitem-0");
+      withRecoveryIndex(db, () => findQueueRecovery(db, "qitem-1"));
+    }))).toBe(1);
+    expect(indexBuilds(() => withRecoveryIndex(db, () => null))).toBe(0);
+  });
+
+  it("a listing reads the recovery rows once, not once per listed row", () => {
+    for (let i = 0; i < 30; i++) {
+      row(`qitem-${i}`, "pending", "[]", at, "orq@r");
+      row(`rec-${i}`, "done", JSON.stringify([recoveryTag(`qitem-${i}`)]));
+    }
+    const outside = recoveryLookups(() => { for (let i = 0; i < 30; i++) findQueueRecovery(db, `qitem-${i}`); });
+    expect(outside).toBe(30);
+    const listing = recoveryLookups(() => repo.list({ destinationSession: "orq@r", state: ["pending"], limit: 1000 }));
+    expect(listing).toBe(0);
+  });
+
+  it("is scoped to the call: after it, a new recovery row is seen at once; nested calls reuse the index", () => {
+    row("qitem-n", "pending");
+    withRecoveryIndex(db, () => {
+      expect(findQueueRecovery(db, "qitem-n")).toBeNull();
+      expect(withRecoveryIndex(db, () => findQueueRecovery(db, "qitem-n"))).toBeNull();
+    });
+    row("rec-n", "pending", JSON.stringify([recoveryTag("qitem-n")]));
+    expect(findQueueRecovery(db, "qitem-n")?.qitemId).toBe("rec-n");
+    expect(withRecoveryIndex(db, () => findQueueRecovery(db, "qitem-n"))?.qitemId).toBe("rec-n");
+  });
+
+  it("another database never reads this one's index", () => {
+    const other = new Database(":memory:"); migrate(other, ALL_MIGRATIONS);
+    try {
+      row("qitem-o", "pending"); row("rec-o", "pending", JSON.stringify([recoveryTag("qitem-o")]));
+      withRecoveryIndex(db, () => expect(findQueueRecovery(other, "qitem-o")).toBeNull());
+    } finally { other.close(); }
+  });
+});

packages/daemon/test/queue-recovery-read-work.test.ts

@@ -4,7 +4,7 @@ import { migrate } from "../src/db/migrate.js";
 import { ALL_MIGRATIONS } from "../src/db/all-migrations.js";
 import { EventBus } from "../src/domain/event-bus.js";
 import { QueueRepository } from "../src/domain/queue-repository.js";
-import { findQueueRecovery, recoveryTag } from "../src/domain/queue-recovery.js";
+import { findQueueRecovery, recoveryTag, RECOVERY_INDEX_SQL } from "../src/domain/queue-recovery.js";
 import { queueRecoveryOwnsWake, readWakeLadderBackstop } from "../src/domain/queue-wake-ladder.js";
 
 describe("recovery lookup work and fresh disposition", () => {
@@ -88,7 +88,9 @@ describe("recovery lookup work and fresh disposition", () => {
     expect(findQueueRecovery(db, "source")?.qitemId).toBe("disposition");
     const prepare = db.prepare.bind(db);
     vi.spyOn(db, "prepare").mockImplementation(sql => {
-      if (sql.includes("json_each(tags)") && sql.includes("ts_updated DESC, qitem_id DESC LIMIT 1")) throw new Error("fixture lookup unavailable");
+      // the per-row lookup, or the listing's one-pass recovery index (withRecoveryIndex) that replaces it in a list
+      if ((sql.includes("json_each(tags)") && sql.includes("ts_updated DESC, qitem_id DESC LIMIT 1"))
+        || sql === RECOVERY_INDEX_SQL) throw new Error("fixture lookup unavailable");
       return prepare(sql);
     });
     expect(() => repo.getById("source")).toThrow("fixture lookup unavailable");