From 2c995acf44b40555322d0f9dfd2cf2f55455245f Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Thu, 17 Sep 2026 16:36:37 +0200 Subject: [PATCH] feat(chord): replace delta tracker with operation log --- packages/chord/README.md | 16 +- packages/chord/src/delta/README.md | 106 +++- packages/chord/src/delta/index.ts | 742 ++++++++++++++++++--------- packages/chord/test/delta.test.ts | 164 +++++- packages/chord/test/services.test.ts | 14 + 5 files changed, 768 insertions(+), 274 deletions(-) diff --git a/packages/chord/README.md b/packages/chord/README.md index ea2392f91..dd9bd9a7a 100644 --- a/packages/chord/README.md +++ b/packages/chord/README.md @@ -38,10 +38,10 @@ The design has a few connected pieces: stream owns independent path-codec state. Replicas become unready on disconnect or replacement until they are rehydrated. -- **Delta tracking** derives compact operations from tracked plain JSON at - flush time. It preserves string append/front-truncation and array-append - behavior without retaining mutation history, supports durable base batches, - and validates untrusted operations as they are applied. +- **Delta tracking** records and coalesces operations over tracked plain JSON. + It preserves common string and array operations, supports durable base + batches, and validates untrusted operations as they are applied. Batches + guarantee convergence but are not canonical or necessarily minimal. - **Remote service sources** advertise services available outside a facet host and open bindings for the services its facets require. Bindings carry logical @@ -193,10 +193,10 @@ receiving host. To reload, load a candidate, pass its facets to `FacetHost.reload()`, dispose the candidate on failure, and dispose the retired `LoadedFacets` only after a -successful cutover. The host activates and validates the candidate while the old -providers remain routed, then replaces each singleton directly without an -unavailable interval. Stable service handles therefore do not become disconnected -during an ordinary reload. Keyed instances +successful cutover. The host activates and validates the candidate while the +currently active providers remain routed, then replaces each singleton directly +without an unavailable interval. Stable service handles therefore do not become +disconnected during an ordinary reload. Keyed instances remain incarnation-specific and replacements receive fresh generations. The bundler writes a complete temporary directory before replacing the previous output, so loaders do not observe partially built generations. diff --git a/packages/chord/src/delta/README.md b/packages/chord/src/delta/README.md index 1911edd5a..ba90f5c19 100644 --- a/packages/chord/src/delta/README.md +++ b/packages/chord/src/delta/README.md @@ -19,8 +19,10 @@ replica = apply(replica, tracker.flush()); ``` The first `flush()` returns one operation containing the complete value. Each -later flush returns the operations needed to transform the previously published -value into the current value. It returns `[]` when the value has not changed. +later flush returns operations whose application transforms the previously +published value into the current value. It returns `[]` when no tracked mutation +is pending, but a mutation window that restores its starting value may still +produce a redundant batch. `applyImmutable()` copies only containers along changed paths and shares unchanged subtrees. It does not mutate, clone, or freeze either complete input. @@ -151,7 +153,7 @@ tracker.state.messages.push(message); delete tracker.state.retry; ``` -Only the value at flush time is published: +Operations are coalesced within a flush window when doing so is cheap and safe: ```ts tracker.state.status = "starting"; @@ -159,8 +161,12 @@ tracker.state.status = "running"; tracker.flush(); // one set to "running" ``` +The operation sequence is not canonical. Equivalent changes may use different +verbs, and mutations that cancel can still produce a nonempty batch. Consumers +must depend on the resulting value, not the exact tuples or their minimality. + Replacing an object or array is valid. Delta compares its properties and elements -with the previously published value: +with the outgoing value at assignment time: ```ts tracker.state.settings = { @@ -180,8 +186,8 @@ Appending text produces an `a` operation: tracker.state.output += "next line\n"; ``` -Moving a bounded text window forward produces `t` followed by `a` when the old -suffix matches the new prefix: +Moving a bounded text window forward produces `t` followed by `a` when the +previous suffix matches the new prefix: ```ts tracker.state.output = tracker.state.output.slice(200) + nextChunk; @@ -199,23 +205,16 @@ tracker.state.messages.push(second); tracker.state.messages.splice(3, 1, replacement); ``` -All `push()` calls before one flush produce one tail `p`. Changes to older -elements remain separate, regardless of whether they happen before or after the -pushes. Changes to newly pushed elements are included in the pushed values. +Adjacent `push()` calls are normally coalesced into one tail `p`. Intervening +operations may keep them separate to preserve ordering. Changes to older +elements remain separate, and changes to newly pushed elements may be folded +into the pushed values when no structural operation intervenes. -Front or middle insertion, removal, sorting, reversing, `fill()`, and -`copyWithin()` are supported. A structural change combined with edits to elements -whose indices moved may compare and publish the retained suffix positionally: - -```ts -tracker.state.items.shift(); -tracker.state.items[0].status = "changed"; -// A shift followed by push in the same flush has the same issue. -``` - -The emitted data can then scale with the retained suffix, or with the complete -array, rather than only the changed element. When batching is under your control, -flush the structural change before editing elements at their new indices. +Front or middle insertion and removal are recorded directly. Edits before and +after an index-changing operation remain ordered against the array generation +they addressed. Sorting, reversing, `fill()`, and `copyWithin()` emit a snapshot +of the affected array; repeated whole-array mutators can therefore produce a +redundant snapshot even when their combined result restores the prior value. Sparse arrays are unsupported. Writing beyond the next index throws. Increasing `length` creates explicit `null` elements; decreasing it removes elements. @@ -223,6 +222,61 @@ Sparse arrays are unsupported. Writing beyond the next index throws. Increasing `fill()` and `copyWithin()` keep normal JavaScript reference semantics. Do not use them to place one mutable object at multiple live paths. +### Large mutation windows + +#### Build once, assign once + +Every object or array assignment is diffed immediately. Do not repeatedly assign +large intermediate values before one flush: + +```ts +// Avoid: traverses every intermediate tree. +for (const frame of frames) tracker.state.view = render(frame); + +// Prefer: only the final tree crosses the tracked boundary. +const nextView = frames.reduce((view, frame) => renderInto(view, frame), initialView); +tracker.state.view = nextView; +``` + +For a few changes, mutate the leaves directly: + +```ts +for (const update of updates) { + tracker.state.view.rows[update.index]!.status = update.status; +} +``` + +#### Do not cancel whole-array mutators + +`sort()`, `reverse()`, `fill()`, and `copyWithin()` record snapshots. Cancelling +them still publishes the final snapshot: + +```ts +// Avoid: final value is unchanged, but a snapshot may still be sent. +tracker.state.items.reverse(); +tracker.state.items.reverse(); + +// Prefer: decide before mutating tracked state. +if (needsReverse) tracker.state.items.reverse(); +``` + +#### Publish large inserts and edit sets in chunks + +Pending inserted values are cloned for replica ownership. Very large unflushed +pushes therefore temporarily retain both the live values and their operation +payloads. Long operation/path histories may also collapse to a complete base +batch, increasing snapshot and wire cost. + +```ts +for (const chunk of chunks(items, 1_000)) { + tracker.state.items.push(...chunk); + replica = apply(replica, tracker.flush()); // send each batch in a real producer +} +``` + +Use the same pattern for large sets of unrelated edits: apply a bounded chunk, +publish it, then continue. + ### Optional properties Optional object properties use absence. They do not require `null`: @@ -292,6 +346,14 @@ tracker.state = replacement; // replace the root; next flush is complete `discard()` intentionally prevents current changes from reaching existing replicas. Use it only when those replicas do not need the discarded changes. +`flush()` guarantees convergence, not a minimal or canonical diff. Any nonempty +batch advances replicated-state sequence numbers and notifies subscribers, even +if applying it leaves the value deeply equal to the previous revision. To bound +pending operation and path history, a sufficiently long mutation window may +collapse to a complete base batch automatically. This bounds accumulated log +metadata, not payload bytes or peak allocation, and trades one full-value +snapshot for an additional recovery point. + `apply()` adopts object and array payloads from its input batch. Do not freeze a batch before applying it, and do not apply one in-memory batch to multiple mutable replicas unless each replica owns that batch. A serialized and decoded diff --git a/packages/chord/src/delta/index.ts b/packages/chord/src/delta/index.ts index 69f6ad021..3aabc51e0 100644 --- a/packages/chord/src/delta/index.ts +++ b/packages/chord/src/delta/index.ts @@ -3,7 +3,7 @@ import type { JsonValue } from "../types.ts"; export type { JsonValue } from "../types.ts"; // ───────────────────────────────────────────────────────────────────────────── -// chord/delta — flush-time change tracking over plain JSON. +// chord/delta — operation-log 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. @@ -20,9 +20,9 @@ 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 flush time, so a root `p` is always a partial modification. + * the root: the type forbids it. `p` may, because a tracked value can itself be + * an array. Operation shape is not canonical: an array may be emptied by either + * a replacement or a root splice. * * `Op` knows nothing about the path dictionary. Interning, id references and * omitted paths live in `WireOp` and exist only between `encode` and `decode`. @@ -130,17 +130,21 @@ const isObj = (value: unknown): value is object => value !== null && typeof valu 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, - }); + const nullProto = Object.getPrototypeOf(value) === null; + const result = (nullProto ? Object.create(null) : {}) as Record; + for (const key of Object.keys(value as Record)) { + const child = (value as Record)[key]!; + const copied = isObj(child) ? cloneJson(child) : child; + if (!nullProto && key in result) { + Object.defineProperty(result, key, { + value: copied, + writable: true, + enumerable: true, + configurable: true, + }); + } else { + result[key] = copied; + } } return result as T; }; @@ -151,10 +155,6 @@ const norm = (target: object, key: string | symbol): Seg | symbol => const MUTATORS = new Set(["push", "pop", "shift", "unshift", "splice", "sort", "reverse", "fill", "copyWithin"]); const MISSING = Symbol("missing"); type MaybeJson = JsonValue | typeof MISSING; -type ArrayDirty = { kind: "append"; start: number } | { kind: "diff" } | { kind: "replace" }; -type DirtyNode = { valueDirty?: true; array?: ArrayDirty; children: Map }; - -const dirtyNode = (): DirtyNode => ({ children: new Map() }); const spliceItems = (target: unknown[], index: number, remove: number, items: JsonValue[]): JsonValue[] => { const removed = Reflect.apply(Array.prototype.splice, target, [index, remove]) as JsonValue[]; @@ -186,11 +186,6 @@ const jsonEqual = (left: JsonValue, right: JsonValue): boolean => { return true; }; -const ownValue = (value: JsonValue, segment: Seg): MaybeJson => { - if (!isObj(value) || !Object.hasOwn(value, segment)) return MISSING; - return (value as Record)[segment]!; -}; - const emitSet = (path: Path, value: JsonValue, out: Op[]): void => { const snapshot = cloneJson(value); if (path.length === 0) out.push(["r", snapshot]); @@ -312,195 +307,453 @@ function diffArray(before: JsonValue[], after: JsonValue[], path: Path, scan: nu } } -const walkDirty = (before: JsonValue, after: JsonValue, node: DirtyNode, path: Path, scan: number, out: Op[]): void => { - if (node.valueDirty) { - diffValue(before, after, path, scan, out); - return; - } - if (node.array !== undefined) { - if (node.array.kind === "replace") { - if (!jsonEqual(before, after)) emitSet(path, after, out); - return; - } - if (!Array.isArray(before) || !Array.isArray(after) || node.array.kind === "diff") { - diffValue(before, after, path, scan, out); - return; - } - const start = node.array.start; - if (before.length !== start || after.length < start) { - diffValue(before, after, path, scan, out); - return; - } - for (const [segment, child] of node.children) { - if (typeof segment !== "number" || segment >= start) continue; - const previous = ownValue(before, segment); - const current = ownValue(after, segment); - if (previous === MISSING || current === MISSING || child.valueDirty) { - diffValue(previous, current, [...path, segment], scan, out); - } else if (isObj(previous) && isObj(current)) { - walkDirty(previous as JsonValue, current as JsonValue, child, [...path, segment], scan, out); - } else { - diffValue(previous, current, [...path, segment], scan, out); - } - } - const items = after.slice(start); - if (items.length > 0) out.push(["p", [...path], start, 0, cloneJson(items)]); - return; - } - for (const [segment, child] of node.children) { - const previous = ownValue(before, segment); - const current = ownValue(after, segment); - if (previous === MISSING || current === MISSING || child.valueDirty) { - diffValue(previous, current, [...path, segment], scan, out); - continue; - } - if (!isObj(previous) || !isObj(current)) { - diffValue(previous, current, [...path, segment], scan, out); - continue; - } - walkDirty(previous as JsonValue, current as JsonValue, child, [...path, segment], scan, out); - } -}; - -/** - * Bring `baseline` up to `root` along the dirty paths by sharing references. - * Returns false — having changed nothing — if any dirty node is an array - * change other than a pure append, so the caller can replay ops instead. - * Cloning a whole array there is O(n) per flush; replay is O(changes). - * - * Strings are immutable, so root's `after` is shared outright, and sharing it - * also means the next flush compares against a flat string rather than a cons. - * Objects are cloned because root keeps mutating them. - */ -const syncBaseline = (baseline: JsonValue, root: JsonValue, node: DirtyNode): boolean => { - if (!canSync(node)) return false; - syncInto(baseline, root, node); - return true; -}; - -const canSync = (node: DirtyNode): boolean => { - if (node.array !== undefined && node.array.kind !== "append") return false; - for (const child of node.children.values()) if (!canSync(child)) return false; - return true; -}; - -const syncInto = (baseline: JsonValue, root: JsonValue, node: DirtyNode): void => { - const parent = baseline as Record; - if (node.array?.kind === "append" && Array.isArray(baseline) && Array.isArray(root)) { - const start = node.array.start; - for (const [index, child] of node.children) { - if (typeof index === "number" && index < start) syncChild(parent, root, index, child); - } - for (let i = start; i < root.length; i++) { - baseline.push(isObj(root[i]) ? cloneJson(root[i] as JsonValue) : (root[i] as JsonValue)); - } - return; - } - for (const [segment, child] of node.children) syncChild(parent, root, segment, child); -}; - -const syncChild = ( - parent: Record, - root: JsonValue, - segment: Seg, - child: DirtyNode, -): void => { - const current = ownValue(root, segment); - const previous = ownValue(parent as JsonValue, segment); - if (current === MISSING) { - if (Array.isArray(parent)) parent.splice(segment as number, 1); - else delete parent[segment]; - return; - } - if (child.valueDirty || !isObj(current) || !isObj(previous) || Array.isArray(current) !== Array.isArray(previous)) { - parent[segment] = isObj(current) ? cloneJson(current as JsonValue) : (current as JsonValue); - return; - } - syncInto(previous as JsonValue, current as JsonValue, child); -}; - -const cloneOp = (op: Op): Op => { - switch (op[0]) { - case "r": - return ["r", cloneJson(op[1])]; - case "s": - return ["s", op[1], cloneJson(op[2])]; - case "p": - return ["p", op[1], op[2], op[3], cloneJson(op[4])]; - default: - return op; - } -}; - export function track(root: T, options: TrackerOptions = {}): Tracker { const scan = options.maxOverlapScan ?? 65_536; - let pending = dirtyNode(); + // Mutations are recorded directly. The active trie supports coalescing while + // retired child generations preserve dominance across array reindexing. + type Slot = { + op: Op; + dead: boolean; + order: number; + index: number; + str?: { anchor: string; value: string }; + }; + type LogNode = { + slots?: Slot[]; + kids?: Map; + retiredKids?: Map[]; + lastOrder?: number; + }; + + let log: (Slot | undefined)[] = []; + let trie: LogNode = {}; + let nextOrder = 0; + let tombstones = 0; + let liveSlots = 0; + let nodeCount = 1; + let lastAddedSlot: Slot | undefined; let hasPending = false; - let baseline: JsonValue | undefined; let forceBase = true; const clearPending = (): void => { - pending = dirtyNode(); + log = []; + trie = {}; + nextOrder = 0; + tombstones = 0; + liveSlots = 0; + nodeCount = 1; + lastAddedSlot = undefined; hasPending = false; }; - const ensureNode = (path: Path): DirtyNode | undefined => { + const logNode = (path: Path): LogNode => { + let at = trie; + for (const segment of path) { + if (!at.kids) at.kids = new Map(); + let next = at.kids.get(segment); + if (next === undefined) { + next = {}; + at.kids.set(segment, next); + nodeCount++; + } + at = next; + } + return at; + }; + + const findLogNode = (path: Path): LogNode | undefined => { + let at = trie; + for (const segment of path) { + const next = at.kids?.get(segment); + if (next === undefined) return undefined; + at = next; + } + return at; + }; + + const compactLog = (): void => { + if (tombstones < 1_024 || tombstones * 2 < log.length) return; + const compacted: Slot[] = []; + for (const slot of log) { + if (slot === undefined) continue; + slot.index = compacted.length; + compacted.push(slot); + } + log = compacted; + tombstones = 0; + }; + + const killSlot = (slot: Slot): void => { + if (slot.dead) return; + slot.dead = true; + liveSlots--; + if (log[slot.index] === slot) { + log[slot.index] = undefined; + tombstones++; + } + }; + + const liveSlot = (at: LogNode): Slot | undefined => { + const slots = at.slots; + if (slots === undefined) return undefined; + while (slots.length > 0 && slots[slots.length - 1]!.dead) slots.pop(); + if (slots.length === 0) { + at.slots = undefined; + return undefined; + } + return slots[slots.length - 1]; + }; + + const killHere = (at: LogNode): void => { + if (at.slots === undefined) return; + for (const slot of at.slots) killSlot(slot); + at.slots = undefined; + }; + + const addSlot = (at: LogNode, slot: Slot): void => { + compactLog(); + slot.order = nextOrder++; + slot.index = log.length; + at.lastOrder = slot.order; + if (at.slots === undefined) at.slots = []; + at.slots.push(slot); + log.push(slot); + liveSlots++; + lastAddedSlot = slot; + }; + + const collapsePending = (): void => { + // Long mutation windows can otherwise retain operations and retired path + // generations indefinitely. Fall back to a complete snapshot once either + // history exceeds a bounded coalescing window. Payload bytes and transient + // allocation remain workload-dependent. + if (forceBase || (liveSlots <= 4_096 && nodeCount <= 4_096)) return; + const value = cloneJson(root as unknown as JsonValue); + clearPending(); hasPending = true; - let node = pending; - for (const segment of path) { - if (node.valueDirty || node.array?.kind === "diff" || node.array?.kind === "replace") return undefined; - let child = node.children.get(segment); - if (child === undefined) child = dirtyNode(); - else node.children.delete(segment); - node.children.set(segment, child); - node = child; + addSlot(trie, { op: ["r", value], dead: false, order: 0, index: 0 }); + }; + + const killSubtree = (at: LogNode): void => { + killHere(at); + if (at.kids !== undefined) { + for (const child of at.kids.values()) killSubtree(child); + at.kids = undefined; } - return node; - }; - - const findNode = (path: Path): DirtyNode | undefined => { - let node = pending; - for (const segment of path) { - const child = node.children.get(segment); - if (child === undefined) return undefined; - node = child; + if (at.retiredKids !== undefined) { + for (const generation of at.retiredKids) { + for (const child of generation.values()) killSubtree(child); + } + at.retiredKids = undefined; } - return node; }; - const markValue = (path: Path): void => { - const node = ensureNode(path); - if (node === undefined) return; - node.valueDirty = true; - node.array = undefined; - node.children.clear(); + const retireKids = (at: LogNode): void => { + if (at.kids === undefined) return; + if (at.retiredKids === undefined) at.retiredKids = []; + at.retiredKids.push(at.kids); + at.kids = undefined; }; - const markArrayAppend = (path: Path, start: number): void => { - const node = ensureNode(path); - if (node === undefined || node.valueDirty || node.array?.kind === "diff" || node.array?.kind === "replace") { + // Deepest live ancestor op carrying a payload we can fold a later write into. + // `s`/`r` carry the whole subtree; `p` carries the items it inserted, so a + // write to one of those indices belongs inside the payload rather than after it. + const foldTarget = (path: Path): { slot: Slot; rest: Path; item?: number } | undefined => { + let at = trie; + let found: { slot: Slot; depth: number; item?: number } | undefined; + let ancestorMax = -1; + for (let depth = 0; depth < path.length; depth++) { + if (found !== undefined && at.lastOrder !== undefined && at.lastOrder > found.slot.order) { + found = undefined; + } + const slot = liveSlot(at); + // A fold is sound only if nothing has been recorded at this path or above it + // since: a later op there (a splice on the same array, a replacement of an + // ancestor) would have to apply after this write, not before it. Writes to + // other branches are irrelevant, which is why this is not "the most recent op". + if (slot !== undefined && slot.order >= ancestorMax && at.lastOrder === slot.order) { + if (slot.op[0] === "s" || slot.op[0] === "r") found = { slot, depth }; + else if (slot.op[0] === "p") { + const index = path[depth]; + const items = slot.op[4] as JsonValue[]; + if (typeof index === "number" && index >= slot.op[2] && index < slot.op[2] + items.length) { + found = { slot, depth: depth + 1, item: index - slot.op[2] }; + } + } + } + if (at.lastOrder !== undefined && at.lastOrder > ancestorMax) ancestorMax = at.lastOrder; + const next = at.kids?.get(path[depth]!); + if (next === undefined) break; + at = next; + } + if (found === undefined) return undefined; + return { slot: found.slot, rest: path.slice(found.depth), item: found.item }; + }; + + const foldInto = (container: JsonValue, rest: Path, op: Op): boolean => { + if (rest.length === 0) return false; + let target: JsonValue = container; + for (let index = 0; index < rest.length - 1; index++) { + if (!isObj(target)) return false; + target = (target as Record)[rest[index]!]!; + } + if (!isObj(target)) return false; + const key = rest[rest.length - 1]!; + const holder = target as Record; + const write = (value: JsonValue): void => { + Object.defineProperty(holder, key, { value, writable: true, enumerable: true, configurable: true }); + }; + switch (op[0]) { + case "s": + if (key === "__proto__") return false; + write(cloneJson(op[2])); + return true; + case "d": + delete holder[key]; + return true; + case "a": { + const value = holder[key]; + if (typeof value !== "string") return false; + write(value + op[2]); + return true; + } + case "t": { + const value = holder[key]; + if (typeof value !== "string") return false; + write(value.slice(op[2])); + return true; + } + case "p": { + const value = holder[key]; + if (!Array.isArray(value)) return false; + spliceItems(value, op[2], op[3], cloneJson(op[4] as JsonValue[])); + return true; + } + default: + return false; + } + }; + + const recordString = (path: Path, previous: string, value: string): void => { + if (forceBase) return; + hasPending = true; + // a string inside a pending payload belongs in that payload, as for any write + const fold = foldTarget(path); + if (fold !== undefined) { + const op: Op = ["s", path as unknown as NonEmptyPath, value]; + if (fold.item !== undefined) { + const items = fold.slot.op[4] as JsonValue[]; + if (fold.rest.length === 0) { + items[fold.item] = value; + return; + } + if (foldInto(items[fold.item]!, fold.rest, op)) return; + } else { + const payload = fold.slot.op[0] === "r" ? fold.slot.op[1] : fold.slot.op[2]; + if (foldInto(payload as JsonValue, fold.rest, op)) return; + } + } + const at = logNode(path); + const live = liveSlot(at); + if (live?.str !== undefined) { + live.str.value = value; return; } - if (node.array === undefined) node.array = { kind: "append", start }; + if (live !== undefined) { + // a pending set or delete at this path already replaced the value; keep that + // op and carry the new value in it rather than anchoring to it + if (live.op[0] === "s" || live.op[0] === "r") { + live.op = live.op[0] === "r" ? ["r", value] : ["s", live.op[1], value]; + return; + } + if (live.op[0] === "d") { + killHere(at); + addSlot(at, { + op: ["s", path as unknown as NonEmptyPath, value], + dead: false, + order: 0, + index: 0, + }); + return; + } + // a truncate/append pair from an earlier string diff: both must go + killHere(at); + } + killSubtree(at); + addSlot(at, { + op: ["s", path as unknown as NonEmptyPath, value], + dead: false, + order: 0, + index: 0, + str: { anchor: previous, value }, + }); }; - const markArrayDiff = (path: Path): void => { - const node = ensureNode(path); - if (node === undefined || node.valueDirty || node.array?.kind === "replace") return; - node.array = { kind: "diff" }; - node.children.clear(); + const record = (op: Op): void => { + if (forceBase) return; + hasPending = true; + const path = (op[0] === "r" ? [] : (op[1] as Path)) as Path; + const existing = findLogNode(path); + const anchored = existing === undefined ? undefined : liveSlot(existing); + if (anchored?.str !== undefined) { + switch (op[0]) { + case "a": + anchored.str.value += op[2]; + return; + case "t": + anchored.str.value = anchored.str.value.slice(op[2]); + return; + case "s": + if (typeof op[2] === "string") { + anchored.str.value = op[2]; + return; + } + } + if (existing !== undefined) killSubtree(existing); + } else if (existing !== undefined && (op[0] === "s" || op[0] === "d" || op[0] === "r")) { + // A replacement absorbed into an ancestor payload must still invalidate + // operations already recorded at and below its destination. + killSubtree(existing); + } + + if (path.length > 0) { + const fold = foldTarget(path); + if (fold !== undefined) { + if (fold.item !== undefined) { + const items = fold.slot.op[4] as JsonValue[]; + if (fold.rest.length === 0) { + if (op[0] === "s") { + items[fold.item] = cloneJson(op[2]); + return; + } + } else if (foldInto(items[fold.item]!, fold.rest, op)) return; + } else { + const payload = fold.slot.op[0] === "r" ? fold.slot.op[1] : fold.slot.op[2]; + if (foldInto(payload as JsonValue, fold.rest, op)) return; + } + } + } + + const at = logNode(path); + const live = liveSlot(at); + if (live !== undefined) { + const previous = live.op; + if (op[0] === "a" && previous[0] === "a") { + live.op = ["a", previous[1], previous[2] + op[2]]; + return; + } + if (op[0] === "a" && (previous[0] === "s" || previous[0] === "r")) { + const value = previous[0] === "r" ? previous[1] : previous[2]; + if (typeof value === "string") { + live.op = previous[0] === "r" ? ["r", value + op[2]] : ["s", previous[1], value + op[2]]; + return; + } + } + if (op[0] === "t" && (previous[0] === "s" || previous[0] === "r")) { + const value = previous[0] === "r" ? previous[1] : previous[2]; + if (typeof value === "string") { + const cut = value.slice(op[2]); + live.op = previous[0] === "r" ? ["r", cut] : ["s", previous[1], cut]; + return; + } + } + if (op[0] === "p" && previous[0] === "p") { + const previousItems = previous[4] as JsonValue[]; + // These rewrites are sound only for adjacent recorded operations. A + // tombstoned operation remains a barrier through lastAddedSlot. + if ( + previous[3] === 0 && + op[3] === 0 && + previous[2] + previousItems.length === op[2] && + lastAddedSlot === live + ) { + const items = op[4] as JsonValue[]; + for (let index = 0; index < items.length; index++) previousItems.push(items[index]!); + return; + } + if ( + previous[3] === 0 && + lastAddedSlot === live && + op[2] >= previous[2] && + op[2] + op[3] <= previous[2] + previousItems.length + ) { + spliceItems(previousItems, op[2] - previous[2], op[3], op[4] as JsonValue[]); + if (previousItems.length === 0 && previous[3] === 0) killSlot(live); + return; + } + if ( + op[3] > 0 && + (op[4] as JsonValue[]).length === 0 && + previousItems.length > 0 && + lastAddedSlot === live + ) { + const from = op[2] - previous[2]; + if (from >= 0 && from + op[3] === previousItems.length) { + previousItems.length = from; + if (previousItems.length === 0 && previous[3] === 0) killSlot(live); + return; + } + } + } + if (op[0] === "s" || op[0] === "d" || op[0] === "r") killHere(at); + } + if (op[0] === "s" || op[0] === "r" || op[0] === "d") { + // Replacements dominate all earlier descendants, including generations + // detached by array splices. + killSubtree(at); + } else if (op[0] === "p") { + // Splices preserve earlier writes but form a barrier for later folding. + retireKids(at); + } + addSlot(at, { op, dead: false, order: 0, index: 0 }); }; - const markArrayReplace = (path: Path): void => { - const node = ensureNode(path); - if (node === undefined || node.valueDirty) return; - node.array = { kind: "replace" }; - node.children.clear(); - }; - - const appendStart = (path: Path): number | undefined => { - const array = findNode(path)?.array; - return array?.kind === "append" ? array.start : undefined; + // Local diff, used when a whole container is assigned: keeps op quality + // without a baseline by comparing the outgoing value with the incoming one at + // the moment of the write. String leaves go through recordString so every + // string path keeps a single anchored slot; otherwise a later write to the + // same string would have to supersede ops whose starting value it no longer + // knows. + const diffInto = (before: JsonValue, after: JsonValue, at: Seg[]): void => { + if (forceBase) return; + hasPending = true; + if (before === after) return; + if (typeof before === "string" && typeof after === "string") { + recordString(at.slice() as Path, before, after); + return; + } + if (Array.isArray(before) && Array.isArray(after) && before.length === after.length) { + for (let index = 0; index < after.length; index++) { + at.push(index); + diffInto(before[index]!, after[index]!, at); + at.pop(); + } + return; + } + if (isObj(before) && isObj(after) && !Array.isArray(before) && !Array.isArray(after)) { + const beforeObject = before as Record; + const afterObject = after as Record; + const beforeKeys = Object.keys(beforeObject); + const afterKeys = Object.keys(afterObject); + if ([...beforeKeys, ...afterKeys].some((key) => RESERVED_SEGMENTS.has(key))) { + record(["s", at.slice() as unknown as NonEmptyPath, cloneJson(after)]); + return; + } + for (const key of afterKeys) { + at.push(key); + if (Object.hasOwn(beforeObject, key)) diffInto(beforeObject[key]!, afterObject[key]!, at); + else record(["s", at.slice() as unknown as NonEmptyPath, cloneJson(afterObject[key]!)]); + at.pop(); + } + for (const key of beforeKeys) { + if (!Object.hasOwn(afterObject, key)) record(["d", [...at, key] as unknown as NonEmptyPath]); + } + return; + } + // arrays of differing length, and everything else: chord's own diff + const out: Op[] = []; + diffValue(before, after, at as Path, scan, out); + for (const op of out) record(op); }; const guard = (segment: Seg | symbol): Seg => { @@ -541,47 +794,51 @@ export function track(root: T, options: TrackerOptions = {}): switch (key) { case "push": { const items = adoptItems(args); - if (items.length > 0) markArrayAppend(path, before); + if (items.length > 0) record(["p", [...path], before, 0, cloneJson(items)]); spliceItems(target, before, 0, items); result = target.length; break; } case "unshift": { const items = adoptItems(args); - if (items.length > 0) markArrayDiff(path); + if (items.length > 0) record(["p", [...path], 0, 0, cloneJson(items)]); spliceItems(target, 0, 0, items); result = target.length; break; } case "pop": - if (before > 0) { - const start = appendStart(path); - if (start === undefined || before - 1 < start) markArrayDiff(path); - } + if (before > 0) record(["p", [...path], before - 1, 1, []]); result = Reflect.apply(Array.prototype.pop, target, args); break; case "shift": - if (before > 0) markArrayDiff(path); + if (before > 0) record(["p", [...path], 0, 1, []]); result = Reflect.apply(Array.prototype.shift, target, args); break; case "splice": { const items = adoptItems(args.slice(2)); const { index, remove } = spliceRange(before, args); if (remove > 0 || items.length > 0) { - const start = appendStart(path); - if (index === 0 && remove === before) markArrayReplace(path); - else if (start !== undefined && index >= start) { - // The final append payload includes all tail edits. - } else if (index === before && remove === 0) markArrayAppend(path, before); - else markArrayDiff(path); + // a splice that clears the whole array is a replacement of it + if (index === 0 && remove === before) { + if (path.length === 0) record(["r", cloneJson(items as JsonValue)]); + else record(["s", [...path] as unknown as NonEmptyPath, cloneJson(items as JsonValue)]); + } else record(["p", [...path], index, remove, cloneJson(items)]); } result = spliceItems(target, index, remove, items); break; } - default: - markArrayDiff(path); + default: { result = Reflect.apply(Array.prototype[key as "sort"], target, args); + if (path.length === 0) record(["r", cloneJson(target as unknown as JsonValue)]); + else + record([ + "s", + [...path] as unknown as NonEmptyPath, + cloneJson(target as unknown as JsonValue), + ]); + } } + collapsePending(); if (key === "pop") childProxies.delete(String(before - 1)); else if (key !== "push") childProxies.clear(); return key === "sort" || key === "reverse" || key === "fill" || key === "copyWithin" ? proxy : result; @@ -619,16 +876,19 @@ export function track(root: T, options: TrackerOptions = {}): return Reflect.set(target, key, value); } if (next < before) { - const start = appendStart(path); - if (next === 0) markArrayReplace(path); - else if (start === undefined || next < start) markArrayDiff(path); + if (next === 0) { + if (path.length === 0) record(["r", []]); + else record(["s", [...path] as unknown as NonEmptyPath, []]); + } else record(["p", [...path], next, before - next, []]); Reflect.set(target, key, next); childProxies.clear(); } else if (next > before) { - markArrayAppend(path, before); target.length = next; target.fill(null, before); + const grown = new Array(next - before).fill(null) as JsonValue[]; + record(["p", [...path], before, 0, grown]); } + collapsePending(); return true; } @@ -643,25 +903,36 @@ export function track(root: T, options: TrackerOptions = {}): if (Array.isArray(target)) { throw new TypeError("undefined would create a sparse array; use splice instead"); } - markValue(at); + if (Object.hasOwn(target, key)) record(["d", at]); childProxies.delete(key); - return Reflect.deleteProperty(target, key); + const deleted = Reflect.deleteProperty(target, key); + if (deleted) collapsePending(); + return deleted; } const previous = (target as Record)[key]; if (previous === value) return true; const cached = childProxies.get(key); if (cached !== undefined && cached.target === previous && cached.proxy === value) return true; - if (Array.isArray(target)) { - const index = segment as number; - if (index === target.length) markArrayAppend(path, target.length); - else { - const start = appendStart(path); - if (start === undefined || index < start) markValue(at); - } - } else markValue(at); + if (Array.isArray(target) && (segment as number) === target.length) { + record(["p", [...path], target.length, 0, [cloneJson(value as JsonValue)]]); + } else if (isObj(previous) && isObj(value)) { + // whole-container assignment: diff locally so a producer that rebuilds + // its partial each frame still emits appends rather than replacements + diffInto(previous as JsonValue, value as JsonValue, [...path, segment] as Seg[]); + } else if (typeof previous === "string" && typeof value === "string") { + // A string path keeps an anchor: the value it had at the first write in + // this window. Every later write updates the anchored value, and flush + // diffs anchor -> final once. That gives the same truncate/append pair a + // baseline diff would, without keeping a baseline for the whole document. + recordString([...path, segment] as Path, previous, value); + } else { + record(["s", at, cloneJson(value as JsonValue)]); + } childProxies.delete(key); - return Reflect.set(target, key, value); + const updated = Reflect.set(target, key, value); + if (updated) collapsePending(); + return updated; }, deleteProperty(target, key) { @@ -671,9 +942,11 @@ export function track(root: T, options: TrackerOptions = {}): if (typeof segment !== "number") throw new UnsafePathError(segment); throw new TypeError("delete would create a sparse array; use splice instead"); } - markValue([...path, segment]); + if (Object.hasOwn(target, key)) record(["d", [...path, segment] as unknown as NonEmptyPath]); childProxies.delete(key); - return Reflect.deleteProperty(target, key); + const deleted = Reflect.deleteProperty(target, key); + if (deleted) collapsePending(); + return deleted; }, defineProperty() { @@ -708,7 +981,6 @@ export function track(root: T, options: TrackerOptions = {}): clearPending(); root = next; state = wrap(root, []); - baseline = undefined; forceBase = true; }, rebase() { @@ -716,31 +988,29 @@ export function track(root: T, options: TrackerOptions = {}): forceBase = true; }, get dirty() { + // conservative: true if anything was written since the last flush, even + // if the writes cancelled out return forceBase || hasPending; }, discard() { - baseline = cloneJson(root as unknown as JsonValue); clearPending(); }, flush() { if (forceBase) { const value = cloneJson(root as unknown as JsonValue); - baseline = cloneJson(root as unknown as JsonValue); forceBase = false; clearPending(); return [["r", value]]; } - if (!hasPending || baseline === undefined) return []; + if (!hasPending) return []; const out: Op[] = []; - walkDirty(baseline, root as unknown as JsonValue, pending, [], scan, out); - // Advance the baseline by SHARING references from root where that is - // cheap and exact — scalars, strings, and array appends. Replaying the - // ops rebuilds every touched string via slice + concat: two window-sized - // allocations per flush and a cons the next flush must flatten. For - // anything the sync cannot express cheaply (a non-append array change), - // it declines and the original replay runs unchanged. - if (!syncBaseline(baseline as JsonValue, root as unknown as JsonValue, pending)) { - if (out.length > 0) baseline = apply(baseline, out.map(cloneOp)); + for (const slot of log) { + if (slot === undefined || slot.dead) continue; + if (slot.str !== undefined) { + diffValue(slot.str.anchor, slot.str.value, slot.op[1] as Path, scan, out); + continue; + } + out.push(slot.op); } clearPending(); return out; diff --git a/packages/chord/test/delta.test.ts b/packages/chord/test/delta.test.ts index e1661c17a..3e9cc38bd 100644 --- a/packages/chord/test/delta.test.ts +++ b/packages/chord/test/delta.test.ts @@ -124,6 +124,36 @@ describe("tracker: intent", () => { expect(apply(structuredClone(initial), ops)).toEqual(t.state); }); + it("updates an anchored string through a later container replacement", () => { + const initial = { xs: [{ k: "a" }] }; + const t = track(structuredClone(initial)); + t.flush(); + t.state.xs[0]!.k += "b"; + t.state.xs = [{ k: "abc" }, { k: "z" }]; + expect(apply(structuredClone(initial), t.flush())).toEqual(t.state); + }); + + it("preserves string history across a null-prototype container replacement", () => { + const initial = { value: { text: "a" } }; + const t = track(structuredClone(initial)); + t.flush(); + const replacement = Object.assign(Object.create(null) as { text: string }, { text: "ab" }); + t.state.value = replacement; + t.state.value.text = "abc"; + const ops = decoder().decode(JSON.parse(JSON.stringify(encoder().encode(t.flush()))) as WireOp[]); + expect(applyImmutable(structuredClone(initial), ops)).toEqual(JSON.parse(JSON.stringify(t.state))); + }); + + it("front-truncates a pending string set from the correct end", () => { + const initial = { xs: [0] as Array }; + const t = track(structuredClone(initial)); + t.flush(); + t.state.xs[0] = "abc"; + t.state.xs = ["bc", 0]; + const ops = decoder().decode(JSON.parse(JSON.stringify(encoder().encode(t.flush()))) as WireOp[]); + expect(applyImmutable(structuredClone(initial), ops)).toEqual(t.state); + }); + it("emits nothing when a replacement is deeply equal", () => { const t = track({ value: { nested: [1, { text: "same" }] } }); t.flush(); @@ -261,6 +291,53 @@ describe("tracker: root ops", () => { expect(apply(structuredClone(initial), t.flush())).toEqual(t.state); }); + it("drops detached element writes when a root array is replaced", () => { + const initial = [{ k: "a" }, { k: "b" }]; + const t = track(structuredClone(initial)); + t.flush(); + t.state[0]!.k += "x"; + t.state.unshift({ k: "head" }); + t.state.splice(0, t.state.length, { k: "final" }); + const ops = t.flush(); + expect(ops[0]?.[0]).toBe("r"); + expect(apply(structuredClone(initial), ops)).toEqual(t.state); + }); + + it("keeps nested writes ordered across an inserted array reindex", () => { + const initial: number[][] = []; + const t = track(structuredClone(initial)); + t.flush(); + t.state.push([10, 20]); + t.state[0]!.shift(); + t.state[0]![0] = 30; + const ops = decoder().decode(JSON.parse(JSON.stringify(encoder().encode(t.flush()))) as WireOp[]); + expect(applyImmutable(structuredClone(initial), ops)).toEqual(t.state); + }); + + it.each([ + [ + "replacement", + (state: JsonValue[]) => { + state[0] = 0; + }, + ], + [ + "clear", + (state: JsonValue[]) => { + (state[0] as JsonValue[]).length = 0; + }, + ], + ] as const)("invalidates nested operations folded into an inserted payload on %s", (_name, replace) => { + const initial: JsonValue[] = []; + const t = track(structuredClone(initial)); + t.flush(); + t.state.push([]); + (t.state[0] as JsonValue[]).push(1); + replace(t.state); + const ops = decoder().decode(JSON.parse(JSON.stringify(encoder().encode(t.flush()))) as WireOp[]); + expect(applyImmutable(structuredClone(initial), ops)).toEqual(t.state); + }); + it("round-trips deterministic element mutations across reindexing splices", () => { let seed = 0x1234abcd; const next = (): number => { @@ -480,7 +557,7 @@ describe("the first flush", () => { expect(t.flush()).toEqual([]); }); - it("discard accepts pending changes into the local baseline", () => { + it("discard accepts pending changes without publishing them", () => { const t = track({ x: 0, y: 0 }); t.flush(); t.state.x = 1; @@ -568,9 +645,9 @@ describe("apply and fan-out", () => { }); }); -describe("flush-time minimization", () => { - // Mutations only mark dirty paths. Flush compares the last published baseline - // with the current value, so repeated writes never accumulate pending ops. +describe("pending operation coalescing", () => { + // Repeated operations are coalesced when ordering and array index semantics + // make that safe. Exact operation shape is not part of the contract. 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)); @@ -642,13 +719,29 @@ describe("flush-time minimization", () => { ).toEqual([["s", ["x"], 5]]); }); - it("derives an append after delete-then-recreate", () => { - const t = track<{ x?: string; y: number }>({ x: "", y: 1 }); + it("converges after delete-then-recreate without requiring an append", () => { + const initial = { x: "", y: 1 }; + const t = track<{ x?: string; y: number }>(structuredClone(initial)); t.flush(); delete t.state.x; t.state.x = "ab"; t.state.x += "cd"; - expect(t.flush()).toEqual([["a", ["x"], "abcd"]]); + const ops = t.flush(); + expect(ops).toEqual([["s", ["x"], "abcd"]]); + expect(apply(structuredClone(initial), ops)).toEqual(t.state); + }); + + it("may publish a redundant batch when mutations restore the prior value", () => { + const initial = { x: 1, xs: [1, 2] }; + const t = track(structuredClone(initial)); + t.flush(); + t.state.x = 2; + t.state.x = 1; + t.state.xs.reverse(); + t.state.xs.reverse(); + const ops = t.flush(); + expect(ops.length).toBeGreaterThan(0); + expect(apply(structuredClone(initial), ops)).toEqual(t.state); }); it("is linear in the number of ops", () => { @@ -682,6 +775,20 @@ describe("flush-time minimization", () => { expect(ops).toHaveLength(3); expect(apply({ a: { b: 1 }, x: 1 }, ops)).toEqual(t.state); }); + + it("bounds long structural windows by collapsing them to a base", () => { + const initial = { xs: [{ value: 0 }, { value: 1 }] }; + const t = track(structuredClone(initial)); + t.flush(); + for (let i = 0; i < 5000; i++) { + t.state.xs[0]!.value = i; + t.state.xs.shift(); + t.state.xs.push({ value: i }); + } + const ops = t.flush(); + expect(isBase(ops)).toBe(true); + expect(apply(structuredClone(initial), ops)).toEqual(t.state); + }); }); describe("flush", () => { @@ -740,6 +847,47 @@ describe("flush", () => { expect(out).toBeDefined(); }); + it("clones tracked values without invoking inherited setters", () => { + const root = {} as Record; + Object.defineProperty(root, "trap", { value: 1, writable: true, enumerable: true, configurable: true }); + Object.defineProperty(Object.prototype, "trap", { + set() { + throw new Error("inherited setter ran"); + }, + configurable: true, + }); + try { + expect(() => track(root).flush()).not.toThrow(); + } finally { + delete (Object.prototype as Record).trap; + } + + let arrayOps: Op[] | undefined; + Object.defineProperty(Array.prototype, "0", { + set() { + throw new Error("inherited array setter ran"); + }, + configurable: true, + }); + try { + arrayOps = track({ values: [1, 2] }).flush(); + } finally { + delete (Array.prototype as unknown as Record)[0]; + } + expect(arrayOps).toEqual([["r", { values: [1, 2] }]]); + }); + + it("replaces the safe parent when an assigned object removes a reserved value key", () => { + const initial: { value: Record } = { + value: JSON.parse('{"constructor":{"label":"data"},"x":1}') as Record, + }; + const t = track(structuredClone(initial)); + t.flush(); + t.state.value = { x: 2 }; + const ops = decoder().decode(JSON.parse(JSON.stringify(encoder().encode(t.flush()))) as WireOp[]); + expect(applyImmutable(structuredClone(initial), ops)).toEqual(t.state); + }); + it("clones and reads 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()); @@ -1031,7 +1179,7 @@ describe("codec: path interning and arity omission", () => { }); }); -describe("property: flush-time tracking", () => { +describe("property: operation-log tracking", () => { it("converges across mixed nested writes, replacements, and array mutations", () => { type State = { rows: { text: string; count: number }[]; meta: { revision: number } }; let seed = 0x5eed1234; diff --git a/packages/chord/test/services.test.ts b/packages/chord/test/services.test.ts index 3a640252c..e3f4ceaca 100644 --- a/packages/chord/test/services.test.ts +++ b/packages/chord/test/services.test.ts @@ -111,6 +111,20 @@ describe("remote services", () => { unsubscribe(); }); + test("publishes a redundant update when tracked mutations restore the prior value", () => { + const state = replicatedState({ value: 1 }); + const deliveries: Array<{ kind: string; sequence: number }> = []; + state.subscribe((_value, _context, delivery) => deliveries.push(delivery)); + state.state.value = 2; + state.state.value = 1; + state.publish(BACKGROUND_CONTEXT); + expect(state.value).toEqual({ value: 1 }); + expect(deliveries).toEqual([ + { kind: "hydrate", sequence: 0 }, + { kind: "update", sequence: 1 }, + ]); + }); + test("flushes pending mutations before hydrating a new state subscriber", () => { const state = replicatedState({ entries: [{ id: "one" }] }); const first: { entries: { id: string }[] }[] = [];