sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

feat(chord): consolidate immutable delta tracking

Browse source
This commit is contained in:
Mario Zechner 2026-09-23 02:22:52 +02:00
parent 898ab80405
commit 9a139c62bf
27 changed files with 2321 additions and 3815 deletions
Download patch file Download diff file Expand all files Collapse all files

90
packages/agent/src/harness/pico3/legacy-tracker.ts Normal file
View file

@ -0,0 +1,90 @@
import {
type Change,
type Tracker as ChordTracker,
type JsonValue,
type Op,
type Prepared,
track as trackChord,
} from "@earendil-works/chord/delta";
/** Temporary compatibility surface for Pico3's flush-based document handling. */
export interface Tracker<T extends object> {
readonly state: T;
readonly target: T;
readonly dirty: boolean;
flush(): Op[];
rebase(): void;
}
class LegacyTracker<T extends object> implements Tracker<T> {
private readonly tracker: ChordTracker<T>;
private change: Change<T> | undefined;
private wrappers = new WeakMap<object, object>();
private forceBase = true;
constructor(initial: T) {
this.tracker = trackChord(initial);
}
get state(): T {
this.change ??= this.tracker.beginChange();
return this.wrap(this.change.state as T);
}
get target(): T {
return this.tracker.value;
}
get dirty(): boolean {
return this.forceBase || this.change !== undefined;
}
flush(): Op[] {
let prepared: Prepared<T> | undefined;
if (this.change !== undefined) {
const change = this.change;
this.change = undefined;
this.wrappers = new WeakMap();
prepared = change.prepare();
// Pico3 adopts before storage. A storage failure faults the owning Session.
this.tracker.adopt(prepared);
}
if (this.forceBase) {
this.forceBase = false;
return [["r", this.tracker.value as unknown as JsonValue]];
}
return (prepared?.ops ?? []) as Op[];
}
rebase(): void {
this.forceBase = true;
}
private wrap<V extends object>(draft: V): V {
const cached = this.wrappers.get(draft);
if (cached !== undefined) return cached as V;
const tracker = this;
const proxy = new Proxy(draft, {
get(target, key, receiver) {
const value = Reflect.get(target, key, receiver);
if (typeof value === "function") {
return (...args: unknown[]) => {
const result = Reflect.apply(value, target, args);
return typeof result === "object" && result !== null ? tracker.wrap(result) : result;
};
}
return typeof value === "object" && value !== null ? tracker.wrap(value) : value;
},
set(target, key, value, receiver) {
if (value === undefined && !Array.isArray(target)) return Reflect.deleteProperty(target, key);
return Reflect.set(target, key, value, receiver);
},
});
this.wrappers.set(draft, proxy);
return proxy;
}
}
export function track<T extends object>(initial: T): Tracker<T> {
return new LegacyTracker(initial);
}

9
packages/agent/src/harness/pico3/session.ts
View file

