From 42cb822f1c5d3153069dd0661bfd297424666cf1 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Fri, 23 Oct 2026 07:05:23 +0000 Subject: [PATCH] feat(chord): shared watch-handle implementation (Watch, NativeWatch); Durable uses it --- packages/chord/CHANGELOG.md | 2 + packages/chord/README.md | 8 +- packages/chord/src/index.ts | 2 +- packages/chord/src/watch.ts | 262 +++++++++- packages/chord/test/watch.test.ts | 490 ++++++++++++++++++ packages/durable/CHANGELOG.md | 4 + packages/durable/src/harness/events.ts | 7 +- packages/durable/src/harness/task-graph.ts | 7 +- packages/durable/src/harness/view.ts | 6 +- packages/durable/src/session/observation.ts | 186 ------- packages/durable/src/session/session.ts | 15 +- packages/durable/test/session-watches.test.ts | 28 +- 12 files changed, 780 insertions(+), 237 deletions(-) create mode 100644 packages/chord/test/watch.test.ts delete mode 100644 packages/durable/src/session/observation.ts diff --git a/packages/chord/CHANGELOG.md b/packages/chord/CHANGELOG.md index c44321d60..5d2dda138 100644 --- a/packages/chord/CHANGELOG.md +++ b/packages/chord/CHANGELOG.md @@ -9,6 +9,8 @@ ### Added - `WatchHandle` 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`, 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` and its implementation `NativeWatch`: 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 diff --git a/packages/chord/README.md b/packages/chord/README.md index d2a136d47..f39bea749 100644 --- a/packages/chord/README.md +++ b/packages/chord/README.md @@ -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`, fed through `advance(value, ops, context)`, +`retire(context)`, `end()` or `end({ error })`, and `observeCancellation(signal)`. +`NativeWatch` implements `NativeWatchHandle`, the same lifecycle for +in-process values: the listener gets `(value, context)` without ops, and an +overflow collapses to the newest value. diff --git a/packages/chord/src/index.ts b/packages/chord/src/index.ts index 08b2f891e..d81b621b5 100644 --- a/packages/chord/src/index.ts +++ b/packages/chord/src/index.ts @@ -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"; diff --git a/packages/chord/src/watch.ts b/packages/chord/src/watch.ts index b4922fb2f..3b8a19044 100644 --- a/packages/chord/src/watch.ts +++ b/packages/chord/src/watch.ts @@ -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 { /** Settle when the watch terminates; an already-running callback remains caller-owned. */ readonly closed: Promise; } + +/** Serialized observation of an in-process value: as `WatchHandle`, but a change is a new value without ops. */ +export interface NativeWatchHandle { + /** 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; + /** Idempotently stop future callbacks and return this watch's terminal result. */ + stop(): Promise; + /** Settle when the watch terminates; an already-running callback remains caller-owned. */ + readonly closed: Promise; +} + +type WatchListener = (value: T, ops: readonly Op[], context: Context) => Promise; + +type NativeWatchListener = (value: T, context: Context) => Promise; + +type WatchFrame = { + 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 { + #detach: (() => void) | undefined; + #report: ((error: unknown) => void) | undefined; + #replace: (() => T) | undefined; + readonly #closed = Promise.withResolvers(); + readonly #pending: WatchFrame[] = []; + #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 { + 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 { + 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, context: Context): Promise; + + /** 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 { + 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 extends WatchLifecycle> implements WatchHandle { + /** 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, context: Context): void { + this.enqueueRetirement(null, RETIREMENT_OPERATIONS, context); + } + + protected invoke(listener: WatchListener, frame: WatchFrame, context: Context): Promise { + 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 extends WatchLifecycle> implements NativeWatchHandle { + 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, frame: WatchFrame, context: Context): Promise { + return listener(frame.value, context); + } + + protected overflowOps(): readonly Op[] { + return NO_OPS; + } +} diff --git a/packages/chord/test/watch.test.ts b/packages/chord/test/watch.test.ts new file mode 100644 index 000000000..3c8282430 --- /dev/null +++ b/packages/chord/test/watch.test.ts @@ -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 { + return new Promise((resolve) => setTimeout(resolve, 0)); +} + +function deferred(): { readonly promise: Promise; readonly resolve: () => void } { + const { promise, resolve } = Promise.withResolvers(); + return { promise, resolve: () => resolve() }; +} + +/** A watch whose detach and reports are recorded. */ +function recordedWatch(value: T, replace?: () => T) { + const recorded = { detached: 0, reported: [] as unknown[] }; + const watch = new Watch( + value, + () => recorded.detached++, + (error) => recorded.reported.push(error), + replace, + ); + return { watch, recorded }; +} + +/** A started watch collecting its frames. */ +function collectingWatch(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 }): { 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 => { + 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({ 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(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(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(0); + const key = createContextKey("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(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(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(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(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(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(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(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, 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(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(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(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(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(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(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(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(0).watch; + aborted.observeCancellation(AbortSignal.abort()); + expect(await aborted.closed).toEqual({ reason: "cancelled" }); + + const cancelled = recordedWatch(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(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(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) => void) => { + const source = { value: 1 as JsonValue }; + const controller = new AbortController(); + const watch = new Watch( + 0, + () => void source, + () => void source, + () => source.value, + ); + const listener = async (): Promise => {}; + 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) => void watch.stop(), + (watch: Watch) => watch.cancel(), + (watch: Watch) => watch.end(), + (watch: Watch) => 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("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( + 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( + 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( + 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( + 0, + () => void source, + () => void source, + ); + const listener = async (): Promise => {}; + 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); + }); +}); diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 96be36d6c..76511515a 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -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 diff --git a/packages/durable/src/harness/events.ts b/packages/durable/src/harness/events.ts index b04c6a4e2..9251f21a5 100644 --- a/packages/durable/src/harness/events.ts +++ b/packages/durable/src/harness/events.ts @@ -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 { const views = conversationViews(harness); - let watch!: CommittedWatch; + let watch!: Watch; 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( + watch = new Watch( [], release, (error) => views.report(error), diff --git a/packages/durable/src/harness/task-graph.ts b/packages/durable/src/harness/task-graph.ts index d7fd62ae1..b44911173 100644 --- a/packages/durable/src/harness/task-graph.ts +++ b/packages/durable/src/harness/task-graph.ts @@ -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; type AnyTaskRecord = TaskRecord; -type Observer = CommittedWatch; +type Observer = Watch; type Mount = { value: TaskGraph; readonly observers: Set; @@ -108,7 +107,7 @@ export class TaskGraphView { /** A serialized exact-frame watch of the graph; cancelling `context` stops it. */ async watch(context: Context): Promise { const { observer } = await this.#attach( - (value, release) => new CommittedWatch(value, release, (error) => this.#session.report(error)), + (value, release) => new Watch(value, release, (error) => this.#session.report(error)), context, ); const signal = context.abortSignal; diff --git a/packages/durable/src/harness/view.ts b/packages/durable/src/harness/view.ts index 813d10ebb..cd4a14f98 100644 --- a/packages/durable/src/harness/view.ts +++ b/packages/durable/src/harness/view.ts @@ -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> { const { observer } = await this.attach( id, - (value, release) => - new CommittedWatch(value, release, (error) => this.#session.report(error)), + (value, release) => new Watch(value, release, (error) => this.#session.report(error)), context, ); const signal = context.abortSignal; diff --git a/packages/durable/src/session/observation.ts b/packages/durable/src/session/observation.ts deleted file mode 100644 index a2adda0ab..000000000 --- a/packages/durable/src/session/observation.ts +++ /dev/null @@ -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 | 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 = { - 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 = (value: T, ops: readonly Op[], context: Context) => Promise; - -/** - * 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 implements WatchHandle { - #detach: (() => void) | undefined; - #report: ((error: unknown) => void) | undefined; - #replace: (() => T) | undefined; - readonly #closed = Promise.withResolvers(); - readonly #pending: WatchFrame[] = []; - #value: T; - #listener: WatchListener | 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 { - return this.#closed.promise; - } - - start(listener: WatchListener): 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 { - 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, 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 { - 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)); -} diff --git a/packages/durable/src/session/session.ts b/packages/durable/src/session/session.ts index 99513ce91..4a6563b70 100644 --- a/packages/durable/src/session/session.ts +++ b/packages/durable/src/session/session.ts @@ -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 | 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 { + #attachDocument(definition: AnyDocToken["definition"], loaded: LoadedDocument): Watch { 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(loaded.tracker.value, detach, (error) => - this.report(error), - ); + const observer = new Watch(loaded.tracker.value, detach, (error) => this.report(error)); const observed = { version: loaded.valueVersion }; unsubscribeCommit = this.observeCommits((publication, context) => { for (const change of publication.changes) { diff --git a/packages/durable/test/session-watches.test.ts b/packages/durable/test/session-watches.test.ts index d0dfbf9d9..7bd9e78d3 100644 --- a/packages/durable/test/session-watches.test.ts +++ b/packages/durable/test/session-watches.test.ts @@ -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( - 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;