fix(watch): keep the audio decoder across a numeric delay change

js/watch/src/audio/decoder.test.ts

@@ -0,0 +1,243 @@
+import { afterEach, beforeEach, describe, expect, it, mock, spyOn } from "bun:test";
+import * as Catalog from "@moq/hang/catalog";
+import * as Container from "@moq/hang/container";
+import * as Moq from "@moq/net";
+import { Time } from "@moq/net";
+import { Signal } from "@moq/signals";
+import type { Broadcast } from "../broadcast";
+import { type Delay, Sync } from "../sync";
+import { SharedRingBuffer } from "./shared-ring-buffer";
+import { Source } from "./source";
+
+// Bun cannot load the blob-URL worklet import.
+mock.module("./render-worklet.ts?worklet", () => ({ default: async () => "blob:fake-render" }));
+const { Decoder } = await import("./decoder");
+
+// Drain reactive work without advancing playback time.
+async function microtasks() {
+	for (let i = 0; i < 100; i++) await Promise.resolve();
+}
+
+class FakeContext extends EventTarget {
+	readonly state = "running";
+	readonly sampleRate: number;
+	readonly audioWorklet = { addModule: async () => {} };
+	constructor(options: AudioContextOptions) {
+		super();
+		this.sampleRate = options.sampleRate ?? 48_000;
+	}
+	resume = async () => {};
+	close = async () => {};
+}
+
+class FakeWorklet {
+	readonly port = Object.assign(new EventTarget(), { postMessage() {}, start() {} });
+	disconnect() {}
+}
+
+class FakeData {
+	readonly sampleRate = 48_000;
+	readonly numberOfChannels = 2;
+	readonly numberOfFrames = 960;
+	readonly timestamp: number;
+	constructor(timestamp: number) {
+		this.timestamp = timestamp;
+	}
+	copyTo() {}
+	close() {}
+}
+
+type Read = NonNullable<Awaited<ReturnType<Container.Consumer["next"]>>>;
+
+let frameTimestamp = 0;
+function frame(): Read {
+	frameTimestamp += 20_000;
+	return {
+		group: 0,
+		discontinuity: 0,
+		continuous: true,
+		frame: { timestamp: Time.Micro(frameTimestamp), payload: new Uint8Array([1]), keyframe: true },
+	};
+}
+
+const globals = ["AudioContext", "AudioWorkletNode", "AudioDecoder", "AudioEncoder", "EncodedAudioChunk"] as const;
+const originals = new Map<string, PropertyDescriptor | undefined>();
+let codecs = 0;
+
+beforeEach(() => {
+	frameTimestamp = 0;
+	codecs = 0;
+	for (const name of globals) originals.set(name, Object.getOwnPropertyDescriptor(globalThis, name));
+
+	class Codec {
+		state = "configured";
+		readonly init: AudioDecoderInit;
+		constructor(init: AudioDecoderInit) {
+			this.init = init;
+			codecs++;
+		}
+		configure() {}
+		reset() {}
+		decode(chunk: { timestamp: number }) {
+			this.init.output(new FakeData(chunk.timestamp) as unknown as AudioData);
+		}
+		close() {
+			this.state = "closed";
+		}
+	}
+	const chunk = class {
+		readonly timestamp: number;
+		constructor(init: EncodedAudioChunkInit) {
+			this.timestamp = init.timestamp;
+		}
+	};
+
+	const values = {
+		AudioContext: FakeContext,
+		AudioWorkletNode: FakeWorklet,
+		AudioDecoder: Codec,
+		AudioEncoder: class {},
+		EncodedAudioChunk: chunk,
+	};
+	for (const name of globals) Object.defineProperty(globalThis, name, { configurable: true, value: values[name] });
+});
+
+afterEach(() => {
+	for (const name of globals) {
+		const original = originals.get(name);
+		if (original) Object.defineProperty(globalThis, name, original);
+		else Reflect.deleteProperty(globalThis, name);
+	}
+	mock.restore();
+});
+
+// Hands frames to whichever container consumer is reading, and ends a consumer's reads on close.
+function feed() {
+	const closed = new WeakSet<Container.Consumer>();
+	const queue: Read[] = [];
+	let waiter: { consumer: Container.Consumer; resolve: (read: Read | undefined) => void } | undefined;
+
+	spyOn(Container.Consumer.prototype, "next").mockImplementation(function (this: Container.Consumer) {
+		if (closed.has(this)) return Promise.resolve(undefined);
+		const read = queue.shift();
+		if (read) return Promise.resolve(read);
+		return new Promise<Read | undefined>((resolve) => {
+			waiter = { consumer: this, resolve };
+		});
+	});
+
+	const close = Container.Consumer.prototype.close;
+	spyOn(Container.Consumer.prototype, "close").mockImplementation(function (this: Container.Consumer) {
+		closed.add(this);
+		if (waiter?.consumer === this) {
+			waiter.resolve(undefined);
+			waiter = undefined;
+		}
+		close.call(this);
+	});
+
+	return (read: Read) => {
+		const current = waiter;
+		waiter = undefined;
+		if (current) current.resolve(read);
+		else queue.push(read);
+	};
+}
+
+async function play(initial: Delay) {
+	const push = feed();
+	const truncate = spyOn(SharedRingBuffer.prototype, "truncate");
+	const reset = spyOn(SharedRingBuffer.prototype, "reset");
+
+	const producer = new Moq.Broadcast.Producer();
+	const consumer = producer.consume();
+	const relativeBroadcast = mock(() => consumer);
+	const audio = Catalog.AudioConfigSchema.parse({
+		codec: "opus",
+		container: { kind: "legacy" },
+		sampleRate: 48_000,
+		numberOfChannels: 2,
+	});
+	const catalog = new Signal<Catalog.Root>({ audio: { renditions: { audio } } });
+	const broadcast = new Signal<Broadcast | undefined>({
+		out: { catalog },
+		relativeBroadcast,
+	} as unknown as Broadcast);
+
+	const delay = new Signal<Delay>(initial);
+	const source = new Source({ broadcast, supported: async () => true });
+	const sync = new Sync({ delay });
+	const decoder = new Decoder({ source, sync });
+	await microtasks();
+
+	// Enough frames to get past the legacy decoder's warm-up, so a handover would truncate.
+	const play = async () => {
+		for (let i = 0; i < 6; i++) {
+			push(frame());
+			await microtasks();
+		}
+	};
+
+	return {
+		delay,
+		play,
+		// Each subscription resolves the rendition's broadcast once.
+		subscriptions: () => relativeBroadcast.mock.calls.length,
+		codecs: () => codecs,
+		truncates: () => truncate.mock.calls.length,
+		resets: () => reset.mock.calls.length,
+		close() {
+			decoder.close();
+			sync.close();
+			source.close();
+			consumer.close();
+			producer.close();
+		},
+	};
+}
+
+describe("Decoder across a delay change", () => {
+	for (const initial of [Time.Milli(100), "auto"] as const) {
+		it(`keeps the subscription and ring when ${initial} becomes a number`, async () => {
+			const playback = await play(initial);
+			try {
+				await playback.play();
+				expect(playback.subscriptions()).toBe(1);
+				expect(playback.codecs()).toBe(1);
+
+				playback.delay.set(Time.Milli(120));
+				await microtasks();
+				await playback.play();
+
+				expect(playback.subscriptions()).toBe(1);
+				expect(playback.codecs()).toBe(1);
+				expect(playback.truncates()).toBe(0);
+				expect(playback.resets()).toBe(0);
+			} finally {
+				playback.close();
+			}
+		});
+	}
+
+	it("rebuilds on a switch to and from instant", async () => {
+		const playback = await play(Time.Milli(100));
+		try {
+			await playback.play();
+
+			playback.delay.set("instant");
+			await microtasks();
+			expect(playback.resets()).toBe(1);
+
+			playback.delay.set(Time.Milli(100));
+			await microtasks();
+			await playback.play();
+
+			expect(playback.subscriptions()).toBe(2);
+			expect(playback.codecs()).toBe(2);
+			// The replacement subscription drops whatever the ring still held from the first.
+			expect(playback.truncates()).toBe(1);
+		} finally {
+			playback.close();
+		}
+	});
+});

