fix(ui-server): take the WAL checkpoint off the journal write path

.changeset/journal-ledger-off-write-path.md

@@ -0,0 +1,9 @@
+---
+"@schlessera/brain-ui-server": patch
+"@schlessera/brain-ui-sdk": patch
+---
+
+Changed: a journal write no longer truncates the SQLite WAL first; admission charges the WAL as it stands and truncates only when that would refuse. Commits stay `synchronous=FULL` (#2789).
+Changed: journal capacity measurement walks only the journal's own SQLite b-trees, not the whole database.
+Changed: a journal row changed outside capacity accounting on the server's connection is refused.
+Changed: the server's maintenance temporary-storage ledger stays open between reservations instead of opening and closing its database each time.

packages/ui-sdk/src/server/maintenance-temp-budget.ts

@@ -76,12 +76,12 @@ function sharedState(domain: MaintenanceTempDomain) {
   const id = createHash("sha256").update(statePath).digest("hex").slice(0, 24);
   return { statePath, parent, authority, mode, file: join(parent, `.brain-maintenance-${id}.db`) };
 }
-function open(directoryPath: string, domain?: MaintenanceTempDomain): Database {
-  let file: string;
+/** Validates (creating if needed) the ledger file; returns it with the inode it names now. */
+function ledgerFile(directoryPath: string, domain?: MaintenanceTempDomain): { file: string; node: { dev: number; ino: number } } {
   if (domain) {
     const shared = sharedState(domain);
     if (resolve(directoryPath) !== shared.parent) throw new Error("maintenance shared ledger directory does not match its state identity");
-    file = shared.file;
+    const file = shared.file;
     let fd: number | undefined;
     try {
       fd = openSync(file, constants.O_CREAT | constants.O_EXCL | constants.O_RDWR | constants.O_NOFOLLOW, 0o600);
@@ -99,18 +99,22 @@ function open(directoryPath: string, domain?: MaintenanceTempDomain): Database {
       || ((node.mode & 0o070) !== 0 && node.gid !== shared.authority.gid)) {
       throw new Error("maintenance shared ledger must preserve existing state access and be an unlinked regular file");
     }
+    return { file, node };
   } else {
     mkdirSync(directoryPath, { mode: 0o700, recursive: true });
     const info = lstatSync(directoryPath);
     if (!info.isDirectory() || info.uid !== (process.getuid?.() ?? 0) || (info.mode & 0o077) !== 0) {
       throw new Error("maintenance admission requires an owned private directory outside the brain");
     }
-    file = join(directoryPath, "reservations.db");
+    const file = join(directoryPath, "reservations.db");
     try { closeSync(openSync(file, constants.O_CREAT | constants.O_EXCL | constants.O_RDWR | constants.O_NOFOLLOW, 0o600)); }
     catch (error) { if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; }
     const node = lstatSync(file);
     if (!node.isFile() || node.nlink !== 1 || node.uid !== info.uid || (node.mode & 0o077) !== 0) throw new Error("maintenance admission database must be an owned private regular file");
+    return { file, node };
   }
+}
+function connect(file: string): Database {
   // Node consumers may import the SDK/Claude entry without running Bun-only
   // maintenance. Resolve its database only when that runtime actually runs.
   const { Database } = createRequire(import.meta.url)("bun:sqlite") as typeof import("bun:sqlite");
@@ -120,6 +124,29 @@ function open(directoryPath: string, domain?: MaintenanceTempDomain): Database {
     return db;
   } catch (error) { db.close(); throw error; }
 }
+function open(directoryPath: string, domain?: MaintenanceTempDomain): Database { return connect(ledgerFile(directoryPath, domain).file); }
+/**
+ * The server keeps each validated ledger open, so a reservation or release
+ * is one commit, not an open, commit, checkpoint and close (#2789). Each use
+ * validates the path again and reopens when it names another file.
+ * Reservations bind live processes of this boot only: an OS crash ends every
+ * owner, and the next boot prunes their rows by boot identity. So commits
+ * need not reach the disk before admission returns (NORMAL); a process crash
+ * still loses no committed reservation.
+ */
+const held = new Map<string, { db: Database; dev: number; ino: number }>();
+function ledger(directoryPath: string, domain?: MaintenanceTempDomain): Database {
+  const { file, node } = ledgerFile(directoryPath, domain);
+  // An open handle outlives a revoked mode; reuse it only while a fresh open could succeed.
+  accessSync(file, constants.R_OK | constants.W_OK);
+  const cached = held.get(file);
+  if (cached && cached.dev === node.dev && cached.ino === node.ino) return cached.db;
+  if (cached) { held.delete(file); cached.db.close(); }
+  const db = connect(file);
+  try { db.exec("PRAGMA synchronous=NORMAL"); } catch (error) { db.close(); throw error; }
+  held.set(file, { db, dev: node.dev, ino: node.ino });
+  return db;
+}
 function prune(db: Database): Entry[] {
   const entries = db.query("SELECT * FROM reservations").all() as Entry[];
   return entries.filter(entry => {
@@ -138,10 +165,9 @@ export function reserveMaintenanceTemp(configured: WorkerStateLimits, domain?: M
   const shared = domain ? sharedState(domain) : undefined;
   const path = shared?.parent ?? directory();
   const selected = shared ? { statePath: shared.statePath } : undefined;
-  const db = open(path, selected);
+  const db = ledger(path, selected);
   const id = randomUUID();
-  try {
-    db.transaction(() => {
+  db.transaction(() => {
       const entries = prune(db);
       const held = entries.reduce((total, entry) => total + BigInt(entry.bytes), 0n);
       if (held + BigInt(limits.allowanceBytes) > BigInt(limits.budgetBytes)) {
@@ -151,19 +177,16 @@ export function reserveMaintenanceTemp(configured: WorkerStateLimits, domain?: M
           + "Nothing ran. Retry after a maintenance command finishes; no agent runtime state was lost.");
       }
       db.query("INSERT INTO reservations (id, bytes, owner) VALUES (?, ?, ?)").run(id, limits.allowanceBytes, JSON.stringify(identity()));
-    }).immediate();
-  } finally { db.close(); }
+  }).immediate();
   let released = false;
   return { id, directory: path, ...(selected ?? {}), release() {
     if (released) return;
-    const db = open(path, selected);
-    try {
-      db.transaction(() => {
-        db.query("UPDATE reservations SET released = 1 WHERE id = ?").run(id);
-        prune(db);
-      }).immediate();
-      released = true;
-    } finally { db.close(); }
+    const db = ledger(path, selected);
+    db.transaction(() => {
+      db.query("UPDATE reservations SET released = 1 WHERE id = ?").run(id);
+      prune(db);
+    }).immediate();
+    released = true;
   } };
 }
 

packages/ui-sdk/tests/maintenance-temp-budget.test.ts

@@ -190,3 +190,25 @@ test("a live owner hidden by procfs is held rather than mistaken for a crashed r
   try { expect(()=>reserveMaintenanceTemp({allowanceBytes:DEFAULT,budgetBytes:DEFAULT}),"UNVERIFIED_LIVE_OWNER_RETAINED").toThrow("busy"); }
   finally { hidden.mockRestore(); first.release(); }
 });
+
+test("server admission keeps its validated ledger open and reopens a replaced ledger file", () => {
+  const root=mkdtempSync("/tmp/odysseus-maintenance-held-");roots.push(root);
+  const statePath=join(root,"ui.db"),limits={allowanceBytes:DEFAULT,budgetBytes:2*DEFAULT};
+  const close=spyOn(Database.prototype,"close");
+  try {
+    for(let index=0;index<3;index++){const reservation=reserveMaintenanceTemp(limits,{statePath});reservation.release();}
+    expect(close,"MAINTENANCE_LEDGER_HELD_OPEN").not.toHaveBeenCalled();
+    const ledger=fs.readdirSync(root).find(name=>/^\.brain-maintenance-[a-f0-9]{24}\.db$/.test(name))!;
+    const held=reserveMaintenanceTemp(limits,{statePath});
+    for(const suffix of ["","-wal","-shm"]) rmSync(join(root,ledger+suffix),{force:true});
+    const replaced=reserveMaintenanceTemp(limits,{statePath});
+    expect(close,"MAINTENANCE_REPLACED_LEDGER_REOPENED").toHaveBeenCalledTimes(1);
+    const db=new Database(join(root,ledger));
+    try { expect(db.query("SELECT id FROM reservations").all(),"MAINTENANCE_ADMISSION_USES_CURRENT_LEDGER").toEqual([{id:replaced.id}]); }
+    finally { db.close(); }
+    fs.chmodSync(join(root,ledger),0o000);
+    try { expect(()=>reserveMaintenanceTemp(limits,{statePath}),"MAINTENANCE_REVOKED_LEDGER_NOT_REUSED").toThrow("EACCES"); }
+    finally { fs.chmodSync(join(root,ledger),0o600); }
+    replaced.release();held.release();
+  } finally { close.mockRestore(); }
+});

packages/ui-server/migrations/063_turn_journal_ledger.sql

@@ -0,0 +1,4 @@
+-- Journal writes without per-write WAL truncation (#2789). `writing` is set
+-- only inside an accounted write and is cleared before that write commits;
+-- connection-local triggers refuse journal rows changed while it is clear.
+ALTER TABLE turn_journal_allocation ADD COLUMN writing INTEGER NOT NULL DEFAULT 0;

packages/ui-server/src/brain/journal-accounting.ts

@@ -11,6 +11,19 @@ export class JournalStorageUnavailable extends Error {
   constructor(message: string) { super(`The change journal cannot safely retain this observation: ${message}`); }
 }
 
+/**
+ * Connection-local triggers: a journal row changed outside an accounted
+ * write is refused, so no journal storage reaches the disk unadmitted.
+ */
+function installEscapeGuard(db: Database) {
+  const tables = (db.query("SELECT name FROM sqlite_schema WHERE type='table' AND name GLOB 'turn_journal_*' AND name != 'turn_journal_allocation'").all() as { name: string }[]).map(row => row.name);
+  for (const table of tables) for (const event of ["INSERT", "UPDATE", "DELETE"]) {
+    db.exec(`CREATE TEMP TRIGGER IF NOT EXISTS ${table}_accounted_${event.toLowerCase()} BEFORE ${event} ON main.${table}
+      WHEN (SELECT writing FROM turn_journal_allocation WHERE id = 1) IS NOT 1
+      BEGIN SELECT RAISE(ABORT, 'journal storage write outside capacity accounting'); END`);
+  }
+}
+
 export function createJournalAccounting(db: Database, dbPath: string, options: {
   capBytes?: number;
   /** An internal deterministic scheduling hook; production does not supply one. */
@@ -26,23 +39,29 @@ export function createJournalAccounting(db: Database, dbPath: string, options: {
   const cap = options.capBytes ?? JOURNAL_CAP_BYTES;
   const memory = dbPath === ":memory:" || dbPath === "";
   const pageSize = (db.query("PRAGMA page_size").get() as { page_size: number }).page_size;
-  const wal = () => { try { return statSync(`${dbPath}-wal`).size; } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return 0; throw error; } };
+  const wal = () => { if (memory) return 0; try { return statSync(`${dbPath}-wal`).size; } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return 0; throw error; } };
+  installEscapeGuard(db);
   function measure() {
     const prior = db.query("SELECT pages FROM turn_journal_allocation WHERE id = 1").get() as { pages: string };
     const logicalPages = (db.query("PRAGMA page_count").get() as { page_count: number }).page_count;
     const extent = memory ? logicalPages : Math.max(logicalPages, Math.ceil(statSync(dbPath).size / pageSize));
     const historical = new Set<number>((JSON.parse(prior.pages) as number[]).filter(page => page <= extent));
-    const names = new Set((db.query("SELECT name FROM sqlite_schema WHERE tbl_name GLOB 'turn_journal_*'").all() as { name: string }[]).map(row => row.name));
-    let owned = 0;
-    // Start with formerly-owned pages. Those still on SQLite's free list stay
-    // charged; a page reassigned to another component no longer belongs here.
-    let retainedFree = historical.size;
-    for (const row of db.query("SELECT pageno, name FROM dbstat").iterate() as Iterable<{ pageno: number; name: string }>) {
-      if (historical.has(row.pageno)) retainedFree--;
-      if (names.has(row.name)) { owned++; historical.add(row.pageno); }
-      else historical.delete(row.pageno);
-    }
+    const names = (db.query("SELECT name FROM sqlite_schema WHERE tbl_name GLOB 'turn_journal_*'").all() as { name: string }[]).map(row => row.name);
+    // Only the journal's own b-trees are walked, never the whole database.
+    const journal = new Set<number>();
+    for (const name of names) for (const row of db.query("SELECT pageno FROM dbstat WHERE name = ?").iterate(name) as Iterable<{ pageno: number }>) journal.add(row.pageno);
+    const owned = journal.size;
     const free = (db.query("PRAGMA freelist_count").get() as { freelist_count: number }).freelist_count;
+    // Formerly-owned pages outside the journal stay charged while they can
+    // still be free. Without walking other b-trees, a page another component
+    // reused looks the same as a free one, so at most as many stay charged as
+    // SQLite has unallocated pages: never fewer than the pages truly free.
+    let released = [...historical].filter(page => !journal.has(page));
+    const unallocated = free + Math.max(0, extent - logicalPages);
+    if (released.length > unallocated) released = released.slice(released.length - unallocated);
+    const retainedFree = released.length;
+    for (const page of released) journal.add(page);
+    const pages = journal;
     // One spill-disabled commit writes each journal page at most once. Global
     // allocation can also touch page 1 and free-list trunk pages, not unrelated
     // Activity payload pages. Keep this temporary capacity reserved even after
@@ -51,47 +70,83 @@ export function createJournalAccounting(db: Database, dbPath: string, options: {
     const physical = memory ? 0 : (owned + retainedFree + 1) * pageSize;
     const objects = (db.query("SELECT coalesce(sum(bytes),0) AS bytes FROM turn_journal_objects WHERE state='present'").get() as { bytes: number }).bytes;
     const pending = (db.query("SELECT coalesce(sum(bytes),0) AS bytes FROM turn_journal_reservations").get() as { bytes: number }).bytes;
-    return { pages: [...historical].sort((a, b) => a - b), walCapacity, bytes: physical + walCapacity + objects + pending };
+    return { pages: [...pages].sort((a, b) => a - b), walCapacity, bytes: physical + walCapacity + objects + pending };
   }
   class CheckpointRace extends Error {}
+  class WalDebt extends Error {}
+  function checkpoint() {
+    if (!memory) {
+      const result = db.query("PRAGMA wal_checkpoint(TRUNCATE)").get() as { busy: number };
+      if (result.busy) throw new JournalStorageUnavailable("a database reader still holds journal write-ahead data");
+    }
+    options.afterCheckpoint?.();
+  }
+  /**
+   * One accounted transaction. Writes no longer truncate the WAL first, so
+   * admission charges the whole WAL file as it stands, whoever wrote it:
+   * earlier journal commits, an abandoned or failed commit's tail, other
+   * components' frames. The measurement's WAL capacity covers this commit.
+   * `fresh` requires an empty WAL under this write lock, right after a
+   * checkpoint, as every write used to.
+   */
+  function admit<T>(operation: () => T, fresh: boolean, retryable: boolean) {
+    return db.transaction(() => {
+      const walBefore = wal();
+      // Checkpoint and BEGIN cannot be one SQLite critical section. A
+      // writer that won between them is detected under this write lock;
+      // roll back before any journal mutation and retry the checkpoint.
+      if (fresh && walBefore !== 0) throw new CheckpointRace();
+      const before = measure();
+      if (retryable && walBefore > 0 && before.bytes + walBefore > cap) throw new WalDebt();
+      db.query("UPDATE turn_journal_allocation SET pages = ?, writing = 1 WHERE id = 1").run(JSON.stringify(before.pages));
+      const result = operation();
+      for (let pass = 0; pass < 10; pass++) {
+        const measured = measure();
+        if (measured.bytes + walBefore > cap) throw retryable && walBefore > 0 ? new WalDebt() : new JournalCapacityError();
+        // The WAL after this commit stays within what it held plus this commit.
+        const pages = JSON.stringify(measured.pages), reserved = walBefore + measured.walCapacity;
+        const prior = db.query("SELECT pages,wal_reserved,writing FROM turn_journal_allocation WHERE id = 1").get() as { pages: string; wal_reserved: number; writing: number };
+        if (prior.pages === pages && prior.wal_reserved === reserved && prior.writing === 0) return result;
+        db.query("UPDATE turn_journal_allocation SET pages = ?,wal_reserved = ?,writing = 0 WHERE id = 1").run(pages, reserved);
+      }
+      throw new JournalStorageUnavailable("allocation accounting did not stabilize");
+    }).immediate();
+  }
+  let established = false;
+  /** Every journal commit stays FULL: it is durable before the turn proceeds. */
   const write = <T>(operation: () => T): T => {
-    if (db.inTransaction) throw new JournalStorageUnavailable("a journal write cannot checkpoint inside another transaction");
+    if (db.inTransaction) throw new JournalStorageUnavailable("a journal write cannot run inside another transaction");
     const spill = (db.query("PRAGMA cache_spill").get() as { cache_spill: number }).cache_spill;
     const synchronous = (db.query("PRAGMA synchronous").get() as { synchronous: number }).synchronous;
     db.exec("PRAGMA synchronous = FULL");
     db.exec("PRAGMA cache_spill = OFF");
     try {
-      for (let attempt = 0; attempt < 3; attempt++) {
-        if (!memory) {
-          const checkpoint = db.query("PRAGMA wal_checkpoint(TRUNCATE)").get() as { busy: number };
-          if (checkpoint.busy) throw new JournalStorageUnavailable("a database reader still holds journal write-ahead data");
-        }
-        options.afterCheckpoint?.();
+      // A ledger's first write starts from an empty WAL. Later writes take no
+      // checkpoint unless the WAL they would carry refuses admission; then
+      // they truncate and retry, as every write used to.
+      let fresh = !established;
+      for (let attempt = 0; attempt < 4; attempt++) {
+        if (fresh) checkpoint();
         try {
-          return db.transaction(() => {
-            // Checkpoint and BEGIN cannot be one SQLite critical section. A
-            // writer that won between them is detected under this write lock;
-            // roll back before any journal mutation and retry the checkpoint.
-            if (!memory && wal() !== 0) throw new CheckpointRace();
-            const before = measure();
-            db.query("UPDATE turn_journal_allocation SET pages = ? WHERE id = 1").run(JSON.stringify(before.pages));
-            const result = operation();
-            for (let pass = 0; pass < 10; pass++) {
-              const measured = measure();
-              if (measured.bytes > cap) throw new JournalCapacityError();
-              const pages = JSON.stringify(measured.pages);
-              const prior = db.query("SELECT pages,wal_reserved FROM turn_journal_allocation WHERE id = 1").get() as { pages: string; wal_reserved: number };
-              if (prior.pages === pages && prior.wal_reserved === measured.walCapacity) return result;
-              db.query("UPDATE turn_journal_allocation SET pages = ?,wal_reserved = ? WHERE id = 1").run(pages, measured.walCapacity);
-            }
-            throw new JournalStorageUnavailable("allocation accounting did not stabilize");
-          }).immediate();
-        } catch (error) { if (!(error instanceof CheckpointRace)) throw error; }
+          const result = admit(operation, fresh, !fresh);
+          established = true;
+          return result;
+        } catch (error) {
+          if (error instanceof WalDebt) { fresh = true; continue; }
+          if (!(error instanceof CheckpointRace)) throw error;
+        }
       }
       throw new JournalStorageUnavailable("another database writer repeatedly won checkpoint admission");
     } finally { db.exec(`PRAGMA cache_spill = ${spill}`); db.exec(`PRAGMA synchronous = ${synchronous}`); }
   };
-  const usage = (): number => measure().bytes;
+  const usage = (): number => {
+    const measured = measure();
+    // The standing charge keeps at least one commit of WAL headroom reserved.
+    return measured.bytes - measured.walCapacity + Math.max(measured.walCapacity, wal());
+  };
+  // Truncate while the server starts. A pinned WAL or an over-cap store
+  // leaves this to the first write, which refuses or recovers as it always did.
+  try { write(() => {}); } catch (error) { if (!(error instanceof JournalStorageUnavailable || error instanceof JournalCapacityError)) throw error; }
   return {
     write, usage,
     openRead(hash, assertOwner) {