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