From c268154ca1171c2f9eefe75d3d7372a68aada2d5 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Sat, 17 Oct 2026 16:48:29 +0000 Subject: [PATCH] feat(chord): move Durable's SQLite drivers into chord/sqlite --- packages/chord/CHANGELOG.md | 4 + packages/chord/README.md | 3 + packages/chord/package.json | 15 ++ packages/chord/src/sqlite/cloudflare.ts | 180 +++++++++++++++ packages/chord/src/sqlite/database.ts | 38 ++++ packages/chord/src/sqlite/index.ts | 3 + packages/chord/src/sqlite/migrations.ts | 43 ++++ packages/chord/src/sqlite/node.ts | 201 +++++++++++++++++ packages/chord/src/sqlite/schema.ts | 9 + packages/chord/test/sqlite/cloudflare.test.ts | 116 ++++++++++ packages/chord/test/sqlite/helpers.ts | 65 ++++++ packages/chord/test/sqlite/migrations.test.ts | 102 +++++++++ packages/chord/test/sqlite/node.test.ts | 169 ++++++++++++++ packages/durable/CHANGELOG.md | 2 + .../durable/src/storage/sqlite/cloudflare.ts | 181 +-------------- .../durable/src/storage/sqlite/database.ts | 39 +--- packages/durable/src/storage/sqlite/index.ts | 2 +- .../durable/src/storage/sqlite/migrations.ts | 41 +--- packages/durable/src/storage/sqlite/node.ts | 208 +----------------- packages/durable/src/storage/sqlite/schema.ts | 14 +- .../durable/src/storage/sqlite/storage.ts | 2 +- .../durable/test/sqlite-cloudflare.test.ts | 180 +-------------- packages/durable/test/sqlite-facade.test.ts | 158 ++----------- .../durable/test/sqlite-migrations.test.ts | 38 ---- tsconfig.json | 3 + 25 files changed, 991 insertions(+), 825 deletions(-) create mode 100644 packages/chord/src/sqlite/cloudflare.ts create mode 100644 packages/chord/src/sqlite/database.ts create mode 100644 packages/chord/src/sqlite/index.ts create mode 100644 packages/chord/src/sqlite/migrations.ts create mode 100644 packages/chord/src/sqlite/node.ts create mode 100644 packages/chord/src/sqlite/schema.ts create mode 100644 packages/chord/test/sqlite/cloudflare.test.ts create mode 100644 packages/chord/test/sqlite/helpers.ts create mode 100644 packages/chord/test/sqlite/migrations.test.ts create mode 100644 packages/chord/test/sqlite/node.test.ts diff --git a/packages/chord/CHANGELOG.md b/packages/chord/CHANGELOG.md index 742c5ce63..49089b5ca 100644 --- a/packages/chord/CHANGELOG.md +++ b/packages/chord/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Added + +- `@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. + ## [1.3.0] - 2026-10-15 ### Breaking Changes diff --git a/packages/chord/README.md b/packages/chord/README.md index 6ba689043..157b4b51d 100644 --- a/packages/chord/README.md +++ b/packages/chord/README.md @@ -19,6 +19,9 @@ Contracts are runtime values described with [TypeBox](https://github.com/sinclai | `@earendil-works/chord/context` | Explicit cancellation and invocation values: `BACKGROUND_CONTEXT`, `withCancel`, `withAbortSignal`, `withoutAbortSignal`, `withContextValue`, `awaitWithContext`. | Node and browsers | | `@earendil-works/chord/delta` | The transactional JSON tracker, structural diff, immutable apply, and the wire encoder/decoder for operation batches. | Node and browsers | | `@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/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/cloudflare` | `DurableObjectSqliteDatabase` and the structural `DurableObjectSqliteStorage` binding type. | Cloudflare Durable Objects | ## Quick start diff --git a/packages/chord/package.json b/packages/chord/package.json index 8c7566537..973063fb5 100644 --- a/packages/chord/package.json +++ b/packages/chord/package.json @@ -31,6 +31,21 @@ "types": "./dist/node.d.ts", "import": "./dist/node.js" }, + "./sqlite": { + "source": "./src/sqlite/index.ts", + "types": "./dist/sqlite/index.d.ts", + "import": "./dist/sqlite/index.js" + }, + "./sqlite/node": { + "source": "./src/sqlite/node.ts", + "types": "./dist/sqlite/node.d.ts", + "import": "./dist/sqlite/node.js" + }, + "./sqlite/cloudflare": { + "source": "./src/sqlite/cloudflare.ts", + "types": "./dist/sqlite/cloudflare.d.ts", + "import": "./dist/sqlite/cloudflare.js" + }, "./package.json": "./package.json" }, "sideEffects": false, diff --git a/packages/chord/src/sqlite/cloudflare.ts b/packages/chord/src/sqlite/cloudflare.ts new file mode 100644 index 000000000..897a660b7 --- /dev/null +++ b/packages/chord/src/sqlite/cloudflare.ts @@ -0,0 +1,180 @@ +import type { SqliteDatabase, SqliteExecutor, SqliteValue } from "./database.ts"; + +/** Values a Durable Object SQL binding accepts and returns. */ +type DurableObjectSqlValue = ArrayBuffer | string | number | null; + +/** + * The parts of a SQLite-backed Durable Object's `ctx.storage` this adapter uses, typed structurally so the package + * needs no Workers type dependency. + */ +export type DurableObjectSqliteStorage = { + readonly sql: { + exec( + query: string, + ...bindings: DurableObjectSqlValue[] + ): { + next(): + | { readonly done?: false; readonly value: Record } + | { readonly done: true; readonly value?: undefined }; + toArray(): Record[]; + }; + }; + transaction(closure: () => Promise): Promise; +}; + +/** Runs synchronous operations immediately when idle and holds queued calls behind a transaction. */ +class SerialQueue { + private tail: Promise = Promise.resolve(); + private pending = 0; + + run(operation: () => T | Promise): Promise { + if (this.pending > 0) return this.enqueue(operation); + try { + return Promise.resolve(operation()); + } catch (error) { + return Promise.reject(error); + } + } + + hold(operation: () => Promise): Promise { + return this.enqueue(operation); + } + + private enqueue(operation: () => T | Promise): Promise { + this.pending++; + const result = this.tail.then(operation).finally(() => { + this.pending--; + }); + this.tail = result.then( + () => {}, + () => {}, + ); + return result; + } +} + +const storageQueues = new WeakMap(); + +function queueFor(storage: DurableObjectSqliteStorage): SerialQueue { + let queue = storageQueues.get(storage); + if (queue === undefined) { + queue = new SerialQueue(); + storageQueues.set(storage, queue); + } + return queue; +} + +/** Durable Object SQL binds numbers as doubles: a `bigint` outside the safe integer range would lose precision. */ +function bindValue(value: SqliteValue): DurableObjectSqlValue { + if (value instanceof Uint8Array) return value.slice().buffer as ArrayBuffer; + if (typeof value !== "bigint") return value; + if (value < BigInt(Number.MIN_SAFE_INTEGER) || value > BigInt(Number.MAX_SAFE_INTEGER)) { + throw new RangeError(`Durable Object SQL cannot bind ${value} without losing precision`); + } + return Number(value); +} + +const bind = (params: readonly SqliteValue[]): DurableObjectSqlValue[] => params.map(bindValue); + +/** A row as returned, with `BLOB` columns converted to `Uint8Array` in place; no other copy is made. */ +function row(row: Record): T { + for (const key in row) { + const value = row[key]; + if (value instanceof ArrayBuffer) (row as Record)[key] = new Uint8Array(value); + } + return row as T; +} + +/** Executes SQL through `ctx.storage.sql`, which has no separate prepare step. */ +class DurableObjectSqliteExecutor implements SqliteExecutor { + protected readonly storage: DurableObjectSqliteStorage; + + constructor(storage: DurableObjectSqliteStorage) { + this.storage = storage; + } + + async exec(sql: string): Promise { + this.check(); + this.storage.sql.exec(sql); + } + + async run(sql: string, ...params: SqliteValue[]): Promise { + this.check(); + this.storage.sql.exec(sql, ...bind(params)); + } + + async get(sql: string, ...params: SqliteValue[]): Promise { + this.check(); + const result = this.storage.sql.exec(sql, ...bind(params)).next(); + return result.done ? undefined : row(result.value); + } + + async all(sql: string, ...params: SqliteValue[]): Promise { + this.check(); + return this.storage.sql + .exec(sql, ...bind(params)) + .toArray() + .map((value) => row(value)); + } + + /** Throws when this handle may no longer run SQL. */ + protected check(): void {} +} + +/** The handle a transaction callback receives; invalid once the callback settles. */ +class DurableObjectSqliteTransaction extends DurableObjectSqliteExecutor { + active = true; + + protected override check(): void { + if (!this.active) throw new Error("SQLite transaction handle is no longer active"); + } +} + +/** + * `SqliteDatabase` adapter for the SQLite storage of a Cloudflare Durable Object. Adapters given the same storage + * object share a queue; direct calls to the underlying storage bypass it. Integers are JavaScript numbers; binding a + * `bigint` outside the safe integer range throws. + */ +export class DurableObjectSqliteDatabase extends DurableObjectSqliteExecutor implements SqliteDatabase { + private readonly queue: SerialQueue; + + constructor(storage: DurableObjectSqliteStorage) { + super(storage); + this.queue = queueFor(storage); + } + + override exec(sql: string): Promise { + return this.queue.run(() => super.exec(sql)); + } + + override run(sql: string, ...params: SqliteValue[]): Promise { + return this.queue.run(() => super.run(sql, ...params)); + } + + override get(sql: string, ...params: SqliteValue[]): Promise { + return this.queue.run(() => super.get(sql, ...params)); + } + + override all(sql: string, ...params: SqliteValue[]): Promise { + return this.queue.run(() => super.all(sql, ...params)); + } + + /** `ctx.storage.transaction()` commits when the callback resolves and rolls back when it rejects. */ + transaction(callback: (transaction: SqliteExecutor) => Promise): Promise { + return this.queue.hold(() => + this.storage.transaction(async () => { + const transaction = new DurableObjectSqliteTransaction(this.storage); + try { + return await callback(transaction); + } finally { + transaction.active = false; + } + }), + ); + } + + /** Waits for queued work. The Durable Object owns its storage, so nothing else closes. */ + close(): Promise { + return this.queue.run(() => {}); + } +} diff --git a/packages/chord/src/sqlite/database.ts b/packages/chord/src/sqlite/database.ts new file mode 100644 index 000000000..fc5a814c6 --- /dev/null +++ b/packages/chord/src/sqlite/database.ts @@ -0,0 +1,38 @@ +/** Values supported by the portable SQLite drivers. */ +export type SqliteValue = null | number | bigint | string | Uint8Array; + +/** + * Asynchronous SQL operations shared by a database and its transaction handles. + * + * `exec` runs SQL text without bindings and may contain several statements. `run`, `get`, and `all` + * execute one statement with positional bindings. Adapters may cache prepared statements by SQL text, + * so callers pass values as bindings instead of interpolating them. + */ +export interface SqliteExecutor { + exec(sql: string): Promise; + run(sql: string, ...params: SqliteValue[]): Promise; + get(sql: string, ...params: SqliteValue[]): Promise; + all(sql: string, ...params: SqliteValue[]): Promise; +} + +/** + * Minimal portable SQLite database facade. + * + * All operations are asynchronous so adapters may execute outside the caller's runtime. + * + * `transaction` passes the callback a transaction handle. All work in the transaction + * must use that handle; the handle is invalid after the callback settles. Adapters must + * queue unrelated operations and other transactions until the transaction finishes. The + * returned promise settles after commit or rollback. Calling the database itself (including + * `transaction` or `close`) from inside a callback therefore waits for that transaction and + * never settles. + * + * When the callback rejects, the adapter must roll the transaction back before rejecting + * with that same error. If rollback fails, it must reject with a different error (for + * example an `AggregateError`) so callers cannot mistake the callback error for a + * guaranteed rollback. + */ +export interface SqliteDatabase extends SqliteExecutor { + transaction(callback: (transaction: SqliteExecutor) => Promise): Promise; + close(): Promise; +} diff --git a/packages/chord/src/sqlite/index.ts b/packages/chord/src/sqlite/index.ts new file mode 100644 index 000000000..322d40236 --- /dev/null +++ b/packages/chord/src/sqlite/index.ts @@ -0,0 +1,3 @@ +export type { SqliteDatabase, SqliteExecutor, SqliteValue } from "./database.ts"; +export { applySqliteMigrations, type SqliteMigration } from "./migrations.ts"; +export { validateSqliteSchemaPrefix } from "./schema.ts"; diff --git a/packages/chord/src/sqlite/migrations.ts b/packages/chord/src/sqlite/migrations.ts new file mode 100644 index 000000000..d4dc7efe8 --- /dev/null +++ b/packages/chord/src/sqlite/migrations.ts @@ -0,0 +1,43 @@ +import type { SqliteDatabase } from "./database.ts"; +import { validateSqliteSchemaPrefix } from "./schema.ts"; + +export type SqliteMigration = { + readonly version: number; + readonly statements: readonly string[]; +}; + +type SchemaRow = { readonly version: number }; + +/** Apply all pending migrations atomically, tracking their version in the named metadata table. */ +export async function applySqliteMigrations( + database: SqliteDatabase, + migrations: readonly SqliteMigration[], + options: { readonly table: string }, +): Promise { + const { table } = options; + validateSqliteSchemaPrefix(table); + for (let index = 0; index < migrations.length; index++) { + if (migrations[index]?.version !== index + 1) { + throw new Error("SQLite migrations must have contiguous versions starting at 1"); + } + } + + await database.transaction(async (transaction) => { + await transaction.exec(`CREATE TABLE IF NOT EXISTS ${table} ( + singleton INTEGER PRIMARY KEY CHECK (singleton = 1), + version INTEGER NOT NULL CHECK (version >= 0) + ) STRICT`); + await transaction.run(`INSERT OR IGNORE INTO ${table} (singleton, version) VALUES (1, 0)`); + const row = await transaction.get(`SELECT version FROM ${table} WHERE singleton = 1`); + if (row === undefined) throw new Error("SQLite schema metadata is missing"); + const currentVersion = migrations.at(-1)?.version ?? 0; + if (row.version > currentVersion) { + throw new Error(`SQLite schema version ${row.version} is newer than supported version ${currentVersion}`); + } + for (const migration of migrations) { + if (migration.version <= row.version) continue; + for (const statement of migration.statements) await transaction.exec(statement); + await transaction.run(`UPDATE ${table} SET version = ? WHERE singleton = 1`, migration.version); + } + }); +} diff --git a/packages/chord/src/sqlite/node.ts b/packages/chord/src/sqlite/node.ts new file mode 100644 index 000000000..b156477e8 --- /dev/null +++ b/packages/chord/src/sqlite/node.ts @@ -0,0 +1,201 @@ +import { mkdir } from "node:fs/promises"; +import { dirname } from "node:path"; +import type { StatementSync } from "node:sqlite"; +import { DatabaseSync } from "node:sqlite"; +import type { SqliteDatabase, SqliteExecutor, SqliteValue } from "./database.ts"; + +/** Node SQLite connection settings for a database file. */ +export type NodeSqliteDatabaseOptions = { + /** SQLite WAL auto-checkpoint threshold. SQLite and this adapter default to 1,000 pages; 0 disables it. */ + readonly walAutoCheckpointPages?: number; + /** Time SQLite waits for a competing file lock. SQLite defaults to 0; this adapter defaults to 5,000 ms. */ + readonly busyTimeoutMs?: number; +}; + +const DEFAULT_WAL_AUTO_CHECKPOINT_PAGES = 1_000; +const DEFAULT_BUSY_TIMEOUT_MS = 5_000; + +type TransactionScope = { active: boolean }; + +const ignore = (): void => {}; + +/** + * Runs operations in call order. An operation starts immediately when nothing is running or waiting; + * otherwise it waits for everything before it. An asynchronous operation holds the queue until it settles. + */ +class SerialOperationQueue { + private tail: Promise = Promise.resolve(); + private pending = 0; + + run(operation: () => T): Promise { + if (this.pending > 0) return this.enqueue(operation); + try { + return Promise.resolve(operation()); + } catch (error) { + return Promise.reject(error); + } + } + + runAsync(operation: () => Promise): Promise { + if (this.pending > 0) return this.enqueue(operation); + this.pending++; + // Publish the barrier before the operation starts, so calls it makes synchronously wait behind it. + const { promise: barrier, resolve: releaseBarrier } = Promise.withResolvers(); + this.tail = barrier; + let started: Promise; + try { + started = operation(); + } catch (error) { + started = Promise.reject(error); + } + return started.finally(() => { + this.pending--; + releaseBarrier(); + }); + } + + private enqueue(operation: () => T | Promise): Promise { + this.pending++; + return this.release(this.tail.then(operation)); + } + + private release(operation: Promise): Promise { + const settled = operation.finally(() => { + this.pending--; + }); + this.tail = settled.then(ignore, ignore); + return settled; + } +} + +/** + * Executes SQL on one connection. Prepared statements are cached per connection by SQL text, so the + * database and its transaction handles share them across transactions. + */ +abstract class NodeSqliteExecutor implements SqliteExecutor { + protected readonly database: DatabaseSync; + protected readonly statements: Map; + + constructor(database: DatabaseSync, statements: Map) { + this.database = database; + this.statements = statements; + } + + exec(sql: string): Promise { + return this.runOperation(() => { + this.database.exec(sql); + }); + } + + run(sql: string, ...params: SqliteValue[]): Promise { + return this.runOperation(() => { + this.statement(sql).run(...params); + }); + } + + get(sql: string, ...params: SqliteValue[]): Promise { + return this.runOperation(() => this.statement(sql).get(...params) as T | undefined); + } + + all(sql: string, ...params: SqliteValue[]): Promise { + return this.runOperation(() => this.statement(sql).all(...params) as T[]); + } + + protected abstract runOperation(operation: () => T): Promise; + + private statement(sql: string): StatementSync { + let statement = this.statements.get(sql); + if (statement === undefined) { + statement = this.database.prepare(sql); + this.statements.set(sql, statement); + } + return statement; + } +} + +class NodeSqliteTransaction extends NodeSqliteExecutor { + private readonly scope: TransactionScope; + + constructor(database: DatabaseSync, statements: Map, scope: TransactionScope) { + super(database, statements); + this.scope = scope; + } + + protected async runOperation(operation: () => T): Promise { + if (!this.scope.active) throw new Error("SQLite transaction handle is no longer active"); + return operation(); + } +} + +/** `SqliteDatabase` adapter backed by Node's built-in `node:sqlite`. */ +export class NodeSqliteDatabase extends NodeSqliteExecutor implements SqliteDatabase { + private readonly access = new SerialOperationQueue(); + private closed = false; + + constructor(database: DatabaseSync) { + super(database, new Map()); + } + + transaction(callback: (transaction: SqliteExecutor) => Promise): Promise { + return this.access.runAsync(async () => { + this.database.exec("BEGIN IMMEDIATE"); + const scope = { active: true }; + try { + const result = await callback(new NodeSqliteTransaction(this.database, this.statements, scope)); + scope.active = false; + this.database.exec("COMMIT"); + return result; + } catch (error) { + scope.active = false; + try { + this.database.exec("ROLLBACK"); + } catch (rollbackError) { + throw new AggregateError([error, rollbackError], "SQLite transaction failed and rollback failed"); + } + throw error; + } + }); + } + + close(): Promise { + return this.access.run(() => { + if (this.closed) return; + this.closed = true; + this.statements.clear(); + try { + this.database.exec("PRAGMA wal_checkpoint(TRUNCATE)"); + } finally { + this.database.close(); + } + }); + } + + protected runOperation(operation: () => T): Promise { + return this.access.run(operation); + } +} + +/** Open and configure a Node-backed SQLite database facade. */ +export async function openNodeSqliteDatabase( + path: string, + options: NodeSqliteDatabaseOptions = {}, +): Promise { + const checkpointPages = options.walAutoCheckpointPages ?? DEFAULT_WAL_AUTO_CHECKPOINT_PAGES; + const timeout = options.busyTimeoutMs ?? DEFAULT_BUSY_TIMEOUT_MS; + if (path !== ":memory:") await mkdir(dirname(path), { recursive: true }); + const database = new DatabaseSync(path, { timeout }); + const adapter = new NodeSqliteDatabase(database); + try { + await adapter.exec("PRAGMA journal_mode = WAL"); + await adapter.exec("PRAGMA synchronous = NORMAL"); + await adapter.exec(`PRAGMA wal_autocheckpoint = ${checkpointPages}`); + return adapter; + } catch (error) { + try { + await adapter.close(); + } catch { + // Preserve the configuration failure. + } + throw error; + } +} diff --git a/packages/chord/src/sqlite/schema.ts b/packages/chord/src/sqlite/schema.ts new file mode 100644 index 000000000..2ee01bc5e --- /dev/null +++ b/packages/chord/src/sqlite/schema.ts @@ -0,0 +1,9 @@ +/** Validate a prefix before using it as part of a SQLite identifier. */ +export function validateSqliteSchemaPrefix(prefix: string): void { + if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(prefix)) { + throw new Error(`Invalid SQLite schema prefix ${JSON.stringify(prefix)}`); + } + if (prefix.toLowerCase().startsWith("sqlite_")) { + throw new Error(`SQLite schema prefix ${JSON.stringify(prefix)} is reserved`); + } +} diff --git a/packages/chord/test/sqlite/cloudflare.test.ts b/packages/chord/test/sqlite/cloudflare.test.ts new file mode 100644 index 000000000..0e7521d25 --- /dev/null +++ b/packages/chord/test/sqlite/cloudflare.test.ts @@ -0,0 +1,116 @@ +import { DatabaseSync } from "node:sqlite"; +import { describe, expect, it } from "vitest"; +import { DurableObjectSqliteDatabase } from "../../src/sqlite/cloudflare.ts"; +import { durableObjectStorage } from "./helpers.ts"; + +describe("Durable Object SQLite driver", () => { + it("binds and returns BLOB values as Uint8Array", async () => { + const database = new DurableObjectSqliteDatabase(durableObjectStorage()); + await database.exec("CREATE TABLE blobs (id INTEGER PRIMARY KEY, data BLOB, n INTEGER)"); + await database.run("INSERT INTO blobs (id, data, n) VALUES (?, ?, ?)", 1, new Uint8Array([1, 2, 3]), 7n); + expect(await database.get("SELECT data, n FROM blobs WHERE id = ?", 1)).toEqual({ + data: new Uint8Array([1, 2, 3]), + n: 7, + }); + }); + + it("reads only the cursor's first row for get", async () => { + const operations: string[] = []; + const database = new DurableObjectSqliteDatabase( + durableObjectStorage(new DatabaseSync(":memory:"), (operation) => operations.push(operation)), + ); + await database.exec("CREATE TABLE items (id INTEGER PRIMARY KEY)"); + await database.run("INSERT INTO items (id) VALUES (1), (2)"); + expect(await database.get("SELECT id FROM items ORDER BY id")).toEqual({ id: 1 }); + expect(operations).toEqual(["next"]); + }); + + it("rejects a bigint binding outside the safe integer range", async () => { + const database = new DurableObjectSqliteDatabase(durableObjectStorage()); + await database.exec("CREATE TABLE numbers (n INTEGER)"); + await expect(database.run("INSERT INTO numbers (n) VALUES (?)", 9_007_199_254_740_993n)).rejects.toThrow( + RangeError, + ); + expect(await database.all("SELECT n FROM numbers")).toEqual([]); + }); + + it("closes after an active transaction settles", async () => { + const database = new DurableObjectSqliteDatabase(durableObjectStorage()); + await database.exec("CREATE TABLE items (id INTEGER PRIMARY KEY)"); + const order: string[] = []; + const { promise: gate, resolve } = Promise.withResolvers(); + const transaction = database.transaction(async (handle) => { + await handle.run("INSERT INTO items (id) VALUES (?)", 1); + await gate; + order.push("transaction"); + }); + const closed = database.close().then(() => order.push("close")); + resolve(); + await Promise.all([transaction, closed]); + expect(order).toEqual(["transaction", "close"]); + }); + + it("runs an uncontended statement before returning its promise", async () => { + const sqlite = new DatabaseSync(":memory:"); + const database = new DurableObjectSqliteDatabase(durableObjectStorage(sqlite)); + await database.exec("CREATE TABLE items (id INTEGER PRIMARY KEY)"); + const written = database.run("INSERT INTO items (id) VALUES (?)", 1); + expect(sqlite.prepare("SELECT id FROM items").all()).toEqual([{ id: 1 }]); + await written; + }); + + it("rolls back a rejected transaction and rejects its handle afterwards", async () => { + const database = new DurableObjectSqliteDatabase(durableObjectStorage()); + await database.exec("CREATE TABLE items (id INTEGER PRIMARY KEY)"); + let handle: Parameters[0]>[0] | undefined; + await expect( + database.transaction(async (transaction) => { + handle = transaction; + await transaction.run("INSERT INTO items (id) VALUES (?)", 1); + throw new Error("abort"); + }), + ).rejects.toThrow("abort"); + expect(await database.all("SELECT id FROM items")).toEqual([]); + await expect(handle!.run("INSERT INTO items (id) VALUES (?)", 2)).rejects.toThrow("no longer active"); + }); + + it("runs operations from another adapter after a transaction, outside its rollback", async () => { + const storage = durableObjectStorage(); + const database = new DurableObjectSqliteDatabase(storage); + const other = new DurableObjectSqliteDatabase(storage); + await database.exec("CREATE TABLE items (id INTEGER PRIMARY KEY)"); + const { promise: gate, resolve } = Promise.withResolvers(); + const transaction = database.transaction(async (handle) => { + await handle.run("INSERT INTO items (id) VALUES (?)", 1); + await gate; + throw new Error("abort"); + }); + const outside = other.run("INSERT INTO items (id) VALUES (?)", 2); + const read = other.all("SELECT id FROM items"); + resolve(); + await expect(transaction).rejects.toThrow("abort"); + await outside; + expect(await read).toEqual([{ id: 2 }]); + await database.close(); + }); + + it("includes direct underlying calls in an active transaction", async () => { + const sqlite = new DatabaseSync(":memory:"); + const storage = durableObjectStorage(sqlite); + const database = new DurableObjectSqliteDatabase(storage); + await database.exec("CREATE TABLE items (id INTEGER PRIMARY KEY)"); + const { promise: opened, resolve: markOpened } = Promise.withResolvers(); + const { promise: gate, resolve } = Promise.withResolvers(); + const transaction = database.transaction(async (handle) => { + await handle.run("INSERT INTO items (id) VALUES (?)", 1); + markOpened(); + await gate; + throw new Error("abort"); + }); + await opened; + storage.sql.exec("INSERT INTO items (id) VALUES (?)", 2); + resolve(); + await expect(transaction).rejects.toThrow("abort"); + expect(sqlite.prepare("SELECT id FROM items").all()).toEqual([]); + }); +}); diff --git a/packages/chord/test/sqlite/helpers.ts b/packages/chord/test/sqlite/helpers.ts new file mode 100644 index 000000000..042696e45 --- /dev/null +++ b/packages/chord/test/sqlite/helpers.ts @@ -0,0 +1,65 @@ +import { DatabaseSync } from "node:sqlite"; +import type { DurableObjectSqliteStorage } from "../../src/sqlite/cloudflare.ts"; + +type SqlValue = ArrayBuffer | string | number | null; + +/** + * A stand-in for a SQLite-backed Durable Object's `ctx.storage` over `node:sqlite`, with its value types: `BLOB` + * bindings and results are `ArrayBuffer`s, and `exec()` without bindings may run several statements. It keeps this + * portable package free of Workers test dependencies; a Cloudflare host can run the same storage conformance suite + * against its binding with `runInDurableObject()` from `cloudflare:test`. + */ +export function durableObjectStorage( + database = new DatabaseSync(":memory:"), + onCursorOperation?: (operation: "next" | "toArray") => void, +): DurableObjectSqliteStorage { + const toNode = (value: SqlValue) => (value instanceof ArrayBuffer ? new Uint8Array(value) : value); + const fromNode = (value: unknown): SqlValue => + value instanceof Uint8Array ? (value.slice().buffer as ArrayBuffer) : (value as SqlValue); + const cursor = (rows: Record[]) => { + let index = 0; + return { + next(): { done: false; value: Record } | { done: true } { + onCursorOperation?.("next"); + const value = rows[index++]; + return value === undefined ? { done: true } : { done: false, value }; + }, + toArray(): Record[] { + onCursorOperation?.("toArray"); + const remaining = rows.slice(index); + index = rows.length; + return remaining; + }, + }; + }; + return { + sql: { + exec(query, ...bindings) { + if (bindings.length === 0 && query.trim().replace(/;\s*$/, "").includes(";")) { + database.exec(query); + return cursor([]); + } + const statement = database.prepare(query); + if (statement.columns().length === 0) { + statement.run(...bindings.map(toNode)); + return cursor([]); + } + const rows = (statement.all(...bindings.map(toNode)) as Record[]).map((row) => + Object.fromEntries(Object.entries(row).map(([key, value]) => [key, fromNode(value)])), + ); + return cursor(rows); + }, + }, + async transaction(closure) { + database.exec("BEGIN"); + try { + const result = await closure(); + database.exec("COMMIT"); + return result; + } catch (error) { + database.exec("ROLLBACK"); + throw error; + } + }, + }; +} diff --git a/packages/chord/test/sqlite/migrations.test.ts b/packages/chord/test/sqlite/migrations.test.ts new file mode 100644 index 000000000..a94e765fe --- /dev/null +++ b/packages/chord/test/sqlite/migrations.test.ts @@ -0,0 +1,102 @@ +import { describe, expect, it } from "vitest"; +import { applySqliteMigrations, type SqliteMigration, validateSqliteSchemaPrefix } from "../../src/sqlite/index.ts"; +import { openNodeSqliteDatabase } from "../../src/sqlite/node.ts"; + +describe("SQLite migrations", () => { + it("tracks independent histories and skips migrations already applied", async () => { + const database = await openNodeSqliteDatabase(":memory:"); + try { + const first: readonly SqliteMigration[] = [ + { version: 1, statements: ["CREATE TABLE notes (body TEXT)", "INSERT INTO notes VALUES ('kept')"] }, + ]; + await applySqliteMigrations(database, first, { table: "notes_schema" }); + await applySqliteMigrations(database, first, { table: "notes_schema" }); + await applySqliteMigrations(database, [], { table: "other_schema" }); + await applySqliteMigrations( + database, + [...first, { version: 2, statements: ["ALTER TABLE notes ADD COLUMN title TEXT"] }], + { table: "notes_schema" }, + ); + expect(await database.all("SELECT body, title FROM notes")).toEqual([{ body: "kept", title: null }]); + expect(await database.get("SELECT version FROM notes_schema")).toEqual({ version: 2 }); + expect(await database.get("SELECT version FROM other_schema")).toEqual({ version: 0 }); + await expect(applySqliteMigrations(database, first, { table: "notes_schema" })).rejects.toThrow( + "SQLite schema version 2 is newer than supported version 1", + ); + } finally { + await database.close(); + } + }); + + it("rejects invalid migration versions before touching the database", async () => { + const database = await openNodeSqliteDatabase(":memory:"); + try { + for (const versions of [[0], [2], [1, 3], [1, 1]]) { + await expect( + applySqliteMigrations( + database, + versions.map((version) => ({ version, statements: [] })), + { table: "app_schema" }, + ), + ).rejects.toThrow("contiguous versions starting at 1"); + } + expect(await database.all("SELECT name FROM sqlite_schema")).toEqual([]); + } finally { + await database.close(); + } + }); + + it("validates prefixes and migration table identifiers before executing SQL", async () => { + const database = await openNodeSqliteDatabase(":memory:"); + try { + for (const prefix of ["", "1bad", "bad-prefix", "bad; DROP TABLE notes", "sqlite_app", "SQLITE_app"]) { + expect(() => validateSqliteSchemaPrefix(prefix)).toThrow(); + await expect(applySqliteMigrations(database, [], { table: prefix })).rejects.toThrow(); + } + for (const prefix of ["pi_", "_private", "app2"]) + expect(() => validateSqliteSchemaPrefix(prefix)).not.toThrow(); + expect(await database.all("SELECT name FROM sqlite_schema")).toEqual([]); + } finally { + await database.close(); + } + }); + + it("rolls initial bootstrap and every pending migration back together", async () => { + const database = await openNodeSqliteDatabase(":memory:"); + try { + const failed: readonly SqliteMigration[] = [ + { + version: 1, + statements: [ + "CREATE TABLE migration_first (value TEXT) STRICT", + "INSERT INTO migration_first (value) VALUES ('retained')", + ], + }, + { + version: 2, + statements: ["CREATE TABLE migration_second (value TEXT) STRICT", "THIS IS NOT SQL"], + }, + ]; + await expect(applySqliteMigrations(database, failed, { table: "app_schema" })).rejects.toThrow(); + expect( + await database.get( + "SELECT count(*) AS count FROM sqlite_schema WHERE name IN ('app_schema', 'migration_first', 'migration_second')", + ), + ).toEqual({ count: 0 }); + + await applySqliteMigrations( + database, + [failed[0], { version: 2, statements: ["CREATE TABLE migration_second (value TEXT) STRICT"] }], + { table: "app_schema" }, + ); + expect(await database.get("SELECT version FROM app_schema WHERE singleton = 1")).toEqual({ + version: 2, + }); + expect(await database.get("SELECT value FROM migration_first")).toEqual({ + value: "retained", + }); + } finally { + await database.close(); + } + }); +}); diff --git a/packages/chord/test/sqlite/node.test.ts b/packages/chord/test/sqlite/node.test.ts new file mode 100644 index 000000000..11098d1c5 --- /dev/null +++ b/packages/chord/test/sqlite/node.test.ts @@ -0,0 +1,169 @@ +import { describe, expect, it } from "vitest"; +import type { SqliteExecutor } from "../../src/sqlite/index.ts"; +import { openNodeSqliteDatabase } from "../../src/sqlite/node.ts"; + +describe("Node SQLite driver", () => { + it("configures connection defaults and accepts explicit settings", async () => { + for (const options of [{}, { busyTimeoutMs: 123, walAutoCheckpointPages: 0 }]) { + const database = await openNodeSqliteDatabase(":memory:", options); + try { + expect(await database.get("PRAGMA busy_timeout")).toEqual({ timeout: options.busyTimeoutMs ?? 5_000 }); + expect(await database.get("PRAGMA wal_autocheckpoint")).toEqual({ + wal_autocheckpoint: options.walAutoCheckpointPages ?? 1_000, + }); + expect(await database.get("PRAGMA synchronous")).toEqual({ synchronous: 1 }); + } finally { + await database.close(); + } + } + }); + + it("commits work done through the transaction handle and closes idempotently", async () => { + const database = await openNodeSqliteDatabase(":memory:"); + await database.transaction(async (transaction) => { + await transaction.exec("CREATE TABLE async_probe (value INTEGER)"); + await transaction.run("INSERT INTO async_probe (value) VALUES (?)", 1); + }); + expect(await database.get("SELECT value FROM async_probe")).toEqual({ value: 1 }); + await database.close(); + await database.close(); + }); + + it("serializes concurrent transactions", async () => { + const database = await openNodeSqliteDatabase(":memory:"); + await database.exec("CREATE TABLE transaction_queue (value INTEGER)"); + let markFirstStarted!: () => void; + const firstStarted = new Promise((resolve) => { + markFirstStarted = resolve; + }); + let releaseFirst!: () => void; + const firstGate = new Promise((resolve) => { + releaseFirst = resolve; + }); + const first = database.transaction(async (transaction) => { + await transaction.exec("INSERT INTO transaction_queue (value) VALUES (1)"); + markFirstStarted(); + await firstGate; + }); + await firstStarted; + + let secondStarted = false; + const second = database.transaction(async (transaction) => { + secondStarted = true; + await transaction.exec("INSERT INTO transaction_queue (value) VALUES (2)"); + }); + await Promise.resolve(); + expect(secondStarted).toBe(false); + + releaseFirst(); + await Promise.all([first, second]); + expect(await database.all("SELECT value FROM transaction_queue ORDER BY value")).toEqual([ + { value: 1 }, + { value: 2 }, + ]); + await database.close(); + }); + + it("runs operations in call order whether they start immediately or wait", async () => { + const database = await openNodeSqliteDatabase(":memory:"); + await database.exec("CREATE TABLE call_order (value INTEGER)"); + // Operations called during a transaction must neither see its uncommitted rows nor join its rollback. + const transaction = database.transaction(async (handle) => { + await handle.run("INSERT INTO call_order (value) VALUES (?)", 1); + await Promise.resolve(); + throw new Error("roll back"); + }); + const beforeWrite = database.all("SELECT value FROM call_order ORDER BY value"); + const write = database.run("INSERT INTO call_order (value) VALUES (?)", 2); + const afterWrite = database.all("SELECT value FROM call_order ORDER BY value"); + await expect(transaction).rejects.toThrow("roll back"); + await write; + expect(await beforeWrite).toEqual([]); + expect(await afterWrite).toEqual([{ value: 2 }]); + + await database.close(); + }); + + it("queues ordinary operations behind an active transaction", async () => { + const database = await openNodeSqliteDatabase(":memory:"); + await database.exec("CREATE TABLE operation_queue (value INTEGER)"); + let markTransactionStarted!: () => void; + const transactionStarted = new Promise((resolve) => { + markTransactionStarted = resolve; + }); + let releaseTransaction!: () => void; + const transactionGate = new Promise((resolve) => { + releaseTransaction = resolve; + }); + const pending = database.transaction(async (transaction) => { + await transaction.exec("INSERT INTO operation_queue (value) VALUES (1)"); + markTransactionStarted(); + await transactionGate; + }); + await transactionStarted; + + let writeSettled = false; + const write = database.exec("INSERT INTO operation_queue (value) VALUES (2)").finally(() => { + writeSettled = true; + }); + let readSettled = false; + const read = database.all("SELECT value FROM operation_queue ORDER BY value").finally(() => { + readSettled = true; + }); + await Promise.resolve(); + expect(writeSettled).toBe(false); + expect(readSettled).toBe(false); + + releaseTransaction(); + await pending; + await write; + await expect(read).resolves.toEqual([{ value: 1 }, { value: 2 }]); + await database.close(); + }); + + it("queues database calls made synchronously by a transaction that started immediately", async () => { + const database = await openNodeSqliteDatabase(":memory:"); + await database.exec("CREATE TABLE barrier_probe (value INTEGER)"); + let outside!: Promise; + const transaction = database.transaction(async (handle) => { + // Misuse: this call must wait for the transaction instead of joining it. + outside = database.run("INSERT INTO barrier_probe (value) VALUES (?)", 2); + await handle.run("INSERT INTO barrier_probe (value) VALUES (?)", 1); + await Promise.resolve(); + throw new Error("roll back"); + }); + await expect(transaction).rejects.toThrow("roll back"); + await outside; + expect(await database.all("SELECT value FROM barrier_probe")).toEqual([{ value: 2 }]); + await database.close(); + }); + + it("rejects a transaction handle used after its transaction settles", async () => { + const database = await openNodeSqliteDatabase(":memory:"); + await database.exec("CREATE TABLE stale_probe (value INTEGER)"); + let handle!: SqliteExecutor; + await database.transaction(async (transaction) => { + handle = transaction; + await transaction.run("INSERT INTO stale_probe (value) VALUES (?)", 1); + }); + const stale = "SQLite transaction handle is no longer active"; + await expect(handle.exec("INSERT INTO stale_probe (value) VALUES (2)")).rejects.toThrow(stale); + await expect(handle.run("INSERT INTO stale_probe (value) VALUES (?)", 3)).rejects.toThrow(stale); + expect(await database.all("SELECT value FROM stale_probe")).toEqual([{ value: 1 }]); + 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)"); + await expect( + database.transaction(async (transaction) => { + await transaction.exec("INSERT INTO rollback_probe (value) VALUES (1)"); + await transaction.exec("COMMIT"); + throw new Error("rejected after an escaped commit"); + }), + ).rejects.toThrow(AggregateError); + expect(await database.get("SELECT value FROM rollback_probe")).toEqual({ value: 1 }); + await database.close(); + }); +}); diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 4e0d75c1d..e587541e6 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -9,6 +9,8 @@ ### Changed +- SQLite drivers, portable interfaces, and the generic migration runner now come from `@earendil-works/chord/sqlite`. Existing Durable SQLite entry points keep re-exporting their public names; storage helpers and Durable schema migrations stay here. + - `ToolControl.terminate` is now typed `boolean` instead of `true`, matching `pi-agent-core` and `pi-coding-agent`; `false` does not end the run ([#10858](https://github.com/earendil-works/pi/issues/10858)). ### Fixed diff --git a/packages/durable/src/storage/sqlite/cloudflare.ts b/packages/durable/src/storage/sqlite/cloudflare.ts index 63412b62b..3bce64417 100644 --- a/packages/durable/src/storage/sqlite/cloudflare.ts +++ b/packages/durable/src/storage/sqlite/cloudflare.ts @@ -1,184 +1,7 @@ -import type { SqliteDatabase, SqliteExecutor, SqliteValue } from "./database.ts"; +import { DurableObjectSqliteDatabase, type DurableObjectSqliteStorage } from "@earendil-works/chord/sqlite/cloudflare"; import { SqliteStorage, type SqliteStorageOptions } from "./storage.ts"; -/** Values a Durable Object SQL binding accepts and returns. */ -type DurableObjectSqlValue = ArrayBuffer | string | number | null; - -/** - * The parts of a SQLite-backed Durable Object's `ctx.storage` this adapter uses, typed structurally so the package - * needs no Workers type dependency. - */ -export type DurableObjectSqliteStorage = { - readonly sql: { - exec( - query: string, - ...bindings: DurableObjectSqlValue[] - ): { - next(): - | { readonly done?: false; readonly value: Record } - | { readonly done: true; readonly value?: undefined }; - toArray(): Record[]; - }; - }; - transaction(closure: () => Promise): Promise; -}; - -/** Runs synchronous operations immediately when idle and holds queued calls behind a transaction. */ -class SerialQueue { - private tail: Promise = Promise.resolve(); - private pending = 0; - - run(operation: () => T | Promise): Promise { - if (this.pending > 0) return this.enqueue(operation); - try { - return Promise.resolve(operation()); - } catch (error) { - return Promise.reject(error); - } - } - - hold(operation: () => Promise): Promise { - return this.enqueue(operation); - } - - private enqueue(operation: () => T | Promise): Promise { - this.pending++; - const result = this.tail.then(operation).finally(() => { - this.pending--; - }); - this.tail = result.then( - () => {}, - () => {}, - ); - return result; - } -} - -const storageQueues = new WeakMap(); - -function queueFor(storage: DurableObjectSqliteStorage): SerialQueue { - let queue = storageQueues.get(storage); - if (queue === undefined) { - queue = new SerialQueue(); - storageQueues.set(storage, queue); - } - return queue; -} - -/** Durable Object SQL binds numbers as doubles: a `bigint` outside the safe integer range would lose precision. */ -function bindValue(value: SqliteValue): DurableObjectSqlValue { - if (value instanceof Uint8Array) return value.slice().buffer as ArrayBuffer; - if (typeof value !== "bigint") return value; - if (value < BigInt(Number.MIN_SAFE_INTEGER) || value > BigInt(Number.MAX_SAFE_INTEGER)) { - throw new RangeError(`Durable Object SQL cannot bind ${value} without losing precision`); - } - return Number(value); -} - -const bind = (params: readonly SqliteValue[]): DurableObjectSqlValue[] => params.map(bindValue); - -/** A row as returned, with `BLOB` columns converted to `Uint8Array` in place; no other copy is made. */ -function row(row: Record): T { - for (const key in row) { - const value = row[key]; - if (value instanceof ArrayBuffer) (row as Record)[key] = new Uint8Array(value); - } - return row as T; -} - -/** Executes SQL through `ctx.storage.sql`, which has no separate prepare step. */ -class DurableObjectSqliteExecutor implements SqliteExecutor { - protected readonly storage: DurableObjectSqliteStorage; - - constructor(storage: DurableObjectSqliteStorage) { - this.storage = storage; - } - - async exec(sql: string): Promise { - this.check(); - this.storage.sql.exec(sql); - } - - async run(sql: string, ...params: SqliteValue[]): Promise { - this.check(); - this.storage.sql.exec(sql, ...bind(params)); - } - - async get(sql: string, ...params: SqliteValue[]): Promise { - this.check(); - const result = this.storage.sql.exec(sql, ...bind(params)).next(); - return result.done ? undefined : row(result.value); - } - - async all(sql: string, ...params: SqliteValue[]): Promise { - this.check(); - return this.storage.sql - .exec(sql, ...bind(params)) - .toArray() - .map((value) => row(value)); - } - - /** Throws when this handle may no longer run SQL. */ - protected check(): void {} -} - -/** The handle a transaction callback receives; invalid once the callback settles. */ -class DurableObjectSqliteTransaction extends DurableObjectSqliteExecutor { - active = true; - - protected override check(): void { - if (!this.active) throw new Error("SQLite transaction handle is no longer active"); - } -} - -/** - * `SqliteDatabase` adapter for the SQLite storage of a Cloudflare Durable Object. Adapters given the same storage - * object share a queue; direct calls to the underlying storage bypass it. Integers are JavaScript numbers; binding a - * `bigint` outside the safe integer range throws. - */ -export class DurableObjectSqliteDatabase extends DurableObjectSqliteExecutor implements SqliteDatabase { - private readonly queue: SerialQueue; - - constructor(storage: DurableObjectSqliteStorage) { - super(storage); - this.queue = queueFor(storage); - } - - override exec(sql: string): Promise { - return this.queue.run(() => super.exec(sql)); - } - - override run(sql: string, ...params: SqliteValue[]): Promise { - return this.queue.run(() => super.run(sql, ...params)); - } - - override get(sql: string, ...params: SqliteValue[]): Promise { - return this.queue.run(() => super.get(sql, ...params)); - } - - override all(sql: string, ...params: SqliteValue[]): Promise { - return this.queue.run(() => super.all(sql, ...params)); - } - - /** `ctx.storage.transaction()` commits when the callback resolves and rolls back when it rejects. */ - transaction(callback: (transaction: SqliteExecutor) => Promise): Promise { - return this.queue.hold(() => - this.storage.transaction(async () => { - const transaction = new DurableObjectSqliteTransaction(this.storage); - try { - return await callback(transaction); - } finally { - transaction.active = false; - } - }), - ); - } - - /** Waits for queued work. The Durable Object owns its storage, so nothing else closes. */ - close(): Promise { - return this.queue.run(() => {}); - } -} +export { DurableObjectSqliteDatabase, type DurableObjectSqliteStorage } from "@earendil-works/chord/sqlite/cloudflare"; /** * Open durable storage on a SQLite-backed Durable Object, optionally under a table prefix. Calls given the same diff --git a/packages/durable/src/storage/sqlite/database.ts b/packages/durable/src/storage/sqlite/database.ts index 64edf8a0c..0af83a65f 100644 --- a/packages/durable/src/storage/sqlite/database.ts +++ b/packages/durable/src/storage/sqlite/database.ts @@ -1,38 +1 @@ -/** Values supported by the portable SQLite storage core. */ -export type SqliteValue = null | number | bigint | string | Uint8Array; - -/** - * Asynchronous SQL operations shared by a database and its transaction handles. - * - * `exec` runs SQL text without bindings and may contain several statements. `run`, `get`, and `all` - * execute one statement with positional bindings. Adapters may cache prepared statements by SQL text, - * so callers pass values as bindings instead of interpolating them. - */ -export interface SqliteExecutor { - exec(sql: string): Promise; - run(sql: string, ...params: SqliteValue[]): Promise; - get(sql: string, ...params: SqliteValue[]): Promise; - all(sql: string, ...params: SqliteValue[]): Promise; -} - -/** - * Minimal database facade required by `SqliteStorage`. - * - * All operations are asynchronous so adapters may execute outside the harness runtime. - * - * `transaction` passes the callback a transaction handle. All work in the transaction - * must use that handle; the handle is invalid after the callback settles. Adapters must - * queue unrelated operations and other transactions until the transaction finishes. The - * returned promise settles after commit or rollback. Calling the database itself (including - * `transaction` or `close`) from inside a callback therefore waits for that transaction and - * never settles. - * - * When the callback rejects, the adapter must roll the transaction back before rejecting - * with that same error. If rollback fails, it must reject with a different error (for - * example an `AggregateError`) so callers cannot mistake the callback error for a - * guaranteed rollback. - */ -export interface SqliteDatabase extends SqliteExecutor { - transaction(callback: (transaction: SqliteExecutor) => Promise): Promise; - close(): Promise; -} +export type { SqliteDatabase, SqliteExecutor, SqliteValue } from "@earendil-works/chord/sqlite"; diff --git a/packages/durable/src/storage/sqlite/index.ts b/packages/durable/src/storage/sqlite/index.ts index 3ae7b068a..2ba0ac928 100644 --- a/packages/durable/src/storage/sqlite/index.ts +++ b/packages/durable/src/storage/sqlite/index.ts @@ -1,4 +1,4 @@ -export type { SqliteDatabase, SqliteExecutor, SqliteValue } from "./database.ts"; +export type { SqliteDatabase, SqliteExecutor, SqliteValue } from "@earendil-works/chord/sqlite"; export { applySqliteMigrations, CURRENT_SQLITE_SCHEMA_VERSION, diff --git a/packages/durable/src/storage/sqlite/migrations.ts b/packages/durable/src/storage/sqlite/migrations.ts index f18928ff9..bd2b6dd89 100644 --- a/packages/durable/src/storage/sqlite/migrations.ts +++ b/packages/durable/src/storage/sqlite/migrations.ts @@ -1,10 +1,11 @@ -import type { SqliteDatabase } from "./database.ts"; +import { + applySqliteMigrations as applyMigrations, + type SqliteDatabase, + type SqliteMigration, +} from "@earendil-works/chord/sqlite"; import { DEFAULT_SQLITE_SCHEMA, type SqliteSchema } from "./schema.ts"; -export type SqliteMigration = { - readonly version: number; - readonly statements: readonly string[]; -}; +export type { SqliteMigration } from "@earendil-works/chord/sqlite"; /** Build immutable, ordered schema history for a set of SQLite identifiers. */ export function createSqliteMigrations(schema: SqliteSchema): readonly SqliteMigration[] { @@ -94,8 +95,6 @@ export const SQLITE_MIGRATIONS: readonly SqliteMigration[] = createSqliteMigrati export const CURRENT_SQLITE_SCHEMA_VERSION = SQLITE_MIGRATIONS.at(-1)?.version ?? 0; -type SchemaRow = { readonly version: number }; - /** Apply all pending schema migrations atomically. */ export async function applySqliteMigrations( database: SqliteDatabase, @@ -110,31 +109,5 @@ export async function applySqliteMigrationsForSchema( schema: SqliteSchema, migrations: readonly SqliteMigration[] = createSqliteMigrations(schema), ): Promise { - for (let index = 0; index < migrations.length; index++) { - if (migrations[index]?.version !== index + 1) { - throw new Error("Durable SQLite migrations must have contiguous versions starting at 1"); - } - } - - const { durableSchema } = schema.tables; - await database.transaction(async (transaction) => { - await transaction.exec(`CREATE TABLE IF NOT EXISTS ${durableSchema} ( - singleton INTEGER PRIMARY KEY CHECK (singleton = 1), - version INTEGER NOT NULL CHECK (version >= 0) - ) STRICT`); - await transaction.run(`INSERT OR IGNORE INTO ${durableSchema} (singleton, version) VALUES (1, 0)`); - const row = await transaction.get(`SELECT version FROM ${durableSchema} WHERE singleton = 1`); - if (row === undefined) throw new Error("Durable SQLite schema metadata is missing"); - const currentVersion = migrations.at(-1)?.version ?? 0; - if (row.version > currentVersion) { - throw new Error( - `Durable SQLite schema version ${row.version} is newer than supported version ${currentVersion}`, - ); - } - for (const migration of migrations) { - if (migration.version <= row.version) continue; - for (const statement of migration.statements) await transaction.exec(statement); - await transaction.run(`UPDATE ${durableSchema} SET version = ? WHERE singleton = 1`, migration.version); - } - }); + await applyMigrations(database, migrations, { table: schema.tables.durableSchema }); } diff --git a/packages/durable/src/storage/sqlite/node.ts b/packages/durable/src/storage/sqlite/node.ts index 1a5e1f57b..210478ae5 100644 --- a/packages/durable/src/storage/sqlite/node.ts +++ b/packages/durable/src/storage/sqlite/node.ts @@ -1,210 +1,16 @@ -import { mkdir } from "node:fs/promises"; -import { dirname } from "node:path"; -import type { StatementSync } from "node:sqlite"; -import { DatabaseSync } from "node:sqlite"; -import type { SqliteDatabase, SqliteExecutor, SqliteValue } from "./database.ts"; -import { validateSqliteSchemaPrefix } from "./schema.ts"; +import { validateSqliteSchemaPrefix } from "@earendil-works/chord/sqlite"; +import { type NodeSqliteDatabaseOptions, openNodeSqliteDatabase } from "@earendil-works/chord/sqlite/node"; import { SqliteStorage, type SqliteStorageOptions } from "./storage.ts"; -/** Node SQLite connection settings for a durable storage file. */ -export type NodeSqliteDatabaseOptions = { - /** SQLite WAL auto-checkpoint threshold. SQLite and this adapter default to 1,000 pages; 0 disables it. */ - readonly walAutoCheckpointPages?: number; - /** Time SQLite waits for a competing file lock. SQLite defaults to 0; this adapter defaults to 5,000 ms. */ - readonly busyTimeoutMs?: number; -}; +export { + NodeSqliteDatabase, + type NodeSqliteDatabaseOptions, + openNodeSqliteDatabase, +} from "@earendil-works/chord/sqlite/node"; /** Node SQLite connection settings plus durable storage options. */ export type NodeSqliteStorageOptions = NodeSqliteDatabaseOptions & SqliteStorageOptions; -const DEFAULT_WAL_AUTO_CHECKPOINT_PAGES = 1_000; -const DEFAULT_BUSY_TIMEOUT_MS = 5_000; - -type TransactionScope = { active: boolean }; - -const ignore = (): void => {}; - -/** - * Runs operations in call order. An operation starts immediately when nothing is running or waiting; - * otherwise it waits for everything before it. An asynchronous operation holds the queue until it settles. - */ -class SerialOperationQueue { - private tail: Promise = Promise.resolve(); - private pending = 0; - - run(operation: () => T): Promise { - if (this.pending > 0) return this.enqueue(operation); - try { - return Promise.resolve(operation()); - } catch (error) { - return Promise.reject(error); - } - } - - runAsync(operation: () => Promise): Promise { - if (this.pending > 0) return this.enqueue(operation); - this.pending++; - // Publish the barrier before the operation starts, so calls it makes synchronously wait behind it. - const { promise: barrier, resolve: releaseBarrier } = Promise.withResolvers(); - this.tail = barrier; - let started: Promise; - try { - started = operation(); - } catch (error) { - started = Promise.reject(error); - } - return started.finally(() => { - this.pending--; - releaseBarrier(); - }); - } - - private enqueue(operation: () => T | Promise): Promise { - this.pending++; - return this.release(this.tail.then(operation)); - } - - private release(operation: Promise): Promise { - const settled = operation.finally(() => { - this.pending--; - }); - this.tail = settled.then(ignore, ignore); - return settled; - } -} - -/** - * Executes SQL on one connection. Prepared statements are cached per connection by SQL text, so the - * database and its transaction handles share them across transactions. - */ -abstract class NodeSqliteExecutor implements SqliteExecutor { - protected readonly database: DatabaseSync; - protected readonly statements: Map; - - constructor(database: DatabaseSync, statements: Map) { - this.database = database; - this.statements = statements; - } - - exec(sql: string): Promise { - return this.runOperation(() => { - this.database.exec(sql); - }); - } - - run(sql: string, ...params: SqliteValue[]): Promise { - return this.runOperation(() => { - this.statement(sql).run(...params); - }); - } - - get(sql: string, ...params: SqliteValue[]): Promise { - return this.runOperation(() => this.statement(sql).get(...params) as T | undefined); - } - - all(sql: string, ...params: SqliteValue[]): Promise { - return this.runOperation(() => this.statement(sql).all(...params) as T[]); - } - - protected abstract runOperation(operation: () => T): Promise; - - private statement(sql: string): StatementSync { - let statement = this.statements.get(sql); - if (statement === undefined) { - statement = this.database.prepare(sql); - this.statements.set(sql, statement); - } - return statement; - } -} - -class NodeSqliteTransaction extends NodeSqliteExecutor { - private readonly scope: TransactionScope; - - constructor(database: DatabaseSync, statements: Map, scope: TransactionScope) { - super(database, statements); - this.scope = scope; - } - - protected async runOperation(operation: () => T): Promise { - if (!this.scope.active) throw new Error("SQLite transaction handle is no longer active"); - return operation(); - } -} - -/** `SqliteDatabase` adapter backed by Node's built-in `node:sqlite`. */ -export class NodeSqliteDatabase extends NodeSqliteExecutor implements SqliteDatabase { - private readonly access = new SerialOperationQueue(); - private closed = false; - - constructor(database: DatabaseSync) { - super(database, new Map()); - } - - transaction(callback: (transaction: SqliteExecutor) => Promise): Promise { - return this.access.runAsync(async () => { - this.database.exec("BEGIN IMMEDIATE"); - const scope = { active: true }; - try { - const result = await callback(new NodeSqliteTransaction(this.database, this.statements, scope)); - scope.active = false; - this.database.exec("COMMIT"); - return result; - } catch (error) { - scope.active = false; - try { - this.database.exec("ROLLBACK"); - } catch (rollbackError) { - throw new AggregateError([error, rollbackError], "SQLite transaction failed and rollback failed"); - } - throw error; - } - }); - } - - close(): Promise { - return this.access.run(() => { - if (this.closed) return; - this.closed = true; - this.statements.clear(); - try { - this.database.exec("PRAGMA wal_checkpoint(TRUNCATE)"); - } finally { - this.database.close(); - } - }); - } - - protected runOperation(operation: () => T): Promise { - return this.access.run(operation); - } -} - -/** Open and configure a Node-backed SQLite database facade. */ -export async function openNodeSqliteDatabase( - path: string, - options: NodeSqliteDatabaseOptions = {}, -): Promise { - const checkpointPages = options.walAutoCheckpointPages ?? DEFAULT_WAL_AUTO_CHECKPOINT_PAGES; - const timeout = options.busyTimeoutMs ?? DEFAULT_BUSY_TIMEOUT_MS; - if (path !== ":memory:") await mkdir(dirname(path), { recursive: true }); - const database = new DatabaseSync(path, { timeout }); - const adapter = new NodeSqliteDatabase(database); - try { - await adapter.exec("PRAGMA journal_mode = WAL"); - await adapter.exec("PRAGMA synchronous = NORMAL"); - await adapter.exec(`PRAGMA wal_autocheckpoint = ${checkpointPages}`); - return adapter; - } catch (error) { - try { - await adapter.close(); - } catch { - // Preserve the configuration failure. - } - throw error; - } -} - /** Open or create file-backed durable storage using Node's built-in SQLite, optionally under a table prefix. */ export async function openNodeSqliteStorage( path: string, diff --git a/packages/durable/src/storage/sqlite/schema.ts b/packages/durable/src/storage/sqlite/schema.ts index fd2fd0c21..b807ba529 100644 --- a/packages/durable/src/storage/sqlite/schema.ts +++ b/packages/durable/src/storage/sqlite/schema.ts @@ -1,3 +1,7 @@ +import { validateSqliteSchemaPrefix } from "@earendil-works/chord/sqlite"; + +export { validateSqliteSchemaPrefix } from "@earendil-works/chord/sqlite"; + /** * Immutable identifier history for migrations and runtime queries. Never change or remove an existing value; a * rename adds a new entry so historical migrations keep rendering their original object names. @@ -40,16 +44,6 @@ export type SqliteSchema = { readonly indexes: SqliteIdentifiers; }; -/** Validate a prefix before using it as part of a SQLite identifier. */ -export function validateSqliteSchemaPrefix(prefix: string): void { - if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(prefix)) { - throw new Error(`Invalid SQLite schema prefix ${JSON.stringify(prefix)}`); - } - if (prefix.toLowerCase().startsWith("sqlite_")) { - throw new Error(`SQLite schema prefix ${JSON.stringify(prefix)} is reserved`); - } -} - function prefixIdentifiers>(names: T, prefix: string): SqliteIdentifiers { return Object.fromEntries( Object.entries(names).map(([key, name]) => [key, `${prefix}${name}`]), diff --git a/packages/durable/src/storage/sqlite/storage.ts b/packages/durable/src/storage/sqlite/storage.ts index 4b13f8763..7dc590382 100644 --- a/packages/durable/src/storage/sqlite/storage.ts +++ b/packages/durable/src/storage/sqlite/storage.ts @@ -1,5 +1,6 @@ import type { Context, JsonValue } from "@earendil-works/chord"; import { apply, type Op } from "@earendil-works/chord/delta"; +import type { SqliteDatabase, SqliteExecutor, SqliteValue } from "@earendil-works/chord/sqlite"; import { StorageRequestError } from "../../errors.ts"; import { idFromNumber, seqFromNumber } from "../../ids.ts"; import type { @@ -33,7 +34,6 @@ import type { TaskRecord, } from "../../types.ts"; import { nextCursor, type ScanStart, scanStart } from "../scan.ts"; -import type { SqliteDatabase, SqliteExecutor, SqliteValue } from "./database.ts"; import { applySqliteMigrationsForSchema } from "./migrations.ts"; import { createSqliteSchema, type SqliteSchema } from "./schema.ts"; diff --git a/packages/durable/test/sqlite-cloudflare.test.ts b/packages/durable/test/sqlite-cloudflare.test.ts index 7546283f7..4c64772db 100644 --- a/packages/durable/test/sqlite-cloudflare.test.ts +++ b/packages/durable/test/sqlite-cloudflare.test.ts @@ -2,77 +2,11 @@ import { DatabaseSync } from "node:sqlite"; import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; import { registerStorageConformance } from "@earendil-works/pi-durable/testing"; import { describe, expect, it } from "vitest"; +import { durableObjectStorage } from "../../chord/test/sqlite/helpers.ts"; import { idFromNumber } from "../src/ids.ts"; -import { - DurableObjectSqliteDatabase, - type DurableObjectSqliteStorage, - openDurableObjectSqliteStorage, -} from "../src/storage/sqlite/cloudflare.ts"; +import { DurableObjectSqliteDatabase, openDurableObjectSqliteStorage } from "../src/storage/sqlite/cloudflare.ts"; import { type EntryId, ROOT_CONVERSATION_ID } from "../src/types.ts"; -type SqlValue = ArrayBuffer | string | number | null; - -/** - * A stand-in for a SQLite-backed Durable Object's `ctx.storage` over `node:sqlite`, with its value types: `BLOB` - * bindings and results are `ArrayBuffer`s, and `exec()` without bindings may run several statements. It keeps this - * portable package free of Workers test dependencies; a Cloudflare host can run the same storage conformance suite - * against its binding with `runInDurableObject()` from `cloudflare:test`. - */ -function durableObjectStorage( - database = new DatabaseSync(":memory:"), - onCursorOperation?: (operation: "next" | "toArray") => void, -): DurableObjectSqliteStorage { - const toNode = (value: SqlValue) => (value instanceof ArrayBuffer ? new Uint8Array(value) : value); - const fromNode = (value: unknown): SqlValue => - value instanceof Uint8Array ? (value.slice().buffer as ArrayBuffer) : (value as SqlValue); - const cursor = (rows: Record[]) => { - let index = 0; - return { - next(): { done: false; value: Record } | { done: true } { - onCursorOperation?.("next"); - const value = rows[index++]; - return value === undefined ? { done: true } : { done: false, value }; - }, - toArray(): Record[] { - onCursorOperation?.("toArray"); - const remaining = rows.slice(index); - index = rows.length; - return remaining; - }, - }; - }; - return { - sql: { - exec(query, ...bindings) { - if (bindings.length === 0 && query.trim().replace(/;\s*$/, "").includes(";")) { - database.exec(query); - return cursor([]); - } - const statement = database.prepare(query); - if (statement.columns().length === 0) { - statement.run(...bindings.map(toNode)); - return cursor([]); - } - const rows = (statement.all(...bindings.map(toNode)) as Record[]).map((row) => - Object.fromEntries(Object.entries(row).map(([key, value]) => [key, fromNode(value)])), - ); - return cursor(rows); - }, - }, - async transaction(closure) { - database.exec("BEGIN"); - try { - const result = await closure(); - database.exec("COMMIT"); - return result; - } catch (error) { - database.exec("ROLLBACK"); - throw error; - } - }, - }; -} - registerStorageConformance({ describe, expect, it }, "SqliteStorage on a prefixed Durable Object", async (use) => use(await openDurableObjectSqliteStorage(durableObjectStorage(), { prefix: "pi_" })), ); @@ -143,114 +77,4 @@ describe("Durable Object SQLite adapter", () => { "reserved", ); }); - - it("binds and returns BLOB values as Uint8Array", async () => { - const database = new DurableObjectSqliteDatabase(durableObjectStorage()); - await database.exec("CREATE TABLE blobs (id INTEGER PRIMARY KEY, data BLOB, n INTEGER)"); - await database.run("INSERT INTO blobs (id, data, n) VALUES (?, ?, ?)", 1, new Uint8Array([1, 2, 3]), 7n); - expect(await database.get("SELECT data, n FROM blobs WHERE id = ?", 1)).toEqual({ - data: new Uint8Array([1, 2, 3]), - n: 7, - }); - }); - - it("reads only the cursor's first row for get", async () => { - const operations: string[] = []; - const database = new DurableObjectSqliteDatabase( - durableObjectStorage(new DatabaseSync(":memory:"), (operation) => operations.push(operation)), - ); - await database.exec("CREATE TABLE items (id INTEGER PRIMARY KEY)"); - await database.run("INSERT INTO items (id) VALUES (1), (2)"); - expect(await database.get("SELECT id FROM items ORDER BY id")).toEqual({ id: 1 }); - expect(operations).toEqual(["next"]); - }); - - it("rejects a bigint binding outside the safe integer range", async () => { - const database = new DurableObjectSqliteDatabase(durableObjectStorage()); - await database.exec("CREATE TABLE numbers (n INTEGER)"); - await expect(database.run("INSERT INTO numbers (n) VALUES (?)", 9_007_199_254_740_993n)).rejects.toThrow( - RangeError, - ); - expect(await database.all("SELECT n FROM numbers")).toEqual([]); - }); - - it("closes after an active transaction settles", async () => { - const database = new DurableObjectSqliteDatabase(durableObjectStorage()); - await database.exec("CREATE TABLE items (id INTEGER PRIMARY KEY)"); - const order: string[] = []; - const { promise: gate, resolve } = Promise.withResolvers(); - const transaction = database.transaction(async (handle) => { - await handle.run("INSERT INTO items (id) VALUES (?)", 1); - await gate; - order.push("transaction"); - }); - const closed = database.close().then(() => order.push("close")); - resolve(); - await Promise.all([transaction, closed]); - expect(order).toEqual(["transaction", "close"]); - }); - - it("runs an uncontended statement before returning its promise", async () => { - const sqlite = new DatabaseSync(":memory:"); - const database = new DurableObjectSqliteDatabase(durableObjectStorage(sqlite)); - await database.exec("CREATE TABLE items (id INTEGER PRIMARY KEY)"); - const written = database.run("INSERT INTO items (id) VALUES (?)", 1); - expect(sqlite.prepare("SELECT id FROM items").all()).toEqual([{ id: 1 }]); - await written; - }); - - it("rolls back a rejected transaction and rejects its handle afterwards", async () => { - const database = new DurableObjectSqliteDatabase(durableObjectStorage()); - await database.exec("CREATE TABLE items (id INTEGER PRIMARY KEY)"); - let handle: Parameters[0]>[0] | undefined; - await expect( - database.transaction(async (transaction) => { - handle = transaction; - await transaction.run("INSERT INTO items (id) VALUES (?)", 1); - throw new Error("abort"); - }), - ).rejects.toThrow("abort"); - expect(await database.all("SELECT id FROM items")).toEqual([]); - await expect(handle!.run("INSERT INTO items (id) VALUES (?)", 2)).rejects.toThrow("no longer active"); - }); - - it("runs operations from another adapter after a transaction, outside its rollback", async () => { - const storage = durableObjectStorage(); - const database = new DurableObjectSqliteDatabase(storage); - const other = new DurableObjectSqliteDatabase(storage); - await database.exec("CREATE TABLE items (id INTEGER PRIMARY KEY)"); - const { promise: gate, resolve } = Promise.withResolvers(); - const transaction = database.transaction(async (handle) => { - await handle.run("INSERT INTO items (id) VALUES (?)", 1); - await gate; - throw new Error("abort"); - }); - const outside = other.run("INSERT INTO items (id) VALUES (?)", 2); - const read = other.all("SELECT id FROM items"); - resolve(); - await expect(transaction).rejects.toThrow("abort"); - await outside; - expect(await read).toEqual([{ id: 2 }]); - await database.close(); - }); - - it("includes direct underlying calls in an active transaction", async () => { - const sqlite = new DatabaseSync(":memory:"); - const storage = durableObjectStorage(sqlite); - const database = new DurableObjectSqliteDatabase(storage); - await database.exec("CREATE TABLE items (id INTEGER PRIMARY KEY)"); - const { promise: opened, resolve: markOpened } = Promise.withResolvers(); - const { promise: gate, resolve } = Promise.withResolvers(); - const transaction = database.transaction(async (handle) => { - await handle.run("INSERT INTO items (id) VALUES (?)", 1); - markOpened(); - await gate; - throw new Error("abort"); - }); - await opened; - storage.sql.exec("INSERT INTO items (id) VALUES (?)", 2); - resolve(); - await expect(transaction).rejects.toThrow("abort"); - expect(sqlite.prepare("SELECT id FROM items").all()).toEqual([]); - }); }); diff --git a/packages/durable/test/sqlite-facade.test.ts b/packages/durable/test/sqlite-facade.test.ts index c3db89910..46662535f 100644 --- a/packages/durable/test/sqlite-facade.test.ts +++ b/packages/durable/test/sqlite-facade.test.ts @@ -1,5 +1,11 @@ import { DatabaseSync, type StatementSync } from "node:sqlite"; import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import { DurableObjectSqliteDatabase as ChordDurableObjectSqliteDatabase } from "@earendil-works/chord/sqlite/cloudflare"; +import { + NodeSqliteDatabase as ChordNodeSqliteDatabase, + openNodeSqliteDatabase as chordOpenNodeSqliteDatabase, +} from "@earendil-works/chord/sqlite/node"; +import { DurableObjectSqliteDatabase } from "@earendil-works/pi-durable/storage/sqlite/cloudflare"; import { describe, expect, it } from "vitest"; import { idFromNumber } from "../src/ids.ts"; import type { SqliteDatabase, SqliteExecutor, SqliteValue } from "../src/storage/sqlite/index.ts"; @@ -79,6 +85,12 @@ class PrepareCountingDatabaseSync extends DatabaseSync { } describe("portable SQLite facade settlement", () => { + it("re-exports the Chord driver classes and Node opener without wrapping them", () => { + expect(NodeSqliteDatabase).toBe(ChordNodeSqliteDatabase); + expect(openNodeSqliteDatabase).toBe(chordOpenNodeSqliteDatabase); + expect(DurableObjectSqliteDatabase).toBe(ChordDurableObjectSqliteDatabase); + }); + it("prepares each storage statement once per connection and reuses it across transactions", async () => { const connection = new PrepareCountingDatabaseSync(":memory:"); const storage = await SqliteStorage.open(new NodeSqliteDatabase(connection)); @@ -109,69 +121,8 @@ describe("portable SQLite facade settlement", () => { await storage.close(BACKGROUND_CONTEXT); }); - it("commits work done through the transaction handle and closes idempotently", async () => { + it("reads storage commits in call order", async () => { const database = await openNodeSqliteDatabase(":memory:"); - await database.transaction(async (transaction) => { - await transaction.exec("CREATE TABLE async_probe (value INTEGER)"); - await transaction.run("INSERT INTO async_probe (value) VALUES (?)", 1); - }); - expect(await database.get("SELECT value FROM async_probe")).toEqual({ value: 1 }); - await database.close(); - await database.close(); - }); - - it("serializes concurrent transactions", async () => { - const database = await openNodeSqliteDatabase(":memory:"); - await database.exec("CREATE TABLE transaction_queue (value INTEGER)"); - let markFirstStarted!: () => void; - const firstStarted = new Promise((resolve) => { - markFirstStarted = resolve; - }); - let releaseFirst!: () => void; - const firstGate = new Promise((resolve) => { - releaseFirst = resolve; - }); - const first = database.transaction(async (transaction) => { - await transaction.exec("INSERT INTO transaction_queue (value) VALUES (1)"); - markFirstStarted(); - await firstGate; - }); - await firstStarted; - - let secondStarted = false; - const second = database.transaction(async (transaction) => { - secondStarted = true; - await transaction.exec("INSERT INTO transaction_queue (value) VALUES (2)"); - }); - await Promise.resolve(); - expect(secondStarted).toBe(false); - - releaseFirst(); - await Promise.all([first, second]); - expect(await database.all("SELECT value FROM transaction_queue ORDER BY value")).toEqual([ - { value: 1 }, - { value: 2 }, - ]); - await database.close(); - }); - - it("runs operations in call order whether they start immediately or wait", async () => { - const database = await openNodeSqliteDatabase(":memory:"); - await database.exec("CREATE TABLE call_order (value INTEGER)"); - // Operations called during a transaction must neither see its uncommitted rows nor join its rollback. - const transaction = database.transaction(async (handle) => { - await handle.run("INSERT INTO call_order (value) VALUES (?)", 1); - await Promise.resolve(); - throw new Error("roll back"); - }); - const beforeWrite = database.all("SELECT value FROM call_order ORDER BY value"); - const write = database.run("INSERT INTO call_order (value) VALUES (?)", 2); - const afterWrite = database.all("SELECT value FROM call_order ORDER BY value"); - await expect(transaction).rejects.toThrow("roll back"); - await write; - expect(await beforeWrite).toEqual([]); - expect(await afterWrite).toEqual([{ value: 2 }]); - const storage = await SqliteStorage.open(database); const commit = storage.commit( [{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], @@ -183,60 +134,6 @@ describe("portable SQLite facade settlement", () => { await storage.close(BACKGROUND_CONTEXT); }); - it("queues ordinary operations behind an active transaction", async () => { - const database = await openNodeSqliteDatabase(":memory:"); - await database.exec("CREATE TABLE operation_queue (value INTEGER)"); - let markTransactionStarted!: () => void; - const transactionStarted = new Promise((resolve) => { - markTransactionStarted = resolve; - }); - let releaseTransaction!: () => void; - const transactionGate = new Promise((resolve) => { - releaseTransaction = resolve; - }); - const pending = database.transaction(async (transaction) => { - await transaction.exec("INSERT INTO operation_queue (value) VALUES (1)"); - markTransactionStarted(); - await transactionGate; - }); - await transactionStarted; - - let writeSettled = false; - const write = database.exec("INSERT INTO operation_queue (value) VALUES (2)").finally(() => { - writeSettled = true; - }); - let readSettled = false; - const read = database.all("SELECT value FROM operation_queue ORDER BY value").finally(() => { - readSettled = true; - }); - await Promise.resolve(); - expect(writeSettled).toBe(false); - expect(readSettled).toBe(false); - - releaseTransaction(); - await pending; - await write; - await expect(read).resolves.toEqual([{ value: 1 }, { value: 2 }]); - await database.close(); - }); - - it("queues database calls made synchronously by a transaction that started immediately", async () => { - const database = await openNodeSqliteDatabase(":memory:"); - await database.exec("CREATE TABLE barrier_probe (value INTEGER)"); - let outside!: Promise; - const transaction = database.transaction(async (handle) => { - // Misuse: this call must wait for the transaction instead of joining it. - outside = database.run("INSERT INTO barrier_probe (value) VALUES (?)", 2); - await handle.run("INSERT INTO barrier_probe (value) VALUES (?)", 1); - await Promise.resolve(); - throw new Error("roll back"); - }); - await expect(transaction).rejects.toThrow("roll back"); - await outside; - expect(await database.all("SELECT value FROM barrier_probe")).toEqual([{ value: 2 }]); - await database.close(); - }); - it("lets admitted multi-query reads finish before storage closes", async () => { const storage = await SqliteStorage.open(await openNodeSqliteDatabase(":memory:")); const entryId = idFromNumber(2); @@ -262,35 +159,6 @@ describe("portable SQLite facade settlement", () => { ).rejects.toThrow("SqliteStorage is closed"); }); - it("rejects a transaction handle used after its transaction settles", async () => { - const database = await openNodeSqliteDatabase(":memory:"); - await database.exec("CREATE TABLE stale_probe (value INTEGER)"); - let handle!: SqliteExecutor; - await database.transaction(async (transaction) => { - handle = transaction; - await transaction.run("INSERT INTO stale_probe (value) VALUES (?)", 1); - }); - const stale = "SQLite transaction handle is no longer active"; - await expect(handle.exec("INSERT INTO stale_probe (value) VALUES (2)")).rejects.toThrow(stale); - await expect(handle.run("INSERT INTO stale_probe (value) VALUES (?)", 3)).rejects.toThrow(stale); - expect(await database.all("SELECT value FROM stale_probe")).toEqual([{ value: 1 }]); - 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)"); - await expect( - database.transaction(async (transaction) => { - await transaction.exec("INSERT INTO rollback_probe (value) VALUES (1)"); - await transaction.exec("COMMIT"); - throw new Error("rejected after an escaped commit"); - }), - ).rejects.toThrow(AggregateError); - expect(await database.get("SELECT value FROM rollback_probe")).toEqual({ value: 1 }); - await database.close(); - }); - it("awaits async transaction settlement and adopts IDs only after success", async () => { const database = new ControlledSettlementDatabase(await openNodeSqliteDatabase(":memory:")); database.controlNextSettlement("delay"); diff --git a/packages/durable/test/sqlite-migrations.test.ts b/packages/durable/test/sqlite-migrations.test.ts index c55117ea6..a1936f626 100644 --- a/packages/durable/test/sqlite-migrations.test.ts +++ b/packages/durable/test/sqlite-migrations.test.ts @@ -57,44 +57,6 @@ describe("durable SQLite migrations", () => { await expect(openNodeSqliteStorage(path)).rejects.toThrow("is newer than supported version"); }); - it("rolls initial bootstrap and every pending migration back together", async () => { - const database = await openNodeSqliteDatabase(await databasePath()); - try { - const failed: readonly SqliteMigration[] = [ - { - version: 1, - statements: [ - "CREATE TABLE migration_first (value TEXT) STRICT", - "INSERT INTO migration_first (value) VALUES ('retained')", - ], - }, - { - version: 2, - statements: ["CREATE TABLE migration_second (value TEXT) STRICT", "THIS IS NOT SQL"], - }, - ]; - await expect(applySqliteMigrations(database, failed)).rejects.toThrow(); - expect( - await database.get( - "SELECT count(*) AS count FROM sqlite_schema WHERE name IN ('durable_schema', 'migration_first', 'migration_second')", - ), - ).toEqual({ count: 0 }); - - await applySqliteMigrations(database, [ - failed[0], - { version: 2, statements: ["CREATE TABLE migration_second (value TEXT) STRICT"] }, - ]); - expect(await database.get("SELECT version FROM durable_schema WHERE singleton = 1")).toEqual({ - version: 2, - }); - expect(await database.get("SELECT value FROM migration_first")).toEqual({ - value: "retained", - }); - } finally { - await database.close(); - } - }); - it("rolls a failed migration back and preserves stored data for a successful retry", async () => { const path = await databasePath(); const storage = await openNodeSqliteStorage(path); diff --git a/tsconfig.json b/tsconfig.json index 39b7cfd3b..8c3b4d133 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -10,6 +10,9 @@ "@earendil-works/chord/delta": ["./packages/chord/src/delta/index.ts"], "@earendil-works/chord/node": ["./packages/chord/src/node.ts"], "@earendil-works/chord/output": ["./packages/chord/src/output.ts"], + "@earendil-works/chord/sqlite": ["./packages/chord/src/sqlite/index.ts"], + "@earendil-works/chord/sqlite/node": ["./packages/chord/src/sqlite/node.ts"], + "@earendil-works/chord/sqlite/cloudflare": ["./packages/chord/src/sqlite/cloudflare.ts"], "@earendil-works/pi-telemetry": ["./packages/telemetry/src/index.ts"], "@earendil-works/pi-telemetry/testing": ["./packages/telemetry/src/testing/index.ts"], "@earendil-works/pi-codemode": ["./packages/codemode/src/index.ts"],