Add media source labels to sync latency tracking (#1222)

Co-authored-by: Claude <noreply@anthropic.com>

Author: kixelated

js/watch/src/audio/mse.ts

@@ -118,7 +118,7 @@ export class Mse implements Backend {
 
 				// Extract the timestamp from the CMAF segment and mark when we received it.
 				const timestamp = Container.Cmaf.decodeTimestamp(frame, timescale);
-				this.source.sync.received(Moq.Time.Milli.fromMicro(timestamp));
+				this.source.sync.received(Moq.Time.Milli.fromMicro(timestamp), "audio");
 
 				await this.#appendBuffer(sourceBuffer, frame);
 
@@ -162,7 +162,8 @@ export class Mse implements Backend {
 				pending = next.frame;
 
 				// Mark that we received this frame for latency calculation.
-				this.source.sync.received(Moq.Time.Milli.fromMicro(pending.timestamp as Moq.Time.Micro));
+				const timestamp = Moq.Time.Milli.fromMicro(pending.timestamp as Moq.Time.Micro);
+				this.source.sync.received(timestamp, "audio");
 
 				break;
 			}
@@ -178,7 +179,8 @@ export class Mse implements Backend {
 					duration = Moq.Time.Micro.sub(frame.timestamp, pending.timestamp);
 
 					// Mark that we received this frame for latency calculation.
-					this.source.sync.received(Moq.Time.Milli.fromMicro(frame.timestamp as Moq.Time.Micro));
+					const timestamp = Moq.Time.Milli.fromMicro(frame.timestamp as Moq.Time.Micro);
+					this.source.sync.received(timestamp, "audio");
 				}
 
 				// Wrap raw frame in moof+mdat

js/watch/src/sync.ts

@@ -29,9 +29,8 @@ export class Sync {
 	// There's probably a way to use Effect, but lets keep it simple for now.
 	#update: PromiseWithResolvers<void>;
 
-	// Batched late-frame tracking: accumulate count and max lateness, log on recovery.
-	#lateCount = 0;
-	#lateMaxMs = 0;
+	// Per-label late-frame tracking: accumulate count and max lateness, flush on recovery.
+	#late = new Map<string, { count: number; maxMs: number }>();
 
 	signals = new Effect();
 
@@ -58,7 +57,7 @@ export class Sync {
 	}
 
 	// Update the reference if this is the earliest frame we've seen, relative to its timestamp.
-	received(timestamp: Time.Milli): void {
+	received(timestamp: Time.Milli, label = ""): void {
 		const now = Time.Milli.now();
 		const ref = Time.Milli.sub(now, timestamp);
 		const currentRef = this.#reference.peek();
@@ -69,12 +68,21 @@ export class Sync {
 			// Otherwise, chained `wait()` calls would cause a false-positive during CPU starvation.
 			const sleep = Time.Milli.add(Time.Milli.sub(currentRef, ref), this.#latency.peek());
 			if (sleep < 0) {
-				this.#lateCount++;
-				this.#lateMaxMs = Math.max(this.#lateMaxMs, -sleep);
-			} else if (this.#lateCount > 0) {
-				console.warn(`sync: ${this.#lateCount} late frame(s), max ${Math.round(this.#lateMaxMs)}ms behind`);
-				this.#lateCount = 0;
-				this.#lateMaxMs = 0;
+				const entry = this.#late.get(label);
+				if (entry) {
+					entry.count++;
+					entry.maxMs = Math.max(entry.maxMs, -sleep);
+				} else {
+					this.#late.set(label, { count: 1, maxMs: -sleep });
+				}
+			} else {
+				const entry = this.#late.get(label);
+				if (entry) {
+					const prefix = label ? `sync[${label}]` : "sync";
+					const behind = Sync.#formatDuration(entry.maxMs);
+					console.warn(`${prefix}: ${entry.count} late frame(s), max ${behind} behind`);
+					this.#late.delete(label);
+				}
 			}
 
 			if (ref >= currentRef) {
@@ -114,6 +122,15 @@ export class Sync {
 		}
 	}
 
+	static #formatDuration(ms: number): string {
+		ms = Math.round(ms);
+		if (ms < 1000) return `${ms}ms`;
+		const s = ms / 1000;
+		if (s < 60) return `${Math.round(s * 10) / 10}s`;
+		const m = s / 60;
+		return `${Math.round(m * 10) / 10}m`;
+	}
+
 	close() {
 		this.signals.close();
 	}

js/watch/src/video/decoder.ts

@@ -320,7 +320,8 @@ class DecoderTrack {
 				}
 
 				// Mark that we received this frame right now.
-				this.source.sync.received(Time.Milli.fromMicro(frame.timestamp as Time.Micro));
+				const timestamp = Time.Milli.fromMicro(frame.timestamp as Time.Micro);
+				this.source.sync.received(timestamp, "video");
 
 				const chunk = new EncodedVideoChunk({
 					type: frame.keyframe ? "key" : "delta",
@@ -398,7 +399,8 @@ class DecoderTrack {
 								});
 
 								// Mark that we received this frame right now.
-								this.source.sync.received(Time.Milli.fromMicro(sample.timestamp as Time.Micro));
+								const timestamp = Time.Milli.fromMicro(sample.timestamp as Time.Micro);
+								this.source.sync.received(timestamp, "video");
 
 								// Track stats
 								this.stats.update((current) => ({

js/watch/src/video/mse.ts

@@ -122,7 +122,7 @@ export class Mse implements Backend {
 
 				// Extract the timestamp from the CMAF segment and mark when we received it.
 				const timestamp = Container.Cmaf.decodeTimestamp(frame, timescale);
-				this.source.sync.received(Moq.Time.Milli.fromMicro(timestamp));
+				this.source.sync.received(Moq.Time.Milli.fromMicro(timestamp), "video");
 
 				await this.#appendBuffer(sourceBuffer, frame);
 
@@ -170,7 +170,8 @@ export class Mse implements Backend {
 				pending = next.frame;
 
 				// Mark that we received this frame right now.
-				this.source.sync.received(Moq.Time.Milli.fromMicro(pending.timestamp as Moq.Time.Micro));
+				const timestamp = Moq.Time.Milli.fromMicro(pending.timestamp as Moq.Time.Micro);
+				this.source.sync.received(timestamp, "video");
 
 				break;
 			}
@@ -185,7 +186,8 @@ export class Mse implements Backend {
 					duration = Moq.Time.Micro.sub(frame.timestamp, pending.timestamp);
 
 					// Mark that we received this frame right now for latency calculation.
-					this.source.sync.received(Moq.Time.Milli.fromMicro(frame.timestamp as Moq.Time.Micro));
+					const timestamp = Moq.Time.Milli.fromMicro(frame.timestamp as Moq.Time.Micro);
+					this.source.sync.received(timestamp, "video");
 				}
 
 				// Wrap raw frame in moof+mdat