From 47667aa747f61200655c909cc38ca20baf2c59ac Mon Sep 17 00:00:00 2001 From: Advait Paliwal Date: Tue, 3 Nov 2026 03:52:55 -0500 Subject: [PATCH 1/6] fix(server): SQLite writes no longer block the event loop The server's writer connection ran node:sqlite DatabaseSync on the main thread, so a slow statement or transaction on a large statev2.sqlite froze every HTTP and relay request until it finished. The writer now runs in a worker thread that reuses the reader worker's message protocol, behind the same SqlClient interface. If the worker cannot start, writes fall back to the main thread as before. --- .../FoundationPersistence.test.ts | 66 +++-- apps/server/src/persistence/Sqlite.ts | 4 +- packages/shared/src/nodeSqliteClient.test.ts | 86 ++++--- packages/shared/src/nodeSqliteClient.ts | 243 ++++++++++++++---- 4 files changed, 272 insertions(+), 127 deletions(-) diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index c5c51646584aa..5fdf701080e2b 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 { @@ -47,8 +46,6 @@ import * as Tracer from "effect/Tracer"; import * as SqlClient from "effect/sql/SqlClient"; import * as Statement from "effect/sql/Statement"; -import { vi } from "vite-plus/test"; - import * as ServerConfig from "../config.ts"; import * as DatabaseStorage from "../persistence/DatabaseStorage.ts"; import * as ServerSettings from "../serverSettings.ts"; @@ -253,46 +250,41 @@ 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.compactEventStore(); + const result = yield* maintenance + .compactEventStore() + .pipe( + Effect.withTracer(tracer), + Effect.provideService(Statement.SpanPropagationEnabled, true), + ); assert.equal(result.deletedEventCount, 148); assert.equal(result.deletedReceiptCount, 100); + const connectionsFor = (pattern: RegExp) => + new Set( + statements + .filter(({ query }) => typeof query === "string" && pattern.test(query)) + .map(({ connection }) => connection), + ); + assert.deepEqual(connectionsFor(/^\s*(SELECT|WITH)\b/), new Set(["reader"])); assert.deepEqual( - reads.filter((query) => /\bSELECT\b/.test(query)), - [], + connectionsFor(/^\s*DELETE FROM orchestration_(events|command_receipts)\b/), + new Set(["writer"]), ); - 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.ts b/apps/server/src/persistence/Sqlite.ts index 9b95422c4f5c1..8eeb0e3b1f81f 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, + // Keeps checkpoint fsyncs off the event loop if the writer has to fall back to the main thread. 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..ec2bd76785444 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,41 @@ 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"); - - yield* Effect.sync(() => other.exec("COMMIT")); - assert.deepEqual(yield* read, [[0]]); - }).pipe(Effect.provide(SqliteClient.layer({ filename }))); + 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)), ); diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index 9531354db1939..114efff5f868d 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 and when the writer runs in a worker, which checkpoints off the event loop itself. */ readonly checkpointWorker?: boolean | undefined; } @@ -194,20 +196,22 @@ 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"); let db; 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; parentPort.postMessage({ type: "ready" }); } catch (error) { @@ -226,7 +230,12 @@ const prepare = (sql, cached) => { statements.set(sql, statement); return statement; }; -parentPort.on("message", ({ id, sql, params, cached, values, raw, safeIntegers }) => { +parentPort.on("message", ({ type, id, sql, params, cached, values, raw, safeIntegers }) => { + if (type === "close") { + if (db !== undefined) db.close(); + parentPort.close(); + return; + } let operation = "open"; try { if (db === undefined) throw new Error("database reader is unavailable"); @@ -290,6 +299,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 +361,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 +421,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 +429,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; }; @@ -487,6 +495,125 @@ 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 = yield* Effect.try( + () => + new NodeWorkerThreads.Worker(connectionWorkerSource, { + eval: true, + workerData: { + filename: options.filename, + readOnly: options.readonly ?? false, + requireWal: false, + allowExtension: false, + prepareCacheSize: 200, + }, + }), + ).pipe(Effect.result); + if (Result.isFailure(created)) { + yield* warn(String(created.failure.cause)); + 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)); + // Like a statement on the main thread, a sent statement finishes before its caller sees an interrupt. + return Effect.uninterruptible( + Effect.callback, SqlError>((resume) => { + const id = nextRequestId++; + pending.set(id, (reply) => 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) }, + }); + } + }), + ); + }); + return workerConnection(request); +}); + const CHECKPOINT_FRAMES = 1000; const CHECKPOINT_IDLE_MS = 100; const CHECKPOINT_BACKPRESSURE_FRAMES = 10 * CHECKPOINT_FRAMES; @@ -1079,7 +1206,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 = From 3676831694ff403183076b954538cc3e5b318abc Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 03:26:20 +0000 Subject: [PATCH 2/6] test(server): reconcile SQLite writer-worker tests with main The compaction test now runs pruneEventStore and allows main's in-transaction prompt check on the writer. The WAL burst test asserts that writes keep moving past a 48 MiB WAL without assuming a checkpoint worker, which the writer worker replaces. AI-assisted; verified with the tests and tools listed above. Co-Authored-By: Claude Opus 5.5 --- .../FoundationPersistence.test.ts | 23 ++++---- apps/server/src/persistence/Sqlite.test.ts | 53 ------------------- 2 files changed, 10 insertions(+), 66 deletions(-) diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index 731d789c3f3f4..0e307efbdcfd2 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -51,6 +51,8 @@ import * as Tracer from "effect/Tracer"; import * as SqlClient from "effect/sql/SqlClient"; import * as Statement from "effect/sql/Statement"; +import { vi } from "vite-plus/test"; + import * as ServerConfig from "../config.ts"; import * as DatabaseStorage from "../persistence/DatabaseStorage.ts"; import * as ServerSettings from "../serverSettings.ts"; @@ -212,25 +214,20 @@ it.effect("discovers on the reader worker and deletes on the writer", () => ); assert.equal(result.deletedEventCount, 150); assert.equal(result.deletedReceiptCount, 100); - const connectionsFor = (matches: (query: string) => boolean) => + // Only the in-transaction check for rows appended during the pass reads on the writer. + assert.deepEqual( new Set( statements - .filter(({ query }) => typeof query === "string" && matches(query)) + .filter( + ({ query }) => + typeof query === "string" && + /^\s*(SELECT|WITH)\b/.test(query) && + !query.includes("AS prompt_sequence"), + ) .map(({ connection }) => connection), - ); - // Only the in-transaction check for rows appended during the pass reads on the writer. - assert.deepEqual( - connectionsFor( - (query) => /^\s*(SELECT|WITH)\b/.test(query) && !query.includes("AS prompt_sequence"), ), new Set(["reader"]), ); - assert.deepEqual( - connectionsFor((query) => - /^\s*DELETE FROM orchestration_(events|command_receipts)\b/.test(query), - ), - new Set(["writer"]), - ); }).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"); From 18a9024b45437d134a60e365ca6e4961423a7af9 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 03:30:10 +0000 Subject: [PATCH 3/6] fix(shared): start the writer worker with Result.try AI-assisted; verified with the tests and tools listed above. Co-Authored-By: Claude Opus 5.5 --- packages/shared/src/nodeSqliteClient.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index 8233158f1ee94..fda547a448670 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -498,7 +498,7 @@ const startWriterWorker = Effect.fnUntraced(function* (options: SqliteClientConf Effect.logWarning("SQLite writer worker unavailable; writes run on the main thread", { reason, }); - const created = yield* Effect.try( + const created = Result.try( () => new NodeWorkerThreads.Worker(connectionWorkerSource, { eval: true, @@ -510,9 +510,9 @@ const startWriterWorker = Effect.fnUntraced(function* (options: SqliteClientConf prepareCacheSize: 200, }, }), - ).pipe(Effect.result); + ); if (Result.isFailure(created)) { - yield* warn(String(created.failure.cause)); + yield* warn(String(created.failure)); return undefined; } const worker = created.success; From 0ff2a9e3bfc0e8fabaaad7ba2968bcd7e3b99c1a Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 03:53:36 +0000 Subject: [PATCH 4/6] fix(shared): writer-worker writes wait for WAL admission like the main-thread writer Moving the writer into a worker bypassed the checkpoint admission that holds writes while a pinned reader keeps the WAL over journal_size_limit or the backpressure frame count, so sustained writes could grow the WAL without bound. The writer worker now classifies statements with an authorizer and, before a write outside a transaction, checkpoints and truncates the WAL or waits until the reader releases it. Blocking happens in the worker, not the event loop. AI-assisted; verified with the tests and tools listed above. Co-Authored-By: Claude Opus 5.5 --- packages/shared/src/nodeSqliteClient.test.ts | 28 ++++++ packages/shared/src/nodeSqliteClient.ts | 91 +++++++++++++++++++- 2 files changed, 116 insertions(+), 3 deletions(-) diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index ec2bd76785444..8e99eab63aa73 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -262,6 +262,34 @@ it.effect("keeps the event loop free while a write waits on another connection's }).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); + const held = yield* Effect.forkChild(sql.withTransaction(blob), { startImmediately: true }); + yield* Effect.sync(() => other.exec("COMMIT")); + 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); + }).pipe( + Effect.provide(SqliteClient.layer({ filename, writerWorker: true, checkpointWorker: true })), + ); + }).pipe(Effect.provide(NodeServices.layer)), +); + it.effect("lets the writer commit while a read-only transaction is open", () => withReaderClient((sql) => Effect.gen(function* () { diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index fda547a448670..215fc949c1d21 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -198,8 +198,10 @@ const hasLargeBinding = (params: ReadonlyArray): boolean => { // Inline worker source also runs in bundles and single-executable builds without module files. 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, { @@ -213,16 +215,87 @@ try { if (mode !== "wal") throw new Error("database uses journal_mode " + mode + ", not wal"); } db = next; + if (workerData.admission !== undefined && typeof db.setAuthorizer === "function") { + admission = { + control: new Int32Array(workerData.admission.control), + 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)"), + }; + db.setAuthorizer((code, action) => { + const writes = + code === constants.SQLITE_TRANSACTION || code === constants.SQLITE_SAVEPOINT + ? action === "BEGIN" + : code !== constants.SQLITE_READ && + code !== constants.SQLITE_SELECT && + code !== constants.SQLITE_FUNCTION && + code !== constants.SQLITE_RECURSIVE && + code !== constants.SQLITE_PRAGMA; + 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. Blocking here leaves the event loop free. +const admit = () => { + const { control } = admission; + const { backpressureFrames, idleMs } = workerData.admission; + while (Atomics.load(control, 0) === 0) { + 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; + const state = admission.checkpoint.get(); + if (!oversized && state.log < backpressureFrames) return; + if (state.checkpointed === state.log) { + 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;"); + let busy; + try { + busy = admission.truncate.get().busy !== 0; + } finally { + db.exec("PRAGMA busy_timeout = " + timeout + ";"); + } + if (!busy) continue; + } + Atomics.wait(control, 0, 0, idleMs); + } +}; 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); @@ -242,6 +315,7 @@ parentPort.on("message", ({ type, id, sql, params, cached, values, raw, safeInte operation = "prepare"; const statement = prepare(sql, cached); operation = "execute"; + if (admission !== undefined && admission.writes.get(statement) && !db.isTransaction) admit(); statement.setReadBigInts(safeIntegers); if (statement.columns().length > 0) { statement.setReturnArrays(values); @@ -494,6 +568,8 @@ 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) { + // Set when the writer closes so a write held for WAL admission stops waiting. + const closing = new Int32Array(new SharedArrayBuffer(4)); const warn = (reason: string) => Effect.logWarning("SQLite writer worker unavailable; writes run on the main thread", { reason, @@ -508,6 +584,13 @@ const startWriterWorker = Effect.fnUntraced(function* (options: SqliteClientConf requireWal: false, allowExtension: false, prepareCacheSize: 200, + admission: options.checkpointWorker + ? { + control: closing.buffer, + backpressureFrames: CHECKPOINT_BACKPRESSURE_FRAMES, + idleMs: CHECKPOINT_IDLE_MS, + } + : undefined, }, }), ); @@ -570,6 +653,8 @@ const startWriterWorker = Effect.fnUntraced(function* (options: SqliteClientConf Effect.promise(() => { if (stoppedMessage !== undefined) return exited; stoppedMessage = "Database writer closed"; + Atomics.store(closing, 0, 1); + Atomics.notify(closing, 0); // 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. From 71c8f69385552ed217b5604c6b02732c042fb442 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 03:56:51 +0000 Subject: [PATCH 5/6] docs(shared): describe checkpointWorker with a writer worker AI-assisted; verified with the tests and tools listed above. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/persistence/Sqlite.ts | 2 +- packages/shared/src/nodeSqliteClient.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/server/src/persistence/Sqlite.ts b/apps/server/src/persistence/Sqlite.ts index 123c7e0ee2fba..3b2962c891a21 100644 --- a/apps/server/src/persistence/Sqlite.ts +++ b/apps/server/src/persistence/Sqlite.ts @@ -56,7 +56,7 @@ export const layerFromPath = Effect.fn("makeSqlitePersistenceLive")(function* (d readerWorker: true, // Run writes off the event loop so slow statements and transactions cannot stall clients. writerWorker: true, - // Keeps checkpoint fsyncs off the event loop if the writer has to fall back to the main thread. + // 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.ts b/packages/shared/src/nodeSqliteClient.ts index 215fc949c1d21..426163d38a625 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -37,7 +37,7 @@ export interface SqliteClientConfig { readonly readerWorker?: boolean | undefined; /** 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 and when the writer runs in a worker, which checkpoints off the event loop itself. */ + /** Ignored for in-memory databases. Bounds WAL growth; a writer worker enforces it in the worker itself. */ readonly checkpointWorker?: boolean | undefined; } From 39f8d400559a0a021fd1a909eb88cde2ca9b6db3 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 04:13:15 +0000 Subject: [PATCH 6/6] fix(shared): writer-worker WAL admission waits are cancellable and cover mutating pragmas A write held for WAL admission now waits on a timer in the worker instead of blocking it, so the caller's interrupt cancels it before it runs; a statement already running still finishes before the interrupt completes. Closing the writer ends the wait. The worker classifies mutating pragmas such as incremental_vacuum and user_version as writes, like the main-thread writer. AI-assisted; verified with the tests and tools listed above. Co-Authored-By: Claude Opus 5.5 --- packages/shared/src/nodeSqliteClient.test.ts | 6 + packages/shared/src/nodeSqliteClient.ts | 181 ++++++++++++------- 2 files changed, 123 insertions(+), 64 deletions(-) diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index 8e99eab63aa73..423e127fce4df 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -279,11 +279,17 @@ it.effect("holds writer-worker writes past the WAL size limit until a pinned rea }); 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")); 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 })), ); diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index 426163d38a625..72059c0c2ab37 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -217,7 +217,6 @@ try { db = next; if (workerData.admission !== undefined && typeof db.setAuthorizer === "function") { admission = { - control: new Int32Array(workerData.admission.control), writes: new WeakMap(), preparing: undefined, sizeLimit: db.prepare("PRAGMA journal_size_limit"), @@ -226,15 +225,25 @@ try { checkpoint: db.prepare("PRAGMA wal_checkpoint(PASSIVE)"), truncate: db.prepare("PRAGMA wal_checkpoint(TRUNCATE)"), }; - db.setAuthorizer((code, action) => { + // 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_READ && - code !== constants.SQLITE_SELECT && - code !== constants.SQLITE_FUNCTION && - code !== constants.SQLITE_RECURSIVE && - code !== constants.SQLITE_PRAGMA; + : 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; }); @@ -264,32 +273,28 @@ const walBytes = () => { } }; // 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. Blocking here leaves the event loop free. -const admit = () => { - const { control } = admission; - const { backpressureFrames, idleMs } = workerData.admission; - while (Atomics.load(control, 0) === 0) { +// 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; + if (!oversized && size < backpressureFrames * frameBytes) return true; const state = admission.checkpoint.get(); - if (!oversized && state.log < backpressureFrames) return; - if (state.checkpointed === state.log) { - 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;"); - let busy; - try { - busy = admission.truncate.get().busy !== 0; - } finally { - db.exec("PRAGMA busy_timeout = " + timeout + ";"); - } - if (!busy) continue; + 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 + ";"); } - Atomics.wait(control, 0, 0, idleMs); } + return false; }; const prepare = (sql, cached) => { if (!cached) return prepareClassified(sql); @@ -303,19 +308,38 @@ const prepare = (sql, cached) => { statements.set(sql, statement); return statement; }; -parentPort.on("message", ({ type, id, sql, params, cached, values, raw, safeIntegers }) => { - if (type === "close") { - if (db !== undefined) db.close(); - parentPort.close(); - return; - } +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) admit(); + 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); @@ -325,17 +349,38 @@ parentPort.on("message", ({ type, id, sql, params, cached, values, raw, safeInte 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); }); `; @@ -568,8 +613,6 @@ 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) { - // Set when the writer closes so a write held for WAL admission stops waiting. - const closing = new Int32Array(new SharedArrayBuffer(4)); const warn = (reason: string) => Effect.logWarning("SQLite writer worker unavailable; writes run on the main thread", { reason, @@ -586,7 +629,6 @@ const startWriterWorker = Effect.fnUntraced(function* (options: SqliteClientConf prepareCacheSize: 200, admission: options.checkpointWorker ? { - control: closing.buffer, backpressureFrames: CHECKPOINT_BACKPRESSURE_FRAMES, idleMs: CHECKPOINT_IDLE_MS, } @@ -653,8 +695,6 @@ const startWriterWorker = Effect.fnUntraced(function* (options: SqliteClientConf Effect.promise(() => { if (stoppedMessage !== undefined) return exited; stoppedMessage = "Database writer closed"; - Atomics.store(closing, 0, 1); - Atomics.notify(closing, 0); // 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. @@ -674,24 +714,37 @@ const startWriterWorker = Effect.fnUntraced(function* (options: SqliteClientConf ); } const safeIntegers = Boolean(Context.get(fiber.context, Client.SafeIntegers)); - // Like a statement on the main thread, a sent statement finishes before its caller sees an interrupt. - return Effect.uninterruptible( - Effect.callback, SqlError>((resume) => { - const id = nextRequestId++; - pending.set(id, (reply) => resume(replyToExit(reply))); - if (pending.size === 1) worker.ref(); - const cached = input.cached && !hasLargeBinding(input.params); + 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({ id, safeIntegers, ...input, cached }); - } catch (cause) { - settle(id, { - id, - error: { operation: "execute", message: String(cause) }, - }); - } - }), - ); + worker.postMessage({ type: "cancel", id }); + } catch {} + return settled; + }); + }); }); return workerConnection(request); });