From f05535517e6b27ff23289c261a73a2636f902a95 Mon Sep 17 00:00:00 2001 From: deepfates Date: Mon, 13 Jul 2026 01:04:12 -0700 Subject: [PATCH 1/6] rename: the package is @deepfates/lync (bin stays lync; format vocabulary unchanged) --- README.md | 66 +++++++++++------------ ROADMAP.md | 2 +- package.json | 2 +- scripts/check-readme-examples.mjs | 2 +- src/index.ts | 2 +- test/cli/sync.test.ts | 2 +- test/cli/synced-store.integration.test.ts | 8 +-- test/client/create.test.ts | 8 +-- test/indexes/memory.test.ts | 2 +- test/relay/attach.test.ts | 6 +-- test/relay/relay.test.ts | 8 +-- test/sync-protocol.test.ts | 4 +- test/synced-store.test.ts | 10 ++-- vitest.config.ts | 16 +++--- 14 files changed, 69 insertions(+), 69 deletions(-) diff --git a/README.md b/README.md index 2f1e5f2..93370e5 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # lync -lync is a file format for append-only interaction history, and `lync-core` is +lync is a file format for append-only interaction history, and `@deepfates/lync` is its reference implementation — one package that ships the parser, event stores, computed views, the loom API, live sync, the `lync` command, and the sync relay. Zero runtime dependencies. @@ -70,7 +70,7 @@ expected-output schema; `generate.py` regenerates digests deterministically. ## The Library ```bash -npm install lync-core +npm install @deepfates/lync ``` Runs in Node (>=22) and the browser. No dependencies. @@ -78,8 +78,8 @@ Runs in Node (>=22) and the browser. No dependencies. ### Parse, union, view ```ts -import { parseLyncFiles } from "lync-core/events"; -import { lyncBranchTreeView, lyncTranscriptView } from "lync-core/views"; +import { parseLyncFiles } from "@deepfates/lync/events"; +import { lyncBranchTreeView, lyncTranscriptView } from "@deepfates/lync/views"; const bytes = new TextEncoder().encode( '{"v":1,"id":"root","kind":"notes/text","at":"2026-07-06T04:12:31Z","author":{"actor":"you"},"parents":[],"payload":{"text":"Once..."}}\n', @@ -105,8 +105,8 @@ loom API gives programs turns and threads instead of raw events, on top of any store: ```ts -import { createLyncLooms } from "lync-core/looms"; -import { createMemoryEventStore } from "lync-core/memory-log"; +import { createLyncLooms } from "@deepfates/lync/looms"; +import { createMemoryEventStore } from "@deepfates/lync/memory-log"; const looms = createLyncLooms({ store: createMemoryEventStore(), @@ -128,8 +128,8 @@ and accepted events without making file order meaningful. An index tracks a collection of looms — upsert entries, subscribe to changes: ```ts -import { loomRef } from "lync-core"; -import { createMemoryLoomIndexes } from "lync-core/indexes/memory"; +import { loomRef } from "@deepfates/lync"; +import { createMemoryLoomIndexes } from "@deepfates/lync/indexes/memory"; const indexes = createMemoryLoomIndexes(); const index = await indexes.create({ title: "My looms" }); @@ -149,10 +149,10 @@ One object that pairs looms with an index and resolves loom/turn/thread/index references to and from URLs: ```ts -import { createLyncLooms } from "lync-core/looms"; -import { createMemoryEventStore } from "lync-core/memory-log"; -import { createMemoryLoomIndexes } from "lync-core/indexes/memory"; -import { createLoomClient } from "lync-core/client"; +import { createLyncLooms } from "@deepfates/lync/looms"; +import { createMemoryEventStore } from "@deepfates/lync/memory-log"; +import { createMemoryLoomIndexes } from "@deepfates/lync/indexes/memory"; +import { createLoomClient } from "@deepfates/lync/client"; const client = createLoomClient({ looms: createLyncLooms({ store: createMemoryEventStore(), author: { actor: "you" } }), @@ -169,7 +169,7 @@ const opened = await client.openReference(client.references.fromUrl(new URL(url) console.log(opened.kind); // "loom" — opened.loom is ready to appendTurn ``` -`lync-core/client/testing` ships `createTestLoomClient`, a fully in-memory +`@deepfates/lync/client/testing` ships `createTestLoomClient`, a fully in-memory client for tests and embedded experiments — deterministic when you pass `createId` and `now`. @@ -182,9 +182,9 @@ collaborators append, because they already recompute through the store's ```ts -import { createMemoryEventStore } from "lync-core/memory-log"; -import { createLyncLooms } from "lync-core/looms"; -import { createSyncedStore, createWebSocketTransport } from "lync-core/synced-store"; +import { createMemoryEventStore } from "@deepfates/lync/memory-log"; +import { createLyncLooms } from "@deepfates/lync/looms"; +import { createSyncedStore, createWebSocketTransport } from "@deepfates/lync/synced-store"; const transport = createWebSocketTransport("wss://host/lync"); const store = createSyncedStore(createMemoryEventStore(), transport, { @@ -206,21 +206,21 @@ in Node. ### Subpath exports -- `lync-core/events` — line parsing, carried-byte export, incremental union -- `lync-core/store` — the event-store contract and serialization -- `lync-core/memory-log`, `lync-core/file-log`, `lync-core/idb-log` — stores +- `@deepfates/lync/events` — line parsing, carried-byte export, incremental union +- `@deepfates/lync/store` — the event-store contract and serialization +- `@deepfates/lync/memory-log`, `@deepfates/lync/file-log`, `@deepfates/lync/idb-log` — stores (`file-log` is node-only; it keeps `node:fs` off the browser path) -- `lync-core/views` — branch tree, transcript, memory, leaderboard -- `lync-core/looms` — the loom/turn API -- `lync-core/references` — loom/turn/thread/index references and URLs -- `lync-core/synced-store` — live sync decorator and WebSocket transport -- `lync-core/sync-protocol` — the five sync frames, encode/decode -- `lync-core/uuid` — zero-dep UUIDv7 for event ids -- `lync-core/indexes`, `lync-core/indexes/entries`, - `lync-core/indexes/memory`, `lync-core/indexes/types` — loom indexes -- `lync-core/client`, `lync-core/client/testing`, `lync-core/client/types` — +- `@deepfates/lync/views` — branch tree, transcript, memory, leaderboard +- `@deepfates/lync/looms` — the loom/turn API +- `@deepfates/lync/references` — loom/turn/thread/index references and URLs +- `@deepfates/lync/synced-store` — live sync decorator and WebSocket transport +- `@deepfates/lync/sync-protocol` — the five sync frames, encode/decode +- `@deepfates/lync/uuid` — zero-dep UUIDv7 for event ids +- `@deepfates/lync/indexes`, `@deepfates/lync/indexes/entries`, + `@deepfates/lync/indexes/memory`, `@deepfates/lync/indexes/types` — loom indexes +- `@deepfates/lync/client`, `@deepfates/lync/client/testing`, `@deepfates/lync/client/types` — the loom client -- `lync-core/relay` — the sync relay (see [The Relay](#the-relay)) +- `@deepfates/lync/relay` — the sync relay (see [The Relay](#the-relay)) ## The Command @@ -228,7 +228,7 @@ The package installs a `lync` bin with seven verbs: `init`, `append`, `verify`, `merge`, `view`, `serve`, and `sync`. ```bash -npm install -g lync-core +npm install -g @deepfates/lync ``` ```bash @@ -273,7 +273,7 @@ client reconnects exactly where it left off. Running a relay is the one thing that needs a WebSocket server, and Node does not ship one — so the relay acquires [`ws`](https://www.npmjs.com/package/ws) -lazily at the moment you construct it. `lync-core` declares no dependency on +lazily at the moment you construct it. `@deepfates/lync` declares no dependency on `ws` at all: install it yourself next to your server (`npm install ws`), and everything else in the package works without it. If you bundle a server that runs the relay, mark `ws` as external — the @@ -283,7 +283,7 @@ Standalone: ```ts -import { startLyncServe } from "lync-core/relay"; +import { startLyncServe } from "@deepfates/lync/relay"; const server = await startLyncServe({ dir: "./rooms", port: 8787 }); console.log("relay on", server.port); @@ -295,7 +295,7 @@ On an existing HTTP server: ```ts import { createServer } from "node:http"; -import { attachLyncServer } from "lync-core/relay"; +import { attachLyncServer } from "@deepfates/lync/relay"; const httpServer = createServer(app); const lync = attachLyncServer(httpServer, { diff --git a/ROADMAP.md b/ROADMAP.md index 147814f..b1d1ca9 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -5,7 +5,7 @@ meaning should land in pacts, not in the envelope. ## Now -- First public release of the one package: `lync-core` (library, indexes, +- First public release of the one package: `@deepfates/lync` (library, indexes, client, relay, and the `lync` command). - Keep `FORMAT.md` and the test vectors aligned as the reference other languages can port. diff --git a/package.json b/package.json index 5568374..67b1c3d 100644 --- a/package.json +++ b/package.json @@ -1,5 +1,5 @@ { - "name": "lync-core", + "name": "@deepfates/lync", "version": "0.3.0", "description": "The lync format: append-only JSONL event logs merged by set union. Parsing, stores, views, looms, live sync, loom client, indexes, the sync relay, and the lync command. Zero dependencies.", "type": "module", diff --git a/scripts/check-readme-examples.mjs b/scripts/check-readme-examples.mjs index 47a8123..97b1099 100644 --- a/scripts/check-readme-examples.mjs +++ b/scripts/check-readme-examples.mjs @@ -14,7 +14,7 @@ // There is no other configuration: a new example is checked by default. // // ts blocks execute from inside the repo root — Node's package self-reference -// resolves "lync-core" and its subpaths exactly like an installed consumer — +// resolves "@deepfates/lync" and its subpaths exactly like an installed consumer — // with cwd in a scratch dir so relative paths never touch the repo. bash // blocks run with the `lync` command token rewritten to the workspace bin; // `npm install` lines are skipped (noted), since installing is the reader's diff --git a/src/index.ts b/src/index.ts index c11208b..5414d37 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,5 +1,5 @@ export * from "./errors.js"; -// The node:fs-backed file store lives only at the explicit "lync-core/file-log" +// The node:fs-backed file store lives only at the explicit "@deepfates/lync/file-log" // subpath so the main barrel stays importable in the browser with zero node builtins. export * from "./idb-log.js"; export * from "./looms.js"; diff --git a/test/cli/sync.test.ts b/test/cli/sync.test.ts index f6897a3..1e5968e 100644 --- a/test/cli/sync.test.ts +++ b/test/cli/sync.test.ts @@ -3,7 +3,7 @@ import { appendFile, mkdtemp, readFile, writeFile } from "node:fs/promises"; import { existsSync } from "node:fs"; import os from "node:os"; import path from "node:path"; -import { startLyncServe, type LyncSyncServer } from "lync-core/relay"; +import { startLyncServe, type LyncSyncServer } from "@deepfates/lync/relay"; import { syncOnce } from "../../src/cli/sync.js"; const quiet = { write: () => true } as const; diff --git a/test/cli/synced-store.integration.test.ts b/test/cli/synced-store.integration.test.ts index 009de56..13a52af 100644 --- a/test/cli/synced-store.integration.test.ts +++ b/test/cli/synced-store.integration.test.ts @@ -2,10 +2,10 @@ import { afterEach, describe, expect, it } from "vitest"; import { mkdtemp } from "node:fs/promises"; import os from "node:os"; import path from "node:path"; -import { createMemoryEventStore } from "lync-core/memory-log"; -import { createLyncLooms, loomRootId } from "lync-core/looms"; -import { createSyncedStore, createWebSocketTransport } from "lync-core/synced-store"; -import { startLyncServe, type LyncSyncServer } from "lync-core/relay"; +import { createMemoryEventStore } from "@deepfates/lync/memory-log"; +import { createLyncLooms, loomRootId } from "@deepfates/lync/looms"; +import { createSyncedStore, createWebSocketTransport } from "@deepfates/lync/synced-store"; +import { startLyncServe, type LyncSyncServer } from "@deepfates/lync/relay"; /** * The embedded browser story, proven end to end: a real relay, two clients diff --git a/test/client/create.test.ts b/test/client/create.test.ts index 56f1e89..ee67750 100644 --- a/test/client/create.test.ts +++ b/test/client/create.test.ts @@ -1,8 +1,8 @@ import { describe, expect, it } from "vitest"; -import { createMemoryEventStore } from "lync-core/memory-log"; -import { createLyncLooms } from "lync-core/looms"; -import { createMemoryLoomIndexes } from "lync-core/indexes/memory"; -import { upsertLoom } from "lync-core/indexes/entries"; +import { createMemoryEventStore } from "@deepfates/lync/memory-log"; +import { createLyncLooms } from "@deepfates/lync/looms"; +import { createMemoryLoomIndexes } from "@deepfates/lync/indexes/memory"; +import { upsertLoom } from "@deepfates/lync/indexes/entries"; import { createLoomClient } from "../../src/client/create.js"; function makeClient() { diff --git a/test/indexes/memory.test.ts b/test/indexes/memory.test.ts index 4f3c541..3cfc26e 100644 --- a/test/indexes/memory.test.ts +++ b/test/indexes/memory.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { loomRef } from "lync-core"; +import { loomRef } from "@deepfates/lync"; import { createMemoryLoomIndexes } from "../../src/indexes/memory.js"; import { upsertLoom } from "../../src/indexes/entries.js"; diff --git a/test/relay/attach.test.ts b/test/relay/attach.test.ts index 3249483..7db23f8 100644 --- a/test/relay/attach.test.ts +++ b/test/relay/attach.test.ts @@ -3,9 +3,9 @@ import { createServer, type Server } from "node:http"; import { mkdtemp } from "node:fs/promises"; import os from "node:os"; import path from "node:path"; -import { createMemoryEventStore } from "lync-core/memory-log"; -import { createLyncLooms, loomRootId } from "lync-core/looms"; -import { createSyncedStore, createWebSocketTransport } from "lync-core/synced-store"; +import { createMemoryEventStore } from "@deepfates/lync/memory-log"; +import { createLyncLooms, loomRootId } from "@deepfates/lync/looms"; +import { createSyncedStore, createWebSocketTransport } from "@deepfates/lync/synced-store"; import { attachLyncServer, type AttachedLyncServer } from "../../src/relay/attach.js"; async function listen(server: Server): Promise { diff --git a/test/relay/relay.test.ts b/test/relay/relay.test.ts index 5ac0c41..3c2d042 100644 --- a/test/relay/relay.test.ts +++ b/test/relay/relay.test.ts @@ -3,9 +3,9 @@ import { createServer, type Server } from "node:http"; import { mkdtemp } from "node:fs/promises"; import os from "node:os"; import path from "node:path"; -import { createMemoryEventStore } from "lync-core/memory-log"; -import { createLyncLooms, loomRootId } from "lync-core/looms"; -import { createSyncedStore, createWebSocketTransport } from "lync-core/synced-store"; +import { createMemoryEventStore } from "@deepfates/lync/memory-log"; +import { createLyncLooms, loomRootId } from "@deepfates/lync/looms"; +import { createSyncedStore, createWebSocketTransport } from "@deepfates/lync/synced-store"; import { createLyncRelay } from "../../src/relay/relay.js"; /** @@ -79,7 +79,7 @@ describe("createLyncRelay durability failures", () => { const os = await import("node:os"); const nodePath = await import("node:path"); const { createServer } = await import("node:http"); - const { createWebSocketTransport } = await import("lync-core/synced-store"); + const { createWebSocketTransport } = await import("@deepfates/lync/synced-store"); const dir = await mkdtemp(nodePath.join(os.tmpdir(), "lync-persist-")); // Read-only dir: recovery (no existing files) succeeds, but every append fails. diff --git a/test/sync-protocol.test.ts b/test/sync-protocol.test.ts index a0770cc..bee445f 100644 --- a/test/sync-protocol.test.ts +++ b/test/sync-protocol.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { decodeFrame, encodeFrame, extractLineId, type SyncFrame } from "lync-core/sync-protocol"; +import { decodeFrame, encodeFrame, extractLineId, type SyncFrame } from "@deepfates/lync/sync-protocol"; describe("lync sync protocol frames", () => { it("round-trips every frame kind", () => { @@ -45,7 +45,7 @@ describe("cursor integrity (dee-inzc blocker)", () => { describe("uuidv7 minting", () => { it("mints valid, time-ordered UUIDv7", async () => { - const { uuidv7 } = await import("lync-core/uuid"); + const { uuidv7 } = await import("@deepfates/lync/uuid"); const a = uuidv7(1_700_000_000_000); const b = uuidv7(1_700_000_000_001); expect(a).toMatch(/^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/); diff --git a/test/synced-store.test.ts b/test/synced-store.test.ts index b83f4ec..9cbbd6a 100644 --- a/test/synced-store.test.ts +++ b/test/synced-store.test.ts @@ -1,14 +1,14 @@ import { describe, expect, it } from "vitest"; -import { createMemoryEventStore } from "lync-core/memory-log"; -import { createLyncLooms } from "lync-core/looms"; +import { createMemoryEventStore } from "@deepfates/lync/memory-log"; +import { createLyncLooms } from "@deepfates/lync/looms"; import { createSyncedStore, type SyncConnectionState, type SyncStatus, type SyncTransport, -} from "lync-core/synced-store"; -import type { SyncFrame } from "lync-core/sync-protocol"; -import { serializeLyncEvent } from "lync-core/store"; +} from "@deepfates/lync/synced-store"; +import type { SyncFrame } from "@deepfates/lync/sync-protocol"; +import { serializeLyncEvent } from "@deepfates/lync/store"; function mockTransport(initial: SyncConnectionState = "online") { const frameHandlers = new Set<(frame: SyncFrame) => void>(); diff --git a/vitest.config.ts b/vitest.config.ts index 11edff1..c4a70ce 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -4,38 +4,38 @@ export default defineConfig({ resolve: { alias: [ { - find: /^lync-core\/profiles\/text-story$/, + find: /^@deepfates\/lync\/profiles\/text-story$/, replacement: new URL( "./src/profiles/text-story.ts", import.meta.url, ).pathname, }, { - find: /^lync-core\/indexes\/([a-z0-9-]+)$/, + find: /^@deepfates\/lync\/indexes\/([a-z0-9-]+)$/, replacement: new URL("./src/indexes/", import.meta.url).pathname + "$1.ts", }, { - find: /^lync-core\/indexes$/, + find: /^@deepfates\/lync\/indexes$/, replacement: new URL("./src/indexes/index.ts", import.meta.url).pathname, }, { - find: /^lync-core\/client\/([a-z0-9-]+)$/, + find: /^@deepfates\/lync\/client\/([a-z0-9-]+)$/, replacement: new URL("./src/client/", import.meta.url).pathname + "$1.ts", }, { - find: /^lync-core\/client$/, + find: /^@deepfates\/lync\/client$/, replacement: new URL("./src/client/index.ts", import.meta.url).pathname, }, { - find: /^lync-core\/relay$/, + find: /^@deepfates\/lync\/relay$/, replacement: new URL("./src/relay/index.ts", import.meta.url).pathname, }, { - find: /^lync-core\/([a-z0-9-]+)$/, + find: /^@deepfates\/lync\/([a-z0-9-]+)$/, replacement: new URL("./src/", import.meta.url).pathname + "$1.ts", }, { - find: /^lync-core$/, + find: /^@deepfates\/lync$/, replacement: new URL("./src/index.ts", import.meta.url).pathname, }, ], From 43034ced1feb21738521bc883e250232a996f3a1 Mon Sep 17 00:00:00 2001 From: deepfates Date: Mon, 13 Jul 2026 01:16:14 -0700 Subject: [PATCH 2/6] =?UTF-8?q?relay:=20log-generation=20id=20=E2=80=94=20?= =?UTF-8?q?a=20cursor=20is=20only=20meaningful=20inside=20the=20generation?= =?UTF-8?q?=20that=20issued=20it=20(dee-u6tq)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit recoverRoom mints a random generation id (never persisted: every restart is a new generation), carried additively on ev/live frames. The CLI sync cursor file stores {seq, generation}; a mismatch resets to 0 and resubscribes — union makes the re-download duplicate no-ops. Stale lives from superseded subs are counted, not trusted. Regression test is the bug's exact shape: persisted events, a broadcast whose disk write failed consuming a seq, a restart, a stale cursor — the client ends with every persisted event (verified failing with the reset neutered). --- src/cli/sync.ts | 62 ++++++++++++++++++++++- src/relay/relay.ts | 17 +++++-- src/sync-protocol.ts | 36 +++++++++++-- test/cli/sync.test.ts | 101 +++++++++++++++++++++++++++++++++++++ test/sync-protocol.test.ts | 27 ++++++++++ 5 files changed, 235 insertions(+), 8 deletions(-) diff --git a/src/cli/sync.ts b/src/cli/sync.ts index 222da47..698a59f 100644 --- a/src/cli/sync.ts +++ b/src/cli/sync.ts @@ -19,6 +19,13 @@ import { decodeFrame, encodeFrame, extractLineId, isCursor } from "../sync-proto * advances only after a received line has reached a durable local state — * appended, recognized as a duplicate, or surfaced as unusable. A sync that * cannot reach `live` within the timeout fails loudly; nothing hangs. + * + * The cursor stores the server's log generation alongside seq: a seq is only + * meaningful inside the generation that issued it (a broadcast whose disk + * write failed still consumed a seq, so after a server restart the recovered + * log can sit behind our cursor). When the server's `gen` differs from the + * stored one, the cursor resets to 0 and we resubscribe from scratch — union + * makes the re-download a set of duplicate no-ops, and nothing is skipped. */ export interface LyncSyncOptions { @@ -50,6 +57,8 @@ interface Cursor { url: string; root: string; seq: number; + /** Server log generation the seq belongs to. Absent: old cursor file or old server. */ + generation?: string; } export async function syncOnce(options: LyncSyncOptions): Promise { @@ -83,8 +92,43 @@ export async function syncOnce(options: LyncSyncOptions): Promise - writeFile(cursorPath, `${JSON.stringify({ url: options.url, root, seq: result.seq } satisfies Cursor, null, 2)}\n`); + writeFile( + cursorPath, + `${JSON.stringify({ url: options.url, root, seq: result.seq, ...(generation !== undefined ? { generation } : {}) } satisfies Cursor, null, 2)}\n`, + ); + + /** + * Returns true when the server's generation differs from the one our cursor + * was saved under — in which case the cursor has been reset to 0 and a fresh + * `sub` from 0 is already on the wire. Frames without gen (old server) never + * trigger a reset. + */ + const generationChanged = (gen: string | undefined): boolean => { + if (gen === undefined || gen === generation) return false; + if (generation === undefined) { + generation = gen; // first sighting: adopt, nothing to reset + return false; + } + options.err.write( + `lync sync: server log generation changed (${generation} -> ${gen}); resyncing ${root} from 0\n`, + ); + generation = gen; + result.seq = 0; + awaitedLives += 1; + socket.send(encodeFrame({ t: "sub", root, since: 0 })); + return true; + }; await new Promise((resolve, reject) => { const timeout = setTimeout(() => { @@ -151,6 +195,10 @@ export async function syncOnce(options: LyncSyncOptions): Promise 0) return; clearTimeout(timeout); result.seq = Math.max(result.seq, frame.seq); if (!options.follow) { @@ -225,6 +280,11 @@ async function readCursor(path: string, url: string, root: string): Promise; @@ -161,9 +170,9 @@ export function createLyncRelay(options: LyncRelayOptions): LyncRelay { send(socket, { t: "err", root: room.root, reason: "recovered-damaged-tail", detail: room.recoveryNote }); } for (let index = frame.since; index < room.lines.length; index += 1) { - send(socket, { t: "ev", root: room.root, seq: index + 1, line: room.lines[index] }); + send(socket, { t: "ev", root: room.root, seq: index + 1, line: room.lines[index], gen: room.generation }); } - send(socket, { t: "live", root: room.root, seq: room.seq }); + send(socket, { t: "live", root: room.root, seq: room.seq, gen: room.generation }); return; } case "ev": { @@ -194,7 +203,7 @@ export function createLyncRelay(options: LyncRelayOptions): LyncRelay { const persisted = await appendSerialized(room, join(options.dir, `${room.root}.lync`), frame.line); // Live delivery is the relay's primary job: fan out even if the disk // write failed. A durability failure is surfaced loudly, never hidden. - broadcast(room, { t: "ev", root: room.root, seq, line: frame.line }); + broadcast(room, { t: "ev", root: room.root, seq, line: frame.line, gen: room.generation }); if (!persisted.ok) { broadcast(room, { t: "err", root: room.root, reason: "persist-failed", detail: id }); } @@ -228,7 +237,7 @@ export function createLyncRelay(options: LyncRelayOptions): LyncRelay { } async function recoverRoom(root: string): Promise { - const room: Room = { root, seq: 0, lines: [], byId: new Map(), subscribers: new Set(), writeChain: Promise.resolve() }; + const room: Room = { root, generation: randomUUID(), seq: 0, lines: [], byId: new Map(), subscribers: new Set(), writeChain: Promise.resolve() }; const path = join(options.dir, `${root}.lync`); if (!existsSync(path)) return room; const text = await readFile(path, "utf8"); diff --git a/src/sync-protocol.ts b/src/sync-protocol.ts index e9c8973..21d816e 100644 --- a/src/sync-protocol.ts +++ b/src/sync-protocol.ts @@ -6,8 +6,8 @@ * union make redundancy harmless. Five frame kinds: * * client → server {"t":"sub", "root": string, "since": number} - * server → client {"t":"ev", "root": string, "seq": number, "line": string} - * server → client {"t":"live", "root": string, "seq": number} + * server → client {"t":"ev", "root": string, "seq": number, "line": string, "gen"?: string} + * server → client {"t":"live", "root": string, "seq": number, "gen"?: string} * client → server {"t":"ev", "root": string, "line": string} * either direction {"t":"presence", "root": string, "data"?: unknown} * either direction {"t":"err", "root"?: string, "reason": string, "detail"?: string} @@ -16,6 +16,17 @@ * event order. The server echoes accepted events to every subscriber of the * root, sender included; echoes are duplicate no-ops under union and still * advance the cursor. This module is pure: frame codecs and guards only. + * + * `gen` is the server's log GENERATION: a random id minted every time a room + * is recovered from disk (so every server restart is a new generation). A + * cursor is only meaningful inside the generation that issued it — a broadcast + * whose disk write failed still consumes a seq, so after a restart the + * recovered log can sit BEHIND a client's saved cursor and the client would + * silently skip the next persisted event forever. A client that sees `gen` + * change from what its cursor was saved under must reset to 0 and resync from + * scratch; union makes the re-download a harmless set of duplicate no-ops. + * The field is additive: frames without it (old servers) decode fine, and old + * clients ignore it. */ export interface SubFrame { @@ -29,12 +40,16 @@ export interface EvFrame { root: string; line: string; seq?: number; + /** Server log generation (see module doc). Absent from old servers and client→server frames. */ + gen?: string; } export interface LiveFrame { t: "live"; root: string; seq: number; + /** Server log generation (see module doc). Absent from old servers. */ + gen?: string; } export interface PresenceFrame { @@ -101,17 +116,32 @@ export function decodeFrame(raw: string | Uint8Array): SyncFrame { if (frame.seq !== undefined && !isCursor(frame.seq)) { return { t: "err", reason: "malformed-ev", detail: "seq must be a nonnegative integer" }; } + // gen is additive: absence is fine (old peers). Present-but-not-a-string + // is malformed — a client resetting its cursor over garbage would be + // acting on noise. + if (frame.gen !== undefined && typeof frame.gen !== "string") { + return { t: "err", reason: "malformed-ev", detail: "gen must be a string" }; + } return { t: "ev", root: frame.root, line: frame.line, ...(frame.seq !== undefined ? { seq: frame.seq as number } : {}), + ...(frame.gen !== undefined ? { gen: frame.gen as string } : {}), }; case "live": if (typeof frame.root !== "string" || !isCursor(frame.seq)) { return { t: "err", reason: "malformed-live" }; } - return { t: "live", root: frame.root, seq: frame.seq as number }; + if (frame.gen !== undefined && typeof frame.gen !== "string") { + return { t: "err", reason: "malformed-live", detail: "gen must be a string" }; + } + return { + t: "live", + root: frame.root, + seq: frame.seq as number, + ...(frame.gen !== undefined ? { gen: frame.gen as string } : {}), + }; case "presence": if (typeof frame.root !== "string") { return { t: "err", reason: "malformed-presence" }; diff --git a/test/cli/sync.test.ts b/test/cli/sync.test.ts index 1e5968e..f9fa612 100644 --- a/test/cli/sync.test.ts +++ b/test/cli/sync.test.ts @@ -258,6 +258,107 @@ describe("cursor corruption recovery (dee-inzc blocker)", () => { }); }); +describe("log generation (dee-u6tq): a cursor is only meaningful inside the generation that issued it", () => { + let server: LyncSyncServer | undefined; + let lockedFile: string | undefined; + + afterEach(async () => { + if (lockedFile) await (await import("node:fs/promises")).chmod(lockedFile, 0o644).catch(() => {}); + lockedFile = undefined; + await server?.close(); + server = undefined; + }); + + it("the bug's exact shape: persisted events, a broadcast whose disk write failed consuming a seq, a restart, a stale cursor — the client ends with EVERY persisted event", async () => { + const { chmod } = await import("node:fs/promises"); + const serverDir = await mkdtemp(path.join(os.tmpdir(), "lync-gen-serve-")); + const clientDir = await mkdtemp(path.join(os.tmpdir(), "lync-gen-client-")); + server = await startLyncServe({ dir: serverDir, log: () => {} }); + const port = server.port; + const url = `ws://localhost:${port}`; + const roomFile = path.join(serverDir, "story.lync"); + + // N events persisted: a producer contributes e1, e2. + const producer = path.join(clientDir, "producer.lync"); + await writeFile(producer, `${eventLine("e1", [], "one")}\n${eventLine("e2", ["e1"], "two")}\n`); + await syncOnce({ file: producer, url, root: "story", out: quiet, err: quiet }); + expect(idsOf(await readFile(roomFile, "utf8"))).toEqual(["e1", "e2"]); + + // The victim client syncs and saves its cursor — with the generation. + const victim = path.join(clientDir, "victim.lync"); + await writeFile(victim, ""); + await syncOnce({ file: victim, url, root: "story", out: quiet, err: quiet }); + const cursor1 = JSON.parse(await readFile(`${victim}.sync.json`, "utf8")) as { seq: number; generation?: string }; + expect(cursor1.seq).toBe(2); + expect(typeof cursor1.generation).toBe("string"); // additive: persisted alongside seq + + // One broadcast with a FAILED disk write consumes seq 3: the room file + // goes read-only, e3 is accepted and fanned out but never persisted. + lockedFile = roomFile; + await chmod(roomFile, 0o444); + await appendFile(producer, `${eventLine("e3", ["e2"], "three, lost to disk")}\n`); + const producerErrs = collect(); + await syncOnce({ file: producer, url, root: "story", out: quiet, err: producerErrs.io }); + expect(idsOf(await readFile(roomFile, "utf8"))).toEqual(["e1", "e2"]); // not on disk + + // The victim, connected during that generation, advances its cursor to 3. + await syncOnce({ file: victim, url, root: "story", out: quiet, err: quiet }); + const cursor2 = JSON.parse(await readFile(`${victim}.sync.json`, "utf8")) as { seq: number; generation?: string }; + expect(cursor2.seq).toBe(3); + expect(idsOf(await readFile(victim, "utf8"))).toEqual(["e1", "e2", "e3"]); + + // Server restart: the disk heals, the relay recovers e1+e2 from disk and + // mints a NEW generation. seq 3 now means something else entirely. + await chmod(roomFile, 0o644); + lockedFile = undefined; + await server.close(); + server = await startLyncServe({ dir: serverDir, port, log: () => {} }); + + // A fresh writer persists e4 in the new generation (its seq: 3). + const writer = path.join(clientDir, "writer.lync"); + await writeFile(writer, `${eventLine("e4", ["e2"], "four, post-restart")}\n`); + await syncOnce({ file: writer, url, root: "story", out: quiet, err: quiet }); + + // The victim reconnects with its stale cursor {seq:3, gen:old}. Pre-fix + // it subscribed since 3 and silently skipped e4 forever. The generation + // mismatch must force a resync from 0 — loudly. + const victimErrs = collect(); + await syncOnce({ file: victim, url, root: "story", out: quiet, err: victimErrs.io }); + expect(victimErrs.text()).toContain("generation changed"); + const victimIds = idsOf(await readFile(victim, "utf8")); + expect(victimIds).toContain("e4"); // the event the old bug skipped forever + expect(victimIds).toEqual(["e1", "e2", "e3", "e4"]); + // Every event persisted on the relay is in the victim's file... + for (const id of idsOf(await readFile(roomFile, "utf8"))) { + expect(victimIds).toContain(id); + } + // ...and the victim's push restored e3 (lost to the dead disk) to it. + expect(idsOf(await readFile(roomFile, "utf8"))).toContain("e3"); + // The cursor now belongs to the new generation. + const cursor3 = JSON.parse(await readFile(`${victim}.sync.json`, "utf8")) as { seq: number; generation?: string }; + expect(cursor3.generation).not.toBe(cursor2.generation); + expect(cursor3.seq).toBeGreaterThanOrEqual(4); + }); + + it("tolerates a pre-generation cursor file (old client state) without resetting", async () => { + const serverDir = await mkdtemp(path.join(os.tmpdir(), "lync-gen-serve-")); + const clientDir = await mkdtemp(path.join(os.tmpdir(), "lync-gen-client-")); + await writeFile(path.join(serverDir, "tale.lync"), `${eventLine("root", [], "one")}\n`); + server = await startLyncServe({ dir: serverDir, log: () => {} }); + const url = `ws://localhost:${server.port}`; + + const file = path.join(clientDir, "tale.lync"); + await writeFile(file, `${eventLine("root", [], "one")}\n`); + // An old cursor file: no generation field at all. + await writeFile(`${file}.sync.json`, `${JSON.stringify({ url, root: "tale", seq: 1 })}\n`); + + const result = await syncOnce({ file, url, root: "tale", out: quiet, err: quiet }); + expect(result.received).toBe(0); // first gen sighting adopts; no spurious resync + const cursor = JSON.parse(await readFile(`${file}.sync.json`, "utf8")) as { seq: number; generation?: string }; + expect(typeof cursor.generation).toBe("string"); // upgraded in place + }); +}); + describe("conflict sidecar durability (dee-inzc major)", () => { let server: LyncSyncServer | undefined; let lockedDir: string | undefined; diff --git a/test/sync-protocol.test.ts b/test/sync-protocol.test.ts index bee445f..bd30fbe 100644 --- a/test/sync-protocol.test.ts +++ b/test/sync-protocol.test.ts @@ -43,6 +43,33 @@ describe("cursor integrity (dee-inzc blocker)", () => { }); }); +describe("log generation field (dee-u6tq)", () => { + it("round-trips gen on ev and live frames", () => { + const frames: SyncFrame[] = [ + { t: "ev", root: "story", line: '{"id":"a"}', seq: 3, gen: "gen-1" }, + { t: "live", root: "story", seq: 7, gen: "gen-1" }, + ]; + for (const frame of frames) { + expect(decodeFrame(encodeFrame(frame))).toEqual(frame); + } + }); + + it("stays tolerant of gen's absence — old peers decode fine, both directions", () => { + // Old server -> new client: no gen on the wire. + expect(decodeFrame('{"t":"ev","root":"r","line":"{}","seq":2}')).toEqual({ t: "ev", root: "r", line: "{}", seq: 2 }); + expect(decodeFrame('{"t":"live","root":"r","seq":7}')).toEqual({ t: "live", root: "r", seq: 7 }); + // New client -> old server: encoding without gen adds nothing. + expect(encodeFrame({ t: "ev", root: "r", line: "{}" })).not.toContain("gen"); + // An unknown extra field from a NEWER peer is dropped, not fatal. + expect(decodeFrame('{"t":"live","root":"r","seq":7,"gen":"g","future":true}')).toEqual({ t: "live", root: "r", seq: 7, gen: "g" }); + }); + + it("rejects a non-string gen — a cursor reset must never act on noise", () => { + expect(decodeFrame('{"t":"ev","root":"r","line":"{}","seq":2,"gen":42}')).toMatchObject({ t: "err", reason: "malformed-ev" }); + expect(decodeFrame('{"t":"live","root":"r","seq":7,"gen":{}}')).toMatchObject({ t: "err", reason: "malformed-live" }); + }); +}); + describe("uuidv7 minting", () => { it("mints valid, time-ordered UUIDv7", async () => { const { uuidv7 } = await import("@deepfates/lync/uuid"); From e65c5bdd16208c14b8d36e1b3b19a099f4ce2624 Mon Sep 17 00:00:00 2001 From: deepfates Date: Mon, 13 Jul 2026 01:18:31 -0700 Subject: [PATCH 3/6] synced-store: await + inspect every union BEFORE the cursor advances (dee-s6dc) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Frames apply strictly in arrival order (serialized chain). A store-write failure freezes the root's cursor — live frames cannot leapfrog it — and screams through the new additive SyncStatus.failures channel; the line is re-fetched on the next resubscribe. Conflicts and garbage are surfaced too. Also carries the generation reset (dee-u6tq) on the synced-store side, with stale-live counting so a superseded sub cannot re-poison a reset cursor. Neuter-verified: freezing disabled makes the regression tests fail. --- src/synced-store.ts | 118 ++++++++++++++++++-- test/synced-store.test.ts | 222 +++++++++++++++++++++++++++++++++++++- 2 files changed, 332 insertions(+), 8 deletions(-) diff --git a/src/synced-store.ts b/src/synced-store.ts index d0e0db7..76b49c2 100644 --- a/src/synced-store.ts +++ b/src/synced-store.ts @@ -25,6 +25,13 @@ export interface SyncStatus { liveRoots: string[]; /** Ids that arrived as same-id-different-body conflicts, surfaced never resolved. */ conflicts: string[]; + /** + * Every ingest failure, surfaced never swallowed: a remote line the local + * store could not durably accept (store write threw) or rejected as + * garbage. A store-write failure also freezes the root's resume cursor so + * the line is re-fetched on the next resubscribe instead of being skipped. + */ + failures: string[]; } export interface SyncTransport { @@ -61,6 +68,24 @@ export function createSyncedStore( const liveRoots = new Set(); const cursors = new Map(); const conflicts = new Set(); + const failures: string[] = []; + // Server log generation per root. A cursor is only meaningful inside the + // generation that issued it (a failed relay disk write still consumes a + // seq, so a restarted server's recovered log can sit BEHIND our cursor). + // On a generation change the cursor resets to 0 and we resubscribe; union + // makes the re-download duplicate no-ops. + const generations = new Map(); + // Roots whose cursor is frozen because a union failed (store write threw): + // the cursor must not advance past the hole, or the line would be skipped + // forever. Cleared on resync — the resubscribe re-fetches from the frozen + // cursor and each refetched line advances it again as its union succeeds. + const stalledRoots = new Set(); + // Outstanding `sub` frames per root: each sub is answered by exactly one + // `live`, in order. While a generation-reset sub is stacked behind an + // earlier one (count > 1), cursor advances are suppressed — frames from the + // superseded sub carry seqs the reset backlog has not re-covered yet, and + // trusting them would re-poison the freshly reset cursor. + const pendingLives = new Map(); let connection: SyncConnectionState = transport.state; const emitStatus = () => { @@ -68,6 +93,7 @@ export function createSyncedStore( connection, liveRoots: [...liveRoots], conflicts: [...conflicts], + failures: [...failures], }); }; @@ -99,6 +125,12 @@ export function createSyncedStore( for (const event of await inner.byRoot(rootId)) { pushLine(rootId, event.bytes); } + // A frozen cursor thaws here: the sub below re-fetches from it, and each + // refetched line advances it again as its union succeeds. + stalledRoots.delete(rootId); + // A fresh connection: any lives owed by subs on the dead connection will + // never arrive, so the count restarts at this sub's one. + pendingLives.set(rootId, 1); transport.send({ t: "sub", root: rootId, since: cursors.get(rootId) ?? 0 }); } @@ -112,19 +144,78 @@ export function createSyncedStore( emitStatus(); }); - transport.onFrame((frame) => { + /** + * Returns true when the server's generation differs from the one this root's + * cursor belongs to — in which case the cursor has been reset to 0 and a + * fresh `sub` from 0 is already on the wire. Frames without gen (old + * servers) never trigger a reset. + */ + const generationChanged = (root: string, gen: string | undefined): boolean => { + if (gen === undefined) return false; + const known = generations.get(root); + if (known === gen) return false; + generations.set(root, gen); + if (known === undefined) return false; // first sighting: adopt + failures.push(`generation changed for ${root} (${known} -> ${gen}); resyncing from 0`); + cursors.set(root, 0); + stalledRoots.delete(root); + pendingLives.set(root, (pendingLives.get(root) ?? 0) + 1); + emitStatus(); + transport.send({ t: "sub", root, since: 0 }); + return true; + }; + + const advanceCursor = (root: string, seq: number) => { + if (stalledRoots.has(root)) return; // frozen behind a failed union + if ((pendingLives.get(root) ?? 0) > 1) return; // a reset-sub's backlog is still owed + cursors.set(root, Math.max(cursors.get(root) ?? 0, seq)); + }; + + async function handleFrame(frame: SyncFrame): Promise { switch (frame.t) { case "ev": { // Remote line: ingest through union WITHOUT re-pushing (the relay has - // already fanned it out). Subscribers fire via the inner store. - void inner.union(frame.line); - if (typeof frame.seq === "number") { - cursors.set(frame.root, Math.max(cursors.get(frame.root) ?? 0, frame.seq)); + // already fanned it out). Subscribers fire via the inner store. The + // union is awaited and inspected BEFORE the cursor advances — a line + // the local store failed to accept must be re-fetched, never skipped. + generationChanged(frame.root, frame.gen); + let outcome: AppendResult; + try { + outcome = await inner.union(frame.line); + } catch (error) { + // Store write failed: the line is NOT durable locally. Freeze the + // cursor so the next resubscribe re-fetches it, and scream. + failures.push(`store failed to ingest a synced line for ${frame.root}: ${String(error)}`); + stalledRoots.add(frame.root); + emitStatus(); + return; + } + switch (outcome.status) { + case "conflict": + conflicts.add(outcome.event.body.id); + emitStatus(); + break; + case "garbage": + // Unusable bytes stay unusable on any re-fetch: surfaced loudly, + // and the cursor may advance past them. + failures.push(`synced line rejected as garbage for ${frame.root}: ${outcome.reason}`); + emitStatus(); + break; + default: + break; // added / duplicate / buffered: durably in the store's hands } + if (typeof frame.seq === "number") advanceCursor(frame.root, frame.seq); return; } case "live": { - cursors.set(frame.root, Math.max(cursors.get(frame.root) ?? 0, frame.seq)); + // A stale live is not live: either its generation is dead (the + // resubscribe from 0 is already on the wire) or it answers a sub a + // generation reset has since superseded. Wait for the real one. + const changed = generationChanged(frame.root, frame.gen); + const outstanding = Math.max(0, (pendingLives.get(frame.root) ?? 1) - 1); + pendingLives.set(frame.root, outstanding); + if (changed || outstanding > 0) return; + advanceCursor(frame.root, frame.seq); liveRoots.add(frame.root); emitStatus(); return; @@ -143,6 +234,19 @@ export function createSyncedStore( default: return; } + } + + // Frames apply strictly in arrival order: each union is awaited before the + // next frame is touched, so a slow union can never let a later frame (or a + // `live` cursor jump) leapfrog a failure. handleFrame never rejects — the + // catch above is the only throw path and it returns — but the chain guards + // anyway so one surprise cannot wedge sync forever. + let frameChain: Promise = Promise.resolve(); + transport.onFrame((frame) => { + frameChain = frameChain.then( + () => handleFrame(frame), + () => handleFrame(frame), + ); }); return { @@ -170,7 +274,7 @@ export function createSyncedStore( ...(inner.diagnostics ? { diagnostics: () => inner.diagnostics!() } : {}), syncRoot: ensureSynced, presence: (root, data) => transport.send({ t: "presence", root, data }), - status: () => ({ connection, liveRoots: [...liveRoots], conflicts: [...conflicts] }), + status: () => ({ connection, liveRoots: [...liveRoots], conflicts: [...conflicts], failures: [...failures] }), close: () => transport.close(), }; } diff --git a/test/synced-store.test.ts b/test/synced-store.test.ts index 9cbbd6a..833f6b3 100644 --- a/test/synced-store.test.ts +++ b/test/synced-store.test.ts @@ -8,7 +8,7 @@ import { type SyncTransport, } from "@deepfates/lync/synced-store"; import type { SyncFrame } from "@deepfates/lync/sync-protocol"; -import { serializeLyncEvent } from "@deepfates/lync/store"; +import { serializeLyncEvent, type AppendResult, type EventStore, type StoredEvent } from "@deepfates/lync/store"; function mockTransport(initial: SyncConnectionState = "online") { const frameHandlers = new Set<(frame: SyncFrame) => void>(); @@ -94,6 +94,9 @@ describe("createSyncedStore", () => { store.syncRoot("r1"); mock.inject({ t: "live", root: "r1", seq: 5 }); mock.inject({ t: "err", root: "r1", reason: "same-id-conflict", detail: "dup-id" }); + // Frames apply in a serialized chain (unions are awaited in arrival + // order), so settle before reading status. + await new Promise((r) => setTimeout(r, 10)); const status = store.status(); expect(status.liveRoots).toContain("r1"); @@ -117,3 +120,220 @@ describe("createSyncedStore", () => { expect(after.some((f) => f.t === "ev")).toBe(true); // backlog re-pushed }); }); + +/** + * An EventStore decorator whose union can be made to fail on command — the + * shape of a full disk or a dead IndexedDB. Records call/finish order so + * ordering tests can prove unions are serialized. + */ +function breakableStore(inner: EventStore) { + let broken = false; + let delayNextMs = 0; + const unionLog: string[] = []; + const store: EventStore = { + append: (ev) => inner.append(ev), + union: async (line: string): Promise => { + unionLog.push(`start:${JSON.parse(line).id}`); + const delay = delayNextMs; + delayNextMs = 0; + if (delay > 0) await new Promise((r) => setTimeout(r, delay)); + if (broken) { + unionLog.push(`fail:${JSON.parse(line).id}`); + throw new Error("injected store-write failure"); + } + const result = await inner.union(line); + unionLog.push(`done:${JSON.parse(line).id}`); + return result; + }, + byId: (id) => inner.byId(id), + byRoot: (rootId) => inner.byRoot(rootId), + subscribe: (rootId, listener) => inner.subscribe(rootId, listener), + roots: (kind) => inner.roots(kind), + }; + return { + store, + unionLog, + setBroken: (b: boolean) => (broken = b), + delayNext: (ms: number) => (delayNextMs = ms), + }; +} + +const settle = () => new Promise((r) => setTimeout(r, 25)); + +describe("awaited union (dee-s6dc): the receive cursor advances only on inspected success", () => { + const line = (id: string, parents: string[], text: string) => serializeLyncEvent(body(id, parents, text)); + + it("a failed union on frame k freezes the cursor at k-1, surfaces the failure, and the event applies after heal + resubscribe", async () => { + const statuses: SyncStatus[] = []; + const mock = mockTransport(); + const flaky = breakableStore(createMemoryEventStore()); + const store = createSyncedStore(flaky.store, mock.transport, { onStatus: (s) => statuses.push(s) }); + store.syncRoot("r1"); + await settle(); + + // Frame 1 lands; the store then breaks; frame 2 (seq k=2) fails; frame 3 + // still applies (arrival order is preserved past the failure). + mock.inject({ t: "ev", root: "r1", seq: 1, line: line("r1", [], "one") }); + await settle(); + flaky.setBroken(true); + mock.inject({ t: "ev", root: "r1", seq: 2, line: line("lost", ["r1"], "two, refused by disk") }); + await settle(); + flaky.setBroken(false); + mock.inject({ t: "ev", root: "r1", seq: 3, line: line("later", ["r1"], "three") }); + mock.inject({ t: "live", root: "r1", seq: 3 }); + await settle(); + + // The failure screamed through onStatus and status(). + expect(store.status().failures.some((f) => f.includes("injected store-write failure"))).toBe(true); + expect(statuses.some((s) => s.failures.some((f) => f.includes("injected store-write failure")))).toBe(true); + // The failed line is NOT in the store; the later one is (order held). + expect(await store.byId("lost")).toBeNull(); + expect(await store.byId("later")).not.toBeNull(); + + // The cursor stayed at k-1 = 1: the resubscribe after reconnect asks the + // relay for everything from there — the lost line gets re-fetched. + mock.setState("offline"); + mock.setState("online"); + mock.open(); + await settle(); + const resub = mock.sent.filter((f) => f.t === "sub" && f.root === "r1").at(-1)!; + expect(resub).toMatchObject({ t: "sub", since: 1 }); + + // The store has healed; the relay replays from the cursor; all applies. + mock.inject({ t: "ev", root: "r1", seq: 2, line: line("lost", ["r1"], "two, refused by disk") }); + mock.inject({ t: "ev", root: "r1", seq: 3, line: line("later", ["r1"], "three") }); + mock.inject({ t: "live", root: "r1", seq: 3 }); + await settle(); + expect(await store.byId("lost")).not.toBeNull(); + + // And the cursor thawed: the next resubscribe resumes past the hole. + mock.setState("offline"); + mock.setState("online"); + mock.open(); + await settle(); + const finalSub = mock.sent.filter((f) => f.t === "sub" && f.root === "r1").at(-1)!; + expect(finalSub).toMatchObject({ t: "sub", since: 3 }); + }); + + it("a live frame cannot leapfrog a failed union: the frozen cursor wins over live's seq", async () => { + const mock = mockTransport(); + const flaky = breakableStore(createMemoryEventStore()); + const store = createSyncedStore(flaky.store, mock.transport, {}); + store.syncRoot("r1"); + await settle(); + + flaky.setBroken(true); + mock.inject({ t: "ev", root: "r1", seq: 1, line: line("r1", [], "refused") }); + mock.inject({ t: "live", root: "r1", seq: 5 }); // relay is far ahead + await settle(); + flaky.setBroken(false); + + mock.setState("offline"); + mock.setState("online"); + mock.open(); + await settle(); + const resub = mock.sent.filter((f) => f.t === "sub" && f.root === "r1").at(-1)!; + expect(resub).toMatchObject({ t: "sub", since: 0 }); // NOT 5 + }); + + it("unions apply strictly in arrival order — a slow union never lets a later frame pass it", async () => { + const mock = mockTransport(); + const flaky = breakableStore(createMemoryEventStore()); + const store = createSyncedStore(flaky.store, mock.transport, {}); + store.syncRoot("r1"); + await settle(); + + flaky.delayNext(60); // frame 1's union is slow + mock.inject({ t: "ev", root: "r1", seq: 1, line: line("r1", [], "slow") }); + mock.inject({ t: "ev", root: "r1", seq: 2, line: line("fast", ["r1"], "fast") }); + await new Promise((r) => setTimeout(r, 150)); + + expect(await store.byId("fast")).not.toBeNull(); + // The second union START comes after the first union DONE: serialized. + const relevant = flaky.unionLog.filter((entry) => entry.endsWith(":r1") || entry.endsWith(":fast")); + expect(relevant).toEqual(["start:r1", "done:r1", "start:fast", "done:fast"]); + }); + + it("garbage from the relay is surfaced, never silently skipped — and never wedges the cursor", async () => { + const statuses: SyncStatus[] = []; + const mock = mockTransport(); + const store = createSyncedStore(createMemoryEventStore(), mock.transport, { onStatus: (s) => statuses.push(s) }); + store.syncRoot("r1"); + await settle(); + + mock.inject({ t: "ev", root: "r1", seq: 1, line: '{"id":"junk","not":"a lync event"}' }); + mock.inject({ t: "ev", root: "r1", seq: 2, line: line("r1", [], "real") }); + mock.inject({ t: "live", root: "r1", seq: 2 }); + await settle(); + + expect(store.status().failures.some((f) => f.includes("garbage"))).toBe(true); + expect(await store.byId("r1")).not.toBeNull(); + // Unusable bytes stay unusable on any re-fetch: the cursor moves past them. + mock.setState("offline"); + mock.setState("online"); + mock.open(); + await settle(); + const resub = mock.sent.filter((f) => f.t === "sub" && f.root === "r1").at(-1)!; + expect(resub).toMatchObject({ t: "sub", since: 2 }); + }); +}); + +describe("generation change in the synced store (dee-u6tq)", () => { + const line = (id: string, parents: string[], text: string) => serializeLyncEvent(body(id, parents, text)); + + it("resets the cursor to 0 and resubscribes when the server's generation changes", async () => { + const statuses: SyncStatus[] = []; + const mock = mockTransport(); + const store = createSyncedStore(createMemoryEventStore(), mock.transport, { onStatus: (s) => statuses.push(s) }); + store.syncRoot("r1"); + await settle(); + + // Generation g1: three events, cursor 3. + mock.inject({ t: "ev", root: "r1", seq: 1, line: line("r1", [], "one"), gen: "g1" }); + mock.inject({ t: "ev", root: "r1", seq: 2, line: line("two", ["r1"], "two"), gen: "g1" }); + mock.inject({ t: "ev", root: "r1", seq: 3, line: line("three", ["r1"], "three"), gen: "g1" }); + mock.inject({ t: "live", root: "r1", seq: 3, gen: "g1" }); + await settle(); + + // The server restarted (lost the unpersisted third event): new generation, + // and its live sits BEHIND our cursor. Pre-fix we would idle forever and + // silently skip the next persisted event. + const before = mock.sent.length; + mock.inject({ t: "live", root: "r1", seq: 2, gen: "g2" }); + await settle(); + + const resub = mock.sent.slice(before).filter((f) => f.t === "sub" && f.root === "r1"); + expect(resub).toEqual([{ t: "sub", root: "r1", since: 0 }]); + // The reset is surfaced, not silent. + expect(store.status().failures.some((f) => f.includes("generation changed"))).toBe(true); + + // The new generation's backlog replays; a NEW event (seq 3 in g2) lands. + mock.inject({ t: "ev", root: "r1", seq: 1, line: line("r1", [], "one"), gen: "g2" }); + mock.inject({ t: "ev", root: "r1", seq: 2, line: line("two", ["r1"], "two"), gen: "g2" }); + mock.inject({ t: "ev", root: "r1", seq: 3, line: line("fresh", ["r1"], "post-restart"), gen: "g2" }); + mock.inject({ t: "live", root: "r1", seq: 3, gen: "g2" }); + await settle(); + expect(await store.byId("fresh")).not.toBeNull(); + + // Cursor now belongs to g2: next resubscribe resumes from 3. + mock.setState("offline"); + mock.setState("online"); + mock.open(); + await settle(); + const finalSub = mock.sent.filter((f) => f.t === "sub" && f.root === "r1").at(-1)!; + expect(finalSub).toMatchObject({ t: "sub", since: 3 }); + }); + + it("frames without gen (old server) never trigger a reset", async () => { + const mock = mockTransport(); + const store = createSyncedStore(createMemoryEventStore(), mock.transport, {}); + store.syncRoot("r1"); + await settle(); + mock.inject({ t: "ev", root: "r1", seq: 1, line: line("r1", [], "one") }); + mock.inject({ t: "live", root: "r1", seq: 1 }); + await settle(); + expect(mock.sent.filter((f) => f.t === "sub" && f.since === 0)).toHaveLength(1); // only the original + expect(store.status().failures).toEqual([]); + }); +}); + From 39a30d4c7c9e63b7d698866f32aa01cd92b8605a Mon Sep 17 00:00:00 2001 From: deepfates Date: Mon, 13 Jul 2026 01:27:12 -0700 Subject: [PATCH 4/6] synced-store: surface every relay-side error, never silently drop it (dee-i1wc) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The synced store swallowed all relay errs except same-id-conflict — a persist-failed durability failure fanned out but never reached the client's status channel, so a store consumer could not see the relay lose a write. Route every non-conflict relay err (persist-failed, conflict-persist-failed, recovered-damaged-tail, line-without-id, server-error, ...) into the failures channel. Nothing fails invisibly. Unit test asserts persist-failed surfaces and is not miscategorised as a conflict. --- src/synced-store.ts | 10 +++++++++- test/synced-store.test.ts | 17 +++++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/src/synced-store.ts b/src/synced-store.ts index 76b49c2..c5fd74c 100644 --- a/src/synced-store.ts +++ b/src/synced-store.ts @@ -227,8 +227,16 @@ export function createSyncedStore( case "err": { if (frame.reason === "same-id-conflict" && frame.detail) { conflicts.add(frame.detail); - emitStatus(); + } else { + // Every other relay-side failure — persist-failed, conflict-persist-failed, + // recovered-damaged-tail, line-without-id, unexpected-live-from-client, + // server-error — is a failure the client must see, never a silent drop. + // A durability failure on the relay reaches the client's status channel. + const where = frame.root ? ` for ${frame.root}` : ""; + const detail = frame.detail ? ` (${frame.detail})` : ""; + failures.push(`relay error${where}: ${frame.reason}${detail}`); } + emitStatus(); return; } default: diff --git a/test/synced-store.test.ts b/test/synced-store.test.ts index 833f6b3..31875a1 100644 --- a/test/synced-store.test.ts +++ b/test/synced-store.test.ts @@ -104,6 +104,23 @@ describe("createSyncedStore", () => { expect(statuses.length).toBeGreaterThanOrEqual(2); }); + it("surfaces every non-conflict relay error into the failures channel, never silently drops it", async () => { + const statuses: SyncStatus[] = []; + const mock = mockTransport(); + const store = createSyncedStore(createMemoryEventStore(), mock.transport, { + onStatus: (s) => statuses.push(s), + }); + store.syncRoot("r1"); + // A relay durability failure: the fan-out landed but the disk write did not. + mock.inject({ t: "err", root: "r1", reason: "persist-failed", detail: "evt-9" }); + await new Promise((r) => setTimeout(r, 10)); + + const status = store.status(); + expect(status.failures.some((f) => f.includes("persist-failed") && f.includes("evt-9"))).toBe(true); + expect(status.conflicts).toEqual([]); // a persist failure is not a conflict + expect(statuses.some((s) => s.failures.some((f) => f.includes("persist-failed")))).toBe(true); + }); + it("re-pushes local backlog and re-subscribes on reconnect", async () => { const mock = mockTransport("online"); const store = createSyncedStore(createMemoryEventStore(), mock.transport); From 48e4690c11a0784390cd2732a4ad515440670241 Mon Sep 17 00:00:00 2001 From: deepfates Date: Mon, 13 Jul 2026 01:27:26 -0700 Subject: [PATCH 5/6] =?UTF-8?q?test:=20the=20loss-free=20trial=20=E2=80=94?= =?UTF-8?q?=20milestone-6=20durability=20proof=20(dee-i1wc)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A real relay plus real synced stores over the global WebSocket, run through the full gauntlet against one shared root: (a) a client's socket drops mid-stream and auto-reconnects; (b) the relay's .lync file goes read-only, so a write is fanned out but refused by disk and surfaced as persist-failed; (c) the server restarts into a new log generation. Final invariant: every event a client successfully appended — including the one the dead disk refused — ends up in every other client's store AND on the relay's on-disk .lync file, and every failure (persist-failed, generation-changed) was surfaced on every client's status channel. Neuter-verified: dropping the err surfacing fails leg (b); neutering the generation reset fails leg (c). --- test/cli/loss-free-trial.integration.test.ts | 231 +++++++++++++++++++ 1 file changed, 231 insertions(+) create mode 100644 test/cli/loss-free-trial.integration.test.ts diff --git a/test/cli/loss-free-trial.integration.test.ts b/test/cli/loss-free-trial.integration.test.ts new file mode 100644 index 0000000..3455e8e --- /dev/null +++ b/test/cli/loss-free-trial.integration.test.ts @@ -0,0 +1,231 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { chmod, mkdtemp, readFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { createMemoryEventStore } from "@deepfates/lync/memory-log"; +import { createSyncedStore, createWebSocketTransport, type SyncStatus } from "@deepfates/lync/synced-store"; +import { startLyncServe, type LyncSyncServer } from "@deepfates/lync/relay"; + +/** + * The loss-free trial (dee-i1wc) — the world-charter milestone-6 proof. + * + * A real relay plus real synced stores over the global WebSocket, run through + * the full durability gauntlet, in three legs against ONE shared root: + * + * (a) a client disconnects mid-stream and reconnects; + * (b) the relay's storage fails mid-run (the .lync file goes read-only) and + * then recovers; + * (c) the server process restarts — a new log generation. + * + * The invariant asserted after the gauntlet, and the surfacing asserted per + * leg: every event any client SUCCESSFULLY appended (local append returned + * `added`) ends up in every other client's store AND in the relay's on-disk + * .lync file — nothing lost, nothing silently skipped — and every failure that + * occurred was surfaced through a status/err channel, not swallowed. + * + * Why leg (b)'s lost-to-disk event only reaches disk after leg (c): a relay + * whose write fails still fans the event out and holds it in memory, so a + * same-generation re-push is a byId duplicate no-op — the line never re-hits + * disk until a restart drops the in-memory copy, mints a new generation, and + * the clients resync from 0 and re-push it fresh. That is the loss-free + * property under a storage failure: the generation reset is what closes it. + */ + +const ROOT = "trial"; + +function idsOf(text: string): string[] { + return text + .split("\n") + .filter((line) => line.length > 0) + .map((line) => { + try { + return (JSON.parse(line) as { id?: string }).id ?? ""; + } catch { + return ""; + } + }); +} + +/** + * A WebSocket subclass that records every instance it constructs, so a test + * can force-close ONE client's live socket — a network drop for that client + * alone, leaving the transport's auto-reconnect to bring it back. + */ +function trackedWebSocket(): { impl: typeof WebSocket; sockets: WebSocket[] } { + const sockets: WebSocket[] = []; + const Real = (globalThis as { WebSocket: typeof WebSocket }).WebSocket; + class Tracked extends Real { + constructor(url: string | URL, protocols?: string | string[]) { + super(url, protocols); + sockets.push(this as unknown as WebSocket); + } + } + return { impl: Tracked as unknown as typeof WebSocket, sockets }; +} + +function makeClient(url: string, actor: string) { + const inner = createMemoryEventStore(); + const tracker = trackedWebSocket(); + const statuses: SyncStatus[] = []; + const transport = createWebSocketTransport(url, { reconnectMs: 30, WebSocketImpl: tracker.impl }); + const store = createSyncedStore(inner, transport, { onStatus: (s) => statuses.push(s) }); + // Every appended id whose LOCAL append returned "added" — the events the + // trial promises never to lose. + const appended = new Set(); + const append = async (id: string, parents: string[], text: string) => { + const result = await store.append({ + v: 1, + id, + kind: "lync/artifact", + at: "2026-07-08T21:00:00Z", + author: { actor }, + parents, + payload: { text }, + }); + if (result.status === "added") appended.add(id); + return result; + }; + const sawFailure = (needle: string) => + statuses.some((s) => s.failures.some((f) => f.includes(needle))) || + store.status().failures.some((f) => f.includes(needle)); + return { actor, store, statuses, appended, append, sawFailure, dropSocket: () => tracker.sockets.at(-1)?.close() }; +} +type Client = ReturnType; + +async function waitFor(check: () => Promise | boolean, timeoutMs = 8_000): Promise { + const deadline = Date.now() + timeoutMs; + let last = "condition not met"; + while (Date.now() < deadline) { + try { + if (await check()) return; + } catch (error) { + last = String(error); + } + await new Promise((r) => setTimeout(r, 25)); + } + throw new Error(`waitFor: ${last} within ${timeoutMs}ms`); +} + +describe("loss-free trial (dee-i1wc): the milestone-6 durability proof", () => { + let server: LyncSyncServer | undefined; + let clients: Client[] = []; + + afterEach(async () => { + for (const c of clients) c.store.close(); + clients = []; + await server?.close(); + server = undefined; + }); + + it("runs the full gauntlet (disconnect, storage failure, restart) and loses nothing, hiding nothing", async () => { + const dir = await mkdtemp(path.join(os.tmpdir(), "lync-loss-free-")); + const roomFile = path.join(dir, `${ROOT}.lync`); + server = await startLyncServe({ dir, log: () => {} }); + const port = server.port; + const url = `ws://localhost:${port}`; + + const a = makeClient(url, "alice"); + const b = makeClient(url, "bob"); + const c = makeClient(url, "carol"); + clients = [a, b, c]; + + // All three sync the shared root and reach live. + for (const cl of clients) cl.store.syncRoot(ROOT); + // The union of everything every client appended — the loss-free promise set. + const promised = () => new Set([...a.appended, ...b.appended, ...c.appended]); + + // Every promised id is durable in every client's store AND on the relay's + // on-disk .lync file. Polls to absorb replication/persist latency. + const assertConverged = async (label: string, extraTimeout = 8_000) => { + const want = [...promised()]; + await waitFor(async () => { + for (const cl of clients) { + for (const id of want) if ((await cl.store.byId(id)) === null) return false; + } + const onDisk = new Set(idsOf(await readFile(roomFile, "utf8"))); + return want.every((id) => onDisk.has(id)); + }, extraTimeout).catch(async (error) => { + // Loud, never a silent skip: report exactly what is missing where. + const onDisk = new Set(idsOf(await readFile(roomFile, "utf8").catch(() => ""))); + const missingDisk = want.filter((id) => !onDisk.has(id)); + const missingStores: string[] = []; + for (const cl of clients) + for (const id of want) if ((await cl.store.byId(id)) === null) missingStores.push(`${cl.actor}:${id}`); + throw new Error(`${label}: not converged — off disk [${missingDisk}], missing in stores [${missingStores}] (${error})`); + }); + }; + + // Root event first (a real root so children have a parent to attach to). + await a.append(ROOT, [], "the trial begins"); + await assertConverged("seed"); + + // ---- Leg (a): a client disconnects mid-stream and reconnects ---------- + // Bob's socket drops while Alice keeps appending. Bob must catch up on + // reconnect with nothing skipped. + b.dropSocket(); + await waitFor(() => b.store.status().connection !== "online"); + await a.append("a1", [ROOT], "appended while bob is dark"); + await a.append("a2", ["a1"], "and another"); + // Bob's transport auto-reconnects (reconnectMs) and resyncs from its cursor. + await waitFor(() => b.store.status().connection === "online"); + await assertConverged("leg-a disconnect/reconnect"); + expect(await b.store.byId("a1")).not.toBeNull(); + expect(await b.store.byId("a2")).not.toBeNull(); + + // ---- Leg (b): the relay's storage fails mid-run, then recovers -------- + // The .lync file goes read-only. Carol appends x1: the relay fans it out + // to every client (so it is in every store) but the disk write fails and + // is surfaced as `persist-failed`. The event is NOT yet on disk — that is + // the point; leg (c)'s generation reset is what restores it. + await chmod(roomFile, 0o444); + const diskBeforeFail = new Set(idsOf(await readFile(roomFile, "utf8"))); + await c.append("x1", ["a2"], "carol's line, refused by the disk"); + // x1 reaches every client's store despite the failed persist... + await waitFor(async () => { + for (const cl of clients) if ((await cl.store.byId("x1")) === null) return false; + return true; + }); + // ...and the durability failure screamed through the status channel on + // every subscriber (asserting the surfacing, not just the recovery). + await waitFor(() => a.sawFailure("persist-failed") && b.sawFailure("persist-failed") && c.sawFailure("persist-failed")); + expect(a.sawFailure("persist-failed")).toBe(true); + expect(b.sawFailure("persist-failed")).toBe(true); + expect(c.sawFailure("persist-failed")).toBe(true); + expect(diskBeforeFail.has("x1")).toBe(false); // never hit disk + expect(new Set(idsOf(await readFile(roomFile, "utf8"))).has("x1")).toBe(false); + // Storage heals. + await chmod(roomFile, 0o644); + + // ---- Leg (c): the server process restarts — a new generation --------- + // The recovered log (from disk) lacks x1 and its seq sits behind the + // clients' cursors. On restart every client detects the generation change, + // resyncs from 0, and re-pushes its backlog — including x1, which the + // fresh room now persists. A post-restart event also flows end to end. + await server.close(); + server = await startLyncServe({ dir, port, log: () => {} }); + await waitFor(() => clients.every((cl) => cl.store.status().connection === "online")); + // The generation reset was surfaced on every client, not silently applied. + await waitFor(() => clients.every((cl) => cl.sawFailure("generation changed"))); + for (const cl of clients) expect(cl.sawFailure("generation changed")).toBe(true); + + // A brand-new event in the new generation, appended by Alice post-restart. + await a.append("post", ["a2"], "after the restart, still one story"); + + // ---- Overall: nothing lost, nothing hidden -------------------------- + // Every promised event — including x1, the one the dead disk refused — is + // now in every client's store AND on the relay's on-disk .lync file. + await assertConverged("overall (post-restart, disk restored)", 12_000); + const finalDisk = new Set(idsOf(await readFile(roomFile, "utf8"))); + for (const id of [ROOT, "a1", "a2", "x1", "post"]) { + expect(finalDisk.has(id)).toBe(true); + } + // The event the dead disk refused survived to disk via the generation reset. + expect(finalDisk.has("x1")).toBe(true); + // And every client converged on the full set. + for (const cl of clients) { + for (const id of [ROOT, "a1", "a2", "x1", "post"]) { + expect(await cl.store.byId(id)).not.toBeNull(); + } + } + }, 30_000); +}); From f9bb1c4965261d2a43a82b77dab4223651908599 Mon Sep 17 00:00:00 2001 From: deepfates Date: Mon, 13 Jul 2026 01:45:16 -0700 Subject: [PATCH 6/6] relay: converge the on-disk log without a restart (dee-1pfp) A relay whose disk write failed kept the line in memory and broadcast it, but a same-generation re-push was a byId duplicate no-op that never retried the write. The durable log stayed silently incomplete until a restart rebuilt byId from disk -- undercutting the format's thesis that the saved log is the truth. Now each room tracks its unpersisted lines (ordered by id = append order). persistPending() drains them, in order, as one serialized unit before the next append to the room, and a same-line re-push retries the write instead of no-oping. A successful flush clears the line; a still-dead disk stops the drain (later lines stay pending, never reordered) and re-surfaces persist-failed. The on-disk log heals on its own, no restart required. Proof: test/cli/loss-free-trial.heal-without-restart.integration.test.ts -- disk fails for X (X reaches all clients, off disk, persist-failed surfaces), disk heals, another client appends Y to the same root with NO restart, and both X and Y land on disk in append order and in every client's store. The retry is load-bearing: neuter the pending flush and the leg fails. --- src/relay/relay.ts | 73 +++++++- ...l.heal-without-restart.integration.test.ts | 169 ++++++++++++++++++ 2 files changed, 237 insertions(+), 5 deletions(-) create mode 100644 test/cli/loss-free-trial.heal-without-restart.integration.test.ts diff --git a/src/relay/relay.ts b/src/relay/relay.ts index 8c08861..b683d65 100644 --- a/src/relay/relay.ts +++ b/src/relay/relay.ts @@ -99,6 +99,14 @@ interface Room { seq: number; lines: string[]; byId: Map; + /** + * Lines accepted into memory (byId/lines) and broadcast, but NOT yet on + * disk because an append failed. Keyed by id, insertion order = append + * order — the relay's durable log must converge on this, in order, with no + * restart. Drained before the next append to the room and on a same-line + * re-push; a line clears only once its bytes reach disk. + */ + unpersisted: Map; subscribers: Set; writeChain: Promise; recoveryNote?: string; @@ -184,7 +192,20 @@ export function createLyncRelay(options: LyncRelayOptions): LyncRelay { } const existing = room.byId.get(id); if (existing !== undefined) { - if (existing === frame.line) return; // duplicate: a no-op by union + if (existing === frame.line) { + // Duplicate under union — normally a pure no-op. But if this line + // is still not on disk (a prior append failed), the re-push is our + // chance to heal without a restart: retry it (and any earlier + // pending line, in order). Nothing fails invisibly — a still-dead + // disk re-surfaces persist-failed. + if (room.unpersisted.has(id)) { + const { failedId } = await persistPending(room); + if (failedId !== undefined) { + broadcast(room, { t: "err", root: room.root, reason: "persist-failed", detail: failedId }); + } + } + return; + } const kept = await appendSerialized(room, join(options.dir, `${room.root}.conflicts`), frame.line); broadcast(room, { t: "err", root: room.root, reason: "same-id-conflict", detail: id }, socket); send(socket, { t: "err", root: room.root, reason: "same-id-conflict", detail: id }); @@ -200,12 +221,17 @@ export function createLyncRelay(options: LyncRelayOptions): LyncRelay { room.lines.push(frame.line); room.seq += 1; const seq = room.seq; - const persisted = await appendSerialized(room, join(options.dir, `${room.root}.lync`), frame.line); + // Before writing this line, first drain any earlier lines that failed + // to persist — the on-disk log converges here, in order, with no + // restart. If an earlier line is still unwritable the disk write for + // this one is deferred too (it must not jump ahead), and this line + // joins the pending set to be retried on the next activity or re-push. + const { failedId } = await persistPending(room, { id, line: frame.line }); // Live delivery is the relay's primary job: fan out even if the disk // write failed. A durability failure is surfaced loudly, never hidden. broadcast(room, { t: "ev", root: room.root, seq, line: frame.line, gen: room.generation }); - if (!persisted.ok) { - broadcast(room, { t: "err", root: room.root, reason: "persist-failed", detail: id }); + if (failedId !== undefined) { + broadcast(room, { t: "err", root: room.root, reason: "persist-failed", detail: failedId }); } return; } @@ -237,7 +263,7 @@ export function createLyncRelay(options: LyncRelayOptions): LyncRelay { } async function recoverRoom(root: string): Promise { - const room: Room = { root, generation: randomUUID(), seq: 0, lines: [], byId: new Map(), subscribers: new Set(), writeChain: Promise.resolve() }; + const room: Room = { root, generation: randomUUID(), seq: 0, lines: [], byId: new Map(), unpersisted: new Map(), subscribers: new Set(), writeChain: Promise.resolve() }; const path = join(options.dir, `${root}.lync`); if (!existsSync(path)) return room; const text = await readFile(path, "utf8"); @@ -281,6 +307,43 @@ export function createLyncRelay(options: LyncRelayOptions): LyncRelay { return attempt; } + // Converge the room's on-disk log with no restart. As ONE serialized unit + // on the room's writeChain, write every currently-unpersisted line in append + // order, then `tail` (a freshly accepted line, if any). Each line that lands + // clears from `unpersisted`; the first failure stops the run and leaves that + // line and every LATER one pending, in order — a later line is never written + // ahead of an earlier one for the same room. Lines already on disk are never + // re-written (only the pending set and the new tail are touched), and each + // append writes one whole `line\n`, so a partial failure never corrupts the + // file. Returns the id of the first line that still could not persist, or + // undefined if everything (including tail) reached disk. + function persistPending(room: Room, tail?: { id: string; line: string }): Promise<{ failedId?: string }> { + const path = join(options.dir, `${room.root}.lync`); + const attempt = room.writeChain.catch(() => undefined).then(async (): Promise<{ failedId?: string }> => { + const queue: Array<[string, string]> = [...room.unpersisted]; + if (tail) queue.push([tail.id, tail.line]); + for (let index = 0; index < queue.length; index += 1) { + const [id, line] = queue[index]; + try { + await appendFile(path, `${line}\n`); + room.unpersisted.delete(id); + } catch (error) { + log(`[lync relay] persist failed for ${path}: ${String(error)}`); + // This line and every later one stay pending, in append order, so + // the next activity retries from here without reordering the log. + for (let rest = index; rest < queue.length; rest += 1) { + const [pendingId, pendingLine] = queue[rest]; + if (!room.unpersisted.has(pendingId)) room.unpersisted.set(pendingId, pendingLine); + } + return { failedId: id }; + } + } + return {}; + }); + room.writeChain = attempt.then(() => undefined, () => undefined); + return attempt; + } + function broadcast(room: Room, frame: SyncFrame, except?: LyncRelaySocket): void { const encoded = encodeFrame(frame); for (const subscriber of room.subscribers) { diff --git a/test/cli/loss-free-trial.heal-without-restart.integration.test.ts b/test/cli/loss-free-trial.heal-without-restart.integration.test.ts new file mode 100644 index 0000000..f762fb3 --- /dev/null +++ b/test/cli/loss-free-trial.heal-without-restart.integration.test.ts @@ -0,0 +1,169 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { chmod, mkdtemp, readFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { createMemoryEventStore } from "@deepfates/lync/memory-log"; +import { createSyncedStore, createWebSocketTransport, type SyncStatus } from "@deepfates/lync/synced-store"; +import { startLyncServe, type LyncSyncServer } from "@deepfates/lync/relay"; + +/** + * Heal-without-restart (dee-1pfp) — the last known durability hole in the relay. + * + * The sibling of the loss-free trial. That trial proves a disk failure heals + * across a server RESTART (the generation reset re-pushes the lost line). This + * proves the relay's on-disk log converges ON ITS OWN, with NO restart: + * + * 1. The relay's `.lync` goes read-only. Alice appends X: it fans out + * to every client (in every store) but the disk write fails and + * `persist-failed` surfaces. X is NOT on disk — the relay holds it as + * pending-unpersisted, in memory, broadcast, but off the durable log. + * 2. The disk heals. NO server restart, NO reconnect, NO new generation. + * 3. Bob appends Y to the SAME root. That next activity first flushes the + * pending X (now that the disk is writable), in append order, then Y. + * + * Assert: BOTH X and Y are on the relay's on-disk .lync file, and BOTH are in + * every client's store. The previously-refused X reached disk via the + * next-activity flush alone — the retry is load-bearing (neuter it and X never + * lands without a restart, and this test fails). + */ + +const ROOT = "heal"; + +function idsOf(text: string): string[] { + return text + .split("\n") + .filter((line) => line.length > 0) + .map((line) => { + try { + return (JSON.parse(line) as { id?: string }).id ?? ""; + } catch { + return ""; + } + }); +} + +function makeClient(url: string, actor: string) { + const inner = createMemoryEventStore(); + const statuses: SyncStatus[] = []; + const transport = createWebSocketTransport(url, { reconnectMs: 30 }); + const store = createSyncedStore(inner, transport, { onStatus: (s) => statuses.push(s) }); + const appended = new Set(); + const append = async (id: string, parents: string[], text: string) => { + const result = await store.append({ + v: 1, + id, + kind: "lync/artifact", + at: "2026-07-08T21:00:00Z", + author: { actor }, + parents, + payload: { text }, + }); + if (result.status === "added") appended.add(id); + return result; + }; + const sawFailure = (needle: string) => + statuses.some((s) => s.failures.some((f) => f.includes(needle))) || + store.status().failures.some((f) => f.includes(needle)); + return { actor, store, statuses, appended, append, sawFailure }; +} +type Client = ReturnType; + +async function waitFor(check: () => Promise | boolean, timeoutMs = 8_000): Promise { + const deadline = Date.now() + timeoutMs; + let last = "condition not met"; + while (Date.now() < deadline) { + try { + if (await check()) return; + } catch (error) { + last = String(error); + } + await new Promise((r) => setTimeout(r, 25)); + } + throw new Error(`waitFor: ${last} within ${timeoutMs}ms`); +} + +describe("heal-without-restart (dee-1pfp): the relay's on-disk log converges with no restart", () => { + let server: LyncSyncServer | undefined; + let clients: Client[] = []; + + afterEach(async () => { + for (const c of clients) c.store.close(); + clients = []; + await server?.close(); + server = undefined; + }); + + it("a transiently-failed line reaches disk on the next activity, no restart", async () => { + const dir = await mkdtemp(path.join(os.tmpdir(), "lync-heal-")); + const roomFile = path.join(dir, `${ROOT}.lync`); + server = await startLyncServe({ dir, log: () => {} }); + const startedPort = server.port; + const url = `ws://localhost:${startedPort}`; + + const a = makeClient(url, "alice"); + const b = makeClient(url, "bob"); + clients = [a, b]; + for (const cl of clients) cl.store.syncRoot(ROOT); + + // Seed a real root, and prove it is durable everywhere before we break the + // disk (so the failure below is the ONLY thing off disk). + await a.append(ROOT, [], "the story begins"); + await waitFor(async () => { + for (const cl of clients) if ((await cl.store.byId(ROOT)) === null) return false; + return new Set(idsOf(await readFile(roomFile, "utf8"))).has(ROOT); + }); + + // ---- Storage fails: Alice's X reaches every client but not disk -------- + await chmod(roomFile, 0o444); + await a.append("X", [ROOT], "alice's line, refused by the disk"); + // X is in every client's store despite the failed persist... + await waitFor(async () => { + for (const cl of clients) if ((await cl.store.byId("X")) === null) return false; + return true; + }); + // ...the durability failure screamed on every subscriber... + await waitFor(() => a.sawFailure("persist-failed") && b.sawFailure("persist-failed")); + expect(a.sawFailure("persist-failed")).toBe(true); + expect(b.sawFailure("persist-failed")).toBe(true); + // ...and X genuinely never hit disk. + expect(new Set(idsOf(await readFile(roomFile, "utf8"))).has("X")).toBe(false); + + // ---- The disk heals. NO restart, NO reconnect, NO new generation. ----- + await chmod(roomFile, 0o644); + + // ---- Next activity on the SAME root, SAME running server -------------- + // Bob appends Y. The relay flushes the pending X first (append order), then + // Y. Both must land on disk without any restart having happened. + await b.append("Y", ["X"], "bob's line, after the disk healed"); + + // The heart of the proof: BOTH X (the refused line) and Y are now on the + // relay's on-disk .lync file. + await waitFor(async () => { + const onDisk = new Set(idsOf(await readFile(roomFile, "utf8"))); + return onDisk.has("X") && onDisk.has("Y") && onDisk.has(ROOT); + }, 8_000).catch(async (error) => { + const onDisk = [...new Set(idsOf(await readFile(roomFile, "utf8").catch(() => "")))]; + throw new Error(`X did not heal to disk without a restart — on disk: [${onDisk}] (${error})`); + }); + const finalDisk = new Set(idsOf(await readFile(roomFile, "utf8"))); + expect(finalDisk.has("X")).toBe(true); + expect(finalDisk.has("Y")).toBe(true); + expect(finalDisk.has(ROOT)).toBe(true); + + // Strict on-disk append order: the earlier line X is never written after Y. + const order = idsOf(await readFile(roomFile, "utf8")); + expect(order.indexOf("X")).toBeLessThan(order.indexOf("Y")); + + // Both lines are in every client's store too. + for (const cl of clients) { + expect(await cl.store.byId("X")).not.toBeNull(); + expect(await cl.store.byId("Y")).not.toBeNull(); + } + + // The server was never restarted: same instance, same port throughout. + expect(server.port).toBe(startedPort); + // No generation change was ever surfaced (a restart would have minted one). + expect(a.sawFailure("generation changed")).toBe(false); + expect(b.sawFailure("generation changed")).toBe(false); + }, 30_000); +});