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,durable): PendingOps folds pending ops for slow consumers; Watch delivers one call per busy period

Browse source 
- chord/delta: PendingOps (push, size, take -> PendingTake): raw pending ops, r drops everything
  before it, a leaf index merges appends and primitive sets per path (cleared by any other op),
  past 10,000 ops only "replace with the current value"; one batch is kept by reference and
  received ops are never mutated
- chord: Watch queues frames in PendingOps while its listener is busy or not started and delivers
  one call with the newest value, the folded ops (or [["r", newest]] after an overflow) and the
  newest frame's context; replaces the 100-frame collapse. NativeWatch is unchanged and takes an
  optional replace()
- durable: watchEvents keeps every event batch in order (NativeWatch with a snapshot replace);
  document, view and task-graph watches fold; spec, README, example 04 and tests updated
- tests: fold fuzzers (tracker, appendText, splices/reorders, root arrays, diffRevisions) with
  codec round trips, adversarial cases, cap, ownership, cost and GC guards
This commit is contained in:
Mario Zechner 2026-10-24 09:37:12 +00:00
parent 1e42501bb3
commit d8d99950ac
20 changed files with 1434 additions and 240 deletions
Download patch file Download diff file Expand all files Collapse all files

7
packages/chord/CHANGELOG.md
View file

