perf(hang): keep a bulk max-delay skip linear in buffered groups

bench/README.md

@@ -479,7 +479,7 @@ reconnects x 64 characters, and keeps its run directory as the `bench-browser`
 artifact.
 
 The in-memory frame, retained-history, gap ledger, bandwidth allocation, cached-group
-replay, and resume-history paths also run directly in Chromium:
+replay, resume-history, and buffered-group paths also run directly in Chromium:
 
 ```bash
 nix develop --command bun bench/browser/micro.ts --bench frames --out .tmp/frames.json base=/path/to/release current=. aa=.
@@ -488,6 +488,7 @@ nix develop --command bun bench/browser/micro.ts --bench ledger --out .tmp/ledge
 nix develop --command bun bench/browser/micro.ts --bench bandwidth --out .tmp/bandwidth.json before=/path/to/main current=. aa=.
 nix develop --command bun bench/browser/micro.ts --bench replay --out .tmp/replay.json before=/path/to/main current=. aa=.
 nix develop --command bun bench/browser/micro.ts --bench resume-history --out .tmp/resume-history.json before=/path/to/main current=. aa=.
+nix develop --command bun bench/browser/micro.ts --bench buffered --out .tmp/buffered.json before=/path/to/main current=. aa=.
 ```
 
 Each variant uses its own workspace source and a persistent, cross-origin isolated
@@ -514,6 +515,15 @@ Its `time_mode` column is 0 for untimed, 1 for millisecond timestamps, and 2
 for mixed millisecond/microsecond timestamps, which exercise exact cross-scale
 comparison.
 
+`buffered` drives every checkout's `@moq/hang` container consumer with one fixture,
+sweeping buffered groups against frames per group, then publishers against players.
+`steady` reads each frame as it lands (its `ns_per_op` is the frame delay);
+`lookahead` and `sparse` decode every group before draining them, in order or
+shuffled with a missing sequence between groups; `burst` times only the arrival
+that skips every buffered group once the max delay drops to zero, per skipped group.
+Each case checks its exact buffered ranges, timestamp order and payloads, and that
+skipped groups are closed; `gaps` and `warnings` must match between variants.
+
 `replay` sweeps retained groups (64, 512, 2048) against subscribers (1, 8) for
 monotonic, reverse, and shuffled arrivals. `join` rows time each subscriber
 joining the cache and replaying every retained group; `append` rows time each of
@@ -542,8 +552,8 @@ route hold, from Chromium's precise memory info; `serve_us` and
 `read_us` include no setup. Bun's collector scans its
 stack conservatively, so the fixture refuses to run there.
 
-The browser smoke lane preserves one measured round of all six fixtures, and drift,
-ledger, bandwidth and replay also run in the nightly JS benchmark leg. Timing is
+The browser smoke lane preserves one measured round of all seven fixtures, and drift,
+ledger, bandwidth, replay and buffered also run in the nightly JS benchmark leg. Timing is
 observational, with the A/A spread showing the host's noise floor.
 
 ## Pattern and container coverage

bench/browser/micro.ts

@@ -23,9 +23,12 @@ const benchmarks: Record<string, { entry: string; rows: number }> = {
 	bandwidth: { entry: "js/net/bench/bandwidth.ts", rows: 44 },
 	replay: { entry: "js/net/bench/replay.ts", rows: 72 },
 	"resume-history": { entry: "js/net/bench/resume-history.ts", rows: 72 },
+	buffered: { entry: "js/hang/bench/buffered.ts", rows: 36 },
 };
 const selected = Object.hasOwn(benchmarks, benchmark) ? benchmarks[benchmark] : undefined;
 if (!selected) throw new Error(`--bench must be one of: ${Object.keys(benchmarks).join(", ")}`);
+// The fixture directory, relative to each checkout; buffered drives @moq/hang's container consumer.
+const dir = benchmark === "buffered" ? "js/hang/bench" : "js/net/bench";
 const rounds = Number(values.rounds);
 if (!Number.isSafeInteger(rounds) || rounds < 1) throw new Error("--rounds must be a positive integer");
 const roots = new Map<string, string>();
