feat(watch): per-track timelines on the shared Sync clock

doc/concept/audio-jitter.md

@@ -383,10 +383,10 @@ them. The cost is over-buffering by the smallest subscribed `delay` when the
 broadcast's earliest rendition is not subscribed.
 
 The playout is recomputed when a rendition is subscribed, unsubscribed, or its
-catalog entry rises, so dropping a slow rendition lowers latency. The shared
-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.
+catalog entry rises, so dropping a slow rendition lowers latency. Each track's
+own arrivals only ever move its part of the shared reference earlier. A track
+that leaves, or whose timeline restarts, releases its part: the earliest
+remaining track's anchor takes over, or the next frame re-anchors.
 
 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

doc/lib/js/watch.md

@@ -49,6 +49,14 @@ suspends the element until corrected or removed. The
 The overlay adds play/pause, volume, fullscreen, a quality selector, a
 buffering indicator, and a stats panel.
 
+Audio, video, and captions play from one clock on local time, anchored at the
+earliest arrival (with `buffer`, at the earliest retained frame once startup
+fills). A congested track falls behind on its own instead of stalling the
+others; in `auto` delay, its rising playout target still raises the delay
+every track plays at. A discontinuity restarts only its own track's timeline,
+and a track that is disabled, deselected, or fails releases its hold on the
+clock.
+
 `el.audio.out.error` and `el.video.out.error` report a fatal decode error.
 The failing track closes its subscription and stops decoding. Select another
 source or rendition to clear the failure; the player does not resubscribe to

js/msf/src/catalog.test.ts

@@ -239,6 +239,21 @@ test("preserves delay through decode and encode", () => {
 	expect(wireTracks[0].delay).toBe(200);
 });
 
