1import * as NodeSqlite from "node:sqlite";
2import type * as NodeWorkerThreads from "node:worker_threads";
3import * as NodeServices from "@effect/platform-node/NodeServices";
4import { assert, it } from "@effect/vitest";
5import * as Deferred from "effect/Deferred";
6import * as Effect from "effect/Effect";
7import * as Fiber from "effect/Fiber";
8import * as FileSystem from "effect/FileSystem";
9import * as Layer from "effect/Layer";
10import * as Logger from "effect/Logger";
11import * as Path from "effect/Path";
12import * as SqlClient from "effect/sql/SqlClient";
13import * as Statement from "effect/sql/Statement";
14import * as TestClock from "effect/testing/TestClock";
15import * as Tracer from "effect/Tracer";
16import { vi } from "vite-plus/test";
17
18import * as SqliteClient from "./nodeSqliteClient.ts";
19
20const startedWorkers = vi.hoisted((): Array<NodeWorkerThreads.Worker> => []);
21// When on, workers run a script that never announces, like a worker stuck opening the file.
22const silentWorkers = vi.hoisted(() => ({ on: false, readyOnTerminate: false }));
23// When set, workers throw this at startup, like a checkpoint worker on SQLite before 3.51.
24const failingWorkers = vi.hoisted(() => ({ message: undefined as string | undefined }));
25
26vi.mock("node:worker_threads", async (importOriginal) => {
27 const actual = await importOriginal<typeof NodeWorkerThreads>();
28 class RecordedWorker extends actual.Worker {
29 constructor(...args: ConstructorParameters<typeof actual.Worker>) {
30 super(
31 ...((silentWorkers.on
32 ? ["setInterval(() => {}, 1 << 30);", { eval: true }]
33 : failingWorkers.message !== undefined
34 ? [`throw new Error(${JSON.stringify(failingWorkers.message)});`, { eval: true }]
35 : args) as ConstructorParameters<typeof actual.Worker>),
36 );
37 startedWorkers.push(this);
38 }
39 override terminate() {
40 // A "ready" already queued when termination starts reaches the parent before "exit".
41 if (silentWorkers.readyOnTerminate) this.emit("message", { type: "ready" });
42 return super.terminate();
43 }
44 }
45 return { ...actual, Worker: RecordedWorker };
46});
47
48const withReaderClient = <A, E>(
49 use: (sql: SqlClient.SqlClient) => Effect.Effect<A, E, SqlClient.SqlClient>,
50) =>
51 Effect.gen(function* () {
52 const fs = yield* FileSystem.FileSystem;
53 const path = yield* Path.Path;
54 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-reader-" });
55 const filename = path.join(directory, "state.sqlite");
56 return yield* Effect.gen(function* () {
57 const sql = yield* SqlClient.SqlClient;
58 yield* sql`PRAGMA journal_mode = WAL`;
59 yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`;
60 yield* sql`INSERT INTO entries VALUES ('first')`;
61 return yield* use(sql);
62 }).pipe(Effect.provide(SqliteClient.layer({ filename, readerWorker: true })));
63 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer));
64
65const countEntries = (sql: SqlClient.SqlClient) =>
66 Effect.map(
67 sql<{ count: number }>`SELECT COUNT(*) AS count FROM entries`,
68 (rows) => rows[0]?.count,
69 );
70
71const layer = it.layer(SqliteClient.layer({ filename: ":memory:" }));
72
73layer("NodeSqliteClient", (it) => {
74 it.effect("retries preparing a query after the missing schema becomes available", () =>
75 Effect.gen(function* () {
76 const sql = yield* SqlClient.SqlClient;
77 const select = sql<{ name: string }>`SELECT name FROM created_after_prepare_failure`;
78 const error = yield* select.pipe(Effect.flip);
79 assert.equal(error._tag, "SqlError");
80 assert.equal(error.reason.operation, "prepare");
81
82 yield* sql`CREATE TABLE created_after_prepare_failure(name TEXT NOT NULL)`;
83 yield* sql`INSERT INTO created_after_prepare_failure VALUES ('recovered')`;
84 assert.deepEqual(yield* select, [{ name: "recovered" }]);
85 assert.deepEqual(yield* select.values, [["recovered"]]);
86 }),
87 );
88
89 it.effect("runs prepared queries and returns positional values", () =>
90 Effect.gen(function* () {
91 const sql = yield* SqlClient.SqlClient;
92
93 yield* sql`CREATE TABLE entries(id INTEGER PRIMARY KEY, name TEXT NOT NULL)`;
94 yield* sql`INSERT INTO entries(name) VALUES (${"alpha"}), (${"beta"})`;
95
96 const rows = yield* sql<{ readonly id: number; readonly name: string }>`
97 SELECT id, name FROM entries ORDER BY id
98 `;
99 assert.equal(rows.length, 2);
100 assert.equal(rows[0]?.name, "alpha");
101 assert.equal(rows[1]?.name, "beta");
102
103 const values = yield* sql`SELECT id, name FROM entries ORDER BY id`.values;
104 assert.equal(values.length, 2);
105 assert.equal(values[0]?.[1], "alpha");
106 assert.equal(values[1]?.[1], "beta");
107
108 const unpreparedValues = yield* sql`SELECT id, name FROM entries ORDER BY id`
109 .valuesUnprepared;
110 assert.deepEqual(unpreparedValues, values);
111 }),
112 );
113
114 it.effect("returns a typed failure when an unprepared statement cannot be prepared", () =>
115 Effect.gen(function* () {
116 const sql = yield* SqlClient.SqlClient;
117 const error = yield* Effect.flip(sql.unsafe("SELECT FROM").unprepared);
118
119 assert.equal(error._tag, "SqlError");
120 assert.equal(error.reason.operation, "prepare");
121 }),
122 );
123
124 it.effect("classifies constraint failures by their SQLite result code", () =>
125 Effect.gen(function* () {
126 const sql = yield* SqlClient.SqlClient;
127 yield* sql`CREATE TABLE constrained(name TEXT NOT NULL UNIQUE)`;
128 yield* sql`INSERT INTO constrained VALUES ('taken')`;
129
130 const duplicate = yield* sql`INSERT INTO constrained VALUES ('taken')`.pipe(Effect.flip);
131 assert(duplicate.reason._tag === "UniqueViolation");
132 assert.equal(duplicate.reason.constraint, "constrained.name");
133
134 const missing = yield* sql`INSERT INTO constrained VALUES (NULL)`.pipe(Effect.flip);
135 assert.equal(missing.reason._tag, "ConstraintError");
136 }),
137 );
138});
139
140const largeText = "x".repeat(64 * 1024 + 1);
141
142it.effect("keeps large text and blob parameters off cached statements", () =>
143 Effect.gen(function* () {
144 const sql = yield* SqlClient.SqlClient;
145 const statements: Array<NodeSqlite.StatementSync> = [];
146 const all = NodeSqlite.StatementSync.prototype.all;
147 yield* Effect.acquireRelease(
148 Effect.sync(() =>
149 vi.spyOn(NodeSqlite.StatementSync.prototype, "all").mockImplementation(function (
150 this: NodeSqlite.StatementSync,
151 ...params
152 ) {
153 statements.push(this);
154 return all.apply(this, params);
155 }),
156 ),
157 (spy) => Effect.sync(() => spy.mockRestore()),
158 );
159
160 const small = "small";
161 const queries = [
162 (value: string | Uint8Array | ArrayBuffer) => sql`SELECT length(${value}) AS size`,
163 (value: string | Uint8Array | ArrayBuffer) =>
164 sql.unsafe("SELECT length($value) AS size", [{ $value: value }]),
165 (value: string | Uint8Array | ArrayBuffer) =>
166 sql.unsafe("SELECT length($value) AS callable", [
167 Object.assign(() => {}, { $value: value }),
168 ]),
169 ];
170 const cached = new Set<NodeSqlite.StatementSync>();
171 const large = new Set<NodeSqlite.StatementSync>();
172 const largeBlobs = [new Uint8Array(64 * 1024 + 1), new ArrayBuffer(64 * 1024 + 1)];
173 for (const value of [small, largeText, ...largeBlobs, small]) {
174 const expected = typeof value === "string" ? value.length : value.byteLength;
175 for (const query of queries) {
176 assert.deepEqual((yield* query(value)).map(Object.values), [[expected]]);
177 assert.deepEqual(yield* query(value).values, [[expected]]);
178 for (const statement of statements.splice(0)) {
179 (value === small ? cached : large).add(statement);
180 }
181 }
182 }
183
184 // Small parameters keep reusing the one cached statement for each query.
185 assert.equal(cached.size, queries.length);
186 // Each large execution gets a statement that nothing holds on to afterwards.
187 assert.equal(large.size, queries.length * 6);
188 assert.isTrue(cached.isDisjointFrom(large));
189 }).pipe(Effect.provide(SqliteClient.layer({ filename: ":memory:" }))),
190);
191
192it.effect("reads integers on the values path as the caller asked, cached or not", () =>
193 Effect.gen(function* () {
194 const sql = yield* SqlClient.SqlClient;
195 for (const value of ["small", largeText]) {
196 const values = sql`SELECT 1 AS id, length(${value}) AS size`.values;
197 assert.deepEqual(yield* values.pipe(Effect.provideService(SqlClient.SafeIntegers, true)), [
198 [1n, BigInt(value.length)],
199 ]);
200 assert.deepEqual(yield* values, [[1, value.length]]);
201 }
202 }).pipe(Effect.provide(SqliteClient.layer({ filename: ":memory:" }))),
203);
204
205it.effect("leaves a parameter it cannot measure for the statement to handle", () =>
206 Effect.gen(function* () {
207 const sql = yield* SqlClient.SqlClient;
208 const detached = new DataView(new ArrayBuffer(8));
209 structuredClone(detached.buffer, { transfer: [detached.buffer] });
210 // What a detached view binds as varies by Node version; match the statement itself.
211 const native = new NodeSqlite.DatabaseSync(":memory:");
212 const expected = native.prepare("SELECT length(?) AS size").all(detached);
213 native.close();
214 assert.deepEqual(yield* sql`SELECT length(${detached}) AS size`, expected);
215
216 const throwing = {
217 get $value(): string {
218 throw new Error("unreadable");
219 },
220 };
221 const error = yield* sql.unsafe("SELECT length($value) AS size", [throwing]).pipe(Effect.flip);
222 assert.equal(error.reason.operation, "execute");
223
224 // Only the statement reads a getter, so it runs once as it would natively.
225 let reads = 0;
226 const counting = {
227 get $value(): number {
228 return ++reads;
229 },
230 };
231 assert.deepEqual(yield* sql.unsafe("SELECT $value AS value", [counting]), [{ value: 1 }]);
232 assert.equal(reads, 1);
233 }).pipe(Effect.provide(SqliteClient.layer({ filename: ":memory:" }))),
234);
235
236const makeTempDatabase = Effect.gen(function* () {
237 const fs = yield* FileSystem.FileSystem;
238 const path = yield* Path.Path;
239 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-transaction-" });
240 const filename = path.join(directory, "state.sqlite");
241 // node:sqlite connections fail a busy statement at once unless given a timeout.
242 const other = yield* Effect.acquireRelease(
243 Effect.sync(() => new NodeSqlite.DatabaseSync(filename)),
244 (database) => Effect.sync(() => database.close()),
245 );
246 yield* Effect.sync(() => {
247 other.exec(`
248 PRAGMA journal_mode = WAL;
249 CREATE TABLE counters(id INTEGER PRIMARY KEY, value INTEGER NOT NULL);
250 INSERT INTO counters VALUES (1, 0);
251 `);
252 });
253 return { filename, other };
254});
255
256it.effect("keeps another connection from committing between a transaction's read and write", () =>
257 Effect.gen(function* () {
258 const { filename, other } = yield* makeTempDatabase;
259 yield* Effect.gen(function* () {
260 const sql = yield* SqlClient.SqlClient;
261 yield* sql.withTransaction(
262 Effect.gen(function* () {
263 const [row] = yield* sql<{ readonly value: number }>`
264 SELECT value FROM counters WHERE id = 1
265 `;
266 // With a deferred BEGIN this commit lands and the write below fails
267 // with SQLITE_BUSY_SNAPSHOT, which no busy timeout can wait out.
268 yield* Effect.sync(() =>
269 assert.throws(
270 () => other.exec("UPDATE counters SET value = value + 1 WHERE id = 1"),
271 /database is locked/,
272 ),
273 );
274 yield* sql`UPDATE counters SET value = ${(row?.value ?? 0) + 10} WHERE id = 1`;
275 }),
276 );
277 yield* Effect.sync(() => other.exec("UPDATE counters SET value = value + 1 WHERE id = 1"));
278 assert.deepEqual(yield* sql`SELECT value FROM counters WHERE id = 1`.values, [[11]]);
279 }).pipe(Effect.provide(SqliteClient.layer({ filename })));
280 }).pipe(Effect.provide(NodeServices.layer)),
281);
282
283it.effect("reports a transaction blocked by another writer as a lock timeout", () =>
284 Effect.gen(function* () {
285 const { filename, other } = yield* makeTempDatabase;
286 yield* Effect.gen(function* () {
287 const sql = yield* SqlClient.SqlClient;
288 yield* sql`PRAGMA busy_timeout = 0`;
289 const read = sql.withTransaction(sql`SELECT value FROM counters WHERE id = 1`.values);
290
291 yield* Effect.sync(() => other.exec("BEGIN IMMEDIATE"));
292 const error = yield* read.pipe(Effect.flip);
293 assert.equal(error.reason._tag, "LockTimeoutError");
294
295 yield* Effect.sync(() => other.exec("COMMIT"));
296 assert.deepEqual(yield* read, [[0]]);
297 }).pipe(Effect.provide(SqliteClient.layer({ filename })));
298 }).pipe(Effect.provide(NodeServices.layer)),
299);
300
301it.effect("returns a typed failure when the database cannot be opened", () =>
302 Effect.gen(function* () {
303 const error = yield* Effect.flip(
304 Layer.build(SqliteClient.layer({ filename: "\0" })).pipe(Effect.scoped),
305 );
306
307 assert.equal(error._tag, "SqlError");
308 assert.equal(error.reason.operation, "open");
309 }),
310);
311
312it.effect(
313 "recovers a prepared query immediately after an exclusive database lock is released",
314 () =>
315 Effect.gen(function* () {
316 const fs = yield* FileSystem.FileSystem;
317 const path = yield* Path.Path;
318 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-prepare-" });
319 const filename = path.join(directory, "state.sqlite");
320 const blocker = yield* Effect.acquireRelease(
321 Effect.sync(() => new NodeSqlite.DatabaseSync(filename)),
322 (database) => Effect.sync(() => database.close()),
323 );
324 yield* Effect.sync(() => {
325 blocker.exec("CREATE TABLE entries(value TEXT); INSERT INTO entries VALUES ('retained')");
326 });
327 yield* Effect.gen(function* () {
328 const sql = yield* SqlClient.SqlClient;
329 yield* Effect.sync(() => blocker.exec("BEGIN EXCLUSIVE"));
330 const select = sql`SELECT value FROM entries`;
331 const error = yield* select.values.pipe(Effect.flip);
332 assert.equal(error._tag, "SqlError");
333 assert.equal(error.reason.operation, "prepare");
334 yield* Effect.sync(() => blocker.exec("ROLLBACK"));
335 assert.deepEqual(yield* select.values, [["retained"]]);
336 assert.deepEqual(yield* select, [{ value: "retained" }]);
337 }).pipe(Effect.provide(SqliteClient.layer({ filename })));
338 }).pipe(Effect.provide(NodeServices.layer)),
339);
340
341it.effect("lets the writer commit while a read-only transaction is open", () =>
342 withReaderClient((sql) =>
343 Effect.gen(function* () {
344 const snapshotTaken = yield* Deferred.make<void>();
345 const writer = yield* Effect.forkChild(
346 Deferred.await(snapshotTaken).pipe(
347 Effect.andThen(sql`INSERT INTO entries VALUES ('second')`),
348 ),
349 );
350 const counts = yield* sql
351 .withTransaction(
352 Effect.gen(function* () {
353 const before = yield* countEntries(sql);
354 yield* Deferred.succeed(snapshotTaken, undefined);
355 // Holding the writer lock here would deadlock this join.
356 yield* Fiber.join(writer);
357 const after = yield* countEntries(sql);
358 return { before, after };
359 }),
360 )
361 .pipe(SqliteClient.readOnly);
362
363 // The reader keeps the snapshot it began with, and later reads see the commit.
364 assert.deepEqual(counts, { before: 1, after: 1 });
365 assert.equal(yield* countEntries(sql).pipe(SqliteClient.readOnly), 2);
366 }),
367 ),
368);
369
370it.effect("rejects a write in a read-only transaction", () =>
371 withReaderClient((sql) =>
372 Effect.gen(function* () {
373 const error = yield* sql
374 .withTransaction(sql`INSERT INTO entries VALUES ('blocked')`)
375 .pipe(SqliteClient.readOnly, Effect.flip);
376
377 assert.equal(error._tag, "SqlError");
378 assert.include(String(error.reason.cause), "readonly database");
379 assert.equal(yield* countEntries(sql), 1);
380 }),
381 ),
382);
383
384it.effect("keeps the reader read-only even if a statement turns query_only off", () =>
385 withReaderClient((sql) =>
386 Effect.gen(function* () {
387 const error = yield* sql
388 .withTransaction(
389 Effect.gen(function* () {
390 yield* sql`PRAGMA query_only = OFF`;
391 yield* sql`INSERT INTO entries VALUES ('blocked')`;
392 }),
393 )
394 .pipe(SqliteClient.readOnly, Effect.flip);
395
396 assert.equal(error._tag, "SqlError");
397 assert.include(String(error.reason.cause), "readonly database");
398 assert.equal(yield* countEntries(sql), 1);
399 }),
400 ),
401);
402
403it.effect("keeps read-only work inside a write transaction on the writer", () =>
404 withReaderClient((sql) =>
405 sql.withTransaction(
406 Effect.gen(function* () {
407 yield* sql`INSERT INTO entries VALUES ('uncommitted')`;
408 // Only the writer can see its own uncommitted row.
409 const count = yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly);
410 assert.equal(count, 2);
411 }),
412 ),
413 ),
414);
415
416it.effect("labels SQL spans with their connection and how long they waited for its lock", () => {
417 const spans: Array<Tracer.NativeSpan> = [];
418 const tracer = Tracer.make({
419 span: (options) => {
420 const span = new Tracer.NativeSpan(options);
421 spans.push(span);
422 return span;
423 },
424 });
425 return withReaderClient((sql) =>
426 Effect.gen(function* () {
427 const held = yield* Deferred.make<void>();
428 const release = yield* Deferred.make<void>();
429 const holder = yield* Effect.forkChild(
430 sql.withTransaction(
431 Deferred.succeed(held, undefined).pipe(Effect.andThen(Deferred.await(release))),
432 ),
433 );
434 yield* Deferred.await(held);
435 // Each waiter runs until it blocks on the writer lock, before the clock moves.
436 const statement = yield* Effect.forkChild(sql`SELECT 1`, { startImmediately: true });
437 const transaction = yield* Effect.forkChild(sql.withTransaction(countEntries(sql)), {
438 startImmediately: true,
439 });
440 yield* TestClock.adjust("25 millis");
441 yield* Deferred.succeed(release, undefined);
442 yield* Fiber.join(holder);
443 yield* Fiber.join(statement);
444 yield* Fiber.join(transaction);
445 yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly);
446 }),
447 ).pipe(
448 // The server turns this on, so a statement's span is current while it waits.
449 Effect.provideService(Statement.SpanPropagationEnabled, true),
450 Effect.withTracer(tracer),
451 Effect.map(() => {
452 // Statements inside a transaction reuse its connection, so only the
453 // statements and transactions that took a lock are labeled.
454 const labeled = spans
455 .filter((span) => span.attributes.has("db.connection"))
456 .map((span) => [
457 span.name,
458 span.attributes.get("db.connection"),
459 span.attributes.get("db.lock_wait_ms"),
460 ]);
461 assert.deepStrictEqual(labeled.slice(3), [
462 ["sql.transaction", "writer", 0],
463 ["sqlite", "writer", 25],
464 ["sql.transaction", "writer", 25],
465 ["sql.transaction", "reader", 0],
466 ]);
467 }),
468 );
469});
470
471it.effect("labels only the span of the statement or transaction that took the lock", () => {
472 const spans: Array<Tracer.NativeSpan> = [];
473 const tracer = Tracer.make({
474 span: (options) => {
475 const span = new Tracer.NativeSpan(options);
476 spans.push(span);
477 return span;
478 },
479 });
480 return Effect.gen(function* () {
481 const sql = yield* SqlClient.SqlClient;
482 const transaction = sql.withTransaction(sql`SELECT 1`);
483 yield* transaction;
484 // Run again nested, the same Effect reuses the outer connection and takes no lock.
485 yield* sql.withTransaction(transaction);
486 // With tracing off, Statement does not make its span current.
487 yield* sql`SELECT 2`.pipe(Effect.withTracerEnabled(false));
488 }).pipe(
489 Effect.withSpan("caller"),
490 Effect.provide(SqliteClient.layer({ filename: ":memory:" })),
491 Effect.provideService(Statement.SpanPropagationEnabled, true),
492 Effect.withTracer(tracer),
493 Effect.map(() => {
494 assert.deepStrictEqual(
495 spans
496 .filter((span) => span.name === "caller" || span.name === "sql.transaction")
497 .map((span) => [span.name, span.attributes.get("db.connection")]),
498 [
499 ["caller", undefined],
500 ["sql.transaction", "writer"],
501 ["sql.transaction", "writer"],
502 ["sql.transaction", undefined],
503 ],
504 );
505 }),
506 );
507});
508
509it.effect("keeps large parameters off the reader worker's cached statements", () =>
510 withReaderClient((sql) =>
511 Effect.gen(function* () {
512 const read = (value: string) =>
513 sql<{ size: number }>`SELECT length(${value}) AS size`.pipe(
514 Effect.map((rows) => rows[0]?.size),
515 );
516 // Start the worker so its requests can be recorded.
517 yield* countEntries(sql).pipe(SqliteClient.readOnly);
518 const worker = startedWorkers.at(-1)!;
519 const cached: Array<boolean> = [];
520 const postMessage = worker.postMessage;
521 yield* Effect.acquireRelease(
522 Effect.sync(() =>
523 vi.spyOn(worker, "postMessage").mockImplementation(function (
524 this: NodeWorkerThreads.Worker,
525 message: { readonly cached: boolean },
526 ) {
527 cached.push(message.cached);
528 return postMessage.call(this, message);
529 }),
530 ),
531 (spy) => Effect.sync(() => spy.mockRestore()),
532 );
533
534 assert.equal(yield* read("small").pipe(SqliteClient.readOnly), 5);
535 assert.equal(yield* read(largeText).pipe(SqliteClient.readOnly), largeText.length);
536 assert.deepEqual(cached, [true, false]);
537 }).pipe(Effect.scoped),
538 ),
539);
540
541it.effect("fails a read-only transaction whose reader stops, then starts a new reader", () =>
542 withReaderClient((sql) =>
543 Effect.gen(function* () {
544 const startedBefore = startedWorkers.length;
545 const error = yield* sql
546 .withTransaction(
547 Effect.gen(function* () {
548 yield* countEntries(sql);
549 yield* Effect.promise(() => startedWorkers.at(-1)!.terminate());
550 return yield* countEntries(sql);
551 }),
552 )
553 .pipe(SqliteClient.readOnly, Effect.flip);
554
555 assert.equal(error._tag, "SqlError");
556 assert.include(String(error.reason.cause), "Database reader exited");
557 assert.equal(yield* countEntries(sql).pipe(SqliteClient.readOnly), 1);
558 assert.equal(startedWorkers.length, startedBefore + 2);
559 }),
560 ),
561);
562
563it.effect("uses the writer for read-only work on an in-memory database", () =>
564 Effect.gen(function* () {
565 const startedBefore = startedWorkers.length;
566 const sql = yield* SqlClient.SqlClient;
567 yield* sql`CREATE TABLE memory_entries(name TEXT NOT NULL)`;
568 yield* sql`INSERT INTO memory_entries VALUES ('kept')`;
569
570 const rows = yield* sql
571 .withTransaction(sql<{ name: string }>`SELECT name FROM memory_entries`)
572 .pipe(SqliteClient.readOnly);
573
574 assert.deepEqual(rows, [{ name: "kept" }]);
575 assert.equal(startedWorkers.length, startedBefore);
576 }).pipe(Effect.provide(SqliteClient.layer({ filename: ":memory:", readerWorker: true }))),
577);
578
579it.effect("keeps a nested transaction inside a read-only transaction on the reader", () =>
580 withReaderClient((sql) =>
581 Effect.gen(function* () {
582 const snapshotTaken = yield* Deferred.make<void>();
583 const writer = yield* Effect.forkChild(
584 Deferred.await(snapshotTaken).pipe(
585 Effect.andThen(sql`INSERT INTO entries VALUES ('second')`),
586 ),
587 );
588 // Callers wrap a store read that opens its own transaction, as the
589 // shell snapshot routes do around getShellSnapshot.
590 const counts = yield* sql
591 .withTransaction(
592 Effect.gen(function* () {
593 const outer = yield* countEntries(sql);
594 yield* Deferred.succeed(snapshotTaken, undefined);
595 yield* Fiber.join(writer);
596 const nested = yield* sql.withTransaction(countEntries(sql));
597 return { outer, nested };
598 }),
599 )
600 .pipe(SqliteClient.readOnly);
601
602 assert.deepEqual(counts, { outer: 1, nested: 1 });
603 }),
604 ),
605);
606
607const withFileClient = <A, E>(
608 setup: (
609 sql: SqlClient.SqlClient,
610 filename: string,
611 ) => Effect.Effect<A, E, SqlClient.SqlClient | FileSystem.FileSystem>,
612) =>
613 Effect.gen(function* () {
614 const fs = yield* FileSystem.FileSystem;
615 const path = yield* Path.Path;
616 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-reader-off-" });
617 const filename = path.join(directory, "state.sqlite");
618 return yield* Effect.gen(function* () {
619 return yield* setup(yield* SqlClient.SqlClient, filename);
620 }).pipe(Effect.provide(SqliteClient.layer({ filename, readerWorker: true })));
621 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer));
622
623it.effect("uses the writer when the reader cannot open the database", () =>
624 withFileClient((sql, filename) =>
625 Effect.gen(function* () {
626 yield* sql`PRAGMA journal_mode = WAL`;
627 yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`;
628 yield* sql`INSERT INTO entries VALUES ('kept')`;
629 // The writer keeps its open file; a read-only open of the path now fails.
630 yield* (yield* FileSystem.FileSystem).remove(filename);
631 const startedBefore = startedWorkers.length;
632
633 assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1);
634 assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1);
635 assert.equal(startedWorkers.length, startedBefore + 1);
636 }),
637 ),
638);
639
640it.effect("uses the writer for a database that is not in WAL mode", () =>
641 withFileClient((sql) =>
642 Effect.gen(function* () {
643 yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`;
644 const startedBefore = startedWorkers.length;
645
646 // A write inside readOnly only succeeds on the writer: the reader is read-only.
647 yield* sql
648 .withTransaction(sql`INSERT INTO entries VALUES ('on the writer')`)
649 .pipe(SqliteClient.readOnly);
650 assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1);
651 assert.equal(startedWorkers.length, startedBefore + 1);
652 }),
653 ),
654);
655
656const captureWarnings = () => {
657 const warnings: Array<unknown> = [];
658 const logger = Logger.make(({ logLevel, message }) => {
659 if (logLevel === "Warn") warnings.push(message);
660 });
661 return { warnings, layer: Logger.layer([logger], { mergeWithExisting: false }) };
662};
663
664const silently = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
665 Effect.acquireUseRelease(
666 Effect.sync(() => {
667 silentWorkers.on = true;
668 }),
669 () => effect,
670 () =>
671 Effect.sync(() => {
672 silentWorkers.on = false;
673 silentWorkers.readyOnTerminate = false;
674 }),
675 );
676
677it.effect("turns the reader off when its worker never opens the database", () => {
678 const { warnings, layer: logs } = captureWarnings();
679 return withFileClient((sql) =>
680 Effect.gen(function* () {
681 yield* sql`PRAGMA journal_mode = WAL`;
682 yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`;
683 yield* sql`INSERT INTO entries VALUES ('kept')`;
684 const read = yield* silently(
685 Effect.gen(function* () {
686 const fiber = yield* Effect.forkChild(
687 sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly),
688 );
689 yield* Effect.yieldNow;
690 yield* TestClock.adjust("10 seconds");
691 return yield* Fiber.join(fiber);
692 }),
693 );
694
695 assert.equal(read, 1);
696 assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1);
697 assert.lengthOf(warnings, 1);
698 }),
699 ).pipe(Effect.provide(logs));
700});
701
702it.effect("lets a read be cancelled while its worker is still starting", () =>
703 withFileClient((sql) =>
704 Effect.gen(function* () {
705 yield* sql`PRAGMA journal_mode = WAL`;
706 yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`;
707 yield* silently(
708 Effect.gen(function* () {
709 const fiber = yield* Effect.forkChild(
710 sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly),
711 );
712 yield* Effect.yieldNow;
713 // Never returns if the wait for the worker were uninterruptible: the
714 // test clock has not moved, so the startup deadline cannot fire.
715 yield* Fiber.interrupt(fiber);
716 }),
717 );
718
719 // The client stays usable: the next read waits out the deadline, then uses the writer.
720 const next = yield* Effect.forkChild(
721 sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly),
722 );
723 yield* Effect.yieldNow;
724 yield* TestClock.adjust("10 seconds");
725 assert.equal(yield* Fiber.join(next), 0);
726 }),
727 ),
728);
729
730it.effect("logs one warning when concurrent first reads find the reader unavailable", () => {
731 const { warnings, layer: logs } = captureWarnings();
732 return withFileClient((sql, filename) =>
733 Effect.gen(function* () {
734 yield* sql`PRAGMA journal_mode = WAL`;
735 yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`;
736 yield* (yield* FileSystem.FileSystem).remove(filename);
737 const startedBefore = startedWorkers.length;
738
739 const counts = yield* Effect.all(
740 [
741 sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly),
742 sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly),
743 ],
744 { concurrency: "unbounded" },
745 );
746
747 assert.deepEqual(counts, [0, 0]);
748 assert.equal(startedWorkers.length, startedBefore + 1);
749 assert.lengthOf(warnings, 1);
750 }),
751 ).pipe(Effect.provide(logs));
752});
753
754it.effect("does not hand out a worker that reports ready after another read timed it out", () =>
755 withFileClient((sql) =>
756 Effect.gen(function* () {
757 yield* sql`PRAGMA journal_mode = WAL`;
758 yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`;
759 yield* sql`INSERT INTO entries VALUES ('kept')`;
760 const read = () => sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly);
761 const counts = yield* silently(
762 Effect.gen(function* () {
763 const first = yield* Effect.forkChild(read());
764 yield* Effect.yieldNow;
765 yield* TestClock.adjust("5 seconds");
766 const second = yield* Effect.forkChild(read());
767 yield* Effect.yieldNow;
768 // The first read reaches its deadline and retires the worker; the second
769 // is still waiting when the worker's "ready" arrives, before it exits.
770 silentWorkers.readyOnTerminate = true;
771 yield* TestClock.adjust("5 seconds");
772 return [yield* Fiber.join(first), yield* Fiber.join(second)];
773 }),
774 );
775
776 assert.deepEqual(counts, [1, 1]);
777 }),
778 ),
779);
780
781const withCheckpointClient = <A, E>(
782 use: (
783 sql: SqlClient.SqlClient,
784 filename: string,
785 ) => Effect.Effect<A, E, SqlClient.SqlClient | FileSystem.FileSystem>,
786) =>
787 Effect.gen(function* () {
788 const fs = yield* FileSystem.FileSystem;
789 const path = yield* Path.Path;
790 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-checkpoint-" });
791 const filename = path.join(directory, "state.sqlite");
792 return yield* Effect.gen(function* () {
793 const sql = yield* SqlClient.SqlClient;
794 yield* sql`PRAGMA journal_mode = WAL`;
795 return yield* use(sql, filename);
796 }).pipe(Effect.provide(SqliteClient.layer({ filename, checkpointWorker: true })));
797 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer));
798
799// Resolves once the worker has handled a request sent now, after every
800// statement so far. Earlier requests from the writer reply with null.
801const checkpointWorkerReply = (worker: NodeWorkerThreads.Worker) =>
802 Effect.promise(
803 () =>
804 new Promise<void>((resolve) => {
805 const onReply = (reply: unknown) => {
806 if (reply !== "barrier") return;
807 worker.off("message", onReply);
808 resolve();
809 };
810 worker.on("message", onReply);
811 // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin.
812 worker.postMessage("barrier");
813 }),
814 );
815
816it.effect("checkpoints the WAL on the worker instead of the writer", () =>
817 withCheckpointClient((sql, filename) =>
818 Effect.gen(function* () {
819 const fs = yield* FileSystem.FileSystem;
820 const walSize = Effect.map(fs.stat(`${filename}-wal`), (info) => Number(info.size));
821 const worker = startedWorkers.at(-1)!;
822 const limit = 1024 * 1024;
823 yield* sql.unsafe(`PRAGMA journal_size_limit = ${limit}`);
824 const autocheckpoint = yield* sql<{
825 readonly wal_autocheckpoint: number;
826 }>`PRAGMA wal_autocheckpoint`;
827 // The writer leaves a WAL of this size to the worker.
828 assert.equal(autocheckpoint[0]?.wal_autocheckpoint, 10_000);
829
830 // About 1200 frames, past the 1000 that start a checkpoint.
831 yield* sql`CREATE TABLE pages(payload BLOB)`;
832 yield* sql`
833 WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < 1200)
834 INSERT INTO pages(payload) SELECT randomblob(4000) FROM n
835 `;
836 yield* checkpointWorkerReply(worker);
837 const wal = yield* sql<{
838 readonly log: number;
839 readonly checkpointed: number;
840 }>`PRAGMA wal_checkpoint(NOOP)`;
841 assert.isAbove(wal[0]!.log, 1000);
842 assert.equal(wal[0]!.checkpointed, wal[0]!.log);
843
844 // Every frame is in the database, so the next commit restarts the WAL and
845 // cuts the file back to the size limit.
846 assert.isAbove(yield* walSize, limit);
847 yield* sql`INSERT INTO pages(payload) VALUES (x'00')`;
848 assert.isAtMost(yield* walSize, limit);
849 }),
850 ),
851);
852
853it.effect("hands checkpoints back to the writer when the worker stops", () =>
854 withCheckpointClient((sql) =>
855 Effect.gen(function* () {
856 yield* Effect.promise(() => startedWorkers.at(-1)!.terminate());
857 const autocheckpoint = yield* sql<{
858 readonly wal_autocheckpoint: number;
859 }>`PRAGMA wal_autocheckpoint`;
860 assert.equal(autocheckpoint[0]?.wal_autocheckpoint, 1000);
861 }),
862 ),
863);
864
865it.effect("warns once and hands checkpoints back when the worker fails at startup", () => {
866 const { warnings, layer: logs } = captureWarnings();
867 const reason = "SQLite 3.50.4 has no NOOP checkpoint";
868 return Effect.acquireUseRelease(
869 Effect.sync(() => {
870 failingWorkers.message = reason;
871 }),
872 () =>
873 withCheckpointClient((sql) =>
874 Effect.gen(function* () {
875 const worker = startedWorkers.at(-1)!;
876 // threadId is -1 once the worker has exited and the client has seen "error" and "exit".
877 if (worker.threadId !== -1) {
878 yield* Effect.promise(() => new Promise((resolve) => worker.once("exit", resolve)));
879 }
880 const autocheckpoint = yield* sql<{
881 readonly wal_autocheckpoint: number;
882 }>`PRAGMA wal_autocheckpoint`;
883 assert.equal(autocheckpoint[0]?.wal_autocheckpoint, 1000);
884 assert.deepEqual(warnings, [
885 [
886 "SQLite checkpoint worker unavailable; checkpoints run on the main connection",
887 { reason },
888 ],
889 ]);
890 }),
891 ),
892 () =>
893 Effect.sync(() => {
894 failingWorkers.message = undefined;
895 }),
896 ).pipe(Effect.provide(logs));
897});
898
899it.effect("does not warn when the client stops its checkpoint worker", () => {
900 const { warnings, layer: logs } = captureWarnings();
901 return Effect.gen(function* () {
902 yield* withCheckpointClient(() => Effect.void);
903 assert.lengthOf(warnings, 0);
904 }).pipe(Effect.provide(logs));
905});