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): SqliteStorage commits a group in a few statements

Browse source 
SqliteStorage.commit(batches) reads the check state of all batches up front,
checks each batch in memory against what the earlier ones left (same
prefix-stop refusals, no savepoints), and writes the accepted batches per
table: small rows as one JSON array parameter (json_each), rows with much
text bound column by column within 100 parameters. A group commit over a
remote SQLite takes a few round trips instead of about eight per batch;
local groups are faster too. Randomized differential test against
MemoryStorage, raw-table and statement-count guards.
This commit is contained in:
Mario Zechner 2026-11-04 19:53:19 +00:00
parent d84123b326
commit fabf916fcf
8 changed files with 1441 additions and 431 deletions
Download patch file Download diff file Expand all files Collapse all files

4
packages/durable/CHANGELOG.md
View file

@ -2,6 +2,10 @@
## [Unreleased]
### Changed
- `SqliteStorage.commit()` costs a few SQL statements per call instead of about eight per batch: it reads the check state of all batches up front, checks them in memory, and writes the accepted batches per table in multi-row statements (rows as one JSON array parameter; rows with much text bound column by column), so a group commit over a remote SQLite takes a few round trips, not hundreds. A stored document record that cannot be parsed now rejects the call as a storage failure instead of refusing the batch.
## [1.5.0] - 2026-11-02
### Breaking Changes

9
packages/durable/docs/spec.md
View file

