diff --git a/apps/server/src/persistence/Sqlite.test.ts b/apps/server/src/persistence/Sqlite.test.ts index 75f9fc3a80b2cf9645a84dbda44e4930629c8a3c..31709abeb2e93371572e490018ea6e9d2c467ac1 100644 --- a/apps/server/src/persistence/Sqlite.test.ts +++ b/apps/server/src/persistence/Sqlite.test.ts @@ -58,31 +58,13 @@ it.effect("waits out a concurrent writer instead of failing with SQLITE_BUSY", ( ); }); -it.effect("shrinks the WAL file back to the size limit after a large write", () => { - const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-sqlite-wal-")); - const dbPath = NodePath.join(tempDir, "state.sqlite"); - const walFileSize = () => NodeFS.statSync(`${dbPath}-wal`).size; - // About 25% more 4 KB rows than the limit holds, in one transaction. - const rowCount = Math.ceil((WAL_SIZE_LIMIT_BYTES * 1.25) / 4000); - - return Effect.gen(function* () { +it.effect("limits the WAL file size in the shared persistence setup", () => + Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; - yield* sql`CREATE TABLE wal_probe(payload BLOB)`; - yield* sql` - WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < ${rowCount}) - INSERT INTO wal_probe(payload) SELECT randomblob(4000) FROM n - `; - assert.isAbove(walFileSize(), WAL_SIZE_LIMIT_BYTES); - - // The auto-checkpoint after the large commit copied every frame into the - // database, so the next commit restarts the WAL and cuts the file back. - yield* sql`INSERT INTO wal_probe(payload) VALUES (x'00')`; - assert.isAtMost(walFileSize(), WAL_SIZE_LIMIT_BYTES); - }).pipe( - Effect.provide(SqlitePersistence.layerFromPath(dbPath).pipe(Layer.provide(NodeServices.layer))), - Effect.ensuring(Effect.sync(() => NodeFS.rmSync(tempDir, { recursive: true, force: true }))), - ); -}); + const rows = yield* sql<{ readonly journal_size_limit: number }>`PRAGMA journal_size_limit`; + assert.equal(rows[0]?.journal_size_limit, WAL_SIZE_LIMIT_BYTES); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); it.effect("collects planner statistics for an existing database on open", () => { const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-sqlite-stats-")); diff --git a/apps/server/src/persistence/Sqlite.ts b/apps/server/src/persistence/Sqlite.ts index 8ba6f8437d4f807494bf47e07d6093b33b422b4b..1c49b578a0bc170272a8d02b5dfca0ecc370b888 100644 --- a/apps/server/src/persistence/Sqlite.ts +++ b/apps/server/src/persistence/Sqlite.ts @@ -50,6 +50,9 @@ export const layerFromPath = Effect.fn("makeSqlitePersistenceLive")(function* (d NodeSqliteClient.layer({ filename: dbPath, readerWorker: true, + // Checkpoints sync the database, with F_FULLFSYNC on macOS. Run them off + // the event loop so a slow flush cannot stall clients. + checkpointWorker: true, spanAttributes: { "db.name": path.basename(dbPath), "service.name": "t3code-server", diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts index 423e1ab6838b2581281dca9923ae602d9b2cee03..18b3de54d8e34322388cb8b70c391cf0d3c00674 100644 --- a/packages/shared/src/nodeSqliteClient.test.ts +++ b/packages/shared/src/nodeSqliteClient.test.ts @@ -678,3 +678,87 @@ it.effect("does not hand out a worker that reports ready after another read time }), ), ); + +const withCheckpointClient = ( + use: ( + sql: SqlClient.SqlClient, + filename: string, + ) => Effect.Effect, +) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-sqlite-checkpoint-" }); + const filename = path.join(directory, "state.sqlite"); + return yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`PRAGMA journal_mode = WAL`; + return yield* use(sql, filename); + }).pipe(Effect.provide(SqliteClient.layer({ filename, checkpointWorker: true }))); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)); + +// Resolves once the worker has handled a request sent now, after every +// statement so far. Earlier requests from the writer reply with null. +const checkpointWorkerReply = (worker: NodeWorkerThreads.Worker) => + Effect.promise( + () => + new Promise((resolve) => { + const onReply = (reply: unknown) => { + if (reply !== "barrier") return; + worker.off("message", onReply); + resolve(); + }; + worker.on("message", onReply); + // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin. + worker.postMessage("barrier"); + }), + ); + +it.effect("checkpoints the WAL on the worker instead of the writer", () => + withCheckpointClient((sql, filename) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const walSize = Effect.map(fs.stat(`${filename}-wal`), (info) => Number(info.size)); + const worker = startedWorkers.at(-1)!; + const limit = 1024 * 1024; + yield* sql.unsafe(`PRAGMA journal_size_limit = ${limit}`); + const autocheckpoint = yield* sql<{ + readonly wal_autocheckpoint: number; + }>`PRAGMA wal_autocheckpoint`; + // The writer leaves a WAL of this size to the worker. + assert.equal(autocheckpoint[0]?.wal_autocheckpoint, 10_000); + + // About 1200 frames, past the 1000 that start a checkpoint. + yield* sql`CREATE TABLE pages(payload BLOB)`; + yield* sql` + WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < 1200) + INSERT INTO pages(payload) SELECT randomblob(4000) FROM n + `; + yield* checkpointWorkerReply(worker); + const wal = yield* sql<{ + readonly log: number; + readonly checkpointed: number; + }>`PRAGMA wal_checkpoint(NOOP)`; + assert.isAbove(wal[0]!.log, 1000); + assert.equal(wal[0]!.checkpointed, wal[0]!.log); + + // Every frame is in the database, so the next commit restarts the WAL and + // cuts the file back to the size limit. + assert.isAbove(yield* walSize, limit); + yield* sql`INSERT INTO pages(payload) VALUES (x'00')`; + assert.isAtMost(yield* walSize, limit); + }), + ), +); + +it.effect("hands checkpoints back to the writer when the worker stops", () => + withCheckpointClient((sql) => + Effect.gen(function* () { + yield* Effect.promise(() => startedWorkers.at(-1)!.terminate()); + const autocheckpoint = yield* sql<{ + readonly wal_autocheckpoint: number; + }>`PRAGMA wal_autocheckpoint`; + assert.equal(autocheckpoint[0]?.wal_autocheckpoint, 1000); + }), + ), +); diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts index 2749ff9cd1b6291f575c315e1a00ca5680efe815..c8ddece4ac9baa9ddc6d4bd4de6256693d954835 100644 --- a/packages/shared/src/nodeSqliteClient.ts +++ b/packages/shared/src/nodeSqliteClient.ts @@ -47,6 +47,12 @@ export interface SqliteClientConfig { readonly readerWorker?: boolean | undefined; /** How long a reader worker may take to open the database before reads use the writer. Default 10 seconds. */ readonly readerStartupTimeout?: Duration.Input | undefined; + /** + * Run WAL checkpoints on a read-write connection in a worker thread, so + * copying pages into the database and syncing it never block the event loop. + * Ignored for in-memory databases. + */ + readonly checkpointWorker?: boolean | undefined; } /** @@ -477,12 +483,115 @@ const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) { }); }); +// SQLite's default wal_autocheckpoint, in WAL frames (pages). +const CHECKPOINT_FRAMES = 1000; +// While the worker runs, the writer still checkpoints a WAL this long. Writes +// with no pause leave the worker behind, and the WAL can only restart once +// every frame is checkpointed, so this keeps the file from growing forever. +const WRITER_BACKSTOP_FRAMES = 10 * CHECKPOINT_FRAMES; +// How long writes must pause before the worker checkpoints the rest of a long WAL. +const CHECKPOINT_IDLE_MS = 100; + +// Each message asks the worker to checkpoint if CHECKPOINT_FRAMES frames are +// waiting, and it replies with the same message when done. Once writes pause, it also checkpoints the +// frames that arrived during its last checkpoint, so the writer can restart the +// WAL. NOOP only counts frames. SQLite before 3.51 reads NOOP as PASSIVE and +// would checkpoint on every request, so the worker refuses to start there and +// the writer keeps checkpointing. +const checkpointWorkerSource = ` +const { parentPort, workerData } = require("node:worker_threads"); +const { DatabaseSync } = require("node:sqlite"); +const db = new DatabaseSync(workerData.filename); +const version = db.prepare("SELECT sqlite_version() AS version").get().version; +const [major, minor] = version.split(".").map(Number); +if (major < 3 || (major === 3 && minor < 51)) throw new Error("SQLite " + version + " has no NOOP checkpoint"); +// macOS fsync() does not flush the drive cache; F_FULLFSYNC does. Other platforms ignore this. +db.exec("PRAGMA checkpoint_fullfsync = ON;"); +const frames = db.prepare("PRAGMA wal_checkpoint(NOOP)"); +const checkpoint = db.prepare("PRAGMA wal_checkpoint(PASSIVE)"); +const finish = () => { + const { log, checkpointed } = frames.get(); + if (log >= workerData.frames && checkpointed < log) checkpoint.get(); +}; +let idle; +parentPort.on("message", (request) => { + const { log, checkpointed } = frames.get(); + if (log - checkpointed >= workerData.frames) checkpoint.get(); + clearTimeout(idle); + idle = setTimeout(finish, workerData.idleMs); + parentPort.postMessage(request); +}); +`; + +/** + * Moves the writer's WAL checkpoints to a worker. Call the returned function + * after each statement on the writer. While the worker runs, the writer only + * checkpoints at WRITER_BACKSTOP_FRAMES. If the worker fails or exits, the + * writer goes back to SQLite's default. + */ +const startCheckpointWorker = Effect.fnUntraced(function* ( + filename: string, + writer: NodeSqlite.DatabaseSync, +) { + let worker: NodeWorkerThreads.Worker; + try { + worker = new NodeWorkerThreads.Worker(checkpointWorkerSource, { + eval: true, + workerData: { filename, frames: CHECKPOINT_FRAMES, idleMs: CHECKPOINT_IDLE_MS }, + }); + } catch { + // No worker threads here: the writer keeps checkpointing. + return undefined; + } + writer.exec(`PRAGMA wal_autocheckpoint = ${WRITER_BACKSTOP_FRAMES};`); + let stopped = false; + let checking = false; + let checkAgain = false; + const handBack = () => { + if (stopped) return; + stopped = true; + try { + writer.exec(`PRAGMA wal_autocheckpoint = ${CHECKPOINT_FRAMES};`); + } catch { + // The writer is already closed. + } + }; + worker.on("error", handBack); + worker.on("exit", handBack); + yield* Effect.addFinalizer(() => Effect.promise(() => worker.terminate())); + + // At most one request is in flight. A statement that runs during it asks for + // one more check, so the frames of a last commit before idle still get one. + const check = () => { + if (stopped) return; + if (checking) { + checkAgain = true; + return; + } + checking = true; + // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin. + worker.postMessage(null); + }; + worker.on("message", () => { + checking = false; + if (checkAgain) { + checkAgain = false; + check(); + } + }); + // After the listeners: adding a "message" listener refs the worker again. + // A checkpoint in progress must not keep the process alive. + worker.unref(); + return check; +}); + const make = Effect.fn("makeWithDatabase")(function* ( options: SqliteClientConfig, ): Effect.fn.Return { yield* checkNodeSqliteCompat(); const compiler = Statement.makeCompilerSqlite(options.transformQueryNames); + const isFileDatabase = options.filename !== ":memory:" && options.filename !== ""; const transformRows = options.transformResultNames ? Statement.defaultTransforms(options.transformResultNames).array : undefined; @@ -510,6 +619,11 @@ const make = Effect.fn("makeWithDatabase")(function* ( }), }).pipe(Effect.orDie), ); + // Started after the close finalizer, so the worker stops before the writer closes. + const checkpoint = + options.checkpointWorker && isFileDatabase + ? yield* startCheckpointWorker(options.filename, db) + : undefined; const statementReaderCache = new WeakMap(); const hasRows = (statement: NodeSqlite.StatementSync): boolean => { @@ -550,9 +664,12 @@ const make = Effect.fn("makeWithDatabase")(function* ( try { statement.setReadBigInts(Boolean(Context.get(fiber.context, Client.SafeIntegers))); if (hasRows(statement)) { - return Effect.succeed(statement.all(...(params as any))); + const rows = statement.all(...(params as any)); + checkpoint?.(); + return Effect.succeed(rows); } const result = statement.run(...(params as any)); + checkpoint?.(); return Effect.succeed(raw ? (result as unknown as ReadonlyArray) : []); } catch (cause) { return Effect.fail( @@ -581,11 +698,14 @@ const make = Effect.fn("makeWithDatabase")(function* ( if (hasRows(statement)) { statement.setReturnArrays(true); // Safe to cast to array after we've setReturnArrays(true) - return statement.all(...(params as any)) as unknown as ReadonlyArray< + const rows = statement.all(...(params as any)) as unknown as ReadonlyArray< ReadonlyArray >; + checkpoint?.(); + return rows; } statement.run(...(params as any)); + checkpoint?.(); return []; }, catch: (cause) => @@ -646,9 +766,7 @@ const make = Effect.fn("makeWithDatabase")(function* ( const semaphore = yield* Semaphore.make(1); const connection = yield* makeConnection; const acquireReader = - options.readerWorker && options.filename !== ":memory:" && options.filename !== "" - ? yield* makeReader(options) - : undefined; + options.readerWorker && isFileDatabase ? yield* makeReader(options) : undefined; const readerSemaphore = yield* Semaphore.make(1); const writer = { lock: semaphore, connection };