Open History to see the last commit
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 Cause from "effect/Cause";
7import * as Effect from "effect/Effect";
8import * as Exit from "effect/Exit";
9import * as Fiber from "effect/Fiber";
10import * as FileSystem from "effect/FileSystem";
11import * as Layer from "effect/Layer";
12import * as Path from "effect/Path";
13import * as SqlClient from "effect/sql/SqlClient";
14import * as SqlError from "effect/sql/SqlError";
15import * as Statement from "effect/sql/Statement";
16import { ChildProcess, ChildProcessSpawner } from "effect/process";
17import * as Stream from "effect/Stream";
18import type * as Tracer from "effect/Tracer";
19import { vi } from "vite-plus/test";
20
21import * as SqliteClient from "./nodeSqliteClient.ts";
22
23const startedWorkers = vi.hoisted((): Array<NodeWorkerThreads.Worker> => []);
24const silentWorkers = vi.hoisted(() => ({ on: false, readyOnTerminate: false }));
25const failingWorkers = vi.hoisted(() => ({ message: undefined as string | undefined }));
26
27vi.mock("node:worker_threads", async (importOriginal) => {
28 const actual = await importOriginal<typeof NodeWorkerThreads>();
29 class RecordedWorker extends actual.Worker {
30 constructor(...args: ConstructorParameters<typeof actual.Worker>) {
31 super(
32 ...((silentWorkers.on
33 ? ["setInterval(() => {}, 1 << 30);", { eval: true }]
34 : failingWorkers.message !== undefined
35 ? [`throw new Error(${JSON.stringify(failingWorkers.message)});`, { eval: true }]
36 : args) as ConstructorParameters<typeof actual.Worker>),
37 );
38 startedWorkers.push(this);
39 }
40 override terminate() {
41 // A "ready" already queued when termination starts reaches the parent before "exit".
42 if (silentWorkers.readyOnTerminate) this.emit("message", { type: "ready" });
43 return super.terminate();
44 }
45 }
46 return { ...actual, Worker: RecordedWorker };
47});
48
49const withReaderClient = <A, E>(
50 use: (sql: SqlClient.SqlClient) => Effect.Effect<A, E, SqlClient.SqlClient>,
51) =>
52 Effect.gen(function* () {
53 const fs = yield* FileSystem.FileSystem;
54 const path = yield* Path.Path;
55 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-reader-" });
56 const filename = path.join(directory, "state.sqlite");
57 return yield* Effect.gen(function* () {
58 const sql = yield* SqlClient.SqlClient;
59 yield* sql`PRAGMA journal_mode = WAL`;
60 yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`;
61 yield* sql`INSERT INTO entries VALUES ('first')`;
62 return yield* use(sql);
63 }).pipe(Effect.provide(SqliteClient.layer({ filename, readerWorker: true })));
64 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer));
65
66const countEntries = (sql: SqlClient.SqlClient) =>
67 Effect.map(
68 sql<{ count: number }>`SELECT COUNT(*) AS count FROM entries`,
69 (rows) => rows[0]?.count,
70 );
71
72const layer = it.layer(SqliteClient.layer({ filename: ":memory:" }));
73
74layer("NodeSqliteClient", (it) => {
75 it.effect("records the writer connection owner on a waiting query", () =>
76 Effect.gen(function* () {
77 const sql = yield* SqlClient.SqlClient;
78 const held = yield* Deferred.make<void>();
79 const release = yield* Deferred.make<void>();
80 const holder = yield* sql
81 .withTransaction(
82 sql`SELECT 1`.pipe(
83 Effect.andThen(Deferred.succeed(held, undefined)),
84 Effect.andThen(Deferred.await(release)),
85 ),
86 )
87 .pipe(Effect.withSpan("test.writerHolder"), Effect.forkChild);
88 yield* Deferred.await(held);
89
90 const waitingSpan = yield* Deferred.make<Tracer.Span>();
91 const observeWaiter: Statement.Transformer = (statement, _sql, _fiber, span) =>
92 Deferred.succeed(waitingSpan, span).pipe(Effect.as(statement));
93 const waiter = yield* sql`SELECT 1`.pipe(
94 Effect.provideService(Statement.CurrentTransformer, observeWaiter),
95 Effect.provideService(Statement.SpanPropagationEnabled, true),
96 (effect) => Effect.forkChild(effect, { startImmediately: true }),
97 );
98 const span = yield* Deferred.await(waitingSpan);
99 yield* Deferred.succeed(release, undefined);
100 yield* Fiber.join(holder);
101 yield* Fiber.join(waiter);
102
103 assert.equal(span.attributes.get("db.lock_owner_at_start.name"), "test.writerHolder");
104 assert.typeOf(span.attributes.get("db.lock_owner_at_start.fiber_id"), "number");
105 assert.typeOf(span.attributes.get("db.lock_owner_at_start.trace_id"), "string");
106 assert.typeOf(span.attributes.get("db.lock_owner_at_start.span_id"), "string");
107 }),
108 );
109
110 it.effect("retries preparing a query after the missing schema becomes available", () =>
111 Effect.gen(function* () {
112 const sql = yield* SqlClient.SqlClient;
113 const select = sql<{ name: string }>`SELECT name FROM created_after_prepare_failure`;
114 const error = yield* select.pipe(Effect.flip);
115 assert.equal(error._tag, "SqlError");
116 assert.equal(error.reason.operation, "prepare");
117
118 yield* sql`CREATE TABLE created_after_prepare_failure(name TEXT NOT NULL)`;
119 yield* sql`INSERT INTO created_after_prepare_failure VALUES ('recovered')`;
120 assert.deepEqual(yield* select, [{ name: "recovered" }]);
121 assert.deepEqual(yield* select.values, [["recovered"]]);
122 }),
123 );
124
125 it.effect("classifies constraint failures by their SQLite result code", () =>
126 Effect.gen(function* () {
127 const sql = yield* SqlClient.SqlClient;
128 yield* sql`CREATE TABLE constrained(name TEXT NOT NULL UNIQUE)`;
129 yield* sql`INSERT INTO constrained VALUES ('taken')`;
130
131 const duplicate = yield* sql`INSERT INTO constrained VALUES ('taken')`.pipe(Effect.flip);
132 assert(duplicate.reason._tag === "UniqueViolation");
133 assert.equal(duplicate.reason.constraint, "constrained.name");
134
135 const missing = yield* sql`INSERT INTO constrained VALUES (NULL)`.pipe(Effect.flip);
136 assert.equal(missing.reason._tag, "ConstraintError");
137 }),
138 );
139});
140
141const largeText = "x".repeat(64 * 1024 + 1);
142
143it.effect("reads integers on the values path as the caller asked, cached or not", () =>
144 Effect.gen(function* () {
145 const sql = yield* SqlClient.SqlClient;
146 for (const value of ["small", largeText]) {
147 const values = sql`SELECT 1 AS id, length(${value}) AS size`.values;
148 assert.deepEqual(yield* values.pipe(Effect.provideService(SqlClient.SafeIntegers, true)), [
149 [1n, BigInt(value.length)],
150 ]);
151 assert.deepEqual(yield* values, [[1, value.length]]);
152 }
153 }).pipe(Effect.provide(SqliteClient.layer({ filename: ":memory:" }))),
154);
155
156const makeTempDatabase = Effect.gen(function* () {
157 const fs = yield* FileSystem.FileSystem;
158 const path = yield* Path.Path;
159 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-transaction-" });
160 const filename = path.join(directory, "state.sqlite");
161 const other = yield* Effect.acquireRelease(
162 Effect.sync(() => new NodeSqlite.DatabaseSync(filename)),
163 (database) => Effect.sync(() => database.close()),
164 );
165 yield* Effect.sync(() => {
166 other.exec(`
167 PRAGMA journal_mode = WAL;
168 CREATE TABLE counters(id INTEGER PRIMARY KEY, value INTEGER NOT NULL);
169 INSERT INTO counters VALUES (1, 0);
170 `);
171 });
172 return { filename, other };
173});
174
175it.effect("keeps another connection from committing between a transaction's read and write", () =>
176 Effect.gen(function* () {
177 const { filename, other } = yield* makeTempDatabase;
178 yield* Effect.gen(function* () {
179 const sql = yield* SqlClient.SqlClient;
180 yield* sql.withTransaction(
181 Effect.gen(function* () {
182 const [row] = yield* sql<{ readonly value: number }>`
183 SELECT value FROM counters WHERE id = 1
184 `;
185 // Deferred BEGIN permits a competing commit, causing SQLITE_BUSY_SNAPSHOT that no timeout can resolve.
186 yield* Effect.sync(() =>
187 assert.throws(
188 () => other.exec("UPDATE counters SET value = value + 1 WHERE id = 1"),
189 /database is locked/,
190 ),
191 );
192 yield* sql`UPDATE counters SET value = ${(row?.value ?? 0) + 10} WHERE id = 1`;
193 }),
194 );
195 yield* Effect.sync(() => other.exec("UPDATE counters SET value = value + 1 WHERE id = 1"));
196 assert.deepEqual(yield* sql`SELECT value FROM counters WHERE id = 1`.values, [[11]]);
197 }).pipe(Effect.provide(SqliteClient.layer({ filename })));
198 }).pipe(Effect.provide(NodeServices.layer)),
199);
200
201it.effect("waits for another process's lock while the database opens", () =>
202 Effect.gen(function* () {
203 const fs = yield* FileSystem.FileSystem;
204 const path = yield* Path.Path;
205 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-open-" });
206 const filename = path.join(directory, "state.sqlite");
207 const seed = new NodeSqlite.DatabaseSync(filename);
208 seed.exec("PRAGMA journal_mode = WAL; CREATE TABLE entries(name TEXT NOT NULL)");
209 seed.close();
210 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
211 const holder = yield* spawner.spawn(
212 ChildProcess.make(process.execPath, [
213 "-e",
214 `const db = new (require("node:sqlite").DatabaseSync)(process.argv[1]);
215 db.exec("PRAGMA locking_mode = EXCLUSIVE; BEGIN EXCLUSIVE");
216 console.log("held");
217 setTimeout(() => db.close(), 300);`,
218 filename,
219 ]),
220 );
221 yield* Stream.runHead(holder.stdout);
222 yield* Effect.gen(function* () {
223 const sql = yield* SqlClient.SqlClient;
224 assert.deepEqual(yield* sql`SELECT COUNT(*) FROM entries`.values, [[0]]);
225 }).pipe(Effect.provide(SqliteClient.layer({ filename, checkpointWorker: true })));
226 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
227);
228
229it.effect("reports a transaction blocked by another writer as a lock timeout", () =>
230 Effect.gen(function* () {
231 const { filename, other } = yield* makeTempDatabase;
232 yield* Effect.gen(function* () {
233 const sql = yield* SqlClient.SqlClient;
234 yield* sql`PRAGMA busy_timeout = 0`;
235 const read = sql.withTransaction(sql`SELECT value FROM counters WHERE id = 1`.values);
236
237 yield* Effect.sync(() => other.exec("BEGIN IMMEDIATE"));
238 const error = yield* read.pipe(Effect.flip);
239 assert.equal(error.reason._tag, "LockTimeoutError");
240
241 yield* Effect.sync(() => other.exec("COMMIT"));
242 assert.deepEqual(yield* read, [[0]]);
243 }).pipe(Effect.provide(SqliteClient.layer({ filename })));
244 }).pipe(Effect.provide(NodeServices.layer)),
245);
246
247it.effect("lets the writer commit while a read-only transaction is open", () =>
248 withReaderClient((sql) =>
249 Effect.gen(function* () {
250 const snapshotTaken = yield* Deferred.make<void>();
251 const writer = yield* Effect.forkChild(
252 Deferred.await(snapshotTaken).pipe(
253 Effect.andThen(sql`INSERT INTO entries VALUES ('second')`),
254 ),
255 );
256 const counts = yield* sql
257 .withTransaction(
258 Effect.gen(function* () {
259 const before = yield* countEntries(sql);
260 yield* Deferred.succeed(snapshotTaken, undefined);
261 // Holding the writer lock here would deadlock this join.
262 yield* Fiber.join(writer);
263 const after = yield* countEntries(sql);
264 return { before, after };
265 }),
266 )
267 .pipe(SqliteClient.readOnly);
268
269 assert.deepEqual(counts, { before: 1, after: 1 });
270 assert.equal(yield* countEntries(sql).pipe(SqliteClient.readOnly), 2);
271 }),
272 ),
273);
274
275it.effect("serves another read while a read-only transaction is open", () =>
276 withReaderClient((sql) =>
277 Effect.gen(function* () {
278 const opened = yield* Deferred.make<void>();
279 const other = yield* Effect.forkChild(
280 Deferred.await(opened).pipe(Effect.andThen(countEntries(sql).pipe(SqliteClient.readOnly))),
281 );
282 yield* sql
283 .withTransaction(
284 Effect.gen(function* () {
285 yield* countEntries(sql);
286 yield* Deferred.succeed(opened, undefined);
287 // One shared reader would make this join wait on the transaction forever.
288 assert.equal(yield* Fiber.join(other), 1);
289 }),
290 )
291 .pipe(SqliteClient.readOnly);
292 }),
293 ),
294);
295
296// regression: #19226 read-only transactions run on the separate reader connection
297it.effect("rejects a write in a read-only transaction", () =>
298 withReaderClient((sql) =>
299 Effect.gen(function* () {
300 const error = yield* sql
301 .withTransaction(sql`INSERT INTO entries VALUES ('blocked')`)
302 .pipe(SqliteClient.readOnly, Effect.flip);
303
304 assert.equal(error._tag, "SqlError");
305 assert.include(String(error.reason.cause), "readonly database");
306 assert.equal(yield* countEntries(sql), 1);
307 }),
308 ),
309);
310
311it.effect("keeps the reader read-only even if a statement turns query_only off", () =>
312 withReaderClient((sql) =>
313 Effect.gen(function* () {
314 const error = yield* sql
315 .withTransaction(
316 Effect.gen(function* () {
317 yield* sql`PRAGMA query_only = OFF`;
318 yield* sql`INSERT INTO entries VALUES ('blocked')`;
319 }),
320 )
321 .pipe(SqliteClient.readOnly, Effect.flip);
322
323 assert.equal(error._tag, "SqlError");
324 assert.include(String(error.reason.cause), "readonly database");
325 assert.equal(yield* countEntries(sql), 1);
326 }),
327 ),
328);
329
330it.effect("keeps read-only work inside a write transaction on the writer", () =>
331 withReaderClient((sql) =>
332 sql.withTransaction(
333 Effect.gen(function* () {
334 yield* sql`INSERT INTO entries VALUES ('uncommitted')`;
335 const count = yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly);
336 assert.equal(count, 2);
337 }),
338 ),
339 ),
340);
341
342it.effect("fails a read-only transaction whose reader stops, then starts a new reader", () =>
343 withReaderClient((sql) =>
344 Effect.gen(function* () {
345 const startedBefore = startedWorkers.length;
346 const error = yield* sql
347 .withTransaction(
348 Effect.gen(function* () {
349 yield* countEntries(sql);
350 yield* Effect.promise(() => startedWorkers.at(-1)!.terminate());
351 return yield* countEntries(sql);
352 }),
353 )
354 .pipe(SqliteClient.readOnly, Effect.flip);
355
356 assert.equal(error._tag, "SqlError");
357 assert.include(String(error.reason.cause), "Database reader exited");
358 assert.equal(yield* countEntries(sql).pipe(SqliteClient.readOnly), 1);
359 assert.equal(startedWorkers.length, startedBefore + 2);
360 }),
361 ),
362);
363
364const withFileClient = <A, E>(
365 setup: (
366 sql: SqlClient.SqlClient,
367 filename: string,
368 ) => Effect.Effect<A, E, SqlClient.SqlClient | FileSystem.FileSystem>,
369) =>
370 Effect.gen(function* () {
371 const fs = yield* FileSystem.FileSystem;
372 const path = yield* Path.Path;
373 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-reader-off-" });
374 const filename = path.join(directory, "state.sqlite");
375 return yield* Effect.gen(function* () {
376 return yield* setup(yield* SqlClient.SqlClient, filename);
377 }).pipe(Effect.provide(SqliteClient.layer({ filename, readerWorker: true })));
378 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer));
379
380it.effect("uses the writer when the reader cannot open the database", () =>
381 withFileClient((sql, filename) =>
382 Effect.gen(function* () {
383 yield* sql`PRAGMA journal_mode = WAL`;
384 yield* sql`CREATE TABLE entries(name TEXT NOT NULL)`;
385 yield* sql`INSERT INTO entries VALUES ('kept')`;
386 yield* (yield* FileSystem.FileSystem).remove(filename);
387 const startedBefore = startedWorkers.length;
388
389 assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1);
390 assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1);
391
392 // A read beside an open read transaction must not start a worker on the next slot.
393 const held = yield* Deferred.make<void>();
394 const release = yield* Deferred.make<void>();
395 const holder = yield* sql
396 .withTransaction(
397 countEntries(sql).pipe(
398 Effect.andThen(Deferred.succeed(held, undefined)),
399 Effect.andThen(Deferred.await(release)),
400 ),
401 )
402 .pipe(SqliteClient.readOnly, Effect.forkChild);
403 yield* Deferred.await(held);
404 const other = yield* countEntries(sql).pipe(SqliteClient.readOnly, (effect) =>
405 Effect.forkChild(effect, { startImmediately: true }),
406 );
407 yield* Deferred.succeed(release, undefined);
408 yield* Fiber.join(holder);
409 assert.equal(yield* Fiber.join(other), 1);
410 assert.equal(startedWorkers.length, startedBefore + 1);
411 }),
412 ),
413);
414
415it.effect("fails with a typed SqlError when the database file is not SQLite", () =>
416 Effect.gen(function* () {
417 const fs = yield* FileSystem.FileSystem;
418 const path = yield* Path.Path;
419 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-notadb-" });
420 const filename = path.join(directory, "state.sqlite");
421 yield* fs.writeFileString(filename, "not a db\n");
422
423 const exit = yield* Layer.build(SqliteClient.layer({ filename, checkpointWorker: true })).pipe(
424 Effect.exit,
425 );
426
427 assert.isTrue(Exit.isFailure(exit));
428 if (Exit.isFailure(exit)) assert.isTrue(SqlError.isSqlError(Cause.squash(exit.cause)));
429 if (Exit.isFailure(exit)) assert.isFalse(Cause.hasDies(exit.cause));
430 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
431);
432
433it.effect("finishes masked database cleanup after its caller is interrupted", () =>
434 Effect.gen(function* () {
435 const sql = yield* SqlClient.SqlClient;
436 yield* sql`CREATE TABLE cleanup_writes(value INTEGER PRIMARY KEY)`;
437 const entered = yield* Deferred.make<void>();
438 const work = yield* Effect.forkChild(
439 Deferred.succeed(entered, undefined).pipe(
440 Effect.andThen(Effect.never),
441 Effect.onInterrupt(() =>
442 Effect.gen(function* () {
443 yield* sql`INSERT INTO cleanup_writes VALUES (1)`;
444 yield* sql.withTransaction(sql`INSERT INTO cleanup_writes VALUES (2)`);
445 }),
446 ),
447 ),
448 { startImmediately: true },
449 );
450 yield* Deferred.await(entered);
451 yield* Fiber.interrupt(work);
452 assert.deepEqual(yield* sql`SELECT value FROM cleanup_writes ORDER BY value`.values, [
453 [1],
454 [2],
455 ]);
456 }).pipe(Effect.provide(SqliteClient.layer({ filename: ":memory:" }))),
457);
458
459// oxlint-disable-next-line t3code/no-real-time-wait-in-tests -- a zero-delay macrotask boundary that lets the worker's queued messages land, not a wait for time
460const eventLoopTurn = Effect.promise(() => new Promise<void>((resolve) => setImmediate(resolve)));
461
462it.effect.each([
463 { reservation: false, masked: false },
464 { reservation: false, masked: true },
465 { reservation: true, masked: false },
466 { reservation: true, masked: true },
467])(
468 "preserves caller cancellation while waiting (reservation=$reservation, masked=$masked)",
469 ({ reservation, masked }) =>
470 Effect.gen(function* () {
471 const sql = yield* SqlClient.SqlClient;
472 yield* sql`CREATE TABLE entries(value INTEGER)`;
473 const held = yield* Deferred.make<void>();
474 const release = yield* Deferred.make<void>();
475 const holder = yield* Effect.forkChild(
476 sql.withTransaction(
477 Deferred.succeed(held, undefined).pipe(Effect.andThen(Deferred.await(release))),
478 ),
479 { startImmediately: true },
480 );
481 yield* Deferred.await(held);
482 const entered = yield* Deferred.make<void>();
483 const body = sql`INSERT INTO entries VALUES (1)`.pipe(
484 Effect.andThen(Deferred.succeed(entered, undefined)),
485 );
486 const acquire = reservation
487 ? Effect.scoped(
488 Effect.gen(function* () {
489 const connection = yield* sql.reserve;
490 yield* connection.executeUnprepared("INSERT INTO entries VALUES (1)", [], undefined);
491 yield* Deferred.succeed(entered, undefined);
492 }),
493 )
494 : sql.withTransaction(body);
495 const waiter = yield* Effect.forkChild(masked ? Effect.uninterruptible(acquire) : acquire, {
496 startImmediately: true,
497 });
498 const cancelled = yield* Effect.forkChild(Fiber.interrupt(waiter), {
499 startImmediately: true,
500 });
501 yield* Effect.gen(function* () {
502 yield* eventLoopTurn;
503 assert.equal(cancelled.pollUnsafe() !== undefined, !masked);
504 assert.isFalse(yield* Deferred.isDone(entered));
505 }).pipe(Effect.ensuring(Deferred.succeed(release, undefined)));
506 yield* Fiber.join(holder);
507 yield* Fiber.join(cancelled);
508 assert.equal(yield* Deferred.isDone(entered), masked);
509 yield* sql.withTransaction(sql`INSERT INTO entries VALUES (2)`);
510 assert.deepEqual(
511 yield* sql`SELECT value FROM entries ORDER BY value`.values,
512 masked ? [[1], [2]] : [[2]],
513 );
514 }).pipe(Effect.provide(SqliteClient.layer({ filename: ":memory:" }))),
515);
516
517const PAYLOAD = new Uint8Array(1024 * 1024).fill(7);
518const WAL_SIZE_LIMIT = 32 * 1024 * 1024;
519
520// Fails instead of hanging when admission holds the write: the worker retries a blocked
521// checkpoint on its own, so two blocked reports for one WAL generation mean a held write.
522const insertUnlessHeld = Effect.fnUntraced(function* (
523 sql: SqlClient.SqlClient,
524 worker: NodeWorkerThreads.Worker,
525 label: string,
526) {
527 const stalled = yield* Deferred.make<void>();
528 let seen: unknown;
529 const onMessage = (message: { type?: unknown; blocked?: unknown; generation?: unknown }) => {
530 if (message?.type !== "checkpoint" || message.blocked !== true) return;
531 if (message.generation === seen) Deferred.doneUnsafe(stalled, Effect.void);
532 seen = message.generation;
533 };
534 worker.on("message", onMessage);
535 const held = yield* Effect.raceFirst(
536 sql`INSERT INTO blobs VALUES (${PAYLOAD})`.pipe(Effect.as(false)),
537 Deferred.await(stalled).pipe(Effect.as(true)),
538 ).pipe(Effect.ensuring(Effect.sync(() => worker.off("message", onMessage))));
539 if (held) assert.fail(`${label} is held by admission while an outside reader pins the WAL`);
540});
541
542// `outside` stands in for `sqlite3 statev2.sqlite` or a backup script reading the database.
543const withOutsideConnection = <A, E>(
544 use: (options: {
545 readonly sql: SqlClient.SqlClient;
546 readonly worker: NodeWorkerThreads.Worker;
547 readonly outside: NodeSqlite.DatabaseSync;
548 readonly walSize: Effect.Effect<number>;
549 }) => Effect.Effect<A, E, SqlClient.SqlClient>,
550) =>
551 Effect.gen(function* () {
552 const fs = yield* FileSystem.FileSystem;
553 const path = yield* Path.Path;
554 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-pinned-wal-" });
555 const filename = path.join(directory, "state.sqlite");
556 const outside = yield* Effect.acquireRelease(
557 Effect.sync(() => new NodeSqlite.DatabaseSync(filename)),
558 (db) =>
559 Effect.sync(() => {
560 if (db.isTransaction) db.exec("COMMIT");
561 db.close();
562 }),
563 );
564 outside.exec(`
565 PRAGMA journal_mode = WAL;
566 PRAGMA wal_autocheckpoint = 0;
567 CREATE TABLE blobs(data BLOB NOT NULL);
568 `);
569 const walSize = fs.stat(`${filename}-wal`).pipe(
570 Effect.map((info) => Number(info.size)),
571 Effect.orDie,
572 );
573 yield* Effect.gen(function* () {
574 const sql = yield* SqlClient.SqlClient;
575 yield* sql.unsafe(`PRAGMA journal_size_limit = ${WAL_SIZE_LIMIT}`);
576 return yield* use({ sql, worker: startedWorkers.at(-1)!, outside, walSize });
577 }).pipe(Effect.provide(SqliteClient.layer({ filename, checkpointWorker: true })));
578 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer));
579
580const openReadTransaction = (db: NodeSqlite.DatabaseSync) => {
581 db.exec("BEGIN");
582 db.prepare("SELECT COUNT(*) FROM blobs").get();
583};
584
585it.effect("keeps admitting writes while an outside read transaction pins the WAL", () =>
586 withOutsideConnection(({ sql, worker, outside }) =>
587 Effect.gen(function* () {
588 yield* sql`INSERT INTO blobs VALUES (${PAYLOAD})`;
589 openReadTransaction(outside);
590 // 140 MiB passes both the 32 MiB size limit and the 32,768-frame ceiling.
591 for (let write = 1; write <= 140; write++) {
592 yield* insertUnlessHeld(sql, worker, `write ${write}`);
593 }
594 const rows = yield* sql<{ count: number }>`SELECT COUNT(*) AS count FROM blobs`;
595 assert.equal(rows[0]?.count, 141);
596 }),
597 ),
598);
599
600it.effect("keeps admitting writes while a reader stops TRUNCATE, then resets the WAL", () =>
601 withOutsideConnection(({ sql, worker, outside, walSize }) =>
602 Effect.gen(function* () {
603 const insert = outside.prepare("INSERT INTO blobs VALUES (?)");
604 for (let write = 1; write <= 40; write++) insert.run(PAYLOAD);
605 // A snapshot of an unbackfilled WAL lets PASSIVE finish but stops TRUNCATE.
606 openReadTransaction(outside);
607 const stopped = yield* Deferred.make<void>();
608 const reset = yield* Deferred.make<void>();
609 worker.on("message", (message: { type?: unknown; truncateId?: unknown; log?: unknown }) => {
610 if (message?.type !== "checkpoint" || message.truncateId === undefined) return;
611 Deferred.doneUnsafe(message.log === 0 ? reset : stopped, Effect.void);
612 });
613 // Starts a checkpoint without appending a frame past the reader's snapshot.
614 yield* sql`DELETE FROM blobs WHERE 0`;
615 yield* Deferred.await(stopped);
616 yield* insertUnlessHeld(sql, worker, "a write after the failed TRUNCATE");
617 assert.isAbove(yield* walSize, WAL_SIZE_LIMIT);
618 outside.exec("COMMIT");
619 // No further write arrives; the worker has to retry on its own.
620 yield* Deferred.await(reset);
621 assert.equal(yield* walSize, 0);
622 }),
623 ),
624);