diff --git a/apps/p2/tsconfig.json b/apps/p2/tsconfig.json index 5641cf3da..afc80b2a6 100644 --- a/apps/p2/tsconfig.json +++ b/apps/p2/tsconfig.json @@ -30,6 +30,7 @@ "@earendil-works/chord/react": ["../../packages/chord/src/react/index.ts"], "@earendil-works/chord/tui": ["../../packages/chord/src/tui/index.ts"], "@earendil-works/chord/tui/launcher": ["../../packages/chord/src/node/tui/index.ts"], + "@earendil-works/chord/sync": ["../../packages/chord/src/sync/index.ts"], "@earendil-works/pi-telemetry": ["../../packages/telemetry/src/index.ts"], "@earendil-works/pi-telemetry/testing": ["../../packages/telemetry/src/testing/index.ts"], "@earendil-works/pi-telemetry/render": ["../../packages/telemetry/src/render.ts"], diff --git a/packages/chord/CHANGELOG.md b/packages/chord/CHANGELOG.md index e3e8c0632..aa89bdc73 100644 --- a/packages/chord/CHANGELOG.md +++ b/packages/chord/CHANGELOG.md @@ -11,6 +11,7 @@ ### Added +- The sync layer (experimental): `@earendil-works/chord/sync` moves a store between devices through a remote, from the store's outbox (`openStore(backend, ctx, { outbox: true })`). Every device continues a store 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 store 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` decides when to upload. `FileSystemSyncRemote`, a remote in a directory, and `FileSyncStateStore`, sync state in a directory, are for development and tests; they take their file operations as a `SyncFileSystem` (Pi Durable's environments fit it as they are). - Chord v2's Node components and client surfaces are exported: `@earendil-works/chord/http`, `/workers`, `/workers/child` (the worker's side, `workerMain` and `processLock`, without esbuild), `/plugins`, `/plugins/runtime` and `/app`; `/surfaces` (the web and terminal surfaces' server side), `/web`, `/react`, `/tui` and `/tui/launcher`. - Chord v2's plugins component gains the plugin model (slice 12): `createPlugins({ api: { name, versions } })` checks every import of the app's versioned API (`@acme/app/v3`, per-kind modules such as `@acme/app/v3/web`): an unlisted version, another kind's module, two versions in one facet and a direct import of Chord are build errors naming the file and line. Roots may be `{ path, required }`: a required root's plugin failing at `load` rejects it with a `PluginError` after the load settled. Files named `..` (`contract.web.tsx`) are of that kind, refused in facets of another kind with the chain of imports named. Single-file plugins (`hello.server.ts` at a root's top, grouped by name, host packages only). `installPlugin({ git, ref } | { npm, version }, root, ctx)` installs plugins from git (a shallow clone of a branch, tag or commit; a repository of several plugins too) or npm; a package lists the plugin folders it provides in its `package.json` (`"chord": { "plugins": ["./notes", "plugins/*", "!plugins/draft"] }`), used by discovery and `installPlugin` (without it: one plugin when the package's top has facet files, else its subfolders with a `package.json`). `testPlugin(dir, { kind, modules, stubs, principal }, ctx)` builds a plugin as production does and loads it on an in-memory host beside stub facets, collecting its contributions to registries nobody serves and listing the contracts it waits for. Agents' guide: `docs/plugin-authoring.md`. - Chord v2 gains the app kit (slice 13, `@earendil-works/chord/app`): `startServer

({ name, dataDir?, root?, listen?, api, kinds, principal, workers?, files?, install?, onError }, ctx)` lays out the data directory (`server/store.sqlite`, `workers//

