diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index c70d1415b..259dbad8c 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -2,6 +2,14 @@ ## [Unreleased] +### Added + +- `SqliteStorage.open(database, { prefix })` can namespace every table and index; `openNodeSqliteStorage()` and `openDurableObjectSqliteStorage()` forward it, for databases shared with host tables. + +### Changed + +- Cloudflare Durable Object SQLite adapters created for the same storage object now coordinate their operations, allowing multiple Pi storages to safely share one Durable Object database. + ### Fixed - The system baseline written after a compaction now leads the model context when the kept history includes an assistant message, as does a first baseline written after earlier answers, instead of following the new input; Anthropic and OpenAI Codex requests sent it as a mid-conversation system message without an initial prompt or tool set. Other system messages after an assistant message stay in place ([#10904](https://github.com/earendil-works/pi/issues/10904)). diff --git a/packages/durable/README.md b/packages/durable/README.md index 884ea5a32..6907b4292 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -662,6 +662,23 @@ const usage = await harness.usage(context); // { models: { "openai/gpt-6-sol": U | 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. Keeps the whole history in memory too, so memory grows with it; use SQLite for long-running agents. | | Cloudflare Durable Object | `openDurableObjectSqliteStorage(ctx.storage)` from `@earendil-works/pi-durable/storage/sqlite/cloudflare` | The SQLite storage of one SQLite-backed Durable Object; commits are its storage transactions. | +SQLite storage can namespace its tables and indexes when sharing a database: pass `{ prefix: "pi_" }` to +`SqliteStorage.open(database, options)`, `openNodeSqliteStorage(path, options)`, or +`openDurableObjectSqliteStorage(ctx.storage, options)`. Storage renders the prefixed names into its SQL, so adapters +need no support. A prefix must be a SQLite identifier and cannot begin with `sqlite_`; a host may reserve more, such +as `_cf_` in Durable Objects. Changing the prefix does not migrate existing data. + +Multiple Pi storages opened through the Cloudflare adapter with the same storage object automatically share an +operation queue. To share the database with host tables, create an adapter and use it for both Pi and all host queries; +direct storage calls bypass the queue and may be rolled back with a Pi transaction: + +```typescript +const database = new DurableObjectSqliteDatabase(ctx.storage); +const piStorage = await SqliteStorage.open(database, { prefix: "pi_" }); +await database.exec("CREATE TABLE IF NOT EXISTS host_notes (text TEXT)"); +await database.run("INSERT INTO host_notes (text) VALUES (?)", text); +``` + 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`, `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, so calling `database` itself inside the callback never settles: diff --git a/packages/durable/src/storage/sqlite/cloudflare.ts b/packages/durable/src/storage/sqlite/cloudflare.ts index 5385133c3..63412b62b 100644 --- a/packages/durable/src/storage/sqlite/cloudflare.ts +++ b/packages/durable/src/storage/sqlite/cloudflare.ts @@ -1,5 +1,5 @@ import type { SqliteDatabase, SqliteExecutor, SqliteValue } from "./database.ts"; -import { SqliteStorage } from "./storage.ts"; +import { SqliteStorage, type SqliteStorageOptions } from "./storage.ts"; /** Values a Durable Object SQL binding accepts and returns. */ type DurableObjectSqlValue = ArrayBuffer | string | number | null; @@ -10,22 +10,61 @@ type DurableObjectSqlValue = ArrayBuffer | string | number | null; */ export type DurableObjectSqliteStorage = { readonly sql: { - exec(query: string, ...bindings: DurableObjectSqlValue[]): { toArray(): Record[] }; + exec( + query: string, + ...bindings: DurableObjectSqlValue[] + ): { + next(): + | { readonly done?: false; readonly value: Record } + | { readonly done: true; readonly value?: undefined }; + toArray(): Record[]; + }; }; transaction(closure: () => Promise): Promise; }; -/** Runs operations one at a time in call order, so a transaction excludes every other operation. */ +/** Runs synchronous operations immediately when idle and holds queued calls behind a transaction. */ class SerialQueue { - private tail: Promise = Promise.resolve(); + private tail: Promise = Promise.resolve(); + private pending = 0; run(operation: () => T | Promise): Promise { - const result = this.tail.then(operation); - this.tail = result.catch(() => {}); + if (this.pending > 0) return this.enqueue(operation); + try { + return Promise.resolve(operation()); + } catch (error) { + return Promise.reject(error); + } + } + + hold(operation: () => Promise): Promise { + return this.enqueue(operation); + } + + private enqueue(operation: () => T | Promise): Promise { + this.pending++; + const result = this.tail.then(operation).finally(() => { + this.pending--; + }); + this.tail = result.then( + () => {}, + () => {}, + ); return result; } } +const storageQueues = new WeakMap(); + +function queueFor(storage: DurableObjectSqliteStorage): SerialQueue { + let queue = storageQueues.get(storage); + if (queue === undefined) { + queue = new SerialQueue(); + storageQueues.set(storage, queue); + } + return queue; +} + /** Durable Object SQL binds numbers as doubles: a `bigint` outside the safe integer range would lose precision. */ function bindValue(value: SqliteValue): DurableObjectSqlValue { if (value instanceof Uint8Array) return value.slice().buffer as ArrayBuffer; @@ -38,15 +77,13 @@ function bindValue(value: SqliteValue): DurableObjectSqlValue { const bind = (params: readonly SqliteValue[]): DurableObjectSqlValue[] => params.map(bindValue); -/** Rows as returned, with `BLOB` columns converted to `Uint8Array` in place; no other copy is made. */ -function rows(rows: Record[]): T[] { - for (const row of rows) { - for (const key in row) { - const value = row[key]; - if (value instanceof ArrayBuffer) (row as Record)[key] = new Uint8Array(value); - } +/** A row as returned, with `BLOB` columns converted to `Uint8Array` in place; no other copy is made. */ +function row(row: Record): T { + for (const key in row) { + const value = row[key]; + if (value instanceof ArrayBuffer) (row as Record)[key] = new Uint8Array(value); } - return rows as T[]; + return row as T; } /** Executes SQL through `ctx.storage.sql`, which has no separate prepare step. */ @@ -69,12 +106,16 @@ class DurableObjectSqliteExecutor implements SqliteExecutor { async get(sql: string, ...params: SqliteValue[]): Promise { this.check(); - return rows(this.storage.sql.exec(sql, ...bind(params)).toArray())[0]; + const result = this.storage.sql.exec(sql, ...bind(params)).next(); + return result.done ? undefined : row(result.value); } async all(sql: string, ...params: SqliteValue[]): Promise { this.check(); - return rows(this.storage.sql.exec(sql, ...bind(params)).toArray()); + return this.storage.sql + .exec(sql, ...bind(params)) + .toArray() + .map((value) => row(value)); } /** Throws when this handle may no longer run SQL. */ @@ -91,11 +132,17 @@ class DurableObjectSqliteTransaction extends DurableObjectSqliteExecutor { } /** - * `SqliteDatabase` adapter for the SQLite storage of a Cloudflare Durable Object. Integers are JavaScript numbers; - * binding a `bigint` outside the safe integer range throws. + * `SqliteDatabase` adapter for the SQLite storage of a Cloudflare Durable Object. Adapters given the same storage + * object share a queue; direct calls to the underlying storage bypass it. Integers are JavaScript numbers; binding a + * `bigint` outside the safe integer range throws. */ export class DurableObjectSqliteDatabase extends DurableObjectSqliteExecutor implements SqliteDatabase { - private readonly queue = new SerialQueue(); + private readonly queue: SerialQueue; + + constructor(storage: DurableObjectSqliteStorage) { + super(storage); + this.queue = queueFor(storage); + } override exec(sql: string): Promise { return this.queue.run(() => super.exec(sql)); @@ -115,7 +162,7 @@ export class DurableObjectSqliteDatabase extends DurableObjectSqliteExecutor imp /** `ctx.storage.transaction()` commits when the callback resolves and rolls back when it rejects. */ transaction(callback: (transaction: SqliteExecutor) => Promise): Promise { - return this.queue.run(() => + return this.queue.hold(() => this.storage.transaction(async () => { const transaction = new DurableObjectSqliteTransaction(this.storage); try { @@ -133,7 +180,13 @@ export class DurableObjectSqliteDatabase extends DurableObjectSqliteExecutor imp } } -/** Open durable storage on a SQLite-backed Durable Object's `ctx.storage`. */ -export function openDurableObjectSqliteStorage(storage: DurableObjectSqliteStorage): Promise { - return SqliteStorage.open(new DurableObjectSqliteDatabase(storage)); +/** + * Open durable storage on a SQLite-backed Durable Object, optionally under a table prefix. Calls given the same + * storage object share an operation queue. + */ +export function openDurableObjectSqliteStorage( + storage: DurableObjectSqliteStorage, + options: SqliteStorageOptions = {}, +): Promise { + return SqliteStorage.open(new DurableObjectSqliteDatabase(storage), options); } diff --git a/packages/durable/src/storage/sqlite/index.ts b/packages/durable/src/storage/sqlite/index.ts index 666541868..3ae7b068a 100644 --- a/packages/durable/src/storage/sqlite/index.ts +++ b/packages/durable/src/storage/sqlite/index.ts @@ -5,4 +5,4 @@ export { SQLITE_MIGRATIONS, type SqliteMigration, } from "./migrations.ts"; -export { SqliteStorage } from "./storage.ts"; +export { SqliteStorage, type SqliteStorageOptions } from "./storage.ts"; diff --git a/packages/durable/src/storage/sqlite/migrations.ts b/packages/durable/src/storage/sqlite/migrations.ts index 541422d53..f18928ff9 100644 --- a/packages/durable/src/storage/sqlite/migrations.ts +++ b/packages/durable/src/storage/sqlite/migrations.ts @@ -1,91 +1,96 @@ import type { SqliteDatabase } from "./database.ts"; +import { DEFAULT_SQLITE_SCHEMA, type SqliteSchema } from "./schema.ts"; export type SqliteMigration = { readonly version: number; readonly statements: readonly string[]; }; -// next_id is TEXT because node:sqlite rejects INTEGER results outside JavaScript's safe integer range. -const INITIAL_SCHEMA: readonly string[] = [ - `CREATE TABLE durable_metadata ( - singleton INTEGER PRIMARY KEY CHECK (singleton = 1), - next_id TEXT NOT NULL, - next_seq INTEGER NOT NULL - ) STRICT`, - `INSERT INTO durable_metadata (singleton, next_id, next_seq) VALUES (1, '2', 1)`, - `CREATE TABLE record_ids ( - id INTEGER PRIMARY KEY, - record_type TEXT NOT NULL CHECK (record_type IN ('conversation', 'entry', 'task', 'submission', 'document')) - ) STRICT`, - `CREATE TABLE conversations ( - id INTEGER PRIMARY KEY, - owner_conversation_id INTEGER, - owner_task_id INTEGER, - record TEXT NOT NULL CHECK (json_valid(record)) - ) STRICT`, - "CREATE INDEX conversations_by_owner_conversation ON conversations (owner_conversation_id, id)", - "CREATE INDEX conversations_by_owner_task ON conversations (owner_task_id, id)", - `CREATE TABLE entries ( - id INTEGER PRIMARY KEY, - conversation_id INTEGER NOT NULL, - head INTEGER, - commit_seq INTEGER NOT NULL, - record TEXT NOT NULL CHECK (json_valid(record)) - ) STRICT`, - "CREATE INDEX entries_by_conversation ON entries (conversation_id, id DESC)", - "CREATE INDEX entry_heads_by_conversation ON entries (conversation_id, id DESC) WHERE head IS NOT NULL", - `CREATE TABLE tasks ( - id INTEGER PRIMARY KEY, - conversation_id INTEGER NOT NULL, - kind TEXT NOT NULL, - status TEXT NOT NULL CHECK (status IN ('pending', 'running', 'waiting', 'completing', 'terminal')), - abort_requested INTEGER NOT NULL CHECK (abort_requested IN (0, 1)), - background INTEGER NOT NULL CHECK (background IN (0, 1)), - record TEXT NOT NULL CHECK (json_valid(record)) - ) STRICT`, - "CREATE INDEX tasks_by_status ON tasks (status, id)", - "CREATE INDEX tasks_by_conversation ON tasks (conversation_id, id)", - "CREATE INDEX tasks_by_kind ON tasks (kind, id)", - "CREATE INDEX tasks_by_abort_requested ON tasks (abort_requested, id)", - "CREATE INDEX tasks_by_background ON tasks (background, id)", - `CREATE TABLE submissions ( - id INTEGER PRIMARY KEY, - conversation_id INTEGER NOT NULL, - request_id TEXT, - status TEXT NOT NULL CHECK (status IN ('queued', 'placed', 'done', 'unanswered')), - record TEXT NOT NULL CHECK (json_valid(record)) - ) STRICT`, - "CREATE INDEX submissions_by_request ON submissions (conversation_id, request_id)", - "CREATE INDEX submissions_by_conversation ON submissions (conversation_id, id)", - "CREATE INDEX submissions_by_status ON submissions (status, id)", - `CREATE TABLE documents ( - id INTEGER PRIMARY KEY, - kind TEXT NOT NULL, - family INTEGER NOT NULL CHECK (family IN (0, 1)), - key_value TEXT NOT NULL, - scope_kind TEXT NOT NULL CHECK (scope_kind IN ('session', 'conversation', 'task')), - owner_id INTEGER NOT NULL, - created_at INTEGER NOT NULL, - retired_at INTEGER, - record TEXT NOT NULL CHECK (json_valid(record)) - ) STRICT`, - `CREATE INDEX documents_by_address - ON documents (kind, scope_kind, owner_id, family, key_value, created_at DESC, retired_at)`, - "CREATE INDEX documents_by_scope ON documents (scope_kind, owner_id, id)", - "CREATE INDEX documents_by_scope_kind ON documents (scope_kind, owner_id, kind, id)", - `CREATE TABLE document_revisions ( - document_id INTEGER NOT NULL, - seq INTEGER NOT NULL, - kind TEXT NOT NULL CHECK (kind IN ('base', 'delta')), - version INTEGER NOT NULL, - content TEXT NOT NULL CHECK (json_valid(content)), - PRIMARY KEY (document_id, seq) - ) STRICT`, - "CREATE INDEX document_revisions_by_kind ON document_revisions (document_id, kind, seq DESC)", -]; +/** Build immutable, ordered schema history for a set of SQLite identifiers. */ +export function createSqliteMigrations(schema: SqliteSchema): readonly SqliteMigration[] { + const { tables, indexes } = schema; + // next_id is TEXT because node:sqlite rejects INTEGER results outside JavaScript's safe integer range. + const initialSchema: readonly string[] = [ + `CREATE TABLE ${tables.durableMetadata} ( + singleton INTEGER PRIMARY KEY CHECK (singleton = 1), + next_id TEXT NOT NULL, + next_seq INTEGER NOT NULL + ) STRICT`, + `INSERT INTO ${tables.durableMetadata} (singleton, next_id, next_seq) VALUES (1, '2', 1)`, + `CREATE TABLE ${tables.recordIds} ( + id INTEGER PRIMARY KEY, + record_type TEXT NOT NULL CHECK (record_type IN ('conversation', 'entry', 'task', 'submission', 'document')) + ) STRICT`, + `CREATE TABLE ${tables.conversations} ( + id INTEGER PRIMARY KEY, + owner_conversation_id INTEGER, + owner_task_id INTEGER, + record TEXT NOT NULL CHECK (json_valid(record)) + ) STRICT`, + `CREATE INDEX ${indexes.conversationsByOwnerConversation} ON ${tables.conversations} (owner_conversation_id, id)`, + `CREATE INDEX ${indexes.conversationsByOwnerTask} ON ${tables.conversations} (owner_task_id, id)`, + `CREATE TABLE ${tables.entries} ( + id INTEGER PRIMARY KEY, + conversation_id INTEGER NOT NULL, + head INTEGER, + commit_seq INTEGER NOT NULL, + record TEXT NOT NULL CHECK (json_valid(record)) + ) STRICT`, + `CREATE INDEX ${indexes.entriesByConversation} ON ${tables.entries} (conversation_id, id DESC)`, + `CREATE INDEX ${indexes.entryHeadsByConversation} ON ${tables.entries} (conversation_id, id DESC) WHERE head IS NOT NULL`, + `CREATE TABLE ${tables.tasks} ( + id INTEGER PRIMARY KEY, + conversation_id INTEGER NOT NULL, + kind TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN ('pending', 'running', 'waiting', 'completing', 'terminal')), + abort_requested INTEGER NOT NULL CHECK (abort_requested IN (0, 1)), + background INTEGER NOT NULL CHECK (background IN (0, 1)), + record TEXT NOT NULL CHECK (json_valid(record)) + ) STRICT`, + `CREATE INDEX ${indexes.tasksByStatus} ON ${tables.tasks} (status, id)`, + `CREATE INDEX ${indexes.tasksByConversation} ON ${tables.tasks} (conversation_id, id)`, + `CREATE INDEX ${indexes.tasksByKind} ON ${tables.tasks} (kind, id)`, + `CREATE INDEX ${indexes.tasksByAbortRequested} ON ${tables.tasks} (abort_requested, id)`, + `CREATE INDEX ${indexes.tasksByBackground} ON ${tables.tasks} (background, id)`, + `CREATE TABLE ${tables.submissions} ( + id INTEGER PRIMARY KEY, + conversation_id INTEGER NOT NULL, + request_id TEXT, + status TEXT NOT NULL CHECK (status IN ('queued', 'placed', 'done', 'unanswered')), + record TEXT NOT NULL CHECK (json_valid(record)) + ) STRICT`, + `CREATE INDEX ${indexes.submissionsByRequest} ON ${tables.submissions} (conversation_id, request_id)`, + `CREATE INDEX ${indexes.submissionsByConversation} ON ${tables.submissions} (conversation_id, id)`, + `CREATE INDEX ${indexes.submissionsByStatus} ON ${tables.submissions} (status, id)`, + `CREATE TABLE ${tables.documents} ( + id INTEGER PRIMARY KEY, + kind TEXT NOT NULL, + family INTEGER NOT NULL CHECK (family IN (0, 1)), + key_value TEXT NOT NULL, + scope_kind TEXT NOT NULL CHECK (scope_kind IN ('session', 'conversation', 'task')), + owner_id INTEGER NOT NULL, + created_at INTEGER NOT NULL, + retired_at INTEGER, + record TEXT NOT NULL CHECK (json_valid(record)) + ) STRICT`, + `CREATE INDEX ${indexes.documentsByAddress} + ON ${tables.documents} (kind, scope_kind, owner_id, family, key_value, created_at DESC, retired_at)`, + `CREATE INDEX ${indexes.documentsByScope} ON ${tables.documents} (scope_kind, owner_id, id)`, + `CREATE INDEX ${indexes.documentsByScopeKind} ON ${tables.documents} (scope_kind, owner_id, kind, id)`, + `CREATE TABLE ${tables.documentRevisions} ( + document_id INTEGER NOT NULL, + seq INTEGER NOT NULL, + kind TEXT NOT NULL CHECK (kind IN ('base', 'delta')), + version INTEGER NOT NULL, + content TEXT NOT NULL CHECK (json_valid(content)), + PRIMARY KEY (document_id, seq) + ) STRICT`, + `CREATE INDEX ${indexes.documentRevisionsByKind} ON ${tables.documentRevisions} (document_id, kind, seq DESC)`, + ]; + return [{ version: 1, statements: initialSchema }]; +} -/** Immutable, ordered schema history. Append new migrations after the initial schema ships. */ -export const SQLITE_MIGRATIONS: readonly SqliteMigration[] = [{ version: 1, statements: INITIAL_SCHEMA }]; +export const SQLITE_MIGRATIONS: readonly SqliteMigration[] = createSqliteMigrations(DEFAULT_SQLITE_SCHEMA); export const CURRENT_SQLITE_SCHEMA_VERSION = SQLITE_MIGRATIONS.at(-1)?.version ?? 0; @@ -95,6 +100,15 @@ type SchemaRow = { readonly version: number }; export async function applySqliteMigrations( database: SqliteDatabase, migrations: readonly SqliteMigration[] = SQLITE_MIGRATIONS, +): Promise { + await applySqliteMigrationsForSchema(database, DEFAULT_SQLITE_SCHEMA, migrations); +} + +/** Apply migrations rendered for a particular schema. */ +export async function applySqliteMigrationsForSchema( + database: SqliteDatabase, + schema: SqliteSchema, + migrations: readonly SqliteMigration[] = createSqliteMigrations(schema), ): Promise { for (let index = 0; index < migrations.length; index++) { if (migrations[index]?.version !== index + 1) { @@ -102,13 +116,14 @@ export async function applySqliteMigrations( } } + const { durableSchema } = schema.tables; await database.transaction(async (transaction) => { - await transaction.exec(`CREATE TABLE IF NOT EXISTS durable_schema ( + await transaction.exec(`CREATE TABLE IF NOT EXISTS ${durableSchema} ( singleton INTEGER PRIMARY KEY CHECK (singleton = 1), version INTEGER NOT NULL CHECK (version >= 0) ) STRICT`); - await transaction.run("INSERT OR IGNORE INTO durable_schema (singleton, version) VALUES (1, 0)"); - const row = await transaction.get("SELECT version FROM durable_schema WHERE singleton = 1"); + await transaction.run(`INSERT OR IGNORE INTO ${durableSchema} (singleton, version) VALUES (1, 0)`); + const row = await transaction.get(`SELECT version FROM ${durableSchema} WHERE singleton = 1`); if (row === undefined) throw new Error("Durable SQLite schema metadata is missing"); const currentVersion = migrations.at(-1)?.version ?? 0; if (row.version > currentVersion) { @@ -119,7 +134,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 transaction.run("UPDATE durable_schema SET version = ? WHERE singleton = 1", migration.version); + await transaction.run(`UPDATE ${durableSchema} 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 eabfa31fe..1a5e1f57b 100644 --- a/packages/durable/src/storage/sqlite/node.ts +++ b/packages/durable/src/storage/sqlite/node.ts @@ -3,16 +3,20 @@ import { dirname } from "node:path"; import type { StatementSync } from "node:sqlite"; import { DatabaseSync } from "node:sqlite"; import type { SqliteDatabase, SqliteExecutor, SqliteValue } from "./database.ts"; -import { SqliteStorage } from "./storage.ts"; +import { validateSqliteSchemaPrefix } from "./schema.ts"; +import { SqliteStorage, type SqliteStorageOptions } from "./storage.ts"; /** Node SQLite connection settings for a durable storage file. */ -export type NodeSqliteStorageOptions = { +export type NodeSqliteDatabaseOptions = { /** SQLite WAL auto-checkpoint threshold. SQLite and this adapter default to 1,000 pages; 0 disables it. */ readonly walAutoCheckpointPages?: number; /** Time SQLite waits for a competing file lock. SQLite defaults to 0; this adapter defaults to 5,000 ms. */ readonly busyTimeoutMs?: number; }; +/** Node SQLite connection settings plus durable storage options. */ +export type NodeSqliteStorageOptions = NodeSqliteDatabaseOptions & SqliteStorageOptions; + const DEFAULT_WAL_AUTO_CHECKPOINT_PAGES = 1_000; const DEFAULT_BUSY_TIMEOUT_MS = 5_000; @@ -179,7 +183,7 @@ export class NodeSqliteDatabase extends NodeSqliteExecutor implements SqliteData /** Open and configure a Node-backed SQLite database facade. */ export async function openNodeSqliteDatabase( path: string, - options: NodeSqliteStorageOptions = {}, + options: NodeSqliteDatabaseOptions = {}, ): Promise { const checkpointPages = options.walAutoCheckpointPages ?? DEFAULT_WAL_AUTO_CHECKPOINT_PAGES; const timeout = options.busyTimeoutMs ?? DEFAULT_BUSY_TIMEOUT_MS; @@ -201,10 +205,12 @@ export async function openNodeSqliteDatabase( } } -/** Open or create file-backed durable storage using Node's built-in SQLite. */ +/** Open or create file-backed durable storage using Node's built-in SQLite, optionally under a table prefix. */ export async function openNodeSqliteStorage( path: string, options: NodeSqliteStorageOptions = {}, ): Promise { - return SqliteStorage.open(await openNodeSqliteDatabase(path, options)); + const { prefix, ...databaseOptions } = options; + if (prefix !== undefined) validateSqliteSchemaPrefix(prefix); + return SqliteStorage.open(await openNodeSqliteDatabase(path, databaseOptions), { prefix }); } diff --git a/packages/durable/src/storage/sqlite/schema.ts b/packages/durable/src/storage/sqlite/schema.ts new file mode 100644 index 000000000..fd2fd0c21 --- /dev/null +++ b/packages/durable/src/storage/sqlite/schema.ts @@ -0,0 +1,69 @@ +/** + * Immutable identifier history for migrations and runtime queries. Never change or remove an existing value; a + * rename adds a new entry so historical migrations keep rendering their original object names. + */ +const TABLE_NAMES = { + durableSchema: "durable_schema", + durableMetadata: "durable_metadata", + recordIds: "record_ids", + conversations: "conversations", + entries: "entries", + tasks: "tasks", + submissions: "submissions", + documents: "documents", + documentRevisions: "document_revisions", +} as const; + +const INDEX_NAMES = { + conversationsByOwnerConversation: "conversations_by_owner_conversation", + conversationsByOwnerTask: "conversations_by_owner_task", + entriesByConversation: "entries_by_conversation", + entryHeadsByConversation: "entry_heads_by_conversation", + tasksByStatus: "tasks_by_status", + tasksByConversation: "tasks_by_conversation", + tasksByKind: "tasks_by_kind", + tasksByAbortRequested: "tasks_by_abort_requested", + tasksByBackground: "tasks_by_background", + submissionsByRequest: "submissions_by_request", + submissionsByConversation: "submissions_by_conversation", + submissionsByStatus: "submissions_by_status", + documentsByAddress: "documents_by_address", + documentsByScope: "documents_by_scope", + documentsByScopeKind: "documents_by_scope_kind", + documentRevisionsByKind: "document_revisions_by_kind", +} as const; + +type SqliteIdentifiers> = { readonly [K in keyof T]: string }; + +export type SqliteSchema = { + readonly tables: SqliteIdentifiers; + readonly indexes: SqliteIdentifiers; +}; + +/** Validate a prefix before using it as part of a SQLite identifier. */ +export function validateSqliteSchemaPrefix(prefix: string): void { + if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(prefix)) { + throw new Error(`Invalid SQLite schema prefix ${JSON.stringify(prefix)}`); + } + if (prefix.toLowerCase().startsWith("sqlite_")) { + throw new Error(`SQLite schema prefix ${JSON.stringify(prefix)} is reserved`); + } +} + +function prefixIdentifiers>(names: T, prefix: string): SqliteIdentifiers { + return Object.fromEntries( + Object.entries(names).map(([key, name]) => [key, `${prefix}${name}`]), + ) as SqliteIdentifiers; +} + +/** Build the identifiers used by every migration and storage query. */ +export function createSqliteSchema(prefix?: string): SqliteSchema { + if (prefix !== undefined) validateSqliteSchemaPrefix(prefix); + const actualPrefix = prefix ?? ""; + return { + tables: prefixIdentifiers(TABLE_NAMES, actualPrefix), + indexes: prefixIdentifiers(INDEX_NAMES, actualPrefix), + }; +} + +export const DEFAULT_SQLITE_SCHEMA = createSqliteSchema(); diff --git a/packages/durable/src/storage/sqlite/storage.ts b/packages/durable/src/storage/sqlite/storage.ts index c524246ef..4b13f8763 100644 --- a/packages/durable/src/storage/sqlite/storage.ts +++ b/packages/durable/src/storage/sqlite/storage.ts @@ -34,7 +34,8 @@ import type { } from "../../types.ts"; import { nextCursor, type ScanStart, scanStart } from "../scan.ts"; import type { SqliteDatabase, SqliteExecutor, SqliteValue } from "./database.ts"; -import { applySqliteMigrations } from "./migrations.ts"; +import { applySqliteMigrationsForSchema } from "./migrations.ts"; +import { createSqliteSchema, type SqliteSchema } from "./schema.ts"; type StoredTask = TaskRecord; type TableName = "conversation" | "entry" | "task" | "submission" | "document"; @@ -60,6 +61,14 @@ type ScopeColumns = { readonly ownerId: number; }; +export type SqliteStorageOptions = { + /** + * Prefix for every table and index in the durable schema. It must be a SQLite identifier and must not use + * SQLite's reserved `sqlite_` prefix. Existing data is not migrated when this changes. + */ + readonly prefix?: string; +}; + const parseJson = (value: string): T => JSON.parse(value) as T; const encodeJson = (value: unknown): string => JSON.stringify(value) as string; // Some SQLite bindings replace lone UTF-16 surrogates. JSON encoding keeps indexed identities lossless. @@ -142,26 +151,29 @@ const writeId = (write: StorageWrite): Id | undefined => { /** Portable SQLite implementation of the Pico storage contract. */ export class SqliteStorage implements Storage { private readonly db: SqliteDatabase; + private readonly schema: SqliteSchema; private nextId: number; private closed = false; private closing: Promise | undefined; private admittedReads = 0; private readsDrained: (() => void) | undefined; - private constructor(db: SqliteDatabase, nextId: number) { + private constructor(db: SqliteDatabase, schema: SqliteSchema, nextId: number) { this.db = db; + this.schema = schema; this.nextId = nextId; } /** Initialize storage over an owned SQLite database facade. */ - static async open(db: SqliteDatabase): Promise { + static async open(db: SqliteDatabase, options: SqliteStorageOptions = {}): Promise { try { - await applySqliteMigrations(db); + const schema = createSqliteSchema(options.prefix); + await applySqliteMigrationsForSchema(db, schema); const metadata = await db.get( - "SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1", + `SELECT next_id, next_seq FROM ${schema.tables.durableMetadata} WHERE singleton = 1`, ); if (metadata === undefined) throw new Error("Durable SQLite metadata is missing"); - return new SqliteStorage(db, Number(metadata.next_id)); + return new SqliteStorage(db, schema, Number(metadata.next_id)); } catch (error) { try { await db.close(); @@ -178,7 +190,7 @@ export class SqliteStorage implements Storage { const candidateNextId = this.candidateNextId(writes); const seq = await this.db.transaction(async (transaction) => { const metadata = await transaction.get( - "SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1", + `SELECT next_id, next_seq FROM ${this.schema.tables.durableMetadata} WHERE singleton = 1`, ); if (metadata === undefined) throw new Error("Durable SQLite metadata is missing"); const committedSeq = seqFromNumber(metadata.next_seq); @@ -187,7 +199,7 @@ export class SqliteStorage implements Storage { 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", + `UPDATE ${this.schema.tables.durableMetadata} SET next_id = ?, next_seq = ? WHERE singleton = 1`, String(Math.max(Number(metadata.next_id), candidateNextId)), committedSeq + 1, ); @@ -205,7 +217,7 @@ export class SqliteStorage implements Storage { async conversation(id: ConversationId, _context: Context): Promise { this.assertOpen(); - const row = await this.db.get("SELECT record FROM conversations WHERE id = ?", id); + const row = await this.db.get(`SELECT record FROM ${this.schema.tables.conversations} WHERE id = ?`, id); return row === undefined ? undefined : parseJson(row.record); } @@ -230,7 +242,7 @@ export class SqliteStorage implements Storage { } params.push(limit + 1); const rows = await this.db.all( - `SELECT record FROM conversations WHERE ${clauses.join(" AND ")} ORDER BY id ${scan.direction} LIMIT ?`, + `SELECT record FROM ${this.schema.tables.conversations} WHERE ${clauses.join(" AND ")} ORDER BY id ${scan.direction} LIMIT ?`, ...params, ); return page( @@ -289,7 +301,10 @@ export class SqliteStorage implements Storage { conversation = await this.readConversation(conversationId); if (conversation === undefined) throw new StorageRequestError(`Unknown conversation: ${conversationId}`); } - const row = await this.db.get("SELECT record, commit_seq FROM entries WHERE id = ?", id); + const row = await this.db.get( + `SELECT record, commit_seq FROM ${this.schema.tables.entries} WHERE id = ?`, + id, + ); if (row === undefined) return undefined; const entry = parseJson(row.record); if (conversation !== undefined) { @@ -315,11 +330,11 @@ export class SqliteStorage implements Storage { const row = upper === undefined ? await this.db.get( - "SELECT record FROM entries WHERE conversation_id = ? AND head IS NOT NULL ORDER BY id DESC LIMIT 1", + `SELECT record FROM ${this.schema.tables.entries} WHERE conversation_id = ? AND head IS NOT NULL ORDER BY id DESC LIMIT 1`, conversation.id, ) : await this.db.get( - "SELECT record FROM entries WHERE conversation_id = ? AND head IS NOT NULL AND id <= ? ORDER BY id DESC LIMIT 1", + `SELECT record FROM ${this.schema.tables.entries} WHERE conversation_id = ? AND head IS NOT NULL AND id <= ? ORDER BY id DESC LIMIT 1`, conversation.id, upper, ); @@ -355,7 +370,7 @@ export class SqliteStorage implements Storage { } params.push(limit + 1 - values.length); const rows = await this.db.all( - `SELECT record FROM entries WHERE ${clauses.join(" AND ")} ORDER BY id DESC LIMIT ?`, + `SELECT record FROM ${this.schema.tables.entries} WHERE ${clauses.join(" AND ")} ORDER BY id DESC LIMIT ?`, ...params, ); values.push(...rows.map((row) => parseJson(row.record))); @@ -400,7 +415,7 @@ export class SqliteStorage implements Storage { } params.push(limit + 1 - values.length); const rows = await this.db.all( - `SELECT record FROM entries WHERE ${clauses.join(" AND ")} ORDER BY id ASC LIMIT ?`, + `SELECT record FROM ${this.schema.tables.entries} WHERE ${clauses.join(" AND ")} ORDER BY id ASC LIMIT ?`, ...params, ); values.push(...rows.map((row) => parseJson(row.record))); @@ -411,7 +426,7 @@ export class SqliteStorage implements Storage { async task(id: TaskId, _context: Context): Promise { this.assertOpen(); - const row = await this.db.get("SELECT record FROM tasks WHERE id = ?", id); + const row = await this.db.get(`SELECT record FROM ${this.schema.tables.tasks} WHERE id = ?`, id); return row === undefined ? undefined : parseJson(row.record); } @@ -448,7 +463,7 @@ export class SqliteStorage implements Storage { } params.push(limit + 1); const rows = await this.db.all( - `SELECT record FROM tasks WHERE ${clauses.join(" AND ")} ORDER BY id ${scan.direction} LIMIT ?`, + `SELECT record FROM ${this.schema.tables.tasks} WHERE ${clauses.join(" AND ")} ORDER BY id ${scan.direction} LIMIT ?`, ...params, ); return page( @@ -460,7 +475,7 @@ export class SqliteStorage implements Storage { async submission(id: SubmissionId, _context: Context): Promise { this.assertOpen(); - const row = await this.db.get("SELECT record FROM submissions WHERE id = ?", id); + const row = await this.db.get(`SELECT record FROM ${this.schema.tables.submissions} WHERE id = ?`, id); return row === undefined ? undefined : parseJson(row.record); } @@ -485,7 +500,7 @@ export class SqliteStorage implements Storage { } params.push(limit + 1); const rows = await this.db.all( - `SELECT record FROM submissions WHERE ${clauses.join(" AND ")} ORDER BY id ${scan.direction} LIMIT ?`, + `SELECT record FROM ${this.schema.tables.submissions} WHERE ${clauses.join(" AND ")} ORDER BY id ${scan.direction} LIMIT ?`, ...params, ); return page( @@ -502,7 +517,7 @@ export class SqliteStorage implements Storage { ): Promise { this.assertOpen(); const row = await this.db.get( - "SELECT record FROM submissions WHERE conversation_id = ? AND request_id = ?", + `SELECT record FROM ${this.schema.tables.submissions} WHERE conversation_id = ? AND request_id = ?`, conversationId, encodeIndexedString(requestId), ); @@ -518,10 +533,10 @@ export class SqliteStorage implements Storage { const parts = addressParts(address); const sql = at === "current" - ? `SELECT record FROM documents + ? `SELECT record FROM ${this.schema.tables.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` - : `SELECT record FROM documents + : `SELECT record FROM ${this.schema.tables.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`; @@ -559,7 +574,7 @@ export class SqliteStorage implements Storage { } params.push(limit + 1); const rows = await this.db.all( - `SELECT record FROM documents WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`, + `SELECT record FROM ${this.schema.tables.documents} WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`, ...params, ); return page( @@ -600,7 +615,7 @@ export class SqliteStorage implements Storage { } private async readConversation(id: ConversationId): Promise { - const row = await this.db.get("SELECT record FROM conversations WHERE id = ?", id); + const row = await this.db.get(`SELECT record FROM ${this.schema.tables.conversations} WHERE id = ?`, id); return row === undefined ? undefined : parseJson(row.record); } @@ -609,7 +624,7 @@ export class SqliteStorage implements Storage { id: DocumentId, at: DocumentPoint, ): Promise { - const row = await executor.get("SELECT record FROM documents WHERE id = ?", id); + const row = await executor.get(`SELECT record FROM ${this.schema.tables.documents} WHERE id = ?`, id); if (row === undefined) return undefined; const record = parseJson(row.record); if (at !== "current" && isCurrentOnly(record)) { @@ -618,7 +633,7 @@ export class SqliteStorage implements Storage { if (!isAliveAt(record, at)) return undefined; const upper = at === "current" ? Number.MAX_SAFE_INTEGER : at; const base = await executor.get( - `SELECT seq, kind, version, content FROM document_revisions + `SELECT seq, kind, version, content FROM ${this.schema.tables.documentRevisions} WHERE document_id = ? AND kind = 'base' AND seq <= ? ORDER BY seq DESC LIMIT 1`, id, upper, @@ -626,7 +641,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 executor.all( - `SELECT seq, kind, version, content FROM document_revisions + `SELECT seq, kind, version, content FROM ${this.schema.tables.documentRevisions} WHERE document_id = ? AND seq > ? AND seq <= ? ORDER BY seq`, id, base.seq, @@ -657,8 +672,9 @@ 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 = (await executor.get("SELECT record_type FROM record_ids WHERE id = ?", id)) - ?.record_type; + const existing = ( + await executor.get(`SELECT record_type FROM ${this.schema.tables.recordIds} 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}`); @@ -728,14 +744,14 @@ export class SqliteStorage implements Storage { if (action.copy !== undefined && actions.has(action.copy.id)) { throw new Error(`Document copy ${id} source is changed in the copy batch`); } - const row = await executor.get("SELECT record FROM documents WHERE id = ?", id); + const row = await executor.get(`SELECT record FROM ${this.schema.tables.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 executor.get<{ readonly version: number }>( - "SELECT version FROM document_revisions WHERE document_id = ? ORDER BY seq DESC LIMIT 1", + `SELECT version FROM ${this.schema.tables.documentRevisions} WHERE document_id = ? ORDER BY seq DESC LIMIT 1`, id, ); if (previous === undefined) throw new Error(`Document ${id} delta has no base`); @@ -763,7 +779,7 @@ export class SqliteStorage implements Storage { const parts = addressParts(address); const id = ( await executor.get( - `SELECT id FROM documents + `SELECT id FROM ${this.schema.tables.documents} WHERE kind = ? AND scope_kind = ? AND owner_id = ? AND family = ? AND key_value = ? AND retired_at IS NULL LIMIT 1`, parts.kind, @@ -781,7 +797,7 @@ export class SqliteStorage implements Storage { case "conversation": await this.claimId(executor, write.value.id, "conversation"); await executor.run( - "INSERT INTO conversations (id, owner_conversation_id, owner_task_id, record) VALUES (?, ?, ?, ?)", + `INSERT INTO ${this.schema.tables.conversations} (id, owner_conversation_id, owner_task_id, record) VALUES (?, ?, ?, ?)`, write.value.id, write.value.owner?.conversationId ?? null, write.value.owner?.taskId ?? null, @@ -791,7 +807,7 @@ export class SqliteStorage implements Storage { case "entry": await this.claimId(executor, write.value.id, "entry"); await executor.run( - "INSERT INTO entries (id, conversation_id, head, commit_seq, record) VALUES (?, ?, ?, ?, ?)", + `INSERT INTO ${this.schema.tables.entries} (id, conversation_id, head, commit_seq, record) VALUES (?, ?, ?, ?, ?)`, write.value.id, write.value.conversationId, write.value.head ?? null, @@ -802,7 +818,7 @@ export class SqliteStorage implements Storage { case "task": await this.claimId(executor, write.value.id, "task"); await executor.run( - `INSERT INTO tasks (id, conversation_id, kind, status, abort_requested, background, record) + `INSERT INTO ${this.schema.tables.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, @@ -819,7 +835,7 @@ export class SqliteStorage implements Storage { case "submission": await this.claimId(executor, write.value.id, "submission"); await executor.run( - `INSERT INTO submissions (id, conversation_id, request_id, status, record) VALUES (?, ?, ?, ?, ?) + `INSERT INTO ${this.schema.tables.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, @@ -838,7 +854,11 @@ export class SqliteStorage implements Storage { } private async claimId(executor: SqliteExecutor, id: Id, table: TableName): Promise { - await executor.run("INSERT OR IGNORE INTO record_ids (id, record_type) VALUES (?, ?)", id, table); + await executor.run( + `INSERT OR IGNORE INTO ${this.schema.tables.recordIds} (id, record_type) VALUES (?, ?)`, + id, + table, + ); } private async applyDocumentActions( @@ -878,7 +898,7 @@ export class SqliteStorage implements Storage { const parts = addressParts(record); await this.claimId(executor, id, "document"); await executor.run( - `INSERT INTO documents + `INSERT INTO ${this.schema.tables.documents} (id, kind, family, key_value, scope_kind, owner_id, created_at, retired_at, record) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, id, @@ -892,17 +912,20 @@ export class SqliteStorage implements Storage { encodeJson(record), ); } else { - const row = (await executor.get("SELECT record FROM documents WHERE id = ?", id))!; + const row = (await executor.get( + `SELECT record FROM ${this.schema.tables.documents} WHERE id = ?`, + id, + ))!; record = parseJson(row.record); } if (content !== undefined) { if (content.kind === "base" && isCurrentOnly(record)) { - await executor.run("DELETE FROM document_revisions WHERE document_id = ?", id); + await executor.run(`DELETE FROM ${this.schema.tables.documentRevisions} WHERE document_id = ?`, id); } const encodedContent = content.kind === "base" ? encodeJson(content.value) : encodeJson(content.ops); await executor.run( - "INSERT INTO document_revisions (document_id, seq, kind, version, content) VALUES (?, ?, ?, ?, ?)", + `INSERT INTO ${this.schema.tables.documentRevisions} (document_id, seq, kind, version, content) VALUES (?, ?, ?, ?, ?)`, id, seq, content.kind, @@ -915,14 +938,14 @@ export class SqliteStorage implements Storage { if (action.create === undefined) { record = { ...record, retiredAt: seq }; await executor.run( - "UPDATE documents SET retired_at = ?, record = ? WHERE id = ?", + `UPDATE ${this.schema.tables.documents} SET retired_at = ?, record = ? WHERE id = ?`, seq, encodeJson(record), id, ); } if (isCurrentOnly(record)) { - await executor.run("DELETE FROM document_revisions WHERE document_id = ?", id); + await executor.run(`DELETE FROM ${this.schema.tables.documentRevisions} WHERE document_id = ?`, id); } } } diff --git a/packages/durable/test/sqlite-cloudflare.test.ts b/packages/durable/test/sqlite-cloudflare.test.ts index 41dbb9110..7546283f7 100644 --- a/packages/durable/test/sqlite-cloudflare.test.ts +++ b/packages/durable/test/sqlite-cloudflare.test.ts @@ -1,41 +1,62 @@ import { DatabaseSync } from "node:sqlite"; +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; import { registerStorageConformance } from "@earendil-works/pi-durable/testing"; import { describe, expect, it } from "vitest"; +import { idFromNumber } from "../src/ids.ts"; import { DurableObjectSqliteDatabase, type DurableObjectSqliteStorage, openDurableObjectSqliteStorage, } from "../src/storage/sqlite/cloudflare.ts"; +import { type EntryId, ROOT_CONVERSATION_ID } from "../src/types.ts"; type SqlValue = ArrayBuffer | string | number | null; /** * A stand-in for a SQLite-backed Durable Object's `ctx.storage` over `node:sqlite`, with its value types: `BLOB` - * bindings and results are `ArrayBuffer`s, and `exec()` without bindings may run several statements. + * bindings and results are `ArrayBuffer`s, and `exec()` without bindings may run several statements. It keeps this + * portable package free of Workers test dependencies; a Cloudflare host can run the same storage conformance suite + * against its binding with `runInDurableObject()` from `cloudflare:test`. */ -function durableObjectStorage(database = new DatabaseSync(":memory:")): DurableObjectSqliteStorage { +function durableObjectStorage( + database = new DatabaseSync(":memory:"), + onCursorOperation?: (operation: "next" | "toArray") => void, +): DurableObjectSqliteStorage { const toNode = (value: SqlValue) => (value instanceof ArrayBuffer ? new Uint8Array(value) : value); const fromNode = (value: unknown): SqlValue => value instanceof Uint8Array ? (value.slice().buffer as ArrayBuffer) : (value as SqlValue); + const cursor = (rows: Record[]) => { + let index = 0; + return { + next(): { done: false; value: Record } | { done: true } { + onCursorOperation?.("next"); + const value = rows[index++]; + return value === undefined ? { done: true } : { done: false, value }; + }, + toArray(): Record[] { + onCursorOperation?.("toArray"); + const remaining = rows.slice(index); + index = rows.length; + return remaining; + }, + }; + }; return { sql: { exec(query, ...bindings) { if (bindings.length === 0 && query.trim().replace(/;\s*$/, "").includes(";")) { database.exec(query); - return { toArray: () => [] }; + return cursor([]); } const statement = database.prepare(query); if (statement.columns().length === 0) { statement.run(...bindings.map(toNode)); - return { toArray: () => [] }; + return cursor([]); } - const rows = statement.all(...bindings.map(toNode)) as Record[]; - return { - toArray: () => - rows.map((row) => - Object.fromEntries(Object.entries(row).map(([key, value]) => [key, fromNode(value)])), - ), - }; + const rows = (statement.all(...bindings.map(toNode)) as Record[]).map((row) => + Object.fromEntries(Object.entries(row).map(([key, value]) => [key, fromNode(value)])), + ); + return cursor(rows); }, }, async transaction(closure) { @@ -52,11 +73,77 @@ function durableObjectStorage(database = new DatabaseSync(":memory:")): DurableO }; } -registerStorageConformance({ describe, expect, it }, "SqliteStorage on a Durable Object", async (use) => - use(await openDurableObjectSqliteStorage(durableObjectStorage())), +registerStorageConformance({ describe, expect, it }, "SqliteStorage on a prefixed Durable Object", async (use) => + use(await openDurableObjectSqliteStorage(durableObjectStorage(), { prefix: "pi_" })), ); describe("Durable Object SQLite adapter", () => { + it("keeps schema names unprefixed by default", async () => { + const sqlite = new DatabaseSync(":memory:"); + await openDurableObjectSqliteStorage(durableObjectStorage(sqlite)); + expect(sqlite.prepare("SELECT version FROM durable_schema").get()).toEqual({ version: 1 }); + expect(sqlite.prepare("SELECT name FROM sqlite_master WHERE name = 'pi_durable_schema'").get()).toBeUndefined(); + }); + + it("prefixes pi's tables and indexes without changing host tables", async () => { + const sqlite = new DatabaseSync(":memory:"); + const storage = durableObjectStorage(sqlite); + const database = new DurableObjectSqliteDatabase(storage); + await database.exec("CREATE TABLE host_notes (text TEXT)"); + await database.exec("CREATE TABLE tasks (text TEXT)"); + await openDurableObjectSqliteStorage(storage, { prefix: "pi_" }); + const names = sqlite + .prepare("SELECT name FROM sqlite_master WHERE type IN ('table', 'index') AND name NOT LIKE 'sqlite_%'") + .all() + .map((row) => String(row.name)); + expect(names).toContain("host_notes"); + expect(names).toContain("tasks"); + expect(names).not.toContain("pi_host_notes"); + const piNames = names.filter((name) => !["host_notes", "tasks"].includes(name)); + expect(piNames.length).toBeGreaterThan(10); + expect(piNames.every((name) => name.startsWith("pi_"))).toBe(true); + expect(piNames).toContain("pi_tasks"); + expect(piNames).toContain("pi_durable_schema"); + }); + + it("isolates concurrently opened prefixes on one Durable Object", async () => { + const storage = durableObjectStorage(); + const [first, second] = await Promise.all([ + openDurableObjectSqliteStorage(storage, { prefix: "first_" }), + openDurableObjectSqliteStorage(storage, { prefix: "second_" }), + ]); + const entryId = idFromNumber(2); + await Promise.all([ + first.commit( + [ + { type: "conversation", value: { id: ROOT_CONVERSATION_ID } }, + { type: "entry", value: { id: entryId, conversationId: ROOT_CONVERSATION_ID, kind: "first" } }, + ], + BACKGROUND_CONTEXT, + ), + second.commit( + [ + { type: "conversation", value: { id: ROOT_CONVERSATION_ID } }, + { type: "entry", value: { id: entryId, conversationId: ROOT_CONVERSATION_ID, kind: "second" } }, + ], + BACKGROUND_CONTEXT, + ), + ]); + expect((await first.entry(entryId, BACKGROUND_CONTEXT))?.entry.kind).toBe("first"); + expect((await second.entry(entryId, BACKGROUND_CONTEXT))?.entry.kind).toBe("second"); + await Promise.all([first.close(BACKGROUND_CONTEXT), second.close(BACKGROUND_CONTEXT)]); + }); + + it("rejects invalid and reserved prefixes", async () => { + await expect(openDurableObjectSqliteStorage(durableObjectStorage(), { prefix: "" })).rejects.toThrow("Invalid"); + await expect(openDurableObjectSqliteStorage(durableObjectStorage(), { prefix: "bad-prefix" })).rejects.toThrow( + "Invalid", + ); + await expect(openDurableObjectSqliteStorage(durableObjectStorage(), { prefix: "sqlite_pi_" })).rejects.toThrow( + "reserved", + ); + }); + it("binds and returns BLOB values as Uint8Array", async () => { const database = new DurableObjectSqliteDatabase(durableObjectStorage()); await database.exec("CREATE TABLE blobs (id INTEGER PRIMARY KEY, data BLOB, n INTEGER)"); @@ -67,6 +154,17 @@ describe("Durable Object SQLite adapter", () => { }); }); + it("reads only the cursor's first row for get", async () => { + const operations: string[] = []; + const database = new DurableObjectSqliteDatabase( + durableObjectStorage(new DatabaseSync(":memory:"), (operation) => operations.push(operation)), + ); + await database.exec("CREATE TABLE items (id INTEGER PRIMARY KEY)"); + await database.run("INSERT INTO items (id) VALUES (1), (2)"); + expect(await database.get("SELECT id FROM items ORDER BY id")).toEqual({ id: 1 }); + expect(operations).toEqual(["next"]); + }); + it("rejects a bigint binding outside the safe integer range", async () => { const database = new DurableObjectSqliteDatabase(durableObjectStorage()); await database.exec("CREATE TABLE numbers (n INTEGER)"); @@ -92,6 +190,15 @@ describe("Durable Object SQLite adapter", () => { expect(order).toEqual(["transaction", "close"]); }); + it("runs an uncontended statement before returning its promise", async () => { + const sqlite = new DatabaseSync(":memory:"); + const database = new DurableObjectSqliteDatabase(durableObjectStorage(sqlite)); + await database.exec("CREATE TABLE items (id INTEGER PRIMARY KEY)"); + const written = database.run("INSERT INTO items (id) VALUES (?)", 1); + expect(sqlite.prepare("SELECT id FROM items").all()).toEqual([{ id: 1 }]); + await written; + }); + it("rolls back a rejected transaction and rejects its handle afterwards", async () => { const database = new DurableObjectSqliteDatabase(durableObjectStorage()); await database.exec("CREATE TABLE items (id INTEGER PRIMARY KEY)"); @@ -107,24 +214,43 @@ describe("Durable Object SQLite adapter", () => { await expect(handle!.run("INSERT INTO items (id) VALUES (?)", 2)).rejects.toThrow("no longer active"); }); - it("queues operations behind an active transaction", async () => { - const database = new DurableObjectSqliteDatabase(durableObjectStorage()); + it("runs operations from another adapter after a transaction, outside its rollback", async () => { + const storage = durableObjectStorage(); + const database = new DurableObjectSqliteDatabase(storage); + const other = new DurableObjectSqliteDatabase(storage); await database.exec("CREATE TABLE items (id INTEGER PRIMARY KEY)"); - const order: string[] = []; const { promise: gate, resolve } = Promise.withResolvers(); const transaction = database.transaction(async (handle) => { await handle.run("INSERT INTO items (id) VALUES (?)", 1); await gate; - order.push("transaction"); - }); - const read = database.all("SELECT id FROM items").then((rows) => { - order.push("read"); - return rows; + throw new Error("abort"); }); + const outside = other.run("INSERT INTO items (id) VALUES (?)", 2); + const read = other.all("SELECT id FROM items"); resolve(); - await transaction; - expect(await read).toEqual([{ id: 1 }]); - expect(order).toEqual(["transaction", "read"]); + await expect(transaction).rejects.toThrow("abort"); + await outside; + expect(await read).toEqual([{ id: 2 }]); await database.close(); }); + + it("includes direct underlying calls in an active transaction", async () => { + const sqlite = new DatabaseSync(":memory:"); + const storage = durableObjectStorage(sqlite); + const database = new DurableObjectSqliteDatabase(storage); + await database.exec("CREATE TABLE items (id INTEGER PRIMARY KEY)"); + const { promise: opened, resolve: markOpened } = Promise.withResolvers(); + const { promise: gate, resolve } = Promise.withResolvers(); + const transaction = database.transaction(async (handle) => { + await handle.run("INSERT INTO items (id) VALUES (?)", 1); + markOpened(); + await gate; + throw new Error("abort"); + }); + await opened; + storage.sql.exec("INSERT INTO items (id) VALUES (?)", 2); + resolve(); + await expect(transaction).rejects.toThrow("abort"); + expect(sqlite.prepare("SELECT id FROM items").all()).toEqual([]); + }); }); diff --git a/packages/durable/test/sqlite-storage.test.ts b/packages/durable/test/sqlite-storage.test.ts index f68dc0a3c..c3db82ab2 100644 --- a/packages/durable/test/sqlite-storage.test.ts +++ b/packages/durable/test/sqlite-storage.test.ts @@ -190,6 +190,46 @@ describe("Pico SqliteStorage", () => { ).toBe(3); }); + it("namespaces tables and indexes with a prefix across reopen", async () => { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-sqlite-")); + tempDirectories.add(directory); + const path = join(directory, "storage.sqlite"); + const host = new DatabaseSync(path); + try { + host.exec("CREATE TABLE tasks (text TEXT)"); + } finally { + host.close(); + } + const storage = await openNodeSqliteStorage(path, { prefix: "pi_" }); + expect(await createRoot(storage)).toBe(1); + await storage.close(context); + + const reopened = await openNodeSqliteStorage(path, { prefix: "pi_" }); + openStorages.add(reopened); + expect(await reopened.conversation(ROOT_CONVERSATION_ID, context)).toEqual({ id: ROOT_CONVERSATION_ID }); + const database = new DatabaseSync(path); + try { + const names = database + .prepare("SELECT name FROM sqlite_master WHERE type IN ('table', 'index') AND name NOT LIKE 'sqlite_%'") + .all() + .map((row) => String(row.name)); + expect(names.filter((name) => !name.startsWith("pi_"))).toEqual(["tasks"]); + expect(names).toContain("pi_tasks"); + } finally { + database.close(); + } + }); + + it("rejects an invalid prefix before creating the database", async () => { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-sqlite-")); + tempDirectories.add(directory); + const databaseDirectory = join(directory, "nested"); + await expect( + openNodeSqliteStorage(join(databaseDirectory, "storage.sqlite"), { prefix: "bad-prefix" }), + ).rejects.toThrow("Invalid SQLite schema prefix"); + await expect(stat(databaseDirectory)).rejects.toMatchObject({ code: "ENOENT" }); + }); + it("rejects persisted metadata corruption on reopen", async () => { const { storage, path } = await createSqliteStorage(); await storage.close(context);