fix(watch,publish): keep the delay across a rename and close frames read during teardown; cover renames in media QA

doc/concept/audio-jitter.md

@@ -337,6 +337,17 @@ reference itself only ever moves earlier, so once the earliest subscribed track
 leaves, the clock stays anchored to it until playback re-anchors; that errs on
 the safe side.
 
+A track that replaces another in the same slot and on the same broadcast path,
+such as the new name a codec change publishes under, continues the outgoing
+track's histogram and startup ramp instead of the prior. The path's arrivals
+did not change with the name; a track on another path starts from the prior.
+Restarting would drop the target to the cold-start 100 ms at the switch, then
+raise it as the lateness is relearned, and the playout follows both moves:
+buffered audio is skipped on the drop and playback parks on each rise. Only the
+reference window restarts, so the new track's frames are measured against its
+own fastest. Its opening backlog would read as seconds late against the old
+track's arrivals.
+
 ## Rise, fall, startup, and the ceiling
 
 There is **no separate rise or fall limiter**. The histogram is the only

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

@@ -694,8 +694,19 @@ test("Consumer measures arrivals as they land", async () => {
 	}
 	expect(consumer.spread.peek()).toBeLessThan(100 as Milli);
 
+	// A replacement on the same path continues the estimate rather than restarting cold.
+	const next = new Track.Producer("next");
+	const replacement = new Consumer(next.subscribe(), {
+		format: new LegacyFormat("audio"),
+		maxDelay: 5000 as Milli,
+		predecessor: consumer,
+	});
+	expect(replacement.spread.peek()).toBe(consumer.spread.peek());
+
 	track.close();
 	consumer.close();
+	next.close();
+	replacement.close();
 });
 
 // --- Ordering ---

js/hang/src/container/consumer.ts

@@ -18,6 +18,13 @@ export interface ConsumerProps {
 	 */
 	// Read-only: a Getter (e.g. another component's output) is accepted directly.
 	maxDelay?: GetterInit<Milli>;
+	/**
+	 * The consumer this one replaces on the same path, such as the previous rendition.
+	 *
+	 * Its arrival estimate carries over instead of restarting from the cold-start prior, so a
+	 * switch neither lowers nor raises {@link Consumer.spread} on its own.
+	 */
+	predecessor?: Consumer;
 }
 
 /** One item of a {@link Consumer.next} read, in delivery order. Mirrors Rust's `container::Read`. */
@@ -122,15 +129,15 @@ export class Consumer {
 	/** The time ranges currently buffered and ready to play. */
 	readonly buffered: Getter<BufferedRanges> = this.#buffered;
 
-	#jitter = new Jitter();
-	#spread = new Signal<Milli>(this.#jitter.measured);
+	#jitter: Jitter;
+	#spread: Signal<Milli>;
 	/**
 	 * How much buffer late arrivals need, measured as frames come off the transport.
 	 *
 	 * The measured term of doc/concept/audio-jitter.md, excluding the codec frame and any advertised
-	 * floor. Starts at the cold-start estimate before any frame arrives.
+	 * floor. Starts at the predecessor's estimate, or the cold-start estimate without one.
 	 */
-	readonly spread: Getter<Milli> = this.#spread;
+	readonly spread: Getter<Milli>;
 
 	#signals = new Effect();
 	#closed = new Once<Error | null>();
@@ -141,6 +148,9 @@ export class Consumer {
 		this.#track = track;
 		this.#format = props.format;
 		this.#maxDelay = getter(props.maxDelay ?? Milli.zero);
+		this.#jitter = props.predecessor ? props.predecessor.#jitter.fork() : new Jitter();
+		this.#spread = new Signal(this.#jitter.measured);
+		this.spread = this.#spread;
 
 		this.#signals.spawn(this.#run.bind(this));
 		this.#signals.cleanup(() => {

js/hang/src/container/jitter.test.ts

@@ -93,4 +93,24 @@ describe("Jitter", () => {
 		}
 		expect(far.measured).toBe(near.measured);
 	});
+
+	it("forks what it learned, but measures the replacement's arrivals against their own", () => {
+		// A path where every tenth frame lands 300ms late.
+		const warm = new Jitter();
+		for (let i = 0; i < 500; i++) {
+			warm.observe((i * 20 + (i % 10 === 0 ? 300 : 0)) as Milli, (i * 20) as Milli);
+		}
+		const learned = warm.measured;
+		expect(learned).toBeGreaterThan(100 as Milli);
+
+		const fork = warm.fork();
+		expect(fork.measured).toBe(learned);
+
+		// The replacement opens with a backlog: older media arriving at once, long after the warm
+		// track's frames. Measured against the warm track's arrivals it would read as seconds late.
+		for (let i = 0; i < 100; i++)
+			fork.observe((20_000 + Math.floor(i / 50) * 1000) as Milli, (8000 + i * 20) as Milli);
+		expect(fork.measured).toBeLessThanOrEqual(learned);
+		expect(warm.measured).toBe(learned);
+	});
 });