@@ -49,12 +52,10 @@ for (const [name, root] of roots) {
 				setup(build) {
 					// Internal modules have no package export, so select each variant's own explicitly.
 					build.onResolve({ filter: /^\.\.\/src\// }, ({ path, importer }) =>
-						importer.endsWith(`/js/net/bench/${benchmark}.ts`)
-							? { path: resolve(root, "js/net/bench", path) }
-							: undefined,
+						importer.endsWith(`/${dir}/${benchmark}.ts`) ? { path: resolve(root, dir, path) } : undefined,
 					);
 					build.onLoad(
-						{ filter: /\/bench\/(frames|drift|ledger|bandwidth|replay|resume-history)\.ts$/ },
+						{ filter: /\/bench\/(frames|drift|ledger|bandwidth|replay|resume-history|buffered)\.ts$/ },
 						async ({ path }) => {
 							const source = await Bun.file(path).text();
 							if (benchmark !== "frames")
@@ -126,7 +127,8 @@ try {
 		pages.set(name, page);
 		rows.set(name, []);
 		page.on("console", (message) => {
-			if (/^(lite|ietf|\d+),/.test(message.text())) rows.get(name)?.push(message.text());
+			if (/^(lite|ietf|steady|lookahead|sparse|burst|\d+),/.test(message.text()))
+				rows.get(name)?.push(message.text());
 		});
 		page.on("pageerror", (error) => {
 			console.error(error);

bench/browser/run.sh

@@ -88,7 +88,7 @@ bun install --frozen-lockfile
 
 cd "$HERE"
 # The in-memory paths share Chromium installation and artifact retention with this lane.
-for benchmark in frames drift ledger bandwidth replay resume-history; do
+for benchmark in frames drift ledger bandwidth replay resume-history buffered; do
     bun micro.ts --bench "$benchmark" --rounds 1 --out "$HARNESS_RUN/$benchmark.json" \
         current="$WORKSPACE" aa="$WORKSPACE"
 done

js/hang/bench/buffered.ts

@@ -0,0 +1,219 @@
+/** Sweep buffered groups, frames per group, publishers and players through the container consumer. */
+import { Milli } from "@moq/clock";
+import { type Group, Time, Track, Varint } from "@moq/net";
+import { Signal } from "@moq/signals";
+import { Consumer } from "../src/container/consumer.ts";
+import { Format } from "../src/container/legacy.ts";
+
+// Far past every fixture's span, so only the burst case skips anything.
+const HOLD = Milli(1_000_000_000);
+// One group per second at 30 fps.
+const FRAME_US = 33_333;
+const GROUP_US = 1_000_000;
+const PAYLOAD = new Uint8Array([7]);
+// Measured operations per case, repeating smaller ones.
+const MIN_OPS = 8192;
+
+type Scenario = "steady" | "lookahead" | "sparse" | "burst";
+
+interface Player {
+	track: Track.Producer;
+	consumer: Consumer;
+	// Every group the consumer received, to check that what it skipped is closed.
+	received: Group.Consumer[];
+}
+
+function timestamp(group: number, frame: number): number {
+	return group * GROUP_US + frame * FRAME_US;
+}
+
+function payload(group: number, frame: number): Uint8Array {
+	const prefix = Varint.encode(timestamp(group, frame));
+	const data = new Uint8Array(prefix.byteLength + PAYLOAD.byteLength);
+	data.set(prefix, 0);
+	data.set(PAYLOAD, prefix.byteLength);
+	return data;
+}
+
+function wire(group: number, frame: number): Group.Frame {
+	return { payload: payload(group, frame), timestamp: Time.Timestamp.fromMicros(timestamp(group, frame)) };
+}
+
+function open(track: Track.Producer, delay: Signal<Track.Delay>): Player {
+	const subscriber = track.subscribe({ delay: { max: HOLD } });
+	const received: Group.Consumer[] = [];
+	const recv = subscriber.recvGroup.bind(subscriber);
+	subscriber.recvGroup = async () => {
+		const group = await recv();
+		if (group) received.push(group);
+		return group;
+	};
+	return { track, consumer: new Consumer(subscriber, { format: new Format("video"), delay }), received };
+}
+
+// The deterministic arrival order of the sparse case: every group exactly once, shuffled.
+function shuffled(count: number): number[] {
+	const order = Array.from({ length: count }, (_, i) => i);
+	let seed = 42;
+	for (let i = count - 1; i > 0; i--) {
+		seed = (Math.imul(seed, 1664525) + 1013904223) >>> 0;
+		const j = seed % (i + 1);
+		[order[i], order[j]] = [order[j], order[i]];
+	}
+	return order;
+}
+
+// Write one frame to every publisher, then wait until every player decoded it, as each arrival would
+// be decoded before the next lands. A decoded frame always changes some buffered range.
+async function arrive(groups: Group.Producer[], players: Player[], index: number, frame: number): Promise<void> {
+	const changed = players.map((player) => player.consumer.buffered.changed());
+	for (const group of groups) group.writeFrame(wire(index, frame));
+	await Promise.all(changed);
+}
+
+// Read every buffered frame, checking timestamp order, and count the gaps between them.
+async function drain(player: Player, expected: number[]): Promise<number> {
+	let gaps = 0;
+	for (const want of expected) {
+		for (;;) {
+			const read = await player.consumer.next();
+			if (!read) throw new Error("frame missing");
+			if (read.kind === "gap") {
+				gaps++;
+				continue;
+			}
+			if (read.kind !== "frame" || read.frame.timestamp !== want || read.frame.payload[0] !== PAYLOAD[0])
+				throw new Error("frame missing or reordered");
+			break;
+		}
+	}
+	return gaps;
+}
+
+function ranges(player: Player): string {
+	return JSON.stringify(player.consumer.buffered.peek());
+}
+
+/** Return elapsed ns, the measured operations, gaps and warnings for one case. */
+async function measure(
+	scenario: Scenario,
+	groups: number,
+	frames: number,
+	publishers: number,
+	readers: number,
+): Promise<{ ns: number; ops: number; gaps: number; warnings: number }> {
+	const delay = new Signal<Track.Delay>({ max: HOLD });
+	// Stamped like a hang publisher's media, so the transport finds its live edge at the newest group.
+	const tracks = Array.from({ length: publishers }, (_, i) =>
+		new Track.Producer(`p${i}`).accept({ timescale: Time.Timescale.MICRO }),
+	);
+	const players = tracks.flatMap((track) => Array.from({ length: readers }, () => open(track, delay)));
+	let warnings = 0;
+	const warn = console.warn;
+	console.warn = () => {
+		warnings++;
+	};
+	try {
+		const sequence = (i: number) => (scenario === "sparse" ? i * 2 : i);
+		const order = scenario === "sparse" ? shuffled(groups) : Array.from({ length: groups }, (_, i) => i);
+		const expected = Array.from({ length: groups * frames }, (_, i) =>
+			timestamp(sequence(Math.floor(i / frames)), i % frames),
+		);
+		let gaps = 0;
+		let start = performance.now();
+		if (scenario === "steady") {
+			// The player reads each frame as it lands, so nothing builds up behind it.
+			for (const i of order) {
+				const sources = tracks.map((track) => track.createGroup(i));
+				for (let frame = 0; frame < frames; frame++) {
+					const reads = players.map((player) => player.consumer.next());
+					for (const source of sources) source.writeFrame(wire(i, frame));
+					for (const read of await Promise.all(reads))
+						if (read?.kind !== "frame" || read.frame.timestamp !== timestamp(i, frame))
+							throw new Error("frame missing or reordered");
+				}
+				for (const source of sources) source.finish();
+			}
+			return { ns: (performance.now() - start) * 1e6, ops: groups * frames * players.length, gaps, warnings };
+		}
+
+		// Everything arrives before the player reads: a stalled reader, or history after a join.
+		for (const i of order) {
+			const sources = tracks.map((track) => track.createGroup(sequence(i)));
+			for (let frame = 0; frame < frames; frame++) await arrive(sources, players, sequence(i), frame);
+			for (const source of sources) source.finish();
+		}
+		// A finished track proves no missing group can still arrive.
+		if (scenario === "sparse") for (const track of tracks) track.finish();
+		await Promise.resolve();
+
+		for (const player of players) {
+			// Finished adjacent groups merge into one range; sparse ones stay one range per group.
+			const want =
+				scenario === "sparse"
+					? Array.from({ length: groups }, (_, i) => ({
+							start: timestamp(sequence(i), 0) / 1000,
+							end: timestamp(sequence(i), frames - 1) / 1000,
+						}))
+					: [{ start: 0, end: timestamp(groups - 1, frames - 1) / 1000 }];
+			if (ranges(player) !== JSON.stringify(want)) throw new Error(`wrong buffered ranges: ${ranges(player)}`);
+		}
+
+		if (scenario === "burst") {
+			// The budget drops to nothing, so the next arrival makes every older group too stale to keep.
+			delay.set({ max: Milli.zero });
+			start = performance.now();
+			const sources = tracks.map((track) => track.createGroup(groups));
+			await arrive(sources, players, groups, 0);
+			const elapsed = (performance.now() - start) * 1e6;
+			const newest = timestamp(groups, 0) / 1000;
+			for (const player of players) {
+				if (ranges(player) !== JSON.stringify([{ start: newest, end: newest }]))
+					throw new Error(`skip left ${ranges(player)}`);
+				if (player.received.slice(0, groups).some((group) => !group.isClosed))
+					throw new Error("skipped group left open");
+				const read = await player.consumer.next();
+				if (read?.kind !== "frame" || read.frame.timestamp !== timestamp(groups, 0))
+					throw new Error("newest frame missing");
+			}
+			return { ns: elapsed, ops: groups * players.length, gaps, warnings };
+		}
+
+		for (const player of players) gaps += await drain(player, expected);
+		return { ns: (performance.now() - start) * 1e6, ops: groups * frames * players.length, gaps, warnings };
+	} finally {
+		console.warn = warn;
+		for (const player of players) player.consumer.close();
+		for (const track of tracks) track.close();
+	}
+}
+
+/**
+ * Print one CSV row per case. `ns_per_op` is per frame and player, except for a burst, where it is per
+ * skipped group and player and covers only the arrival that skips them.
+ */
+export async function run(): Promise<void> {
+	console.log("scenario,groups,frames,publishers,players,ops,gaps,warnings,ns_per_op");
+	const cases: [Scenario, number, number, number, number][] = [];
+	for (const scenario of ["steady", "lookahead", "sparse", "burst"] as const)
+		for (const groups of [1, 16, 256, 1024])
+			for (const frames of [1, 30]) cases.push([scenario, groups, frames, 1, 1]);
+	for (const publishers of [1, 4])
+		for (const players of [1, 8]) cases.push(["lookahead", 64, 30, publishers, players]);
+	for (const [scenario, groups, frames, publishers, players] of cases) {
+		// Small cases repeat, so a coarse browser clock still resolves them.
+		const total = { ns: 0, ops: 0, gaps: 0, warnings: 0 };
+		while (total.ops < MIN_OPS) {
+			const { ns, ops, gaps, warnings } = await measure(scenario, groups, frames, publishers, players);
+			total.ns += ns;
+			total.ops += ops;
+			total.gaps += gaps;
+			total.warnings += warnings;
+		}
+		console.log(
+			`${scenario},${groups},${frames},${publishers},${players},${total.ops},${total.gaps},${total.warnings},${(total.ns / total.ops).toFixed(0)}`,
+		);
+	}
+}
+
+if (import.meta.main) await run();

js/hang/src/container/consumer.skips.test.ts

@@ -139,3 +139,49 @@ test("a backlog start waits on a successor without frames", async () => {
 		warn.mockRestore();
 	}
 });
+
+test("Consumer measures each skip's span over every group left, not just the new oldest", async () => {
+	const warn = spyOn(console, "warn").mockImplementation(() => {});
+	const track = new Track.Producer("tone");
+	const delay = new Signal<Track.Delay>({ max: Milli(30_000) });
+	const consumer = new Consumer(track.subscribe({ delay: { max: Milli(30_000) } }), {
+		format: new Format("data"),
+		delay,
+	});
+	const write = (sequence: number, timestamp: number) => {
+		const group = track.createGroup(sequence);
+		group.writeFrame({ payload: Varint.encode(timestamp), timestamp: Time.Timestamp.now() });
+	};
+	const buffered = async (timestamp: number) => {
+		while (consumer.buffered.peek().at(-1)?.end !== timestamp) await consumer.buffered.changed();
+	};
+	try {
+		// Group 2 starts below group 1, so the span after skipping group 0 still reaches back to it.
+		for (const [sequence, timestamp] of [
+			[0, 0],
+			[1, 10_000],
+			[2, 1_000],
+			[3, 12_000],
+		])
+			write(sequence, timestamp);
+		await buffered(12);
+		// Reading the first frame warms the reader, so the age skip runs on the max delay alone.
+		expect(await consumer.next()).toMatchObject({ kind: "frame", group: 0 });
+		delay.set({ max: Milli(5) });
+		write(4, 13_000);
+		await buffered(13);
+		expect(warn).toHaveBeenLastCalledWith("skipping slow groups: track=tone 0 -> 3 count=3");
+		expect(consumer.buffered.peek()).toEqual([
+			{ start: Milli(12), end: Milli(12) },
+			{ start: Milli(13), end: Milli(13) },
+		]);
+		// The skipped groups held media past the first frame, so their span reads as a gap first.
+		expect(await consumer.next()).toMatchObject({ kind: "gap", timestamp: 0, duration: 12_000 });
+		expect(await consumer.next()).toMatchObject({ kind: "frame", group: 3 });
+	} finally {
+		consumer.close();
+		track.finish();
+		track.close();
+		warn.mockRestore();
+	}
+});