sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

feat(chord): move Durable's SQLite drivers into chord/sqlite

Browse source
This commit is contained in:
Mario Zechner 2026-10-17 16:48:29 +00:00
parent 8efd314fa3
commit c268154ca1
25 changed files with 991 additions and 825 deletions
Download patch file Download diff file Expand all files Collapse all files

4
packages/chord/CHANGELOG.md
View file

@ -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

3
packages/chord/README.md
View file

@ -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

15
packages/chord/package.json
View file

@ -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,

180
packages/chord/src/sqlite/cloudflare.ts Normal file
View file

@ -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<string, DurableObjectSqlValue> }
| { readonly done: true; readonly value?: undefined };
toArray(): Record<string, DurableObjectSqlValue>[];
};
};
transaction<T>(closure: () => Promise<T>): Promise<T>;
};
/** Runs synchronous operations immediately when idle and holds queued calls behind a transaction. */
class SerialQueue {
private tail: Promise<void> = Promise.resolve();
private pending = 0;
run<T>(operation: () => T | Promise<T>): Promise<T> {
if (this.pending > 0) return this.enqueue(operation);
try {
return Promise.resolve(operation());
} catch (error) {
return Promise.reject(error);
}
}
hold<T>(operation: () => Promise<T>): Promise<T> {
return this.enqueue(operation);
}
private enqueue<T>(operation: () => T | Promise<T>): Promise<T> {
this.pending++;
const result = this.tail.then(operation).finally(() => {
this.pending--;
});
this.tail = result.then(
() => {},
() => {},
);
return result;
}
}
const storageQueues = new WeakMap<DurableObjectSqliteStorage, SerialQueue>();
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<T>(row: Record<string, DurableObjectSqlValue>): T {
for (const key in row) {
const value = row[key];
if (value instanceof ArrayBuffer) (row as Record<string, unknown>)[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<void> {
this.check();
this.storage.sql.exec(sql);
}
async run(sql: string, ...params: SqliteValue[]): Promise<void> {
this.check();
this.storage.sql.exec(sql, ...bind(params));
}
async get<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T | undefined> {
this.check();
const result = this.storage.sql.exec(sql, ...bind(params)).next();
return result.done ? undefined : row<T>(result.value);
}
async all<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T[]> {
this.check();
return this.storage.sql
.exec(sql, ...bind(params))
.toArray()
.map((value) => row<T>(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<void> {
return this.queue.run(() => super.exec(sql));
}
override run(sql: string, ...params: SqliteValue[]): Promise<void> {
return this.queue.run(() => super.run(sql, ...params));
}
override get<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T | undefined> {
return this.queue.run(() => super.get<T>(sql, ...params));
}
override all<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T[]> {
return this.queue.run(() => super.all<T>(sql, ...params));
}
/** `ctx.storage.transaction()` commits when the callback resolves and rolls back when it rejects. */
transaction<T>(callback: (transaction: SqliteExecutor) => Promise<T>): Promise<T> {
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<void> {
return this.queue.run(() => {});
}
}

38
packages/chord/src/sqlite/database.ts Normal file
View file

@ -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<void>;
run(sql: string, ...params: SqliteValue[]): Promise<void>;
get<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T | undefined>;
all<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T[]>;
}
/**
* 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<T>(callback: (transaction: SqliteExecutor) => Promise<T>): Promise<T>;
close(): Promise<void>;
}

3
packages/chord/src/sqlite/index.ts Normal file
View file

@ -0,0 +1,3 @@
export type { SqliteDatabase, SqliteExecutor, SqliteValue } from "./database.ts";
export { applySqliteMigrations, type SqliteMigration } from "./migrations.ts";
export { validateSqliteSchemaPrefix } from "./schema.ts";

43
packages/chord/src/sqlite/migrations.ts Normal file
View file

@ -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<void> {
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<SchemaRow>(`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);
}
});
}

201
packages/chord/src/sqlite/node.ts Normal file
View file

@ -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<void> = Promise.resolve();
private pending = 0;
run<T>(operation: () => T): Promise<T> {
if (this.pending > 0) return this.enqueue(operation);
try {
return Promise.resolve(operation());
} catch (error) {
return Promise.reject(error);
}
}
runAsync<T>(operation: () => Promise<T>): Promise<T> {
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<void>();
this.tail = barrier;
let started: Promise<T>;
try {
started = operation();
} catch (error) {
started = Promise.reject(error);
}
return started.finally(() => {
this.pending--;
releaseBarrier();
});
}
private enqueue<T>(operation: () => T | Promise<T>): Promise<T> {
this.pending++;
return this.release(this.tail.then(operation));
}
private release<T>(operation: Promise<T>): Promise<T> {
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<string, StatementSync>;
constructor(database: DatabaseSync, statements: Map<string, StatementSync>) {
this.database = database;
this.statements = statements;
}
exec(sql: string): Promise<void> {
return this.runOperation(() => {
this.database.exec(sql);
});
}
run(sql: string, ...params: SqliteValue[]): Promise<void> {
return this.runOperation(() => {
this.statement(sql).run(...params);
});
}
get<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T | undefined> {
return this.runOperation(() => this.statement(sql).get(...params) as T | undefined);
}
all<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T[]> {
return this.runOperation(() => this.statement(sql).all(...params) as T[]);
}
protected abstract runOperation<T>(operation: () => T): Promise<T>;
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<string, StatementSync>, scope: TransactionScope) {
super(database, statements);
this.scope = scope;
}
protected async runOperation<T>(operation: () => T): Promise<T> {
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<T>(callback: (transaction: SqliteExecutor) => Promise<T>): Promise<T> {
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<void> {
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<T>(operation: () => T): Promise<T> {
return this.access.run(operation);
}
}
/** Open and configure a Node-backed SQLite database facade. */
export async function openNodeSqliteDatabase(
path: string,
options: NodeSqliteDatabaseOptions = {},
): Promise<NodeSqliteDatabase> {
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;
}
}

9
packages/chord/src/sqlite/schema.ts Normal file
View file

@ -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`);
}
}

116
packages/chord/test/sqlite/cloudflare.test.ts Normal file
View file

@ -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<void>();
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<Parameters<DurableObjectSqliteDatabase["transaction"]>[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<void>();
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<void>();
const { promise: gate, resolve } = Promise.withResolvers<void>();
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([]);
});
});

65
packages/chord/test/sqlite/helpers.ts Normal file
View file

@ -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<string, SqlValue>[]) => {
let index = 0;
return {
next(): { done: false; value: Record<string, SqlValue> } | { done: true } {
onCursorOperation?.("next");
const value = rows[index++];
return value === undefined ? { done: true } : { done: false, value };
},
toArray(): Record<string, SqlValue>[] {
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<string, unknown>[]).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;
}
},
};
}

102
packages/chord/test/sqlite/migrations.test.ts Normal file
View file

@ -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();
}
});
});

169
packages/chord/test/sqlite/node.test.ts Normal file
View file

@ -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<void>((resolve) => {
markFirstStarted = resolve;
});
let releaseFirst!: () => void;
const firstGate = new Promise<void>((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<void>((resolve) => {
markTransactionStarted = resolve;
});
let releaseTransaction!: () => void;
const transactionGate = new Promise<void>((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<void>;
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();
});
});

2
packages/durable/CHANGELOG.md
View file

@ -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

181
packages/durable/src/storage/sqlite/cloudflare.ts
View file

@ -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<string, DurableObjectSqlValue> }
| { readonly done: true; readonly value?: undefined };
toArray(): Record<string, DurableObjectSqlValue>[];
};
};
transaction<T>(closure: () => Promise<T>): Promise<T>;
};
/** Runs synchronous operations immediately when idle and holds queued calls behind a transaction. */
class SerialQueue {
private tail: Promise<void> = Promise.resolve();
private pending = 0;
run<T>(operation: () => T | Promise<T>): Promise<T> {
if (this.pending > 0) return this.enqueue(operation);
try {
return Promise.resolve(operation());
} catch (error) {
return Promise.reject(error);
}
}
hold<T>(operation: () => Promise<T>): Promise<T> {
return this.enqueue(operation);
}
private enqueue<T>(operation: () => T | Promise<T>): Promise<T> {
this.pending++;
const result = this.tail.then(operation).finally(() => {
this.pending--;
});
this.tail = result.then(
() => {},
() => {},
);
return result;
}
}
const storageQueues = new WeakMap<DurableObjectSqliteStorage, SerialQueue>();
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<T>(row: Record<string, DurableObjectSqlValue>): T {
for (const key in row) {
const value = row[key];
if (value instanceof ArrayBuffer) (row as Record<string, unknown>)[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<void> {
this.check();
this.storage.sql.exec(sql);
}
async run(sql: string, ...params: SqliteValue[]): Promise<void> {
this.check();
this.storage.sql.exec(sql, ...bind(params));
}
async get<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T | undefined> {
this.check();
const result = this.storage.sql.exec(sql, ...bind(params)).next();
return result.done ? undefined : row<T>(result.value);
}
async all<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T[]> {
this.check();
return this.storage.sql
.exec(sql, ...bind(params))
.toArray()
.map((value) => row<T>(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<void> {
return this.queue.run(() => super.exec(sql));
}
override run(sql: string, ...params: SqliteValue[]): Promise<void> {
return this.queue.run(() => super.run(sql, ...params));
}
override get<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T | undefined> {
return this.queue.run(() => super.get<T>(sql, ...params));
}
override all<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T[]> {
return this.queue.run(() => super.all<T>(sql, ...params));
}
/** `ctx.storage.transaction()` commits when the callback resolves and rolls back when it rejects. */
transaction<T>(callback: (transaction: SqliteExecutor) => Promise<T>): Promise<T> {
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<void> {
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

39
packages/durable/src/storage/sqlite/database.ts
View file

@ -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<void>;
run(sql: string, ...params: SqliteValue[]): Promise<void>;
get<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T | undefined>;
all<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T[]>;
}
/**
* 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<T>(callback: (transaction: SqliteExecutor) => Promise<T>): Promise<T>;
close(): Promise<void>;
}
export type { SqliteDatabase, SqliteExecutor, SqliteValue } from "@earendil-works/chord/sqlite";

2
packages/durable/src/storage/sqlite/index.ts
View file

@ -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,

41
packages/durable/src/storage/sqlite/migrations.ts
View file

@ -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<void> {
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<SchemaRow>(`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 });
}

208
packages/durable/src/storage/sqlite/node.ts
View file

@ -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<void> = Promise.resolve();
private pending = 0;
run<T>(operation: () => T): Promise<T> {
if (this.pending > 0) return this.enqueue(operation);
try {
return Promise.resolve(operation());
} catch (error) {
return Promise.reject(error);
}
}
runAsync<T>(operation: () => Promise<T>): Promise<T> {
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<void>();
this.tail = barrier;
let started: Promise<T>;
try {
started = operation();
} catch (error) {
started = Promise.reject(error);
}
return started.finally(() => {
this.pending--;
releaseBarrier();
});
}
private enqueue<T>(operation: () => T | Promise<T>): Promise<T> {
this.pending++;
return this.release(this.tail.then(operation));
}
private release<T>(operation: Promise<T>): Promise<T> {
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<string, StatementSync>;
constructor(database: DatabaseSync, statements: Map<string, StatementSync>) {
this.database = database;
this.statements = statements;
}
exec(sql: string): Promise<void> {
return this.runOperation(() => {
this.database.exec(sql);
});
}
run(sql: string, ...params: SqliteValue[]): Promise<void> {
return this.runOperation(() => {
this.statement(sql).run(...params);
});
}
get<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T | undefined> {
return this.runOperation(() => this.statement(sql).get(...params) as T | undefined);
}
all<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T[]> {
return this.runOperation(() => this.statement(sql).all(...params) as T[]);
}
protected abstract runOperation<T>(operation: () => T): Promise<T>;
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<string, StatementSync>, scope: TransactionScope) {
super(database, statements);
this.scope = scope;
}
protected async runOperation<T>(operation: () => T): Promise<T> {
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<T>(callback: (transaction: SqliteExecutor) => Promise<T>): Promise<T> {
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<void> {
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<T>(operation: () => T): Promise<T> {
return this.access.run(operation);
}
}
/** Open and configure a Node-backed SQLite database facade. */
export async function openNodeSqliteDatabase(
path: string,
options: NodeSqliteDatabaseOptions = {},
): Promise<NodeSqliteDatabase> {
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,

14
packages/durable/src/storage/sqlite/schema.ts
View file

@ -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<typeof INDEX_NAMES>;
};
/** 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<T extends Record<string, string>>(names: T, prefix: string): SqliteIdentifiers<T> {
return Object.fromEntries(
Object.entries(names).map(([key, name]) => [key, `${prefix}${name}`]),

2
packages/durable/src/storage/sqlite/storage.ts
View file

@ -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";

180
packages/durable/test/sqlite-cloudflare.test.ts
View file

@ -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<string, SqlValue>[]) => {
let index = 0;
return {
next(): { done: false; value: Record<string, SqlValue> } | { done: true } {
onCursorOperation?.("next");
const value = rows[index++];
return value === undefined ? { done: true } : { done: false, value };
},
toArray(): Record<string, SqlValue>[] {
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<string, unknown>[]).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<void>();
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<Parameters<DurableObjectSqliteDatabase["transaction"]>[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<void>();
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<void>();
const { promise: gate, resolve } = Promise.withResolvers<void>();
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([]);
});
});

158
packages/durable/test/sqlite-facade.test.ts
View file

@ -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<void>((resolve) => {
markFirstStarted = resolve;
});
let releaseFirst!: () => void;
const firstGate = new Promise<void>((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<void>((resolve) => {
markTransactionStarted = resolve;
});
let releaseTransaction!: () => void;
const transactionGate = new Promise<void>((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<void>;
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<EntryId>(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");

38
packages/durable/test/sqlite-migrations.test.ts
View file

@ -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);

3
tsconfig.json
View file

@ -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"],