fix(watch): keep media holes on the current timeline

js/watch/src/audio/buffer.ts

@@ -133,7 +133,7 @@ export class Backpressure {
 		this.#release();
 	}
 
-	/** Note that the decoder closed, dropping every frame still decoding. */
+	/** Note that the decoder closed or reset, dropping every frame still decoding. */
 	dropped(): void {
 		this.#decoding = [];
 		this.#release();
@@ -344,7 +344,7 @@ export class AudioBuffer {
 		this.#backpressure.decoded(end, landed);
 	}
 
-	/** Note that the decoder closed, dropping every frame still decoding. */
+	/** Note that the decoder closed or reset, dropping every frame still decoding. */
 	dropped(): void {
 		this.#backpressure.dropped();
 	}

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

@@ -154,12 +154,18 @@ let codecs = 0;
 let codecErrors: AudioDecoderInit["error"][] = [];
 // Thrown by the next `decode`, as WebCodecs does synchronously for a chunk it refuses.
 let decodeThrows: Error | undefined;
+let deferAudioOutput = false;
+let pendingAudio: (() => void)[] = [];
+let audioInputs: number[] = [];
 
 beforeEach(() => {
 	frameTimestamp = 0;
 	codecs = 0;
 	codecErrors = [];
 	decodeThrows = undefined;
+	deferAudioOutput = false;
+	pendingAudio = [];
+	audioInputs = [];
 	contexts = [];
 	worklets = [];
 	moduleLoaded = Promise.resolve();
@@ -179,12 +185,16 @@ beforeEach(() => {
 		}
 		reset() {
 			this.needsKey = true;
+			pendingAudio = [];
 		}
 		decode(chunk: { timestamp: number; type: EncodedAudioChunkType }) {
 			if (decodeThrows) throw decodeThrows;
 			if (this.needsKey && chunk.type !== "key") throw new DOMException("key chunk required", "DataError");
 			this.needsKey = false;
-			this.init.output(new FakeData(chunk.timestamp) as unknown as AudioData);
+			audioInputs.push(chunk.timestamp);
+			const emit = () => this.init.output(new FakeData(chunk.timestamp) as unknown as AudioData);
+			if (deferAudioOutput) pendingAudio.push(emit);
+			else emit();
 		}
 		close() {
 			this.state = "closed";
@@ -380,6 +390,40 @@ async function play(initial: Delay, buffered = false, cmaf = false) {
 	};
 }
 
+for (const cmaf of [false, true]) {
+	it(`a ${cmaf ? "CMAF" : "legacy"} gap releases accounting for codec outputs it discarded`, async () => {
+		const clock = new Manual();
+		const installed = clock.install();
+		deferAudioOutput = true;
+		const playback = await play(Milli(100), true, cmaf);
+		try {
+			for (let n = 0; n < 6; n++) {
+				playback.push({ ...frame(), group: n === 0 ? 0 : 1 });
+				await microtasks();
+			}
+			expect(audioInputs).toHaveLength(6);
+			expect(pendingAudio).toHaveLength(6);
+			expect(playback.inserted()).toHaveLength(0);
+			playback.push({ kind: "gap", timestamp: Micro(140_000), duration: Micro(20_000) });
+			await microtasks();
+			expect(pendingAudio).toHaveLength(0);
+			deferAudioOutput = false;
+			frameTimestamp = 160_000;
+			const resumed = frame();
+			playback.push({ ...resumed, group: 2 });
+			await microtasks();
+			expect(audioInputs).toHaveLength(7);
+			expect(playback.inserted()).toContain(resumed.frame.timestamp);
+			await playback.play(5);
+			expect(playback.sync.playing.peek()).toBe(true);
+		} finally {
+			playback.close();
+			await microtasks();
+			installed.close();
+		}
+	});
+}
+
 it("decodes an audio packet inside a group after a gap", async () => {
 	const playback = await play(Milli(100));
 	try {
@@ -683,20 +727,42 @@ describe("Decoder across a rendition's absence", () => {
 		}
 	});
 
-	it("keeps the graph when only the container changes", async () => {
-		const playback = await play(Milli(100));
-		try {
-			await playback.play();
-			playback.restore({ container: { kind: "loc" } });
-			await microtasks();
-			await playback.play();
-
-			expect(playback.subscriptions()).toBe(2);
-			expect(contexts).toHaveLength(1);
-		} finally {
-			playback.close();
-		}
-	});
+	for (const [name, patch] of Object.entries({
+		container: { container: { kind: "loc" } },
+		codec: { codec: "mp4a.40.2" },
+	})) {
+		it(`keeps compatible PCM and Sync when the ${name} changes on the same instance`, async () => {
+			const clock = new Manual();
+			const installed = clock.install();
+			const playback = await play(Milli(100), true);
+			const reset = spyOn(playback.sync, "reset");
+			try {
+				await playback.play(7);
+				const ring = playback.rings().at(-1);
+				if (!ring) throw new Error("missing audio ring");
+				const held = ring.length;
+				const reference = playback.sync.out.reference.peek();
+				expect(held).toBeGreaterThan(0);
+				playback.restore(patch);
+				await microtasks();
+				expect(playback.subscriptions()).toBe(2);
+				expect(playback.codecs()).toBe(2);
+				expect(contexts).toHaveLength(1);
+				expect(playback.rings()).toEqual([ring]);
+				expect(ring.length).toBe(held);
+				expect(playback.resets()).toBe(0);
+				expect(reset).not.toHaveBeenCalled();
+				expect(playback.sync.out.reference.peek()).toBe(reference);
+				expect(ring.read([new Float32Array(960), new Float32Array(960)])).toBe(960);
+				await playback.play();
+				expect(reset).not.toHaveBeenCalled();
+			} finally {
+				playback.close();
+				await microtasks();
+				installed.close();
+			}
+		});
+	}
 
 	it("rebuilds the graph on a new channel count", async () => {
 		const playback = await play(Milli(100));
@@ -930,6 +996,14 @@ async function livePlayback(burst: number) {
 	return {
 		sync,
 		error: decoder.out.error,
+		arrivals: () => decoder.arrivals(),
+		// Lose one whole group without changing the following media timestamps or arrival cadence.
+		async hole() {
+			if (held.length) throw new Error("a hole needs an empty delivery burst");
+			sequence++;
+			frame++;
+			await clock.advance(FRAME);
+		},
 		counts,
 		ring,
 		get burst() {
@@ -993,6 +1067,36 @@ async function livePlayback(burst: number) {
 	};
 }
 
+it("a missing live group re-primes without clearing decoded PCM or resetting Sync", async () => {
+	const playback = await livePlayback(1);
+	const reset = spyOn(playback.sync, "reset");
+	const prime = spyOn(AudioDecoder.prototype, "reset");
+	const arrivals = playback.arrivals();
+	const received: { kind: string }[] = [];
+	const reading = (async () => {
+		for await (const arrival of arrivals) received.push(arrival);
+	})();
+	try {
+		await playback.run(1000);
+		const ring = playback.ring();
+		expect(ring?.stalled).toBe(false);
+		const buffered = ring?.length;
+		await playback.hole();
+		expect(ring?.length).toBe(buffered);
+		await playback.run(2000);
+		expect(received.some((arrival) => arrival.kind === "gap")).toBe(true);
+		expect(prime).toHaveBeenCalledTimes(1);
+		expect(reset).not.toHaveBeenCalled();
+		expect(playback.ring()).toBe(ring);
+		expect(ring?.stalled).toBe(false);
+		expect(playback.error.peek()).toBeUndefined();
+	} finally {
+		arrivals.close();
+		await reading;
+		await playback.close();
+	}
+});
+
 describe("Decoder across a rename", () => {
 	// A codec change renames the audio track, and the replacement's consumer reads the same path.
 	// Clearing the estimate at teardown and restarting it from the cold-start prior dropped the
@@ -1179,30 +1283,47 @@ it("waits for video support before allowing audio-only startup", async () => {
 	}
 });
 
-it("does not count flushed audio toward a new startup floor after a gap", async () => {
-	const playback = await play(Milli(100), true);
-	try {
-		await playback.play(7);
-		expect(playback.sync.playing.peek()).toBe(true);
-		const openingRing = playback.rings().at(-1);
-		if (!openingRing) throw new Error("missing audio ring");
-		openingRing.read([new Float32Array(960), new Float32Array(960)]);
-		await microtasks();
-		playback.push({ kind: "gap", timestamp: Micro(160_000), duration: Micro(840_000) });
-		await microtasks();
-		frameTimestamp = 1_000_000;
-		await playback.play(1);
-		expect(playback.sync.playing.peek()).toBe(false);
-		const ring = playback.rings().at(-1);
-		if (!ring) throw new Error("missing audio ring");
-		expect(ring.read([new Float32Array(960), new Float32Array(960)])).toBe(0);
-		await playback.play(5);
-		expect(playback.sync.playing.peek()).toBe(true);
-	} finally {
-		playback.close();
-		await microtasks();
-	}
-});
+for (const cmaf of [false, true]) {
+	it(`a ${cmaf ? "CMAF" : "legacy"} hole re-primes the codec while keeping PCM and the clock`, async () => {
+		const clock = new Manual();
+		const installed = clock.install();
+		const playback = await play(Milli(100), true, cmaf);
+		const reset = spyOn(playback.sync, "reset");
+		const prime = spyOn(AudioDecoder.prototype, "reset");
+		try {
+			await playback.play(7);
+			expect(playback.sync.playing.peek()).toBe(true);
+			const ring = playback.rings().at(-1);
+			if (!ring) throw new Error("missing audio ring");
+			ring.read([new Float32Array(960), new Float32Array(960)]);
+			await microtasks();
+			const held = ring.length;
+			const reference = playback.sync.out.reference.peek();
+			const resets = playback.resets();
+			const parked = playback.sync.wait(Milli(1000));
+			expect(held).toBeGreaterThan(0);
+			playback.push({ kind: "gap", timestamp: Micro(160_000), duration: Micro(20_000) });
+			await microtasks();
+			expect(prime).toHaveBeenCalledTimes(1);
+			expect(reset).not.toHaveBeenCalled();
+			expect(playback.resets()).toBe(resets);
+			expect(ring.length).toBe(held);
+			expect(playback.sync.out.reference.peek()).toBe(reference);
+			expect(playback.sync.playing.peek()).toBe(true);
+			expect(Bun.peek.status(parked)).toBe("pending");
+			frameTimestamp = 160_000;
+			await playback.play(1);
+			expect(playback.sync.playing.peek()).toBe(true);
+			expect(ring.read([new Float32Array(960), new Float32Array(960)])).toBe(960);
+			await clock.advance(Nano.fromMilli(Milli(1200)));
+			expect(await parked).toBe(true);
+		} finally {
+			playback.close();
+			await microtasks();
+			installed.close();
+		}
+	});
+}
 
 it.each([100, 1000])(
 	"closing during shared startup at %dms releases the audio graph and backpressured reader",

js/watch/src/audio/decoder.ts

@@ -512,9 +512,7 @@ export class Decoder {
 					if (!(await play(order.flush()))) return;
 					if (!next) break;
 					if (this.#onRead(next)) {
-						decoder.reset();
-						decoder.configure(decoderConfig);
-						order.reset();
+						this.#reprime(decoder, decoderConfig, order);
 					}
 					if (next.kind === "gap") this.#arrivals.write(gap(next));
 					continue;
@@ -601,9 +599,7 @@ export class Decoder {
 					if (!(await play(order.flush()))) return;
 					if (!next) break;
 					if (this.#onRead(next)) {
-						decoder.reset();
-						decoder.configure(decoderConfig);
-						order.reset();
+						this.#reprime(decoder, decoderConfig, order);
 					}
 					if (next.kind === "gap") this.#arrivals.write(gap(next));
 					continue;
@@ -639,16 +635,22 @@ export class Decoder {
 	): Promise<boolean> {
 		for (const read of reads) {
 			if (this.#onRead(read)) {
-				decoder.reset();
-				decoder.configure(config);
-				order.reset();
+				this.#reprime(decoder, config, order);
 			}
 			if (!order.admit(read)) continue;
 			if (!(await this.#decode(effect, decoder, read.frame))) return false;
 		}
 		return true;
 	}
 
+	// A reset discards queued codec output, so release its accounting without clearing retained PCM.
+	#reprime(decoder: AudioDecoder, config: AudioDecoderConfig, order: Order<FrameRead>): void {
+		decoder.reset();
+		this.#ring.peek()?.dropped();
+		decoder.configure(config);
+		order.reset();
+	}
+
 	// Decode one frame once backpressure admits it, reporting whether to keep going.
 	async #decode(effect: Effect, decoder: AudioDecoder, frame: Container.ReceivedFrame): Promise<boolean> {
 		const ring = await this.#ready(effect);
@@ -891,8 +893,11 @@ export class Decoder {
 	// a gap, or a later group after a declared end.
 	#onRead(read: Container.Read): boolean {
 		if (!this.#terminal.update(read)) return false;
-		this.sync.reset();
-		this.reset();
+		// A hole restarts codec prediction on the same timeline, keeping decoded PCM and parked video.
+		if (read.kind === "frame") {
+			this.sync.reset();
+			this.reset();
+		}
 		return true;
 	}
 

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

@@ -100,3 +100,15 @@ test("an endpoint does not trim media resumed in a later group", () => {
 	expect(terminal.end).toBeUndefined();
 	expect(terminal.span({ timestamp: 5_000_000, sampleRate: 48_000, numberOfFrames: 960 }).frames).toBe(960);
 });
+
+test("a hole preserves a declared endpoint until a later group starts a new timeline", () => {
+	const terminal = new Terminal();
+	terminal.update(frame(0, 0));
+	terminal.update(end(1, 20_000));
+	expect(terminal.update(gap)).toBe(true);
+	expect(terminal.end).toBe(20_000 as Micro);
+	expect(terminal.update(frame(1, 10_000))).toBe(false);
+	expect(terminal.span({ timestamp: 10_000, sampleRate: 48_000, numberOfFrames: 960 }).frames).toBe(480);
+	expect(terminal.update(frame(2, 1_000_000))).toBe(true);
+	expect(terminal.end).toBeUndefined();
+});