From 48c8f322844087ff6cd80885f093ad44557647d1 Mon Sep 17 00:00:00 2001 From: Armin Ronacher Date: Tue, 20 Oct 2026 15:09:02 +0200 Subject: [PATCH] feat(chord): exclusive SQLite operations and a replicated connection wrapper NodeSqliteDatabase.exclusive() runs an operation alone on the connection, outside a transaction, and rolls back a transaction the operation left open. The internal replicated connection wrapper opens a database file with automatic checkpoints off, reports every operation, and runs a hook before SQLite checkpoints and deletes the WAL on close. It is not exported from the package yet. --- packages/chord/CHANGELOG.md | 1 + packages/chord/README.md | 2 +- packages/chord/src/sqlite/node.ts | 60 +++++++- packages/chord/src/sqlite/replication/node.ts | 139 +++++++++++++++++ packages/chord/test/sqlite/node.test.ts | 63 ++++++++ .../test/sqlite/replication-node.test.ts | 142 ++++++++++++++++++ 6 files changed, 400 insertions(+), 7 deletions(-) create mode 100644 packages/chord/src/sqlite/replication/node.ts create mode 100644 packages/chord/test/sqlite/replication-node.test.ts diff --git a/packages/chord/CHANGELOG.md b/packages/chord/CHANGELOG.md index 59ea796d3..fcf7e6881 100644 --- a/packages/chord/CHANGELOG.md +++ b/packages/chord/CHANGELOG.md @@ -11,6 +11,7 @@ - `@earendil-works/chord/telemetry-sqlite`: portable SQLite telemetry recording with bounded settled-span queues, ten-second metric aggregates, periodic flushes, and retention cleanup. - `Context.telemetry` is an explicit, non-optional telemetry parent (no-op by default). `withTelemetry(telemetry, context)` derives it without changing values or cancellation; `traceIn(context, options, fn)` starts a child span and passes its derived context to the callback. - `@earendil-works/chord/sqlite`: portable SQLite interfaces, atomic migration runner, and schema-prefix validation, with Node (`/sqlite/node`) and Cloudflare Durable Object (`/sqlite/cloudflare`) drivers. Moved from Durable, which keeps its storage implementations and re-exports the moved public names. +- `@earendil-works/chord/sqlite/node`: `NodeSqliteDatabase.exclusive(operation)` runs an operation alone on the connection, outside a transaction, after every earlier call and before every later one. - `@earendil-works/chord/output`: `Progress` takes an optional `ProgressClock` (`now`, `setTimer`) to pace commits on, by default `Date.now` and `setTimeout`. ## [1.3.0] - 2026-10-15 diff --git a/packages/chord/README.md b/packages/chord/README.md index 20f595ccf..2b732c448 100644 --- a/packages/chord/README.md +++ b/packages/chord/README.md @@ -21,7 +21,7 @@ Contracts are runtime values described with [TypeBox](https://github.com/sinclai | `@earendil-works/chord/output` | Bounded output streams and paced progress: `OutputBuffer` (head or tail window of a stream, `OutputLimits`, `OutputSkip`), `boundOutput`, `sanitizeOutput`, `utf8ByteLength`, and `Progress` (adaptive commits of a changing value). | Node and browsers | | `@earendil-works/chord/telemetry-sqlite` | `createSqliteTelemetry(db, { retentionMs?, maxQueuedSpans? })`: buffered recording and backend-independent `TelemetryStore` queries. | Any SQLite adapter | | `@earendil-works/chord/sqlite` | Portable `SqliteDatabase`, `SqliteExecutor`, and `SqliteValue`, atomic `applySqliteMigrations(database, migrations, { table })`, and `validateSqliteSchemaPrefix`. The metadata table must be a valid, non-reserved SQLite identifier; migration versions start at 1 and are contiguous. | Any SQLite adapter | -| `@earendil-works/chord/sqlite/node` | `NodeSqliteDatabase`, `openNodeSqliteDatabase`, and `NodeSqliteDatabaseOptions`. | Node | +| `@earendil-works/chord/sqlite/node` | `NodeSqliteDatabase` (with `exclusive(operation)` for work that must run alone on the connection outside a transaction, such as checkpoints), `openNodeSqliteDatabase`, and `NodeSqliteDatabaseOptions`. | Node | | `@earendil-works/chord/sqlite/cloudflare` | `DurableObjectSqliteDatabase` and the structural `DurableObjectSqliteStorage` binding type. | Cloudflare Durable Objects | ## Quick start diff --git a/packages/chord/src/sqlite/node.ts b/packages/chord/src/sqlite/node.ts index b156477e8..818f5ac87 100644 --- a/packages/chord/src/sqlite/node.ts +++ b/packages/chord/src/sqlite/node.ts @@ -15,7 +15,7 @@ export type NodeSqliteDatabaseOptions = { const DEFAULT_WAL_AUTO_CHECKPOINT_PAGES = 1_000; const DEFAULT_BUSY_TIMEOUT_MS = 5_000; -type TransactionScope = { active: boolean }; +type HandleScope = { active: boolean }; const ignore = (): void => {}; @@ -113,16 +113,19 @@ abstract class NodeSqliteExecutor implements SqliteExecutor { } } -class NodeSqliteTransaction extends NodeSqliteExecutor { - private readonly scope: TransactionScope; +/** The handle passed to a transaction or exclusive operation; it runs SQL directly and is invalid once its scope ends. */ +class NodeSqliteScopedExecutor extends NodeSqliteExecutor { + private readonly scope: HandleScope; + private readonly kind: string; - constructor(database: DatabaseSync, statements: Map, scope: TransactionScope) { + constructor(database: DatabaseSync, statements: Map, scope: HandleScope, kind: string) { super(database, statements); this.scope = scope; + this.kind = kind; } protected async runOperation(operation: () => T): Promise { - if (!this.scope.active) throw new Error("SQLite transaction handle is no longer active"); + if (!this.scope.active) throw new Error(`SQLite ${this.kind} handle is no longer active`); return operation(); } } @@ -141,7 +144,9 @@ export class NodeSqliteDatabase extends NodeSqliteExecutor implements SqliteData this.database.exec("BEGIN IMMEDIATE"); const scope = { active: true }; try { - const result = await callback(new NodeSqliteTransaction(this.database, this.statements, scope)); + const result = await callback( + new NodeSqliteScopedExecutor(this.database, this.statements, scope, "transaction"), + ); scope.active = false; this.database.exec("COMMIT"); return result; @@ -157,6 +162,49 @@ export class NodeSqliteDatabase extends NodeSqliteExecutor implements SqliteData }); } + /** + * Runs `operation` alone on the connection, outside a transaction: it starts after every earlier operation and + * transaction has settled, and later calls wait until it settles. The handle runs SQL immediately and is invalid + * afterwards. Use it for work that must not interleave with commits but cannot run inside a transaction, such as + * `PRAGMA wal_checkpoint`, or to hold back all later operations while awaiting something else. As with + * `transaction`, calling the database itself from inside the operation waits for the operation and never settles. + * + * An operation that begins a transaction must end it. Otherwise later statements would silently join it and + * never commit, so a transaction still open when the operation settles is rolled back and the call rejects, with + * the operation's own error as the `cause` if it failed. + */ + exclusive(operation: (executor: SqliteExecutor) => Promise): Promise { + return this.access.runAsync(async () => { + const scope = { active: true }; + let outcome: { readonly ok: true; readonly value: T } | { readonly ok: false; readonly error: unknown }; + try { + const handle = new NodeSqliteScopedExecutor(this.database, this.statements, scope, "exclusive"); + outcome = { ok: true, value: await operation(handle) }; + } catch (error) { + outcome = { ok: false, error }; + } finally { + scope.active = false; + } + if (this.database.isTransaction) { + const leftOpen = new Error( + "SQLite exclusive operation left a transaction open; it was rolled back", + outcome.ok ? undefined : { cause: outcome.error }, + ); + try { + this.database.exec("ROLLBACK"); + } catch (rollbackError) { + throw new AggregateError( + [leftOpen, rollbackError], + "SQLite exclusive operation left a transaction open and rollback failed", + ); + } + throw leftOpen; + } + if (!outcome.ok) throw outcome.error; + return outcome.value; + }); + } + close(): Promise { return this.access.run(() => { if (this.closed) return; diff --git a/packages/chord/src/sqlite/replication/node.ts b/packages/chord/src/sqlite/replication/node.ts new file mode 100644 index 000000000..51d0431c0 --- /dev/null +++ b/packages/chord/src/sqlite/replication/node.ts @@ -0,0 +1,139 @@ +import { resolve } from "node:path"; +import type { SqliteDatabase, SqliteExecutor, SqliteValue } from "../database.ts"; +import { type NodeSqliteDatabase, openNodeSqliteDatabase } from "../node.ts"; + +/** What the replication engine observes on a replicated connection. See `ReplicatedNodeSqliteDatabase`. */ +export interface ReplicationHooks { + /** + * Called synchronously after any operation through the connection settles, whether it succeeded or not, because + * it may have committed. Must not throw and must not call the database. + */ + afterOperation(): void; + /** + * Runs during `close`, alone on the connection, after every earlier operation and before SQLite checkpoints and + * deletes the WAL. Everything the engine has not taken out of the WAL by the time this settles is lost to it. + */ + beforeClose(executor: SqliteExecutor): Promise; +} + +/** Connection settings for a replicated database file. Automatic WAL checkpoints are always off. */ +export type ReplicatedNodeSqliteDatabaseOptions = { + /** Time SQLite waits for a competing file lock; see `NodeSqliteDatabaseOptions.busyTimeoutMs`. */ + readonly busyTimeoutMs?: number; +}; + +/** + * A Node SQLite connection whose write-ahead log (WAL) the replication engine reads from outside SQLite. + * + * The engine ships the pages SQLite appends to `{path}-wal`. That only works if no page leaves the WAL before the + * engine has read it, and a page leaves the WAL whenever anything checkpoints: SQLite's automatic checkpoint after a + * commit, a checkpoint by another connection, and the checkpoint SQLite runs when the last connection closes. This + * wrapper exists so that the engine, and only the engine, decides when that happens on this connection: + * + * - It opens the file itself, with automatic checkpoints disabled, so that the engine is attached before anyone else + * can write. pi-durable runs schema migrations on open, and those commits must be replicated like any other. + * - It owns the underlying connection and never hands it out, so every operation goes through the wrapper. After + * each one it calls `afterOperation`, which is how the engine learns that the WAL may have grown. That includes + * statements outside a transaction, which commit on their own. + * - It exposes `exclusive` so the engine can checkpoint, or hold back every later operation, without interleaving + * with a commit. + * - `close` first runs `beforeClose` in the connection's queue, after everything queued earlier, and only then lets + * the connection close. Operations called after `close` reject, so nothing can commit between the two steps. + * + * The wrapper does not read the WAL, schedule flushes or checkpoint; that is the engine's work, through the hooks. + * It also cannot stop another process from opening the file, which may checkpoint behind the engine's back. The + * engine detects that from the WAL salts and recovers by taking a new base. + * + * The class is exported as a type only, so the only way to get an instance is `openReplicatedNodeSqliteDatabase`. + * A constructor that accepted any `NodeSqliteDatabase` would let a caller pass one with automatic checkpoints on, or + * keep writing through the connection directly, and either would drop WAL pages before the engine read them. + */ +class ReplicatedNodeSqliteDatabase implements SqliteDatabase { + /** Absolute path of the database file. The WAL is `${path}-wal`. */ + readonly path: string; + private readonly database: NodeSqliteDatabase; + private readonly hooks: ReplicationHooks; + private closing: Promise | undefined; + + constructor(path: string, database: NodeSqliteDatabase, hooks: ReplicationHooks) { + this.path = path; + this.database = database; + this.hooks = hooks; + } + + exec(sql: string): Promise { + return this.observe(() => this.database.exec(sql)); + } + + run(sql: string, ...params: SqliteValue[]): Promise { + return this.observe(() => this.database.run(sql, ...params)); + } + + get(sql: string, ...params: SqliteValue[]): Promise { + return this.observe(() => this.database.get(sql, ...params)); + } + + all(sql: string, ...params: SqliteValue[]): Promise { + return this.observe(() => this.database.all(sql, ...params)); + } + + transaction(callback: (transaction: SqliteExecutor) => Promise): Promise { + return this.observe(() => this.database.transaction(callback)); + } + + /** Runs `operation` alone on the connection; see `NodeSqliteDatabase.exclusive`. */ + exclusive(operation: (executor: SqliteExecutor) => Promise): Promise { + if (this.closing !== undefined) return Promise.reject(closedError()); + return this.database.exclusive(operation); + } + + close(): Promise { + this.closing ??= this.closeOnce(); + return this.closing; + } + + private async closeOnce(): Promise { + let hookError: unknown; + let hookFailed = false; + try { + await this.database.exclusive((executor) => this.hooks.beforeClose(executor)); + } catch (error) { + hookFailed = true; + hookError = error; + } + try { + await this.database.close(); + } catch (closeError) { + if (hookFailed) throw new AggregateError([hookError, closeError], "Replication close hook and close failed"); + throw closeError; + } + if (hookFailed) throw hookError; + } + + // Delegating synchronously keeps call order: the operation enters the connection's queue before this returns. + private observe(operation: () => Promise): Promise { + if (this.closing !== undefined) return Promise.reject(closedError()); + return operation().finally(() => this.hooks.afterOperation()); + } +} + +export type { ReplicatedNodeSqliteDatabase }; + +const closedError = (): Error => new Error("Replicated SQLite database is closed"); + +/** Open a database file for replication. The file needs a WAL, so in-memory and temporary databases are refused. */ +export async function openReplicatedNodeSqliteDatabase( + path: string, + hooks: ReplicationHooks, + options: ReplicatedNodeSqliteDatabaseOptions = {}, +): Promise { + if (path === "" || path === ":memory:" || path.startsWith("file:")) { + throw new Error(`Replication needs a database file path, got ${JSON.stringify(path)}`); + } + const absolute = resolve(path); + const database = await openNodeSqliteDatabase(absolute, { + walAutoCheckpointPages: 0, + busyTimeoutMs: options.busyTimeoutMs, + }); + return new ReplicatedNodeSqliteDatabase(absolute, database, hooks); +} diff --git a/packages/chord/test/sqlite/node.test.ts b/packages/chord/test/sqlite/node.test.ts index 11098d1c5..d9edc14c8 100644 --- a/packages/chord/test/sqlite/node.test.ts +++ b/packages/chord/test/sqlite/node.test.ts @@ -153,6 +153,69 @@ describe("Node SQLite driver", () => { await database.close(); }); + it("runs an exclusive operation alone, outside a transaction, after earlier calls", async () => { + const database = await openNodeSqliteDatabase(":memory:"); + await database.exec("CREATE TABLE exclusive_probe (value INTEGER)"); + const earlier = database.run("INSERT INTO exclusive_probe (value) VALUES (?)", 1); + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + let handle!: SqliteExecutor; + const exclusive = database.exclusive(async (executor) => { + handle = executor; + const seen = await executor.all("SELECT value FROM exclusive_probe"); + // Not inside a transaction: the handle may begin and commit one itself. + await executor.exec("BEGIN; INSERT INTO exclusive_probe (value) VALUES (2); COMMIT"); + await gate; + return seen; + }); + let laterSettled = false; + const later = database.all("SELECT value FROM exclusive_probe ORDER BY value").finally(() => { + laterSettled = true; + }); + await earlier; + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(laterSettled).toBe(false); + + release(); + await expect(exclusive).resolves.toEqual([{ value: 1 }]); + await expect(later).resolves.toEqual([{ value: 1 }, { value: 2 }]); + await expect(handle.exec("SELECT 1")).rejects.toThrow("SQLite exclusive handle is no longer active"); + await expect(database.exclusive(async () => Promise.reject(new Error("failed")))).rejects.toThrow("failed"); + expect(await database.get("SELECT count(*) AS count FROM exclusive_probe")).toEqual({ count: 2 }); + await database.close(); + }); + + it("rolls back a transaction an exclusive operation left open and rejects", async () => { + const database = await openNodeSqliteDatabase(":memory:"); + await database.exec("CREATE TABLE left_open (value INTEGER)"); + const leftOpen = "SQLite exclusive operation left a transaction open; it was rolled back"; + + await expect( + database.exclusive(async (executor) => { + await executor.exec("BEGIN; INSERT INTO left_open (value) VALUES (1)"); + return "returned without committing"; + }), + ).rejects.toThrow(leftOpen); + + const failure = new Error("failed mid-transaction"); + const rejected = database.exclusive(async (executor) => { + await executor.exec("BEGIN IMMEDIATE; INSERT INTO left_open (value) VALUES (2)"); + throw failure; + }); + await expect(rejected).rejects.toThrow(leftOpen); + await expect(rejected).rejects.toMatchObject({ cause: failure }); + + // Later calls run outside any leftover transaction: autocommit writes commit and transactions can begin. + await database.run("INSERT INTO left_open (value) VALUES (?)", 3); + await database.transaction(async (transaction) => { + await transaction.run("INSERT INTO left_open (value) VALUES (?)", 4); + }); + expect(await database.all("SELECT value FROM left_open ORDER BY value")).toEqual([{ value: 3 }, { value: 4 }]); + await database.close(); + }); + it("does not preserve the callback error when rollback itself fails", async () => { const database = await openNodeSqliteDatabase(":memory:"); await database.exec("CREATE TABLE rollback_probe (value INTEGER)"); diff --git a/packages/chord/test/sqlite/replication-node.test.ts b/packages/chord/test/sqlite/replication-node.test.ts new file mode 100644 index 000000000..5b333f7d7 --- /dev/null +++ b/packages/chord/test/sqlite/replication-node.test.ts @@ -0,0 +1,142 @@ +import { existsSync, mkdtempSync, readFileSync, rmSync, statSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import type { SqliteExecutor } from "../../src/sqlite/index.ts"; +import * as replication from "../../src/sqlite/replication/node.ts"; +import { openReplicatedNodeSqliteDatabase, type ReplicationHooks } from "../../src/sqlite/replication/node.ts"; + +const WAL_HEADER_BYTES = 32; +const WAL_FRAME_HEADER_BYTES = 24; + +/** Committed frames in a WAL file, counted from its header without validating checksums. */ +function walCommitFrames(path: string): number { + const wal = readFileSync(path); + if (wal.length < WAL_HEADER_BYTES) return 0; + const pageSize = wal.readUInt32BE(8); + let commits = 0; + for (let offset = WAL_HEADER_BYTES; offset + WAL_FRAME_HEADER_BYTES + pageSize <= wal.length; ) { + if (wal.readUInt32BE(offset + 4) !== 0) commits++; + offset += WAL_FRAME_HEADER_BYTES + pageSize; + } + return commits; +} + +function recordingHooks(beforeClose: (executor: SqliteExecutor) => Promise = async () => {}) { + const hooks = { + operations: 0, + afterOperation() { + hooks.operations++; + }, + beforeClose, + } satisfies ReplicationHooks & { operations: number }; + return hooks; +} + +describe("replicated Node SQLite connection", () => { + let directory: string; + let path: string; + + beforeEach(() => { + directory = mkdtempSync(join(tmpdir(), "chord-replication-")); + path = join(directory, "replica.sqlite"); + }); + + afterEach(() => { + rmSync(directory, { recursive: true, force: true }); + }); + + it("can only be created by opening a file", () => { + // A constructor would accept connections with automatic checkpoints on, or ones the caller still writes through. + expect(Object.keys(replication).sort()).toEqual(["openReplicatedNodeSqliteDatabase"]); + }); + + it("refuses databases without a WAL file", async () => { + for (const target of ["", ":memory:", "file::memory:"]) { + await expect(openReplicatedNodeSqliteDatabase(target, recordingHooks())).rejects.toThrow( + "Replication needs a database file path", + ); + } + }); + + it("opens in WAL mode and never checkpoints on its own", async () => { + const database = await openReplicatedNodeSqliteDatabase(path, recordingHooks()); + expect(database.path).toBe(path); + expect(await database.get("PRAGMA journal_mode")).toEqual({ journal_mode: "wal" }); + expect(await database.get("PRAGMA wal_autocheckpoint")).toEqual({ wal_autocheckpoint: 0 }); + await database.exec("CREATE TABLE pages (value BLOB)"); + // SQLite's default would checkpoint once the WAL passes 1,000 pages; write well past that. + for (let batch = 0; batch < 12; batch++) { + await database.transaction(async (transaction) => { + for (let row = 0; row < 100; row++) + await transaction.run("INSERT INTO pages (value) VALUES (randomblob(4000))"); + }); + } + expect(walCommitFrames(`${path}-wal`)).toBe(13); + expect(statSync(`${path}-wal`).size).toBeGreaterThan(1_200 * 4_096); + await database.close(); + }); + + it("reports every operation, including failed ones and statements outside transactions", async () => { + const hooks = recordingHooks(); + const database = await openReplicatedNodeSqliteDatabase(path, hooks); + await database.exec("CREATE TABLE notes (value TEXT NOT NULL)"); + await database.run("INSERT INTO notes (value) VALUES (?)", "autocommit"); + await database.get("SELECT value FROM notes"); + await database.all("SELECT value FROM notes"); + await database.transaction(async (transaction) => { + await transaction.run("INSERT INTO notes (value) VALUES (?)", "transaction"); + }); + await expect(database.run("INSERT INTO notes (value) VALUES (NULL)")).rejects.toThrow(); + expect(hooks.operations).toBe(6); + await database.exclusive(async (executor) => { + await executor.exec("PRAGMA wal_checkpoint(TRUNCATE)"); + }); + expect(hooks.operations).toBe(6); + await database.close(); + }); + + it("runs the close hook after earlier operations, while the WAL holds every commit", async () => { + let framesAtClose = -1; + let rowsAtClose: unknown; + const database = await openReplicatedNodeSqliteDatabase( + path, + recordingHooks(async (executor) => { + framesAtClose = walCommitFrames(`${path}-wal`); + rowsAtClose = await executor.get("SELECT count(*) AS count FROM notes"); + }), + ); + await database.exec("CREATE TABLE notes (value TEXT)"); + // Called before close and still queued when close is called: these must commit before the hook runs. + const pending = [1, 2, 3].map((value) => database.run("INSERT INTO notes (value) VALUES (?)", String(value))); + const closed = database.close(); + expect(database.close()).toBe(closed); + await Promise.all(pending); + await closed; + + expect(framesAtClose).toBe(4); + expect(rowsAtClose).toEqual({ count: 3 }); + // SQLite checkpoints and removes the WAL when the last connection closes. + expect(existsSync(`${path}-wal`)).toBe(false); + await expect(database.run("INSERT INTO notes (value) VALUES (?)", "late")).rejects.toThrow( + "Replicated SQLite database is closed", + ); + await expect(database.exclusive(async () => {})).rejects.toThrow("Replicated SQLite database is closed"); + }); + + it("closes the connection even when the close hook fails", async () => { + const database = await openReplicatedNodeSqliteDatabase( + path, + recordingHooks(async () => { + throw new Error("spool failed"); + }), + ); + await database.exec("CREATE TABLE notes (value TEXT)"); + await expect(database.close()).rejects.toThrow("spool failed"); + expect(existsSync(`${path}-wal`)).toBe(false); + + const reopened = await openReplicatedNodeSqliteDatabase(path, recordingHooks()); + expect(await reopened.all("SELECT value FROM notes")).toEqual([]); + await reopened.close(); + }); +});