js/hang/src/container/jitter.ts

@@ -106,6 +106,18 @@ export class Jitter {
 		this.#peak = delay;
 	}
 
+	/**
+	 * A new estimator that keeps what this one learned but measures arrivals afresh, for a track that
+	 * replaces this one on the same path. Its frames are measured against their own fastest, never
+	 * against this track's.
+	 */
+	fork(): Jitter {
+		const fork = new Jitter();
+		fork.#histogram = Float64Array.from(this.#histogram);
+		fork.#count = this.#count;
+		return fork;
+	}
+
 	/** The measured term: the upper edge of the bucket holding the 95th percentile. */
 	get measured(): Milli {
 		let cumulative = 0;

js/publish/src/preview.ts

@@ -155,14 +155,21 @@ export class Renderer {
 			});
 		});
 
+		// Not raced against teardown, which would drop a frame whose read had already resolved.
 		effect.spawn(async () => {
 			for (;;) {
-				const next = await effect.race(reader.read());
-				if (!next?.value) break;
+				const { value: frame } = await reader.read();
+				if (!frame) break;
+
+				// The cleanup above already released the latest frame, so keep nothing past it.
+				if (effect.abort.aborted) {
+					frame.close();
+					break;
+				}
 
 				this.#latest.update((prev) => {
 					prev?.close();
-					return next.value;
+					return frame;
 				});
 			}
 		});
@@ -265,14 +272,15 @@ export class Transcode {
 				reader.cancel().catch(() => {});
 			});
 
+			// Not raced against teardown, which would drop a frame whose read had already resolved.
 			inner.spawn(async () => {
 				for (;;) {
-					const next = await inner.race(reader.read());
-					if (!next?.value) break;
+					const { value: frame } = await reader.read();
+					if (!frame) break;
 
 					// Ours now, so close it once the encoder has taken what it needs.
-					const frame = next.value;
 					try {
+						if (inner.abort.aborted) break;
 						if (encoder.state !== "configured") continue;
 
 						// Mirror Encoder.serve: default to a 2s GOP unless the rendition overrides it.

js/publish/src/ui/components/stats-tab.ts

@@ -114,13 +114,14 @@ export function statsTab(parent: Effect, publish: MoqPublish): HTMLElement {
 			reader.cancel().catch(() => {});
 		});
 
+		// Not raced against teardown, which would drop a frame whose read had already resolved.
 		effect.spawn(async () => {
 			for (;;) {
-				const next = await effect.race(reader.read());
-				if (!next?.value) break;
+				const { value: frame } = await reader.read();
+				if (!frame) break;
 
 				frames++;
-				next.value.close();
+				frame.close();
 			}
 		});
 	});

js/publish/src/video/capture.test.ts

@@ -8,6 +8,7 @@ import type { StreamTrack } from "./types";
 mock.module("./capture-worker.ts?worklet", () => ({ default: async () => "blob:fake-worker" }));
 const { Capture } = await import("./capture");
 const { Encoder } = await import("./encoder");
+const { Renderer } = await import("../preview");
 
 class Frame {
 	// Every frame built, so a test can assert none is left open.
@@ -185,8 +186,9 @@ function frameSource() {
 	};
 }
 
-// Let the capture pump drain everything pushed so far.
-const flush = () => new Promise((resolve) => setTimeout(resolve, 10));
+// Let the capture pump drain everything pushed so far. The pipeline runs on promises alone, so one
+// macrotask turn settles every microtask queued before it, with no wall-clock wait.
+const flush = () => new Promise((resolve) => setImmediate(resolve));
 
 const open = () => Frame.all.filter((frame) => !frame.closed);
 
@@ -462,3 +464,101 @@ test("an encoder resuming on a still source encodes a keyframe of the current pi
 		else Reflect.deleteProperty(globalThis, "VideoEncoder");
 	}
 });
