diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index acf3cc27f6211..0e307efbdcfd2 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -1,4 +1,3 @@ -import * as NodeSqlite from "node:sqlite"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import { @@ -191,47 +190,44 @@ it.effect("discovers on the reader worker and deletes on the writer", () => const path = yield* Path.Path; const filename = path.join(yield* fs.makeTempDirectoryScoped(), "fixture.sqlite"); yield* Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; yield* seedCompaction(); - // Establish the reader before recording main-thread statements. - yield* sql`SELECT 1`.pipe(NodeSqliteClient.readOnly); - const reads: string[] = []; - const writes: string[] = []; - const all = NodeSqlite.StatementSync.prototype.all; - const run = NodeSqlite.StatementSync.prototype.run; - yield* Effect.acquireRelease( - Effect.sync(() => { - const readSpy = vi - .spyOn(NodeSqlite.StatementSync.prototype, "all") - .mockImplementation(function (this: NodeSqlite.StatementSync, ...params) { - reads.push(this.sourceSQL); - return all.apply(this, params); - }); - const writeSpy = vi - .spyOn(NodeSqlite.StatementSync.prototype, "run") - .mockImplementation(function (this: NodeSqlite.StatementSync, ...params) { - writes.push(this.sourceSQL); - return run.apply(this, params); - }); - return { readSpy, writeSpy }; - }), - ({ readSpy, writeSpy }) => - Effect.sync(() => { - readSpy.mockRestore(); - writeSpy.mockRestore(); - }), - ); + const statements: Array<{ readonly query: unknown; readonly connection: unknown }> = []; + const tracer = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan(options); + const end = span.end.bind(span); + span.end = (time, exit) => { + end(time, exit); + const query = span.attributes.get("db.query.text"); + if (query === undefined) return; + statements.push({ query, connection: span.attributes.get("db.connection") }); + }; + return span; + }, + }); const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; - const result = yield* maintenance.pruneEventStore(); + const result = yield* maintenance + .pruneEventStore() + .pipe( + Effect.withTracer(tracer), + Effect.provideService(Statement.SpanPropagationEnabled, true), + ); assert.equal(result.deletedEventCount, 150); assert.equal(result.deletedReceiptCount, 100); // Only the in-transaction check for rows appended during the pass reads on the writer. assert.deepEqual( - reads.filter((query) => /\bSELECT\b/.test(query) && !query.includes("AS prompt_sequence")), - [], + new Set( + statements + .filter( + ({ query }) => + typeof query === "string" && + /^\s*(SELECT|WITH)\b/.test(query) && + !query.includes("AS prompt_sequence"), + ) + .map(({ connection }) => connection), + ), + new Set(["reader"]), ); - assert(writes.some((query) => query.includes("DELETE FROM orchestration_events"))); - assert(writes.some((query) => query.includes("DELETE FROM orchestration_command_receipts"))); }).pipe( Effect.provide( ProjectionMaintenance.layer.pipe( diff --git a/apps/server/src/persistence/Sqlite.test.ts b/apps/server/src/persistence/Sqlite.test.ts index a2d29f12f1810..86ec0214aae90 100644 --- a/apps/server/src/persistence/Sqlite.test.ts +++ b/apps/server/src/persistence/Sqlite.test.ts @@ -3,7 +3,6 @@ import * as NodeFS from "node:fs"; import * as NodeOS from "node:os"; import * as NodePath from "node:path"; import * as NodeSqlite from "node:sqlite"; -import type * as NodeWorkerThreads from "node:worker_threads"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; @@ -11,23 +10,9 @@ import * as HostProcess from "@t3tools/shared/HostProcess"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as SqlClient from "effect/sql/SqlClient"; -import { vi } from "vite-plus/test"; import * as SqlitePersistence from "./Sqlite.ts"; -const workers = vi.hoisted((): Array => []); - -vi.mock("node:worker_threads", async (importOriginal) => { - const actual = await importOriginal(); - class RecordedWorker extends actual.Worker { - constructor(...args: ConstructorParameters) { - super(...args); - workers.push(this); - } - } - return { ...actual, Worker: RecordedWorker }; -}); - it.effect("keeps writes moving past a 48 MiB WAL while a reader retains its snapshot", () => { const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-sqlite-burst-")); const dbPath = NodePath.join(tempDir, "state.sqlite"); @@ -37,7 +22,6 @@ it.effect("keeps writes moving past a 48 MiB WAL while a reader retains its snap return Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; - const checkpointWorker = workers.at(-1)!; yield* sql`CREATE TABLE burst_probe(id INTEGER PRIMARY KEY, payload BLOB)`; yield* sql`INSERT INTO burst_probe VALUES (1, NULL)`; yield* sql`PRAGMA wal_checkpoint(TRUNCATE)`; @@ -51,43 +35,6 @@ it.effect("keeps writes moving past a 48 MiB WAL while a reader retains its snap yield* sql.withTransaction( sql`INSERT INTO burst_probe VALUES (2, zeroblob(${48 * 1024 * 1024}))`, ); - const state = yield* Effect.callback<{ - readonly log: number; - readonly checkpointed: number; - readonly blocked: boolean; - }>((resume) => { - const onMessage = (message: { - readonly type?: string; - readonly log?: number; - readonly checkpointed?: number; - readonly blocked?: boolean; - }) => { - if ( - message.type === "checkpoint" && - typeof message.log === "number" && - message.log > 10_000 && - typeof message.checkpointed === "number" && - typeof message.blocked === "boolean" - ) { - checkpointWorker.off("message", onMessage); - resume( - Effect.succeed({ - log: message.log, - checkpointed: message.checkpointed, - blocked: message.blocked, - }), - ); - } - }; - checkpointWorker.on("message", onMessage); - checkpointWorker.ref(); - // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin. - checkpointWorker.postMessage("barrier"); - return Effect.sync(() => checkpointWorker.off("message", onMessage)); - }); - assert.isBelow(state.checkpointed, state.log); - assert.isFalse(state.blocked); - yield* sql.withTransaction(sql`INSERT INTO burst_probe VALUES (3, NULL)`); assert.equal(reader.prepare("SELECT COUNT(*) AS count FROM burst_probe").get()?.count, 1); reader.exec("COMMIT"); diff --git a/apps/server/src/persistence/Sqlite.ts b/apps/server/src/persistence/Sqlite.ts index 38913376cd123..3b2962c891a21 100644 --- a/apps/server/src/persistence/Sqlite.ts +++ b/apps/server/src/persistence/Sqlite.ts @@ -54,7 +54,9 @@ export const layerFromPath = Effect.fn("makeSqlitePersistenceLive")(function* (d NodeSqliteClient.layer({ filename: dbPath, readerWorker: true, - // Run checkpoint fsyncs off the event loop so slow disk flushes cannot stall clients. + // Run writes off the event loop so slow statements and transactions cannot stall clients. + writerWorker: true, + // Bounds WAL growth while readers pin snapshots, in the writer worker or on a main-thread fallback. checkpointWorker: true, spanAttributes: { "db.name": path.basename(dbPath), diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index 32b6e8fdf158f..423e127fce4df 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -168,30 +168,32 @@ const makeTempDatabase = Effect.gen(function* () { return { filename, other }; }); -it.effect("keeps another connection from committing between a transaction's read and write", () => - Effect.gen(function* () { - const { filename, other } = yield* makeTempDatabase; - yield* Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - yield* sql.withTransaction( - Effect.gen(function* () { - const [row] = yield* sql<{ readonly value: number }>` +it.effect.each([{ writerWorker: false }, { writerWorker: true }])( + "keeps another connection from committing between a transaction's read and write (writerWorker=$writerWorker)", + ({ writerWorker }) => + Effect.gen(function* () { + const { filename, other } = yield* makeTempDatabase; + yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql.withTransaction( + Effect.gen(function* () { + const [row] = yield* sql<{ readonly value: number }>` SELECT value FROM counters WHERE id = 1 `; - // Deferred BEGIN permits a competing commit, causing SQLITE_BUSY_SNAPSHOT that no timeout can resolve. - yield* Effect.sync(() => - assert.throws( - () => other.exec("UPDATE counters SET value = value + 1 WHERE id = 1"), - /database is locked/, - ), - ); - yield* sql`UPDATE counters SET value = ${(row?.value ?? 0) + 10} WHERE id = 1`; - }), - ); - yield* Effect.sync(() => other.exec("UPDATE counters SET value = value + 1 WHERE id = 1")); - assert.deepEqual(yield* sql`SELECT value FROM counters WHERE id = 1`.values, [[11]]); - }).pipe(Effect.provide(SqliteClient.layer({ filename }))); - }).pipe(Effect.provide(NodeServices.layer)), + // Deferred BEGIN permits a competing commit, causing SQLITE_BUSY_SNAPSHOT that no timeout can resolve. + yield* Effect.sync(() => + assert.throws( + () => other.exec("UPDATE counters SET value = value + 1 WHERE id = 1"), + /database is locked/, + ), + ); + yield* sql`UPDATE counters SET value = ${(row?.value ?? 0) + 10} WHERE id = 1`; + }), + ); + yield* Effect.sync(() => other.exec("UPDATE counters SET value = value + 1 WHERE id = 1")); + assert.deepEqual(yield* sql`SELECT value FROM counters WHERE id = 1`.values, [[11]]); + }).pipe(Effect.provide(SqliteClient.layer({ filename, writerWorker }))); + }).pipe(Effect.provide(NodeServices.layer)), ); it.effect("waits for another process's lock while the database opens", () => @@ -222,21 +224,75 @@ it.effect("waits for another process's lock while the database opens", () => }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), ); -it.effect("reports a transaction blocked by another writer as a lock timeout", () => +it.effect.each([{ writerWorker: false }, { writerWorker: true }])( + "reports a transaction blocked by another writer as a lock timeout (writerWorker=$writerWorker)", + ({ writerWorker }) => + Effect.gen(function* () { + const { filename, other } = yield* makeTempDatabase; + yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`PRAGMA busy_timeout = 0`; + const read = sql.withTransaction(sql`SELECT value FROM counters WHERE id = 1`.values); + + yield* Effect.sync(() => other.exec("BEGIN IMMEDIATE")); + const error = yield* read.pipe(Effect.flip); + assert.equal(error.reason._tag, "LockTimeoutError"); + + yield* Effect.sync(() => other.exec("COMMIT")); + assert.deepEqual(yield* read, [[0]]); + }).pipe(Effect.provide(SqliteClient.layer({ filename, writerWorker }))); + }).pipe(Effect.provide(NodeServices.layer)), +); + +it.effect("keeps the event loop free while a write waits on another connection's lock", () => Effect.gen(function* () { const { filename, other } = yield* makeTempDatabase; yield* Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; - yield* sql`PRAGMA busy_timeout = 0`; - const read = sql.withTransaction(sql`SELECT value FROM counters WHERE id = 1`.values); - yield* Effect.sync(() => other.exec("BEGIN IMMEDIATE")); - const error = yield* read.pipe(Effect.flip); - assert.equal(error.reason._tag, "LockTimeoutError"); + const write = yield* Effect.forkChild( + sql`UPDATE counters SET value = value + 1 WHERE id = 1`, + { startImmediately: true }, + ); + // A writer on this thread would spin in busy_timeout, so the holder could never commit. + yield* Effect.sync(() => other.exec("UPDATE counters SET value = 10 WHERE id = 1; COMMIT")); + yield* Fiber.join(write); + assert.deepEqual(yield* sql`SELECT value FROM counters WHERE id = 1`.values, [[11]]); + }).pipe(Effect.provide(SqliteClient.layer({ filename, writerWorker: true }))); + }).pipe(Effect.provide(NodeServices.layer)), +); +it.effect("holds writer-worker writes past the WAL size limit until a pinned reader releases", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const { filename, other } = yield* makeTempDatabase; + yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const limit = 2 * 1024 * 1024; + yield* sql`CREATE TABLE IF NOT EXISTS blobs(payload BLOB)`; + yield* sql.unsafe(`PRAGMA journal_size_limit = ${limit}`); + yield* sql`PRAGMA wal_checkpoint(TRUNCATE)`; + const blob = sql`INSERT INTO blobs VALUES (zeroblob(${1536 * 1024}))`; + yield* Effect.sync(() => { + other.exec("BEGIN"); + other.prepare("SELECT COUNT(*) FROM counters").get(); + }); + yield* sql.withTransaction(blob); + yield* sql.withTransaction(blob); + // Held writes, mutating pragmas included, can be cancelled before they run. + for (const cancelled of [blob, sql`PRAGMA user_version = 7`]) { + yield* Fiber.interrupt(yield* Effect.forkChild(cancelled, { startImmediately: true })); + } + const held = yield* Effect.forkChild(sql.withTransaction(blob), { startImmediately: true }); yield* Effect.sync(() => other.exec("COMMIT")); - assert.deepEqual(yield* read, [[0]]); - }).pipe(Effect.provide(SqliteClient.layer({ filename }))); + yield* Fiber.join(held); + // Without admission the third write appends to a WAL the pinned reader kept from resetting. + assert.isBelow(Number((yield* fs.stat(`${filename}-wal`)).size), limit); + assert.deepEqual(yield* sql`SELECT COUNT(*) FROM blobs`.values, [[3]]); + assert.deepEqual(yield* sql`PRAGMA user_version`.values, [[0]]); + }).pipe( + Effect.provide(SqliteClient.layer({ filename, writerWorker: true, checkpointWorker: true })), + ); }).pipe(Effect.provide(NodeServices.layer)), ); diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index 4c2c2fe9677af..72059c0c2ab37 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -35,7 +35,9 @@ export interface SqliteClientConfig { readonly spanAttributes?: Record | undefined; /** Ignored for in-memory databases, which cannot share a second connection. */ readonly readerWorker?: boolean | undefined; - /** Ignored for in-memory databases. */ + /** Ignored for in-memory databases. Falls back to the main thread when the worker cannot start. */ + readonly writerWorker?: boolean | undefined; + /** Ignored for in-memory databases. Bounds WAL growth; a writer worker enforces it in the worker itself. */ readonly checkpointWorker?: boolean | undefined; } @@ -194,31 +196,111 @@ const hasLargeBinding = (params: ReadonlyArray): boolean => { }; // Inline worker source also runs in bundles and single-executable builds without module files. -const readerWorkerSource = ` +const connectionWorkerSource = ` const { parentPort, workerData } = require("node:worker_threads"); -const { DatabaseSync } = require("node:sqlite"); +const { DatabaseSync, constants } = require("node:sqlite"); +const { statSync } = require("node:fs"); let db; +let admission; try { // A read-only open cannot be undone by SQL, unlike PRAGMA query_only. const next = new DatabaseSync(workerData.filename, { - readOnly: true, + readOnly: workerData.readOnly, 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"); + if (workerData.requireWal) { + 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; + if (workerData.admission !== undefined && typeof db.setAuthorizer === "function") { + admission = { + writes: new WeakMap(), + preparing: undefined, + sizeLimit: db.prepare("PRAGMA journal_size_limit"), + pageSize: db.prepare("PRAGMA page_size"), + busyTimeout: db.prepare("PRAGMA busy_timeout"), + checkpoint: db.prepare("PRAGMA wal_checkpoint(PASSIVE)"), + truncate: db.prepare("PRAGMA wal_checkpoint(TRUNCATE)"), + }; + // Same classification as the main-thread writer's checkpoint admission. + const valuePragmas = new Set(["application_id", "auto_vacuum", "journal_mode", "schema_version", "user_version"]); + const mutatesPragma = (name, value) => { + const pragma = name === null ? undefined : name.toLowerCase(); + if (pragma === undefined) return false; + if (pragma === "incremental_vacuum" || pragma === "optimize") return true; + if (pragma === "wal_checkpoint") return value === null || value.toLowerCase() !== "noop"; + return value !== null && valuePragmas.has(pragma); + }; + db.setAuthorizer((code, action, argument) => { + const writes = + code === constants.SQLITE_TRANSACTION || code === constants.SQLITE_SAVEPOINT + ? action === "BEGIN" + : code === constants.SQLITE_PRAGMA + ? mutatesPragma(action, argument) + : code !== constants.SQLITE_READ && + code !== constants.SQLITE_SELECT && + code !== constants.SQLITE_FUNCTION && + code !== constants.SQLITE_RECURSIVE; + if (writes && admission.preparing !== undefined) admission.preparing.writes = true; + return constants.SQLITE_OK; + }); + } parentPort.postMessage({ type: "ready" }); } catch (error) { parentPort.postMessage({ type: "unavailable", message: String(error && error.message ? error.message : error) }); } const statements = new Map(); +const prepareClassified = (sql) => { + if (admission === undefined) return db.prepare(sql); + const classification = { writes: false }; + admission.preparing = classification; + try { + const statement = db.prepare(sql); + admission.writes.set(statement, classification.writes); + return statement; + } finally { + admission.preparing = undefined; + } +}; +const walBytes = () => { + try { + return statSync(workerData.filename + "-wal").size; + } catch { + return 0; + } +}; +// Mirrors the checkpoint worker's admission: a write waits while the WAL is over its bound and +// a pinned reader keeps checkpoints from shrinking it. The wait polls, so it can be cancelled. +const admitted = () => { + const { backpressureFrames } = workerData.admission; + for (let attempt = 0; attempt < 2; attempt++) { + const limit = Number(admission.sizeLimit.get().journal_size_limit); + const frameBytes = Number(admission.pageSize.get().page_size) + 24; + const size = walBytes(); + const oversized = limit >= 0 && size > limit; + if (!oversized && size < backpressureFrames * frameBytes) return true; + const state = admission.checkpoint.get(); + if (!oversized && state.log < backpressureFrames) return true; + if (state.checkpointed !== state.log) return false; + const timeout = Number(admission.busyTimeout.get().timeout); + // A pinned reader must not make truncation wait; retry after the idle delay instead. + db.exec("PRAGMA busy_timeout = 0;"); + try { + if (admission.truncate.get().busy !== 0) return false; + } finally { + db.exec("PRAGMA busy_timeout = " + timeout + ";"); + } + } + return false; +}; const prepare = (sql, cached) => { - if (!cached) return db.prepare(sql); + if (!cached) return prepareClassified(sql); let statement = statements.get(sql); if (statement === undefined) { - statement = db.prepare(sql); + statement = prepareClassified(sql); if (statements.size >= workerData.prepareCacheSize) statements.delete(statements.keys().next().value); } else { statements.delete(sql); @@ -226,13 +308,38 @@ const prepare = (sql, cached) => { statements.set(sql, statement); return statement; }; -parentPort.on("message", ({ id, sql, params, cached, values, raw, safeIntegers }) => { +const reply = (id, operation, 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, + }, + }); +let waiting; +const backlog = []; +const run = (request) => { + const { id, sql, params, cached, values, raw, safeIntegers } = request; let operation = "open"; try { if (db === undefined) throw new Error("database reader is unavailable"); operation = "prepare"; const statement = prepare(sql, cached); operation = "execute"; + if (admission !== undefined && admission.writes.get(statement) && !db.isTransaction && !admitted()) { + waiting = { + id, + timer: setTimeout(() => { + waiting = undefined; + run(request); + while (waiting === undefined && backlog.length > 0) run(backlog.shift()); + }, workerData.admission.idleMs), + }; + return; + } statement.setReadBigInts(safeIntegers); if (statement.columns().length > 0) { statement.setReturnArrays(values); @@ -242,17 +349,38 @@ parentPort.on("message", ({ id, sql, params, cached, values, raw, safeIntegers } 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, - }, - }); + reply(id, operation, error); } +}; +const stopWaiting = (message) => { + if (waiting === undefined) return false; + clearTimeout(waiting.timer); + reply(waiting.id, "execute", message); + waiting = undefined; + return true; +}; +parentPort.on("message", (request) => { + if (request.type === "close") { + stopWaiting("Database writer closed"); + for (const queued of backlog) reply(queued.id, "execute", "Database writer closed"); + if (db !== undefined) db.close(); + parentPort.close(); + return; + } + if (request.type === "cancel") { + // A request that is not waiting has already run; its reply is on the way. + const queued = backlog.findIndex((entry) => entry.id === request.id); + if (queued !== -1) { + reply(request.id, "execute", "Statement cancelled before it ran"); + backlog.splice(queued, 1); + } else if (waiting !== undefined && waiting.id === request.id) { + stopWaiting("Statement cancelled before it ran"); + while (waiting === undefined && backlog.length > 0) run(backlog.shift()); + } + return; + } + if (waiting !== undefined) backlog.push(request); + else run(request); }); `; @@ -290,6 +418,47 @@ const readerFailureMessages = { const failed = (cause: unknown, message: string, operation: string) => new SqlError({ reason: classifyError(cause, message, operation) }); +const replyToExit = (reply: ReaderReply): Exit.Exit, SqlError> => { + if ("rows" in reply) return Exit.succeed(reply.rows); + const { operation, message, ...codes } = reply.error; + return Exit.fail( + failed(Object.assign(new Error(message), codes), readerFailureMessages[operation], operation), + ); +}; + +const workerConnection = ( + request: (input: ReaderRequest) => Effect.Effect, SqlError>, + rewriteUnprepared: (sql: string) => string = identity, +) => + 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) { + const effect = request({ + sql: rewriteUnprepared(sql), + params: params ?? [], + cached: false, + values: false, + raw: false, + }); + return rowTransform ? Effect.map(effect, rowTransform) : effect; + }, + executeStream(_sql, _params) { + return Stream.die(new UnsupportedNodeSqliteOperationError()); + }, + }); + // Read-transaction cleanup succeeds after worker exit because it can no longer change anything. const endsReadTransaction = (sql: string) => /^\s*ROLLBACK\b|^\s*COMMIT\b/i.test(sql); @@ -311,10 +480,12 @@ const makeReader = Effect.fnUntraced(function* ( let nextRequestId = 0; const start = () => { - const worker = new NodeWorkerThreads.Worker(readerWorkerSource, { + const worker = new NodeWorkerThreads.Worker(connectionWorkerSource, { eval: true, workerData: { filename: options.filename, + readOnly: true, + requireWal: true, allowExtension: false, prepareCacheSize: 200, }, @@ -369,25 +540,7 @@ const makeReader = Effect.fnUntraced(function* ( 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, - ), - ), - ); - }, - }); + pending.set(id, { endsTransaction, resume: (reply) => resume(replyToExit(reply)) }); if (pending.size === 1) worker.ref(); const cached = input.cached && !hasLargeBinding(input.params); // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin. @@ -395,36 +548,10 @@ const makeReader = Effect.fnUntraced(function* ( }); }); - 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) { - // Read snapshots need deferred BEGIN; readonly connections reject IMMEDIATE and EXCLUSIVE. - 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()); - }, - }); + // Read snapshots need deferred BEGIN; readonly connections reject IMMEDIATE and EXCLUSIVE. + const connection = workerConnection(request, (sql) => + /^\s*BEGIN\b/i.test(sql) ? "BEGIN" : sql, + ); current = { worker, connection, started }; return current; }; @@ -484,6 +611,144 @@ const makeReader = Effect.fnUntraced(function* ( }); }); +// The writer cannot be replaced after an exit: a new connection would lack setup PRAGMAs and any open transaction. +const startWriterWorker = Effect.fnUntraced(function* (options: SqliteClientConfig) { + const warn = (reason: string) => + Effect.logWarning("SQLite writer worker unavailable; writes run on the main thread", { + reason, + }); + const created = Result.try( + () => + new NodeWorkerThreads.Worker(connectionWorkerSource, { + eval: true, + workerData: { + filename: options.filename, + readOnly: options.readonly ?? false, + requireWal: false, + allowExtension: false, + prepareCacheSize: 200, + admission: options.checkpointWorker + ? { + backpressureFrames: CHECKPOINT_BACKPRESSURE_FRAMES, + idleMs: CHECKPOINT_IDLE_MS, + } + : undefined, + }, + }), + ); + if (Result.isFailure(created)) { + yield* warn(String(created.failure)); + return undefined; + } + const worker = created.success; + const runFork = Effect.runForkWith(yield* Effect.context()); + let announce: (startup: ReaderStartup) => void = () => {}; + const started = new Promise((resolve) => { + announce = resolve; + }); + const exited = new Promise((resolve) => worker.once("exit", () => resolve())); + const pending = new Map void>(); + let nextRequestId = 0; + let ready = false; + let stoppedMessage: string | undefined; + const settle = (id: number, reply: ReaderReply) => { + const resume = pending.get(id); + if (resume === undefined) return; + pending.delete(id); + if (pending.size === 0) worker.unref(); + resume(reply); + }; + const stop = (message: string) => { + if (stoppedMessage !== undefined) return; + stoppedMessage = message; + announce({ type: "unavailable", message }); + for (const id of pending.keys()) settle(id, { id, error: { operation: "execute", message } }); + if (ready) runFork(Effect.logError("SQLite writer worker stopped", { reason: message })); + }; + worker.on("message", (reply: ReaderReply | ReaderStartup) => { + if ("type" in reply) { + announce(reply); + if (pending.size === 0) worker.unref(); + return; + } + settle(reply.id, reply); + }); + worker.on("error", (cause) => stop(`Database writer failed: ${cause.message}`)); + worker.on("exit", (code) => stop(`Database writer exited with code ${code}`)); + + const startupTimeout = Duration.seconds(10); + const startup = yield* Effect.timeoutOption( + Effect.promise(() => started), + startupTimeout, + ); + if (Option.isNone(startup) || startup.value.type !== "ready") { + void worker.terminate(); + yield* warn( + Option.isSome(startup) && startup.value.type === "unavailable" + ? startup.value.message + : `worker did not open the database within ${Duration.toMillis(startupTimeout)} ms`, + ); + return undefined; + } + ready = true; + yield* Effect.addFinalizer(() => + Effect.promise(() => { + if (stoppedMessage !== undefined) return exited; + stoppedMessage = "Database writer closed"; + // Queued after in-flight statements; the ref keeps the process alive until the database closes. + worker.ref(); + // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin. + worker.postMessage({ type: "close" }); + return exited; + }), + ); + + const request = (input: ReaderRequest) => + Effect.withFiber, SqlError>((fiber) => { + if (stoppedMessage !== undefined) { + // The exited connection rolled back whatever was open; nothing else can succeed. + return /^\s*ROLLBACK\b/i.test(input.sql) + ? 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++; + let replied: () => void = () => {}; + const settled = new Promise((resolve) => { + replied = resolve; + }); + pending.set(id, (reply) => { + replied(); + resume(replyToExit(reply)); + }); + if (pending.size === 1) worker.ref(); + const cached = input.cached && !hasLargeBinding(input.params); + try { + // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin. + worker.postMessage({ id, safeIntegers, ...input, cached }); + } catch (cause) { + settle(id, { + id, + error: { operation: "execute", message: String(cause) }, + }); + } + // Only a write still waiting for WAL admission is cancelled; a running statement finishes first. + return Effect.promise(() => { + if (!pending.has(id)) return settled; + try { + // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin. + worker.postMessage({ type: "cancel", id }); + } catch {} + return settled; + }); + }); + }); + return workerConnection(request); +}); + const CHECKPOINT_FRAMES = 1000; const CHECKPOINT_IDLE_MS = 100; const CHECKPOINT_BACKPRESSURE_FRAMES = 32_768; @@ -1076,7 +1341,9 @@ const make = Effect.fn("makeWithDatabase")(function* ( }); const semaphore = yield* Semaphore.make(1); - const connection = yield* makeConnection; + const writerConnection = + options.writerWorker && isFileDatabase ? yield* startWriterWorker(options) : undefined; + const connection = writerConnection ?? (yield* makeConnection); // A reader that cannot start turns off the whole pool, so routing never picks a dead slot. const readerPool = { unavailable: false }; const readers =