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): shared watch-handle implementation (Watch, NativeWatch); Durable uses it

Browse source
This commit is contained in:
Mario Zechner 2026-10-23 07:05:23 +00:00
parent 936e5f40ae
commit 42cb822f1c
12 changed files with 780 additions and 237 deletions
Download patch file Download diff file Expand all files Collapse all files

2
packages/chord/CHANGELOG.md
View file

@ -9,6 +9,8 @@
### Added
- `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.
- `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.
## [1.4.0] - 2026-10-21

8
packages/chord/README.md
View file

@ -14,7 +14,7 @@ Contracts are runtime values described with [TypeBox](https://github.com/sinclai
| Import | Contents | Runs in |
|--------|----------|---------|
| `@earendil-works/chord` | Contracts (`local`, `remote`, `method`, `opaque`), declarations (through the factory's `self`: `state`, `derive`, `resource`, `on`, `task`), hosts (`createHost`, `defineFacet`, `defineRegistry`), links (`memoryLinkPair`, `ipcLink`, `webSocketLink`, `acceptWebSockets`), errors (`ChordError`, `isChordError`, `forbidden`, `invalid`). Also the shared base: `Context`, `JsonValue`, `JsonRepresentation`, `Draft`, `copyJson`, `isJsonValue`, `replicatedState` with its types, and the watch types `WatchHandle` and `WatchEnd`. | Node and browsers |
| `@earendil-works/chord` | Contracts (`local`, `remote`, `method`, `opaque`), declarations (through the factory's `self`: `state`, `derive`, `resource`, `on`, `task`), hosts (`createHost`, `defineFacet`, `defineRegistry`), links (`memoryLinkPair`, `ipcLink`, `webSocketLink`, `acceptWebSockets`), errors (`ChordError`, `isChordError`, `forbidden`, `invalid`). Also the shared base: `Context`, `JsonValue`, `JsonRepresentation`, `Draft`, `copyJson`, `isJsonValue`, `replicatedState` with its types, and the watch types `WatchHandle`, `NativeWatchHandle`, and `WatchEnd` with their implementations `Watch` and `NativeWatch`. | Node and browsers |
| `@earendil-works/chord/node` | Child host processes: `createHostManager`, `forkDriver`, `childControl`, `processLock`; the facet module loader `nodeLoader`. | Node |
| `@earendil-works/chord/context` | Explicit cancellation, telemetry, and invocation values: `BACKGROUND_CONTEXT`, `withCancel`, `withAbortSignal`, `withoutAbortSignal`, `withTelemetry`, `traceIn`, `withContextValue`, `awaitWithContext`. | Node and browsers |
| `@earendil-works/chord/delta` | The transactional JSON tracker, structural diff, immutable apply, and the wire encoder/decoder for operation batches. | Node and browsers |
@ -224,3 +224,9 @@ frames, and neither aborts nor awaits a running call; it and `closed` give the
An ended watch drops every reference into its source, so a stopped watch that is
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.

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

@ -78,6 +78,6 @@ export {
replicatedState,
} from "./state.ts";
export type { Ctx, GetOptions, Readable, RefStatus, Shape, StateEvent } from "./types.ts";
export type { WatchEnd, WatchHandle } from "./watch.ts";
export { NativeWatch, type NativeWatchHandle, Watch, type WatchEnd, type WatchHandle } from "./watch.ts";
export { acceptWebSockets, type WebSocketAcceptOptions, type WebSocketServerLike } from "./websocket.ts";
export { type Frame, PROTOCOL_VERSION } from "./wire.ts";

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

@ -1,5 +1,7 @@
import type { Context } from "./context/index.ts";
// 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 { JsonValue } from "./json.ts";
/**
* Terminal result of one watch.
@ -27,3 +29,261 @@ export interface WatchHandle<T> {
/** Settle when the watch terminates; an already-running callback remains caller-owned. */
readonly closed: Promise<WatchEnd>;
}
/** Serialized observation of an in-process value: as `WatchHandle`, but a change is a new value without ops. */
export interface NativeWatchHandle<T> {
/** Acquisition value before start; latest delivered value afterward. */
readonly value: T;
/** Install the sole asynchronous listener. Never invokes it inline. */
start(listener: (value: T, context: Context) => Promise<void>): void;
/** Idempotently stop future callbacks and return this watch's terminal result. */
stop(): Promise<WatchEnd>;
/** Settle when the watch terminates; an already-running callback remains caller-owned. */
readonly closed: Promise<WatchEnd>;
}
type WatchListener<T> = (value: T, ops: readonly Op[], context: Context) => Promise<void>;
type NativeWatchListener<T> = (value: T, context: Context) => Promise<void>;
type WatchFrame<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;
};
/** 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.
*/
abstract class WatchLifecycle<T, L> {
#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) {
this.#value = value;
this.#detach = detach;
this.#report = report;
this.#replace = replace;
}
get value(): T {
return this.#value;
}
get closed(): Promise<WatchEnd> {
return this.#closed.promise;
}
start(listener: L): void {
if (this.#started) throw new Error("Watch is already started");
if (this.#end !== undefined) throw new Error("Watch is stopped");
this.#started = true;
this.#listener = listener;
if (this.#pending.length > 0) this.#schedule();
}
stop(): Promise<WatchEnd> {
this.#terminate({ reason: "stopped" });
return this.#closed.promise;
}
/** End the watch `cancelled` when `signal` aborts; installed once. */
observeCancellation(signal: AbortSignal): void {
if (this.#cancellationSignal !== undefined) throw new Error("Watch cancellation is already installed");
if (this.#end !== undefined) return;
this.#cancellationSignal = signal;
this.#cancellationListener = () => this.cancel();
signal.addEventListener("abort", this.#cancellationListener, { once: true });
if (signal.aborted) this.cancel();
}
/** End the watch `cancelled`: its acquisition context was cancelled. */
cancel(): void {
this.#terminate({ reason: "cancelled" });
}
/** The source closed: end with `ended`, or with `failed` and the error when the source failed. */
end(failure?: { readonly error: unknown }): void {
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();
}
/**
* 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 });
if (this.#started) this.#schedule();
}
#schedule(): void {
if (this.#scheduled || this.#running || this.#end !== undefined) return;
this.#scheduled = true;
queueMicrotask(() => {
this.#scheduled = false;
void this.#drain();
});
}
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);
try {
await this.invoke(listener, frame, deliveryContext);
} catch (error) {
if (this.#end === undefined) {
this.#report?.(error);
this.#terminate({ reason: "listener_error", error: toError(error) });
}
break;
}
if (frame.retires === true) {
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
* callback already running keeps running.
*/
#terminate(end: WatchEnd): void {
if (this.#end !== undefined) return;
this.#end = end;
const detach = this.#detach;
this.#detach = undefined;
this.#report = undefined;
this.#replace = undefined;
this.#listener = undefined;
this.#pending.length = 0;
if (this.#cancellationListener !== undefined) {
this.#cancellationSignal?.removeEventListener("abort", this.#cancellationListener);
}
this.#cancellationSignal = undefined;
this.#cancellationListener = undefined;
detach?.();
this.#closed.resolve(end);
}
}
function toError(error: unknown): Error {
if (error instanceof Error) return error;
try {
return new Error(String(error));
} catch {
return new Error(Object.prototype.toString.call(error));
}
}
/** 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.
*/
export class Watch<T> extends WatchLifecycle<T, WatchListener<T>> implements WatchHandle<T> {
/** 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);
}
/**
* The source retired: queue the last frame, `[["r", null]]` with value `null`, then end `retired`. Typed for watches
* whose values include `null`.
*/
retire(this: Watch<T | null>, context: Context): void {
this.enqueueRetirement(null, RETIREMENT_OPERATIONS, context);
}
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[] {
// 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 NO_OPS: readonly Op[] = [];
/**
* The watch handle of a native source: values in order, an overflow collapses to 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);
}
/** Queue `value`, the source's new value. */
advance(value: T, context: Context): void {
this.enqueue(value, NO_OPS, context);
}
protected invoke(listener: NativeWatchListener<T>, frame: WatchFrame<T>, context: Context): Promise<void> {
return listener(frame.value, context);
}
protected overflowOps(): readonly Op[] {
return NO_OPS;
}
}

490
packages/chord/test/watch.test.ts Normal file
View file

@ -0,0 +1,490 @@
// 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 {
BACKGROUND_CONTEXT,
type Context,
createContextKey,
withAbortSignal,
withContextValue,
} from "../src/context/index.ts";
import type { Op } from "../src/delta/index.ts";
import type { JsonValue } from "../src/json.ts";
import { NativeWatch, Watch, type WatchEnd } from "../src/watch.ts";
const context = BACKGROUND_CONTEXT;
function flush(): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, 0));
}
function deferred(): { readonly promise: Promise<void>; readonly resolve: () => void } {
const { promise, resolve } = Promise.withResolvers<void>();
return { promise, resolve: () => resolve() };
}
/** A watch whose detach and reports are recorded. */
function recordedWatch<T>(value: T, replace?: () => 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);
const frames: Array<{ value: T; ops: readonly Op[] }> = [];
watch.start(async (delivered, ops) => {
frames.push({ value: delivered, ops });
});
return { watch, recorded, frames };
}
/** The end of `closed`, or undefined while the watch is open. */
function endOf(watch: { readonly closed: Promise<WatchEnd> }): { readonly end: WatchEnd | undefined } {
const settled: { end: WatchEnd | undefined } = { end: undefined };
void watch.closed.then((end) => {
settled.end = end;
});
return settled;
}
const collect = (() => {
setFlagsFromString("--expose-gc");
const gc = runInNewContext("gc") as () => void;
return async (): Promise<void> => {
for (let round = 0; round < 5; round++) {
await flush();
gc();
}
};
})();
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);
await flush();
expect(watch.value).toEqual({ n: 0 });
const frames: Array<{ value: JsonValue; ops: readonly Op[] }> = [];
watch.start(async (value, ops) => {
frames.push({ value, ops });
});
expect(frames).toEqual([]);
await flush();
expect(frames).toEqual([
{ value: { n: 1 }, ops: [["s", ["n"], 1]] },
{ value: { n: 2 }, ops: [["s", ["n"], 2]] },
]);
expect(watch.value).toEqual({ n: 2 });
});
it("never invokes the listener inline", async () => {
const { watch, frames } = collectingWatch<JsonValue>(0);
watch.advance(1, [["r", 1]], context);
expect(frames).toEqual([]);
await flush();
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);
const gate = deferred();
const calls: string[] = [];
watch.start(async (value) => {
calls.push(`start ${value}`);
if (value === 1) await gate.promise;
calls.push(`end ${value}`);
});
watch.advance(1, [["r", 1]], context);
await flush();
watch.advance(2, [["r", 2]], context);
watch.advance(3, [["r", 3]], context);
await flush();
expect(calls).toEqual(["start 1"]);
gate.resolve();
await flush();
expect(calls).toEqual(["start 1", "end 1", "start 2", "end 2", "start 3", "end 3"]);
});
it("delivers a context with the frame's values and 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;
watch.start(async (_value, _ops, deliveryContext) => {
delivered = deliveryContext;
});
watch.advance(1, [["r", 1]], frameContext);
await flush();
expect(delivered?.value(key)).toBe("first");
expect(delivered?.abortSignal).toBeUndefined();
});
});
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 });
});
await flush();
expect(frames).toEqual([{ value: 101, ops: [["r", 101]] }]);
});
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);
const gate = deferred();
const values: JsonValue[] = [];
watch.start(async (value) => {
values.push(value);
if (value === 1) await gate.promise;
});
watch.advance(1, [["r", 1]], context);
await flush();
for (let n = 2; n <= 102; n++) watch.advance(n, [["r", n]], context);
gate.resolve();
await flush();
expect(values).toEqual([1, 102]);
});
});
describe("Watch retirement", () => {
it('delivers `[["r", null]]` as the last frame and then ends retired', async () => {
const { watch, frames, recorded } = collectingWatch<JsonValue>(1);
const settled = endOf(watch);
watch.advance(2, [["r", 2]], context);
watch.retire(context);
watch.advance(3, [["r", 3]], context);
await flush();
expect(frames).toEqual([
{ value: 2, ops: [["r", 2]] },
{ value: null, ops: [["r", null]] },
]);
expect(watch.value).toBeNull();
expect(settled.end).toEqual({ reason: "retired" });
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("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);
watch.advance(2, [["r", 2]], context);
await flush();
expect(frames).toEqual([
{ value: null, ops: [["r", null]] },
{ value: 2, ops: [["r", 2]] },
]);
expect(settled.end).toBeUndefined();
expect(await watch.stop()).toEqual({ reason: "stopped" });
});
it("is only available to watches whose values include null", () => {
const check = (numbers: Watch<number>, documents: Watch<{ n: number } | null>) => {
// @ts-expect-error `retire` delivers `null`, which a number watch cannot hold.
numbers.retire(context);
documents.retire(context);
};
expectTypeOf(check).toBeFunction();
});
it("discards a retirement still pending when the source ends", async () => {
const { watch } = recordedWatch<JsonValue>(1);
watch.retire(context);
watch.end();
expect(await watch.closed).toEqual({ reason: "ended" });
expect(watch.value).toBe(1);
});
});
describe("Watch ends", () => {
it("stops idempotently and rejects a repeated or late start", async () => {
const { watch, recorded } = recordedWatch<JsonValue>(0);
watch.start(async () => {});
expect(() => watch.start(async () => {})).toThrow("Watch is already started");
expect(await watch.stop()).toEqual({ reason: "stopped" });
expect(await watch.stop()).toEqual({ reason: "stopped" });
expect(recorded.detached).toBe(1);
const late = recordedWatch<JsonValue>(0).watch;
await late.stop();
expect(() => late.start(async () => {})).toThrow("Watch is stopped");
});
it("ends `ended` or `failed` from its source, dropping pending frames", async () => {
const { watch, frames } = collectingWatch<JsonValue>(0);
watch.advance(1, [["r", 1]], context);
watch.end();
await flush();
expect(frames).toEqual([]);
expect(await watch.closed).toEqual({ reason: "ended" });
const failure = new Error("source failed");
const failed = recordedWatch<JsonValue>(0).watch;
failed.end({ error: failure });
failed.end();
expect(await failed.closed).toEqual({ reason: "failed", error: failure });
});
it("stops future delivery without aborting or awaiting an in-flight callback", async () => {
const { watch } = recordedWatch<JsonValue>(0);
const gate = deferred();
const calls: string[] = [];
watch.start(async (value) => {
calls.push(`start ${value}`);
await gate.promise;
calls.push(`end ${value}`);
});
watch.advance(1, [["r", 1]], context);
await flush();
watch.advance(2, [["r", 2]], context);
expect(await watch.stop()).toEqual({ reason: "stopped" });
gate.resolve();
await flush();
expect(calls).toEqual(["start 1", "end 1"]);
});
it("ends `cancelled` when its acquisition signal aborts, also when already aborted", async () => {
const controller = new AbortController();
const { watch, recorded } = recordedWatch<JsonValue>(0);
watch.observeCancellation(controller.signal);
expect(() => watch.observeCancellation(controller.signal)).toThrow("Watch cancellation is already installed");
controller.abort();
expect(await watch.closed).toEqual({ reason: "cancelled" });
expect(recorded.detached).toBe(1);
const aborted = recordedWatch<JsonValue>(0).watch;
aborted.observeCancellation(AbortSignal.abort());
expect(await aborted.closed).toEqual({ reason: "cancelled" });
const cancelled = recordedWatch<JsonValue>(0).watch;
cancelled.cancel();
expect(await cancelled.closed).toEqual({ reason: "cancelled" });
});
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 values: JsonValue[] = [];
watch.start(async (value) => {
values.push(value);
throw failure;
});
watch.advance(1, [["r", 1]], context);
watch.advance(2, [["r", 2]], context);
expect(await watch.closed).toEqual({ reason: "listener_error", error: failure });
expect(values).toEqual([1]);
expect(recorded.reported).toEqual([failure]);
expect(recorded.detached).toBe(1);
});
it("wraps a thrown non-Error in an Error", async () => {
const { watch } = recordedWatch<JsonValue>(0);
watch.start(() => Promise.reject("plain"));
watch.advance(1, [["r", 1]], context);
const end = await watch.closed;
expect(end.reason).toBe("listener_error");
expect(end.reason === "listener_error" && end.error.message).toBe("plain");
});
});
describe("Watch references", () => {
it("retains neither its source, its listener, nor its signal once ended, while still referenced", async () => {
/** A watch over a source only its closures reference; `end` ends it before returning. */
const acquire = async (end: (watch: Watch<JsonValue>) => void) => {
const source = { value: 1 as JsonValue };
const controller = new AbortController();
const watch = new Watch<JsonValue>(
0,
() => void source,
() => void source,
() => source.value,
);
const listener = async (): Promise<void> => {};
watch.observeCancellation(controller.signal);
watch.start(listener);
watch.advance(1, [["r", 1]], context);
await flush();
end(watch);
return {
watch,
source: new WeakRef(source),
listener: new WeakRef(listener),
signal: new WeakRef(controller.signal),
};
};
// Control: an open watch keeps its source, so the collections below are real.
const open = await acquire(() => {});
await collect();
expect(open.source.deref()).toBeDefined();
expect(open.listener.deref()).toBeDefined();
expect(open.signal.deref()).toBeDefined();
await open.watch.stop();
for (const end of [
(watch: Watch<JsonValue>) => void watch.stop(),
(watch: Watch<JsonValue>) => watch.cancel(),
(watch: Watch<JsonValue>) => watch.end(),
(watch: Watch<JsonValue>) => watch.retire(context),
]) {
const ended = await acquire(end);
await flush();
await collect();
expect(ended.source.deref()).toBeUndefined();
expect(ended.listener.deref()).toBeUndefined();
expect(ended.signal.deref()).toBeUndefined();
}
});
});
describe("NativeWatch", () => {
it("delivers values by identity in order with a context, never inline", async () => {
const first = { run: () => 1 };
const second = { run: () => 2 };
const watch = new NativeWatch<{ run: () => number }>(
{ run: () => 0 },
() => {},
() => {},
);
const key = createContextKey<string>("frame");
const delivered: Array<{ value: { run: () => number }; frame: string | undefined }> = [];
watch.start(async (value, deliveryContext) => {
delivered.push({ value, frame: deliveryContext.value(key) });
});
watch.advance(first, withContextValue(key, "first", context));
watch.advance(second, withContextValue(key, "second", context));
expect(delivered).toEqual([]);
await flush();
expect(delivered).toHaveLength(2);
expect(delivered[0].value).toBe(first);
expect(delivered[1].value).toBe(second);
expect(delivered.map((entry) => entry.frame)).toEqual(["first", "second"]);
expect(watch.value).toBe(second);
});
it("collapses 101 pending values to the newest", async () => {
const watch = new NativeWatch<number>(
0,
() => {},
() => {},
);
for (let n = 1; n <= 101; n++) watch.advance(n, context);
const values: number[] = [];
watch.start(async (value) => {
values.push(value);
});
await flush();
expect(values).toEqual([101]);
});
it("ends with every reason of its lifecycle", async () => {
const create = () =>
new NativeWatch<number>(
0,
() => {},
() => {},
);
const stopped = create();
expect(await stopped.stop()).toEqual({ reason: "stopped" });
const cancelled = create();
cancelled.observeCancellation(AbortSignal.abort());
expect(await cancelled.closed).toEqual({ reason: "cancelled" });
const ended = create();
ended.end();
expect(await ended.closed).toEqual({ reason: "ended" });
const error = new Error("source failed");
const failed = create();
failed.end({ error });
expect(await failed.closed).toEqual({ reason: "failed", error });
const reported: unknown[] = [];
const throwing = new NativeWatch<number>(
0,
() => {},
(thrown) => reported.push(thrown),
);
const failure = new Error("listener failed");
throwing.start(() => Promise.reject(failure));
throwing.advance(1, context);
expect(await throwing.closed).toEqual({ reason: "listener_error", error: failure });
expect(reported).toEqual([failure]);
});
it("retains neither its source nor its listener once stopped, while still referenced", async () => {
const acquire = async (stop: boolean) => {
const source = { value: 1 };
const watch = new NativeWatch<number>(
0,
() => void source,
() => void source,
);
const listener = async (): Promise<void> => {};
watch.start(listener);
watch.advance(source.value, context);
await flush();
if (stop) await watch.stop();
return { watch, source: new WeakRef(source), listener: new WeakRef(listener) };
};
const open = await acquire(false);
await collect();
expect(open.source.deref()).toBeDefined();
await open.watch.stop();
const stopped = await acquire(true);
await collect();
expect(stopped.source.deref()).toBeUndefined();
expect(stopped.listener.deref()).toBeUndefined();
expect(stopped.watch.value).toBe(1);
});
});

