refactor(net): encapsulate Group.state behind non-blocking read methods

js/json/src/consumer.ts

@@ -7,8 +7,9 @@ import type { Config } from "./producer.ts";
 /**
  * Consumes a JSON value from a track, reconstructing it from snapshots and deltas.
  *
- * Reads each group's snapshot (frame 0) and applies the following frames as merge patches,
- * yielding the reconstructed value after each one.
+ * Reads each group's snapshot (frame 0) and applies the following frames as merge patches. A live
+ * consumer yields each update as it arrives; a consumer that has fallen behind (or just joined)
+ * collapses the buffered backlog and yields only the latest value. See {@link next}.
  */
 export class Consumer<T> {
 	#track: Moq.Track;
@@ -28,7 +29,15 @@ export class Consumer<T> {
 		this.#decompress = config.compression ?? false;
 	}
 
-	/** Get the next reconstructed value, or `undefined` once the track ends. */
+	/**
+	 * Get the next reconstructed value, or `undefined` once the track ends.
+	 *
+	 * Applies every frame already buffered in the group but yields only the latest reconstructed
+	 * value: the intermediate reconstructions are stale, so a late joiner (or any consumer that has
+	 * fallen behind) catches up to the head in one step instead of replaying every superseded state.
+	 * Frames are still decoded in order (the DEFLATE window and merge patches are sequential); only
+	 * the per-frame yield is skipped.
+	 */
 	async next(): Promise<T | undefined> {
 		for (;;) {
 			if (!this.#group) {
@@ -41,6 +50,16 @@ export class Consumer<T> {
 				this.#decoder = undefined;
 			}
 
+			// Drain every frame already buffered, keeping only the latest reconstructed value.
+			let value: T | undefined;
+			let advanced = false;
+			for (let frame = this.#group.tryReadFrame(); frame !== undefined; frame = this.#group.tryReadFrame()) {
+				value = this.#apply(frame);
+				advanced = true;
+			}
+			if (advanced) return value;
+
+			// Nothing buffered: block for the next frame (or the group's end).
 			const frame = await this.#group.readFrame();
 			if (frame === undefined) {
 				// The group is exhausted; advance to the next one.

js/json/src/json.test.ts

@@ -186,6 +186,19 @@ test("array change is a wholesale delta", async () => {
 	expect(await structure(track)).toEqual([2]);
 });
 
+test("late joiner collapses a buffered backlog to the latest value", async () => {
+	const track = new Track("test");
+	const producer = new Producer<Value>(track, { deltaRatio: 100 });
+	for (let n = 0; n <= 20; n++) {
+		producer.update({ n });
+	}
+	producer.finish();
+
+	// A whole group's worth of snapshot + deltas is buffered before the consumer reads, so it applies
+	// them all but yields only the latest value once, not every superseded state.
+	expect(await drain(track)).toEqual([{ n: 20 }]);
+});
+
 test("frame cap rolls snapshot", async () => {
 	const track = new Track("test");
 	const producer = new Producer<Value>(track, { deltaRatio: 1_000_000 });

js/net/src/group.test.ts

@@ -0,0 +1,77 @@
+import { expect, test } from "bun:test";
+import { Group } from "./group.ts";
+
+const dec = new TextDecoder();
+
+test("tryReadFrame drains buffered frames then returns undefined", () => {
+	const group = new Group(0);
+	group.writeString("a");
+	group.writeString("b");
+
+	expect(dec.decode(group.tryReadFrame())).toBe("a");
+	expect(dec.decode(group.tryReadFrame())).toBe("b");
+	// Nothing buffered: undefined, and the group is not closed so this is not end-of-group.
+	expect(group.tryReadFrame()).toBeUndefined();
+});
+
+test("tryReadFrameSequence reports per-frame sequence numbers", () => {
+	const group = new Group(7);
+	group.writeString("a");
+	group.writeString("b");
+
+	expect(group.tryReadFrameSequence()).toEqual({ sequence: 0, data: new TextEncoder().encode("a") });
+	expect(group.tryReadFrameSequence()).toEqual({ sequence: 1, data: new TextEncoder().encode("b") });
+	expect(group.tryReadFrameSequence()).toBeUndefined();
+});
+
+test("done distinguishes a finished group from one that is merely empty", () => {
+	const group = new Group(0);
+	// Open and empty: not done (more frames may arrive), and tryReadFrame is undefined.
+	expect(group.tryReadFrame()).toBeUndefined();
+	expect(group.done).toBe(false);
+
+	group.writeString("a");
+	// Buffered but closed: still not done until the frame is drained.
+	group.close();
+	expect(group.done).toBe(false);
+
+	group.tryReadFrame();
+	// Drained and closed: now done.
+	expect(group.tryReadFrame()).toBeUndefined();
+	expect(group.done).toBe(true);
+});
+
+test("readable resolves once a frame is buffered", async () => {
+	const group = new Group(0);
+	// No frame yet: readable() must stay pending for an empty, open group.
+	const readable = group.readable();
+	let settled = false;
+	void readable.then(() => {
+		settled = true;
+	});
+	await Promise.resolve();
+	expect(settled).toBe(false);
+
+	// Writing makes it resolve.
+	group.writeString("hi");
+	await readable; // must not hang
+	expect(dec.decode(group.tryReadFrame())).toBe("hi");
+});
+
+test("readable resolves once the group closes, even with nothing buffered", async () => {
+	const group = new Group(0);
+	const readable = group.readable();
+	group.close();
+	await readable; // resolves on close so callers don't wait forever
+	expect(group.tryReadFrame()).toBeUndefined();
+});
+
+test("buffered frames are still readable after the group closes", async () => {
+	const group = new Group(0);
+	group.writeString("a");
+	group.close();
+
+	// Closing doesn't discard buffered frames; the blocking reader drains them before ending.
+	expect(await group.readString()).toBe("a");
+	expect(await group.readFrame()).toBeUndefined();
+});

js/net/src/group.ts

@@ -1,7 +1,7 @@
 import { Signal } from "@moq/signals";
 
 /** Reactive backing state for a {@link Group}: buffered frames, a closed flag, and the running frame count. */
-export class GroupState {
+class GroupState {
 	frames = new Signal<Uint8Array[]>([]);
 	closed = new Signal<boolean | Error>(false);
 	total = new Signal<number>(0); // The total number of frames in the group thus far
@@ -12,8 +12,9 @@ export class Group {
 	/** Sequence number of this group within its track. */
 	readonly sequence: number;
 
-	/** Reactive backing state. */
-	state = new GroupState();
+	// Reactive backing state, deliberately private: read through the read* methods so callers can't
+	// poke the signals directly (and so the internal representation can change).
+	#state = new GroupState();
 
 	/** Resolves with the abort error (or undefined) once closed. */
 	readonly closed: Promise<Error | undefined>;
@@ -23,7 +24,7 @@ export class Group {
 
 		// Cache the closed promise to avoid recreating it every time.
 		this.closed = new Promise((resolve) => {
-			const dispose = this.state.closed.subscribe((closed) => {
+			const dispose = this.#state.closed.subscribe((closed) => {
 				if (!closed) return;
 				resolve(closed instanceof Error ? closed : undefined);
 				dispose();
@@ -36,13 +37,13 @@ export class Group {
 	 * @param frame - The frame to write
 	 */
 	writeFrame(frame: Uint8Array) {
-		if (this.state.closed.peek()) throw new Error("group is closed");
+		if (this.#state.closed.peek()) throw new Error("group is closed");
 
-		this.state.frames.mutate((frames) => {
+		this.#state.frames.mutate((frames) => {
 			frames.push(frame);
 		});
 
-		this.state.total.update((total) => total + 1);
+		this.#state.total.update((total) => total + 1);
 	}
 
 	/** Write a string as a single UTF-8 encoded frame. */
@@ -60,36 +61,67 @@ export class Group {
 		this.writeFrame(new Uint8Array([bool ? 1 : 0]));
 	}
 
+	/** True once no further frames can be read: the group has closed and every buffered frame is read. */
+	get done(): boolean {
+		return this.#state.frames.peek().length === 0 && this.#state.closed.peek() !== false;
+	}
+
 	/**
-	 * Reads the next frame from the group.
-	 * @returns A promise that resolves to the next frame or undefined
+	 * Reads the next already-buffered frame without blocking.
+	 *
+	 * Returns `undefined` when nothing is buffered right now. That is *not* by itself end-of-group:
+	 * check {@link done} to tell "no frame buffered yet" (more may arrive) from "the group finished".
+	 * Drain a backlog by looping until this returns `undefined`, then branch on {@link done}: if not
+	 * done, {@link readable} resolves when the next frame arrives.
 	 */
-	async readFrame(): Promise<Uint8Array | undefined> {
-		for (;;) {
-			const frames = this.state.frames.peek();
-			const frame = frames.shift();
-			if (frame) return frame;
+	tryReadFrame(): Uint8Array | undefined {
+		return this.tryReadFrameSequence()?.data;
+	}
 
-			const closed = this.state.closed.peek();
-			if (closed instanceof Error) throw closed;
-			if (closed) return;
+	/** Like {@link tryReadFrame} but also reports the frame's sequence number within the group. */
+	tryReadFrameSequence(): { sequence: number; data: Uint8Array } | undefined {
+		const frames = this.#state.frames.peek();
+		const data = frames.shift();
+		if (data === undefined) return undefined;
+		return { sequence: this.#state.total.peek() - frames.length - 1, data };
+	}
 
-			await Signal.race(this.state.frames, this.state.closed);
+	/**
+	 * Resolves once {@link readFrame} would not block: a frame is buffered, or the group has closed.
+	 * Always settles (never hangs), so on a finished group it resolves immediately; pair it with
+	 * {@link done} to avoid re-waiting on a group that has nothing left.
+	 *
+	 * Lets a caller fold "this group has a frame" into a larger wait (e.g. racing it against a new
+	 * group arriving) without touching the group's internal signals.
+	 */
+	async readable(): Promise<void> {
+		for (;;) {
+			if (this.#state.frames.peek().length > 0) return;
+			if (this.#state.closed.peek()) return;
+			await Signal.race(this.#state.frames, this.#state.closed);
 		}
 	}
 
+	/**
+	 * Reads the next frame from the group.
+	 * @returns A promise that resolves to the next frame or undefined
+	 */
+	async readFrame(): Promise<Uint8Array | undefined> {
+		return (await this.readFrameSequence())?.data;
+	}
+
 	/** Reads the next frame along with its sequence number within the group. */
 	async readFrameSequence(): Promise<{ sequence: number; data: Uint8Array } | undefined> {
 		for (;;) {
-			const frames = this.state.frames.peek();
-			const frame = frames.shift();
-			if (frame) return { sequence: this.state.total.peek() - frames.length - 1, data: frame };
+			const next = this.tryReadFrameSequence();
+			if (next) return next;
 
-			const closed = this.state.closed.peek();
+			// Drain buffered frames before observing the close, so a closed group still yields them.
+			const closed = this.#state.closed.peek();
 			if (closed instanceof Error) throw closed;
-			if (closed) return;
+			if (closed) return undefined;
 
-			await Signal.race(this.state.frames, this.state.closed);
+			await this.readable();
 		}
 	}
 
@@ -113,6 +145,6 @@ export class Group {
 
 	/** Closes the group, optionally with an error to abort readers. */
 	close(abort?: Error) {
-		this.state.closed.set(abort ?? true);
+		this.#state.closed.set(abort ?? true);
 	}
 }

js/net/src/track.test.ts

@@ -49,3 +49,23 @@ test("nextGroupOrdered returns undefined when track closes", async () => {
 	track.close();
 	expect(await track.nextGroupOrdered()).toBeUndefined();
 });
+
+test("readFrame does not livelock when a sole group finishes before the next arrives", async () => {
+	const track = new Track("test");
+
+	// A group is appended then finished empty while the track stays open. A finished group's
+	// readable() resolves immediately, so the reader must not busy-wait on it (which would starve the
+	// macrotask queue and never observe the next group).
+	const g0 = track.appendGroup();
+	g0.close();
+
+	// The next group arrives via a macrotask; if the reader livelocks on microtasks it never runs.
+	setTimeout(() => {
+		const g1 = track.appendGroup();
+		g1.writeString("hello");
+		g1.close();
+		track.close();
+	}, 10);
+
+	expect(await track.readString()).toBe("hello");
+}, 2000);

js/net/src/track.ts

@@ -162,11 +162,9 @@ export class Track {
 
 			// Discard old groups.
 			while (groups.length > 1) {
-				const frames = groups[0].state.frames.peek();
-				const next = frames.shift();
+				const next = groups[0].tryReadFrameSequence();
 				if (next) {
-					const frame = groups[0].state.total.peek() - frames.length - 1;
-					return { group: groups[0].sequence, frame, data: next };
+					return { group: groups[0].sequence, frame: next.sequence, data: next.data };
 				}
 
 				// Skip this old group
@@ -185,20 +183,24 @@ export class Track {
 
 			// If there's a group, wait for a frame.
 			const group = groups[0];
-			const frames = group.state.frames.peek();
-			const next = frames.shift();
+			const next = group.tryReadFrameSequence();
 			if (next) {
-				const frame = group.state.total.peek() - frames.length - 1;
-				return { group: group.sequence, frame, data: next };
+				return { group: group.sequence, frame: next.sequence, data: next.data };
 			}
 
 			// If the track is closed, return undefined.
 			const closed = this.state.closed.peek();
 			if (closed instanceof Error) throw closed;
 			if (closed) return undefined;
 
-			// NOTE: We don't care if the latest group was closed or not.
-			await Signal.race(this.state.groups, this.state.closed, group.state.frames);
+			// Wake on a new group or the track closing. Only fold in the current group's readability
+			// while it can still deliver frames: a finished group's readable() resolves immediately, so
+			// racing it would busy-loop until a newer group arrives instead of waiting.
+			if (group.done) {
+				await Signal.race(this.state.groups, this.state.closed);
+			} else {
+				await Promise.race([Signal.race(this.state.groups, this.state.closed), group.readable()]);
+			}
 		}
 	}