From f3e68e8eaf3274d8111796b9df4e6b43e2f8e58a Mon Sep 17 00:00:00 2001 From: Christian Klotz Date: Wed, 30 Sep 2026 11:10:50 +0100 Subject: [PATCH 1/4] feat(durable): make SQLite storage asynchronous --- packages/durable/README.md | 12 +- .../durable/src/storage/sqlite/database.ts | 33 +- .../durable/src/storage/sqlite/migrations.ts | 16 +- packages/durable/src/storage/sqlite/node.ts | 124 +++++--- .../durable/src/storage/sqlite/storage.ts | 292 ++++++++++-------- packages/durable/test/sqlite-facade.test.ts | 181 +++++++++-- .../durable/test/sqlite-migrations.test.ts | 60 ++-- 7 files changed, 462 insertions(+), 256 deletions(-) diff --git a/packages/durable/README.md b/packages/durable/README.md index e47a08a49..e0c3fbf3c 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -360,7 +360,17 @@ const usage = await harness.usage(context); // { models: { "openai/gpt-6-sol": U | SQLite | `openNodeSqliteStorage(file)` from `@earendil-works/pi-durable/storage/sqlite/node` | One database file. WAL mode with `synchronous = NORMAL`: commits survive process crashes; the newest may be lost on power or host failure. | | JSONL | `openNodeJsonlStorage(directory, context)` from `@earendil-works/pi-durable/storage/jsonl/node` | Append-only files in one directory. Pass `{ fsync: true }` to flush before each commit marker. | -One process owns a storage at a time; there is no cross-process locking. The portable SQLite and JSONL cores (`/storage/sqlite`, `/storage/jsonl`) run without Node APIs, for example on Bun or in Cloudflare Durable Objects, given a synchronous SQLite database or a `FileSystem` from `@earendil-works/pi-durable/env`. +One process owns a storage at a time; there is no cross-process locking. The portable SQLite and JSONL cores (`/storage/sqlite`, `/storage/jsonl`) run without Node APIs, for example on Bun or in Cloudflare Durable Objects, given an asynchronous `SqliteDatabase` facade or a `FileSystem` from `@earendil-works/pi-durable/env`. + +SQLite adapters implement promise-based `exec`, `prepare`, statement `run`/`get`/`all`, transactions, and close. Operations started by a transaction callback use that transaction; adapters must queue unrelated operations and other transactions until the callback settles: + +```typescript +await database.transaction(async () => { + await database.exec("CREATE TABLE example (value TEXT)"); + const insert = await database.prepare("INSERT INTO example (value) VALUES (?)"); + await insert.run("stored atomically"); +}); +``` Custom backends can run the shared conformance suite with any Vitest- or Jest-compatible runner: diff --git a/packages/durable/src/storage/sqlite/database.ts b/packages/durable/src/storage/sqlite/database.ts index f7dd4ed56..6fb551fba 100644 --- a/packages/durable/src/storage/sqlite/database.ts +++ b/packages/durable/src/storage/sqlite/database.ts @@ -1,31 +1,26 @@ /** Values supported by the portable SQLite storage core. */ export type SqliteValue = null | number | bigint | string | Uint8Array; -/** - * A prepared synchronous SQLite statement. - * Implementations must support repeated execution with new bindings across transactions. - */ +/** A prepared SQLite statement that supports repeated execution with new bindings across transactions. */ export interface SqliteStatement { - run(...params: SqliteValue[]): void; - get(...params: SqliteValue[]): T | undefined; - all(...params: SqliteValue[]): T[]; + run(...params: SqliteValue[]): Promise; + get(...params: SqliteValue[]): Promise; + all(...params: SqliteValue[]): Promise; } /** * Minimal database facade required by `SqliteStorage`. * - * Queries and transaction callbacks are synchronous so the same storage core can - * run on Node, Bun, and Cloudflare Durable Object SQLite. An adapter may return a - * promise from `transaction` while it waits for the transaction to settle. - * When the callback throws, the adapter must roll the transaction back before - * rethrowing that same error. If rollback fails, it must throw a different error - * (for example an `AggregateError`) so callers cannot mistake the callback error - * for a guaranteed rollback. Callers must not close the database while a - * returned settlement is pending. + * All operations are asynchronous so adapters may execute outside the harness runtime. + * Operations started by a transaction callback belong to that transaction; adapters + * must queue unrelated operations until the callback settles. When a 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 { - exec(sql: string): void; - prepare(sql: string): SqliteStatement; - transaction(callback: () => T): T | Promise; - close(): void | Promise; + exec(sql: string): Promise; + prepare(sql: string): Promise; + transaction(callback: () => Promise): Promise; + close(): Promise; } diff --git a/packages/durable/src/storage/sqlite/migrations.ts b/packages/durable/src/storage/sqlite/migrations.ts index d86738cfa..dcadc797f 100644 --- a/packages/durable/src/storage/sqlite/migrations.ts +++ b/packages/durable/src/storage/sqlite/migrations.ts @@ -102,13 +102,15 @@ export async function applySqliteMigrations( } } - await database.transaction(() => { - database.exec(`CREATE TABLE IF NOT EXISTS durable_schema ( + await database.transaction(async () => { + await database.exec(`CREATE TABLE IF NOT EXISTS durable_schema ( singleton INTEGER PRIMARY KEY CHECK (singleton = 1), version INTEGER NOT NULL CHECK (version >= 0) ) STRICT`); - database.prepare("INSERT OR IGNORE INTO durable_schema (singleton, version) VALUES (1, 0)").run(); - const row = database.prepare("SELECT version FROM durable_schema WHERE singleton = 1").get(); + await (await database.prepare("INSERT OR IGNORE INTO durable_schema (singleton, version) VALUES (1, 0)")).run(); + const row = await ( + await database.prepare("SELECT version FROM durable_schema WHERE singleton = 1") + ).get(); if (row === undefined) throw new Error("Durable SQLite schema metadata is missing"); const currentVersion = migrations.at(-1)?.version ?? 0; if (row.version > currentVersion) { @@ -118,8 +120,10 @@ export async function applySqliteMigrations( } for (const migration of migrations) { if (migration.version <= row.version) continue; - for (const statement of migration.statements) database.exec(statement); - database.prepare("UPDATE durable_schema SET version = ? WHERE singleton = 1").run(migration.version); + for (const statement of migration.statements) await database.exec(statement); + await (await database.prepare("UPDATE durable_schema SET version = ? WHERE singleton = 1")).run( + migration.version, + ); } }); } diff --git a/packages/durable/src/storage/sqlite/node.ts b/packages/durable/src/storage/sqlite/node.ts index 2844fde27..3ab973ade 100644 --- a/packages/durable/src/storage/sqlite/node.ts +++ b/packages/durable/src/storage/sqlite/node.ts @@ -1,3 +1,4 @@ +import { AsyncLocalStorage } from "node:async_hooks"; import { mkdir } from "node:fs/promises"; import { dirname } from "node:path"; import type { SQLInputValue, StatementSync } from "node:sqlite"; @@ -16,74 +17,115 @@ export type NodeSqliteStorageOptions = { const DEFAULT_WAL_AUTO_CHECKPOINT_PAGES = 1_000; const DEFAULT_BUSY_TIMEOUT_MS = 5_000; +type RunOperation = (operation: () => T | Promise) => Promise; + +class SerialOperationQueue { + private tail = Promise.resolve(); + + async run(operation: () => T | Promise): Promise { + const previous = this.tail; + let release: () => void; + this.tail = new Promise((resolve) => { + release = resolve; + }); + await previous; + try { + return await operation(); + } finally { + release!(); + } + } +} + class NodeSqliteStatement implements SqliteStatement { private readonly statement: StatementSync; + private readonly runOperation: RunOperation; - constructor(statement: StatementSync) { + constructor(statement: StatementSync, runOperation: RunOperation) { this.statement = statement; + this.runOperation = runOperation; } - run(...params: SqliteValue[]): void { - this.statement.run(...(params as SQLInputValue[])); + run(...params: SqliteValue[]): Promise { + return this.runOperation(() => { + this.statement.run(...(params as SQLInputValue[])); + }); } - get(...params: SqliteValue[]): T | undefined { - return this.statement.get(...(params as SQLInputValue[])) as T | undefined; + get(...params: SqliteValue[]): Promise { + return this.runOperation(() => this.statement.get(...(params as SQLInputValue[])) as T | undefined); } - all(...params: SqliteValue[]): T[] { - return this.statement.all(...(params as SQLInputValue[])) as T[]; + all(...params: SqliteValue[]): Promise { + return this.runOperation(() => this.statement.all(...(params as SQLInputValue[])) as T[]); } } /** `SqliteDatabase` adapter backed by Node's built-in `node:sqlite`. */ export class NodeSqliteDatabase implements SqliteDatabase { private readonly database: DatabaseSync; + private readonly access = new SerialOperationQueue(); + private readonly transactionContext = new AsyncLocalStorage<{ active: boolean }>(); private closed = false; constructor(database: DatabaseSync) { this.database = database; } - exec(sql: string): void { - this.database.exec(sql); + exec(sql: string): Promise { + return this.runOperation(() => { + this.database.exec(sql); + }); } - prepare(sql: string): SqliteStatement { - return new NodeSqliteStatement(this.database.prepare(sql)); + prepare(sql: string): Promise { + return this.runOperation( + () => new NodeSqliteStatement(this.database.prepare(sql), (operation) => this.runOperation(operation)), + ); } - transaction(callback: () => T): T { - this.database.exec("BEGIN IMMEDIATE"); - try { - const result = callback(); - if ( - result !== null && - (typeof result === "object" || typeof result === "function") && - typeof Reflect.get(result, "then") === "function" - ) { - throw new TypeError("SQLite transaction callbacks must be synchronous"); - } - this.database.exec("COMMIT"); - return result; - } catch (error) { + transaction(callback: () => Promise): Promise { + if (this.transactionContext.getStore()?.active === true) { + return Promise.reject(new Error("Nested SQLite transactions are not supported")); + } + return this.access.run(async () => { + this.database.exec("BEGIN IMMEDIATE"); + const context = { active: true }; try { - this.database.exec("ROLLBACK"); - } catch (rollbackError) { - throw new AggregateError([error, rollbackError], "SQLite transaction failed and rollback failed"); + const result = await this.transactionContext.run(context, callback); + context.active = false; + this.database.exec("COMMIT"); + return result; + } catch (error) { + context.active = false; + try { + this.database.exec("ROLLBACK"); + } catch (rollbackError) { + throw new AggregateError([error, rollbackError], "SQLite transaction failed and rollback failed"); + } + throw error; } - throw error; - } + }); } - close(): void { - if (this.closed) return; - this.closed = true; - try { - this.database.exec("PRAGMA wal_checkpoint(TRUNCATE)"); - } finally { - this.database.close(); + close(): Promise { + if (this.transactionContext.getStore()?.active === true) { + return Promise.reject(new Error("Cannot close SQLite during an active transaction")); } + return this.access.run(() => { + if (this.closed) return; + this.closed = true; + try { + this.database.exec("PRAGMA wal_checkpoint(TRUNCATE)"); + } finally { + this.database.close(); + } + }); + } + + private async runOperation(operation: () => T | Promise): Promise { + if (this.transactionContext.getStore()?.active === true) return operation(); + return this.access.run(operation); } } @@ -98,13 +140,13 @@ export async function openNodeSqliteDatabase( const database = new DatabaseSync(path, { timeout }); const adapter = new NodeSqliteDatabase(database); try { - adapter.exec("PRAGMA journal_mode = WAL"); - adapter.exec("PRAGMA synchronous = NORMAL"); - adapter.exec(`PRAGMA wal_autocheckpoint = ${checkpointPages}`); + 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 { - adapter.close(); + await adapter.close(); } catch { // Preserve the configuration failure. } diff --git a/packages/durable/src/storage/sqlite/storage.ts b/packages/durable/src/storage/sqlite/storage.ts index c4ae7a550..8402582d2 100644 --- a/packages/durable/src/storage/sqlite/storage.ts +++ b/packages/durable/src/storage/sqlite/storage.ts @@ -63,11 +63,16 @@ const encodeJson = (value: unknown): string => JSON.stringify(value) as string; // Some SQLite bindings replace lone UTF-16 surrogates. JSON encoding keeps indexed identities lossless. const encodeIndexedString = (value: string): string => JSON.stringify(value); -const getRow = (statement: SqliteStatement, ...params: SqliteValue[]): T | undefined => - statement.get(...params); +const getRow = async ( + statement: Promise, + ...params: SqliteValue[] +): Promise => (await statement).get(...params); -const allRows = (statement: SqliteStatement, ...params: SqliteValue[]): T[] => - statement.all(...params); +const allRows = async (statement: Promise, ...params: SqliteValue[]): Promise => + (await statement).all(...params); + +const runStatement = async (statement: Promise, ...params: SqliteValue[]): Promise => + (await statement).run(...params); const cursorId = >(cursor: Cursor | undefined): I | undefined => { const after = cursor?.after; @@ -140,24 +145,23 @@ class StatementCachingDatabase implements SqliteDatabase { this.database = database; } - exec(sql: string): void { - this.database.exec(sql); + exec(sql: string): Promise { + return this.database.exec(sql); } - prepare(sql: string): SqliteStatement { - let statement = this.statements.get(sql); - if (statement === undefined) { - statement = this.database.prepare(sql); - this.statements.set(sql, statement); - } + async prepare(sql: string): Promise { + const cached = this.statements.get(sql); + if (cached !== undefined) return cached; + const statement = await this.database.prepare(sql); + this.statements.set(sql, statement); return statement; } - transaction(callback: () => T): T | Promise { + transaction(callback: () => Promise): Promise { return this.database.transaction(callback); } - close(): void | Promise { + close(): Promise { this.statements.clear(); return this.database.close(); } @@ -178,7 +182,7 @@ export class SqliteStorage implements Storage { static async open(db: SqliteDatabase): Promise { try { await applySqliteMigrations(db); - const metadata = getRow( + const metadata = await getRow( db.prepare("SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1"), ); if (metadata === undefined) throw new Error("Durable SQLite metadata is missing"); @@ -197,19 +201,21 @@ export class SqliteStorage implements Storage { this.assertOpen(); const documentActions = this.prepareDocumentActions(writes); const candidateNextId = this.candidateNextId(writes); - const seq = await this.db.transaction(() => { - const metadata = getRow( + const seq = await this.db.transaction(async () => { + const metadata = await getRow( this.db.prepare("SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1"), ); if (metadata === undefined) throw new Error("Durable SQLite metadata is missing"); const committedSeq = seqFromNumber(metadata.next_seq); - this.checkGlobalIds(writes); - this.checkDocumentActions(documentActions); - for (const write of writes) this.applyTableWrite(write, committedSeq); - this.applyDocumentActions(documentActions, committedSeq); - this.db - .prepare("UPDATE durable_metadata SET next_id = ?, next_seq = ? WHERE singleton = 1") - .run(String(Math.max(Number(metadata.next_id), candidateNextId)), committedSeq + 1); + await this.checkGlobalIds(writes); + await this.checkDocumentActions(documentActions); + for (const write of writes) await this.applyTableWrite(write, committedSeq); + await this.applyDocumentActions(documentActions, committedSeq); + await runStatement( + this.db.prepare("UPDATE durable_metadata SET next_id = ?, next_seq = ? WHERE singleton = 1"), + String(Math.max(Number(metadata.next_id), candidateNextId)), + committedSeq + 1, + ); return committedSeq; }); this.nextId = Math.max(this.nextId, candidateNextId); @@ -224,7 +230,7 @@ export class SqliteStorage implements Storage { async conversation(id: ConversationId, _context: Context): Promise { this.assertOpen(); - const row = getRow(this.db.prepare("SELECT record FROM conversations WHERE id = ?"), id); + const row = await getRow(this.db.prepare("SELECT record FROM conversations WHERE id = ?"), id); return row === undefined ? undefined : parseJson(row.record); } @@ -246,7 +252,7 @@ export class SqliteStorage implements Storage { params.push(query.ownerTaskId); } params.push(limit + 1); - const rows = allRows( + const rows = await allRows( this.db.prepare(`SELECT record FROM conversations WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`), ...params, ); @@ -278,10 +284,13 @@ export class SqliteStorage implements Storage { let conversation: ConversationRecord | undefined; if (context !== undefined) { const conversationId = idFromNumber(idOrConversationId); - conversation = this.readConversation(conversationId); + conversation = await this.readConversation(conversationId); if (conversation === undefined) throw new Error(`Unknown conversation: ${conversationId}`); } - const row = getRow(this.db.prepare("SELECT record, commit_seq FROM entries WHERE id = ?"), id); + const row = await getRow( + this.db.prepare("SELECT record, commit_seq FROM entries WHERE id = ?"), + id, + ); if (row === undefined) return undefined; const entry = parseJson(row.record); if (conversation !== undefined) { @@ -289,7 +298,7 @@ export class SqliteStorage implements Storage { while (conversation.id !== entry.conversationId) { if (conversation.parent === undefined) return undefined; upperEntryId = Math.min(upperEntryId, conversation.parent.at); - conversation = this.readConversation(conversation.parent.conversationId)!; + conversation = (await this.readConversation(conversation.parent.conversationId))!; } if (entry.id > upperEntryId) return undefined; } @@ -302,19 +311,19 @@ export class SqliteStorage implements Storage { _context: Context, ): Promise<(EntryRecord & { readonly head: EntryId }) | undefined> { this.assertOpen(); - let conversation = this.readConversation(conversationId); + let conversation = await this.readConversation(conversationId); if (conversation === undefined) throw new Error(`Unknown conversation: ${conversationId}`); let upper: number | undefined = atOrBeforeEntryId; while (true) { const row = upper === undefined - ? getRow( + ? await getRow( this.db.prepare( "SELECT record FROM entries WHERE conversation_id = ? AND head IS NOT NULL ORDER BY id DESC LIMIT 1", ), conversation.id, ) - : getRow( + : await getRow( this.db.prepare( "SELECT record FROM entries WHERE conversation_id = ? AND head IS NOT NULL AND id <= ? ORDER BY id DESC LIMIT 1", ), @@ -324,7 +333,7 @@ export class SqliteStorage implements Storage { if (row !== undefined) return parseJson(row.record); if (conversation.parent === undefined) return undefined; upper = upper === undefined ? conversation.parent.at : Math.min(upper, conversation.parent.at); - conversation = this.readConversation(conversation.parent.conversationId)!; + conversation = (await this.readConversation(conversation.parent.conversationId))!; } } @@ -335,7 +344,7 @@ export class SqliteStorage implements Storage { _context: Context, ): Promise> { this.assertOpen(); - let conversation = this.readConversation(query.conversationId); + let conversation = await this.readConversation(query.conversationId); if (conversation === undefined) throw new Error(`Unknown conversation: ${query.conversationId}`); const after = cursorId(cursor); let upper: number | undefined = query.maxEntryId; @@ -353,7 +362,7 @@ export class SqliteStorage implements Storage { params.push(upper); } params.push(limit + 1 - values.length); - const rows = allRows( + const rows = await allRows( this.db.prepare(`SELECT record FROM entries WHERE ${clauses.join(" AND ")} ORDER BY id DESC LIMIT ?`), ...params, ); @@ -361,14 +370,14 @@ export class SqliteStorage implements Storage { if (values.length > limit || conversation.parent === undefined) break; upper = upper === undefined ? conversation.parent.at : Math.min(upper, conversation.parent.at); if (query.minEntryId !== undefined && upper < query.minEntryId) break; - conversation = this.readConversation(conversation.parent.conversationId)!; + conversation = (await this.readConversation(conversation.parent.conversationId))!; } return page(values, limit); } async task(id: TaskId, _context: Context): Promise { this.assertOpen(); - const row = getRow(this.db.prepare("SELECT record FROM tasks WHERE id = ?"), id); + const row = await getRow(this.db.prepare("SELECT record FROM tasks WHERE id = ?"), id); return row === undefined ? undefined : parseJson(row.record); } @@ -402,7 +411,7 @@ export class SqliteStorage implements Storage { params.push(query.background ? 1 : 0); } params.push(limit + 1); - const rows = allRows( + const rows = await allRows( this.db.prepare(`SELECT record FROM tasks WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`), ...params, ); @@ -414,7 +423,7 @@ export class SqliteStorage implements Storage { async submission(id: SubmissionId, _context: Context): Promise { this.assertOpen(); - const row = getRow(this.db.prepare("SELECT record FROM submissions WHERE id = ?"), id); + const row = await getRow(this.db.prepare("SELECT record FROM submissions WHERE id = ?"), id); return row === undefined ? undefined : parseJson(row.record); } @@ -436,7 +445,7 @@ export class SqliteStorage implements Storage { params.push(query.status); } params.push(limit + 1); - const rows = allRows( + const rows = await allRows( this.db.prepare(`SELECT record FROM submissions WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`), ...params, ); @@ -452,7 +461,7 @@ export class SqliteStorage implements Storage { _context: Context, ): Promise { this.assertOpen(); - const row = getRow( + const row = await getRow( this.db.prepare("SELECT record FROM submissions WHERE conversation_id = ? AND request_id = ?"), conversationId, encodeIndexedString(requestId), @@ -478,7 +487,7 @@ export class SqliteStorage implements Storage { ORDER BY created_at DESC LIMIT 1`); const params: SqliteValue[] = [parts.kind, parts.scopeKind, parts.ownerId, parts.family, parts.keyValue]; if (at !== "current") params.push(at, at); - const row = getRow(statement, ...params); + const row = await getRow(statement, ...params); return row === undefined ? undefined : parseJson(row.record); } @@ -508,7 +517,7 @@ export class SqliteStorage implements Storage { params.push(query.at, query.at); } params.push(limit + 1); - const rows = allRows( + const rows = await allRows( this.db.prepare(`SELECT record FROM documents WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`), ...params, ); @@ -524,13 +533,13 @@ export class SqliteStorage implements Storage { await this.db.close(); } - private readConversation(id: ConversationId): ConversationRecord | undefined { - const row = getRow(this.db.prepare("SELECT record FROM conversations WHERE id = ?"), id); + private async readConversation(id: ConversationId): Promise { + const row = await getRow(this.db.prepare("SELECT record FROM conversations WHERE id = ?"), id); return row === undefined ? undefined : parseJson(row.record); } - private materializeDocument(id: DocumentId, at: DocumentPoint): StoredDocument | undefined { - const row = getRow(this.db.prepare("SELECT record FROM documents WHERE id = ?"), id); + private async materializeDocument(id: DocumentId, at: DocumentPoint): Promise { + const row = await getRow(this.db.prepare("SELECT record FROM documents WHERE id = ?"), id); if (row === undefined) return undefined; const record = parseJson(row.record); if (at !== "current" && isCurrentOnly(record)) { @@ -538,7 +547,7 @@ export class SqliteStorage implements Storage { } if (!isAliveAt(record, at)) return undefined; const upper = at === "current" ? Number.MAX_SAFE_INTEGER : at; - const base = getRow( + const base = await getRow( this.db.prepare(`SELECT seq, kind, version, content FROM document_revisions WHERE document_id = ? AND kind = 'base' AND seq <= ? ORDER BY seq DESC LIMIT 1`), id, @@ -546,7 +555,7 @@ export class SqliteStorage implements Storage { ); if (base === undefined) throw new Error(`Document ${id} is missing a required base`); let value = parseJson(base.content); - const tail = allRows( + const tail = await allRows( this.db.prepare(`SELECT seq, kind, version, content FROM document_revisions WHERE document_id = ? AND seq > ? AND seq <= ? ORDER BY seq`), id, @@ -571,7 +580,7 @@ export class SqliteStorage implements Storage { return nextId; } - private checkGlobalIds(writes: readonly StorageWrite[]): void { + private async checkGlobalIds(writes: readonly StorageWrite[]): Promise { const claimed = new Map, TableName>(); const lookup = this.db.prepare("SELECT record_type FROM record_ids WHERE id = ?"); for (const write of writes) { @@ -579,7 +588,7 @@ export class SqliteStorage implements Storage { const document = write.type === "document.create" || write.type === "document.copy"; const table: TableName = document ? "document" : write.type; const id = document ? write.record.id : write.value.id; - const existing = getRow(lookup, id)?.record_type; + const existing = (await getRow(lookup, id))?.record_type; const earlier = claimed.get(id); if (table === "conversation" || table === "entry" || table === "document") { if (existing !== undefined) throw new Error(`ID ${id} already belongs to ${existing}`); @@ -640,19 +649,19 @@ export class SqliteStorage implements Storage { return actions; } - private checkDocumentActions(actions: ReadonlyMap): void { + private async checkDocumentActions(actions: ReadonlyMap): Promise { const liveCounts = new Map(); for (const [id, action] of actions) { if (action.copy !== undefined && actions.has(action.copy.id)) { throw new StorageRejected(`Document copy ${id} source is changed in the copy batch`); } - const row = getRow(this.db.prepare("SELECT record FROM documents WHERE id = ?"), id); + const row = await getRow(this.db.prepare("SELECT record FROM documents WHERE id = ?"), id); const existing = row === undefined ? undefined : parseJson(row.record); if (action.create === undefined && existing === undefined) throw new Error(`Unknown document: ${id}`); if (action.create !== undefined && existing !== undefined) throw new Error(`Document ${id} already exists`); if (existing?.retiredAt !== undefined) throw new Error(`Document ${id} is retired`); if (action.content?.kind === "delta") { - const previous = getRow<{ readonly version: number }>( + const previous = await getRow<{ readonly version: number }>( this.db.prepare( "SELECT version FROM document_revisions WHERE document_id = ? ORDER BY seq DESC LIMIT 1", ), @@ -666,7 +675,7 @@ export class SqliteStorage implements Storage { const record = action.create ?? existing!; const key = addressKey(record); let live = liveCounts.get(key); - if (live === undefined) live = this.currentDocumentId(record) === undefined ? 0 : 1; + if (live === undefined) live = (await this.currentDocumentId(record)) === undefined ? 0 : 1; if (action.retire && existing !== undefined) live--; if (action.create !== undefined && !action.retire) live++; liveCounts.set(key, live); @@ -676,73 +685,81 @@ export class SqliteStorage implements Storage { } } - private currentDocumentId(address: DocumentAddress | DocumentCreate | DocumentRecord): DocumentId | undefined { + private async currentDocumentId( + address: DocumentAddress | DocumentCreate | DocumentRecord, + ): Promise { const parts = addressParts(address); - const id = getRow( - this.db.prepare(`SELECT id FROM documents + const id = ( + await getRow( + this.db.prepare(`SELECT id FROM documents WHERE kind = ? AND scope_kind = ? AND owner_id = ? AND family = ? AND key_value = ? AND retired_at IS NULL LIMIT 1`), - parts.kind, - parts.scopeKind, - parts.ownerId, - parts.family, - parts.keyValue, + parts.kind, + parts.scopeKind, + parts.ownerId, + parts.family, + parts.keyValue, + ) )?.id; return id === undefined ? undefined : idFromNumber(id); } - private applyTableWrite(write: StorageWrite, seq: Seq): void { + private async applyTableWrite(write: StorageWrite, seq: Seq): Promise { switch (write.type) { case "conversation": - this.claimId(write.value.id, "conversation"); - this.db - .prepare( + await this.claimId(write.value.id, "conversation"); + await runStatement( + this.db.prepare( "INSERT INTO conversations (id, owner_conversation_id, owner_task_id, record) VALUES (?, ?, ?, ?)", - ) - .run( - write.value.id, - write.value.owner?.conversationId ?? null, - write.value.owner?.taskId ?? null, - encodeJson(write.value), - ); + ), + write.value.id, + write.value.owner?.conversationId ?? null, + write.value.owner?.taskId ?? null, + encodeJson(write.value), + ); break; case "entry": - this.claimId(write.value.id, "entry"); - this.db - .prepare("INSERT INTO entries (id, conversation_id, head, commit_seq, record) VALUES (?, ?, ?, ?, ?)") - .run(write.value.id, write.value.conversationId, write.value.head ?? null, seq, encodeJson(write.value)); + await this.claimId(write.value.id, "entry"); + await runStatement( + this.db.prepare( + "INSERT INTO entries (id, conversation_id, head, commit_seq, record) VALUES (?, ?, ?, ?, ?)", + ), + write.value.id, + write.value.conversationId, + write.value.head ?? null, + seq, + encodeJson(write.value), + ); break; case "task": - this.claimId(write.value.id, "task"); - this.db - .prepare(`INSERT INTO tasks (id, conversation_id, kind, status, abort_requested, background, record) + await this.claimId(write.value.id, "task"); + await runStatement( + this.db.prepare(`INSERT INTO tasks (id, conversation_id, kind, status, abort_requested, background, record) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET conversation_id = excluded.conversation_id, kind = excluded.kind, status = excluded.status, abort_requested = excluded.abort_requested, - background = excluded.background, record = excluded.record`) - .run( - write.value.id, - write.value.conversationId, - encodeIndexedString(write.value.kind), - write.value.state.status, - write.value.abortRequested ? 1 : 0, - write.value.background ? 1 : 0, - encodeJson(write.value), - ); + background = excluded.background, record = excluded.record`), + write.value.id, + write.value.conversationId, + encodeIndexedString(write.value.kind), + write.value.state.status, + write.value.abortRequested ? 1 : 0, + write.value.background ? 1 : 0, + encodeJson(write.value), + ); break; case "submission": - this.claimId(write.value.id, "submission"); - this.db - .prepare(`INSERT INTO submissions (id, conversation_id, request_id, status, record) VALUES (?, ?, ?, ?, ?) + await this.claimId(write.value.id, "submission"); + await runStatement( + this.db.prepare(`INSERT INTO submissions (id, conversation_id, request_id, status, record) VALUES (?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET conversation_id = excluded.conversation_id, - request_id = excluded.request_id, status = excluded.status, record = excluded.record`) - .run( - write.value.id, - write.value.conversationId, - write.value.requestId === undefined ? null : encodeIndexedString(write.value.requestId), - write.value.status, - encodeJson(write.value), - ); + request_id = excluded.request_id, status = excluded.status, record = excluded.record`), + write.value.id, + write.value.conversationId, + write.value.requestId === undefined ? null : encodeIndexedString(write.value.requestId), + write.value.status, + encodeJson(write.value), + ); break; case "document.create": case "document.copy": @@ -752,16 +769,20 @@ export class SqliteStorage implements Storage { } } - private claimId(id: Id, table: TableName): void { - this.db.prepare("INSERT OR IGNORE INTO record_ids (id, record_type) VALUES (?, ?)").run(id, table); + private async claimId(id: Id, table: TableName): Promise { + await runStatement( + this.db.prepare("INSERT OR IGNORE INTO record_ids (id, record_type) VALUES (?, ?)"), + id, + table, + ); } - private applyDocumentActions(actions: ReadonlyMap, seq: Seq): void { + private async applyDocumentActions(actions: ReadonlyMap, seq: Seq): Promise { for (const [id, action] of actions) { let content = action.content; if (action.copy !== undefined) { try { - const stored = this.materializeDocument(action.copy.id, action.copy.at); + const stored = await this.materializeDocument(action.copy.id, action.copy.at); if (stored === undefined) throw new Error(`Fork source document ${action.copy.id} cannot be read`); const create = action.create!; if ( @@ -788,47 +809,56 @@ export class SqliteStorage implements Storage { ...(action.retire ? { retiredAt: seq } : {}), }; const parts = addressParts(record); - this.claimId(id, "document"); - this.db - .prepare(`INSERT INTO documents + await this.claimId(id, "document"); + await runStatement( + this.db.prepare(`INSERT INTO documents (id, kind, family, key_value, scope_kind, owner_id, created_at, retired_at, record) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`) - .run( - id, - parts.kind, - parts.family, - parts.keyValue, - parts.scopeKind, - parts.ownerId, - seq, - action.retire ? seq : null, - encodeJson(record), - ); + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`), + id, + parts.kind, + parts.family, + parts.keyValue, + parts.scopeKind, + parts.ownerId, + seq, + action.retire ? seq : null, + encodeJson(record), + ); } else { - const row = getRow(this.db.prepare("SELECT record FROM documents WHERE id = ?"), id)!; + const row = (await getRow(this.db.prepare("SELECT record FROM documents WHERE id = ?"), id))!; record = parseJson(row.record); } if (content !== undefined) { if (content.kind === "base" && isCurrentOnly(record)) { - this.db.prepare("DELETE FROM document_revisions WHERE document_id = ?").run(id); + await runStatement(this.db.prepare("DELETE FROM document_revisions WHERE document_id = ?"), id); } const encodedContent = content.kind === "base" ? encodeJson(content.value) : encodeJson(content.ops); - this.db - .prepare( + await runStatement( + this.db.prepare( "INSERT INTO document_revisions (document_id, seq, kind, version, content) VALUES (?, ?, ?, ?, ?)", - ) - .run(id, seq, content.kind, content.version, encodedContent); + ), + id, + seq, + content.kind, + content.version, + encodedContent, + ); } if (action.retire) { if (action.create === undefined) { record = { ...record, retiredAt: seq }; - this.db - .prepare("UPDATE documents SET retired_at = ?, record = ? WHERE id = ?") - .run(seq, encodeJson(record), id); + await runStatement( + this.db.prepare("UPDATE documents SET retired_at = ?, record = ? WHERE id = ?"), + seq, + encodeJson(record), + id, + ); + } + if (isCurrentOnly(record)) { + await runStatement(this.db.prepare("DELETE FROM document_revisions WHERE document_id = ?"), id); } - if (isCurrentOnly(record)) this.db.prepare("DELETE FROM document_revisions WHERE document_id = ?").run(id); } } } diff --git a/packages/durable/test/sqlite-facade.test.ts b/packages/durable/test/sqlite-facade.test.ts index fe24eeaf2..90129bde4 100644 --- a/packages/durable/test/sqlite-facade.test.ts +++ b/packages/durable/test/sqlite-facade.test.ts @@ -1,10 +1,13 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; import { describe, expect, it } from "vitest"; import { StorageRejected } from "../src/errors.ts"; import { idFromNumber } from "../src/ids.ts"; import type { SqliteDatabase, SqliteStatement } from "../src/storage/sqlite/index.ts"; import { SqliteStorage } from "../src/storage/sqlite/index.ts"; -import { type NodeSqliteDatabase, openNodeSqliteDatabase } from "../src/storage/sqlite/node.ts"; +import { type NodeSqliteDatabase, openNodeSqliteDatabase, openNodeSqliteStorage } from "../src/storage/sqlite/node.ts"; import { type EntryId, ROOT_CONVERSATION_ID } from "../src/types.ts"; type SettlementMode = "immediate" | "delay" | "reject"; @@ -19,37 +22,31 @@ class ControlledSettlementDatabase implements SqliteDatabase { this.delegate = delegate; } - exec(sql: string): void { - this.delegate.exec(sql); + exec(sql: string): Promise { + return this.delegate.exec(sql); } - prepare(sql: string): SqliteStatement { + prepare(sql: string): Promise { this.prepareCounts.set(sql, (this.prepareCounts.get(sql) ?? 0) + 1); return this.delegate.prepare(sql); } - transaction(callback: () => T): T | Promise { + transaction(callback: () => Promise): Promise { const mode = this.mode; this.mode = "immediate"; if (mode === "immediate") return this.delegate.transaction(callback); - try { - const result = this.delegate.transaction(() => { - const value = callback(); - if (mode === "reject") throw new Error("controlled settlement rejection"); - return value; - }); - return new Promise((resolve) => { - this.pendingSettlement = () => resolve(result); - }); - } catch (error) { - return new Promise((_resolve, reject) => { - this.pendingSettlement = () => reject(error); - }); - } + const settlement = this.delegate.transaction(async () => { + const value = await callback(); + if (mode === "reject") throw new Error("controlled settlement rejection"); + return value; + }); + return new Promise((resolve, reject) => { + this.pendingSettlement = () => void settlement.then(resolve, reject); + }); } - close(): void { - this.delegate.close(); + close(): Promise { + return this.delegate.close(); } prepareCount(sql: string): number { @@ -106,27 +103,143 @@ describe("portable SQLite facade settlement", () => { await storage.close(BACKGROUND_CONTEXT); }); - it("rejects asynchronous Node transaction callbacks and closes idempotently", async () => { + it("supports asynchronous Node transaction callbacks and closes idempotently", async () => { const database = await openNodeSqliteDatabase(":memory:"); - expect(() => database.transaction(() => Promise.resolve())).toThrow( - "SQLite transaction callbacks must be synchronous", + await database.transaction(async () => { + await database.exec("CREATE TABLE async_probe (value INTEGER)"); + await (await database.prepare("INSERT INTO async_probe (value) VALUES (?)")).run(1); + }); + expect(await (await database.prepare("SELECT value FROM async_probe")).get()).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 () => { + await database.exec("INSERT INTO transaction_queue (value) VALUES (1)"); + markFirstStarted(); + await firstGate; + }); + await firstStarted; + + let secondStarted = false; + const second = database.transaction(async () => { + secondStarted = true; + await database.exec("INSERT INTO transaction_queue (value) VALUES (2)"); + }); + await Promise.resolve(); + expect(secondStarted).toBe(false); + + releaseFirst(); + await Promise.all([first, second]); + expect(await (await database.prepare("SELECT value FROM transaction_queue ORDER BY value")).all()).toEqual([ + { value: 1 }, + { 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)"); + const readStatement = await database.prepare("SELECT value FROM operation_queue ORDER BY value"); + let markTransactionStarted!: () => void; + const transactionStarted = new Promise((resolve) => { + markTransactionStarted = resolve; + }); + let releaseTransaction!: () => void; + const transactionGate = new Promise((resolve) => { + releaseTransaction = resolve; + }); + const transaction = database.transaction(async () => { + await database.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 = readStatement.all().finally(() => { + readSettled = true; + }); + await Promise.resolve(); + expect(writeSettled).toBe(false); + expect(readSettled).toBe(false); + + releaseTransaction(); + await transaction; + await write; + await expect(read).resolves.toEqual([{ value: 1 }, { value: 2 }]); + await database.close(); + }); + + it("does not share an in-flight prepare across transaction contexts after reopening", async () => { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-sqlite-cache-")); + const path = join(directory, "storage.sqlite"); + try { + const initial = await openNodeSqliteStorage(path); + await initial.close(BACKGROUND_CONTEXT); + + const database = await openNodeSqliteDatabase(path); + const storage = await SqliteStorage.open(database); + let markTransactionStarted!: () => void; + const transactionStarted = new Promise((resolve) => { + markTransactionStarted = resolve; + }); + let releaseTransaction!: () => void; + const transactionGate = new Promise((resolve) => { + releaseTransaction = resolve; + }); + const transaction = database.transaction(async () => { + markTransactionStarted(); + await transactionGate; + return storage.conversation(ROOT_CONVERSATION_ID, BACKGROUND_CONTEXT); + }); + await transactionStarted; + + const concurrentRead = storage.conversation(ROOT_CONVERSATION_ID, BACKGROUND_CONTEXT); + releaseTransaction(); + await expect(Promise.all([transaction, concurrentRead])).resolves.toEqual([undefined, undefined]); + await storage.close(BACKGROUND_CONTEXT); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + + it("rejects close from an active transaction", async () => { + const database = await openNodeSqliteDatabase(":memory:"); + await expect(database.transaction(async () => database.close())).rejects.toThrow( + "Cannot close SQLite during an active transaction", ); - database.close(); - database.close(); + await database.close(); }); it("does not preserve a guaranteed rejection when rollback itself fails", async () => { const database = await openNodeSqliteDatabase(":memory:"); - database.exec("CREATE TABLE rollback_probe (value INTEGER)"); - expect(() => - database.transaction(() => { - database.exec("INSERT INTO rollback_probe (value) VALUES (1)"); - database.exec("COMMIT"); + await database.exec("CREATE TABLE rollback_probe (value INTEGER)"); + await expect( + database.transaction(async () => { + await database.exec("INSERT INTO rollback_probe (value) VALUES (1)"); + await database.exec("COMMIT"); throw new StorageRejected("rejected after an escaped commit"); }), - ).toThrow(AggregateError); - expect(database.prepare("SELECT value FROM rollback_probe").get()).toEqual({ value: 1 }); - database.close(); + ).rejects.toThrow(AggregateError); + expect(await (await database.prepare("SELECT value FROM rollback_probe")).get()).toEqual({ value: 1 }); + await database.close(); }); it("awaits async transaction settlement and adopts IDs only after success", async () => { diff --git a/packages/durable/test/sqlite-migrations.test.ts b/packages/durable/test/sqlite-migrations.test.ts index 0e2e0665e..535475641 100644 --- a/packages/durable/test/sqlite-migrations.test.ts +++ b/packages/durable/test/sqlite-migrations.test.ts @@ -32,15 +32,19 @@ describe("durable SQLite migrations", () => { try { await applySqliteMigrations(database); await applySqliteMigrations(database); - expect(database.prepare("SELECT version FROM durable_schema WHERE singleton = 1").get()).toEqual({ - version: CURRENT_SQLITE_SCHEMA_VERSION, - }); - expect(database.prepare("SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1").get()).toEqual({ + expect(await (await database.prepare("SELECT version FROM durable_schema WHERE singleton = 1")).get()).toEqual( + { + version: CURRENT_SQLITE_SCHEMA_VERSION, + }, + ); + expect( + await (await database.prepare("SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1")).get(), + ).toEqual({ next_id: "2", next_seq: 1, }); } finally { - database.close(); + await database.close(); } }); @@ -48,10 +52,10 @@ describe("durable SQLite migrations", () => { const path = await databasePath(); const database = await openNodeSqliteDatabase(path); await applySqliteMigrations(database); - database - .prepare("UPDATE durable_schema SET version = ? WHERE singleton = 1") - .run(CURRENT_SQLITE_SCHEMA_VERSION + 1); - database.close(); + await (await database.prepare("UPDATE durable_schema SET version = ? WHERE singleton = 1")).run( + CURRENT_SQLITE_SCHEMA_VERSION + 1, + ); + await database.close(); await expect(openNodeSqliteStorage(path)).rejects.toThrow("is newer than supported version"); }); @@ -74,23 +78,27 @@ describe("durable SQLite migrations", () => { ]; await expect(applySqliteMigrations(database, failed)).rejects.toThrow(); expect( - database - .prepare( + await ( + await database.prepare( "SELECT count(*) AS count FROM sqlite_schema WHERE name IN ('durable_schema', 'migration_first', 'migration_second')", ) - .get(), + ).get(), ).toEqual({ count: 0 }); await applySqliteMigrations(database, [ failed[0], { version: 2, statements: ["CREATE TABLE migration_second (value TEXT) STRICT"] }, ]); - expect(database.prepare("SELECT version FROM durable_schema WHERE singleton = 1").get()).toEqual({ - version: 2, + expect(await (await database.prepare("SELECT version FROM durable_schema WHERE singleton = 1")).get()).toEqual( + { + version: 2, + }, + ); + expect(await (await database.prepare("SELECT value FROM migration_first")).get()).toEqual({ + value: "retained", }); - expect(database.prepare("SELECT value FROM migration_first").get()).toEqual({ value: "retained" }); } finally { - database.close(); + await database.close(); } }); @@ -124,13 +132,15 @@ describe("durable SQLite migrations", () => { }, ]; await expect(applySqliteMigrations(database, failedMigrations)).rejects.toThrow(); - expect(database.prepare("SELECT version FROM durable_schema WHERE singleton = 1").get()).toEqual({ + expect(await (await database.prepare("SELECT version FROM durable_schema WHERE singleton = 1")).get()).toEqual({ version: CURRENT_SQLITE_SCHEMA_VERSION, }); expect( - database - .prepare("SELECT count(*) AS count FROM sqlite_schema WHERE type = 'table' AND name = 'migration_probe'") - .get(), + await ( + await database.prepare( + "SELECT count(*) AS count FROM sqlite_schema WHERE type = 'table' AND name = 'migration_probe'", + ) + ).get(), ).toEqual({ count: 0 }); const successfulMigrations: readonly SqliteMigration[] = [ @@ -138,10 +148,10 @@ describe("durable SQLite migrations", () => { { version: nextVersion, statements: ["CREATE TABLE migration_probe (value TEXT) STRICT"] }, ]; await applySqliteMigrations(database, successfulMigrations); - expect(database.prepare("SELECT version FROM durable_schema WHERE singleton = 1").get()).toEqual({ + expect(await (await database.prepare("SELECT version FROM durable_schema WHERE singleton = 1")).get()).toEqual({ version: nextVersion, }); - expect(database.prepare("SELECT record, commit_seq FROM entries WHERE id = 2").get()).toEqual({ + expect(await (await database.prepare("SELECT record, commit_seq FROM entries WHERE id = 2")).get()).toEqual({ record: JSON.stringify({ id: 2, conversationId: ROOT_CONVERSATION_ID, @@ -150,10 +160,12 @@ describe("durable SQLite migrations", () => { }), commit_seq: 1, }); - expect(database.prepare("SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1").get()).toEqual({ + expect( + await (await database.prepare("SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1")).get(), + ).toEqual({ next_id: "3", next_seq: 2, }); - database.close(); + await database.close(); }); }); From fd3af5ee2bb27b51d24de52be77723b41834798d Mon Sep 17 00:00:00 2001 From: Christian Klotz Date: Wed, 30 Sep 2026 11:39:02 +0100 Subject: [PATCH 2/4] feat(durable): pass a transaction handle to SQLite transaction callbacks --- packages/durable/README.md | 8 +- .../durable/src/storage/sqlite/database.ts | 29 ++-- packages/durable/src/storage/sqlite/index.ts | 2 +- .../durable/src/storage/sqlite/migrations.ts | 14 +- packages/durable/src/storage/sqlite/node.ts | 59 +++++-- .../durable/src/storage/sqlite/storage.ts | 153 ++++++++++-------- packages/durable/test/sqlite-facade.test.ts | 126 ++++++++------- 7 files changed, 226 insertions(+), 165 deletions(-) diff --git a/packages/durable/README.md b/packages/durable/README.md index e0c3fbf3c..9c43ca9a4 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -362,12 +362,12 @@ const usage = await harness.usage(context); // { models: { "openai/gpt-6-sol": U One process owns a storage at a time; there is no cross-process locking. The portable SQLite and JSONL cores (`/storage/sqlite`, `/storage/jsonl`) run without Node APIs, for example on Bun or in Cloudflare Durable Objects, given an asynchronous `SqliteDatabase` facade or a `FileSystem` from `@earendil-works/pi-durable/env`. -SQLite adapters implement promise-based `exec`, `prepare`, statement `run`/`get`/`all`, transactions, and close. Operations started by a transaction callback use that transaction; adapters must queue unrelated operations and other transactions until the callback settles: +SQLite adapters implement promise-based `exec`, `prepare`, statement `run`/`get`/`all`, transactions, and close. A transaction callback receives a transaction handle; all work in the transaction must use it, and the handle expires when the callback settles. Adapters must queue unrelated operations and other transactions until the transaction finishes: ```typescript -await database.transaction(async () => { - await database.exec("CREATE TABLE example (value TEXT)"); - const insert = await database.prepare("INSERT INTO example (value) VALUES (?)"); +await database.transaction(async (transaction) => { + await transaction.exec("CREATE TABLE example (value TEXT)"); + const insert = await transaction.prepare("INSERT INTO example (value) VALUES (?)"); await insert.run("stored atomically"); }); ``` diff --git a/packages/durable/src/storage/sqlite/database.ts b/packages/durable/src/storage/sqlite/database.ts index 6fb551fba..0c55841dd 100644 --- a/packages/durable/src/storage/sqlite/database.ts +++ b/packages/durable/src/storage/sqlite/database.ts @@ -1,26 +1,35 @@ /** Values supported by the portable SQLite storage core. */ export type SqliteValue = null | number | bigint | string | Uint8Array; -/** A prepared SQLite statement that supports repeated execution with new bindings across transactions. */ +/** A prepared SQLite statement that supports repeated execution with new bindings. */ export interface SqliteStatement { run(...params: SqliteValue[]): Promise; get(...params: SqliteValue[]): Promise; all(...params: SqliteValue[]): Promise; } +/** Asynchronous SQL operations shared by a database and its transaction handles. */ +export interface SqliteExecutor { + exec(sql: string): Promise; + prepare(sql: string): Promise; +} + /** * Minimal database facade required by `SqliteStorage`. * * All operations are asynchronous so adapters may execute outside the harness runtime. - * Operations started by a transaction callback belong to that transaction; adapters - * must queue unrelated operations until the callback settles. When a 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. + * + * `transaction` passes the callback a transaction handle. All work in the transaction + * must use that handle; the handle and statements prepared through it are 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. + * + * 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 { - exec(sql: string): Promise; - prepare(sql: string): Promise; - transaction(callback: () => Promise): Promise; +export interface SqliteDatabase extends SqliteExecutor { + transaction(callback: (transaction: SqliteExecutor) => Promise): Promise; close(): Promise; } diff --git a/packages/durable/src/storage/sqlite/index.ts b/packages/durable/src/storage/sqlite/index.ts index 5d3a4d352..86805bc23 100644 --- a/packages/durable/src/storage/sqlite/index.ts +++ b/packages/durable/src/storage/sqlite/index.ts @@ -1,4 +1,4 @@ -export type { SqliteDatabase, SqliteStatement, SqliteValue } from "./database.ts"; +export type { SqliteDatabase, SqliteExecutor, SqliteStatement, SqliteValue } from "./database.ts"; 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 dcadc797f..d5e124c96 100644 --- a/packages/durable/src/storage/sqlite/migrations.ts +++ b/packages/durable/src/storage/sqlite/migrations.ts @@ -102,14 +102,16 @@ export async function applySqliteMigrations( } } - await database.transaction(async () => { - await database.exec(`CREATE TABLE IF NOT EXISTS durable_schema ( + await database.transaction(async (transaction) => { + await transaction.exec(`CREATE TABLE IF NOT EXISTS durable_schema ( singleton INTEGER PRIMARY KEY CHECK (singleton = 1), version INTEGER NOT NULL CHECK (version >= 0) ) STRICT`); - await (await database.prepare("INSERT OR IGNORE INTO durable_schema (singleton, version) VALUES (1, 0)")).run(); + await ( + await transaction.prepare("INSERT OR IGNORE INTO durable_schema (singleton, version) VALUES (1, 0)") + ).run(); const row = await ( - await database.prepare("SELECT version FROM durable_schema WHERE singleton = 1") + await transaction.prepare("SELECT version FROM durable_schema WHERE singleton = 1") ).get(); if (row === undefined) throw new Error("Durable SQLite schema metadata is missing"); const currentVersion = migrations.at(-1)?.version ?? 0; @@ -120,8 +122,8 @@ export async function applySqliteMigrations( } for (const migration of migrations) { if (migration.version <= row.version) continue; - for (const statement of migration.statements) await database.exec(statement); - await (await database.prepare("UPDATE durable_schema SET version = ? WHERE singleton = 1")).run( + for (const statement of migration.statements) await transaction.exec(statement); + await (await transaction.prepare("UPDATE durable_schema SET version = ? WHERE singleton = 1")).run( migration.version, ); } diff --git a/packages/durable/src/storage/sqlite/node.ts b/packages/durable/src/storage/sqlite/node.ts index 3ab973ade..7c3d1aef6 100644 --- a/packages/durable/src/storage/sqlite/node.ts +++ b/packages/durable/src/storage/sqlite/node.ts @@ -3,7 +3,7 @@ import { mkdir } from "node:fs/promises"; import { dirname } from "node:path"; import type { SQLInputValue, StatementSync } from "node:sqlite"; import { DatabaseSync } from "node:sqlite"; -import type { SqliteDatabase, SqliteStatement, SqliteValue } from "./database.ts"; +import type { SqliteDatabase, SqliteExecutor, SqliteStatement, SqliteValue } from "./database.ts"; import { SqliteStorage } from "./storage.ts"; /** Node SQLite connection settings for a durable storage file. */ @@ -18,6 +18,7 @@ const DEFAULT_WAL_AUTO_CHECKPOINT_PAGES = 1_000; const DEFAULT_BUSY_TIMEOUT_MS = 5_000; type RunOperation = (operation: () => T | Promise) => Promise; +type TransactionScope = { active: boolean }; class SerialOperationQueue { private tail = Promise.resolve(); @@ -61,11 +62,39 @@ class NodeSqliteStatement implements SqliteStatement { } } +class NodeSqliteTransaction implements SqliteExecutor { + private readonly database: DatabaseSync; + private readonly scope: TransactionScope; + + constructor(database: DatabaseSync, scope: TransactionScope) { + this.database = database; + this.scope = scope; + } + + exec(sql: string): Promise { + return this.runOperation(() => { + this.database.exec(sql); + }); + } + + prepare(sql: string): Promise { + return this.runOperation( + () => new NodeSqliteStatement(this.database.prepare(sql), (operation) => this.runOperation(operation)), + ); + } + + private async runOperation(operation: () => T | Promise): 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 implements SqliteDatabase { private readonly database: DatabaseSync; private readonly access = new SerialOperationQueue(); - private readonly transactionContext = new AsyncLocalStorage<{ active: boolean }>(); + /** Detects database calls from inside a transaction callback, which would otherwise wait for that transaction forever. */ + private readonly transactionScope = new AsyncLocalStorage(); private closed = false; constructor(database: DatabaseSync) { @@ -84,20 +113,22 @@ export class NodeSqliteDatabase implements SqliteDatabase { ); } - transaction(callback: () => Promise): Promise { - if (this.transactionContext.getStore()?.active === true) { + transaction(callback: (transaction: SqliteExecutor) => Promise): Promise { + if (this.insideTransaction()) { return Promise.reject(new Error("Nested SQLite transactions are not supported")); } return this.access.run(async () => { this.database.exec("BEGIN IMMEDIATE"); - const context = { active: true }; + const scope = { active: true }; try { - const result = await this.transactionContext.run(context, callback); - context.active = false; + const result = await this.transactionScope.run(scope, () => + callback(new NodeSqliteTransaction(this.database, scope)), + ); + scope.active = false; this.database.exec("COMMIT"); return result; } catch (error) { - context.active = false; + scope.active = false; try { this.database.exec("ROLLBACK"); } catch (rollbackError) { @@ -109,7 +140,7 @@ export class NodeSqliteDatabase implements SqliteDatabase { } close(): Promise { - if (this.transactionContext.getStore()?.active === true) { + if (this.insideTransaction()) { return Promise.reject(new Error("Cannot close SQLite during an active transaction")); } return this.access.run(() => { @@ -123,10 +154,16 @@ export class NodeSqliteDatabase implements SqliteDatabase { }); } - private async runOperation(operation: () => T | Promise): Promise { - if (this.transactionContext.getStore()?.active === true) return operation(); + private runOperation(operation: () => T | Promise): Promise { + if (this.insideTransaction()) { + return Promise.reject(new Error("Use the transaction handle inside a transaction callback")); + } return this.access.run(operation); } + + private insideTransaction(): boolean { + return this.transactionScope.getStore()?.active === true; + } } /** Open and configure a Node-backed SQLite database facade. */ diff --git a/packages/durable/src/storage/sqlite/storage.ts b/packages/durable/src/storage/sqlite/storage.ts index 8402582d2..c75b7f161 100644 --- a/packages/durable/src/storage/sqlite/storage.ts +++ b/packages/durable/src/storage/sqlite/storage.ts @@ -31,7 +31,7 @@ import type { TaskQuery, TaskRecord, } from "../../types.ts"; -import type { SqliteDatabase, SqliteStatement, SqliteValue } from "./database.ts"; +import type { SqliteDatabase, SqliteExecutor, SqliteStatement, SqliteValue } from "./database.ts"; import { applySqliteMigrations } from "./migrations.ts"; type StoredTask = TaskRecord; @@ -137,44 +137,42 @@ const writeId = (write: StorageWrite): Id | undefined => { } }; -class StatementCachingDatabase implements SqliteDatabase { - private readonly database: SqliteDatabase; +/** + * Reuses prepared statements by SQL text. Stores only resolved statements, so a pending + * prepare is never shared between callers. + */ +class StatementCache implements SqliteExecutor { + private readonly executor: SqliteExecutor; private readonly statements = new Map(); - constructor(database: SqliteDatabase) { - this.database = database; + constructor(executor: SqliteExecutor) { + this.executor = executor; } exec(sql: string): Promise { - return this.database.exec(sql); + return this.executor.exec(sql); } async prepare(sql: string): Promise { const cached = this.statements.get(sql); if (cached !== undefined) return cached; - const statement = await this.database.prepare(sql); + const statement = await this.executor.prepare(sql); this.statements.set(sql, statement); return statement; } - - transaction(callback: () => Promise): Promise { - return this.database.transaction(callback); - } - - close(): Promise { - this.statements.clear(); - return this.database.close(); - } } /** Portable SQLite implementation of the Pico storage contract. */ export class SqliteStorage implements Storage { private readonly db: SqliteDatabase; + /** Statements for reads outside transactions. Transaction statements are cached per transaction. */ + private readonly reads: StatementCache; private nextId: number; private closed = false; private constructor(db: SqliteDatabase, nextId: number) { - this.db = new StatementCachingDatabase(db); + this.db = db; + this.reads = new StatementCache(db); this.nextId = nextId; } @@ -201,18 +199,19 @@ export class SqliteStorage implements Storage { this.assertOpen(); const documentActions = this.prepareDocumentActions(writes); const candidateNextId = this.candidateNextId(writes); - const seq = await this.db.transaction(async () => { + const seq = await this.db.transaction(async (transaction) => { + const executor = new StatementCache(transaction); const metadata = await getRow( - this.db.prepare("SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1"), + executor.prepare("SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1"), ); if (metadata === undefined) throw new Error("Durable SQLite metadata is missing"); const committedSeq = seqFromNumber(metadata.next_seq); - await this.checkGlobalIds(writes); - await this.checkDocumentActions(documentActions); - for (const write of writes) await this.applyTableWrite(write, committedSeq); - await this.applyDocumentActions(documentActions, committedSeq); + await this.checkGlobalIds(executor, writes); + await this.checkDocumentActions(executor, documentActions); + for (const write of writes) await this.applyTableWrite(executor, write, committedSeq); + await this.applyDocumentActions(executor, documentActions, committedSeq); await runStatement( - this.db.prepare("UPDATE durable_metadata SET next_id = ?, next_seq = ? WHERE singleton = 1"), + executor.prepare("UPDATE durable_metadata SET next_id = ?, next_seq = ? WHERE singleton = 1"), String(Math.max(Number(metadata.next_id), candidateNextId)), committedSeq + 1, ); @@ -230,7 +229,7 @@ export class SqliteStorage implements Storage { async conversation(id: ConversationId, _context: Context): Promise { this.assertOpen(); - const row = await getRow(this.db.prepare("SELECT record FROM conversations WHERE id = ?"), id); + const row = await getRow(this.reads.prepare("SELECT record FROM conversations WHERE id = ?"), id); return row === undefined ? undefined : parseJson(row.record); } @@ -253,7 +252,7 @@ export class SqliteStorage implements Storage { } params.push(limit + 1); const rows = await allRows( - this.db.prepare(`SELECT record FROM conversations WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`), + this.reads.prepare(`SELECT record FROM conversations WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`), ...params, ); return page( @@ -288,7 +287,7 @@ export class SqliteStorage implements Storage { if (conversation === undefined) throw new Error(`Unknown conversation: ${conversationId}`); } const row = await getRow( - this.db.prepare("SELECT record, commit_seq FROM entries WHERE id = ?"), + this.reads.prepare("SELECT record, commit_seq FROM entries WHERE id = ?"), id, ); if (row === undefined) return undefined; @@ -318,13 +317,13 @@ export class SqliteStorage implements Storage { const row = upper === undefined ? await getRow( - this.db.prepare( + this.reads.prepare( "SELECT record FROM entries WHERE conversation_id = ? AND head IS NOT NULL ORDER BY id DESC LIMIT 1", ), conversation.id, ) : await getRow( - this.db.prepare( + this.reads.prepare( "SELECT record FROM entries WHERE conversation_id = ? AND head IS NOT NULL AND id <= ? ORDER BY id DESC LIMIT 1", ), conversation.id, @@ -363,7 +362,7 @@ export class SqliteStorage implements Storage { } params.push(limit + 1 - values.length); const rows = await allRows( - this.db.prepare(`SELECT record FROM entries WHERE ${clauses.join(" AND ")} ORDER BY id DESC LIMIT ?`), + this.reads.prepare(`SELECT record FROM entries WHERE ${clauses.join(" AND ")} ORDER BY id DESC LIMIT ?`), ...params, ); values.push(...rows.map((row) => parseJson(row.record))); @@ -377,7 +376,7 @@ export class SqliteStorage implements Storage { async task(id: TaskId, _context: Context): Promise { this.assertOpen(); - const row = await getRow(this.db.prepare("SELECT record FROM tasks WHERE id = ?"), id); + const row = await getRow(this.reads.prepare("SELECT record FROM tasks WHERE id = ?"), id); return row === undefined ? undefined : parseJson(row.record); } @@ -412,7 +411,7 @@ export class SqliteStorage implements Storage { } params.push(limit + 1); const rows = await allRows( - this.db.prepare(`SELECT record FROM tasks WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`), + this.reads.prepare(`SELECT record FROM tasks WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`), ...params, ); return page( @@ -423,7 +422,7 @@ export class SqliteStorage implements Storage { async submission(id: SubmissionId, _context: Context): Promise { this.assertOpen(); - const row = await getRow(this.db.prepare("SELECT record FROM submissions WHERE id = ?"), id); + const row = await getRow(this.reads.prepare("SELECT record FROM submissions WHERE id = ?"), id); return row === undefined ? undefined : parseJson(row.record); } @@ -446,7 +445,7 @@ export class SqliteStorage implements Storage { } params.push(limit + 1); const rows = await allRows( - this.db.prepare(`SELECT record FROM submissions WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`), + this.reads.prepare(`SELECT record FROM submissions WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`), ...params, ); return page( @@ -462,7 +461,7 @@ export class SqliteStorage implements Storage { ): Promise { this.assertOpen(); const row = await getRow( - this.db.prepare("SELECT record FROM submissions WHERE conversation_id = ? AND request_id = ?"), + this.reads.prepare("SELECT record FROM submissions WHERE conversation_id = ? AND request_id = ?"), conversationId, encodeIndexedString(requestId), ); @@ -478,10 +477,10 @@ export class SqliteStorage implements Storage { const parts = addressParts(address); const statement = at === "current" - ? this.db.prepare(`SELECT record FROM documents + ? this.reads.prepare(`SELECT record FROM documents WHERE kind = ? AND scope_kind = ? AND owner_id = ? AND family = ? AND key_value = ? AND retired_at IS NULL ORDER BY created_at DESC LIMIT 1`) - : this.db.prepare(`SELECT record FROM documents + : this.reads.prepare(`SELECT record FROM documents WHERE kind = ? AND scope_kind = ? AND owner_id = ? AND family = ? AND key_value = ? AND created_at <= ? AND (retired_at IS NULL OR retired_at > ?) ORDER BY created_at DESC LIMIT 1`); @@ -493,7 +492,7 @@ export class SqliteStorage implements Storage { async document(id: DocumentId, at: DocumentPoint, _context: Context): Promise { this.assertOpen(); - return this.materializeDocument(id, at); + return this.materializeDocument(this.reads, id, at); } async scanDocuments( @@ -518,7 +517,7 @@ export class SqliteStorage implements Storage { } params.push(limit + 1); const rows = await allRows( - this.db.prepare(`SELECT record FROM documents WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`), + this.reads.prepare(`SELECT record FROM documents WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`), ...params, ); return page( @@ -534,12 +533,16 @@ export class SqliteStorage implements Storage { } private async readConversation(id: ConversationId): Promise { - const row = await getRow(this.db.prepare("SELECT record FROM conversations WHERE id = ?"), id); + const row = await getRow(this.reads.prepare("SELECT record FROM conversations WHERE id = ?"), id); return row === undefined ? undefined : parseJson(row.record); } - private async materializeDocument(id: DocumentId, at: DocumentPoint): Promise { - const row = await getRow(this.db.prepare("SELECT record FROM documents WHERE id = ?"), id); + private async materializeDocument( + executor: SqliteExecutor, + id: DocumentId, + at: DocumentPoint, + ): Promise { + const row = await getRow(executor.prepare("SELECT record FROM documents WHERE id = ?"), id); if (row === undefined) return undefined; const record = parseJson(row.record); if (at !== "current" && isCurrentOnly(record)) { @@ -548,7 +551,7 @@ export class SqliteStorage implements Storage { if (!isAliveAt(record, at)) return undefined; const upper = at === "current" ? Number.MAX_SAFE_INTEGER : at; const base = await getRow( - this.db.prepare(`SELECT seq, kind, version, content FROM document_revisions + executor.prepare(`SELECT seq, kind, version, content FROM document_revisions WHERE document_id = ? AND kind = 'base' AND seq <= ? ORDER BY seq DESC LIMIT 1`), id, upper, @@ -556,7 +559,7 @@ export class SqliteStorage implements Storage { if (base === undefined) throw new Error(`Document ${id} is missing a required base`); let value = parseJson(base.content); const tail = await allRows( - this.db.prepare(`SELECT seq, kind, version, content FROM document_revisions + executor.prepare(`SELECT seq, kind, version, content FROM document_revisions WHERE document_id = ? AND seq > ? AND seq <= ? ORDER BY seq`), id, base.seq, @@ -580,9 +583,9 @@ export class SqliteStorage implements Storage { return nextId; } - private async checkGlobalIds(writes: readonly StorageWrite[]): Promise { + private async checkGlobalIds(executor: SqliteExecutor, writes: readonly StorageWrite[]): Promise { const claimed = new Map, TableName>(); - const lookup = this.db.prepare("SELECT record_type FROM record_ids WHERE id = ?"); + const lookup = executor.prepare("SELECT record_type FROM record_ids WHERE id = ?"); for (const write of writes) { if (write.type === "document.change" || write.type === "document.retire") continue; const document = write.type === "document.create" || write.type === "document.copy"; @@ -649,20 +652,23 @@ export class SqliteStorage implements Storage { return actions; } - private async checkDocumentActions(actions: ReadonlyMap): Promise { + private async checkDocumentActions( + executor: SqliteExecutor, + actions: ReadonlyMap, + ): Promise { const liveCounts = new Map(); for (const [id, action] of actions) { if (action.copy !== undefined && actions.has(action.copy.id)) { throw new StorageRejected(`Document copy ${id} source is changed in the copy batch`); } - const row = await getRow(this.db.prepare("SELECT record FROM documents WHERE id = ?"), id); + const row = await getRow(executor.prepare("SELECT record FROM documents WHERE id = ?"), id); const existing = row === undefined ? undefined : parseJson(row.record); if (action.create === undefined && existing === undefined) throw new Error(`Unknown document: ${id}`); if (action.create !== undefined && existing !== undefined) throw new Error(`Document ${id} already exists`); if (existing?.retiredAt !== undefined) throw new Error(`Document ${id} is retired`); if (action.content?.kind === "delta") { const previous = await getRow<{ readonly version: number }>( - this.db.prepare( + executor.prepare( "SELECT version FROM document_revisions WHERE document_id = ? ORDER BY seq DESC LIMIT 1", ), id, @@ -675,7 +681,7 @@ export class SqliteStorage implements Storage { const record = action.create ?? existing!; const key = addressKey(record); let live = liveCounts.get(key); - if (live === undefined) live = (await this.currentDocumentId(record)) === undefined ? 0 : 1; + if (live === undefined) live = (await this.currentDocumentId(executor, record)) === undefined ? 0 : 1; if (action.retire && existing !== undefined) live--; if (action.create !== undefined && !action.retire) live++; liveCounts.set(key, live); @@ -686,12 +692,13 @@ export class SqliteStorage implements Storage { } private async currentDocumentId( + executor: SqliteExecutor, address: DocumentAddress | DocumentCreate | DocumentRecord, ): Promise { const parts = addressParts(address); const id = ( await getRow( - this.db.prepare(`SELECT id FROM documents + executor.prepare(`SELECT id FROM documents WHERE kind = ? AND scope_kind = ? AND owner_id = ? AND family = ? AND key_value = ? AND retired_at IS NULL LIMIT 1`), parts.kind, @@ -704,12 +711,12 @@ export class SqliteStorage implements Storage { return id === undefined ? undefined : idFromNumber(id); } - private async applyTableWrite(write: StorageWrite, seq: Seq): Promise { + private async applyTableWrite(executor: SqliteExecutor, write: StorageWrite, seq: Seq): Promise { switch (write.type) { case "conversation": - await this.claimId(write.value.id, "conversation"); + await this.claimId(executor, write.value.id, "conversation"); await runStatement( - this.db.prepare( + executor.prepare( "INSERT INTO conversations (id, owner_conversation_id, owner_task_id, record) VALUES (?, ?, ?, ?)", ), write.value.id, @@ -719,9 +726,9 @@ export class SqliteStorage implements Storage { ); break; case "entry": - await this.claimId(write.value.id, "entry"); + await this.claimId(executor, write.value.id, "entry"); await runStatement( - this.db.prepare( + executor.prepare( "INSERT INTO entries (id, conversation_id, head, commit_seq, record) VALUES (?, ?, ?, ?, ?)", ), write.value.id, @@ -732,9 +739,9 @@ export class SqliteStorage implements Storage { ); break; case "task": - await this.claimId(write.value.id, "task"); + await this.claimId(executor, write.value.id, "task"); await runStatement( - this.db.prepare(`INSERT INTO tasks (id, conversation_id, kind, status, abort_requested, background, record) + executor.prepare(`INSERT INTO tasks (id, conversation_id, kind, status, abort_requested, background, record) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET conversation_id = excluded.conversation_id, kind = excluded.kind, status = excluded.status, abort_requested = excluded.abort_requested, @@ -749,9 +756,9 @@ export class SqliteStorage implements Storage { ); break; case "submission": - await this.claimId(write.value.id, "submission"); + await this.claimId(executor, write.value.id, "submission"); await runStatement( - this.db.prepare(`INSERT INTO submissions (id, conversation_id, request_id, status, record) VALUES (?, ?, ?, ?, ?) + executor.prepare(`INSERT INTO submissions (id, conversation_id, request_id, status, record) VALUES (?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET conversation_id = excluded.conversation_id, request_id = excluded.request_id, status = excluded.status, record = excluded.record`), write.value.id, @@ -769,20 +776,24 @@ export class SqliteStorage implements Storage { } } - private async claimId(id: Id, table: TableName): Promise { + private async claimId(executor: SqliteExecutor, id: Id, table: TableName): Promise { await runStatement( - this.db.prepare("INSERT OR IGNORE INTO record_ids (id, record_type) VALUES (?, ?)"), + executor.prepare("INSERT OR IGNORE INTO record_ids (id, record_type) VALUES (?, ?)"), id, table, ); } - private async applyDocumentActions(actions: ReadonlyMap, seq: Seq): Promise { + private async applyDocumentActions( + executor: SqliteExecutor, + actions: ReadonlyMap, + seq: Seq, + ): Promise { for (const [id, action] of actions) { let content = action.content; if (action.copy !== undefined) { try { - const stored = await this.materializeDocument(action.copy.id, action.copy.at); + const stored = await this.materializeDocument(executor, action.copy.id, action.copy.at); if (stored === undefined) throw new Error(`Fork source document ${action.copy.id} cannot be read`); const create = action.create!; if ( @@ -809,9 +820,9 @@ export class SqliteStorage implements Storage { ...(action.retire ? { retiredAt: seq } : {}), }; const parts = addressParts(record); - await this.claimId(id, "document"); + await this.claimId(executor, id, "document"); await runStatement( - this.db.prepare(`INSERT INTO documents + executor.prepare(`INSERT INTO documents (id, kind, family, key_value, scope_kind, owner_id, created_at, retired_at, record) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`), id, @@ -825,17 +836,17 @@ export class SqliteStorage implements Storage { encodeJson(record), ); } else { - const row = (await getRow(this.db.prepare("SELECT record FROM documents WHERE id = ?"), id))!; + const row = (await getRow(executor.prepare("SELECT record FROM documents WHERE id = ?"), id))!; record = parseJson(row.record); } if (content !== undefined) { if (content.kind === "base" && isCurrentOnly(record)) { - await runStatement(this.db.prepare("DELETE FROM document_revisions WHERE document_id = ?"), id); + await runStatement(executor.prepare("DELETE FROM document_revisions WHERE document_id = ?"), id); } const encodedContent = content.kind === "base" ? encodeJson(content.value) : encodeJson(content.ops); await runStatement( - this.db.prepare( + executor.prepare( "INSERT INTO document_revisions (document_id, seq, kind, version, content) VALUES (?, ?, ?, ?, ?)", ), id, @@ -850,14 +861,14 @@ export class SqliteStorage implements Storage { if (action.create === undefined) { record = { ...record, retiredAt: seq }; await runStatement( - this.db.prepare("UPDATE documents SET retired_at = ?, record = ? WHERE id = ?"), + executor.prepare("UPDATE documents SET retired_at = ?, record = ? WHERE id = ?"), seq, encodeJson(record), id, ); } if (isCurrentOnly(record)) { - await runStatement(this.db.prepare("DELETE FROM document_revisions WHERE document_id = ?"), id); + await runStatement(executor.prepare("DELETE FROM document_revisions WHERE document_id = ?"), id); } } } diff --git a/packages/durable/test/sqlite-facade.test.ts b/packages/durable/test/sqlite-facade.test.ts index 90129bde4..fa69595ef 100644 --- a/packages/durable/test/sqlite-facade.test.ts +++ b/packages/durable/test/sqlite-facade.test.ts @@ -1,13 +1,10 @@ -import { mkdtemp, rm } from "node:fs/promises"; -import { tmpdir } from "node:os"; -import { join } from "node:path"; import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; import { describe, expect, it } from "vitest"; import { StorageRejected } from "../src/errors.ts"; import { idFromNumber } from "../src/ids.ts"; -import type { SqliteDatabase, SqliteStatement } from "../src/storage/sqlite/index.ts"; +import type { SqliteDatabase, SqliteExecutor, SqliteStatement } from "../src/storage/sqlite/index.ts"; import { SqliteStorage } from "../src/storage/sqlite/index.ts"; -import { type NodeSqliteDatabase, openNodeSqliteDatabase, openNodeSqliteStorage } from "../src/storage/sqlite/node.ts"; +import { type NodeSqliteDatabase, openNodeSqliteDatabase } from "../src/storage/sqlite/node.ts"; import { type EntryId, ROOT_CONVERSATION_ID } from "../src/types.ts"; type SettlementMode = "immediate" | "delay" | "reject"; @@ -17,6 +14,7 @@ class ControlledSettlementDatabase implements SqliteDatabase { private mode: SettlementMode = "immediate"; private pendingSettlement: (() => void) | undefined; private readonly prepareCounts = new Map(); + private readonly transactionPrepareCounts = new Map(); constructor(delegate: NodeSqliteDatabase) { this.delegate = delegate; @@ -31,12 +29,20 @@ class ControlledSettlementDatabase implements SqliteDatabase { return this.delegate.prepare(sql); } - transaction(callback: () => Promise): Promise { + transaction(callback: (transaction: SqliteExecutor) => Promise): Promise { const mode = this.mode; this.mode = "immediate"; - if (mode === "immediate") return this.delegate.transaction(callback); - const settlement = this.delegate.transaction(async () => { - const value = await callback(); + const counted = (transaction: SqliteExecutor): Promise => + callback({ + exec: (sql) => transaction.exec(sql), + prepare: (sql) => { + this.transactionPrepareCounts.set(sql, (this.transactionPrepareCounts.get(sql) ?? 0) + 1); + return transaction.prepare(sql); + }, + }); + if (mode === "immediate") return this.delegate.transaction(counted); + const settlement = this.delegate.transaction(async (transaction) => { + const value = await counted(transaction); if (mode === "reject") throw new Error("controlled settlement rejection"); return value; }); @@ -53,6 +59,10 @@ class ControlledSettlementDatabase implements SqliteDatabase { return this.prepareCounts.get(sql) ?? 0; } + transactionPrepareCount(sql: string): number { + return this.transactionPrepareCounts.get(sql) ?? 0; + } + controlNextSettlement(mode: Exclude): void { if (this.pendingSettlement !== undefined) throw new Error("A settlement is already pending"); this.mode = mode; @@ -67,7 +77,7 @@ class ControlledSettlementDatabase implements SqliteDatabase { } describe("portable SQLite facade settlement", () => { - it("prepares each storage statement once and rebinds it across commits", async () => { + it("reuses read statements across calls and write statements within a transaction", async () => { const database = new ControlledSettlementDatabase(await openNodeSqliteDatabase(":memory:")); const storage = await SqliteStorage.open(database); await storage.commit([{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], BACKGROUND_CONTEXT); @@ -94,20 +104,20 @@ describe("portable SQLite facade settlement", () => { ).rejects.toThrow("ID 1 already belongs to conversation"); expect((await storage.entry(idFromNumber(2), BACKGROUND_CONTEXT))?.entry.kind).toBe("cached"); expect(database.prepareCount("SELECT record, commit_seq FROM entries WHERE id = ?")).toBe(1); - expect(database.prepareCount("INSERT OR IGNORE INTO record_ids (id, record_type) VALUES (?, ?)")).toBe(1); + // Two committing transactions insert 101 entries; each prepares the insert once. expect( - database.prepareCount( + database.transactionPrepareCount( "INSERT INTO entries (id, conversation_id, head, commit_seq, record) VALUES (?, ?, ?, ?, ?)", ), - ).toBe(1); + ).toBe(2); await storage.close(BACKGROUND_CONTEXT); }); - it("supports asynchronous Node transaction callbacks and closes idempotently", async () => { + it("commits work done through the transaction handle and closes idempotently", async () => { const database = await openNodeSqliteDatabase(":memory:"); - await database.transaction(async () => { - await database.exec("CREATE TABLE async_probe (value INTEGER)"); - await (await database.prepare("INSERT INTO async_probe (value) VALUES (?)")).run(1); + await database.transaction(async (transaction) => { + await transaction.exec("CREATE TABLE async_probe (value INTEGER)"); + await (await transaction.prepare("INSERT INTO async_probe (value) VALUES (?)")).run(1); }); expect(await (await database.prepare("SELECT value FROM async_probe")).get()).toEqual({ value: 1 }); await database.close(); @@ -125,17 +135,17 @@ describe("portable SQLite facade settlement", () => { const firstGate = new Promise((resolve) => { releaseFirst = resolve; }); - const first = database.transaction(async () => { - await database.exec("INSERT INTO transaction_queue (value) VALUES (1)"); + 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 () => { + const second = database.transaction(async (transaction) => { secondStarted = true; - await database.exec("INSERT INTO transaction_queue (value) VALUES (2)"); + await transaction.exec("INSERT INTO transaction_queue (value) VALUES (2)"); }); await Promise.resolve(); expect(secondStarted).toBe(false); @@ -161,8 +171,8 @@ describe("portable SQLite facade settlement", () => { const transactionGate = new Promise((resolve) => { releaseTransaction = resolve; }); - const transaction = database.transaction(async () => { - await database.exec("INSERT INTO operation_queue (value) VALUES (1)"); + const pending = database.transaction(async (transaction) => { + await transaction.exec("INSERT INTO operation_queue (value) VALUES (1)"); markTransactionStarted(); await transactionGate; }); @@ -181,60 +191,52 @@ describe("portable SQLite facade settlement", () => { expect(readSettled).toBe(false); releaseTransaction(); - await transaction; + await pending; await write; await expect(read).resolves.toEqual([{ value: 1 }, { value: 2 }]); await database.close(); }); - it("does not share an in-flight prepare across transaction contexts after reopening", async () => { - const directory = await mkdtemp(join(tmpdir(), "pi-durable-sqlite-cache-")); - const path = join(directory, "storage.sqlite"); - try { - const initial = await openNodeSqliteStorage(path); - await initial.close(BACKGROUND_CONTEXT); - - const database = await openNodeSqliteDatabase(path); - const storage = await SqliteStorage.open(database); - let markTransactionStarted!: () => void; - const transactionStarted = new Promise((resolve) => { - markTransactionStarted = resolve; - }); - let releaseTransaction!: () => void; - const transactionGate = new Promise((resolve) => { - releaseTransaction = resolve; - }); - const transaction = database.transaction(async () => { - markTransactionStarted(); - await transactionGate; - return storage.conversation(ROOT_CONVERSATION_ID, BACKGROUND_CONTEXT); - }); - await transactionStarted; - - const concurrentRead = storage.conversation(ROOT_CONVERSATION_ID, BACKGROUND_CONTEXT); - releaseTransaction(); - await expect(Promise.all([transaction, concurrentRead])).resolves.toEqual([undefined, undefined]); - await storage.close(BACKGROUND_CONTEXT); - } finally { - await rm(directory, { recursive: true, force: true }); - } - }); - - it("rejects close from an active transaction", async () => { + it("rejects database operations from inside a transaction callback instead of waiting forever", async () => { const database = await openNodeSqliteDatabase(":memory:"); - await expect(database.transaction(async () => database.close())).rejects.toThrow( + await database.exec("CREATE TABLE misuse_probe (value INTEGER)"); + const statement = await database.prepare("SELECT value FROM misuse_probe"); + const misuse = "Use the transaction handle inside a transaction callback"; + await expect(database.transaction(() => database.exec("SELECT 1"))).rejects.toThrow(misuse); + await expect(database.transaction(() => database.prepare("SELECT 1"))).rejects.toThrow(misuse); + await expect(database.transaction(() => statement.all())).rejects.toThrow(misuse); + await expect(database.transaction(() => database.transaction(async () => undefined))).rejects.toThrow( + "Nested SQLite transactions are not supported", + ); + await expect(database.transaction(() => database.close())).rejects.toThrow( "Cannot close SQLite during an active transaction", ); 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; + let statement!: SqliteStatement; + await database.transaction(async (transaction) => { + handle = transaction; + statement = await transaction.prepare("INSERT INTO stale_probe (value) VALUES (?)"); + }); + const stale = "SQLite transaction handle is no longer active"; + await expect(handle.exec("INSERT INTO stale_probe (value) VALUES (1)")).rejects.toThrow(stale); + await expect(statement.run(1)).rejects.toThrow(stale); + expect(await (await database.prepare("SELECT value FROM stale_probe")).all()).toEqual([]); + await database.close(); + }); + it("does not preserve a guaranteed rejection when rollback itself fails", async () => { const database = await openNodeSqliteDatabase(":memory:"); await database.exec("CREATE TABLE rollback_probe (value INTEGER)"); await expect( - database.transaction(async () => { - await database.exec("INSERT INTO rollback_probe (value) VALUES (1)"); - await database.exec("COMMIT"); + database.transaction(async (transaction) => { + await transaction.exec("INSERT INTO rollback_probe (value) VALUES (1)"); + await transaction.exec("COMMIT"); throw new StorageRejected("rejected after an escaped commit"); }), ).rejects.toThrow(AggregateError); From cd0ba2fabcebd1c234d1009595cebf37a1f6c5e6 Mon Sep 17 00:00:00 2001 From: Christian Klotz Date: Wed, 30 Sep 2026 19:49:55 +0100 Subject: [PATCH 3/4] feat(durable): execute SQLite queries by SQL text and cache statements per connection Replace prepare/SqliteStatement with run/get/all(sql, ...params) on the database and transaction handles. The Node adapter caches prepared statements per connection, so transactions reuse them instead of preparing again on every commit. Read documents in one transaction so a concurrent commit cannot replace the base between the record and revision queries. --- packages/durable/README.md | 5 +- .../durable/src/storage/sqlite/database.ts | 25 ++- packages/durable/src/storage/sqlite/index.ts | 2 +- .../durable/src/storage/sqlite/migrations.ts | 12 +- packages/durable/src/storage/sqlite/node.ts | 101 ++++----- .../durable/src/storage/sqlite/storage.ts | 210 +++++++----------- packages/durable/test/sqlite-facade.test.ts | 120 +++++----- .../durable/test/sqlite-migrations.test.ts | 51 ++--- 8 files changed, 233 insertions(+), 293 deletions(-) diff --git a/packages/durable/README.md b/packages/durable/README.md index 9c43ca9a4..2d33a44d0 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -362,13 +362,12 @@ const usage = await harness.usage(context); // { models: { "openai/gpt-6-sol": U One process owns a storage at a time; there is no cross-process locking. The portable SQLite and JSONL cores (`/storage/sqlite`, `/storage/jsonl`) run without Node APIs, for example on Bun or in Cloudflare Durable Objects, given an asynchronous `SqliteDatabase` facade or a `FileSystem` from `@earendil-works/pi-durable/env`. -SQLite adapters implement promise-based `exec`, `prepare`, statement `run`/`get`/`all`, transactions, and close. A transaction callback receives a transaction handle; all work in the transaction must use it, and the handle expires when the callback settles. Adapters must queue unrelated operations and other transactions until the transaction finishes: +SQLite adapters implement promise-based `exec`, `run`, `get`, `all`, `transaction`, and `close`. `run`, `get`, and `all` take SQL text plus positional bindings; adapters may cache prepared statements by SQL text. A transaction callback receives a transaction handle; all work in the transaction must use it, and the handle expires when the callback settles. Adapters must queue unrelated operations and other transactions until the transaction finishes: ```typescript await database.transaction(async (transaction) => { await transaction.exec("CREATE TABLE example (value TEXT)"); - const insert = await transaction.prepare("INSERT INTO example (value) VALUES (?)"); - await insert.run("stored atomically"); + await transaction.run("INSERT INTO example (value) VALUES (?)", "stored atomically"); }); ``` diff --git a/packages/durable/src/storage/sqlite/database.ts b/packages/durable/src/storage/sqlite/database.ts index 0c55841dd..35a0f39b6 100644 --- a/packages/durable/src/storage/sqlite/database.ts +++ b/packages/durable/src/storage/sqlite/database.ts @@ -1,17 +1,18 @@ /** Values supported by the portable SQLite storage core. */ export type SqliteValue = null | number | bigint | string | Uint8Array; -/** A prepared SQLite statement that supports repeated execution with new bindings. */ -export interface SqliteStatement { - run(...params: SqliteValue[]): Promise; - get(...params: SqliteValue[]): Promise; - all(...params: SqliteValue[]): Promise; -} - -/** Asynchronous SQL operations shared by a database and its transaction handles. */ +/** + * 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; - prepare(sql: string): Promise; + run(sql: string, ...params: SqliteValue[]): Promise; + get(sql: string, ...params: SqliteValue[]): Promise; + all(sql: string, ...params: SqliteValue[]): Promise; } /** @@ -20,9 +21,9 @@ export interface SqliteExecutor { * 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 and statements prepared through it are 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. + * 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. * * 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 diff --git a/packages/durable/src/storage/sqlite/index.ts b/packages/durable/src/storage/sqlite/index.ts index 86805bc23..666541868 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, SqliteStatement, SqliteValue } from "./database.ts"; +export type { SqliteDatabase, SqliteExecutor, SqliteValue } from "./database.ts"; 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 d5e124c96..4b1287e43 100644 --- a/packages/durable/src/storage/sqlite/migrations.ts +++ b/packages/durable/src/storage/sqlite/migrations.ts @@ -107,12 +107,8 @@ export async function applySqliteMigrations( singleton INTEGER PRIMARY KEY CHECK (singleton = 1), version INTEGER NOT NULL CHECK (version >= 0) ) STRICT`); - await ( - await transaction.prepare("INSERT OR IGNORE INTO durable_schema (singleton, version) VALUES (1, 0)") - ).run(); - const row = await ( - await transaction.prepare("SELECT version FROM durable_schema WHERE singleton = 1") - ).get(); + await transaction.run("INSERT OR IGNORE INTO durable_schema (singleton, version) VALUES (1, 0)"); + const row = await transaction.get("SELECT version FROM durable_schema 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) { @@ -123,9 +119,7 @@ export async function applySqliteMigrations( for (const migration of migrations) { if (migration.version <= row.version) continue; for (const statement of migration.statements) await transaction.exec(statement); - await (await transaction.prepare("UPDATE durable_schema SET version = ? WHERE singleton = 1")).run( - migration.version, - ); + await transaction.run("UPDATE durable_schema SET version = ? WHERE singleton = 1", migration.version); } }); } diff --git a/packages/durable/src/storage/sqlite/node.ts b/packages/durable/src/storage/sqlite/node.ts index 7c3d1aef6..558c1dcfc 100644 --- a/packages/durable/src/storage/sqlite/node.ts +++ b/packages/durable/src/storage/sqlite/node.ts @@ -3,7 +3,7 @@ import { mkdir } from "node:fs/promises"; import { dirname } from "node:path"; import type { SQLInputValue, StatementSync } from "node:sqlite"; import { DatabaseSync } from "node:sqlite"; -import type { SqliteDatabase, SqliteExecutor, SqliteStatement, SqliteValue } from "./database.ts"; +import type { SqliteDatabase, SqliteExecutor, SqliteValue } from "./database.ts"; import { SqliteStorage } from "./storage.ts"; /** Node SQLite connection settings for a durable storage file. */ @@ -17,7 +17,6 @@ export type NodeSqliteStorageOptions = { const DEFAULT_WAL_AUTO_CHECKPOINT_PAGES = 1_000; const DEFAULT_BUSY_TIMEOUT_MS = 5_000; -type RunOperation = (operation: () => T | Promise) => Promise; type TransactionScope = { active: boolean }; class SerialOperationQueue { @@ -38,37 +37,17 @@ class SerialOperationQueue { } } -class NodeSqliteStatement implements SqliteStatement { - private readonly statement: StatementSync; - private readonly runOperation: RunOperation; +/** + * 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(statement: StatementSync, runOperation: RunOperation) { - this.statement = statement; - this.runOperation = runOperation; - } - - run(...params: SqliteValue[]): Promise { - return this.runOperation(() => { - this.statement.run(...(params as SQLInputValue[])); - }); - } - - get(...params: SqliteValue[]): Promise { - return this.runOperation(() => this.statement.get(...(params as SQLInputValue[])) as T | undefined); - } - - all(...params: SqliteValue[]): Promise { - return this.runOperation(() => this.statement.all(...(params as SQLInputValue[])) as T[]); - } -} - -class NodeSqliteTransaction implements SqliteExecutor { - private readonly database: DatabaseSync; - private readonly scope: TransactionScope; - - constructor(database: DatabaseSync, scope: TransactionScope) { + constructor(database: DatabaseSync, statements: Map) { this.database = database; - this.scope = scope; + this.statements = statements; } exec(sql: string): Promise { @@ -77,40 +56,55 @@ class NodeSqliteTransaction implements SqliteExecutor { }); } - prepare(sql: string): Promise { - return this.runOperation( - () => new NodeSqliteStatement(this.database.prepare(sql), (operation) => this.runOperation(operation)), - ); + run(sql: string, ...params: SqliteValue[]): Promise { + return this.runOperation(() => { + this.statement(sql).run(...(params as SQLInputValue[])); + }); } - private async runOperation(operation: () => T | Promise): Promise { + get(sql: string, ...params: SqliteValue[]): Promise { + return this.runOperation(() => this.statement(sql).get(...(params as SQLInputValue[])) as T | undefined); + } + + all(sql: string, ...params: SqliteValue[]): Promise { + return this.runOperation(() => this.statement(sql).all(...(params as SQLInputValue[])) 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 implements SqliteDatabase { - private readonly database: DatabaseSync; +export class NodeSqliteDatabase extends NodeSqliteExecutor implements SqliteDatabase { private readonly access = new SerialOperationQueue(); /** Detects database calls from inside a transaction callback, which would otherwise wait for that transaction forever. */ private readonly transactionScope = new AsyncLocalStorage(); private closed = false; constructor(database: DatabaseSync) { - this.database = database; - } - - exec(sql: string): Promise { - return this.runOperation(() => { - this.database.exec(sql); - }); - } - - prepare(sql: string): Promise { - return this.runOperation( - () => new NodeSqliteStatement(this.database.prepare(sql), (operation) => this.runOperation(operation)), - ); + super(database, new Map()); } transaction(callback: (transaction: SqliteExecutor) => Promise): Promise { @@ -122,7 +116,7 @@ export class NodeSqliteDatabase implements SqliteDatabase { const scope = { active: true }; try { const result = await this.transactionScope.run(scope, () => - callback(new NodeSqliteTransaction(this.database, scope)), + callback(new NodeSqliteTransaction(this.database, this.statements, scope)), ); scope.active = false; this.database.exec("COMMIT"); @@ -146,6 +140,7 @@ export class NodeSqliteDatabase implements SqliteDatabase { return this.access.run(() => { if (this.closed) return; this.closed = true; + this.statements.clear(); try { this.database.exec("PRAGMA wal_checkpoint(TRUNCATE)"); } finally { @@ -154,7 +149,7 @@ export class NodeSqliteDatabase implements SqliteDatabase { }); } - private runOperation(operation: () => T | Promise): Promise { + protected runOperation(operation: () => T): Promise { if (this.insideTransaction()) { return Promise.reject(new Error("Use the transaction handle inside a transaction callback")); } diff --git a/packages/durable/src/storage/sqlite/storage.ts b/packages/durable/src/storage/sqlite/storage.ts index c75b7f161..0fe598352 100644 --- a/packages/durable/src/storage/sqlite/storage.ts +++ b/packages/durable/src/storage/sqlite/storage.ts @@ -31,7 +31,7 @@ import type { TaskQuery, TaskRecord, } from "../../types.ts"; -import type { SqliteDatabase, SqliteExecutor, SqliteStatement, SqliteValue } from "./database.ts"; +import type { SqliteDatabase, SqliteExecutor, SqliteValue } from "./database.ts"; import { applySqliteMigrations } from "./migrations.ts"; type StoredTask = TaskRecord; @@ -63,17 +63,6 @@ const encodeJson = (value: unknown): string => JSON.stringify(value) as string; // Some SQLite bindings replace lone UTF-16 surrogates. JSON encoding keeps indexed identities lossless. const encodeIndexedString = (value: string): string => JSON.stringify(value); -const getRow = async ( - statement: Promise, - ...params: SqliteValue[] -): Promise => (await statement).get(...params); - -const allRows = async (statement: Promise, ...params: SqliteValue[]): Promise => - (await statement).all(...params); - -const runStatement = async (statement: Promise, ...params: SqliteValue[]): Promise => - (await statement).run(...params); - const cursorId = >(cursor: Cursor | undefined): I | undefined => { const after = cursor?.after; if (after === undefined) return undefined; @@ -137,42 +126,14 @@ const writeId = (write: StorageWrite): Id | undefined => { } }; -/** - * Reuses prepared statements by SQL text. Stores only resolved statements, so a pending - * prepare is never shared between callers. - */ -class StatementCache implements SqliteExecutor { - private readonly executor: SqliteExecutor; - private readonly statements = new Map(); - - constructor(executor: SqliteExecutor) { - this.executor = executor; - } - - exec(sql: string): Promise { - return this.executor.exec(sql); - } - - async prepare(sql: string): Promise { - const cached = this.statements.get(sql); - if (cached !== undefined) return cached; - const statement = await this.executor.prepare(sql); - this.statements.set(sql, statement); - return statement; - } -} - /** Portable SQLite implementation of the Pico storage contract. */ export class SqliteStorage implements Storage { private readonly db: SqliteDatabase; - /** Statements for reads outside transactions. Transaction statements are cached per transaction. */ - private readonly reads: StatementCache; private nextId: number; private closed = false; private constructor(db: SqliteDatabase, nextId: number) { this.db = db; - this.reads = new StatementCache(db); this.nextId = nextId; } @@ -180,8 +141,8 @@ export class SqliteStorage implements Storage { static async open(db: SqliteDatabase): Promise { try { await applySqliteMigrations(db); - const metadata = await getRow( - db.prepare("SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1"), + const metadata = await db.get( + "SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1", ); if (metadata === undefined) throw new Error("Durable SQLite metadata is missing"); return new SqliteStorage(db, Number(metadata.next_id)); @@ -200,18 +161,17 @@ export class SqliteStorage implements Storage { const documentActions = this.prepareDocumentActions(writes); const candidateNextId = this.candidateNextId(writes); const seq = await this.db.transaction(async (transaction) => { - const executor = new StatementCache(transaction); - const metadata = await getRow( - executor.prepare("SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1"), + const metadata = await transaction.get( + "SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1", ); if (metadata === undefined) throw new Error("Durable SQLite metadata is missing"); const committedSeq = seqFromNumber(metadata.next_seq); - await this.checkGlobalIds(executor, writes); - await this.checkDocumentActions(executor, documentActions); - for (const write of writes) await this.applyTableWrite(executor, write, committedSeq); - await this.applyDocumentActions(executor, documentActions, committedSeq); - await runStatement( - executor.prepare("UPDATE durable_metadata SET next_id = ?, next_seq = ? WHERE singleton = 1"), + await this.checkGlobalIds(transaction, writes); + await this.checkDocumentActions(transaction, documentActions); + for (const write of writes) await this.applyTableWrite(transaction, write, committedSeq); + await this.applyDocumentActions(transaction, documentActions, committedSeq); + await transaction.run( + "UPDATE durable_metadata SET next_id = ?, next_seq = ? WHERE singleton = 1", String(Math.max(Number(metadata.next_id), candidateNextId)), committedSeq + 1, ); @@ -229,7 +189,7 @@ export class SqliteStorage implements Storage { async conversation(id: ConversationId, _context: Context): Promise { this.assertOpen(); - const row = await getRow(this.reads.prepare("SELECT record FROM conversations WHERE id = ?"), id); + const row = await this.db.get("SELECT record FROM conversations WHERE id = ?", id); return row === undefined ? undefined : parseJson(row.record); } @@ -251,8 +211,8 @@ export class SqliteStorage implements Storage { params.push(query.ownerTaskId); } params.push(limit + 1); - const rows = await allRows( - this.reads.prepare(`SELECT record FROM conversations WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`), + const rows = await this.db.all( + `SELECT record FROM conversations WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`, ...params, ); return page( @@ -286,10 +246,7 @@ export class SqliteStorage implements Storage { conversation = await this.readConversation(conversationId); if (conversation === undefined) throw new Error(`Unknown conversation: ${conversationId}`); } - const row = await getRow( - this.reads.prepare("SELECT record, commit_seq FROM entries WHERE id = ?"), - id, - ); + const row = await this.db.get("SELECT record, commit_seq FROM entries WHERE id = ?", id); if (row === undefined) return undefined; const entry = parseJson(row.record); if (conversation !== undefined) { @@ -316,16 +273,12 @@ export class SqliteStorage implements Storage { while (true) { const row = upper === undefined - ? await getRow( - this.reads.prepare( - "SELECT record FROM entries WHERE conversation_id = ? AND head IS NOT NULL ORDER BY id DESC LIMIT 1", - ), + ? await this.db.get( + "SELECT record FROM entries WHERE conversation_id = ? AND head IS NOT NULL ORDER BY id DESC LIMIT 1", conversation.id, ) - : await getRow( - this.reads.prepare( - "SELECT record FROM entries WHERE conversation_id = ? AND head IS NOT NULL AND id <= ? ORDER BY id DESC LIMIT 1", - ), + : await this.db.get( + "SELECT record FROM entries WHERE conversation_id = ? AND head IS NOT NULL AND id <= ? ORDER BY id DESC LIMIT 1", conversation.id, upper, ); @@ -361,8 +314,8 @@ export class SqliteStorage implements Storage { params.push(upper); } params.push(limit + 1 - values.length); - const rows = await allRows( - this.reads.prepare(`SELECT record FROM entries WHERE ${clauses.join(" AND ")} ORDER BY id DESC LIMIT ?`), + const rows = await this.db.all( + `SELECT record FROM entries WHERE ${clauses.join(" AND ")} ORDER BY id DESC LIMIT ?`, ...params, ); values.push(...rows.map((row) => parseJson(row.record))); @@ -376,7 +329,7 @@ export class SqliteStorage implements Storage { async task(id: TaskId, _context: Context): Promise { this.assertOpen(); - const row = await getRow(this.reads.prepare("SELECT record FROM tasks WHERE id = ?"), id); + const row = await this.db.get("SELECT record FROM tasks WHERE id = ?", id); return row === undefined ? undefined : parseJson(row.record); } @@ -410,8 +363,8 @@ export class SqliteStorage implements Storage { params.push(query.background ? 1 : 0); } params.push(limit + 1); - const rows = await allRows( - this.reads.prepare(`SELECT record FROM tasks WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`), + const rows = await this.db.all( + `SELECT record FROM tasks WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`, ...params, ); return page( @@ -422,7 +375,7 @@ export class SqliteStorage implements Storage { async submission(id: SubmissionId, _context: Context): Promise { this.assertOpen(); - const row = await getRow(this.reads.prepare("SELECT record FROM submissions WHERE id = ?"), id); + const row = await this.db.get("SELECT record FROM submissions WHERE id = ?", id); return row === undefined ? undefined : parseJson(row.record); } @@ -444,8 +397,8 @@ export class SqliteStorage implements Storage { params.push(query.status); } params.push(limit + 1); - const rows = await allRows( - this.reads.prepare(`SELECT record FROM submissions WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`), + const rows = await this.db.all( + `SELECT record FROM submissions WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`, ...params, ); return page( @@ -460,8 +413,8 @@ export class SqliteStorage implements Storage { _context: Context, ): Promise { this.assertOpen(); - const row = await getRow( - this.reads.prepare("SELECT record FROM submissions WHERE conversation_id = ? AND request_id = ?"), + const row = await this.db.get( + "SELECT record FROM submissions WHERE conversation_id = ? AND request_id = ?", conversationId, encodeIndexedString(requestId), ); @@ -475,24 +428,25 @@ export class SqliteStorage implements Storage { ): Promise { this.assertOpen(); const parts = addressParts(address); - const statement = + const sql = at === "current" - ? this.reads.prepare(`SELECT record FROM documents + ? `SELECT record FROM documents WHERE kind = ? AND scope_kind = ? AND owner_id = ? AND family = ? AND key_value = ? - AND retired_at IS NULL ORDER BY created_at DESC LIMIT 1`) - : this.reads.prepare(`SELECT record FROM documents + AND retired_at IS NULL ORDER BY created_at DESC LIMIT 1` + : `SELECT record FROM documents WHERE kind = ? AND scope_kind = ? AND owner_id = ? AND family = ? AND key_value = ? AND created_at <= ? AND (retired_at IS NULL OR retired_at > ?) - ORDER BY created_at DESC LIMIT 1`); + ORDER BY created_at DESC LIMIT 1`; const params: SqliteValue[] = [parts.kind, parts.scopeKind, parts.ownerId, parts.family, parts.keyValue]; if (at !== "current") params.push(at, at); - const row = await getRow(statement, ...params); + const row = await this.db.get(sql, ...params); return row === undefined ? undefined : parseJson(row.record); } async document(id: DocumentId, at: DocumentPoint, _context: Context): Promise { this.assertOpen(); - return this.materializeDocument(this.reads, id, at); + // The record and revision queries must observe one committed state; a commit between them can replace the base. + return this.db.transaction((transaction) => this.materializeDocument(transaction, id, at)); } async scanDocuments( @@ -516,8 +470,8 @@ export class SqliteStorage implements Storage { params.push(query.at, query.at); } params.push(limit + 1); - const rows = await allRows( - this.reads.prepare(`SELECT record FROM documents WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`), + const rows = await this.db.all( + `SELECT record FROM documents WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`, ...params, ); return page( @@ -533,7 +487,7 @@ export class SqliteStorage implements Storage { } private async readConversation(id: ConversationId): Promise { - const row = await getRow(this.reads.prepare("SELECT record FROM conversations WHERE id = ?"), id); + const row = await this.db.get("SELECT record FROM conversations WHERE id = ?", id); return row === undefined ? undefined : parseJson(row.record); } @@ -542,7 +496,7 @@ export class SqliteStorage implements Storage { id: DocumentId, at: DocumentPoint, ): Promise { - const row = await getRow(executor.prepare("SELECT record FROM documents WHERE id = ?"), id); + const row = await executor.get("SELECT record FROM documents WHERE id = ?", id); if (row === undefined) return undefined; const record = parseJson(row.record); if (at !== "current" && isCurrentOnly(record)) { @@ -550,17 +504,17 @@ export class SqliteStorage implements Storage { } if (!isAliveAt(record, at)) return undefined; const upper = at === "current" ? Number.MAX_SAFE_INTEGER : at; - const base = await getRow( - executor.prepare(`SELECT seq, kind, version, content FROM document_revisions - WHERE document_id = ? AND kind = 'base' AND seq <= ? ORDER BY seq DESC LIMIT 1`), + const base = await executor.get( + `SELECT seq, kind, version, content FROM document_revisions + WHERE document_id = ? AND kind = 'base' AND seq <= ? ORDER BY seq DESC LIMIT 1`, id, upper, ); if (base === undefined) throw new Error(`Document ${id} is missing a required base`); let value = parseJson(base.content); - const tail = await allRows( - executor.prepare(`SELECT seq, kind, version, content FROM document_revisions - WHERE document_id = ? AND seq > ? AND seq <= ? ORDER BY seq`), + const tail = await executor.all( + `SELECT seq, kind, version, content FROM document_revisions + WHERE document_id = ? AND seq > ? AND seq <= ? ORDER BY seq`, id, base.seq, upper, @@ -585,13 +539,13 @@ export class SqliteStorage implements Storage { private async checkGlobalIds(executor: SqliteExecutor, writes: readonly StorageWrite[]): Promise { const claimed = new Map, TableName>(); - const lookup = executor.prepare("SELECT record_type FROM record_ids WHERE id = ?"); for (const write of writes) { if (write.type === "document.change" || write.type === "document.retire") continue; const document = write.type === "document.create" || write.type === "document.copy"; const table: TableName = document ? "document" : write.type; const id = document ? write.record.id : write.value.id; - const existing = (await getRow(lookup, id))?.record_type; + const existing = (await executor.get("SELECT record_type FROM record_ids WHERE id = ?", id)) + ?.record_type; const earlier = claimed.get(id); if (table === "conversation" || table === "entry" || table === "document") { if (existing !== undefined) throw new Error(`ID ${id} already belongs to ${existing}`); @@ -661,16 +615,14 @@ export class SqliteStorage implements Storage { if (action.copy !== undefined && actions.has(action.copy.id)) { throw new StorageRejected(`Document copy ${id} source is changed in the copy batch`); } - const row = await getRow(executor.prepare("SELECT record FROM documents WHERE id = ?"), id); + const row = await executor.get("SELECT record FROM documents WHERE id = ?", id); const existing = row === undefined ? undefined : parseJson(row.record); if (action.create === undefined && existing === undefined) throw new Error(`Unknown document: ${id}`); if (action.create !== undefined && existing !== undefined) throw new Error(`Document ${id} already exists`); if (existing?.retiredAt !== undefined) throw new Error(`Document ${id} is retired`); if (action.content?.kind === "delta") { - const previous = await getRow<{ readonly version: number }>( - executor.prepare( - "SELECT version FROM document_revisions WHERE document_id = ? ORDER BY seq DESC LIMIT 1", - ), + const previous = await executor.get<{ readonly version: number }>( + "SELECT version FROM document_revisions WHERE document_id = ? ORDER BY seq DESC LIMIT 1", id, ); if (previous === undefined) throw new Error(`Document ${id} delta has no base`); @@ -697,10 +649,10 @@ export class SqliteStorage implements Storage { ): Promise { const parts = addressParts(address); const id = ( - await getRow( - executor.prepare(`SELECT id FROM documents + await executor.get( + `SELECT id FROM documents WHERE kind = ? AND scope_kind = ? AND owner_id = ? AND family = ? AND key_value = ? AND retired_at IS NULL - LIMIT 1`), + LIMIT 1`, parts.kind, parts.scopeKind, parts.ownerId, @@ -715,10 +667,8 @@ export class SqliteStorage implements Storage { switch (write.type) { case "conversation": await this.claimId(executor, write.value.id, "conversation"); - await runStatement( - executor.prepare( - "INSERT INTO conversations (id, owner_conversation_id, owner_task_id, record) VALUES (?, ?, ?, ?)", - ), + await executor.run( + "INSERT INTO conversations (id, owner_conversation_id, owner_task_id, record) VALUES (?, ?, ?, ?)", write.value.id, write.value.owner?.conversationId ?? null, write.value.owner?.taskId ?? null, @@ -727,10 +677,8 @@ export class SqliteStorage implements Storage { break; case "entry": await this.claimId(executor, write.value.id, "entry"); - await runStatement( - executor.prepare( - "INSERT INTO entries (id, conversation_id, head, commit_seq, record) VALUES (?, ?, ?, ?, ?)", - ), + await executor.run( + "INSERT INTO entries (id, conversation_id, head, commit_seq, record) VALUES (?, ?, ?, ?, ?)", write.value.id, write.value.conversationId, write.value.head ?? null, @@ -740,12 +688,12 @@ export class SqliteStorage implements Storage { break; case "task": await this.claimId(executor, write.value.id, "task"); - await runStatement( - executor.prepare(`INSERT INTO tasks (id, conversation_id, kind, status, abort_requested, background, record) + await executor.run( + `INSERT INTO tasks (id, conversation_id, kind, status, abort_requested, background, record) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET conversation_id = excluded.conversation_id, kind = excluded.kind, status = excluded.status, abort_requested = excluded.abort_requested, - background = excluded.background, record = excluded.record`), + background = excluded.background, record = excluded.record`, write.value.id, write.value.conversationId, encodeIndexedString(write.value.kind), @@ -757,10 +705,10 @@ export class SqliteStorage implements Storage { break; case "submission": await this.claimId(executor, write.value.id, "submission"); - await runStatement( - executor.prepare(`INSERT INTO submissions (id, conversation_id, request_id, status, record) VALUES (?, ?, ?, ?, ?) + await executor.run( + `INSERT INTO submissions (id, conversation_id, request_id, status, record) VALUES (?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET conversation_id = excluded.conversation_id, - request_id = excluded.request_id, status = excluded.status, record = excluded.record`), + request_id = excluded.request_id, status = excluded.status, record = excluded.record`, write.value.id, write.value.conversationId, write.value.requestId === undefined ? null : encodeIndexedString(write.value.requestId), @@ -777,11 +725,7 @@ export class SqliteStorage implements Storage { } private async claimId(executor: SqliteExecutor, id: Id, table: TableName): Promise { - await runStatement( - executor.prepare("INSERT OR IGNORE INTO record_ids (id, record_type) VALUES (?, ?)"), - id, - table, - ); + await executor.run("INSERT OR IGNORE INTO record_ids (id, record_type) VALUES (?, ?)", id, table); } private async applyDocumentActions( @@ -821,10 +765,10 @@ export class SqliteStorage implements Storage { }; const parts = addressParts(record); await this.claimId(executor, id, "document"); - await runStatement( - executor.prepare(`INSERT INTO documents + await executor.run( + `INSERT INTO documents (id, kind, family, key_value, scope_kind, owner_id, created_at, retired_at, record) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`), + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, id, parts.kind, parts.family, @@ -836,19 +780,17 @@ export class SqliteStorage implements Storage { encodeJson(record), ); } else { - const row = (await getRow(executor.prepare("SELECT record FROM documents WHERE id = ?"), id))!; + const row = (await executor.get("SELECT record FROM documents WHERE id = ?", id))!; record = parseJson(row.record); } if (content !== undefined) { if (content.kind === "base" && isCurrentOnly(record)) { - await runStatement(executor.prepare("DELETE FROM document_revisions WHERE document_id = ?"), id); + await executor.run("DELETE FROM document_revisions WHERE document_id = ?", id); } const encodedContent = content.kind === "base" ? encodeJson(content.value) : encodeJson(content.ops); - await runStatement( - executor.prepare( - "INSERT INTO document_revisions (document_id, seq, kind, version, content) VALUES (?, ?, ?, ?, ?)", - ), + await executor.run( + "INSERT INTO document_revisions (document_id, seq, kind, version, content) VALUES (?, ?, ?, ?, ?)", id, seq, content.kind, @@ -860,15 +802,15 @@ export class SqliteStorage implements Storage { if (action.retire) { if (action.create === undefined) { record = { ...record, retiredAt: seq }; - await runStatement( - executor.prepare("UPDATE documents SET retired_at = ?, record = ? WHERE id = ?"), + await executor.run( + "UPDATE documents SET retired_at = ?, record = ? WHERE id = ?", seq, encodeJson(record), id, ); } if (isCurrentOnly(record)) { - await runStatement(executor.prepare("DELETE FROM document_revisions WHERE document_id = ?"), id); + await executor.run("DELETE FROM document_revisions WHERE document_id = ?", id); } } } diff --git a/packages/durable/test/sqlite-facade.test.ts b/packages/durable/test/sqlite-facade.test.ts index fa69595ef..b9a311b26 100644 --- a/packages/durable/test/sqlite-facade.test.ts +++ b/packages/durable/test/sqlite-facade.test.ts @@ -1,11 +1,12 @@ +import { DatabaseSync, type StatementSync } from "node:sqlite"; import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; import { describe, expect, it } from "vitest"; import { StorageRejected } from "../src/errors.ts"; import { idFromNumber } from "../src/ids.ts"; -import type { SqliteDatabase, SqliteExecutor, SqliteStatement } from "../src/storage/sqlite/index.ts"; +import type { SqliteDatabase, SqliteExecutor, SqliteValue } from "../src/storage/sqlite/index.ts"; import { SqliteStorage } from "../src/storage/sqlite/index.ts"; -import { type NodeSqliteDatabase, openNodeSqliteDatabase } from "../src/storage/sqlite/node.ts"; -import { type EntryId, ROOT_CONVERSATION_ID } from "../src/types.ts"; +import { NodeSqliteDatabase, openNodeSqliteDatabase } from "../src/storage/sqlite/node.ts"; +import { type DocumentId, type EntryId, ROOT_CONVERSATION_ID } from "../src/types.ts"; type SettlementMode = "immediate" | "delay" | "reject"; @@ -13,8 +14,6 @@ class ControlledSettlementDatabase implements SqliteDatabase { private readonly delegate: NodeSqliteDatabase; private mode: SettlementMode = "immediate"; private pendingSettlement: (() => void) | undefined; - private readonly prepareCounts = new Map(); - private readonly transactionPrepareCounts = new Map(); constructor(delegate: NodeSqliteDatabase) { this.delegate = delegate; @@ -24,25 +23,24 @@ class ControlledSettlementDatabase implements SqliteDatabase { return this.delegate.exec(sql); } - prepare(sql: string): Promise { - this.prepareCounts.set(sql, (this.prepareCounts.get(sql) ?? 0) + 1); - return this.delegate.prepare(sql); + run(sql: string, ...params: SqliteValue[]): Promise { + return this.delegate.run(sql, ...params); + } + + get(sql: string, ...params: SqliteValue[]): Promise { + return this.delegate.get(sql, ...params); + } + + all(sql: string, ...params: SqliteValue[]): Promise { + return this.delegate.all(sql, ...params); } transaction(callback: (transaction: SqliteExecutor) => Promise): Promise { const mode = this.mode; this.mode = "immediate"; - const counted = (transaction: SqliteExecutor): Promise => - callback({ - exec: (sql) => transaction.exec(sql), - prepare: (sql) => { - this.transactionPrepareCounts.set(sql, (this.transactionPrepareCounts.get(sql) ?? 0) + 1); - return transaction.prepare(sql); - }, - }); - if (mode === "immediate") return this.delegate.transaction(counted); + if (mode === "immediate") return this.delegate.transaction(callback); const settlement = this.delegate.transaction(async (transaction) => { - const value = await counted(transaction); + const value = await callback(transaction); if (mode === "reject") throw new Error("controlled settlement rejection"); return value; }); @@ -55,14 +53,6 @@ class ControlledSettlementDatabase implements SqliteDatabase { return this.delegate.close(); } - prepareCount(sql: string): number { - return this.prepareCounts.get(sql) ?? 0; - } - - transactionPrepareCount(sql: string): number { - return this.transactionPrepareCounts.get(sql) ?? 0; - } - controlNextSettlement(mode: Exclude): void { if (this.pendingSettlement !== undefined) throw new Error("A settlement is already pending"); this.mode = mode; @@ -76,10 +66,23 @@ class ControlledSettlementDatabase implements SqliteDatabase { } } +class PrepareCountingDatabaseSync extends DatabaseSync { + private readonly prepareCounts = new Map(); + + override prepare(sql: string): StatementSync { + this.prepareCounts.set(sql, (this.prepareCounts.get(sql) ?? 0) + 1); + return super.prepare(sql); + } + + repeatedPrepares(): string[] { + return [...this.prepareCounts].filter(([, count]) => count > 1).map(([sql]) => sql); + } +} + describe("portable SQLite facade settlement", () => { - it("reuses read statements across calls and write statements within a transaction", async () => { - const database = new ControlledSettlementDatabase(await openNodeSqliteDatabase(":memory:")); - const storage = await SqliteStorage.open(database); + 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)); await storage.commit([{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], BACKGROUND_CONTEXT); await storage.commit( Array.from({ length: 100 }, (_, index) => ({ @@ -103,13 +106,7 @@ describe("portable SQLite facade settlement", () => { storage.commit([{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], BACKGROUND_CONTEXT), ).rejects.toThrow("ID 1 already belongs to conversation"); expect((await storage.entry(idFromNumber(2), BACKGROUND_CONTEXT))?.entry.kind).toBe("cached"); - expect(database.prepareCount("SELECT record, commit_seq FROM entries WHERE id = ?")).toBe(1); - // Two committing transactions insert 101 entries; each prepares the insert once. - expect( - database.transactionPrepareCount( - "INSERT INTO entries (id, conversation_id, head, commit_seq, record) VALUES (?, ?, ?, ?, ?)", - ), - ).toBe(2); + expect(connection.repeatedPrepares()).toEqual([]); await storage.close(BACKGROUND_CONTEXT); }); @@ -117,9 +114,9 @@ describe("portable SQLite facade settlement", () => { const database = await openNodeSqliteDatabase(":memory:"); await database.transaction(async (transaction) => { await transaction.exec("CREATE TABLE async_probe (value INTEGER)"); - await (await transaction.prepare("INSERT INTO async_probe (value) VALUES (?)")).run(1); + await transaction.run("INSERT INTO async_probe (value) VALUES (?)", 1); }); - expect(await (await database.prepare("SELECT value FROM async_probe")).get()).toEqual({ value: 1 }); + expect(await database.get("SELECT value FROM async_probe")).toEqual({ value: 1 }); await database.close(); await database.close(); }); @@ -152,7 +149,7 @@ describe("portable SQLite facade settlement", () => { releaseFirst(); await Promise.all([first, second]); - expect(await (await database.prepare("SELECT value FROM transaction_queue ORDER BY value")).all()).toEqual([ + expect(await database.all("SELECT value FROM transaction_queue ORDER BY value")).toEqual([ { value: 1 }, { value: 2 }, ]); @@ -162,7 +159,6 @@ describe("portable SQLite facade settlement", () => { it("queues ordinary operations behind an active transaction", async () => { const database = await openNodeSqliteDatabase(":memory:"); await database.exec("CREATE TABLE operation_queue (value INTEGER)"); - const readStatement = await database.prepare("SELECT value FROM operation_queue ORDER BY value"); let markTransactionStarted!: () => void; const transactionStarted = new Promise((resolve) => { markTransactionStarted = resolve; @@ -183,7 +179,7 @@ describe("portable SQLite facade settlement", () => { writeSettled = true; }); let readSettled = false; - const read = readStatement.all().finally(() => { + const read = database.all("SELECT value FROM operation_queue ORDER BY value").finally(() => { readSettled = true; }); await Promise.resolve(); @@ -200,11 +196,9 @@ describe("portable SQLite facade settlement", () => { it("rejects database operations from inside a transaction callback instead of waiting forever", async () => { const database = await openNodeSqliteDatabase(":memory:"); await database.exec("CREATE TABLE misuse_probe (value INTEGER)"); - const statement = await database.prepare("SELECT value FROM misuse_probe"); const misuse = "Use the transaction handle inside a transaction callback"; await expect(database.transaction(() => database.exec("SELECT 1"))).rejects.toThrow(misuse); - await expect(database.transaction(() => database.prepare("SELECT 1"))).rejects.toThrow(misuse); - await expect(database.transaction(() => statement.all())).rejects.toThrow(misuse); + await expect(database.transaction(() => database.all("SELECT value FROM misuse_probe"))).rejects.toThrow(misuse); await expect(database.transaction(() => database.transaction(async () => undefined))).rejects.toThrow( "Nested SQLite transactions are not supported", ); @@ -218,15 +212,14 @@ describe("portable SQLite facade settlement", () => { const database = await openNodeSqliteDatabase(":memory:"); await database.exec("CREATE TABLE stale_probe (value INTEGER)"); let handle!: SqliteExecutor; - let statement!: SqliteStatement; await database.transaction(async (transaction) => { handle = transaction; - statement = await transaction.prepare("INSERT INTO stale_probe (value) VALUES (?)"); + 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 (1)")).rejects.toThrow(stale); - await expect(statement.run(1)).rejects.toThrow(stale); - expect(await (await database.prepare("SELECT value FROM stale_probe")).all()).toEqual([]); + 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(); }); @@ -240,7 +233,7 @@ describe("portable SQLite facade settlement", () => { throw new StorageRejected("rejected after an escaped commit"); }), ).rejects.toThrow(AggregateError); - expect(await (await database.prepare("SELECT value FROM rollback_probe")).get()).toEqual({ value: 1 }); + expect(await database.get("SELECT value FROM rollback_probe")).toEqual({ value: 1 }); await database.close(); }); @@ -295,4 +288,31 @@ describe("portable SQLite facade settlement", () => { expect(await storage.entry(idFromNumber(200), BACKGROUND_CONTEXT)).toBeUndefined(); await storage.close(BACKGROUND_CONTEXT); }); + + it("reads a document from one committed state while a commit replaces its base", async () => { + const id = idFromNumber(5); + // Each yield count starts the commit at a different point of the read's record and revision queries. + for (let yields = 0; yields < 16; yields++) { + const storage = await SqliteStorage.open(await openNodeSqliteDatabase(":memory:")); + await storage.commit( + [ + { + type: "document.create", + record: { id, kind: "replaced", scope: { kind: "session" } }, + content: { kind: "base", version: 1, value: { value: 1 } }, + }, + ], + BACKGROUND_CONTEXT, + ); + const read = storage.document(id, "current", BACKGROUND_CONTEXT); + for (let index = 0; index < yields; index++) await Promise.resolve(); + const replace = storage.commit( + [{ type: "document.change", id, content: { kind: "base", version: 1, value: { value: 2 } } }], + BACKGROUND_CONTEXT, + ); + const [stored] = await Promise.all([read, replace]); + expect([{ value: 1 }, { value: 2 }]).toContainEqual(stored?.value); + await storage.close(BACKGROUND_CONTEXT); + } + }); }); diff --git a/packages/durable/test/sqlite-migrations.test.ts b/packages/durable/test/sqlite-migrations.test.ts index 535475641..c55117ea6 100644 --- a/packages/durable/test/sqlite-migrations.test.ts +++ b/packages/durable/test/sqlite-migrations.test.ts @@ -32,14 +32,10 @@ describe("durable SQLite migrations", () => { try { await applySqliteMigrations(database); await applySqliteMigrations(database); - expect(await (await database.prepare("SELECT version FROM durable_schema WHERE singleton = 1")).get()).toEqual( - { - version: CURRENT_SQLITE_SCHEMA_VERSION, - }, - ); - expect( - await (await database.prepare("SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1")).get(), - ).toEqual({ + expect(await database.get("SELECT version FROM durable_schema WHERE singleton = 1")).toEqual({ + version: CURRENT_SQLITE_SCHEMA_VERSION, + }); + expect(await database.get("SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1")).toEqual({ next_id: "2", next_seq: 1, }); @@ -52,7 +48,8 @@ describe("durable SQLite migrations", () => { const path = await databasePath(); const database = await openNodeSqliteDatabase(path); await applySqliteMigrations(database); - await (await database.prepare("UPDATE durable_schema SET version = ? WHERE singleton = 1")).run( + await database.run( + "UPDATE durable_schema SET version = ? WHERE singleton = 1", CURRENT_SQLITE_SCHEMA_VERSION + 1, ); await database.close(); @@ -78,23 +75,19 @@ describe("durable SQLite migrations", () => { ]; await expect(applySqliteMigrations(database, failed)).rejects.toThrow(); expect( - await ( - await database.prepare( - "SELECT count(*) AS count FROM sqlite_schema WHERE name IN ('durable_schema', 'migration_first', 'migration_second')", - ) - ).get(), + 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 (await database.prepare("SELECT version FROM durable_schema WHERE singleton = 1")).get()).toEqual( - { - version: 2, - }, - ); - expect(await (await database.prepare("SELECT value FROM migration_first")).get()).toEqual({ + 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 { @@ -132,15 +125,13 @@ describe("durable SQLite migrations", () => { }, ]; await expect(applySqliteMigrations(database, failedMigrations)).rejects.toThrow(); - expect(await (await database.prepare("SELECT version FROM durable_schema WHERE singleton = 1")).get()).toEqual({ + expect(await database.get("SELECT version FROM durable_schema WHERE singleton = 1")).toEqual({ version: CURRENT_SQLITE_SCHEMA_VERSION, }); expect( - await ( - await database.prepare( - "SELECT count(*) AS count FROM sqlite_schema WHERE type = 'table' AND name = 'migration_probe'", - ) - ).get(), + await database.get( + "SELECT count(*) AS count FROM sqlite_schema WHERE type = 'table' AND name = 'migration_probe'", + ), ).toEqual({ count: 0 }); const successfulMigrations: readonly SqliteMigration[] = [ @@ -148,10 +139,10 @@ describe("durable SQLite migrations", () => { { version: nextVersion, statements: ["CREATE TABLE migration_probe (value TEXT) STRICT"] }, ]; await applySqliteMigrations(database, successfulMigrations); - expect(await (await database.prepare("SELECT version FROM durable_schema WHERE singleton = 1")).get()).toEqual({ + expect(await database.get("SELECT version FROM durable_schema WHERE singleton = 1")).toEqual({ version: nextVersion, }); - expect(await (await database.prepare("SELECT record, commit_seq FROM entries WHERE id = 2")).get()).toEqual({ + expect(await database.get("SELECT record, commit_seq FROM entries WHERE id = 2")).toEqual({ record: JSON.stringify({ id: 2, conversationId: ROOT_CONVERSATION_ID, @@ -160,9 +151,7 @@ describe("durable SQLite migrations", () => { }), commit_seq: 1, }); - expect( - await (await database.prepare("SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1")).get(), - ).toEqual({ + expect(await database.get("SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1")).toEqual({ next_id: "3", next_seq: 2, }); From 77ac62f22a8198608ea6d15b48547cfe2df150fe Mon Sep 17 00:00:00 2001 From: Christian Klotz Date: Wed, 30 Sep 2026 20:10:39 +0100 Subject: [PATCH 4/4] refactor(durable): simplify Node SQLite queue and bindings --- packages/durable/src/storage/sqlite/node.ts | 16 +++++++--------- 1 file changed, 7 insertions(+), 9 deletions(-) diff --git a/packages/durable/src/storage/sqlite/node.ts b/packages/durable/src/storage/sqlite/node.ts index 558c1dcfc..74c3cdb1e 100644 --- a/packages/durable/src/storage/sqlite/node.ts +++ b/packages/durable/src/storage/sqlite/node.ts @@ -1,7 +1,7 @@ import { AsyncLocalStorage } from "node:async_hooks"; import { mkdir } from "node:fs/promises"; import { dirname } from "node:path"; -import type { SQLInputValue, StatementSync } from "node:sqlite"; +import type { StatementSync } from "node:sqlite"; import { DatabaseSync } from "node:sqlite"; import type { SqliteDatabase, SqliteExecutor, SqliteValue } from "./database.ts"; import { SqliteStorage } from "./storage.ts"; @@ -24,15 +24,13 @@ class SerialOperationQueue { async run(operation: () => T | Promise): Promise { const previous = this.tail; - let release: () => void; - this.tail = new Promise((resolve) => { - release = resolve; - }); + const { promise, resolve: release } = Promise.withResolvers(); + this.tail = promise; await previous; try { return await operation(); } finally { - release!(); + release(); } } } @@ -58,16 +56,16 @@ abstract class NodeSqliteExecutor implements SqliteExecutor { run(sql: string, ...params: SqliteValue[]): Promise { return this.runOperation(() => { - this.statement(sql).run(...(params as SQLInputValue[])); + this.statement(sql).run(...params); }); } get(sql: string, ...params: SqliteValue[]): Promise { - return this.runOperation(() => this.statement(sql).get(...(params as SQLInputValue[])) as T | undefined); + 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 SQLInputValue[])) as T[]); + return this.runOperation(() => this.statement(sql).all(...params) as T[]); } protected abstract runOperation(operation: () => T): Promise;