From f7079d56281b4408f7b963b80c956ccde940d0a4 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Mon, 31 Aug 2026 02:04:17 +0200 Subject: [PATCH] feat(chord): add JSON delta tracking --- packages/chord/PLANNING.md | 2 +- packages/chord/README.md | 24 + packages/chord/package.json | 5 + packages/chord/src/delta/index.ts | 1080 +++++++++++++++++++++++++++++ packages/chord/test/delta.test.ts | 1045 ++++++++++++++++++++++++++++ 5 files changed, 2155 insertions(+), 1 deletion(-) create mode 100644 packages/chord/src/delta/index.ts create mode 100644 packages/chord/test/delta.test.ts diff --git a/packages/chord/PLANNING.md b/packages/chord/PLANNING.md index 4274f09ac..7c8fb4b05 100644 --- a/packages/chord/PLANNING.md +++ b/packages/chord/PLANNING.md @@ -404,7 +404,7 @@ Explicit non-goals: - automatic unchanged-value suppression; and - high-frequency stream transport. -A later delta primitive requires a concrete feature and a separate snapshot/delta/gap-recovery design. +Chord now exposes a standalone intent-preserving JSON delta primitive for durable output and other explicit consumers. Replicated service state still uses complete snapshots and ordered latest-value updates; adopting deltas there requires a separate snapshot/delta/gap-recovery design. ## 9. Symmetric RPC plumbing diff --git a/packages/chord/README.md b/packages/chord/README.md index 21b89c599..fa1dc5ac7 100644 --- a/packages/chord/README.md +++ b/packages/chord/README.md @@ -36,6 +36,10 @@ The design has a few connected pieces: ordered updates, and become unready on disconnect or replacement until they are rehydrated. +- **Delta tracking** records mutations to plain JSON as compact operations. It + preserves append, front-truncation, and array-splice intent, supports durable + base batches, and validates untrusted operations before applying them. + - **Remote service sources** advertise services available outside a facet host and open bindings for the services its facets require. Bindings carry logical calls and subscriptions through an application-supplied adapter. Chord @@ -57,6 +61,26 @@ the root API. Chord-owned identifiers use the `chord.*` namespace and its reserved service prefix is `$chord.*`. +## Tracking JSON deltas + +Import the standalone delta primitive from `@earendil-works/chord/delta`: + +```ts +import { apply, track } from "@earendil-works/chord/delta"; + +const changes = track({ output: "", count: 0 }); +changes.flush(); // opening base batch +changes.state.output += "done\n"; +changes.state.count += 1; + +const ops = changes.flush(); +const replica = apply({ output: "", count: 0 }, ops); +``` + +The first flush is always a complete base batch. Later flushes contain path-based +changes. String assignments preserve pure appends and rolling-window movement as +append and front-truncate operations; unrelated rewrites fall back to a set. + ## Bundling and loading facets `@earendil-works/chord/bundler` uses esbuild to turn ESM or TypeScript application diff --git a/packages/chord/package.json b/packages/chord/package.json index 8a0287af9..870bd0f11 100644 --- a/packages/chord/package.json +++ b/packages/chord/package.json @@ -16,6 +16,11 @@ "types": "./dist/context/index.d.ts", "import": "./dist/context/index.js" }, + "./delta": { + "source": "./src/delta/index.ts", + "types": "./dist/delta/index.d.ts", + "import": "./dist/delta/index.js" + }, "./bundler": { "source": "./src/bundler.ts", "types": "./dist/bundler.d.ts", diff --git a/packages/chord/src/delta/index.ts b/packages/chord/src/delta/index.ts new file mode 100644 index 000000000..26cd587b4 --- /dev/null +++ b/packages/chord/src/delta/index.ts @@ -0,0 +1,1080 @@ +import type { JsonValue } from "../types.ts"; + +export type { JsonValue } from "../types.ts"; + +// ───────────────────────────────────────────────────────────────────────────── +// chord/delta — intent-recording change tracking over plain JSON. +// +// Depends on nothing else in the harness. Session storage, the runtime and the +// facet host consume it; keep the arrows pointing that way. +// ───────────────────────────────────────────────────────────────────────────── + +export type Seg = string | number; +export type Path = readonly Seg[]; +export type NonEmptyPath = readonly [Seg, ...Seg[]]; + +/** A path inline, or an id assigned by the encoder on second use. */ +export type PathRef

