fix(shared): the SQLite reader pool comes back after a failed start

packages/shared/src/nodeSqliteClient.test.ts

@@ -4,12 +4,15 @@ import * as NodeServices from "@effect/platform-node/NodeServices";
 import { assert, it } from "@effect/vitest";
 import * as Deferred from "effect/Deferred";
 import * as Cause from "effect/Cause";
+import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
 import * as Exit from "effect/Exit";
 import * as Fiber from "effect/Fiber";
 import * as FileSystem from "effect/FileSystem";
 import * as Layer from "effect/Layer";
+import * as Logger from "effect/Logger";
 import * as Path from "effect/Path";
+import * as Queue from "effect/Queue";
 import * as SqlClient from "effect/sql/SqlClient";
 import * as SqlError from "effect/sql/SqlError";
 import * as Statement from "effect/sql/Statement";
@@ -22,6 +25,7 @@ import { vi } from "vite-plus/test";
 import * as SqliteClient from "./nodeSqliteClient.ts";
 
 const startedWorkers = vi.hoisted((): Array<NodeWorkerThreads.Worker> => []);
+const workerExits = vi.hoisted(() => new WeakMap<NodeWorkerThreads.Worker, Promise<void>>());
 const silentWorkers = vi.hoisted(() => ({ on: false, readyOnTerminate: false }));
 const failingWorkers = vi.hoisted(() => ({ message: undefined as string | undefined }));
 const checkpointWorkers = vi.hoisted((): Array<NodeWorkerThreads.Worker> => []);
@@ -38,6 +42,7 @@ vi.mock("node:worker_threads", async (importOriginal) => {
             : args) as ConstructorParameters<typeof actual.Worker>),
       );
       startedWorkers.push(this);
+      workerExits.set(this, new Promise((resolve) => this.once("exit", () => resolve())));
       if (args[1]?.workerData?.frames !== undefined) checkpointWorkers.push(this);
     }
     override terminate() {
@@ -620,6 +625,197 @@ it.effect("uses the writer when the reader cannot open the database", () =>
   ),
 );
 