4
packages/durable/CHANGELOG.md
View file

@ -13,6 +13,10 @@
- `HarnessOptions.readOnly` opens a Harness as a viewer of a session another process drives: open leaves running tasks as committed instead of recovering them, every commit with writes rejects with `ReadOnlySession` before it reaches Storage, and `resume()` and every call that would run tasks reject with it.
### Changed
- Document, conversation view, task graph, and event watches use Chord's `Watch` implementation; behaviour is unchanged.
## [1.4.0] - 2026-10-21
### Breaking Changes

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

@ -1,7 +1,6 @@
import type { Context, JsonValue, WatchEnd } from "@earendil-works/chord";
import { type Context, type JsonValue, Watch, 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 { CommittedWatch } from "../session/observation.ts";
import type {
CommitChange,
CommitPublication,
@ -168,7 +167,7 @@ export async function watchEvents(
context: Context,
): Promise<AgentEventStream> {
const views = conversationViews(harness);
let watch!: CommittedWatch<readonly AgentEvent[]>;
let watch!: Watch<readonly AgentEvent[]>;
let snapshot!: SnapshotEvent;
await views.attach(
conversationId,
@ -180,7 +179,7 @@ export async function watchEvents(
let current = initial;
snapshot = snapshotOf(initial);
// Batches are the watch's values; an overflow delivers a snapshot of the newest view instead.
watch = new CommittedWatch<readonly AgentEvent[]>(
watch = new Watch<readonly AgentEvent[]>(
[],
release,
(error) => views.report(error),

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

@ -1,7 +1,6 @@
import type { Context, JsonValue, WatchHandle } from "@earendil-works/chord";
import { type Context, type JsonValue, Watch, type WatchHandle } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT, withoutAbortSignal } from "@earendil-works/chord/context";
import { applyImmutable, type Op } from "@earendil-works/chord/delta";
import { CommittedWatch } from "../session/observation.ts";
import { contained, type SessionImpl } from "../session/session.ts";
import type {
CommitPublication,
@ -56,7 +55,7 @@ export type TaskGraph = {
export type TaskGraphWatch = WatchHandle<TaskGraph>;
type AnyTaskRecord = TaskRecord<JsonValue, JsonValue, JsonValue>;
type Observer = CommittedWatch<TaskGraph>;
type Observer = Watch<TaskGraph>;
type Mount = {
value: TaskGraph;
readonly observers: Set<Observer>;
@ -108,7 +107,7 @@ export class TaskGraphView {
/** A serialized exact-frame watch of the graph; cancelling `context` stops it. */
async watch(context: Context): Promise<TaskGraphWatch> {
const { observer } = await this.#attach(
(value, release) => new CommittedWatch<TaskGraph>(value, release, (error) => this.#session.report(error)),
(value, release) => new Watch<TaskGraph>(value, release, (error) => this.#session.report(error)),
context,
);
const signal = context.abortSignal;

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

@ -1,7 +1,6 @@
import type { Context, JsonValue, WatchHandle } from "@earendil-works/chord";
import { type Context, type JsonValue, Watch, type WatchHandle } from "@earendil-works/chord";
import { withoutAbortSignal } from "@earendil-works/chord/context";
import { applyImmutable, type NonEmptyPath, type Op, type Path } from "@earendil-works/chord/delta";
import { CommittedWatch } from "../session/observation.ts";
import { contained, type SessionImpl } from "../session/session.ts";
import type {
CommitPublication,
@ -111,8 +110,7 @@ export class ConversationViews {
async watch(id: ConversationId, context: Context): Promise<WatchHandle<ConversationView>> {
const { observer } = await this.attach(
id,
(value, release) =>
new CommittedWatch<ConversationView>(value, release, (error) => this.#session.report(error)),
(value, release) => new Watch<ConversationView>(value, release, (error) => this.#session.report(error)),
context,
);
const signal = context.abortSignal;

186
packages/durable/src/session/observation.ts
View file

@ -1,186 +0,0 @@
import type { Context, JsonValue, WatchEnd, WatchHandle } from "@earendil-works/chord";
import { withoutAbortSignal } from "@earendil-works/chord/context";
import type { Op } from "@earendil-works/chord/delta";
import { errorMessage } from "../errors.ts";
import type { JsonObject } from "../types.ts";
export type ObservedDocumentValue = Readonly<JsonObject> | null;
/** Maximum exact committed frames retained behind one unavailable watch listener. */
const MAX_PENDING_WATCH_FRAMES = 100;
/** Canonical terminal frame of a retired document incarnation. */
const RETIREMENT_OPERATIONS: readonly Op[] = [["r", null]];
type WatchFrame<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;
};
type WatchListener<T> = (value: T, ops: readonly Op[], context: Context) => Promise<void>;
/**
* Serialized exact-frame watch bound to one document incarnation or conversation view. Its source ends it explicitly:
* `retire()` for a retired document, `end()` for a closed or failed Session. 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.
* @internal
*/
export class CommittedWatch<T = ObservedDocumentValue> implements WatchHandle<T> {
#detach: (() => void) | undefined;
#report: ((error: unknown) => void) | undefined;
#replace: (() => T) | undefined;
readonly #closed = Promise.withResolvers<WatchEnd>();
readonly #pending: WatchFrame<T>[] = [];
#value: T;
#listener: WatchListener<T> | undefined;
#started = false;
#scheduled = false;
#running = false;
#retired = false;
/** Set once, by `#terminate()`. */
#end: WatchEnd | undefined;
#cancellationSignal: AbortSignal | undefined;
#cancellationListener: (() => void) | undefined;
/**
* `report` gets a listener's throw, which also ends the watch. `replace` gives the value an overflow delivers; by
* default the newest value.
*/
constructor(value: T, detach: () => void, report: (error: unknown) => void, replace?: () => T) {
this.#value = value;
this.#detach = detach;
this.#report = report;
this.#replace = replace;
}
get value(): T {
return this.#value;
}
get closed(): Promise<WatchEnd> {
return this.#closed.promise;
}
start(listener: WatchListener<T>): void {
if (this.#started) throw new Error("Watch is already started");
if (this.#end !== undefined) throw new Error("Watch is stopped");
this.#started = true;
this.#listener = listener;
if (this.#pending.length > 0) this.#schedule();
}
stop(): Promise<WatchEnd> {
this.#terminate({ reason: "stopped" });
return this.#closed.promise;
}
observeCancellation(signal: AbortSignal): void {
if (this.#cancellationSignal !== undefined) throw new Error("Watch cancellation is already installed");
if (this.#end !== undefined) return;
this.#cancellationSignal = signal;
this.#cancellationListener = () => this.cancel();
signal.addEventListener("abort", this.#cancellationListener, { once: true });
if (signal.aborted) this.cancel();
}
cancel(): void {
this.#terminate({ reason: "cancelled" });
}
/** The source closed: end with `ended`, or with `failed` and the error when the Session failed. */
end(failure?: { readonly error: unknown }): void {
this.#terminate(failure === undefined ? { reason: "ended" } : { reason: "failed", error: failure.error });
}
advance(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: [["r", replacement as JsonValue]], context });
} else {
this.#pending.push({ value, ops, context });
}
if (this.#started) this.#schedule();
}
/**
* The document incarnation retired: queue the last frame, `[["r", null]]` with value `null`, after which the watch
* ends `retired`. An overflowing backlog is dropped, since the frame replaces the root anyway.
*/
retire(this: CommittedWatch<ObservedDocumentValue>, 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: null, ops: RETIREMENT_OPERATIONS, context, retires: true });
if (this.#started) this.#schedule();
}
#schedule(): void {
if (this.#scheduled || this.#running || this.#end !== undefined) return;
this.#scheduled = true;
queueMicrotask(() => {
this.#scheduled = false;
void this.#drain();
});
}
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);
try {
await listener(frame.value, frame.ops, deliveryContext);
} catch (error) {
if (this.#end === undefined) {
this.#report?.(error);
this.#terminate({ reason: "listener_error", error: toError(error) });
}
break;
}
if (frame.retires === true) {
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
* callback already running keeps running.
*/
#terminate(end: WatchEnd): void {
if (this.#end !== undefined) return;
this.#end = end;
const detach = this.#detach;
this.#detach = undefined;
this.#report = undefined;
this.#replace = undefined;
this.#listener = undefined;
this.#pending.length = 0;
if (this.#cancellationListener !== undefined) {
this.#cancellationSignal?.removeEventListener("abort", this.#cancellationListener);
}
this.#cancellationSignal = undefined;
this.#cancellationListener = undefined;
detach?.();
this.#closed.resolve(end);
}
}
function toError(error: unknown): Error {
return error instanceof Error ? error : new Error(errorMessage(error));
}

15
packages/durable/src/session/session.ts
View file

@ -1,4 +1,4 @@
import type { Context, JsonValue } from "@earendil-works/chord";
import { type Context, type JsonValue, Watch } from "@earendil-works/chord";
import { awaitWithContext, BACKGROUND_CONTEXT, withoutAbortSignal } from "@earendil-works/chord/context";
import { type Op, track } from "@earendil-works/chord/delta";
import {
@ -56,9 +56,11 @@ import type {
TaskRecord,
Tx,
} from "../types.ts";
import { CommittedWatch, type ObservedDocumentValue } from "./observation.ts";
import { type LoadedDocument, Transaction, type TransactionHost, type TransactionScope } from "./transaction.ts";
/** What a document watch observes: the committed value of its incarnation, `null` once it retired. */
type ObservedDocumentValue = Readonly<JsonObject> | null;
/** The default source of a report: a listener, which never fails what ran it. */
const LISTENER: ReportInfo = { type: "listener_error" };
@ -775,10 +777,7 @@ export class SessionImpl implements Session {
* Attach a watch to one committed incarnation: check the definition, then forward this incarnation's committed
* changes, its retirement, and close. The watch's detach removes both subscriptions.
*/
#attachDocument(
definition: AnyDocToken["definition"],
loaded: LoadedDocument,
): CommittedWatch<ObservedDocumentValue> {
#attachDocument(definition: AnyDocToken["definition"], loaded: LoadedDocument): Watch<ObservedDocumentValue> {
checkRecordScope(definition, loaded.record);
checkRecordVersion(definition, loaded.record, loaded.storedVersion);
let unsubscribeCommit = (): void => {};
@ -790,9 +789,7 @@ export class SessionImpl implements Session {
unsubscribeClose();
unpin();
};
const observer = new CommittedWatch<ObservedDocumentValue>(loaded.tracker.value, detach, (error) =>
this.report(error),
);
const observer = new Watch<ObservedDocumentValue>(loaded.tracker.value, detach, (error) => this.report(error));
const observed = { version: loaded.valueVersion };
unsubscribeCommit = this.observeCommits((publication, context) => {
for (const change of publication.changes) {

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

@ -1,6 +1,6 @@
import { setFlagsFromString } from "node:v8";
import { runInNewContext } from "node:vm";
import type { Context, JsonValue } from "@earendil-works/chord";
import type { Context } from "@earendil-works/chord";
import {
BACKGROUND_CONTEXT,
createContextKey,
@ -11,7 +11,6 @@ import {
import { applyImmutable } from "@earendil-works/chord/delta";
import { type DocumentWatch, defineDoc, MemoryStorage } from "@earendil-works/pi-durable";
import { describe, expect, it } from "vitest";
import { CommittedWatch } from "../src/session/observation.ts";
import { SessionImpl } from "../src/session/session.ts";
import { context, documentChanges, flush, openTestSession } from "./session-support.ts";
@ -508,31 +507,6 @@ describe("Session document watches", () => {
await watch.stop();
});
it("ends retired only on its source's retirement, never on a null value", async () => {
const watch = new CommittedWatch<JsonValue>(
1,
() => {},
() => {},
);
const frames: Array<{ value: JsonValue; ops: unknown }> = [];
watch.start(async (value, ops) => {
frames.push({ value, ops });
});
let end: unknown;
void watch.closed.then((settled) => {
end = settled;
});
watch.advance(null, [["r", null]], context);
watch.advance(2, [["r", 2]], context);
await flush();
expect(frames).toEqual([
{ value: null, ops: [["r", null]] },
{ value: 2, ops: [["r", 2]] },
]);
expect(end).toBeUndefined();
expect(await watch.stop()).toEqual({ reason: "stopped" });
});
it("retains neither its Session nor its listener once stopped, while still referenced", async () => {
setFlagsFromString("--expose-gc");
const gc = runInNewContext("gc") as () => void;