+
+// Detaches a reader at every microtask offset after a push, since a demand gap or a hidden preview can
+// land anywhere while a frame crosses the capture. A frame already read is the reader's to close, even
+// when the teardown wins, so only the capture's held picture stays open between readers.
+async function sweep(input: ReturnType<typeof frameSource>, attach: (attached: boolean) => void): Promise<void> {
+	let timestamp = 2_000_000;
+	for (let yields = 0; yields < 40; yields++) {
+		attach(true);
+		await flush();
+
+		timestamp += 100_000;
+		input.push(timestamp);
+		for (let i = 0; i < yields; i++) await Promise.resolve();
+		attach(false);
+		await flush();
+
+		expect({ yields, open: open().length }).toEqual({ yields, open: 1 });
+	}
+}
+
+test("a preview hidden while a frame is in flight closes that frame", async () => {
+	using input = frameSource();
+	const capture = new Capture({ source: input.source });
+	const enabled = new Signal(false);
+	const preview = new Renderer({ frames: capture.out.frames, enabled });
+
+	try {
+		input.push(1_000_000);
+		await flush();
+		await sweep(input, (attached) => enabled.set(attached));
+	} finally {
+		preview.close();
+		capture.close();
+	}
+	expect(open()).toEqual([]);
+});
+
+test("an encoder torn down while a frame is in flight closes that frame", async () => {
+	using input = frameSource();
+
+	class NullVideoEncoder {
+		state: CodecState = "unconfigured";
+		#output: VideoEncoderInit["output"];
+		#codec?: string;
+
+		constructor(init: VideoEncoderInit) {
+			this.#output = init.output;
+		}
+
+		static async isConfigSupported(config: VideoEncoderConfig): Promise<{ supported: boolean }> {
+			return { supported: config.codec.startsWith("avc1") };
+		}
+
+		configure(config: VideoEncoderConfig): void {
+			this.state = "configured";
+			this.#codec = config.codec;
+		}
+
+		encode(): void {}
+
+		// Only the probe flushes; report the configured codec, as Chrome does.
+		async flush(): Promise<void> {
+			const chunk = { type: "key", timestamp: 0, byteLength: 1, copyTo: () => {} };
+			this.#output(chunk as never, { decoderConfig: { codec: this.#codec } } as never);
+		}
+
+		close(): void {
+			this.state = "closed";
+		}
+	}
+	const original = Object.getOwnPropertyDescriptor(globalThis, "VideoEncoder");
+	Object.defineProperty(globalThis, "VideoEncoder", { configurable: true, value: NullVideoEncoder, writable: true });
+
+	const capture = new Capture({ source: input.source });
+	const track = new Moq.Track.Producer("video").accept({ timescale: Moq.Time.Timescale.MILLI });
+	const live = new Signal<Moq.Track.Producer | undefined>(undefined);
+	const rendition = { config: new Signal(undefined), track: live, close: () => {} };
+	const encoder = new Encoder("video", {
+		enabled: true,
+		broadcast: { video: () => rendition, baseline: new Baseline() } as never,
+		capture,
+	});
+
+	try {
+		input.push(1_000_000);
+		for (let i = 0; i < 50 && !encoder.out.resolved.peek(); i++) await flush();
+		expect(encoder.out.resolved.peek()).toBeDefined();
+
+		await sweep(input, (attached) => live.set(attached ? track : undefined));
+	} finally {
+		encoder.close();
+		capture.close();
+		track.close();
+		if (original) Object.defineProperty(globalThis, "VideoEncoder", original);
+		else Reflect.deleteProperty(globalThis, "VideoEncoder");
+	}
+	expect(open()).toEqual([]);
+});

js/publish/src/video/capture.ts

@@ -189,6 +189,7 @@ class Frames extends Fanout<VideoFrame> {
 		effect.cleanup(release);
 
 		const reader = live.getReader();
+		let cancelled = false;
 		return new ReadableStream<VideoFrame>(
 			{
 				pull: async (controller) => {
@@ -203,7 +204,13 @@ class Frames extends Fanout<VideoFrame> {
 					for (;;) {
 						const { value } = await reader.read();
 						if (!value) {
-							controller.close();
+							if (!cancelled) controller.close();
+							return;
+						}
+
+						// Cancelled while this read was resolving, so the stream can no longer take it.
+						if (cancelled) {
+							value.close();
 							return;
 						}
 
@@ -217,6 +224,7 @@ class Frames extends Fanout<VideoFrame> {
 					}
 				},
 				cancel: async (reason) => {
+					cancelled = true;
 					release();
 					await reader.cancel(reason);
 				},

js/publish/src/video/encoder.ts

@@ -426,20 +426,21 @@ export class Encoder {
 				if (!fanout) return;
 
 				// Our own stream off the shared capture, so the preview or another rendition reading
-				// slowly can't take frames from this one.
+				// slowly can't take frames from this one. Teardown ends a pending read by cancelling it;
+				// racing the read against teardown would drop a frame that had already resolved, unclosed.
 				const reader = fanout.subscribe(effect).getReader();
 				effect.cleanup(() => {
 					reader.cancel().catch(() => {});
 				});
 
 				effect.spawn(async () => {
 					for (;;) {
-						const next = await effect.race(reader.read());
-						if (!next?.value) break;
+						const { value: frame } = await reader.read();
+						if (!frame) break;
 
 						// Ours now: every path below has to close it.
-						const frame = next.value;
 						try {
+							if (effect.abort.aborted) break;
 							if (encoder.state !== "configured") continue;
 
 							// This doesn't need to be reactive.