@ -6261,7 +6261,14 @@ after a committed base or retirement.
### 11.2 SQLite
One SQL transaction is one storage call: the batches of a `commit()` call,
each checked before it writes. SQLite stores:
each checked before it writes. The checks read the stored state of all the
call's batches in a few statements and run in memory, each batch against what
the batches before it left; the accepted batches are written per table in
multi-row statements (rows as one JSON array parameter, rows with much text
bound column by column within Durable Objects' 100 parameters), so a call
costs a few statements whatever its number of batches (a remote SQLite pays a
round trip per statement). A copy reads its source from the
database, so the batches before it are written first. SQLite stores:
- conversation, entry, task, and submission records;
- document records;

486
packages/durable/src/storage/sqlite/commit.ts Normal file
View file

@ -0,0 +1,486 @@
// One `SqliteStorage.commit()` call in a few statements whatever its number of batches: the check state of all its
// batches read up front (`CommitState`), each batch checked in memory against what the batches before it left, and the
// accepted batches written per table (`PendingWrites`, through `statements.ts`). Over a remote SQLite every statement
// is a round trip while the write lock is held; a group of commits must not cost one round trip per write.
import type { SqliteExecutor } from "@earendil-works/chord/sqlite";
import type {
DocumentContent,
DocumentCreate,
DocumentId,
DocumentRecord,
Id,
Seq,
StorageWrite,
} from "../../types.ts";
import {
addressColumnsKey,
addressKey,
addressParts,
type Cell,
documentRow,
encodeIndexedString,
encodeJson,
isCurrentOnly,
parseJson,
} from "./rows.ts";
import type { SqliteSchema } from "./schema.ts";
import { insertRows, readRows, runForEach } from "./statements.ts";
/** The table a record ID belongs to, as `record_ids` stores it. */
export type TableName = "conversation" | "entry" | "task" | "submission" | "document";
/** What one batch does to one document: its creation or copy, its new content, its retirement. */
export type DocumentAction = {
create?: DocumentCreate;
copy?: Extract<StorageWrite, { readonly type: "document.copy" }>["source"];
content?: DocumentContent;
retire: boolean;
};
type Tables = SqliteSchema["tables"];
/** The record ID a write claims and its table, if it claims one. */
export const claimOf = (write: StorageWrite): { readonly id: Id<string>; readonly table: TableName } | undefined => {
switch (write.type) {
case "conversation":
case "entry":
case "task":
case "submission":
return { id: write.value.id, table: write.type };
case "document.create":
case "document.copy":
return { id: write.record.id, table: "document" };
case "document.change":
case "document.retire":
return undefined;
}
};
/** One batch's writes per document; throws when a document gets two content commands or two retirements. */
export function prepareDocumentActions(writes: readonly StorageWrite[]): Map<DocumentId, DocumentAction> {
const actions = new Map<DocumentId, DocumentAction>();
for (const write of writes) {
if (
write.type !== "document.create" &&
write.type !== "document.copy" &&
write.type !== "document.change" &&
write.type !== "document.retire"
) {
continue;
}
const id = write.type === "document.create" || write.type === "document.copy" ? write.record.id : write.id;
let action = actions.get(id);
if (action === undefined) {
action = { retire: false };
actions.set(id, action);
}
switch (write.type) {
case "document.create":
if (action.create !== undefined || action.content !== undefined || action.copy !== undefined) {
throw new Error(`Document ${id} has more than one content command`);
}
action.create = write.record;
action.content = write.content;
break;
case "document.copy":
if (action.create !== undefined || action.content !== undefined || action.copy !== undefined) {
throw new Error(`Document ${id} has more than one content command`);
}
action.create = write.record;
action.copy = write.source;
break;
case "document.change":
if (action.content !== undefined || action.copy !== undefined) {
throw new Error(`Document ${id} has more than one content command`);
}
action.content = write.content;
break;
case "document.retire":
if (action.retire) throw new Error(`Document ${id} is retired more than once`);
action.retire = true;
break;
}
}
return actions;
}
type RecordIdRow = { readonly id: number; readonly record_type: TableName };
type DocumentRow = { readonly record: string };
type VersionRow = { readonly document_id: number; readonly version: number | null };
type AddressRow = {
readonly id: number;
readonly kind: string;
readonly scope_kind: string;
readonly owner_id: number;
readonly family: number;
readonly key_value: string;
};
/**
* What the checks of one call's batches read: the stored state of every ID, document and address the batches touch,
* read in a few statements when the call starts, then advanced in memory by each accepted batch.
*/
export class CommitState {
/** The table of each claimed ID; absent: unclaimed. */
readonly #recordIds: Map<number, TableName>;
/** Each touched document's record; absent: no such document. */
readonly #documents: Map<number, DocumentRecord>;
/** The version of each delta-changed document's latest revision; absent: no revision. */
readonly #versions: Map<number, number>;
/** The current incarnation of each created address (`addressKey`); `undefined`: none. */
readonly #current: Map<string, DocumentId | undefined>;
private constructor(
recordIds: Map<number, TableName>,
documents: Map<number, DocumentRecord>,
versions: Map<number, number>,
current: Map<string, DocumentId | undefined>,
) {
this.#recordIds = recordIds;
this.#documents = documents;
this.#versions = versions;
this.#current = current;
}
/** Reads the stored state the checks of `batches` need: one statement per kind of state (split only by size). */
static async read(
executor: SqliteExecutor,
tables: Tables,
batches: readonly (readonly StorageWrite[])[],
): Promise<CommitState> {
const ids = new Set<number>();
const documentIds = new Set<number>();
const deltaIds = new Set<number>();
const addresses = new Map<string, ReturnType<typeof addressParts>>();
for (const writes of batches) {
for (const write of writes) {
const claim = claimOf(write);
if (claim !== undefined) ids.add(claim.id);
switch (write.type) {
case "document.create":
case "document.copy":
documentIds.add(write.record.id);
addresses.set(addressKey(write.record), addressParts(write.record));
break;
case "document.change":
documentIds.add(write.id);
if (write.content.kind === "delta") deltaIds.add(write.id);
break;
case "document.retire":
documentIds.add(write.id);
break;
default:
break;
}
}
}
const recordIds = new Map<number, TableName>();
for (const row of await readRows<RecordIdRow>(
executor,
`SELECT id, record_type FROM ${tables.recordIds} WHERE id IN (SELECT value FROM json_each(?))`,
[...ids],
)) {
recordIds.set(row.id, row.record_type);
}
const documents = new Map<number, DocumentRecord>();
for (const row of await readRows<DocumentRow>(
executor,
`SELECT record FROM ${tables.documents} WHERE id IN (SELECT value FROM json_each(?))`,
[...documentIds],
)) {
const record = parseJson<DocumentRecord>(row.record);
documents.set(record.id, record);
}
const versions = new Map<number, number>();
// Driven by the requested IDs: each latest revision is one descending probe of the primary key.
for (const row of await readRows<VersionRow>(
executor,
`SELECT requested.value AS document_id, (
SELECT version FROM ${tables.documentRevisions} WHERE document_id = requested.value ORDER BY seq DESC LIMIT 1
) AS version FROM json_each(?) AS requested`,
[...deltaIds],
)) {
if (row.version !== null) versions.set(row.document_id, row.version);
}
return new CommitState(recordIds, documents, versions, await readCurrent(executor, tables, addresses));
}
/** Durable's own checks of one batch that need no copy source: throws the refusal. */
check(writes: readonly StorageWrite[], actions: ReadonlyMap<DocumentId, DocumentAction>): void {
this.#checkIds(writes);
this.#checkDocuments(actions);
}
/** Advances the state past an accepted batch stored at `seq`, its copies resolved to their content. */
apply(writes: readonly StorageWrite[], actions: ReadonlyMap<DocumentId, DocumentAction>, seq: Seq): void {
for (const write of writes) {
// An accepted batch claims an ID only for its stored table, if it has one (`#checkIds`).
const claim = claimOf(write);
if (claim !== undefined) this.#recordIds.set(claim.id, claim.table);
}
for (const [id, action] of actions) {
if (action.create !== undefined) {
const record = { ...action.create, createdAt: seq, ...(action.retire ? { retiredAt: seq } : {}) };
this.#documents.set(id, record);
if (!action.retire) this.#current.set(addressKey(record), id);
} else if (action.retire) {
const record = { ...this.#documents.get(id)!, retiredAt: seq };
this.#documents.set(id, record);
// The batch may have created the address's next incarnation already: only this one leaves.
const key = addressKey(record);
if (this.#current.get(key) === id) this.#current.set(key, undefined);
}
// A retired document's versions are never read again: every later action on it is refused first.
if (action.content !== undefined) this.#versions.set(id, action.content.version);
}
}
/** A document's record as the accepted batches left it. */
document(id: DocumentId): DocumentRecord | undefined {
return this.#documents.get(id);
}
#checkIds(writes: readonly StorageWrite[]): void {
const claimed = new Map<Id<string>, TableName>();
for (const write of writes) {
const claim = claimOf(write);
if (claim === undefined) continue;
const { id, table } = claim;
const existing = this.#recordIds.get(id);
const earlier = claimed.get(id);
if (table === "conversation" || table === "entry" || table === "document") {
if (existing !== undefined) throw new Error(`ID ${id} already belongs to ${existing}`);
if (earlier !== undefined) throw new Error(`ID ${id} is written more than once`);
} else {
if (existing !== undefined && existing !== table)
throw new Error(`ID ${id} already belongs to ${existing}`);
if (earlier !== undefined && earlier !== table) throw new Error(`ID ${id} is written as two record types`);
}
claimed.set(id, table);
}
}
#checkDocuments(actions: ReadonlyMap<DocumentId, DocumentAction>): void {
const liveCounts = new Map<string, number>();
for (const [id, action] of actions) {
if (action.copy !== undefined && actions.has(action.copy.id)) {
throw new Error(`Document copy ${id} source is changed in the copy batch`);
}
const existing = this.#documents.get(id);
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 = this.#versions.get(id);
if (previous === undefined) throw new Error(`Document ${id} delta has no base`);
if (previous !== action.content.version)
throw new Error(`Document ${id} version transition requires a base`);
}
// A plain change keeps its address's live count: only creations and retirements are counted.
if (action.create === undefined && !action.retire) continue;
const key = addressKey(action.create ?? existing!);
let live = liveCounts.get(key);
// Only created addresses were read: an address no batch creates only loses incarnations, never exceeds one.
if (live === undefined) live = this.#current.get(key) === undefined ? 0 : 1;
if (action.retire && existing !== undefined) live--;
if (action.create !== undefined && !action.retire) live++;
liveCounts.set(key, live);
}
for (const live of liveCounts.values()) {
if (live > 1) throw new Error("Document address already has a current incarnation");
}
}
}
/** The current incarnation of each of `addresses`, `undefined` for one without. */
async function readCurrent(
executor: SqliteExecutor,
tables: Tables,
addresses: ReadonlyMap<string, ReturnType<typeof addressParts>>,
): Promise<Map<string, DocumentId | undefined>> {
const current = new Map<string, DocumentId | undefined>();
for (const key of addresses.keys()) current.set(key, undefined);
const rows = await readRows<AddressRow>(
executor,
`SELECT document.id, document.kind, document.scope_kind, document.owner_id, document.family, document.key_value
FROM json_each(?) AS address JOIN ${tables.documents} AS document
ON document.kind = json_extract(address.value, '$[0]') AND document.scope_kind = json_extract(address.value, '$[1]')
AND document.owner_id = json_extract(address.value, '$[2]') AND document.family = json_extract(address.value, '$[3]')
AND document.key_value = json_extract(address.value, '$[4]')
WHERE document.retired_at IS NULL`,
[...addresses.values()].map((part) => [part.kind, part.scopeKind, part.ownerId, part.family, part.keyValue]),
);
for (const row of rows) {
current.set(
addressColumnsKey(row.kind, row.scope_kind, row.owner_id, row.family, row.key_value),
row.id as DocumentId,
);
}
return current;
}
/**
* The writes of the accepted batches not yet written, per table, as their net effect: a task or submission written
* twice is written once, as last written; a current-only document's revisions before its latest base or retirement
* are dropped (and its stored ones deleted).
*/
export class PendingWrites {
readonly #recordIds = new Map<number, TableName>();
readonly #conversations: Cell[][] = [];
readonly #entries: Cell[][] = [];
readonly #tasks = new Map<number, Cell[]>();
readonly #submissions = new Map<number, Cell[]>();
/** Each created or retired document's record. */
readonly #documents = new Map<number, DocumentRecord>();
/** The documents whose stored revisions go. */
readonly #cleared = new Set<number>();
/** Each document's new revisions, in seq order. */
readonly #revisions = new Map<number, Cell[][]>();
/** Adds a batch stored at `seq`, after `state` applied it. */
add(
writes: readonly StorageWrite[],
actions: ReadonlyMap<DocumentId, DocumentAction>,
seq: Seq,
state: CommitState,
): void {
for (const write of writes) this.#addWrite(write, seq);
for (const [id, action] of actions) this.#addDocument(id, action, seq, state.document(id)!);
}
#addWrite(write: StorageWrite, seq: Seq): void {
const claim = claimOf(write);
if (claim !== undefined) this.#recordIds.set(claim.id, claim.table);
switch (write.type) {
case "conversation":
this.#conversations.push([
write.value.id,
write.value.owner?.conversationId ?? null,
write.value.owner?.taskId ?? null,
encodeJson(write.value),
]);
break;
case "entry":
this.#entries.push([
write.value.id,
write.value.conversationId,
write.value.head ?? null,
seq,
encodeJson(write.value),
]);
break;
case "task":
this.#tasks.set(write.value.id, [
write.value.id,
write.value.conversationId,
encodeIndexedString(write.value.kind),
write.value.state.status,
write.value.abortRequested ? 1 : 0,
write.value.background ? 1 : 0,
encodeJson(write.value),
]);
break;
case "submission":
this.#submissions.set(write.value.id, [
write.value.id,
write.value.conversationId,
write.value.requestId === undefined ? null : encodeIndexedString(write.value.requestId),
write.value.status,
encodeJson(write.value),
]);
break;
default:
break;
}
}
#addDocument(id: DocumentId, action: DocumentAction, seq: Seq, record: DocumentRecord): void {
if (action.create !== undefined || action.retire) this.#documents.set(id, record);
const content = action.content;
if (content !== undefined) {
if (content.kind === "base" && isCurrentOnly(record)) this.#clear(id);
const encoded = content.kind === "base" ? encodeJson(content.value) : encodeJson(content.ops);
let revisions = this.#revisions.get(id);
if (revisions === undefined) {
revisions = [];
this.#revisions.set(id, revisions);
}
revisions.push([id, seq, content.kind, content.version, encoded]);
}
if (action.retire && isCurrentOnly(record)) this.#clear(id);
}
/** Drops a document's revisions: the pending ones, and the stored ones when written. */
#clear(id: number): void {
this.#cleared.add(id);
this.#revisions.delete(id);
}
/** Writes everything pending, a few statements per table (`insertRows`), and forgets it. */
async flush(executor: SqliteExecutor, tables: Tables): Promise<void> {
await insertRows(
executor,
tables.recordIds,
["id", "record_type"],
[...this.#recordIds].map(([id, table]) => [id, table]),
"ON CONFLICT(id) DO NOTHING",
);
await insertRows(
executor,
tables.conversations,
["id", "owner_conversation_id", "owner_task_id", "record"],
this.#conversations,
);
await insertRows(
executor,
tables.entries,
["id", "conversation_id", "head", "commit_seq", "record"],
this.#entries,
);
await insertRows(
executor,
tables.tasks,
["id", "conversation_id", "kind", "status", "abort_requested", "background", "record"],
[...this.#tasks.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`,
);
await insertRows(
executor,
tables.submissions,
["id", "conversation_id", "request_id", "status", "record"],
[...this.#submissions.values()],
`ON CONFLICT(id) DO UPDATE SET conversation_id = excluded.conversation_id,
request_id = excluded.request_id, status = excluded.status, record = excluded.record`,
);
// A created document is inserted, a stored one retired: one upsert writes both.
await insertRows(
executor,
tables.documents,
["id", "kind", "family", "key_value", "scope_kind", "owner_id", "created_at", "retired_at", "record"],
[...this.#documents.values()].map(documentRow),
"ON CONFLICT(id) DO UPDATE SET retired_at = excluded.retired_at, record = excluded.record",
);
await runForEach(
executor,
`DELETE FROM ${tables.documentRevisions} WHERE document_id IN (SELECT value FROM json_each(?))`,
[...this.#cleared],
);
await insertRows(
executor,
tables.documentRevisions,
["document_id", "seq", "kind", "version", "content"],
[...this.#revisions.values()].flat(),
);
this.#recordIds.clear();
this.#conversations.length = 0;
this.#entries.length = 0;
this.#tasks.clear();
this.#submissions.clear();
this.#documents.clear();
this.#cleared.clear();
this.#revisions.clear();
}
}

79
packages/durable/src/storage/sqlite/rows.ts Normal file
View file

@ -0,0 +1,79 @@
// How Durable's records map to SQLite: JSON text, indexed strings, and a document's scope and address columns.
import type { DocumentAddress, DocumentCreate, DocumentRecord } from "../../types.ts";
/** A value of a stored cell, as Durable's rows hold them. */
export type Cell = string | number | null;
/** A document's scope as its `scope_kind` and `owner_id` columns. */
export type ScopeColumns = {
readonly scopeKind: DocumentRecord["scope"]["kind"];
readonly ownerId: number;
};
/** Parses a stored JSON column. */
export const parseJson = <T>(value: string): T => JSON.parse(value) as T;
/** Encodes a value as a stored JSON column. */
export const encodeJson = (value: unknown): string => JSON.stringify(value) as string;
/** Encodes an indexed string column. Some SQLite bindings replace lone UTF-16 surrogates; JSON keeps it lossless. */
export const encodeIndexedString = (value: string): string => JSON.stringify(value);
/** A document scope's columns. */
export const scopeColumns = (scope: DocumentRecord["scope"]): ScopeColumns => {
switch (scope.kind) {
case "session":
return { scopeKind: "session", ownerId: 0 };
case "conversation":
return { scopeKind: "conversation", ownerId: scope.conversationId };
case "task":
return { scopeKind: "task", ownerId: scope.taskId };
}
};
/** A document address's columns, as `documents` stores and indexes them. */
export const addressParts = (address: DocumentAddress | DocumentCreate | DocumentRecord) => {
const scope = scopeColumns(address.scope);
return {
kind: encodeIndexedString(address.kind),
...scope,
family: address.key === undefined ? 0 : 1,
keyValue: encodeIndexedString(address.key ?? ""),
};
};
/** One string per document address, from its stored columns. */
export const addressColumnsKey = (
kind: string,
scopeKind: string,
ownerId: number,
family: number,
keyValue: string,
): string => JSON.stringify([kind, scopeKind, ownerId, family, keyValue]);
/** One string per document address. */
export const addressKey = (address: DocumentAddress | DocumentCreate | DocumentRecord): string => {
const parts = addressParts(address);
return addressColumnsKey(parts.kind, parts.scopeKind, parts.ownerId, parts.family, parts.keyValue);
};
/** Whether a document keeps only its current state: its revisions go at each new base and at retirement. */
export const isCurrentOnly = (record: DocumentRecord): boolean =>
record.scope.kind !== "conversation" || record.history === "latest";
/** A document record's `documents` row: id, kind, family, key, scope, owner, created and retired seqs, record. */
export function documentRow(record: DocumentRecord): Cell[] {
const parts = addressParts(record);
return [
record.id,
parts.kind,
parts.family,
parts.keyValue,
parts.scopeKind,
parts.ownerId,
record.createdAt,
record.retiredAt ?? null,
encodeJson(record),
];
}

105
packages/durable/src/storage/sqlite/statements.ts Normal file
View file

@ -0,0 +1,105 @@
// The multi-row statements of a commit call, few whatever the number of rows: rows as one JSON array parameter read
// through `json_each(?)` (one SQL text per table, no bound-parameter limit), and large rows bound column by column in
// multi-row `VALUES` of a few fixed sizes, so their already JSON-encoded records are not escaped a second time.
import type { SqliteExecutor } from "@earendil-works/chord/sqlite";
import type { Cell } from "./rows.ts";
/** The rows `sql` reads for `values`, given as its one JSON array parameter (`json_each(?)`): one statement per chunk. */
export async function readRows<T extends object>(
executor: SqliteExecutor,
sql: string,
values: readonly (Cell | readonly Cell[])[],
): Promise<T[]> {
const rows: T[] = [];
for (const array of jsonArrays(values.map((value) => JSON.stringify(value)))) {
rows.push(...(await executor.all<T>(sql, array)));
}
return rows;
}
/** Runs `sql` (a deletion) for `values`, given as its one JSON array parameter: one statement per chunk. */
export async function runForEach(executor: SqliteExecutor, sql: string, values: readonly Cell[]): Promise<void> {
for (const array of jsonArrays(values.map((value) => JSON.stringify(value)))) await executor.run(sql, array);
}
// Rows whose string cells hold more characters than this (before escaping) go bound: a second JSON escaping of their
// records costs more than the statements they need.
const LARGE_ROW_CHARACTERS = 8 * 1024;
/**
* Inserts `rows` (one cell per column, in order) into `table`, then `conflict` (an upsert's clause): the rows of little
* text from one JSON array parameter per chunk, the same SQL text whatever their number; the large ones bound column by
* column.
*/
export async function insertRows(
executor: SqliteExecutor,
table: string,
columns: readonly string[],
rows: readonly (readonly Cell[])[],
conflict = "",
): Promise<void> {
const small: string[] = [];
const large: (readonly Cell[])[] = [];
for (const row of rows) {
let text = 0;
for (const cell of row) if (typeof cell === "string") text += cell.length;
if (text > LARGE_ROW_CHARACTERS) large.push(row);
else small.push(JSON.stringify(row));
}
const names = columns.join(", ");
if (small.length > 0) {
const cells = columns.map((_, index) => `json_extract(item.value, '$[${index}]')`).join(", ");
// `WHERE true` keeps SQLite from reading an upsert's `ON` as a join constraint of the `SELECT`.
const sql = `INSERT INTO ${table} (${names}) SELECT ${cells} FROM json_each(?) AS item WHERE true ${conflict}`;
for (const array of jsonArrays(small)) await executor.run(sql, array);
}
await insertBound(executor, `INSERT INTO ${table} (${names}) VALUES`, columns.length, large, conflict);
}
// Durable Objects bind at most 100 parameters per statement.
const MAX_PARAMETERS = 100;
/**
* Inserts `rows` with `head` (`INSERT INTO t (columns) VALUES`), `width` cells each, in statements of 1, 2, 4, ... rows
* up to `MAX_PARAMETERS` cells: few statements, and at most a handful of SQL texts per table for prepared statements.
*/
async function insertBound(
executor: SqliteExecutor,
head: string,
width: number,
rows: readonly (readonly Cell[])[],
conflict: string,
): Promise<void> {
const row = `(${new Array(width).fill("?").join(", ")})`;
let size = 1;
while (size * 2 * width <= MAX_PARAMETERS) size *= 2;
// `size` only shrinks: halved before a chunk to fit the rows left, it is also what the loop then steps over.
for (let start = 0; start < rows.length; start += size) {
while (size > rows.length - start) size /= 2;
const chunk = rows.slice(start, start + size);
await executor.run(`${head} ${new Array(size).fill(row).join(", ")} ${conflict}`, ...chunk.flat());
}
}
// Durable Objects refuse a string over 2 MB, a bound parameter too; 512 Ki characters are at most 1.5 MB of UTF-8.
const MAX_ARRAY_CHARACTERS = 512 * 1024;
/** JSON values `parts`, in order, as JSON arrays of at most `MAX_ARRAY_CHARACTERS` each (a longer value alone). */
function jsonArrays(parts: readonly string[]): string[] {
const arrays: string[] = [];
let chunk: string[] = [];
// The array's text: its brackets, its parts and the commas between them.
let length = 2;
for (const part of parts) {
if (chunk.length > 0 && length + 1 + part.length > MAX_ARRAY_CHARACTERS) {
arrays.push(`[${chunk.join(",")}]`);
chunk = [];
length = 2;
}
length += (chunk.length > 0 ? 1 : 0) + part.length;
chunk.push(part);
}
if (chunk.length > 0) arrays.push(`[${chunk.join(",")}]`);
return arrays;
}

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

@ -12,7 +12,6 @@ import type {
Cursor,
DocumentAddress,
DocumentContent,
DocumentCreate,
DocumentId,
DocumentPoint,
DocumentQuery,
@ -36,12 +35,12 @@ import type {
TaskRecord,
} from "../../types.ts";
import { nextCursor, type ScanStart, scanStart } from "../scan.ts";
import { CommitState, claimOf, type DocumentAction, PendingWrites, prepareDocumentActions } from "./commit.ts";
import { applySqliteMigrationsForSchema } from "./migrations.ts";
import { addressParts, encodeIndexedString, isCurrentOnly, parseJson, scopeColumns } from "./rows.ts";
import { createSqliteSchema, type SqliteSchema } from "./schema.ts";
type StoredTask = TaskRecord<JsonValue, JsonValue, JsonValue>;
type TableName = "conversation" | "entry" | "task" | "submission" | "document";
type RecordIdRow = { readonly record_type: TableName };
type JsonRow = { readonly record: string };
type EntryJsonRow = { readonly record: string; readonly commit_seq: number };
type RevisionRow = {
@ -50,18 +49,7 @@ type RevisionRow = {
readonly version: number;
readonly content: string;
};
type IdRow = { readonly id: number };
type MetadataRow = { readonly next_id: string; readonly next_seq: number };
type DocumentAction = {
create?: DocumentCreate;
copy?: Extract<StorageWrite, { readonly type: "document.copy" }>["source"];
content?: DocumentContent;
retire: boolean;
};
type ScopeColumns = {
readonly scopeKind: DocumentRecord["scope"]["kind"];
readonly ownerId: number;
};
export type SqliteStorageOptions = {
/**
@ -71,11 +59,6 @@ export type SqliteStorageOptions = {
readonly prefix?: string;
};
const parseJson = <T>(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.
const encodeIndexedString = (value: string): string => JSON.stringify(value);
const cursorId = <I extends Id<string>>(cursor: Cursor | undefined): I | undefined => {
const after = cursor?.after;
if (after === undefined) return undefined;
@ -100,56 +83,11 @@ const scanSql = (start: ScanStart): { readonly clause: string; readonly param: n
? { clause: "id > ?", param: start.after ?? -1, direction: "ASC" }
: { clause: "id < ?", param: start.after ?? Number.MAX_SAFE_INTEGER, direction: "DESC" };
const scopeColumns = (scope: DocumentRecord["scope"]): ScopeColumns => {
switch (scope.kind) {
case "session":
return { scopeKind: "session", ownerId: 0 };
case "conversation":
return { scopeKind: "conversation", ownerId: scope.conversationId };
case "task":
return { scopeKind: "task", ownerId: scope.taskId };
}
};
const addressParts = (address: DocumentAddress | DocumentCreate | DocumentRecord) => {
const scope = scopeColumns(address.scope);
return {
kind: encodeIndexedString(address.kind),
...scope,
family: address.key === undefined ? 0 : 1,
keyValue: encodeIndexedString(address.key ?? ""),
};
};
const addressKey = (address: DocumentAddress | DocumentCreate | DocumentRecord): string => {
const parts = addressParts(address);
return JSON.stringify([parts.kind, parts.scopeKind, parts.ownerId, parts.family, parts.keyValue]);
};
const isAliveAt = (record: DocumentRecord, at: DocumentPoint): boolean => {
if (at === "current") return record.retiredAt === undefined;
return record.createdAt <= at && (record.retiredAt === undefined || at < record.retiredAt);
};
const isCurrentOnly = (record: DocumentRecord): boolean =>
record.scope.kind !== "conversation" || record.history === "latest";
const writeId = (write: StorageWrite): Id<string> | undefined => {
switch (write.type) {
case "conversation":
case "entry":
case "task":
case "submission":
return write.value.id;
case "document.create":
case "document.copy":
return write.record.id;
case "document.change":
case "document.retire":
return undefined;
}
};
/** Whether a database call of the checks failed, whatever it threw or rejected with. */
type CheckCalls = { failed: boolean };
@ -214,7 +152,11 @@ export class SqliteStorage implements Storage {
/**
* Commit `batches` in one SQL transaction, each with its own seq. Each batch is checked before it writes anything,
* against the state the earlier ones left, and the first one refused ends the batch: the transaction commits the ones
* before it. No savepoints, which Durable Objects do not allow: the checks run before the writes instead.
* before it. No savepoints, which Durable Objects do not allow: the checks run before the writes instead. The checks
* read the stored state of all batches up front and run in memory, and the accepted batches are written per table in
* multi-row statements (`commit.ts`), so a call costs a few statements whatever its number of batches. A batch with a
* copy reads the copy's source from the database: the batches before it are written first, so a call of many fork
* batches costs a few statements per fork.
*/
async commit(
batches: readonly (readonly StorageWrite[])[],
@ -228,27 +170,22 @@ export class SqliteStorage implements Storage {
await this.db.transaction(async (transaction) => {
const metadata = await this.readMetadata(transaction);
let next = metadata.next_seq;
// Whether a database call of a batch's checks failed: then the failure is an unknown outcome, not a refusal.
const calls: CheckCalls = { failed: false };
const checked = watchCalls(transaction, calls);
// A failure of the database rejects the call and rolls back the batches before it with the transaction.
const state = await CommitState.read(transaction, this.schema.tables, batches);
const pending = new PendingWrites();
for (const writes of batches) {
let documentActions: Map<DocumentId, DocumentAction>;
// Only Durable's own checks refuse a batch, before it wrote anything: the ones before it stand. A failure of
// the database rejects the call, and rolls back the batches before it with the transaction.
calls.failed = false;
try {
documentActions = this.prepareDocumentActions(writes);
await this.checkBatch(checked, writes, documentActions);
} catch (error) {
if (calls.failed) throw error;
refused = { error };
const checked = await this.checkBatch(transaction, state, pending, writes);
if ("refused" in checked) {
refused = { error: checked.refused };
break;
}
const seq = seqFromNumber(next++);
await this.applyBatch(transaction, writes, documentActions, seq);
state.apply(writes, checked.actions, seq);
pending.add(writes, checked.actions, seq, state);
seqs.push(seq);
candidateNextId = Math.max(candidateNextId, this.candidateNextId(writes));
}
await pending.flush(transaction, this.schema.tables);
if (seqs.length > 0) {
await this.writeMetadata(transaction, Math.max(Number(metadata.next_id), candidateNextId), next);
}
@ -275,25 +212,34 @@ export class SqliteStorage implements Storage {
);
}
/** Durable's own checks of one batch, copy resolution included, before it writes anything. */
/**
* Durable's own checks of one batch, against what the batches before it left: its document actions, or its refusal.
* Only these checks refuse; a failure of a database call rejects. A copy reads its source from the database, so the
* batches before it are written first.
*/
private async checkBatch(
executor: SqliteExecutor,
transaction: SqliteExecutor,
state: CommitState,
pending: PendingWrites,
writes: readonly StorageWrite[],
documentActions: Map<DocumentId, DocumentAction>,
): Promise<void> {
await this.checkGlobalIds(executor, writes);
await this.checkDocumentActions(executor, documentActions);
await this.resolveCopies(executor, documentActions);
}
private async applyBatch(
executor: SqliteExecutor,
writes: readonly StorageWrite[],
documentActions: ReadonlyMap<DocumentId, DocumentAction>,
seq: Seq,
): Promise<void> {
for (const write of writes) await this.applyTableWrite(executor, write, seq);
await this.applyDocumentActions(executor, documentActions, seq);
): Promise<{ readonly actions: Map<DocumentId, DocumentAction> } | { readonly refused: unknown }> {
let actions: Map<DocumentId, DocumentAction>;
try {
actions = prepareDocumentActions(writes);
state.check(writes, actions);
} catch (error) {
return { refused: error };
}
if (![...actions.values()].some((action) => action.copy !== undefined)) return { actions };
await pending.flush(transaction, this.schema.tables);
const calls: CheckCalls = { failed: false };
try {
await this.resolveCopies(watchCalls(transaction, calls), actions);
} catch (error) {
if (calls.failed) throw error;
return { refused: error };
}
return { actions };
}
async mintId<I extends Id<string>>(): Promise<I> {
@ -746,208 +692,12 @@ export class SqliteStorage implements Storage {
private candidateNextId(writes: readonly StorageWrite[]): number {
let nextId = this.nextId;
for (const write of writes) {
const id = writeId(write);
const id = claimOf(write)?.id;
if (id !== undefined) nextId = Math.max(nextId, id + 1);
}
return nextId;
}
private async checkGlobalIds(executor: SqliteExecutor, writes: readonly StorageWrite[]): Promise<void> {
const claimed = new Map<Id<string>, TableName>();
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 executor.get<RecordIdRow>(`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}`);
if (earlier !== undefined) throw new Error(`ID ${id} is written more than once`);
} else {
if (existing !== undefined && existing !== table)
throw new Error(`ID ${id} already belongs to ${existing}`);
if (earlier !== undefined && earlier !== table) throw new Error(`ID ${id} is written as two record types`);
}
claimed.set(id, table);
}
}
private prepareDocumentActions(writes: readonly StorageWrite[]): Map<DocumentId, DocumentAction> {
const actions = new Map<DocumentId, DocumentAction>();
for (const write of writes) {
if (
write.type !== "document.create" &&
write.type !== "document.copy" &&
write.type !== "document.change" &&
write.type !== "document.retire"
) {
continue;
}
const id = write.type === "document.create" || write.type === "document.copy" ? write.record.id : write.id;
let action = actions.get(id);
if (action === undefined) {
action = { retire: false };
actions.set(id, action);
}
switch (write.type) {
case "document.create":
if (action.create !== undefined || action.content !== undefined || action.copy !== undefined) {
throw new Error(`Document ${id} has more than one content command`);
}
action.create = write.record;
action.content = write.content;
break;
case "document.copy":
if (action.create !== undefined || action.content !== undefined || action.copy !== undefined) {
throw new Error(`Document ${id} has more than one content command`);
}
action.create = write.record;
action.copy = write.source;
break;
case "document.change":
if (action.content !== undefined || action.copy !== undefined) {
throw new Error(`Document ${id} has more than one content command`);
}
action.content = write.content;
break;
case "document.retire":
if (action.retire) throw new Error(`Document ${id} is retired more than once`);
action.retire = true;
break;
}
}
return actions;
}
private async checkDocumentActions(
executor: SqliteExecutor,
actions: ReadonlyMap<DocumentId, DocumentAction>,
): Promise<void> {
const liveCounts = new Map<string, number>();
for (const [id, action] of actions) {
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<JsonRow>(`SELECT record FROM ${this.schema.tables.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 executor.get<{ readonly version: number }>(
`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`);
if (previous.version !== action.content.version) {
throw new Error(`Document ${id} version transition requires a base`);
}
}
const record = action.create ?? existing!;
const key = addressKey(record);
let live = liveCounts.get(key);
if (live === undefined) live = (await this.currentDocumentId(executor, record)) === undefined ? 0 : 1;
if (action.retire && existing !== undefined) live--;
if (action.create !== undefined && !action.retire) live++;
liveCounts.set(key, live);
}
for (const live of liveCounts.values()) {
if (live > 1) throw new Error("Document address already has a current incarnation");
}
}
private async currentDocumentId(
executor: SqliteExecutor,
address: DocumentAddress | DocumentCreate | DocumentRecord,
): Promise<DocumentId | undefined> {
const parts = addressParts(address);
const id = (
await executor.get<IdRow>(
`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,
parts.scopeKind,
parts.ownerId,
parts.family,
parts.keyValue,
)
)?.id;
return id === undefined ? undefined : idFromNumber<DocumentId>(id);
}
private async applyTableWrite(executor: SqliteExecutor, write: StorageWrite, seq: Seq): Promise<void> {
switch (write.type) {
case "conversation":
await this.claimId(executor, write.value.id, "conversation");
await executor.run(
`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,
encodeJson(write.value),
);
break;
case "entry":
await this.claimId(executor, write.value.id, "entry");
await executor.run(
`INSERT INTO ${this.schema.tables.entries} (id, conversation_id, head, commit_seq, record) VALUES (?, ?, ?, ?, ?)`,
write.value.id,
write.value.conversationId,
write.value.head ?? null,
seq,
encodeJson(write.value),
);
break;
case "task":
await this.claimId(executor, write.value.id, "task");
await executor.run(
`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,
background = excluded.background, record = excluded.record`,
write.value.id,
write.value.conversationId,
encodeIndexedString(write.value.kind),
write.value.state.status,
write.value.abortRequested ? 1 : 0,
write.value.background ? 1 : 0,
encodeJson(write.value),
);
break;
case "submission":
await this.claimId(executor, write.value.id, "submission");
await executor.run(
`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,
write.value.conversationId,
write.value.requestId === undefined ? null : encodeIndexedString(write.value.requestId),
write.value.status,
encodeJson(write.value),
);
break;
case "document.create":
case "document.copy":
case "document.change":
case "document.retire":
break;
}
}
private async claimId(executor: SqliteExecutor, id: Id<string>, table: TableName): Promise<void> {
await executor.run(
`INSERT OR IGNORE INTO ${this.schema.tables.recordIds} (id, record_type) VALUES (?, ?)`,
id,
table,
);
}
/** Resolve each copy to a base of its source's state before this batch, which may not change the source. */
private async resolveCopies(executor: SqliteExecutor, actions: Map<DocumentId, DocumentAction>): Promise<void> {
for (const [id, action] of actions) {
@ -973,77 +723,6 @@ export class SqliteStorage implements Storage {
}
}
private async applyDocumentActions(
executor: SqliteExecutor,
actions: ReadonlyMap<DocumentId, DocumentAction>,
seq: Seq,
): Promise<void> {
for (const [id, action] of actions) {
// Copies are resolved to their content by `resolveCopies()`.
const content = action.content;
let record: DocumentRecord;
if (action.create !== undefined) {
record = {
...action.create,
createdAt: seq,
...(action.retire ? { retiredAt: seq } : {}),
};
const parts = addressParts(record);
await this.claimId(executor, id, "document");
await executor.run(
`INSERT INTO ${this.schema.tables.documents}
(id, kind, family, key_value, scope_kind, owner_id, created_at, retired_at, record)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
id,
parts.kind,
parts.family,
parts.keyValue,
parts.scopeKind,
parts.ownerId,
seq,
action.retire ? seq : null,
encodeJson(record),
);
} else {
const row = (await executor.get<JsonRow>(
`SELECT record FROM ${this.schema.tables.documents} WHERE id = ?`,
id,
))!;
record = parseJson<DocumentRecord>(row.record);
}
if (content !== undefined) {
if (content.kind === "base" && isCurrentOnly(record)) {
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 ${this.schema.tables.documentRevisions} (document_id, seq, kind, version, content) VALUES (?, ?, ?, ?, ?)`,
id,
seq,
content.kind,
content.version,
encodedContent,
);
}
if (action.retire) {
if (action.create === undefined) {
record = { ...record, retiredAt: seq };
await executor.run(
`UPDATE ${this.schema.tables.documents} SET retired_at = ?, record = ? WHERE id = ?`,
seq,
encodeJson(record),
id,
);
}
if (isCurrentOnly(record)) {
await executor.run(`DELETE FROM ${this.schema.tables.documentRevisions} WHERE document_id = ?`, id);
}
}
}
}
private assertOpen(): void {
if (this.closed) throw new Error("SqliteStorage is closed");
}

623
packages/durable/test/sqlite-commit.test.ts Normal file
View file

@ -0,0 +1,623 @@
// SqliteStorage.commit() over its batch path (src/storage/sqlite/commit.ts): random batch calls compared against the
// memory reference storage, and the statement count of a call, which must not grow with its batches.
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import type { JsonValue } from "@earendil-works/chord/json";
import type { SqliteDatabase, SqliteExecutor, SqliteValue } from "@earendil-works/chord/sqlite";
import { afterEach, describe, expect, it } from "vitest";
import { idFromNumber } from "../src/ids.ts";
import { MemoryStorage } from "../src/storage/memory.ts";
import { openNodeSqliteDatabase } from "../src/storage/sqlite/node.ts";
import { SqliteStorage } from "../src/storage/sqlite/storage.ts";
import type {
ConversationId,
DocumentContent,
DocumentCreate,
DocumentId,
EntryId,
Seq,
Storage,
StorageWrite,
SubmissionId,
TaskId,
TaskRecord,
} from "../src/types.ts";
import { ROOT_CONVERSATION_ID } from "../src/types.ts";
const context = BACKGROUND_CONTEXT;
const openStorages = new Set<Storage>();
const directories = new Set<string>();
afterEach(async () => {
for (const storage of openStorages) await storage.close(context);
openStorages.clear();
for (const directory of directories) await rm(directory, { recursive: true, force: true });
directories.clear();
});
/** A SQLite database in a fresh directory, optionally seen through `wrap`. */
async function openDatabase(): Promise<SqliteDatabase> {
const directory = await mkdtemp(join(tmpdir(), "pi-durable-sqlite-commit-"));
directories.add(directory);
return openNodeSqliteDatabase(join(directory, "storage.sqlite"));
}
async function openStorage(db: SqliteDatabase): Promise<SqliteStorage> {
const storage = await SqliteStorage.open(db);
openStorages.add(storage);
return storage;
}
/** Deterministic pseudo-random numbers (mulberry32), so a failing seed reproduces. */
function random(seed: number) {
let state = seed >>> 0;
const next = (): number => {
state = (state + 0x6d2b79f5) >>> 0;
let t = state;
t = Math.imul(t ^ (t >>> 15), t | 1);
t ^= t + Math.imul(t ^ (t >>> 7), t | 61);
return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
};
const int = (max: number): number => Math.floor(next() * max);
const pick = <T>(values: readonly T[]): T => values[int(values.length)]!;
const chance = (p: number): boolean => next() < p;
return { int, pick, chance };
}
type Random = ReturnType<typeof random>;
/** What the generator has proposed so far: the pools later writes reuse, valid or not. */
type World = {
nextId: number;
readonly conversations: ConversationId[];
readonly entries: EntryId[];
readonly tasks: TaskId<JsonValue>[];
readonly submissions: SubmissionId[];
readonly documents: { readonly id: DocumentId; readonly record: DocumentCreate }[];
readonly seqs: Seq[];
/**
* The documents the current call created or changed, with the version it gave them last, and the addresses it
* retired: later batches of the call target them, so batches depend on the ones before them in the same call.
*/
readonly call: Map<DocumentId, number>;
readonly retired: DocumentCreate[];
};
const TASK_STATES = [
{ status: "pending", checkpoint: { phase: "ready" } },
{ status: "running", checkpoint: { phase: "ready" } },
{ status: "terminal", outcome: { kind: "returned", value: 1 } },
] as const;
function fresh(world: World): number {
return world.nextId++;
}
/** Any ID the world knows, of any table: a write reusing it is often refused. */
function anyId(world: World, rng: Random): number {
const pools: readonly (readonly number[])[] = [
world.conversations,
world.entries,
world.tasks,
world.submissions,
world.documents.map((document) => document.id),
].filter((pool) => pool.length > 0);
return pools.length === 0 ? fresh(world) : rng.pick(rng.pick(pools));
}
function randomAddress(world: World, rng: Random, id: DocumentId): DocumentCreate {
const kind = rng.pick(["a", "b"]);
const key = rng.chance(0.5) ? undefined : rng.pick(["x", "y"]);
const base = { id, kind, ...(key === undefined ? {} : { key }) };
const scope = rng.int(3);
if (scope === 0) return { ...base, scope: { kind: "session" } };
if (scope === 1 && world.tasks.length > 0)
return { ...base, scope: { kind: "task", taskId: rng.pick(world.tasks) } };
const conversationId = rng.pick(world.conversations);
return rng.chance(0.5)
? { ...base, scope: { kind: "conversation", conversationId }, history: "latest", fork: "current" }
: { ...base, scope: { kind: "conversation", conversationId }, history: "rewindable", fork: "asOf" };
}
function randomContent(rng: Random, version: number): DocumentContent {
return rng.chance(0.4)
? { kind: "base", version: rng.chance(0.8) ? version : version + 1, value: { n: rng.int(100), at: rng.int(5) } }
: { kind: "delta", version: rng.chance(0.85) ? version : version + 1, ops: [["s", ["n"], rng.int(100)]] };
}
/** One random write, biased toward valid ones, with `reference` telling which documents are live. */
async function randomWrite(world: World, rng: Random, reference: MemoryStorage): Promise<StorageWrite> {
switch (rng.int(9)) {
case 0: {
const id = rng.chance(0.85)
? idFromNumber<ConversationId>(fresh(world))
: (anyId(world, rng) as ConversationId);
world.conversations.push(id);
return { type: "conversation", value: { id } };
}
case 1:
case 2: {
const id = rng.chance(0.9) ? idFromNumber<EntryId>(fresh(world)) : (anyId(world, rng) as EntryId);
world.entries.push(id);
const head = rng.chance(0.2) && world.entries.length > 0 ? { head: rng.pick(world.entries) } : {};
return {
type: "entry",
value: { id, conversationId: rng.pick(world.conversations), kind: "note", data: rng.int(1000), ...head },
};
}
case 3: {
const id =
world.tasks.length > 0 && rng.chance(0.6)
? rng.pick(world.tasks)
: rng.chance(0.9)
? idFromNumber<TaskId<JsonValue>>(fresh(world))
: (anyId(world, rng) as TaskId<JsonValue>);
world.tasks.push(id);
const task = {
id,
conversationId: rng.pick(world.conversations),
kind: rng.pick(["t.one", "t.two"]),
version: 1,
input: { value: rng.int(10) },
state: rng.pick(TASK_STATES),
background: rng.chance(0.5),
abortRequested: rng.chance(0.2),
} as TaskRecord<JsonValue, JsonValue, JsonValue>;
return { type: "task", value: task };
}
case 4: {
const id =
world.submissions.length > 0 && rng.chance(0.5)
? rng.pick(world.submissions)
: rng.chance(0.9)
? idFromNumber<SubmissionId>(fresh(world))
: (anyId(world, rng) as SubmissionId);
world.submissions.push(id);
// One request ID per submission: storages differ on duplicates, which the Session never writes.
const requestId = rng.chance(0.5) ? { requestId: `r${id}` } : {};
return {
type: "submission",
value: { id, conversationId: rng.pick(world.conversations), type: "input", status: "queued", ...requestId },
};
}
case 5: {
const id = rng.chance(0.9) ? idFromNumber<DocumentId>(fresh(world)) : (anyId(world, rng) as DocumentId);
// Often at an address the call retired or another document has: the address's live count decides.
const record =
world.retired.length > 0 && rng.chance(0.5)
? { ...rng.pick(world.retired), id }
: world.documents.length > 0 && rng.chance(0.3)
? { ...rng.pick(world.documents).record, id }
: randomAddress(world, rng, id);
world.documents.push({ id, record });
world.call.set(id, 1);
return { type: "document.create", record, content: { kind: "base", version: 1, value: { n: 0, at: 0 } } };
}
case 6:
case 7: {
if (world.documents.length === 0) return randomWrite(world, rng, reference);
const id =
world.call.size > 0 && rng.chance(0.4)
? rng.pick([...world.call.keys()])
: (await pickDocument(rng, reference, world.documents)).id;
if (rng.chance(0.25)) {
const record = world.documents.find((document) => document.id === id)!.record;
world.retired.push(record);
world.call.delete(id);
return { type: "document.retire", id };
}
const version = world.call.get(id) ?? (await reference.document(id, "current", context))?.version ?? 1;
const content = randomContent(rng, version);
world.call.set(id, content.version);
return { type: "document.change", id, content };
}
default: {
const sources = world.documents.filter((document) => document.record.scope.kind === "conversation");
if (sources.length === 0) return randomWrite(world, rng, reference);
const source = await pickDocument(rng, reference, sources);
const id = idFromNumber<DocumentId>(fresh(world));
const conversationId = rng.pick(world.conversations);
const record = rng.chance(0.85)
? ({ ...source.record, id, scope: { kind: "conversation", conversationId } } as DocumentCreate)
: randomAddress(world, rng, id);
world.documents.push({ id, record });
const historical = source.record.history === "rewindable" ? 0.4 : 0.05;
const at = world.seqs.length > 0 && rng.chance(historical) ? rng.pick(world.seqs) : "current";
return { type: "document.copy", record, source: { id: source.id, at } };
}
}
}
/**
* A document's replacement in one batch: a new incarnation at its address and its retirement, in either order, so the
* address's current incarnation changes within the batch.
*/
async function replacement(world: World, rng: Random, reference: MemoryStorage): Promise<StorageWrite[]> {
if (world.documents.length === 0) return [];
const target = await pickDocument(rng, reference, world.documents);
const id = idFromNumber<DocumentId>(fresh(world));
const record = { ...target.record, id };
world.documents.push({ id, record });
world.call.set(id, 1);
world.call.delete(target.id);
world.retired.push(target.record);
const pair: StorageWrite[] = [
{ type: "document.create", record, content: { kind: "base", version: 1, value: { n: 0, at: 0 } } },
{ type: "document.retire", id: target.id },
];
return rng.chance(0.5) ? pair : pair.reverse();
}
/** One of `documents`, mostly a live one (as `reference` stores it). */
async function pickDocument<T extends { readonly id: DocumentId }>(
rng: Random,
reference: MemoryStorage,
documents: readonly T[],
): Promise<T> {
if (rng.chance(0.15)) return rng.pick(documents);
const live: T[] = [];
for (const document of documents) {
if ((await reference.document(document.id, "current", context)) !== undefined) live.push(document);
}
return live.length > 0 ? rng.pick(live) : rng.pick(documents);
}
/** A read's result, or its error's message: both storages must agree on either. */
async function settle(read: () => Promise<unknown>): Promise<unknown> {
try {
return { value: await read() };
} catch (error) {
return { error: error instanceof Error ? error.message : String(error) };
}
}
/** Everything a storage exposes about the world's records, at `"current"` and at `seqs`. */
async function snapshot(storage: Storage, world: World, seqs: readonly Seq[]): Promise<unknown> {
const all = 10_000;
const documentIds = [...new Set(world.documents.map((document) => document.id))];
const scopes = [
{ kind: "session" } as const,
...world.conversations.map((conversationId) => ({ kind: "conversation", conversationId }) as const),
...world.tasks.map((taskId) => ({ kind: "task", taskId }) as const),
];
return {
conversations: (await storage.scanConversations({}, all, undefined, context)).items,
entries: await Promise.all([...new Set(world.entries)].map((id) => settle(() => storage.entry(id, context)))),
transcripts: await Promise.all(
[...new Set(world.conversations)].map((conversationId) =>
settle(() => storage.scanEntries({ conversationId, order: "ascending" }, all, undefined, context)),
),
),
tasks: (await storage.scanTasks({}, all, undefined, context)).items,
submissions: (await storage.scanSubmissions({}, all, undefined, context)).items,
requests: await Promise.all(
[...new Set(world.conversations)].flatMap((conversationId) =>
[...new Set(world.submissions)].map((id) =>
settle(() => storage.submissionByRequest(conversationId, `r${id}`, context)),
),
),
),
documents: await Promise.all(
documentIds.flatMap((id) =>
(["current", ...seqs] as const).map((at) => settle(() => storage.document(id, at, context))),
),
),
scopes: await Promise.all(
scopes.map((scope) => settle(() => storage.scanDocuments({ scope, at: "current" }, all, undefined, context))),
),
addresses: await Promise.all(
world.documents.map(({ record }) =>
settle(() =>
storage.findDocument(
{ kind: record.kind, scope: record.scope, ...(record.key === undefined ? {} : { key: record.key }) },
"current",
context,
),
),
),
),
};
}
/**
* Every row of a SQLite storage's tables, in key order: also what reads do not show, such as the revisions a
* current-only document dropped.
*/
async function dumpTables(db: SqliteDatabase): Promise<Record<string, unknown>> {
const tables = ["durable_metadata", "record_ids", "conversations", "entries", "tasks", "submissions", "documents"];
const dump: Record<string, unknown> = {};
for (const table of [...tables, "document_revisions"])
dump[table] = await db.all(`SELECT * FROM ${table} ORDER BY 1, 2`);
return dump;
}
/**
* The revisions a current-only document should have dropped, which no read shows: one older than its latest base, or
* any once it is retired.
*/
function staleRevisions(db: SqliteDatabase): Promise<unknown[]> {
return db.all(
`SELECT revision.document_id, revision.seq FROM document_revisions AS revision
JOIN documents AS document ON document.id = revision.document_id
WHERE (document.scope_kind != 'conversation' OR json_extract(document.record, '$.history') = 'latest')
AND (document.retired_at IS NOT NULL OR revision.seq < (
SELECT MAX(seq) FROM document_revisions WHERE document_id = revision.document_id AND kind = 'base'
))`,
);
}
/** Commits `batches` one call each until one is refused: what one call of all of them must store. */
async function commitEach(storage: Storage, batches: readonly (readonly StorageWrite[])[]): Promise<void> {
for (const batch of batches) {
if ((await storage.commit([batch], () => {}, context)).stored === 0) return;
}
}
/** Commits `batches` and returns what the call reported: stored count, refused or not, and the receipts. */
async function commitOutcome(storage: Storage, batches: readonly (readonly StorageWrite[])[]) {
const receipts: [number, Seq][] = [];
const result = await storage.commit(batches, (index, seq) => receipts.push([index, seq]), context);
return { stored: result.stored, refused: "refused" in result, receipts };
}
describe("SqliteStorage batch commits against the memory reference", () => {
for (const seed of [1, 2, 3, 4, 5, 6]) {
it(`agrees with MemoryStorage on random batch calls (seed ${seed})`, async () => {
const rng = random(seed);
const reference = new MemoryStorage();
openStorages.add(reference);
const groupedDb = await openDatabase();
const sqlite = await openStorage(groupedDb);
// The same batches, one call each: a group must leave exactly the rows its batches leave alone.
const singleDb = await openDatabase();
const single = await openStorage(singleDb);
const world: World = {
nextId: 2,
conversations: [ROOT_CONVERSATION_ID],
entries: [],
tasks: [],
submissions: [],
documents: [],
seqs: [],
call: new Map(),
retired: [],
};
const root: StorageWrite[][] = [[{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }]];
expect(await commitOutcome(sqlite, root)).toEqual(await commitOutcome(reference, root));
await commitEach(single, root);
let refusals = 0;
let prefixes = 0;
let groups = 0;
for (let round = 0; round < 120; round++) {
const batches: StorageWrite[][] = [];
world.call.clear();
world.retired.length = 0;
const count = 1 + rng.int(6);
for (let index = 0; index < count; index++) {
const writes: StorageWrite[] = [];
const size = 1 + rng.int(4);
for (let write = 0; write < size; write++) writes.push(await randomWrite(world, rng, reference));
if (rng.chance(0.2)) writes.push(...(await replacement(world, rng, reference)));
batches.push(writes);
}
const expected = await commitOutcome(reference, batches);
const actual = await commitOutcome(sqlite, batches);
expect(actual, `round ${round}: ${JSON.stringify(batches)}`).toEqual(expected);
await commitEach(single, batches);
expect(await dumpTables(groupedDb), `round ${round} tables`).toEqual(await dumpTables(singleDb));
expect(await staleRevisions(groupedDb), `round ${round} revisions`).toEqual([]);
if (expected.refused) refusals++;
if (expected.refused && expected.stored > 0) prefixes++;
if (!expected.refused && batches.length > 1) groups++;
world.seqs.push(...expected.receipts.map(([, seq]) => seq));
const seqs = world.seqs.length > 0 ? [world.seqs[rng.int(world.seqs.length)]!] : [];
expect(await snapshot(sqlite, world, seqs), `round ${round}`).toEqual(
await snapshot(reference, world, seqs),
);
}
expect(await snapshot(sqlite, world, world.seqs)).toEqual(await snapshot(reference, world, world.seqs));
// The generator must exercise every path: refusals, refused prefixes, accepted groups.
expect(refusals).toBeGreaterThan(10);
expect(prefixes).toBeGreaterThan(10);
expect(groups).toBeGreaterThan(10);
});
}
});
/** A statement a transaction ran, with its parameters. */
type Ran = { readonly sql: string; readonly params: readonly SqliteValue[] };
/** `db` recording the statements its transactions run. */
function recordStatements(db: SqliteDatabase): { readonly db: SqliteDatabase; readonly ran: Ran[] } {
const ran: Ran[] = [];
const recorded = (executor: SqliteExecutor): SqliteExecutor => ({
exec: (sql) => {
ran.push({ sql, params: [] });
return executor.exec(sql);
},
run: (sql, ...params) => {
ran.push({ sql, params });
return executor.run(sql, ...params);
},
runCounted: (sql, ...params) => {
ran.push({ sql, params });
return executor.runCounted(sql, ...params);
},
get: (sql, ...params) => {
ran.push({ sql, params });
return executor.get(sql, ...params);
},
all: (sql, ...params) => {
ran.push({ sql, params });
return executor.all(sql, ...params);
},
});
return {
db: {
...recorded(db),
transaction: (callback) => db.transaction((tx) => callback(recorded(tx))),
close: () => db.close(),
},
ran,
};
}
describe("SqliteStorage batch commit statements", () => {
/**
* One call of `count` batches, each a document delta and an entry (as a Session turn writes), after a call that
* creates the documents: the statements of the second call, and the database.
*/
async function groupOf(count: number, data: JsonValue = 0) {
const database = await openDatabase();
const recording = recordStatements(database);
const storage = await openStorage(recording.db);
const setup: StorageWrite[] = [{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }];
const documents: DocumentId[] = [];
for (let index = 0; index < count; index++) {
const id = await storage.mintId<DocumentId>();
documents.push(id);
setup.push({
type: "document.create",
record: { id, kind: "counter", scope: { kind: "session" }, key: String(id) },
content: { kind: "base", version: 1, value: { n: 0 } },
});
}
await storage.commit([setup], () => {}, context);
const batches: StorageWrite[][] = [];
for (const id of documents) {
batches.push([
{ type: "document.change", id, content: { kind: "delta", version: 1, ops: [["s", ["n"], 1]] } },
{
type: "entry",
value: { id: await storage.mintId<EntryId>(), conversationId: ROOT_CONVERSATION_ID, kind: "note", data },
},
]);
}
const before = recording.ran.length;
expect((await storage.commit(batches, () => {}, context)).stored).toBe(count);
return { statements: recording.ran.slice(before), database };
}
it("runs the same eight statements for one batch and for a group of 150", async () => {
// Metadata read, three check reads (IDs, documents, latest versions), three inserts, metadata write.
expect((await groupOf(1)).statements).toHaveLength(8);
const group = await groupOf(150);
expect(group.statements).toHaveLength(8);
// One SQL text per table whatever the group's size: prepared statements are reused.
expect(group.statements.map((statement) => statement.sql)).toEqual(
(await groupOf(3)).statements.map((statement) => statement.sql),
);
});
it("splits the JSON array of small rows only by size: at most 512 Ki characters each", async () => {
// 100 entries of 7,000 characters: under the large-row bound each, 700 K characters together.
const { statements } = await groupOf(100, "x".repeat(7_000));
const inserts = statements.filter((statement) => statement.sql.startsWith("INSERT INTO entries"));
expect(inserts.map((statement) => statement.sql.includes("json_each"))).toEqual([true, true]);
expect(inserts.every((statement) => (statement.params[0] as string).length <= 512 * 1024)).toBe(true);
expect(statements).toHaveLength(9);
});
it("binds rows with much text column by column, 1, 2, 4 ... rows per statement within 100 parameters", async () => {
// Records full of quotes, which a JSON array would escape a second time: 2 of 400,000 quotes, then 19 of 10,000.
const quotes = '"'.repeat(400_000);
const { statements, database } = await groupOf(2, quotes);
const inserts = statements.filter((statement) => statement.sql.startsWith("INSERT INTO entries"));
expect(inserts.map((statement) => statement.params.length)).toEqual([10]);
const rows = await database.all<{ record: string }>("SELECT record FROM entries");
expect(rows.map((row) => (JSON.parse(row.record) as { data: string }).data === quotes)).toEqual([true, true]);
// 19 rows of 5 cells: 16 + 2 + 1, the largest power of two within 100 parameters first.
const many = (await groupOf(19, '"'.repeat(10_000))).statements;
const bound = many.filter((statement) => statement.sql.startsWith("INSERT INTO entries"));
expect(bound.map((statement) => statement.params.length)).toEqual([80, 10, 5]);
});
it("reads and writes through indexes: no statement of a call scans a stored table", async () => {
const database = await openDatabase();
const recording = recordStatements(database);
const storage = await openStorage(recording.db);
const conversationId = ROOT_CONVERSATION_ID;
const record = (id: DocumentId): DocumentCreate => ({
id,
kind: "plan",
scope: { kind: "conversation", conversationId },
history: "latest",
fork: "current",
});
const first = await storage.mintId<DocumentId>();
const second = await storage.mintId<DocumentId>();
await storage.commit(
[
[{ type: "conversation", value: { id: conversationId } }],
[
{
type: "document.create",
record: record(first),
content: { kind: "base", version: 1, value: { n: 0 } },
},
],
],
() => {},
context,
);
const before = recording.ran.length;
// Every read and write the batch path has: IDs, documents, versions, addresses, the upsert, revision drops.
await storage.commit(
[
[{ type: "document.change", id: first, content: { kind: "delta", version: 1, ops: [["s", ["n"], 1]] } }],
[
{ type: "document.change", id: first, content: { kind: "base", version: 1, value: { n: 2 } } },
{ type: "entry", value: { id: await storage.mintId<EntryId>(), conversationId, kind: "note" } },
],
[
{ type: "document.retire", id: first },
{ type: "document.create", record: record(second), content: { kind: "base", version: 1, value: {} } },
],
],
() => {},
context,
);
const batched = recording.ran.slice(before).filter((statement) => statement.sql.includes("json_each"));
expect(batched.length).toBeGreaterThan(5);
for (const { sql, params } of batched) {
const plan = await database.all<{ detail: string }>(`EXPLAIN QUERY PLAN ${sql}`, ...params);
const scans = plan.filter((row) => row.detail.startsWith("SCAN") && !row.detail.includes("VIRTUAL TABLE"));
expect(scans, sql).toEqual([]);
}
});
});
describe("SqliteStorage batch commit regressions", () => {
for (const order of ["creation first", "retirement first"] as const) {
it(`refuses a third incarnation after a replacement in an earlier batch of the call (${order})`, async () => {
// Review finding: retiring A after creating B in one batch must leave B the address's current incarnation.
const outcomes = [];
for (const storage of [new MemoryStorage(), await openStorage(await openDatabase())]) {
openStorages.add(storage);
const id = (n: number) => idFromNumber<DocumentId>(n);
const at = (n: number): DocumentCreate => ({ id: id(n), kind: "slot", scope: { kind: "session" } });
const create = (n: number): StorageWrite => ({
type: "document.create",
record: at(n),
content: { kind: "base", version: 1, value: {} },
});
await storage.commit([[create(10)]], () => {}, context);
const replace: StorageWrite[] = [create(11), { type: "document.retire", id: id(10) }];
const result = await storage.commit(
[order === "creation first" ? replace : replace.reverse(), [create(12)]],
() => {},
context,
);
outcomes.push({ stored: result.stored, refused: String(result.refused) });
}
expect(outcomes[1]).toEqual({
stored: 1,
refused: "Error: Document address already has a current incarnation",
});
expect(outcomes[1]).toEqual(outcomes[0]);
});
}
});

159
packages/durable/test/sqlite-storage.test.ts
View file

@ -5,7 +5,7 @@ import { DatabaseSync } from "node:sqlite";
import type { Context } from "@earendil-works/chord/context";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import type { JsonValue } from "@earendil-works/chord/json";
import type { SqliteDatabase } from "@earendil-works/chord/sqlite";
import type { SqliteDatabase, SqliteValue } from "@earendil-works/chord/sqlite";
import { registerStorageConformance } from "@earendil-works/pi-durable/testing";
import { afterEach, describe, expect, it } from "vitest";
import { idFromNumber } from "../src/ids.ts";
@ -691,46 +691,92 @@ describe("Pico SqliteStorage", () => {
});
describe("SqliteStorage batches", () => {
/** `database` whose transactions' `get` and `all` calls fail where `fail` returns a failure for the call. */
function faultyDatabase(
database: SqliteDatabase,
fail: (method: "get" | "all", sql: string) => (() => Promise<never>) | undefined,
): SqliteDatabase {
return {
exec: (sql) => database.exec(sql),
run: (sql, ...params) => database.run(sql, ...params),
runCounted: (sql, ...params) => database.runCounted(sql, ...params),
get: (sql, ...params) => database.get(sql, ...params),
all: (sql, ...params) => database.all(sql, ...params),
close: () => database.close(),
transaction: (callback) =>
database.transaction((transaction) =>
callback({
exec: (sql) => transaction.exec(sql),
run: (sql, ...params) => transaction.run(sql, ...params),
runCounted: (sql, ...params) => transaction.runCounted(sql, ...params),
get: <T extends object>(sql: string, ...params: SqliteValue[]) => {
const failure = fail("get", sql);
return failure === undefined ? transaction.get<T>(sql, ...params) : failure();
},
all: <T extends object>(sql: string, ...params: SqliteValue[]) => {
const failure = fail("all", sql);
return failure === undefined ? transaction.all<T>(sql, ...params) : failure();
},
}),
),
};
}
async function openDatabase(): Promise<SqliteDatabase> {
const directory = await mkdtemp(join(tmpdir(), "pi-durable-sqlite-"));
tempDirectories.add(directory);
return openNodeSqliteDatabase(join(directory, "storage.sqlite"));
}
for (const mode of ["undefined rejection", "synchronous throw"] as const) {
it(`rejects the whole call, the batches before it included, when a later batch's check read fails with a ${mode}`, async () => {
// Sol's: the first batch passes its checks and is written; the second batch's check read fails.
const directory = await mkdtemp(join(tmpdir(), "pi-durable-sqlite-"));
tempDirectories.add(directory);
const database = await openNodeSqliteDatabase(join(directory, "storage.sqlite"));
let failCheck = false;
let reads = 0;
const faulty: SqliteDatabase = {
exec: (sql) => database.exec(sql),
run: (sql, ...params) => database.run(sql, ...params),
runCounted: (sql, ...params) => database.runCounted(sql, ...params),
get: (sql, ...params) => database.get(sql, ...params),
all: (sql, ...params) => database.all(sql, ...params),
close: () => database.close(),
transaction: (callback) =>
database.transaction((transaction) =>
callback({
exec: (sql) => transaction.exec(sql),
run: (sql, ...params) => transaction.run(sql, ...params),
runCounted: (sql, ...params) => transaction.runCounted(sql, ...params),
all: (sql, ...params) => transaction.all(sql, ...params),
get: (sql, ...params) => {
if (failCheck && sql.includes("record_type") && ++reads === 2) {
if (mode === "synchronous throw") throw new Error("sync driver failure");
return Promise.reject(undefined);
}
return transaction.get(sql, ...params);
},
}),
),
};
const storage = await SqliteStorageClass.open(faulty);
it(`rejects the whole call, the batches written before it included, when a copy's source read fails with a ${mode}`, async () => {
// The first batch passes its checks; the second batch's copy reads its source from the database, so the first
// is written before that read, which fails.
let failCopy = false;
const storage = await SqliteStorageClass.open(
faultyDatabase(await openDatabase(), (method, sql) => {
if (!failCopy || method !== "get" || !sql.includes("SELECT record FROM documents WHERE id = ?")) return;
return () => {
if (mode === "synchronous throw") throw new Error("sync driver failure");
return Promise.reject(undefined);
};
}),
);
openStorages.add(storage);
await commitWrites(storage, [{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], context);
const ids = [await storage.mintId<EntryId>(), await storage.mintId<EntryId>()];
const batches: StorageWrite[][] = ids.map((id) => [
{ type: "entry", value: { id, conversationId: ROOT_CONVERSATION_ID, kind: "note" } },
]);
failCheck = true;
const record = (id: DocumentId, conversationId: ConversationId): DocumentCreate => ({
id,
kind: "copied",
scope: { kind: "conversation", conversationId },
history: "rewindable",
fork: "asOf",
});
const sourceId = await storage.mintId<DocumentId>();
await commitWrites(
storage,
[
{
type: "document.create",
record: record(sourceId, ROOT_CONVERSATION_ID),
content: { kind: "base", version: 1, value: { n: 1 } },
},
],
context,
);
const entryId = await storage.mintId<EntryId>();
const childId = await storage.mintId<ConversationId>();
const batches: StorageWrite[][] = [
[{ type: "entry", value: { id: entryId, conversationId: ROOT_CONVERSATION_ID, kind: "note" } }],
[
{ type: "conversation", value: { id: childId } },
{
type: "document.copy",
record: record(await storage.mintId<DocumentId>(), childId),
source: { id: sourceId, at: "current" },
},
],
];
failCopy = true;
const receipts: number[] = [];
const outcome = await storage
.commit(batches, (index) => receipts.push(index), context)
@ -738,10 +784,11 @@ describe("SqliteStorage batches", () => {
() => "resolved",
() => "rejected",
);
failCheck = false;
failCopy = false;
expect(outcome).toBe("rejected");
expect(receipts).toEqual([]);
expect(await storage.entry(ids[0]!, context)).toBeUndefined();
expect(await storage.entry(entryId, context)).toBeUndefined();
expect(await storage.conversation(childId, context)).toBeUndefined();
});
}
@ -756,33 +803,13 @@ describe("SqliteStorage batches", () => {
};
for (const [name, fail] of Object.entries(failures)) {
it(`rejects a batch call whose check read fails in the database with ${name}, and refuses only for its own checks`, async () => {
const directory = await mkdtemp(join(tmpdir(), "pi-durable-sqlite-"));
tempDirectories.add(directory);
const database = await openNodeSqliteDatabase(join(directory, "storage.sqlite"));
// A driver failure in a check's read, armed for the next batch call.
// A driver failure in the checks' read of record types, armed for the next batch call.
let failCheck = false;
const faulty: SqliteDatabase = {
exec: (sql) => database.exec(sql),
run: (sql, ...params) => database.run(sql, ...params),
runCounted: (sql, ...params) => database.runCounted(sql, ...params),
get: (sql, ...params) => database.get(sql, ...params),
all: (sql, ...params) => database.all(sql, ...params),
close: () => database.close(),
transaction: (callback) =>
database.transaction((transaction) =>
callback({
exec: (sql) => transaction.exec(sql),
run: (sql, ...params) => transaction.run(sql, ...params),
runCounted: (sql, ...params) => transaction.runCounted(sql, ...params),
all: (sql, ...params) => transaction.all(sql, ...params),
get: (sql, ...params) => {
if (failCheck && sql.includes("record_type")) return fail();
return transaction.get(sql, ...params);
},
}),
),
};
const storage = await SqliteStorageClass.open(faulty);
const storage = await SqliteStorageClass.open(
faultyDatabase(await openDatabase(), (_method, sql) =>
failCheck && sql.includes("record_type") ? fail : undefined,
),
);
openStorages.add(storage);
await commitWrites(storage, [{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], context);
const first = await storage.mintId<EntryId>();