From 4c25119f5432dbaa4942ad329642aa0cfecae385 Mon Sep 17 00:00:00 2001 From: Armin Ronacher Date: Wed, 21 Oct 2026 14:26:19 +0200 Subject: [PATCH] feat(durable): replicate storages as a commit log Every built-in storage implements ReplayableStorage: head(), replay() at an explicit sequence, and export() of the current state as a log that replays to the same reads. SQLite schema version 3 adds sequence indexes so export streams. @earendil-works/pi-durable/sync adds ReplicatingStorage, which appends each committed batch to an outbox with fork copies resolved, compactRecords(), and versioned JSON log lines. --- packages/durable/CHANGELOG.md | 4 +- packages/durable/README.md | 18 ++ packages/durable/docs/spec.md | 27 +++ packages/durable/package.json | 6 + packages/durable/src/index.ts | 3 + packages/durable/src/storage/export.ts | 135 +++++++++++ packages/durable/src/storage/indexeddb.ts | 96 +++++++- packages/durable/src/storage/jsonl/storage.ts | 31 ++- packages/durable/src/storage/memory.ts | 59 ++++- .../durable/src/storage/sqlite/migrations.ts | 8 + packages/durable/src/storage/sqlite/schema.ts | 4 + .../durable/src/storage/sqlite/storage.ts | 173 +++++++++++++- packages/durable/src/sync/compact.ts | 127 +++++++++++ packages/durable/src/sync/index.ts | 3 + packages/durable/src/sync/log.ts | 101 +++++++++ packages/durable/src/sync/replicating.ts | 157 +++++++++++++ packages/durable/src/types.ts | 31 +++ packages/durable/test-workers/storage.test.ts | 4 +- packages/durable/test/replay-support.ts | 214 ++++++++++++++++++ .../durable/test/sqlite-cloudflare.test.ts | 2 +- .../durable/test/sqlite-migrations.test.ts | 2 +- packages/durable/test/storage-replay.test.ts | 198 ++++++++++++++++ packages/durable/test/sync.test.ts | 211 +++++++++++++++++ scripts/check-entry-graphs.mjs | 5 +- 24 files changed, 1586 insertions(+), 33 deletions(-) create mode 100644 packages/durable/src/storage/export.ts create mode 100644 packages/durable/src/sync/compact.ts create mode 100644 packages/durable/src/sync/index.ts create mode 100644 packages/durable/src/sync/log.ts create mode 100644 packages/durable/src/sync/replicating.ts create mode 100644 packages/durable/test/replay-support.ts create mode 100644 packages/durable/test/storage-replay.test.ts create mode 100644 packages/durable/test/sync.test.ts diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 900351d0b..ca1c6c9e4 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -6,13 +6,15 @@ - `TaskRuntime.sleep()` is removed: it waited inside a phase, which holds an invocation and, after a restart, reruns the phase and restarts the wait. A task now ends its phase asleep with `runtime.sleepUntil(until, checkpoint, context)` or by returning `{ status: "sleeping", until, checkpoint }` from `runtime.commit()`; move the work after the wait into the next phase and carry what it needs in the checkpoint. `TaskRuntime.setTimer()` paces work inside a phase without waiting for it. - Tasks have a `sleeping` status, so code that handles every `TaskState`, `TaskGraphState`, or `TaskInspection` state needs a case for it. -- Opening a SQLite storage migrates it to schema version 2 automatically, rebuilding its task table. A migrated database cannot be opened by 1.3.0 or earlier, so a deployment cannot roll back past this release. JSONL storage needs no migration, but 1.3.0 or earlier never resumes a sleeping task it finds (a run's conversation stays busy). +- Opening a SQLite storage migrates it to schema version 3 automatically, rebuilding its task table and adding sequence indexes for log export. A migrated database cannot be opened by 1.3.0 or earlier, so a deployment cannot roll back past this release. JSONL storage needs no migration, but 1.3.0 or earlier never resumes a sleeping task it finds (a run's conversation stays busy). - `TaskRuntime` has `sleepUntil()` and `setTimer()`, and `ToolExecutionApi` has `now()`: code that builds either object by hand, such as a test fake or a tool wrapper that does not spread the original, must add them. - A task cut off by a crash 5 times in a row without committing a new state is parked instead of run again (see the crash-loop guard under Added); set `settings.restartLimit` to `Infinity` to keep the old behavior. `TaskInspection`'s blocked reasons include `restart_limit`. - `defineDoc()` and `defineDocFamily()` require an explicit `checkpointWhen` predicate on every definition; there is no default. Use `() => false` to keep ordinary changes as deltas, or supply a policy that periodically writes a base. ### Added +- Storages can be replicated as a commit log: every built-in storage implements `ReplayableStorage`, with `head()`, `replay(record)` (a commit at an explicit sequence, gaps allowed), and `export()` (the current state as a log that replays to the same reads, historical document reads included; SQLite streams it through sequence indexes). `LogRecord` and `LogWrite` describe the log; fork copies appear as the creations they became. +- `@earendil-works/pi-durable/sync`: `ReplicatingStorage` appends every committed batch to an `Outbox` after it commits, with fork copies resolved and current-only documents marked; `compactRecords()` drops superseded task records and document content before a newer base; `encodeLogLine()` and `decodeLogLines()` read and write versioned JSON lines and report a torn last line. - Added `IndexedDbStorage` (`@earendil-works/pi-durable/storage/indexeddb`), a browser storage backed by IndexedDB. - `HarnessOptions.telemetry` emits the content-free `DURABLE_TELEMETRY_SCHEMA`: public submission admission spans, task phase spans (including extension tasks), committed entry and submission events, running phase-change checkpoints, and nested-tool replay events. Optional propagation stores opaque carriers on submissions and tasks so queued inputs and reopened tasks retain their trace; `TaskOptions.trace` accepts a carrier override or `"root"` (absence inherits the creator's carrier); creation-time `links` connect inputs sharing a run, while steer admissions link to the running trace. Internal admissions emit events without a submit span. Optional metrics record phase duration and live task counts. Failed phases record fixed error-status categories (`task_faulted`, `task_failed`, `task_orphaned`) and diagnostics capped at 500 characters; aborted phases keep status `ok`. Every `onReport` report also emits a fixed-field `pi.durable.report` event on its current span (dropped by the NOOP span when there is none), plus a `pi.durable.reports` counter by type; report telemetry contains no error payload and leaves the callback unchanged. Clock failures report without a phase context. Generation and compaction forward their phase context to pi-ai requests. - `settings.compaction.modelOverrides` sets compaction thresholds per model, keyed by exact `"provider/modelId"`, so one harness can serve small- and large-window models; the threshold check, overflow recovery, and the compaction task use the entry of the model they act for ([#10942](https://github.com/earendil-works/pi/issues/10942)). diff --git a/packages/durable/README.md b/packages/durable/README.md index 71aa68376..7e759885d 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -798,6 +798,24 @@ await database.transaction(async (transaction) => { }); ``` +### Replication + +Every built-in storage is a `ReplayableStorage`: `export()` turns its state into a log of `{ seq, writes }` records +that replays into any other storage with the same reads, `replay(record)` writes one record at its sequence, and +`head()` is the last sequence. `ReplicatingStorage` from `@earendil-works/pi-durable/sync` wraps a storage and appends +each committed batch to an `Outbox` as such a record, for a host that ships it elsewhere: + +```typescript +const outbox = new MemoryOutbox(); // or a host outbox that persists records +const storage = new ReplicatingStorage(await openNodeSqliteStorage(file), outbox); +const harness = await Harness.open(storage, options, context); +// Elsewhere: for (const record of records) await replica.replay(record, context); +``` + +`compactRecords(records)` drops what later records in the same run supersede (earlier task records, document content +before a newer base) before the records are shipped. Only the device that drives a session opens a Harness on it: +`Harness.open` commits, so a replica that keeps replaying records must be read through its storage. + Custom backends can run the shared conformance suite with any Vitest- or Jest-compatible runner: ```typescript diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index 3e206cf8e..1de2d948e 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -5279,6 +5279,33 @@ a stored version boundary; a version transition must be a base. The semantic conformance suite covers memory, SQLite, and JSONL. +Every built-in backend is also a `ReplayableStorage`: + +```ts +type LogWrite = Exclude; +type LogRecord = { readonly seq: Seq; readonly writes: readonly LogWrite[] }; + +interface ReplayableStorage extends Storage { + head(context: Context): Promise; + replay(record: LogRecord, context: Context): Promise; + export(context: Context): AsyncIterable; +} +``` + +`replay()` is `commit()` at an explicit sequence above the head; later commits +continue after it. A log never contains a copy: a copy is logged as the +creation it became, with its materialized base, so replay never reads a source. +`export()` writes the current state as a log in sequence order that, replayed +into an empty backend, answers every read the same, historical document reads +included. Documents keep their stamps: creation at `createdAt`, every retained +revision at its sequence, retirement at `retiredAt`. A current-only document +whose creation base was reclaimed is created with the placeholder base +`{ version: 1, value: {} }`, which no read can reach. Conversations, tasks, and +submissions carry no sequence and go into the first record; Storage checks no +references between records, and nothing reads them at a past point. The last +record is at the head, empty when nothing else is there. A commit while an +export is consumed makes it reject. + ### 10.1 Failures One rule per kind of error, so a host knows what happens without reading code. diff --git a/packages/durable/package.json b/packages/durable/package.json index c15e96ad4..90ac8ea16 100644 --- a/packages/durable/package.json +++ b/packages/durable/package.json @@ -60,6 +60,12 @@ "import": "./dist/images/cloudflare.js", "default": "./dist/images/cloudflare.js" }, + "./sync": { + "source": "./src/sync/index.ts", + "types": "./dist/sync/index.d.ts", + "import": "./dist/sync/index.js", + "default": "./dist/sync/index.js" + }, "./storage/memory": { "source": "./src/storage/memory.ts", "types": "./dist/storage/memory.d.ts", diff --git a/packages/durable/src/index.ts b/packages/durable/src/index.ts index 95d9606ae..cb6caa1c8 100644 --- a/packages/durable/src/index.ts +++ b/packages/durable/src/index.ts @@ -168,9 +168,12 @@ export type { JoinPolicy, JsonObject, LatestConversationSemantics, + LogRecord, + LogWrite, NextTaskState, Page, PhaseHandler, + ReplayableStorage, ReportFrom, ReportInfo, ReportTask, diff --git a/packages/durable/src/storage/export.ts b/packages/durable/src/storage/export.ts new file mode 100644 index 000000000..9c9d1a13b --- /dev/null +++ b/packages/durable/src/storage/export.ts @@ -0,0 +1,135 @@ +import type { + DocumentContent, + DocumentCreate, + DocumentId, + DocumentRecord, + LogRecord, + LogWrite, + Seq, + StorageWrite, +} from "../types.ts"; + +/** Whether a document keeps only its newest base: session, task, and `history: "latest"` conversation documents. */ +export const isCurrentOnly = (record: DocumentCreate | DocumentRecord): boolean => + record.scope.kind !== "conversation" || record.history === "latest"; + +/** Reject a log record holding a fork copy: a log carries copies as the creations they became. */ +export function assertLogRecord(record: LogRecord): void { + if (record.writes.some((write) => (write as StorageWrite).type === "document.copy")) { + throw new Error("A log record cannot contain a document copy"); + } +} + +/** One write of an export at the sequence it belongs to. */ +export type SequencedWrite = { readonly seq: Seq; readonly write: LogWrite }; + +/** A stored revision of a document with the sequence that wrote it. */ +export type StoredRevision = DocumentContent & { readonly seq: Seq }; + +/** + * Content of a document creation whose own base was reclaimed: a current-only document keeps only its newest base, + * and nothing reads its content before that base. Export, compaction, and JSONL recovery write it. + */ +export const RECLAIMED_BASE: Extract = Object.freeze({ + kind: "base", + version: 1, + value: Object.freeze({}), +}); + +/** The creation record a document was committed with. */ +export function documentCreate(record: DocumentRecord): DocumentCreate { + const { createdAt: _createdAt, retiredAt: _retiredAt, ...create } = record; + return create as DocumentCreate; +} + +/** + * The creation of a document at `createdAt`, carrying the base stored with it, or a placeholder when that base was + * reclaimed. `atCreation` is the stored revision at `createdAt`, if any. + */ +export function creationWrite(record: DocumentRecord, atCreation: StoredRevision | undefined): SequencedWrite { + if (atCreation !== undefined && atCreation.kind !== "base") { + throw new Error(`Document ${record.id} has a delta at its creation`); + } + const content: Extract = + atCreation === undefined + ? RECLAIMED_BASE + : { kind: "base", version: atCreation.version, value: atCreation.value }; + return { seq: record.createdAt, write: { type: "document.create", record: documentCreate(record), content } }; +} + +/** A stored revision after the creation, as the change that wrote it. */ +export function revisionWrite(id: DocumentId, revision: StoredRevision): SequencedWrite { + const content: DocumentContent = + revision.kind === "base" + ? { kind: "base", version: revision.version, value: revision.value } + : { kind: "delta", version: revision.version, ops: revision.ops }; + return { seq: revision.seq, write: { type: "document.change", id, content } }; +} + +/** Every write of one document, unordered: creation, later revisions, and retirement. */ +export function documentWrites(record: DocumentRecord, revisions: readonly StoredRevision[]): SequencedWrite[] { + const atCreation = revisions.find((revision) => revision.seq === record.createdAt); + const writes = [creationWrite(record, atCreation)]; + for (const revision of revisions) { + if (revision.seq !== record.createdAt) writes.push(revisionWrite(record.id, revision)); + } + if (record.retiredAt !== undefined) { + writes.push({ seq: record.retiredAt, write: { type: "document.retire", id: record.id } }); + } + return writes; +} + +/** Writes sorted by sequence, keeping the given order within one sequence. */ +export function sortBySeq(writes: SequencedWrite[]): SequencedWrite[] { + return writes.sort((a, b) => a.seq - b.seq); +} + +/** + * Group sequence-ordered streams into log records. Writes without a sequence (conversations, tasks, submissions) go + * into the first record: Storage checks no references between records, and nothing reads those at a past point. A + * last record at `head` follows, empty when nothing else is there, so a log continued after `head` follows on. + */ +export async function* groupExport( + head: Seq | undefined, + unsequenced: readonly LogWrite[], + streams: readonly AsyncIterable[], +): AsyncGenerator { + if (head === undefined) { + if (unsequenced.length > 0) throw new Error("Storage without commits holds records"); + return; + } + const cursors = streams.map((stream) => stream[Symbol.asyncIterator]()); + const current: (SequencedWrite | undefined)[] = []; + for (const cursor of cursors) current.push(await advance(cursor)); + + let first: LogWrite[] | undefined = [...unsequenced]; + let last: Seq | undefined; + while (true) { + let seq: Seq | undefined; + for (const item of current) if (item !== undefined && (seq === undefined || item.seq < seq)) seq = item.seq; + if (seq === undefined) break; + if (seq > head) throw new Error(`Exported write at ${seq} is past the head ${head}`); + const writes: LogWrite[] = first ?? []; + first = undefined; + for (let index = 0; index < cursors.length; index++) { + while (current[index]?.seq === seq) { + writes.push(current[index]!.write); + current[index] = await advance(cursors[index]!); + } + } + yield { seq, writes }; + last = seq; + } + if (last !== head) yield { seq: head, writes: first ?? [] }; +} + +async function advance(cursor: AsyncIterator): Promise { + const next = await cursor.next(); + if (next.done) return undefined; + return next.value; +} + +/** Iterate an array as a stream. */ +export async function* streamOf(items: Iterable): AsyncGenerator { + yield* items; +} diff --git a/packages/durable/src/storage/indexeddb.ts b/packages/durable/src/storage/indexeddb.ts index 6b2121444..7d6df5d62 100644 --- a/packages/durable/src/storage/indexeddb.ts +++ b/packages/durable/src/storage/indexeddb.ts @@ -21,10 +21,12 @@ import type { EntryRecord, Id, JsonObject, + LogRecord, + LogWrite, Page, + ReplayableStorage, ScanOrder, Seq, - Storage, StorageWrite, StoredDocument, SubmissionId, @@ -34,6 +36,16 @@ import type { TaskQuery, TaskRecord, } from "../types.ts"; +import { + assertLogRecord, + documentWrites, + groupExport, + isCurrentOnly, + type SequencedWrite, + type StoredRevision, + sortBySeq, + streamOf, +} from "./export.ts"; import { nextCursor, type ScanStart, scanStart } from "./scan.ts"; const DATABASE_VERSION = 1; @@ -313,9 +325,6 @@ const isAliveAt = (record: DocumentRecord, at: DocumentPoint): boolean => { 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 | undefined => { switch (write.type) { case "conversation": @@ -375,7 +384,7 @@ function createSchema(database: IDBDatabase): void { } /** Browser IndexedDB implementation of the durable storage contract. One owner must serialize access to each database. */ -export class IndexedDbStorage implements Storage { +export class IndexedDbStorage implements ReplayableStorage { private readonly database: IDBDatabase; private closed = false; private closing: Promise | undefined; @@ -418,13 +427,88 @@ export class IndexedDbStorage implements Storage { } async commit(writes: readonly StorageWrite[], _context: Context): Promise { + return this.commitAt(writes, undefined); + } + + async head(_context: Context): Promise { + this.assertOpen(); + return this.transaction(STORES.meta, "readonly", async (transaction) => { + const metadata = await requestResult( + transaction.objectStore(STORES.meta).get(META_KEY) as IDBRequest, + ); + if (metadata === undefined) throw new Error("Durable IndexedDB metadata is missing"); + return metadata.nextSeq === 1 ? undefined : seqFromNumber(metadata.nextSeq - 1); + }); + } + + async replay(record: LogRecord, _context: Context): Promise { + assertLogRecord(record); + await this.commitAt(record.writes, record.seq); + } + + /** Reads every store in one transaction and builds the log in memory: browser sessions are small. */ + async *export(context: Context): AsyncGenerator { + this.assertOpen(); + const snapshot = await this.transaction(ALL_STORES, "readonly", async (transaction) => { + const all = (store: string) => requestResult(transaction.objectStore(store).getAll() as IDBRequest); + const metadata = await requestResult( + transaction.objectStore(STORES.meta).get(META_KEY) as IDBRequest, + ); + if (metadata === undefined) throw new Error("Durable IndexedDB metadata is missing"); + return { + head: metadata.nextSeq === 1 ? undefined : seqFromNumber(metadata.nextSeq - 1), + conversations: await all(STORES.conversations), + tasks: await all(STORES.tasks), + submissions: await all(STORES.submissions), + entries: await all(STORES.entries), + documents: await all(STORES.documents), + revisions: await all(STORES.documentRevisions), + }; + }); + const unsequenced: LogWrite[] = [ + ...snapshot.conversations.map((row): LogWrite => ({ type: "conversation", value: row.record })), + ...snapshot.tasks.map((row): LogWrite => ({ type: "task", value: row.record })), + ...snapshot.submissions.map((row): LogWrite => ({ type: "submission", value: row.record })), + ]; + const entries = snapshot.entries.map( + (row): SequencedWrite => ({ seq: row.commitSeq, write: { type: "entry", value: row.record } }), + ); + const revisionsByDocument = new Map(); + for (const row of snapshot.revisions) { + const revision: StoredRevision = + row.kind === "base" + ? { seq: row.seq, kind: "base", version: row.version, value: row.content as JsonObject } + : { seq: row.seq, kind: "delta", version: row.version, ops: row.content as readonly Op[] }; + const revisions = revisionsByDocument.get(row.documentId); + if (revisions === undefined) revisionsByDocument.set(row.documentId, [revision]); + else revisions.push(revision); + } + const documents: SequencedWrite[] = []; + for (const row of snapshot.documents) { + documents.push(...documentWrites(row.record, revisionsByDocument.get(row.id) ?? [])); + } + const records = groupExport(snapshot.head, unsequenced, [ + streamOf(sortBySeq(entries)), + streamOf(sortBySeq(documents)), + ]); + for await (const record of records) { + if ((await this.head(context)) !== snapshot.head) throw new Error("Storage changed during export"); + yield record; + } + } + + /** Commit at `seq`, or at the next sequence when it is undefined. */ + private async commitAt(writes: readonly StorageWrite[], at: Seq | undefined): Promise { this.assertOpen(); const documentActions = this.prepareDocumentActions(writes); return this.transaction(ALL_STORES, "readwrite", async (transaction) => { const metaStore = transaction.objectStore(STORES.meta); const metadata = await requestResult(metaStore.get(META_KEY) as IDBRequest); if (metadata === undefined) throw new Error("Durable IndexedDB metadata is missing"); - const seq = seqFromNumber(metadata.nextSeq); + if (at !== undefined && (!Number.isSafeInteger(at) || at < metadata.nextSeq)) { + throw new Error(`Commit sequence ${at} does not strictly increase`); + } + const seq = at ?? seqFromNumber(metadata.nextSeq); await this.checkGlobalIds(transaction, writes); await this.checkDocumentActions(transaction, documentActions); for (const write of writes) await this.applyTableWrite(transaction, write, seq); diff --git a/packages/durable/src/storage/jsonl/storage.ts b/packages/durable/src/storage/jsonl/storage.ts index 4b03fd5be..5d2e83605 100644 --- a/packages/durable/src/storage/jsonl/storage.ts +++ b/packages/durable/src/storage/jsonl/storage.ts @@ -15,6 +15,8 @@ import type { EntryId, EntryQuery, Id, + LogRecord, + ReplayableStorage, Seq, Storage, StorageWrite, @@ -24,7 +26,8 @@ import type { TaskQuery, TaskRecord, } from "../../types.ts"; -import { MemoryStorage } from "../memory.ts"; +import { assertLogRecord, isCurrentOnly, RECLAIMED_BASE } from "../export.ts"; +import { MemoryStorage, type PreparedMemoryCommit } from "../memory.ts"; const FORMAT_VERSION = 1; const MAIN_FILE = "main.jsonl"; @@ -104,9 +107,6 @@ const isSidecarFileName = (name: string): boolean => /^(?:doc|task)-(?:0|[1-9]\d const isReclaimFileName = (name: string): boolean => /^(?:doc|task)-(?:0|[1-9]\d*)\.jsonl\.reclaim$/.test(name); -const isCurrentOnly = (record: DocumentCreate): boolean => - record.scope.kind !== "conversation" || record.history === "latest"; - const sidecarKey = (file: string, seq: Seq, ordinal: number): string => JSON.stringify([file, seq, ordinal]); const jsonLine = (value: MainMarker | SidecarRecord): string => `${JSON.stringify(value)}\n`; @@ -239,7 +239,7 @@ const parseSidecarRecord = (text: string, file: string, line: number): SidecarRe }; /** Portable JSONL implementation of the storage contract. */ -export class JsonlStorage implements Storage { +export class JsonlStorage implements ReplayableStorage { private readonly fs: FileSystem; private readonly directory: string; private readonly mainPath: string; @@ -277,7 +277,24 @@ export class JsonlStorage implements Storage { async commit(writes: readonly StorageWrite[], context: Context): Promise { this.assertUsable(); - const prepared = this.memory.prepareCommit(writes); + return this.persist(this.memory.prepareCommit(writes), context); + } + + async head(context: Context): Promise { + return this.store.head(context); + } + + async replay(record: LogRecord, context: Context): Promise { + this.assertUsable(); + assertLogRecord(record); + await this.persist(this.memory.prepareCommit(record.writes, record.seq), context); + } + + export(context: Context): AsyncIterable { + return this.store.export(context); + } + + private async persist(prepared: PreparedMemoryCommit, context: Context): Promise { const encoded = this.encodeCommit(prepared.seq, prepared.writes); const reclamations = this.planReclamations(prepared.writes, encoded); const sidecars = await Promise.all( @@ -695,7 +712,7 @@ export class JsonlStorage implements Storage { writes.push({ type: "document.create", record: operation.record, - content: reclaimed || content === undefined ? { kind: "base", version: 1, value: {} } : content, + content: reclaimed || content === undefined ? RECLAIMED_BASE : content, }); } else if (!reclaimed && content !== undefined) { writes.push({ type: "document.change", id, content }); diff --git a/packages/durable/src/storage/memory.ts b/packages/durable/src/storage/memory.ts index 7f379cc29..896a23ed5 100644 --- a/packages/durable/src/storage/memory.ts +++ b/packages/durable/src/storage/memory.ts @@ -19,10 +19,12 @@ import type { EntryRecord, Id, JsonObject, + LogRecord, + LogWrite, Page, + ReplayableStorage, ScanOrder, Seq, - Storage, StorageWrite, StoredDocument, SubmissionId, @@ -32,6 +34,15 @@ import type { TaskQuery, TaskRecord, } from "../types.ts"; +import { + assertLogRecord, + documentWrites, + groupExport, + isCurrentOnly, + type SequencedWrite, + sortBySeq, + streamOf, +} from "./export.ts"; import { nextCursor, type ScanStart, scanStart } from "./scan.ts"; type StoredTask = TaskRecord; @@ -194,9 +205,6 @@ const isAliveAt = (record: DocumentRecord, at: DocumentPoint): boolean => { return record.createdAt <= at && (record.retiredAt === undefined || at < record.retiredAt); }; -const isCurrentOnly = (record: DocumentRecord): boolean => - record.scope.kind !== "conversation" || record.history === "latest"; - const tableContaining = (state: State, id: Id): TableName | undefined => state.recordTypes.get(id); const page = }>( @@ -234,7 +242,7 @@ export interface PreparedMemoryCommit { * Reads and retained writes are cloned intentionally to match the ownership boundary * of serialization-backed stores. This is backend conformance, not validation. */ -export class MemoryStorage implements Storage { +export class MemoryStorage implements ReplayableStorage { private readonly state: State = { recordTypes: new Map(), conversations: new Map(), @@ -430,6 +438,47 @@ export class MemoryStorage implements Storage { return idFromNumber(this.nextId++); } + async head(_context: Context): Promise { + this.assertOpen(); + return this.headSeq(); + } + + async replay(record: LogRecord, _context: Context): Promise { + assertLogRecord(record); + this.prepareCommit(record.writes, record.seq).apply(); + } + + /** Builds the whole log from a consistent view when iteration starts; a later commit makes it reject. */ + async *export(_context: Context): AsyncGenerator { + this.assertOpen(); + const head = this.headSeq(); + const unsequenced: LogWrite[] = []; + for (const id of this.state.conversationIds) { + unsequenced.push({ type: "conversation", value: clone(this.state.conversations.get(id)!) }); + } + for (const id of this.state.taskIds) unsequenced.push({ type: "task", value: clone(this.state.tasks.get(id)!) }); + for (const id of this.state.submissionIds) { + unsequenced.push({ type: "submission", value: clone(this.state.submissions.get(id)!) }); + } + const entries: SequencedWrite[] = []; + for (const [id, value] of this.state.entries) { + entries.push({ seq: this.state.entryCommitSeqs.get(id)!, write: { type: "entry", value: clone(value) } }); + } + const documents: SequencedWrite[] = []; + for (const stored of this.state.documents.values()) { + documents.push(...documentWrites(clone(stored.record), clone(stored.revisions))); + } + const records = groupExport(head, unsequenced, [streamOf(sortBySeq(entries)), streamOf(sortBySeq(documents))]); + for await (const record of records) { + if (this.headSeq() !== head) throw new Error("Storage changed during export"); + yield record; + } + } + + private headSeq(): Seq | undefined { + return this.nextSeq === 1 ? undefined : seqFromNumber(this.nextSeq - 1); + } + async conversation(id: ConversationId, _context: Context): Promise { this.assertOpen(); const value = this.state.conversations.get(id); diff --git a/packages/durable/src/storage/sqlite/migrations.ts b/packages/durable/src/storage/sqlite/migrations.ts index a5ee244ae..78967d213 100644 --- a/packages/durable/src/storage/sqlite/migrations.ts +++ b/packages/durable/src/storage/sqlite/migrations.ts @@ -110,9 +110,17 @@ export function createSqliteMigrations(schema: SqliteSchema): readonly SqliteMig `CREATE INDEX ${indexes.tasksByAbortRequested} ON ${tables.tasks} (abort_requested, id)`, `CREATE INDEX ${indexes.tasksByBackground} ON ${tables.tasks} (background, id)`, ]; + // Version 3 orders rows by commit sequence, so a log export streams through the session instead of sorting it. + const sequenceIndexes: readonly string[] = [ + `CREATE INDEX ${indexes.entriesBySeq} ON ${tables.entries} (commit_seq, id)`, + `CREATE INDEX ${indexes.documentsByCreation} ON ${tables.documents} (created_at, id)`, + `CREATE INDEX ${indexes.documentsByRetirement} ON ${tables.documents} (retired_at, id) WHERE retired_at IS NOT NULL`, + `CREATE INDEX ${indexes.documentRevisionsBySeq} ON ${tables.documentRevisions} (seq, document_id)`, + ]; return [ { version: 1, statements: initialSchema }, { version: 2, statements: sleepingStatus }, + { version: 3, statements: sequenceIndexes }, ]; } diff --git a/packages/durable/src/storage/sqlite/schema.ts b/packages/durable/src/storage/sqlite/schema.ts index b807ba529..73656433e 100644 --- a/packages/durable/src/storage/sqlite/schema.ts +++ b/packages/durable/src/storage/sqlite/schema.ts @@ -35,6 +35,10 @@ const INDEX_NAMES = { documentsByScope: "documents_by_scope", documentsByScopeKind: "documents_by_scope_kind", documentRevisionsByKind: "document_revisions_by_kind", + entriesBySeq: "entries_by_seq", + documentsByCreation: "documents_by_creation", + documentsByRetirement: "documents_by_retirement", + documentRevisionsBySeq: "document_revisions_by_seq", } as const; type SqliteIdentifiers> = { readonly [K in keyof T]: string }; diff --git a/packages/durable/src/storage/sqlite/storage.ts b/packages/durable/src/storage/sqlite/storage.ts index 7dc590382..3704f2e68 100644 --- a/packages/durable/src/storage/sqlite/storage.ts +++ b/packages/durable/src/storage/sqlite/storage.ts @@ -20,10 +20,12 @@ import type { EntryRecord, Id, JsonObject, + LogRecord, + LogWrite, Page, + ReplayableStorage, ScanOrder, Seq, - Storage, StorageWrite, StoredDocument, SubmissionId, @@ -33,6 +35,15 @@ import type { TaskQuery, TaskRecord, } from "../../types.ts"; +import { + assertLogRecord, + creationWrite, + groupExport, + isCurrentOnly, + revisionWrite, + type SequencedWrite, + type StoredRevision, +} from "../export.ts"; import { nextCursor, type ScanStart, scanStart } from "../scan.ts"; import { applySqliteMigrationsForSchema } from "./migrations.ts"; import { createSqliteSchema, type SqliteSchema } from "./schema.ts"; @@ -56,6 +67,19 @@ type DocumentAction = { content?: DocumentContent; retire: boolean; }; +/** Columns of an export scan: rows are read in pages ordered by `(k1, k2)`. */ +type KeysetRow = { readonly k1: number; readonly k2: number }; +type ExportRevisionRow = KeysetRow & { + readonly kind: DocumentContent["kind"]; + readonly version: number; + readonly content: string; +}; +type ExportCreationRow = KeysetRow & { + readonly record: string; + readonly kind: DocumentContent["kind"] | null; + readonly version: number | null; + readonly content: string | null; +}; type ScopeColumns = { readonly scopeKind: DocumentRecord["scope"]["kind"]; readonly ownerId: number; @@ -69,6 +93,15 @@ export type SqliteStorageOptions = { readonly prefix?: string; }; +const EXPORT_PAGE_SIZE = 256; + +const revisionFromRow = (row: RevisionRow): StoredRevision => { + const seq = seqFromNumber(row.seq); + return row.kind === "base" + ? { seq, kind: "base", version: row.version, value: parseJson(row.content) } + : { seq, kind: "delta", version: row.version, ops: parseJson(row.content) }; +}; + const parseJson = (value: string): T => JSON.parse(value) as T; const encodeJson = (value: unknown): string => JSON.stringify(value) as string; // Some SQLite bindings replace lone UTF-16 surrogates. JSON encoding keeps indexed identities lossless. @@ -129,9 +162,6 @@ const isAliveAt = (record: DocumentRecord, at: DocumentPoint): boolean => { 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 | undefined => { switch (write.type) { case "conversation": @@ -149,7 +179,7 @@ const writeId = (write: StorageWrite): Id | undefined => { }; /** Portable SQLite implementation of the Pico storage contract. */ -export class SqliteStorage implements Storage { +export class SqliteStorage implements ReplayableStorage { private readonly db: SqliteDatabase; private readonly schema: SqliteSchema; private nextId: number; @@ -185,15 +215,142 @@ export class SqliteStorage implements Storage { } async commit(writes: readonly StorageWrite[], _context: Context): Promise { + return this.commitAt(writes, undefined); + } + + async head(_context: Context): Promise { + this.assertOpen(); + return this.readHead(); + } + + async replay(record: LogRecord, _context: Context): Promise { + assertLogRecord(record); + await this.commitAt(record.writes, record.seq); + } + + /** + * Streams in pages ordered by sequence through indexes, so the session is never held in memory, except the records + * without a sequence (conversations, tasks, submissions), which the first log record holds. + */ + async *export(_context: Context): AsyncGenerator { + this.assertOpen(); + const head = await this.readHead(); + const unchanged = async (): Promise => { + this.assertOpen(); + if ((await this.readHead()) !== head) throw new Error("Storage changed during export"); + }; + const { tables } = this.schema; + const unsequenced: LogWrite[] = []; + for await (const row of this.keyset( + `SELECT id AS k1, 0 AS k2, record FROM ${tables.conversations} WHERE (id, 0) > (?, ?) ORDER BY id LIMIT ?`, + unchanged, + )) { + unsequenced.push({ type: "conversation", value: parseJson(row.record) }); + } + for await (const row of this.keyset( + `SELECT id AS k1, 0 AS k2, record FROM ${tables.tasks} WHERE (id, 0) > (?, ?) ORDER BY id LIMIT ?`, + unchanged, + )) { + unsequenced.push({ type: "task", value: parseJson(row.record) }); + } + for await (const row of this.keyset( + `SELECT id AS k1, 0 AS k2, record FROM ${tables.submissions} WHERE (id, 0) > (?, ?) ORDER BY id LIMIT ?`, + unchanged, + )) { + unsequenced.push({ type: "submission", value: parseJson(row.record) }); + } + const entries = this.mapRows( + this.keyset( + `SELECT commit_seq AS k1, id AS k2, record FROM ${tables.entries} + WHERE (commit_seq, id) > (?, ?) ORDER BY commit_seq, id LIMIT ?`, + unchanged, + ), + (row): SequencedWrite => ({ + seq: seqFromNumber(row.k1), + write: { type: "entry", value: parseJson(row.record) }, + }), + ); + const creations = this.mapRows( + this.keyset( + `SELECT d.created_at AS k1, d.id AS k2, d.record, r.kind, r.version, r.content + FROM ${tables.documents} d + LEFT JOIN ${tables.documentRevisions} r ON r.document_id = d.id AND r.seq = d.created_at + WHERE (d.created_at, d.id) > (?, ?) ORDER BY d.created_at, d.id LIMIT ?`, + unchanged, + ), + (row) => + creationWrite( + parseJson(row.record), + row.kind === null + ? undefined + : revisionFromRow({ seq: row.k1, kind: row.kind, version: row.version!, content: row.content! }), + ), + ); + const revisions = this.mapRows( + this.keyset( + `SELECT r.seq AS k1, r.document_id AS k2, r.kind, r.version, r.content + FROM ${tables.documentRevisions} r JOIN ${tables.documents} d ON d.id = r.document_id + WHERE (r.seq, r.document_id) > (?, ?) AND r.seq <> d.created_at + ORDER BY r.seq, r.document_id LIMIT ?`, + unchanged, + ), + (row) => revisionWrite(idFromNumber(row.k2), revisionFromRow({ ...row, seq: row.k1 })), + ); + const retirements = this.mapRows( + this.keyset( + `SELECT retired_at AS k1, id AS k2 FROM ${tables.documents} + WHERE retired_at IS NOT NULL AND (retired_at, id) > (?, ?) ORDER BY retired_at, id LIMIT ?`, + unchanged, + ), + (row): SequencedWrite => ({ + seq: seqFromNumber(row.k1), + write: { type: "document.retire", id: idFromNumber(row.k2) }, + }), + ); + for await (const record of groupExport(head, unsequenced, [entries, creations, revisions, retirements])) { + await unchanged(); + yield record; + } + } + + /** Rows of `sql` in `(k1, k2)` order, a page at a time; `sql` binds the last key pair and the page size. */ + private async *keyset(sql: string, unchanged: () => Promise): AsyncGenerator { + let after: KeysetRow = { k1: -1, k2: -1 }; + while (true) { + const rows = await this.db.all(sql, after.k1, after.k2, EXPORT_PAGE_SIZE); + await unchanged(); + yield* rows; + if (rows.length < EXPORT_PAGE_SIZE) return; + after = rows.at(-1)!; + } + } + + private async *mapRows(rows: AsyncIterable, map: (row: R) => T): AsyncGenerator { + for await (const row of rows) yield map(row); + } + + private async readHead(): Promise { + const metadata = await this.db.get( + `SELECT next_id, next_seq FROM ${this.schema.tables.durableMetadata} WHERE singleton = 1`, + ); + if (metadata === undefined) throw new Error("Durable SQLite metadata is missing"); + return metadata.next_seq === 1 ? undefined : seqFromNumber(metadata.next_seq - 1); + } + + /** Commit at `seq`, or at the next sequence when it is undefined. */ + private async commitAt(writes: readonly StorageWrite[], seq: Seq | undefined): Promise { this.assertOpen(); const documentActions = this.prepareDocumentActions(writes); const candidateNextId = this.candidateNextId(writes); - const seq = await this.db.transaction(async (transaction) => { + const committed = await this.db.transaction(async (transaction) => { const metadata = await transaction.get( `SELECT next_id, next_seq FROM ${this.schema.tables.durableMetadata} WHERE singleton = 1`, ); if (metadata === undefined) throw new Error("Durable SQLite metadata is missing"); - const committedSeq = seqFromNumber(metadata.next_seq); + if (seq !== undefined && (!Number.isSafeInteger(seq) || seq < metadata.next_seq)) { + throw new Error(`Commit sequence ${seq} does not strictly increase`); + } + const committedSeq = seq ?? seqFromNumber(metadata.next_seq); await this.checkGlobalIds(transaction, writes); await this.checkDocumentActions(transaction, documentActions); for (const write of writes) await this.applyTableWrite(transaction, write, committedSeq); @@ -206,7 +363,7 @@ export class SqliteStorage implements Storage { return committedSeq; }); this.nextId = Math.max(this.nextId, candidateNextId); - return seq; + return committed; } async mintId>(): Promise { diff --git a/packages/durable/src/sync/compact.ts b/packages/durable/src/sync/compact.ts new file mode 100644 index 000000000..1a459c24f --- /dev/null +++ b/packages/durable/src/sync/compact.ts @@ -0,0 +1,127 @@ +import { isCurrentOnly, RECLAIMED_BASE } from "../storage/export.ts"; +import type { DocumentId, Id, LogWrite } from "../types.ts"; +import type { OutboxRecord } from "./log.ts"; + +/** Whether a document keeps only its newest base; `undefined` when unknown, which keeps all of its content. */ +export type CurrentOnlyLookup = (id: DocumentId) => boolean | undefined; + +type Position = { readonly record: number; readonly write: number }; + +const before = (a: Position, b: Position): boolean => + a.record < b.record || (a.record === b.record && a.write < b.write); + +type DocumentWrites = { + create?: Position; + /** Content writes in order: the creation's base and every change. */ + content: { readonly position: Position; readonly base: boolean }[]; + retired: boolean; +}; + +/** + * Compact a run of consecutive records into the records that replay to the same reads: + * + * - entries and conversations stay; + * - of each task and submission only the last record stays, at the sequence that wrote it; + * - a current-only document loses content before its newest base in the run, and all of it when the run retires it; + * a creation whose base is dropped keeps an empty placeholder base, as export writes it; + * - rewindable and unknown documents keep everything; + * - records left without writes are dropped, except the last, which marks where the run ends. + * + * Pure: the same rules compact one flush, merge pending batches offline, and fold uploaded batches remotely. + */ +export function compactRecords( + records: readonly OutboxRecord[], + lookup: CurrentOnlyLookup = () => undefined, +): OutboxRecord[] { + if (records.length === 0) return []; + const currentOnly = new Map(); + const lastRecord = new Map, Position>(); + const documents = new Map(); + const documentWrites = (id: DocumentId): DocumentWrites => { + let writes = documents.get(id); + if (writes === undefined) { + writes = { content: [], retired: false }; + documents.set(id, writes); + } + return writes; + }; + + records.forEach((record, recordIndex) => { + for (const id of record.currentOnly ?? []) currentOnly.set(id, true); + record.writes.forEach((write, writeIndex) => { + const position = { record: recordIndex, write: writeIndex }; + switch (write.type) { + case "task": + case "submission": + lastRecord.set(write.value.id, position); + break; + case "document.create": { + currentOnly.set(write.record.id, isCurrentOnly(write.record)); + const writes = documentWrites(write.record.id); + writes.create = position; + writes.content.push({ position, base: true }); + break; + } + case "document.change": + documentWrites(write.id).content.push({ position, base: write.content.kind === "base" }); + break; + case "document.retire": + documentWrites(write.id).retired = true; + break; + case "conversation": + case "entry": + break; + } + }); + }); + + /** Content positions to drop, and creations whose base becomes the placeholder. */ + const dropped = new Set(); + const placeholders = new Set(); + const key = (position: Position): string => `${position.record}:${position.write}`; + for (const [id, writes] of documents) { + if (!(currentOnly.get(id) ?? lookup(id) ?? false)) continue; + let keepFrom: Position | undefined; + if (!writes.retired) { + for (const content of writes.content) if (content.base) keepFrom = content.position; + // Without a base in the run, its deltas build on content before the run. + if (keepFrom === undefined) continue; + } + for (const content of writes.content) { + if (keepFrom !== undefined && !before(content.position, keepFrom)) continue; + if (writes.create !== undefined && key(content.position) === key(writes.create)) { + placeholders.add(key(content.position)); + } else { + dropped.add(key(content.position)); + } + } + } + + const compacted: OutboxRecord[] = []; + records.forEach((record, recordIndex) => { + const writes: LogWrite[] = []; + record.writes.forEach((write, writeIndex) => { + const position = { record: recordIndex, write: writeIndex }; + const positionKey = key(position); + if (write.type === "task" || write.type === "submission") { + const last = lastRecord.get(write.value.id)!; + if (last.record !== recordIndex || last.write !== writeIndex) return; + } + if (dropped.has(positionKey)) return; + if (write.type === "document.create" && placeholders.has(positionKey)) { + writes.push({ type: "document.create", record: write.record, content: RECLAIMED_BASE }); + return; + } + writes.push(write); + }); + const last = recordIndex === records.length - 1; + if (writes.length === 0 && !last) return; + const hints = (record.currentOnly ?? []).filter((id) => + writes.some( + (write) => (write.type === "document.change" || write.type === "document.retire") && write.id === id, + ), + ); + compacted.push({ seq: record.seq, writes, ...(hints.length === 0 ? {} : { currentOnly: hints }) }); + }); + return compacted; +} diff --git a/packages/durable/src/sync/index.ts b/packages/durable/src/sync/index.ts new file mode 100644 index 000000000..97db6cddc --- /dev/null +++ b/packages/durable/src/sync/index.ts @@ -0,0 +1,3 @@ +export { type CurrentOnlyLookup, compactRecords } from "./compact.ts"; +export { decodeLogLines, encodeLogLine, LOG_VERSION, LogFormatError, type OutboxRecord } from "./log.ts"; +export { MemoryOutbox, type Outbox, ReplicatingStorage, type ReplicatingStorageOptions } from "./replicating.ts"; diff --git a/packages/durable/src/sync/log.ts b/packages/durable/src/sync/log.ts new file mode 100644 index 000000000..6c0c4e292 --- /dev/null +++ b/packages/durable/src/sync/log.ts @@ -0,0 +1,101 @@ +import { seqFromNumber } from "../ids.ts"; +import type { DocumentId, LogRecord, LogWrite } from "../types.ts"; + +/** Version of the log line envelope. It changes when a `LogWrite` variant is added, removed, or changes shape. */ +export const LOG_VERSION = 1; + +/** + * A committed record as the local outbox keeps it. `currentOnly` lists the documents this record changes or retires + * without creating them whose history is current-only, which compaction needs and the writes alone do not say. + * Created documents describe themselves. + */ +export type OutboxRecord = LogRecord & { + readonly currentOnly?: readonly DocumentId[]; +}; + +type LogLine = { + readonly v: typeof LOG_VERSION; + readonly seq: number; + readonly writes: readonly LogWrite[]; + readonly currentOnly?: readonly DocumentId[]; +}; + +export class LogFormatError extends Error { + constructor(message: string, options?: ErrorOptions) { + super(message, options); + this.name = "LogFormatError"; + } +} + +/** One JSON line, terminated by a newline. */ +export function encodeLogLine(record: OutboxRecord): string { + const line: LogLine = { + v: LOG_VERSION, + seq: record.seq, + writes: record.writes, + ...(record.currentOnly === undefined || record.currentOnly.length === 0 + ? {} + : { currentOnly: record.currentOnly }), + }; + return `${JSON.stringify(line)}\n`; +} + +/** + * Decode JSON lines. A last line without its newline is a torn append and is returned as `torn` instead of a + * record; anything else that does not decode is an error. A line of an unknown version is an error too: an older pi + * must not replay writes it does not understand. + */ +export function decodeLogLines(text: string): { readonly records: OutboxRecord[]; readonly torn: number } { + const records: OutboxRecord[] = []; + let start = 0; + let previous = 0; + while (start < text.length) { + const end = text.indexOf("\n", start); + if (end === -1) return { records, torn: text.length - start }; + const record = decodeLine(text.slice(start, end), records.length + 1); + if (record.seq <= previous) + throw new LogFormatError(`Log line ${records.length + 1} does not increase the sequence`); + previous = record.seq; + records.push(record); + start = end + 1; + } + return { records, torn: 0 }; +} + +function decodeLine(text: string, number: number): OutboxRecord { + let value: unknown; + try { + value = JSON.parse(text); + } catch (error) { + throw new LogFormatError(`Log line ${number} is not JSON`, { cause: error }); + } + if (typeof value !== "object" || value === null || Array.isArray(value)) { + throw new LogFormatError(`Log line ${number} is not an object`); + } + const line = value as Partial & { readonly v?: unknown }; + if (line.v !== LOG_VERSION) throw new LogFormatError(`Log line ${number} has unknown version ${String(line.v)}`); + if (typeof line.seq !== "number" || !Number.isSafeInteger(line.seq) || line.seq < 1) { + throw new LogFormatError(`Log line ${number} has no valid sequence`); + } + if (!Array.isArray(line.writes)) throw new LogFormatError(`Log line ${number} has no writes`); + for (const write of line.writes as readonly { readonly type?: unknown }[]) { + if (!LOG_WRITE_TYPES.has(write?.type)) { + throw new LogFormatError(`Log line ${number} has a write of unknown type ${String(write?.type)}`); + } + } + return { + seq: seqFromNumber(line.seq), + writes: line.writes, + ...(line.currentOnly === undefined ? {} : { currentOnly: line.currentOnly }), + }; +} + +const LOG_WRITE_TYPES: ReadonlySet = new Set([ + "conversation", + "entry", + "task", + "submission", + "document.create", + "document.change", + "document.retire", +]); diff --git a/packages/durable/src/sync/replicating.ts b/packages/durable/src/sync/replicating.ts new file mode 100644 index 000000000..d56b46c81 --- /dev/null +++ b/packages/durable/src/sync/replicating.ts @@ -0,0 +1,157 @@ +import type { Context } from "@earendil-works/chord"; +import { isCurrentOnly } from "../storage/export.ts"; +import type { + ConversationId, + DocumentId, + EntryId, + Id, + LogRecord, + LogWrite, + ReplayableStorage, + Seq, + Storage, + StorageWrite, +} from "../types.ts"; +import type { OutboxRecord } from "./log.ts"; + +/** Where committed records wait for upload. Appends happen one at a time, in commit order. */ +export interface Outbox { + append(record: OutboxRecord, context: Context): Promise; +} + +export type ReplicatingStorageOptions = { + /** Called after each record is in the outbox, on the Session line: must not block. A flush scheduler listens here. */ + readonly onRecord?: (record: OutboxRecord) => void; +}; + +/** + * A storage that appends every committed batch to an outbox after it commits, as a log record. + * + * Fork copies are resolved into the creations they become before the commit, so the log never needs a copy's + * source. Records replayed through `replay()` came from elsewhere, such as a pull-back, and are not logged again. + * Wrapping the storage rather than observing the Session also logs what the Session writes straight to storage, such + * as the tasks a close stops. + * + * The append follows the commit: a commit that fails has no defined outcome and must not reach the log. A crash + * between the two leaves the storage ahead of the outbox, which the host detects on open by comparing heads. + */ +export class ReplicatingStorage implements ReplayableStorage { + readonly inner: ReplayableStorage; + readonly #outbox: Outbox; + readonly #onRecord: ((record: OutboxRecord) => void) | undefined; + /** History of documents seen so far, so a change knows whether it may be compacted away later. */ + readonly #currentOnly = new Map(); + + constructor(inner: ReplayableStorage, outbox: Outbox, options: ReplicatingStorageOptions = {}) { + this.inner = inner; + this.#outbox = outbox; + this.#onRecord = options.onRecord; + } + + async commit(writes: readonly StorageWrite[], context: Context): Promise { + const logged = await this.#resolveCopies(writes, context); + const currentOnly = await this.#currentOnlyChanges(logged, context); + const seq = await this.inner.commit(writes, context); + for (const write of logged) { + if (write.type === "document.create") this.#currentOnly.set(write.record.id, isCurrentOnly(write.record)); + else if (write.type === "document.retire") this.#currentOnly.delete(write.id); + } + const record: OutboxRecord = { seq, writes: logged, ...(currentOnly.length === 0 ? {} : { currentOnly }) }; + await this.#outbox.append(record, context); + this.#onRecord?.(record); + return seq; + } + + /** Write records that came from elsewhere; they are not logged again. */ + replay(record: LogRecord, context: Context): Promise { + return this.inner.replay(record, context); + } + + head(context: Context): Promise { + return this.inner.head(context); + } + + export(context: Context): AsyncIterable { + return this.inner.export(context); + } + + async #resolveCopies(writes: readonly StorageWrite[], context: Context): Promise { + const resolved: LogWrite[] = []; + for (const write of writes) { + if (write.type !== "document.copy") { + resolved.push(write); + continue; + } + // The Session line is held and a batch may not change a copy's source, so this reads what the commit copies. + const source = await this.inner.document(write.source.id, write.source.at, context); + if (source === undefined) throw new Error(`Fork source document ${write.source.id} cannot be read`); + resolved.push({ + type: "document.create", + record: write.record, + content: { kind: "base", version: source.version, value: source.value }, + }); + } + return resolved; + } + + /** Current-only documents this batch changes or retires without creating them. */ + async #currentOnlyChanges(writes: readonly LogWrite[], context: Context): Promise { + const created = new Set(); + for (const write of writes) if (write.type === "document.create") created.add(write.record.id); + const ids: DocumentId[] = []; + for (const write of writes) { + if (write.type !== "document.change" && write.type !== "document.retire") continue; + if (created.has(write.id) || ids.includes(write.id)) continue; + let currentOnly = this.#currentOnly.get(write.id); + if (currentOnly === undefined) { + // The document is alive before this commit; its record says what history it keeps. Read once per document. + const stored = await this.inner.document(write.id, "current", context); + if (stored === undefined) continue; + currentOnly = isCurrentOnly(stored.record); + this.#currentOnly.set(write.id, currentOnly); + } + if (currentOnly) ids.push(write.id); + } + return ids; + } + + mintId>(): Promise { + return this.inner.mintId(); + } + conversation: Storage["conversation"] = (id, context) => this.inner.conversation(id, context); + scanConversations: Storage["scanConversations"] = (query, limit, cursor, context) => + this.inner.scanConversations(query, limit, cursor, context); + entry(id: EntryId, context: Context): ReturnType; + entry(conversationId: ConversationId, id: EntryId, context: Context): ReturnType; + entry(idOrConversationId: EntryId | ConversationId, idOrContext: EntryId | Context, context?: Context) { + return context === undefined + ? this.inner.entry(idOrConversationId as EntryId, idOrContext as Context) + : this.inner.entry(idOrConversationId as ConversationId, idOrContext as EntryId, context); + } + findLatestHeadMarker: Storage["findLatestHeadMarker"] = (conversationId, atOrBefore, context) => + this.inner.findLatestHeadMarker(conversationId, atOrBefore, context); + scanEntries: Storage["scanEntries"] = (query, limit, cursor, context) => + this.inner.scanEntries(query, limit, cursor, context); + task: Storage["task"] = (id, context) => this.inner.task(id, context); + scanTasks: Storage["scanTasks"] = (query, limit, cursor, context) => + this.inner.scanTasks(query, limit, cursor, context); + submission: Storage["submission"] = (id, context) => this.inner.submission(id, context); + scanSubmissions: Storage["scanSubmissions"] = (query, limit, cursor, context) => + this.inner.scanSubmissions(query, limit, cursor, context); + submissionByRequest: Storage["submissionByRequest"] = (conversationId, requestId, context) => + this.inner.submissionByRequest(conversationId, requestId, context); + findDocument: Storage["findDocument"] = (address, at, context) => this.inner.findDocument(address, at, context); + document: Storage["document"] = (id, at, context) => this.inner.document(id, at, context); + scanDocuments: Storage["scanDocuments"] = (query, limit, cursor, context) => + this.inner.scanDocuments(query, limit, cursor, context); + close: Storage["close"] = (context) => this.inner.close(context); +} + +/** An outbox in memory, for tests and hosts that persist records themselves. */ +export class MemoryOutbox implements Outbox { + readonly records: OutboxRecord[] = []; + + async append(record: OutboxRecord, _context: Context): Promise { + this.records.push(record); + } +} diff --git a/packages/durable/src/types.ts b/packages/durable/src/types.ts index fe69f8e20..aa9522b24 100644 --- a/packages/durable/src/types.ts +++ b/packages/durable/src/types.ts @@ -1272,3 +1272,34 @@ export interface Storage { /** Release backend resources; all later operations must reject. */ close(context: Context): Promise; } + +/** + * A write as it appears in a commit log. Fork copies are logged as the creation they became, with the materialized + * base, so replaying a log never needs to read a copy's source. + */ +export type LogWrite = Exclude; + +/** One committed batch at its sequence. Sequences strictly increase within a log and may have gaps. */ +export type LogRecord = { + readonly seq: Seq; + readonly writes: readonly LogWrite[]; +}; + +/** Storage that can write a log at explicit sequences and turn its state back into a log. */ +export interface ReplayableStorage extends Storage { + /** Sequence of the last committed batch; `undefined` for a storage without commits. */ + head(context: Context): Promise; + + /** + * Atomically commit `record.writes` at `record.seq`, which must exceed the head. Validation, atomicity, and stamps + * are those of `commit()`; later commits continue after `record.seq`. + */ + replay(record: LogRecord, context: Context): Promise; + + /** + * The current state as a log in sequence order: replayed into an empty storage, it answers every read the same, + * historical document reads included. Its last record is at the head. No commit may run while it is consumed; one + * that does makes the export reject. + */ + export(context: Context): AsyncIterable; +} diff --git a/packages/durable/test-workers/storage.test.ts b/packages/durable/test-workers/storage.test.ts index 9f45ec600..c7dc290fd 100644 --- a/packages/durable/test-workers/storage.test.ts +++ b/packages/durable/test-workers/storage.test.ts @@ -18,7 +18,7 @@ registerStorageConformance({ describe, expect, it }, "SqliteStorage on Durable O ); describe("Durable Object SQLite schema", () => { - it("migrates a version 1 schema to admit sleeping tasks, keeping tasks and indexes", async () => { + it("migrates a version 1 schema to admit sleeping tasks, keeping tasks and indexes, then adds sequence indexes", async () => { await runInDurableObject(fresh(), async (_instance, state) => { const database = new DurableObjectSqliteDatabase(state.storage); const schema = createSqliteSchema("legacy_"); @@ -35,7 +35,7 @@ describe("Durable Object SQLite schema", () => { await expect(insert(6, "sleeping")).rejects.toThrow(); await openDurableObjectSqliteStorage(state.storage, { prefix: "legacy_" }); - expect(await database.get(`SELECT version FROM ${schema.tables.durableSchema}`)).toEqual({ version: 2 }); + expect(await database.get(`SELECT version FROM ${schema.tables.durableSchema}`)).toEqual({ version: 3 }); expect(await database.get(`SELECT status FROM ${tasks} WHERE id = 5`)).toEqual({ status: "waiting" }); await insert(6, "sleeping"); const indexes = await database.all<{ name: string }>( diff --git a/packages/durable/test/replay-support.ts b/packages/durable/test/replay-support.ts new file mode 100644 index 000000000..9cfb31aff --- /dev/null +++ b/packages/durable/test/replay-support.ts @@ -0,0 +1,214 @@ +import type { JsonValue } from "@earendil-works/chord"; +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import { idFromNumber, seqFromNumber } from "../src/ids.ts"; +import { + type ConversationId, + type DocumentCreate, + type DocumentId, + type DocumentPoint, + type DocumentRecord, + type EntryId, + type LogRecord, + type LogWrite, + type ReplayableStorage, + ROOT_CONVERSATION_ID, + type Storage, + type StorageWrite, + type SubmissionId, + type SubmissionRecord, + type TaskId, + type TaskRecord, +} from "../src/types.ts"; + +const context = BACKGROUND_CONTEXT; +type StoredTask = TaskRecord; + +export const conversationId = (id: number) => idFromNumber(id); +export const entryId = (id: number) => idFromNumber(id); +export const taskId = (id: number) => idFromNumber>(id); +export const documentId = (id: number) => idFromNumber(id); + +export function task(id: number, status: "pending" | "running" | "terminal", phase = "work"): StoredTask { + const base = { + id: taskId(id), + conversationId: ROOT_CONVERSATION_ID, + kind: "test.task", + version: 1, + input: { id }, + background: false, + abortRequested: false, + }; + return status === "terminal" + ? { ...base, state: { status, outcome: { status: "completed", result: { phase } } } } + : { ...base, state: { status, checkpoint: { phase } } }; +} + +function submission(id: number, status: "queued" | "placed", entry?: number): SubmissionRecord { + return status === "queued" + ? { id: idFromNumber(id), conversationId: ROOT_CONVERSATION_ID, type: "input", status } + : { + id: idFromNumber(id), + conversationId: ROOT_CONVERSATION_ID, + type: "input", + status, + entry: entryId(entry!), + }; +} + +export const entry = (id: number, conversation = ROOT_CONVERSATION_ID, extra: object = {}): LogWrite => ({ + type: "entry", + value: { id: entryId(id), conversationId: conversation, kind: "message", data: { id }, ...extra }, +}); + +export const sessionDoc = (id: number, kind = "test.session"): DocumentCreate => ({ + id: documentId(id), + kind, + scope: { kind: "session" }, +}); +export const rewindableDoc = ( + id: number, + conversation = ROOT_CONVERSATION_ID, + kind = "test.rewindable", +): DocumentCreate => ({ + id: documentId(id), + kind, + scope: { kind: "conversation", conversationId: conversation }, + history: "rewindable", + fork: "asOf", +}); +export const latestDoc = (id: number, conversation = ROOT_CONVERSATION_ID, kind = "test.latest"): DocumentCreate => ({ + id: documentId(id), + kind, + scope: { kind: "conversation", conversationId: conversation }, + history: "latest", + fork: "current", +}); + +export const create = (record: DocumentCreate, value: Record, version = 1): LogWrite => ({ + type: "document.create", + record, + content: { kind: "base", version, value }, +}); +export const delta = (id: number, path: string, value: JsonValue, version = 1): LogWrite => ({ + type: "document.change", + id: documentId(id), + content: { kind: "delta", version, ops: [["s", [path], value]] }, +}); +export const base = (id: number, value: Record, version = 1): LogWrite => ({ + type: "document.change", + id: documentId(id), + content: { kind: "base", version, value }, +}); + +/** + * A history that exercises everything export has to preserve: reclaimed current-only content, rewindable history + * across a version transition, task documents, fork copies, retire and recreate, an empty lifetime, upserted records, + * and a head that no stamp records. + */ +export async function seed(storage: ReplayableStorage): Promise { + const batches: StorageWrite[][] = [ + [{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], + [ + entry(2), + entry(3), + { type: "task", value: task(4, "pending") }, + { type: "submission", value: submission(5, "queued") }, + create(sessionDoc(6), { n: 0 }), + create(rewindableDoc(7), { a: 1 }), + ], + [delta(6, "n", 1), delta(7, "b", 1), { type: "task", value: task(4, "running") }, entry(8)], + [base(6, { n: 5 }), base(7, { a: 2 }, 2), { type: "submission", value: submission(5, "placed", 8) }], + [ + create({ id: documentId(9), kind: "test.task-doc", scope: { kind: "task", taskId: taskId(4) } }, { t: 1 }), + create(latestDoc(10), { x: 1 }), + ], + [ + { + type: "conversation", + value: { id: conversationId(11), parent: { conversationId: ROOT_CONVERSATION_ID, at: entryId(8) } }, + }, + { + type: "document.copy", + record: rewindableDoc(12, conversationId(11)), + source: { id: documentId(7), at: seqFromNumber(3) }, + }, + { + type: "document.copy", + record: latestDoc(13, conversationId(11)), + source: { id: documentId(10), at: "current" }, + }, + entry(14, conversationId(11), { head: entryId(14) }), + ], + [base(10, { x: 2 }), delta(7, "c", 1, 2)], + [ + { type: "task", value: task(4, "terminal") }, + delta(9, "t", 2), + { type: "document.retire", id: documentId(9) }, + delta(10, "y", 1), + ], + [ + { type: "document.retire", id: documentId(7) }, + create(rewindableDoc(15), { a: 3 }), + create(latestDoc(16, ROOT_CONVERSATION_ID, "test.flash"), { gone: true }), + { type: "document.retire", id: documentId(16) }, + delta(13, "copied", true), + ], + [{ type: "task", value: task(17, "pending") }], + ]; + for (const batch of batches) await storage.commit(batch, context); +} + +type Reads = Record; + +/** Every read a Session can make, at every point of the history, as one comparable value. */ +export async function reads(storage: Storage & Pick): Promise { + const head = await storage.head(context); + const out: Reads = { head }; + const conversations = (await storage.scanConversations({}, 1000, undefined, context)).items; + out.conversations = conversations; + for (const conversation of conversations) { + const entries = ( + await storage.scanEntries({ conversationId: conversation.id, order: "ascending" }, 1000, undefined, context) + ).items; + out[`entries:${conversation.id}`] = await Promise.all(entries.map((item) => storage.entry(item.id, context))); + out[`head-marker:${conversation.id}`] = await storage.findLatestHeadMarker(conversation.id, undefined, context); + } + const tasks = (await storage.scanTasks({}, 1000, undefined, context)).items; + out.tasks = tasks; + out.submissions = (await storage.scanSubmissions({}, 1000, undefined, context)).items; + out.submissionByStatus = (await storage.scanSubmissions({ status: "placed" }, 1000, undefined, context)).items; + out.tasksByStatus = (await storage.scanTasks({ status: "terminal" }, 1000, undefined, context)).items; + + const scopes: DocumentRecord["scope"][] = [ + { kind: "session" }, + ...conversations.map((conversation) => ({ kind: "conversation" as const, conversationId: conversation.id })), + ...tasks.map((item) => ({ kind: "task" as const, taskId: item.id })), + ]; + const points: DocumentPoint[] = ["current"]; + for (let seq = 1; seq <= (head ?? 0); seq++) points.push(seqFromNumber(seq)); + for (const scope of scopes) { + for (const at of points) { + const documents = (await storage.scanDocuments({ scope, at }, 1000, undefined, context)).items; + out[`documents:${JSON.stringify(scope)}:${at}`] = documents; + for (const record of documents) { + out[`document:${record.id}:${at}`] = await storage + .document(record.id, at, context) + .catch((error: unknown) => ({ rejected: error instanceof Error ? error.name : "unknown" })); + const address = { + kind: record.kind, + scope: record.scope, + ...(record.key === undefined ? {} : { key: record.key }), + }; + out[`find:${record.id}:${at}`] = await storage.findDocument(address, at, context); + } + } + } + out.nextId = await storage.mintId(); + return out; +} + +export async function collect(log: AsyncIterable): Promise { + const records: LogRecord[] = []; + for await (const record of log) records.push(record); + return records; +} diff --git a/packages/durable/test/sqlite-cloudflare.test.ts b/packages/durable/test/sqlite-cloudflare.test.ts index 0bf9c101d..99638165a 100644 --- a/packages/durable/test/sqlite-cloudflare.test.ts +++ b/packages/durable/test/sqlite-cloudflare.test.ts @@ -15,7 +15,7 @@ describe("Durable Object SQLite adapter", () => { it("keeps schema names unprefixed by default", async () => { const sqlite = new DatabaseSync(":memory:"); await openDurableObjectSqliteStorage(durableObjectStorage(sqlite)); - expect(sqlite.prepare("SELECT version FROM durable_schema").get()).toEqual({ version: 2 }); + expect(sqlite.prepare("SELECT version FROM durable_schema").get()).toEqual({ version: 3 }); expect(sqlite.prepare("SELECT name FROM sqlite_master WHERE name = 'pi_durable_schema'").get()).toBeUndefined(); }); diff --git a/packages/durable/test/sqlite-migrations.test.ts b/packages/durable/test/sqlite-migrations.test.ts index 0006e8a25..4b2ef8284 100644 --- a/packages/durable/test/sqlite-migrations.test.ts +++ b/packages/durable/test/sqlite-migrations.test.ts @@ -73,7 +73,7 @@ describe("durable SQLite migrations", () => { await storage.close(BACKGROUND_CONTEXT); const migrated = await openNodeSqliteDatabase(path); try { - expect(await migrated.get("SELECT version FROM durable_schema WHERE singleton = 1")).toEqual({ version: 2 }); + expect(await migrated.get("SELECT version FROM durable_schema WHERE singleton = 1")).toEqual({ version: 3 }); expect(await migrated.get("SELECT status, record FROM tasks WHERE id = 5")).toEqual({ status: "waiting", record: JSON.stringify(waiting), diff --git a/packages/durable/test/storage-replay.test.ts b/packages/durable/test/storage-replay.test.ts new file mode 100644 index 000000000..7c6221b9d --- /dev/null +++ b/packages/durable/test/storage-replay.test.ts @@ -0,0 +1,198 @@ +/// + +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 { IDBKeyRange as FakeIDBKeyRange, IDBFactory } from "fake-indexeddb"; +import { afterEach, describe, expect, it } from "vitest"; +import { seqFromNumber } from "../src/ids.ts"; +import { IndexedDbStorage } from "../src/storage/indexeddb.ts"; +import { openNodeJsonlStorage } from "../src/storage/jsonl/node.ts"; +import { MemoryStorage } from "../src/storage/memory.ts"; +import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts"; +import { type LogRecord, type ReplayableStorage, ROOT_CONVERSATION_ID, type StorageWrite } from "../src/types.ts"; +import { collect, create, delta, documentId, entry, entryId, latestDoc, reads, seed, task } from "./replay-support.ts"; + +Object.defineProperty(globalThis, "IDBKeyRange", { configurable: true, value: FakeIDBKeyRange }); + +const context = BACKGROUND_CONTEXT; + +const cleanups: (() => Promise)[] = []; + +afterEach(async () => { + for (const cleanup of cleanups.splice(0).reverse()) await cleanup(); +}); + +async function temporaryDirectory(): Promise { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-replay-")); + cleanups.push(() => rm(directory, { recursive: true, force: true })); + return directory; +} + +type Backend = { + readonly name: string; + open(): Promise<{ readonly storage: ReplayableStorage; reopen(): Promise }>; +}; + +function track(storage: ReplayableStorage): ReplayableStorage { + cleanups.push(() => storage.close(context).catch(() => {})); + return storage; +} + +const backends: readonly Backend[] = [ + { + name: "memory", + async open() { + const storage = new MemoryStorage(); + return { storage, reopen: async () => storage }; + }, + }, + { + name: "sqlite", + async open() { + const path = join(await temporaryDirectory(), "session.sqlite"); + let storage = track(await openNodeSqliteStorage(path)); + return { + storage, + reopen: async () => { + await storage.close(context); + storage = track(await openNodeSqliteStorage(path)); + return storage; + }, + }; + }, + }, + { + name: "jsonl", + async open() { + const directory = await temporaryDirectory(); + let storage = track(await openNodeJsonlStorage(directory, context)); + return { + storage, + reopen: async () => { + await storage.close(context); + storage = track(await openNodeJsonlStorage(directory, context)); + return storage; + }, + }; + }, + }, + { + name: "indexeddb", + async open() { + const indexedDB = new IDBFactory(); + let storage = track(await IndexedDbStorage.open("session", { indexedDB })); + return { + storage, + reopen: async () => { + await storage.close(context); + storage = track(await IndexedDbStorage.open("session", { indexedDB })); + return storage; + }, + }; + }, + }, +]; + +describe("export and replay", () => { + for (const source of backends) { + for (const target of backends) { + it(`replays ${source.name} into ${target.name} with the same reads`, async () => { + const opened = await source.open(); + await seed(opened.storage); + const reopened = await opened.reopen(); + const log = await collect(reopened.export(context)); + expect(log.at(-1)?.seq).toBe(await reopened.head(context)); + for (const record of log) { + expect(record.writes.some((write) => (write as StorageWrite).type === "document.copy")).toBe(false); + } + + const replica = (await target.open()).storage; + for (const record of log) await replica.replay(record, context); + expect(await reads(replica)).toEqual(await reads(reopened)); + }); + } + } + + for (const backend of backends) { + describe(backend.name, () => { + it("exports nothing from an empty storage", async () => { + const { storage } = await backend.open(); + expect(await storage.head(context)).toBeUndefined(); + expect(await collect(storage.export(context))).toEqual([]); + }); + + it("puts records without a sequence into the first record and ends at the head", async () => { + const { storage } = await backend.open(); + await storage.commit([{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], context); + await storage.commit([{ type: "task", value: task(2, "pending") }], context); + expect(await collect(storage.export(context))).toEqual([ + { + seq: 2, + writes: [ + { type: "conversation", value: { id: ROOT_CONVERSATION_ID } }, + { type: "task", value: task(2, "pending") }, + ], + }, + ]); + }); + + it("replays at explicit sequences with gaps and continues after them", async () => { + const { storage } = await backend.open(); + await storage.replay( + { seq: seqFromNumber(3), writes: [{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }] }, + context, + ); + await storage.replay({ seq: seqFromNumber(7), writes: [entry(9)] }, context); + expect(await storage.head(context)).toBe(7); + expect((await storage.entry(entryId(9), context))?.commitSeq).toBe(7); + expect(await storage.mintId()).toBe(10); + expect(await storage.commit([entry(10)], context)).toBe(8); + await expect(storage.replay({ seq: seqFromNumber(8), writes: [entry(11)] }, context)).rejects.toThrow( + "does not strictly increase", + ); + expect(await storage.entry(entryId(11), context)).toBeUndefined(); + }); + + it("refuses document copies in a log record", async () => { + const { storage } = await backend.open(); + await storage.commit([{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], context); + const copy = { + type: "document.copy", + record: latestDoc(3), + source: { id: documentId(2), at: "current" }, + } as unknown as LogRecord["writes"][number]; + await expect(storage.replay({ seq: seqFromNumber(2), writes: [copy] }, context)).rejects.toThrow( + "cannot contain a document copy", + ); + }); + + it("rejects an export that a commit interrupts", async () => { + const { storage } = await backend.open(); + await seed(storage); + const iterator = storage.export(context)[Symbol.asyncIterator](); + await iterator.next(); + await storage.commit([entry(100)], context); + await expect(async () => { + while (!(await iterator.next()).done); + }).rejects.toThrow("changed during export"); + }); + }); + } + + it("streams a SQLite export across pages", async () => { + const sqlite = backends.find((backend) => backend.name === "sqlite")!; + const { storage } = await sqlite.open(); + await storage.commit([{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], context); + await storage.commit([create(latestDoc(2), { n: 0 })], context); + for (let id = 3; id < 1000; id += 2) await storage.commit([entry(id), delta(2, "n", id)], context); + const log = await collect(storage.export(context)); + // The conversation has no sequence; it joins the first record, at the document's creation. + expect(log).toHaveLength(500); + + const replica = new MemoryStorage(); + for (const record of log) await replica.replay(record, context); + expect(await reads(replica)).toEqual(await reads(storage)); + }); +}); diff --git a/packages/durable/test/sync.test.ts b/packages/durable/test/sync.test.ts new file mode 100644 index 000000000..a6507b909 --- /dev/null +++ b/packages/durable/test/sync.test.ts @@ -0,0 +1,211 @@ +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import { describe, expect, it } from "vitest"; +import { seqFromNumber } from "../src/ids.ts"; +import { MemoryStorage } from "../src/storage/memory.ts"; +import { + compactRecords, + decodeLogLines, + encodeLogLine, + LogFormatError, + MemoryOutbox, + type OutboxRecord, + ReplicatingStorage, +} from "../src/sync/index.ts"; +import { type LogRecord, ROOT_CONVERSATION_ID, type StorageWrite } from "../src/types.ts"; +import { collect, create, delta, documentId, entry, latestDoc, reads, seed, task } from "./replay-support.ts"; + +const context = BACKGROUND_CONTEXT; + +async function replayed(records: readonly LogRecord[]): Promise { + const storage = new MemoryStorage(); + for (const record of records) await storage.replay(record, context); + return storage; +} + +async function replicated(): Promise<{ readonly storage: ReplicatingStorage; readonly outbox: MemoryOutbox }> { + const outbox = new MemoryOutbox(); + const storage = new ReplicatingStorage(new MemoryStorage(), outbox); + await seed(storage); + return { storage, outbox }; +} + +describe("ReplicatingStorage", () => { + it("logs every commit so that replaying the log gives the same reads", async () => { + const { storage, outbox } = await replicated(); + expect(outbox.records.map((record) => record.seq)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]); + expect(await reads(await replayed(outbox.records))).toEqual(await reads(storage)); + }); + + it("logs fork copies as the creations they became", async () => { + const { outbox } = await replicated(); + const fork = outbox.records.find((record) => record.seq === 6)!; + const writes = fork.writes as readonly StorageWrite[]; + expect(writes.some((write) => write.type === "document.copy")).toBe(false); + // The rewindable source as of seq 3, and the latest one as it was when the fork committed. + expect(writes.filter((write) => write.type === "document.create")).toMatchObject([ + { record: { id: 12 }, content: { kind: "base", version: 1, value: { a: 1, b: 1 } } }, + { record: { id: 13 }, content: { kind: "base", version: 1, value: { x: 1 } } }, + ]); + }); + + it("marks changes of current-only documents created earlier", async () => { + const { outbox } = await replicated(); + const hints = Object.fromEntries(outbox.records.map((record) => [record.seq, record.currentOnly ?? []])); + expect(hints).toMatchObject({ 3: [6], 4: [6], 7: [10], 8: [9, 10], 9: [13] }); + expect(hints[2]).toEqual([]); + }); + + it("does not log replayed records", async () => { + const outbox = new MemoryOutbox(); + const storage = new ReplicatingStorage(new MemoryStorage(), outbox); + await storage.replay( + { seq: seqFromNumber(1), writes: [{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }] }, + context, + ); + expect(outbox.records).toEqual([]); + expect(await storage.commit([entry(2)], context)).toBe(2); + expect(outbox.records.map((record) => record.seq)).toEqual([2]); + }); + + it("appends after the commit, so a rejected commit is never logged", async () => { + const outbox = new MemoryOutbox(); + const storage = new ReplicatingStorage(new MemoryStorage(), outbox); + await storage.commit([{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], context); + await expect( + storage.commit([{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], context), + ).rejects.toThrow("already belongs"); + expect(outbox.records).toHaveLength(1); + }); + + it("tells a listener about each record once it is in the outbox", async () => { + const outbox = new MemoryOutbox(); + const seen: number[] = []; + const storage = new ReplicatingStorage(new MemoryStorage(), outbox, { + onRecord: (record) => seen.push(record.seq, outbox.records.length), + }); + await storage.commit([{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], context); + expect(seen).toEqual([1, 1]); + }); +}); + +describe("compactRecords", () => { + it("keeps the reads of every split of the log into batches", async () => { + const { storage, outbox } = await replicated(); + const expected = await reads(storage); + const records = outbox.records; + for (let cut = 0; cut <= records.length; cut++) { + const batches = [records.slice(0, cut), records.slice(cut)]; + const compacted = batches.flatMap((batch) => compactRecords(batch)); + expect(await reads(await replayed(compacted)), `cut at ${cut}`).toEqual(expected); + } + }); + + it("keeps the reads when every record is its own batch, or the whole log is one", async () => { + const { storage, outbox } = await replicated(); + const expected = await reads(storage); + expect(await reads(await replayed(outbox.records.flatMap((record) => compactRecords([record]))))).toEqual( + expected, + ); + expect(await reads(await replayed(compactRecords(outbox.records)))).toEqual(expected); + }); + + it("compacts the export of a storage too", async () => { + const { storage } = await replicated(); + const exported = await collect(storage.export(context)); + expect(await reads(await replayed(compactRecords(exported)))).toEqual(await reads(storage)); + }); + + it("drops superseded task records and streaming deltas", async () => { + const live = latestDoc(2); + const records: OutboxRecord[] = [ + { + seq: seqFromNumber(10), + writes: [{ type: "task", value: task(5, "running") }, delta(2, "partial", "a")], + currentOnly: [documentId(2)], + }, + { seq: seqFromNumber(11), writes: [delta(2, "partial", "ab")], currentOnly: [documentId(2)] }, + { seq: seqFromNumber(12), writes: [entry(6)] }, + { + seq: seqFromNumber(13), + writes: [ + { type: "task", value: task(5, "terminal") }, + { type: "document.change", id: documentId(2), content: { kind: "base", version: 1, value: {} } }, + ], + currentOnly: [documentId(2)], + }, + { seq: seqFromNumber(14), writes: [delta(2, "partial", "c")], currentOnly: [documentId(2)] }, + ]; + expect(compactRecords(records)).toEqual([ + { seq: 12, writes: [entry(6)] }, + { + seq: 13, + writes: [ + { type: "task", value: task(5, "terminal") }, + { type: "document.change", id: documentId(2), content: { kind: "base", version: 1, value: {} } }, + ], + currentOnly: [documentId(2)], + }, + { seq: 14, writes: [delta(2, "partial", "c")], currentOnly: [documentId(2)] }, + ]); + // Without the hint the document's history is unknown, so its deltas stay. + const unhinted = records.map(({ currentOnly: _currentOnly, ...record }) => record); + expect(compactRecords(unhinted).flatMap((record) => record.writes)).toHaveLength(6); + expect(live.history).toBe("latest"); + }); + + it("keeps an empty last record so the batch ends where the run ended", () => { + const records: OutboxRecord[] = [ + { seq: seqFromNumber(3), writes: [{ type: "task", value: task(5, "pending") }] }, + { seq: seqFromNumber(4), writes: [{ type: "task", value: task(5, "running") }] }, + { seq: seqFromNumber(5), writes: [{ type: "task", value: task(5, "terminal") }] }, + { + seq: seqFromNumber(6), + writes: [create(latestDoc(7), { gone: true }), { type: "document.retire", id: documentId(7) }], + }, + { seq: seqFromNumber(7), writes: [{ type: "task", value: task(8, "pending") }] }, + { seq: seqFromNumber(8), writes: [{ type: "task", value: task(8, "running") }] }, + ]; + expect(compactRecords(records)).toEqual([ + { seq: 5, writes: [{ type: "task", value: task(5, "terminal") }] }, + { + seq: 6, + writes: [ + { type: "document.create", record: latestDoc(7), content: { kind: "base", version: 1, value: {} } }, + { type: "document.retire", id: documentId(7) }, + ], + }, + { seq: 8, writes: [{ type: "task", value: task(8, "running") }] }, + ]); + expect(compactRecords([records[0]!, { seq: seqFromNumber(4), writes: [] }])).toEqual([ + { seq: 3, writes: [{ type: "task", value: task(5, "pending") }] }, + { seq: 4, writes: [] }, + ]); + }); +}); + +describe("log lines", () => { + it("round-trips records with their hints", async () => { + const { outbox } = await replicated(); + const text = outbox.records.map(encodeLogLine).join(""); + expect(decodeLogLines(text)).toEqual({ records: outbox.records, torn: 0 }); + }); + + it("reports a torn last line instead of failing", () => { + const line = encodeLogLine({ seq: seqFromNumber(1), writes: [entry(2)] }); + const torn = encodeLogLine({ seq: seqFromNumber(2), writes: [entry(3)] }).slice(0, 10); + expect(decodeLogLines(line + torn)).toEqual({ records: [{ seq: 1, writes: [entry(2)] }], torn: 10 }); + }); + + it("refuses unknown versions, unknown writes, and sequences that do not increase", () => { + expect(() => decodeLogLines('{"v":2,"seq":1,"writes":[]}\n')).toThrow(LogFormatError); + expect(() => decodeLogLines('{"v":1,"seq":1,"writes":[{"type":"document.copy"}]}\n')).toThrow( + "unknown type document.copy", + ); + expect(() => decodeLogLines('{"v":1,"seq":2,"writes":[]}\n{"v":1,"seq":2,"writes":[]}\n')).toThrow( + "does not increase", + ); + expect(() => decodeLogLines("not json\n")).toThrow("is not JSON"); + // The parser's own error says where and why. + expect(() => decodeLogLines("not json\n")).toThrow(expect.objectContaining({ cause: expect.any(SyntaxError) })); + }); +}); diff --git a/scripts/check-entry-graphs.mjs b/scripts/check-entry-graphs.mjs index f53303d99..79503592c 100644 --- a/scripts/check-entry-graphs.mjs +++ b/scripts/check-entry-graphs.mjs @@ -64,8 +64,9 @@ const BUDGETS = { // memory and SQLite storages share scan order and cursor handling. Values from Chord's root (copyJson, // replicatedState) reach its host runtime through the root barrel. Chord's context reaches pi-telemetry's root // (index, noop, metrics, memory) for its NOOP telemetry default; harness/telemetry.ts is Durable's schema. - // The Harness's request identity reaches pi-ai's utils/identity.ts and its version constant. - maxFiles: 73, + // The Harness's request identity reaches pi-ai's utils/identity.ts and its version constant. The storages + // share log export (storage/export.ts). + maxFiles: 74, forbid: [ "packages/ai/src/index.ts", "packages/ai/src/utils/typebox-helpers.ts",