From e667d990c0bf4177ea30ed18fc6bf8196fe01f09 Mon Sep 17 00:00:00 2001 From: Navid Nadali Date: Wed, 4 Nov 2026 10:44:49 +0000 Subject: [PATCH 1/5] fix(shared): the SQLite reader pool comes back after a failed start When a reader worker failed to start, the client turned the reader pool off for the rest of the process, so every read ran on the writer connection until a restart. A failed start now turns the pool off with one warning, and a background retry (1 s, doubling to 60 s) brings it back. A database that is not in WAL mode keeps readers off without retrying. A gauge, t3_sqlite_reader_pool_available, shows the state. Co-Authored-By: Claude Opus 5.5 --- packages/shared/src/nodeSqliteClient.test.ts | 185 +++++++++++++++++++ packages/shared/src/nodeSqliteClient.ts | 168 ++++++++++++----- 2 files changed, 305 insertions(+), 48 deletions(-) diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index 32b6e8fdf158f..6f7266bc3a34b 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -3,12 +3,17 @@ import type * as NodeWorkerThreads from "node:worker_threads"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; +import * as Logger from "effect/Logger"; +import * as Metric from "effect/Metric"; import * as Path from "effect/Path"; +import * as Queue from "effect/Queue"; import * as SqlClient from "effect/sql/SqlClient"; import * as Statement from "effect/sql/Statement"; +import * as TestClock from "effect/testing/TestClock"; import { ChildProcess, ChildProcessSpawner } from "effect/process"; import * as Stream from "effect/Stream"; import type * as Tracer from "effect/Tracer"; @@ -17,6 +22,7 @@ import { vi } from "vite-plus/test"; import * as SqliteClient from "./nodeSqliteClient.ts"; const startedWorkers = vi.hoisted((): Array => []); +const workerExits = vi.hoisted(() => new WeakMap>()); const silentWorkers = vi.hoisted(() => ({ on: false, readyOnTerminate: false })); const failingWorkers = vi.hoisted(() => ({ message: undefined as string | undefined })); @@ -32,6 +38,7 @@ vi.mock("node:worker_threads", async (importOriginal) => { : args) as ConstructorParameters), ); startedWorkers.push(this); + workerExits.set(this, new Promise((resolve) => this.once("exit", () => resolve()))); } override terminate() { // A "ready" already queued when termination starts reaches the parent before "exit". @@ -408,6 +415,184 @@ it.effect("uses the writer when the reader cannot open the database", () => ), ); +const withLogLines = ( + use: (lines: Queue.Queue>) => Effect.Effect, +) => + Effect.gen(function* () { + const lines = yield* Queue.unbounded>(); + const logger = Logger.make(({ logLevel, message }) => { + Queue.offerUnsafe(lines, [logLevel, ...(message as ReadonlyArray)]); + }); + return yield* use(lines).pipe( + Effect.provide(Logger.layer([logger], { mergeWithExisting: false })), + ); + }); + +const readConnection = (sql: SqlClient.SqlClient) => + Effect.gen(function* () { + const spans: Array = []; + 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 = ( + mode: { readonly failWith?: string; readonly silent?: boolean }, + effect: Effect.Effect, +) => + 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 main connection"; + +const notWal = "database uses journal_mode delete, not wal"; + +const readerPoolAvailable = Effect.map(Metric.snapshot, (snapshots) => + snapshots.flatMap((snapshot) => + snapshot.id === "t3_sqlite_reader_pool_available" && snapshot.type === "Gauge" + ? [snapshot.state.value] + : [], + ), +); + +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: "worker 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* () { + // The failed read's own metric attributes must not label the pool's gauge. + 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); + }).pipe(Effect.provideService(Metric.CurrentMetricAttributes, { read: "failed" })), + ); + 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 }]); + assert.deepEqual(yield* readerPoolAvailable, [0]); + + 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* readerPoolAvailable, [1]); + assert.deepEqual(yield* readConnection(sql), { count: 1, connection: ["reader"] }); + assert.equal(Queue.sizeUnsafe(lines), 0); + }), + ), + ).pipe(Effect.provideService(Metric.MetricRegistry, new Map())), +); + +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([ + { + 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 main connection", + { 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); + assert.deepEqual(yield* readerPoolAvailable, [0]); + }), + ), + ).pipe(Effect.provideService(Metric.MetricRegistry, new Map())), +); + it.effect("finishes masked database cleanup after its caller is interrupted", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index 9531354db1939..17421b32d0234 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -11,6 +11,7 @@ import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import { identity } from "effect/Function"; import * as Layer from "effect/Layer"; +import * as Metric from "effect/Metric"; import * as Option from "effect/Option"; import * as Predicate from "effect/Predicate"; import * as References from "effect/References"; @@ -198,6 +199,7 @@ const readerWorkerSource = ` const { parentPort, workerData } = require("node:worker_threads"); const { DatabaseSync } = require("node:sqlite"); let db; +let permanent = false; try { // A read-only open cannot be undone by SQL, unlike PRAGMA query_only. const next = new DatabaseSync(workerData.filename, { @@ -207,11 +209,12 @@ try { next.exec("PRAGMA busy_timeout = 5000;"); // Without WAL a reader's shared lock blocks the writer's commits, so refuse. const mode = String(next.prepare("PRAGMA journal_mode").get().journal_mode).toLowerCase(); - if (mode !== "wal") throw new Error("database uses journal_mode " + mode + ", not wal"); + permanent = mode !== "wal"; + if (permanent) throw new Error("database uses journal_mode " + mode + ", not wal"); db = next; parentPort.postMessage({ type: "ready" }); } catch (error) { - parentPort.postMessage({ type: "unavailable", message: String(error && error.message ? error.message : error) }); + parentPort.postMessage({ type: "unavailable", message: String(error && error.message ? error.message : error), permanent }); } const statements = new Map(); const prepare = (sql, cached) => { @@ -266,7 +269,7 @@ interface ReaderRequest { type ReaderStartup = | { readonly type: "ready" } - | { readonly type: "unavailable"; readonly message: string }; + | { readonly type: "unavailable"; readonly message: string; readonly permanent?: boolean }; type ReaderReply = | { readonly id: number; readonly rows: ReadonlyArray } @@ -295,19 +298,29 @@ 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_RETRY_DELAY = Duration.seconds(1); +const READER_RETRY_MAX_DELAY = Duration.minutes(1); + +const readerPoolAvailable = Metric.gauge("t3_sqlite_reader_pool_available", { + description: "1 while SQLite reads can use reader workers, 0 while they use the main connection.", +}); // Bind transactions to one worker; fail after its exit instead of resuming outside the transaction on a replacement. const makeReader = Effect.fnUntraced(function* ( options: SqliteClientConfig, pool: { unavailable: boolean }, ) { - let current: - | { - readonly worker: NodeWorkerThreads.Worker; - readonly connection: Connection; - readonly started: Promise; - } - | undefined; + type Reader = { + readonly worker: NodeWorkerThreads.Worker; + readonly connection: Connection; + readonly started: Promise; + }; + type StartFailure = { + readonly reader?: Reader; + readonly reason: string; + readonly retry: boolean; + }; + let current: Reader | undefined; let nextRequestId = 0; const start = () => { @@ -323,6 +336,7 @@ const makeReader = Effect.fnUntraced(function* ( const started = new Promise((resolve) => { announce = resolve; }); + let ready = false; let stoppedMessage: string | undefined; const pending = new Map< number, @@ -331,7 +345,8 @@ const makeReader = Effect.fnUntraced(function* ( const stop = (message: string) => { stoppedMessage = message; announce({ type: "unavailable", message }); - if (current?.worker === worker) current = undefined; + // A worker that stops while starting stays in place until whoever awaits its startup retires it. + if (ready && current?.worker === worker) current = undefined; for (const [id, request] of pending) { request.resume( request.endsTransaction @@ -343,6 +358,7 @@ const makeReader = Effect.fnUntraced(function* ( }; worker.on("message", (reply: ReaderReply | ReaderStartup) => { if ("type" in reply) { + ready = reply.type === "ready"; announce(reply); if (pending.size === 0) worker.unref(); return; @@ -435,54 +451,109 @@ const makeReader = Effect.fnUntraced(function* ( }), ); + const scope = yield* Effect.scope; + const context = yield* Effect.context(); const startupTimeout = Duration.seconds(10); - const turnOff = (reader: typeof current, reason: string) => + const open = Effect.suspend((): Effect.Effect => { + let reader: Reader; + try { + reader = current ?? start(); + } catch (cause) { + return Effect.fail({ reason: String(cause), retry: true }); + } + return Effect.flatMap( + Effect.timeoutOption( + Effect.promise(() => reader.started), + startupTimeout, + ), + (startup) => + Option.isNone(startup) + ? Effect.fail({ + reader, + reason: `worker did not open the database within ${Duration.toMillis(startupTimeout)} ms`, + retry: true, + }) + : startup.value.type === "ready" + ? Effect.succeed(reader) + : Effect.fail({ + reader, + reason: startup.value.message, + retry: startup.value.permanent !== true, + }), + ); + }); + + // Only the slot's current reader may change the pool; a reader retired earlier was already reported. + const retire = (reader: Reader | undefined) => { + if (reader === undefined) return true; + void reader.worker.terminate(); + if (current !== reader) return false; + current = undefined; + return true; + }; + + const recover = (delay: Duration.Duration): Effect.Effect => + Effect.sleep(delay).pipe( + Effect.andThen(open), + Effect.matchEffect({ + onSuccess: () => + Effect.suspend(() => { + pool.unavailable = false; + return Metric.update(readerPoolAvailable, 1).pipe( + Effect.andThen( + Effect.logInfo( + "SQLite reader worker available again; reads use the reader workers", + ), + ), + ); + }), + onFailure: (failure) => + failure.retry + ? Effect.andThen( + Effect.sync(() => retire(failure.reader)), + recover(Duration.min(Duration.times(delay, 2), READER_RETRY_MAX_DELAY)), + ) + : Effect.andThen( + Effect.logInfo( + "SQLite reader worker retries stopped permanently; reads use the main connection", + { reason: failure.reason }, + ), + Effect.sync(() => retire(failure.reader)), + ), + }), + ); + + const turnOff = (failure: StartFailure) => Effect.suspend(() => { - if (reader !== undefined) { - if (current === reader) current = undefined; - void reader.worker.terminate(); - } // @effect-diagnostics-next-line effectSucceedWithVoid:off - The connection result distinguishes undefined from void. - if (pool.unavailable) return Effect.succeed(undefined); + if (!retire(failure.reader) || pool.unavailable) return Effect.succeed(undefined); pool.unavailable = true; + // Pool state belongs to the client: never log, meter or retry with the failed read's services. return Effect.logWarning("SQLite reader worker unavailable; reads use the main connection", { - reason, - }).pipe(Effect.as(undefined)); + reason: failure.reason, + }).pipe( + Effect.andThen(Metric.update(readerPoolAvailable, 0)), + Effect.andThen( + failure.retry ? Effect.forkIn(recover(READER_RETRY_DELAY), scope) : Effect.void, + ), + Effect.updateContext((_: Context.Context) => context), + Effect.as(undefined), + ); }); return Effect.suspend((): Effect.Effect => { // @effect-diagnostics-next-line effectSucceedWithVoid:off - The connection result distinguishes undefined from void. if (pool.unavailable) return Effect.succeed(undefined); - let reader: NonNullable; - try { - reader = current ?? start(); - } catch (cause) { - return turnOff(undefined, String(cause)); - } - return Effect.flatMap( - Effect.withFiber((fiber) => { - const restore = Context.get(fiber.context, ReadOnly) ?? identity; - return restore( - Effect.timeoutOption( - Effect.promise(() => reader.started), - startupTimeout, - ), - ); + return Effect.withFiber((fiber) => + (Context.get(fiber.context, ReadOnly) ?? identity)(open), + ).pipe( + Effect.matchEffect({ + onFailure: turnOff, + onSuccess: (reader) => + // A late ready event must not hand out a worker retired by another acquire timeout. + Effect.succeed(pool.unavailable || current !== reader ? undefined : reader.connection), }), - (startup): Effect.Effect => - // A late ready event must not hand out a worker retired by another acquire timeout. - pool.unavailable || current !== reader - ? // @effect-diagnostics-next-line effectSucceedWithVoid:off - The connection result distinguishes undefined from void. - Effect.succeed(undefined) - : Option.isSome(startup) && startup.value.type === "ready" - ? Effect.succeed(reader.connection) - : turnOff( - reader, - Option.isSome(startup) && startup.value.type === "unavailable" - ? startup.value.message - : `worker did not open the database within ${Duration.toMillis(startupTimeout)} ms`, - ), ); }); }); @@ -1080,7 +1151,7 @@ const make = Effect.fn("makeWithDatabase")(function* ( const semaphore = yield* Semaphore.make(1); const connection = yield* makeConnection; - // A reader that cannot start turns off the whole pool, so routing never picks a dead slot. + // A reader that cannot start turns off the whole pool until a retry starts it, so routing never picks a dead slot. const readerPool = { unavailable: false }; const readers = options.readerWorker && isFileDatabase @@ -1095,6 +1166,7 @@ const make = Effect.fn("makeWithDatabase")(function* ( ), ) : []; + if (readers.length > 0) yield* Metric.update(readerPoolAvailable, 1); const writer = { name: "writer" as const, lock: semaphore, connection }; let activeWriter: ConnectionOwner | undefined; From db37135e0710d1bab482199d30f097983f1d1494 Mon Sep 17 00:00:00 2001 From: Navid Nadali Date: Wed, 4 Nov 2026 16:56:33 +0000 Subject: [PATCH 2/5] test(shared): the SQLite reader retry stops when the client closes Closing the client while the reader retry waits, or while a retried reader is starting, leaves no reader starting afterwards. Co-Authored-By: Claude Opus 5.5 --- packages/shared/src/nodeSqliteClient.test.ts | 25 ++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index 6f7266bc3a34b..2436635b1511e 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -545,6 +545,31 @@ it.effect("backs off a reader that never starts without logging each retry", () ), ); +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", From 429d7af3c75456d79933abbabaf06c2979d24000 Mon Sep 17 00:00:00 2001 From: Navid Nadali Date: Sat, 14 Nov 2026 19:15:55 +0000 Subject: [PATCH 3/5] test(shared): the SQLite reader pool should come back after a failed start A reader that fails to start turns the pool off for the rest of the process. These tests expect a background retry with backoff to bring it back, a gauge that shows the pool's state, no retries for a database that is not in WAL mode, and no retries after the client closes. They fail on main. Co-Authored-By: Claude Opus 5.5 --- packages/shared/src/nodeSqliteClient.test.ts | 209 +++++++++++++++++++ 1 file changed, 209 insertions(+) diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index aea1c956391df..9cae602f13833 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -4,12 +4,16 @@ 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 Metric from "effect/Metric"; 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 +26,7 @@ import { vi } from "vite-plus/test"; import * as SqliteClient from "./nodeSqliteClient.ts"; const startedWorkers = vi.hoisted((): Array => []); +const workerExits = vi.hoisted(() => new WeakMap>()); const silentWorkers = vi.hoisted(() => ({ on: false, readyOnTerminate: false })); const failingWorkers = vi.hoisted(() => ({ message: undefined as string | undefined })); const checkpointWorkers = vi.hoisted((): Array => []); @@ -38,6 +43,7 @@ vi.mock("node:worker_threads", async (importOriginal) => { : args) as ConstructorParameters), ); 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 +626,209 @@ it.effect("uses the writer when the reader cannot open the database", () => ), ); +const withLogLines = ( + use: (lines: Queue.Queue>) => Effect.Effect, +) => + Effect.gen(function* () { + const lines = yield* Queue.unbounded>(); + const logger = Logger.make(({ logLevel, message }) => { + Queue.offerUnsafe(lines, [logLevel, ...(message as ReadonlyArray)]); + }); + return yield* use(lines).pipe( + Effect.provide(Logger.layer([logger], { mergeWithExisting: false })), + ); + }); + +const readConnection = (sql: SqlClient.SqlClient) => + Effect.gen(function* () { + const spans: Array = []; + 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 = ( + mode: { readonly failWith?: string; readonly silent?: boolean }, + effect: Effect.Effect, +) => + 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"; + +const readerPoolAvailable = Effect.map(Metric.snapshot, (snapshots) => + snapshots.flatMap((snapshot) => + snapshot.id === "t3_sqlite_reader_pool_available" && snapshot.type === "Gauge" + ? [snapshot.state.value] + : [], + ), +); + +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* () { + // The failed read's own metric attributes must not label the pool's gauge. + 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); + }).pipe(Effect.provideService(Metric.CurrentMetricAttributes, { read: "failed" })), + ); + 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 }]); + assert.deepEqual(yield* readerPoolAvailable, [0]); + + 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* readerPoolAvailable, [1]); + assert.deepEqual(yield* readConnection(sql), { count: 1, connection: ["reader"] }); + assert.equal(Queue.sizeUnsafe(lines), 0); + }), + ), + ).pipe(Effect.provideService(Metric.MetricRegistry, new Map())), +); + +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); + assert.deepEqual(yield* readerPoolAvailable, [0]); + }), + ), + ).pipe(Effect.provideService(Metric.MetricRegistry, new Map())), +); + it.effect("fails with a typed SqlError when the database file is not SQLite", () => Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; From 9269085990d4fa5a4ef21f6515319f96885f6d86 Mon Sep 17 00:00:00 2001 From: Navid Nadali Date: Sat, 14 Nov 2026 19:18:53 +0000 Subject: [PATCH 4/5] fix(shared): the SQLite reader pool comes back after a failed start When a reader worker failed to start, the reader pool turned off for the rest of the process, so every read ran on the writer until a restart. A failed start now turns the pool off with one warning, and one background retry (1 s, doubling to 60 s) brings it back, with one log line. A database that is not in WAL mode keeps readers off without retrying. The retry belongs to the client and stops when it closes. A gauge, t3_sqlite_reader_pool_available, shows the state. The writer is unchanged. Co-Authored-By: Claude Opus 5.5 --- packages/shared/src/nodeSqliteClient.test.ts | 1 + packages/shared/src/nodeSqliteClient.ts | 70 ++++++++++++++++++-- 2 files changed, 66 insertions(+), 5 deletions(-) diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index 9cae602f13833..30d8135e1c2c9 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -703,6 +703,7 @@ it.effect.each([ withLogLines((lines) => withReaderClient((sql) => Effect.gen(function* () { + assert.deepEqual(yield* readerPoolAvailable, [1]); // The failed read's own metric attributes must not label the pool's gauge. const failed = yield* withWorkers( workers, diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index 5cc46c542fbf5..d27a39558c8a8 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -11,6 +11,7 @@ import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import { identity } from "effect/Function"; import * as Layer from "effect/Layer"; +import * as Metric from "effect/Metric"; import * as Option from "effect/Option"; import * as Predicate from "effect/Predicate"; import * as References from "effect/References"; @@ -193,6 +194,9 @@ const hasLargeBinding = (params: ReadonlyArray): 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 +218,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 +351,12 @@ 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); + +const readerPoolAvailable = Metric.gauge("t3_sqlite_reader_pool_available", { + description: "1 while SQLite reads can use reader workers, 0 while they use the writer.", +}); // 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 +578,48 @@ 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(); + + const recover = (delay: Duration.Duration): Effect.Effect => + Effect.sleep(delay).pipe( + Effect.andThen(acquire), + Effect.matchEffect({ + onSuccess: () => + Effect.suspend(() => { + pool.unavailable = false; + return Metric.update(readerPoolAvailable, 1).pipe( + Effect.andThen( + 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 => pool.unavailable ? Effect.undefined @@ -577,10 +628,18 @@ 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, meter 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(Metric.update(readerPoolAvailable, 0)), + Effect.andThen( + permanent ? Effect.void : Effect.forkIn(recover(READER_RETRY_DELAY), scope), + ), + Effect.updateContext((_: Context.Context) => context), + Effect.as(undefined), + ); }), ), ); @@ -880,6 +939,7 @@ const make = Effect.fn("makeWithDatabase")(function* ( ), ) : []; + if (readers.length > 0) yield* Metric.update(readerPoolAvailable, 1); // One permit per slot, so a queued read waits for any reader, not one it picked while all were busy. const readerAdmission = yield* Semaphore.make(readers.length); From c9dcd47e49fca1e7cbac980a4e3c0f27079d668b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 23:47:29 +0000 Subject: [PATCH 5/5] refactor(shared): drop the reader pool gauge Every reader-pool client shares one unlabelled gauge, so the OpenCode usage reader overwrites the state database's value. The warn and info log lines already report when the pool goes down and comes back. Co-Authored-By: Claude Opus 5.5 --- packages/shared/src/nodeSqliteClient.test.ts | 20 +++----------------- packages/shared/src/nodeSqliteClient.ts | 17 +++-------------- 2 files changed, 6 insertions(+), 31 deletions(-) diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index 30d8135e1c2c9..5c7b086642a42 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -11,7 +11,6 @@ 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 Metric from "effect/Metric"; import * as Path from "effect/Path"; import * as Queue from "effect/Queue"; import * as SqlClient from "effect/sql/SqlClient"; @@ -676,14 +675,6 @@ const unavailableWarning = "SQLite reader worker unavailable; reads use the writ const notWal = "database uses journal_mode delete, not wal"; -const readerPoolAvailable = Effect.map(Metric.snapshot, (snapshots) => - snapshots.flatMap((snapshot) => - snapshot.id === "t3_sqlite_reader_pool_available" && snapshot.type === "Gauge" - ? [snapshot.state.value] - : [], - ), -); - it.effect.each([ { failure: "a crash", @@ -703,8 +694,6 @@ it.effect.each([ withLogLines((lines) => withReaderClient((sql) => Effect.gen(function* () { - assert.deepEqual(yield* readerPoolAvailable, [1]); - // The failed read's own metric attributes must not label the pool's gauge. const failed = yield* withWorkers( workers, Effect.gen(function* () { @@ -713,14 +702,13 @@ it.effect.each([ }); yield* TestClock.adjust(startup); return yield* Fiber.join(read); - }).pipe(Effect.provideService(Metric.CurrentMetricAttributes, { read: "failed" })), + }), ); 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 }]); - assert.deepEqual(yield* readerPoolAvailable, [0]); yield* TestClock.adjust("1 second"); assert.equal(startedWorkers.length, startedBefore + 1); @@ -728,12 +716,11 @@ it.effect.each([ "Info", "SQLite reader worker available again; reads use the reader workers", ]); - assert.deepEqual(yield* readerPoolAvailable, [1]); assert.deepEqual(yield* readConnection(sql), { count: 1, connection: ["reader"] }); assert.equal(Queue.sizeUnsafe(lines), 0); }), ), - ).pipe(Effect.provideService(Metric.MetricRegistry, new Map())), + ), ); it.effect("backs off a reader that never starts without logging each retry", () => @@ -824,10 +811,9 @@ it.effect.each([ assert.deepEqual(yield* readConnection(sql), { count: 1, connection: ["writer"] }); assert.equal(startedWorkers.length, startedBefore + starts); assert.deepEqual(yield* Queue.clear(lines), logged); - assert.deepEqual(yield* readerPoolAvailable, [0]); }), ), - ).pipe(Effect.provideService(Metric.MetricRegistry, new Map())), + ), ); it.effect("fails with a typed SqlError when the database file is not SQLite", () => diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index d27a39558c8a8..51e47c4bb5530 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -11,7 +11,6 @@ import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import { identity } from "effect/Function"; import * as Layer from "effect/Layer"; -import * as Metric from "effect/Metric"; import * as Option from "effect/Option"; import * as Predicate from "effect/Predicate"; import * as References from "effect/References"; @@ -354,10 +353,6 @@ const READER_STARTUP_TIMEOUT = Duration.seconds(10); const READER_RETRY_DELAY = Duration.seconds(1); const READER_RETRY_MAX_DELAY = Duration.minutes(1); -const readerPoolAvailable = Metric.gauge("t3_sqlite_reader_pool_available", { - description: "1 while SQLite reads can use reader workers, 0 while they use the writer.", -}); - // Bind transactions to one worker; fail after its exit instead of resuming outside the transaction on a replacement. const makeWorkerConnection = Effect.fnUntraced(function* (options: { readonly filename: string; @@ -600,12 +595,8 @@ const makeReader = Effect.fnUntraced(function* (filename: string, pool: { unavai onSuccess: () => Effect.suspend(() => { pool.unavailable = false; - return Metric.update(readerPoolAvailable, 1).pipe( - Effect.andThen( - Effect.logInfo( - "SQLite reader worker available again; reads use the reader workers", - ), - ), + return Effect.logInfo( + "SQLite reader worker available again; reads use the reader workers", ); }), onFailure: (error) => { @@ -629,11 +620,10 @@ const makeReader = Effect.fnUntraced(function* (filename: string, pool: { unavai if (pool.unavailable) return Effect.undefined; pool.unavailable = true; const { reason, permanent } = startFailure(error); - // Pool state belongs to the client: never log, meter or retry with the failed read's services. + // 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, }).pipe( - Effect.andThen(Metric.update(readerPoolAvailable, 0)), Effect.andThen( permanent ? Effect.void : Effect.forkIn(recover(READER_RETRY_DELAY), scope), ), @@ -939,7 +929,6 @@ const make = Effect.fn("makeWithDatabase")(function* ( ), ) : []; - if (readers.length > 0) yield* Metric.update(readerPoolAvailable, 1); // One permit per slot, so a queued read waits for any reader, not one it picked while all were busy. const readerAdmission = yield* Semaphore.make(readers.length);