+const withLogLines = <A, E, R>(
+  use: (lines: Queue.Queue<ReadonlyArray<unknown>>) => Effect.Effect<A, E, R>,
+) =>
+  Effect.gen(function* () {
+    const lines = yield* Queue.unbounded<ReadonlyArray<unknown>>();
+    const logger = Logger.make(({ logLevel, message }) => {
+      Queue.offerUnsafe(lines, [logLevel, ...(message as ReadonlyArray<unknown>)]);
+    });
+    return yield* use(lines).pipe(
+      Effect.provide(Logger.layer([logger], { mergeWithExisting: false })),
+    );
+  });
+
+const readConnection = (sql: SqlClient.SqlClient) =>
+  Effect.gen(function* () {
+    const spans: Array<Tracer.Span> = [];
+    const observe: Statement.Transformer = (statement, _sql, _fiber, span) =>
+      Effect.sync(() => {
+        spans.push(span);
+        return statement;
+      });
+    const count = yield* countEntries(sql).pipe(
+      SqliteClient.readOnly,
+      Effect.provideService(Statement.CurrentTransformer, observe),
+      Effect.provideService(Statement.SpanPropagationEnabled, true),
+    );
+    return { count, connection: spans.map((span) => span.attributes.get("db.connection")) };
+  });
+
+const withWorkers = <A, E, R>(
+  mode: { readonly failWith?: string; readonly silent?: boolean },
+  effect: Effect.Effect<A, E, R>,
+) =>
+  Effect.acquireUseRelease(
+    Effect.sync(() => {
+      failingWorkers.message = mode.failWith;
+      silentWorkers.on = mode.silent ?? false;
+    }),
+    () => effect,
+    () =>
+      Effect.sync(() => {
+        failingWorkers.message = undefined;
+        silentWorkers.on = false;
+      }),
+  );
+
+const unavailableWarning = "SQLite reader worker unavailable; reads use the writer";
+
+const notWal = "database uses journal_mode delete, not wal";
+
+it.effect.each([
+  {
+    failure: "a crash",
+    workers: { failWith: "reader start failed" },
+    startup: Duration.zero,
+    reason: "Database reader failed: reader start failed",
+  },
+  {
+    failure: "a startup timeout",
+    workers: { silent: true },
+    startup: Duration.seconds(10),
+    reason: "Database reader did not open the database within 10000 ms",
+  },
+])(
+  "reads on the writer after $failure until a retry starts the reader again",
+  ({ workers, startup, reason }) =>
+    withLogLines((lines) =>
+      withReaderClient((sql) =>
+        Effect.gen(function* () {
+          const failed = yield* withWorkers(
+            workers,
+            Effect.gen(function* () {
+              const read = yield* Effect.forkChild(readConnection(sql), {
+                startImmediately: true,
+              });
+              yield* TestClock.adjust(startup);
+              return yield* Fiber.join(read);
+            }),
+          );
+          assert.deepEqual(failed, { count: 1, connection: ["writer"] });
+          const startedBefore = startedWorkers.length;
+          assert.deepEqual(yield* readConnection(sql), { count: 1, connection: ["writer"] });
+          assert.equal(startedWorkers.length, startedBefore);
+          assert.deepEqual(yield* Queue.take(lines), ["Warn", unavailableWarning, { reason }]);
+
+          yield* TestClock.adjust("1 second");
+          assert.equal(startedWorkers.length, startedBefore + 1);
+          assert.deepEqual(yield* Queue.take(lines), [
+            "Info",
+            "SQLite reader worker available again; reads use the reader workers",
+          ]);
+          assert.deepEqual(yield* readConnection(sql), { count: 1, connection: ["reader"] });
+          assert.equal(Queue.sizeUnsafe(lines), 0);
+        }),
+      ),
+    ),
+);
+
+it.effect("backs off a reader that never starts without logging each retry", () =>
+  withLogLines((lines) =>
+    withReaderClient((sql) =>
+      withWorkers(
+        { silent: true },
+        Effect.gen(function* () {
+          const read = yield* Effect.forkChild(readConnection(sql), { startImmediately: true });
+          yield* TestClock.adjust("10 seconds");
+          assert.deepEqual(yield* Fiber.join(read), { count: 1, connection: ["writer"] });
+
+          const startedBefore = startedWorkers.length;
+          // Each retry waits 10 s for startup; the gaps between them double from 1 s up to a minute.
+          yield* TestClock.adjust("10 minutes");
+          assert.equal(startedWorkers.length - startedBefore, 12);
+          assert.equal(Queue.sizeUnsafe(lines), 1);
+        }),
+      ),
+    ),
+  ),
+);
+
+it.effect.each([
+  { phase: "the retry waits", beforeClose: Duration.zero, retries: 0 },
+  { phase: "a retried reader starts", beforeClose: Duration.seconds(1), retries: 1 },
+])("stops retrying the reader when the client closes while $phase", ({ beforeClose, retries }) =>
+  withWorkers(
+    { silent: true },
+    Effect.gen(function* () {
+      const startedAtClose = yield* withReaderClient((sql) =>
+        Effect.gen(function* () {
+          const read = yield* Effect.forkChild(readConnection(sql), { startImmediately: true });
+          yield* TestClock.adjust("10 seconds");
+          assert.deepEqual(yield* Fiber.join(read), { count: 1, connection: ["writer"] });
+
+          const startedBefore = startedWorkers.length;
+          yield* TestClock.adjust(beforeClose);
+          assert.equal(startedWorkers.length - startedBefore, retries);
+          return startedWorkers.length;
+        }),
+      );
+      yield* TestClock.adjust("10 minutes");
+      assert.equal(startedWorkers.length, startedAtClose);
+    }),
+  ),
+);
+
+it.effect.each([
+  {
+    when: "the database is not in WAL mode",
+    journalMode: "DELETE",
+    workers: {},
+    starts: 1,
+    logged: [["Warn", unavailableWarning, { reason: notWal }]],
+  },
+  {
+    when: "a retry finds the database out of WAL mode",
+    journalMode: "WAL",
+    workers: { failWith: "reader start failed" },
+    starts: 2,
+    logged: [
+      ["Warn", unavailableWarning, { reason: "Database reader failed: reader start failed" }],
+      [
+        "Info",
+        "SQLite reader worker retries stopped permanently; reads use the writer",
+        { reason: notWal },
+      ],
+    ],
+  },
+])(
+  "keeps readers off without further retries when $when",
+  ({ journalMode, workers, starts, logged }) =>
+    withLogLines((lines) =>
+      withFileClient((sql) =>
+        Effect.gen(function* () {
+          yield* sql.unsafe(`PRAGMA journal_mode = ${journalMode}`);
+          yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`;
+          yield* sql`INSERT INTO entries VALUES ('kept')`;
+          const startedBefore = startedWorkers.length;
+
+          const failed = yield* withWorkers(workers, readConnection(sql));
+          assert.deepEqual(failed, { count: 1, connection: ["writer"] });
+          yield* sql`PRAGMA journal_mode = DELETE`;
+          yield* TestClock.adjust("1 second");
+          yield* Effect.promise(() => workerExits.get(startedWorkers.at(-1)!)!);
+          yield* TestClock.adjust("1 hour");
+          assert.deepEqual(yield* readConnection(sql), { count: 1, connection: ["writer"] });
+          assert.equal(startedWorkers.length, startedBefore + starts);
+          assert.deepEqual(yield* Queue.clear(lines), logged);
+        }),
+      ),
+    ),
+);
+
 it.effect("fails with a typed SqlError when the database file is not SQLite", () =>
   Effect.gen(function* () {
     const fs = yield* FileSystem.FileSystem;

packages/shared/src/nodeSqliteClient.ts

@@ -193,6 +193,9 @@ const hasLargeBinding = (params: ReadonlyArray<unknown>): boolean => {
   }
 };
 
+// A reader refuses a database outside WAL mode with this code; retrying cannot change that.
+const NOT_WAL = "ERR_SQLITE_NOT_WAL";
+
 // Inline worker source also runs in bundles and single-executable builds without module files.
 const workerSource = `
 const { parentPort, workerData } = require("node:worker_threads");
@@ -214,7 +217,9 @@ try {
   // Reads the header, so a file that is not a database fails here rather than on first use.
   const mode = String(next.prepare("PRAGMA journal_mode").get().journal_mode).toLowerCase();
   // Without WAL a reader's shared lock blocks the writer's commits, so refuse.
-  if (workerData.requireWal && mode !== "wal") throw new Error("database uses journal_mode " + mode + ", not wal");
+  if (workerData.requireWal && mode !== "wal") {
+    throw Object.assign(new Error("database uses journal_mode " + mode + ", not wal"), { code: ${JSON.stringify(NOT_WAL)} });
+  }
   db = next;
   parentPort.postMessage({ type: "ready" });
 } catch (error) {
@@ -345,6 +350,8 @@ const endsReadTransaction = (sql: string) => /^\s*ROLLBACK\b|^\s*COMMIT\b/i.test
 // WAL readers run concurrently; one worker serialized every read behind the slowest transaction.
 const READER_POOL_SIZE = 4;
 const READER_STARTUP_TIMEOUT = Duration.seconds(10);
+const READER_RETRY_DELAY = Duration.seconds(1);
+const READER_RETRY_MAX_DELAY = Duration.minutes(1);
 
 // Bind transactions to one worker; fail after its exit instead of resuming outside the transaction on a replacement.
 const makeWorkerConnection = Effect.fnUntraced(function* (options: {
@@ -566,9 +573,44 @@ const makeWorkerConnection = Effect.fnUntraced(function* (options: {
   });
 });
 
-// A reader that cannot start turns off the whole pool, so routing never picks a dead slot.
+const startFailure = (error: SqlError) => {
+  const cause = error.reason.cause;
+  return {
+    reason: cause instanceof Error ? cause.message : error.message,
+    permanent: Predicate.hasProperty(cause, "code") && cause.code === NOT_WAL,
+  };
+};
+
+// A reader that cannot start turns off the whole pool, so routing never picks a dead slot, until
+// one background retry starts it again.
 const makeReader = Effect.fnUntraced(function* (filename: string, pool: { unavailable: boolean }) {
   const acquire = yield* makeWorkerConnection({ filename, role: "reader", readOnly: true });
+  const scope = yield* Effect.scope;
+  const context = yield* Effect.context<never>();
+
+  const recover = (delay: Duration.Duration): Effect.Effect<void> =>
+    Effect.sleep(delay).pipe(
+      Effect.andThen(acquire),
+      Effect.matchEffect({
+        onSuccess: () =>
+          Effect.suspend(() => {
+            pool.unavailable = false;
+            return Effect.logInfo(
+              "SQLite reader worker available again; reads use the reader workers",
+            );
+          }),
+        onFailure: (error) => {
+          const { reason, permanent } = startFailure(error);
+          return permanent
+            ? Effect.logInfo(
+                "SQLite reader worker retries stopped permanently; reads use the writer",
+                { reason },
+              )
+            : recover(Duration.min(Duration.times(delay, 2), READER_RETRY_MAX_DELAY));
+        },
+      }),
+    );
+
   return Effect.suspend((): Effect.Effect<Connection | undefined> =>
     pool.unavailable
       ? Effect.undefined
@@ -577,10 +619,17 @@ const makeReader = Effect.fnUntraced(function* (filename: string, pool: { unavai
           Effect.catch((error) => {
             if (pool.unavailable) return Effect.undefined;
             pool.unavailable = true;
-            const cause = error.reason.cause;
+            const { reason, permanent } = startFailure(error);
+            // Pool state belongs to the client: never log or retry with the failed read's services.
             return Effect.logWarning("SQLite reader worker unavailable; reads use the writer", {
-              reason: cause instanceof Error ? cause.message : error.message,
-            }).pipe(Effect.as(undefined));
+              reason,
+            }).pipe(
+              Effect.andThen(
+                permanent ? Effect.void : Effect.forkIn(recover(READER_RETRY_DELAY), scope),
+              ),
+              Effect.updateContext((_: Context.Context<never>) => context),
+              Effect.as(undefined),
+            );
           }),
         ),
   );