perf(daemon): reuse the waiting view's prepared statements per connection

lastMeaningfulTransition, lastMeaningfulTransitionMany and WaitingStatements
prepared their fixed SQL on every call, and these run on every waiting view,
sweep and ladder read. Statements are now prepared once per connection and
reused: a WeakMap keyed by the connection, so they go with it, holding at most
32 texts from this module's fixed set.

A prepared statement still reads the current data each time it runs, and
SQLite re-prepares it itself after a schema change, so no result changes. The
schema checks stay per call, and a WaitingStatements instance still memoizes
its own table checks.

The listing-reads test now counts runs of the transition read rather than
prepares, since the statement is prepared once and reused.

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

@@ -12,15 +12,35 @@ export type WaitingActivityReader = (session: string) => Pick<ArbitratedSeatStat
 /** A blocked row's backstop when no live timer or watchdog is armed and its blocker isn't a row that can finish. */
 export const NO_TIMED_BACKSTOP = "UNVERIFIED: no timed backstop";
 
+/** This module's statements, prepared once per connection and reused. Preparing parses and plans the SQL every time,
+ *  and these few fixed texts run on every waiting view, sweep and ladder read. A prepared statement still reads the
+ *  current data each time it runs, and SQLite re-prepares it itself after a schema change, so reuse changes no result.
+ *  Keyed by the connection, so they go with it, and by SQL text from this module's fixed set, capped as a backstop. */
+const REUSED_STATEMENT_CAP = 32;
+const reusedStatements = new WeakMap<Database.Database, Map<string, Database.Statement>>();
+export function reusedStatement(db: Database.Database, sql: string): Database.Statement {
+  let byText = reusedStatements.get(db);
+  if (!byText) {
+    byText = new Map();
+    reusedStatements.set(db, byText);
+  }
+  let statement = byText.get(sql);
+  if (!statement) {
+    statement = db.prepare(sql);
+    if (byText.size < REUSED_STATEMENT_CAP) byText.set(sql, statement);
+  }
+  return statement;
+}
+
 /** A read of the existing transition log, not another progress receipt. Machine
  * actors and typed wake receipts are bookkeeping, never task progress. Historical
  * untyped author notes remain author testimony; their text is not classified.
  * Delivery bookkeeping uses the reserved daemon actors and must not reset the
  * ladder before it can join its dispatch marker to the delivery receipt. */
 export function lastMeaningfulTransition(db: Database.Database, id: string): { id: number; at: string } | null {
-  if (!db.prepare("SELECT 1 FROM sqlite_master WHERE name = 'queue_transitions'").get()) return null;
-  const hasWakes = db.prepare("SELECT 1 FROM sqlite_master WHERE name = 'queue_transition_wakes'").get();
-  const row = db.prepare(`SELECT t.transition_id, t.ts FROM (
+  if (!reusedStatement(db, "SELECT 1 FROM sqlite_master WHERE name = 'queue_transitions'").get()) return null;
+  const hasWakes = reusedStatement(db, "SELECT 1 FROM sqlite_master WHERE name = 'queue_transition_wakes'").get();
+  const row = reusedStatement(db, `SELECT t.transition_id, t.ts FROM (
       SELECT *, LAG(state) OVER (ORDER BY transition_id) AS previous_state FROM queue_transitions WHERE qitem_id = ?
     ) t WHERE (t.actor_session NOT IN ('watchdog@system', 'wake-ladder@system', 'daemon@kernel', 'daemon@system') OR t.state IS NOT t.previous_state)
     ${hasWakes ? "AND NOT EXISTS (SELECT 1 FROM queue_transition_wakes w WHERE w.transition_id = t.transition_id AND w.phase = 'fired')" : ""}
@@ -34,9 +54,9 @@ export function lastMeaningfulTransitionMany(db: Database.Database, ids: readonl
   const unique = [...new Set(ids)];
   const out = new Map<string, { id: number; at: string } | null>(unique.map((id) => [id, null]));
   if (unique.length === 0) return out;
-  if (!db.prepare("SELECT 1 FROM sqlite_master WHERE name = 'queue_transitions'").get()) return out;
-  const hasWakes = db.prepare("SELECT 1 FROM sqlite_master WHERE name = 'queue_transition_wakes'").get();
-  const rows = db.prepare(`SELECT qitem_id, transition_id, ts FROM (
+  if (!reusedStatement(db, "SELECT 1 FROM sqlite_master WHERE name = 'queue_transitions'").get()) return out;
+  const hasWakes = reusedStatement(db, "SELECT 1 FROM sqlite_master WHERE name = 'queue_transition_wakes'").get();
+  const rows = reusedStatement(db, `SELECT qitem_id, transition_id, ts FROM (
       SELECT t.qitem_id, t.transition_id, t.ts, ROW_NUMBER() OVER (PARTITION BY t.qitem_id ORDER BY t.transition_id DESC) AS newest FROM (
         SELECT *, LAG(state) OVER (PARTITION BY qitem_id ORDER BY transition_id) AS previous_state FROM queue_transitions
          WHERE qitem_id IN (SELECT value FROM json_each(?))
@@ -47,21 +67,15 @@ export function lastMeaningfulTransitionMany(db: Database.Database, ids: readonl
   return out;
 }
 
-/** The waiting view's statements, each prepared once, and the schema checks they depend on, each made once. A listing
- * holds one for all its rows; a single read uses a fresh one. A listing reads synchronously, so neither the statements
- * nor the schema checks can go stale under it. */
+/** The waiting view's statements (reused per connection) and the schema checks they depend on, each made once per
+ * instance. A listing holds one for all its rows; a single read uses a fresh one. A listing reads synchronously, so its
+ * schema checks can't go stale under it, and a fresh instance checks again. */
 export class WaitingStatements {
-  private readonly prepared = new Map<string, Database.Statement>();
   private readonly tables = new Map<string, boolean>();
   constructor(private readonly db: Database.Database) {}
 
   statement(sql: string): Database.Statement {
-    let prepared = this.prepared.get(sql);
-    if (!prepared) {
-      prepared = this.db.prepare(sql);
-      this.prepared.set(sql, prepared);
-    }
-    return prepared;
+    return reusedStatement(this.db, sql);
   }
 
   hasTable(name: string): boolean {

packages/daemon/test/queue-listing-reads.test.ts

@@ -366,14 +366,19 @@ describe("show (getById) is the listing projection of one row", () => {
 
   it("reads the row's last meaningful transition once, not once for the pickup and once for the waiting view", async () => {
     const q = (await repo.create({ sourceSession: "orch@r", destinationSession: "worker@r", body: "work", nudge: false } as never)).qitemId;
-    const prepare = db.prepare.bind(db);
+    // Counts runs of the read, not prepares: the statement is prepared once per connection and reused.
+    type Run = (this: { source: string }, ...args: unknown[]) => unknown;
+    const proto = Object.getPrototypeOf(db.prepare("SELECT 1")) as { get: Run; all: Run };
     let n = 0;
-    const spy = vi.spyOn(db, "prepare").mockImplementation(((sql: string) => {
-      if (sql.includes("actor_session NOT IN ('watchdog@system'")) n++;
-      return prepare(sql);
-    }) as typeof db.prepare);
+    const spies = (["get", "all"] as const).map((method) => {
+      const original = proto[method];
+      return vi.spyOn(proto, method).mockImplementation(function (this: { source: string }, ...args: unknown[]) {
+        if (this.source.includes("actor_session NOT IN ('watchdog@system'")) n++;
+        return original.apply(this, args);
+      });
+    });
     repo.getById(q);
-    spy.mockRestore();
+    for (const spy of spies) spy.mockRestore();
     expect(n).toBe(1);
   });
 });

packages/daemon/test/queue-waiting-statements.test.ts

@@ -0,0 +1,74 @@
+import { afterEach, describe, expect, it } from "vitest";
+import Database from "better-sqlite3";
+import { WaitingStatements, lastMeaningfulTransition, lastMeaningfulTransitionMany, reusedStatement } from "../src/domain/queue-waiting.js";
+
+// The waiting view reuses its statements per connection. These pin what that reuse must not change: every read sees
+// current data, the schema checks run each call, the cached texts are bounded, and statements go with their connection.
+describe("the waiting view's reused statements", () => {
+  const open: Database.Database[] = [];
+  const connection = (wakes = true) => {
+    const db = new Database(":memory:");
+    open.push(db);
+    db.exec(`CREATE TABLE queue_transitions (transition_id INTEGER PRIMARY KEY AUTOINCREMENT, qitem_id TEXT NOT NULL, ts TEXT NOT NULL,
+      state TEXT NOT NULL, actor_session TEXT NOT NULL)`);
+    if (wakes) db.exec("CREATE TABLE queue_transition_wakes (transition_id INTEGER NOT NULL, qitem_id TEXT NOT NULL, phase TEXT NOT NULL)");
+    return db;
+  };
+  const transition = (db: Database.Database, qitemId: string, state: string, actor = "a-one@rig") =>
+    Number(db.prepare("INSERT INTO queue_transitions (qitem_id, ts, state, actor_session) VALUES (?, ?, ?, ?)")
+      .run(qitemId, new Date().toISOString(), state, actor).lastInsertRowid);
+  afterEach(() => {
+    for (const db of open.splice(0)) if (db.open) db.close();
+  });
+
+  it("reads current data on every call through the same statement", () => {
+    const db = connection();
+    const first = transition(db, "q-1", "pending");
+    expect(lastMeaningfulTransition(db, "q-1")?.id).toBe(first);
+    const second = transition(db, "q-1", "in-progress");
+    expect(lastMeaningfulTransition(db, "q-1")?.id).toBe(second);
+    transition(db, "q-1", "in-progress", "watchdog@system"); // bookkeeping, not progress
+    expect(lastMeaningfulTransition(db, "q-1")?.id).toBe(second);
+    expect(lastMeaningfulTransitionMany(db, ["q-1", "q-2"])).toEqual(new Map([["q-1", { id: second, at: expect.any(String) }], ["q-2", null]]));
+    const third = transition(db, "q-2", "pending");
+    expect(lastMeaningfulTransitionMany(db, ["q-1", "q-2"]).get("q-2")?.id).toBe(third);
+  });
+
+  it("checks the schema each call, and keeps working across a schema change", () => {
+    const db = connection(false);
+    const id = transition(db, "q-1", "pending");
+    expect(lastMeaningfulTransition(db, "q-1")?.id).toBe(id);
+    // The wakes table arrives later: the next call sees it and leaves out a fired wake's transition.
+    db.exec("CREATE TABLE queue_transition_wakes (transition_id INTEGER NOT NULL, qitem_id TEXT NOT NULL, phase TEXT NOT NULL)");
+    const woken = transition(db, "q-1", "blocked", "a-one@rig");
+    db.prepare("INSERT INTO queue_transition_wakes (transition_id, qitem_id, phase) VALUES (?, 'q-1', 'fired')").run(woken);
+    expect(lastMeaningfulTransition(db, "q-1")?.id).toBe(id);
+    // A column added under a reused statement: SQLite re-prepares it, and the result is unchanged.
+    db.exec("ALTER TABLE queue_transitions ADD COLUMN note TEXT");
+    expect(lastMeaningfulTransition(db, "q-1")?.id).toBe(id);
+    // A WaitingStatements instance memoizes its table checks; a fresh one checks again.
+    const before = new WaitingStatements(db);
+    expect(before.hasTable("late_table")).toBe(false);
+    db.exec("CREATE TABLE late_table (x INTEGER)");
+    expect(before.hasTable("late_table")).toBe(false);
+    expect(new WaitingStatements(db).hasTable("late_table")).toBe(true);
+  });
+
+  it("reuses one statement per text and connection, never across connections, and caps the cached texts", () => {
+    const one = connection();
+    const two = connection();
+    const sql = "SELECT 1 FROM sqlite_master WHERE name = 'queue_transitions'";
+    expect(reusedStatement(one, sql)).toBe(reusedStatement(one, sql));
+    expect(new WaitingStatements(one).statement(sql)).toBe(reusedStatement(one, sql));
+    expect(reusedStatement(two, sql)).not.toBe(reusedStatement(one, sql));
+    // Past the cap a text is prepared fresh each time, never kept.
+    const texts = Array.from({ length: 40 }, (_, i) => `SELECT ${i} AS n`);
+    for (const text of texts) reusedStatement(two, text);
+    expect(reusedStatement(two, texts[39]!)).not.toBe(reusedStatement(two, texts[39]!));
+    expect(reusedStatement(two, texts[0]!)).toBe(reusedStatement(two, texts[0]!));
+    // A closed connection fails as a prepare on it would; the other keeps working.
+    one.close();
+    expect(() => lastMeaningfulTransition(one, "q-1")).toThrow();
+    expect(lastMeaningfulTransition(two, "q-1")).toBeNull();
+  });
+});