diff --git a/apps/server/src/orchestration-v2/ShellStream.ts b/apps/server/src/orchestration-v2/ShellStream.ts index cec809f45ad4209bb7fa9c9e195a356e87315592..9730ca5016f683472d5c3502a569df804b54f4ac 100644 --- a/apps/server/src/orchestration-v2/ShellStream.ts +++ b/apps/server/src/orchestration-v2/ShellStream.ts @@ -12,6 +12,7 @@ import { OrchestrationProjectShell as ProjectShellSchema, OrchestrationV2ThreadShell as ThreadShellSchema, } from "@t3tools/contracts"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import * as Clock from "effect/Clock"; import * as Effect from "effect/Effect"; import * as Schema from "effect/Schema"; @@ -46,13 +47,16 @@ export const loadShellSnapshotParts = (input: { readonly latestSequence: Effect.Effect; }) => Effect.gen(function* () { - const read = yield* input.sql.withTransaction( - Effect.all({ - decodeThreads: input.readThreads, - projects: input.listProjects, - snapshotSequence: input.latestSequence, - }), - ); + // Read on the reader worker, off the event loop and the writer. + const read = yield* input.sql + .withTransaction( + Effect.all({ + decodeThreads: input.readThreads, + projects: input.listProjects, + snapshotSequence: input.latestSequence, + }), + ) + .pipe(NodeSqliteClient.readOnly); return { projects: read.projects, threads: yield* read.decodeThreads, diff --git a/apps/server/src/orchestration-v2/ThreadPullRequestService.ts b/apps/server/src/orchestration-v2/ThreadPullRequestService.ts index 71d884fc1819185f62dc56297a2018b0be22a5d3..68a3070a6b2ea21d30ae4e19957611d9e0136a2b 100644 --- a/apps/server/src/orchestration-v2/ThreadPullRequestService.ts +++ b/apps/server/src/orchestration-v2/ThreadPullRequestService.ts @@ -11,6 +11,7 @@ import { type ThreadLinkedPullRequest, } from "@t3tools/contracts"; import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as Crypto from "effect/Crypto"; @@ -127,10 +128,14 @@ export const make = Effect.gen(function* () { */ const readThreadSnapshot = ({ threadId, backfill }: RefreshRequest) => threadId === null - ? orchestrator.getShellSnapshot({ - location: "active", - unsettledOnly: !(backfill || pendingBackfill.size > 0), - }) + ? orchestrator + .getShellSnapshot({ + location: "active", + unsettledOnly: !(backfill || pendingBackfill.size > 0), + }) + // Reads every active thread once a minute: run it off the event loop. Writes keep + // landing during the read, and the snapshot's sequence guards the sync against them. + .pipe(NodeSqliteClient.readOnly) : Effect.gen(function* () { // Read the sequence first. The thread is then at least this new, so a // sync guarded by the sequence is rejected rather than missing a change. diff --git a/apps/server/src/persistence/Sqlite.ts b/apps/server/src/persistence/Sqlite.ts index d6adf41f3d957a702f306daf516b679fbbf62a5c..d3760db007c10a327587caaf5eb909b32b60609a 100644 --- a/apps/server/src/persistence/Sqlite.ts +++ b/apps/server/src/persistence/Sqlite.ts @@ -45,6 +45,7 @@ export const layerFromPath = Effect.fn("makeSqlitePersistenceLive")(function* (d layerSetup, NodeSqliteClient.layer({ filename: dbPath, + readerWorker: true, spanAttributes: { "db.name": path.basename(dbPath), "service.name": "t3code-server", diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index 83fe851ac8ad534f5399b153fe69dd7fb1439124..040baa7ba1304e7629a960b468729779d84a5b82 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -1,14 +1,67 @@ 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 Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +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 SqlClient from "effect/sql/SqlClient"; +import * as TestClock from "effect/testing/TestClock"; +import { vi } from "vite-plus/test"; import * as SqliteClient from "./nodeSqliteClient.ts"; +const startedWorkers = vi.hoisted((): Array => []); +// When on, workers run a script that never announces, like a worker stuck opening the file. +const silentWorkers = vi.hoisted(() => ({ on: false, readyOnTerminate: false })); + +vi.mock("node:worker_threads", async (importOriginal) => { + const actual = await importOriginal(); + class RecordedWorker extends actual.Worker { + constructor(...args: ConstructorParameters) { + super( + ...((silentWorkers.on + ? ["setInterval(() => {}, 1 << 30);", { eval: true }] + : args) as ConstructorParameters), + ); + startedWorkers.push(this); + } + override terminate() { + // A "ready" already queued when termination starts reaches the parent before "exit". + if (silentWorkers.readyOnTerminate) this.emit("message", { type: "ready" }); + return super.terminate(); + } + } + return { ...actual, Worker: RecordedWorker }; +}); + +const withReaderClient = ( + use: (sql: SqlClient.SqlClient) => Effect.Effect, +) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-reader-" }); + const filename = path.join(directory, "state.sqlite"); + return yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`PRAGMA journal_mode = WAL`; + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + yield* sql`INSERT INTO entries VALUES ('first')`; + return yield* use(sql); + }).pipe(Effect.provide(SqliteClient.layer({ filename, readerWorker: true }))); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)); + +const countEntries = (sql: SqlClient.SqlClient) => + Effect.map( + sql<{ count: number }>`SELECT COUNT(*) AS count FROM entries`, + (rows) => rows[0]?.count, + ); + const layer = it.layer(SqliteClient.layer({ filename: ":memory:" })); layer("NodeSqliteClient", (it) => { @@ -182,3 +235,318 @@ it.effect( }).pipe(Effect.provide(SqliteClient.layer({ filename }))); }).pipe(Effect.provide(NodeServices.layer)), ); + +it.effect("lets the writer commit while a read-only transaction is open", () => + withReaderClient((sql) => + Effect.gen(function* () { + const snapshotTaken = yield* Deferred.make(); + const writer = yield* Effect.forkChild( + Deferred.await(snapshotTaken).pipe( + Effect.andThen(sql`INSERT INTO entries VALUES ('second')`), + ), + ); + const counts = yield* sql + .withTransaction( + Effect.gen(function* () { + const before = yield* countEntries(sql); + yield* Deferred.succeed(snapshotTaken, undefined); + // Holding the writer lock here would deadlock this join. + yield* Fiber.join(writer); + const after = yield* countEntries(sql); + return { before, after }; + }), + ) + .pipe(SqliteClient.readOnly); + + // The reader keeps the snapshot it began with, and later reads see the commit. + assert.deepEqual(counts, { before: 1, after: 1 }); + assert.equal(yield* countEntries(sql).pipe(SqliteClient.readOnly), 2); + }), + ), +); + +it.effect("rejects a write in a read-only transaction", () => + withReaderClient((sql) => + Effect.gen(function* () { + const error = yield* sql + .withTransaction(sql`INSERT INTO entries VALUES ('blocked')`) + .pipe(SqliteClient.readOnly, Effect.flip); + + assert.equal(error._tag, "SqlError"); + assert.include(String(error.reason.cause), "readonly database"); + assert.equal(yield* countEntries(sql), 1); + }), + ), +); + +it.effect("keeps the reader read-only even if a statement turns query_only off", () => + withReaderClient((sql) => + Effect.gen(function* () { + const error = yield* sql + .withTransaction( + Effect.gen(function* () { + yield* sql`PRAGMA query_only = OFF`; + yield* sql`INSERT INTO entries VALUES ('blocked')`; + }), + ) + .pipe(SqliteClient.readOnly, Effect.flip); + + assert.equal(error._tag, "SqlError"); + assert.include(String(error.reason.cause), "readonly database"); + assert.equal(yield* countEntries(sql), 1); + }), + ), +); + +it.effect("keeps read-only work inside a write transaction on the writer", () => + withReaderClient((sql) => + sql.withTransaction( + Effect.gen(function* () { + yield* sql`INSERT INTO entries VALUES ('uncommitted')`; + // Only the writer can see its own uncommitted row. + const count = yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly); + assert.equal(count, 2); + }), + ), + ), +); + +it.effect("fails a read-only transaction whose reader stops, then starts a new reader", () => + withReaderClient((sql) => + Effect.gen(function* () { + const startedBefore = startedWorkers.length; + const error = yield* sql + .withTransaction( + Effect.gen(function* () { + yield* countEntries(sql); + yield* Effect.promise(() => startedWorkers.at(-1)!.terminate()); + return yield* countEntries(sql); + }), + ) + .pipe(SqliteClient.readOnly, Effect.flip); + + assert.equal(error._tag, "SqlError"); + assert.include(String(error.reason.cause), "Database reader exited"); + assert.equal(yield* countEntries(sql).pipe(SqliteClient.readOnly), 1); + assert.equal(startedWorkers.length, startedBefore + 2); + }), + ), +); + +it.effect("uses the writer for read-only work on an in-memory database", () => + Effect.gen(function* () { + const startedBefore = startedWorkers.length; + const sql = yield* SqlClient.SqlClient; + yield* sql`CREATE TABLE memory_entries(name TEXT NOT NULL)`; + yield* sql`INSERT INTO memory_entries VALUES ('kept')`; + + const rows = yield* sql + .withTransaction(sql<{ name: string }>`SELECT name FROM memory_entries`) + .pipe(SqliteClient.readOnly); + + assert.deepEqual(rows, [{ name: "kept" }]); + assert.equal(startedWorkers.length, startedBefore); + }).pipe(Effect.provide(SqliteClient.layer({ filename: ":memory:", readerWorker: true }))), +); + +it.effect("keeps a nested transaction inside a read-only transaction on the reader", () => + withReaderClient((sql) => + Effect.gen(function* () { + const snapshotTaken = yield* Deferred.make(); + const writer = yield* Effect.forkChild( + Deferred.await(snapshotTaken).pipe( + Effect.andThen(sql`INSERT INTO entries VALUES ('second')`), + ), + ); + // Callers wrap a store read that opens its own transaction, as the + // shell snapshot routes do around getShellSnapshot. + const counts = yield* sql + .withTransaction( + Effect.gen(function* () { + const outer = yield* countEntries(sql); + yield* Deferred.succeed(snapshotTaken, undefined); + yield* Fiber.join(writer); + const nested = yield* sql.withTransaction(countEntries(sql)); + return { outer, nested }; + }), + ) + .pipe(SqliteClient.readOnly); + + assert.deepEqual(counts, { outer: 1, nested: 1 }); + }), + ), +); + +const withFileClient = ( + setup: ( + sql: SqlClient.SqlClient, + filename: string, + ) => Effect.Effect, +) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-reader-off-" }); + const filename = path.join(directory, "state.sqlite"); + return yield* Effect.gen(function* () { + return yield* setup(yield* SqlClient.SqlClient, filename); + }).pipe(Effect.provide(SqliteClient.layer({ filename, readerWorker: true }))); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)); + +it.effect("uses the writer when the reader cannot open the database", () => + withFileClient((sql, filename) => + Effect.gen(function* () { + yield* sql`PRAGMA journal_mode = WAL`; + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + yield* sql`INSERT INTO entries VALUES ('kept')`; + // The writer keeps its open file; a read-only open of the path now fails. + yield* (yield* FileSystem.FileSystem).remove(filename); + const startedBefore = startedWorkers.length; + + assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1); + assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1); + assert.equal(startedWorkers.length, startedBefore + 1); + }), + ), +); + +it.effect("uses the writer for a database that is not in WAL mode", () => + withFileClient((sql) => + Effect.gen(function* () { + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + const startedBefore = startedWorkers.length; + + // A write inside readOnly only succeeds on the writer: the reader is read-only. + yield* sql + .withTransaction(sql`INSERT INTO entries VALUES ('on the writer')`) + .pipe(SqliteClient.readOnly); + assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1); + assert.equal(startedWorkers.length, startedBefore + 1); + }), + ), +); + +const captureWarnings = () => { + const warnings: Array = []; + const logger = Logger.make(({ logLevel, message }) => { + if (logLevel === "Warn") warnings.push(message); + }); + return { warnings, layer: Logger.layer([logger], { mergeWithExisting: false }) }; +}; + +const silently = (effect: Effect.Effect) => + Effect.acquireUseRelease( + Effect.sync(() => { + silentWorkers.on = true; + }), + () => effect, + () => + Effect.sync(() => { + silentWorkers.on = false; + silentWorkers.readyOnTerminate = false; + }), + ); + +it.effect("turns the reader off when its worker never opens the database", () => { + const { warnings, layer: logs } = captureWarnings(); + return withFileClient((sql) => + Effect.gen(function* () { + yield* sql`PRAGMA journal_mode = WAL`; + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + yield* sql`INSERT INTO entries VALUES ('kept')`; + const read = yield* silently( + Effect.gen(function* () { + const fiber = yield* Effect.forkChild( + sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), + ); + yield* Effect.yieldNow; + yield* TestClock.adjust("10 seconds"); + return yield* Fiber.join(fiber); + }), + ); + + assert.equal(read, 1); + assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1); + assert.lengthOf(warnings, 1); + }), + ).pipe(Effect.provide(logs)); +}); + +it.effect("lets a read be cancelled while its worker is still starting", () => + withFileClient((sql) => + Effect.gen(function* () { + yield* sql`PRAGMA journal_mode = WAL`; + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + yield* silently( + Effect.gen(function* () { + const fiber = yield* Effect.forkChild( + sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), + ); + yield* Effect.yieldNow; + // Never returns if the wait for the worker were uninterruptible: the + // test clock has not moved, so the startup deadline cannot fire. + yield* Fiber.interrupt(fiber); + }), + ); + + // The client stays usable: the next read waits out the deadline, then uses the writer. + const next = yield* Effect.forkChild( + sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), + ); + yield* Effect.yieldNow; + yield* TestClock.adjust("10 seconds"); + assert.equal(yield* Fiber.join(next), 0); + }), + ), +); + +it.effect("logs one warning when concurrent first reads find the reader unavailable", () => { + const { warnings, layer: logs } = captureWarnings(); + return withFileClient((sql, filename) => + Effect.gen(function* () { + yield* sql`PRAGMA journal_mode = WAL`; + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + yield* (yield* FileSystem.FileSystem).remove(filename); + const startedBefore = startedWorkers.length; + + const counts = yield* Effect.all( + [ + sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), + sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), + ], + { concurrency: "unbounded" }, + ); + + assert.deepEqual(counts, [0, 0]); + assert.equal(startedWorkers.length, startedBefore + 1); + assert.lengthOf(warnings, 1); + }), + ).pipe(Effect.provide(logs)); +}); + +it.effect("does not hand out a worker that reports ready after another read timed it out", () => + withFileClient((sql) => + Effect.gen(function* () { + yield* sql`PRAGMA journal_mode = WAL`; + yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`; + yield* sql`INSERT INTO entries VALUES ('kept')`; + const read = () => sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly); + const counts = yield* silently( + Effect.gen(function* () { + const first = yield* Effect.forkChild(read()); + yield* Effect.yieldNow; + yield* TestClock.adjust("5 seconds"); + const second = yield* Effect.forkChild(read()); + yield* Effect.yieldNow; + // The first read reaches its deadline and retires the worker; the second + // is still waiting when the worker's "ready" arrives, before it exits. + silentWorkers.readyOnTerminate = true; + yield* TestClock.adjust("5 seconds"); + return [yield* Fiber.join(first), yield* Fiber.join(second)]; + }), + ); + + assert.deepEqual(counts, [1, 1]); + }), + ), +); diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index 6b05d8ac6ca11c5285abad382a762c48ec4df47d..dd00bccd1e476342a63507b8ded590f60bb5c7b0 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -5,6 +5,7 @@ * @module SqliteClient */ import * as NodeSqlite from "node:sqlite"; +import * as NodeWorkerThreads from "node:worker_threads"; import * as Cache from "effect/Cache"; import * as Duration from "effect/Duration"; @@ -13,6 +14,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 Option from "effect/Option"; import * as Predicate from "effect/Predicate"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; @@ -36,8 +38,33 @@ export interface SqliteClientConfig { readonly spanAttributes?: Record | undefined; readonly transformResultNames?: ((str: string) => string) | undefined; readonly transformQueryNames?: ((str: string) => string) | undefined; + /** + * Run transactions marked with `readOnly` on a second connection in a worker + * thread, so long reads block neither the event loop nor writes. Ignored for + * in-memory databases, which a second connection cannot share. + */ + readonly readerWorker?: boolean | undefined; + /** How long a reader worker may take to open the database before reads use the writer. Default 10 seconds. */ + readonly readerStartupTimeout?: Duration.Input | undefined; } +/** + * Set by `readOnly`. A top-level transaction started while it is true runs on + * the reader worker when the client has one, and on the writer otherwise. + */ +const ReadOnly = Context.Reference("@t3tools/shared/nodeSqliteClient/ReadOnly", { + defaultValue: () => false, +}); + +/** + * Marks the SQL in `effect` as read-only. Wrap the whole transaction, as in + * `sql.withTransaction(reads).pipe(readOnly)`. Inside an open write + * transaction the reads keep using the writer, and a write attempted on the + * reader fails with "attempt to write a readonly database". + */ +export const readOnly = (effect: Effect.Effect): Effect.Effect => + Effect.provideService(effect, ReadOnly, true); + export class UnsupportedNodeSqliteVersionError extends Schema.TaggedError()( "UnsupportedNodeSqliteVersionError", { @@ -99,6 +126,306 @@ const classifyError = (cause: unknown, message: string, operation: string) => { return classifySqliteError(cause, { message, operation }); }; +// Evaluated source instead of a worker file, so the reader also starts from +// bundles and single-executable builds that have no module file on disk. +const readerWorkerSource = ` +const { parentPort, workerData } = require("node:worker_threads"); +const { DatabaseSync } = require("node:sqlite"); +let db; +try { + // A read-only open cannot be undone by SQL, unlike PRAGMA query_only. + const next = new DatabaseSync(workerData.filename, { + readOnly: true, + allowExtension: workerData.allowExtension, + }); + 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"); + db = next; + parentPort.postMessage({ type: "ready" }); +} catch (error) { + parentPort.postMessage({ type: "unavailable", message: String(error && error.message ? error.message : error) }); +} +const statements = new Map(); +const prepare = (sql, cached) => { + if (!cached) return db.prepare(sql); + let statement = statements.get(sql); + if (statement === undefined) { + statement = db.prepare(sql); + if (statements.size >= workerData.prepareCacheSize) statements.delete(statements.keys().next().value); + } else { + statements.delete(sql); + } + statements.set(sql, statement); + return statement; +}; +parentPort.on("message", ({ id, sql, params, cached, values, raw, safeIntegers }) => { + let operation = "open"; + try { + if (db === undefined) throw new Error("database reader is unavailable"); + operation = "prepare"; + const statement = prepare(sql, cached); + operation = "execute"; + statement.setReadBigInts(safeIntegers); + if (statement.columns().length > 0) { + statement.setReturnArrays(values); + parentPort.postMessage({ id, rows: statement.all(...params) }); + } else { + const result = statement.run(...params); + parentPort.postMessage({ id, rows: raw ? result : [] }); + } + } catch (error) { + parentPort.postMessage({ + id, + error: { + operation, + message: String(error && error.message ? error.message : error), + code: error && error.code, + errcode: error && error.errcode, + errstr: error && error.errstr, + }, + }); + } +}); +`; + +interface ReaderRequest { + readonly sql: string; + readonly params: ReadonlyArray; + readonly cached: boolean; + readonly values: boolean; + readonly raw: boolean; +} + +type ReaderStartup = + | { readonly type: "ready" } + | { readonly type: "unavailable"; readonly message: string }; + +type ReaderReply = + | { readonly id: number; readonly rows: ReadonlyArray } + | { + readonly id: number; + readonly error: { + readonly operation: "open" | "prepare" | "execute"; + readonly message: string; + readonly code?: unknown; + readonly errcode?: unknown; + readonly errstr?: unknown; + }; + }; + +const readerFailureMessages = { + open: "Failed to open database", + prepare: "Failed to prepare statement", + execute: "Failed to execute statement", +} as const; + +const failed = (cause: unknown, message: string, operation: string) => + new SqlError({ reason: classifyError(cause, message, operation) }); + +// COMMIT, ROLLBACK and ROLLBACK TO SAVEPOINT. On a reader they change nothing, +// so they succeed once the worker is gone instead of failing the cleanup. +const endsReadTransaction = (sql: string) => /^\s*ROLLBACK\b|^\s*COMMIT\b/i.test(sql); + +/** + * Read-only connections served by one worker thread at a time. Each worker + * gets its own `Connection`, so a transaction stays on the worker it began on: + * once that worker exits, its statements fail rather than silently continuing + * outside the transaction on a replacement. The next acquire starts a new one. + * A worker that cannot open the database read-only, or finds it outside WAL + * mode, turns the reader off for the life of the client: reads use the writer. + */ +const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) { + let current: + | { + readonly worker: NodeWorkerThreads.Worker; + readonly connection: Connection; + readonly started: Promise; + } + | undefined; + let unavailable = false; + let nextRequestId = 0; + + const start = () => { + const worker = new NodeWorkerThreads.Worker(readerWorkerSource, { + eval: true, + workerData: { + filename: options.filename, + allowExtension: options.allowExtension ?? false, + prepareCacheSize: options.prepareCacheSize ?? 200, + }, + }); + let announce: (startup: ReaderStartup) => void = () => {}; + const started = new Promise((resolve) => { + announce = resolve; + }); + let stoppedMessage: string | undefined; + const pending = new Map< + number, + { readonly endsTransaction: boolean; readonly resume: (reply: ReaderReply) => void } + >(); + const stop = (message: string) => { + stoppedMessage = message; + announce({ type: "unavailable", message }); + if (current?.worker === worker) current = undefined; + for (const [id, request] of pending) { + request.resume( + request.endsTransaction + ? { id, rows: [] } + : { id, error: { operation: "execute", message } }, + ); + } + pending.clear(); + }; + worker.on("message", (reply: ReaderReply | ReaderStartup) => { + if ("type" in reply) { + announce(reply); + // An idle reader must not keep the process alive. + if (pending.size === 0) worker.unref(); + return; + } + const request = pending.get(reply.id); + if (request === undefined) return; + pending.delete(reply.id); + if (pending.size === 0) worker.unref(); + request.resume(reply); + }); + worker.on("error", (cause) => stop(`Database reader failed: ${cause.message}`)); + worker.on("exit", (code) => stop(`Database reader exited with code ${code}`)); + + const request = (input: ReaderRequest) => + Effect.withFiber, SqlError>((fiber) => { + const endsTransaction = endsReadTransaction(input.sql); + if (stoppedMessage !== undefined) { + return endsTransaction + ? Effect.succeed([]) + : Effect.fail( + failed(new Error(stoppedMessage), readerFailureMessages.execute, "execute"), + ); + } + const safeIntegers = Boolean(Context.get(fiber.context, Client.SafeIntegers)); + return Effect.callback, SqlError>((resume) => { + const id = nextRequestId++; + pending.set(id, { + endsTransaction, + resume: (reply) => { + if ("rows" in reply) { + resume(Effect.succeed(reply.rows)); + return; + } + const { operation, message, ...codes } = reply.error; + resume( + Effect.fail( + failed( + Object.assign(new Error(message), codes), + readerFailureMessages[operation], + operation, + ), + ), + ); + }, + }); + if (pending.size === 1) worker.ref(); + // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin. + worker.postMessage({ id, safeIntegers, ...input }); + }); + }); + + const connection = identity({ + execute(sql, params, rowTransform) { + const effect = request({ sql, params, cached: true, values: false, raw: false }); + return rowTransform ? Effect.map(effect, rowTransform) : effect; + }, + executeRaw(sql, params) { + return request({ sql, params, cached: true, values: false, raw: true }); + }, + executeValues(sql, params) { + return request({ sql, params, cached: true, values: true, raw: false }); + }, + executeValuesUnprepared(sql, params) { + return request({ sql, params: params ?? [], cached: false, values: true, raw: false }); + }, + executeUnprepared(sql, params, rowTransform) { + // A read-only connection rejects BEGIN IMMEDIATE and BEGIN EXCLUSIVE, + // and a deferred BEGIN is all a read snapshot needs. + const statement = /^\s*BEGIN\b/i.test(sql) ? "BEGIN" : sql; + const effect = request({ + sql: statement, + params: params ?? [], + cached: false, + values: false, + raw: false, + }); + return rowTransform ? Effect.map(effect, rowTransform) : effect; + }, + executeStream(_sql, _params) { + return Stream.die(new UnsupportedNodeSqliteOperationError()); + }, + }); + current = { worker, connection, started }; + return current; + }; + + yield* Effect.addFinalizer(() => + Effect.promise(async () => { + await current?.worker.terminate(); + }), + ); + + const startupTimeout = Duration.fromInputUnsafe( + options.readerStartupTimeout ?? Duration.seconds(10), + ); + + // Runs once: concurrent first acquires share one startup and may all see it fail. + const turnOff = (reader: typeof current, reason: string) => + Effect.suspend(() => { + if (reader !== undefined) { + if (current === reader) current = undefined; + void reader.worker.terminate(); + } + if (unavailable) return Effect.succeed(undefined); + unavailable = true; + return Effect.logWarning("SQLite reader worker unavailable; reads use the main connection", { + reason, + }).pipe(Effect.as(undefined)); + }); + + /** The live worker's connection once it is ready, or undefined to use the writer. */ + return Effect.suspend((): Effect.Effect => { + if (unavailable) return Effect.succeed(undefined); + let reader: NonNullable; + try { + reader = current ?? start(); + } catch (cause) { + return turnOff(undefined, String(cause)); + } + // SqlClient acquires transaction connections uninterruptibly. Waiting for the + // worker happens before any lock is taken, so let a cancelled request leave. + return Effect.flatMap( + Effect.interruptible( + Effect.timeoutOption( + Effect.promise(() => reader.started), + startupTimeout, + ), + ), + (startup): Effect.Effect => + // Another acquire may have timed out and retired this worker while this one + // waited; a late "ready" must not hand out a worker that is being terminated. + unavailable || current !== reader + ? 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`, + ), + ); + }); +}); + const make = Effect.fn("makeWithDatabase")(function* ( options: SqliteClientConfig, ): Effect.fn.Return { @@ -261,16 +588,40 @@ const make = Effect.fn("makeWithDatabase")(function* ( const semaphore = yield* Semaphore.make(1); const connection = yield* makeConnection; + const acquireReader = + options.readerWorker && options.filename !== ":memory:" && options.filename !== "" + ? yield* makeReader(options) + : undefined; + const readerSemaphore = yield* Semaphore.make(1); - const acquirer = semaphore.withPermits(1)(Effect.succeed(connection)); - const transactionAcquirer = Effect.uninterruptibleMask((restore) => { - const fiber = Fiber.getCurrent()!; - const scope = Context.getUnsafe(fiber.context, Scope.Scope); - return Effect.as( - Effect.tap(restore(semaphore.take(1)), () => Scope.addFinalizer(scope, semaphore.release(1))), - connection, - ); - }); + const writer = { lock: semaphore, connection }; + // Picks the reader only for `readOnly` work, and the writer whenever the + // reader is turned off. + const route = Effect.withFiber((fiber) => + acquireReader !== undefined && Context.get(fiber.context, ReadOnly) + ? Effect.map(acquireReader, (readerConnection) => + readerConnection === undefined + ? writer + : { lock: readerSemaphore, connection: readerConnection }, + ) + : Effect.succeed(writer), + ); + + const acquirer = Effect.flatMap(route, (target) => + target.lock.withPermits(1)(Effect.succeed(target.connection)), + ); + const transactionAcquirer = Effect.flatMap(route, (target) => + Effect.uninterruptibleMask((restore) => { + const fiber = Fiber.getCurrent()!; + const scope = Context.getUnsafe(fiber.context, Scope.Scope); + return Effect.as( + Effect.tap(restore(target.lock.take(1)), () => + Scope.addFinalizer(scope, target.lock.release(1)), + ), + target.connection, + ); + }), + ); return yield* Client.make({ acquirer,