diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 14750ec5a..1b2ca0a02 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Added + +- `@earendil-works/pi-durable/sync` (experimental) moves sessions between devices through a remote store, from the outbox of the session's Chord store (`openStore(backend, ctx, { outbox: true })`, Durable's records in it through `StoreStorage`). Every device continues a session as its own: `uploadSession()` uploads the outbox commits after the last sequence the copy shares with the remote as one gzip batch into a slot the remote creates once per start sequence, trims the outbox through what the remote acknowledged, and reports a copy as `diverged` when another device's upload created that slot first, or `behind` when its head fell below what it had uploaded. A session whose outbox does not reach back to its first commit starts with an archive slot (`store.export()`). `pullSession()` catches a copy up through `applyBatch()`, resumes an interrupted pull, or restores into an empty backend (`importStore()` for an archive); `foldSession()` folds the history into a snapshot archive, keeping folded slots taken. `FlushScheduler` and `isUrgentCommit()` decide when to upload. `FileSystemSyncRemote` is a remote in a directory; `FileSyncStateStore` keeps sync state in a directory. + ## [1.5.0] - 2026-11-02 ### Breaking Changes diff --git a/packages/durable/README.md b/packages/durable/README.md index b9e54a068..ecd172545 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -889,6 +889,32 @@ await database.transaction(async (transaction) => { }); ``` +### Sync (experimental) + +`@earendil-works/pi-durable/sync` moves sessions between devices through a remote store. It syncs a session's Chord +store, not a `Storage`: open the store with its outbox, `openStore(backend, ctx, { outbox: true })`, and Durable's +records in it with `StoreStorage.open(store, owner, ctx)`. The outbox holds every commit after its floor, written in +the commit's own transaction, so an upload reads `outbox.read(after)` and trims what the remote acknowledged. + +Every device continues a session as its own. A remote keeps one slot per start sequence and creates it once, and only +where the history ends, so the device whose upload creates a slot continues the session from there. A device that finds +the slot taken by another device has diverged: the host renames its copy to a new session and restores the session +fresh. A session whose outbox does not reach back to its first commit (turned on later, or trimmed) starts with an +archive (`store.export()`). The remote folds the history into snapshot archives, keeping folded slots taken. + +```typescript +const store = await openStore(sqliteBackend(db), context, { outbox: true }); +const result = await uploadSession({ sessionId, device, source: store, store: remote, state, context }); +if (result.type === "diverged") forkLocalCopy(sessionId, result.at); // host: rename, upload under the new ID +// Catch up, resume, or restore into an empty backend; nothing may hold the store's claims (no StoreStorage open). +const { store: pulled } = await pullSession({ sessionId, store: remote, local: { kind: "store", store }, state, context }); +``` + +Pulling refuses a copy with commits it has not uploaded (`UnsyncedCommitsError`): upload first. Uploading refuses a +copy that lost commits it had uploaded (`behind`): it pulls them back first. `FlushScheduler` and `isUrgentCommit()` +decide when to upload; `FileSystemSyncRemote` is a remote in a directory and `FileSyncStateStore` keeps sync state in +one, for development and tests. + A custom backend's `commit(batches, receive, context)` stores `batches` in order, each as one atomic commit with its own seq, checked against the state the earlier ones left, in as few round trips as it can. It stops at the first batch its own checks refuse (global IDs, document actions, copy resolution): the batches before it stay stored, it and the later ones are not. Once they are durable it calls `receive(index, seq)` per stored batch, in order and synchronously, then resolves with `{ stored, refused? }`; it rejects for an unknown outcome (connection lost, a failed statement or `COMMIT`). `receive` is a durability receipt: the Session publishes each batch later, one at a time, so memory and every view never show a partial log. `receive` calls no Storage method and never throws. Custom backends can run the shared conformance suite, which checks this too, with any Vitest- or Jest-compatible runner: ```typescript diff --git a/packages/durable/src/sync/batch.ts b/packages/durable/src/sync/batch.ts index a26a4f3e1..e15cd96a5 100644 --- a/packages/durable/src/sync/batch.ts +++ b/packages/durable/src/sync/batch.ts @@ -1,21 +1,49 @@ -import type { LogRecord } from "../types.ts"; +import type { ArchivePart, OutboxCommit } from "@earendil-works/chord/next"; import { decodeLogLines, encodeLogLine, LogFormatError } from "./lines.ts"; /** - * A batch as uploaded: the log records after one sequence as gzip-compressed JSON lines. gzip, because + * A batch as uploaded: outbox commits after one sequence as gzip-compressed JSON lines. gzip, because * `CompressionStream` offers it in Node, Workers, and browsers alike, and a browser viewer reads batches too. */ -export async function encodeBatch(records: readonly LogRecord[]): Promise { - return transform(new TextEncoder().encode(records.map(encodeLogLine).join("")), new CompressionStream("gzip")); +export async function encodeBatch(commits: readonly OutboxCommit[]): Promise { + return gzip(new TextEncoder().encode(commits.map(encodeLogLine).join(""))); } -export async function decodeBatch(bytes: Uint8Array): Promise { - const text = new TextDecoder("utf-8", { fatal: true }).decode( - await transform(bytes, new DecompressionStream("gzip")), - ); - const { records, torn } = decodeLogLines(text); +export async function decodeBatch(bytes: Uint8Array): Promise { + const { commits, torn } = decodeLogLines(await gunzipText(bytes)); if (torn > 0) throw new LogFormatError("Batch ends inside a log line"); - return records; + return commits; +} + +/** A snapshot as uploaded: a store archive (`store.export()`), one part per JSON line, gzip-compressed. */ +export async function encodeSnapshot(parts: AsyncIterable): Promise { + let text = ""; + for await (const part of parts) text += `${JSON.stringify(part)}\n`; + return gzip(new TextEncoder().encode(text)); +} + +/** The parts of a snapshot, unchecked: `importStore()` checks them (a trust boundary). */ +export async function decodeSnapshot(bytes: Uint8Array): Promise { + const text = await gunzipText(bytes); + if (text.length > 0 && !text.endsWith("\n")) throw new LogFormatError("Snapshot ends inside a line"); + return text + .split("\n") + .filter((line) => line.length > 0) + .map((line, index) => { + try { + return JSON.parse(line) as unknown; + } catch (error) { + throw new LogFormatError(`Snapshot line ${index + 1} is not JSON`, { cause: error }); + } + }); +} + +function gzip(bytes: Uint8Array): Promise { + return transform(bytes, new CompressionStream("gzip")); +} + +async function gunzipText(bytes: Uint8Array): Promise { + return new TextDecoder("utf-8", { fatal: true }).decode(await transform(bytes, new DecompressionStream("gzip"))); } async function transform(bytes: Uint8Array, stream: CompressionStream | DecompressionStream): Promise { diff --git a/packages/durable/src/sync/client.ts b/packages/durable/src/sync/client.ts index ebc588fb5..f45eb0419 100644 --- a/packages/durable/src/sync/client.ts +++ b/packages/durable/src/sync/client.ts @@ -1,17 +1,29 @@ import type { Context } from "@earendil-works/chord"; -import { LogInterruptedError } from "../errors.ts"; -import { LOG_START } from "../storage/log.ts"; -import type { CommitPublication, LogRecord, ReplayableStorage, Seq } from "../types.ts"; -import { decodeBatch, encodeBatch } from "./batch.ts"; -import { compactRecords } from "./compact.ts"; import { + applyBatch, + type Backend, + type ChordStore, + importStore, + type OutboxCommit, + openStore, + type Store, +} from "@earendil-works/chord/next"; +import { seqFromNumber } from "../ids.ts"; +import type { CommitPublication, Seq } from "../types.ts"; +import { decodeBatch, decodeSnapshot, encodeBatch, encodeSnapshot } from "./batch.ts"; +import { + type BatchInfo, HistoryUnavailableError, + LOG_START, remoteHead, remoteHistory, type SessionSyncMaintenance, type SessionSyncStore, } from "./remote.ts"; +/** Commits read from the outbox per uploaded batch. */ +const BATCH_COMMITS = 1000; + /** * What a host keeps for each copy of a session it syncs: the last sequence the copy shares with the remote. Commits * after it are the copy's own and not uploaded yet. When this is lost, a copy without commits of its own takes the @@ -20,16 +32,16 @@ import { export type SyncState = { readonly syncedHead: Seq | undefined; /** - * Set while a pull replays a batch or snapshot ending here: a local head up to it holds the remote's records, not + * Set while a pull applies a batch or snapshot ending here: a local head up to it holds the remote's commits, not * the copy's own, even when the pull stopped partway. */ readonly pulling?: Seq; }; -/** The last sequence a copy shares with the remote, counting records an interrupted pull replayed. */ -function sharedHead(state: SyncState | undefined, local: Seq | undefined): Seq | undefined { - if (state?.pulling !== undefined && local !== undefined && local <= state.pulling) return local; - return state?.syncedHead; +/** The last sequence a copy shares with the remote, counting commits an interrupted pull applied. */ +function sharedHead(state: SyncState | undefined, local: Seq): Seq { + if (state?.pulling !== undefined && local <= state.pulling) return local; + return state?.syncedHead ?? LOG_START; } export interface SyncStateStore { @@ -37,27 +49,35 @@ export interface SyncStateStore { set(sessionId: string, state: SyncState, context: Context): Promise; } -/** Where the log comes from: the storage, or a host's proxy to the process that owns it. */ -export type LogSource = Pick; +/** A store's head as a sequence: `lastSeq`, 0 for a store without commits. */ +export function storeHead(store: Store): Seq { + return seqFromNumber(store.lastSeq); +} -/** Read the log after `after`, again when a commit interrupts it. */ -export async function readLog( - source: LogSource, - after: Seq | undefined, - context: Context, - attempts = 5, -): Promise { - for (let attempt = 1; ; attempt++) { - const records: LogRecord[] = []; - try { - for await (const record of source.log(after, context)) records.push(record); - return records; - } catch (error) { - if (!(error instanceof LogInterruptedError) || attempt >= attempts) throw error; - } +/** The store keeps no outbox, or its outbox no longer holds the commits a sync needs. */ +export class OutboxUnavailableError extends Error { + constructor(message: string) { + super(message); + this.name = "OutboxUnavailableError"; } } +function outboxOf(store: Store) { + if (store.outbox === undefined) { + throw new OutboxUnavailableError("The session's store has no outbox: open it with { outbox: true } to sync it"); + } + return store.outbox; +} + +/** The outbox commits after `after`, a page of at most `limit`. */ +export async function readOutbox(store: Store, after: Seq, limit: number, context: Context): Promise { + const outbox = outboxOf(store); + if (after < outbox.floor) { + throw new OutboxUnavailableError(`The outbox starts after ${outbox.floor}, past the synced head ${after}`); + } + return outbox.read(after, limit, context); +} + export type UploadResult = | { readonly type: "uploaded"; readonly head: Seq } /** Nothing was committed since the last upload. */ @@ -65,19 +85,23 @@ export type UploadResult = /** * Another device continued the session from `at` first. This copy's commits after `at` are a different history: * the host renames the copy to a new session (forked from this one at `at`) and uploads it there, and the session - * itself restores into a fresh storage. + * itself restores into a fresh store. */ | { readonly type: "diverged"; readonly at: Seq } /** * The copy lost commits it had uploaded, for example to a power loss: its head is below `syncedHead`. Its next * commits would reuse their sequences, so nothing is uploaded. Pull what it lost before the copy commits again; a pull - * replays from the local head. + * applies from the local head. */ - | { readonly type: "behind"; readonly head: Seq | undefined; readonly syncedHead: Seq }; + | { readonly type: "behind"; readonly head: Seq; readonly syncedHead: Seq }; /** - * Upload what this copy committed since it last shared the remote's history. Every device continues a session as its - * own; whichever creates the slot first continues the session, so an upload that is acknowledged stays. + * Upload what this copy committed since it last shared the remote's history, from the store's outbox, and trim the + * outbox through what the remote acknowledged. Every device continues a session as its own; whichever creates the slot + * first continues the session, so an upload that is acknowledged stays. + * + * A session the remote does not have yet, whose outbox does not reach back to its first commit (turned on later, or + * trimmed, as for a copy renamed after it diverged), starts with an archive slot: the store's `export()` at its head. * * A slot this device created already is an upload whose response was lost: the remote has the history up to its end, * and the upload goes on from there. @@ -85,124 +109,205 @@ export type UploadResult = export async function uploadSession(options: { readonly sessionId: string; readonly device: string; - readonly source: LogSource; + readonly source: Store; readonly store: SessionSyncStore; readonly state: SyncStateStore; readonly context: Context; }): Promise { const { sessionId, device, source, store, state, context } = options; - let synced = sharedHead(await state.get(sessionId, context), await source.head(context)); + const outbox = outboxOf(source); + let synced = sharedHead(await state.get(sessionId, context), storeHead(source)); + let uploaded: Seq | undefined; while (true) { // Read again on every pass: a slot this device already holds may end past what the copy still has. - const head = await source.head(context); - if (synced !== undefined && (head ?? LOG_START) < synced) return { type: "behind", head, syncedHead: synced }; - const records = await readLog(source, synced, context); - if (records.length === 0) return { type: "idle" }; - const after = synced ?? LOG_START; - const to = records.at(-1)!.seq; - const result = await store.putBatch(sessionId, after, to, device, await encodeBatch(records), context); + const head = storeHead(source); + if (head < synced) return { type: "behind", head, syncedHead: synced }; + let to: Seq; + let body: Uint8Array; + let archive = false; + if (synced === LOG_START && outbox.floor > 0) { + const snapshot = await exportSnapshot(source, context); + to = snapshot.to; + body = snapshot.body; + archive = true; + } else { + const commits = await readOutbox(source, synced, BATCH_COMMITS, context); + if (commits.length === 0) + return uploaded === undefined ? { type: "idle" } : { type: "uploaded", head: uploaded }; + to = seqFromNumber(commits.at(-1)!.seq); + body = await encodeBatch(commits); + } + const after = synced; + const result = await store.putBatch(sessionId, after, to, device, body, context, archive ? { archive } : {}); // A history that does not end at `after` is another one too: the remote never held this copy's past `after`. if (result.type === "not_head") return { type: "diverged", at: after }; if (result.type === "exists" && result.device !== device) return { type: "diverged", at: after }; synced = result.type === "created" ? to : result.to; await state.set(sessionId, { syncedHead: synced }, context); - if (result.type === "created") return { type: "uploaded", head: to }; + const trimmable = Math.min(synced, source.lastSeq); + if (trimmable > outbox.floor) await outbox.trim(trimmable, context); + if (result.type === "created") uploaded = to; } } +/** The store's state at its head as a snapshot body. `export()` holds the commit queue, so the head cannot move. */ +async function exportSnapshot( + store: Store, + context: Context, +): Promise<{ readonly to: Seq; readonly body: Uint8Array }> { + let to: Seq | undefined; + const parts = store.export(context); + const body = await encodeSnapshot( + (async function* () { + for await (const part of parts) { + if (part.kind === "manifest") to = seqFromNumber(part.lastSeq); + yield part; + } + })(), + ); + if (to === undefined) throw new Error("A store export has no manifest"); + return { to, body }; +} + /** This copy has commits it has not uploaded: upload them first, which may find that the copy diverged. */ export class UnsyncedCommitsError extends Error { - constructor(sessionId: string, localHead: Seq, syncedHead: Seq | undefined) { + constructor(sessionId: string, localHead: Seq, syncedHead: Seq) { super( - `This copy of session ${sessionId} is at ${localHead}, past what it shares with the remote (${syncedHead ?? 0})`, + `This copy of session ${sessionId} is at ${localHead}, past what it shares with the remote (${syncedHead})`, ); this.name = "UnsyncedCommitsError"; } } +/** The copy a pull applies to: an open store, or an empty backend to restore the session into. */ +export type LocalCopy = + | { readonly kind: "store"; readonly store: Store } + | { readonly kind: "empty"; readonly backend: Backend }; + /** - * Replay what the remote has after the local head into `storage`; into an empty storage, that restores the session. - * A pull that stopped inside a batch resumes it. No Harness may have the storage open: its memory would be behind. + * Apply what the remote has after the local head; into an empty backend, that restores the session (from an archive + * when the history starts with one) and opens its store with an outbox. Resolves with the store, which the caller + * owns and closes. Nothing may hold the store's claims (no `StoreStorage`, no Harness): `applyBatch()` refuses then. * - * A copy with commits it has not uploaded rejects with `UnsyncedCommitsError` instead of getting the remote's records + * A copy with commits it has not uploaded rejects with `UnsyncedCommitsError` instead of getting the remote's commits * spliced onto its own. A copy whose history folding removed rejects with `HistoryUnavailableError` and restores into - * a fresh storage. + * a fresh backend. */ export async function pullSession(options: { readonly sessionId: string; readonly store: SessionSyncStore; - readonly storage: ReplayableStorage; + readonly local: LocalCopy; readonly state: SyncStateStore; readonly context: Context; -}): Promise { - const { sessionId, store, storage, state, context } = options; - const listing = await store.list(sessionId, context); - const local = await storage.head(context); - const synced = sharedHead(await state.get(sessionId, context), local); - if (local !== undefined && local > (synced ?? LOG_START)) throw new UnsyncedCommitsError(sessionId, local, synced); - const history = remoteHistory(listing, local); - const bodies: { readonly to: Seq; read(): Promise }[] = []; - if (history.snapshot !== undefined) { - const snapshot = history.snapshot; - bodies.push({ to: snapshot, read: () => store.getSnapshot(sessionId, snapshot, context) }); - } - for (const batch of history.batches) { - bodies.push({ to: batch.to, read: () => store.getBatch(sessionId, batch.after, context) }); - } - for (const body of bodies) { - const bytes = await body.read(); - const at = await storage.head(context); - if (bytes === undefined) throw new HistoryUnavailableError(at ?? LOG_START); - await state.set(sessionId, { syncedHead: at, pulling: body.to }, context); - for (const record of await decodeBatch(bytes)) { - if (record.seq > (at ?? LOG_START)) await storage.replay(record, context); +}): Promise<{ readonly store: Store; readonly head: Seq }> { + const { sessionId, store: remote, local, state, context } = options; + const listing = await remote.list(sessionId, context); + let target: Store | undefined = local.kind === "store" ? local.store : undefined; + const at = () => (target === undefined ? LOG_START : storeHead(target)); + const synced = sharedHead(await state.get(sessionId, context), at()); + if (at() > synced) throw new UnsyncedCommitsError(sessionId, at(), synced); + const history = remoteHistory(listing, at()); + try { + const bodies: { readonly to: Seq; readonly archive: boolean; read(): Promise }[] = []; + if (history.snapshot !== undefined) { + const snapshot = history.snapshot; + bodies.push({ to: snapshot, archive: true, read: () => remote.getSnapshot(sessionId, snapshot, context) }); } - await state.set(sessionId, { syncedHead: await storage.head(context) }, context); + for (const batch of history.batches) { + bodies.push({ + to: batch.to, + archive: batch.archive === true, + read: () => remote.getBatch(sessionId, batch.after, context), + }); + } + for (const body of bodies) { + if (body.to <= at()) continue; + const bytes = await body.read(); + if (bytes === undefined) throw new HistoryUnavailableError(at()); + await state.set(sessionId, { syncedHead: at(), pulling: body.to }, context); + if (body.archive) { + // An archive restores a whole store: only into an empty backend. + if (target !== undefined || local.kind !== "empty") throw new HistoryUnavailableError(at()); + target = await importStore(local.backend, await decodeSnapshot(bytes), context, { outbox: true }); + } else { + if (target === undefined) { + if (local.kind !== "empty") throw new Error("unreachable: a store copy always has a target"); + target = await openStore(local.backend, context, { outbox: true }); + } + const from = target.lastSeq; + const commits = (await decodeBatch(bytes)).filter((commit) => commit.seq > from); + if (commits.length > 0) await applyBatch(target as ChordStore, commits, context); + } + await state.set(sessionId, { syncedHead: at() }, context); + } + if (target === undefined) { + if (local.kind !== "empty") throw new Error("unreachable: a store copy always has a target"); + target = await openStore(local.backend, context, { outbox: true }); + } + const head = at(); + if (head < remoteHead(listing)) throw new HistoryUnavailableError(head); + await state.set(sessionId, { syncedHead: head }, context); + // What the remote holds needs no upload: the outbox keeps only this copy's own commits. + const outbox = target.outbox; + if (outbox !== undefined && head > outbox.floor) await outbox.trim(head, context); + return { store: target, head }; + } catch (error) { + // A store this pull opened is the caller's only on success. + if (local.kind === "empty" && target !== undefined) await target.close(context).catch(() => {}); + throw error; } - const head = await storage.head(context); - if ((head ?? LOG_START) < remoteHead(listing)) throw new HistoryUnavailableError(head ?? LOG_START); - await state.set(sessionId, { syncedHead: head }, context); - return head; } /** - * Fold a session's history into one snapshot: drop the records of the batches it covers, keeping their slots taken, - * and delete older snapshots. The remote runs this, not clients. The rules of `compactRecords()` need no storage, - * since a log holds no document copies. + * Fold a session's history into one snapshot: restore it into `scratch` (an empty backend, for example an in-memory + * SQLite database), upload its `export()` as a snapshot, drop the commits of the batches it covers, keeping their + * slots taken, and delete older snapshots. The remote runs this, not clients. */ export async function foldSession(options: { readonly sessionId: string; readonly store: SessionSyncMaintenance; + readonly scratch: Backend; readonly context: Context; }): Promise { - const { sessionId, store, context } = options; + const { sessionId, store, scratch, context } = options; const listing = await store.list(sessionId, context); const history = remoteHistory(listing); if (history.batches.length + (history.snapshot === undefined ? 0 : 1) < 2) return undefined; - const records: LogRecord[] = []; - const read = async (bytes: Uint8Array | undefined, what: string) => { - // Only the remote folds, and it folds one session at a time: what it just listed cannot be gone. - if (bytes === undefined) throw new Error(`${what} of session ${sessionId} vanished while folding`); - records.push(...(await decodeBatch(bytes))); - }; - if (history.snapshot !== undefined) { - await read(await store.getSnapshot(sessionId, history.snapshot, context), `Snapshot ${history.snapshot}`); - } - for (const batch of history.batches) { - await read(await store.getBatch(sessionId, batch.after, context), `Batch after ${batch.after}`); - } - await store.putSnapshot(sessionId, history.head, await encodeBatch(compactRecords(records)), context); - await store.foldBatches( + const restored = await pullSession({ sessionId, - history.batches.map((batch) => batch.after), + store, + local: { kind: "empty", backend: scratch }, + state: new ScratchState(), context, - ); - await store.deleteSnapshots( - sessionId, - listing.snapshots.filter((snapshot) => snapshot < history.head), - context, - ); - return history.head; + }); + try { + const snapshot = await exportSnapshot(restored.store, context); + await store.putSnapshot(sessionId, snapshot.to, snapshot.body, context); + await store.foldBatches( + sessionId, + history.batches.filter((batch: BatchInfo) => batch.to <= snapshot.to).map((batch) => batch.after), + context, + ); + await store.deleteSnapshots( + sessionId, + listing.snapshots.filter((at) => at < snapshot.to), + context, + ); + return snapshot.to; + } finally { + await restored.store.close(context); + } +} + +/** Sync state for a restore nobody continues. */ +class ScratchState implements SyncStateStore { + #state: SyncState | undefined; + async get(): Promise { + return this.#state; + } + async set(_sessionId: string, state: SyncState): Promise { + this.#state = state; + } } /** Whether a commit should reach the remote soon: what the user typed, a finished answer, a settled submission. */ diff --git a/packages/durable/src/sync/compact.ts b/packages/durable/src/sync/compact.ts deleted file mode 100644 index c47eb482e..000000000 --- a/packages/durable/src/sync/compact.ts +++ /dev/null @@ -1,121 +0,0 @@ -import { isCurrentOnly, RECLAIMED_BASE } from "../storage/log.ts"; -import type { DocumentId, Id, LogRecord, LogWrite } from "../types.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 log records into the records that replay to the same reads, as a storage's own log - * of the same changes would hold them: - * - * - 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 the placeholder base; - * - rewindable documents keep everything, and so do documents created before the run unless `lookup` knows them; - * - records left without writes are dropped, except the last, which marks where the run ends. - * - * Pure: it folds a snapshot and the batches after it into a new snapshot without a storage, for example remotely. - */ -export function compactRecords( - records: readonly LogRecord[], - lookup: CurrentOnlyLookup = () => undefined, -): LogRecord[] { - 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) => { - 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: LogRecord[] = []; - 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; - compacted.push({ seq: record.seq, writes }); - }); - return compacted; -} diff --git a/packages/durable/src/sync/filesystem.ts b/packages/durable/src/sync/filesystem.ts index 362dbf5f5..5d8a8a411 100644 --- a/packages/durable/src/sync/filesystem.ts +++ b/packages/durable/src/sync/filesystem.ts @@ -34,9 +34,9 @@ function unwrap(result: Result): T { * without the atomicity of the real remote across processes: for development and tests, with one process writing. * * ```text - * {root}/{session}/slots/{after}.json { after, to, device, folded? }, written once - * {root}/{session}/batches/{after}.batch the records, until folded - * {root}/{session}/snapshots/{to}.batch + * {root}/{session}/slots/{after}.json { after, to, device, archive?, folded? }, written once + * {root}/{session}/batches/{after}.batch the commits (or an archive), until folded + * {root}/{session}/snapshots/{to}.batch a store archive * {root}/{session}/meta.json the host's description, last write wins * ``` */ @@ -56,6 +56,7 @@ export class FileSystemSyncRemote implements SessionSyncMaintenance { device: string, body: Uint8Array, context: Context, + options: { readonly archive?: boolean } = {}, ): Promise { const existing = await this.#slot(sessionId, after, context); if (existing !== undefined) return { type: "exists", to: existing.to, device: existing.device }; @@ -63,7 +64,7 @@ export class FileSystemSyncRemote implements SessionSyncMaintenance { if (after !== head) return { type: "not_head", head }; await this.#write(sessionId, ["batches"], `${pad(after)}.batch`, body, context); // The slot is written last: it is what makes the batch exist. - const slot: BatchInfo = { after, to, device }; + const slot: BatchInfo = { after, to, device, ...(options.archive === true ? { archive: true as const } : {}) }; await this.#write( sessionId, ["slots"], diff --git a/packages/durable/src/sync/index.ts b/packages/durable/src/sync/index.ts index d13153cde..4db8ca543 100644 --- a/packages/durable/src/sync/index.ts +++ b/packages/durable/src/sync/index.ts @@ -1,24 +1,26 @@ -export { decodeBatch, encodeBatch } from "./batch.ts"; +export { decodeBatch, decodeSnapshot, encodeBatch, encodeSnapshot } from "./batch.ts"; export { FlushScheduler, type FlushSchedulerOptions, foldSession, isUrgentCommit, - type LogSource, + type LocalCopy, + OutboxUnavailableError, pullSession, - readLog, + readOutbox, type SyncState, type SyncStateStore, + storeHead, UnsyncedCommitsError, type UploadResult, uploadSession, } from "./client.ts"; -export { type CurrentOnlyLookup, compactRecords } from "./compact.ts"; export { FileSystemSyncRemote } from "./filesystem.ts"; export { decodeLogLines, encodeLogLine, LOG_VERSION, LogFormatError } from "./lines.ts"; export { type BatchInfo, HistoryUnavailableError, + LOG_START, type PutBatchResult, type RemoteHistory, type RemoteListing, diff --git a/packages/durable/src/sync/lines.ts b/packages/durable/src/sync/lines.ts index 101e9a244..8228c0201 100644 --- a/packages/durable/src/sync/lines.ts +++ b/packages/durable/src/sync/lines.ts @@ -1,13 +1,16 @@ -import { seqFromNumber } from "../ids.ts"; -import type { LogRecord, LogWrite } from "../types.ts"; +import type { OutboxCommit } from "@earendil-works/chord/next"; -/** Version of the log line envelope. It changes when a `LogWrite` variant is added, removed, or changes shape. */ -export const LOG_VERSION = 1; +/** + * Version of the line envelope. It changes when the envelope changes; the writes inside carry the store outbox's own + * `format`, which `applyBatch()` checks. + */ +export const LOG_VERSION = 2; type LogLine = { readonly v: typeof LOG_VERSION; readonly seq: number; - readonly writes: readonly LogWrite[]; + readonly nextId: number; + readonly writes: OutboxCommit["writes"]; }; export class LogFormatError extends Error { @@ -17,35 +20,36 @@ export class LogFormatError extends Error { } } -/** One JSON line, terminated by a newline. */ -export function encodeLogLine(record: LogRecord): string { - const line: LogLine = { v: LOG_VERSION, seq: record.seq, writes: record.writes }; +/** One outbox commit as a JSON line, terminated by a newline. */ +export function encodeLogLine(commit: OutboxCommit): string { + const line: LogLine = { v: LOG_VERSION, seq: commit.seq, nextId: commit.nextId, writes: commit.writes }; 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. + * Decode JSON lines. A last line without its newline is a torn append and is returned as `torn` instead of a commit; + * anything else that does not decode is an error. A line of an unknown version is an error too: an older pi must not + * apply writes it does not understand. The writes themselves are checked by `applyBatch()`, a trust boundary. */ -export function decodeLogLines(text: string): { readonly records: LogRecord[]; readonly torn: number } { - const records: LogRecord[] = []; +export function decodeLogLines(text: string): { readonly commits: OutboxCommit[]; readonly torn: number } { + const commits: OutboxCommit[] = []; 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); + if (end === -1) return { commits, torn: text.length - start }; + const commit = decodeLine(text.slice(start, end), commits.length + 1); + if (commit.seq <= previous) { + throw new LogFormatError(`Log line ${commits.length + 1} does not increase the sequence`); + } + previous = commit.seq; + commits.push(commit); start = end + 1; } - return { records, torn: 0 }; + return { commits, torn: 0 }; } -function decodeLine(text: string, number: number): LogRecord { +function decodeLine(text: string, number: number): OutboxCommit { let value: unknown; try { value = JSON.parse(text); @@ -60,21 +64,9 @@ function decodeLine(text: string, number: number): LogRecord { 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)}`); - } + if (typeof line.nextId !== "number" || !Number.isSafeInteger(line.nextId)) { + throw new LogFormatError(`Log line ${number} has no valid next id`); } - return { seq: seqFromNumber(line.seq), writes: line.writes }; + if (!Array.isArray(line.writes)) throw new LogFormatError(`Log line ${number} has no writes`); + return { seq: line.seq, nextId: line.nextId, writes: line.writes }; } - -const LOG_WRITE_TYPES: ReadonlySet = new Set([ - "conversation", - "entry", - "task", - "submission", - "document.create", - "document.change", - "document.retire", -]); diff --git a/packages/durable/src/sync/remote.ts b/packages/durable/src/sync/remote.ts index b18797415..a1b474d6f 100644 --- a/packages/durable/src/sync/remote.ts +++ b/packages/durable/src/sync/remote.ts @@ -1,9 +1,12 @@ import type { Context } from "@earendil-works/chord"; -import { LOG_START } from "../storage/log.ts"; +import { seqFromNumber } from "../ids.ts"; import type { JsonObject, Seq } from "../types.ts"; +/** The sequence before every commit: a store without commits is at it. */ +export const LOG_START = seqFromNumber(0); + /** - * A slot of a session's remote history: the batch of log records after `after`, up to `to`, uploaded by `device`. A + * A slot of a session's remote history: the batch of outbox commits after `after`, up to `to`, uploaded by `device`. A * store creates each slot once and never replaces it, so the device whose upload creates it continues the session from * `after`, and a device that finds it taken by another one holds a different history. */ @@ -11,11 +14,16 @@ export type BatchInfo = { readonly after: Seq; readonly to: Seq; readonly device: string; - /** Folding dropped its records into a snapshot; the slot stays taken. */ + /** Folding dropped its commits into a snapshot; the slot stays taken. */ readonly folded?: true; + /** + * The body is a store archive at `to` instead of commits: the first upload of a session whose outbox does not reach + * back to its first commit. Only the slot after 0 can be one. + */ + readonly archive?: true; }; -/** What a remote store holds for one session. A snapshot is a complete log up to its sequence, folded from batches. */ +/** What a remote store holds for one session. A snapshot is a store archive at its sequence, folded from batches. */ export type RemoteListing = { readonly batches: readonly BatchInfo[]; readonly snapshots: readonly Seq[]; @@ -58,8 +66,9 @@ export interface SessionSyncStore { device: string, body: Uint8Array, context: Context, + options?: { readonly archive?: boolean }, ): Promise; - /** A batch's records; `undefined` when its slot is folded or absent. */ + /** A batch's body; `undefined` when its slot is folded or absent. */ getBatch(sessionId: string, after: Seq, context: Context): Promise; getSnapshot(sessionId: string, to: Seq, context: Context): Promise; list(sessionId: string, context: Context): Promise; @@ -79,25 +88,25 @@ export interface SessionSyncStore { /** The remote's own side: folding history into snapshots, which clients never do. */ export interface SessionSyncMaintenance extends SessionSyncStore { putSnapshot(sessionId: string, to: Seq, body: Uint8Array, context: Context): Promise; - /** Drop the records of these slots, keeping the slots taken so nobody can create them again. */ + /** Drop the bodies of these slots, keeping the slots taken so nobody can create them again. */ foldBatches(sessionId: string, afters: readonly Seq[], context: Context): Promise; deleteSnapshots(sessionId: string, snapshots: readonly Seq[], context: Context): Promise; } -/** What a copy replays to reach the remote's head. */ +/** What a copy applies to reach the remote's head. */ export type RemoteHistory = { /** The snapshot a restore starts from. */ readonly snapshot?: Seq; /** - * The batches to replay in order. The first may start before the copy's head, when a pull stopped inside it; its - * records at or below the head are skipped. + * The batches to apply in order. The first may start before the copy's head, when a pull stopped inside it; its + * commits at or below the head are skipped. */ readonly batches: readonly BatchInfo[]; - /** Where replaying them ends. */ + /** Where applying them ends. */ readonly head: Seq; }; -/** The remote's history no longer continues from a copy's head, after folding: restore into a fresh storage. */ +/** The remote's history no longer continues from a copy's head, after folding: restore into a fresh store. */ export class HistoryUnavailableError extends Error { constructor(head: Seq) { super(`The remote history does not continue from ${head}`); @@ -106,7 +115,7 @@ export class HistoryUnavailableError extends Error { } /** - * The history a copy at `from` replays, or a restore without `from`. Slots are created once, so at most one batch + * The history a copy at `from` applies, or a restore without `from`. Slots are created once, so at most one batch * starts at each sequence and the history is a single chain. */ export function remoteHistory(listing: RemoteListing, from?: Seq): RemoteHistory { diff --git a/packages/durable/test/sync-e2e.test.ts b/packages/durable/test/sync-e2e.test.ts index ab5a0d304..8274b654c 100644 --- a/packages/durable/test/sync-e2e.test.ts +++ b/packages/durable/test/sync-e2e.test.ts @@ -1,16 +1,19 @@ -// Two devices continue one session through a remote folder: SQLite files, real Harnesses, the faux model. The hosts' -// side as p2 will run it: a Harness only on a device's own file, upload and pull through their own handles. +// Two devices continue one session through a remote folder: SQLite files holding a Chord store with its outbox, +// Durable's storage on it, real Harnesses, the faux model. The hosts' side as p2 will run it: a Harness only on a +// device's own file, upload and pull through the store while no Harness holds it. import { mkdtemp, rename, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { openStore, type Store } from "@earendil-works/chord/next"; import { fauxAssistantMessage } from "@earendil-works/pi-ai"; import { afterEach, describe, expect, it } from "vitest"; import { NodeExecutionEnv } from "../src/env/node.ts"; -import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts"; -import { FileSyncStateStore, FileSystemSyncRemote, pullSession, readLog, uploadSession } from "../src/sync/index.ts"; -import type { Seq } from "../src/types.ts"; +import { StoreStorage } from "../src/storage/store/index.ts"; +import { FileSyncStateStore, FileSystemSyncRemote, pullSession, readOutbox, uploadSession } from "../src/sync/index.ts"; +import type { Seq, Storage } from "../src/types.ts"; import { allEntries, type ChatSetup, chatSetup, openChat, textOf, unanswered } from "./chat-support.ts"; import { context } from "./session-support.ts"; +import { backend, SESSION_OWNER } from "./sync-support.ts"; const SESSION = "s1"; const directories: string[] = []; @@ -25,29 +28,47 @@ async function temp(prefix: string): Promise { return directory; } +/** Durable's storage on a store that the storage's close closes too: the Harness owns the store while it runs. */ +async function harnessStorage(store: Store): Promise { + const storage = await StoreStorage.open(store, SESSION_OWNER, context); + const close = storage.close.bind(storage); + return Object.assign(storage, { + close: async (ctx: typeof context) => { + await close(ctx); + await store.close(ctx); + }, + }); +} + /** One device: its data directory, its sync state and its faux model. */ async function device(name: string, remote: FileSystemSyncRemote) { const dir = await temp(`pi-durable-e2e-${name}-`); const state = new FileSyncStateStore(new NodeExecutionEnv({ cwd: dir }), join(dir, "sync")); const setup: ChatSetup = chatSetup(); const file = (sessionId: string) => join(dir, `${sessionId}.sqlite`); - /** Runs `fn` on a storage handle of its own, as the host's uploader does next to a worker. */ - const withStorage = async ( - sessionId: string, - fn: (storage: Awaited>) => Promise, - ): Promise => { - const storage = await openNodeSqliteStorage(file(sessionId)); + const exists = async (sessionId: string) => + (await import("node:fs/promises")).access(file(sessionId)).then( + () => true, + () => false, + ); + /** Runs `fn` on the session's store with its outbox, as the host's uploader does when no worker runs it. */ + const withStore = async (sessionId: string, fn: (store: Store) => Promise): Promise => { + const store = await openStore(await backend(file(sessionId)), context, { outbox: true }); try { - return await fn(storage); + return await fn(store); } finally { - await storage.close(context); + await store.close(context); } }; return { name, setup, state, - open: async (sessionId = SESSION) => openChat(await openNodeSqliteStorage(file(sessionId)), setup), + open: async (sessionId = SESSION) => + openChat( + await harnessStorage(await openStore(await backend(file(sessionId)), context, { outbox: true })), + setup, + ), /** Renames a stopped copy. The new ID has no remote history, so no sync state either. */ rename: async (from: string, to: string) => { for (const suffix of ["", "-wal", "-shm"]) @@ -56,16 +77,29 @@ async function device(name: string, remote: FileSystemSyncRemote) { }); await state.delete(from, context); }, - log: (after: Seq | undefined, sessionId = SESSION) => - withStorage(sessionId, (storage) => readLog(storage, after, context)), - head: (sessionId = SESSION): Promise => - withStorage(sessionId, (storage) => storage.head(context)), + /** Removes a stopped copy so the next pull restores it fresh. */ + forget: async (sessionId = SESSION) => { + for (const suffix of ["", "-wal", "-shm"]) await rm(file(sessionId) + suffix, { force: true }); + await state.delete(sessionId, context); + }, + outbox: (after: Seq, sessionId = SESSION) => + withStore(sessionId, (store) => readOutbox(store, after, 1000, context)), + head: (sessionId = SESSION): Promise => withStore(sessionId, async (store) => store.lastSeq), upload: (sessionId = SESSION) => - withStorage(sessionId, (source) => + withStore(sessionId, (source) => uploadSession({ sessionId, device: name, source, store: remote, state, context }), ), - pull: (sessionId = SESSION) => - withStorage(sessionId, (storage) => pullSession({ sessionId, store: remote, storage, state, context })), + pull: async (sessionId = SESSION) => { + const local = (await exists(sessionId)) + ? ({ + kind: "store", + store: await openStore(await backend(file(sessionId)), context, { outbox: true }), + } as const) + : ({ kind: "empty", backend: await backend(file(sessionId)) } as const); + const result = await pullSession({ sessionId, store: remote, local, state, context }); + await result.store.close(context); + return result.head; + }, }; } @@ -157,11 +191,10 @@ describe("session sync end to end", () => { const viewer = await desktop.open(); expect((await allEntries(viewer.root)).map((entry) => entry.kind)).toEqual(["pi.user"]); await viewer.harness.close(context); - const written = await desktop.log(before); + // Opening the Harness wrote one commit: recovery resets the running generation to pending. + const written = await desktop.outbox(before as Seq); expect(written).toHaveLength(1); - expect(written[0]!.writes).toMatchObject([ - { type: "task", value: { state: { status: "pending" }, restarts: 1 } }, - ]); + expect(JSON.stringify(written[0]!.writes)).toContain('"restarts":1'); expect((await desktop.upload()).type).toBe("uploaded"); await root.abort(context); @@ -186,6 +219,7 @@ describe("session sync end to end", () => { // The desktop renames its copy, uploads all of it under the new ID, and restores the session from the remote. await desktop.rename(SESSION, "s1-fork"); expect((await desktop.upload("s1-fork")).type).toBe("uploaded"); + await desktop.forget(); await desktop.pull(); const laptopTurn = ["user:hi", "assistant:one", "user:laptop", "assistant:from laptop"]; diff --git a/packages/durable/test/sync-remote.test.ts b/packages/durable/test/sync-remote.test.ts index bdee7fd66..10f5e1e06 100644 --- a/packages/durable/test/sync-remote.test.ts +++ b/packages/durable/test/sync-remote.test.ts @@ -1,13 +1,14 @@ +// The sync protocol over stores with an outbox: slots created once, uploads from the outbox, pulls through +// `applyBatch()`, archives for sessions whose outbox does not reach back, folds into snapshots. import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import type { Context } from "@earendil-works/chord"; +import type { Context, JsonValue } from "@earendil-works/chord"; import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import { openStore, type Store } from "@earendil-works/chord/next"; import { afterEach, describe, expect, it } from "vitest"; import { NodeExecutionEnv } from "../src/env/node.ts"; -import { seqFromNumber } from "../src/ids.ts"; -import { LOG_START } from "../src/storage/log.ts"; -import { MemoryStorage } from "../src/storage/memory.ts"; +import { idFromNumber, seqFromNumber } from "../src/ids.ts"; import { type BatchInfo, decodeBatch, @@ -18,7 +19,10 @@ import { foldSession, HistoryUnavailableError, isUrgentCommit, + LOG_START, + OutboxUnavailableError, pullSession, + readOutbox, remoteHead, remoteHistory, type SessionSyncMaintenance, @@ -27,14 +31,25 @@ import { UnsyncedCommitsError, uploadSession, } from "../src/sync/index.ts"; -import type { CommitPublication, LogRecord, ReplayableStorage, Seq } from "../src/types.ts"; -import { collect, entry, reads, seedBatches } from "./replay-support.ts"; +import { + type CommitPublication, + type DocumentCreate, + type DocumentId, + type EntryId, + ROOT_CONVERSATION_ID, + type Seq, + type StorageWrite, + type TaskId, +} from "../src/types.ts"; +import { backend, commitAll, reads, sessionStore } from "./sync-support.ts"; const context = BACKGROUND_CONTEXT; const SESSION = "session-1"; const directories: string[] = []; +const stores: Store[] = []; afterEach(async () => { + for (const store of stores.splice(0)) await store.close(context); for (const directory of directories.splice(0)) await rm(directory, { recursive: true, force: true }); }); @@ -54,21 +69,139 @@ class MemoryState implements SyncStateStore { } } -/** One device with its copy of a session, which it continues as its own. */ -function device(name: string, store: FileSystemSyncRemote, storage: ReplayableStorage = new MemoryStorage()) { - const state = new MemoryState(); - return { - name, - storage, - state, - upload: (sessionId = SESSION) => - uploadSession({ sessionId, device: name, source: storage, store, state, context }), - pull: (sessionId = SESSION) => pullSession({ sessionId, store, storage, state, context }), - }; +/** `store` with some methods replaced: the remote keeps private fields, so the rest stay bound to it. */ +function overriding(store: FileSystemSyncRemote, overrides: Partial): SessionSyncMaintenance { + return new Proxy(store, { + get(target, property) { + const own = (overrides as Record)[property]; + if (own !== undefined) return own; + const value = Reflect.get(target, property, target); + return typeof value === "function" ? value.bind(target) : value; + }, + }); } -async function commitAll(storage: ReplayableStorage, batches: readonly (readonly unknown[])[]): Promise { - for (const batch of batches) await storage.commit(batch as never, () => {}, context); +const tracked = (store: Store): Store => { + stores.push(store); + return store; +}; + +/** + * One device with its copy of a session, which it continues as its own. A copy starts empty: its first pull restores + * it into a fresh backend, or its first commit opens a store with an outbox. + */ +function device(name: string, remoteStore: FileSystemSyncRemote) { + const state = new MemoryState(); + const copies = new Map(); + const self = { + name, + state, + /** The copy's store, opened on first use. */ + async store(sessionId = SESSION): Promise { + let store = copies.get(sessionId); + if (store === undefined) { + store = tracked(await sessionStore()); + copies.set(sessionId, store); + } + return store; + }, + has: (sessionId = SESSION) => copies.has(sessionId), + async commit(batches: readonly (readonly StorageWrite[])[], sessionId = SESSION): Promise { + await commitAll(await self.store(sessionId), batches); + }, + async upload(sessionId = SESSION) { + return uploadSession({ + sessionId, + device: name, + source: await self.store(sessionId), + store: remoteStore, + state, + context, + }); + }, + async pull(sessionId = SESSION): Promise { + const existing = copies.get(sessionId); + const result = await pullSession({ + sessionId, + store: remoteStore, + local: + existing === undefined + ? { kind: "empty", backend: await backend() } + : { kind: "store", store: existing }, + state, + context, + }); + if (existing === undefined) copies.set(sessionId, tracked(result.store)); + return result.head; + }, + /** Drops the copy, as a host does before restoring it fresh. */ + async forget(sessionId = SESSION): Promise { + const store = copies.get(sessionId); + copies.delete(sessionId); + await state.set(sessionId, { syncedHead: undefined }, context); + if (store !== undefined) { + stores.splice(stores.indexOf(store), 1); + await store.close(context); + } + }, + /** Renames a copy: the new session id has no remote history, so no sync state either. */ + rename(from: string, to: string): void { + const store = copies.get(from)!; + copies.delete(from); + copies.set(to, store); + state.states.delete(from); + }, + }; + return self; +} + +const entryId = (id: number) => idFromNumber(id); +const documentId = (id: number) => idFromNumber(id); +const entry = (id: number) => + ({ + type: "entry", + value: { id: entryId(id), conversationId: ROOT_CONVERSATION_ID, kind: "message", data: { id } }, + }) as const satisfies StorageWrite; +const sessionDoc = (id: number): DocumentCreate => ({ + id: documentId(id), + kind: "test.session", + scope: { kind: "session" }, +}); +const task = (id: number, status: "pending" | "running") => + ({ + type: "task", + value: { + id: idFromNumber>(id), + conversationId: ROOT_CONVERSATION_ID, + kind: "test.task", + version: 1, + input: { id }, + background: false, + abortRequested: false, + state: { status, checkpoint: { status } }, + }, + }) as const satisfies StorageWrite; + +/** Ten commits of entries, a task updated in place, and a session document with deltas and a new base. */ +function seedBatches(): StorageWrite[][] { + return [ + [{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], + [entry(2), task(3, "pending")], + [ + { + type: "document.create", + record: sessionDoc(4), + content: { kind: "base", version: 1, value: { n: 0 } }, + }, + ], + [entry(5), task(3, "running")], + [{ type: "document.change", id: documentId(4), content: { kind: "delta", version: 1, ops: [["s", ["n"], 1]] } }], + [entry(6)], + [{ type: "document.change", id: documentId(4), content: { kind: "base", version: 1, value: { n: 5 } } }], + [entry(7)], + [entry(8), task(9, "pending")], + [entry(10)], + ]; } const slot = (after: number, to: number, device: string): BatchInfo => ({ @@ -78,13 +211,14 @@ const slot = (after: number, to: number, device: string): BatchInfo => ({ }); describe("batches", () => { - it("round-trip through gzip JSON lines", async () => { - const storage = new MemoryStorage(); - await commitAll(storage, seedBatches()); - const records = await collect(storage.log(undefined, context)); - const bytes = await encodeBatch(records); + it("round-trip outbox commits through gzip JSON lines", async () => { + const store = tracked(await sessionStore()); + await commitAll(store, seedBatches()); + const commits = await readOutbox(store, LOG_START, 1000, context); + expect(commits.length).toBeGreaterThan(0); + const bytes = await encodeBatch(commits); expect(bytes[0]).toBe(0x1f); - expect(await decodeBatch(bytes)).toEqual(records); + expect(await decodeBatch(bytes)).toEqual(commits); }); }); @@ -114,16 +248,22 @@ describe("remoteHistory", () => { }); describe("syncing a session between devices", () => { - it("uploads what was committed since the last upload, and nothing when nothing was", async () => { + it("uploads what was committed since the last upload, trims the outbox, and uploads nothing when nothing was", async () => { const store = await remote(); const laptop = device("laptop", store); const batches = seedBatches(); - await commitAll(laptop.storage, batches.slice(0, 4)); - expect(await laptop.upload()).toEqual({ type: "uploaded", head: 4 }); + await laptop.commit(batches.slice(0, 4)); + const four = (await laptop.store()).lastSeq; + expect(await laptop.upload()).toEqual({ type: "uploaded", head: four }); + expect((await laptop.store()).outbox!.floor).toBe(four); expect(await laptop.upload()).toEqual({ type: "idle" }); - await commitAll(laptop.storage, batches.slice(4)); - expect(await laptop.upload()).toEqual({ type: "uploaded", head: 10 }); - expect((await store.list(SESSION, context)).batches).toEqual([slot(0, 4, "laptop"), slot(4, 10, "laptop")]); + await laptop.commit(batches.slice(4)); + const ten = (await laptop.store()).lastSeq; + expect(await laptop.upload()).toEqual({ type: "uploaded", head: ten }); + expect((await store.list(SESSION, context)).batches).toEqual([ + slot(0, four, "laptop"), + slot(four, ten, "laptop"), + ]); }); it("hands a session back and forth: each device continues it as its own", async () => { @@ -131,37 +271,37 @@ describe("syncing a session between devices", () => { const laptop = device("laptop", store); const phone = device("phone", store); const batches = seedBatches(); - await commitAll(laptop.storage, batches.slice(0, 5)); + await laptop.commit(batches.slice(0, 5)); await laptop.upload(); - expect(await phone.pull()).toBe(5); - expect(await reads(phone.storage)).toEqual(await reads(laptop.storage)); - await commitAll(phone.storage, batches.slice(5, 7)); - expect(await phone.upload()).toEqual({ type: "uploaded", head: 7 }); + expect(await phone.pull()).toBe((await laptop.store()).lastSeq); + expect(await reads(await phone.store())).toEqual(await reads(await laptop.store())); + await phone.commit(batches.slice(5, 7)); + expect((await phone.upload()).type).toBe("uploaded"); - expect(await laptop.pull()).toBe(7); - expect(await reads(laptop.storage)).toEqual(await reads(phone.storage)); - await commitAll(laptop.storage, batches.slice(7)); - expect(await laptop.upload()).toEqual({ type: "uploaded", head: 10 }); - expect(await phone.pull()).toBe(10); - expect(await reads(phone.storage)).toEqual(await reads(laptop.storage)); - expect((await store.list(SESSION, context)).batches).toEqual([ - slot(0, 5, "laptop"), - slot(5, 7, "phone"), - slot(7, 10, "laptop"), + await laptop.pull(); + expect(await reads(await laptop.store())).toEqual(await reads(await phone.store())); + await laptop.commit(batches.slice(7)); + expect((await laptop.upload()).type).toBe("uploaded"); + await phone.pull(); + expect(await reads(await phone.store())).toEqual(await reads(await laptop.store())); + expect((await store.list(SESSION, context)).batches.map((batch) => batch.device)).toEqual([ + "laptop", + "phone", + "laptop", ]); }); - it("keeps a copy that only pulls in step", async () => { + it("keeps a copy that only pulls in step, with nothing left to upload", async () => { const store = await remote(); const laptop = device("laptop", store); const viewer = device("viewer", store); for (const writes of seedBatches()) { - await laptop.storage.commit(writes, () => {}, context); + await laptop.commit([writes]); await laptop.upload(); await viewer.pull(); } - expect(await reads(viewer.storage)).toEqual(await reads(laptop.storage)); + expect(await reads(await viewer.store())).toEqual(await reads(await laptop.store())); expect(await viewer.upload()).toEqual({ type: "idle" }); }); @@ -173,28 +313,31 @@ describe("syncing a session between devices", () => { const store = await remote(); const devices = { laptop: device("laptop", store), phone: device("phone", store) }; const batches = seedBatches(); - await commitAll(devices.laptop.storage, batches.slice(0, 5)); + await devices.laptop.commit(batches.slice(0, 5)); await devices.laptop.upload(); await devices.phone.pull(); - // Both continue from 5 without hearing from each other. - await commitAll(devices.laptop.storage, batches.slice(5, 7)); - await commitAll(devices.phone.storage, [[entry(100)]]); + const shared = (await devices.laptop.store()).lastSeq; + // Both continue without hearing from each other. + await devices.laptop.commit(batches.slice(5, 7)); + await devices.phone.commit([[entry(100)]]); const winner = devices[first]; const loser = devices[second]; expect((await winner.upload()).type).toBe("uploaded"); - // The loser's acknowledged uploads stay; only the commits nobody acknowledged are a different history. - expect(await loser.upload()).toEqual({ type: "diverged", at: 5 }); + expect(await loser.upload()).toEqual({ type: "diverged", at: shared }); await expect(loser.pull()).rejects.toThrow(UnsyncedCommitsError); - // The loser's copy becomes a new session, uploaded whole; the session restores fresh on that device. - expect(await loser.upload("fork-1")).toEqual({ type: "uploaded", head: await loser.storage.head(context) }); + // The loser's copy becomes a new session, uploaded whole as an archive (its outbox was trimmed through + // `shared`); the session restores fresh on that device. + loser.rename(SESSION, "fork-1"); + const forked = await loser.store("fork-1"); + expect(await loser.upload("fork-1")).toEqual({ type: "uploaded", head: forked.lastSeq }); + expect((await store.list("fork-1", context)).batches[0]).toMatchObject({ after: 0, archive: true }); const fork = device(second, store); await fork.pull("fork-1"); - expect(await reads(fork.storage)).toEqual(await reads(loser.storage)); - const restored = device(second, store); - await restored.pull(); - expect(await reads(restored.storage)).toEqual(await reads(winner.storage)); + expect(await reads(await fork.store("fork-1"))).toEqual(await reads(forked)); + await loser.pull(); + expect(await reads(await loser.store())).toEqual(await reads(await winner.store())); }); } @@ -203,91 +346,139 @@ describe("syncing a session between devices", () => { const laptop = device("laptop", store); const replica = device("replica", store); const batches = seedBatches(); - await commitAll(laptop.storage, batches.slice(0, 5)); + await laptop.commit(batches.slice(0, 5)); await laptop.upload(); - await commitAll(laptop.storage, batches.slice(5, 7)); - await laptop.upload(); - // The response of that upload was lost: the laptop still believes the remote is at 5. - await laptop.state.set(SESSION, { syncedHead: seqFromNumber(5) }, context); - await commitAll(laptop.storage, batches.slice(7)); - expect(await laptop.upload()).toEqual({ type: "uploaded", head: 10 }); - + const five = (await laptop.store()).lastSeq; + await laptop.commit(batches.slice(5, 7)); + const seven = (await laptop.store()).lastSeq; + // The remote creates the slot, but its response is lost: no sync state is written, nothing is trimmed. + const lossy: SessionSyncMaintenance = overriding(store, { + putBatch: async (...args: Parameters) => { + await store.putBatch(...args); + throw new Error("response lost"); + }, + }); + await expect( + uploadSession({ + sessionId: SESSION, + device: "laptop", + source: await laptop.store(), + store: lossy, + state: laptop.state, + context, + }), + ).rejects.toThrow("response lost"); + expect(await laptop.state.get(SESSION, context)).toEqual({ syncedHead: five }); + await laptop.commit(batches.slice(7)); + const ten = (await laptop.store()).lastSeq; + expect(await laptop.upload()).toEqual({ type: "uploaded", head: ten }); expect((await store.list(SESSION, context)).batches).toEqual([ - slot(0, 5, "laptop"), - slot(5, 7, "laptop"), - slot(7, 10, "laptop"), + slot(0, five, "laptop"), + slot(five, seven, "laptop"), + slot(seven, ten, "laptop"), ]); await replica.pull(); - expect(await reads(replica.storage)).toEqual(await reads(laptop.storage)); + expect(await reads(await replica.store())).toEqual(await reads(await laptop.store())); }); it("refuses to upload a copy that lost uploaded commits, and pulls them back", async () => { const store = await remote(); const batches = seedBatches(); const laptop = device("laptop", store); - await commitAll(laptop.storage, batches.slice(0, 7)); + await laptop.commit(batches.slice(0, 7)); await laptop.upload(); - // A power loss kept only the first 5 commits; the sync state still says 7. - const survived = new MemoryStorage(); + const uploaded = (await laptop.store()).lastSeq; + // A power loss kept only the first commits: a store that holds less than the sync state says. + const survived = tracked(await sessionStore()); await commitAll(survived, batches.slice(0, 5)); const upload = () => uploadSession({ sessionId: SESSION, device: "laptop", source: survived, store, state: laptop.state, context }); - expect(await upload()).toEqual({ type: "behind", head: 5, syncedHead: 7 }); + expect(await upload()).toEqual({ type: "behind", head: survived.lastSeq, syncedHead: uploaded }); - await pullSession({ sessionId: SESSION, store, storage: survived, state: laptop.state, context }); - expect(await reads(survived)).toEqual(await reads(laptop.storage)); + await pullSession({ + sessionId: SESSION, + store, + local: { kind: "store", store: survived }, + state: laptop.state, + context, + }); + expect(await reads(survived)).toEqual(await reads(await laptop.store())); await commitAll(survived, batches.slice(7)); - expect(await upload()).toEqual({ type: "uploaded", head: 10 }); + expect((await upload()).type).toBe("uploaded"); }); it("creates slots only where the history ends", async () => { const store = await remote(); const batches = seedBatches(); const laptop = device("laptop", store); - await commitAll(laptop.storage, batches.slice(0, 5)); + await laptop.commit(batches.slice(0, 5)); await laptop.upload(); - // A copy that believes it shares 3 with the remote: no slot starts there, and the history ends at 5. + const head = (await laptop.store()).lastSeq; + // A copy that believes it shares 1 with the remote: no slot starts there, and the history ends further on. const desktop = device("desktop", store); - await commitAll(desktop.storage, batches.slice(0, 4)); - await desktop.state.set(SESSION, { syncedHead: seqFromNumber(3) }, context); - const records = await collect(desktop.storage.log(seqFromNumber(3), context)); + await desktop.commit(batches.slice(0, 2)); + await desktop.state.set(SESSION, { syncedHead: seqFromNumber(1) }, context); + const commits = await readOutbox(await desktop.store(), seqFromNumber(1), 100, context); expect( await store.putBatch( SESSION, - seqFromNumber(3), - seqFromNumber(4), + seqFromNumber(1), + seqFromNumber(commits.at(-1)!.seq), "desktop", - await encodeBatch(records), + await encodeBatch(commits), context, ), - ).toEqual({ type: "not_head", head: 5 }); - expect(await desktop.upload()).toEqual({ type: "diverged", at: 3 }); - expect((await store.list(SESSION, context)).batches).toEqual([slot(0, 5, "laptop")]); + ).toEqual({ type: "not_head", head }); + expect(await desktop.upload()).toEqual({ type: "diverged", at: 1 }); + expect((await store.list(SESSION, context)).batches).toEqual([slot(0, head, "laptop")]); }); it("resumes a pull that stopped inside a batch", async () => { const store = await remote(); const laptop = device("laptop", store); - await commitAll(laptop.storage, seedBatches().slice(0, 3)); + await laptop.commit(seedBatches().slice(0, 3)); await laptop.upload(); - await commitAll(laptop.storage, seedBatches().slice(3)); + await laptop.commit(seedBatches().slice(3)); await laptop.upload(); - const storage = new MemoryStorage(); - let replayed = 0; - const failing = Object.assign(Object.create(storage) as MemoryStorage, { - replay: async (record: LogRecord, replayContext: Context) => { - if (++replayed === 5) throw new Error("connection lost"); - return storage.replay(record, replayContext); + // The viewer applied the first batch, then lost the connection while reading the second. + const viewer = device("viewer", store); + const flaky: SessionSyncMaintenance = overriding(store, { + getBatch: async (sessionId: string, after: Seq, ctx: Context) => { + if (after > 0) throw new Error("connection lost"); + return store.getBatch(sessionId, after, ctx); }, }); - const viewer = device("viewer", store, failing); - await expect(viewer.pull()).rejects.toThrow("connection lost"); - expect(await storage.head(context)).toBe(4); - // The records past what it shares are the remote's, not its own: nothing to upload, and the pull resumes. - expect(await viewer.upload()).toEqual({ type: "idle" }); - await viewer.pull(); - expect(await reads(storage)).toEqual(await reads(laptop.storage)); + const local = tracked(await openStore(await backend(), context, { outbox: true })); + await expect( + pullSession({ + sessionId: SESSION, + store: flaky, + local: { kind: "store", store: local }, + state: viewer.state, + context, + }), + ).rejects.toThrow("connection lost"); + expect(local.lastSeq).toBeGreaterThan(0); + // The commits past what it shares are the remote's, not its own: nothing to upload, and the pull resumes. + expect( + await uploadSession({ + sessionId: SESSION, + device: "viewer", + source: local, + store, + state: viewer.state, + context, + }), + ).toEqual({ type: "idle" }); + await pullSession({ + sessionId: SESSION, + store, + local: { kind: "store", store: local }, + state: viewer.state, + context, + }); + expect(await reads(local)).toEqual(await reads(await laptop.store())); }); it("folds the history into a snapshot that restores the session, and keeps its slots taken", async () => { @@ -296,54 +487,94 @@ describe("syncing a session between devices", () => { const stale = device("stale", store); const replica = device("replica", store); for (const [index, writes] of seedBatches().entries()) { - await laptop.storage.commit(writes, () => {}, context); + await laptop.commit([writes]); await laptop.upload(); if (index === 3) await stale.pull(); } - expect(await foldSession({ sessionId: SESSION, store, context })).toBe(10); - expect((await store.list(SESSION, context)).snapshots).toEqual([10]); + const head = (await laptop.store()).lastSeq; + expect(await foldSession({ sessionId: SESSION, store, scratch: await backend(), context })).toBe(head); + expect((await store.list(SESSION, context)).snapshots).toEqual([head]); await replica.pull(); - expect(await reads(replica.storage)).toEqual(await reads(laptop.storage)); - // A copy at 4 cannot continue through folded batches: it restores instead. + expect(await reads(await replica.store())).toEqual(await reads(await laptop.store())); + // A copy behind the fold cannot continue through folded batches: it restores instead. + const at = (await stale.store()).lastSeq; await expect(stale.pull()).rejects.toThrow(HistoryUnavailableError); - // Nor can it take a folded slot: its own commits from 4 on are a different history. - await stale.storage.commit([entry(300)], () => {}, context); - expect(await stale.upload()).toEqual({ type: "diverged", at: 4 }); + // Nor can it take a folded slot: its own commits are a different history. + await stale.commit([[entry(300)]]); + expect(await stale.upload()).toEqual({ type: "diverged", at }); + await stale.forget(); + await stale.pull(); + expect(await reads(await stale.store())).toEqual(await reads(await laptop.store())); // Uploads go on after the fold, and the replica follows them; so does a second fold and a fresh restore. - await commitAll(laptop.storage, [[entry(100)], [entry(101)]]); + await laptop.commit([[entry(100)], [entry(101)]]); await laptop.upload(); - await laptop.storage.commit([entry(102)], () => {}, context); + await laptop.commit([[entry(102)]]); await laptop.upload(); await replica.pull(); - expect(await reads(replica.storage)).toEqual(await reads(laptop.storage)); - expect(await foldSession({ sessionId: SESSION, store, context })).toBe(13); + expect(await reads(await replica.store())).toEqual(await reads(await laptop.store())); + expect(await foldSession({ sessionId: SESSION, store, scratch: await backend(), context })).toBe( + (await laptop.store()).lastSeq, + ); const restored = device("restored", store); await restored.pull(); - expect(await reads(restored.storage)).toEqual(await reads(laptop.storage)); + expect(await reads(await restored.store())).toEqual(await reads(await laptop.store())); }); it("reports a batch that vanished while folding", async () => { const store = await remote(); const laptop = device("laptop", store); for (const writes of seedBatches().slice(0, 3)) { - await laptop.storage.commit(writes, () => {}, context); + await laptop.commit([writes]); await laptop.upload(); } - const vanishing: SessionSyncMaintenance = { - putBatch: (...args) => store.putBatch(...args), - putSnapshot: (...args) => store.putSnapshot(...args), - getSnapshot: (...args) => store.getSnapshot(...args), - foldBatches: (...args) => store.foldBatches(...args), - deleteSnapshots: (...args) => store.deleteSnapshots(...args), - list: (...args) => store.list(...args), - sessions: (...args) => store.sessions(...args), - putMeta: (...args) => store.putMeta(...args), - getMeta: (...args) => store.getMeta(...args), + const vanishing: SessionSyncMaintenance = overriding(store, { getBatch: async () => undefined, - }; - await expect(foldSession({ sessionId: SESSION, store: vanishing, context })).rejects.toThrow("vanished"); + }); + await expect( + foldSession({ sessionId: SESSION, store: vanishing, scratch: await backend(), context }), + ).rejects.toThrow(HistoryUnavailableError); + }); + + it("refuses a store without an outbox", async () => { + const store = await remote(); + const plain = tracked(await openStore(await backend(), context)); + await expect( + uploadSession({ + sessionId: SESSION, + device: "laptop", + source: plain, + store, + state: new MemoryState(), + context, + }), + ).rejects.toThrow(OutboxUnavailableError); + }); + + it("uploads an archive first for a session whose outbox was turned on late", async () => { + const store = await remote(); + const directory = await mkdtemp(join(tmpdir(), "pi-durable-sync-late-")); + directories.push(directory); + const file = join(directory, "session.sqlite"); + const before = await openStore(await backend(file), context); + await commitAll(before, seedBatches().slice(0, 4)); + await before.close(context); + // Reopening with the outbox turns it on from the current head: the earlier commits are not in it. + const source = tracked(await openStore(await backend(file), context, { outbox: true })); + expect(source.outbox!.floor).toBe(source.lastSeq); + await commitAll(source, seedBatches().slice(4)); + const state = new MemoryState(); + expect(await uploadSession({ sessionId: SESSION, device: "laptop", source, store, state, context })).toEqual({ + type: "uploaded", + head: source.lastSeq, + }); + expect((await store.list(SESSION, context)).batches).toEqual([ + { ...slot(0, source.lastSeq, "laptop"), archive: true }, + ]); + const replica = device("replica", store); + await replica.pull(); + expect(await reads(await replica.store())).toEqual(await reads(source)); }); }); @@ -352,13 +583,13 @@ describe("host pieces", () => { const store = await remote(); expect(await store.sessions(undefined, context)).toEqual({ sessions: [], cursor: "", more: false }); const laptop = device("laptop", store); - await laptop.storage.commit([entry(1)], () => {}, context); + await laptop.commit([[entry(1)]], "s2"); await laptop.upload("s2"); await store.putMeta("s1", { title: "first" }, context); await store.putMeta("s1", { title: "renamed", origin: "laptop" }, context); expect((await store.sessions(undefined, context)).sessions).toEqual([ { id: "s1", head: LOG_START, meta: { title: "renamed", origin: "laptop" } }, - { id: "s2", head: 1, meta: undefined }, + { id: "s2", head: (await laptop.store("s2")).lastSeq, meta: undefined }, ]); expect(await store.getMeta("s1", context)).toEqual({ title: "renamed", origin: "laptop" }); expect(await store.getMeta("s2", context)).toBeUndefined(); @@ -493,6 +724,6 @@ describe("FlushScheduler", () => { publication([{ type: "entry", value: { id: 1, conversationId: 1, kind: "pi.user" } }] as never), ), ).toBe(true); - expect(isUrgentCommit(publication([entry(1)] as never))).toBe(false); + expect(isUrgentCommit(publication([{ type: "entry", value: entry(1).value }] as never))).toBe(false); }); }); diff --git a/packages/durable/test/sync-support.ts b/packages/durable/test/sync-support.ts new file mode 100644 index 000000000..48bc7423e --- /dev/null +++ b/packages/durable/test/sync-support.ts @@ -0,0 +1,75 @@ +// A session's copy for the sync tests: a Chord store over a SQLite database with its outbox on, and Durable's storage +// in it under one owner name. The store is what syncs; the storage is opened on it only while something commits. + +import type { Context } from "@earendil-works/chord"; +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import { type Backend, openStore, type Store, sqliteBackend } from "@earendil-works/chord/next"; +import { openNodeSqliteDatabase } from "@earendil-works/chord/sqlite/node"; +import { StoreStorage } from "../src/storage/store/index.ts"; +import { commitWrites } from "../src/testing/commit-writes.ts"; +import type { Storage, StorageWrite } from "../src/types.ts"; + +const context = BACKGROUND_CONTEXT; + +/** The owner name Durable's storage takes in a session store. */ +export const SESSION_OWNER = "pi.session"; + +/** A backend over the SQLite file at `path`, `:memory:` by default. */ +export async function backend(path = ":memory:"): Promise { + return sqliteBackend(await openNodeSqliteDatabase(path)); +} + +/** A session store with its outbox on. */ +export async function sessionStore(path = ":memory:"): Promise { + return openStore(await backend(path), context, { outbox: true }); +} + +/** Runs `fn` on Durable's storage in `store`, released afterwards so the store can be pulled into again. */ +export async function withStorage(store: Store, fn: (storage: Storage) => Promise, ctx: Context = context) { + const storage = await StoreStorage.open(store, SESSION_OWNER, ctx); + try { + return await fn(storage); + } finally { + await storage.close(ctx); + } +} + +/** Commits each batch through Durable's storage in `store`. */ +export async function commitAll(store: Store, batches: readonly (readonly StorageWrite[])[]): Promise { + await withStorage(store, async (storage) => { + for (const batch of batches) await commitWrites(storage, batch, context); + }); +} + +/** + * What a Session reads of the store, as one comparable value: its records and every live document. Two copies that + * read the same hold the same session. + */ +export async function reads(store: Store): Promise> { + return withStorage(store, async (storage) => { + const out: Record = { head: store.lastSeq }; + 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}`] = entries; + } + const tasks = (await storage.scanTasks({}, 1000, undefined, context)).items; + out.tasks = tasks; + out.submissions = (await storage.scanSubmissions({}, 1000, undefined, context)).items; + const scopes = [ + { kind: "session" as const }, + ...conversations.map((conversation) => ({ kind: "conversation" as const, conversationId: conversation.id })), + ...tasks.map((item) => ({ kind: "task" as const, taskId: item.id })), + ]; + for (const scope of scopes) { + const documents = (await storage.scanDocuments({ scope, at: "current" }, 1000, undefined, context)).items; + out[`documents:${JSON.stringify(scope)}`] = await Promise.all( + documents.map(async (record) => ({ record, value: await storage.document(record.id, "current", context) })), + ); + } + return out; + }); +} diff --git a/packages/durable/test/sync.test.ts b/packages/durable/test/sync.test.ts index a9200ed4c..b4accfb3a 100644 --- a/packages/durable/test/sync.test.ts +++ b/packages/durable/test/sync.test.ts @@ -1,148 +1,73 @@ 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 } from "../src/sync/index.ts"; -import type { LogRecord, StorageWrite } from "../src/types.ts"; -import { collect, create, delta, documentId, entry, latestDoc, reads, seedBatches, task } from "./replay-support.ts"; +import { idFromNumber } from "../src/ids.ts"; +import { + decodeLogLines, + decodeSnapshot, + encodeLogLine, + encodeSnapshot, + LOG_START, + LogFormatError, + readOutbox, +} from "../src/sync/index.ts"; +import { type EntryId, ROOT_CONVERSATION_ID } from "../src/types.ts"; +import { commitAll, sessionStore } from "./sync-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 seeded() { + const store = await sessionStore(); + await commitAll(store, [ + [{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], + [ + { + type: "entry", + value: { id: idFromNumber(2), conversationId: ROOT_CONVERSATION_ID, kind: "message", data: {} }, + }, + ], + ]); + return store; } -/** - * Commit `batches` and read the storage's log after each one, as a flush after every commit would upload it; with - * `snapshotAfter`, the first that many commits are read as one complete log instead. - */ -async function uploaded( - batches: readonly (readonly StorageWrite[])[], - snapshotAfter = 0, -): Promise<{ readonly storage: MemoryStorage; readonly snapshot: LogRecord[]; readonly records: LogRecord[] }> { - const storage = new MemoryStorage(); - for (const batch of batches.slice(0, snapshotAfter)) await storage.commit(batch, () => {}, context); - const snapshot = await collect(storage.log(undefined, context)); - const records: LogRecord[] = []; - for (const batch of batches.slice(snapshotAfter)) { - const head = await storage.head(context); - await storage.commit(batch, () => {}, context); - records.push(...(await collect(storage.log(head, context)))); - } - return { storage, snapshot, records }; -} - -describe("compactRecords", () => { - it("folds a snapshot and the batches after it into one that replays to the same reads, at every split", async () => { - const batches = seedBatches(); - for (let split = 0; split <= batches.length; split++) { - const { storage, snapshot, records } = await uploaded(batches, split); - const folded = compactRecords([...snapshot, ...records]); - expect(await reads(await replayed(folded)), `split at ${split}`).toEqual(await reads(storage)); - expect(folded.at(-1)?.seq).toBe(await storage.head(context)); - } - }); - - it("keeps the reads of every run of uploaded batches", async () => { - const { storage, records } = await uploaded(seedBatches()); - const expected = await reads(storage); - for (let cut = 0; cut <= records.length; cut++) { - const compacted = [records.slice(0, cut), records.slice(cut)].flatMap((run) => compactRecords(run)); - expect(await reads(await replayed(compacted)), `cut at ${cut}`).toEqual(expected); - } - }); - - it("leaves a storage's complete log as it is", async () => { - const { storage } = await uploaded(seedBatches()); - const log = await collect(storage.log(undefined, context)); - expect(compactRecords(log)).toEqual(log); - }); - - it("drops superseded task records and document content before a newer base", () => { - const records: LogRecord[] = [ - { seq: seqFromNumber(10), writes: [{ type: "task", value: task(5, "running") }, delta(2, "partial", "a")] }, - { seq: seqFromNumber(11), writes: [delta(2, "partial", "ab")] }, - { 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: {} } }, - ], - }, - { seq: seqFromNumber(14), writes: [delta(2, "partial", "c")] }, - ]; - // Document 2 was created before the run; the lookup says it is current-only, like pi.live. - expect(compactRecords(records, (id) => (id === documentId(2) ? true : undefined))).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: {} } }, - ], - }, - { seq: 14, writes: [delta(2, "partial", "c")] }, - ]); - // Without the lookup its history is unknown, so its content stays. - expect(compactRecords(records).flatMap((record) => record.writes)).toHaveLength(6); - }); - - it("keeps an empty last record so the batch ends where the run ended", () => { - const records: LogRecord[] = [ - { 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", async () => { - const { records } = await uploaded(seedBatches()); - const text = records.map(encodeLogLine).join(""); - expect(decodeLogLines(text)).toEqual({ records, torn: 0 }); + it("round-trips outbox commits", async () => { + const store = await seeded(); + try { + const commits = await readOutbox(store, LOG_START, 100, context); + // The storage's table declarations, then the two commits. + expect(commits.length).toBe(3); + expect(decodeLogLines(commits.map(encodeLogLine).join(""))).toEqual({ commits, torn: 0 }); + } finally { + await store.close(context); + } }); 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 }); + const line = encodeLogLine({ seq: 1, nextId: 2, writes: [] }); + expect(decodeLogLines(line + line.slice(0, 5).replace('"seq":1', '"seq":2'))).toEqual({ + commits: [{ seq: 1, nextId: 2, writes: [] }], + torn: 5, + }); }); - it("refuses unknown versions, unknown writes, and sequences that do not increase", () => { + it("refuses unknown versions, a missing next id, and sequences that do not increase", () => { + expect(() => decodeLogLines('{"v":1,"seq":1,"nextId":2,"writes":[]}\n')).toThrow(LogFormatError); 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) })); + const line = encodeLogLine({ seq: 3, nextId: 4, writes: [] }); + expect(() => decodeLogLines(line + line)).toThrow("does not increase"); + }); +}); + +describe("snapshots", () => { + it("round-trip a store archive", async () => { + const store = await seeded(); + try { + const parts: unknown[] = []; + for await (const part of store.export(context)) parts.push(part); + expect(await decodeSnapshot(await encodeSnapshot(store.export(context)))).toEqual(parts); + expect(parts[0]).toMatchObject({ kind: "manifest", lastSeq: store.lastSeq }); + } finally { + await store.close(context); + } }); });