/`, `plugins/`, `logs/`, `owner-token`; one server per data directory), creates the server's store and host, HTTP, one workers component per worker kind and the plugins component over the app's required root and the user's root, and loads the server's plugins with the kit's objects; `startWorker({ api, kinds, busy?, onError })` is a worker's entry (its directory and store, the link to the server first, its plugins). The start-data objects `AppInfo` and `WorkerInfo` serve process values to plugins. Sessions (`Sessions`: `issue`, `verify`, `revoke`, `list`; tokens kept as SHA-256 hashes, compared in constant time; persisted ones end after 400 days unused and their browser cookie is renewed daily while used, ephemeral ones live in memory, at most `limits.ephemeralSessions`) authenticate a signed `HttpOnly`, `SameSite=Strict` cookie or `Authorization: Bearer`; a one-time owner link per start (`app.ownerLink()`, the token in the URL's fragment, single use, 10 minutes), an owner token file (mode 0600, written atomically) and device pairing (`POST /_chord/pair`, owner approval, a poll returning the token once; bounded in all and per client address) sign the owner in. The data directory, its stores and secret files are made private (0700, 0600) at every start; the home directory, a file system's root and another user's directory are refused. `Files` per host: small and chunked writes at a stored offset, ranged reads, a watchable index, caps, at most `limits.uploads` uploads under way. `app.exportData(address, path, ctx)` and `app.importData(archive, address, ctx)` move the data a worker created (its store and files) as a tar without stopping anything: a running worker writes its own store at one consistent point, else the server reads it under the worker's process lock (import refusing an existing address, another app's archive, every path leaving the target, and archives past `limits.importBytes` or `limits.importEntries`). `app.applyChanges(ctx)` and `app.plugins.install(source, ctx)`. Optional plugins as exported facets: `defineUsers(profile)`, `inviteFacet()`, `ephemeralLoginFacet()`. diff --git a/packages/chord/README.md b/packages/chord/README.md index 51a96d912..9b27d475d 100644 --- a/packages/chord/README.md +++ b/packages/chord/README.md @@ -36,6 +36,7 @@ The design, the rule book for all of this, is [docs/design.md](docs/design.md). | `@earendil-works/chord/sqlite` | Portable `SqliteDatabase`, `SqliteExecutor`, and `SqliteValue`, atomic `applySqliteMigrations(database, migrations, { table })`, and `validateSqliteSchemaPrefix`. The metadata table must be a valid, non-reserved SQLite identifier; migration versions start at 1 and are contiguous. | Any SQLite adapter | | `@earendil-works/chord/sqlite/node` | `NodeSqliteDatabase`, `openNodeSqliteDatabase`, and `NodeSqliteDatabaseOptions`. | Node | | `@earendil-works/chord/sqlite/cloudflare` | `DurableObjectSqliteDatabase` and the structural `DurableObjectSqliteStorage` binding type. | Cloudflare Durable Objects | +| `@earendil-works/chord/sync` | Sync (experimental): a store's outbox commits uploaded to a remote into slots created once (`uploadSession`), applied on another copy (`pullSession`), folded into snapshots (`foldSession`); `FlushScheduler`; the remote and sync-state interfaces, and directory-backed ones over a `SyncFileSystem` for development and tests. | Node and browsers | `esbuild` (for `/plugins`, `/surfaces` and `/workers`), `ws` (for `/http` and `/tui/launcher`) and `react` (for `/react`) are optional peer dependencies, imported only there; `/app` uses all but `react`. diff --git a/packages/chord/package.json b/packages/chord/package.json index c0076c7e9..2faec4edc 100644 --- a/packages/chord/package.json +++ b/packages/chord/package.json @@ -116,6 +116,11 @@ "types": "./dist/node/tui/index.d.ts", "import": "./dist/node/tui/index.js" }, + "./sync": { + "source": "./src/sync/index.ts", + "types": "./dist/sync/index.d.ts", + "import": "./dist/sync/index.js" + }, "./package.json": "./package.json" }, "sideEffects": false, diff --git a/packages/chord/src/sync/batch.ts b/packages/chord/src/sync/batch.ts new file mode 100644 index 000000000..bc6196ea0 --- /dev/null +++ b/packages/chord/src/sync/batch.ts @@ -0,0 +1,139 @@ +// Bodies on the remote, encoded and decoded as streams, so neither side holds a whole batch or snapshot in memory: +// gzip-compressed JSON lines, one outbox commit (a batch) or one archive part (a snapshot) per line. gzip, because +// `CompressionStream` offers it in Node, Workers, and browsers alike, and a browser viewer reads batches too. + +import type { ArchivePart } from "../store/archive.ts"; +import { decodeJsonl, encodeJsonl } from "../store/jsonl.ts"; +import type { OutboxCommit } from "../store/outbox.ts"; +import { decodeLogLine, encodeLogLine, LogFormatError } from "./lines.ts"; + +/** A body on the remote: its bytes in chunks, read once. */ +export type SyncBody = AsyncIterable; + +/** The bytes a chunk of encoded lines collects before it is compressed. */ +const CHUNK = 64 * 1024; + +/** A batch body: `commits` as gzip-compressed JSON lines, read from `commits` only as the body is read. */ +export function encodeBatch(commits: Iterable | AsyncIterable): SyncBody { + return encodeLines( + (async function* () { + for await (const commit of commits) yield encodeLogLine(commit); + })(), + ); +} + +/** The commits of a batch body, in order, as they arrive. */ +export async function* decodeBatch(body: SyncBody): AsyncGenerator { + for await (const { commit } of decodeBatchLines(body)) yield commit; +} + +/** The commits of a batch body with the size of each line, for callers that bound what they hold. */ +export async function* decodeBatchLines( + body: SyncBody, +): AsyncGenerator<{ readonly commit: OutboxCommit; readonly bytes: number }> { + let previous = 0; + for await (const { text, number } of lines(gunzip(body))) { + const commit = decodeLogLine(text, number); + if (commit.seq <= previous) throw new LogFormatError(`Log line ${number} does not increase the sequence`); + previous = commit.seq; + yield { commit, bytes: text.length + 1 }; + } +} + +/** A snapshot body: a store archive (`store.export()`), one part per JSON line, gzip-compressed. */ +export function encodeSnapshot(parts: AsyncIterable): SyncBody { + return gzip(encodeJsonl(parts)); +} + +/** The parts of a snapshot body as they arrive, unchecked: `importStore()` checks them (a trust boundary). */ +export function decodeSnapshot(body: SyncBody): AsyncIterable { + return decodeJsonl(gunzip(body)); +} + +/** Text lines, already newline-terminated, gzip-compressed in chunks of whole lines. */ +function encodeLines(lines: AsyncIterable): SyncBody { + return gzip( + (async function* () { + const encoder = new TextEncoder(); + let pending = ""; + for await (const line of lines) { + pending += line; + if (pending.length >= CHUNK) { + yield encoder.encode(pending); + pending = ""; + } + } + if (pending.length > 0) yield encoder.encode(pending); + })(), + ); +} + +/** The lines of decompressed `chunks` without their newlines; a body that ends inside a line is an error. */ +async function* lines(chunks: AsyncIterable): AsyncGenerator<{ readonly text: string; number: number }> { + const decoder = new TextDecoder("utf-8", { fatal: true }); + let rest = ""; + let number = 0; + const decode = (chunk?: Uint8Array): string => { + try { + return chunk === undefined ? decoder.decode() : decoder.decode(chunk, { stream: true }); + } catch (error) { + throw new LogFormatError(`Log line ${number + 1} is not UTF-8`, { cause: error }); + } + }; + for await (const chunk of chunks) { + rest += decode(chunk); + let start = 0; + for (let end = rest.indexOf("\n"); end >= 0; end = rest.indexOf("\n", start)) { + yield { text: rest.slice(start, end), number: ++number }; + start = end + 1; + } + rest = rest.slice(start); + } + rest += decode(); + if (rest.length > 0) throw new LogFormatError(`The body ends inside log line ${number + 1}`); +} + +function gzip(chunks: AsyncIterable): SyncBody { + return pipe(chunks, new CompressionStream("gzip")); +} + +function gunzip(chunks: SyncBody): AsyncIterable { + return pipe(chunks, new DecompressionStream("gzip")); +} + +/** + * `chunks` through `transform`, as they come: the source is written while the output is read, so each holds no more + * than the stream's buffers. Stopping early, or a failure on either side, ends the source too (its `return()`, which + * for an export gives the store's commit queue back). + */ +async function* pipe( + chunks: AsyncIterable, + transform: CompressionStream | DecompressionStream, +): AsyncGenerator { + const writer = transform.writable.getWriter(); + const reader = transform.readable.getReader(); + const writing = (async () => { + try { + for await (const chunk of chunks) await writer.write(chunk as Uint8Array); + await writer.close(); + } catch (error) { + await writer.abort(error).catch(() => {}); + throw error; + } + })(); + // Its failure reaches the reader too (the abort errors the readable side); this only keeps it from going unhandled. + writing.catch(() => {}); + let done = false; + try { + while (true) { + const result = await reader.read(); + if (result.done) break; + yield result.value; + } + done = true; + await writing; + } finally { + if (!done) await reader.cancel().catch(() => {}); + reader.releaseLock(); + } +} diff --git a/packages/chord/src/sync/client.ts b/packages/chord/src/sync/client.ts new file mode 100644 index 000000000..b20626bd9 --- /dev/null +++ b/packages/chord/src/sync/client.ts @@ -0,0 +1,536 @@ +// Session sync over a Chord store's outbox: upload the commits after the last sequence a copy shares with a remote +// into a slot the remote creates once, pull a copy forward with `applyBatch`, restore from an archive, fold history +// into snapshots, and decide when to upload. The remote and the host's sync state are interfaces (`remote.ts`). +// +// Bounded memory: bodies are streams (`batch.ts`). An upload reads the outbox a page at a time and stops a batch at +// `BATCH_BYTES`; a pull applies a batch in chunks of at most `APPLY_BYTES`; a snapshot is exported and imported as it +// streams. Nothing holds a whole batch, snapshot or store, so a Durable Object's isolate (128 MB) can run it. + +import type { Context } from "../context/index.ts"; +import { applyBatch } from "../store/apply-batch.ts"; +import type { Backend } from "../store/backend.ts"; +import { importStore } from "../store/import.ts"; +import type { OutboxCommit } from "../store/outbox.ts"; +import { type ChordStore, openStore, type Store } from "../store/store.ts"; +import { decodeBatchLines, decodeSnapshot, encodeBatch, encodeSnapshot, type SyncBody } from "./batch.ts"; +import { + type BatchInfo, + HistoryUnavailableError, + LOG_START, + type PutBatchResult, + remoteHead, + remoteHistory, + type Seq, + type SessionSyncMaintenance, + type SessionSyncStore, +} from "./remote.ts"; + +/** Commits read from the outbox per page at most. */ +const OUTBOX_PAGE = 100; +/** About how much one outbox page holds: the count of the next page follows the size of the commits read. */ +const PAGE_BYTES = 1024 * 1024; +/** + * Where an upload ends a batch: once its lines (before compression) reach this size. One commit larger than this is + * a batch of its own. + */ +export const BATCH_BYTES = 8 * 1024 * 1024; +/** How much of a batch a pull holds before it applies it: commits are applied in chunks of about this size. */ +export const APPLY_BYTES = 4 * 1024 * 1024; + +/** + * 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 + * remote's head; one with commits finds out by uploading. + */ +export type SyncState = { + readonly syncedHead: Seq | undefined; + /** + * 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 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 { + get(sessionId: string, context: Context): Promise; + set(sessionId: string, state: SyncState, context: Context): Promise; +} + +/** A store's head as a sequence: `lastSeq`, 0 for a store without commits. */ +export function storeHead(store: Store): Seq { + return store.lastSeq; +} + +/** 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. */ + | { readonly type: "idle" } + /** + * 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 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 + * applies from the local head. + */ + | { readonly type: "behind"; readonly head: Seq; readonly syncedHead: Seq } + /** + * The remote refused this device's epoch: another device holds the right to write the session now. Like + * `diverged`, the copy's commits after `at` are a different history, so the host forks the copy (S7). + */ + | { readonly type: "stale_epoch"; readonly at: Seq; readonly epoch: number }; + +/** + * 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. + */ +export async function uploadSession(options: { + readonly sessionId: string; + readonly device: string; + readonly source: Store; + readonly store: SessionSyncStore; + readonly state: SyncStateStore; + /** The writer's epoch, from whatever grants it (the remote, Radius); sent with every slot. */ + readonly epoch?: number; + readonly context: Context; +}): Promise { + const { sessionId, device, source, store, state, epoch, context } = options; + 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 = storeHead(source); + if (head < synced) return { type: "behind", head, syncedHead: synced }; + let to: Seq; + let body: SyncBody; + let archive = false; + let close: (() => Promise) | undefined; + if (synced === LOG_START && outbox.floor > 0) { + const snapshot = await exportSnapshot(source, context); + to = snapshot.to; + body = snapshot.body; + close = snapshot.close; + archive = true; + } else { + const range = await outboxRange(source, synced, context); + if (range === undefined) + return uploaded === undefined ? { type: "idle" } : { type: "uploaded", head: uploaded }; + to = range.to; + body = encodeBatch(outboxCommits(source, synced, to, context)); + } + const after = synced; + let result: PutBatchResult; + try { + result = await store.putBatch(sessionId, after, to, device, body, context, { + ...(archive ? { archive } : {}), + ...(epoch === undefined ? {} : { epoch }), + }); + } finally { + // A remote that answered without reading the archive leaves the export holding the commit queue. + await close?.(); + } + // A history that does not end at `after` is another one too: the remote never held this copy's past `after`. + if (result.type === "stale_epoch") return { type: "stale_epoch", at: after, epoch: result.epoch }; + 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); + const trimmable = Math.min(synced, source.lastSeq); + if (trimmable > outbox.floor) await outbox.trim(trimmable, context); + if (result.type === "created") uploaded = to; + } +} + +/** + * Where the next batch after `after` ends: the outbox read a page at a time, holding one page, until the lines reach + * `BATCH_BYTES` (one commit at least). Undefined when the outbox has nothing after `after`. + */ +async function outboxRange(store: Store, after: Seq, context: Context): Promise<{ readonly to: Seq } | undefined> { + let to: Seq | undefined; + let bytes = 0; + const pages = new OutboxPages(store, after, context); + for (let page = await pages.next(); page.length > 0; page = await pages.next()) { + for (const commit of page) { + if (to !== undefined && bytes + encodedSize(commit) > BATCH_BYTES) return { to }; + bytes += encodedSize(commit); + to = commit.seq; + } + if (bytes >= BATCH_BYTES) break; + } + return to === undefined ? undefined : { to }; +} + +/** The outbox commits after `after` through `to`, a page at a time, as the body is read. */ +async function* outboxCommits(store: Store, after: Seq, to: Seq, context: Context): AsyncGenerator { + const pages = new OutboxPages(store, after, context); + let at = after; + while (at < to) { + const page = await pages.next(); + if (page.length === 0) throw new OutboxUnavailableError(`The outbox ends at ${at}, before ${to}`); + for (const commit of page) { + if (commit.seq > to) return; + yield commit; + at = commit.seq; + } + } +} + +/** + * Pages of the outbox after a sequence, each about `PAGE_BYTES`: the outbox reads by count, so the next page's count + * follows the commits read so far (`OUTBOX_PAGE` to start, one commit at least). A run of large commits reads one at a + * time instead of `OUTBOX_PAGE` of them at once. + */ +class OutboxPages { + readonly #store: Store; + readonly #context: Context; + #after: Seq; + #limit = OUTBOX_PAGE; + #done = false; + + constructor(store: Store, after: Seq, context: Context) { + this.#store = store; + this.#after = after; + this.#context = context; + } + + async next(): Promise { + if (this.#done) return []; + const page = await readOutbox(this.#store, this.#after, this.#limit, this.#context); + if (page.length < this.#limit) this.#done = true; + if (page.length === 0) return page; + this.#after = page.at(-1)!.seq; + const average = page.reduce((size, commit) => size + encodedSize(commit), 0) / page.length; + this.#limit = Math.max(1, Math.min(OUTBOX_PAGE, Math.floor(PAGE_BYTES / average))); + return page; + } +} + +/** A commit's line size before compression, as `BATCH_BYTES` counts it. */ +function encodedSize(commit: OutboxCommit): number { + return JSON.stringify(commit.writes).length + 64; +} + +/** A snapshot of a store being exported: where it ends, its body, and `close()` for when the body is not read to the end. */ +type ExportedSnapshot = { readonly to: Seq; readonly body: SyncBody; close(): Promise }; + +/** + * The store's state as a streamed snapshot body. The export holds the store's commit queue from its first part until + * the body is read to the end or `close()` is called, so `to` (its manifest's `lastSeq`) is exactly what it holds. + */ +async function exportSnapshot(store: Store, context: Context): Promise { + const parts = store.export(context); + const first = await parts.next(); + if (first.done || first.value.kind !== "manifest") { + await parts.return?.(undefined); + throw new Error("A store export does not start with its manifest"); + } + const manifest = first.value; + return { + to: manifest.lastSeq, + body: encodeSnapshot( + (async function* () { + yield manifest; + yield* parts; + })(), + ), + close: async () => { + await parts.return?.(undefined); + }, + }; +} + +/** + * Apply a batch body to `store` as it streams, in chunks of about `APPLY_BYTES`, skipping the commits the store has + * (a pull resuming inside a batch). Each chunk is one `applyBatch()` transaction; a pull stopped between chunks + * resumes after the last applied one. + */ +async function applyBody(store: ChordStore, body: SyncBody, context: Context): Promise { + let chunk: OutboxCommit[] = []; + let bytes = 0; + for await (const line of decodeBatchLines(body)) { + if (line.commit.seq <= store.lastSeq) continue; + chunk.push(line.commit); + bytes += line.bytes; + if (bytes >= APPLY_BYTES) { + await applyBatch(store, chunk, context); + chunk = []; + bytes = 0; + } + } + if (chunk.length > 0) await applyBatch(store, chunk, context); +} + +/** 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) { + super( + `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 }; + +/** + * 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 commits + * spliced onto its own. A copy whose history folding removed rejects with `HistoryUnavailableError` and restores into + * a fresh backend. + */ +export async function pullSession(options: { + readonly sessionId: string; + readonly store: SessionSyncStore; + readonly local: LocalCopy; + readonly state: SyncStateStore; + readonly context: 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) }); + } + 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, 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 }); + } + await applyBody(target as ChordStore, bytes, 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; + } +} + +/** + * Fold a session's history into one snapshot: restore it into `scratch` (an empty backend; a file or a Durable Object's + * SQLite storage, not `:memory:`, so the restored store is not held in memory), 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, 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 restored = await pullSession({ + sessionId, + store, + local: { kind: "empty", backend: scratch }, + state: new ScratchState(), + context, + }); + try { + const snapshot = await exportSnapshot(restored.store, context); + try { + await store.putSnapshot(sessionId, snapshot.to, snapshot.body, context); + } finally { + await snapshot.close(); + } + 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; + } +} + +export type FlushSchedulerOptions = { + readonly flush: () => Promise; + /** Arms a timer and returns its disarm. */ + readonly setTimer: (callback: () => void, delayMs: number) => () => void; + readonly now: () => number; + /** A failed flush; the scheduler retries it with backoff. */ + readonly onError: (error: unknown) => void; + /** Delay of an ordinary flush after a commit. Default 5 seconds. */ + readonly intervalMs?: number; + /** Delay of a flush after an urgent commit. Default 200 milliseconds. */ + readonly urgentMs?: number; + /** First retry delay after a failed flush, doubling up to `maxRetryMs`. Default 1 second. */ + readonly retryMs?: number; + /** Default 1 minute. */ + readonly maxRetryMs?: number; +}; + +/** + * Decides when to flush: some time after a commit, sooner after an urgent one, one flush at a time. Commits that arrive + * during a flush keep their urgency and schedule the next one when it ends. A failed flush is retried with backoff + * until one succeeds, so the last commits of a turn do not wait for the next commit. + */ +export class FlushScheduler { + readonly #options: FlushSchedulerOptions; + #due: number | undefined; + #disarm: (() => void) | undefined; + #running: Promise | undefined; + /** When the commits that arrived during a flush want the next one. */ + #pendingDue: number | undefined; + #failures = 0; + #closed = false; + + constructor(options: FlushSchedulerOptions) { + this.#options = options; + } + + /** A commit landed. */ + notify(urgent: boolean): void { + if (this.#closed) return; + const due = this.#options.now() + (urgent ? (this.#options.urgentMs ?? 200) : (this.#options.intervalMs ?? 5000)); + if (this.#running !== undefined) { + this.#pendingDue = Math.min(this.#pendingDue ?? due, due); + return; + } + this.#arm(due); + } + + /** Flush now, after a flush underway, and stop scheduling. A failure rejects. */ + async close(): Promise { + this.#closed = true; + this.#disarm?.(); + this.#disarm = undefined; + this.#due = undefined; + await this.#running; + await this.#options.flush(); + } + + #arm(due: number): void { + if (this.#due !== undefined && this.#due <= due) return; + this.#disarm?.(); + this.#due = due; + this.#disarm = this.#options.setTimer(() => void this.#run(), Math.max(0, due - this.#options.now())); + } + + #run(): Promise { + this.#disarm = undefined; + this.#due = undefined; + this.#running = (async () => { + let retry: number | undefined; + try { + await this.#options.flush(); + this.#failures = 0; + } catch (error) { + this.#options.onError(error); + const delay = (this.#options.retryMs ?? 1000) * 2 ** this.#failures; + this.#failures++; + retry = this.#options.now() + Math.min(delay, this.#options.maxRetryMs ?? 60_000); + } finally { + this.#running = undefined; + } + if (this.#closed) return; + const pending = this.#pendingDue; + this.#pendingDue = undefined; + // A retry waits for its backoff even when commits arrived: they flush with it. + const due = retry ?? pending; + if (due !== undefined) this.#arm(due); + })(); + return this.#running; + } +} diff --git a/packages/chord/src/sync/files.ts b/packages/chord/src/sync/files.ts new file mode 100644 index 000000000..0daa8528c --- /dev/null +++ b/packages/chord/src/sync/files.ts @@ -0,0 +1,52 @@ +// The file operations the directory-backed sync pieces need, as an interface: any environment that has them (Node, +// a container, a remote host) can back a sync remote or the host's sync state. Shaped as Pi Durable's `FileSystem` +// (`@earendil-works/pi-durable/env`), so its environments pass as they are. + +import type { Context } from "../context/index.ts"; + +/** An operation's outcome: its value, or the error it failed with. */ +export type FileResult = { readonly ok: true; readonly value: T } | { readonly ok: false; readonly error: Error }; + +/** A directory entry. */ +export type SyncFileInfo = { readonly name: string; readonly kind: "file" | "directory" | "symlink" }; + +/** A file opened for reads at offsets: what streaming a body back needs. */ +export interface SyncFileReader { + /** Up to `length` bytes at `offset`; fewer only at end of file. */ + read(offset: number, length: number, context: Context): Promise>; + close(context: Context): Promise; +} + +/** File operations for the directory-backed sync pieces. A missing file or directory fails with `code: "not_found"`. */ +export interface SyncFileSystem { + joinPath(parts: string[], context: Context): Promise>; + readTextFile(path: string, context: Context): Promise>; + readBinaryFile(path: string, context: Context): Promise>; + /** Open a file for bounded reads, so a body streams back without being read whole. */ + openBinaryReader( + path: string, + options: { noFollow?: boolean } | undefined, + context: Context, + ): Promise>; + writeFile(path: string, content: string | Uint8Array, context: Context): Promise>; + /** Append to a file, creating it: a body is written chunk by chunk as it streams. */ + appendFile(path: string, content: string | Uint8Array, context: Context): Promise>; + renameFile(sourcePath: string, destinationPath: string, context: Context): Promise>; + listDir(path: string, context: Context): Promise>; + createDir(path: string, options: { recursive?: boolean } | undefined, context: Context): Promise>; + remove( + path: string, + options: { recursive?: boolean; force?: boolean } | undefined, + context: Context, + ): Promise>; +} + +/** Whether `result` failed because the file or directory is missing. */ +export function isNotFound(result: FileResult): boolean { + return !result.ok && (result.error as { readonly code?: unknown }).code === "not_found"; +} + +export function unwrap(result: FileResult): T { + if (!result.ok) throw result.error; + return result.value; +} diff --git a/packages/chord/src/sync/filesystem.ts b/packages/chord/src/sync/filesystem.ts new file mode 100644 index 000000000..2d9e3593d --- /dev/null +++ b/packages/chord/src/sync/filesystem.ts @@ -0,0 +1,233 @@ +import type { Context } from "../context/index.ts"; +import type { SyncBody } from "./batch.ts"; +import { isNotFound, type SyncFileSystem, unwrap } from "./files.ts"; +import { + type BatchInfo, + type JsonObject, + type PutBatchResult, + type RemoteListing, + remoteHead, + type Seq, + type SessionChanges, + type SessionSyncMaintenance, +} from "./remote.ts"; + +const pad = (value: number): string => String(value).padStart(16, "0"); +/** How much of a body is read from its file at a time. */ +const READ_CHUNK = 256 * 1024; + +/** One buffer as a body. */ +async function* once(bytes: Uint8Array): SyncBody { + yield bytes; +} +const SLOT_NAME = /^(\d{16})\.json$/; +const SNAPSHOT_NAME = /^(\d{16})\.batch$/; +const SESSION_ID = /^[A-Za-z0-9_-][A-Za-z0-9._-]*$/; + +/** A session ID names a file or directory: refuse anything that could leave the root. */ +export function checkSessionId(sessionId: string): string { + if (!SESSION_ID.test(sessionId)) throw new Error(`Invalid session ID for a sync directory: ${sessionId}`); + return sessionId; +} + +/** + * A remote store in a directory, laid out as the real remote is: per session, one slot per batch start, holding where + * the batch ends and which device created it, and snapshots named by their end. As the real remote, it creates a slot + * only where the history ends. Creating a slot is check-then-write, + * 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, 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 + * ``` + */ +export class FileSystemSyncRemote implements SessionSyncMaintenance { + readonly #fs: SyncFileSystem; + readonly #root: string; + + constructor(fs: SyncFileSystem, root: string) { + this.#fs = fs; + this.#root = root; + } + + async putBatch( + sessionId: string, + after: Seq, + to: Seq, + device: string, + body: SyncBody, + 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 }; + const head = remoteHead(await this.list(sessionId, context)); + 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, ...(options.archive === true ? { archive: true as const } : {}) }; + await this.#write( + sessionId, + ["slots"], + `${pad(after)}.json`, + once(new TextEncoder().encode(JSON.stringify(slot))), + context, + ); + return { type: "created" }; + } + + async getBatch(sessionId: string, after: Seq, context: Context): Promise { + const slot = await this.#slot(sessionId, after, context); + if (slot === undefined || slot.folded) return undefined; + return this.#stream(context, sessionId, "batches", `${pad(after)}.batch`); + } + + getSnapshot(sessionId: string, to: Seq, context: Context): Promise { + return this.#stream(context, sessionId, "snapshots", `${pad(to)}.batch`); + } + + async putSnapshot(sessionId: string, to: Seq, body: SyncBody, context: Context): Promise { + await this.#write(sessionId, ["snapshots"], `${pad(to)}.batch`, body, context); + } + + async foldBatches(sessionId: string, afters: readonly Seq[], context: Context): Promise { + for (const after of afters) { + const slot = await this.#slot(sessionId, after, context); + if (slot === undefined || slot.folded) continue; + const folded: BatchInfo = { ...slot, folded: true }; + await this.#write( + sessionId, + ["slots"], + `${pad(after)}.json`, + once(new TextEncoder().encode(JSON.stringify(folded))), + context, + ); + const body = await this.#path(context, sessionId, "batches", `${pad(after)}.batch`); + unwrap(await this.#fs.remove(body, { force: true }, context)); + } + } + + async deleteSnapshots(sessionId: string, snapshots: readonly Seq[], context: Context): Promise { + for (const to of snapshots) { + const path = await this.#path(context, sessionId, "snapshots", `${pad(to)}.batch`); + unwrap(await this.#fs.remove(path, { force: true }, context)); + } + } + + async list(sessionId: string, context: Context): Promise { + const batches: BatchInfo[] = []; + for (const name of await this.#names(context, sessionId, "slots")) { + const match = SLOT_NAME.exec(name); + if (match === null) continue; + const slot = await this.#slot(sessionId, Number(match[1]), context); + if (slot !== undefined) batches.push(slot); + } + const snapshots: Seq[] = []; + for (const name of await this.#names(context, sessionId, "snapshots")) { + const match = SNAPSHOT_NAME.exec(name); + if (match !== null) snapshots.push(Number(match[1])); + } + return { batches, snapshots }; + } + + /** Every session, every time: a directory keeps no change order. */ + async sessions(_cursor: string | undefined, context: Context): Promise { + const result = await this.#fs.listDir(this.#root, context); + const names = isNotFound(result) + ? [] + : unwrap(result) + .filter((info) => info.kind === "directory" && SESSION_ID.test(info.name)) + .map((info) => info.name) + .sort(); + const sessions = []; + for (const id of names) { + sessions.push({ + id, + head: remoteHead(await this.list(id, context)), + meta: await this.getMeta(id, context), + }); + } + return { sessions, cursor: "", more: false }; + } + + async putMeta(sessionId: string, meta: JsonObject, context: Context): Promise { + await this.#write(sessionId, [], "meta.json", once(new TextEncoder().encode(JSON.stringify(meta))), context); + } + + async getMeta(sessionId: string, context: Context): Promise { + const bytes = await this.#read(context, sessionId, "meta.json"); + return bytes === undefined ? undefined : (JSON.parse(new TextDecoder().decode(bytes)) as JsonObject); + } + + async #slot(sessionId: string, after: Seq, context: Context): Promise { + const bytes = await this.#read(context, sessionId, "slots", `${pad(after)}.json`); + return bytes === undefined ? undefined : (JSON.parse(new TextDecoder().decode(bytes)) as BatchInfo); + } + + async #names(context: Context, sessionId: string, ...parts: string[]): Promise { + const result = await this.#fs.listDir(await this.#path(context, sessionId, ...parts), context); + if (isNotFound(result)) return []; + return unwrap(result).map((info) => info.name); + } + + async #read(context: Context, sessionId: string, ...parts: string[]): Promise { + const result = await this.#fs.readBinaryFile(await this.#path(context, sessionId, ...parts), context); + if (isNotFound(result)) return undefined; + return unwrap(result); + } + + /** A body streamed back from its file, a chunk at a time; undefined when the file is missing. */ + async #stream(context: Context, sessionId: string, ...parts: string[]): Promise { + const opened = await this.#fs.openBinaryReader( + await this.#path(context, sessionId, ...parts), + undefined, + context, + ); + if (isNotFound(opened)) return undefined; + const reader = unwrap(opened); + return (async function* () { + try { + for (let offset = 0; ; ) { + const chunk = unwrap(await reader.read(offset, READ_CHUNK, context)); + if (chunk.length === 0) return; + offset += chunk.length; + yield chunk; + } + } finally { + await reader.close(context); + } + })(); + } + + /** + * Write a body through a temporary file and a rename, so a reader never sees half a file; the body is appended chunk + * by chunk as it streams. + */ + async #write( + sessionId: string, + directory: readonly string[], + name: string, + body: SyncBody, + context: Context, + ): Promise { + unwrap( + await this.#fs.createDir(await this.#path(context, sessionId, ...directory), { recursive: true }, context), + ); + const path = await this.#path(context, sessionId, ...directory, name); + const temporary = await this.#path(context, sessionId, ...directory, `${name}.tmp`); + unwrap(await this.#fs.writeFile(temporary, new Uint8Array(), context)); + try { + for await (const chunk of body) unwrap(await this.#fs.appendFile(temporary, chunk, context)); + } catch (error) { + await this.#fs.remove(temporary, { force: true }, context); + throw error; + } + unwrap(await this.#fs.renameFile(temporary, path, context)); + } + + async #path(context: Context, sessionId: string, ...parts: string[]): Promise { + return unwrap(await this.#fs.joinPath([this.#root, checkSessionId(sessionId), ...parts], context)); + } +} diff --git a/packages/chord/src/sync/index.ts b/packages/chord/src/sync/index.ts new file mode 100644 index 000000000..64de26764 --- /dev/null +++ b/packages/chord/src/sync/index.ts @@ -0,0 +1,48 @@ +// Session sync over a store's outbox (`@earendil-works/chord/sync`, experimental). Portable: the remote and the host's +// sync state are interfaces, and the directory-backed ones take their file operations (`SyncFileSystem`). + +export { + decodeBatch, + decodeBatchLines, + decodeSnapshot, + encodeBatch, + encodeSnapshot, + type SyncBody, +} from "./batch.ts"; +export { + APPLY_BYTES, + BATCH_BYTES, + FlushScheduler, + type FlushSchedulerOptions, + foldSession, + type LocalCopy, + OutboxUnavailableError, + pullSession, + readOutbox, + type SyncState, + type SyncStateStore, + storeHead, + UnsyncedCommitsError, + type UploadResult, + uploadSession, +} from "./client.ts"; +export type { FileResult, SyncFileInfo, SyncFileReader, SyncFileSystem } from "./files.ts"; +export { checkSessionId, FileSystemSyncRemote } from "./filesystem.ts"; +export { decodeLogLine, encodeLogLine, LOG_VERSION, LogFormatError } from "./lines.ts"; +export { + type BatchInfo, + HistoryUnavailableError, + type JsonObject, + LOG_START, + type PutBatchResult, + type RemoteHistory, + type RemoteListing, + remoteHead, + remoteHistory, + type Seq, + type SessionChanges, + type SessionSummary, + type SessionSyncMaintenance, + type SessionSyncStore, +} from "./remote.ts"; +export { FileSyncStateStore } from "./state.ts"; diff --git a/packages/chord/src/sync/lines.ts b/packages/chord/src/sync/lines.ts new file mode 100644 index 000000000..3860c58dd --- /dev/null +++ b/packages/chord/src/sync/lines.ts @@ -0,0 +1,54 @@ +import type { OutboxCommit } from "../store/outbox.ts"; + +/** + * 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 nextId: number; + readonly writes: OutboxCommit["writes"]; +}; + +export class LogFormatError extends Error { + constructor(message: string, options?: ErrorOptions) { + super(message, options); + this.name = "LogFormatError"; + } +} + +/** 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`; +} + +/** + * One line (without its newline) as an outbox commit; `number` names the line in errors. A line of an unknown version + * is an error: an older pi must not apply writes it does not understand. The writes themselves are checked by + * `applyBatch()`, a trust boundary. + */ +export function decodeLogLine(text: string, number: number): OutboxCommit { + 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 (typeof line.nextId !== "number" || !Number.isSafeInteger(line.nextId)) { + throw new LogFormatError(`Log line ${number} has no valid next id`); + } + if (!Array.isArray(line.writes)) throw new LogFormatError(`Log line ${number} has no writes`); + return { seq: line.seq, nextId: line.nextId, writes: line.writes }; +} diff --git a/packages/chord/src/sync/remote.ts b/packages/chord/src/sync/remote.ts new file mode 100644 index 000000000..a674d234d --- /dev/null +++ b/packages/chord/src/sync/remote.ts @@ -0,0 +1,168 @@ +import type { Context } from "../context/index.ts"; +import type { SyncBody } from "./batch.ts"; + +/** A store's commit sequence (`store.lastSeq`, an outbox commit's `seq`); 0 is before every commit. */ +export type Seq = number; + +/** A JSON object: a session's description on the remote. */ +export type JsonObject = { [key: string]: unknown }; + +/** The sequence before every commit: a store without commits is at it. */ +export const LOG_START: Seq = 0; + +/** + * 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. + */ +export type BatchInfo = { + readonly after: Seq; + readonly to: Seq; + readonly device: string; + /** 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 store archive at its sequence, folded from batches. */ +export type RemoteListing = { + readonly batches: readonly BatchInfo[]; + readonly snapshots: readonly Seq[]; +}; + +export type PutBatchResult = + | { readonly type: "created" } + /** The slot is taken; by this device, after an upload whose response was lost, or by another one. */ + | { readonly type: "exists"; readonly to: Seq; readonly device: string } + /** + * No slot starts at `after`, and the history does not end there: a slot there would branch the chain inside a batch + * or leave a gap. The uploader's history is not the remote's from `after` on. + */ + | { readonly type: "not_head"; readonly head: Seq } + /** + * The upload's `epoch` is not the session's current one: another device was given the right to write the session + * since this one was. Who grants it, and when, is the remote's (RFC 0074's single writer); Chord only reports it. + */ + | { readonly type: "stale_epoch"; readonly epoch: number }; + +/** A session as the remote's listing describes it. */ +export type SessionSummary = { + readonly id: string; + /** Where its history ends; `LOG_START` for a session with a description and no history. */ + readonly head: Seq; + readonly meta: JsonObject | undefined; +}; + +/** Sessions changed after a cursor, and the cursor to continue from. */ +export type SessionChanges = { + readonly sessions: readonly SessionSummary[]; + /** Pass it to the next call; opaque. */ + readonly cursor: string; + /** More changes follow right away: call again with `cursor`. */ + readonly more: boolean; +}; + +/** The client's view of a remote store. Clients never delete or fold. */ +export interface SessionSyncStore { + /** + * Create the slot after `after` unless it exists or the history does not end at `after`. `body` is a stream read once; + * a remote that answers without reading it (`exists`, `not_head`) leaves it unread. + */ + putBatch( + sessionId: string, + after: Seq, + to: Seq, + device: string, + body: SyncBody, + context: Context, + options?: { + readonly archive?: boolean; + /** The writer's epoch: a remote that issues epochs refuses a stale one (`stale_epoch`); others ignore it. */ + readonly epoch?: number; + }, + ): Promise; + /** 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; + /** + * The sessions changed after `cursor` (all of them without one). A remote may also list sessions that did not change, + * so a caller compares what it gets with what it has. + */ + sessions(cursor: string | undefined, context: Context): Promise; + /** + * The host's description of a session (title, location, which device started it), outside its history: the last + * write wins, and it may be ahead of or behind the batches. + */ + putMeta(sessionId: string, meta: JsonObject, context: Context): Promise; + getMeta(sessionId: string, context: Context): Promise; +} + +/** The remote's own side: folding history into snapshots, which clients never do. */ +export interface SessionSyncMaintenance extends SessionSyncStore { + putSnapshot(sessionId: string, to: Seq, body: SyncBody, context: Context): Promise; + /** 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 applies to reach the remote's head. */ +export type RemoteHistory = { + /** The snapshot a restore starts from. */ + readonly snapshot?: Seq; + /** + * 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 applying them ends. */ + readonly head: Seq; +}; + +/** 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}`); + this.name = "HistoryUnavailableError"; + } +} + +/** + * 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 { + const chain = new Map(listing.batches.map((batch) => [batch.after, batch])); + let snapshot: Seq | undefined; + let at: Seq; + const batches: BatchInfo[] = []; + if (from === undefined || from === LOG_START) { + snapshot = listing.snapshots.length === 0 ? undefined : (Math.max(...listing.snapshots) as Seq); + at = snapshot ?? LOG_START; + } else { + at = from; + // A pull that stopped inside a batch finishes it; only one batch can contain the head. + const inside = listing.batches.find((batch) => batch.after < from && from < batch.to); + if (inside !== undefined) { + batches.push(inside); + at = inside.to; + } + } + for (let next = chain.get(at); next !== undefined; next = chain.get(at)) { + batches.push(next); + at = next.to; + } + return { ...(snapshot === undefined ? {} : { snapshot }), batches, head: at }; +} + +/** The furthest head the remote's history reaches. */ +export function remoteHead(listing: RemoteListing): Seq { + const chain = new Map(listing.batches.map((batch) => [batch.after, batch])); + let at: Seq = LOG_START; + for (let next = chain.get(at); next !== undefined; next = chain.get(at)) at = next.to; + return at; +} diff --git a/packages/chord/src/sync/state.ts b/packages/chord/src/sync/state.ts new file mode 100644 index 000000000..49fb8ac8f --- /dev/null +++ b/packages/chord/src/sync/state.ts @@ -0,0 +1,47 @@ +import type { Context } from "../context/index.ts"; +import type { SyncState, SyncStateStore } from "./client.ts"; +import { isNotFound, type SyncFileSystem, unwrap } from "./files.ts"; +import { checkSessionId } from "./filesystem.ts"; +import type { Seq } from "./remote.ts"; + +/** + * Sync state in a directory, one JSON file per session, replaced through a temporary file and a rename. Losing the + * directory loses nothing that matters: a copy without commits of its own takes the remote's head, and one with + * commits finds out by uploading. + */ +export class FileSyncStateStore implements SyncStateStore { + readonly #fs: SyncFileSystem; + readonly #directory: string; + + constructor(fs: SyncFileSystem, directory: string) { + this.#fs = fs; + this.#directory = directory; + } + + async get(sessionId: string, context: Context): Promise { + const result = await this.#fs.readTextFile(await this.#path(sessionId, ".json", context), context); + if (isNotFound(result)) return undefined; + const stored = JSON.parse(unwrap(result)) as { syncedHead?: Seq; pulling?: Seq }; + return { + syncedHead: stored.syncedHead, + ...(stored.pulling === undefined ? {} : { pulling: stored.pulling }), + }; + } + + async set(sessionId: string, state: SyncState, context: Context): Promise { + unwrap(await this.#fs.createDir(this.#directory, { recursive: true }, context)); + const path = await this.#path(sessionId, ".json", context); + const temporary = await this.#path(sessionId, ".json.tmp", context); + unwrap(await this.#fs.writeFile(temporary, JSON.stringify(state), context)); + unwrap(await this.#fs.renameFile(temporary, path, context)); + } + + /** Forget a session, as when its copy is renamed or deleted. */ + async delete(sessionId: string, context: Context): Promise { + unwrap(await this.#fs.remove(await this.#path(sessionId, ".json", context), { force: true }, context)); + } + + async #path(sessionId: string, suffix: string, context: Context): Promise { + return unwrap(await this.#fs.joinPath([this.#directory, `${checkSessionId(sessionId)}${suffix}`], context)); + } +} diff --git a/packages/chord/test/bench/sync-memory.ts b/packages/chord/test/bench/sync-memory.ts new file mode 100644 index 000000000..fd4daed52 --- /dev/null +++ b/packages/chord/test/bench/sync-memory.ts @@ -0,0 +1,169 @@ +// Memory benchmark of the sync layer inside Durable Objects (RFC 0074's budget: 96 to 128 MB per isolate): workerd +// through Miniflare, the objects of `sync-worker.ts` on their own SQLite storage, and the isolate's V8 heap sampled +// through the inspector while each step runs. Steps per session size: fill the origin's store (commits with its +// outbox), upload (outbox to remote, bodies streamed into the remote's SQLite), pull (restore a replica through the +// origin's remote over RPC), fold (restore into scratch, export a snapshot back). Each step starts after a full GC; its +// baseline is the used heap then, and its peak the highest `usedSize` sampled until it answers. +// +// workerd runs with `--expose-gc` and `--max-old-space-size=128` (Cloudflare's per-isolate limit), so a step that +// needed more than 128 MB would fail, not just report a big number. +// +// Run by hand from packages/chord (not in CI; about three minutes): +// node --conditions=source --experimental-strip-types test/bench/sync-memory.ts + +import { build } from "esbuild"; +import { convertV4MiniflareOptions, Miniflare } from "miniflare"; +import WebSocket from "ws"; + +/** Session histories: total MB of row text, and the size of each commit's row. */ +const RUNS = [ + { sizeMb: 8, rowBytes: 64 * 1024 }, + { sizeMb: 32, rowBytes: 64 * 1024 }, + { sizeMb: 128, rowBytes: 64 * 1024 }, + { sizeMb: 256, rowBytes: 64 * 1024 }, + // Few large commits: each one close to a whole batch, the case that holds the most at once. + { sizeMb: 64, rowBytes: 4 * 1024 * 1024 }, +]; +const INSPECTOR_PORT = 9341; +const SAMPLE_MS = 10; +const MB = 1024 * 1024; + +process.env.MINIFLARE_WORKERD_V8_FLAGS = "--expose-gc --max-old-space-size=128"; + +const bundled = await build({ + entryPoints: [new URL("./sync-worker.ts", import.meta.url).pathname], + bundle: true, + format: "esm", + platform: "neutral", + write: false, + external: ["cloudflare:*"], + conditions: ["source", "workerd", "worker"], + mainFields: ["module", "main"], + target: "es2023", +}); + +const mf = new Miniflare( + convertV4MiniflareOptions({ + inspectorPort: INSPECTOR_PORT, + workers: [ + { + name: "sync-bench", + compatibilityDate: "2026-10-01", + compatibilityFlags: ["nodejs_compat"], + modules: [{ type: "ESModule", path: "worker.mjs", contents: bundled.outputFiles[0]!.text }], + durableObjects: { + ORIGIN: { className: "Origin", useSQLite: true }, + REPLICA: { className: "Replica", useSQLite: true }, + FOLDER: { className: "Folder", useSQLite: true }, + }, + }, + ], + }), +); + +try { + await mf.ready; + await mf.dispatchFetch("http://bench/unknown"); + const [target] = (await (await fetch(`http://127.0.0.1:${INSPECTOR_PORT}/json/list`)).json()) as { + webSocketDebuggerUrl: string; + }[]; + const inspector = await connect(target!.webSocketDebuggerUrl); + + const rows: string[] = []; + for (const { sizeMb, rowBytes } of RUNS) { + const run = `run${sizeMb}x${rowBytes}`; + const count = (sizeMb * MB) / rowBytes; + const label = `${sizeMb} MB in ${mb(rowBytes)} MB rows`; + for (const step of ["fill", "upload", "pull", "fold"] as const) { + const query = step === "fill" ? `&count=${count}&size=${rowBytes}` : ""; + const measured = await measure(inspector, () => mf.dispatchFetch(`http://bench/${step}?run=${run}${query}`)); + const body = (await measured.response.json()) as { ms: number; result: unknown }; + if (!measured.response.ok) throw new Error(`${step} ${label} failed: ${JSON.stringify(body)}`); + rows.push( + `| ${label} | ${step} | ${mb(measured.baseline)} | ${mb(measured.peak)} | ${mb(measured.peak - measured.baseline)} | ${body.ms} | ${summary(step, body.result)} |`, + ); + console.log(rows.at(-1)); + } + } + inspector.close(); + console.log( + [ + "", + "| history | step | heap before (MB) | peak heap (MB) | peak above before (MB) | ms | result |", + "|---|---|---|---|---|---|---|", + ...rows, + ].join("\n"), + ); +} finally { + await mf.dispose(); +} + +/** What a step's answer must offer (Miniflare's own `Response` type, not the DOM one). */ +type StepResponse = { readonly ok: boolean; json(): Promise; text(): Promise }; + +type Inspector = { call(method: string): Promise<{ result?: { usedSize?: number } }>; close(): void }; + +async function connect(url: string): Promise { + const ws = new WebSocket(url); + await new Promise((resolve, reject) => { + ws.once("open", resolve); + ws.once("error", reject); + }); + let id = 0; + const pending = new Map void>(); + ws.on("message", (raw) => { + const message = JSON.parse(String(raw)) as { id?: number }; + if (message.id !== undefined) pending.get(message.id)?.(message as never); + pending.delete(message.id ?? -1); + }); + const inspector: Inspector = { + call: (method) => + new Promise((resolve) => { + const n = ++id; + pending.set(n, resolve as (value: never) => void); + ws.send(JSON.stringify({ id: n, method })); + }), + close: () => ws.close(), + }; + await inspector.call("Runtime.enable"); + return inspector; +} + +/** + * Runs `step` from a settled heap: a full GC in the isolate first, its used heap then as the baseline, then samples + * every `SAMPLE_MS` until the step answers; the peak is the highest sample. + */ +async function measure( + inspector: Inspector, + step: () => Promise, +): Promise<{ response: StepResponse; baseline: number; peak: number }> { + const used = async () => (await inspector.call("Runtime.getHeapUsage")).result?.usedSize ?? 0; + await (await mf.dispatchFetch("http://bench/gc")).text(); + const baseline = await used(); + let peak = baseline; + let running = true; + const sampling = (async () => { + while (running) { + peak = Math.max(peak, await used()); + await new Promise((resolve) => setTimeout(resolve, SAMPLE_MS)); + } + })(); + const response = await step(); + running = false; + await sampling; + peak = Math.max(peak, await used()); + return { response, baseline, peak }; +} + +function mb(bytes: number): string { + return (bytes / MB).toFixed(1); +} + +function summary(step: string, result: unknown): string { + const value = result as Record; + if (step === "upload") { + const stored = value.stored as number; + return `${value.slots} slots, ${mb(stored)} MB stored`; + } + return JSON.stringify(result); +} diff --git a/packages/chord/test/bench/sync-worker.ts b/packages/chord/test/bench/sync-worker.ts new file mode 100644 index 000000000..5392cb70f --- /dev/null +++ b/packages/chord/test/bench/sync-worker.ts @@ -0,0 +1,450 @@ +// The Durable Objects of the sync memory benchmark (`sync-memory.ts`), one Chord store each on the object's own SQLite +// storage (`chord/sqlite/cloudflare`), as a deployment has them: +// - `Origin`: the copy that commits and uploads, and the remote, its slots and bodies in its SQLite (`SqlRemote`); +// - `Replica`: a copy restored by a pull, through the origin's remote over RPC; +// - `Folder`: folds the origin's history into a snapshot, its scratch store on its own SQLite. +// Bodies cross objects as byte streams and are stored in parts, so what an object holds is what the sync layer holds. +// The benchmark drives the objects over `fetch` and reads the isolate's heap through the inspector. + +import { DurableObject } from "cloudflare:workers"; +import { BACKGROUND_CONTEXT as context } from "../../src/context/index.ts"; +import { int, str } from "../../src/contracts/index.ts"; +import { DurableObjectSqliteDatabase } from "../../src/sqlite/cloudflare.ts"; +import { sqliteBackend } from "../../src/store/backend.ts"; +import { LiveQueries } from "../../src/store/live-query.ts"; +import type { StagedWrite } from "../../src/store/step.ts"; +import { type ChordStore, openStore } from "../../src/store/store.ts"; +import { defineTable } from "../../src/store/table.ts"; +import { TableChanges } from "../../src/store/table-changes.ts"; +import { TableHandle } from "../../src/store/table-handle.ts"; +import { prepareTables } from "../../src/store/table-migration.ts"; +import { + type BatchInfo, + foldSession, + type JsonObject, + type PutBatchResult, + pullSession, + type RemoteListing, + remoteHead, + type Seq, + type SessionChanges, + type SessionSyncMaintenance, + type SyncBody, + type SyncState, + type SyncStateStore, + uploadSession, +} from "../../src/sync/index.ts"; + +// The parts of the Workers runtime this bench uses, typed here: full Workers types would clash with the repository's +// Node types, and `DurableObject` (packages/codemode/src/cloudflare-modules.d.ts) leaves `ctx` untyped. +type SqlValue = string | number | null | ArrayBuffer | Uint8Array; +type SqlCursor = { toArray(): T[]; one(): T }; +type SqlStorage = { exec>(query: string, ...bindings: SqlValue[]): SqlCursor }; +type ObjectStorage = { + readonly sql: SqlStorage; + transaction(closure: () => Promise): Promise; + transactionSync(closure: () => T): T; +}; +type ObjectState = { readonly storage: ObjectStorage }; +type ObjectId = unknown; +type Namespace = { idFromName(name: string): ObjectId; get(id: ObjectId): T }; + +/** A Durable Object of this bench: the runtime's base class with its `ctx` typed. */ +const BenchObject = DurableObject as unknown as abstract new ( + ctx: ObjectState, + env: Env, +) => { readonly ctx: ObjectState; readonly env: Env }; + +type Env = { + readonly ORIGIN: Namespace; + readonly REPLICA: Namespace; + readonly FOLDER: Namespace; +}; + +const Rows = defineTable("rows", { version: 1, columns: { id: str(), n: int(), text: str() }, key: ["id"] }); +const TABLE = "bench.sync:rows"; +const OPEN = { admit: () => true, adopt: () => {} }; +const owner = { label: "bench", scope: {}, queries: new LiveQueries(), assertLive: () => {}, fail: () => {} }; +/** Where a stored body is split into rows: below the object's 2 MB value limit. */ +const PART = 512 * 1024; + +/** The isolate's own GC (the benchmark starts workerd with `--expose-gc`), so a step starts from a settled heap. */ +const collect = (): void => (globalThis as { gc?: () => void }).gc?.(); + +/** A body as a byte stream, for RPC. */ +function toStream(body: SyncBody): ReadableStream { + const iterator = body[Symbol.asyncIterator](); + return new ReadableStream({ + async pull(controller) { + const next = await iterator.next(); + if (next.done) controller.close(); + else controller.enqueue(next.value); + }, + async cancel() { + await iterator.return?.(); + }, + }); +} + +/** A byte stream from RPC as a body. */ +async function* fromStream(stream: ReadableStream): SyncBody { + const reader = stream.getReader(); + try { + while (true) { + const next = await reader.read(); + if (next.done) return; + yield next.value; + } + } finally { + reader.releaseLock(); + } +} + +type SlotRow = { after: number; upto: number; device: string; archive: number; folded: number }; + +/** + * A remote in an object's SQLite storage, in tables of its own beside the store's: slots, and bodies written and read + * back a part at a time as they stream. + */ +class SqlRemote implements SessionSyncMaintenance { + readonly #sql: SqlStorage; + + constructor(sql: SqlStorage) { + this.#sql = sql; + sql.exec(`CREATE TABLE IF NOT EXISTS r_slots (after INTEGER PRIMARY KEY, upto INTEGER, device TEXT, archive INTEGER, folded INTEGER); + CREATE TABLE IF NOT EXISTS r_parts (kind TEXT, at INTEGER, part INTEGER, data BLOB, PRIMARY KEY (kind, at, part)); + CREATE TABLE IF NOT EXISTS r_snapshots (upto INTEGER PRIMARY KEY);`); + } + + async putBatch( + _session: string, + after: Seq, + to: Seq, + device: string, + body: SyncBody, + _context: unknown, + options: { readonly archive?: boolean } = {}, + ): Promise { + const existing = this.#sql + .exec<{ upto: number; device: string }>("SELECT upto, device FROM r_slots WHERE after = ?", after) + .toArray()[0]; + if (existing) return { type: "exists", to: existing.upto, device: existing.device }; + const head = remoteHead(await this.list()); + if (after !== head) return { type: "not_head", head }; + await this.#write("batch", after, body); + this.#sql.exec("INSERT INTO r_slots VALUES (?, ?, ?, ?, 0)", after, to, device, options.archive ? 1 : 0); + return { type: "created" }; + } + + async getBatch(_session: string, after: Seq): Promise { + const slot = this.#sql.exec<{ folded: number }>("SELECT folded FROM r_slots WHERE after = ?", after).toArray()[0]; + return slot && !slot.folded ? this.#read("batch", after) : undefined; + } + + async getSnapshot(_session: string, to: Seq): Promise { + return this.#sql.exec("SELECT 1 FROM r_snapshots WHERE upto = ?", to).toArray().length > 0 + ? this.#read("snapshot", to) + : undefined; + } + + async putSnapshot(_session: string, to: Seq, body: SyncBody): Promise { + await this.#write("snapshot", to, body); + this.#sql.exec("INSERT OR REPLACE INTO r_snapshots VALUES (?)", to); + } + + async foldBatches(_session: string, afters: readonly Seq[]): Promise { + for (const after of afters) { + this.#sql.exec("UPDATE r_slots SET folded = 1 WHERE after = ?", after); + this.#sql.exec("DELETE FROM r_parts WHERE kind = 'batch' AND at = ?", after); + } + } + + async deleteSnapshots(_session: string, snapshots: readonly Seq[]): Promise { + for (const to of snapshots) { + this.#sql.exec("DELETE FROM r_snapshots WHERE upto = ?", to); + this.#sql.exec("DELETE FROM r_parts WHERE kind = 'snapshot' AND at = ?", to); + } + } + + async list(): Promise { + const batches = this.#sql + .exec("SELECT after, upto, device, archive, folded FROM r_slots ORDER BY after") + .toArray() + .map( + (slot): BatchInfo => ({ + after: slot.after, + to: slot.upto, + device: slot.device, + ...(slot.archive ? { archive: true as const } : {}), + ...(slot.folded ? { folded: true as const } : {}), + }), + ); + const snapshots = this.#sql + .exec<{ upto: number }>("SELECT upto FROM r_snapshots") + .toArray() + .map((row) => row.upto); + return { batches, snapshots }; + } + + async sessions(): Promise { + return { sessions: [], cursor: "", more: false }; + } + async putMeta(): Promise {} + async getMeta(): Promise { + return undefined; + } + + /** Bytes the remote stores, all bodies together. */ + stored(): number { + return this.#sql.exec<{ n: number }>("SELECT coalesce(sum(length(data)), 0) AS n FROM r_parts").one().n; + } + + /** The body's chunks gathered into parts of `PART` bytes, one row each, as it streams. */ + async #write(kind: string, at: number, body: SyncBody): Promise { + this.#sql.exec("DELETE FROM r_parts WHERE kind = ? AND at = ?", kind, at); + let part = 0; + let pending: Uint8Array[] = []; + let size = 0; + const flush = () => { + const data = new Uint8Array(size); + let offset = 0; + for (const chunk of pending) { + data.set(chunk, offset); + offset += chunk.length; + } + this.#sql.exec("INSERT INTO r_parts VALUES (?, ?, ?, ?)", kind, at, part++, data); + pending = []; + size = 0; + }; + for await (const chunk of body) { + pending.push(chunk); + size += chunk.length; + if (size >= PART) flush(); + } + if (size > 0) flush(); + } + + /** The body's parts, read back one row at a time. */ + #read(kind: string, at: number): SyncBody { + const sql = this.#sql; + return (async function* () { + for (let part = 0; ; part++) { + const row = sql + .exec<{ data: ArrayBuffer }>( + "SELECT data FROM r_parts WHERE kind = ? AND at = ? AND part = ?", + kind, + at, + part, + ) + .toArray()[0]; + if (!row) return; + yield new Uint8Array(row.data); + } + })(); + } +} + +/** The origin's remote seen from another object: its calls over RPC, bodies as byte streams. */ +class RpcRemote implements SessionSyncMaintenance { + readonly #origin: Origin; + + constructor(origin: Origin) { + this.#origin = origin; + } + + async putBatch(): Promise { + throw new Error("The benchmark uploads only at the origin"); + } + async getBatch(_session: string, after: Seq): Promise { + const stream = await this.#origin.getBatch(after); + return stream === null ? undefined : fromStream(stream); + } + async getSnapshot(_session: string, to: Seq): Promise { + const stream = await this.#origin.getSnapshot(to); + return stream === null ? undefined : fromStream(stream); + } + async putSnapshot(_session: string, to: Seq, body: SyncBody): Promise { + await this.#origin.putSnapshot(to, toStream(body)); + } + async foldBatches(_session: string, afters: readonly Seq[]): Promise { + await this.#origin.foldBatches([...afters]); + } + async deleteSnapshots(_session: string, snapshots: readonly Seq[]): Promise { + await this.#origin.deleteSnapshots([...snapshots]); + } + async list(): Promise { + return (await this.#origin.list()) as RemoteListing; + } + async sessions(): Promise { + return { sessions: [], cursor: "", more: false }; + } + async putMeta(): Promise {} + async getMeta(): Promise { + return undefined; + } +} + +class MemoryState implements SyncStateStore { + #state: SyncState | undefined; + async get(): Promise { + return this.#state; + } + async set(_session: string, state: SyncState): Promise { + this.#state = state; + } +} + +/** + * This object's SQLite storage as a store backend. Durable Object SQLite refuses `PRAGMA encoding` and + * `pragma_encoding` (`SQLITE_AUTH`), which `openStore` reads for its UTF-8 check; its databases are always UTF-8, so + * the benchmark answers that one query itself. (A store bug, to fix in `openStore` or the Cloudflare adapter.) + */ +function backendOf(storage: ObjectStorage) { + const rewrite = (query: string) => + query + .replace(/^PRAGMA encoding$/, "SELECT 'UTF-8' AS encoding") + .replace(/pragma_encoding e\b/, "(SELECT 'UTF-8' AS encoding) e"); + const sql = storage.sql; + const wrapped = { + sql: { exec: (query: string, ...bindings: SqlValue[]) => sql.exec(rewrite(query), ...bindings) }, + transaction: (closure: () => Promise) => storage.transaction(closure), + transactionSync: (closure: () => T) => storage.transactionSync(closure), + }; + return sqliteBackend(new DurableObjectSqliteDatabase(wrapped as never)); +} + +/** Deterministic text that compresses about as transcript text does (about 2:1), not to nothing. */ +function text(size: number, seed: number): string { + let state = seed * 2654435761 + 1; + const words = ["the", "agent", "reads", "a", "file", "and", "writes", "tests", "for", "it", "then", "runs"]; + let out = ""; + while (out.length < size) { + state ^= state << 13; + state ^= state >>> 17; + state ^= state << 5; + out += `${words[(state >>> 0) % words.length]} ${(state >>> 8) & 0xffff} `; + } + return out.slice(0, size); +} + +export class Origin extends BenchObject { + #remote = new SqlRemote(this.ctx.storage.sql); + + /** `count` commits of one row of `size` characters each. */ + async fill(count: number, size: number): Promise<{ lastSeq: number }> { + collect(); + const store = await openStore(backendOf(this.ctx.storage), context, { outbox: true }); + try { + const chord = store as ChordStore; + await prepareTables([{ definition: Rows, table: TABLE }], chord, OPEN, "bench.sync", context); + const handle = new TableHandle(Rows, TABLE, chord, owner); + for (let index = 0; index < count; index++) { + const staged: StagedWrite[] = []; + new TableChanges(handle, staged).put({ id: `r${index}`, n: index, text: text(size, index) }); + await chord.commit(staged, OPEN, context); + } + return { lastSeq: store.lastSeq }; + } finally { + await store.close(context); + } + } + + async upload(): Promise { + collect(); + const store = await openStore(backendOf(this.ctx.storage), context, { outbox: true }); + try { + const result = await uploadSession({ + sessionId: "s", + device: "origin", + source: store, + store: this.#remote, + state: new MemoryState(), + context, + }); + return { result, slots: (await this.#remote.list()).batches.length, stored: this.#remote.stored() }; + } finally { + await store.close(context); + } + } + + async getBatch(after: number): Promise | null> { + const body = await this.#remote.getBatch("s", after); + return body === undefined ? null : toStream(body); + } + async getSnapshot(to: number): Promise | null> { + const body = await this.#remote.getSnapshot("s", to); + return body === undefined ? null : toStream(body); + } + async putSnapshot(to: number, stream: ReadableStream): Promise { + await this.#remote.putSnapshot("s", to, fromStream(stream)); + } + async foldBatches(afters: number[]): Promise { + await this.#remote.foldBatches("s", afters); + } + async deleteSnapshots(snapshots: number[]): Promise { + await this.#remote.deleteSnapshots("s", snapshots); + } + async list(): Promise { + return this.#remote.list(); + } +} + +export class Replica extends BenchObject { + /** Restores the session from the origin's remote into this object's empty storage. */ + async pull(origin: string): Promise { + collect(); + const remote = new RpcRemote(this.env.ORIGIN.get(this.env.ORIGIN.idFromName(origin))); + const pulled = await pullSession({ + sessionId: "s", + store: remote, + local: { kind: "empty", backend: backendOf(this.ctx.storage) }, + state: new MemoryState(), + context, + }); + await pulled.store.close(context); + return { head: pulled.head }; + } +} + +export class Folder extends BenchObject { + /** Folds the origin's history into a snapshot, restoring it into this object's empty storage as scratch. */ + async fold(origin: string): Promise { + collect(); + const remote = new RpcRemote(this.env.ORIGIN.get(this.env.ORIGIN.idFromName(origin))); + return { + folded: await foldSession({ sessionId: "s", store: remote, scratch: backendOf(this.ctx.storage), context }), + }; + } +} + +export default { + async fetch(request: Request, env: Env): Promise { + const url = new URL(request.url); + const run = url.searchParams.get("run") ?? "bench"; + const count = Number(url.searchParams.get("count") ?? 0); + const size = Number(url.searchParams.get("size") ?? 0); + const started = Date.now(); + let result: unknown; + switch (url.pathname) { + case "/gc": + // Every object of the benchmark shares this isolate: one full GC here settles the heap for the next step. + collect(); + return new Response("ok"); + case "/fill": + result = await env.ORIGIN.get(env.ORIGIN.idFromName(run)).fill(count, size); + break; + case "/upload": + result = await env.ORIGIN.get(env.ORIGIN.idFromName(run)).upload(); + break; + case "/pull": + result = await env.REPLICA.get(env.REPLICA.idFromName(run)).pull(run); + break; + case "/fold": + result = await env.FOLDER.get(env.FOLDER.idFromName(run)).fold(run); + break; + default: + return new Response("unknown step", { status: 404 }); + } + return Response.json({ ms: Date.now() - started, result }); + }, +}; diff --git a/packages/chord/test/sync/lines.test.ts b/packages/chord/test/sync/lines.test.ts new file mode 100644 index 000000000..b1690758f --- /dev/null +++ b/packages/chord/test/sync/lines.test.ts @@ -0,0 +1,146 @@ +// The sync wire format, as streams: outbox commits as versioned, gzip-compressed JSON lines, snapshots as store +// archives, and decoding that holds one line at a time. +import { describe, expect, it } from "vitest"; +import { BACKGROUND_CONTEXT as context } from "../../src/context/index.ts"; +import { createDocument } from "../../src/store/engine.ts"; +import type { ChordStore } from "../../src/store/store.ts"; +import { + decodeBatch, + decodeLogLine, + decodeSnapshot, + encodeBatch, + encodeLogLine, + encodeSnapshot, + LOG_START, + LogFormatError, + readOutbox, + type SyncBody, +} from "../../src/sync/index.ts"; +import { memoryStore } from "../support.ts"; + +const OPEN = { admit: () => true, adopt: () => {} }; + +/** A store with its outbox on and two commits: two documents created one at a time. */ +async function seeded(): Promise { + const store = await memoryStore({ outbox: true }); + for (const key of ["a", "b"]) { + const base = { version: 1, fingerprint: "f1", value: { key } }; + await store.commit( + [ + createDocument( + { namespace: "chord", owner: "app.sync", kind: "note", key }, + base, + JSON.stringify(base.value), + ), + ], + OPEN, + context, + ); + } + return store; +} + +async function collect(items: AsyncIterable): Promise { + const out: T[] = []; + for await (const item of items) out.push(item); + return out; +} + +/** A body's bytes, joined. */ +async function bytesOf(body: SyncBody): Promise { + const chunks = await collect(body); + const out = new Uint8Array(chunks.reduce((size, chunk) => size + chunk.length, 0)); + let offset = 0; + for (const chunk of chunks) { + out.set(chunk, offset); + offset += chunk.length; + } + return out; +} + +/** Plain bytes gzip-compressed, as a body: for bodies the encoders would never write. */ +function gzipped(text: string): SyncBody { + return (async function* () { + const stream = new Blob([text]).stream().pipeThrough(new CompressionStream("gzip")); + yield new Uint8Array(await new Response(stream).arrayBuffer()); + })(); +} + +/** `bytes` as a body of chunks of `size`: decoding must not depend on where chunks split. */ +async function* chunked(bytes: Uint8Array, size: number): SyncBody { + for (let offset = 0; offset < bytes.length; offset += size) yield bytes.subarray(offset, offset + size); +} + +describe("batch bodies", () => { + it("round-trip outbox commits, whatever the chunks", async () => { + const store = await seeded(); + try { + const commits = await readOutbox(store, LOG_START, 100, context); + expect(commits).toHaveLength(2); + const bytes = await bytesOf(encodeBatch(commits)); + expect(bytes[0]).toBe(0x1f); + for (const size of [1, 7, bytes.length]) { + expect(await collect(decodeBatch(chunked(bytes, size)))).toEqual(commits); + } + } finally { + await store.close(context); + } + }); + + it("refuse a body that ends inside a line", async () => { + const line = encodeLogLine({ seq: 1, nextId: 2, writes: [] }); + await expect(collect(decodeBatch(gzipped(line + line.slice(0, 5))))).rejects.toThrow("ends inside log line 2"); + }); + + it("refuse sequences that do not increase", async () => { + const line = encodeLogLine({ seq: 3, nextId: 4, writes: [] }); + await expect(collect(decodeBatch(gzipped(line + line)))).rejects.toThrow("does not increase"); + }); + + it("decode a commit only when it is read", async () => { + const lines = [1, 2, 3].map((seq) => encodeLogLine({ seq, nextId: seq + 1, writes: [] })).join(""); + const commits = decodeBatch(gzipped(`${lines}not json\n`)); + const iterator = commits[Symbol.asyncIterator](); + expect((await iterator.next()).value).toMatchObject({ seq: 1 }); + await iterator.return?.(undefined); + }); +}); + +describe("log lines", () => { + it("refuse unknown versions and a missing next id", () => { + expect(() => decodeLogLine('{"v":1,"seq":1,"nextId":2,"writes":[]}', 1)).toThrow(LogFormatError); + expect(() => decodeLogLine('{"v":2,"seq":1,"writes":[]}', 1)).toThrow(LogFormatError); + expect(decodeLogLine('{"v":2,"seq":1,"nextId":2,"writes":[]}', 1)).toEqual({ seq: 1, nextId: 2, writes: [] }); + }); +}); + +describe("snapshot bodies", () => { + it("round-trip a store archive", async () => { + const store = await seeded(); + try { + const parts = await collect(store.export(context)); + expect(await collect(decodeSnapshot(encodeSnapshot(store.export(context))))).toEqual(parts); + expect(parts[0]).toMatchObject({ kind: "manifest", lastSeq: store.lastSeq }); + } finally { + await store.close(context); + } + }); + + it("give the store's commit queue back when the body is not read to the end", async () => { + const store = await seeded(); + try { + const body = encodeSnapshot(store.export(context))[Symbol.asyncIterator](); + await body.next(); + await body.return?.(); + const base = { version: 1, fingerprint: "f1", value: { key: "c" } }; + const committed = store.commit( + [createDocument({ namespace: "chord", owner: "app.sync", kind: "note", key: "c" }, base, "{}")], + OPEN, + context, + ); + await expect(committed).resolves.toBeGreaterThan(0); + } finally { + await store.close(context); + } + }); +}); diff --git a/packages/chord/test/sync/node-files.ts b/packages/chord/test/sync/node-files.ts new file mode 100644 index 000000000..bd1faa4ec --- /dev/null +++ b/packages/chord/test/sync/node-files.ts @@ -0,0 +1,59 @@ +// `SyncFileSystem` over Node's file system, for the sync tests (a host passes its own environment, such as Pi Durable's +// `NodeExecutionEnv`). + +import { appendFile, mkdir, open, readdir, readFile, rename, rm, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import type { FileResult, SyncFileInfo, SyncFileReader, SyncFileSystem } from "../../src/sync/index.ts"; + +/** An error with a `code`: `not_found` for a missing file or directory, as `SyncFileSystem` reports it. */ +class FileError extends Error { + readonly code: string; + constructor(code: string, cause: unknown) { + super(cause instanceof Error ? cause.message : String(cause), { cause }); + this.code = code; + } +} + +async function attempt(operation: () => Promise): Promise> { + try { + return { ok: true, value: await operation() }; + } catch (error) { + const code = (error as NodeJS.ErrnoException).code === "ENOENT" ? "not_found" : "unknown"; + return { ok: false, error: new FileError(code, error) }; + } +} + +export const nodeFiles: SyncFileSystem = { + joinPath: async (parts) => ({ ok: true, value: join(...parts) }), + readTextFile: (path) => attempt(() => readFile(path, "utf8")), + readBinaryFile: (path) => attempt(async () => new Uint8Array(await readFile(path))), + openBinaryReader: (path) => + attempt(async (): Promise => { + const file = await open(path, "r"); + return { + read: (offset, length) => + attempt(async () => { + const buffer = new Uint8Array(length); + const { bytesRead } = await file.read(buffer, 0, length, offset); + return buffer.subarray(0, bytesRead); + }), + close: async () => { + await file.close(); + }, + }; + }), + writeFile: (path, content) => attempt(() => writeFile(path, content)), + appendFile: (path, content) => attempt(() => appendFile(path, content)), + renameFile: (source, destination) => attempt(() => rename(source, destination)), + listDir: (path) => + attempt(async () => + (await readdir(path, { withFileTypes: true })).map( + (entry): SyncFileInfo => ({ + name: entry.name, + kind: entry.isDirectory() ? "directory" : entry.isSymbolicLink() ? "symlink" : "file", + }), + ), + ), + createDir: (path, options) => attempt(async () => void (await mkdir(path, options))), + remove: (path, options) => attempt(() => rm(path, options)), +}; diff --git a/packages/chord/test/sync/sync.test.ts b/packages/chord/test/sync/sync.test.ts new file mode 100644 index 000000000..c4058a35c --- /dev/null +++ b/packages/chord/test/sync/sync.test.ts @@ -0,0 +1,814 @@ +// The sync protocol over stores with an outbox: slots created once, uploads from the outbox, pulls through +// `applyBatch()`, archives for stores whose outbox does not reach back, folds into snapshots. The stores hold a plain +// table: sync knows nothing of what a store keeps. +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { BACKGROUND_CONTEXT, type Context } from "../../src/context/index.ts"; +import { int, opt, str } from "../../src/contracts/index.ts"; +import { openNodeSqliteDatabase } from "../../src/sqlite/node.ts"; +import { type Backend, sqliteBackend } from "../../src/store/backend.ts"; +import type { StagedWrite } from "../../src/store/step.ts"; +import { type ChordStore, openStore, type Store } from "../../src/store/store.ts"; +import { defineTable } from "../../src/store/table.ts"; +import { TableChanges } from "../../src/store/table-changes.ts"; +import { TableHandle } from "../../src/store/table-handle.ts"; +import { prepareTables } from "../../src/store/table-migration.ts"; +import { + type BatchInfo, + decodeBatch, + encodeBatch, + FileSyncStateStore, + FileSystemSyncRemote, + FlushScheduler, + foldSession, + HistoryUnavailableError, + LOG_START, + OutboxUnavailableError, + pullSession, + readOutbox, + remoteHead, + remoteHistory, + type Seq, + type SessionSyncMaintenance, + type SyncState, + type SyncStateStore, + UnsyncedCommitsError, + uploadSession, +} from "../../src/sync/index.ts"; +import { tableOwner } from "../support.ts"; +import { nodeFiles } from "./node-files.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 }); +}); + +async function remote(): Promise { + const directory = await mkdtemp(join(tmpdir(), "chord-sync-")); + directories.push(directory); + return new FileSystemSyncRemote(nodeFiles, directory); +} + +/** A backend over the SQLite file at `path`, `:memory:` by default. */ +async function backend(path = ":memory:"): Promise { + return sqliteBackend(await openNodeSqliteDatabase(path)); +} + +/** A store with its outbox on. */ +async function sessionStore(path = ":memory:"): Promise { + return openStore(await backend(path), context, { outbox: true }); +} + +/** The table every copy writes: a counter per id. */ +const Counts = defineTable("counts", { + version: 1, + columns: { id: str(), n: int(), pad: opt(str()) }, + key: ["id"], +}); +const TABLE = "app.sync:counts"; +const OPEN = { admit: () => true, adopt: () => {} }; + +/** One batch of the test's history: rows to put. */ +type Rows = readonly { readonly id: string; readonly n: number; readonly pad?: string }[]; + +/** Commits each batch of rows to `store`, preparing the table first (its creation is a commit of its own). */ +async function commitAll(store: Store, batches: readonly Rows[]): Promise { + const chord = store as ChordStore; + await prepareTables([{ definition: Counts, table: TABLE }], chord, OPEN, "app.sync", context); + const handle = new TableHandle(Counts, TABLE, chord, tableOwner()); + for (const rows of batches) { + const staged: StagedWrite[] = []; + const changes = new TableChanges(handle, staged); + for (const row of rows) changes.put(row); + await chord.commit(staged, OPEN, context); + } +} + +/** What a copy holds, as one comparable value: every part of its export. */ +async function reads(store: Store): Promise { + const parts: unknown[] = []; + for await (const part of store.export(context)) parts.push(part); + return parts; +} + +class MemoryState implements SyncStateStore { + readonly states = new Map(); + async get(sessionId: string, _context: Context): Promise { + return this.states.get(sessionId); + } + async set(sessionId: string, state: SyncState, _context: Context): Promise { + this.states.set(sessionId, state); + } +} + +/** `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; + }, + }); +} + +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 Rows[], 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; +} + +/** Ten batches of rows: new ids and updates of earlier ones. */ +function seedBatches(): Rows[] { + return [ + [{ id: "a", n: 1 }], + [ + { id: "b", n: 1 }, + { id: "c", n: 1 }, + ], + [{ id: "a", n: 2 }], + [{ id: "d", n: 1 }], + [{ id: "b", n: 2 }], + [{ id: "e", n: 1 }], + [{ id: "a", n: 3 }], + [{ id: "f", n: 1 }], + [ + { id: "g", n: 1 }, + { id: "c", n: 2 }, + ], + [{ id: "h", n: 1 }], + ]; +} + +/** One row a single device adds. */ +const entry = (id: number): Rows => [{ id: `x${id}`, n: id }]; + +const slot = (after: number, to: number, device: string): BatchInfo => ({ + after: after, + to: to, + device, +}); + +describe("batches", () => { + 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 decoded = []; + for await (const commit of decodeBatch(encodeBatch(commits))) decoded.push(commit); + expect(decoded).toEqual(commits); + }); +}); + +describe("remoteHistory", () => { + const listing = { + batches: [{ ...slot(0, 4, "a"), folded: true as const }, slot(4, 10, "a"), slot(10, 12, "b")], + snapshots: [4], + }; + + it("restores from the newest snapshot along the chain of slots", () => { + expect(remoteHistory(listing)).toEqual({ snapshot: 4, batches: [slot(4, 10, "a"), slot(10, 12, "b")], head: 12 }); + expect(remoteHead(listing)).toBe(12); + expect(remoteHistory({ batches: [], snapshots: [] })).toEqual({ batches: [], head: 0 }); + }); + + it("continues a copy from its head, also from inside a batch", () => { + expect(remoteHistory(listing, 4)).toEqual({ + batches: [slot(4, 10, "a"), slot(10, 12, "b")], + head: 12, + }); + expect(remoteHistory(listing, 7)).toEqual({ + batches: [slot(4, 10, "a"), slot(10, 12, "b")], + head: 12, + }); + expect(remoteHistory(listing, 12)).toEqual({ batches: [], head: 12 }); + }); +}); + +describe("syncing a session between devices", () => { + 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 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 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 () => { + const store = await remote(); + const laptop = device("laptop", store); + const phone = device("phone", store); + const batches = seedBatches(); + await laptop.commit(batches.slice(0, 5)); + await laptop.upload(); + + 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"); + + 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, 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.commit([writes]); + await laptop.upload(); + await viewer.pull(); + } + expect(await reads(await viewer.store())).toEqual(await reads(await laptop.store())); + expect(await viewer.upload()).toEqual({ type: "idle" }); + }); + + for (const [first, second] of [ + ["phone", "laptop"], + ["laptop", "phone"], + ] as const) { + it(`lets the ${first} keep the session when both continued it and the ${first} uploaded first`, async () => { + const store = await remote(); + const devices = { laptop: device("laptop", store), phone: device("phone", store) }; + const batches = seedBatches(); + await devices.laptop.commit(batches.slice(0, 5)); + await devices.laptop.upload(); + await devices.phone.pull(); + 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"); + 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 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(await fork.store("fork-1"))).toEqual(await reads(forked)); + await loser.pull(); + expect(await reads(await loser.store())).toEqual(await reads(await winner.store())); + }); + } + + it("goes on after an upload whose response was lost", async () => { + const store = await remote(); + const laptop = device("laptop", store); + const replica = device("replica", store); + const batches = seedBatches(); + await laptop.commit(batches.slice(0, 5)); + await laptop.upload(); + 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, five, "laptop"), + slot(five, seven, "laptop"), + slot(seven, ten, "laptop"), + ]); + await replica.pull(); + expect(await reads(await replica.store())).toEqual(await reads(await laptop.store())); + }); + + it("passes the writer's epoch to the remote and reports a refused one as stale_epoch", async () => { + const store = await remote(); + const laptop = device("laptop", store); + await laptop.commit(seedBatches().slice(0, 2)); + const shared = (await laptop.store()).lastSeq; + let current = 1; + const epochs: (number | undefined)[] = []; + // A remote that issues epochs (Radius): refuses an upload whose epoch is not the current one. + const issuing = overriding(store, { + putBatch: async (...args: Parameters) => { + const epoch = args[6]?.epoch; + epochs.push(epoch); + if (epoch !== current) return { type: "stale_epoch", epoch: current }; + return store.putBatch(...args); + }, + }); + const source = await laptop.store(); + const upload = (epoch: number) => + uploadSession({ + sessionId: SESSION, + device: "laptop", + source, + store: issuing, + state: laptop.state, + epoch, + context, + }); + expect((await upload(1)).type).toBe("uploaded"); + // The remote gave the session to another writer: this device's epoch is stale. + current = 2; + await laptop.commit([entry(100)]); + expect(await upload(1)).toEqual({ type: "stale_epoch", at: shared, epoch: 2 }); + expect(epochs).toEqual([1, 1]); + }); + + 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 laptop.commit(batches.slice(0, 7)); + await laptop.upload(); + 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: survived.lastSeq, syncedHead: uploaded }); + + 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()).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 laptop.commit(batches.slice(0, 5)); + await laptop.upload(); + 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 desktop.commit(batches.slice(0, 2)); + await desktop.state.set(SESSION, { syncedHead: 1 }, context); + const commits = await readOutbox(await desktop.store(), 1, 100, context); + expect( + await store.putBatch(SESSION, 1, commits.at(-1)!.seq, "desktop", await encodeBatch(commits), context), + ).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 laptop.commit(seedBatches().slice(0, 3)); + await laptop.upload(); + await laptop.commit(seedBatches().slice(3)); + await laptop.upload(); + + // 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 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 () => { + const store = await remote(); + const laptop = device("laptop", store); + const stale = device("stale", store); + const replica = device("replica", store); + for (const [index, writes] of seedBatches().entries()) { + await laptop.commit([writes]); + await laptop.upload(); + if (index === 3) await stale.pull(); + } + 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(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 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 laptop.commit([entry(100), entry(101)]); + await laptop.upload(); + await laptop.commit([entry(102)]); + await laptop.upload(); + await replica.pull(); + 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(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.commit([writes]); + await laptop.upload(); + } + const vanishing: SessionSyncMaintenance = overriding(store, { + getBatch: async () => undefined, + }); + 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(), "chord-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)); + }); +}); + +describe("bounded memory", () => { + /** Rows of `size` characters each, one per commit: about `count * size` bytes of history. */ + function bigBatches(count: number, size: number): Rows[] { + return Array.from({ length: count }, (_, index) => [{ id: `big${index}`, n: index, pad: "x".repeat(size) }]); + } + + it("ends a batch at BATCH_BYTES, so a large history uploads in several slots", async () => { + const store = await remote(); + const laptop = device("laptop", store); + // 24 commits of ~1 MB: about three batches of BATCH_BYTES (8 MB). + await laptop.commit(bigBatches(24, 1024 * 1024)); + expect(await laptop.upload()).toEqual({ type: "uploaded", head: (await laptop.store()).lastSeq }); + const slots = (await store.list(SESSION, context)).batches; + expect(slots.length).toBeGreaterThanOrEqual(3); + for (const [index, slot] of slots.entries()) if (index > 0) expect(slot.after).toBe(slots[index - 1]!.to); + const replica = device("replica", store); + await replica.pull(); + expect(await reads(await replica.store())).toEqual(await reads(await laptop.store())); + }); + + it("applies a large batch in chunks of APPLY_BYTES, and a pull stopped between them resumes", async () => { + const store = await remote(); + const laptop = device("laptop", store); + // 6 commits of ~1 MB in one batch: about two apply chunks of APPLY_BYTES (4 MB). + await laptop.commit(bigBatches(6, 1024 * 1024)); + await laptop.upload(); + expect((await store.list(SESSION, context)).batches).toHaveLength(1); + const viewer = device("viewer", store); + const local = tracked(await sessionStore()); + const chord = local as ChordStore; + let applied = 0; + const apply = chord.replay.bind(chord); + chord.replay = ((...args: Parameters) => { + if (++applied === 2) return Promise.reject(new Error("power cut")); + return apply(...args); + }) as ChordStore["replay"]; + const pull = () => + pullSession({ + sessionId: SESSION, + store, + local: { kind: "store", store: local }, + state: viewer.state, + context, + }); + await expect(pull()).rejects.toThrow("power cut"); + const midway = local.lastSeq; + expect(midway).toBeGreaterThan(0); + expect(midway).toBeLessThan((await laptop.store()).lastSeq); + await pull(); + // The first chunk was applied, the second refused, then the resumed pull applied the rest in more chunks. + expect(applied).toBeGreaterThanOrEqual(3); + expect(await reads(local)).toEqual(await reads(await laptop.store())); + }); +}); + +describe("host pieces", () => { + it("lists sessions and keeps the last metadata written", async () => { + const store = await remote(); + expect(await store.sessions(undefined, context)).toEqual({ sessions: [], cursor: "", more: false }); + const laptop = device("laptop", store); + 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: (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(); + }); + + it("refuses session IDs that leave the root", async () => { + const store = await remote(); + for (const id of ["..", "../x", "a/b", "", ".hidden"]) { + await expect(store.list(id, context)).rejects.toThrow("Invalid session ID"); + } + }); + + it("keeps sync state in files", async () => { + const directory = await mkdtemp(join(tmpdir(), "chord-sync-state-")); + directories.push(directory); + const state = new FileSyncStateStore(nodeFiles, join(directory, "sync")); + expect(await state.get(SESSION, context)).toBeUndefined(); + await state.set(SESSION, { syncedHead: 3, pulling: 7 }, context); + expect(await state.get(SESSION, context)).toEqual({ syncedHead: 3, pulling: 7 }); + await state.set(SESSION, { syncedHead: undefined }, context); + expect(await state.get(SESSION, context)).toEqual({ syncedHead: undefined }); + await state.delete(SESSION, context); + expect(await state.get(SESSION, context)).toBeUndefined(); + }); +}); + +describe("FlushScheduler", () => { + function clock() { + let now = 0; + const timers: { at: number; callback: () => void; armed: boolean }[] = []; + return { + now: () => now, + setTimer: (callback: () => void, delayMs: number) => { + const timer = { at: now + delayMs, callback, armed: true }; + timers.push(timer); + return () => { + timer.armed = false; + }; + }, + /** Fire due timers at their own times, letting each callback's promises settle before the next. */ + async advance(ms: number) { + const end = now + ms; + while (true) { + const next = timers.filter((item) => item.armed && item.at <= end).sort((a, b) => a.at - b.at)[0]; + if (next === undefined) break; + now = next.at; + next.armed = false; + next.callback(); + await new Promise((resolve) => setTimeout(resolve, 0)); + } + now = end; + }, + }; + } + + it("flushes some time after a commit, sooner after an urgent one, once per window", async () => { + const time = clock(); + const flushes: number[] = []; + const scheduler = new FlushScheduler({ + ...time, + flush: async () => void flushes.push(time.now()), + onError: (error) => { + throw error; + }, + }); + scheduler.notify(false); + scheduler.notify(false); + await time.advance(4000); + expect(flushes).toEqual([]); + scheduler.notify(true); + await time.advance(200); + expect(flushes).toEqual([4200]); + await time.advance(5000); + expect(flushes).toEqual([4200]); + await scheduler.close(); + expect(flushes).toEqual([4200, 9200]); + }); + + it("keeps the urgency of commits that arrived during a flush", async () => { + const time = clock(); + let release: () => void = () => {}; + const flushes: number[] = []; + const scheduler = new FlushScheduler({ + ...time, + flush: () => { + flushes.push(time.now()); + return new Promise((resolve) => { + release = resolve; + }); + }, + onError: (error) => { + throw error; + }, + }); + scheduler.notify(false); + await time.advance(5000); + scheduler.notify(false); + scheduler.notify(true); + release(); + await new Promise((resolve) => setTimeout(resolve, 0)); + await time.advance(200); + expect(flushes).toEqual([5000, 5200]); + }); + + it("retries a failed flush with backoff until it succeeds, without another commit", async () => { + const time = clock(); + const attempts: number[] = []; + const errors: unknown[] = []; + const scheduler = new FlushScheduler({ + ...time, + flush: async () => { + attempts.push(time.now()); + if (attempts.length < 3) throw new Error("offline"); + }, + onError: (error) => errors.push(error), + }); + scheduler.notify(true); + await time.advance(5000); + expect(attempts).toEqual([200, 1200, 3200]); + expect(errors).toHaveLength(2); + await time.advance(60_000); + expect(attempts).toHaveLength(3); + }); +}); diff --git a/packages/coding-agent/examples/tsconfig.json b/packages/coding-agent/examples/tsconfig.json index 3fedb9266..d9775419c 100644 --- a/packages/coding-agent/examples/tsconfig.json +++ b/packages/coding-agent/examples/tsconfig.json @@ -70,6 +70,9 @@ "@earendil-works/chord/tui/launcher": [ "../../../packages/chord/src/node/tui/index.ts" ], + "@earendil-works/chord/sync": [ + "../../../packages/chord/src/sync/index.ts" + ], "@test/contracts/*": [ "../../chord/test/node/fixtures/plugins/contracts/*.ts" ], diff --git a/tsconfig.json b/tsconfig.json index 2e2155052..3f4ecf822 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -26,6 +26,7 @@ "@earendil-works/chord/react": ["./packages/chord/src/react/index.ts"], "@earendil-works/chord/tui": ["./packages/chord/src/tui/index.ts"], "@earendil-works/chord/tui/launcher": ["./packages/chord/src/node/tui/index.ts"], + "@earendil-works/chord/sync": ["./packages/chord/src/sync/index.ts"], "@acme/app/*": ["./packages/chord/test/examples/49-plugin-api/app/*.ts"], "@notes/app/*": ["./packages/chord/test/examples/53-app/app/*.ts"], "@board/app/*": ["./packages/chord/test/examples/56-web/app/*.ts"],