fix(server): SQLite writes no longer block the event loop

apps/server/src/orchestration-v2/FoundationPersistence.test.ts

@@ -1,4 +1,3 @@
-import * as NodeSqlite from "node:sqlite";
 import * as NodeServices from "@effect/platform-node/NodeServices";
 import { assert, it } from "@effect/vitest";
 import {
@@ -191,47 +190,44 @@ it.effect("discovers on the reader worker and deletes on the writer", () =>
     const path = yield* Path.Path;
     const filename = path.join(yield* fs.makeTempDirectoryScoped(), "fixture.sqlite");
     yield* Effect.gen(function* () {
-      const sql = yield* SqlClient.SqlClient;
       yield* seedCompaction();
-      // Establish the reader before recording main-thread statements.
-      yield* sql`SELECT 1`.pipe(NodeSqliteClient.readOnly);
-      const reads: string[] = [];
-      const writes: string[] = [];
-      const all = NodeSqlite.StatementSync.prototype.all;
-      const run = NodeSqlite.StatementSync.prototype.run;
-      yield* Effect.acquireRelease(
-        Effect.sync(() => {
-          const readSpy = vi
-            .spyOn(NodeSqlite.StatementSync.prototype, "all")
-            .mockImplementation(function (this: NodeSqlite.StatementSync, ...params) {
-              reads.push(this.sourceSQL);
-              return all.apply(this, params);
-            });
-          const writeSpy = vi
-            .spyOn(NodeSqlite.StatementSync.prototype, "run")
-            .mockImplementation(function (this: NodeSqlite.StatementSync, ...params) {
-              writes.push(this.sourceSQL);
-              return run.apply(this, params);
-            });
-          return { readSpy, writeSpy };
-        }),
-        ({ readSpy, writeSpy }) =>
-          Effect.sync(() => {
-            readSpy.mockRestore();
-            writeSpy.mockRestore();
-          }),
-      );
+      const statements: Array<{ readonly query: unknown; readonly connection: unknown }> = [];
+      const tracer = Tracer.make({
+        span(options) {
+          const span = new Tracer.NativeSpan(options);
+          const end = span.end.bind(span);
+          span.end = (time, exit) => {
+            end(time, exit);
+            const query = span.attributes.get("db.query.text");
+            if (query === undefined) return;
+            statements.push({ query, connection: span.attributes.get("db.connection") });
+          };
+          return span;
+        },
+      });
       const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
-      const result = yield* maintenance.pruneEventStore();
+      const result = yield* maintenance
+        .pruneEventStore()
+        .pipe(
+          Effect.withTracer(tracer),
+          Effect.provideService(Statement.SpanPropagationEnabled, true),
+        );
       assert.equal(result.deletedEventCount, 150);
       assert.equal(result.deletedReceiptCount, 100);
       // Only the in-transaction check for rows appended during the pass reads on the writer.
       assert.deepEqual(
-        reads.filter((query) => /\bSELECT\b/.test(query) && !query.includes("AS prompt_sequence")),
-        [],
+        new Set(
+          statements
+            .filter(
+              ({ query }) =>
+                typeof query === "string" &&
+                /^\s*(SELECT|WITH)\b/.test(query) &&
+                !query.includes("AS prompt_sequence"),
+            )
+            .map(({ connection }) => connection),
+        ),
+        new Set(["reader"]),
       );
-      assert(writes.some((query) => query.includes("DELETE FROM orchestration_events")));
-      assert(writes.some((query) => query.includes("DELETE FROM orchestration_command_receipts")));
     }).pipe(
       Effect.provide(
         ProjectionMaintenance.layer.pipe(

apps/server/src/persistence/Sqlite.test.ts

@@ -3,31 +3,16 @@ import * as NodeFS from "node:fs";
 import * as NodeOS from "node:os";
 import * as NodePath from "node:path";
 import * as NodeSqlite from "node:sqlite";
-import type * as NodeWorkerThreads from "node:worker_threads";
 
 import * as NodeServices from "@effect/platform-node/NodeServices";
 import { assert, it } from "@effect/vitest";
 import * as HostProcess from "@t3tools/shared/HostProcess";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as SqlClient from "effect/sql/SqlClient";
-import { vi } from "vite-plus/test";
 
 import * as SqlitePersistence from "./Sqlite.ts";
 
-const workers = vi.hoisted((): Array<NodeWorkerThreads.Worker> => []);
-
-vi.mock("node:worker_threads", async (importOriginal) => {
-  const actual = await importOriginal<typeof NodeWorkerThreads>();
-  class RecordedWorker extends actual.Worker {
-    constructor(...args: ConstructorParameters<typeof actual.Worker>) {
-      super(...args);
-      workers.push(this);
-    }
-  }
-  return { ...actual, Worker: RecordedWorker };
-});
-
 it.effect("keeps writes moving past a 48 MiB WAL while a reader retains its snapshot", () => {
   const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-sqlite-burst-"));
   const dbPath = NodePath.join(tempDir, "state.sqlite");
@@ -37,7 +22,6 @@ it.effect("keeps writes moving past a 48 MiB WAL while a reader retains its snap
 
   return Effect.gen(function* () {
     const sql = yield* SqlClient.SqlClient;
-    const checkpointWorker = workers.at(-1)!;
     yield* sql`CREATE TABLE burst_probe(id INTEGER PRIMARY KEY, payload BLOB)`;
     yield* sql`INSERT INTO burst_probe VALUES (1, NULL)`;
     yield* sql`PRAGMA wal_checkpoint(TRUNCATE)`;
@@ -51,43 +35,6 @@ it.effect("keeps writes moving past a 48 MiB WAL while a reader retains its snap
     yield* sql.withTransaction(
       sql`INSERT INTO burst_probe VALUES (2, zeroblob(${48 * 1024 * 1024}))`,
     );
-    const state = yield* Effect.callback<{
-      readonly log: number;
-      readonly checkpointed: number;
-      readonly blocked: boolean;
-    }>((resume) => {
-      const onMessage = (message: {
-        readonly type?: string;
-        readonly log?: number;
-        readonly checkpointed?: number;
-        readonly blocked?: boolean;
-      }) => {
-        if (
-          message.type === "checkpoint" &&
-          typeof message.log === "number" &&
-          message.log > 10_000 &&
-          typeof message.checkpointed === "number" &&
-          typeof message.blocked === "boolean"
-        ) {
-          checkpointWorker.off("message", onMessage);
-          resume(
-            Effect.succeed({
-              log: message.log,
-              checkpointed: message.checkpointed,
-              blocked: message.blocked,
-            }),
-          );
-        }
-      };
-      checkpointWorker.on("message", onMessage);
-      checkpointWorker.ref();
-      // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin.
-      checkpointWorker.postMessage("barrier");
-      return Effect.sync(() => checkpointWorker.off("message", onMessage));
-    });
-    assert.isBelow(state.checkpointed, state.log);
-    assert.isFalse(state.blocked);
-
     yield* sql.withTransaction(sql`INSERT INTO burst_probe VALUES (3, NULL)`);
     assert.equal(reader.prepare("SELECT COUNT(*) AS count FROM burst_probe").get()?.count, 1);
     reader.exec("COMMIT");

apps/server/src/persistence/Sqlite.ts

@@ -54,7 +54,9 @@ export const layerFromPath = Effect.fn("makeSqlitePersistenceLive")(function* (d
     NodeSqliteClient.layer({
       filename: dbPath,
       readerWorker: true,
-      // Run checkpoint fsyncs off the event loop so slow disk flushes cannot stall clients.
+      // Run writes off the event loop so slow statements and transactions cannot stall clients.
+      writerWorker: true,
+      // Bounds WAL growth while readers pin snapshots, in the writer worker or on a main-thread fallback.
       checkpointWorker: true,
       spanAttributes: {
         "db.name": path.basename(dbPath),

packages/shared/src/nodeSqliteClient.test.ts

@@ -168,30 +168,32 @@ const makeTempDatabase = Effect.gen(function* () {
   return { filename, other };
 });
 
-it.effect("keeps another connection from committing between a transaction's read and write", () =>
-  Effect.gen(function* () {
-    const { filename, other } = yield* makeTempDatabase;
-    yield* Effect.gen(function* () {
-      const sql = yield* SqlClient.SqlClient;
-      yield* sql.withTransaction(
-        Effect.gen(function* () {
-          const [row] = yield* sql<{ readonly value: number }>`
+it.effect.each([{ writerWorker: false }, { writerWorker: true }])(
+  "keeps another connection from committing between a transaction's read and write (writerWorker=$writerWorker)",
+  ({ writerWorker }) =>
+    Effect.gen(function* () {
+      const { filename, other } = yield* makeTempDatabase;
+      yield* Effect.gen(function* () {
+        const sql = yield* SqlClient.SqlClient;
+        yield* sql.withTransaction(
+          Effect.gen(function* () {
+            const [row] = yield* sql<{ readonly value: number }>`
             SELECT value FROM counters WHERE id = 1
           `;
-          // Deferred BEGIN permits a competing commit, causing SQLITE_BUSY_SNAPSHOT that no timeout can resolve.
-          yield* Effect.sync(() =>
-            assert.throws(
-              () => other.exec("UPDATE counters SET value = value + 1 WHERE id = 1"),
-              /database is locked/,
-            ),
-          );
-          yield* sql`UPDATE counters SET value = ${(row?.value ?? 0) + 10} WHERE id = 1`;
-        }),
-      );
-      yield* Effect.sync(() => other.exec("UPDATE counters SET value = value + 1 WHERE id = 1"));
-      assert.deepEqual(yield* sql`SELECT value FROM counters WHERE id = 1`.values, [[11]]);
-    }).pipe(Effect.provide(SqliteClient.layer({ filename })));
-  }).pipe(Effect.provide(NodeServices.layer)),
+            // Deferred BEGIN permits a competing commit, causing SQLITE_BUSY_SNAPSHOT that no timeout can resolve.
+            yield* Effect.sync(() =>
+              assert.throws(
+                () => other.exec("UPDATE counters SET value = value + 1 WHERE id = 1"),
+                /database is locked/,
+              ),
+            );
+            yield* sql`UPDATE counters SET value = ${(row?.value ?? 0) + 10} WHERE id = 1`;
+          }),
+        );
+        yield* Effect.sync(() => other.exec("UPDATE counters SET value = value + 1 WHERE id = 1"));
+        assert.deepEqual(yield* sql`SELECT value FROM counters WHERE id = 1`.values, [[11]]);
+      }).pipe(Effect.provide(SqliteClient.layer({ filename, writerWorker })));
+    }).pipe(Effect.provide(NodeServices.layer)),
 );
 
 it.effect("waits for another process's lock while the database opens", () =>
@@ -222,21 +224,75 @@ it.effect("waits for another process's lock while the database opens", () =>
   }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
 );
 
-it.effect("reports a transaction blocked by another writer as a lock timeout", () =>
+it.effect.each([{ writerWorker: false }, { writerWorker: true }])(
+  "reports a transaction blocked by another writer as a lock timeout (writerWorker=$writerWorker)",
+  ({ writerWorker }) =>
+    Effect.gen(function* () {
+      const { filename, other } = yield* makeTempDatabase;
+      yield* Effect.gen(function* () {
+        const sql = yield* SqlClient.SqlClient;
+        yield* sql`PRAGMA busy_timeout = 0`;
+        const read = sql.withTransaction(sql`SELECT value FROM counters WHERE id = 1`.values);
+
+        yield* Effect.sync(() => other.exec("BEGIN IMMEDIATE"));
+        const error = yield* read.pipe(Effect.flip);
+        assert.equal(error.reason._tag, "LockTimeoutError");
+
+        yield* Effect.sync(() => other.exec("COMMIT"));
+        assert.deepEqual(yield* read, [[0]]);
+      }).pipe(Effect.provide(SqliteClient.layer({ filename, writerWorker })));
+    }).pipe(Effect.provide(NodeServices.layer)),
+);
+
+it.effect("keeps the event loop free while a write waits on another connection's lock", () =>
   Effect.gen(function* () {
     const { filename, other } = yield* makeTempDatabase;
     yield* Effect.gen(function* () {
       const sql = yield* SqlClient.SqlClient;
-      yield* sql`PRAGMA busy_timeout = 0`;
-      const read = sql.withTransaction(sql`SELECT value FROM counters WHERE id = 1`.values);
-
       yield* Effect.sync(() => other.exec("BEGIN IMMEDIATE"));
-      const error = yield* read.pipe(Effect.flip);
-      assert.equal(error.reason._tag, "LockTimeoutError");
+      const write = yield* Effect.forkChild(
+        sql`UPDATE counters SET value = value + 1 WHERE id = 1`,
+        { startImmediately: true },
+      );
+      // A writer on this thread would spin in busy_timeout, so the holder could never commit.
+      yield* Effect.sync(() => other.exec("UPDATE counters SET value = 10 WHERE id = 1; COMMIT"));
+      yield* Fiber.join(write);
+      assert.deepEqual(yield* sql`SELECT value FROM counters WHERE id = 1`.values, [[11]]);
+    }).pipe(Effect.provide(SqliteClient.layer({ filename, writerWorker: true })));
+  }).pipe(Effect.provide(NodeServices.layer)),
+);
 
+it.effect("holds writer-worker writes past the WAL size limit until a pinned reader releases", () =>
+  Effect.gen(function* () {
+    const fs = yield* FileSystem.FileSystem;
+    const { filename, other } = yield* makeTempDatabase;
+    yield* Effect.gen(function* () {
+      const sql = yield* SqlClient.SqlClient;
+      const limit = 2 * 1024 * 1024;
+      yield* sql`CREATE TABLE IF NOT EXISTS blobs(payload BLOB)`;
+      yield* sql.unsafe(`PRAGMA journal_size_limit = ${limit}`);
+      yield* sql`PRAGMA wal_checkpoint(TRUNCATE)`;
+      const blob = sql`INSERT INTO blobs VALUES (zeroblob(${1536 * 1024}))`;
+      yield* Effect.sync(() => {
+        other.exec("BEGIN");
+        other.prepare("SELECT COUNT(*) FROM counters").get();
+      });
+      yield* sql.withTransaction(blob);
+      yield* sql.withTransaction(blob);
+      // Held writes, mutating pragmas included, can be cancelled before they run.
+      for (const cancelled of [blob, sql`PRAGMA user_version = 7`]) {
+        yield* Fiber.interrupt(yield* Effect.forkChild(cancelled, { startImmediately: true }));
+      }
+      const held = yield* Effect.forkChild(sql.withTransaction(blob), { startImmediately: true });
       yield* Effect.sync(() => other.exec("COMMIT"));
-      assert.deepEqual(yield* read, [[0]]);
-    }).pipe(Effect.provide(SqliteClient.layer({ filename })));
+      yield* Fiber.join(held);
+      // Without admission the third write appends to a WAL the pinned reader kept from resetting.
+      assert.isBelow(Number((yield* fs.stat(`${filename}-wal`)).size), limit);
+      assert.deepEqual(yield* sql`SELECT COUNT(*) FROM blobs`.values, [[3]]);
+      assert.deepEqual(yield* sql`PRAGMA user_version`.values, [[0]]);
+    }).pipe(
+      Effect.provide(SqliteClient.layer({ filename, writerWorker: true, checkpointWorker: true })),
+    );
   }).pipe(Effect.provide(NodeServices.layer)),
 );