authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-14 00:14:56-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-14 00:14:56-07:00
log8659fcf145c3372161b360b74b094b8e0ab80bec
tree7732bfb83ad9664d07d468ac8dfd3903e2a57441
parent30d1010443d92d7c17f9d773639375a4c6de77bc
signature Signed by PGP key B5690EEEBB952194

fix(server): thread list reads no longer freeze the server on large databases (#19226)

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Co-authored-by: Alex Southwell <saphid@gmail.com>

5 files changed, 749 insertions(+), 20 deletions(-)

apps/server/src/orchestration-v2/ShellStream.ts+11-7
......@@ -12,6 +12,7 @@ import {
1212 OrchestrationProjectShell as ProjectShellSchema,
1313 OrchestrationV2ThreadShell as ThreadShellSchema,
1414} from "@t3tools/contracts";
15import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
1516import * as Clock from "effect/Clock";
1617import * as Effect from "effect/Effect";
1718import * as Schema from "effect/Schema";
......@@ -46,13 +47,16 @@ export const loadShellSnapshotParts = <E1, E2, E3, E4>(input: {
4647 readonly latestSequence: Effect.Effect<number, E4>;
4748}) =>
4849 Effect.gen(function* () {
49 const read = yield* input.sql.withTransaction(
50 Effect.all({
51 decodeThreads: input.readThreads,
52 projects: input.listProjects,
53 snapshotSequence: input.latestSequence,
54 }),
55 );
50 // Read on the reader worker, off the event loop and the writer.
51 const read = yield* input.sql
52 .withTransaction(
53 Effect.all({
54 decodeThreads: input.readThreads,
55 projects: input.listProjects,
56 snapshotSequence: input.latestSequence,
57 }),
58 )
59 .pipe(NodeSqliteClient.readOnly);
5660 return {
5761 projects: read.projects,
5862 threads: yield* read.decodeThreads,
apps/server/src/orchestration-v2/ThreadPullRequestService.ts+9-4
......@@ -11,6 +11,7 @@ import {
1111 type ThreadLinkedPullRequest,
1212} from "@t3tools/contracts";
1313import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
14import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
1415import * as Cause from "effect/Cause";
1516import * as Context from "effect/Context";
1617import * as Crypto from "effect/Crypto";
......@@ -127,10 +128,14 @@ export const make = Effect.gen(function* () {
127128 */
128129 const readThreadSnapshot = ({ threadId, backfill }: RefreshRequest) =>
129130 threadId === null
130 ? orchestrator.getShellSnapshot({
131 location: "active",
132 unsettledOnly: !(backfill || pendingBackfill.size > 0),
133 })
131 ? orchestrator
132 .getShellSnapshot({
133 location: "active",
134 unsettledOnly: !(backfill || pendingBackfill.size > 0),
135 })
136 // Reads every active thread once a minute: run it off the event loop. Writes keep
137 // landing during the read, and the snapshot's sequence guards the sync against them.
138 .pipe(NodeSqliteClient.readOnly)
134139 : Effect.gen(function* () {
135140 // Read the sequence first. The thread is then at least this new, so a
136141 // sync guarded by the sequence is rejected rather than missing a change.
apps/server/src/persistence/Sqlite.ts+1
......@@ -45,6 +45,7 @@ export const layerFromPath = Effect.fn("makeSqlitePersistenceLive")(function* (d
4545 layerSetup,
4646 NodeSqliteClient.layer({
4747 filename: dbPath,
48 readerWorker: true,
4849 spanAttributes: {
4950 "db.name": path.basename(dbPath),
5051 "service.name": "t3code-server",
packages/shared/src/nodeSqliteClient.test.ts+368
......@@ -1,14 +1,67 @@
11import * as NodeSqlite from "node:sqlite";
2import type * as NodeWorkerThreads from "node:worker_threads";
23import * as NodeServices from "@effect/platform-node/NodeServices";
34import { assert, it } from "@effect/vitest";
5import * as Deferred from "effect/Deferred";
46import * as Effect from "effect/Effect";
7import * as Fiber from "effect/Fiber";
58import * as FileSystem from "effect/FileSystem";
69import * as Layer from "effect/Layer";
10import * as Logger from "effect/Logger";
711import * as Path from "effect/Path";
812import * as SqlClient from "effect/sql/SqlClient";
13import * as TestClock from "effect/testing/TestClock";
14import { vi } from "vite-plus/test";
915
1016import * as SqliteClient from "./nodeSqliteClient.ts";
1117
18const startedWorkers = vi.hoisted((): Array<NodeWorkerThreads.Worker> => []);
19// When on, workers run a script that never announces, like a worker stuck opening the file.
20const silentWorkers = vi.hoisted(() => ({ on: false, readyOnTerminate: false }));
21
22vi.mock("node:worker_threads", async (importOriginal) => {
23 const actual = await importOriginal<typeof NodeWorkerThreads>();
24 class RecordedWorker extends actual.Worker {
25 constructor(...args: ConstructorParameters<typeof actual.Worker>) {
26 super(
27 ...((silentWorkers.on
28 ? ["setInterval(() => {}, 1 << 30);", { eval: true }]
29 : args) as ConstructorParameters<typeof actual.Worker>),
30 );
31 startedWorkers.push(this);
32 }
33 override terminate() {
34 // A "ready" already queued when termination starts reaches the parent before "exit".
35 if (silentWorkers.readyOnTerminate) this.emit("message", { type: "ready" });
36 return super.terminate();
37 }
38 }
39 return { ...actual, Worker: RecordedWorker };
40});
41
42const withReaderClient = <A, E>(
43 use: (sql: SqlClient.SqlClient) => Effect.Effect<A, E, SqlClient.SqlClient>,
44) =>
45 Effect.gen(function* () {
46 const fs = yield* FileSystem.FileSystem;
47 const path = yield* Path.Path;
48 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-reader-" });
49 const filename = path.join(directory, "state.sqlite");
50 return yield* Effect.gen(function* () {
51 const sql = yield* SqlClient.SqlClient;
52 yield* sql`PRAGMA journal_mode = WAL`;
53 yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`;
54 yield* sql`INSERT INTO entries VALUES ('first')`;
55 return yield* use(sql);
56 }).pipe(Effect.provide(SqliteClient.layer({ filename, readerWorker: true })));
57 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer));
58
59const countEntries = (sql: SqlClient.SqlClient) =>
60 Effect.map(
61 sql<{ count: number }>`SELECT COUNT(*) AS count FROM entries`,
62 (rows) => rows[0]?.count,
63 );
64
1265const layer = it.layer(SqliteClient.layer({ filename: ":memory:" }));
1366
1467layer("NodeSqliteClient", (it) => {
......@@ -182,3 +235,318 @@ it.effect(
182235 }).pipe(Effect.provide(SqliteClient.layer({ filename })));
183236 }).pipe(Effect.provide(NodeServices.layer)),
184237);
238
239it.effect("lets the writer commit while a read-only transaction is open", () =>
240 withReaderClient((sql) =>
241 Effect.gen(function* () {
242 const snapshotTaken = yield* Deferred.make<void>();
243 const writer = yield* Effect.forkChild(
244 Deferred.await(snapshotTaken).pipe(
245 Effect.andThen(sql`INSERT INTO entries VALUES ('second')`),
246 ),
247 );
248 const counts = yield* sql
249 .withTransaction(
250 Effect.gen(function* () {
251 const before = yield* countEntries(sql);
252 yield* Deferred.succeed(snapshotTaken, undefined);
253 // Holding the writer lock here would deadlock this join.
254 yield* Fiber.join(writer);
255 const after = yield* countEntries(sql);
256 return { before, after };
257 }),
258 )
259 .pipe(SqliteClient.readOnly);
260
261 // The reader keeps the snapshot it began with, and later reads see the commit.
262 assert.deepEqual(counts, { before: 1, after: 1 });
263 assert.equal(yield* countEntries(sql).pipe(SqliteClient.readOnly), 2);
264 }),
265 ),
266);
267
268it.effect("rejects a write in a read-only transaction", () =>
269 withReaderClient((sql) =>
270 Effect.gen(function* () {
271 const error = yield* sql
272 .withTransaction(sql`INSERT INTO entries VALUES ('blocked')`)
273 .pipe(SqliteClient.readOnly, Effect.flip);
274
275 assert.equal(error._tag, "SqlError");
276 assert.include(String(error.reason.cause), "readonly database");
277 assert.equal(yield* countEntries(sql), 1);
278 }),
279 ),
280);
281
282it.effect("keeps the reader read-only even if a statement turns query_only off", () =>
283 withReaderClient((sql) =>
284 Effect.gen(function* () {
285 const error = yield* sql
286 .withTransaction(
287 Effect.gen(function* () {
288 yield* sql`PRAGMA query_only = OFF`;
289 yield* sql`INSERT INTO entries VALUES ('blocked')`;
290 }),
291 )
292 .pipe(SqliteClient.readOnly, Effect.flip);
293
294 assert.equal(error._tag, "SqlError");
295 assert.include(String(error.reason.cause), "readonly database");
296 assert.equal(yield* countEntries(sql), 1);
297 }),
298 ),
299);
300
301it.effect("keeps read-only work inside a write transaction on the writer", () =>
302 withReaderClient((sql) =>
303 sql.withTransaction(
304 Effect.gen(function* () {
305 yield* sql`INSERT INTO entries VALUES ('uncommitted')`;
306 // Only the writer can see its own uncommitted row.
307 const count = yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly);
308 assert.equal(count, 2);
309 }),
310 ),
311 ),
312);
313
314it.effect("fails a read-only transaction whose reader stops, then starts a new reader", () =>
315 withReaderClient((sql) =>
316 Effect.gen(function* () {
317 const startedBefore = startedWorkers.length;
318 const error = yield* sql
319 .withTransaction(
320 Effect.gen(function* () {
321 yield* countEntries(sql);
322 yield* Effect.promise(() => startedWorkers.at(-1)!.terminate());
323 return yield* countEntries(sql);
324 }),
325 )
326 .pipe(SqliteClient.readOnly, Effect.flip);
327
328 assert.equal(error._tag, "SqlError");
329 assert.include(String(error.reason.cause), "Database reader exited");
330 assert.equal(yield* countEntries(sql).pipe(SqliteClient.readOnly), 1);
331 assert.equal(startedWorkers.length, startedBefore + 2);
332 }),
333 ),
334);
335
336it.effect("uses the writer for read-only work on an in-memory database", () =>
337 Effect.gen(function* () {
338 const startedBefore = startedWorkers.length;
339 const sql = yield* SqlClient.SqlClient;
340 yield* sql`CREATE TABLE memory_entries(name TEXT NOT NULL)`;
341 yield* sql`INSERT INTO memory_entries VALUES ('kept')`;
342
343 const rows = yield* sql
344 .withTransaction(sql<{ name: string }>`SELECT name FROM memory_entries`)
345 .pipe(SqliteClient.readOnly);
346
347 assert.deepEqual(rows, [{ name: "kept" }]);
348 assert.equal(startedWorkers.length, startedBefore);
349 }).pipe(Effect.provide(SqliteClient.layer({ filename: ":memory:", readerWorker: true }))),
350);
351
352it.effect("keeps a nested transaction inside a read-only transaction on the reader", () =>
353 withReaderClient((sql) =>
354 Effect.gen(function* () {
355 const snapshotTaken = yield* Deferred.make<void>();
356 const writer = yield* Effect.forkChild(
357 Deferred.await(snapshotTaken).pipe(
358 Effect.andThen(sql`INSERT INTO entries VALUES ('second')`),
359 ),
360 );
361 // Callers wrap a store read that opens its own transaction, as the
362 // shell snapshot routes do around getShellSnapshot.
363 const counts = yield* sql
364 .withTransaction(
365 Effect.gen(function* () {
366 const outer = yield* countEntries(sql);
367 yield* Deferred.succeed(snapshotTaken, undefined);
368 yield* Fiber.join(writer);
369 const nested = yield* sql.withTransaction(countEntries(sql));
370 return { outer, nested };
371 }),
372 )
373 .pipe(SqliteClient.readOnly);
374
375 assert.deepEqual(counts, { outer: 1, nested: 1 });
376 }),
377 ),
378);
379
380const withFileClient = <A, E>(
381 setup: (
382 sql: SqlClient.SqlClient,
383 filename: string,
384 ) => Effect.Effect<A, E, SqlClient.SqlClient | FileSystem.FileSystem>,
385) =>
386 Effect.gen(function* () {
387 const fs = yield* FileSystem.FileSystem;
388 const path = yield* Path.Path;
389 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-reader-off-" });
390 const filename = path.join(directory, "state.sqlite");
391 return yield* Effect.gen(function* () {
392 return yield* setup(yield* SqlClient.SqlClient, filename);
393 }).pipe(Effect.provide(SqliteClient.layer({ filename, readerWorker: true })));
394 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer));
395
396it.effect("uses the writer when the reader cannot open the database", () =>
397 withFileClient((sql, filename) =>
398 Effect.gen(function* () {
399 yield* sql`PRAGMA journal_mode = WAL`;
400 yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`;
401 yield* sql`INSERT INTO entries VALUES ('kept')`;
402 // The writer keeps its open file; a read-only open of the path now fails.
403 yield* (yield* FileSystem.FileSystem).remove(filename);
404 const startedBefore = startedWorkers.length;
405
406 assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1);
407 assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1);
408 assert.equal(startedWorkers.length, startedBefore + 1);
409 }),
410 ),
411);
412
413it.effect("uses the writer for a database that is not in WAL mode", () =>
414 withFileClient((sql) =>
415 Effect.gen(function* () {
416 yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`;
417 const startedBefore = startedWorkers.length;
418
419 // A write inside readOnly only succeeds on the writer: the reader is read-only.
420 yield* sql
421 .withTransaction(sql`INSERT INTO entries VALUES ('on the writer')`)
422 .pipe(SqliteClient.readOnly);
423 assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1);
424 assert.equal(startedWorkers.length, startedBefore + 1);
425 }),
426 ),
427);
428
429const captureWarnings = () => {
430 const warnings: Array<unknown> = [];
431 const logger = Logger.make(({ logLevel, message }) => {
432 if (logLevel === "Warn") warnings.push(message);
433 });
434 return { warnings, layer: Logger.layer([logger], { mergeWithExisting: false }) };
435};
436
437const silently = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
438 Effect.acquireUseRelease(
439 Effect.sync(() => {
440 silentWorkers.on = true;
441 }),
442 () => effect,
443 () =>
444 Effect.sync(() => {
445 silentWorkers.on = false;
446 silentWorkers.readyOnTerminate = false;
447 }),
448 );
449
450it.effect("turns the reader off when its worker never opens the database", () => {
451 const { warnings, layer: logs } = captureWarnings();
452 return withFileClient((sql) =>
453 Effect.gen(function* () {
454 yield* sql`PRAGMA journal_mode = WAL`;
455 yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`;
456 yield* sql`INSERT INTO entries VALUES ('kept')`;
457 const read = yield* silently(
458 Effect.gen(function* () {
459 const fiber = yield* Effect.forkChild(
460 sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly),
461 );
462 yield* Effect.yieldNow;
463 yield* TestClock.adjust("10 seconds");
464 return yield* Fiber.join(fiber);
465 }),
466 );
467
468 assert.equal(read, 1);
469 assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1);
470 assert.lengthOf(warnings, 1);
471 }),
472 ).pipe(Effect.provide(logs));
473});
474
475it.effect("lets a read be cancelled while its worker is still starting", () =>
476 withFileClient((sql) =>
477 Effect.gen(function* () {
478 yield* sql`PRAGMA journal_mode = WAL`;
479 yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`;
480 yield* silently(
481 Effect.gen(function* () {
482 const fiber = yield* Effect.forkChild(
483 sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly),
484 );
485 yield* Effect.yieldNow;
486 // Never returns if the wait for the worker were uninterruptible: the
487 // test clock has not moved, so the startup deadline cannot fire.
488 yield* Fiber.interrupt(fiber);
489 }),
490 );
491
492 // The client stays usable: the next read waits out the deadline, then uses the writer.
493 const next = yield* Effect.forkChild(
494 sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly),
495 );
496 yield* Effect.yieldNow;
497 yield* TestClock.adjust("10 seconds");
498 assert.equal(yield* Fiber.join(next), 0);
499 }),
500 ),
501);
502
503it.effect("logs one warning when concurrent first reads find the reader unavailable", () => {
504 const { warnings, layer: logs } = captureWarnings();
505 return withFileClient((sql, filename) =>
506 Effect.gen(function* () {
507 yield* sql`PRAGMA journal_mode = WAL`;
508 yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`;
509 yield* (yield* FileSystem.FileSystem).remove(filename);
510 const startedBefore = startedWorkers.length;
511
512 const counts = yield* Effect.all(
513 [
514 sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly),
515 sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly),
516 ],
517 { concurrency: "unbounded" },
518 );
519
520 assert.deepEqual(counts, [0, 0]);
521 assert.equal(startedWorkers.length, startedBefore + 1);
522 assert.lengthOf(warnings, 1);
523 }),
524 ).pipe(Effect.provide(logs));
525});
526
527it.effect("does not hand out a worker that reports ready after another read timed it out", () =>
528 withFileClient((sql) =>
529 Effect.gen(function* () {
530 yield* sql`PRAGMA journal_mode = WAL`;
531 yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`;
532 yield* sql`INSERT INTO entries VALUES ('kept')`;
533 const read = () => sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly);
534 const counts = yield* silently(
535 Effect.gen(function* () {
536 const first = yield* Effect.forkChild(read());
537 yield* Effect.yieldNow;
538 yield* TestClock.adjust("5 seconds");
539 const second = yield* Effect.forkChild(read());
540 yield* Effect.yieldNow;
541 // The first read reaches its deadline and retires the worker; the second
542 // is still waiting when the worker's "ready" arrives, before it exits.
543 silentWorkers.readyOnTerminate = true;
544 yield* TestClock.adjust("5 seconds");
545 return [yield* Fiber.join(first), yield* Fiber.join(second)];
546 }),
547 );
548
549 assert.deepEqual(counts, [1, 1]);
550 }),
551 ),
552);
packages/shared/src/nodeSqliteClient.ts+360-9
......@@ -5,6 +5,7 @@
55 * @module SqliteClient
66 */
77import * as NodeSqlite from "node:sqlite";
8import * as NodeWorkerThreads from "node:worker_threads";
89
910import * as Cache from "effect/Cache";
1011import * as Duration from "effect/Duration";
......@@ -13,6 +14,7 @@ import * as Exit from "effect/Exit";
1314import * as Fiber from "effect/Fiber";
1415import { identity } from "effect/Function";
1516import * as Layer from "effect/Layer";
17import * as Option from "effect/Option";
1618import * as Predicate from "effect/Predicate";
1719import * as Schema from "effect/Schema";
1820import * as Scope from "effect/Scope";
......@@ -36,8 +38,33 @@ export interface SqliteClientConfig {
3638 readonly spanAttributes?: Record<string, unknown> | undefined;
3739 readonly transformResultNames?: ((str: string) => string) | undefined;
3840 readonly transformQueryNames?: ((str: string) => string) | undefined;
41 /**
42 * Run transactions marked with `readOnly` on a second connection in a worker
43 * thread, so long reads block neither the event loop nor writes. Ignored for
44 * in-memory databases, which a second connection cannot share.
45 */
46 readonly readerWorker?: boolean | undefined;
47 /** How long a reader worker may take to open the database before reads use the writer. Default 10 seconds. */
48 readonly readerStartupTimeout?: Duration.Input | undefined;
3949}
4050
51/**
52 * Set by `readOnly`. A top-level transaction started while it is true runs on
53 * the reader worker when the client has one, and on the writer otherwise.
54 */
55const ReadOnly = Context.Reference<boolean>("@t3tools/shared/nodeSqliteClient/ReadOnly", {
56 defaultValue: () => false,
57});
58
59/**
60 * Marks the SQL in `effect` as read-only. Wrap the whole transaction, as in
61 * `sql.withTransaction(reads).pipe(readOnly)`. Inside an open write
62 * transaction the reads keep using the writer, and a write attempted on the
63 * reader fails with "attempt to write a readonly database".
64 */
65export const readOnly = <A, E, R>(effect: Effect.Effect<A, E, R>): Effect.Effect<A, E, R> =>
66 Effect.provideService(effect, ReadOnly, true);
67
4168export class UnsupportedNodeSqliteVersionError extends Schema.TaggedError<UnsupportedNodeSqliteVersionError>()(
4269 "UnsupportedNodeSqliteVersionError",
4370 {
......@@ -99,6 +126,306 @@ const classifyError = (cause: unknown, message: string, operation: string) => {
99126 return classifySqliteError(cause, { message, operation });
100127};
101128
129// Evaluated source instead of a worker file, so the reader also starts from
130// bundles and single-executable builds that have no module file on disk.
131const readerWorkerSource = `
132const { parentPort, workerData } = require("node:worker_threads");
133const { DatabaseSync } = require("node:sqlite");
134let db;
135try {
136 // A read-only open cannot be undone by SQL, unlike PRAGMA query_only.
137 const next = new DatabaseSync(workerData.filename, {
138 readOnly: true,
139 allowExtension: workerData.allowExtension,
140 });
141 next.exec("PRAGMA busy_timeout = 5000;");
142 // Without WAL a reader's shared lock blocks the writer's commits, so refuse.
143 const mode = String(next.prepare("PRAGMA journal_mode").get().journal_mode).toLowerCase();
144 if (mode !== "wal") throw new Error("database uses journal_mode " + mode + ", not wal");
145 db = next;
146 parentPort.postMessage({ type: "ready" });
147} catch (error) {
148 parentPort.postMessage({ type: "unavailable", message: String(error && error.message ? error.message : error) });
149}
150const statements = new Map();
151const prepare = (sql, cached) => {
152 if (!cached) return db.prepare(sql);
153 let statement = statements.get(sql);
154 if (statement === undefined) {
155 statement = db.prepare(sql);
156 if (statements.size >= workerData.prepareCacheSize) statements.delete(statements.keys().next().value);
157 } else {
158 statements.delete(sql);
159 }
160 statements.set(sql, statement);
161 return statement;
162};
163parentPort.on("message", ({ id, sql, params, cached, values, raw, safeIntegers }) => {
164 let operation = "open";
165 try {
166 if (db === undefined) throw new Error("database reader is unavailable");
167 operation = "prepare";
168 const statement = prepare(sql, cached);
169 operation = "execute";
170 statement.setReadBigInts(safeIntegers);
171 if (statement.columns().length > 0) {
172 statement.setReturnArrays(values);
173 parentPort.postMessage({ id, rows: statement.all(...params) });
174 } else {
175 const result = statement.run(...params);
176 parentPort.postMessage({ id, rows: raw ? result : [] });
177 }
178 } catch (error) {
179 parentPort.postMessage({
180 id,
181 error: {
182 operation,
183 message: String(error && error.message ? error.message : error),
184 code: error && error.code,
185 errcode: error && error.errcode,
186 errstr: error && error.errstr,
187 },
188 });
189 }
190});
191`;
192
193interface ReaderRequest {
194 readonly sql: string;
195 readonly params: ReadonlyArray<unknown>;
196 readonly cached: boolean;
197 readonly values: boolean;
198 readonly raw: boolean;
199}
200
201type ReaderStartup =
202 | { readonly type: "ready" }
203 | { readonly type: "unavailable"; readonly message: string };
204
205type ReaderReply =
206 | { readonly id: number; readonly rows: ReadonlyArray<any> }
207 | {
208 readonly id: number;
209 readonly error: {
210 readonly operation: "open" | "prepare" | "execute";
211 readonly message: string;
212 readonly code?: unknown;
213 readonly errcode?: unknown;
214 readonly errstr?: unknown;
215 };
216 };
217
218const readerFailureMessages = {
219 open: "Failed to open database",
220 prepare: "Failed to prepare statement",
221 execute: "Failed to execute statement",
222} as const;
223
224const failed = (cause: unknown, message: string, operation: string) =>
225 new SqlError({ reason: classifyError(cause, message, operation) });
226
227// COMMIT, ROLLBACK and ROLLBACK TO SAVEPOINT. On a reader they change nothing,
228// so they succeed once the worker is gone instead of failing the cleanup.
229const endsReadTransaction = (sql: string) => /^\s*ROLLBACK\b|^\s*COMMIT\b/i.test(sql);
230
231/**
232 * Read-only connections served by one worker thread at a time. Each worker
233 * gets its own `Connection`, so a transaction stays on the worker it began on:
234 * once that worker exits, its statements fail rather than silently continuing
235 * outside the transaction on a replacement. The next acquire starts a new one.
236 * A worker that cannot open the database read-only, or finds it outside WAL
237 * mode, turns the reader off for the life of the client: reads use the writer.
238 */
239const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) {
240 let current:
241 | {
242 readonly worker: NodeWorkerThreads.Worker;
243 readonly connection: Connection;
244 readonly started: Promise<ReaderStartup>;
245 }
246 | undefined;
247 let unavailable = false;
248 let nextRequestId = 0;
249
250 const start = () => {
251 const worker = new NodeWorkerThreads.Worker(readerWorkerSource, {
252 eval: true,
253 workerData: {
254 filename: options.filename,
255 allowExtension: options.allowExtension ?? false,
256 prepareCacheSize: options.prepareCacheSize ?? 200,
257 },
258 });
259 let announce: (startup: ReaderStartup) => void = () => {};
260 const started = new Promise<ReaderStartup>((resolve) => {
261 announce = resolve;
262 });
263 let stoppedMessage: string | undefined;
264 const pending = new Map<
265 number,
266 { readonly endsTransaction: boolean; readonly resume: (reply: ReaderReply) => void }
267 >();
268 const stop = (message: string) => {
269 stoppedMessage = message;
270 announce({ type: "unavailable", message });
271 if (current?.worker === worker) current = undefined;
272 for (const [id, request] of pending) {
273 request.resume(
274 request.endsTransaction
275 ? { id, rows: [] }
276 : { id, error: { operation: "execute", message } },
277 );
278 }
279 pending.clear();
280 };
281 worker.on("message", (reply: ReaderReply | ReaderStartup) => {
282 if ("type" in reply) {
283 announce(reply);
284 // An idle reader must not keep the process alive.
285 if (pending.size === 0) worker.unref();
286 return;
287 }
288 const request = pending.get(reply.id);
289 if (request === undefined) return;
290 pending.delete(reply.id);
291 if (pending.size === 0) worker.unref();
292 request.resume(reply);
293 });
294 worker.on("error", (cause) => stop(`Database reader failed: ${cause.message}`));
295 worker.on("exit", (code) => stop(`Database reader exited with code ${code}`));
296
297 const request = (input: ReaderRequest) =>
298 Effect.withFiber<ReadonlyArray<any>, SqlError>((fiber) => {
299 const endsTransaction = endsReadTransaction(input.sql);
300 if (stoppedMessage !== undefined) {
301 return endsTransaction
302 ? Effect.succeed([])
303 : Effect.fail(
304 failed(new Error(stoppedMessage), readerFailureMessages.execute, "execute"),
305 );
306 }
307 const safeIntegers = Boolean(Context.get(fiber.context, Client.SafeIntegers));
308 return Effect.callback<ReadonlyArray<any>, SqlError>((resume) => {
309 const id = nextRequestId++;
310 pending.set(id, {
311 endsTransaction,
312 resume: (reply) => {
313 if ("rows" in reply) {
314 resume(Effect.succeed(reply.rows));
315 return;
316 }
317 const { operation, message, ...codes } = reply.error;
318 resume(
319 Effect.fail(
320 failed(
321 Object.assign(new Error(message), codes),
322 readerFailureMessages[operation],
323 operation,
324 ),
325 ),
326 );
327 },
328 });
329 if (pending.size === 1) worker.ref();
330 // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin.
331 worker.postMessage({ id, safeIntegers, ...input });
332 });
333 });
334
335 const connection = identity<Connection>({
336 execute(sql, params, rowTransform) {
337 const effect = request({ sql, params, cached: true, values: false, raw: false });
338 return rowTransform ? Effect.map(effect, rowTransform) : effect;
339 },
340 executeRaw(sql, params) {
341 return request({ sql, params, cached: true, values: false, raw: true });
342 },
343 executeValues(sql, params) {
344 return request({ sql, params, cached: true, values: true, raw: false });
345 },
346 executeValuesUnprepared(sql, params) {
347 return request({ sql, params: params ?? [], cached: false, values: true, raw: false });
348 },
349 executeUnprepared(sql, params, rowTransform) {
350 // A read-only connection rejects BEGIN IMMEDIATE and BEGIN EXCLUSIVE,
351 // and a deferred BEGIN is all a read snapshot needs.
352 const statement = /^\s*BEGIN\b/i.test(sql) ? "BEGIN" : sql;
353 const effect = request({
354 sql: statement,
355 params: params ?? [],
356 cached: false,
357 values: false,
358 raw: false,
359 });
360 return rowTransform ? Effect.map(effect, rowTransform) : effect;
361 },
362 executeStream(_sql, _params) {
363 return Stream.die(new UnsupportedNodeSqliteOperationError());
364 },
365 });
366 current = { worker, connection, started };
367 return current;
368 };
369
370 yield* Effect.addFinalizer(() =>
371 Effect.promise(async () => {
372 await current?.worker.terminate();
373 }),
374 );
375
376 const startupTimeout = Duration.fromInputUnsafe(
377 options.readerStartupTimeout ?? Duration.seconds(10),
378 );
379
380 // Runs once: concurrent first acquires share one startup and may all see it fail.
381 const turnOff = (reader: typeof current, reason: string) =>
382 Effect.suspend(() => {
383 if (reader !== undefined) {
384 if (current === reader) current = undefined;
385 void reader.worker.terminate();
386 }
387 if (unavailable) return Effect.succeed(undefined);
388 unavailable = true;
389 return Effect.logWarning("SQLite reader worker unavailable; reads use the main connection", {
390 reason,
391 }).pipe(Effect.as(undefined));
392 });
393
394 /** The live worker's connection once it is ready, or undefined to use the writer. */
395 return Effect.suspend((): Effect.Effect<Connection | undefined> => {
396 if (unavailable) return Effect.succeed(undefined);
397 let reader: NonNullable<typeof current>;
398 try {
399 reader = current ?? start();
400 } catch (cause) {
401 return turnOff(undefined, String(cause));
402 }
403 // SqlClient acquires transaction connections uninterruptibly. Waiting for the
404 // worker happens before any lock is taken, so let a cancelled request leave.
405 return Effect.flatMap(
406 Effect.interruptible(
407 Effect.timeoutOption(
408 Effect.promise(() => reader.started),
409 startupTimeout,
410 ),
411 ),
412 (startup): Effect.Effect<Connection | undefined> =>
413 // Another acquire may have timed out and retired this worker while this one
414 // waited; a late "ready" must not hand out a worker that is being terminated.
415 unavailable || current !== reader
416 ? Effect.succeed(undefined)
417 : Option.isSome(startup) && startup.value.type === "ready"
418 ? Effect.succeed(reader.connection)
419 : turnOff(
420 reader,
421 Option.isSome(startup) && startup.value.type === "unavailable"
422 ? startup.value.message
423 : `worker did not open the database within ${Duration.toMillis(startupTimeout)} ms`,
424 ),
425 );
426 });
427});
428
102429const make = Effect.fn("makeWithDatabase")(function* (
103430 options: SqliteClientConfig,
104431): Effect.fn.Return<Client.SqlClient, SqlError, Scope.Scope | Reactivity.Reactivity> {
......@@ -261,16 +588,40 @@ const make = Effect.fn("makeWithDatabase")(function* (
261588
262589 const semaphore = yield* Semaphore.make(1);
263590 const connection = yield* makeConnection;
591 const acquireReader =
592 options.readerWorker && options.filename !== ":memory:" && options.filename !== ""
593 ? yield* makeReader(options)
594 : undefined;
595 const readerSemaphore = yield* Semaphore.make(1);
264596
265 const acquirer = semaphore.withPermits(1)(Effect.succeed(connection));
266 const transactionAcquirer = Effect.uninterruptibleMask((restore) => {
267 const fiber = Fiber.getCurrent()!;
268 const scope = Context.getUnsafe(fiber.context, Scope.Scope);
269 return Effect.as(
270 Effect.tap(restore(semaphore.take(1)), () => Scope.addFinalizer(scope, semaphore.release(1))),
271 connection,
272 );
273 });
597 const writer = { lock: semaphore, connection };
598 // Picks the reader only for `readOnly` work, and the writer whenever the
599 // reader is turned off.
600 const route = Effect.withFiber((fiber) =>
601 acquireReader !== undefined && Context.get(fiber.context, ReadOnly)
602 ? Effect.map(acquireReader, (readerConnection) =>
603 readerConnection === undefined
604 ? writer
605 : { lock: readerSemaphore, connection: readerConnection },
606 )
607 : Effect.succeed(writer),
608 );
609
610 const acquirer = Effect.flatMap(route, (target) =>
611 target.lock.withPermits(1)(Effect.succeed(target.connection)),
612 );
613 const transactionAcquirer = Effect.flatMap(route, (target) =>
614 Effect.uninterruptibleMask((restore) => {
615 const fiber = Fiber.getCurrent()!;
616 const scope = Context.getUnsafe(fiber.context, Scope.Scope);
617 return Effect.as(
618 Effect.tap(restore(target.lock.take(1)), () =>
619 Scope.addFinalizer(scope, target.lock.release(1)),
620 ),
621 target.connection,
622 );
623 }),
624 );
274625
275626 return yield* Client.make({
276627 acquirer,