diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index ca1c6c9e4..900351d0b 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -6,15 +6,13 @@ - `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 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). +- 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). - `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 7e759885d..71aa68376 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -798,24 +798,6 @@ 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 1de2d948e..3e206cf8e 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -5279,33 +5279,6 @@ 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 90ac8ea16..c15e96ad4 100644 --- a/packages/durable/package.json +++ b/packages/durable/package.json @@ -60,12 +60,6 @@ "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 cb6caa1c8..95d9606ae 100644 --- a/packages/durable/src/index.ts +++ b/packages/durable/src/index.ts @@ -168,12 +168,9 @@ 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 deleted file mode 100644 index 9c9d1a13b..000000000 --- a/packages/durable/src/storage/export.ts +++ /dev/null @@ -1,135 +0,0 @@ -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 7d6df5d62..6b2121444 100644 --- a/packages/durable/src/storage/indexeddb.ts +++ b/packages/durable/src/storage/indexeddb.ts @@ -21,12 +21,10 @@ import type { EntryRecord, Id, JsonObject, - LogRecord, - LogWrite, Page, - ReplayableStorage, ScanOrder, Seq, + Storage, StorageWrite, StoredDocument, SubmissionId, @@ -36,16 +34,6 @@ 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; @@ -325,6 +313,9 @@ 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": @@ -384,7 +375,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 ReplayableStorage { +export class IndexedDbStorage implements Storage { private readonly database: IDBDatabase; private closed = false; private closing: Promise | undefined; @@ -427,88 +418,13 @@ export class IndexedDbStorage implements ReplayableStorage { } 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"); - 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); + const seq = 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 5d2e83605..4b03fd5be 100644 --- a/packages/durable/src/storage/jsonl/storage.ts +++ b/packages/durable/src/storage/jsonl/storage.ts @@ -15,8 +15,6 @@ import type { EntryId, EntryQuery, Id, - LogRecord, - ReplayableStorage, Seq, Storage, StorageWrite, @@ -26,8 +24,7 @@ import type { TaskQuery, TaskRecord, } from "../../types.ts"; -import { assertLogRecord, isCurrentOnly, RECLAIMED_BASE } from "../export.ts"; -import { MemoryStorage, type PreparedMemoryCommit } from "../memory.ts"; +import { MemoryStorage } from "../memory.ts"; const FORMAT_VERSION = 1; const MAIN_FILE = "main.jsonl"; @@ -107,6 +104,9 @@ 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 ReplayableStorage { +export class JsonlStorage implements Storage { private readonly fs: FileSystem; private readonly directory: string; private readonly mainPath: string; @@ -277,24 +277,7 @@ export class JsonlStorage implements ReplayableStorage { async commit(writes: readonly StorageWrite[], context: Context): Promise { this.assertUsable(); - 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 prepared = this.memory.prepareCommit(writes); const encoded = this.encodeCommit(prepared.seq, prepared.writes); const reclamations = this.planReclamations(prepared.writes, encoded); const sidecars = await Promise.all( @@ -712,7 +695,7 @@ export class JsonlStorage implements ReplayableStorage { writes.push({ type: "document.create", record: operation.record, - content: reclaimed || content === undefined ? RECLAIMED_BASE : content, + content: reclaimed || content === undefined ? { kind: "base", version: 1, value: {} } : 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 896a23ed5..7f379cc29 100644 --- a/packages/durable/src/storage/memory.ts +++ b/packages/durable/src/storage/memory.ts @@ -19,12 +19,10 @@ import type { EntryRecord, Id, JsonObject, - LogRecord, - LogWrite, Page, - ReplayableStorage, ScanOrder, Seq, + Storage, StorageWrite, StoredDocument, SubmissionId, @@ -34,15 +32,6 @@ 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; @@ -205,6 +194,9 @@ 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 = }>( @@ -242,7 +234,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 ReplayableStorage { +export class MemoryStorage implements Storage { private readonly state: State = { recordTypes: new Map(), conversations: new Map(), @@ -438,47 +430,6 @@ export class MemoryStorage implements ReplayableStorage { 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 78967d213..a5ee244ae 100644 --- a/packages/durable/src/storage/sqlite/migrations.ts +++ b/packages/durable/src/storage/sqlite/migrations.ts @@ -110,17 +110,9 @@ 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 73656433e..b807ba529 100644 --- a/packages/durable/src/storage/sqlite/schema.ts +++ b/packages/durable/src/storage/sqlite/schema.ts @@ -35,10 +35,6 @@ 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 3704f2e68..7dc590382 100644 --- a/packages/durable/src/storage/sqlite/storage.ts +++ b/packages/durable/src/storage/sqlite/storage.ts @@ -20,12 +20,10 @@ import type { EntryRecord, Id, JsonObject, - LogRecord, - LogWrite, Page, - ReplayableStorage, ScanOrder, Seq, + Storage, StorageWrite, StoredDocument, SubmissionId, @@ -35,15 +33,6 @@ 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"; @@ -67,19 +56,6 @@ 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; @@ -93,15 +69,6 @@ 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. @@ -162,6 +129,9 @@ 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": @@ -179,7 +149,7 @@ const writeId = (write: StorageWrite): Id | undefined => { }; /** Portable SQLite implementation of the Pico storage contract. */ -export class SqliteStorage implements ReplayableStorage { +export class SqliteStorage implements Storage { private readonly db: SqliteDatabase; private readonly schema: SqliteSchema; private nextId: number; @@ -215,142 +185,15 @@ export class SqliteStorage implements ReplayableStorage { } 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 committed = await this.db.transaction(async (transaction) => { + const seq = 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"); - 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); + const committedSeq = 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); @@ -363,7 +206,7 @@ export class SqliteStorage implements ReplayableStorage { return committedSeq; }); this.nextId = Math.max(this.nextId, candidateNextId); - return committed; + return seq; } async mintId>(): Promise { diff --git a/packages/durable/src/sync/compact.ts b/packages/durable/src/sync/compact.ts deleted file mode 100644 index 1a459c24f..000000000 --- a/packages/durable/src/sync/compact.ts +++ /dev/null @@ -1,127 +0,0 @@ -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 deleted file mode 100644 index 97db6cddc..000000000 --- a/packages/durable/src/sync/index.ts +++ /dev/null @@ -1,3 +0,0 @@ -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 deleted file mode 100644 index 6c0c4e292..000000000 --- a/packages/durable/src/sync/log.ts +++ /dev/null @@ -1,101 +0,0 @@ -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 deleted file mode 100644 index d56b46c81..000000000 --- a/packages/durable/src/sync/replicating.ts +++ /dev/null @@ -1,157 +0,0 @@ -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 aa9522b24..fe69f8e20 100644 --- a/packages/durable/src/types.ts +++ b/packages/durable/src/types.ts @@ -1272,34 +1272,3 @@ 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 c7dc290fd..9f45ec600 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, then adds sequence indexes", async () => { + it("migrates a version 1 schema to admit sleeping tasks, keeping tasks and 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: 3 }); + expect(await database.get(`SELECT version FROM ${schema.tables.durableSchema}`)).toEqual({ version: 2 }); 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 deleted file mode 100644 index 9cfb31aff..000000000 --- a/packages/durable/test/replay-support.ts +++ /dev/null @@ -1,214 +0,0 @@ -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 99638165a..0bf9c101d 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: 3 }); + expect(sqlite.prepare("SELECT version FROM durable_schema").get()).toEqual({ version: 2 }); 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 4b2ef8284..0006e8a25 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: 3 }); + expect(await migrated.get("SELECT version FROM durable_schema WHERE singleton = 1")).toEqual({ version: 2 }); 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 deleted file mode 100644 index 7c6221b9d..000000000 --- a/packages/durable/test/storage-replay.test.ts +++ /dev/null @@ -1,198 +0,0 @@ -/// - -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 deleted file mode 100644 index a6507b909..000000000 --- a/packages/durable/test/sync.test.ts +++ /dev/null @@ -1,211 +0,0 @@ -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 79503592c..f53303d99 100644 --- a/scripts/check-entry-graphs.mjs +++ b/scripts/check-entry-graphs.mjs @@ -64,9 +64,8 @@ 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. The storages - // share log export (storage/export.ts). - maxFiles: 74, + // The Harness's request identity reaches pi-ai's utils/identity.ts and its version constant. + maxFiles: 73, forbid: [ "packages/ai/src/index.ts", "packages/ai/src/utils/typebox-helpers.ts",