= P | number; + +/** + * Tuples are the form — in memory, on the wire, on disk. + * + * `r` is the ONLY op that replaces a whole value. `s`/`d`/`a`/`t` cannot target + * the root: the type forbids it. `p` may, and only because a tracked value can + * itself be an array — but a `p` that replaces its entire target is normalised to + * `r`/`s` at record time, so a root `p` is always a partial modification. + * + * `Op` knows nothing about the path dictionary. Interning, id references and + * omitted paths live in `WireOp` and exist only between `encode` and `decode`. + */ +export type Op = + | readonly ["r", JsonValue] + | readonly ["s", NonEmptyPath, JsonValue] + | readonly ["d", NonEmptyPath] + | readonly ["a", NonEmptyPath, string] + | readonly ["t", NonEmptyPath, number] + | readonly ["p", Path, number, number, JsonValue[]]; + +/** + * What crosses a boundary. Adds two compressions and nothing else: + * + * ["#", id, path] defines an id, emitted on a path's SECOND use + * a numeric PathRef references a previously defined id + * a shortened tuple reuses the previous op's path; arity disambiguates + * + * ["r", value] carries no path, so it encodes to itself — which is why isBase + * works unchanged on either vocabulary. + */ +export type WireOp = + | readonly ["r", JsonValue] + | readonly ["s", PathRef, JsonValue] + | readonly ["s", JsonValue] + | readonly ["d", PathRef] + | readonly ["d"] + | readonly ["a", PathRef, string] + | readonly ["a", string] + | readonly ["t", PathRef, number] + | readonly ["t", number] + | readonly ["p", PathRef, number, number, JsonValue[]] + | readonly ["p", number, number, JsonValue[]] + | readonly ["#", number, Path]; + +// ─── Classification ────────────────────────────────────────────────────────── + +export const isReplace = (op: Op | WireOp): boolean => op[0] === "r"; + +/** + * A batch begins with a replacement. Flush guarantees `r` is at index 0 or absent, + * so this is exact rather than a heuristic. + */ +export const isBase = (ops: readonly (Op | WireOp)[]): boolean => ops.length > 0 && ops[0]![0] === "r"; + +// ─── Overlap ───────────────────────────────────────────────────────────────── + +/** + * Longest suffix of `a` that is a prefix of `b`. Probes with indexOf and verifies + * exact substring equality, so the hot loops are native. A hand-written KMP is + * asymptotically equivalent and much slower in practice. + * + * Always correct: the returned n satisfies a.slice(a.length - n) === b.slice(0, n). + */ +export function overlap(a: string, b: string, scan: number, probe = 64, maxCandidates = 8): number { + if (a.length === 0 || b.length === 0 || scan === 0) return 0; + const tail = a.length > scan ? a.slice(a.length - scan) : a; + + // A probe of length h can only find overlaps of at least h — the head must + // actually occur in `a`. So try a long head first (few candidates, and it + // catches the large overlaps a rolling window produces), then fall back to one + // character, which finds any overlap at the cost of more candidates. + // + // Candidates are bounded because repetitive output — a build log, or any run of + // one character — makes a long head match at thousands of positions. Giving up + // returns 0, which emits a set: larger, never wrong. + for (const h of [Math.min(probe, b.length), 1]) { + const head = b.slice(0, h); + let tried = 0; + for (let k = tail.indexOf(head); k !== -1; k = tail.indexOf(head, k + 1)) { + if (++tried > maxCandidates) break; + const n = tail.length - k; + if (n <= b.length && tail.slice(k) === b.slice(0, n)) return n; + } + if (h === 1) break; + } + return 0; +} + +// ─── Tracker ───────────────────────────────────────────────────────────────── + +export interface TrackerOptions { + maxOverlapScan?: number; + coalesce?: boolean; +} + +export interface Tracker { + /** + * Mutate this. Plain TS: assignment, push, splice, delete, all of it. + * + * Assigning to it replaces the whole value: emits `["r", next]` and discards + * ops recorded before it, which describe a value that no longer exists. + * + * `tracker.state = tracker.target` is therefore a checkpoint — it forces a base + * batch without changing anything. Discarding the pending ops is correct + * because the proxy mutates the target directly, so the value already carries + * them. + */ + state: T; + /** The untracked backing object. Reads only — writing here emits nothing. */ + readonly target: T; + flush(): Op[]; + /** + * Clear pending ops and make the next flush a base batch, without changing + * the value. + * + * Recovery replays from the last base batch, so a producer that wants "at most + * N batches to replay" calls this every N. The pending ops are already + * reflected in the value — the proxy mutates the target directly — so + * discarding them loses nothing. + */ + rebase(): void; + /** Discard without emitting. */ + discard(): void; + readonly dirty: boolean; +} + +const isObj = (v: unknown): v is object => v !== null && typeof v === "object"; +const cloneJson = (value: T): T => { + if (!isObj(value)) return value; + if (Array.isArray(value)) return value.map((item) => cloneJson(item)) as T; + const result = Object.create(Object.getPrototypeOf(value) === null ? null : Object.prototype) as Record< + string, + JsonValue + >; + for (const [key, child] of Object.entries(value)) { + Object.defineProperty(result, key, { + value: cloneJson(child), + writable: true, + enumerable: true, + configurable: true, + }); + } + return result as T; +}; +const INDEX = /^(?:0|[1-9]\d*)$/; +const norm = (t: object, k: string | symbol): Seg | symbol => + typeof k === "symbol" ? k : Array.isArray(t) && INDEX.test(k) ? Number(k) : k; +const toIntegerOrInfinity = (value: unknown): number => { + const number = Number(value); + if (Number.isNaN(number) || number === 0) return 0; + return Number.isFinite(number) ? Math.trunc(number) : number; +}; +const MUTATORS = new Set(["push", "pop", "shift", "unshift", "splice", "sort", "reverse", "fill", "copyWithin"]); +const TRACKED_PROXIES = new WeakSet(); + +export function track(root: T, options: TrackerOptions = {}): Tracker { + validateJsonValue(root); + const scan = options.maxOverlapScan ?? 65_536; + const doCoalesce = options.coalesce ?? true; + + let ops: Op[] = []; + + const emit = (op: Op) => { + ops.push(op); + }; + + /** + * Reject reserved keys at record time, not just on apply. + * + * `JSON.parse('{"__proto__":{}}')` yields an OWN property, so a producer can + * hold state containing one without doing anything unusual. The tracker would + * then emit a path the applier refuses — state that cannot be replicated. Fail + * where the mistake is, at the write. + */ + const guard = (seg: Seg | symbol): Seg => { + if (typeof seg === "symbol") throw new UnsafePathError(String(seg)); + if (typeof seg === "string" && RESERVED_SEGMENTS.has(seg)) throw new UnsafePathError(seg); + return seg; + }; + + /** + * A splice covering the whole target is a replacement written the long way. + * Only the interception knows the pre-splice length, so this is a record-time + * rewrite, not something flush can recover. + */ + const spliceOrReplace = (path: Path, before: number, index: number, remove: number, items: JsonValue[]) => { + if (index === 0 && remove === before) { + if (path.length === 0) emit(["r", items]); + else emit(["s", path as NonEmptyPath, items]); + return; + } + emit(["p", [...path], index, remove, items]); + }; + + const assertNoTrackedProxy = (value: unknown, seen = new Set()): void => { + if (!isObj(value) || seen.has(value)) return; + if (TRACKED_PROXIES.has(value)) throw new TypeError("tracked proxies cannot be assigned as values"); + seen.add(value); + for (const child of Array.isArray(value) ? value : Object.values(value)) assertNoTrackedProxy(child, seen); + }; + + const wrap = (obj: V, path: Path): V => { + const childProxies = new Map(); + const proxy = new Proxy(obj, { + get(t, k, r) { + if (Array.isArray(t) && typeof k === "string" && MUTATORS.has(k)) { + return (...args: unknown[]) => { + const before = t.length; + const cloneItems = (values: readonly unknown[]): JsonValue[] => { + for (const value of values) { + assertNoTrackedProxy(value); + assertJsonValue(value); + } + return values.map((value) => cloneJson(value as JsonValue)); + }; + let result: unknown; + switch (k) { + case "push": { + const items = cloneItems(args); + result = Reflect.apply(Array.prototype.push, t, items); + spliceOrReplace(path, before, before, 0, cloneJson(items)); + break; + } + case "unshift": { + const items = cloneItems(args); + result = Reflect.apply(Array.prototype.unshift, t, items); + spliceOrReplace(path, before, 0, 0, cloneJson(items)); + break; + } + case "pop": + result = Reflect.apply(Array.prototype.pop, t, args); + if (before > 0) spliceOrReplace(path, before, before - 1, 1, []); + break; + case "shift": + result = Reflect.apply(Array.prototype.shift, t, args); + if (before > 0) spliceOrReplace(path, before, 0, 1, []); + break; + case "splice": { + const items = cloneItems(args.slice(2)); + const rawStart = args.length === 0 ? 0 : toIntegerOrInfinity(args[0]); + const index = rawStart < 0 ? Math.max(0, before + rawStart) : Math.min(rawStart, before); + const remove = + args.length === 0 + ? 0 + : args.length === 1 + ? before - index + : Math.max(0, Math.min(toIntegerOrInfinity(args[1]), before - index)); + result = Reflect.apply(Array.prototype.splice, t, [index, remove, ...items]); + spliceOrReplace(path, before, index, remove, cloneJson(items)); + break; + } + // sort/reverse/fill/copyWithin permute wholesale; intent is not a splice. + default: { + let methodArgs = args; + let fillValue: JsonValue | undefined; + if (k === "fill") { + assertNoTrackedProxy(args[0]); + assertJsonValue(args[0]); + fillValue = cloneJson(args[0]); + methodArgs = [fillValue, ...args.slice(1)]; + } + result = Reflect.apply(Array.prototype[k as "sort"], t, methodArgs); + if (isObj(fillValue)) { + for (let index = 0; index < t.length; index++) { + if (t[index] === fillValue) t[index] = cloneJson(fillValue as JsonValue); + } + } else if (k === "copyWithin") { + for (let index = 0; index < t.length; index++) { + if (isObj(t[index])) t[index] = cloneJson(t[index] as JsonValue); + } + } + assertJsonValue(t); + if (path.length === 0) emit(["r", cloneJson(t as JsonValue)]); + else emit(["s", path as NonEmptyPath, cloneJson(t as JsonValue)]); + } + } + if (k === "pop") childProxies.delete(String(before - 1)); + else if (k !== "push") childProxies.clear(); + return k === "sort" || k === "reverse" || k === "fill" || k === "copyWithin" ? proxy : result; + }; + } + const value = Reflect.get(t, k, r); + if (!isObj(value)) return value; + const cached = childProxies.get(k); + if (cached?.target === value) return cached.proxy; + const child = wrap(value, [...path, guard(norm(t, k))]); + childProxies.set(k, { target: value, proxy: child }); + return child; + }, + + set(t, k, v, r) { + // `length` is a real mutation but not a document location. + if (Array.isArray(t) && k === "length") { + const before = t.length; + const next = Number(v); + if (!Number.isSafeInteger(next) || next < 0 || next > 4_294_967_295) return Reflect.set(t, k, v, r); + if (next < before) { + Reflect.set(t, k, next, r); + childProxies.clear(); + spliceOrReplace(path, before, next, before - next, []); + } else if (next > before) { + const items = Array(next - before).fill(null); + t.length = next; + t.fill(null, before); + emit(["p", [...path], before, 0, items]); + } + return true; + } + + const segment = guard(norm(t, k)); + if (Array.isArray(t)) { + if (typeof segment !== "number") throw new UnsafePathError(segment); + if (segment > t.length) throw new UnsafePathError(segment); + } + const at = [...path, segment] as unknown as NonEmptyPath; + + if (v === undefined) { + // JSON has no undefined, and deleting an array element would create a hole. + if (Array.isArray(t)) throw new TypeError("undefined would create a sparse array; use splice instead"); + emit(["d", at]); + childProxies.delete(k); + return Reflect.deleteProperty(t, k); + } + + const prev = (t as Record)[k]; + if (typeof prev === "string" && typeof v === "string") { + if (prev === v) return true; + if (v.length > prev.length && v.startsWith(prev)) { + emit(["a", at, v.slice(prev.length)]); // fast path, no probe + } else { + const ov = overlap(prev, v, scan); + if (ov === 0) emit(["s", at, v]); + else { + emit(["t", at, prev.length - ov]); + if (v.length > ov) emit(["a", at, v.slice(ov)]); + } + } + } else { + assertNoTrackedProxy(v); + assertJsonValue(v); + const stored = cloneJson(v); + emit(["s", at, cloneJson(stored)]); + childProxies.delete(k); + return Reflect.set(t, k, stored, r); + } + childProxies.delete(k); + return Reflect.set(t, k, v, r); + }, + + deleteProperty(t, k) { + const segment = guard(norm(t, k)); + if (Array.isArray(t)) { + if (typeof segment !== "number") throw new UnsafePathError(segment); + throw new TypeError("delete would create a sparse array; use splice instead"); + } + emit(["d", [...path, segment] as unknown as NonEmptyPath]); + childProxies.delete(k); + return Reflect.deleteProperty(t, k); + }, + }); + + TRACKED_PROXIES.add(proxy); + return proxy as V; + }; + + let state = wrap(root, []); + // The first flush is always a base batch. A consumer starts with nothing, so + // a stream that opens with deltas has nothing to apply them to — and that + // fails at runtime, in the consumer, rather than where the mistake is. + let forceBase = true; + + return { + get state() { + return state; + }, + // A setter, so the obvious thing works. Without it `tracker.state = next` + // silently swaps the proxy for a plain object and stops tracking. + set state(next: T) { + if (next === state) { + ops = []; + forceBase = true; + return; + } + if (TRACKED_PROXIES.has(next)) throw new TypeError("tracker state cannot be replaced with a tracked proxy"); + validateJsonValue(next); + ops = []; + root = next; + state = wrap(root, []); + forceBase = true; + }, + get target() { + return root; + }, + + // Same effect as assigning the current value back, minus the re-wrap: the + // value has not changed, so the proxy cache is still valid. + rebase() { + ops = []; + forceBase = true; + }, + + get dirty() { + return forceBase || ops.length > 0; + }, + discard() { + ops = []; + }, + flush() { + if (forceBase) { + forceBase = false; + ops = []; + return [["r", cloneJson(root as unknown as JsonValue)]]; + } + if (ops.length === 0) return []; + let out = ops; + ops = []; + + // Everything before a replacement is dead. + let lastReplace = -1; + for (let index = out.length - 1; index >= 0; index--) { + if (!isReplace(out[index]!)) continue; + lastReplace = index; + break; + } + if (lastReplace > 0) out = out.slice(lastReplace); + + out = dropDead(out); + return doCoalesce ? coalesce(out) : out; + }, + }; +} + +const samePath = (a: Op, b: Op): boolean => { + const left = pathOf(a); + const right = pathOf(b); + if (left === undefined || right === undefined || left.length !== right.length) return false; + for (let index = 0; index < left.length; index++) { + if (left[index] !== right[index]) return false; + } + return true; +}; + +function pathOf(op: Op): Path | undefined { + return op[0] === "r" ? undefined : op[1]; +} + +/** + * Drop ops that a later op makes unreachable. + * + * Ops apply in order, so an op is dead if a later one overwrites the whole + * subtree it lives in. Only `s`, `d` and `r` do that — `a`, `t` and `p` modify + * what is already there, so they never dominate. + * + * O(n * d): walk backwards keeping a map of overwritten paths, and for each op + * test its own prefixes against that map. `d` is the path depth, which is small + * and bounded. The obvious formulation — compare each op against every + * dominator — is O(n^2 * d) and degrades on exactly the wide-flush input this + * pass exists to clean up. Measured: 12-14 microseconds per op, flat from 100 to + * 5000 ops, and 10001 redundant writes collapse to 2 emitted ops. + * + * OBJECT KEY ORDER IS NOT PRESERVED. A producer that does + * `set c; set a; delete c; set c` leaves `c` last, because deleting and + * reinserting moves it. The replica receives only the surviving `set c`, applies + * it to a base where `c` never existed, and `c` lands first. Values are + * identical — verified over 1823 randomised sequences with zero value + * mismatches — but the serialised forms can differ. + * + * This is fine because nothing compares a replica against its producer: the + * consumer holds the state and nobody diffs it back. The one visible case is a + * consumer that iterates keys for display, where two views could order rows + * differently. It requires a delete-and-recreate of the same key within one + * flush. Do not hash or content-address a replicated value. + */ +function dropDead(ops: Op[]): Op[] { + type Overwrite = { kind: "s" | "d"; path: Path }; + const overwritten = new Map(); + const kept: Op[] = []; + const rootKey = JSON.stringify([]); + + for (let i = ops.length - 1; i >= 0; i--) { + const op = ops[i]!; + if (op[0] === "r") { + kept.push(op); + overwritten.clear(); + overwritten.set(rootKey, { kind: "s", path: [] }); + continue; + } + + const path = opPath(op); + if (path === undefined) { + kept.push(op); + continue; + } + + let dead = overwritten.has(rootKey); + for (let length = 1; length <= path.length && !dead; length++) { + const overwrite = overwritten.get(JSON.stringify(path.slice(0, length))); + if (overwrite === undefined) continue; + // A `d` at exactly a later `s`'s path is NOT dead. Dropping it leaves the + // key in its original slot, while the producer's delete-then-set moved it + // to the end: same value, different serialisation. + if (length === path.length && op[0] === "d" && overwrite.kind === "s") break; + dead = true; + } + if (dead) continue; + + kept.push(op); + if (op[0] === "p") { + // A splice can reindex every descendant, so path equality below it cannot + // prove that a later overwrite dominates an earlier operation. + for (const [key, overwrite] of overwritten) { + if ( + overwrite.path.length > path.length && + path.every((segment, index) => overwrite.path[index] === segment) + ) { + overwritten.delete(key); + } + } + } else if (op[0] === "s" || op[0] === "d") { + overwritten.set(JSON.stringify(path), { kind: op[0], path }); + } + } + return kept.reverse(); +} + +/** Total, because `Op` paths are always inline. */ +function opPath(op: Op): Path | undefined { + return op[0] === "r" ? undefined : op[1]; +} + +/** Merge adjacent ops describing the same path. Lossless. */ +function coalesce(ops: Op[]): Op[] { + const out: Op[] = []; + for (const op of ops) { + const last = out[out.length - 1]; + if (last !== undefined && samePath(last, op)) { + if (last[0] === "a" && op[0] === "a") { + out[out.length - 1] = ["a", last[1], `${last[2]}${op[2]}`]; + continue; + } + if (last[0] === "s" && op[0] === "s") { + // later write wins + out[out.length - 1] = op; + continue; + } + if (last[0] === "s" && op[0] === "a" && typeof last[2] === "string") { + out[out.length - 1] = ["s", last[1], `${last[2]}${op[2]}`]; + continue; + } + } + out.push(op); + } + return out; +} + +// ─── Path safety ───────────────────────────────────────────────────────────── + +/** + * Segments that reach the prototype chain. + * + * `JSON.parse` is safe on its own — it makes `__proto__` an own property. What is + * not safe is `parent[key] = value`, which is exactly what an applier does, and + * paths are data: `["s", ["__proto__", "isAdmin"], true]` pollutes + * `Object.prototype` for the whole process. + * + * Ops arrive from a facet, a plugin compartment, or a tool whose details may echo + * model output, so none of it is trusted input. + */ +export const RESERVED_SEGMENTS: ReadonlySet = new Set(["__proto__", "constructor", "prototype"]); + +export class UnsafePathError extends Error { + // Not a parameter property: Node's --experimental-strip-types rejects those, + // and these files are meant to run under it directly. + readonly segment: Seg; + constructor(segment: Seg) { + super(`unsafe path segment: ${String(segment)}`); + this.segment = segment; + this.name = "UnsafePathError"; + } +} + +/** + * Rejects anything JSON.stringify would not round-trip and object aliases that + * would lose identity when serialized. Tracked state is a JSON tree, not an + * arbitrary object graph. + */ +export function assertJsonValue(value: unknown): asserts value is JsonValue { + const visit = (candidate: unknown, seen: Set): void => { + if (candidate === null) return; + const type = typeof candidate; + if (type === "string" || type === "boolean") return; + if (type === "number") { + if (!Number.isFinite(candidate as number)) throw new TypeError("non-finite number is not JSON"); + return; + } + if (type !== "object") throw new TypeError(`${type} is not a JsonValue`); + if (seen.has(candidate)) throw new TypeError("shared or cyclic object is not JSON state"); + seen.add(candidate); + if (Array.isArray(candidate)) { + for (const item of candidate) visit(item, seen); + } else { + const prototype = Object.getPrototypeOf(candidate); + if (prototype !== Object.prototype && prototype !== null) { + throw new TypeError("non-plain object is not a JsonValue"); + } + if (Object.getOwnPropertySymbols(candidate).length > 0) { + throw new TypeError("symbol keys are not JSON"); + } + for (const item of Object.values(candidate as object)) visit(item, seen); + } + }; + visit(value, new Set()); +} + +function validateJsonValue(value: unknown): void { + assertJsonValue(value); +} + +/** + * Verb, arity and payload shape for a **decoded** op: paths inline, no `#`, no + * short forms. `apply` uses this. + * + * Validating `Op` against the wire grammar would be laxer than the type: a + * two-element `["s", value]` would pass, and `apply` would then read the value as + * a path. Each vocabulary gets the validator that matches it. + */ +export function assertValidOp(op: unknown): asserts op is Op { + if (!Array.isArray(op) || op.length === 0) throw new TypeError("op is not a tuple"); + switch (op[0]) { + case "r": + if (op.length !== 2) throw new TypeError("r arity"); + assertJsonValue(op[1]); + return; + case "s": + if (op.length !== 3) throw new TypeError("s arity"); + assertPathArg(op[1], true); + assertJsonValue(op[2]); + return; + case "d": + if (op.length !== 2) throw new TypeError("d arity"); + assertPathArg(op[1], true); + return; + case "a": + if (op.length !== 3 || typeof op[2] !== "string") throw new TypeError("a shape"); + assertPathArg(op[1], true); + return; + case "t": + if (op.length !== 3 || !Number.isInteger(op[2]) || op[2] < 0) throw new TypeError("t shape"); + assertPathArg(op[1], true); + return; + case "p": { + if (op.length !== 5) throw new TypeError("p arity"); + assertPathArg(op[1]); + if (!Number.isInteger(op[2]) || op[2] < 0) throw new TypeError("p index"); + if (!Number.isInteger(op[3]) || op[3] < 0) throw new TypeError("p remove"); + if (!Array.isArray(op[4])) throw new TypeError("p items"); + for (const v of op[4]) assertJsonValue(v); + return; + } + // Silently skipping an unknown verb is how a newer producer's op vanishes. + default: + throw new TypeError(`unknown op verb: ${String(op[0])}`); + } +} + +function assertPathArg(p: unknown, nonEmpty = false): void { + if (!Array.isArray(p)) throw new TypeError("path is not an array"); + if (nonEmpty && p.length === 0) throw new TypeError("path is empty"); + assertSafePath(p as Path); +} + +/** The same, for the wire grammar: ids and short forms are legal here. */ +export function assertValidWireOp(op: unknown): asserts op is WireOp { + if (!Array.isArray(op) || op.length === 0) throw new TypeError("op is not a tuple"); + const [verb] = op as unknown[]; + const okRef = (r: unknown): void => { + if (typeof r === "number") { + if (!Number.isInteger(r) || r < 0) throw new TypeError("bad path id"); + return; + } + // A string is not a path. Unchecked, `"a".slice(0, -1)` is `""`, so it + // resolves to the ROOT and writes there — a path that is not a path, accepted. + if (!Array.isArray(r)) throw new TypeError("path is not an array"); + assertSafePath(r as Path); + }; + switch (verb) { + case "r": + if (op.length !== 2) throw new TypeError("r arity"); + assertJsonValue(op[1]); + return; + case "s": + if (op.length === 3) { + okRef(op[1]); + assertJsonValue(op[2]); + } else if (op.length === 2) assertJsonValue(op[1]); + else throw new TypeError("s arity"); + return; + case "d": + if (op.length === 2) okRef(op[1]); + else if (op.length !== 1) throw new TypeError("d arity"); + return; + case "a": + if (op.length === 3) { + okRef(op[1]); + if (typeof op[2] !== "string") throw new TypeError("a value"); + } else if (op.length === 2) { + if (typeof op[1] !== "string") throw new TypeError("a value"); + } else throw new TypeError("a arity"); + return; + case "t": + if (op.length === 3) { + okRef(op[1]); + if (!Number.isInteger(op[2]) || (op[2] as number) < 0) throw new TypeError("t count"); + } else if (op.length === 2) { + if (!Number.isInteger(op[1]) || (op[1] as number) < 0) throw new TypeError("t count"); + } else throw new TypeError("t arity"); + return; + case "p": { + const [i, r, items] = op.length === 5 ? [op[2], op[3], op[4]] : op.length === 4 ? [op[1], op[2], op[3]] : []; + if (items === undefined) throw new TypeError("p arity"); + if (op.length === 5) okRef(op[1]); + if (!Number.isInteger(i) || (i as number) < 0) throw new TypeError("p index"); + if (!Number.isInteger(r) || (r as number) < 0) throw new TypeError("p remove"); + if (!Array.isArray(items)) throw new TypeError("p items"); + for (const v of items) assertJsonValue(v); + return; + } + case "#": { + if (op.length !== 3 || !Number.isInteger(op[1]) || (op[1] as number) < 0 || !Array.isArray(op[2])) { + throw new TypeError("# shape"); + } + assertSafePath(op[2] as Path); + return; + } + // Silently skipping an unknown verb is how a newer producer's op vanishes. + default: + throw new TypeError(`unknown op verb: ${String(verb)}`); + } +} + +export function assertSafePath(path: Path): void { + for (const seg of path) { + if (typeof seg === "string") { + if (RESERVED_SEGMENTS.has(seg)) throw new UnsafePathError(seg); + } else if (!Number.isInteger(seg) || seg < 0) { + throw new UnsafePathError(seg); + } + } +} + +/** + * An index may address an existing element or append exactly one past the end. + * + * This is not an arbitrary cap — it is what keeps the value a `JsonValue`. A + * sparse array does not survive a JSON round trip: holes serialise to `null` and + * return as real properties, so `arr[7] = x` on a length-3 array already produces + * state a replica cannot match. Rejecting the write is more honest than silently + * diverging. + * + * It also removes the denial of service it would otherwise permit: + * `["s", ["xs", 4294967290], 1]` allocates a 4.29-billion-entry array from one op. + * Growth stays possible and stays proportional — the tracker already emits + * `arr.length = n` as a splice of explicit nulls, whose op size grows with the + * gap, so a large growth costs a large op rather than a small one. + */ +function assertIndexInRange(parent: readonly unknown[], index: number): void { + if (index > parent.length) throw new UnsafePathError(index); +} + +// ─── Applier ───────────────────────────────────────────────────────────────── + +export class PathError extends Error { + readonly path: Path | number; + constructor(path: Path | number) { + super(`unresolvable path: ${JSON.stringify(path)}`); + this.path = path; + this.name = "PathError"; + } +} + +/** + * Apply ops to a plain mutable value. Returns the value, because `r` replaces it + * outright and cannot be done in place. + * + * Takes decoded ops. Path ids and omitted paths are a wire concern — run + * `decode` first if the ops came from a boundary. + */ +export function apply(target: T | undefined, ops: readonly Op[]): T { + let root = target as unknown as JsonValue; + + for (const op of ops) { + assertValidOp(op); + + if (op[0] === "r") { + // Adopted, not copied. The consumer owns the batch it was handed. + // + // Fanning one batch out to several consumers in-process therefore makes + // their replicas alias each other. That is an ownership rule, not a + // defect: copy the batch at the fan-out point, or let each consumer + // decode its own. A batch that crosses a real boundary is already + // distinct, because serialisation produces fresh objects. + root = op[1]; + continue; + } + + const path = op[1]; + assertSafePath(path); + + if (op[0] === "p") { + const target_ = path.length === 0 ? root : resolve(root, path); + if (!Array.isArray(target_)) throw new PathError(path); + target_.splice(op[2], op[3]); + const chunkSize = 10_000; + for (let offset = 0; offset < op[4].length; offset += chunkSize) { + target_.splice(op[2] + offset, 0, ...op[4].slice(offset, offset + chunkSize)); + } + continue; + } + + // s/d/a/t can never target the root — the type forbids it. + const parent = resolve(root, path.slice(0, -1)) as Record; + const key = path[path.length - 1]!; + if (Array.isArray(parent)) { + if (typeof key !== "number") throw new UnsafePathError(key); + assertIndexInRange(parent, key); + } + // defineProperty rather than assignment: a setter inherited from the prototype + // chain would otherwise run on write. + const write = (value: JsonValue) => { + Object.defineProperty(parent, key, { value, writable: true, enumerable: true, configurable: true }); + }; + const read = (): unknown => (Object.hasOwn(parent, key) ? parent[key] : undefined); + switch (op[0]) { + case "s": + write(op[2]); + break; + case "d": + if (Array.isArray(parent)) { + if (typeof key !== "number" || key >= parent.length) throw new PathError(path); + (parent as unknown as JsonValue[]).splice(key, 1); + } else delete parent[key]; + break; + case "a": { + const current = read(); + if (typeof current !== "string") throw new PathError(path); + write(`${current}${op[2]}`); + break; + } + case "t": { + const current = read(); + if (typeof current !== "string") throw new PathError(path); + write(current.slice(op[2])); + break; + } + } + } + return root as unknown as T; +} + +function resolveValue(root: JsonValue, path: Path): JsonValue { + let node: JsonValue = root; + for (const seg of path) { + if (!isObj(node)) throw new PathError(path); + if (Array.isArray(node) && typeof seg !== "number") throw new UnsafePathError(seg); + // Own properties only: an inherited getter must not run, and a walk must not + // escape the value into the prototype chain. + if (!Object.hasOwn(node, seg as PropertyKey)) throw new PathError(path); + node = (node as Record)[seg]!; + } + return node; +} + +function resolve(root: JsonValue, path: Path): JsonValue { + const node = resolveValue(root, path); + if (!isObj(node)) throw new PathError(path); + return node; +} + +// ─── Codec ─────────────────────────────────────────────────────────────────── +// +// Path interning and arity omission live between the tracker and a boundary; +// `Op` and `apply` know nothing about them. +// +// ONE PAIR PER STREAM. The table spans an entire subscription or file: a path +// interned in batch 3 is referenced in batch 40. A second consumer that +// subscribes at batch 40 has never seen the definition, so it needs its own +// encoder — sharing one across consumers hands the late subscriber ids it cannot +// resolve. A base batch does not rescue it, because `["r", value]` carries no +// path refs and leaves the table empty. + +const pathKey = (path: Path): string => JSON.stringify(path); + +export interface Encoder { + encode(ops: readonly Op[]): WireOp[]; +} + +/** + * Intern on SECOND use. A definition costs more than the path it replaces, so + * interning on first use loses on the many paths written exactly once. + */ +export function encoder(): Encoder { + const seen = new Set(); + const ids = new Map(); + let nextId = 0; + let previous: string | undefined; // last path in THIS batch + + return { + encode(ops) { + // Arity omission is scoped to a batch. Letting it span batches would make + // a batch's first op depend on the previous batch's last one, so a reader + // that skips or reorders a batch decodes into the wrong path. Ids are the + // only cross-batch state, and the dictionary makes those explicit. + previous = undefined; + const out: WireOp[] = []; + for (const op of ops) { + if (op[0] === "r") { + out.push(op); + // A base batch is a RECOVERY POINT: a reader replays from the last one + // with a fresh decoder. So everything after it must be self-contained. + // Keeping ids across a replacement emits references to definitions the + // reader never saw — recovery fails with an unresolvable path id. + seen.clear(); + ids.clear(); + nextId = 0; + previous = undefined; + continue; + } + const path = op[1]; + const key = pathKey(path); + + // Same path as the previous op: drop the ref entirely. + if (key === previous) { + switch (op[0]) { + case "s": + out.push(["s", op[2]]); + break; + case "d": + out.push(["d"]); + break; + case "a": + out.push(["a", op[2]]); + break; + case "t": + out.push(["t", op[2]]); + break; + case "p": + out.push(["p", op[2], op[3], op[4]]); + break; + } + continue; + } + + let ref: PathRef = path; + const existing = ids.get(key); + if (existing !== undefined) { + ref = existing; + } else if (seen.has(key)) { + const id = nextId++; + ids.set(key, id); + out.push(["#", id, path]); // second use: define, then reference + ref = id; + } else { + seen.add(key); // first use: inline + } + + switch (op[0]) { + case "s": + out.push(["s", ref as PathRef, op[2]]); + break; + case "d": + out.push(["d", ref as PathRef]); + break; + case "a": + out.push(["a", ref as PathRef, op[2]]); + break; + case "t": + out.push(["t", ref as PathRef, op[2]]); + break; + case "p": + out.push(["p", ref, op[2], op[3], op[4]]); + break; + } + previous = key; + } + return out; + }, + }; +} + +export interface Decoder { + decode(wire: readonly WireOp[]): Op[]; +} + +export function decoder(): Decoder { + const paths = new Map(); + + return { + decode(wire) { + let previous: Path | undefined; // scoped to the batch, as in encode + const out: Op[] = []; + for (const op of wire) { + assertValidWireOp(op); + if (op[0] === "#") { + assertSafePath(op[2]); + paths.set(op[1], op[2]); + continue; + } + if (op[0] === "r") { + out.push(op); + paths.clear(); + previous = undefined; + continue; + } + + // Arity tells us whether a ref is present: the short forms omit it. + const short = + (op[0] === "d" && op.length === 1) || + (op[0] !== "d" && op[0] !== "p" && op.length === 2) || + (op[0] === "p" && op.length === 4); + + let path: Path; + if (short) { + if (previous === undefined) throw new PathError([]); + path = previous; + } else { + const ref = op[1] as PathRef; + if (typeof ref === "number") { + const resolved = paths.get(ref); + if (resolved === undefined) throw new PathError(ref); + path = resolved; + } else { + path = ref; + } + previous = path; + } + + if (op[0] !== "p" && path.length === 0) throw new PathError(path); + switch (op[0]) { + case "s": + out.push(["s", path as NonEmptyPath, (short ? op[1] : op[2]) as JsonValue]); + break; + case "d": + out.push(["d", path as NonEmptyPath]); + break; + case "a": + out.push(["a", path as NonEmptyPath, (short ? op[1] : op[2]) as string]); + break; + case "t": + out.push(["t", path as NonEmptyPath, (short ? op[1] : op[2]) as number]); + break; + case "p": { + const [i, r, items] = short + ? [op[1] as number, op[2] as number, op[3] as JsonValue[]] + : [op[2] as number, op[3] as number, op[4] as JsonValue[]]; + out.push(["p", path, i, r, items]); + break; + } + } + } + return out; + }, + }; +} diff --git a/packages/chord/test/delta.test.ts b/packages/chord/test/delta.test.ts new file mode 100644 index 000000000..dc58e1aa7 --- /dev/null +++ b/packages/chord/test/delta.test.ts @@ -0,0 +1,1045 @@ +import { describe, expect, it } from "vitest"; +import { + apply, + assertJsonValue, + assertValidOp, + assertValidWireOp, + decoder, + encoder, + isBase, + type JsonValue, + type Op, + overlap, + track, + type WireOp, +} from "../src/delta/index.ts"; + +const PAD = "p".repeat(400); // see "adaptive emission": a delta must be able to win + +describe("overlap", () => { + it("finds an overlap shorter than the long probe", () => { + expect(overlap("abcdefgh", "defghxyz", 65_536)).toBe(5); + }); + + it("honors a disabled scan", () => { + expect(overlap("abcdef", "defghi", 0)).toBe(0); + }); +}); + +describe("tracker: intent", () => { + it("records an append, not a replacement", () => { + const t = track({ s: "", pad: PAD }); + t.flush(); // drain the base batch + t.state.s += "ab"; + t.state.s += "cd"; + const ops = t.flush(); + expect(ops).toEqual([["a", ["s"], "abcd"]]); + expect(apply({ s: "", pad: PAD }, ops)).toEqual({ s: "abcd", pad: PAD }); + }); + + it("recovers truncate+append from a rolling window", () => { + // The case every effect-recording library degrades to a whole-value set. + const t = track({ s: "abcdefgh", pad: PAD }); + t.flush(); // drain the base batch + t.state.s = `${t.state.s.slice(3)}xyz`; + const ops = t.flush(); + expect(ops).toEqual([ + ["t", ["s"], 3], + ["a", ["s"], "xyz"], + ]); + expect(apply({ s: "abcdefgh", pad: PAD }, ops)).toEqual({ s: "defghxyz", pad: PAD }); + }); + + it("caches nested proxies while replacing changed children", () => { + const t = track({ child: { value: 1 } }); + const first = t.state.child; + expect(t.state.child).toBe(first); + t.state.child = { value: 2 }; + expect(t.state.child).not.toBe(first); + expect(t.state.child).toBe(t.state.child); + }); + + it("records array intent as one splice", () => { + const t = track({ xs: [1, 2], pad: PAD }); + t.flush(); // drain the base batch + t.state.xs.push(3); + expect(t.flush()).toEqual([["p", ["xs"], 2, 0, [3]]]); + }); + + it("normalises undefined to a delete", () => { + const t = track({ a: 1, pad: PAD }); + t.flush(); // drain the base batch + (t.state as Record).a = undefined; + expect(t.flush()).toEqual([["d", ["a"]]]); + }); + + it("round-trips interleaved writes", () => { + const initial = { out: "x".repeat(500), total: 0 }; + const t = track(structuredClone(initial)); + t.flush(); + for (let i = 0; i < 1_000; i++) { + t.state.out = `${t.state.out.slice(10)}${String(i).padStart(10, "0")}`; + t.state.total += 10; + } + const ops = t.flush(); + expect(apply(structuredClone(initial), ops)).toEqual(t.target); + }); + + it("invalidates a pending child when its parent is overwritten", () => { + const initial = { a: { x: 1 } as Record }; + const t = track(structuredClone(initial)); + t.flush(); + t.state.a.b = 99; + t.state.a = { c: 2 }; + expect(apply(structuredClone(initial), t.flush())).toEqual(t.target); + }); +}); + +describe("tracker: root ops", () => { + it("splices a value that is itself an array", () => { + const t = track([1, 2, 3, PAD]); + t.flush(); // drain the base batch + t.state.push(4); + const ops = t.flush(); + expect(ops).toEqual([["p", [], 4, 0, [4]]]); + expect(apply([1, 2, 3, PAD], ops)).toEqual([1, 2, 3, PAD, 4]); + }); + + it("normalises a splice covering the whole root to a replacement", () => { + const t = track([1, 2, 3]); + t.flush(); // drain the base batch + t.state.splice(0, 3, 9); + const ops = t.flush(); + expect(ops).toEqual([["r", [9]]]); + expect(isBase(ops)).toBe(true); + }); + + it("normalises a nested splice-all to a set, not a replacement", () => { + const t = track({ xs: [1, 2, 3], pad: PAD }); + t.flush(); // drain the base batch + t.state.xs.splice(0, 3, 9); + expect(t.flush()).toEqual([["s", ["xs"], [9]]]); + }); + + it("matches splice with no arguments", () => { + const initial = { xs: [1, 2, 3] }; + const t = track(structuredClone(initial)); + t.flush(); + (t.state.xs.splice as (...args: never[]) => number[])(); + expect(t.target).toEqual(initial); + expect(apply(structuredClone(initial), t.flush())).toEqual(t.target); + }); + + it("normalises splice arguments to integers", () => { + const initial = { xs: [1, 2, 3] }; + const t = track(structuredClone(initial)); + t.flush(); + t.state.xs.splice(0, undefined as unknown as number, 9); + t.state.xs.splice(1.5, 1); + const ops = t.flush(); + for (const op of ops) { + if (op[0] !== "p") continue; + expect(Number.isInteger(op[2])).toBe(true); + expect(Number.isInteger(op[3])).toBe(true); + } + expect(apply(structuredClone(initial), ops)).toEqual(t.target); + }); + + it("normalises length = 0 on the root", () => { + const t = track([1, 2, 3]); + t.flush(); // drain the base batch + t.state.length = 0; + const ops = t.flush(); + expect(ops).toEqual([["r", []]]); + expect(apply([1, 2, 3], ops)).toEqual([]); + }); + + it("round-trips repeated nested splices", () => { + const initial = { xs: [1] }; + const t = track(structuredClone(initial)); + t.flush(); + for (let i = 2; i <= 100; i++) t.state.xs.push(i); + const ops = t.flush(); + expect(apply(structuredClone(initial), ops)).toEqual(t.target); + }); + + it("does not fold a child path across a parent splice", () => { + const initial = { xs: ["ab", "cd"] }; + const t = track(structuredClone(initial)); + t.flush(); + t.state.xs[0] += "x"; + t.state.xs.shift(); + t.state.xs[0] += "y"; + expect(apply(structuredClone(initial), t.flush())).toEqual(t.target); + }); + + it("drops child ops when repeated parent splices collapse to a snapshot", () => { + const initial = { xs: ["ab"] }; + const t = track(structuredClone(initial)); + t.flush(); + t.state.xs.push("q"); + t.state.xs[0] += "cd"; + t.state.xs.push("z"); + expect(apply(structuredClone(initial), t.flush())).toEqual(t.target); + }); + + it("does not let a post-splice set dominate an earlier element append", () => { + const initial = { xs: ["ab"] }; + const t = track(structuredClone(initial)); + t.flush(); + t.state.xs[0] += "x"; + t.state.xs.unshift("q"); + t.state.xs[0] = "Z"; + expect(apply(structuredClone(initial), t.flush())).toEqual(t.target); + }); + + it("does not discard a nested append through a reindexed set", () => { + const initial = { xs: [{ k: "a" }] as ({ k: string } | number)[] }; + const t = track(structuredClone(initial)); + t.flush(); + (t.state.xs[0] as { k: string }).k += "x"; + t.state.xs.unshift(9); + t.state.xs[0] = 7; + expect(apply(structuredClone(initial), t.flush())).toEqual(t.target); + }); + + it("preserves element operations across middle-array insertions", () => { + const initial = { xs: ["a", "b"] }; + const t = track(structuredClone(initial)); + t.flush(); + t.state.xs[1] += "x"; + t.state.xs.splice(1, 0, "inserted"); + t.state.xs[1] = "changed"; + expect(apply(structuredClone(initial), t.flush())).toEqual(t.target); + }); + + it("round-trips deterministic element mutations across reindexing splices", () => { + let seed = 0x1234abcd; + const next = (): number => { + seed = (seed * 1_664_525 + 1_013_904_223) >>> 0; + return seed; + }; + for (let round = 0; round < 200; round++) { + const initial = { xs: [{ k: "a" }, { k: "b" }, { k: "c" }] }; + const t = track(structuredClone(initial)); + t.flush(); + for (let step = 0; step < 30; step++) { + const index = next() % t.state.xs.length; + switch (next() % 5) { + case 0: + t.state.xs[index]!.k += String.fromCharCode(97 + (next() % 26)); + break; + case 1: + t.state.xs[index] = { k: `set-${round}-${step}` }; + break; + case 2: + t.state.xs.unshift({ k: `head-${round}-${step}` }); + break; + case 3: + if (t.state.xs.length > 1) t.state.xs.shift(); + break; + default: { + const at = next() % (t.state.xs.length + 1); + const remove = t.state.xs.length > 1 && next() % 2 === 0 ? 1 : 0; + t.state.xs.splice(at, remove, { k: `mid-${round}-${step}` }); + } + } + if (t.state.xs.length > 10) t.state.xs.shift(); + } + expect(apply(structuredClone(initial), t.flush())).toEqual(t.target); + } + }); + + it("keeps mutator chaining tracked", () => { + const initial = { xs: [3, 1, 2] }; + const t = track(structuredClone(initial)); + t.flush(); + t.state.xs.sort().push(4); + expect(apply(structuredClone(initial), t.flush())).toEqual(t.target); + }); + + it("detaches objects supplied to fill", () => { + const supplied = { value: 1 }; + const t = track({ xs: [{ value: 0 }, { value: 0 }] }); + t.flush(); + t.state.xs.fill(supplied); + supplied.value = 2; + expect(t.target.xs).toEqual([{ value: 1 }, { value: 1 }]); + expect(t.target.xs[0]).not.toBe(t.target.xs[1]); + expect(apply({ xs: [{ value: 0 }, { value: 0 }] }, t.flush())).toEqual(t.target); + }); + + it("keeps copyWithin results as a JSON tree", () => { + const t = track({ xs: [{ value: 1 }, { value: 2 }] }); + t.flush(); + t.state.xs.copyWithin(1, 0, 1); + expect(t.target.xs).toEqual([{ value: 1 }, { value: 1 }]); + expect(t.target.xs[0]).not.toBe(t.target.xs[1]); + expect(apply({ xs: [{ value: 1 }, { value: 2 }] }, t.flush())).toEqual(t.target); + }); + + it("grows arrays with explicit null values", () => { + const initial = { xs: [1] as (number | null)[] }; + const t = track(structuredClone(initial)); + t.flush(); + t.state.xs.length = 4; + expect(t.target.xs).toEqual([1, null, null, null]); + expect(apply(structuredClone(initial), t.flush())).toEqual(t.target); + }); +}); + +describe("replacing the whole value", () => { + // The proxy cannot observe `tracker.state = next`: `state` belongs to the + // tracker, not to the tracked object, so assigning to it swaps the proxy for a + // plain object and silently stops tracking. Replacement must be a method. + it("assigning to state emits a base batch and keeps tracking", () => { + const t = track>({ p: 1, q: PAD }); + t.flush(); // drain the base batch + t.state.p = 2; + t.state = { r: 9, s: "new" }; + const ops = t.flush(); + expect(ops).toEqual([["r", { r: 9, s: "new" }]]); + expect(isBase(ops)).toBe(true); + expect(apply({}, ops)).toEqual(t.target); + + t.state.r = 10; // the new value is tracked + expect(t.flush()).toEqual([["s", ["r"], 10]]); + }); + + it("assigning the tracked root to state rebases without poisoning the tracker", () => { + const t = track({ value: 1 }); + t.flush(); + const trackedRoot = t.state; + t.state = trackedRoot; + expect(t.flush()).toEqual([["r", { value: 1 }]]); + t.state.value = 2; + expect(t.flush()).toEqual([["s", ["value"], 2]]); + }); + + it("rejects replacing state with a nested tracked proxy", () => { + const t = track<{ nested: Record }>({ nested: { value: 1 } }); + const nested = t.state.nested; + expect(() => { + t.state = nested as unknown as { nested: Record }; + }).toThrow(); + }); + + it("rejects assigning a nested tracked proxy into state", () => { + const t = track<{ + left: Record; + right?: Record; + wrapped?: { inner: Record }; + items?: Record[]; + }>({ left: { value: 1 } }); + t.flush(); + expect(() => { + t.state.right = t.state.left; + }).toThrow(/tracked prox/); + expect(() => { + t.state.wrapped = { inner: t.state.left }; + }).toThrow(/tracked prox/); + t.state.items = []; + expect(() => t.state.items?.push(t.state.left)).toThrow(/tracked prox/); + const other = track({ value: 2 }); + expect(() => { + t.state.wrapped = { inner: other.state as unknown as Record }; + }).toThrow(/tracked prox/); + }); + + it("assigning to state discards ops recorded before it", () => { + // They describe a value that no longer exists. + const t = track>({ p: 1, q: PAD }); + t.flush(); // drain the base batch + t.state.p = 2; + t.state = { z: 1 }; + expect(t.flush()).toEqual([["r", { z: 1 }]]); + }); + + it("a partial rewrite keeps its ops", () => { + const t = track>({ a: "x".repeat(200), b: "y".repeat(200) }); + t.flush(); // drain the base batch + t.state.a = "p"; + const ops = t.flush(); + expect(isBase(ops)).toBe(false); + expect(ops).toEqual([["s", ["a"], "p"]]); + }); + + it("rebase() forces a base batch without changing the value", () => { + // This is the checkpoint. Recovery replays from the last base batch, so a + // producer must be able to bound that rather than depend on the adaptive + // size rule, which compares each flush independently. + const t = track>({ p: 1, pad: PAD }); + t.flush(); // drain the base batch + t.rebase(); + const ops = t.flush(); + expect(isBase(ops)).toBe(true); + expect(apply({}, ops)).toEqual(t.target); + }); + + it("rebase() applies once, not to every later flush", () => { + const t = track>({ p: 1, pad: PAD }); + t.flush(); // drain the base batch + t.rebase(); + t.flush(); + t.state.p = 2; + expect(t.flush()).toEqual([["s", ["p"], 2]]); + }); +}); + +describe("the first flush", () => { + it("is always a base batch", () => { + // A consumer starts with nothing, so a stream opening with deltas has + // nothing to apply them to. Requiring the producer to remember would fail + // at runtime, in the consumer, rather than where the mistake is. + const t = track>({ x: 0 }); + t.state.x = 100; + const ops = t.flush(); + expect(isBase(ops)).toBe(true); + expect(ops).toEqual([["r", { x: 100 }]]); // carries mutations made before it + }); + + it("is followed by deltas", () => { + const t = track>({ x: 0, pad: PAD }); + t.flush(); + t.state.x = 1; + expect(t.flush()).toEqual([["s", ["x"], 1]]); + }); + + it("emits nothing when untouched", () => { + // An unmutated tracker has nothing to say; the base batch is owed to a + // consumer, and there is no consumer until someone flushes. + const t = track>({ x: 0 }); + expect(t.dirty).toBe(true); + expect(t.flush()).toEqual([["r", { x: 0 }]]); + expect(t.dirty).toBe(false); + expect(t.flush()).toEqual([]); + }); + + it("discard clears pending deltas", () => { + const t = track({ x: 0 }); + t.flush(); + t.state.x = 1; + expect(t.dirty).toBe(true); + t.discard(); + expect(t.dirty).toBe(false); + expect(t.flush()).toEqual([]); + }); +}); + +describe("apply and fan-out", () => { + it("adopts an `r` payload rather than copying it", () => { + // The consumer owns the batch it was handed. Copying every base batch + // defensively doubles the memory of the one op carrying the whole value. + // + // The rule that follows: do not hand one batch to two in-process + // consumers. This test pins the aliasing so nobody "fixes" it by adding a + // clone back. + const t = track>({ n: 0 }); + const batch = t.flush(); + const a = apply>(undefined, batch); + const b = apply>(undefined, batch); + a.n = 1; + expect(b.n).toBe(1); + }); + + it("does not alias the producer", () => { + const t = track>({ x: 0 }); + const replica = apply>(undefined, t.flush()); + t.state.x = 999; + t.flush(); + expect(replica.x).toBe(0); + }); + + it("does not preserve aliases assigned to separate paths", () => { + const t = track<{ a?: { value: number }; b?: { value: number } }>({}); + const replica = apply<{ a?: { value: number }; b?: { value: number } }>(undefined, t.flush()); + const shared = { value: 0 }; + t.state.a = shared; + t.state.b = shared; + t.state.a.value = 1; + const next = apply(replica, t.flush()); + expect(next).toEqual(t.target); + expect(t.target.b?.value).toBe(0); + }); + + it("does not alias pushed values with an in-process consumer", () => { + const t = track({ xs: [] as { value: number }[] }); + const replica = apply<{ xs: { value: number }[] }>(undefined, t.flush()); + t.state.xs.push({ value: 1 }); + const next = apply(replica, t.flush()); + next.xs[0]!.value = 2; + expect(t.target.xs[0]!.value).toBe(1); + }); + + it("folds a whole stream without a base-batch branch", () => { + // `apply` handles `r` by replacing, and tolerates an undefined target, + // so a consumer needs no isBase check and no clone of its own. + const t = track({ x: 0, l: [] as string[] }); + const enc = encoder(); + const dec = decoder(); + let replica: unknown; + const send = () => { + replica = apply(replica, dec.decode(enc.encode(t.flush()))); + }; + + t.state.x = 100; + t.state.l.push("xyz"); + send(); + t.state.x = 101; + send(); + expect(replica).toEqual(t.target); + }); +}); + +describe("dead-op elimination", () => { + // Ops apply in order, so an op is dead if a later one overwrites the whole + // subtree it lives in. Only s/d/r dominate; a/t/p modify what is there. + type DeadOpState = { a: Record; x?: number; y?: number }; + const run = (mutate: (state: DeadOpState) => void, init: DeadOpState = { a: { b: 1 }, x: 1, y: 2 }) => { + const t = track(structuredClone(init)); + t.flush(); // drain the base batch + mutate(t.state); + return t.flush(); + }; + + it("collapses repeated writes to one field", () => { + expect( + run((s) => { + s.x = 1; + s.x = 2; + s.x = 3; + }), + ).toEqual([["s", ["x"], 3]]); + }); + + it("drops a write superseded by a later, non-adjacent one", () => { + expect( + run((s) => { + s.x = 10; + s.y = 20; + s.x = 30; + }), + ).toEqual([ + ["s", ["y"], 20], + ["s", ["x"], 30], + ]); + }); + + it("drops a child write when the parent is replaced after it", () => { + expect( + run((s) => { + s.a.b = 99; + s.a = { c: 5 }; + }), + ).toEqual([["s", ["a"], { c: 5 }]]); + }); + + it("keeps a child write that follows the parent replacement", () => { + expect( + run((s) => { + s.a = { b: 1 }; + s.a.b = 7; + }), + ).toEqual([ + ["s", ["a"], { b: 1 }], + ["s", ["a", "b"], 7], + ]); + }); + + it("collapses set-then-delete to the delete", () => { + expect( + run((s) => { + s.x = 5; + delete s.x; + }), + ).toEqual([["d", ["x"]]]); + }); + + it("keeps delete-then-set, to preserve key position", () => { + // Dropping the delete would leave the key in its original slot while the + // producer moved it to the end. Same value, different serialisation. + expect( + run((s) => { + delete s.x; + s.x = 5; + }), + ).toEqual([ + ["d", ["x"]], + ["s", ["x"], 5], + ]); + }); + + it("keeps the delete when a later append folds into the set", () => { + const t = track<{ x?: string; y: number }>({ x: "", y: 1 }); + t.flush(); + delete t.state.x; + t.state.x = "ab"; + t.state.x += "cd"; + expect(t.flush()).toEqual([ + ["d", ["x"]], + ["s", ["x"], "abcd"], + ]); + }); + + it("is linear in the number of ops", () => { + // The naive formulation compares every op against every dominator, which + // is quadratic and degrades on exactly the wide flush this pass cleans up. + const wide = (n: number) => { + const root: Record = {}; + for (let i = 0; i < n; i++) root[`f${i}`] = i; + const t = track(root); + t.flush(); // drain the base batch + const started = performance.now(); + for (let i = 0; i < n; i++) t.state[`f${i}`] = i + 1; + t.flush(); + return performance.now() - started; + }; + wide(200); // warm + const small = Math.max(wide(250), 0.1); + const large = wide(2500); + expect(large / small).toBeLessThan(40); // linear would be ~10x + }); + + it("collapses a pathological redundant producer", () => { + const t = track<{ a: Record; x: number }>({ a: { b: 1 }, x: 1 }); + t.flush(); // drain the base batch + for (let i = 0; i < 2000; i++) { + t.state.x = i; + t.state.a.b = i; + } + t.state.a = { done: true }; + expect(t.flush()).toHaveLength(2); + }); +}); + +describe("flush", () => { + it("drops everything before a replacement", () => { + const t = track([1, 2]); + t.flush(); // drain the base batch + t.state.push(3); + t.state.splice(0, 3, 7); + expect(t.flush()).toEqual([["r", [7]]]); + }); + + it("keeps the prefix when the replacement is nested", () => { + // `s` on a subtree is not a root replacement, so earlier ops stay live. + const t = track({ a: 1, xs: [1, 2, 3], pad: PAD }); + t.flush(); // drain the base batch + t.state.a = 2; + t.state.xs.splice(0, 3, 9); + expect(t.flush()).toEqual([ + ["s", ["a"], 2], + ["s", ["xs"], [9]], + ]); + }); + + it("rejects a constructor walk", () => { + // ({}).constructor.constructor is Function — the classic escape ladder. + expect(() => apply({}, JSON.parse('[["s",["constructor","prototype","gadget"],true]]'))).toThrow(); + expect(({} as Record).gadget).toBeUndefined(); + }); + + it("rejects a forbidden path reached through an interned id", () => { + const wire = [ + ["#", 0, ["__proto__", "w"]], + ["s", 0, true], + ] as unknown as WireOp[]; + expect(() => decoder().decode(wire)).toThrow(); + }); + + it("does not run an inherited setter", () => { + Object.defineProperty(Object.prototype, "trap", { + get() { + throw new Error("inherited getter ran"); + }, + configurable: true, + }); + try { + expect(() => apply({}, [["s", ["trap"], 1]])).not.toThrow(); + } finally { + delete (Object.prototype as Record).trap; + } + }); + + it("allows a reserved name as a VALUE key", () => { + // Reserved as segments, not as values. + const out = apply({}, [["s", ["a"], JSON.parse('{"__proto__":{"z":1}}')]]); + expect(({} as Record).z).toBeUndefined(); + expect(out).toBeDefined(); + }); + + it("clones reserved value keys without invoking prototype setters", () => { + const t = track({ value: JSON.parse('{"__proto__":{"z":1}}') as JsonValue }); + const out = apply<{ value: JsonValue }>(undefined, t.flush()); + expect(Object.hasOwn(out.value as object, "__proto__")).toBe(true); + expect(({} as Record).z).toBeUndefined(); + }); +}); + +describe("safety: array indices", () => { + // An index may address an existing element or append exactly one past the end. + // Not an arbitrary cap: a sparse array does not survive a JSON round trip, so + // arr[7] = x on a length-3 array already produces unreplicable state. + it("writes an existing index", () => { + expect(apply({ xs: [1, 2, 3] }, [["s", ["xs", 1], 9]])).toEqual({ xs: [1, 9, 3] }); + }); + it("appends one past the end", () => { + expect(apply({ xs: [1, 2, 3] }, [["s", ["xs", 3], 9]])).toEqual({ xs: [1, 2, 3, 9] }); + }); + it("rejects a gap", () => { + expect(() => apply({ xs: [1, 2, 3] }, [["s", ["xs", 5], 9]])).toThrow(); + }); + it("rejects a huge index", () => { + // Would otherwise allocate 4.29 billion entries from one op. + expect(() => apply({ xs: [] }, [["s", ["xs", 4_294_967_290], 1]])).toThrow(); + }); + it("rejects string-spelled array indices at the consumer", () => { + expect(() => apply({ xs: [1, 2, 3] }, [["s", ["xs", "7"], 9]])).toThrow(); + expect(() => apply({ xs: ["a"] }, [["a", ["xs", "0"], "b"]])).toThrow(); + }); + it("rejects sparse writes at the producer", () => { + const t = track({ xs: [1, 2, 3] }); + t.flush(); + expect(() => { + t.state.xs[5] = 9; + }).toThrow(); + expect(t.target.xs).toEqual([1, 2, 3]); + }); + it("rejects array deletes at the producer", () => { + const t = track({ xs: [1, 2, 3] }); + t.flush(); + expect(() => { + delete t.state.xs[1]; + }).toThrow(); + expect(t.target.xs).toEqual([1, 2, 3]); + }); + it("rejects undefined array elements at the producer", () => { + const t = track<{ xs: (number | undefined)[] }>({ xs: [1, 2, 3] }); + t.flush(); + expect(() => { + t.state.xs[1] = undefined; + }).toThrow(); + expect(t.target.xs).toEqual([1, 2, 3]); + }); + it("allows explicit growth with nulls", () => { + expect(apply({ xs: [1] }, [["p", ["xs"], 1, 0, [null, null, 9]]])).toEqual({ xs: [1, null, null, 9] }); + }); + it("rejects deleting one past an array's end", () => { + expect(() => apply({ xs: [1] }, [["d", ["xs", 1]]])).toThrow(); + }); + it("applies large splice payloads without spreading them at once", () => { + const items = Array(300_000).fill(null); + const result = apply({ xs: [] as JsonValue[] }, [["p", ["xs"], 0, 0, items]]); + expect(result.xs).toHaveLength(items.length); + }); +}); + +describe("safety: op structure", () => { + // The CVE-2025-55182 lesson: a decoder must not trust tuple shape. + it("rejects an unknown verb rather than skipping it", () => { + // Silently skipping is how a newer producer's op vanishes and a replica drifts. + expect(() => apply({ a: 1 }, [["ZZZ", ["a"], 9] as never])).toThrow(); + }); + it("rejects non-array splice items", () => { + // Unvalidated, this spreads a string: ["n","o","t","-","a",…] + expect(() => apply({ xs: [1] }, [["p", ["xs"], 0, 0, "not-an-array"] as never])).toThrow(); + }); + it("rejects a string path", () => { + // Unvalidated, "a".slice(0,-1) is "", so it resolved to the ROOT and wrote. + expect(() => apply({ a: 1 }, [["s", "a", 9] as never])).toThrow(); + }); + it("rejects a non-tuple op", () => { + expect(() => apply({ a: 1 }, [{ op: "s" } as never])).toThrow(); + expect(() => apply({ a: 1 }, [null as never])).toThrow(); + }); + it("rejects append to a missing or non-string value", () => { + expect(() => apply({ a: 1 }, [["a", ["missing"], "x"]])).toThrow(); + expect(() => apply({ a: 1 }, [["a", ["a"], "x"]])).toThrow(); + }); + it("rejects negative truncation", () => { + expect(() => apply({ a: "abc" }, [["t", ["a"], -1]])).toThrow(); + expect(() => decoder().decode([["t", ["a"], -1]])).toThrow(); + }); + it("clamps a splice remove past the end, as Array.prototype.splice does", () => { + // Deterministic and identical on both sides — not a hole. + expect(apply({ xs: [1, 2] }, [["p", ["xs"], 0, 1e9, []]])).toEqual({ xs: [] }); + }); +}); + +describe("safety: what a facet can put on the wire", () => { + // A facet cannot forge an op — it mutates plain objects and the tracker builds + // the tuples. Values and keys are the leak, and structuredClone is NOT a JSON + // check: it clones a Map happily, and the op then carries {} while the producer + // keeps a real Map. + const rejects = (name: string, mutate: (s: Record) => void) => + it(`rejects ${name}`, () => { + const t = track>({ pad: PAD }); + t.flush(); // drain the base batch + expect(() => mutate(t.state)).toThrow(); + }); + + rejects("a function", (s) => { + s.f = () => {}; + }); + rejects("a Map", (s) => { + s.m = new Map([[1, 2]]); + }); + rejects("a Date", (s) => { + s.d = new Date(0); + }); + rejects("a BigInt", (s) => { + s.b = 1n; + }); + rejects("a cycle", (s) => { + const a: Record = {}; + a.self = a; + s.c = a; + }); + rejects("a symbol key", (s) => { + s[Symbol("k") as unknown as string] = 1; + }); + + it("rejects invalid array items before mutation", () => { + const t = track<{ xs: JsonValue[] }>({ xs: [] }); + t.flush(); + expect(() => t.state.xs.push(new Map() as unknown as JsonValue)).toThrow(); + expect(t.target.xs).toEqual([]); + }); + + it("allows a getter on state, recording the computed result", () => { + const t = track>({ pad: PAD }); + t.flush(); // drain the base batch + Object.defineProperty(t.target, "g", { get: () => 42, enumerable: true }); + t.state.x = 1; + expect(t.flush()).toEqual([["s", ["x"], 1]]); + }); + + it("accepts a value that merely looks like an op", () => { + // Nested inside ["s", path, value]; nothing flattens, so it cannot be read + // as a top-level op. + const t = track>({ pad: PAD }); + t.flush(); // drain the base batch + t.state.x = ["r", { evil: true }]; + expect(t.flush()).toEqual([["s", ["x"], ["r", { evil: true }]]]); + }); +}); + +describe("assertions", () => { + it("assertJsonValue rejects non-JSON", () => { + const shared = { value: 1 }; + for (const bad of [ + new Map(), + new Set(), + new Date(), + /re/, + () => {}, + Number.NaN, + Infinity, + { left: shared, right: shared }, + ]) { + expect(() => assertJsonValue(bad)).toThrow(); + } + for (const good of [null, true, 1, "s", [1, "a"], { a: { b: [null] } }]) { + expect(() => assertJsonValue(good)).not.toThrow(); + } + }); + it("assertValidOp accepts decoded ops and rejects wire forms", () => { + // Validating Op against the wire grammar would be laxer than the type: a + // two-element ["s", value] would pass, and apply would read the value as a + // path. Each vocabulary gets the validator that matches it. + for (const op of [ + ["r", { a: 1 }], + ["s", ["a"], 1], + ["d", ["a"]], + ["a", ["a"], "x"], + ["t", ["a"], 2], + ["p", ["a"], 0, 0, []], + ] as never[]) { + expect(() => assertValidOp(op)).not.toThrow(); + } + for (const wireOnly of [ + ["s", 1], + ["d"], + ["a", "x"], + ["t", 2], + ["p", 0, 0, []], + ["#", 0, ["a"]], + ["s", 0, 1], + ] as never[]) { + expect(() => assertValidOp(wireOnly)).toThrow(); + expect(() => assertValidWireOp(wireOnly)).not.toThrow(); + } + }); +}); + +describe("codec: path interning and arity omission", () => { + // One pair per stream. The table spans a whole subscription or file, so a + // second consumer joining later needs its own encoder. + const roundTrip = (batches: Op[][]) => { + const enc = encoder(); + const dec = decoder(); + return batches.map((ops) => dec.decode(enc.encode(ops))); + }; + + it("round-trips a stream exactly", () => { + const t = track({ a: { deep: "" }, b: { deep: "" } }); + t.flush(); // drain the base batch + const batches: Op[][] = []; + for (let i = 0; i < 6; i++) { + t.state.a.deep += `x${i}`; + t.state.b.deep += `y${i}`; + batches.push(t.flush()); + } + expect(roundTrip(batches)).toEqual(batches); + }); + + it("interns on second use, not first", () => { + const enc = encoder(); + const p = ["a", "deep"] as const; + const first = enc.encode([["a", p, "1"]]); + const second = enc.encode([["a", p, "2"]]); + expect(first).toEqual([["a", p, "1"]]); // inline + expect(second).toEqual([ + ["#", 0, p], + ["a", 0, "2"], + ]); // define, then use + }); + + it("omits the path when it repeats", () => { + const enc = encoder(); + const p = ["a"] as const; + expect( + enc.encode([ + ["s", p, 1], + ["s", p, 2], + ]), + ).toEqual([ + ["s", p, 1], + ["s", 2], + ]); + }); + + it("does not collide paths containing null characters", () => { + const first = ["a\u0000b"] as const; + const second = ["a", "b"] as const; + const ops: Op[] = [ + ["s", first, 1], + ["s", second, 2], + ]; + expect(decoder().decode(encoder().encode(ops))).toEqual(ops); + }); + + it("rejects a short form without a previous path", () => { + expect(() => decoder().decode([["a", "x"]])).toThrow(); + }); + + it("clears decoder ids on a base batch", () => { + const dec = decoder(); + dec.decode([ + ["#", 0, ["a"]], + ["a", 0, "1"], + ]); + dec.decode([["r", { a: "" }]]); + expect(() => dec.decode([["a", 0, "2"]])).toThrow(); + }); + + it("resets the table on a base batch, so recovery is self-contained", () => { + // A reader replays from the LAST base batch with a fresh decoder. Ids + // defined before it were never seen, so keeping them breaks recovery with + // an unresolvable path id. + const enc = encoder(); + const p = ["a", "deep"] as const; + enc.encode([["a", p, "1"]]); + enc.encode([["a", p, "2"]]); // id 0 assigned + const base = enc.encode([["r", { a: { deep: "x" } }]]); + const after = enc.encode([["a", p, "3"]]); + expect(base).toEqual([["r", { a: { deep: "x" } }]]); + expect(after).toEqual([["a", p, "3"]]); // inline again + + const dec = decoder(); + expect(() => dec.decode(base)).not.toThrow(); + expect(dec.decode(after)).toEqual([["a", p, "3"]]); + }); + + it("survives recovery from the last base batch", () => { + const enc = encoder(); + const t = track({ a: { deep: "" }, b: { deep: "" } }); + t.flush(); // drain the base batch + const wire: WireOp[][] = []; + for (let i = 0; i < 8; i++) { + t.state.a.deep += `x${i}`; + t.state.b.deep += `y${i}`; + if (i === 5) t.rebase(); + wire.push(enc.encode(t.flush())); + } + const lastBase = wire.map(isBase).lastIndexOf(true); + const dec = decoder(); + let replica: unknown; + for (const w of wire.slice(lastBase)) { + const ops = dec.decode(w); + replica = replica === undefined ? structuredClone((ops[0] as ["r", unknown])[1]) : apply(replica, ops); + } + expect(replica).toEqual(t.target); + }); + + it("round-trips random streams", () => { + for (let round = 0; round < 300; round++) { + const t = track<{ a: { p: string; q: string }; b: JsonValue[]; c?: number }>({ + a: { p: "", q: "" }, + b: [], + c: 0, + }); + t.flush(); // drain the base batch + const batches: Op[][] = []; + for (let i = 0; i < 8; i++) { + const r = Math.random(); + if (r < 0.3) t.state.a.p += "x"; + else if (r < 0.5) t.state.a.q += "y"; + else if (r < 0.65) t.state.b.push(i); + else if (r < 0.8) t.state.c = i; + else if (r < 0.9) delete t.state.c; + else t.rebase(); + const ops = t.flush(); + if (ops.length > 0) batches.push(ops); + } + expect(roundTrip(batches)).toEqual(batches); + } + }); +}); + +describe("property: random round-trip", () => { + // 3000 random sequences found the root-splice hole on the first run, before any + // hand-written case did. + const rnd = (d = 0): unknown => { + const r = Math.random(); + if (d > 2 || r < 0.3) return Math.floor(Math.random() * 5); + if (r < 0.45) return ["x", "y", null, true][Math.floor(Math.random() * 4)]; + if (r < 0.7) return Array.from({ length: Math.floor(Math.random() * 4) }, () => rnd(d + 1)); + const o: Record = {}; + for (const k of ["a", "b", "c"]) if (Math.random() < 0.6) o[k] = rnd(d + 1); + return o; + }; + + it("a replica always matches the producer", () => { + let checked = 0; + for (let i = 0; i < 3000; i++) { + const base = rnd(); + if (typeof base !== "object" || base === null) continue; + const t = track(structuredClone(base) as object); + t.flush(); // drain the base batch + const next = rnd(); + if (Array.isArray(t.state) && Array.isArray(next)) { + (t.state as unknown[]).splice(0, (t.state as unknown[]).length, ...next); + } else if (!Array.isArray(t.state) && typeof next === "object" && next !== null && !Array.isArray(next)) { + const s = t.state as Record; + for (const k of Object.keys(s)) if (!(k in next)) delete s[k]; + for (const [k, v] of Object.entries(next)) s[k] = v; + } else continue; + expect(apply(structuredClone(base), t.flush())).toEqual(t.target); + checked++; + } + // Most random pairs are shape-incompatible and skipped; this only guards + // against the loop silently checking nothing. + expect(checked).toBeGreaterThan(200); + }); +});