@ -1,7 +1,8 @@
import { type Context, isJsonValue } from "@earendil-works/chord";
import { createContextKey, withContextValue, withoutAbortSignal } from "@earendil-works/chord/context";
import { isBase, type Op, type Tracker, track } from "@earendil-works/chord/delta";
import { isBase, type Op } from "@earendil-works/chord/delta";
import { deriveContext } from "./context.ts";
import { type Tracker, track } from "./legacy-tracker.ts";
import { Membrane } from "./membrane.ts";
import {
type AnyKind,
@ -509,7 +510,7 @@ class TxImpl implements CoreTx {
snapshot(ref: { doc: "session" }): SessionState;
snapshot(ref: DocRef): object {
if (ref.doc !== "session") this.assertScope(ref.conversationId, "snapshot");
const value = structuredClone(this.doc(ref).target) as JsonObject;
const value = plain(this.doc(ref).state) as JsonObject;
if (ref.doc !== "session") {
this.defaults.fill(ref.doc, value);
if (this.invoker.type === "task") {
@ -637,8 +638,8 @@ class TxImpl implements CoreTx {
get: (key: string) => {
this.assertSurfaceActive();
const { doc, fallback } = definition(key);
const document = this.doc({ doc, conversationId }).target as JsonObject;
return key in document ? structuredClone(document[key]) : structuredClone(fallback);
const document = this.doc({ doc, conversationId }).state as JsonObject;
return key in document ? plain(document[key]) : fallback === undefined ? undefined : plain(fallback);
},
set: (key: string, value: JsonValue) => {
this.assertSurfaceActive();

19
packages/agent/src/harness/pico3/view.ts
View file

@ -1,4 +1,5 @@
import { applyImmutable, type Op, type Tracker, track } from "@earendil-works/chord/delta";
import { applyImmutable, type Op } from "@earendil-works/chord/delta";
import { type Tracker, track } from "./legacy-tracker.ts";
import type { CommitResult, Session } from "./session.ts";
import type {
AnyKind,
@ -86,8 +87,8 @@ export class ViewManager {
.map((scoped) => scoped.event);
if (appended.length === 0 && documentChanges.length === 0 && !taskChanged && events.length === 0) continue;
const entryOps = applyEntries(record.tracker, appended);
const state = record.tracker.state;
applyEntries(state.entries, appended);
const rewindableChanged = documentChanges.some(
(change) => change.ref.doc === "rewindable" && change.ref.conversationId === record.conversationId,
);
@ -117,7 +118,7 @@ export class ViewManager {
syncRecord(state.plugins, this.plugins(rewindable!, sticky!, session));
}
const ops = [...entryOps, ...record.tracker.flush()];
const ops = record.tracker.flush();
if (ops.length === 0 && events.length === 0) continue;
const envelope = Object.freeze({
revision: ++record.revision,
@ -334,23 +335,15 @@ class WatchImpl implements Watch {
}
}
function applyEntries(tracker: Tracker<ConversationView>, appended: EntryList): Op[] {
const ops: Op[] = [];
const entries = tracker.state.entries;
function applyEntries(entries: EntryList, appended: EntryList): void {
for (const entry of appended) {
if (entry.head !== undefined) {
const retained = entries.findIndex((candidate) => candidate.id >= entry.head!);
const remove = retained < 0 ? entries.length : retained;
if (remove > 0) {
ops.push(...tracker.flush());
entries.splice(0, remove);
tracker.flush();
ops.push(["p", ["entries"], 0, remove, []]);
}
if (remove > 0) entries.splice(0, remove);
}
entries.push(structuredClone(entry));
}
return ops;
}
function inputIds(task: Task): Id[] {

17
packages/agent/test/harness/pico3/spec-view-events.test.ts
View file

@ -170,7 +170,7 @@ test("late joiner reconstructs streaming generation state without replayed event
watch.stop();
});
test("a head commit is represented by transcript splice ops and matching entry/head events", async () => {
test("a self-head commit rewrites one transcript entry with matching entry/head events", async () => {
const env = await open({});
onTestFinished(() => env.close());
await env.root.write({ kind: "before" }, ctx);
@ -181,10 +181,17 @@ test("a head commit is represented by transcript splice ops and matching entry/h
});
await env.root.reset("fresh", ctx);
assert.ok(envelope);
const entryOps = envelope.ops.filter((op) => op[0] === "p" && op[1]?.[0] === "entries");
assert.equal(entryOps.length, 2);
assert.equal(entryOps[0]?.[2], 0);
assert.ok((entryOps[0]?.[3] as number) > 0);
const entryOps = envelope.ops.filter((op) => op[0] !== "r" && op[1][0] === "entries");
assert.deepEqual(
entryOps
.map((op) => {
if (op[0] !== "s") throw new Error(`Unexpected transcript operation ${op[0]}`);
assert.equal(op[1][1], 0);
return op[1][2];
})
.sort(),
["head", "id", "kind", "model"],
);
assert.deepEqual(
envelope.events.map((event) => event.type),
["head.moved", "entry.added"],

22
packages/chord/src/api.ts
View file

@ -1,8 +1,9 @@
import { FacetKernel } from "./facets/host.ts";
import { disposeLoadedFacets } from "./facets/loader.ts";
import { RemoteServiceBindingImpl } from "./services/consumer.ts";
import { MutableReplicatedStateImpl } from "./services/state.ts";
import { attachReplicatedStateSource, MutableReplicatedStateImpl } from "./services/state.ts";
import type {
AttachedReplicatedState,
Facet,
FacetHost,
FacetLoader,
@ -12,6 +13,8 @@ import type {
RemoteServiceBinding,
RemoteServiceBindingOptions,
RemoteServiceContract,
ReplicatedStateSource,
ReplicatedStateSourceOptions,
Service,
} from "./types.ts";
@ -85,6 +88,19 @@ export function createRemoteServiceBinding(options: RemoteServiceBindingOptions)
return new RemoteServiceBindingImpl(options);
}
export function replicatedState<T extends object>(initial: T): MutableReplicatedState<T> {
return new MutableReplicatedStateImpl(initial);
export function replicatedState<T>(
source: ReplicatedStateSource<T>,
options?: ReplicatedStateSourceOptions,
): AttachedReplicatedState<T>;
export function replicatedState<T extends object>(initial: T): MutableReplicatedState<T>;
export function replicatedState(
initialOrSource: object | ReplicatedStateSource<unknown>,
options?: ReplicatedStateSourceOptions,
): MutableReplicatedState<object> | AttachedReplicatedState<unknown> {
if (isReplicatedStateSource(initialOrSource)) return attachReplicatedStateSource(initialOrSource, options);
return new MutableReplicatedStateImpl(initialOrSource);
}
function isReplicatedStateSource(value: object): value is ReplicatedStateSource<unknown> {
return typeof (value as { readonly attach?: unknown }).attach === "function";
}

523
packages/chord/src/delta/diff.ts Normal file
View file

@ -0,0 +1,523 @@
import type { JsonValue } from "../types.ts";
import { type NonEmptyPath, type Op, overlap, type Path, RESERVED_SEGMENTS, type Seg } from "./index.ts";
const DEFAULT_OVERLAP_SCAN = 65_536;
const MAX_DELTA_OPERATIONS = 4_096;
const overflowedBatches = new WeakSet<Op[]>();
const emitOperation = (operations: Op[], operation: Op): void => {
if (overflowedBatches.has(operations)) return;
if (operations.length >= MAX_DELTA_OPERATIONS) {
overflowedBatches.add(operations);
return;
}
operations.push(operation);
};
const isContainer = (value: JsonValue): value is JsonValue[] | Record<string, JsonValue> =>
value !== null && typeof value === "object";
const emitSet = (path: Path, value: JsonValue, operations: Op[]): void => {
if (path.length === 0) emitOperation(operations, ["r", value]);
else emitOperation(operations, ["s", path as NonEmptyPath, value]);
};
const equalJson = (left: JsonValue, right: JsonValue): boolean => {
if (left === right) return true;
if (!isContainer(left) || !isContainer(right) || Array.isArray(left) !== Array.isArray(right)) return false;
if (Array.isArray(left)) {
const other = right as JsonValue[];
if (left.length !== other.length) return false;
for (let index = 0; index < left.length; index++) {
if (!equalJson(left[index]!, other[index]!)) return false;
}
return true;
}
const other = right as Record<string, JsonValue>;
const keys = Object.keys(left);
if (keys.length !== Object.keys(other).length) return false;
for (const key of keys) {
if (!Object.hasOwn(other, key) || !equalJson(left[key]!, other[key]!)) return false;
}
return true;
};
const permutation = (before: readonly JsonValue[], after: readonly JsonValue[]): number[] | undefined => {
if (before.length !== after.length) return undefined;
const positions = new Map<JsonValue, { indices: number[]; used: number }>();
for (let index = 0; index < before.length; index++) {
const value = before[index]!;
const entry = positions.get(value);
if (entry === undefined) positions.set(value, { indices: [index], used: 0 });
else entry.indices.push(index);
}
const result = new Array<number>(after.length);
for (let index = 0; index < after.length; index++) {
const entry = positions.get(after[index]!);
if (entry === undefined || entry.used === entry.indices.length) return undefined;
result[index] = entry.indices[entry.used++]!;
}
return result;
};
const emitString = (before: string, after: string, path: NonEmptyPath, operations: Op[]): void => {
if (before === after) return;
if (after.length > before.length && after.slice(0, before.length) === before) {
emitOperation(operations, ["a", path, after.slice(before.length)]);
return;
}
const shared = overlap(before, after, DEFAULT_OVERLAP_SCAN);
if (shared === 0) {
emitOperation(operations, ["s", path, after]);
return;
}
emitOperation(operations, ["t", path, before.length - shared]);
if (after.length > shared) emitOperation(operations, ["a", path, after.slice(shared)]);
};
type ArrayMatch = readonly [before: number, after: number];
type MatchCandidate = {
before: number;
after: number;
previous: number;
};
const MAX_IDENTITY_CANDIDATES = 200_000;
const MAX_SEMANTIC_CELLS = 65_536;
const sameValue = (left: JsonValue, right: JsonValue): boolean => left === right || equalJson(left, right);
const lcsMatches = (
before: readonly JsonValue[],
after: readonly JsonValue[],
equal: (left: JsonValue, right: JsonValue) => boolean,
maxCells: number,
): ArrayMatch[] | undefined => {
if (before.length === 0 || after.length === 0) return [];
if (before.length * after.length > maxCells) return undefined;
const width = after.length + 1;
const lengths = new Uint32Array((before.length + 1) * width);
for (let left = before.length - 1; left >= 0; left--) {
for (let right = after.length - 1; right >= 0; right--) {
const at = left * width + right;
lengths[at] = equal(before[left]!, after[right]!)
? lengths[(left + 1) * width + right + 1]! + 1
: Math.max(lengths[(left + 1) * width + right]!, lengths[left * width + right + 1]!);
}
}
const matches: ArrayMatch[] = [];
let left = 0;
let right = 0;
while (left < before.length && right < after.length) {
if (
equal(before[left]!, after[right]!) &&
lengths[left * width + right] === lengths[(left + 1) * width + right + 1]! + 1
) {
matches.push([left++, right++]);
} else if (lengths[(left + 1) * width + right]! >= lengths[left * width + right + 1]!) left += 1;
else right += 1;
}
return matches;
};
const semanticallyAligned = (left: JsonValue, right: JsonValue): boolean => {
if (sameValue(left, right)) return true;
if (!isContainer(left) || !isContainer(right) || Array.isArray(left) !== Array.isArray(right)) return false;
if (Array.isArray(left) && Array.isArray(right)) {
if (left.length !== right.length) return false;
return left.some((value, index) => isContainer(value) && value === right[index]);
}
const leftObject = left as Record<string, JsonValue>;
const rightObject = right as Record<string, JsonValue>;
for (const key of Object.keys(leftObject)) {
const value = leftObject[key];
if (isContainer(value) && Object.hasOwn(rightObject, key) && value === rightObject[key]) return true;
}
return false;
};
const lowerBound = (values: readonly number[], value: number): number => {
let low = 0;
let high = values.length;
while (low < high) {
const middle = (low + high) >>> 1;
if (values[middle]! < value) low = middle + 1;
else high = middle;
}
return low;
};
const identitySubsequence = (
before: readonly JsonValue[],
beforeStart: number,
beforeEnd: number,
after: readonly JsonValue[],
afterStart: number,
afterEnd: number,
): ArrayMatch[] | undefined => {
const beforeCount = beforeEnd - beforeStart;
const afterCount = afterEnd - afterStart;
const matches: ArrayMatch[] = [];
if (afterCount < beforeCount) {
let beforeIndex = beforeStart;
for (let afterIndex = afterStart; afterIndex < afterEnd; afterIndex++) {
while (beforeIndex < beforeEnd && before[beforeIndex] !== after[afterIndex]) beforeIndex += 1;
if (beforeIndex === beforeEnd) return undefined;
matches.push([beforeIndex++, afterIndex]);
}
return matches;
}
if (beforeCount < afterCount) {
let afterIndex = afterStart;
for (let beforeIndex = beforeStart; beforeIndex < beforeEnd; beforeIndex++) {
while (afterIndex < afterEnd && before[beforeIndex] !== after[afterIndex]) afterIndex += 1;
if (afterIndex === afterEnd) return undefined;
matches.push([beforeIndex, afterIndex++]);
}
return matches;
}
return undefined;
};
const greedyIdentityAnchors = (
positions: ReadonlyMap<JsonValue, readonly number[]>,
after: readonly JsonValue[],
afterStart: number,
afterEnd: number,
): ArrayMatch[] => {
const matches: ArrayMatch[] = [];
let previous = -1;
for (let afterIndex = afterStart; afterIndex < afterEnd; afterIndex++) {
const candidates = positions.get(after[afterIndex]!);
if (candidates === undefined) continue;
const at = lowerBound(candidates, previous + 1);
const beforeIndex = candidates[at];
if (beforeIndex === undefined) continue;
matches.push([beforeIndex, afterIndex]);
previous = beforeIndex;
}
return matches;
};
const identityAnchors = (
before: readonly JsonValue[],
beforeStart: number,
beforeEnd: number,
after: readonly JsonValue[],
afterStart: number,
afterEnd: number,
): ArrayMatch[] => {
const positions = new Map<JsonValue, number[]>();
for (let index = beforeStart; index < beforeEnd; index++) {
const value = before[index]!;
const existing = positions.get(value);
if (existing === undefined) positions.set(value, [index]);
else existing.push(index);
}
let candidateCount = 0;
for (let index = afterStart; index < afterEnd; index++) {
candidateCount += positions.get(after[index]!)?.length ?? 0;
if (candidateCount > MAX_IDENTITY_CANDIDATES) {
return greedyIdentityAnchors(positions, after, afterStart, afterEnd);
}
}
if (candidateCount === 0) return [];
const candidates: MatchCandidate[] = [];
const tails: number[] = [];
const tailValues: number[] = [];
for (let afterIndex = afterStart; afterIndex < afterEnd; afterIndex++) {
const beforePositions = positions.get(after[afterIndex]!);
if (beforePositions === undefined) continue;
for (let index = beforePositions.length - 1; index >= 0; index--) {
const beforeIndex = beforePositions[index]!;
const at = lowerBound(tailValues, beforeIndex);
const candidateIndex = candidates.length;
candidates.push({ before: beforeIndex, after: afterIndex, previous: at === 0 ? -1 : tails[at - 1]! });
tails[at] = candidateIndex;
tailValues[at] = beforeIndex;
}
}
const matches: ArrayMatch[] = [];
let candidateIndex = tails[tails.length - 1] ?? -1;
while (candidateIndex >= 0) {
const candidate = candidates[candidateIndex]!;
matches.push([candidate.before, candidate.after]);
candidateIndex = candidate.previous;
}
matches.reverse();
return matches;
};
const processArrayMatches = (
before: JsonValue[],
after: JsonValue[],
path: Path,
operations: Op[],
beforeStart: number,
beforeEnd: number,
afterStart: number,
afterEnd: number,
outputStart: number,
matches: readonly ArrayMatch[],
): void => {
let beforeAt = beforeStart;
let afterAt = afterStart;
let outputAt = outputStart;
for (const [beforeMatch, afterMatch] of matches) {
if (overflowedBatches.has(operations)) return;
diffArrayRegion(before, after, path, operations, beforeAt, beforeMatch, afterAt, afterMatch, outputAt);
outputAt += afterMatch - afterAt;
if (!sameValue(before[beforeMatch]!, after[afterMatch]!)) {
diffValue(before[beforeMatch]!, after[afterMatch]!, [...path, outputAt], operations);
}
outputAt += 1;
beforeAt = beforeMatch + 1;
afterAt = afterMatch + 1;
}
if (!overflowedBatches.has(operations)) {
diffArrayRegion(before, after, path, operations, beforeAt, beforeEnd, afterAt, afterEnd, outputAt);
}
};
function diffArrayRegion(
before: JsonValue[],
after: JsonValue[],
path: Path,
operations: Op[],
beforeStart: number,
beforeEnd: number,
afterStart: number,
afterEnd: number,
outputStart: number,
): void {
if (overflowedBatches.has(operations)) return;
while (beforeStart < beforeEnd && afterStart < afterEnd && sameValue(before[beforeStart]!, after[afterStart]!)) {
beforeStart += 1;
afterStart += 1;
outputStart += 1;
}
while (beforeStart < beforeEnd && afterStart < afterEnd && sameValue(before[beforeEnd - 1]!, after[afterEnd - 1]!)) {
beforeEnd -= 1;
afterEnd -= 1;
}
const beforeCount = beforeEnd - beforeStart;
const afterCount = afterEnd - afterStart;
if (beforeCount === 0 && afterCount === 0) return;
if (beforeCount === 0 || afterCount === 0) {
emitOperation(operations, ["p", path, outputStart, beforeCount, after.slice(afterStart, afterEnd)]);
return;
}
if (beforeCount === afterCount) {
const positional: ArrayMatch[] = [];
for (let offset = 0; offset < beforeCount; offset++) {
if (sameValue(before[beforeStart + offset]!, after[afterStart + offset]!)) {
positional.push([beforeStart + offset, afterStart + offset]);
}
}
if (positional.length > 0) {
processArrayMatches(
before,
after,
path,
operations,
beforeStart,
beforeEnd,
afterStart,
afterEnd,
outputStart,
positional,
);
return;
}
}
const subsequence = identitySubsequence(before, beforeStart, beforeEnd, after, afterStart, afterEnd);
if (subsequence !== undefined && subsequence.length > 0) {
processArrayMatches(
before,
after,
path,
operations,
beforeStart,
beforeEnd,
afterStart,
afterEnd,
outputStart,
subsequence,
);
return;
}
const identity = identityAnchors(before, beforeStart, beforeEnd, after, afterStart, afterEnd);
if (identity.length > 0) {
processArrayMatches(
before,
after,
path,
operations,
beforeStart,
beforeEnd,
afterStart,
afterEnd,
outputStart,
identity,
);
return;
}
const semantic = lcsMatches(
before.slice(beforeStart, beforeEnd),
after.slice(afterStart, afterEnd),
semanticallyAligned,
MAX_SEMANTIC_CELLS,
);
if (semantic !== undefined && semantic.length > 0) {
const absolute = semantic.map(
([beforeIndex, afterIndex]) => [beforeStart + beforeIndex, afterStart + afterIndex] as const,
);
processArrayMatches(
before,
after,
path,
operations,
beforeStart,
beforeEnd,
afterStart,
afterEnd,
outputStart,
absolute,
);
return;
}
if (beforeCount === 1 && afterCount === 1) {
diffValue(before[beforeStart]!, after[afterStart]!, [...path, outputStart], operations);
return;
}
emitOperation(operations, ["p", path, outputStart, beforeCount, after.slice(afterStart, afterEnd)]);
}
const diffArray = (before: JsonValue[], after: JsonValue[], path: Path, operations: Op[]): void => {
if (before === after || equalJson(before, after)) return;
if (
before.length === after.length &&
before.length > 1 &&
!sameValue(before[0]!, after[0]!) &&
!sameValue(before[before.length - 1]!, after[after.length - 1]!)
) {
const order = permutation(before, after);
if (order !== undefined) {
emitOperation(operations, ["m", path, order]);
return;
}
}
diffArrayRegion(before, after, path, operations, 0, before.length, 0, after.length, 0);
};
const diffObject = (
before: Record<string, JsonValue>,
after: Record<string, JsonValue>,
path: Path,
operations: Op[],
): void => {
const beforeKeys = Object.keys(before);
const afterKeys = Object.keys(after);
if ([...beforeKeys, ...afterKeys].some((key) => RESERVED_SEGMENTS.has(key))) {
if (!equalJson(before, after)) emitSet(path, after, operations);
return;
}
for (const key of afterKeys) {
if (overflowedBatches.has(operations)) return;
if (Object.hasOwn(before, key)) diffValue(before[key]!, after[key]!, [...path, key], operations);
else emitSet([...path, key], after[key]!, operations);
}
for (const key of beforeKeys) {
if (overflowedBatches.has(operations)) return;
if (!Object.hasOwn(after, key)) emitOperation(operations, ["d", [...path, key] as unknown as NonEmptyPath]);
}
};
const diffValue = (before: JsonValue, after: JsonValue, path: Path, operations: Op[]): void => {
if (before === after || overflowedBatches.has(operations)) return;
if (typeof before === "string" && typeof after === "string" && path.length > 0) {
emitString(before, after, path as NonEmptyPath, operations);
return;
}
if (Array.isArray(before) && Array.isArray(after)) {
diffArray(before, after, path, operations);
return;
}
if (isContainer(before) && isContainer(after) && !Array.isArray(before) && !Array.isArray(after)) {
diffObject(before, after, path, operations);
return;
}
emitSet(path, after, operations);
};
const jsonCost = (value: JsonValue): number => {
if (value === null) return 4;
if (typeof value === "string") return value.length + 2;
if (typeof value === "number") return String(value).length;
if (typeof value === "boolean") return value ? 4 : 5;
if (Array.isArray(value)) {
let cost = 2;
for (let index = 0; index < value.length; index++) cost += jsonCost(value[index]!) + (index === 0 ? 0 : 1);
return cost;
}
let cost = 2;
let index = 0;
for (const key of Object.keys(value)) {
cost += key.length + 3 + jsonCost(value[key]!) + (index++ === 0 ? 0 : 1);
}
return cost;
};
const pathCost = (path: Path): number => {
let cost = 2;
for (let index = 0; index < path.length; index++) {
const segment: Seg = path[index]!;
cost += (typeof segment === "string" ? segment.length + 2 : String(segment).length) + (index === 0 ? 0 : 1);
}
return cost;
};
const operationCost = (operation: Op): number => {
switch (operation[0]) {
case "r":
return 6 + jsonCost(operation[1]);
case "s":
return 7 + pathCost(operation[1]) + jsonCost(operation[2]);
case "d":
return 6 + pathCost(operation[1]);
case "a":
return 7 + pathCost(operation[1]) + operation[2].length + 2;
case "t":
return 7 + pathCost(operation[1]) + String(operation[2]).length;
case "p":
return (
10 +
pathCost(operation[1]) +
String(operation[2]).length +
String(operation[3]).length +
jsonCost(operation[4])
);
case "m":
return 7 + pathCost(operation[1]) + jsonCost(operation[2]);
}
};
/** Compute a compact operation batch from two immutable JSON revisions. */
export function diffRevisions(before: JsonValue, after: JsonValue): Op[] {
const operations: Op[] = [];
diffValue(before, after, [], operations);
if (overflowedBatches.has(operations)) return [["r", after]];
if (operations.length === 0 || operations[0]?.[0] === "r") return operations;
let deltaCost = 2;
for (const operation of operations) deltaCost += operationCost(operation) + 1;
if (deltaCost < 65_536) return operations;
const snapshotCost = jsonCost(after) + 6;
return deltaCost >= snapshotCost ? [["r", after]] : operations;
}

88
packages/chord/src/state/draft.ts → packages/chord/src/delta/draft.ts
View file

@ -35,9 +35,16 @@ type DraftContext = {
const RELEASED_BASE: Container = Object.freeze({});
const ARRAY_MUTATORS = new Set(["push", "pop", "shift", "unshift", "splice", "sort", "reverse", "fill", "copyWithin"]);
const MAX_NATIVE_ARRAY_INSERT_ITEMS = 10_000;
/** Run a mutation recipe and return its value with transaction ownership metadata. */
export function produceWithMetadata<T extends object>(base: T, recipe: (draft: Draft<T>) => void): ProduceMetadata<T> {
export type DraftTransaction<T extends object> = {
readonly state: Draft<T>;
finish(): ProduceMetadata<T>;
abort(): void;
};
/** Create a revocable copy-on-write draft whose lifetime is controlled by its owner. */
export function createDraftTransaction<T extends object>(base: T): DraftTransaction<T> {
const context: DraftContext = {
active: true,
created: [],
@ -46,17 +53,8 @@ export function produceWithMetadata<T extends object>(base: T, recipe: (draft: D
states: new WeakMap<object, DraftState>(),
};
const root = getState(context, base as Container);
try {
const outcome = (recipe as (draft: Draft<T>) => unknown)(root.proxy as Draft<T>);
if (isPromiseLike(outcome)) {
void Promise.resolve(outcome).catch(() => undefined);
throw new TypeError("Replicated state change callbacks must be synchronous");
}
return {
value: finalize(root, new Set<DraftState>(), new WeakMap<DraftState, Container>()) as T,
owned: context.owned,
};
} finally {
const release = (): void => {
assertActive(context);
context.active = false;
for (const state of context.created) {
state.base = RELEASED_BASE;
@ -68,10 +66,45 @@ export function produceWithMetadata<T extends object>(base: T, recipe: (draft: D
context.created.length = 0;
context.proxies = new WeakMap();
context.states = new WeakMap();
};
return {
state: root.proxy as Draft<T>,
finish() {
assertActive(context);
try {
return {
value: finalize(root, new Set<DraftState>(), new WeakMap<DraftState, Container>()) as T,
owned: context.owned,
};
} finally {
release();
}
},
abort: release,
};
}
/** @internal Compatibility helper for the existing replicated-state service. */
export function produceWithMetadata<T extends object>(base: T, recipe: (draft: Draft<T>) => void): ProduceMetadata<T> {
const transaction = createDraftTransaction(base);
try {
const outcome = (recipe as (draft: Draft<T>) => unknown)(transaction.state);
if (isPromiseLike(outcome)) {
void Promise.resolve(outcome).catch(() => undefined);
throw new TypeError("Replicated state change callbacks must be synchronous");
}
return transaction.finish();
} catch (error) {
try {
transaction.abort();
} catch {
// finish() already released the transaction before propagating an error.
}
throw error;
}
}
/** Run a mutation recipe against a copy-on-write draft of `base`. */
/** @internal Compatibility helper for the existing replicated-state service. */
export function produce<T extends object>(base: T, recipe: (draft: Draft<T>) => void): T {
return produceWithMetadata(base, recipe).value;
}
@ -238,11 +271,14 @@ function mutateArray(state: DraftState, target: unknown[], property: string, arg
return draftValue(context, Reflect.apply(Array.prototype.shift, target, []));
case "unshift": {
const items = cloneArrayItems(args, context);
if (hasInheritedGrowthIndex(target, target.length + items.length)) {
spliceArray(target, items, 0, 0, context);
return target.length;
if (
items.length <= MAX_NATIVE_ARRAY_INSERT_ITEMS &&
!hasInheritedGrowthIndex(target, target.length + items.length)
) {
return Reflect.apply(Array.prototype.unshift, target, items);
}
return Reflect.apply(Array.prototype.unshift, target, items);
spliceArray(target, items, 0, 0, context);
return target.length;
}
case "splice": {
const length = target.length;
@ -255,12 +291,14 @@ function mutateArray(state: DraftState, target: unknown[], property: string, arg
: Math.min(Math.max(toIntegerOrInfinity(args[1]), 0), length - start);
const items = cloneArrayItems(args.slice(2), context);
if (
items.length <= MAX_NATIVE_ARRAY_INSERT_ITEMS &&
!mayRunCoercionCode(args[0]) &&
!mayRunCoercionCode(args[1]) &&
!hasInheritedGrowthIndex(target, length - remove + items.length)
) {
const removed = Reflect.apply(Array.prototype.splice, target, [start, remove, ...items]) as unknown[];
return removed.map((item) => draftValue(context, item));
return Reflect.apply(Array.prototype.splice, target, [start, remove, ...items]).map((item) =>
draftValue(context, item),
);
}
return spliceArray(target, items, start, remove, context, length);
}
@ -570,8 +608,14 @@ function assertDenseArray(value: unknown[]): void {
throw new TypeError("Draft arrays must be dense and contain only indexed entries");
}
for (let index = 0; index < value.length; index++) {
if (!Object.hasOwn(value, index) || value[index] === undefined) {
throw new TypeError("Draft arrays cannot contain holes or undefined entries");
const descriptor = Object.getOwnPropertyDescriptor(value, String(index));
if (
descriptor === undefined ||
!descriptor.enumerable ||
!("value" in descriptor) ||
descriptor.value === undefined
) {
throw new TypeError("Draft arrays must contain enumerable indexed data properties with defined values");
}
}
}

1334
packages/chord/src/delta/index.ts
View file

File diff suppressed because it is too large Load diff

189
packages/chord/src/delta/tracker.ts Normal file
View file

@ -0,0 +1,189 @@
import type { JsonValue } from "../types.ts";
import { diffRevisions } from "./diff.ts";
import { createDraftTransaction, type Draft, type DraftTransaction } from "./draft.ts";
import type { Op } from "./index.ts";
import { JsonRevisionStore } from "./value.ts";
export interface Prepared<T extends object> {
readonly base: T;
readonly value: T;
readonly ops: readonly Op[];
}
export interface Change<T extends object> {
readonly state: Draft<T>;
prepare(): Prepared<T>;
abort(): void;
}
export interface Tracker<T extends object> {
readonly value: T;
beginChange(): Change<T>;
prepareReplace(value: T): Prepared<T>;
adopt(prepared: Prepared<T>): void;
}
type PreparedMetadata = {
readonly owner: object;
readonly revision: number;
status: "prepared" | "consumed" | "aborted";
};
const preparedMetadata = new WeakMap<object, PreparedMetadata>();
class PreparedImpl<T extends object> implements Prepared<T> {
readonly base: T;
readonly value: T;
readonly ops: readonly Op[];
constructor(metadata: PreparedMetadata, base: T, value: T, ops: readonly Op[]) {
this.base = base;
this.value = value;
this.ops = ops;
preparedMetadata.set(this, metadata);
Object.freeze(this);
}
}
class ChangeImpl<T extends object> implements Change<T> {
readonly state: Draft<T>;
#tracker: TrackerImpl<T> | undefined;
#transaction: DraftTransaction<T> | undefined;
#preparedMetadata: PreparedMetadata | undefined;
#status: "open" | "prepared" | "aborted" = "open";
constructor(tracker: TrackerImpl<T>, transaction: DraftTransaction<T>) {
this.#tracker = tracker;
this.#transaction = transaction;
this.state = transaction.state;
}
prepare(): Prepared<T> {
if (this.#status !== "open") throw new Error("Change has already been settled");
const tracker = this.#tracker!;
const transaction = this.#transaction!;
try {
const prepared = tracker.prepareDraft(transaction);
this.#preparedMetadata = preparedMetadata.get(prepared)!;
this.#status = "prepared";
return prepared;
} catch (error) {
this.#status = "aborted";
throw error;
} finally {
this.#tracker = undefined;
this.#transaction = undefined;
}
}
abort(): void {
if (this.#status === "aborted") return;
if (this.#status === "prepared") {
if (this.#preparedMetadata?.status === "prepared") this.#preparedMetadata.status = "aborted";
this.#preparedMetadata = undefined;
this.#status = "aborted";
return;
}
this.#status = "aborted";
const tracker = this.#tracker!;
const transaction = this.#transaction!;
this.#tracker = undefined;
this.#transaction = undefined;
try {
transaction.abort();
} finally {
tracker.releaseChange();
}
}
}
class TrackerImpl<T extends object> implements Tracker<T> {
readonly #store = new JsonRevisionStore();
readonly #owner = {};
#value: T;
#revision = 0;
#changing = false;
constructor(initial: T) {
this.#value = this.#store.import(initial);
}
get value(): T {
return this.#value;
}
beginChange(): Change<T> {
this.#reserveChange();
try {
return new ChangeImpl(this, createDraftTransaction(this.#value));
} catch (error) {
this.#changing = false;
throw error;
}
}
prepareReplace(value: T): Prepared<T> {
this.#reserveChange();
try {
return this.#prepare(this.#store.import(value));
} finally {
this.#changing = false;
}
}
adopt(prepared: Prepared<T>): void {
const metadata = prepared instanceof PreparedImpl ? preparedMetadata.get(prepared) : undefined;
if (metadata?.owner !== this.#owner) throw new Error("Prepared change belongs to a different tracker");
if (metadata.status === "consumed") throw new Error("Prepared change has already been used");
if (metadata.status === "aborted") throw new Error("Prepared change has been aborted");
if (this.#changing) throw new Error("Cannot adopt while a change is active");
if (metadata.revision !== this.#revision || this.#value !== prepared.base) {
throw new Error("Prepared change is stale");
}
metadata.status = "consumed";
this.#value = prepared.value;
this.#revision += 1;
}
prepareDraft(transaction: DraftTransaction<T>): PreparedImpl<T> {
try {
const produced = transaction.finish();
return this.#prepare(this.#store.commit(produced.value, produced.owned));
} finally {
this.#changing = false;
}
}
releaseChange(): void {
this.#changing = false;
}
#reserveChange(): void {
if (this.#changing) throw new Error("Tracker already has an active change");
this.#changing = true;
}
#prepare(candidate: T): PreparedImpl<T> {
const base = this.#value;
const operations = diffRevisions(base as unknown as JsonValue, candidate as unknown as JsonValue);
const value = operations.length === 0 ? base : candidate;
const ops = freezeOperations(operations);
return new PreparedImpl({ owner: this.#owner, revision: this.#revision, status: "prepared" }, base, value, ops);
}
}
/** Create a tracker around an imported immutable JSON revision. */
export function track<T extends object>(initial: T): Tracker<T> {
return new TrackerImpl(initial);
}
function freezeOperations(operations: Op[]): readonly Op[] {
for (const operation of operations) freezeMetadata(operation);
return Object.freeze(operations);
}
function freezeMetadata(value: unknown): void {
if (value === null || typeof value !== "object" || Object.isFrozen(value)) return;
for (const child of Object.values(value)) freezeMetadata(child);
Object.freeze(value);
}

12
packages/chord/src/state/value.ts → packages/chord/src/delta/value.ts
View file

@ -164,8 +164,16 @@ function assertDenseArray(value: readonly unknown[]): void {
throw new TypeError("Replicated state arrays must be dense and contain only indexed entries");
}
for (let index = 0; index < value.length; index++) {
if (!Object.hasOwn(value, index) || value[index] === undefined) {
throw new TypeError("Replicated state arrays cannot contain holes or undefined entries");
const descriptor = Object.getOwnPropertyDescriptor(value, String(index));
if (
descriptor === undefined ||
!descriptor.enumerable ||
!("value" in descriptor) ||
descriptor.value === undefined
) {
throw new TypeError(
"Replicated state arrays must contain enumerable indexed data properties with defined values",
);
}
}
}

7
packages/chord/src/index.ts
View file

@ -10,6 +10,7 @@ export {
defineService,
replicatedState,
} from "./api.ts";
export type { Draft } from "./delta/index.ts";
export { isJsonValue } from "./json.ts";
export {
isRemoteServiceErrorCode,
@ -46,8 +47,8 @@ export {
type WireServiceProviderUpdate,
type WireServiceSubscriptionSnapshot,
} from "./services/wire.ts";
export type { Draft } from "./state/draft.ts";
export type {
AttachedReplicatedState,
Context,
ContextKey,
Facet,
@ -66,6 +67,10 @@ export type {
RemoteServiceTransport,
ReplicatedState,
ReplicatedStateDelivery,
ReplicatedStateSource,
ReplicatedStateSourceAttachment,
ReplicatedStateSourceFrame,
ReplicatedStateSourceOptions,
Service,
ServiceCall,
ServiceCatalogueEntry,

5
packages/chord/src/services/provider.ts
View file

@ -437,11 +437,12 @@ export class RemoteServiceProvider {
if (member.kind === "method") {
members.push({ name, kind: "method" });
} else {
const snapshot = member.state.snapshot();
members.push({
name,
kind: "state",
sequence: member.state.sequence,
ops: [["r", member.state.value as JsonValue]],
sequence: snapshot.sequence,
ops: [["r", snapshot.value as JsonValue]],
});
}
}

4
packages/chord/src/services/state-internals.ts
View file

@ -2,8 +2,8 @@ import type { Op } from "../delta/index.ts";
import type { Context } from "../types.ts";
export interface ReplicatedStateInternals {
readonly sequence: number;
readonly value: unknown;
/** Atomically capture the immutable value and its matching publication sequence. */
snapshot(): { readonly value: unknown; readonly sequence: number };
subscribe(listener: (ops: readonly Op[], sequence: number, context: Context) => void): () => void;
}

291
packages/chord/src/services/state.ts
View file

@ -1,71 +1,53 @@
import { BACKGROUND_CONTEXT } from "../context/index.ts";
import { applyImmutable, isBase, type Op } from "../delta/index.ts";
import { diffRevisions } from "../state/diff.ts";
import { produceWithMetadata } from "../state/draft.ts";
import { JsonRevisionStore } from "../state/value.ts";
import type { Context, JsonValue, MutableReplicatedState, ReplicatedState, ReplicatedStateDelivery } from "../types.ts";
import { applyImmutable, isBase, type Op, type Prepared, type Tracker, track } from "../delta/index.ts";
import { JsonRevisionStore } from "../delta/value.ts";
import type {
AttachedReplicatedState,
Context,
JsonValue,
MutableReplicatedState,
ReplicatedState,
ReplicatedStateDelivery,
ReplicatedStateSource,
ReplicatedStateSourceAttachment,
ReplicatedStateSourceFrame,
ReplicatedStateSourceOptions,
} from "../types.ts";
import { registerReplicatedStateInternals } from "./state-internals.ts";
type StateListener<T> = (value: T, context: Context, delivery: ReplicatedStateDelivery) => void;
type SourceListener = (ops: readonly Op[], sequence: number, context: Context) => void;
type Publication<T> = {
value: T;
ops: readonly Op[];
sequence: number;
context: Context;
readonly value: T;
readonly ops: readonly Op[];
readonly sequence: number;
readonly context: Context;
};
export class MutableReplicatedStateImpl<T extends object> implements MutableReplicatedState<T> {
readonly #listeners = new Map<(value: T, context: Context, delivery: ReplicatedStateDelivery) => void, number>();
readonly #sourceListeners = new Set<(ops: readonly Op[], sequence: number, context: Context) => void>();
readonly #store = new JsonRevisionStore();
/** Maintains local publication order independently of how revisions are produced. */
class ReplicatedStatePublisher<T> {
readonly #listeners = new Map<StateListener<T>, number>();
readonly #sourceListeners = new Set<SourceListener>();
readonly #publications: Publication<T>[] = [];
#value: T;
#sequence = 0;
#changing = false;
#delivering = false;
constructor(initial: T) {
this.#value = this.#store.import(initial);
const thisSource = this;
registerReplicatedStateInternals(this, {
get sequence() {
return thisSource.#sequence;
},
get value() {
return thisSource.#value;
},
subscribe: (listener) => {
thisSource.#sourceListeners.add(listener);
return () => thisSource.#sourceListeners.delete(listener);
},
});
this.#value = initial;
}
get value(): T {
return this.#value;
}
change(context: Context, mutate: Parameters<MutableReplicatedState<T>["change"]>[1]): void {
if (this.#changing) throw new Error("Replicated state cannot be changed reentrantly from a change callback");
this.#changing = true;
let next: T;
try {
const produced = produceWithMetadata(this.#value, mutate);
if (produced.value === this.#value) return;
next = this.#store.commit(produced.value, produced.owned);
} finally {
this.#changing = false;
}
this.#commit(next, context);
snapshot(): { readonly value: T; readonly sequence: number } {
return { value: this.#value, sequence: this.#sequence };
}
replace(context: Context, value: T): void {
if (this.#changing) throw new Error("Replicated state cannot be replaced from a change callback");
this.#commit(this.#store.import(value), context);
}
subscribe(listener: (value: T, context: Context, delivery: ReplicatedStateDelivery) => void): () => void {
const sequence = this.#sequence;
const value = this.#value;
subscribe(listener: StateListener<T>): () => void {
const { value, sequence } = this.snapshot();
this.#listeners.set(listener, sequence);
try {
listener(value, serviceDeliveryContext(), { kind: "hydrate", sequence });
@ -76,13 +58,18 @@ export class MutableReplicatedStateImpl<T extends object> implements MutableRepl
return () => this.#listeners.delete(listener);
}
#commit(next: T, context: Context): void {
const ops = diffRevisions(this.#value as unknown as JsonValue, next as unknown as JsonValue);
if (ops.length === 0) return;
this.#value = next;
subscribeSource(listener: SourceListener): () => void {
this.#sourceListeners.add(listener);
return () => this.#sourceListeners.delete(listener);
}
/** Publish an already-prepared immutable revision and return isolated listener failures. */
publish(value: T, ops: readonly Op[], context: Context): unknown[] {
this.#value = value;
this.#sequence += 1;
this.#publications.push({ value: next, ops, sequence: this.#sequence, context });
if (this.#delivering) return;
this.#publications.push({ value, ops, sequence: this.#sequence, context });
if (this.#delivering) return [];
this.#delivering = true;
const errors: unknown[] = [];
try {
@ -111,14 +98,172 @@ export class MutableReplicatedStateImpl<T extends object> implements MutableRepl
} finally {
this.#delivering = false;
}
if (errors.length === 1) throw errors[0];
if (errors.length > 1) throw new AggregateError(errors, "Replicated state listeners failed");
return errors;
}
}
export class MutableReplicatedStateImpl<T extends object> implements MutableReplicatedState<T> {
readonly #tracker: Tracker<T>;
readonly #publisher: ReplicatedStatePublisher<T>;
#changing = false;
constructor(initial: T) {
this.#tracker = track(initial);
this.#publisher = new ReplicatedStatePublisher(this.#tracker.value);
registerReplicatedStateInternals(this, {
snapshot: () => this.#publisher.snapshot(),
subscribe: (listener) => this.#publisher.subscribeSource(listener),
});
}
get value(): T {
return this.#tracker.value;
}
change(context: Context, mutate: Parameters<MutableReplicatedState<T>["change"]>[1]): void {
if (this.#changing) throw new Error("Replicated state cannot be changed reentrantly from a change callback");
this.#changing = true;
let prepared: Prepared<T>;
try {
const change = this.#tracker.beginChange();
try {
const outcome = (mutate as (draft: typeof change.state) => unknown)(change.state);
if (isPromiseLike(outcome)) {
void Promise.resolve(outcome).catch(() => undefined);
throw new TypeError("Replicated state change callbacks must be synchronous");
}
prepared = change.prepare();
} catch (error) {
change.abort();
throw error;
}
} finally {
this.#changing = false;
}
this.#tracker.adopt(prepared);
if (prepared.ops.length === 0) return;
throwCollectedErrors(
this.#publisher.publish(prepared.value, prepared.ops, context),
"Replicated state listeners failed",
);
}
replace(context: Context, value: T): void {
if (this.#changing) throw new Error("Replicated state cannot be replaced from a change callback");
const prepared = this.#tracker.prepareReplace(value);
this.#tracker.adopt(prepared);
if (prepared.ops.length === 0) return;
throwCollectedErrors(
this.#publisher.publish(prepared.value, prepared.ops, context),
"Replicated state listeners failed",
);
}
subscribe(listener: StateListener<T>): () => void {
return this.#publisher.subscribe(listener);
}
}
class AttachedReplicatedStateImpl<T> implements AttachedReplicatedState<T> {
readonly #publisher: ReplicatedStatePublisher<T>;
readonly #attachment: ReplicatedStateSourceAttachment<T>;
readonly #reportError: (error: Error) => void;
#cursor: number;
#disposed = false;
constructor(attachment: ReplicatedStateSourceAttachment<T>, options: ReplicatedStateSourceOptions) {
const { snapshot } = attachment;
assertCursor(snapshot.cursor, "snapshot");
this.#attachment = attachment;
this.#cursor = snapshot.cursor;
this.#publisher = new ReplicatedStatePublisher(snapshot.value);
this.#reportError = options.onError ?? reportErrorAsync;
registerReplicatedStateInternals(this, {
snapshot: () => this.#publisher.snapshot(),
subscribe: (listener) => this.#publisher.subscribeSource(listener),
});
}
get value(): T {
return this.#publisher.value;
}
subscribe(listener: StateListener<T>): () => void {
return this.#publisher.subscribe(listener);
}
activate(): void {
this.#attachment.activate((frame) => this.#receive(frame));
}
dispose(): void {
if (this.#disposed) return;
this.#disposed = true;
this.#attachment.dispose();
}
#receive(frame: ReplicatedStateSourceFrame<T>): void {
if (this.#disposed) return;
try {
assertCursor(frame.cursor, "frame");
const expected = this.#cursor + 1;
if (frame.cursor !== expected) {
throw new Error(`Replicated state source cursor has a gap: expected ${expected}, received ${frame.cursor}`);
}
this.#cursor = frame.cursor;
const errors = this.#publisher.publish(frame.value, frame.ops, frame.context);
if (errors.length === 1) this.#report(errors[0]);
else if (errors.length > 1) this.#report(new AggregateError(errors, "Replicated state listeners failed"));
} catch (error) {
this.#fail(toError(error));
}
}
#fail(error: Error): void {
if (this.#disposed) return;
this.#disposed = true;
try {
this.#attachment.dispose();
} catch (disposeError) {
this.#report(new AggregateError([error, disposeError], "Replicated state source contract failed"));
return;
}
this.#report(error);
}
#report(error: unknown): void {
try {
this.#reportError(toError(error));
} catch (reportError) {
reportErrorAsync(toError(reportError));
}
}
}
/** Attach a publication-only replicated state to one authoritative immutable source stream. */
export function attachReplicatedStateSource<T>(
source: ReplicatedStateSource<T>,
options: ReplicatedStateSourceOptions = {},
): AttachedReplicatedState<T> {
const attachment = source.attach();
try {
const state = new AttachedReplicatedStateImpl(attachment, options);
state.activate();
return state;
} catch (error) {
try {
attachment.dispose();
} catch (disposeError) {
throw new AggregateError([error, disposeError], "Failed to attach replicated state source");
}
throw error;
}
}
/** A cold read-only state used by service consumers until a complete snapshot arrives. */
export class ReplicatedStateReplica<T extends JsonValue = JsonValue> implements ReplicatedState<T> {
readonly #listeners = new Set<(value: T, context: Context, delivery: ReplicatedStateDelivery) => void>();
readonly #listeners = new Set<StateListener<T>>();
readonly #reportError: (error: Error) => void;
readonly #store = new JsonRevisionStore();
#value: T | undefined;
@ -132,7 +277,7 @@ export class ReplicatedStateReplica<T extends JsonValue = JsonValue> implements
return this.#value;
}
subscribe(listener: (value: T, context: Context, delivery: ReplicatedStateDelivery) => void): () => void {
subscribe(listener: StateListener<T>): () => void {
this.#listeners.add(listener);
if (this.#value !== undefined) {
this.#deliver(listener, this.#value, serviceDeliveryContext(), {
@ -187,12 +332,7 @@ export class ReplicatedStateReplica<T extends JsonValue = JsonValue> implements
for (const listener of this.#listeners) this.#deliver(listener, this.#value, context, delivery);
}
#deliver(
listener: (value: T, context: Context, delivery: ReplicatedStateDelivery) => void,
value: T,
context: Context,
delivery: ReplicatedStateDelivery,
): void {
#deliver(listener: StateListener<T>, value: T, context: Context, delivery: ReplicatedStateDelivery): void {
try {
listener(value, context, delivery);
} catch (error) {
@ -207,6 +347,29 @@ export function serviceDeliveryContext(): Context {
return BACKGROUND_CONTEXT;
}
function assertCursor(cursor: number, kind: "snapshot" | "frame"): void {
if (!Number.isSafeInteger(cursor))
throw new TypeError(`Replicated state source ${kind} cursor must be a safe integer`);
}
function isPromiseLike(value: unknown): value is PromiseLike<unknown> {
return (
((typeof value === "object" && value !== null) || typeof value === "function") &&
typeof (value as { then?: unknown }).then === "function"
);
}
function throwCollectedErrors(errors: readonly unknown[], message: string): void {
if (errors.length === 1) throw errors[0];
if (errors.length > 1) throw new AggregateError(errors, message);
}
function reportErrorAsync(error: Error): void {
queueMicrotask(() => {
throw error;
});
}
function toError(error: unknown): Error {
return error instanceof Error ? error : new Error(String(error));
}

249
packages/chord/src/state/diff.ts
View file

@ -1,249 +0,0 @@
import { type NonEmptyPath, type Op, overlap, type Path, RESERVED_SEGMENTS, type Seg } from "../delta/index.ts";
import type { JsonValue } from "../types.ts";
const DEFAULT_OVERLAP_SCAN = 65_536;
const MAX_DELTA_OPERATIONS = 4_096;
const isContainer = (value: JsonValue): value is JsonValue[] | Record<string, JsonValue> =>
value !== null && typeof value === "object";
const emitSet = (path: Path, value: JsonValue, operations: Op[]): void => {
if (path.length === 0) operations.push(["r", value]);
else operations.push(["s", path as NonEmptyPath, value]);
};
const equalJson = (left: JsonValue, right: JsonValue): boolean => {
if (left === right) return true;
if (!isContainer(left) || !isContainer(right) || Array.isArray(left) !== Array.isArray(right)) return false;
if (Array.isArray(left)) {
const other = right as JsonValue[];
if (left.length !== other.length) return false;
for (let index = 0; index < left.length; index++) {
if (!equalJson(left[index]!, other[index]!)) return false;
}
return true;
}
const other = right as Record<string, JsonValue>;
const keys = Object.keys(left);
if (keys.length !== Object.keys(other).length) return false;
for (const key of keys) {
if (!Object.hasOwn(other, key) || !equalJson(left[key]!, other[key]!)) return false;
}
return true;
};
const permutation = (before: readonly JsonValue[], after: readonly JsonValue[]): number[] | undefined => {
if (before.length !== after.length) return undefined;
const positions = new Map<JsonValue, { indices: number[]; used: number }>();
for (let index = 0; index < before.length; index++) {
const value = before[index]!;
const entry = positions.get(value);
if (entry === undefined) positions.set(value, { indices: [index], used: 0 });
else entry.indices.push(index);
}
const result = new Array<number>(after.length);
for (let index = 0; index < after.length; index++) {
const entry = positions.get(after[index]!);
if (entry === undefined || entry.used === entry.indices.length) return undefined;
result[index] = entry.indices[entry.used++]!;
}
return result;
};
const emitString = (before: string, after: string, path: NonEmptyPath, operations: Op[]): void => {
if (before === after) return;
if (after.length > before.length && after.slice(0, before.length) === before) {
operations.push(["a", path, after.slice(before.length)]);
return;
}
const shared = overlap(before, after, DEFAULT_OVERLAP_SCAN);
if (shared === 0) {
operations.push(["s", path, after]);
return;
}
operations.push(["t", path, before.length - shared]);
if (after.length > shared) operations.push(["a", path, after.slice(shared)]);
};
const emitSameLengthShift = (
before: readonly JsonValue[],
after: readonly JsonValue[],
path: Path,
operations: Op[],
): boolean => {
const length = before.length;
if (length < 2) return false;
let candidates = 0;
for (let removed = 1; removed < length && candidates < 16; removed++) {
if (before[removed] !== after[0]) continue;
candidates += 1;
let matches = true;
for (let index = removed; index < length; index++) {
if (before[index] !== after[index - removed]) {
matches = false;
break;
}
}
if (!matches) continue;
operations.push(["p", path, 0, removed, []]);
operations.push(["p", path, length - removed, 0, after.slice(length - removed)]);
return true;
}
candidates = 0;
for (let inserted = 1; inserted < length && candidates < 16; inserted++) {
if (after[inserted] !== before[0]) continue;
candidates += 1;
let matches = true;
for (let index = inserted; index < length; index++) {
if (after[index] !== before[index - inserted]) {
matches = false;
break;
}
}
if (!matches) continue;
operations.push(["p", path, 0, 0, after.slice(0, inserted)]);
operations.push(["p", path, length, inserted, []]);
return true;
}
return false;
};
const diffArray = (before: JsonValue[], after: JsonValue[], path: Path, operations: Op[]): void => {
if (before.length === after.length) {
let mismatches = 0;
for (let index = 0; index < after.length; index++) {
if (before[index] !== after[index]) mismatches += 1;
}
if (mismatches === 0) return;
if (mismatches > 1) {
const order = permutation(before, after);
if (order !== undefined) {
operations.push(["m", path, order]);
return;
}
if (emitSameLengthShift(before, after, path, operations)) return;
}
for (let index = 0; index < after.length; index++) {
if (operations.length > MAX_DELTA_OPERATIONS) return;
diffValue(before[index]!, after[index]!, [...path, index], operations);
}
return;
}
let prefix = 0;
const shortest = Math.min(before.length, after.length);
while (prefix < shortest && equalJson(before[prefix]!, after[prefix]!)) prefix += 1;
let suffix = 0;
while (
suffix < shortest - prefix &&
equalJson(before[before.length - 1 - suffix]!, after[after.length - 1 - suffix]!)
) {
suffix += 1;
}
operations.push(["p", path, prefix, before.length - prefix - suffix, after.slice(prefix, after.length - suffix)]);
};
const diffObject = (
before: Record<string, JsonValue>,
after: Record<string, JsonValue>,
path: Path,
operations: Op[],
): void => {
const beforeKeys = Object.keys(before);
const afterKeys = Object.keys(after);
if ([...beforeKeys, ...afterKeys].some((key) => RESERVED_SEGMENTS.has(key))) {
if (!equalJson(before, after)) emitSet(path, after, operations);
return;
}
for (const key of afterKeys) {
if (operations.length > MAX_DELTA_OPERATIONS) return;
if (Object.hasOwn(before, key)) diffValue(before[key]!, after[key]!, [...path, key], operations);
else emitSet([...path, key], after[key]!, operations);
}
for (const key of beforeKeys) {
if (operations.length > MAX_DELTA_OPERATIONS) return;
if (!Object.hasOwn(after, key)) operations.push(["d", [...path, key] as unknown as NonEmptyPath]);
}
};
const diffValue = (before: JsonValue, after: JsonValue, path: Path, operations: Op[]): void => {
if (before === after || operations.length > MAX_DELTA_OPERATIONS) return;
if (typeof before === "string" && typeof after === "string" && path.length > 0) {
emitString(before, after, path as NonEmptyPath, operations);
return;
}
if (Array.isArray(before) && Array.isArray(after)) {
diffArray(before, after, path, operations);
return;
}
if (isContainer(before) && isContainer(after) && !Array.isArray(before) && !Array.isArray(after)) {
diffObject(before, after, path, operations);
return;
}
emitSet(path, after, operations);
};
const jsonCost = (value: JsonValue): number => {
if (value === null) return 4;
if (typeof value === "string") return value.length + 2;
if (typeof value === "number") return String(value).length;
if (typeof value === "boolean") return value ? 4 : 5;
if (Array.isArray(value)) {
let cost = 2;
for (let index = 0; index < value.length; index++) cost += jsonCost(value[index]!) + (index === 0 ? 0 : 1);
return cost;
}
let cost = 2;
let index = 0;
for (const key of Object.keys(value)) {
cost += key.length + 3 + jsonCost(value[key]!) + (index++ === 0 ? 0 : 1);
}
return cost;
};
const pathCost = (path: Path): number => {
let cost = 2;
for (let index = 0; index < path.length; index++) {
const segment: Seg = path[index]!;
cost += (typeof segment === "string" ? segment.length + 2 : String(segment).length) + (index === 0 ? 0 : 1);
}
return cost;
};
const operationCost = (operation: Op): number => {
switch (operation[0]) {
case "r":
return 6 + jsonCost(operation[1]);
case "s":
return 7 + pathCost(operation[1]) + jsonCost(operation[2]);
case "d":
return 6 + pathCost(operation[1]);
case "a":
return 7 + pathCost(operation[1]) + operation[2].length + 2;
case "t":
return 7 + pathCost(operation[1]) + String(operation[2]).length;
case "p":
return (
10 +
pathCost(operation[1]) +
String(operation[2]).length +
String(operation[3]).length +
jsonCost(operation[4])
);
case "m":
return 7 + pathCost(operation[1]) + jsonCost(operation[2]);
}
};
/** Compute a compact operation batch from two immutable JSON revisions. */
export function diffRevisions(before: JsonValue, after: JsonValue): Op[] {
const operations: Op[] = [];
diffValue(before, after, [], operations);
if (operations.length > MAX_DELTA_OPERATIONS) return [["r", after]];
if (operations.length === 0 || operations[0]?.[0] === "r") return operations;
let deltaCost = 2;
for (const operation of operations) deltaCost += operationCost(operation) + 1;
if (deltaCost < 65_536) return operations;
const snapshotCost = jsonCost(after) + 6;
return deltaCost >= snapshotCost ? [["r", after]] : operations;
}

51
packages/chord/src/types.ts
View file

@ -1,6 +1,5 @@
import type { Op } from "./delta/index.ts";
import type { Draft, Op } from "./delta/index.ts";
import type { RemoteServiceProvider } from "./services/provider.ts";
import type { Draft } from "./state/draft.ts";
export type { RemoteServiceError } from "./services/errors.ts";
export type { RemoteServiceProvider } from "./services/provider.ts";
@ -59,6 +58,54 @@ export interface MutableReplicatedState<T extends object> extends ReplicatedStat
replace(context: Context, value: T): void;
}
/** One immutable authoritative revision committed after an attachment snapshot. */
export interface ReplicatedStateSourceFrame<T> {
/** Monotonic source cursor. The first frame after a snapshot must be `snapshot.cursor + 1`. */
readonly cursor: number;
/** The exact immutable value produced by this commit. */
readonly value: T;
/** The exact immutable operation batch that produced `value` from the preceding source revision. */
readonly ops: readonly Op[];
readonly context: Context;
}
export interface ReplicatedStateSourceAttachment<T> {
/** Fixed immutable snapshot captured at the atomic attachment boundary. */
readonly snapshot: { readonly value: T; readonly cursor: number };
/**
* Install the sole listener and synchronously drain every buffered frame in source commit order.
* This method is single-use. After it begins, every new committed frame must also be delivered in order
* until disposal, including commits made reentrantly while a prior frame is being delivered.
*/
activate(listener: (frame: ReplicatedStateSourceFrame<T>) => void): void;
/** Stop delivery and release source resources. Disposal must be idempotent. */
dispose(): void;
}
/**
* An authoritative immutable revision source.
*
* `attach()` must synchronously and atomically capture one snapshot and register the returned attachment to buffer
* every later committed frame. The snapshot must include every commit before that boundary; buffered frames must
* include every commit after it, with no overlap or gap. Snapshot values, frame values, and operation batches are
* immutable and remain valid after delivery. Chord only publishes these references; it never applies or re-diffs them.
*/
export interface ReplicatedStateSource<T> {
attach(): ReplicatedStateSourceAttachment<T>;
}
export interface ReplicatedStateSourceOptions {
/** Receives source-contract and publication-listener failures without throwing them into the source. */
readonly onError?: (error: Error) => void;
}
/** A synchronously hydrated publication-only state backed by one source attachment. */
export interface AttachedReplicatedState<T> extends ReplicatedState<T> {
readonly value: T;
/** Idempotently release the source attachment. The last published value remains readable. */
dispose(): void;
}
declare const SERVICE_TYPE: unique symbol;
export type ServiceMode = "singleton" | "keyed";

124
packages/chord/test/delta-clone.test.ts
View file

@ -1,76 +1,80 @@
import { describe, expect, it } from "vitest";
import { apply, track } from "../src/delta/index.ts";
describe("delta payload cloning", () => {
it("deeply detaches mixed objects and arrays in a base snapshot", () => {
describe("tracker ownership", () => {
it("deeply detaches and freezes the imported revision", () => {
const input = {
point: { x: 3, y: 7, pressure: 0.1 },
rows: [{ values: [0, false, null, "text", { n: 1 }] }],
};
const producer = track(input);
const copy = apply<typeof input>(undefined, producer.flush());
expect(copy).toEqual(input);
expect(copy).not.toBe(input);
expect(copy.point).not.toBe(input.point);
expect(copy.rows).not.toBe(input.rows);
expect(copy.rows[0]).not.toBe(input.rows[0]);
expect(copy.rows[0].values).not.toBe(input.rows[0].values);
expect(copy.rows[0].values[4]).not.toBe(input.rows[0].values[4]);
copy.point.pressure = 0.9;
(copy.rows[0].values[4] as { n: number }).n = 2;
expect(input.point.pressure).toBe(0.1);
expect(input.rows[0].values[4]).toEqual({ n: 1 });
producer.state.point.x = 99;
expect(copy.point.x).toBe(3);
const tracker = track(input);
expect(tracker.value).toEqual(input);
expect(tracker.value).not.toBe(input);
expect(tracker.value.point).not.toBe(input.point);
expect(tracker.value.rows[0]!.values[4]).not.toBe(input.rows[0]!.values[4]);
expect(Object.isFrozen(tracker.value.rows[0]!.values)).toBe(true);
input.point.x = 99;
expect(tracker.value.point.x).toBe(3);
});
it("preserves null prototypes at the root and inside arrays", () => {
it("preserves null prototypes and expands aliases", () => {
type Dictionary = { enabled: boolean; child: { n: number } };
const dictionary = Object.assign(Object.create(null) as Dictionary, { enabled: true, child: { n: 1 } });
const rootCopy = apply<Dictionary>(undefined, track(dictionary).flush());
const nestedCopy = apply<{ rows: Dictionary[] }>(undefined, track({ rows: [dictionary] }).flush());
expect(Object.getPrototypeOf(rootCopy)).toBeNull();
expect(Object.getPrototypeOf(nestedCopy.rows[0])).toBeNull();
expect(Object.getPrototypeOf(rootCopy.child)).toBe(Object.prototype);
expect(rootCopy.child).not.toBe(dictionary.child);
expect(nestedCopy.rows[0].child).not.toBe(dictionary.child);
rootCopy.child.n = 2;
expect(dictionary.child.n).toBe(1);
expect(nestedCopy.rows[0].child.n).toBe(1);
});
it("recursively clones own data properties shadowing inherited names", () => {
const input = { constructor: { n: 1 }, toString: [{ n: 2 }], hasOwnProperty: { n: 3 } };
const copy = apply<typeof input>(undefined, track(input).flush());
expect(Object.getPrototypeOf(copy)).toBe(Object.prototype);
for (const key of Object.keys(input)) expect(Object.hasOwn(copy, key)).toBe(true);
expect(copy.constructor).not.toBe(input.constructor);
expect(copy.toString[0]).not.toBe(input.toString[0]);
expect(copy.hasOwnProperty).not.toBe(input.hasOwnProperty);
copy.constructor.n = 9;
expect(input.constructor.n).toBe(1);
});
it("continues expanding shared input values into independent JSON payloads", () => {
const shared = { nested: [{ n: 1 }] };
const copy = apply<{ a: typeof shared; b: typeof shared }>(undefined, track({ a: shared, b: shared }).flush());
expect(copy.a).toEqual(copy.b);
expect(copy.a).not.toBe(copy.b);
expect(copy.a.nested).not.toBe(copy.b.nested);
expect(copy.a.nested[0]).not.toBe(copy.b.nested[0]);
const tracker = track({ dictionary, left: shared, right: shared });
expect(Object.getPrototypeOf(tracker.value.dictionary)).toBeNull();
expect(tracker.value.dictionary).not.toBe(dictionary);
expect(tracker.value.left).not.toBe(tracker.value.right);
expect(tracker.value.left.nested[0]).not.toBe(tracker.value.right.nested[0]);
});
it("keeps published insertion payloads independent of subsequent producer edits", () => {
type Item = { point: { x: number; y: number; pressure: number }; tags: { name: string }[] };
const producer = track({ items: [] as Item[] });
const replica = apply<typeof producer.state>(undefined, producer.flush());
producer.state.items.push({ point: { x: 1, y: 2, pressure: 0.5 }, tags: [{ name: "original" }] });
const batch = producer.flush();
producer.state.items[0].point.x = 9;
producer.state.items[0].tags[0].name = "changed";
const published = apply(replica, batch);
expect(published.items[0].point.x).toBe(1);
expect(published.items[0].tags[0].name).toBe("original");
expect(published.items[0].point).not.toBe(producer.target.items[0].point);
it("copies assigned and inserted values immediately", () => {
const tracker = track({ rows: [] as { nested: { value: number } }[] });
const assigned = { nested: { value: 1 } };
const change = tracker.beginChange();
change.state.rows.push(assigned);
assigned.nested.value = 9;
expect(change.state.rows[0]!.nested.value).toBe(1);
const prepared = change.prepare();
expect(prepared.value.rows[0]!.nested.value).toBe(1);
const replica = apply(structuredClone(prepared.base), prepared.ops);
expect(replica).toEqual(prepared.value);
});
it("rejects cycles, accessors, classes, and non-finite numbers", () => {
const cyclic: { self?: object } = {};
cyclic.self = cyclic;
expect(() => track(cyclic)).toThrow(/cycles/);
expect(() => track({ value: Number.NaN })).toThrow(/strict JSON/);
expect(() => track({ value: new Date() })).toThrow(/plain objects/);
const accessor = Object.defineProperty({}, "value", { enumerable: true, get: () => 1 });
expect(() => track(accessor)).toThrow(/data properties/);
});
it("rejects indexed accessors and non-enumerable array entries without invoking getters", () => {
let reads = 0;
const accessor: number[] = [];
Object.defineProperty(accessor, "0", {
enumerable: true,
configurable: true,
get() {
reads += 1;
return 1;
},
});
expect(() => track({ values: accessor })).toThrow(/data properties/);
expect(reads).toBe(0);
const hidden: number[] = [];
Object.defineProperty(hidden, "0", { value: 1, enumerable: false, configurable: true, writable: true });
expect(() => track({ values: hidden })).toThrow(/enumerable/);
const tracker = track({ values: [] as number[] });
const change = tracker.beginChange();
expect(() => {
change.state.values = accessor;
}).toThrow(/data properties/);
expect(reads).toBe(0);
change.abort();
});
});

18
packages/chord/test/delta-retention.test.ts
View file

@ -2,21 +2,9 @@ import { spawnSync } from "node:child_process";
import { fileURLToPath } from "node:url";
import { describe, expect, it } from "vitest";
describe("delta proxy retention across GC jobs", () => {
it.each([
"unheld",
"tracker-disposal",
"raw-root-rebase",
"array-subtree-reattach",
"held-descendant",
"held-identity",
"held-array-reacquire",
"alias-rewrap",
"detach-reinsert",
"root-array-windows",
"blocked-view",
])(
"preserves semantics without retaining unused proxies: %s",
describe("delta change retention across GC jobs", () => {
it.each(["aborted-payload", "unadopted-prepared", "draft-proxies", "settled-lifecycle"])(
"releases %s",
(scenario) => {
const child = spawnSync(
process.execPath,

275
packages/chord/test/delta-retention.worker.ts
View file

@ -1,223 +1,82 @@
// Isolated --expose-gc scenarios; invoked by delta-retention.test.ts.
import assert from "node:assert/strict";
import { setImmediate } from "node:timers/promises";
import { apply, type Tracker, track } from "../src/delta/index.ts";
import { type Tracker, track } from "../src/delta/index.ts";
async function collect(): Promise<void> {
const retainedSettledObjects: object[] = [];
async function collect(ref: WeakRef<object>): Promise<void> {
assert.ok(global.gc, "worker requires --expose-gc");
// WeakRef keeps dereferenced/created targets alive through the current job.
await setImmediate();
global.gc();
await setImmediate();
global.gc();
await setImmediate();
}
async function waitForCollection(ref: WeakRef<object>): Promise<void> {
for (let attempt = 0; attempt < 20; attempt++) {
await collect();
for (let attempt = 0; attempt < 30; attempt++) {
await setImmediate();
global.gc();
if (ref.deref() === undefined) return;
}
assert.fail("an otherwise unreferenced child proxy is still retained");
}
function sampleChild(t: Tracker<{ rows: { n: number }[] }>): WeakRef<object> {
return new WeakRef(t.state.rows[0]);
}
function fillCaches(state: { n: number }[]): WeakRef<object> {
let sum = 0;
for (let i = 0; i < state.length; i++) sum += state[i].n;
assert.ok(sum >= 0);
return new WeakRef(state[0]);
}
function select(t: Tracker<{ rows: { text: string }[]; selected: { text: string } | null }>): WeakRef<object> {
const child = t.state.rows[0];
t.state.selected = child;
return new WeakRef(child);
assert.fail("released change data is still retained");
}
// Separate frames prevent VM temporaries from pinning the objects under test.
function disposableTracker(): [WeakRef<object>, WeakRef<object>] {
const t = track({ rows: [{ n: 1 }] });
t.flush();
void t.state.rows[0].n;
return [new WeakRef(t.state), new WeakRef(t.target)];
function abortedPayload(tracker: Tracker<{ payload: { rows: { value: number }[] } | null }>): WeakRef<object> {
const change = tracker.beginChange();
change.state.payload = { rows: Array.from({ length: 50_000 }, (_, value) => ({ value })) };
const reference = new WeakRef(change.state.payload.rows);
change.abort();
return reference;
}
function blockedView(t: Tracker<{ holder: { constructor: { n: number } }; safe: { n: number } }>): WeakRef<object> {
const view = t.state.holder.constructor;
assert.equal(view.n, 1);
return new WeakRef(view);
function unadoptedPrepared(tracker: Tracker<{ payload: { rows: { value: number }[] } | null }>): WeakRef<object> {
const change = tracker.beginChange();
change.state.payload = { rows: Array.from({ length: 50_000 }, (_, value) => ({ value })) };
const prepared = change.prepare();
return new WeakRef(prepared.value.payload!.rows);
}
function draftProxies(tracker: Tracker<{ payload: { rows: { value: number }[] } | null }>): WeakRef<object> {
const change = tracker.beginChange();
change.state.payload = { rows: [{ value: 1 }] };
const reference = new WeakRef(change.state.payload.rows[0]!);
change.abort();
return reference;
}
function settledLifecycle(): { tracker: WeakRef<object>; future: WeakRef<object> } {
const tracker = track<{ payload: { rows: { value: number }[] } | null }>({ payload: null });
const change = tracker.beginChange();
change.state.payload = { rows: [{ value: 1 }] };
const retainedPrepared = change.prepare();
tracker.adopt(retainedPrepared);
retainedSettledObjects.push(change, retainedPrepared);
const futurePrepared = tracker.prepareReplace({
payload: { rows: Array.from({ length: 50_000 }, (_, value) => ({ value })) },
});
tracker.adopt(futurePrepared);
return {
tracker: new WeakRef(tracker),
future: new WeakRef(futurePrepared.value.payload!.rows),
};
}
const scenario = process.argv[2];
switch (scenario) {
case "tracker-disposal": {
const [proxy, raw] = disposableTracker();
await waitForCollection(proxy);
await waitForCollection(raw);
break;
if (scenario === "settled-lifecycle") {
const references = settledLifecycle();
await collect(references.tracker);
await collect(references.future);
} else {
const tracker = track<{ payload: { rows: { value: number }[] } | null }>({ payload: null });
let reference: WeakRef<object>;
switch (scenario) {
case "aborted-payload":
reference = abortedPayload(tracker);
break;
case "unadopted-prepared":
reference = unadoptedPrepared(tracker);
break;
case "draft-proxies":
reference = draftProxies(tracker);
break;
default:
throw new Error(`unknown retention scenario: ${scenario}`);
}
case "raw-root-rebase": {
const t = track([1]);
let replica = apply<number[]>(undefined, t.flush());
t.state = t.target;
replica = apply(replica, t.flush());
t.state.push(2);
replica = apply(replica, t.flush());
assert.deepEqual(replica, t.target);
break;
}
case "array-subtree-reattach": {
const t = track({ rows: [[{ nested: { n: 1 } }, { nested: { n: 2 } }], []] });
let replica = apply<typeof t.state>(undefined, t.flush());
const held = t.state.rows[0];
const leaf = held[0].nested;
held.reverse(); // clears the array's index cache, but not held descendants
replica = apply(replica, t.flush());
t.state.rows.shift();
replica = apply(replica, t.flush());
await collect();
t.state.rows.push(held);
replica = apply(replica, t.flush());
await collect();
leaf.n = 42;
replica = apply(replica, t.flush());
assert.deepEqual(replica, t.target);
break;
}
case "unheld": {
const t = track({ rows: [{ n: 1 }] });
t.flush();
const ref = sampleChild(t);
await waitForCollection(ref);
assert.equal(t.target.rows[0].n, 1);
let replica = apply<typeof t.state>(undefined, [["r", structuredClone(t.target)]]);
t.state.rows[0].n = 2;
replica = apply(replica, t.flush());
assert.deepEqual(replica, t.target);
break;
}
case "held-descendant": {
const t = track({
rows: [
{ nested: { text: "a" }, values: [1] },
{ nested: { text: "b" }, values: [2] },
],
});
let replica = apply<typeof t.state>(undefined, t.flush());
const held = t.state.rows[1].nested;
const values = t.state.rows[1].values;
for (let i = 0; i < 8; i++) {
await collect();
t.state.rows.unshift({ nested: { text: `head-${i}` }, values: [] });
replica = apply(replica, t.flush());
await collect();
held.text += "x";
values.push(i);
assert.equal(t.state.rows[i + 2].nested, held);
assert.equal(t.state.rows[i + 2].values, values);
replica = apply(replica, t.flush());
assert.deepEqual(replica, t.target);
}
t.state.rows.reverse();
replica = apply(replica, t.flush());
await collect();
held.text += "reversed";
values.push(99);
replica = apply(replica, t.flush());
assert.deepEqual(replica, t.target);
break;
}
case "held-array-reacquire": {
const t = track({ rows: [[1], [2]] });
let replica = apply<typeof t.state>(undefined, t.flush());
const held = t.state.rows[1];
for (let i = 0; i < 5; i++) {
t.state.rows.unshift([]);
assert.equal(t.state.rows[i + 2], held);
held.push(10 + i);
replica = apply(replica, t.flush());
assert.deepEqual(replica, t.target);
await collect();
}
break;
}
case "held-identity": {
const t = track({ rows: [{ n: 1 }, { n: 2 }] });
let replica = apply<typeof t.state>(undefined, t.flush());
const held = t.state.rows[1];
for (let i = 0; i < 10; i++) {
await collect();
assert.equal(t.state.rows[1], held);
held.n++;
replica = apply(replica, t.flush());
assert.deepEqual(replica, t.target);
}
break;
}
case "alias-rewrap": {
const t = track<{ rows: { text: string }[]; selected: { text: string } | null }>({
rows: [{ text: "a" }],
selected: null,
});
let replica = apply<typeof t.state>(undefined, t.flush());
const ref = select(t);
replica = apply(replica, t.flush());
await waitForCollection(ref);
for (let i = 0; i < 5; i++) {
t.state.selected!.text += "x";
replica = apply(replica, t.flush());
assert.deepEqual(replica, JSON.parse(JSON.stringify(t.target)));
assert.equal(t.state.selected, t.state.rows[0]);
await collect();
}
break;
}
case "detach-reinsert": {
const t = track({ rows: [{ nested: { n: 1 } }, { nested: { n: 2 } }] });
let replica = apply<typeof t.state>(undefined, t.flush());
const held = t.state.rows[1];
t.state.rows.pop();
replica = apply(replica, t.flush());
await collect();
held.nested.n = 9;
assert.deepEqual(t.flush(), []);
t.state.rows.unshift(held);
replica = apply(replica, t.flush());
await collect();
held.nested.n = 10;
replica = apply(replica, t.flush());
assert.deepEqual(replica, JSON.parse(JSON.stringify(t.target)));
break;
}
case "root-array-windows": {
const t = track(Array.from({ length: 10_000 }, (_, n) => ({ n })));
t.flush();
await collect();
const start = process.memoryUsage().heapUsed;
for (let round = 0; round < 12; round++) {
const ref = fillCaches(t.state);
await waitForCollection(ref);
}
for (let i = 0; i < 4; i++) await collect();
const added = process.memoryUsage().heapUsed - start;
// Generous bound for engine/version variance, not a throughput assertion.
assert.ok(added < 12 * 1024 * 1024, `read/GC windows retained ${added} bytes of bookkeeping`);
assert.equal(t.state[9999].n, 9999);
console.log(JSON.stringify({ retainedBookkeepingBytes: added }));
break;
}
case "blocked-view": {
const t = track({ holder: { constructor: { n: 1 } }, safe: { n: 0 } });
t.flush();
const blocked = blockedView(t);
await waitForCollection(blocked);
assert.throws(() => {
t.state.holder.constructor.n = 2;
});
assert.equal(t.state.holder.constructor.n, 1);
assert.deepEqual(t.flush(), []);
break;
}
default:
throw new Error(`unknown retention scenario: ${scenario}`);
await collect(reference);
assert.deepEqual(tracker.value, { payload: null });
}
console.log(JSON.stringify({ scenario, passed: true }));

31
packages/chord/test/delta-traversal.bench.ts
View file

@ -53,18 +53,22 @@ function membrane<T extends object>(input: T): T {
};
return wrap(input) as T;
}
function setup(mode: Mode, input: Drawing): { state: Drawing; flush(): unknown } {
function setup(mode: Mode, input: Drawing): { state: Drawing; finish(): void } {
if (mode === "delta") {
const t = track(input);
return { state: t.state, flush: () => t.flush() };
const tracker = track(input);
const change = tracker.beginChange();
return {
state: change.state,
finish() {
tracker.adopt(change.prepare());
},
};
}
return { state: mode === "proxy" ? membrane(input) : input, flush: () => null };
return { state: mode === "proxy" ? membrane(input) : input, finish() {} };
}
// Keep temporary raw inputs and initial wire batches out of the measuring frame.
// Keep temporary raw inputs out of the measuring frame.
function prepare(mode: Mode, strokes: number, points: number) {
const t = setup(mode, fixture(strokes, points));
t.flush();
return t;
return setup(mode, fixture(strokes, points));
}
function traverse(state: Drawing): number {
let sum = 0;
@ -81,10 +85,18 @@ function traverse(state: Drawing): number {
function warm(mode: Mode): void {
const t = prepare(mode, 10, 10);
for (let i = 0; i < 10; i++) traverse(t.state);
t.finish();
}
function sources() {
return Object.fromEntries(
["../src/delta/index.ts", "./delta-traversal.bench.ts"].map((path) => [
[
"../src/delta/index.ts",
"../src/delta/tracker.ts",
"../src/delta/draft.ts",
"../src/delta/diff.ts",
"../src/delta/value.ts",
"./delta-traversal.bench.ts",
].map((path) => [
path,
createHash("sha256")
.update(readFileSync(new URL(path, import.meta.url)))
@ -120,6 +132,7 @@ async function measure(mode: Mode, strokes: number, points: number, trial: numbe
await gc();
const retained = process.memoryUsage().heapUsed;
assert.equal(t.state.strokes.length, strokes);
t.finish();
return {
mode,
strokes,

1755
packages/chord/test/delta.test.ts
View file

File diff suppressed because it is too large Load diff

52
packages/chord/test/services.test.ts
View file

@ -8,7 +8,10 @@ import {
RemoteServiceProvider,
type RemoteServiceTransport,
type ReplicatedState,
type ReplicatedStateSource,
type ReplicatedStateSourceFrame,
replicatedState,
type ServiceProviderUpdate,
} from "../src/index.ts";
import { createLoopbackServiceTransport } from "./helpers.ts";
@ -39,8 +42,9 @@ interface Echo {
const Echo = defineService<Echo>("test.echo");
type TimelineState = { entries: { id: string }[]; retained: { value: number } };
interface Timeline {
readonly state: ReplicatedState<{ entries: { id: string }[]; retained: { value: number } }>;
readonly state: ReplicatedState<TimelineState>;
}
const Timeline = defineService<Timeline>("test.timeline");
@ -278,6 +282,52 @@ describe("remote services", () => {
provider.dispose();
});
test("publishes authoritative source references through services without re-diffing", () => {
const initial: TimelineState = Object.freeze({ entries: [{ id: "one" }], retained: { value: 1 } });
let publish: ((frame: ReplicatedStateSourceFrame<TimelineState>) => void) | undefined;
let disposed = false;
const source: ReplicatedStateSource<TimelineState> = {
attach() {
return {
snapshot: { value: initial, cursor: 20 },
activate(listener) {
publish = listener;
},
dispose() {
disposed = true;
},
};
},
};
const state = replicatedState(source);
const provider = new RemoteServiceProvider([Timeline]);
provider.provide(Timeline, { state });
const updates: ServiceProviderUpdate[] = [];
const subscription = provider.subscribe(Timeline.id, "singleton", (update) => updates.push(update));
const snapshotMember = subscription.snapshot.instances[0]?.members[0];
expect(snapshotMember?.kind).toBe("state");
if (snapshotMember?.kind === "state") expect(snapshotMember.ops[0]?.[1]).toBe(initial);
subscription.activate();
const next: TimelineState = Object.freeze({
entries: [{ id: "one" }, { id: "two" }],
retained: initial.retained,
});
const ops: ReplicatedStateSourceFrame<TimelineState>["ops"] = Object.freeze([
["p", ["entries"], 1, 0, [{ id: "two" }]],
]);
publish?.({ cursor: 21, value: next, ops, context: BACKGROUND_CONTEXT });
expect(updates).toHaveLength(1);
const update = updates[0];
expect(update?.type).toBe("state");
if (update?.type === "state") expect(update.ops).toBe(ops);
subscription.close();
provider.dispose();
state.dispose();
expect(disposed).toBe(true);
});
test("keeps singleton facades stable when their provider is replaced", async () => {
const provider = new RemoteServiceProvider([Models]);
provider.provide(Models, {

160
packages/chord/test/state-diff.test.ts
View file

@ -1,9 +1,16 @@
import { describe, expect, it } from "vitest";
import { applyImmutable, assertValidOp, assertValidWireOp, decoder, encoder, type Op } from "../src/delta/index.ts";
import { diffRevisions } from "../src/state/diff.ts";
import {
applyImmutable,
assertValidOp,
assertValidWireOp,
decoder,
diffRevisions,
encoder,
type Op,
} from "../src/delta/index.ts";
import type { JsonValue } from "../src/types.ts";
const expectDiff = (before: JsonValue, after: JsonValue, expected: JsonValue): void => {
const expectDiff = (before: JsonValue, after: JsonValue, expected: unknown): void => {
const operations = diffRevisions(before, after);
expect(operations).toEqual(expected);
expect(applyImmutable(before, operations)).toEqual(after);
@ -53,6 +60,13 @@ describe("immutable revision diff", () => {
expectDiff({ values: [a, b, c] }, { values: [c, a, b] }, [["m", ["values"], [2, 0, 1]]]);
});
it("normalizes reordered distinct deeply-equal objects to a no-op", () => {
const first = { nested: { value: 1 } };
const second = { nested: { value: 1 } };
expect(first).not.toBe(second);
expect(diffRevisions({ values: [first, second] }, { values: [second, first] })).toEqual([]);
});
it("validates and encodes permutations", () => {
const operations: Op[] = [
["m", ["values"], [2, 0, 1]],
@ -93,6 +107,146 @@ describe("immutable revision diff", () => {
);
});
it("emits payload-free splices for scattered removals", () => {
const [a, b, c, d, e] = ["a", "b", "c", "d", "e"].map((id) => ({ id }));
expectDiff({ values: [a, b, c, d, e] }, { values: [a, c, e] }, [
["p", ["values"], 1, 1, []],
["p", ["values"], 2, 1, []],
]);
});
it.each([
["front", [1, 2, 3, 4], [2, 3, 4], [["p", ["values"], 0, 1, []]]],
["tail", [1, 2, 3, 4], [1, 2, 3], [["p", ["values"], 3, 1, []]]],
["middle", [1, 2, 3, 4], [1, 3, 4], [["p", ["values"], 1, 1, []]]],
["all", [1, 2, 3, 4], [], [["p", ["values"], 0, 4, []]]],
["none", [1, 2, 3, 4], [1, 2, 3, 4], []],
] as const)("encodes %s removal canonically", (_name, before, after, expected) => {
expectDiff({ values: [...before] }, { values: [...after] }, expected);
});
it("does not field-diff unrelated shifted objects with common fields", () => {
const before = [1, 2, 3].map((value) => ({ type: "row", value }));
const after = [2, 3, 4].map((value) => ({ type: "row", value }));
expectDiff({ values: before }, { values: after }, [
["p", ["values"], 0, 1, []],
["p", ["values"], 2, 0, [{ type: "row", value: 4 }]],
]);
});
it("does not treat coincidental id or key fields as structural identity", () => {
const left = { value: "left" };
const right = { value: "right" };
const before = [left, { id: 1, key: "a", value: "first" }, { id: 2, key: "b", value: "second" }, right];
const replacements = [
{ id: 2, key: "b", value: "edited-second" },
{ id: 1, key: "a", value: "edited-first" },
];
const after = [left, ...replacements, right];
expectDiff({ values: before }, { values: after }, [["p", ["values"], 1, 2, replacements]]);
});
it("combines removals, append, and a shared-subtree survivor edit", () => {
const a = { id: "a", stable: {}, detail: { text: "a" } };
const b = { id: "b", stable: {}, detail: { text: "b" } };
const c = { id: "c", stable: {}, detail: { text: "c" } };
const d = { id: "d", stable: {}, detail: { text: "d" } };
const changedC = { id: "c", stable: c.stable, detail: { text: "changed" } };
const appended = { id: "e", stable: {}, detail: { text: "e" } };
expectDiff({ values: [a, b, c, d] }, { values: [a, changedC, d, appended] }, [
["p", ["values"], 1, 1, []],
["a", ["values", 1, "detail", "text"], "hanged"],
["p", ["values"], 3, 0, [appended]],
]);
});
it.each([256 * 1024, 1024 * 1024])("does not retain a %i-byte removed-neighbor payload", (size) => {
const payload = "x".repeat(size);
const retained = [
{ id: "a", payload },
{ id: "b", payload },
{ id: "c", payload },
{ id: "d", payload },
{ id: "e", payload },
];
const before = Object.freeze({ values: Object.freeze(retained.slice()) });
const after = Object.freeze({ values: Object.freeze([retained[0]!, retained[2]!, retained[4]!]) });
const operations = diffRevisions(before as unknown as JsonValue, after as unknown as JsonValue);
expect(operations).toEqual([
["p", ["values"], 1, 1, []],
["p", ["values"], 2, 1, []],
]);
expect(JSON.stringify(operations).length).toBeLessThan(100);
expect(applyImmutable(before, operations)).toEqual(after);
expect(before.values).toEqual(retained);
});
it.each([1_001, 10_000])("keeps push, pop, and middle removal narrow at %,i items", (size) => {
const values = Array.from({ length: size }, (_, value) => ({ value }));
const appended = { value: size };
expectDiff({ values }, { values: [...values, appended] }, [["p", ["values"], size, 0, [appended]]]);
expectDiff({ values }, { values: values.slice(0, -1) }, [["p", ["values"], size - 1, 1, []]]);
const middle = Math.floor(size / 2);
expectDiff({ values }, { values: [...values.slice(0, middle), ...values.slice(middle + 1)] }, [
["p", ["values"], middle, 1, []],
]);
});
it("keeps forty-thousand-row sparse edits narrow", () => {
const values = Array.from({ length: 40_000 }, (_, value) => ({ value, stable: { value } }));
const after = values.slice();
const changed: number[] = [];
for (let index = 100; index < after.length; index += 400) {
after[index] = { value: -index, stable: values[index]!.stable };
changed.push(index);
}
const operations = diffRevisions({ values }, { values: after });
expect(operations).toEqual(changed.map((index) => ["s", ["values", index, "value"], -index]));
expect(JSON.stringify(operations).length).toBeLessThan(7_500);
expect(applyImmutable({ values }, operations)).toEqual({ values: after });
});
it("keeps a reconstructed large-array leaf edit narrow", () => {
const before = Array.from({ length: 1_000 }, (_, value) => ({ value, label: `row-${value}` }));
const after = structuredClone(before);
after[700]!.label = "changed";
const operations = diffRevisions({ values: before }, { values: after });
expect(operations).toEqual([["s", ["values", 700, "label"], "changed"]]);
expect(JSON.stringify(operations).length).toBeLessThan(100);
expect(applyImmutable({ values: before }, operations)).toEqual({ values: after });
});
it("splices an ambiguous equal-count moved-and-edited gap", () => {
const left = { value: "left" };
const right = { value: "right" };
const before = [left, { id: 1, value: "a" }, { id: 2, value: "b" }, right];
const replacements = [
{ id: 2, value: "edited" },
{ id: 1, value: "also-edited" },
];
const after = [left, ...replacements, right];
expectDiff({ values: before }, { values: after }, [["p", ["values"], 1, 2, replacements]]);
});
it("encodes five hundred unshifts without snapshotting retained rows", () => {
const retained = Array.from({ length: 10_000 }, (_, value) => ({ value, payload: "x".repeat(100) }));
const inserted = Array.from({ length: 500 }, (_, value) => ({ value: -value - 1 }));
const operations = diffRevisions({ values: retained }, { values: [...inserted, ...retained] });
expect(operations).toEqual([["p", ["values"], 0, 0, inserted]]);
expect(JSON.stringify(operations).length).toBeLessThan(20_000);
expect(applyImmutable({ values: retained }, operations)).toEqual({ values: [...inserted, ...retained] });
});
it("bounds wide-object operation emission with a root replacement", () => {
const before: Record<string, JsonValue> = {};
const after: Record<string, JsonValue> = {};
for (let index = 0; index < 20_000; index++) {
before[`field${index}`] = 0;
after[`field${index}`] = 1;
}
expect(diffRevisions(before, after)).toEqual([["r", after]]);
});
it("falls back to a base operation when leaf operations are larger", () => {
const before = { values: Array.from({ length: 40_000 }, () => 0) };
const after = { values: Array.from({ length: 40_000 }, () => 1) };

515
packages/chord/test/state-draft.test.ts
View file

@ -1,463 +1,104 @@
import { describe, expect, it } from "vitest";
import { type Draft, produce, produceWithMetadata } from "../src/state/draft.ts";
import { applyImmutable, type Draft, track } from "../src/delta/index.ts";
describe("produce", () => {
it("returns the base when the recipe makes no semantic write", () => {
const base = { count: 1, nested: { value: "same" }, values: [1, 2, 3] };
const result = produce(base, (draft) => {
draft.count = 1;
draft.nested.value = "same";
delete (draft as { missing?: boolean }).missing;
draft.values.sort((left, right) => left - right);
});
expect(result).toBe(base);
});
it("copies only changed branches and keeps stable draft identities", () => {
const base = {
describe("copy-on-write change drafts", () => {
it("copies only changed branches", () => {
const tracker = track({
changed: { count: 1, sibling: { value: "kept" } },
untouched: { value: 2 },
};
let first: Draft<typeof base.changed> | undefined;
const { value, owned } = produceWithMetadata(base, (draft) => {
first = draft.changed;
expect(draft.changed).toBe(first);
draft.changed.count = 2;
draft.changed.count = 3;
});
expect(value).not.toBe(base);
expect(value.changed).not.toBe(base.changed);
expect(value.changed.sibling).toBe(base.changed.sibling);
expect(value.untouched).toBe(base.untouched);
expect(value.changed.count).toBe(3);
expect(owned.has(value)).toBe(true);
expect(owned.has(value.changed)).toBe(true);
expect(owned.has(value.changed.sibling)).toBe(false);
expect(owned.has(value.untouched)).toBe(false);
expect(owned.has(base)).toBe(false);
const base = tracker.value;
const change = tracker.beginChange();
const first = change.state.changed;
expect(change.state.changed).toBe(first);
change.state.changed.count = 3;
const prepared = change.prepare();
expect(prepared.value).not.toBe(base);
expect(prepared.value.changed).not.toBe(base.changed);
expect(prepared.value.changed.sibling).toBe(base.changed.sibling);
expect(prepared.value.untouched).toBe(base.untouched);
});
it("supports property assignment and deletion without mutating a frozen base", () => {
const nested: Readonly<{ old: boolean }> = Object.freeze({ old: true });
const base: Readonly<{
name: string;
nested: Readonly<{ old: boolean }>;
optional?: string;
}> = Object.freeze({ name: "before", nested, optional: "remove" });
const result = produce(base, (draft) => {
draft.name = "after";
delete draft.optional;
draft.nested.old = false;
});
expect(result).toEqual({ name: "after", nested: { old: false } });
expect(base).toEqual({ name: "before", nested: { old: true }, optional: "remove" });
expect(nested.old).toBe(true);
it("supports frozen bases, deletion, and native array mutators", () => {
const tracker = track({ optional: "remove" as string | undefined, values: [3, 1, 2] });
const change = tracker.beginChange();
delete change.state.optional;
change.state.values.push(4);
expect(change.state.values.pop()).toBe(4);
change.state.values.unshift(0);
expect(change.state.values.shift()).toBe(0);
expect(change.state.values.splice(1, 1, 5, 4)).toEqual([1]);
change.state.values.sort((left, right) => left - right);
change.state.values.reverse();
change.state.values.fill(9, 1, 3);
change.state.values.copyWithin(1, 0, 2);
const prepared = change.prepare();
expect(prepared.value).toEqual({ values: [5, 5, 9, 2] });
});
it("assigns base containers by value instead of creating aliases", () => {
const shared = { value: 1 };
const base: { source: { value: number }; alias: { value: number } | null } = { source: shared, alias: null };
const { value, owned } = produceWithMetadata(base, (draft) => {
draft.alias = base.source;
draft.alias.value = 2;
});
expect(base.source.value).toBe(1);
expect(value.source).toBe(base.source);
expect(value.source.value).toBe(1);
expect(value.alias).toEqual({ value: 2 });
expect(value.alias).not.toBe(value.source);
expect(owned.has(value)).toBe(true);
expect(owned.has(value.alias!)).toBe(true);
expect(owned.has(value.source)).toBe(false);
});
it("preserves tuple positions in draft types", () => {
const base = { tuple: ["value", 1] as readonly [string, number] };
produce(base, (draft) => {
const text: string = draft.tuple[0];
const count: number = draft.tuple[1];
draft.tuple = [text, count + 1];
});
});
it("rejects asynchronous recipes without changing the base", () => {
const base = { value: 1 };
expect(() =>
produce(base, async (draft) => {
draft.value = 2;
await Promise.resolve();
}),
).toThrow(/synchronous/);
expect(base).toEqual({ value: 1 });
});
it("rolls back and revokes every created draft when the recipe throws", () => {
const base = { child: { count: 1 }, values: [1, 2] };
const failure = new Error("stop");
let escapedChild: Draft<typeof base.child> | undefined;
let escapedArray: Draft<typeof base.values> | undefined;
expect(() =>
produce(base, (draft) => {
escapedChild = draft.child;
escapedArray = draft.values;
draft.child.count = 9;
draft.values.push(3);
throw failure;
}),
).toThrow(failure);
expect(base).toEqual({ child: { count: 1 }, values: [1, 2] });
expect(() => escapedChild?.count).toThrow(TypeError);
expect(() => escapedArray?.length).toThrow(TypeError);
});
it("revokes escaped drafts after a successful recipe", () => {
const base = { child: { count: 1 } };
let escapedRoot: Draft<typeof base> | undefined;
let escapedChild: Draft<typeof base.child> | undefined;
const result = produce(base, (draft) => {
escapedRoot = draft;
escapedChild = draft.child;
draft.child.count = 2;
});
expect(result.child.count).toBe(2);
expect(() => escapedRoot?.child).toThrow(TypeError);
expect(() => escapedChild?.count).toThrow(TypeError);
});
it("supports native mutating array methods", () => {
const base = { values: [3, 1, 2] };
const result = produce(base, (draft) => {
draft.values.push(4);
expect(draft.values.pop()).toBe(4);
draft.values.unshift(0);
expect(draft.values.shift()).toBe(0);
expect(draft.values.splice(1, 1, 5, 4)).toEqual([1]);
draft.values.sort((left, right) => left - right);
draft.values.reverse();
draft.values.fill(9, 1, 3);
draft.values.copyWithin(1, 0, 2);
});
expect(result.values).toEqual([5, 5, 9, 2]);
expect(base.values).toEqual([3, 1, 2]);
});
it("copies inserted draft values without detaching their original handle", () => {
const base = { values: [{ value: 1 }, { value: 2 }] };
const result = produce(base, (draft) => {
const held = draft.values[0]!;
draft.values.unshift(held);
held.value = 9;
});
expect(result.values).toEqual([{ value: 1 }, { value: 9 }, { value: 2 }]);
expect(result.values[0]).not.toBe(result.values[1]);
});
it("fill reads its value after coercing its range", () => {
const base = { values: [{ value: 0 }], other: { value: 1 } };
const result = produce(base, (draft) => {
draft.values.fill(draft.other, {
valueOf() {
draft.other.value = 2;
return 0;
},
} as unknown as number);
});
expect(result.values).toEqual([{ value: 2 }]);
expect(
produce(base, (draft) => {
draft.values.fill(undefined as unknown as { value: number }, 0, 0);
}),
).toBe(base);
});
it("copies object values for fill and copyWithin", () => {
const base = { values: [{ value: 1 }, { value: 2 }, { value: 3 }] };
const result = produce(base, (draft) => {
draft.values.copyWithin(1, 0, 1);
draft.values[0]!.value = 9;
});
expect(result.values).toEqual([{ value: 9 }, { value: 1 }, { value: 3 }]);
expect(result.values[0]).not.toBe(result.values[1]);
const filled = produce({ ...base, other: { value: 4 } }, (draft) => {
draft.values.fill(draft.other);
draft.values[0]!.value = 5;
});
expect(filled.values).toEqual([{ value: 5 }, { value: 4 }, { value: 4 }]);
expect(new Set([...filled.values, filled.other]).size).toBe(4);
});
it("uses the actual receiver for borrowed array mutators", () => {
const base = { left: [{ value: 1 }, { value: 2 }], right: [{ value: 3 }] };
const external = [{ value: 0 }];
const result = produce(base, (draft) => {
const held = draft.left[0]!;
draft.right.reverse.call(draft.left);
held.value = 9;
draft.right.push.call(external, { value: 2 });
expect(() => {
const push = draft.right.push;
push({ value: 4 });
}).toThrow(TypeError);
});
expect(result.left).toEqual([{ value: 2 }, { value: 9 }]);
expect(external).toEqual([{ value: 0 }, { value: 2 }]);
});
it("copies user assignments made while an array mutator coerces arguments", () => {
const base = { values: [{ value: 1 }], other: { value: 2 } };
const result = produce(base, (draft) => {
draft.values.splice(
{
valueOf() {
draft.values[0] = draft.other;
return 0;
},
} as unknown as number,
0,
);
draft.values[0]!.value = 3;
});
expect(result).toEqual({ values: [{ value: 3 }], other: { value: 2 } });
expect(result.values[0]).not.toBe(result.other);
const sameArray = produce({ values: [{ value: 1 }, { value: 2 }] }, (draft) => {
draft.values.splice(
{
valueOf() {
draft.values[1] = draft.values[0]!;
return 0;
},
} as unknown as number,
0,
);
draft.values[0]!.value = 9;
});
expect(sameArray.values).toEqual([{ value: 9 }, { value: 1 }]);
expect(sameArray.values[0]).not.toBe(sameArray.values[1]);
});
it("copyWithin preserves handles across coercion side effects", () => {
const base = { values: [{ value: 1 }, { value: 2 }, { value: 3 }] };
const result = produce(base, (draft) => {
const held = draft.values[2]!;
draft.values.copyWithin(
0,
{
valueOf() {
draft.values.reverse();
return 0;
},
} as unknown as number,
0,
);
held.value = 9;
});
expect(result.values).toEqual([{ value: 9 }, { value: 2 }, { value: 1 }]);
});
it("preserves native coercion semantics in array arguments", () => {
const spliced = produce({ values: [1, 2, 3] }, (draft) => {
draft.values.splice(Number.POSITIVE_INFINITY, 0, 4);
});
expect(spliced.values).toEqual([1, 2, 3, 4]);
const filled = produce({ values: [1, 2, 3] }, (draft) => {
draft.values.fill(9, Number.POSITIVE_INFINITY);
draft.values.copyWithin(Number.POSITIVE_INFINITY, 0);
});
expect(filled.values).toEqual([1, 2, 3]);
const explicitUndefined = produce({ values: [1, 2, 3], copied: [1, 2, 3] }, (draft) => {
draft.values.fill(9, 0, undefined as unknown as number);
draft.copied.copyWithin(1, 0, undefined as unknown as number);
});
expect(explicitUndefined.values).toEqual([9, 9, 9]);
expect(explicitUndefined.copied).toEqual([1, 1, 2]);
const capturedLength = produce({ values: [1, 2] }, (draft) => {
draft.values.splice(
{
valueOf() {
draft.values.push(3);
return 1;
},
} as unknown as number,
0,
);
});
expect(capturedLength.values).toEqual([1, 2]);
const removedLength = produce({ values: [1, 2, 3], removed: 0 }, (draft) => {
const removed = draft.values.splice(
{
valueOf() {
draft.values.length = 0;
return 0;
},
} as unknown as number,
3,
);
draft.removed = removed.length;
});
expect(removedLength).toEqual({ values: [], removed: 3 });
});
it("default sort observes pending nested edits", () => {
const base = { values: [[2], [1]] };
const result = produce(base, (draft) => {
draft.values[0]![0] = 0;
draft.values.sort();
});
expect(result.values).toEqual([[0], [1]]);
it("copies inserted draft values without aliasing their original handle", () => {
const tracker = track({ values: [{ value: 1 }, { value: 2 }] });
const change = tracker.beginChange();
const held = change.state.values[0]!;
change.state.values.unshift(held);
held.value = 9;
const prepared = change.prepare();
expect(prepared.value.values).toEqual([{ value: 1 }, { value: 9 }, { value: 2 }]);
expect(prepared.value.values[0]).not.toBe(prepared.value.values[1]);
});
it("keeps comparator edits when sorting object drafts", () => {
const base = {
const tracker = track({
rows: [
{ rank: 2, comparisons: 0 },
{ rank: 1, comparisons: 0 },
],
};
const result = produce(base, (draft) => {
draft.rows.sort((left, right) => {
left.comparisons++;
right.comparisons++;
return left.rank - right.rank;
});
});
expect(result.rows.map((row) => row.rank)).toEqual([1, 2]);
expect(result.rows.map((row) => row.comparisons)).toEqual([1, 1]);
expect(result.rows[0]).not.toBe(base.rows[1]);
expect(result.rows[1]).not.toBe(base.rows[0]);
expect(base.rows).toEqual([
{ rank: 2, comparisons: 0 },
{ rank: 1, comparisons: 0 },
]);
const change = tracker.beginChange();
change.state.rows.sort((left, right) => {
left.comparisons++;
right.comparisons++;
return left.rank - right.rank;
});
const prepared = change.prepare();
expect(prepared.value.rows.map((row) => row.rank)).toEqual([1, 2]);
expect(prepared.value.rows.map((row) => row.comparisons)).toEqual([1, 1]);
});
it("drops writes through detached child handles", () => {
const kept = { value: "kept" };
const removed = { value: "removed" };
const base = { child: removed, items: [removed, kept] };
const result = produce(base, (draft) => {
const child = draft.child;
const shifted = draft.items.shift();
delete (draft as { child?: Draft<typeof removed>; items: Draft<typeof base.items> }).child;
child.value = "detached child";
if (shifted !== undefined) shifted.value = "detached item";
});
expect(result).toEqual({ items: [kept] });
expect(result.items[0]).toBe(kept);
expect(removed.value).toBe("removed");
const tracker = track({ child: { value: "removed" }, items: [{ value: "removed" }, { value: "kept" }] });
const change = tracker.beginChange();
const child = change.state.child;
const shifted = change.state.items.shift();
delete (change.state as { child?: Draft<{ value: string }>; items: Draft<{ value: string }[]> }).child;
child.value = "detached child";
if (shifted !== undefined) shifted.value = "detached item";
expect(change.prepare().value).toEqual({ items: [{ value: "kept" }] });
});
it("deep-clones assigned values, expands aliases, and reports every clone as owned", () => {
const shared = { value: 2 };
const external = { left: shared, right: shared };
const base: { payload: typeof external | null; untouched: { value: number } } = {
payload: null,
untouched: { value: 1 },
};
const { value, owned } = produceWithMetadata(base, (draft) => {
draft.payload = external;
draft.payload.left.value = 3;
});
expect(external.left.value).toBe(2);
expect(value.payload).not.toBe(external);
expect(value.payload?.left).not.toBe(shared);
expect(value.payload?.right).not.toBe(shared);
expect(value.payload?.left).not.toBe(value.payload?.right);
expect(value.payload).toEqual({ left: { value: 3 }, right: { value: 2 } });
expect(value.untouched).toBe(base.untouched);
expect(owned.has(value)).toBe(true);
expect(owned.has(value.payload!)).toBe(true);
expect(owned.has(value.payload!.left)).toBe(true);
expect(owned.has(value.payload!.right)).toBe(true);
expect(owned.has(external)).toBe(false);
expect(owned.has(shared)).toBe(false);
it.each(["unshift", "splice"] as const)("inserts 100,000 items with %s without argument overflow", (method) => {
const tracker = track({ values: [-1] });
const change = tracker.beginChange();
const items = Array.from({ length: 100_000 }, (_, value) => value);
if (method === "unshift") Reflect.apply(change.state.values.unshift, change.state.values, items);
else Reflect.apply(change.state.values.splice, change.state.values, [1, 0, ...items]);
const prepared = change.prepare();
expect(prepared.value.values).toHaveLength(100_001);
expect(prepared.value.values[method === "unshift" ? 0 : 1]).toBe(0);
expect(prepared.value.values[method === "unshift" ? 99_999 : 100_000]).toBe(99_999);
expect(applyImmutable(prepared.base, prepared.ops)).toEqual(prepared.value);
});
it("rejects cycles in assigned values without changing the base", () => {
const cyclic: { self?: unknown } = {};
cyclic.self = cyclic;
const base: { payload: object | null } = { payload: null };
expect(() =>
produce(base, (draft) => {
draft.payload = cyclic;
}),
).toThrow(/cycles/);
expect(base.payload).toBeNull();
});
it("rejects sparse or undefined array results", () => {
const base = { values: [1, 2] };
expect(() =>
produce(base, (draft) => {
delete draft.values[0];
}),
).toThrow(/dense|holes/);
expect(() =>
produce(base, (draft) => {
draft.values[0] = undefined as unknown as number;
}),
).toThrow(/undefined/);
expect(() =>
produce(base, (draft) => {
draft.values.length = 4;
}),
).toThrow(/dense|holes/);
expect(base.values).toEqual([1, 2]);
});
it("writes own properties without invoking inherited setters", () => {
Object.defineProperty(Object.prototype, "trap", {
set() {
throw new Error("inherited setter ran");
},
configurable: true,
});
try {
const result = produce({ nested: {} as Record<string, number> }, (draft) => {
draft.nested.trap = 1;
});
expect(result).toEqual({ nested: { trap: 1 } });
} finally {
delete (Object.prototype as Record<string, unknown>).trap;
}
let pushed: { values: number[] } | undefined;
let unshifted: { values: number[] } | undefined;
Object.defineProperty(Array.prototype, "3", {
set() {
throw new Error("inherited array setter ran");
},
configurable: true,
});
try {
pushed = produce({ values: [1, 2, 3] }, (draft) => {
draft.values.push(4);
});
unshifted = produce({ values: [1, 2, 3] }, (draft) => {
draft.values.unshift(0);
});
} finally {
delete (Array.prototype as unknown as Record<string, unknown>)["3"];
}
expect(pushed).toEqual({ values: [1, 2, 3, 4] });
expect(unshifted).toEqual({ values: [0, 1, 2, 3] });
});
it("rejects reflective mutations and symbol writes", () => {
const base = { value: 1 };
expect(() => produce(base, (draft) => Object.defineProperty(draft, "other", { value: 2 }))).toThrow(TypeError);
expect(() => produce(base, (draft) => Object.setPrototypeOf(draft, null))).toThrow(TypeError);
expect(() => produce(base, (draft) => Object.preventExtensions(draft))).toThrow(TypeError);
expect(() => produce(base, (draft) => Reflect.set(draft, Symbol("key"), 2))).toThrow(/Symbol/);
expect(base).toEqual({ value: 1 });
it("rejects invalid writes without mutating the committed base", () => {
const tracker = track({ values: [1, 2] });
let change = tracker.beginChange();
expect(() => delete change.state.values[0]).toThrow(/dense|holes/);
change.abort();
change = tracker.beginChange();
expect(() => {
change.state.values[0] = undefined as unknown as number;
}).toThrow(/undefined/);
change.abort();
expect(tracker.value.values).toEqual([1, 2]);
});
});

29
packages/chord/test/state-fuzz.test.ts
View file

@ -1,9 +1,5 @@
import { expect, it } from "vitest";
import { BACKGROUND_CONTEXT } from "../src/context/index.ts";
import { applyImmutable, type Op } from "../src/delta/index.ts";
import { replicatedState } from "../src/index.ts";
import { getReplicatedStateInternals } from "../src/services/state-internals.ts";
import type { Draft } from "../src/state/draft.ts";
import { applyImmutable, type Draft, type Op, track } from "../src/delta/index.ts";
type Item = { id: number; text: string; score: number };
type Document = {
@ -75,7 +71,7 @@ const mutate = (document: MutableDocument, choice: number, value: number): void
}
};
it("converges across randomized transactional revisions", () => {
it("converges across randomized prepared revisions", () => {
for (let seed = 1; seed <= 100; seed++) {
const rng = random(seed);
const initial: Document = {
@ -83,23 +79,22 @@ it("converges across randomized transactional revisions", () => {
text: "start",
meta: { revision: 0 },
};
const state = replicatedState(initial);
const tracker = track(initial);
const expected = clone(initial);
let replica = clone(state.value);
let latest: readonly Op[] = [];
getReplicatedStateInternals(state)!.subscribe((operations) => {
latest = operations;
});
let replica = clone(tracker.value);
for (let step = 0; step < 100; step++) {
const choice = Math.floor(rng() * 14);
const value = seed * 1_000 + step;
mutate(expected, choice, value);
latest = [];
state.change(BACKGROUND_CONTEXT, (draft) => mutate(draft, choice, value));
replica = applyImmutable(replica, latest);
expect(state.value, `state seed ${seed} step ${step} choice ${choice}`).toEqual(expected);
const change = tracker.beginChange();
mutate(change.state, choice, value);
const prepared = change.prepare();
const operations: readonly Op[] = prepared.ops;
replica = applyImmutable(replica, operations);
tracker.adopt(prepared);
expect(tracker.value, `state seed ${seed} step ${step} choice ${choice}`).toEqual(expected);
expect(replica, `replica seed ${seed} step ${step} choice ${choice}`).toEqual(expected);
expect(Object.isFrozen(state.value)).toBe(true);
expect(Object.isFrozen(tracker.value)).toBe(true);
}
}
});

53
packages/chord/test/state-value.test.ts
View file

@ -1,40 +1,37 @@
import { describe, expect, it } from "vitest";
import { produceWithMetadata } from "../src/state/draft.ts";
import { JsonRevisionStore } from "../src/state/value.ts";
import { track } from "../src/delta/index.ts";
describe("JSON revision ownership", () => {
describe("immutable revision ownership", () => {
it("imports detached frozen JSON and expands aliases", () => {
const shared = { value: 1 };
const input = { left: shared, right: shared };
const store = new JsonRevisionStore();
const value = store.import(input);
expect(value).toEqual(input);
expect(value).not.toBe(input);
expect(value.left).not.toBe(value.right);
expect(Object.isFrozen(value)).toBe(true);
expect(Object.isFrozen(value.left)).toBe(true);
const tracker = track(input);
expect(tracker.value).toEqual(input);
expect(tracker.value).not.toBe(input);
expect(tracker.value.left).not.toBe(tracker.value.right);
expect(Object.isFrozen(tracker.value)).toBe(true);
expect(Object.isFrozen(tracker.value.left)).toBe(true);
});
it("commits transaction copies in place while sharing unchanged branches", () => {
const store = new JsonRevisionStore();
const base = store.import({ changed: { value: 1 }, retained: { value: 2 } });
const produced = produceWithMetadata(base, (draft) => {
draft.changed.value = 3;
});
const next = store.commit(produced.value, produced.owned);
expect(next).toBe(produced.value);
expect(next.changed).toBe(produced.value.changed);
expect(next.retained).toBe(base.retained);
expect(Object.isFrozen(next)).toBe(true);
expect(Object.isFrozen(next.changed)).toBe(true);
const tracker = track({ changed: { value: 1 }, retained: { value: 2 } });
const base = tracker.value;
const change = tracker.beginChange();
change.state.changed.value = 3;
const prepared = change.prepare();
expect(prepared.value.changed).not.toBe(base.changed);
expect(prepared.value.retained).toBe(base.retained);
expect(Object.isFrozen(prepared.value.changed)).toBe(true);
tracker.adopt(prepared);
expect(tracker.value).toBe(prepared.value);
});
it("rejects non-JSON values and cycles", () => {
const store = new JsonRevisionStore();
expect(() => store.import({ value: Number.NaN })).toThrow(/strict JSON/);
expect(() => store.import({ value: new Date() })).toThrow(/plain objects/);
const cyclic: { self?: object } = {};
cyclic.self = cyclic;
expect(() => store.import(cyclic)).toThrow(/cycles/);
it("makes repeated placements independent", () => {
const tracker = track({ values: [] as { value: number }[] });
const change = tracker.beginChange();
const shared = { value: 1 };
change.state.values.push(shared, shared);
const prepared = change.prepare();
expect(prepared.value.values[0]).not.toBe(prepared.value.values[1]);
});
});

214
packages/chord/test/state.test.ts
View file

@ -1,10 +1,16 @@
import { describe, expect, it } from "vitest";
import { BACKGROUND_CONTEXT } from "../src/context/index.ts";
import { replicatedState } from "../src/index.ts";
import type { Draft, Op } from "../src/delta/index.ts";
import {
type Context,
type JsonValue,
type ReplicatedStateSource,
type ReplicatedStateSourceAttachment,
type ReplicatedStateSourceFrame,
replicatedState,
} from "../src/index.ts";
import { ReplicatedStateReplica } from "../src/services/state.ts";
import { getReplicatedStateInternals } from "../src/services/state-internals.ts";
import type { Draft } from "../src/state/draft.ts";
import type { JsonValue } from "../src/types.ts";
describe("transactional replicated state", () => {
it("publishes one immutable structurally shared revision", () => {
@ -40,6 +46,18 @@ describe("transactional replicated state", () => {
expect(() => escaped?.value).toThrow(TypeError);
});
it("rejects PromiseLike callbacks and aborts their draft", () => {
const state = replicatedState({ value: 0 });
const previous = state.value;
expect(() =>
state.change(BACKGROUND_CONTEXT, ((draft: Draft<{ value: number }>) => {
draft.value = 1;
return Promise.resolve();
}) as (draft: Draft<{ value: number }>) => void),
).toThrow(/synchronous/);
expect(state.value).toBe(previous);
});
it("rejects nested changes and replacements without losing the outer rollback", () => {
const state = replicatedState({ left: 0, right: 0 });
expect(() =>
@ -192,3 +210,193 @@ describe("transactional replicated state", () => {
expect(state.value.retained).not.toBe(previous.retained);
});
});
type SourceValue = Readonly<{ value: number }>;
class TestSource<T> implements ReplicatedStateSource<T> {
readonly attachments = new Set<TestSourceAttachment<T>>();
onAttach: (() => void) | undefined;
#value: T;
#cursor: number;
constructor(value: T, cursor = 0) {
this.#value = value;
this.#cursor = cursor;
}
attach(): ReplicatedStateSourceAttachment<T> {
const attachment = new TestSourceAttachment(Object.freeze({ value: this.#value, cursor: this.#cursor }), () =>
this.attachments.delete(attachment),
);
this.attachments.add(attachment);
this.onAttach?.();
return attachment;
}
commit(value: T, ops: readonly Op[], context: Context = BACKGROUND_CONTEXT, cursor = this.#cursor + 1): void {
this.#value = value;
this.#cursor = cursor;
const frame = Object.freeze({ cursor, value, ops, context });
for (const attachment of [...this.attachments]) attachment.publish(frame);
}
}
class TestSourceAttachment<T> implements ReplicatedStateSourceAttachment<T> {
readonly snapshot: { readonly value: T; readonly cursor: number };
readonly #onDispose: () => void;
readonly #buffer: ReplicatedStateSourceFrame<T>[] = [];
#listener: ((frame: ReplicatedStateSourceFrame<T>) => void) | undefined;
#activated = false;
#disposed = false;
constructor(snapshot: { readonly value: T; readonly cursor: number }, onDispose: () => void) {
this.snapshot = snapshot;
this.#onDispose = onDispose;
}
activate(listener: (frame: ReplicatedStateSourceFrame<T>) => void): void {
if (this.#activated) throw new Error("attachment is already active");
this.#activated = true;
this.#listener = listener;
for (const frame of this.#buffer.splice(0)) listener(frame);
}
publish(frame: ReplicatedStateSourceFrame<T>): void {
if (this.#disposed) return;
if (this.#listener === undefined) this.#buffer.push(frame);
else this.#listener(frame);
}
dispose(): void {
if (this.#disposed) return;
this.#disposed = true;
this.#buffer.length = 0;
this.#onDispose();
}
}
describe("authoritative replicated state sources", () => {
it("captures before activation and drains queued commits in order", () => {
const initial = Object.freeze({ value: 0 });
const first = Object.freeze({ value: 1 });
const second = Object.freeze({ value: 2 });
const firstOps = Object.freeze([["s", ["value"], 1] as const]);
const secondOps = Object.freeze([["s", ["value"], 2] as const]);
const source = new TestSource<SourceValue>(initial, 10);
source.onAttach = () => {
source.commit(first, firstOps);
source.commit(second, secondOps);
};
const state = replicatedState(source);
const deliveries: Array<{ value: SourceValue; sequence: number }> = [];
state.subscribe((value, _context, delivery) => deliveries.push({ value, sequence: delivery.sequence }));
expect(state.value).toBe(second);
expect(deliveries).toEqual([{ value: second, sequence: 2 }]);
expect(getReplicatedStateInternals(state)?.snapshot()).toEqual({ value: second, sequence: 2 });
});
it("hydrates at sequence zero when attaching after existing commits", () => {
const initial = Object.freeze({ value: 0 });
const current = Object.freeze({ value: 1 });
const source = new TestSource<SourceValue>(initial, 40);
source.commit(current, Object.freeze([["s", ["value"], 1] as const]));
const state = replicatedState(source);
const deliveries: number[] = [];
state.subscribe((_value, _context, delivery) => deliveries.push(delivery.sequence));
expect(state.value).toBe(current);
expect(deliveries).toEqual([0]);
});
it("publishes exact source value and operation references without applying or re-diffing", () => {
const source = new TestSource<SourceValue>(Object.freeze({ value: 0 }));
const state = replicatedState(source);
const next = Object.freeze({ value: 1 });
const ops = Object.freeze([["s", ["value"], 1] as const]);
let publishedOps: readonly Op[] | undefined;
getReplicatedStateInternals(state)?.subscribe((received) => {
publishedOps = received;
});
let publishedValue: SourceValue | undefined;
state.subscribe((value, _context, delivery) => {
if (delivery.kind === "update") publishedValue = value;
});
source.commit(next, ops);
expect(state.value).toBe(next);
expect(publishedValue).toBe(next);
expect(publishedOps).toBe(ops);
});
it("buffers reentrant frames and skips updates covered by a late hydration", () => {
const source = new TestSource<SourceValue>(Object.freeze({ value: 0 }));
const state = replicatedState(source);
const received: number[] = [];
const late: Array<{ kind: string; sequence: number; value: number }> = [];
let nested = false;
getReplicatedStateInternals(state)?.subscribe((_ops, sequence) => {
if (sequence !== 1 || nested) return;
nested = true;
source.commit(Object.freeze({ value: 2 }), Object.freeze([["s", ["value"], 2] as const]));
state.subscribe((value, _context, delivery) => late.push({ ...delivery, value: value.value }));
});
state.subscribe((value, _context, delivery) => {
if (delivery.kind === "update") received.push(value.value);
});
source.commit(Object.freeze({ value: 1 }), Object.freeze([["s", ["value"], 1] as const]));
expect(received).toEqual([1, 2]);
expect(late).toEqual([{ kind: "hydrate", sequence: 2, value: 2 }]);
});
it("reports listener failures without throwing them into the source", () => {
const source = new TestSource<SourceValue>(Object.freeze({ value: 0 }));
const errors: Error[] = [];
const state = replicatedState(source, { onError: (error) => errors.push(error) });
const received: number[] = [];
state.subscribe((_value, _context, delivery) => {
if (delivery.kind === "update") throw new Error("listener failed");
});
state.subscribe((value, _context, delivery) => {
if (delivery.kind === "update") received.push(value.value);
});
expect(() =>
source.commit(Object.freeze({ value: 1 }), Object.freeze([["s", ["value"], 1] as const])),
).not.toThrow();
source.commit(Object.freeze({ value: 2 }), Object.freeze([["s", ["value"], 2] as const]));
expect(received).toEqual([1, 2]);
expect(errors.map((error) => error.message)).toEqual(["listener failed", "listener failed"]);
});
it("reports cursor gaps, disposes the broken attachment, and ignores later frames", () => {
const source = new TestSource<SourceValue>(Object.freeze({ value: 0 }), 5);
const errors: Error[] = [];
const state = replicatedState(source, { onError: (error) => errors.push(error) });
source.commit(Object.freeze({ value: 2 }), Object.freeze([["s", ["value"], 2] as const]), BACKGROUND_CONTEXT, 7);
expect(errors[0]?.message).toContain("expected 6, received 7");
expect(source.attachments.size).toBe(0);
expect(state.value).toEqual({ value: 0 });
source.commit(Object.freeze({ value: 3 }), Object.freeze([["s", ["value"], 3] as const]), BACKGROUND_CONTEXT, 8);
expect(state.value).toEqual({ value: 0 });
});
it("keeps attachments independent and disposes each idempotently", () => {
const source = new TestSource<SourceValue>(Object.freeze({ value: 0 }));
const first = replicatedState(source);
const second = replicatedState(source);
expect(source.attachments.size).toBe(2);
source.commit(Object.freeze({ value: 1 }), Object.freeze([["s", ["value"], 1] as const]));
expect(first.value.value).toBe(1);
expect(second.value.value).toBe(1);
first.dispose();
first.dispose();
expect(source.attachments.size).toBe(1);
source.commit(Object.freeze({ value: 2 }), Object.freeze([["s", ["value"], 2] as const]));
expect(first.value.value).toBe(1);
expect(second.value.value).toBe(2);
second.dispose();
expect(source.attachments.size).toBe(0);
});
});