From 2d80bb24a6d9bf8dd0e62f383eb3d45178ba58c7 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Mon, 5 Oct 2026 22:08:30 +1100 Subject: [PATCH 1/2] fix(server): thread list reads no longer freeze the server on large databases getShellSnapshot held the only SQLite connection on the main thread for the whole read, blocking keepalives, requests and writes for seconds on large databases. nodeSqliteClient gets a readerWorker option and a readOnly marker: a transaction wrapped in readOnly runs on a second connection in a worker thread. getShellSnapshot and the three routes that wrap it in a transaction are marked. The reader connection is opened read-only, because PRAGMA query_only can be turned off by a statement on the same connection. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../src/orchestration-v2/ProjectionStore.ts | 8 +- .../src/orchestration-v2/ShellStream.ts | 18 +- apps/server/src/persistence/Sqlite.ts | 1 + packages/shared/src/nodeSqliteClient.test.ts | 181 +++++++++++ packages/shared/src/nodeSqliteClient.ts | 284 +++++++++++++++++- 5 files changed, 481 insertions(+), 11 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index f33484383527..eac9f086ee73 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -69,6 +69,7 @@ import { isOrchestrationV2SupersededInterrupt, } from "@t3tools/shared/orchestrationV2Timeline"; import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -5864,7 +5865,12 @@ export const layer: Layer.Layer = const getShellSnapshot: ProjectionStoreV2Shape["getShellSnapshot"] = (options) => sql .withTransaction(readShellSnapshotRows(options)) - .pipe(Effect.mapError(shellSnapshotReadError), Effect.flatMap(decodeShellSnapshot)); + .pipe( + // Reads every thread, so it runs off the event loop and never holds the writer. + NodeSqliteClient.readOnly, + Effect.mapError(shellSnapshotReadError), + Effect.flatMap(decodeShellSnapshot), + ); // Per-thread shell for the live shell streams: reads only the target thread // plus its fork-source chain instead of materializing every thread. Returns diff --git a/apps/server/src/orchestration-v2/ShellStream.ts b/apps/server/src/orchestration-v2/ShellStream.ts index cec809f45ad4..9730ca5016f6 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/persistence/Sqlite.ts b/apps/server/src/persistence/Sqlite.ts index d6adf41f3d95..d3760db007c1 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 83fe851ac8ad..a4b37b61fcf7 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -1,14 +1,54 @@ 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 Path from "effect/Path"; import * as SqlClient from "effect/sql/SqlClient"; +import { vi } from "vite-plus/test"; import * as SqliteClient from "./nodeSqliteClient.ts"; +const startedWorkers = vi.hoisted((): Array => []); + +vi.mock("node:worker_threads", async (importOriginal) => { + const actual = await importOriginal(); + class RecordedWorker extends actual.Worker { + constructor(...args: ConstructorParameters) { + super(...args); + startedWorkers.push(this); + } + } + 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 +222,144 @@ 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 }); + }), + ), +); diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index 6b05d8ac6ca1..26d6ef579d50 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"; @@ -36,8 +37,31 @@ 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; } +/** + * 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 +123,237 @@ 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; +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) { + // A read-only open cannot be undone by SQL, unlike PRAGMA query_only. The + // writer connection opened the database first, so its WAL files exist. + const next = new DatabaseSync(workerData.filename, { + readOnly: true, + allowExtension: workerData.allowExtension, + }); + next.exec("PRAGMA busy_timeout = 5000;"); + db = next; + } + 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 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. + */ +const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) { + let current: + | { readonly worker: NodeWorkerThreads.Worker; readonly connection: Connection } + | undefined; + 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, + }, + }); + // An idle reader must not keep the process alive. + worker.unref(); + let stoppedMessage: string | undefined; + const pending = new Map< + number, + { readonly endsTransaction: boolean; readonly resume: (reply: ReaderReply) => void } + >(); + const stop = (message: string) => { + stoppedMessage = 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) => { + 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 }; + return current; + }; + + yield* Effect.addFinalizer(() => + Effect.promise(async () => { + await current?.worker.terminate(); + }), + ); + + /** The live worker's connection, or undefined when no worker can start. */ + return (): Connection | undefined => { + try { + return (current ?? start()).connection; + } catch { + return undefined; + } + }; +}); + const make = Effect.fn("makeWithDatabase")(function* ( options: SqliteClientConfig, ): Effect.fn.Return { @@ -261,14 +516,37 @@ 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); + + // Picks the reader only for `readOnly` work, and falls back to the writer + // when the reader worker cannot start. + const route = (fiber: Fiber.Fiber) => { + const readerConnection = + acquireReader !== undefined && Context.get(fiber.context, ReadOnly) + ? acquireReader() + : undefined; + return readerConnection === undefined + ? { lock: semaphore, connection } + : { lock: readerSemaphore, connection: readerConnection }; + }; - const acquirer = semaphore.withPermits(1)(Effect.succeed(connection)); + const acquirer = Effect.withFiber((fiber) => { + const target = route(fiber); + return target.lock.withPermits(1)(Effect.succeed(target.connection)); + }); const transactionAcquirer = Effect.uninterruptibleMask((restore) => { const fiber = Fiber.getCurrent()!; const scope = Context.getUnsafe(fiber.context, Scope.Scope); + const target = route(fiber); return Effect.as( - Effect.tap(restore(semaphore.take(1)), () => Scope.addFinalizer(scope, semaphore.release(1))), - connection, + Effect.tap(restore(target.lock.take(1)), () => + Scope.addFinalizer(scope, target.lock.release(1)), + ), + target.connection, ); }); From febd5c2c3ca54bbae86127c987abd641014b7f15 Mon Sep 17 00:00:00 2001 From: Alex Southwell Date: Tue, 6 Oct 2026 09:49:44 +1100 Subject: [PATCH 2/2] fix(server): fall back to the main connection when the SQLite reader cannot start The reader worker opened the database lazily, so a read-only open failure broke every thread list read for good, and a database outside WAL mode let a reader's shared lock stall the writer. The worker now opens the file read-only at startup, checks for WAL, and reports ready before any read is routed to it. If it cannot, does not report within readerStartupTimeout (10 s), or a request is cancelled while it starts, reads use the main connection as before; the reader turns off once, with one warning. A late "ready" from a worker another read already retired is not handed out. getShellSnapshot is no longer marked inside the store. Callers opt in where a snapshot that keeps moving is fine: the three snapshot routes and the once-a-minute pull request discovery pass, which guards its writes with the snapshot's sequence. Restore-safety and project-deletion checks read on the main connection, as on main. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../src/orchestration-v2/ProjectionStore.ts | 8 +- .../ThreadPullRequestService.ts | 13 +- packages/shared/src/nodeSqliteClient.test.ts | 189 +++++++++++++++++- packages/shared/src/nodeSqliteClient.ts | 167 +++++++++++----- 4 files changed, 318 insertions(+), 59 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index eac9f086ee73..f33484383527 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -69,7 +69,6 @@ import { isOrchestrationV2SupersededInterrupt, } from "@t3tools/shared/orchestrationV2Timeline"; import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; -import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -5865,12 +5864,7 @@ export const layer: Layer.Layer = const getShellSnapshot: ProjectionStoreV2Shape["getShellSnapshot"] = (options) => sql .withTransaction(readShellSnapshotRows(options)) - .pipe( - // Reads every thread, so it runs off the event loop and never holds the writer. - NodeSqliteClient.readOnly, - Effect.mapError(shellSnapshotReadError), - Effect.flatMap(decodeShellSnapshot), - ); + .pipe(Effect.mapError(shellSnapshotReadError), Effect.flatMap(decodeShellSnapshot)); // Per-thread shell for the live shell streams: reads only the target thread // plus its fork-source chain instead of materializing every thread. Returns diff --git a/apps/server/src/orchestration-v2/ThreadPullRequestService.ts b/apps/server/src/orchestration-v2/ThreadPullRequestService.ts index 7385f3ff5d25..a891fa764ac7 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/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index a4b37b61fcf7..040baa7ba130 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -7,21 +7,34 @@ 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(...args); + 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 }; }); @@ -363,3 +376,177 @@ it.effect("keeps a nested transaction inside a read-only transaction on the read }), ), ); + +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 26d6ef579d50..dd00bccd1e47 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -14,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"; @@ -43,6 +44,8 @@ export interface SqliteClientConfig { * 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; } /** @@ -129,6 +132,21 @@ 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); @@ -145,16 +163,7 @@ const prepare = (sql, cached) => { parentPort.on("message", ({ id, sql, params, cached, values, raw, safeIntegers }) => { let operation = "open"; try { - if (db === undefined) { - // A read-only open cannot be undone by SQL, unlike PRAGMA query_only. The - // writer connection opened the database first, so its WAL files exist. - const next = new DatabaseSync(workerData.filename, { - readOnly: true, - allowExtension: workerData.allowExtension, - }); - next.exec("PRAGMA busy_timeout = 5000;"); - db = next; - } + if (db === undefined) throw new Error("database reader is unavailable"); operation = "prepare"; const statement = prepare(sql, cached); operation = "execute"; @@ -189,6 +198,10 @@ interface ReaderRequest { readonly raw: boolean; } +type ReaderStartup = + | { readonly type: "ready" } + | { readonly type: "unavailable"; readonly message: string }; + type ReaderReply = | { readonly id: number; readonly rows: ReadonlyArray } | { @@ -220,11 +233,18 @@ const endsReadTransaction = (sql: string) => /^\s*ROLLBACK\b|^\s*COMMIT\b/i.test * 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 worker: NodeWorkerThreads.Worker; + readonly connection: Connection; + readonly started: Promise; + } | undefined; + let unavailable = false; let nextRequestId = 0; const start = () => { @@ -236,8 +256,10 @@ const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) { prepareCacheSize: options.prepareCacheSize ?? 200, }, }); - // An idle reader must not keep the process alive. - worker.unref(); + let announce: (startup: ReaderStartup) => void = () => {}; + const started = new Promise((resolve) => { + announce = resolve; + }); let stoppedMessage: string | undefined; const pending = new Map< number, @@ -245,6 +267,7 @@ const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) { >(); const stop = (message: string) => { stoppedMessage = message; + announce({ type: "unavailable", message }); if (current?.worker === worker) current = undefined; for (const [id, request] of pending) { request.resume( @@ -255,7 +278,13 @@ const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) { } pending.clear(); }; - worker.on("message", (reply: ReaderReply) => { + 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); @@ -334,7 +363,7 @@ const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) { return Stream.die(new UnsupportedNodeSqliteOperationError()); }, }); - current = { worker, connection }; + current = { worker, connection, started }; return current; }; @@ -344,14 +373,57 @@ const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) { }), ); - /** The live worker's connection, or undefined when no worker can start. */ - return (): Connection | undefined => { + 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 { - return (current ?? start()).connection; - } catch { - return undefined; + 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* ( @@ -522,33 +594,34 @@ const make = Effect.fn("makeWithDatabase")(function* ( : undefined; const readerSemaphore = yield* Semaphore.make(1); - // Picks the reader only for `readOnly` work, and falls back to the writer - // when the reader worker cannot start. - const route = (fiber: Fiber.Fiber) => { - const readerConnection = - acquireReader !== undefined && Context.get(fiber.context, ReadOnly) - ? acquireReader() - : undefined; - return readerConnection === undefined - ? { lock: semaphore, connection } - : { lock: readerSemaphore, connection: readerConnection }; - }; + 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.withFiber((fiber) => { - const target = route(fiber); - return target.lock.withPermits(1)(Effect.succeed(target.connection)); - }); - const transactionAcquirer = Effect.uninterruptibleMask((restore) => { - const fiber = Fiber.getCurrent()!; - const scope = Context.getUnsafe(fiber.context, Scope.Scope); - const target = route(fiber); - return Effect.as( - Effect.tap(restore(target.lock.take(1)), () => - Scope.addFinalizer(scope, target.lock.release(1)), - ), - target.connection, - ); - }); + 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,