1// @effect-diagnostics nodeBuiltinImport:off
2import * as NodeChildProcess from "node:child_process";
3import * as NodeFS from "node:fs";
4import * as NodeOS from "node:os";
5import * as NodePath from "node:path";
6
7import * as NodeServices from "@effect/platform-node/NodeServices";
8import { assert, it } from "@effect/vitest";
9import * as Effect from "effect/Effect";
10import * as Layer from "effect/Layer";
11import * as SqlClient from "effect/sql/SqlClient";
12
13import { WAL_SIZE_LIMIT_BYTES } from "./Sqlite.ts";
14import * as SqlitePersistence from "./Sqlite.ts";
15
16const lockHolderSource = `
17const { DatabaseSync } = require("node:sqlite");
18const db = new DatabaseSync(process.argv[1]);
19db.exec("BEGIN IMMEDIATE");
20process.stdout.write("locked\\n");
21setTimeout(() => {
22 db.exec("COMMIT");
23 db.close();
24}, Number(process.argv[2]));
25`;
26
27const spawnWriteLockHolder = (dbPath: string, holdMs: number) =>
28 Effect.promise(
29 () =>
30 new Promise<void>((resolve, reject) => {
31 const holder = NodeChildProcess.spawn(
32 process.execPath,
33 ["-e", lockHolderSource, dbPath, String(holdMs)],
34 { stdio: ["ignore", "pipe", "ignore"] },
35 );
36 holder.stdout.once("data", () => resolve());
37 holder.on("error", reject);
38 holder.on("exit", () =>
39 reject(new Error("lock holder exited before acquiring the write lock")),
40 );
41 }),
42 );
43
44it.effect("waits out a concurrent writer instead of failing with SQLITE_BUSY", () => {
45 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-sqlite-busy-"));
46 const dbPath = NodePath.join(tempDir, "state.sqlite");
47
48 return Effect.gen(function* () {
49 const sql = yield* SqlClient.SqlClient;
50 yield* sql`CREATE TABLE busy_probe(id INTEGER PRIMARY KEY)`;
51 yield* spawnWriteLockHolder(dbPath, 300);
52 yield* sql`INSERT INTO busy_probe(id) VALUES (${1})`;
53 const rows = yield* sql<{ readonly id: number }>`SELECT id FROM busy_probe`;
54 assert.deepEqual([...rows], [{ id: 1 }]);
55 }).pipe(
56 Effect.provide(SqlitePersistence.layerFromPath(dbPath).pipe(Layer.provide(NodeServices.layer))),
57 Effect.ensuring(Effect.sync(() => NodeFS.rmSync(tempDir, { recursive: true, force: true }))),
58 );
59});
60
61it.effect("limits the WAL file size in the shared persistence setup", () =>
62 Effect.gen(function* () {
63 const sql = yield* SqlClient.SqlClient;
64 const rows = yield* sql<{ readonly journal_size_limit: number }>`PRAGMA journal_size_limit`;
65 assert.equal(rows[0]?.journal_size_limit, WAL_SIZE_LIMIT_BYTES);
66 }).pipe(Effect.provide(SqlitePersistence.layerMemory)),
67);
68
69it.effect("collects planner statistics for an existing database on open", () => {
70 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-sqlite-stats-"));
71 const dbPath = NodePath.join(tempDir, "state.sqlite");
72 const persistence = () =>
73 SqlitePersistence.layerFromPath(dbPath).pipe(Layer.provide(NodeServices.layer));
74
75 return Effect.gen(function* () {
76 yield* Effect.gen(function* () {
77 const sql = yield* SqlClient.SqlClient;
78 yield* sql`CREATE TABLE stats_probe(id INTEGER PRIMARY KEY, thread_id TEXT NOT NULL)`;
79 yield* sql`CREATE INDEX stats_probe_thread_idx ON stats_probe(thread_id)`;
80 yield* sql`
81 WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < 1000)
82 INSERT INTO stats_probe(thread_id) SELECT 'thread-' || (i % 10) FROM n
83 `;
84 }).pipe(Effect.provide(persistence()));
85
86 // A database written before statistics existed gets them on the next open.
87 yield* Effect.gen(function* () {
88 const sql = yield* SqlClient.SqlClient;
89 const rows = yield* sql<{ readonly stat: string }>`
90 SELECT stat FROM sqlite_stat1 WHERE idx = 'stats_probe_thread_idx'
91 `;
92 assert.deepEqual([...rows], [{ stat: "1000 100" }]);
93 }).pipe(Effect.provide(persistence()));
94 }).pipe(
95 Effect.ensuring(Effect.sync(() => NodeFS.rmSync(tempDir, { recursive: true, force: true }))),
96 );
97});
98
99it.effect("only fsyncs the database file at WAL checkpoints", () => {
100 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-sqlite-sync-"));
101 const dbPath = NodePath.join(tempDir, "state.sqlite");
102
103 return Effect.gen(function* () {
104 const sql = yield* SqlClient.SqlClient;
105 const rows = yield* sql<{ readonly synchronous: number }>`PRAGMA synchronous`;
106 // 1 is NORMAL.
107 assert.equal(rows[0]?.synchronous, 1);
108 }).pipe(
109 Effect.provide(SqlitePersistence.layerFromPath(dbPath).pipe(Layer.provide(NodeServices.layer))),
110 Effect.ensuring(Effect.sync(() => NodeFS.rmSync(tempDir, { recursive: true, force: true }))),
111 );
112});
113
114it.effect("applies busy_timeout in the shared persistence setup", () =>
115 Effect.gen(function* () {
116 const sql = yield* SqlClient.SqlClient;
117 const rows = yield* sql<{ readonly timeout: number }>`PRAGMA busy_timeout`;
118 assert.equal(rows[0]?.timeout, 5000);
119 }).pipe(Effect.provide(SqlitePersistence.layerMemory)),
120);
121
122it.effect("enables checkpoint_fullfsync in the shared persistence setup", () =>
123 Effect.gen(function* () {
124 const sql = yield* SqlClient.SqlClient;
125 const rows = yield* sql<{
126 readonly checkpoint_fullfsync: number;
127 }>`PRAGMA checkpoint_fullfsync`;
128 assert.equal(rows[0]?.checkpoint_fullfsync, 1);
129 }).pipe(Effect.provide(SqlitePersistence.layerMemory)),
130);