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

feat(durable): execute SQLite queries by SQL text and cache statements per connection

Browse source 
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.
This commit is contained in:
Christian Klotz 2026-09-30 19:49:55 +01:00
parent fd3af5ee2b
commit cd0ba2fabc
No known key found for this signature in database
8 changed files with 233 additions and 293 deletions
Download patch file Download diff file Expand all files Collapse all files

5
packages/durable/README.md
View file

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

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

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

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

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

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

@ -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<SchemaRow>();
await transaction.run("INSERT OR IGNORE INTO durable_schema (singleton, version) VALUES (1, 0)");
const row = await transaction.get<SchemaRow>("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);
}
});
}

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

@ -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 = <T>(operation: () => T | Promise<T>) => Promise<T>;
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<string, StatementSync>;
constructor(statement: StatementSync, runOperation: RunOperation) {
this.statement = statement;
this.runOperation = runOperation;
}
run(...params: SqliteValue[]): Promise<void> {
return this.runOperation(() => {
this.statement.run(...(params as SQLInputValue[]));
});
}
get<T extends object>(...params: SqliteValue[]): Promise<T | undefined> {
return this.runOperation(() => this.statement.get(...(params as SQLInputValue[])) as T | undefined);
}
all<T extends object>(...params: SqliteValue[]): Promise<T[]> {
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<string, StatementSync>) {
this.database = database;
this.scope = scope;
this.statements = statements;
}
exec(sql: string): Promise<void> {
@ -77,40 +56,55 @@ class NodeSqliteTransaction implements SqliteExecutor {
});
}
prepare(sql: string): Promise<SqliteStatement> {
return this.runOperation(
() => new NodeSqliteStatement(this.database.prepare(sql), (operation) => this.runOperation(operation)),
);
run(sql: string, ...params: SqliteValue[]): Promise<void> {
return this.runOperation(() => {
this.statement(sql).run(...(params as SQLInputValue[]));
});
}
private async runOperation<T>(operation: () => T | Promise<T>): Promise<T> {
get<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T | undefined> {
return this.runOperation(() => this.statement(sql).get(...(params as SQLInputValue[])) as T | undefined);
}
all<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T[]> {
return this.runOperation(() => this.statement(sql).all(...(params as SQLInputValue[])) as T[]);
}
protected abstract runOperation<T>(operation: () => T): Promise<T>;
private statement(sql: string): StatementSync {
let statement = this.statements.get(sql);
if (statement === undefined) {
statement = this.database.prepare(sql);
this.statements.set(sql, statement);
}
return statement;
}
}
class NodeSqliteTransaction extends NodeSqliteExecutor {
private readonly scope: TransactionScope;
constructor(database: DatabaseSync, statements: Map<string, StatementSync>, scope: TransactionScope) {
super(database, statements);
this.scope = scope;
}
protected async runOperation<T>(operation: () => T): Promise<T> {
if (!this.scope.active) throw new Error("SQLite transaction handle is no longer active");
return operation();
}
}
/** `SqliteDatabase` adapter backed by Node's built-in `node:sqlite`. */
export class NodeSqliteDatabase 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<TransactionScope>();
private closed = false;
constructor(database: DatabaseSync) {
this.database = database;
}
exec(sql: string): Promise<void> {
return this.runOperation(() => {
this.database.exec(sql);
});
}
prepare(sql: string): Promise<SqliteStatement> {
return this.runOperation(
() => new NodeSqliteStatement(this.database.prepare(sql), (operation) => this.runOperation(operation)),
);
super(database, new Map());
}
transaction<T>(callback: (transaction: SqliteExecutor) => Promise<T>): Promise<T> {
@ -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<T>(operation: () => T | Promise<T>): Promise<T> {
protected runOperation<T>(operation: () => T): Promise<T> {
if (this.insideTransaction()) {
return Promise.reject(new Error("Use the transaction handle inside a transaction callback"));
}

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

@ -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<JsonValue, JsonValue, JsonValue>;
@ -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 <T extends object>(
statement: Promise<SqliteStatement>,
...params: SqliteValue[]
): Promise<T | undefined> => (await statement).get<T>(...params);
const allRows = async <T extends object>(statement: Promise<SqliteStatement>, ...params: SqliteValue[]): Promise<T[]> =>
(await statement).all<T>(...params);
const runStatement = async (statement: Promise<SqliteStatement>, ...params: SqliteValue[]): Promise<void> =>
(await statement).run(...params);
const cursorId = <I extends Id<string>>(cursor: Cursor | undefined): I | undefined => {
const after = cursor?.after;
if (after === undefined) return undefined;
@ -137,42 +126,14 @@ const writeId = (write: StorageWrite): Id<string> | 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<string, SqliteStatement>();
constructor(executor: SqliteExecutor) {
this.executor = executor;
}
exec(sql: string): Promise<void> {
return this.executor.exec(sql);
}
async prepare(sql: string): Promise<SqliteStatement> {
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<SqliteStorage> {
try {
await applySqliteMigrations(db);
const metadata = await getRow<MetadataRow>(
db.prepare("SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1"),
const metadata = await db.get<MetadataRow>(
"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<MetadataRow>(
executor.prepare("SELECT next_id, next_seq FROM durable_metadata WHERE singleton = 1"),
const metadata = await transaction.get<MetadataRow>(
"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<ConversationRecord | undefined> {
this.assertOpen();
const row = await getRow<JsonRow>(this.reads.prepare("SELECT record FROM conversations WHERE id = ?"), id);
const row = await this.db.get<JsonRow>("SELECT record FROM conversations WHERE id = ?", id);
return row === undefined ? undefined : parseJson<ConversationRecord>(row.record);
}
@ -251,8 +211,8 @@ export class SqliteStorage implements Storage {
params.push(query.ownerTaskId);
}
params.push(limit + 1);
const rows = await allRows<JsonRow>(
this.reads.prepare(`SELECT record FROM conversations WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`),
const rows = await this.db.all<JsonRow>(
`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<EntryJsonRow>(
this.reads.prepare("SELECT record, commit_seq FROM entries WHERE id = ?"),
id,
);
const row = await this.db.get<EntryJsonRow>("SELECT record, commit_seq FROM entries WHERE id = ?", id);
if (row === undefined) return undefined;
const entry = parseJson<EntryRecord>(row.record);
if (conversation !== undefined) {
@ -316,16 +273,12 @@ export class SqliteStorage implements Storage {
while (true) {
const row =
upper === undefined
? await getRow<JsonRow>(
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<JsonRow>(
"SELECT record FROM entries WHERE conversation_id = ? AND head IS NOT NULL ORDER BY id DESC LIMIT 1",
conversation.id,
)
: await getRow<JsonRow>(
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<JsonRow>(
"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<JsonRow>(
this.reads.prepare(`SELECT record FROM entries WHERE ${clauses.join(" AND ")} ORDER BY id DESC LIMIT ?`),
const rows = await this.db.all<JsonRow>(
`SELECT record FROM entries WHERE ${clauses.join(" AND ")} ORDER BY id DESC LIMIT ?`,
...params,
);
values.push(...rows.map((row) => parseJson<EntryRecord>(row.record)));
@ -376,7 +329,7 @@ export class SqliteStorage implements Storage {
async task(id: TaskId, _context: Context): Promise<StoredTask | undefined> {
this.assertOpen();
const row = await getRow<JsonRow>(this.reads.prepare("SELECT record FROM tasks WHERE id = ?"), id);
const row = await this.db.get<JsonRow>("SELECT record FROM tasks WHERE id = ?", id);
return row === undefined ? undefined : parseJson<StoredTask>(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<JsonRow>(
this.reads.prepare(`SELECT record FROM tasks WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`),
const rows = await this.db.all<JsonRow>(
`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<SubmissionRecord | undefined> {
this.assertOpen();
const row = await getRow<JsonRow>(this.reads.prepare("SELECT record FROM submissions WHERE id = ?"), id);
const row = await this.db.get<JsonRow>("SELECT record FROM submissions WHERE id = ?", id);
return row === undefined ? undefined : parseJson<SubmissionRecord>(row.record);
}
@ -444,8 +397,8 @@ export class SqliteStorage implements Storage {
params.push(query.status);
}
params.push(limit + 1);
const rows = await allRows<JsonRow>(
this.reads.prepare(`SELECT record FROM submissions WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`),
const rows = await this.db.all<JsonRow>(
`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<SubmissionRecord | undefined> {
this.assertOpen();
const row = await getRow<JsonRow>(
this.reads.prepare("SELECT record FROM submissions WHERE conversation_id = ? AND request_id = ?"),
const row = await this.db.get<JsonRow>(
"SELECT record FROM submissions WHERE conversation_id = ? AND request_id = ?",
conversationId,
encodeIndexedString(requestId),
);
@ -475,24 +428,25 @@ export class SqliteStorage implements Storage {
): Promise<DocumentRecord | undefined> {
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<JsonRow>(statement, ...params);
const row = await this.db.get<JsonRow>(sql, ...params);
return row === undefined ? undefined : parseJson<DocumentRecord>(row.record);
}
async document(id: DocumentId, at: DocumentPoint, _context: Context): Promise<StoredDocument | undefined> {
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<JsonRow>(
this.reads.prepare(`SELECT record FROM documents WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`),
const rows = await this.db.all<JsonRow>(
`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<ConversationRecord | undefined> {
const row = await getRow<JsonRow>(this.reads.prepare("SELECT record FROM conversations WHERE id = ?"), id);
const row = await this.db.get<JsonRow>("SELECT record FROM conversations WHERE id = ?", id);
return row === undefined ? undefined : parseJson<ConversationRecord>(row.record);
}
@ -542,7 +496,7 @@ export class SqliteStorage implements Storage {
id: DocumentId,
at: DocumentPoint,
): Promise<StoredDocument | undefined> {
const row = await getRow<JsonRow>(executor.prepare("SELECT record FROM documents WHERE id = ?"), id);
const row = await executor.get<JsonRow>("SELECT record FROM documents WHERE id = ?", id);
if (row === undefined) return undefined;
const record = parseJson<DocumentRecord>(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<RevisionRow>(
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<RevisionRow>(
`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<JsonObject>(base.content);
const tail = await allRows<RevisionRow>(
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<RevisionRow>(
`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<void> {
const claimed = new Map<Id<string>, 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<RecordIdRow>(lookup, id))?.record_type;
const existing = (await executor.get<RecordIdRow>("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<JsonRow>(executor.prepare("SELECT record FROM documents WHERE id = ?"), id);
const row = await executor.get<JsonRow>("SELECT record FROM documents WHERE id = ?", id);
const existing = row === undefined ? undefined : parseJson<DocumentRecord>(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<DocumentId | undefined> {
const parts = addressParts(address);
const id = (
await getRow<IdRow>(
executor.prepare(`SELECT id FROM documents
await executor.get<IdRow>(
`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<string>, table: TableName): Promise<void> {
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<JsonRow>(executor.prepare("SELECT record FROM documents WHERE id = ?"), id))!;
const row = (await executor.get<JsonRow>("SELECT record FROM documents WHERE id = ?", id))!;
record = parseJson<DocumentRecord>(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);
}
}
}

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

@ -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<string, number>();
private readonly transactionPrepareCounts = new Map<string, number>();
constructor(delegate: NodeSqliteDatabase) {
this.delegate = delegate;
@ -24,25 +23,24 @@ class ControlledSettlementDatabase implements SqliteDatabase {
return this.delegate.exec(sql);
}
prepare(sql: string): Promise<SqliteStatement> {
this.prepareCounts.set(sql, (this.prepareCounts.get(sql) ?? 0) + 1);
return this.delegate.prepare(sql);
run(sql: string, ...params: SqliteValue[]): Promise<void> {
return this.delegate.run(sql, ...params);
}
get<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T | undefined> {
return this.delegate.get<T>(sql, ...params);
}
all<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T[]> {
return this.delegate.all<T>(sql, ...params);
}
transaction<T>(callback: (transaction: SqliteExecutor) => Promise<T>): Promise<T> {
const mode = this.mode;
this.mode = "immediate";
const counted = (transaction: SqliteExecutor): Promise<T> =>
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<SettlementMode, "immediate">): 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<string, number>();
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<EntryId>(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<void>((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<EntryId>(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<DocumentId>(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);
}
});
});

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

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