diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index aea1c956391df..5c7b086642a42 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/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 => []); +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 +42,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 +625,197 @@ 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"; + +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; diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index 5cc46c542fbf5..51e47c4bb5530 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -193,6 +193,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 +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(); + + const recover = (delay: Duration.Duration): Effect.Effect => + 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 => 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) => context), + Effect.as(undefined), + ); }), ), );