@ -9,10 +9,11 @@
### Added
- `NativeWatchable<T>`: an in-process value that changes and can be watched (`value`, `watch(context)` returning a `NativeWatchHandle<T>`), the in-process counterpart of `Watchable<T>`.
- `WatchHandle<T>` and `WatchEnd`, the one watch API, moved from Durable: a serialized exact-frame observation with `value`, `start(listener)`, `stop()`, and `closed`. `WatchEnd` reasons are `stopped`, `cancelled`, `retired`, `ended`, `listener_error`, and `failed`.
- `Watch<T>`, the one watch handle implementation, moved from Durable: a source creates one per watch and feeds it with `advance(value, ops, context)`, `retire(context)` (last frame `[["r", null]]`, then `retired`), `end()` or `end({ error })`, and `observeCancellation(signal)`. More than 100 undelivered frames collapse to one root replacement `[["r", newest]]`; termination drops the listener and every reference into the source.
- `WatchHandle<T>` and `WatchEnd`, the one watch API, moved from Durable: a serialized observation with `value`, `start(listener)`, `stop()`, and `closed`. `WatchEnd` reasons are `stopped`, `cancelled`, `retired`, `ended`, `listener_error`, and `failed`.
- `Watch<T>`, the one watch handle implementation, moved from Durable: a source creates one per watch and feeds it with `advance(value, ops, context)`, `retire(context)` (last frame `[["r", null]]`, then `retired`), `end()` or `end({ error })`, and `observeCancellation(signal)`. Frames that arrive while the listener runs (or before `start`) are delivered in one call with the newest value, their ops folded by `PendingOps` and the newest frame's context; past 10,000 pending ops after folding, the call gets one root replacement `[["r", newest]]` instead (replaces the collapse after 100 undelivered frames); termination drops the listener and every reference into the source.
- `Watchable<T>`: a value that changes and can be watched (`value`, `watch(context)` returning a `WatchHandle<T>`), the common interface of documents, derives and other observable values.
- `NativeWatchHandle<T>` and its implementation `NativeWatch<T>`: the same lifecycle for in-process values, with listener `(value, context)` and no ops; an overflow collapses to the newest value.
- `NativeWatchHandle<T>` and its implementation `NativeWatch<T>`: the same lifecycle for in-process values, with listener `(value, context)` and no ops: one call per value in order, and past 100 pending values one call with the value of the optional `replace()` or the newest value.
- `PendingOps` and `PendingTake` in `@earendil-works/chord/delta`: the pending ops of one slow consumer, kept raw and folded while they wait (appends to one path join, a primitive set replaces earlier leaf ops on its path, `r` drops everything before it); `size` counts the pending ops after folding; `take()` empties the queue and hands over the folded ops, or `replace` past 10,000 pending ops. Received batches and op tuples are never mutated.
- `ownedSlice(text, start, end?)` in `@earendil-works/chord/delta`: a slice of a string that does not keep the whole string alive ([#9265](https://github.com/earendil-works/pi/issues/9265))
- `appendText(draft, key, text)` in `@earendil-works/chord/delta`: append to a string in a draft at O(text) cost, recorded as an append op without comparing the old and new strings (a plain `draft[key] += text` makes `prepare()` read the whole string, O(length) per change); `StringKeys<T>` names the keys it accepts ([#9265](https://github.com/earendil-works/pi/issues/9265))

15
packages/chord/README.md
View file

@ -210,10 +210,12 @@ model; see [State](docs/guide.md#state) in the usage guide.
Durable's documents, conversation views, and task graph. `watch.value` is the value
at acquisition until `start(listener)` installs the sole asynchronous listener; each
later change arrives as `(value, ops, context)`, one call at a time, and `value`
advances to it. A source keeps at most 100 undelivered frames and then replaces them
with one root replacement `[["r", newest]]`. `stop()` is idempotent, drops pending
frames, and neither aborts nor awaits a running call; it and `closed` give the
`WatchEnd`:
advances to it. Changes made while the listener runs (or before `start`) arrive in
one call: the newest value, the ops of all of them folded by Delta's `PendingOps`,
and the newest change's context. Past 10,000 pending ops after folding the call
gets one root replacement `[["r", newest]]` instead. `stop()` is idempotent, drops
pending frames, and neither aborts nor awaits a running call; it and `closed` give
the `WatchEnd`:
- `stopped`: `stop()` was called;
- `cancelled`: the context the watch was acquired with was cancelled;
@ -228,5 +230,6 @@ still referenced keeps only its last value alive.
Sources implement handles with `Watch<T>`, fed through `advance(value, ops, context)`,
`retire(context)`, `end()` or `end({ error })`, and `observeCancellation(signal)`.
`NativeWatch<T>` implements `NativeWatchHandle<T>`, the same lifecycle for
in-process values: the listener gets `(value, context)` without ops, and an
overflow collapses to the newest value.
in-process values: the listener gets `(value, context)` without ops, one call per
value in order; past 100 pending values they collapse to one, the value of the
optional `replace()` or the newest.

29
packages/chord/src/delta/README.md
View file

@ -82,6 +82,35 @@ Parsing a separately serialized message per replica is also a valid detachment.
`decode()` does not copy payloads, so decoding one in-memory `WireOp[]` for two
mutable replicas makes them alias each other.
## Slow consumers
`PendingOps` holds the batches one consumer has not taken yet (a watch listener
that is still running, a link whose window is full) and folds them while they
wait, so what it hands over replays to the same value with fewer ops:
```ts
import { PendingOps } from "@earendil-works/chord/delta";
const pending = new PendingOps();
pending.push(prepared.ops); // for each adopted batch, while the consumer is busy
// When the consumer is ready again:
const batch = pending.take();
const ops = batch.kind === "ops" ? batch.ops : [["r", tracker.value]];
```
- A leaf op (an append, or a set of a primitive) folds into the last pending leaf
op on its path: `a` + `a` join, `s` replaces `s` or `a`, an `a` after a string `s`
appends into it. `["a", ["text"], "he"]`, `["s", ["n"], 1]`, `["a", ["text"], "y"]`
becomes `["a", ["text"], "hey"]`, `["s", ["n"], 1]`.
- Any other op (`p`, `m`, `d`, `t`, `s` of an object or array) ends folding into
the ops before it, as it may change what a path names; `r` drops them.
- Past 10,000 pending ops after folding, only `replace` stays: the owner sends the
current value as `[["r", value]]`.
- One batch is handed over as received; received batches and op tuples are never
mutated, so the same batch can feed other consumers.
- It folds ops of the tracker, `diffRevisions()` and `appendText()`; it does not
validate or normalize hand-written ops.
## Do and don't
- Do build a fresh root and then transfer it; validate untrusted roots at your ingestion boundary.

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

@ -125,6 +125,10 @@ export { diffRevisions } from "./diff.ts";
export type { Draft } from "./draft.ts";
export { appendText, type Change, type Prepared, type StringKeys, type Tracker, track } from "./tracker.ts";
// ─── Pending ops of a slow consumer ──────────────────────────────────────────
export { PendingOps, type PendingTake } from "./pending.ts";
// ─── Path safety ─────────────────────────────────────────────────────────────
/**

173
packages/chord/src/delta/pending.ts Normal file
View file

@ -0,0 +1,173 @@
// The pending queue: the ops a slow consumer of one state has not taken yet, folded while they wait.
import type { JsonValue } from "../json.ts";
import type { NonEmptyPath, Op, Path } from "./index.ts";
/**
* What `PendingOps.take()` hands over: the pending ops, folded, or `replace` after an overflow, which the owner of the
* queue turns into a root replacement with the current value.
*/
export type PendingTake = { readonly kind: "ops"; readonly ops: readonly Op[] } | { readonly kind: "replace" };
/** Pending ops past which only "replace with the current value" stays pending. */
export const MAX_PENDING_OPS = 10_000;
/** A leaf op: an append, or a set of a primitive. The leaf index points at these only. */
type LeafOp = readonly ["a", NonEmptyPath, string] | readonly ["s", NonEmptyPath, JsonValue];
type Folding = {
readonly kind: "folding";
/** Owned by the queue until `take()` hands it over; never one of the received op arrays. */
readonly ops: Op[];
/** Path key to the position of the last pending leaf op on that path. */
readonly leaves: Map<string, number>;
};
type State =
| { readonly kind: "empty" }
/** One batch, kept by reference: a consumer that keeps up never pays for folding. */
| { readonly kind: "batch"; readonly ops: readonly Op[] }
| Folding
/** Overflowed: nothing more accumulates until `take()` or a batch that starts with `r`. */
| { readonly kind: "replace" };
const EMPTY: State = { kind: "empty" };
const REPLACE: State = { kind: "replace" };
const NO_OPS: PendingTake = { kind: "ops", ops: [] };
const REPLACE_TAKE: PendingTake = { kind: "replace" };
/**
* Pending ops of one state for one consumer, in order and raw (never encoded while they wait). `r` drops everything
* before it. A leaf op merges into the last pending leaf op on the same path (`a` + `a` joined, `s` replaces `s` or `a`,
* `a` after a string `s` appends into it); any other op clears the leaf index. Leaf ops on different paths commute,
* except after an ancestor primitive `s`, after which the descendant cannot be written until a structural op clears the
* index. Past `MAX_PENDING_OPS` only "replace with the current value" stays pending.
*
* Ops are only those of Delta's tracker, `diffRevisions()` and `appendText()`; received op arrays and tuples are never
* mutated (they feed other consumers too), a merge makes a new tuple.
*/
export class PendingOps {
#state: State = EMPTY;
/** Queue the ops of one commit. */
push(ops: readonly Op[]): void {
if (ops.length === 0) return;
const state = this.#state;
// A batch that starts with `r` (the tracker's and `diffRevisions()`'s only place for it) drops everything
// pending, also an overflow, and is kept as received.
if (ops[0]![0] === "r") {
this.#state = { kind: "batch", ops };
return;
}
switch (state.kind) {
case "empty":
this.#state = { kind: "batch", ops };
return;
case "batch": {
const folding: Folding = { kind: "folding", ops: [], leaves: new Map() };
this.#state = folding;
if (this.#fold(state.ops, folding)) this.#fold(ops, folding);
return;
}
case "folding":
this.#fold(ops, state);
return;
case "replace":
return;
}
}
/**
* Pending ops after folding; 1 once overflowed (the replacement the owner sends). One batch is counted as received,
* so a single commit larger than `MAX_PENDING_OPS` counts in full until a second one arrives and overflows.
*/
get size(): number {
const state = this.#state;
if (state.kind === "empty") return 0;
return state.kind === "replace" ? 1 : state.ops.length;
}
/** Hand over everything pending and start empty. */
take(): PendingTake {
const state = this.#state;
this.#state = EMPTY;
switch (state.kind) {
case "empty":
return NO_OPS;
case "batch":
case "folding":
return { kind: "ops", ops: state.ops };
case "replace":
return REPLACE_TAKE;
}
}
/** Fold `ops` into `folding`; false once the queue overflowed into `replace`. */
#fold(ops: readonly Op[], folding: Folding): boolean {
for (const op of ops) {
switch (op[0]) {
case "r":
folding.ops.length = 0;
folding.leaves.clear();
break;
case "a":
if (mergeLeaf(op, folding)) continue;
break;
case "s":
if (isPrimitive(op[2])) {
if (mergeLeaf(op, folding)) continue;
} else if (folding.leaves.size > 0) folding.leaves.clear();
break;
default:
if (folding.leaves.size > 0) folding.leaves.clear();
}
folding.ops.push(op);
if (folding.ops.length > MAX_PENDING_OPS) {
this.#state = REPLACE;
return false;
}
}
return true;
}
}
const isPrimitive = (value: JsonValue): boolean => value === null || typeof value !== "object";
/**
* Merge the leaf op `op` into the last pending leaf op on its path, or index it as that op at the position it is about
* to be pushed to. True when merged.
*/
function mergeLeaf(op: LeafOp, folding: Folding): boolean {
const key = pathKey(op[1]);
const at = folding.leaves.get(key);
if (at !== undefined) {
// The index points at leaf ops only.
const merged = merge(folding.ops[at] as LeafOp, op);
if (merged !== undefined) {
folding.ops[at] = merged;
return true;
}
}
folding.leaves.set(key, folding.ops.length);
return false;
}
/**
* The key of a path in the leaf index. Injective, as merging two distinct paths would lose a write: `["a.b"]` and
* `["a", "b"]` are different keys. A number and its string (`1`, `"1"`) may share one: within one index lifetime a
* parent is either an array or an object, as changing it is a structural op that clears the index.
*/
function pathKey(path: Path): string {
return JSON.stringify(path);
}
/** `next` folded into `previous` (same path), or undefined when they do not fold. */
function merge(previous: LeafOp, next: LeafOp): LeafOp | undefined {
if (next[0] === "s") return next;
if (previous[0] === "a") return ["a", previous[1], previous[2] + next[2]];
// Only a string set takes an append; the tracker never appends to another primitive.
if (typeof previous[2] === "string") return ["s", previous[1], previous[2] + next[2]];
return undefined;
}

198
packages/chord/src/watch.ts
View file

@ -1,6 +1,6 @@
// Watch handles: the watch types and the one implementation of their lifecycle, for JSON and native values.
import { type Context, withoutAbortSignal } from "./context/index.ts";
import type { Op } from "./delta/index.ts";
import { type Op, PendingOps } from "./delta/index.ts";
import type { JsonValue } from "./json.ts";
/**
@ -18,7 +18,10 @@ export type WatchEnd =
| { readonly reason: "listener_error"; readonly error: Error }
| { readonly reason: "failed"; readonly error: unknown };
/** Serialized exact-frame observation of an immutable value with bounded pending delivery. */
/**
* Serialized observation of an immutable value: one listener call at a time; changes made meanwhile arrive folded into
* the next call.
*/
export interface WatchHandle<T> {
/** Acquisition revision before start; latest delivered immutable revision afterward. */
readonly value: T;
@ -32,12 +35,13 @@ export interface WatchHandle<T> {
/**
* A value that changes and can be watched: documents, derives, members of references. Values are immutable by
* convention (JSON for state); each watch starts at the current value and then gets one frame per change.
* convention (JSON for state); each watch starts at the current value and then gets the changes, folded while its
* listener is busy.
*/
export interface Watchable<T> {
/** The current value. */
readonly value: T;
/** A watch holding the current value, then one frame per change; the context's cancellation ends it. */
/** A watch holding the current value, then its changes; the context's cancellation ends it. */
watch(context: Context): Promise<WatchHandle<T>>;
}
@ -52,7 +56,10 @@ export interface NativeWatchable<T> {
watch(context: Context): Promise<NativeWatchHandle<T>>;
}
/** Serialized observation of an in-process value: as `WatchHandle`, but a change is a new value without ops. */
/**
* Serialized observation of an in-process value: as `WatchHandle`, but each new value is its own call, in order, without
* ops.
*/
export interface NativeWatchHandle<T> {
/** Acquisition value before start; latest delivered value afterward. */
readonly value: T;
@ -68,45 +75,37 @@ type WatchListener<T> = (value: T, ops: readonly Op[], context: Context) => Prom
type NativeWatchListener<T> = (value: T, context: Context) => Promise<void>;
type WatchFrame<T> = {
/** One listener call: the value it delivers and the context of the newest frame it covers. */
type Delivery<T> = {
readonly value: T;
readonly ops: readonly Op[];
readonly context: Context;
/** The retirement frame: the watch ends `retired` once it is delivered. */
readonly retires?: true;
/** The retirement: the watch ends `retired` once it is delivered. */
readonly retires: boolean;
};
/** Maximum frames retained behind one unavailable watch listener. */
const MAX_PENDING_WATCH_FRAMES = 100;
/**
* The lifecycle both handles share: pending frames, serialized delivery, and termination. A source feeds it and ends
* it explicitly; termination drops every reference the watch holds into its source, so a stopped watch that is still
* referenced keeps nothing but its last value alive. Subclasses supply only how a listener is called and the ops of an
* overflow frame.
* The lifecycle both handles share: serialized delivery and termination. A source feeds it and ends it explicitly;
* termination drops every reference the watch holds into its source, so a stopped watch that is still referenced keeps
* nothing but its last value alive. Subclasses keep what is pending and say how a listener is called.
*/
abstract class WatchLifecycle<T, L> {
abstract class WatchLifecycle<T, L, D extends Delivery<T>> {
#detach: (() => void) | undefined;
#report: ((error: unknown) => void) | undefined;
#replace: (() => T) | undefined;
readonly #closed = Promise.withResolvers<WatchEnd>();
readonly #pending: WatchFrame<T>[] = [];
#value: T;
#listener: L | undefined;
#started = false;
#scheduled = false;
#running = false;
#retired = false;
/** Set once, by `#terminate()`. */
#end: WatchEnd | undefined;
#cancellationSignal: AbortSignal | undefined;
#cancellationListener: (() => void) | undefined;
constructor(value: T, detach: () => void, report: (error: unknown) => void, replace?: () => T) {
constructor(value: T, detach: () => void, report: (error: unknown) => void) {
this.#value = value;
this.#detach = detach;
this.#report = report;
this.#replace = replace;
}
get value(): T {
@ -122,7 +121,7 @@ abstract class WatchLifecycle<T, L> {
if (this.#end !== undefined) throw new Error("Watch is stopped");
this.#started = true;
this.#listener = listener;
if (this.#pending.length > 0) this.#schedule();
this.#schedule();
}
stop(): Promise<WatchEnd> {
@ -150,34 +149,22 @@ abstract class WatchLifecycle<T, L> {
this.#terminate(failure === undefined ? { reason: "ended" } : { reason: "failed", error: failure.error });
}
/** Call `listener` with one frame. */
protected abstract invoke(listener: L, frame: WatchFrame<T>, context: Context): Promise<void>;
/** The ops of the frame an overflow delivers instead of the backlog, whose value is `value`. */
protected abstract overflowOps(value: T): readonly Op[];
/** Queue one frame; past the bound the backlog collapses to one frame of `replace()` or the newest value. */
protected enqueue(value: T, ops: readonly Op[], context: Context): void {
if (this.#end !== undefined || this.#retired) return;
if (this.#pending.length >= MAX_PENDING_WATCH_FRAMES) {
this.#pending.length = 0;
const replacement = this.#replace?.() ?? value;
this.#pending.push({ value: replacement, ops: this.overflowOps(replacement), context });
} else {
this.#pending.push({ value, ops, context });
}
if (this.#started) this.#schedule();
/** Whether the watch still takes frames: it has not ended. */
protected get open(): boolean {
return this.#end === undefined;
}
/**
* Queue the last frame, after which the watch ends `retired`. An overflowing backlog is dropped, since the frame
* replaces the root anyway.
*/
protected enqueueRetirement(value: T, ops: readonly Op[], context: Context): void {
if (this.#end !== undefined || this.#retired) return;
this.#retired = true;
if (this.#pending.length >= MAX_PENDING_WATCH_FRAMES) this.#pending.length = 0;
this.#pending.push({ value, ops, context, retires: true });
/** Take the next delivery, or undefined when nothing is pending. */
protected abstract next(): D | undefined;
/** Call `listener` with one delivery. */
protected abstract invoke(listener: L, delivery: D, context: Context): Promise<void>;
/** Drop everything pending and every reference into the source; called once, on termination. */
protected abstract release(): void;
/** Something became pending: deliver it once started, never inline. */
protected pending(): void {
if (this.#started) this.#schedule();
}
@ -190,18 +177,19 @@ abstract class WatchLifecycle<T, L> {
});
}
/** Deliver until nothing is pending; what arrives during a call is taken after it returns. */
async #drain(): Promise<void> {
if (this.#running || this.#end !== undefined || !this.#started) return;
this.#running = true;
try {
while (this.#end === undefined) {
const frame = this.#pending.shift();
const listener = this.#listener;
if (frame === undefined || listener === undefined) break;
this.#value = frame.value;
const deliveryContext = withoutAbortSignal(frame.context);
if (listener === undefined) break;
const delivery = this.next();
if (delivery === undefined) break;
this.#value = delivery.value;
try {
await this.invoke(listener, frame, deliveryContext);
await this.invoke(listener, delivery, withoutAbortSignal(delivery.context));
} catch (error) {
if (this.#end === undefined) {
this.#report?.(error);
@ -209,19 +197,18 @@ abstract class WatchLifecycle<T, L> {
}
break;
}
if (frame.retires === true) {
if (delivery.retires) {
this.#terminate({ reason: "retired" });
break;
}
}
} finally {
this.#running = false;
if (this.#end === undefined && this.#pending.length > 0) this.#schedule();
}
}
/**
* End the watch, once: detach, drop pending frames and every reference into the source, and settle `closed`. A
* End the watch, once: detach, drop what is pending and every reference into the source, and settle `closed`. A
* callback already running keeps running.
*/
#terminate(end: WatchEnd): void {
@ -230,9 +217,8 @@ abstract class WatchLifecycle<T, L> {
const detach = this.#detach;
this.#detach = undefined;
this.#report = undefined;
this.#replace = undefined;
this.#listener = undefined;
this.#pending.length = 0;
this.release();
if (this.#cancellationListener !== undefined) {
this.#cancellationSignal?.removeEventListener("abort", this.#cancellationListener);
}
@ -252,18 +238,29 @@ function toError(error: unknown): Error {
}
}
type WatchDelivery<T> = Delivery<T> & { readonly ops: readonly Op[] };
/** Canonical terminal frame of a retired source. */
const RETIREMENT_OPERATIONS: readonly Op[] = [["r", null]];
/**
* The watch handle of a JSON source: exact frames in order, an overflow becomes one root replacement `[["r", newest]]`.
* `detach` unsubscribes it from its source on termination; `report` gets a listener's throw, which also ends the
* watch; `replace` gives the value an overflow delivers, by default the newest value.
* The watch handle of a JSON source. Frames that arrive before `start()` or while the listener runs wait in one
* `PendingOps` queue; the next call gets the newest value, the folded ops of every frame since the previous call (or
* `[["r", newest]]` after the queue overflowed) and the newest frame's context. `detach` unsubscribes it from its
* source on termination; `report` gets a listener's throw, which also ends the watch.
*/
export class Watch<T> extends WatchLifecycle<T, WatchListener<T>> implements WatchHandle<T> {
export class Watch<T> extends WatchLifecycle<T, WatchListener<T>, WatchDelivery<T>> implements WatchHandle<T> {
readonly #ops = new PendingOps();
/** The newest frame, while a call is pending. */
#newest: { readonly value: T; readonly context: Context } | undefined;
#retiring = false;
/** Queue the frame `ops`, which turn the previous value into `value`. */
advance(value: T, ops: readonly Op[], context: Context): void {
this.enqueue(value, ops, context);
if (!this.open || this.#retiring) return;
this.#ops.push(ops);
this.#newest = { value, context };
this.pending();
}
/**
@ -271,41 +268,76 @@ export class Watch<T> extends WatchLifecycle<T, WatchListener<T>> implements Wat
* whose values include `null`.
*/
retire(this: Watch<T | null>, context: Context): void {
this.enqueueRetirement(null, RETIREMENT_OPERATIONS, context);
if (!this.open || this.#retiring) return;
this.#retiring = true;
this.#ops.push(RETIREMENT_OPERATIONS);
this.#newest = { value: null, context };
this.pending();
}
protected invoke(listener: WatchListener<T>, frame: WatchFrame<T>, context: Context): Promise<void> {
return listener(frame.value, frame.ops, context);
}
protected overflowOps(value: T): readonly Op[] {
protected next(): WatchDelivery<T> | undefined {
const newest = this.#newest;
if (newest === undefined) return undefined;
this.#newest = undefined;
const batch = this.#ops.take();
// Values of a JSON watch are JSON; `T` is the source's narrower or wider view of them (e.g. a typed document).
return [["r", value as JsonValue]];
const ops: readonly Op[] = batch.kind === "ops" ? batch.ops : [["r", newest.value as JsonValue]];
// Nothing is queued after a retirement, so a retiring watch's pending call is its last.
return { value: newest.value, ops, context: newest.context, retires: this.#retiring };
}
protected invoke(listener: WatchListener<T>, delivery: WatchDelivery<T>, context: Context): Promise<void> {
return listener(delivery.value, delivery.ops, context);
}
protected release(): void {
this.#ops.take();
this.#newest = undefined;
}
}
const NO_OPS: readonly Op[] = [];
/** Maximum values retained behind one unavailable native watch listener. */
const MAX_PENDING_NATIVE_VALUES = 100;
/**
* The watch handle of a native source: values in order, an overflow collapses to the newest value. `detach` and
* `report` as for `Watch`.
* The watch handle of a native source: every value in order, one call each; past 100 pending values they collapse to
* one, `replace()` or by default the newest value. `detach` and `report` as for `Watch`.
*/
export class NativeWatch<T> extends WatchLifecycle<T, NativeWatchListener<T>> implements NativeWatchHandle<T> {
constructor(value: T, detach: () => void, report: (error: unknown) => void) {
// No replacement: an overflow delivers the newest value.
super(value, detach, report, undefined);
export class NativeWatch<T>
extends WatchLifecycle<T, NativeWatchListener<T>, Delivery<T>>
implements NativeWatchHandle<T>
{
readonly #pending: Delivery<T>[] = [];
#replace: (() => T) | undefined;
constructor(value: T, detach: () => void, report: (error: unknown) => void, replace?: () => T) {
super(value, detach, report);
this.#replace = replace;
}
/** Queue `value`, the source's new value. */
advance(value: T, context: Context): void {
this.enqueue(value, NO_OPS, context);
if (!this.open) return;
if (this.#pending.length < MAX_PENDING_NATIVE_VALUES) {
this.#pending.push({ value, context, retires: false });
} else {
this.#pending.length = 0;
const replacement = this.#replace === undefined ? value : this.#replace();
this.#pending.push({ value: replacement, context, retires: false });
}
this.pending();
}
protected invoke(listener: NativeWatchListener<T>, frame: WatchFrame<T>, context: Context): Promise<void> {
return listener(frame.value, context);
protected next(): Delivery<T> | undefined {
return this.#pending.shift();
}
protected overflowOps(): readonly Op[] {
return NO_OPS;
protected invoke(listener: NativeWatchListener<T>, delivery: Delivery<T>, context: Context): Promise<void> {
return listener(delivery.value, context);
}
protected release(): void {
this.#pending.length = 0;
this.#replace = undefined;
}
}

714
packages/chord/test/delta-pending.test.ts Normal file
View file

@ -0,0 +1,714 @@
// Tests for Delta's pending queue: folding is invisible to a replica, the cap, ownership of received ops, cost, GC.
import { setFlagsFromString } from "node:v8";
import { runInNewContext } from "node:vm";
import { afterEach, describe, expect, it, vi } from "vitest";
import {
appendText,
applyImmutable,
decoder,
diffRevisions,
encoder,
type Op,
PendingOps,
type PendingTake,
type Prepared,
track,
} from "../src/delta/index.ts";
import { MAX_PENDING_OPS } from "../src/delta/pending.ts";
import type { JsonValue } from "../src/json.ts";
type JsonObject = { [key: string]: JsonValue };
type Container = JsonValue[] | JsonObject;
/** A seeded generator, so a failing sequence reproduces from its seed. */
function random(seed: number) {
let state = seed >>> 0;
const next = (): number => {
state = (Math.imul(state, 1664525) + 1013904223) >>> 0;
return state / 4294967296;
};
const int = (max: number): number => Math.floor(next() * max);
const pick = <T>(items: readonly T[]): T => items[int(items.length)]!;
// Surrogate halves and separators: an append may split a code point, and paths hold any string.
const texts = [
"a",
"bc",
" tok",
"é",
"😀",
"\u2028",
"\ud83d",
"\ude00",
"日本語",
'"q"',
"\\",
"",
"x".repeat(30),
];
const keys = ["a", "b", "0", "1", "01", "a.b", "x y", "text", "list"];
const text = (): string => pick(texts);
const primitive = (): JsonValue => pick([() => int(100), text, () => next() < 0.5, () => null])();
const value = (depth = 0): JsonValue => {
const roll = next();
if (depth > 2 || roll < 0.4) return primitive();
if (roll < 0.7) return Array.from({ length: int(4) }, () => value(depth + 1));
const object: JsonObject = {};
for (let count = int(4); count >= 0; count--) object[pick(keys)] = value(depth + 1);
return object;
};
const root = (): JsonObject => {
const object: JsonObject = {};
for (let count = int(4) + 1; count >= 0; count--) object[pick(keys)] = value();
return object;
};
return { next, int, pick, text, primitive, value, root };
}
type Random = ReturnType<typeof random>;
/** Every (parent, key) slot under `node`, parents first. */
function slots(node: JsonValue, out: Array<[Container, string | number]> = []): Array<[Container, string | number]> {
if (node === null || typeof node !== "object") return out;
const keys: Array<string | number> = Array.isArray(node) ? node.map((_, index) => index) : Object.keys(node);
for (const key of keys) {
out.push([node, key]);
slots((node as JsonObject)[key]!, out);
}
return out;
}
const read = (parent: Container, key: string | number): JsonValue => (parent as JsonObject)[key]!;
const write = (parent: Container, key: string | number, value: JsonValue): void => {
(parent as JsonObject)[key] = value;
};
type Append = (parent: JsonObject, key: string, text: string) => void;
/** Appends through `appendText()`, which a draft records without comparing strings. */
const appendToDraft: Append = (parent, key, text) => appendText(parent as never, key as never, text);
const appendToValue: Append = (parent, key, text) => {
parent[key] = `${parent[key] as string}${text}`;
};
/** One random edit anywhere in `draft`: string appends and trims, array splices and reorders, object edits. */
function mutateAnywhere(draft: JsonObject, rnd: Random, append: Append): void {
const all = slots(draft);
if (all.length === 0) {
draft[rnd.pick(["a", "b"])] = rnd.value();
return;
}
const [parent, key] = rnd.pick(all);
const current = read(parent, key);
const roll = rnd.next();
if (typeof current === "string") {
if (roll < 0.3 && !Array.isArray(parent)) append(parent, key as string, rnd.text());
else if (roll < 0.6) write(parent, key, current + rnd.text());
else if (roll < 0.8) write(parent, key, current.slice(Math.min(current.length, rnd.int(3))));
else write(parent, key, rnd.primitive());
} else if (Array.isArray(current)) {
if (roll < 0.3) current.push(rnd.value());
else if (roll < 0.5)
current.splice(rnd.int(current.length + 1), rnd.int(2), ...(rnd.next() < 0.5 ? [rnd.value()] : []));
else if (roll < 0.65 && current.length > 0) current[rnd.int(current.length)] = rnd.value();
else if (roll < 0.75 && current.length > 1) current.reverse();
else if (roll < 0.85 && current.length > 1)
current.sort((a, b) => JSON.stringify(a).localeCompare(JSON.stringify(b)));
else if (roll < 0.95) write(parent, key, rnd.value());
else current.length = 0;
} else if (current !== null && typeof current === "object") {
const keys = Object.keys(current);
if (roll < 0.4) current[rnd.pick(["a", "b", "1", "x y"])] = rnd.value();
else if (roll < 0.6 && keys.length > 0) delete current[rnd.pick(keys)];
else write(parent, key, rnd.value());
} else {
write(parent, key, rnd.next() < 0.7 ? rnd.primitive() : rnd.value());
}
}
/** Leaf edits of a list of elements interleaved with splices, pushes and reorders of the list. */
function mutateList(draft: JsonObject, rnd: Random): void {
const list = draft.list;
if (!Array.isArray(list)) {
draft.list = [
{ x: "", n: 0 },
{ x: "", n: 0 },
{ x: "", n: 0 },
];
return;
}
const roll = rnd.next();
if (roll < 0.45 && list.length > 0) {
const element = list[rnd.int(list.length)];
if (element !== null && typeof element === "object" && !Array.isArray(element)) {
if (rnd.next() < 0.3) appendText(element as never, "x" as never, rnd.text());
else if (rnd.next() < 0.5) element.x = `${element.x as string}${rnd.text()}`;
else element.n = rnd.int(9);
} else list[rnd.int(list.length)] = { x: "", n: 0 };
} else if (roll < 0.6)
list.splice(rnd.int(list.length + 1), rnd.int(2), ...(rnd.next() < 0.6 ? [{ x: rnd.text(), n: 0 }] : []));
else if (roll < 0.7) list.push({ x: "", n: 0 });
else if (roll < 0.8 && list.length > 1) list.reverse();
else if (roll < 0.9 && list.length > 0) list[rnd.int(list.length)] = rnd.primitive();
else if (list.length > 0) list.splice(rnd.int(list.length), 1);
}
/** Edits of a root array: pushes, splices, reorders, element sets and shifts. */
function mutateRootArray(draft: JsonValue[], rnd: Random): void {
switch (rnd.int(7)) {
case 0:
draft.push(rnd.value());
break;
case 1:
draft.splice(rnd.int(draft.length + 1), rnd.int(4), rnd.value());
break;
case 2:
draft.reverse();
break;
case 3:
draft.sort((a, b) => JSON.stringify(a).localeCompare(JSON.stringify(b)));
break;
case 4:
if (draft.length > 0) draft[rnd.int(draft.length)] = rnd.value();
break;
case 5:
draft.shift();
break;
default:
draft.splice(rnd.int(draft.length + 1), rnd.int(4));
}
}
/** One sequence of commits from a value: the commits' op batches and the value after each. */
type Sequence = { readonly initial: JsonValue; readonly commits: Array<{ ops: readonly Op[]; value: JsonValue }> };
function trackedSequence<T extends object>(
initial: T,
commits: number,
rnd: Random,
mutate: (draft: T) => void,
replace: () => T,
): Sequence {
const tracker = track(structuredClone(initial));
const out: Sequence["commits"] = [];
for (let index = 0; index < commits; index++) {
let prepared: Prepared<T>;
if (rnd.next() < 0.03) prepared = tracker.prepareReplace(replace());
else {
const change = tracker.beginChange();
for (let edits = 1 + rnd.int(4); edits > 0; edits--) mutate(change.state as T);
prepared = change.prepare();
}
tracker.adopt(prepared);
out.push({ ops: prepared.ops, value: tracker.value as JsonValue });
}
return { initial: initial as JsonValue, commits: out };
}
/** Diffs between values, each an edit of the previous one or a new value. */
function diffSequence(rnd: Random, commits: number): Sequence {
const initial = rnd.value();
const out: Sequence["commits"] = [];
let value = initial;
for (let index = 0; index < commits; index++) {
const edited: JsonObject = { value: structuredClone(value) };
for (let edits = 1 + rnd.int(3); edits > 0; edits--) mutateAnywhere(edited, rnd, appendToValue);
const next = rnd.next() < 0.2 ? rnd.value() : edited.value!;
out.push({ ops: diffRevisions(value, next), value: next });
value = next;
}
return { initial, commits: out };
}
/**
* Feed the sequence through one queue, taking at random points as a slow consumer would, and replay what was taken
* on a replica, once as ops and once through Delta's encoder and decoder. Both must reach the final value; the received
* ops must be untouched.
*/
function checkSequence(sequence: Sequence, rnd: Random): { ops: number; folded: number } {
for (const commit of sequence.commits) {
// Frozen: a queue that wrote into a received batch or op tuple throws here.
for (const op of commit.ops) Object.freeze(op);
Object.freeze(commit.ops);
}
const received = JSON.stringify(sequence.commits.map((commit) => commit.ops));
const queue = new PendingOps();
const taken: Array<{ batch: PendingTake; value: JsonValue }> = [];
for (const [index, commit] of sequence.commits.entries()) {
queue.push(commit.ops);
if (rnd.next() < 0.1 || index === sequence.commits.length - 1)
taken.push({ batch: queue.take(), value: commit.value });
}
const final = sequence.commits.length === 0 ? sequence.initial : sequence.commits.at(-1)!.value;
let replica = structuredClone(sequence.initial);
const encode = encoder();
const decode = decoder();
let decoded = applyImmutable<JsonValue>(
null,
decode.decode(JSON.parse(JSON.stringify(encode.encode([["r", replica]])))),
);
let folded = 0;
for (const { batch, value } of taken) {
const ops: readonly Op[] = batch.kind === "ops" ? batch.ops : [["r", value]];
folded += ops.length;
replica = applyImmutable(replica, ops);
decoded = applyImmutable(decoded, decode.decode(JSON.parse(JSON.stringify(encode.encode(ops)))));
}
expect(replica).toEqual(final);
expect(decoded).toEqual(final);
expect(JSON.stringify(sequence.commits.map((commit) => commit.ops))).toBe(received);
return { ops: sequence.commits.reduce((sum, commit) => sum + commit.ops.length, 0), folded };
}
describe("PendingOps folding", () => {
it("is invisible to a replica: tracker edits anywhere, appendText, splices and reorders", () => {
const rnd = random(1);
let ops = 0;
let folded = 0;
for (let run = 0; run < 1500; run++) {
const sequence = trackedSequence(
rnd.root(),
1 + rnd.int(30),
rnd,
(draft) => mutateAnywhere(draft, rnd, appendToDraft),
rnd.root,
);
const result = checkSequence(sequence, rnd);
ops += result.ops;
folded += result.folded;
}
// The sequences fold: a fuzzer whose ops never merge proves nothing.
expect(folded).toBeLessThan(ops * 0.9);
});
it("is invisible to a replica: leaf edits of list elements between splices and reorders of the list", () => {
const rnd = random(2);
let ops = 0;
let folded = 0;
for (let run = 0; run < 1500; run++) {
const initial: JsonObject = {
list: [
{ x: "a", n: 0 },
{ x: "b", n: 1 },
],
};
const sequence = trackedSequence(
initial,
5 + rnd.int(40),
rnd,
(draft) => mutateList(draft, rnd),
() => ({ list: [] }),
);
const result = checkSequence(sequence, rnd);
ops += result.ops;
folded += result.folded;
}
expect(folded).toBeLessThan(ops * 0.9);
});
it("is invisible to a replica: a root array", () => {
const rnd = random(3);
for (let run = 0; run < 500; run++) {
const initial = Array.from({ length: rnd.int(6) }, () => rnd.value());
const replace = () => Array.from({ length: rnd.int(6) }, () => rnd.value());
checkSequence(
trackedSequence(initial, 40, rnd, (draft) => mutateRootArray(draft, rnd), replace),
rnd,
);
}
});
it("is invisible to a replica: diffRevisions between arbitrary values", () => {
const rnd = random(4);
for (let run = 0; run < 800; run++) checkSequence(diffSequence(rnd, 30), rnd);
});
});
/** Push each op as its own batch, take, and check the replay; returns the folded ops. */
function foldOneByOne(initial: JsonValue, ops: readonly Op[]): readonly Op[] {
const queue = new PendingOps();
for (const op of ops) queue.push([op]);
const batch = queue.take();
if (batch.kind !== "ops") throw new Error("unexpected overflow");
const expected = applyImmutable(structuredClone(initial), ops);
expect(applyImmutable(structuredClone(initial), batch.ops)).toEqual(expected);
const encode = encoder();
const decode = decoder();
const base = applyImmutable(null, decode.decode(JSON.parse(JSON.stringify(encode.encode([["r", initial]])))));
expect(applyImmutable(base, decode.decode(JSON.parse(JSON.stringify(encode.encode(batch.ops)))))).toEqual(expected);
return batch.ops;
}
describe("PendingOps adversarial sequences", () => {
it("a leaf set in an array element, a splice before it, the same path set again", () => {
const ops: Op[] = [
["s", ["arr", 2, "x"], 1],
["p", ["arr"], 0, 0, [{ x: 9 }]],
["s", ["arr", 2, "x"], 2],
];
expect(foldOneByOne({ arr: [{ x: 0 }, { x: 0 }, { x: 0 }] }, ops)).toEqual(ops);
const appends: Op[] = [
["a", ["arr", 0, "x"], "1"],
["p", ["arr"], 0, 0, [{ x: "new" }]],
["a", ["arr", 0, "x"], "2"],
];
expect(foldOneByOne({ arr: [{ x: "a" }, { x: "b" }] }, appends)).toEqual(appends);
});
it("`t` then `a`, and appends around a trim", () => {
expect(
foldOneByOne({ s: "hello" }, [
["t", ["s"], 2],
["a", ["s"], "!"],
]),
).toHaveLength(2);
expect(
foldOneByOne({ s: "hello" }, [
["a", ["s"], "AB"],
["t", ["s"], 3],
["a", ["s"], "C"],
]),
).toHaveLength(3);
});
it("an object `s` at a parent, then leaf sets below it", () => {
expect(
foldOneByOne({ o: { a: 1 } }, [
["s", ["o", "a"], 2],
["s", ["o"], { a: 5, b: 6 }],
["s", ["o", "a"], 7],
]),
).toHaveLength(3);
expect(
foldOneByOne({ x: { s: "a" } }, [
["s", ["x"], { s: "b" }],
["a", ["x", "s"], "c"],
["s", ["x", "s"], "d"],
]),
).toEqual([
["s", ["x"], { s: "b" }],
["s", ["x", "s"], "d"],
]);
});
it("an ancestor primitive `s` between leaf ops below it", () => {
expect(
foldOneByOne({ o: { a: "x", b: "y" } }, [
["a", ["o", "a"], "1"],
["s", ["o", "b"], "q"],
["s", ["o"], "flat"],
["a", ["o"], "!"],
]),
).toEqual([
["a", ["o", "a"], "1"],
["s", ["o", "b"], "q"],
["s", ["o"], "flat!"],
]);
expect(
foldOneByOne({ a: { b: 0 } }, [
["s", ["a", "b"], 1],
["s", ["a"], null],
["s", ["a"], "x"],
]),
).toEqual([
["s", ["a", "b"], 1],
["s", ["a"], "x"],
]);
});
it("deletes, moves and splices shift what a path names", () => {
expect(
foldOneByOne({ arr: ["a", "b", "c"] }, [
["s", ["arr", 2], "C"],
["d", ["arr", 0]],
["s", ["arr", 1], "Z"],
]),
).toHaveLength(3);
expect(
foldOneByOne({ arr: [1, 2, 3] }, [
["s", ["arr", 0], 10],
["m", ["arr"], [2, 1, 0]],
["s", ["arr", 0], 20],
]),
).toHaveLength(3);
expect(
foldOneByOne({ arr: [{ x: 0 }] }, [
["s", ["arr", 0, "x"], 1],
["p", ["arr"], 1, 0, [{ x: 0 }]],
["s", ["arr", 0, "x"], 2],
["s", ["arr", 1, "x"], 3],
]),
).toHaveLength(4);
});
it("keys of one parent never collide: `a.b` and `a`/`b`, `1` on an object and on an array", () => {
foldOneByOne({ a: { b: "x" }, "a.b": "z" }, [
["a", ["a", "b"], "1"],
["s", ["a.b"], "y"],
["a", ["a", "b"], "2"],
]);
foldOneByOne({ o: { "1": "a" }, arr: ["x", "y"] }, [
["s", ["o", "1"], "b"],
["s", ["arr", 1], "z"],
["s", ["o", "1"], "c"],
["s", ["arr", 1], "w"],
]);
});
it("an array set at the end appends, and a later set of that index folds into it", () => {
expect(
foldOneByOne({ arr: [1] }, [
["s", ["arr", 1], 2],
["s", ["arr", 0], 9],
["s", ["arr", 1], 3],
]),
).toEqual([
["s", ["arr", 1], 3],
["s", ["arr", 0], 9],
]);
});
it("`r` drops everything before it", () => {
expect(
foldOneByOne({ a: 1 }, [
["s", ["a"], 2],
["r", { b: "x" }],
["a", ["b"], "y"],
]),
).toEqual([
["r", { b: "x" }],
["a", ["b"], "y"],
]);
});
it("joins appends, folds appends into a string set, and keeps split surrogates", () => {
expect(
foldOneByOne({ s: "" }, [
["s", ["s"], "ab"],
["a", ["s"], "c"],
["s", ["t"], 1],
["a", ["s"], "d"],
]),
).toEqual([
["s", ["s"], "abcd"],
["s", ["t"], 1],
]);
expect(
foldOneByOne({ s: "" }, [
["a", ["s"], "\ud83d"],
["s", ["n"], 1],
["a", ["s"], "\ude00"],
["a", ["s"], "é"],
]),
).toEqual([
["a", ["s"], "😀é"],
["s", ["n"], 1],
]);
});
it("interleaved leaves, a structural op, then the same leaves again", () => {
expect(
foldOneByOne({ a: "", b: "", list: [0] }, [
["a", ["a"], "1"],
["a", ["b"], "1"],
["a", ["a"], "2"],
["p", ["list"], 1, 0, [1]],
["a", ["b"], "2"],
["a", ["a"], "3"],
]),
).toEqual([
["a", ["a"], "12"],
["a", ["b"], "1"],
["p", ["list"], 1, 0, [1]],
["a", ["b"], "2"],
["a", ["a"], "3"],
]);
});
});
/** `count` set ops on distinct paths. */
const distinctSets = (count: number, from = 0): Op[] =>
Array.from({ length: count }, (_, index): Op => ["s", ["k", from + index], index]);
describe("PendingOps cap", () => {
it(`keeps ${MAX_PENDING_OPS} pending ops and replaces past them`, () => {
const queue = new PendingOps();
queue.push([]);
queue.push(distinctSets(MAX_PENDING_OPS));
const exact = queue.take();
expect(exact.kind === "ops" && exact.ops.length).toBe(MAX_PENDING_OPS);
queue.push(distinctSets(MAX_PENDING_OPS / 2));
queue.push(distinctSets(MAX_PENDING_OPS / 2, MAX_PENDING_OPS / 2));
const full = queue.take();
expect(full.kind === "ops" && full.ops.length).toBe(MAX_PENDING_OPS);
queue.push(distinctSets(MAX_PENDING_OPS / 2));
queue.push(distinctSets(MAX_PENDING_OPS / 2 + 1, MAX_PENDING_OPS / 2));
expect(queue.take()).toEqual({ kind: "replace" });
});
it("counts ops after folding", () => {
const queue = new PendingOps();
for (let index = 0; index < 3 * MAX_PENDING_OPS; index++) queue.push([["a", ["text"], "x"]]);
expect(queue.take()).toEqual({ kind: "ops", ops: [["a", ["text"], "x".repeat(3 * MAX_PENDING_OPS)]] });
});
it("accumulates nothing once replaced, until taken or a batch starting with `r`", () => {
const queue = new PendingOps();
queue.push(distinctSets(1));
queue.push(distinctSets(MAX_PENDING_OPS + 1));
queue.push(distinctSets(3));
expect(queue.take()).toEqual({ kind: "replace" });
expect(queue.take()).toEqual({ kind: "ops", ops: [] });
queue.push(distinctSets(1));
queue.push(distinctSets(MAX_PENDING_OPS + 1));
const fresh: readonly Op[] = [["r", { k: [0] }]];
queue.push(fresh);
const taken = queue.take();
expect(taken.kind === "ops" && taken.ops).toBe(fresh);
queue.push(distinctSets(1));
queue.push(fresh);
queue.push([["s", ["n"], 1]]);
expect(queue.take()).toEqual({ kind: "ops", ops: [...fresh, ["s", ["n"], 1]] });
});
it("hands one batch over as received, whatever its size", () => {
const queue = new PendingOps();
const batch = distinctSets(MAX_PENDING_OPS + 1);
queue.push(batch);
const taken = queue.take();
expect(taken.kind === "ops" && taken.ops).toBe(batch);
queue.push(batch);
queue.push([["s", ["n"], 1]]);
expect(queue.take()).toEqual({ kind: "replace" });
});
});
describe("PendingOps ownership", () => {
it("hands over a received batch by identity and never mutates batches or op tuples", () => {
const queue = new PendingOps();
const first: readonly Op[] = Object.freeze([Object.freeze(["a", ["s"], "1"] as const)]);
queue.push(first);
const alone = queue.take();
expect(alone.kind === "ops" && alone.ops).toBe(first);
const second: readonly Op[] = Object.freeze([Object.freeze(["a", ["s"], "2"] as const)]);
queue.push(first);
queue.push(second);
expect(queue.take()).toEqual({ kind: "ops", ops: [["a", ["s"], "12"]] });
expect(first).toEqual([["a", ["s"], "1"]]);
expect(second).toEqual([["a", ["s"], "2"]]);
});
});
describe("PendingOps size", () => {
it("counts pending ops after folding, the replacement once overflowed, and nothing once taken", () => {
const queue = new PendingOps();
expect(queue.size).toBe(0);
queue.push([]);
expect(queue.size).toBe(0);
queue.push([
["a", ["text"], "a"],
["s", ["n"], 1],
]);
expect(queue.size).toBe(2);
queue.push([
["a", ["text"], "b"],
["s", ["n"], 2],
["p", ["list"], 0, 0, [1]],
]);
expect(queue.size).toBe(3);
queue.take();
expect(queue.size).toBe(0);
queue.push(distinctSets(MAX_PENDING_OPS));
expect(queue.size).toBe(MAX_PENDING_OPS);
queue.push([["s", ["n"], 1]]);
expect(queue.size).toBe(1);
expect(queue.take()).toEqual({ kind: "replace" });
expect(queue.size).toBe(0);
});
});
describe("PendingOps cost", () => {
afterEach(() => {
vi.restoreAllMocks();
});
it("computes one path key per leaf op once folding, none for one batch or structural ops", () => {
const stringify = vi.spyOn(JSON, "stringify");
const queue = new PendingOps();
const leaves: Op[] = [
["a", ["entries", 0, "text"], "a"],
["s", ["meta", "tokens"], 1],
];
queue.push(leaves);
queue.take();
expect(stringify).toHaveBeenCalledTimes(0);
queue.push(leaves);
expect(stringify).toHaveBeenCalledTimes(0);
queue.push([...leaves, ["p", ["entries"], 1, 0, [{ text: "" }]], ["s", ["meta"], { tokens: 0 }], ["d", ["x"]]]);
expect(stringify).toHaveBeenCalledTimes(4);
});
it("keeps one op per leaf path however many commits write it", () => {
const queue = new PendingOps();
for (let index = 0; index < 50_000; index++) {
queue.push([
["a", ["entries", 0, "text"], "tok "],
["s", ["meta", "tokens"], index],
]);
}
const taken = queue.take();
expect(taken.kind === "ops" && taken.ops).toEqual([
["a", ["entries", 0, "text"], "tok ".repeat(50_000)],
["s", ["meta", "tokens"], 49_999],
]);
});
});
const collect = (() => {
setFlagsFromString("--expose-gc");
const gc = runInNewContext("gc") as () => void;
return async (): Promise<void> => {
for (let round = 0; round < 5; round++) {
await new Promise((resolve) => setTimeout(resolve, 0));
gc();
}
};
})();
describe("PendingOps references", () => {
it("releases a folded-over op, and the index and everything pending once taken", async () => {
const queue = new PendingOps();
const pending = (): { payload: WeakRef<object>; replaced: WeakRef<object> } => {
const payload = { big: "x".repeat(1000) };
const replaced: Op = ["s", ["n"], 1];
queue.push([replaced]);
queue.push([
["s", ["n"], 2],
["s", ["o"], payload],
]);
return { payload: new WeakRef(payload), replaced: new WeakRef(replaced) };
};
const first = pending();
await collect();
// The set of `n` replaced the first one: only its newest op is held.
expect(first.replaced.deref()).toBeUndefined();
expect(first.payload.deref()).toBeDefined();
queue.take();
await collect();
expect(first.payload.deref()).toBeUndefined();
// Taken while the folding state holds the index: nothing stays behind in the queue.
const second = pending();
queue.push([["p", ["list"], 0, 0, [1]]]);
queue.take();
await collect();
expect(second.payload.deref()).toBeUndefined();
});
});

318
packages/chord/test/watch.test.ts
View file

@ -1,7 +1,7 @@
// Tests for Chord's watch handles: the shared lifecycle through `Watch`, and what `NativeWatch` does differently.
import { setFlagsFromString } from "node:v8";
import { runInNewContext } from "node:vm";
import { describe, expect, expectTypeOf, it } from "vitest";
import { afterEach, describe, expect, expectTypeOf, it, vi } from "vitest";
import {
BACKGROUND_CONTEXT,
type Context,
@ -9,7 +9,8 @@ import {
withAbortSignal,
withContextValue,
} from "../src/context/index.ts";
import type { Op } from "../src/delta/index.ts";
import { applyImmutable, type Op, track } from "../src/delta/index.ts";
import { MAX_PENDING_OPS } from "../src/delta/pending.ts";
import type { JsonValue } from "../src/json.ts";
import { NativeWatch, Watch, type WatchEnd } from "../src/watch.ts";
@ -25,20 +26,19 @@ function deferred(): { readonly promise: Promise<void>; readonly resolve: () =>
}
/** A watch whose detach and reports are recorded. */
function recordedWatch<T>(value: T, replace?: () => T) {
function recordedWatch<T>(value: T) {
const recorded = { detached: 0, reported: [] as unknown[] };
const watch = new Watch<T>(
value,
() => recorded.detached++,
(error) => recorded.reported.push(error),
replace,
);
return { watch, recorded };
}
/** A started watch collecting its frames. */
function collectingWatch<T>(value: T, replace?: () => T) {
const { watch, recorded } = recordedWatch(value, replace);
/** A started watch collecting its calls. */
function collectingWatch<T>(value: T) {
const { watch, recorded } = recordedWatch(value);
const frames: Array<{ value: T; ops: readonly Op[] }> = [];
watch.start(async (delivered, ops) => {
frames.push({ value: delivered, ops });
@ -67,12 +67,20 @@ const collect = (() => {
})();
describe("Watch delivery", () => {
it("keeps the acquisition value until start and delivers exact frames in order", async () => {
const { watch } = recordedWatch<JsonValue>({ n: 0 });
watch.advance({ n: 1 }, [["s", ["n"], 1]], context);
watch.advance({ n: 2 }, [["s", ["n"], 2]], context);
it("keeps the acquisition value until start, then delivers the frames since in one call", async () => {
const { watch } = recordedWatch<JsonValue>({ n: 0, text: "" });
watch.advance({ n: 1, text: "" }, [["s", ["n"], 1]], context);
watch.advance(
{ n: 2, text: "a" },
[
["s", ["n"], 2],
["a", ["text"], "a"],
],
context,
);
watch.advance({ n: 2, text: "ab" }, [["a", ["text"], "b"]], context);
await flush();
expect(watch.value).toEqual({ n: 0 });
expect(watch.value).toEqual({ n: 0, text: "" });
const frames: Array<{ value: JsonValue; ops: readonly Op[] }> = [];
watch.start(async (value, ops) => {
@ -81,10 +89,32 @@ describe("Watch delivery", () => {
expect(frames).toEqual([]);
await flush();
expect(frames).toEqual([
{ value: { n: 1 }, ops: [["s", ["n"], 1]] },
{ value: { n: 2 }, ops: [["s", ["n"], 2]] },
{
value: { n: 2, text: "ab" },
ops: [
["s", ["n"], 2],
["a", ["text"], "ab"],
],
},
]);
expect(watch.value).toEqual({ n: 2 });
expect(watch.value).toEqual({ n: 2, text: "ab" });
});
it("passes a frame's ops on by identity to a listener that keeps up", async () => {
const { watch } = recordedWatch<JsonValue>(0);
const delivered: Array<readonly Op[]> = [];
watch.start(async (_value, ops) => {
delivered.push(ops);
});
const first: readonly Op[] = [["r", 1]];
const second: readonly Op[] = [["r", 2]];
watch.advance(1, first, context);
await flush();
watch.advance(2, second, context);
await flush();
expect(delivered).toHaveLength(2);
expect(delivered[0]).toBe(first);
expect(delivered[1]).toBe(second);
});
it("never invokes the listener inline", async () => {
@ -95,98 +125,159 @@ describe("Watch delivery", () => {
expect(frames).toEqual([{ value: 1, ops: [["r", 1]] }]);
});
it("awaits each callback and buffers frames advanced while one is in flight", async () => {
const { watch } = recordedWatch<JsonValue>(0);
it("awaits each callback and folds frames advanced while one is in flight into the next call", async () => {
const { watch } = recordedWatch<JsonValue>({ text: "" });
const gate = deferred();
const calls: string[] = [];
watch.start(async (value) => {
calls.push(`start ${value}`);
if (value === 1) await gate.promise;
calls.push(`end ${value}`);
const frames: Array<{ value: JsonValue; ops: readonly Op[] }> = [];
watch.start(async (value, ops) => {
calls.push(`start ${JSON.stringify(value)}`);
frames.push({ value, ops });
if (frames.length === 1) await gate.promise;
calls.push(`end ${JSON.stringify(value)}`);
});
watch.advance(1, [["r", 1]], context);
watch.advance({ text: "a" }, [["a", ["text"], "a"]], context);
await flush();
watch.advance(2, [["r", 2]], context);
watch.advance(3, [["r", 3]], context);
watch.advance({ text: "ab" }, [["a", ["text"], "b"]], context);
watch.advance({ text: "abc" }, [["a", ["text"], "c"]], context);
await flush();
expect(calls).toEqual(["start 1"]);
expect(calls).toEqual(['start {"text":"a"}']);
gate.resolve();
await flush();
expect(calls).toEqual(["start 1", "end 1", "start 2", "end 2", "start 3", "end 3"]);
expect(calls).toEqual(['start {"text":"a"}', 'end {"text":"a"}', 'start {"text":"abc"}', 'end {"text":"abc"}']);
expect(frames[1]).toEqual({ value: { text: "abc" }, ops: [["a", ["text"], "bc"]] });
});
it("delivers a context with the frame's values and without its abort signal", async () => {
it("delivers the newest frame's context, without its abort signal", async () => {
const { watch } = recordedWatch<JsonValue>(0);
const key = createContextKey<string>("frame");
const controller = new AbortController();
const frameContext = withContextValue(key, "first", withAbortSignal(controller.signal, context));
let delivered: Context | undefined;
const delivered: Context[] = [];
watch.start(async (_value, _ops, deliveryContext) => {
delivered = deliveryContext;
delivered.push(deliveryContext);
});
watch.advance(1, [["r", 1]], frameContext);
watch.advance(1, [["r", 1]], withContextValue(key, "first", context));
watch.advance(2, [["r", 2]], withContextValue(key, "second", withAbortSignal(controller.signal, context)));
await flush();
expect(delivered?.value(key)).toBe("first");
expect(delivered?.abortSignal).toBeUndefined();
expect(delivered).toHaveLength(1);
expect(delivered[0]?.value(key)).toBe("second");
expect(delivered[0]?.abortSignal).toBeUndefined();
});
});
describe("Watch with a slow listener", () => {
afterEach(() => {
vi.useRealTimers();
});
it("gets one call per batch whose ops turn the previous delivered value into the newest", async () => {
vi.useFakeTimers();
type Doc = { entries: Array<{ text: string; status: string }>; tokens: number };
const tracker = track<Doc>({ entries: [], tokens: 0 });
const { watch } = recordedWatch<Doc>(tracker.value);
const commit = (edit: (draft: Doc) => void): void => {
const change = tracker.beginChange();
edit(change.state as Doc);
const prepared = change.prepare();
tracker.adopt(prepared);
watch.advance(tracker.value, prepared.ops, context);
};
let replica: Doc = tracker.value;
const calls: Array<{ commits: number; ops: number }> = [];
let commits = 0;
watch.start(async (value, ops) => {
replica = applyImmutable(replica, ops);
expect(replica).toEqual(value);
calls.push({ commits, ops: ops.length });
// A listener taking a second per call, as a slow client would.
await new Promise((resolve) => setTimeout(resolve, 1000));
});
// Ten seconds of a streamed reply at 60 tokens per second, a new entry every second.
for (let tick = 0; tick < 600; tick++) {
if (tick % 60 === 0) commit((draft) => void draft.entries.push({ text: "", status: "streaming" }));
commit((draft) => {
draft.entries.at(-1)!.text += "tok ";
draft.tokens++;
});
commits++;
await vi.advanceTimersByTimeAsync(1000 / 60);
}
await vi.advanceTimersByTimeAsync(2000);
expect(replica).toEqual(tracker.value);
// One call per second, each carrying at most one op per path written since the previous one.
expect(calls.length).toBeGreaterThanOrEqual(10);
expect(calls.length).toBeLessThanOrEqual(12);
for (const call of calls) expect(call.ops).toBeLessThanOrEqual(5);
await watch.stop();
});
});
/** Advance `watch` by one frame per element of `{ k: [0, ..., count - 1] }`, each setting its own path. */
function advanceDistinct(watch: Watch<JsonValue>, count: number): JsonValue {
const values: number[] = [];
for (let index = 0; index < count; index++) {
values.push(index);
watch.advance({ k: [...values] }, [["s", ["k", index], index]], context);
}
return { k: values };
}
describe("Watch overflow", () => {
it("collapses 101 pending frames to one root replacement of the newest value", async () => {
const { watch } = recordedWatch<JsonValue>(0);
for (let n = 1; n <= 101; n++) watch.advance(n, [["r", n]], context);
const frames: Array<{ value: JsonValue; ops: readonly Op[] }> = [];
watch.start(async (value, ops) => {
frames.push({ value, ops });
});
it(`keeps ${MAX_PENDING_OPS} pending ops after folding and replaces the root past them`, async () => {
const exact = collectingWatch<JsonValue>({ k: [] });
advanceDistinct(exact.watch, MAX_PENDING_OPS);
await flush();
expect(frames).toEqual([{ value: 101, ops: [["r", 101]] }]);
expect(exact.frames).toHaveLength(1);
expect(exact.frames[0]?.ops).toHaveLength(MAX_PENDING_OPS);
const over = collectingWatch<JsonValue>({ k: [] });
const newest = advanceDistinct(over.watch, MAX_PENDING_OPS + 1);
await flush();
expect(over.frames).toEqual([{ value: newest, ops: [["r", newest]] }]);
});
it("keeps 100 pending frames exact", async () => {
const { watch } = recordedWatch<JsonValue>(0);
for (let n = 1; n <= 100; n++) watch.advance(n, [["r", n]], context);
const values: JsonValue[] = [];
watch.start(async (value) => {
values.push(value);
});
await flush();
expect(values).toEqual(Array.from({ length: 100 }, (_, index) => index + 1));
});
it("delivers the value of `replace` on overflow", async () => {
const { watch } = recordedWatch<JsonValue>(0, () => "replacement");
for (let n = 1; n <= 101; n++) watch.advance(n, [], context);
const frames: Array<{ value: JsonValue; ops: readonly Op[] }> = [];
watch.start(async (value, ops) => {
frames.push({ value, ops });
});
await flush();
expect(frames).toEqual([{ value: "replacement", ops: [["r", "replacement"]] }]);
});
it("never folds the in-flight frame into an overflow", async () => {
const { watch } = recordedWatch<JsonValue>(0);
it("never folds the in-flight call into an overflow", async () => {
const { watch } = recordedWatch<JsonValue>({ k: [] });
const gate = deferred();
const values: JsonValue[] = [];
watch.start(async (value) => {
values.push(value);
if (value === 1) await gate.promise;
const frames: Array<{ value: JsonValue; ops: readonly Op[] }> = [];
watch.start(async (value, ops) => {
frames.push({ value, ops });
if (frames.length === 1) await gate.promise;
});
watch.advance(1, [["r", 1]], context);
watch.advance({ k: [], first: true }, [["s", ["first"], true]], context);
await flush();
for (let n = 2; n <= 102; n++) watch.advance(n, [["r", n]], context);
const newest = advanceDistinct(watch, MAX_PENDING_OPS + 1);
gate.resolve();
await flush();
expect(values).toEqual([1, 102]);
expect(frames).toEqual([
{ value: { k: [], first: true }, ops: [["s", ["first"], true]] },
{ value: newest, ops: [["r", newest]] },
]);
});
it("starts folding again after an overflow was delivered", async () => {
const { watch, frames } = collectingWatch<JsonValue>({ k: [] });
advanceDistinct(watch, MAX_PENDING_OPS + 1);
await flush();
watch.advance({ n: 1 }, [["r", { n: 1 }]], context);
watch.advance({ n: 2 }, [["s", ["n"], 2]], context);
await flush();
expect(frames.at(-1)).toEqual({
value: { n: 2 },
ops: [
["r", { n: 1 }],
["s", ["n"], 2],
],
});
});
});
describe("Watch retirement", () => {
it('delivers `[["r", null]]` as the last frame and then ends retired', async () => {
it('delivers `[["r", null]]` as the last call and then ends retired', async () => {
const { watch, frames, recorded } = collectingWatch<JsonValue>(1);
const settled = endOf(watch);
watch.advance(2, [["r", 2]], context);
await flush();
watch.retire(context);
watch.advance(3, [["r", 3]], context);
await flush();
@ -199,22 +290,25 @@ describe("Watch retirement", () => {
expect(recorded.detached).toBe(1);
});
it("drops an overflowing backlog for the retirement frame", async () => {
const { watch } = recordedWatch<JsonValue>(0);
for (let n = 1; n <= 100; n++) watch.advance(n, [["r", n]], context);
watch.retire(context);
const frames: Array<{ value: JsonValue; ops: readonly Op[] }> = [];
watch.start(async (value, ops) => {
frames.push({ value, ops });
});
expect(await watch.closed).toEqual({ reason: "retired" });
expect(frames).toEqual([{ value: null, ops: [["r", null]] }]);
it("drops pending frames, also an overflow, for the retirement", async () => {
for (const pending of [3, MAX_PENDING_OPS + 1]) {
const { watch } = recordedWatch<JsonValue>({ k: [] });
advanceDistinct(watch, pending);
watch.retire(context);
const frames: Array<{ value: JsonValue; ops: readonly Op[] }> = [];
watch.start(async (value, ops) => {
frames.push({ value, ops });
});
expect(await watch.closed).toEqual({ reason: "retired" });
expect(frames).toEqual([{ value: null, ops: [["r", null]] }]);
}
});
it("ends retired only on its source's retirement, never on a null value", async () => {
const { watch, frames } = collectingWatch<JsonValue>(1);
const settled = endOf(watch);
watch.advance(null, [["r", null]], context);
await flush();
watch.advance(2, [["r", 2]], context);
await flush();
expect(frames).toEqual([
@ -311,13 +405,17 @@ describe("Watch ends", () => {
it("ends `listener_error` on a throw, reports it, and delivers nothing more", async () => {
const { watch, recorded } = recordedWatch<JsonValue>(0);
const failure = new Error("listener failed");
const gate = deferred();
const values: JsonValue[] = [];
watch.start(async (value) => {
values.push(value);
await gate.promise;
throw failure;
});
watch.advance(1, [["r", 1]], context);
await flush();
watch.advance(2, [["r", 2]], context);
gate.resolve();
expect(await watch.closed).toEqual({ reason: "listener_error", error: failure });
expect(values).toEqual([1]);
expect(recorded.reported).toEqual([failure]);
@ -344,7 +442,6 @@ describe("Watch references", () => {
0,
() => void source,
() => void source,
() => source.value,
);
const listener = async (): Promise<void> => {};
watch.observeCancellation(controller.signal);
@ -382,6 +479,43 @@ describe("Watch references", () => {
expect(ended.signal.deref()).toBeUndefined();
}
});
it("releases its pending queue and leaf index once delivered or ended", async () => {
/** Two pending frames, folded, whose ops alone reference `payload`; `settle` delivers or ends them. */
const pending = async (settle: (watch: Watch<JsonValue>, gate: () => void) => Promise<void>) => {
const payload = { big: "x".repeat(1000) };
const watch = new Watch<JsonValue>(
0,
() => {},
() => {},
);
const gate = deferred();
let calls = 0;
watch.start(async () => {
if (++calls === 1) await gate.promise;
});
watch.advance(1, [["s", ["n"], 1]], context);
await flush();
watch.advance(2, [["s", ["o"], payload]], context);
watch.advance(3, [["s", ["n"], 3]], context);
await settle(watch, gate.resolve);
return { watch, payload: new WeakRef(payload) };
};
const delivered = await pending(async (_watch, gate) => {
gate();
await flush();
});
await collect();
expect(delivered.payload.deref()).toBeUndefined();
expect(delivered.watch.value).toBe(3);
const stopped = await pending(async (watch) => {
await watch.stop();
});
await collect();
expect(stopped.payload.deref()).toBeUndefined();
});
});
describe("NativeWatch", () => {
@ -424,6 +558,22 @@ describe("NativeWatch", () => {
expect(values).toEqual([101]);
});
it("collapses 101 pending values to the value of `replace`", async () => {
const watch = new NativeWatch<string>(
"",
() => {},
() => {},
() => "replacement",
);
for (let n = 1; n <= 101; n++) watch.advance(`value ${n}`, context);
const values: string[] = [];
watch.start(async (value) => {
values.push(value);
});
await flush();
expect(values).toEqual(["replacement"]);
});
it("ends with every reason of its lifecycle", async () => {
const create = () =>
new NativeWatch<number>(

3
packages/durable/CHANGELOG.md
View file

@ -7,6 +7,7 @@
- `Storage.commit(writes, adopt, context)` takes an `adopt(seq)` callback, the Session's in-memory adoption of the batch: a storage calls it exactly once, synchronously, once the batch is durable and before its next commit starts, then resolves with `seq`, and never for a commit that fails before the batch is durable. Memory, SQLite, JSONL, and IndexedDB storages do; custom storages and wrappers must call or forward it, and the storage conformance suite checks it. A commit that rejects after its adoption fails the Session without rolling the adopted changes back.
- `Storage.mintId()`'s contract is narrowed to IDs greater than every ID minted since the storage was opened and every ID stored, also across reopen. The bundled storages already behave so.
- Watches use Chord's `WatchHandle` and `WatchEnd` (Durable re-exports both). End reasons are `stopped`, `cancelled`, `retired`, `ended` (was `session_closed`), `listener_error`, and `failed` (was `session_failed`). Retirement is an explicit event of the watched document incarnation (last frame `[["r", null]]`, then `retired`), never inferred from a `null` value. An ended watch drops its listener and every reference into its Session, so a stopped watch that is still referenced keeps only its last value.
- Document, conversation view, and task graph watches deliver the commits made while their listener runs (or before `start()`) in one callback: the newest value, the commits' operations folded (appends to one string join, sets of one field keep the last) and the newest commit's Context. Past 10,000 pending operations the callback gets one root replacement `[["r", newest]]`. Before, each commit was its own callback and more than 100 pending commits became a root replacement. `watchEvents()` still delivers every event batch in order, with one `snapshot` past 100 pending batches.
- Removed the Chord state getters `Session.documentState()`, `Conversation.viewState()`, and `Harness.taskGraph()`, and the `DocumentState` type; use `watchDoc()`, `Conversation.watch()`, and `Harness.watchTaskGraph()`. Durable no longer uses Chord's `replicatedState()`.
### Added
@ -15,7 +16,7 @@
### Changed
- Document, conversation view, task graph, and event watches use Chord's `Watch` implementation; behaviour is unchanged.
- Document, conversation view, task graph, and event watches use Chord's watch handles (`Watch`, and `NativeWatch` for event batches).
### Fixed

4
packages/durable/README.md
View file

@ -473,7 +473,7 @@ Work that already started keeps the code it took: a running tool call finishes u
## Watching a Conversation
Everything a UI needs is committed state. `watch()` returns a watch of the conversation's structural view: its value at attachment, then each commit that touches it, with the exact Chord operations of that commit, one callback at a time:
Everything a UI needs is committed state. `watch()` returns a watch of the conversation's structural view: its value at attachment, then each commit that touches it, with the Chord operations of that commit, one callback at a time:
```typescript
const watch = await root.watch(context);
@ -490,7 +490,7 @@ watch.start(async (value, ops) => {
Watches are Chord's `WatchHandle` (exported by `@earendil-works/chord`; Durable re-exports it). A watch ends once, and `closed` (or `stop()`) gives the reason: `stopped`, `cancelled` (its acquisition context was cancelled), `retired` (a document watch whose incarnation retired, after it delivered `[["r", null]]`), `ended` (the Harness closed), `listener_error`, or `failed` (the Harness failed). An ended watch keeps its last value and no longer references the Harness.
A slow watch keeps at most 100 undelivered frames. After that, the pending frames are replaced by one frame holding the whole newest view. A client that joins late or reconnects starts from the current view; nothing is replayed.
Commits that land while a callback runs are delivered together in the next callback: the newest view and the operations of all of them, folded (appends to one string join, sets of one field keep the last). Past 10,000 pending operations the callback gets one root replacement `[["r", view]]` holding the whole newest view instead. A client that joins late or reconnects starts from the current view; nothing is replayed.
Partial answers and tool output are committed at most every 100 ms by default, so a crash loses at most that window. `settings.progress` changes the intervals; a host whose storage is remote can commit less often, for example `{ partialIntervalMs: 500, outputIntervalMs: 500 }`.

51
packages/durable/docs/spec.md
View file

@ -4674,7 +4674,7 @@ of the loaded tracker cache. Reopen creates a new watch lifetime.
Acquisition captures the value and registers for later commits atomically, on
the Session line: the acquisition value includes every commit before
acquisition, and every later exact frame is buffered until `start()`. This
acquisition, and every later commit is buffered until `start()`. This
prevents a value from being paired with operations based on a newer unseen
revision.
@ -4686,7 +4686,7 @@ Session code.
On the Session line, `watchDoc()` resolves one existing concrete incarnation,
captures its current immutable tracker revision in O(1), and registers for every
later exact committed revision and operation batch. It returns `undefined` when
later committed revision and its operation batch. It returns `undefined` when
absent. Before `start()`, `watch.value` remains the acquisition revision while
later frames buffer.
@ -4705,29 +4705,34 @@ try {
```
`start()` installs the sole listener and schedules delivery; it never invokes
user code inline. Each callback receives the exact immutable committed value and
the exact operation batch that produced it. Immediately before invocation,
user code inline. Each callback receives the newest immutable committed value and
operations that turn the previously delivered value into it: the exact operation
batch of its commit when the listener kept up, or the folded batches of every
commit since the previous callback. Immediately before invocation,
`watch.value` advances to that value. The watch awaits the listener before
starting its next callback, but Session commits never wait for callback
settlement.
A watch retains at most 100 pending committed frames, excluding the frame already
being delivered. Adding frame 101 replaces the complete undelivered suffix with
one self-contained root replacement `[["r", newestValue]]`, using the newest
exact immutable revision and that commit's Context. Later exact frames follow
that replacement normally. Another overflow replaces the pending suffix again.
No serialized-byte measurement, value copy, operation replay, or re-diff occurs.
This is convergent observation, not an audit stream; consumers requiring every
transition must persist those facts separately.
Commits that land while a callback runs, or before `start()`, wait in Chord's
pending queue (`PendingOps`) as raw operations and are delivered together in one
callback with the newest value and the newest commit's Context. While waiting,
an operation on a leaf (an append, or a set of a primitive) folds into the last
pending one on the same path: appends join, and a set replaces an earlier set or
append. Any other operation keeps the order of everything before it, and a root
replacement drops everything before it. Past 10,000 pending operations after
folding, the callback gets one self-contained root replacement
`[["r", newestValue]]` instead. No serialized-byte measurement, value copy,
operation replay, or re-diff occurs. This is convergent observation, not an audit
stream; consumers requiring every transition must persist those facts separately.
`watch.value` and every previously returned revision remain stable forever under
the trusted immutability contract. Consumers must not mutate them or retained
descendants. The selected commit Context's values are preserved without inheriting the
producer's cancellation. A callback owns its own asynchronous work.
A watch remains bound to its original incarnation. Retirement queues the exact
terminal `[["r", null]]` frame, or folds it into an overflow replacement with
value `null`. After that callback settles, the watch closes as `retired` and never
A watch remains bound to its original incarnation. Retirement drops what is
pending and queues the terminal `[["r", null]]` with value `null`. After that
callback settles, the watch closes as `retired` and never
follows a replacement incarnation.
`stop()` is idempotent. It synchronously unregisters, discards pending frames,
@ -4785,9 +4790,8 @@ redundant nonempty batch remains a real publication. Mounted document revisions
are structurally shared under the trusted immutability contract. The mount
performs no semantic projection and owns no persistence.
`Conversation.watch()` exposes the mount through the serialized exact-frame
watch of section 9.2, with bounded pending frames and full-value
overflow replacements.
`Conversation.watch()` exposes the mount through the serialized watch of
section 9.2, with folded pending commits and a full-value overflow replacement.
### 9.4 Agent events (experimental)
@ -5034,8 +5038,8 @@ that sets or deletes `tasks[id]`, or sets a node's `conversations`. A
publication that changes no node creates no revision. Task and conversation
records are immutable values; a node shares nothing with them.
`watchTaskGraph()` returns the mount through the serialized exact-frame watch
of section 9.2. It reads only: it never enables scheduling. Close ends it at its
`watchTaskGraph()` returns the mount through the serialized watch of section
9.2. It reads only: it never enables scheduling. Close ends it at its
seal like every other watch (section 2.2).
```ts
@ -5524,9 +5528,10 @@ These are contracts, not invitations to add defensive machinery:
acquisition revision. Initialize the consumer from it first. After start, the
property advances to each delivered immutable revision before its callback;
every earlier reference remains stable.
- **Buffered watches:** slow or unstarted watches retain up to 100 exact committed
frames. Overflow replaces the undelivered suffix with one full-value root
replacement, so intermediate committed states may be omitted. A consumer that
- **Buffered watches:** slow or unstarted watches deliver the commits made
meanwhile in one callback with folded operations, and past 10,000 pending
operations one full-value root replacement, so intermediate committed states
are omitted. A consumer that
must audit every transition must persist each fact in an immutable entry or
journal and scan that history explicitly.
- **Watch stop:** a listener may call and await `stop()`; it prevents future

13
packages/durable/src/harness/events.ts
View file

@ -1,4 +1,4 @@
import { type Context, type JsonValue, Watch, type WatchEnd } from "@earendil-works/chord";
import { type Context, type JsonValue, NativeWatch, type WatchEnd } from "@earendil-works/chord";
import type { Op, Path } from "@earendil-works/chord/delta";
import type { AssistantMessage, Message, Usage } from "@earendil-works/pi-ai";
import type {
@ -167,7 +167,7 @@ export async function watchEvents(
context: Context,
): Promise<AgentEventStream> {
const views = conversationViews(harness);
let watch!: Watch<readonly AgentEvent[]>;
let watch!: NativeWatch<readonly AgentEvent[]>;
let snapshot!: SnapshotEvent;
await views.attach(
conversationId,
@ -178,8 +178,9 @@ export async function watchEvents(
const held = new Set<TaskId>(completing.map((record) => record.id));
let current = initial;
snapshot = snapshotOf(initial);
// Batches are the watch's values; an overflow delivers a snapshot of the newest view instead.
watch = new Watch<readonly AgentEvent[]>(
// Batches are the watch's values, each delivered (folding would drop events); an overflow delivers a snapshot of
// the newest view instead.
watch = new NativeWatch<readonly AgentEvent[]>(
[],
release,
(error) => views.report(error),
@ -189,7 +190,7 @@ export async function watchEvents(
publication: (before, after, ops, publication, commitContext) => {
current = after;
const events = translate(conversationId, before, after, ops, publication, held);
if (events.length > 0) watch.advance(events, [], commitContext);
if (events.length > 0) watch.advance(events, commitContext);
},
end: (failure) => watch.end(failure),
};
@ -205,7 +206,7 @@ export async function watchEvents(
if (signal !== undefined) watch.observeCancellation(signal);
return {
snapshot,
start: (listener) => watch.start((events, _ops, deliveryContext) => listener(events, deliveryContext)),
start: (listener) => watch.start(listener),
stop: () => watch.stop(),
closed: watch.closed,
};

2
packages/durable/src/harness/task-graph.ts
View file

@ -104,7 +104,7 @@ export class TaskGraphView {
});
}
/** A serialized exact-frame watch of the graph; cancelling `context` stops it. */
/** A serialized watch of the graph; cancelling `context` stops it. */
async watch(context: Context): Promise<TaskGraphWatch> {
const { observer } = await this.#attach(
(value, release) => new Watch<TaskGraph>(value, release, (error) => this.#session.report(error)),

6
packages/durable/src/harness/types.ts
View file

@ -778,7 +778,7 @@ export interface Conversation {
* owned, transitively, by its non-background tasks.
*/
waitForIdle(context: Context): Promise<void>;
/** The structural view (spec §9.3) as a serialized exact-frame watch with bounded pending frames. */
/** The structural view (spec §9.3) as a serialized watch; commits made while its listener runs arrive folded. */
watch(context: Context): Promise<ConversationWatch>;
}
@ -837,8 +837,8 @@ export interface Harness extends Session {
/** Session total: every conversation's `pi.usage` summed. */
usage(context: Context): Promise<UsageState>;
/**
* Every live task with its owner edge, status, and owned conversations (spec §9.5), as a serialized exact-frame
* watch with bounded pending frames.
* Every live task with its owner edge, status, and owned conversations (spec §9.5), as a serialized watch; commits
* made while its listener runs arrive folded.
*/
watchTaskGraph(context: Context): Promise<TaskGraphWatch>;
}

2
packages/durable/src/harness/view.ts
View file

@ -106,7 +106,7 @@ export class ConversationViews {
this.#session.report(error);
}
/** A serialized exact-frame watch of the view; cancelling `context` stops it. */
/** A serialized watch of the view; cancelling `context` stops it. */
async watch(id: ConversationId, context: Context): Promise<WatchHandle<ConversationView>> {
const { observer } = await this.attach(
id,

30
packages/durable/test/examples/04-watches.ts
View file

@ -22,21 +22,33 @@ const chat = await session.commit(async (tx) => {
return conversation;
}, context);
// A watch starts from one stable acquisition revision. Slow callbacks never
// overlap; exact committed frames buffer, with a full-value reset after 100.
// A watch starts from one stable acquisition revision. Callbacks never overlap:
// commits made while one runs arrive folded into the next call, which gets the
// newest value and ops that turn the previously delivered value into it (one
// full-value replacement `[["r", value]]` once more than 10,000 ops pile up).
const notesWatch = await session.watchDoc(Notes, chat.id, context);
if (notesWatch === undefined) throw new Error("notes are absent");
console.log("watch baseline:", notesWatch.value);
const delivered = new Promise<void>((resolve) => {
notesWatch.start(async (value, _ops, _deliveryContext) => {
console.log("watch update:", value);
resolve();
});
const busy = Promise.withResolvers<void>();
const done = Promise.withResolvers<void>();
let calls = 0;
notesWatch.start(async (value, ops, _deliveryContext) => {
console.log("watch update:", value, JSON.stringify(ops));
// The first call stays busy, as a slow consumer would, while three more commits land.
if (++calls === 1) await busy.promise;
else done.resolve();
});
await session.commit(async (tx) => {
(await tx.doc(Notes, chat.id)).text = "observed asynchronously";
(await tx.doc(Notes, chat.id)).text = "observed";
}, context);
await delivered;
for (const more of [" asynchronously", ", folded", "!"]) {
await session.commit(async (tx) => {
(await tx.doc(Notes, chat.id)).text += more;
}, context);
}
busy.resolve();
await done.promise;
console.log("listener calls:", calls);
await notesWatch.stop();
await session.close(context);

19
packages/durable/test/harness-events.test.ts
View file

@ -252,6 +252,25 @@ describe("agent events", () => {
await harness.close(context);
});
it("delivers each of 100 pending batches in its own call, in order, unlike state watches", async () => {
const { harness, root } = await openChat(new MemoryStorage(), chatSetup());
const stream = await watchEvents(harness, root.id, context);
for (let index = 0; index < 100; index++) {
await root.commit((tx) => tx.appendEntry(root.id, { kind: "note", data: { index } }), context);
}
const batches: AgentEvent[][] = [];
stream.start(async (events) => {
batches.push([...events]);
});
await drained();
expect(batches).toHaveLength(100);
expect(
batches.map((events) => events.map((event) => (event.type === "entry_appended" ? event.entry.data : event))),
).toEqual(Array.from({ length: 100 }, (_, index) => [{ index }]));
await stream.stop();
await harness.close(context);
});
it("replaces undelivered batches with one snapshot after 100 pending batches", async () => {
const { harness, root } = await openChat(new MemoryStorage(), chatSetup());
const stream = await watchEvents(harness, root.id, context);

6
packages/durable/test/harness-view.test.ts
View file

@ -228,9 +228,10 @@ describe("conversation view", () => {
await harness.close(context);
});
it("replaces undelivered frames with the newest view after 100 pending frames", async () => {
it("folds undelivered commits into one call whose ops reach the newest view", async () => {
const { harness, root } = await openChat(new MemoryStorage(), chatSetup());
const watch = await root.watch(context);
const initial = watch.value;
for (let index = 0; index < 101; index++) {
await root.commit((tx) => tx.appendEntry(root.id, { kind: "note" }), context);
}
@ -240,7 +241,8 @@ describe("conversation view", () => {
});
await drained();
expect(frames).toHaveLength(1);
expect(frames[0]!.ops).toEqual([["r", frames[0]!.value]]);
expect(frames[0]!.ops[0]?.[0]).not.toBe("r");
expect(applyImmutable(initial, frames[0]!.ops)).toEqual(frames[0]!.value);
expect(frames[0]!.value.entries).toHaveLength(101);
await watch.stop();
await harness.close(context);

75
packages/durable/test/session-watches.test.ts
View file

@ -24,6 +24,16 @@ const StateDoc = defineDoc<State>({
initial: () => ({ value: 0, items: ["a", "b"], retained: { label: "stable" } }),
});
type Keys = { keys: Record<string, number> };
const KeysDoc = defineDoc<Keys>({
checkpointWhen: () => false,
kind: "watch.keys",
version: 1,
scope: "session",
initial: () => ({ keys: {} }),
});
type Deferred = { readonly promise: Promise<void>; readonly resolve: () => void };
function deferred(): Deferred {
@ -49,7 +59,7 @@ describe("Session document watches", () => {
expect(storage.mintCount).toBe(0);
});
it("keeps the acquisition revision until start and delivers exact committed frames", async () => {
it("keeps the acquisition revision until start, then delivers the commits since in one call", async () => {
const { session, publications } = await createState();
const watch = (await session.watchDoc(StateDoc, context))!;
const initial = watch.value;
@ -75,27 +85,38 @@ describe("Session document watches", () => {
inline = false;
expect(deliveries).toHaveLength(0);
await flush();
expect(deliveries.map(({ value }) => value?.value)).toEqual([1, 2]);
expect(deliveries[0]!.value).toBe(firstPublished.value);
expect(deliveries[0]!.ops).toBe(firstPublished.ops);
expect(deliveries[1]!.value).toBe(secondPublished.value);
expect(deliveries[1]!.ops).toBe(secondPublished.ops);
// Both commits set `value`: folded, the second set replaces the first.
expect(firstPublished.ops).toEqual([["s", ["value"], 1]]);
expect(deliveries).toEqual([{ value: secondPublished.value, ops: secondPublished.ops }]);
expect(deliveries[0]!.value).toBe(secondPublished.value);
expect(initial?.value).toBe(0);
// A listener that keeps up gets each commit's ops as published.
await session.commit(async (tx) => {
(await tx.doc(StateDoc)).value = 3;
}, context);
await flush();
const thirdPublished = documentChanges(publications.at(-1)!)[0]!;
expect(deliveries).toHaveLength(2);
expect(deliveries[1]!.value).toBe(thirdPublished.value);
expect(deliveries[1]!.ops).toBe(thirdPublished.ops);
await watch.stop();
});
it("serializes callbacks and buffers exact frames committed while one is in flight", async () => {
it("serializes callbacks and folds the commits made while one is in flight into the next", async () => {
const { session } = await createState();
const watch = (await session.watchDoc(StateDoc, context))!;
const entered = deferred();
const release = deferred();
const values: number[] = [];
const batches: unknown[] = [];
let active = 0;
let maxActive = 0;
watch.start(async (value) => {
watch.start(async (value, ops) => {
active++;
maxActive = Math.max(maxActive, active);
values.push(value?.value ?? -1);
batches.push(ops);
if (values.length === 1) {
entered.resolve();
await release.promise;
@ -115,7 +136,8 @@ describe("Session document watches", () => {
release.resolve();
await flush();
expect(maxActive).toBe(1);
expect(values).toEqual(Array.from({ length: 20 }, (_, index) => index + 1));
expect(values).toEqual([1, 20]);
expect(batches[1]).toEqual([["s", ["value"], 20]]);
await watch.stop();
});
@ -143,7 +165,7 @@ describe("Session document watches", () => {
await watch.stop();
});
it("collapses 101 pending commits to one root replacement", async () => {
it("folds 101 pending commits of one field into one set", async () => {
const { session } = await createState();
const watch = (await session.watchDoc(StateDoc, context))!;
for (let value = 1; value <= 101; value++) {
@ -156,11 +178,36 @@ describe("Session document watches", () => {
deliveries.push({ value: value?.value ?? -1, ops });
});
await flush();
expect(deliveries).toEqual([{ value: 101, ops: [["r", watch.value]] }]);
expect(deliveries).toEqual([{ value: 101, ops: [["s", ["value"], 101]] }]);
await watch.stop();
});
it("never folds the in-flight frame into an overflow reset", async () => {
it("replaces the root once more than 10,000 ops are pending after folding", async () => {
const { session } = openTestSession();
await session.commit((tx) => tx.doc(KeysDoc).then(() => undefined), context);
const watch = (await session.watchDoc(KeysDoc, context))!;
// Sets of distinct keys, in commits small enough to stay ops: 4,000 + 4,000 + 2,001 pending ops.
for (const [from, to] of [
[0, 4_000],
[4_000, 8_000],
[8_000, 10_001],
] as const) {
await session.commit(async (tx) => {
const state = await tx.doc(KeysDoc);
for (let index = from; index < to; index++) state.keys[`k${index}`] = index;
}, context);
}
const deliveries: Array<{ value: Readonly<Keys> | null; ops: unknown }> = [];
watch.start(async (value, ops) => {
deliveries.push({ value, ops });
});
await flush();
expect(deliveries).toEqual([{ value: watch.value, ops: [["r", watch.value]] }]);
expect(Object.keys(watch.value?.keys ?? {})).toHaveLength(10_001);
await watch.stop();
});
it("never folds the in-flight frame into the next call", async () => {
const { session } = await createState();
const watch = (await session.watchDoc(StateDoc, context))!;
const entered = deferred();
@ -185,12 +232,12 @@ describe("Session document watches", () => {
release.resolve();
await flush();
expect(deliveries[0]!.value).toBe(1);
expect(deliveries[1]).toEqual({ value: 102, ops: [["r", watch.value]] });
expect(deliveries[1]).toEqual({ value: 102, ops: [["s", ["value"], 102]] });
expect(deliveries).toHaveLength(2);
await watch.stop();
});
it("folds retirement into an overflow reset and then closes", async () => {
it("folds retirement over pending commits and then closes", async () => {
const { session } = await createState();
const watch = (await session.watchDoc(StateDoc, context))!;
for (let value = 1; value <= 100; value++) {

5
scripts/check-entry-graphs.mjs
View file

@ -66,8 +66,9 @@ const BUDGETS = {
// (index, noop, metrics, memory) for its NOOP telemetry default; harness/telemetry.ts is Durable's schema.
// The Harness's request identity reaches pi-ai's utils/identity.ts and its version constant. Partial commits
// parse streamed tool-call arguments with pi-ai's utils/streaming-json.ts (and its json-parse.ts) to append what
// argument strings grew by (harness/partial-changes.ts).
maxFiles: 76,
// argument strings grew by (harness/partial-changes.ts). Chord's watches fold pending ops with Delta's
// pending queue (delta/pending.ts).
maxFiles: 77,
forbid: [
"packages/ai/src/index.ts",
"packages/ai/src/utils/typebox-helpers.ts",