From 7297af3b699e35cb5cad5d9605d8cdcd6b349d32 Mon Sep 17 00:00:00 2001 From: Cristina Poncela Cubeiro <140309543+cristinaponcela@users.noreply.github.com> Date: Wed, 4 Nov 2026 14:09:20 +0100 Subject: [PATCH] chore: move to Chord --- apps/p2/tsconfig.json | 2 + packages/chord/CHANGELOG.md | 1 + packages/chord/README.md | 2 + packages/chord/package.json | 10 + packages/{durable => chord}/src/sync/batch.ts | 3 +- .../{durable => chord}/src/sync/client.ts | 39 ++-- packages/{durable => chord}/src/sync/index.ts | 8 +- packages/{durable => chord}/src/sync/lines.ts | 2 +- packages/chord/src/sync/node/index.ts | 5 + packages/chord/src/sync/node/remote.ts | 180 ++++++++++++++++ packages/chord/src/sync/node/state.ts | 49 +++++ .../{durable => chord}/src/sync/remote.ts | 12 +- .../test/sync/lines.test.ts} | 46 ++-- .../test/sync/sync.test.ts} | 198 ++++++++--------- packages/coding-agent/examples/tsconfig.json | 6 + packages/durable/CHANGELOG.md | 2 +- packages/durable/README.md | 27 +-- packages/durable/package.json | 6 - packages/durable/src/index.ts | 1 + packages/durable/src/sync-urgency.ts | 12 ++ packages/durable/src/sync/filesystem.ts | 199 ------------------ packages/durable/src/sync/state.ts | 52 ----- packages/durable/test/sync-e2e.test.ts | 21 +- packages/durable/test/sync-support.ts | 75 ------- packages/durable/test/sync-urgency.test.ts | 20 ++ tsconfig.json | 2 + 26 files changed, 459 insertions(+), 521 deletions(-) rename packages/{durable => chord}/src/sync/batch.ts (95%) rename packages/{durable => chord}/src/sync/client.ts (94%) rename packages/{durable => chord}/src/sync/index.ts (73%) rename packages/{durable => chord}/src/sync/lines.ts (97%) create mode 100644 packages/chord/src/sync/node/index.ts create mode 100644 packages/chord/src/sync/node/remote.ts create mode 100644 packages/chord/src/sync/node/state.ts rename packages/{durable => chord}/src/sync/remote.ts (94%) rename packages/{durable/test/sync.test.ts => chord/test/sync/lines.test.ts} (61%) rename packages/{durable/test/sync-remote.test.ts => chord/test/sync/sync.test.ts} (84%) create mode 100644 packages/durable/src/sync-urgency.ts delete mode 100644 packages/durable/src/sync/filesystem.ts delete mode 100644 packages/durable/src/sync/state.ts delete mode 100644 packages/durable/test/sync-support.ts create mode 100644 packages/durable/test/sync-urgency.test.ts diff --git a/apps/p2/tsconfig.json b/apps/p2/tsconfig.json index 0dc1bd562..102ce4971 100644 --- a/apps/p2/tsconfig.json +++ b/apps/p2/tsconfig.json @@ -29,6 +29,8 @@ "@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/chord/sync/node": ["../../packages/chord/src/sync/node/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 02257c937..4276181e9 100644 --- a/packages/chord/CHANGELOG.md +++ b/packages/chord/CHANGELOG.md @@ -9,6 +9,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. `@earendil-works/chord/sync/node` has `FileSystemSyncRemote`, a remote in a directory, and `FileSyncStateStore`, sync state in a directory, for development and tests. - Chord v2's Node components and client surfaces are exported: `@earendil-works/chord/http`, `/workers`, `/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 c5aa13d1e..bcd53570c 100644 --- a/packages/chord/README.md +++ b/packages/chord/README.md @@ -35,6 +35,8 @@ 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. | Node and browsers | +| `@earendil-works/chord/sync/node` | `FileSystemSyncRemote` and `FileSyncStateStore`: a remote and sync state in a directory, for development and tests. | Node | `esbuild` (for `/plugins` and `/surfaces`), `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 9b335c0db..a85a5890a 100644 --- a/packages/chord/package.json +++ b/packages/chord/package.json @@ -106,6 +106,16 @@ "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" + }, + "./sync/node": { + "source": "./src/sync/node/index.ts", + "types": "./dist/sync/node/index.d.ts", + "import": "./dist/sync/node/index.js" + }, "./package.json": "./package.json" }, "sideEffects": false, diff --git a/packages/durable/src/sync/batch.ts b/packages/chord/src/sync/batch.ts similarity index 95% rename from packages/durable/src/sync/batch.ts rename to packages/chord/src/sync/batch.ts index 8d26272fb..a486bfd66 100644 --- a/packages/durable/src/sync/batch.ts +++ b/packages/chord/src/sync/batch.ts @@ -1,4 +1,5 @@ -import type { ArchivePart, OutboxCommit } from "@earendil-works/chord"; +import type { ArchivePart } from "../store/archive.ts"; +import type { OutboxCommit } from "../store/outbox.ts"; import { decodeLogLines, encodeLogLine, LogFormatError } from "./lines.ts"; /** diff --git a/packages/durable/src/sync/client.ts b/packages/chord/src/sync/client.ts similarity index 94% rename from packages/durable/src/sync/client.ts rename to packages/chord/src/sync/client.ts index 5e333a2cd..18f506146 100644 --- a/packages/durable/src/sync/client.ts +++ b/packages/chord/src/sync/client.ts @@ -1,15 +1,14 @@ -import { - applyBatch, - type Backend, - type ChordStore, - importStore, - type OutboxCommit, - openStore, - type Store, -} from "@earendil-works/chord"; -import type { Context } from "@earendil-works/chord/context"; -import { seqFromNumber } from "../ids.ts"; -import type { CommitPublication, Seq } from "../types.ts"; +// 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`); a +// directory-backed pair for Node is in `./node`. + +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 { decodeBatch, decodeSnapshot, encodeBatch, encodeSnapshot } from "./batch.ts"; import { type BatchInfo, @@ -17,6 +16,7 @@ import { LOG_START, remoteHead, remoteHistory, + type Seq, type SessionSyncMaintenance, type SessionSyncStore, } from "./remote.ts"; @@ -51,7 +51,7 @@ export interface SyncStateStore { /** A store's head as a sequence: `lastSeq`, 0 for a store without commits. */ export function storeHead(store: Store): Seq { - return seqFromNumber(store.lastSeq); + return store.lastSeq; } /** The store keeps no outbox, or its outbox no longer holds the commits a sync needs. */ @@ -134,7 +134,7 @@ export async function uploadSession(options: { const commits = await readOutbox(source, synced, BATCH_COMMITS, context); if (commits.length === 0) return uploaded === undefined ? { type: "idle" } : { type: "uploaded", head: uploaded }; - to = seqFromNumber(commits.at(-1)!.seq); + to = commits.at(-1)!.seq; body = await encodeBatch(commits); } const after = synced; @@ -160,7 +160,7 @@ async function exportSnapshot( const body = await encodeSnapshot( (async function* () { for await (const part of parts) { - if (part.kind === "manifest") to = seqFromNumber(part.lastSeq); + if (part.kind === "manifest") to = part.lastSeq; yield part; } })(), @@ -310,15 +310,6 @@ class ScratchState implements SyncStateStore { } } -/** Whether a commit should reach the remote soon: what the user typed, a finished answer, a settled submission. */ -export function isUrgentCommit(publication: CommitPublication): boolean { - return publication.changes.some( - (change) => - (change.type === "entry" && (change.value.kind === "pi.user" || change.value.kind === "pi.assistant")) || - (change.type === "submission" && (change.value.status === "done" || change.value.status === "unanswered")), - ); -} - export type FlushSchedulerOptions = { readonly flush: () => Promise; /** Arms a timer and returns its disarm. */ diff --git a/packages/durable/src/sync/index.ts b/packages/chord/src/sync/index.ts similarity index 73% rename from packages/durable/src/sync/index.ts rename to packages/chord/src/sync/index.ts index 4db8ca543..3abf984aa 100644 --- a/packages/durable/src/sync/index.ts +++ b/packages/chord/src/sync/index.ts @@ -1,9 +1,11 @@ +// Session sync over a store's outbox (`@earendil-works/chord/sync`, experimental). Portable: the remote and the host's +// sync state are interfaces; `@earendil-works/chord/sync/node` has directory-backed ones for development and tests. + export { decodeBatch, decodeSnapshot, encodeBatch, encodeSnapshot } from "./batch.ts"; export { FlushScheduler, type FlushSchedulerOptions, foldSession, - isUrgentCommit, type LocalCopy, OutboxUnavailableError, pullSession, @@ -15,20 +17,20 @@ export { type UploadResult, uploadSession, } from "./client.ts"; -export { FileSystemSyncRemote } from "./filesystem.ts"; export { decodeLogLines, 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/durable/src/sync/lines.ts b/packages/chord/src/sync/lines.ts similarity index 97% rename from packages/durable/src/sync/lines.ts rename to packages/chord/src/sync/lines.ts index 93a5abb5a..a267a91ea 100644 --- a/packages/durable/src/sync/lines.ts +++ b/packages/chord/src/sync/lines.ts @@ -1,4 +1,4 @@ -import type { OutboxCommit } from "@earendil-works/chord"; +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 diff --git a/packages/chord/src/sync/node/index.ts b/packages/chord/src/sync/node/index.ts new file mode 100644 index 000000000..85fef15ef --- /dev/null +++ b/packages/chord/src/sync/node/index.ts @@ -0,0 +1,5 @@ +// Directory-backed sync pieces for Node (`@earendil-works/chord/sync/node`): a remote and the host's sync state, for +// development and tests. + +export { checkSessionId, FileSystemSyncRemote } from "./remote.ts"; +export { FileSyncStateStore } from "./state.ts"; diff --git a/packages/chord/src/sync/node/remote.ts b/packages/chord/src/sync/node/remote.ts new file mode 100644 index 000000000..e212434cb --- /dev/null +++ b/packages/chord/src/sync/node/remote.ts @@ -0,0 +1,180 @@ +// A sync remote in a local directory, for development and tests (Node). + +import { mkdir, readdir, readFile, rename, rm, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import type { Context } from "../../context/index.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"); +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; +} + +/** Whether `error` is a missing file or directory. */ +export function isNotFound(error: unknown): boolean { + return (error as NodeJS.ErrnoException | undefined)?.code === "ENOENT"; +} + +/** Write through a temporary file and a rename, so a reader never sees half a file. */ +export async function writeAtomically(directory: string, name: string, body: Uint8Array | string): Promise { + await mkdir(directory, { recursive: true }); + const temporary = join(directory, `${name}.tmp`); + await writeFile(temporary, body); + await rename(temporary, join(directory, name)); +} + +/** + * 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 #root: string; + + constructor(root: string) { + this.#root = root; + } + + async putBatch( + sessionId: string, + after: Seq, + to: Seq, + device: string, + body: Uint8Array, + context: Context, + options: { readonly archive?: boolean } = {}, + ): Promise { + const existing = await this.#slot(sessionId, after); + 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 writeAtomically(this.#path(sessionId, "batches"), `${pad(after)}.batch`, body); + // 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 writeAtomically(this.#path(sessionId, "slots"), `${pad(after)}.json`, JSON.stringify(slot)); + return { type: "created" }; + } + + async getBatch(sessionId: string, after: Seq, _context: Context): Promise { + const slot = await this.#slot(sessionId, after); + if (slot === undefined || slot.folded) return undefined; + return this.#read(this.#path(sessionId, "batches", `${pad(after)}.batch`)); + } + + getSnapshot(sessionId: string, to: Seq, _context: Context): Promise { + return this.#read(this.#path(sessionId, "snapshots", `${pad(to)}.batch`)); + } + + async putSnapshot(sessionId: string, to: Seq, body: Uint8Array, _context: Context): Promise { + await writeAtomically(this.#path(sessionId, "snapshots"), `${pad(to)}.batch`, body); + } + + async foldBatches(sessionId: string, afters: readonly Seq[], _context: Context): Promise { + for (const after of afters) { + const slot = await this.#slot(sessionId, after); + if (slot === undefined || slot.folded) continue; + const folded: BatchInfo = { ...slot, folded: true }; + await writeAtomically(this.#path(sessionId, "slots"), `${pad(after)}.json`, JSON.stringify(folded)); + await rm(this.#path(sessionId, "batches", `${pad(after)}.batch`), { force: true }); + } + } + + async deleteSnapshots(sessionId: string, snapshots: readonly Seq[], _context: Context): Promise { + for (const to of snapshots) await rm(this.#path(sessionId, "snapshots", `${pad(to)}.batch`), { force: true }); + } + + async list(sessionId: string, _context: Context): Promise { + const batches: BatchInfo[] = []; + for (const name of await this.#names(this.#path(sessionId, "slots"))) { + const match = SLOT_NAME.exec(name); + if (match === null) continue; + const slot = await this.#slot(sessionId, Number(match[1])); + if (slot !== undefined) batches.push(slot); + } + const snapshots: Seq[] = []; + for (const name of await this.#names(this.#path(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 { + let entries: import("node:fs").Dirent[]; + try { + entries = await readdir(this.#root, { withFileTypes: true }); + } catch (error) { + if (!isNotFound(error)) throw error; + entries = []; + } + const names = entries + .filter((entry) => entry.isDirectory() && SESSION_ID.test(entry.name)) + .map((entry) => entry.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 writeAtomically(this.#path(sessionId), "meta.json", JSON.stringify(meta)); + } + + async getMeta(sessionId: string, _context: Context): Promise { + const bytes = await this.#read(this.#path(sessionId, "meta.json")); + return bytes === undefined ? undefined : (JSON.parse(new TextDecoder().decode(bytes)) as JsonObject); + } + + async #slot(sessionId: string, after: Seq): Promise { + const bytes = await this.#read(this.#path(sessionId, "slots", `${pad(after)}.json`)); + return bytes === undefined ? undefined : (JSON.parse(new TextDecoder().decode(bytes)) as BatchInfo); + } + + async #names(directory: string): Promise { + try { + return await readdir(directory); + } catch (error) { + if (isNotFound(error)) return []; + throw error; + } + } + + async #read(path: string): Promise { + try { + return new Uint8Array(await readFile(path)); + } catch (error) { + if (isNotFound(error)) return undefined; + throw error; + } + } + + #path(sessionId: string, ...parts: string[]): string { + return join(this.#root, checkSessionId(sessionId), ...parts); + } +} diff --git a/packages/chord/src/sync/node/state.ts b/packages/chord/src/sync/node/state.ts new file mode 100644 index 000000000..d91038522 --- /dev/null +++ b/packages/chord/src/sync/node/state.ts @@ -0,0 +1,49 @@ +// Sync state in a local directory (Node). + +import { readFile, rm } from "node:fs/promises"; +import { join } from "node:path"; +import type { Context } from "../../context/index.ts"; +import type { SyncState, SyncStateStore } from "../client.ts"; +import type { Seq } from "../remote.ts"; +import { checkSessionId, isNotFound, writeAtomically } 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 #directory: string; + + constructor(directory: string) { + this.#directory = directory; + } + + async get(sessionId: string, _context: Context): Promise { + let text: string; + try { + text = await readFile(this.#path(sessionId), "utf8"); + } catch (error) { + if (isNotFound(error)) return undefined; + throw error; + } + const stored = JSON.parse(text) as { syncedHead?: Seq; pulling?: Seq }; + return { + syncedHead: stored.syncedHead, + ...(stored.pulling === undefined ? {} : { pulling: stored.pulling }), + }; + } + + async set(sessionId: string, state: SyncState, _context: Context): Promise { + await writeAtomically(this.#directory, `${checkSessionId(sessionId)}.json`, JSON.stringify(state)); + } + + /** Forget a session, as when its copy is renamed or deleted. */ + async delete(sessionId: string, _context: Context): Promise { + await rm(this.#path(sessionId), { force: true }); + } + + #path(sessionId: string): string { + return join(this.#directory, `${checkSessionId(sessionId)}.json`); + } +} diff --git a/packages/durable/src/sync/remote.ts b/packages/chord/src/sync/remote.ts similarity index 94% rename from packages/durable/src/sync/remote.ts rename to packages/chord/src/sync/remote.ts index d8e415e0e..b32ccdddd 100644 --- a/packages/durable/src/sync/remote.ts +++ b/packages/chord/src/sync/remote.ts @@ -1,9 +1,13 @@ -import type { Context } from "@earendil-works/chord/context"; -import { seqFromNumber } from "../ids.ts"; -import type { JsonObject, Seq } from "../types.ts"; +import type { Context } from "../context/index.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 = seqFromNumber(0); +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 diff --git a/packages/durable/test/sync.test.ts b/packages/chord/test/sync/lines.test.ts similarity index 61% rename from packages/durable/test/sync.test.ts rename to packages/chord/test/sync/lines.test.ts index b4accfb3a..8e15d07b2 100644 --- a/packages/durable/test/sync.test.ts +++ b/packages/chord/test/sync/lines.test.ts @@ -1,6 +1,8 @@ -import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +// The sync wire format: outbox commits as versioned JSON lines, and snapshots as store archives. import { describe, expect, it } from "vitest"; -import { idFromNumber } from "../src/ids.ts"; +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 { decodeLogLines, decodeSnapshot, @@ -9,23 +11,28 @@ import { LOG_START, LogFormatError, readOutbox, -} from "../src/sync/index.ts"; -import { type EntryId, ROOT_CONVERSATION_ID } from "../src/types.ts"; -import { commitAll, sessionStore } from "./sync-support.ts"; +} from "../../src/sync/index.ts"; +import { memoryStore } from "../support.ts"; -const context = BACKGROUND_CONTEXT; +const OPEN = { admit: () => true, adopt: () => {} }; -async function seeded() { - const store = await sessionStore(); - await commitAll(store, [ - [{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], - [ - { - type: "entry", - value: { id: idFromNumber(2), conversationId: ROOT_CONVERSATION_ID, kind: "message", data: {} }, - }, - ], - ]); +/** 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; } @@ -34,8 +41,7 @@ describe("log lines", () => { const store = await seeded(); try { const commits = await readOutbox(store, LOG_START, 100, context); - // The storage's table declarations, then the two commits. - expect(commits.length).toBe(3); + expect(commits).toHaveLength(2); expect(decodeLogLines(commits.map(encodeLogLine).join(""))).toEqual({ commits, torn: 0 }); } finally { await store.close(context); @@ -44,7 +50,7 @@ describe("log lines", () => { it("reports a torn last line instead of failing", () => { const line = encodeLogLine({ seq: 1, nextId: 2, writes: [] }); - expect(decodeLogLines(line + line.slice(0, 5).replace('"seq":1', '"seq":2'))).toEqual({ + expect(decodeLogLines(line + line.slice(0, 5))).toEqual({ commits: [{ seq: 1, nextId: 2, writes: [] }], torn: 5, }); diff --git a/packages/durable/test/sync-remote.test.ts b/packages/chord/test/sync/sync.test.ts similarity index 84% rename from packages/durable/test/sync-remote.test.ts rename to packages/chord/test/sync/sync.test.ts index c09fd9a37..3d564412e 100644 --- a/packages/durable/test/sync-remote.test.ts +++ b/packages/chord/test/sync/sync.test.ts @@ -1,48 +1,42 @@ // The sync protocol over stores with an outbox: slots created once, uploads from the outbox, pulls through -// `applyBatch()`, archives for sessions whose outbox does not reach back, folds into snapshots. +// `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 { openStore, type Store } from "@earendil-works/chord"; -import type { Context } from "@earendil-works/chord/context"; -import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; -import type { JsonValue } from "@earendil-works/chord/json"; import { afterEach, describe, expect, it } from "vitest"; -import { NodeExecutionEnv } from "../src/env/node.ts"; -import { idFromNumber, seqFromNumber } from "../src/ids.ts"; +import { BACKGROUND_CONTEXT, type Context } from "../../src/context/index.ts"; +import { int, 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, - isUrgentCommit, LOG_START, OutboxUnavailableError, pullSession, readOutbox, remoteHead, remoteHistory, + type Seq, type SessionSyncMaintenance, type SyncState, type SyncStateStore, UnsyncedCommitsError, uploadSession, -} from "../src/sync/index.ts"; -import { - type CommitPublication, - type DocumentCreate, - type DocumentId, - type EntryId, - ROOT_CONVERSATION_ID, - type Seq, - type StorageWrite, - type TaskId, -} from "../src/types.ts"; -import { backend, commitAll, reads, sessionStore } from "./sync-support.ts"; +} from "../../src/sync/index.ts"; +import { FileSyncStateStore, FileSystemSyncRemote } from "../../src/sync/node/index.ts"; +import { tableOwner } from "../support.ts"; const context = BACKGROUND_CONTEXT; const SESSION = "session-1"; @@ -55,9 +49,47 @@ afterEach(async () => { }); async function remote(): Promise { - const directory = await mkdtemp(join(tmpdir(), "pi-durable-sync-")); + const directory = await mkdtemp(join(tmpdir(), "chord-sync-")); directories.push(directory); - return new FileSystemSyncRemote(new NodeExecutionEnv({ cwd: directory }), directory); + return new FileSystemSyncRemote(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() }, 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 }[]; + +/** 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 { @@ -107,7 +139,7 @@ function device(name: string, remoteStore: FileSystemSyncRemote) { return store; }, has: (sessionId = SESSION) => copies.has(sessionId), - async commit(batches: readonly (readonly StorageWrite[])[], sessionId = SESSION): Promise { + async commit(batches: readonly Rows[], sessionId = SESSION): Promise { await commitAll(await self.store(sessionId), batches); }, async upload(sessionId = SESSION) { @@ -156,58 +188,34 @@ function device(name: string, remoteStore: FileSystemSyncRemote) { return self; } -const entryId = (id: number) => idFromNumber(id); -const documentId = (id: number) => idFromNumber(id); -const entry = (id: number) => - ({ - type: "entry", - value: { id: entryId(id), conversationId: ROOT_CONVERSATION_ID, kind: "message", data: { id } }, - }) as const satisfies StorageWrite; -const sessionDoc = (id: number): DocumentCreate => ({ - id: documentId(id), - kind: "test.session", - scope: { kind: "session" }, -}); -const task = (id: number, status: "pending" | "running") => - ({ - type: "task", - value: { - id: idFromNumber>(id), - conversationId: ROOT_CONVERSATION_ID, - kind: "test.task", - version: 1, - input: { id }, - background: false, - abortRequested: false, - state: { status, checkpoint: { status } }, - }, - }) as const satisfies StorageWrite; - -/** Ten commits of entries, a task updated in place, and a session document with deltas and a new base. */ -function seedBatches(): StorageWrite[][] { +/** Ten batches of rows: new ids and updates of earlier ones. */ +function seedBatches(): Rows[] { return [ - [{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }], - [entry(2), task(3, "pending")], + [{ id: "a", n: 1 }], [ - { - type: "document.create", - record: sessionDoc(4), - content: { kind: "base", version: 1, value: { n: 0 } }, - }, + { id: "b", n: 1 }, + { id: "c", n: 1 }, ], - [entry(5), task(3, "running")], - [{ type: "document.change", id: documentId(4), content: { kind: "delta", version: 1, ops: [["s", ["n"], 1]] } }], - [entry(6)], - [{ type: "document.change", id: documentId(4), content: { kind: "base", version: 1, value: { n: 5 } } }], - [entry(7)], - [entry(8), task(9, "pending")], - [entry(10)], + [{ 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: seqFromNumber(after), - to: seqFromNumber(to), + after: after, + to: to, device, }); @@ -226,7 +234,7 @@ describe("batches", () => { describe("remoteHistory", () => { const listing = { batches: [{ ...slot(0, 4, "a"), folded: true as const }, slot(4, 10, "a"), slot(10, 12, "b")], - snapshots: [seqFromNumber(4)], + snapshots: [4], }; it("restores from the newest snapshot along the chain of slots", () => { @@ -236,15 +244,15 @@ describe("remoteHistory", () => { }); it("continues a copy from its head, also from inside a batch", () => { - expect(remoteHistory(listing, seqFromNumber(4))).toEqual({ + expect(remoteHistory(listing, 4)).toEqual({ batches: [slot(4, 10, "a"), slot(10, 12, "b")], head: 12, }); - expect(remoteHistory(listing, seqFromNumber(7))).toEqual({ + expect(remoteHistory(listing, 7)).toEqual({ batches: [slot(4, 10, "a"), slot(10, 12, "b")], head: 12, }); - expect(remoteHistory(listing, seqFromNumber(12))).toEqual({ batches: [], head: 12 }); + expect(remoteHistory(listing, 12)).toEqual({ batches: [], head: 12 }); }); }); @@ -320,7 +328,7 @@ describe("syncing a session between devices", () => { 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)]]); + await devices.phone.commit([entry(100)]); const winner = devices[first]; const loser = devices[second]; @@ -418,17 +426,10 @@ describe("syncing a session between devices", () => { // 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: seqFromNumber(1) }, context); - const commits = await readOutbox(await desktop.store(), seqFromNumber(1), 100, context); + await desktop.state.set(SESSION, { syncedHead: 1 }, context); + const commits = await readOutbox(await desktop.store(), 1, 100, context); expect( - await store.putBatch( - SESSION, - seqFromNumber(1), - seqFromNumber(commits.at(-1)!.seq), - "desktop", - await encodeBatch(commits), - context, - ), + 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")]); @@ -502,16 +503,16 @@ describe("syncing a session between devices", () => { 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)]]); + 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.commit([entry(100), entry(101)]); await laptop.upload(); - await laptop.commit([[entry(102)]]); + await laptop.commit([entry(102)]); await laptop.upload(); await replica.pull(); expect(await reads(await replica.store())).toEqual(await reads(await laptop.store())); @@ -584,7 +585,7 @@ describe("host pieces", () => { const store = await remote(); expect(await store.sessions(undefined, context)).toEqual({ sessions: [], cursor: "", more: false }); const laptop = device("laptop", store); - await laptop.commit([[entry(1)]], "s2"); + 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); @@ -606,10 +607,10 @@ describe("host pieces", () => { it("keeps sync state in files", async () => { const directory = await mkdtemp(join(tmpdir(), "pi-durable-sync-state-")); directories.push(directory); - const state = new FileSyncStateStore(new NodeExecutionEnv({ cwd: directory }), join(directory, "sync")); + const state = new FileSyncStateStore(join(directory, "sync")); expect(await state.get(SESSION, context)).toBeUndefined(); - await state.set(SESSION, { syncedHead: seqFromNumber(3), pulling: seqFromNumber(7) }, context); - expect(await state.get(SESSION, context)).toEqual({ syncedHead: seqFromNumber(3), pulling: seqFromNumber(7) }); + 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); @@ -714,17 +715,4 @@ describe("FlushScheduler", () => { await time.advance(60_000); expect(attempts).toHaveLength(3); }); - - it("marks commits with input, answers, or settled submissions as urgent", () => { - const publication = (changes: CommitPublication["changes"]): CommitPublication => ({ - seq: seqFromNumber(1) as Seq, - changes, - }); - expect( - isUrgentCommit( - publication([{ type: "entry", value: { id: 1, conversationId: 1, kind: "pi.user" } }] as never), - ), - ).toBe(true); - expect(isUrgentCommit(publication([{ type: "entry", value: entry(1).value }] as never))).toBe(false); - }); }); diff --git a/packages/coding-agent/examples/tsconfig.json b/packages/coding-agent/examples/tsconfig.json index b4c50cf05..c4e29fdf2 100644 --- a/packages/coding-agent/examples/tsconfig.json +++ b/packages/coding-agent/examples/tsconfig.json @@ -67,6 +67,12 @@ "@earendil-works/chord/tui/launcher": [ "../../../packages/chord/src/node/tui/index.ts" ], + "@earendil-works/chord/sync": [ + "../../../packages/chord/src/sync/index.ts" + ], + "@earendil-works/chord/sync/node": [ + "../../../packages/chord/src/sync/node/index.ts" + ], "@test/contracts/*": [ "../../chord/test/node/fixtures/plugins/contracts/*.ts" ], diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 1b2ca0a02..c12c9b0c0 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -4,7 +4,7 @@ ### Added -- `@earendil-works/pi-durable/sync` (experimental) moves sessions between devices through a remote store, from the outbox of the session's Chord store (`openStore(backend, ctx, { outbox: true })`, Durable's records in it through `StoreStorage`). Every device continues a session as its own: `uploadSession()` uploads the outbox commits after the last sequence the copy shares with the remote as one gzip batch into a slot the remote creates once per start sequence, trims the outbox through what the remote acknowledged, and reports a copy as `diverged` when another device's upload created that slot first, or `behind` when its head fell below what it had uploaded. A session whose outbox does not reach back to its first commit starts with an archive slot (`store.export()`). `pullSession()` catches a copy up through `applyBatch()`, resumes an interrupted pull, or restores into an empty backend (`importStore()` for an archive); `foldSession()` folds the history into a snapshot archive, keeping folded slots taken. `FlushScheduler` and `isUrgentCommit()` decide when to upload. `FileSystemSyncRemote` is a remote in a directory; `FileSyncStateStore` keeps sync state in a directory. +- `isUrgentCommit(publication)`: whether a Durable commit should reach a sync remote soon (a user message, an answer, a settled submission), for `FlushScheduler` in Chord's sync layer (`@earendil-works/chord/sync`). ## [1.5.0] - 2026-11-02 diff --git a/packages/durable/README.md b/packages/durable/README.md index 22f9e8efb..6bd14cd4f 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -891,29 +891,10 @@ await database.transaction(async (transaction) => { ### Sync (experimental) -`@earendil-works/pi-durable/sync` moves sessions between devices through a remote store. It syncs a session's Chord -store, not a `Storage`: open the store with its outbox, `openStore(backend, ctx, { outbox: true })`, and Durable's -records in it with `StoreStorage.open(store, owner, ctx)`. The outbox holds every commit after its floor, written in -the commit's own transaction, so an upload reads `outbox.read(after)` and trims what the remote acknowledged. - -Every device continues a session as its own. A remote keeps one slot per start sequence and creates it once, and only -where the history ends, so the device whose upload creates a slot continues the session from there. A device that finds -the slot taken by another device has diverged: the host renames its copy to a new session and restores the session -fresh. A session whose outbox does not reach back to its first commit (turned on later, or trimmed) starts with an -archive (`store.export()`). The remote folds the history into snapshot archives, keeping folded slots taken. - -```typescript -const store = await openStore(sqliteBackend(db), context, { outbox: true }); -const result = await uploadSession({ sessionId, device, source: store, store: remote, state, context }); -if (result.type === "diverged") forkLocalCopy(sessionId, result.at); // host: rename, upload under the new ID -// Catch up, resume, or restore into an empty backend; nothing may hold the store's claims (no StoreStorage open). -const { store: pulled } = await pullSession({ sessionId, store: remote, local: { kind: "store", store }, state, context }); -``` - -Pulling refuses a copy with commits it has not uploaded (`UnsyncedCommitsError`): upload first. Uploading refuses a -copy that lost commits it had uploaded (`behind`): it pulls them back first. `FlushScheduler` and `isUrgentCommit()` -decide when to upload; `FileSystemSyncRemote` is a remote in a directory and `FileSyncStateStore` keeps sync state in -one, for development and tests. +Sessions move between devices through Chord's sync layer, `@earendil-works/chord/sync`: open a session's store with +its outbox, `openStore(backend, ctx, { outbox: true })`, keep Durable's records in it with +`StoreStorage.open(store, owner, ctx)`, and sync the store. `isUrgentCommit(publication)` tells Chord's +`FlushScheduler` which Durable commits to upload soon: a user message, an answer, a settled submission. A custom backend's `commit(batches, receive, context)` stores `batches` in order, each as one atomic commit with its own seq, checked against the state the earlier ones left, in as few round trips as it can. It stops at the first batch its own checks refuse (global IDs, document actions, copy resolution): the batches before it stay stored, it and the later ones are not. Once they are durable it calls `receive(index, seq)` per stored batch, in order and synchronously, then resolves with `{ stored, refused? }`; it rejects for an unknown outcome (connection lost, a failed statement or `COMMIT`). `receive` is a durability receipt: the Session publishes each batch later, one at a time, so memory and every view never show a partial log. `receive` calls no Storage method and never throws. Custom backends can run the shared conformance suite, which checks this too, with any Vitest- or Jest-compatible runner: diff --git a/packages/durable/package.json b/packages/durable/package.json index ebe451a3d..c32091c32 100644 --- a/packages/durable/package.json +++ b/packages/durable/package.json @@ -60,12 +60,6 @@ "import": "./dist/images/cloudflare.js", "default": "./dist/images/cloudflare.js" }, - "./sync": { - "source": "./src/sync/index.ts", - "types": "./dist/sync/index.d.ts", - "import": "./dist/sync/index.js", - "default": "./dist/sync/index.js" - }, "./storage/memory": { "source": "./src/storage/memory.ts", "types": "./dist/storage/memory.d.ts", diff --git a/packages/durable/src/index.ts b/packages/durable/src/index.ts index 1a3babbee..b6adad00e 100644 --- a/packages/durable/src/index.ts +++ b/packages/durable/src/index.ts @@ -138,6 +138,7 @@ export { UsageDoc, type UsageState } from "./harness/usage.ts"; export type { ConversationView, ConversationWindow } from "./harness/view.ts"; export { createSession } from "./session/session.ts"; export { MemoryStorage } from "./storage/memory.ts"; +export { isUrgentCommit } from "./sync-urgency.ts"; export { defineTask } from "./tasks.ts"; export type { CheckpointInfo, diff --git a/packages/durable/src/sync-urgency.ts b/packages/durable/src/sync-urgency.ts new file mode 100644 index 000000000..fb2a844ca --- /dev/null +++ b/packages/durable/src/sync-urgency.ts @@ -0,0 +1,12 @@ +// When a Durable commit should reach a sync remote soon, for Chord's `FlushScheduler` (`@earendil-works/chord/sync`). + +import type { CommitPublication } from "./types.ts"; + +/** Whether a commit should reach the remote soon: what the user typed, a finished answer, a settled submission. */ +export function isUrgentCommit(publication: CommitPublication): boolean { + return publication.changes.some( + (change) => + (change.type === "entry" && (change.value.kind === "pi.user" || change.value.kind === "pi.assistant")) || + (change.type === "submission" && (change.value.status === "done" || change.value.status === "unanswered")), + ); +} diff --git a/packages/durable/src/sync/filesystem.ts b/packages/durable/src/sync/filesystem.ts deleted file mode 100644 index 24af2bbad..000000000 --- a/packages/durable/src/sync/filesystem.ts +++ /dev/null @@ -1,199 +0,0 @@ -import type { Context } from "@earendil-works/chord/context"; -import type { FileError, FileSystem, Result } from "../env/index.ts"; -import { seqFromNumber } from "../ids.ts"; -import type { JsonObject, Seq } from "../types.ts"; -import { - type BatchInfo, - type PutBatchResult, - type RemoteListing, - remoteHead, - type SessionChanges, - type SessionSyncMaintenance, -} from "./remote.ts"; - -const pad = (value: number): string => String(value).padStart(16, "0"); -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; -} - -function unwrap(result: Result): T { - if (!result.ok) throw result.error; - return result.value; -} - -/** - * 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: FileSystem; - readonly #root: string; - - constructor(fs: FileSystem, root: string) { - this.#fs = fs; - this.#root = root; - } - - async putBatch( - sessionId: string, - after: Seq, - to: Seq, - device: string, - body: Uint8Array, - context: Context, - options: { readonly archive?: boolean } = {}, - ): Promise { - const existing = await this.#slot(sessionId, after, context); - if (existing !== undefined) return { type: "exists", to: existing.to, device: existing.device }; - 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`, - 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.#read(context, sessionId, "batches", `${pad(after)}.batch`); - } - - getSnapshot(sessionId: string, to: Seq, context: Context): Promise { - return this.#read(context, sessionId, "snapshots", `${pad(to)}.batch`); - } - - async putSnapshot(sessionId: string, to: Seq, body: Uint8Array, 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`, - 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, seqFromNumber(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(seqFromNumber(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 = - !result.ok && result.error.code === "not_found" - ? [] - : 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", 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 (!result.ok && result.error.code === "not_found") 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 (!result.ok && result.error.code === "not_found") return undefined; - return unwrap(result); - } - - /** Write through a temporary file and a rename, so a reader never sees half a file. */ - async #write( - sessionId: string, - directory: readonly string[], - name: string, - body: Uint8Array, - 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, body, context)); - 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/durable/src/sync/state.ts b/packages/durable/src/sync/state.ts deleted file mode 100644 index b02ace9ca..000000000 --- a/packages/durable/src/sync/state.ts +++ /dev/null @@ -1,52 +0,0 @@ -import type { Context } from "@earendil-works/chord/context"; -import type { FileError, FileSystem, Result } from "../env/index.ts"; -import type { Seq } from "../types.ts"; -import type { SyncState, SyncStateStore } from "./client.ts"; -import { checkSessionId } from "./filesystem.ts"; - -function unwrap(result: Result): T { - if (!result.ok) throw result.error; - return result.value; -} - -/** - * 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: FileSystem; - readonly #directory: string; - - constructor(fs: FileSystem, 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 (!result.ok && result.error.code === "not_found") 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/durable/test/sync-e2e.test.ts b/packages/durable/test/sync-e2e.test.ts index 1a8d953d5..ea9f1b63a 100644 --- a/packages/durable/test/sync-e2e.test.ts +++ b/packages/durable/test/sync-e2e.test.ts @@ -4,18 +4,25 @@ import { mkdtemp, rename, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { openStore, type Store } from "@earendil-works/chord"; +import { type Backend, openStore, type Store, sqliteBackend } from "@earendil-works/chord"; +import { openNodeSqliteDatabase } from "@earendil-works/chord/sqlite/node"; +import { pullSession, readOutbox, type Seq, uploadSession } from "@earendil-works/chord/sync"; +import { FileSyncStateStore, FileSystemSyncRemote } from "@earendil-works/chord/sync/node"; import { fauxAssistantMessage } from "@earendil-works/pi-ai"; import { afterEach, describe, expect, it } from "vitest"; -import { NodeExecutionEnv } from "../src/env/node.ts"; import { StoreStorage } from "../src/storage/store/index.ts"; -import { FileSyncStateStore, FileSystemSyncRemote, pullSession, readOutbox, uploadSession } from "../src/sync/index.ts"; -import type { Seq, Storage } from "../src/types.ts"; +import type { Storage } from "../src/types.ts"; import { allEntries, type ChatSetup, chatSetup, openChat, textOf, unanswered } from "./chat-support.ts"; import { context } from "./session-support.ts"; -import { backend, SESSION_OWNER } from "./sync-support.ts"; const SESSION = "s1"; +/** The owner name Durable's storage takes in a session store. */ +const SESSION_OWNER = "pi.session"; + +/** A backend over the SQLite file at `path`. */ +async function backend(path: string): Promise { + return sqliteBackend(await openNodeSqliteDatabase(path)); +} const directories: string[] = []; afterEach(async () => { @@ -43,7 +50,7 @@ async function harnessStorage(store: Store): Promise { /** One device: its data directory, its sync state and its faux model. */ async function device(name: string, remote: FileSystemSyncRemote) { const dir = await temp(`pi-durable-e2e-${name}-`); - const state = new FileSyncStateStore(new NodeExecutionEnv({ cwd: dir }), join(dir, "sync")); + const state = new FileSyncStateStore(join(dir, "sync")); const setup: ChatSetup = chatSetup(); const file = (sessionId: string) => join(dir, `${sessionId}.sqlite`); const exists = async (sessionId: string) => @@ -105,7 +112,7 @@ async function device(name: string, remote: FileSystemSyncRemote) { async function setupDevices() { const root = await temp("pi-durable-e2e-remote-"); - const remote = new FileSystemSyncRemote(new NodeExecutionEnv({ cwd: root }), root); + const remote = new FileSystemSyncRemote(root); return { remote, laptop: await device("laptop", remote), desktop: await device("desktop", remote) }; } diff --git a/packages/durable/test/sync-support.ts b/packages/durable/test/sync-support.ts deleted file mode 100644 index c7120a8bd..000000000 --- a/packages/durable/test/sync-support.ts +++ /dev/null @@ -1,75 +0,0 @@ -// A session's copy for the sync tests: a Chord store over a SQLite database with its outbox on, and Durable's storage -// in it under one owner name. The store is what syncs; the storage is opened on it only while something commits. - -import { type Backend, openStore, type Store, sqliteBackend } from "@earendil-works/chord"; -import type { Context } from "@earendil-works/chord/context"; -import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; -import { openNodeSqliteDatabase } from "@earendil-works/chord/sqlite/node"; -import { StoreStorage } from "../src/storage/store/index.ts"; -import { commitWrites } from "../src/testing/commit-writes.ts"; -import type { Storage, StorageWrite } from "../src/types.ts"; - -const context = BACKGROUND_CONTEXT; - -/** The owner name Durable's storage takes in a session store. */ -export const SESSION_OWNER = "pi.session"; - -/** A backend over the SQLite file at `path`, `:memory:` by default. */ -export async function backend(path = ":memory:"): Promise { - return sqliteBackend(await openNodeSqliteDatabase(path)); -} - -/** A session store with its outbox on. */ -export async function sessionStore(path = ":memory:"): Promise { - return openStore(await backend(path), context, { outbox: true }); -} - -/** Runs `fn` on Durable's storage in `store`, released afterwards so the store can be pulled into again. */ -export async function withStorage(store: Store, fn: (storage: Storage) => Promise, ctx: Context = context) { - const storage = await StoreStorage.open(store, SESSION_OWNER, ctx); - try { - return await fn(storage); - } finally { - await storage.close(ctx); - } -} - -/** Commits each batch through Durable's storage in `store`. */ -export async function commitAll(store: Store, batches: readonly (readonly StorageWrite[])[]): Promise { - await withStorage(store, async (storage) => { - for (const batch of batches) await commitWrites(storage, batch, context); - }); -} - -/** - * What a Session reads of the store, as one comparable value: its records and every live document. Two copies that - * read the same hold the same session. - */ -export async function reads(store: Store): Promise> { - return withStorage(store, async (storage) => { - const out: Record = { head: store.lastSeq }; - const conversations = (await storage.scanConversations({}, 1000, undefined, context)).items; - out.conversations = conversations; - for (const conversation of conversations) { - const entries = ( - await storage.scanEntries({ conversationId: conversation.id, order: "ascending" }, 1000, undefined, context) - ).items; - out[`entries:${conversation.id}`] = entries; - } - const tasks = (await storage.scanTasks({}, 1000, undefined, context)).items; - out.tasks = tasks; - out.submissions = (await storage.scanSubmissions({}, 1000, undefined, context)).items; - const scopes = [ - { kind: "session" as const }, - ...conversations.map((conversation) => ({ kind: "conversation" as const, conversationId: conversation.id })), - ...tasks.map((item) => ({ kind: "task" as const, taskId: item.id })), - ]; - for (const scope of scopes) { - const documents = (await storage.scanDocuments({ scope, at: "current" }, 1000, undefined, context)).items; - out[`documents:${JSON.stringify(scope)}`] = await Promise.all( - documents.map(async (record) => ({ record, value: await storage.document(record.id, "current", context) })), - ); - } - return out; - }); -} diff --git a/packages/durable/test/sync-urgency.test.ts b/packages/durable/test/sync-urgency.test.ts new file mode 100644 index 000000000..f5aef7c74 --- /dev/null +++ b/packages/durable/test/sync-urgency.test.ts @@ -0,0 +1,20 @@ +// `isUrgentCommit`: which Durable commits a host uploads soon (Chord's `FlushScheduler`). +import { describe, expect, it } from "vitest"; +import { idFromNumber } from "../src/ids.ts"; +import { isUrgentCommit } from "../src/index.ts"; +import { type CommitPublication, type EntryId, ROOT_CONVERSATION_ID, type Seq } from "../src/types.ts"; + +const publication = (changes: CommitPublication["changes"]): CommitPublication => ({ seq: 1 as Seq, changes }); +const entry = (kind: string) => + ({ + type: "entry", + value: { id: idFromNumber(1), conversationId: ROOT_CONVERSATION_ID, kind, data: {} }, + }) as never; + +describe("isUrgentCommit", () => { + it("marks commits with input, answers, or settled submissions as urgent", () => { + expect(isUrgentCommit(publication([entry("pi.user")]))).toBe(true); + expect(isUrgentCommit(publication([entry("pi.assistant")]))).toBe(true); + expect(isUrgentCommit(publication([entry("message")]))).toBe(false); + }); +}); diff --git a/tsconfig.json b/tsconfig.json index 611472c26..9659f278f 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -25,6 +25,8 @@ "@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/chord/sync/node": ["./packages/chord/src/sync/node/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"],