js/watch/src/audio/decoder.ts

@@ -221,7 +221,7 @@ export class Decoder {
 	#runEnabled(effect: Effect): void {
 		const enabled = effect.get(this.in.enabled);
 		if (!enabled) return;
-		if (effect.get(this.sync.in.delay) === "instant") {
+		if (effect.get(this.sync.out.instant)) {
 			this.reset();
 			return;
 		}
@@ -253,7 +253,7 @@ export class Decoder {
 	#runDecoder(effect: Effect): void {
 		const enabled = effect.get(this.in.enabled);
 		if (!enabled) return;
-		if (effect.get(this.sync.in.delay) === "instant") return;
+		if (effect.get(this.sync.out.instant)) return;
 
 		const broadcast = effect.get(this.source.in.broadcast);
 		if (!broadcast) return;

js/watch/src/sync.test.ts

@@ -60,6 +60,7 @@ describe("delay and buffer", () => {
 	it("holds nothing when instant, whatever the buffer says", async () => {
 		const sync = new Sync({ delay: "instant", buffer: 30_000 as Time.Milli });
 		await flush();
+		expect(sync.out.instant.peek()).toBe(true);
 		expect(sync.out.buffered.peek()).toBe(false);
 		expect(sync.out.delay.peek()).toBe(0 as Time.Milli);
 		expect(sync.out.maxAge.peek()).toBe(0 as Time.Milli);
@@ -118,6 +119,12 @@ describe("delay and buffer", () => {
 });
 
 describe("wait", () => {
+	it("returns at once when constructed instant, before effects flush", async () => {
+		const sync = new Sync({ delay: "instant" });
+		await sync.wait(Time.Milli.zero);
+		sync.close();
+	});
+
 	it("wakes a sleeping wait when the delay switches to instant", async () => {
 		const delay = new Signal<Delay>(10_000 as Time.Milli);
 		const sync = new Sync({ delay });

js/watch/src/sync.ts

@@ -19,8 +19,7 @@ import {
  * `"instant"` drops the buffer and the pacing together: nothing is held, and {@link Sync.wait}
  * returns without sleeping, so a frame presents as soon as it exists. It also overrides
  * {@link SyncInput.buffer}, since holding a lookahead would contradict holding nothing. Audio is
- * the owner's business: a ring with no depth underruns, so whoever wants this mode is expected to
- * turn audio off.
+ * off: a ring with no depth underruns, so the audio decoder unsubscribes and flushes its ring.
  */
 export type Delay = "instant" | "auto" | Time.Milli;
 
@@ -49,6 +48,10 @@ export type SyncInput = {
 };
 
 type SyncOutput = {
+	// Derived: whether the delay is "instant". Numeric and "auto" delays share `false`, so readers
+	// that only care about the mode don't rerun when the number changes.
+	instant: Signal<boolean>;
+
 	// The earliest time we've received a frame, relative to its timestamp.
 	// This will keep being updated as we catch up to the live playhead then will be relatively static.
 	reference: Signal<Time.Milli | undefined>;
@@ -78,6 +81,7 @@ export class Sync {
 	readonly in: Readonlys<SyncInput>;
 
 	readonly #out: SyncOutput = {
+		instant: new Signal<boolean>(false),
 		reference: new Signal<Time.Milli | undefined>(undefined),
 		delay: new Signal<Time.Milli>(Time.Milli.zero),
 		jitter: new Signal<Time.Milli>(FALLBACK_JITTER),
@@ -104,6 +108,7 @@ export class Sync {
 			probe: getter(props?.probe),
 		};
 
+		this.#signals.run(this.#runInstant.bind(this));
 		this.#signals.run(this.#runJitter.bind(this));
 		this.#signals.run(this.#runDelay.bind(this));
 		this.#signals.run(this.#runMaxAge.bind(this));
@@ -116,11 +121,15 @@ export class Sync {
 		return () => this.#media.update((media) => media.filter((candidate) => candidate !== registered));
 	}
 
+	#runInstant(effect: Effect): void {
+		this.#out.instant.set(effect.get(this.in.delay) === "instant");
+	}
+
 	// Derive `buffered` / `maxAge` from the resolved delay and the configured lookahead.
 	#runMaxAge(effect: Effect): void {
 		const delay = effect.get(this.#out.delay);
 		// "instant" holds nothing, so a configured lookahead doesn't apply.
-		const buffer = effect.get(this.in.delay) === "instant" ? Time.Milli.zero : effect.get(this.in.buffer);
+		const buffer = effect.get(this.#out.instant) ? Time.Milli.zero : effect.get(this.in.buffer);
 
 		this.#out.buffered.set(buffer > 0);
 		this.#out.maxAge.set(Time.Milli.add(delay, buffer));
@@ -169,7 +178,7 @@ export class Sync {
 
 		// A zero delay still holds the rendition's own delay, which is a frame interval at 60fps.
 		// "instant" holds nothing at all.
-		const instant = effect.get(this.in.delay) === "instant";
+		const instant = effect.get(this.#out.instant);
 		const delay = instant ? Time.Milli.zero : Time.Milli.add(media, jitter);
 		this.#out.delay.set(delay);
 	}
@@ -243,6 +252,7 @@ export class Sync {
 	async wait(timestamp: Time.Milli): Promise<void> {
 		// A zero delay still sleeps: the sleep comes from the reference, which holds an early frame
 		// until its timestamp comes up. "instant" is the only thing that skips the wait itself.
+		// Peek the input, not `out.instant`, which lags a same-turn change until effects flush.
 		if (this.in.delay.peek() === "instant") return;
 
 		const reference = this.#out.reference.peek();

quest/m0/README.md

@@ -56,7 +56,6 @@ a published `@moq/watch` break.
 - [Broadcast epochs](/quest/m0/broadcast-epoch/README.md) - every first-party publisher that can restart mints a fresh `@<uuidv7>` epoch, viewers follow the newest live one, and bare names still resolve on every version
 - [Audio jitter target](/quest/m0/audio-jitter-target/README.md) - the audio playout target is a measured estimate of arrival timing in both languages, not a round-trip guess
 - [Opus DTX](/quest/m0/opus-dtx.md) - voice audio publishes without DTX, so its timeline follows the capture clock through silence
-- [Delay rebuild](/quest/m0/watch-delay-rebuild.md) - a numeric audio delay change keeps the decoder and its ring
 
 ## Related
 

quest/m0/audio-jitter-target/watch.md

@@ -104,8 +104,3 @@ quest lands on `main`, so it adds the spread inputs beside `probe` and stops
 reading `probe`; removing it is part of the `SyncInput` reshape in
 [Plan: A/V clock](/quest/m1/av-clock.md). Land the estimator so
 that quest can adopt it without a second estimator change.
-
-
-## Related
-
-- [Delay rebuild](/quest/m0/watch-delay-rebuild.md) - lands first, so preset steps stop rebuilding the decoder