+test("rejects a negative delay", () => {
+	const raw = encodeJson({
+		version: "draft-01",
+		tracks: [{ name: "video0", packaging: "loc", isLive: true, role: "video", codec: "avc1.640028", delay: -1 }],
+	});
+	expect(() => decode(raw)).toThrow();
+	expect(() =>
+		encode({
+			tracks: [
+				{ name: "video0", packaging: "loc", isLive: true, role: "video", codec: "avc1.640028", delay: -1 },
+			],
+		}),
+	).toThrow();
+});
+
 test("carries extension root members through decode and encode", () => {
 	// An extension section (here `mpegts`) rides the catalog root untouched, and a
 	// section this build has never heard of survives the same way.

js/msf/src/catalog.ts

@@ -56,7 +56,8 @@ const trackShape = {
 
 	// Non-standard: how far this rendition trails the broadcast's earliest, in milliseconds.
 	// Mirrors hang `delay`. A player holds `delay + jitter` and does not subtract across renditions.
-	delay: z.optional(z.number()),
+	// A rendition cannot trail the earliest by a negative amount, so that is malformed.
+	delay: z.optional(z.number().check(z.nonnegative())),
 };
 
 /** Zod schema describing a single track entry in an MSF catalog. */
@@ -127,6 +128,9 @@ const WireCatalogSchema = z.object({
 
 /** Serialize a catalog to its JSON wire representation (draft-01). */
 export function encode(catalog: Catalog): Uint8Array {
+	// Refuse what `decode` would reject, so a publisher fails here instead of its subscribers.
+	CatalogSchema.parse(catalog);
+
 	// Hoist inline init payloads into a shared, deduplicated initDataList and
 	// reference each from its track via initRef (the draft-01 wire shape).
 	const initDataList: z.infer<typeof InitDataSchema>[] = [];

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

@@ -1302,7 +1302,7 @@ it("keeps audio's opening buffered while an enabled video track fills", async ()
 	const installed = clock.install();
 	const playback = await play(Milli(100), true);
 	const video = new Signal<{ start: Milli; end: Milli } | null>(null);
-	const remove = playback.sync.register(new Signal(Milli(100)), video);
+	const remove = playback.sync.track({ target: new Signal(Milli(100)), startup: video });
 	try {
 		await playback.play(8);
 		await clock.advance(Nano.zero);
@@ -1317,7 +1317,7 @@ it("keeps audio's opening buffered while an enabled video track fills", async ()
 		expect(ring.read([new Float32Array(960), new Float32Array(960)])).toBe(960);
 		expect(ring.ring?.timestamp).toBe(Micro((opening ?? 0) + 20_000));
 	} finally {
-		remove();
+		remove.close();
 		playback.close();
 		await microtasks();
 		installed.close();
@@ -1329,7 +1329,7 @@ it("starts above the decoded audio floor without expanding PCM storage", async (
 	const installed = clock.install();
 	const playback = await play(Milli(1000), true);
 	const video = new Signal<{ start: Milli; end: Milli } | null>(null);
-	const remove = playback.sync.register(new Signal(Milli(1000)), video);
+	const remove = playback.sync.track({ target: new Signal(Milli(1000)), startup: video });
 	try {
 		await playback.play(80);
 		await clock.advance(Nano.zero);
@@ -1346,7 +1346,7 @@ it("starts above the decoded audio floor without expanding PCM storage", async (
 		expect(ring.read([new Float32Array(960), new Float32Array(960)])).toBe(960);
 		expect(ring.ring?.timestamp).toBe(Micro((opening ?? 0) + 20_000));
 	} finally {
-		remove();
+		remove.close();
 		playback.close();
 		await microtasks();
 		installed.close();
@@ -1397,6 +1397,7 @@ for (const cmaf of [false, true]) {
 		const installed = clock.install();
 		const playback = await play(Milli(100), true, cmaf);
 		const reset = spyOn(playback.sync, "reset");
+		const video = playback.sync.track({ target: new Signal<Milli | undefined>(undefined) });
 		const prime = spyOn(AudioDecoder.prototype, "reset");
 		try {
 			await playback.play(7);
@@ -1408,7 +1409,7 @@ for (const cmaf of [false, true]) {
 			const held = ring.length;
 			const reference = playback.sync.out.reference.peek();
 			const resets = playback.resets();
-			const parked = playback.sync.wait(Milli(1000));
+			const parked = video.wait(Milli(1000));
 			expect(held).toBeGreaterThan(0);
 			playback.push({ kind: "gap", timestamp: Micro(160_000), duration: Micro(20_000) });
 			await microtasks();
@@ -1438,7 +1439,7 @@ it.each([100, 1000])(
 	async (delay) => {
 		const playback = await play(Milli(delay), true);
 		const video = new Signal<{ start: Milli; end: Milli } | null>(null);
-		const remove = playback.sync.register(new Signal(Milli(delay)), video);
+		const remove = playback.sync.track({ target: new Signal(Milli(delay)), startup: video });
 		const disconnect = spyOn(FakeWorklet.prototype, "disconnect");
 		const closeTrack = spyOn(Moq.Track.Subscriber.prototype, "close");
 		try {
@@ -1457,7 +1458,7 @@ it.each([100, 1000])(
 			await microtasks();
 			expect(playback.inserted()).toHaveLength(decoded);
 		} finally {
-			remove();
+			remove.close();
 			playback.close();
 			await microtasks();
 			disconnect.mockRestore();
@@ -1468,7 +1469,10 @@ it.each([100, 1000])(
 
 it("a rejected audio support probe does not leave buffered video waiting on a pending track", async () => {
 	const sync = new Sync({ delay: Milli(100), buffer: Milli(1000) });
-	const remove = sync.register(new Signal(Milli(100)), new Signal({ start: Milli(0), end: Milli(100) }));
+	const remove = sync.track({
+		target: new Signal(Milli(100)),
+		startup: new Signal({ start: Milli(0), end: Milli(100) }),
+	});
 	let reject!: (error: Error) => void;
 	const failure = new Error("audio support probe failed");
 	const logged = spyOn(console, "warn").mockImplementation(() => {});
@@ -1498,7 +1502,7 @@ it("a rejected audio support probe does not leave buffered video waiting on a pe
 	const decoder = new Decoder({ source, sync });
 	try {
 		await microtasks();
-		sync.received(Milli(0));
+		remove.received(Milli(0));
 		await microtasks();
 		expect(sync.playing.peek()).toBe(false);
 		reject(failure);
@@ -1510,7 +1514,7 @@ it("a rejected audio support probe does not leave buffered video waiting on a pe
 	} finally {
 		decoder.close();
 		source.close();
-		remove();
+		remove.close();
 		sync.close();
 		logged.mockRestore();
 	}
@@ -1521,7 +1525,7 @@ it.each([false, true])("releases shared startup across a hole in queued audio (C
 	const installed = clock.install();
 	const playback = await play(Milli(1500), true, cmaf);
 	const video = new Signal<{ start: Milli; end: Milli } | null>({ start: Milli(20), end: Milli(1520) });
-	const remove = playback.sync.register(new Signal(Milli(1500)), video);
+	const remove = playback.sync.track({ target: new Signal(Milli(1500)), startup: video });
 	try {
 		await playback.play(25);
 		// The audio after the hole counts, the hole itself does not.
@@ -1547,7 +1551,7 @@ it.each([false, true])("releases shared startup across a hole in queued audio (C
 		expect(playback.sync.playing.peek()).toBe(true);
 		expect(ring.read([new Float32Array(960), new Float32Array(960)])).toBe(960);
 	} finally {
-		remove();
+		remove.close();
 		playback.close();
 		await microtasks();
 		installed.close();

js/watch/src/audio/decoder.ts

@@ -20,7 +20,7 @@ import { base64ToBytes } from "../base64";
 import { arrivedAt, type Counter, type Key, type Received, Tally } from "../feedback";
 import { type Failure, failureMoved, nextMedia, subscribeMedia } from "../media";
 import { supportCacheKey } from "../support-probe";
-import type { Sync } from "../sync";
+import type { Sync, Track } from "../sync";
 import { AudioBuffer } from "./buffer";
 import {
 	type DecoderConfig,
@@ -127,6 +127,8 @@ export class Decoder {
 	readonly source: Source;
 	/** The clock that schedules playback. */
 	readonly sync: Sync;
+	// Audio's hold on the clock: its target, startup span, and arrivals.
+	readonly #track: Track;
 
 	readonly #out: DecoderOutput = {
 		error: new Signal<Error | undefined>(undefined),
@@ -232,7 +234,8 @@ export class Decoder {
 			const hold = Milli.fromMicro(ringHold(effect.get(this.sync.out.delay)));
 			return { start: span.start, end: Milli(span.end + Math.min(queued, hold)) };
 		});
-		this.#signals.cleanup(this.sync.register(playout, startup));
+		this.#track = this.sync.track({ target: playout, startup });
+		this.#signals.cleanup(() => this.#track.close());
 		this.#identity = this.#signals.computed((effect) => {
 			const config = effect.get(this.source.out.config);
 			return config ? playbackIdentity(config) : undefined;
@@ -614,7 +617,7 @@ export class Decoder {
 
 				// Classify ingress before decoder backpressure changes the delivery clock.
 				const timestamp = Milli.fromMicro(frame.timestamp as Micro);
-				const late = this.sync.received(timestamp, frame.arrival);
+				const late = this.#track.received(timestamp, frame.arrival);
 				this.#arrivals.write({ kind: "frame", timestamp, arrival: frame.arrival, late });
 
 				const duration = packetDuration(config.codec, frame);
@@ -700,7 +703,7 @@ export class Decoder {
 				const { frame } = next;
 
 				const timestamp = Milli.fromMicro(frame.timestamp);
-				const late = this.sync.received(timestamp, frame.arrival);
+				const late = this.#track.received(timestamp, frame.arrival);
 				this.#arrivals.write({ kind: "frame", timestamp, arrival: frame.arrival, late });
 
 				const duration = packetDuration(config.codec, frame);
@@ -841,6 +844,8 @@ export class Decoder {
 	#retire(): void {
 		this.#consumer = undefined;
 		this.#measured.set(undefined);
+		// Its arrivals stop holding the clock's reference.
+		this.#track.reset();
 	}
 
 	// The ring, once the worklet has loaded. Decoding waits for it, so the frames that arrive first

js/watch/src/audio/replay.ts

@@ -150,7 +150,8 @@ export async function* replay(trace: Arrival[], options: Options): AsyncGenerato
 			delay: config.delay !== undefined ? Milli(config.delay) : undefined,
 		}),
 	);
-	signals.cleanup(sync.register(playout));
+	const audio = sync.track({ target: playout });
+	signals.cleanup(() => audio.close());
 	await settle();
 
 	let delay = sync.out.delay.peek();
@@ -190,7 +191,7 @@ export async function* replay(trace: Arrival[], options: Options): AsyncGenerato
 			const { timestamp, arrival } = next.frame;
 			read.set(timestamp);
 			const media = Milli.fromMicro(timestamp);
-			const late = sync.received(media, arrival);
+			const late = audio.received(media, arrival);
 			arrivals.push({ kind: "frame", timestamp: media, arrival, late });
 			const entry = recorded.get(timestamp);
 			if (!entry) throw new Error(`the consumer delivered an unrecorded frame at ${timestamp} us`);

js/watch/src/index.ts

@@ -22,7 +22,7 @@ export { assets } from "./assets";
 export * as Audio from "./audio";
 export * from "./broadcast";
 export * from "./player";
-export * from "./sync";
+export { type Delay, Sync, type SyncInput, type SyncOutput } from "./sync";
 /** Caption and subtitle playback. */
 export * as Text from "./text";
 /** Video rendition selection, decoding, and rendering. */

js/watch/src/msf.test.ts

@@ -50,7 +50,7 @@ test.each(["mp4a.40.2", "flac", "future-audio"])("keeps MSF decoder warmup absen
 	expect(audio?.warmup).toBeUndefined();
 });
 
-test("rounds a fractional MSF delay up, and drops zero", () => {
+test("rounds a fractional MSF delay up, drops zero, and rejects a negative one", () => {
 	const catalog: Msf.Catalog = {
 		tracks: [
 			{
@@ -81,6 +81,14 @@ test("rounds a fractional MSF delay up, and drops zero", () => {
 	expect(hang.video?.renditions.video?.delay).toBe(u53(201));
 	expect(hang.audio?.renditions.audio?.delay).toBe(u53(1));
 	expect(hang.audio?.renditions.early?.delay).toBeUndefined();
+
+	// A rendition cannot trail the earliest by a negative amount, even between -1 and 0.
+	for (const delay of [-1, -0.5]) {
+		const negative: Msf.Catalog = {
+			tracks: [{ name: "audio", packaging: "loc", role: "audio", codec: "opus", delay }],
+		};
+		expect(() => toHang(negative)).toThrow();
+	}
 });
 
 test("rounds a fractional MSF jitter up, keeps zero, and rejects a negative one", () => {

js/watch/src/msf.ts

@@ -49,10 +49,11 @@ function toContainer(track: Msf.Track): ContainerInfo | undefined {
 }
 
 // Hang stores delay as whole milliseconds rounded up, and zero means the rendition is not behind.
-// MSF writes a fractional millisecond, which u53 rejects.
+// MSF writes a fractional millisecond, which u53 rejects. A negative value goes to u53 unrounded, so
+// it is rejected like a negative jitter rather than read as absent.
 function delayMillis(value: number): ReturnType<typeof u53> | undefined {
-	if (value <= 0) return undefined;
-	return u53(Math.ceil(value));
+	if (value === 0) return undefined;
+	return u53(value < 0 ? value : Math.ceil(value));
 }
 
 // Hang stores jitter as whole milliseconds; round MSF's fractional value up. A negative value goes

js/watch/src/retention.test.ts

@@ -59,14 +59,15 @@ async function pendingListeners(run: () => Promise<void>): Promise<number> {
 // piles it up for the life of the player.
 test("waits on a stable clock leave nothing behind", async () => {
 	const sync = new Sync({ delay: Milli(10) });
+	const video = sync.track({ target: new Signal<Milli | undefined>(undefined) });
 	// Let the delay effect run before anchoring.
 	await yieldNow();
-	sync.received(Nano.toMilli(Nano(now())));
+	video.received(Nano.toMilli(Nano(now())));
 
 	const wait = async () => {
 		for (let round = 0; round < 10; round++) {
 			const at = Nano.toMilli(Nano(now()));
-			await Promise.all(Array.from({ length: 100 }, () => sync.wait(at)));
+			await Promise.all(Array.from({ length: 100 }, () => video.wait(at)));
 		}
 	};
 	expect(await pendingReactions(wait)).toBe(0);
@@ -87,6 +88,7 @@ test("a long subscription through the player path leaves nothing behind", async
 	const producer = new Container.Legacy.Producer(track, format);
 
 	const sync = new Sync({ delay: Milli(10) });
+	const audio = sync.track({ target: new Signal<Milli | undefined>(undefined) });
 	const effect = new Effect();
 	const sub = subscribeMedia(effect, {
 		broadcast: broadcast.consume(),
@@ -111,8 +113,8 @@ test("a long subscription through the player path leaves nothing behind", async
 			if (!next) throw new Error("the track ended");
 
 			const timestamp = Milli.fromMicro(next.frame.timestamp);
-			sync.received(timestamp);
-			const presented = effect.race(sync.wait(timestamp));
+			audio.received(timestamp);
+			const presented = effect.race(audio.wait(timestamp));
 			presenting.add(presented);
 			void presented.finally(() => presenting.delete(presented));
 		}