test(media): measure late join against the published GOP

test/interop/README.md

@@ -126,7 +126,10 @@ Each run covers, against a real local relay:
   The fake device is not physical hardware, and the headless permission decision
   is not a person clicking a browser prompt.
 - **pause and resume**, **unsubscribe and rejoin**, **detach and reattach**,
-  **publisher stop and same-path republish**, and **late join**.
+  **publisher stop and same-path republish**, and **late join**. The late join
+  must present the newest published GOP or a newer one. Its lower bound is the
+  encoded keyframe timestamp sampled before the existing viewer closes, since
+  painting the canvas does not mean capture and encoding have finished.
 - **resources return to baseline** - the page wraps `WebTransport`, `WebSocket`,
   `AudioContext`, and `Worker` to count live instances, so a detach that leaks a
   session is visible rather than merely invisible.

test/interop/clients/js/harness.ts

@@ -256,15 +256,18 @@ export async function readFixtureState(page: Page): Promise<FixtureState> {
 }
 
 /** Invoke one of the page's {@link InteropControl} commands. */
-export async function command(page: Page, name: keyof InteropControl): Promise<void> {
-	await page.evaluate(
+export async function command<K extends keyof InteropControl>(
+	page: Page,
+	name: K,
+): Promise<Awaited<ReturnType<InteropControl[K]>>> {
+	return (await page.evaluate(
 		([key, fn]) => {
-			const control = (window as unknown as Record<string, Record<string, () => void> | undefined>)[key];
+			const control = (window as unknown as Record<string, Record<string, () => unknown> | undefined>)[key];
 			if (!control?.[fn]) throw new Error(`the page exposes no ${fn} command`);
-			control[fn]();
+			return control[fn]();
 		},
 		[CONTROL, name] as const,
-	);
+	)) as Awaited<ReturnType<InteropControl[K]>>;
 }
 
 /** Read the page's live resource counts, which outlive the player element. */

test/interop/clients/js/media.ts

@@ -47,7 +47,7 @@ import {
 import {
 	type CaptureState,
 	FAULTS,
-	KEYFRAME_INTERVAL_MS,
+	lateJoinStartsLive,
 	leakedPlayerStarted,
 	SAMPLE_MS,
 	SAMPLE_RATE,
@@ -115,9 +115,6 @@ const MIN_RATE = 0.5;
  */
 const MAX_SKEW_STEPS = 1;
 
-/** The first decodable frame may start at the current GOP's keyframe, but never in older history. */
-const MAX_LATE_JOIN_LAG_FRAMES = Math.ceil((Pattern.FPS * KEYFRAME_INTERVAL_MS) / 1000);
-
 const percentile = (values: number[], p: number) => {
 	if (values.length === 0) return Number.NaN;
 	const sorted = [...values].sort((a, b) => a - b);
@@ -660,8 +657,8 @@ try {
 	// at the live edge rather than replay what it missed.
 	if (wants("late-join")) {
 		console.error("=== late join ===");
+		const live = await command(publisher, "liveGop");
 		await player.close();
-		const live = await readFixtureState(publisher);
 		[player, playerErrors] = await subscriber(broadcast, "latecomer");
 		await gesture(player);
 		const joined = await waitForState(player, playerErrors, {
@@ -670,17 +667,16 @@ try {
 			description: "the latecomer to present the fixture",
 			predicate: (state) => state.frameId !== undefined && state.audioContext === "running",
 		});
-		// The fixture sample names the frame painted immediately before the page opens. The first
-		// decodable frame can be the keyframe at the start of the current GOP, so require it to be
-		// within that GOP rather than requiring an impossible zero-frame capture/encode delay.
-		const lag = live.frameId - (joined.frameId ?? 0);
+		// requestFrame() is asynchronous, and capture timestamps decide keyframes. The painted
+		// counter can therefore be over 15 ticks ahead of a still-current encoded GOP under load.
+		// Compare against its actual published keyframe, sampled while the old viewer held demand.
 		check(
-			lag <= MAX_LATE_JOIN_LAG_FRAMES,
+			lateJoinStartsLive(live, joined.videoTimestamp),
 			"late join starts live",
 			() =>
-				`joined at frame ${joined.frameId}, ${lag} frames behind the ${live.frameId} already published when it opened (one GOP is ${MAX_LATE_JOIN_LAG_FRAMES})`,
+				`joined at frame ${joined.frameId}, timestamp ${joined.videoTimestamp}ms before the current GOP's ${live.timestamp}ms keyframe`,
 		);
-		console.error(`  joined at frame ${joined.frameId}, live edge was ${live.frameId}`);
+		console.error(`  joined at frame ${joined.frameId}, timestamp ${joined.videoTimestamp}ms, current GOP began at ${live.timestamp}ms`);
 		assertMedia(await collect(player, playerErrors, WINDOW_MS), "late join");
 	}
 

test/interop/clients/js/src/contract.test.ts

@@ -1,5 +1,6 @@
 import { expect, test } from "bun:test";
-import { leakedPlayerStarted, type Resources } from "./contract";
+import * as Moq from "@moq/net";
+import { lateJoinStartsLive, leakedPlayerStarted, readLiveGop, type Resources } from "./contract";
 
 const playing: Resources = { transports: 1, sockets: 0, audioContexts: 1, workers: 0 };
 
@@ -17,3 +18,37 @@ test("a leaked player is visible when it reuses a pooled websocket fallback", ()
 test("unchanged counts are not a leak start", () => {
 	expect(leakedPlayerStarted(playing, playing)).toBe(false);
 });
+
+
+test("late join uses the published GOP when painting is ahead of capture", async () => {
+	const broadcast = new Moq.Broadcast.Producer();
+	const track = broadcast.createTrack("video");
+	try {
+		const old = track.appendGroup();
+		old.writeFrame({ payload: new Uint8Array([1]), timestamp: Moq.Time.Timestamp.fromMillis(3000) });
+		old.close();
+		const current = track.appendGroup();
+		current.writeFrame({ payload: new Uint8Array([2]), timestamp: Moq.Time.Timestamp.fromMillis(3700) });
+		current.writeFrame({ payload: new Uint8Array([3]), timestamp: Moq.Time.Timestamp.fromMillis(4199) });
+
+		// Painted counter 127 can precede capture while GOP 111 remains current. The old
+		// 127 - 111 <= 15 check rejects a valid join; an earlier published GOP must fail.
+		const viewer = track.subscribe();
+		const gop = await readLiveGop(broadcast.track("video"));
+		expect(gop.timestamp).toBe(3700);
+		expect(lateJoinStartsLive(gop, 3700)).toBe(true);
+		expect(lateJoinStartsLive(gop, 4199)).toBe(true);
+		expect(lateJoinStartsLive(gop, 3000)).toBe(false);
+		expect(lateJoinStartsLive(gop, undefined)).toBe(false);
+		expect(track.used.peek()).toBe(true);
+		const viewed = await viewer.recvGroup();
+		expect((await viewed?.readFrame())?.payload).toEqual(new Uint8Array([2]));
+		expect((await viewed?.readFrame())?.payload).toEqual(new Uint8Array([3]));
+		viewed?.close();
+		viewer.close();
+		await track.unused();
+		expect(track.used.peek()).toBe(false);
+	} finally {
+		broadcast.close();
+	}
+});

test/interop/clients/js/src/contract.ts

@@ -8,6 +8,8 @@
  * @module
  */
 
+import type * as Moq from "@moq/net";
+
 // ── the deterministic publisher ─────────────────────────────────────────────
 
 /** A deliberate defect, used to prove an assertion can fail. */
@@ -43,6 +45,35 @@ export type FixtureState = {
 	encodedFrames: number;
 };
 
+/** The newest published video's GOP, sampled independently of canvas capture and encoding. */
+export type LiveGop = {
+	/** The timestamp of its keyframe, in milliseconds on the publisher's media clock. */
+	timestamp: number;
+};
+
+/** Read the newest published GOP without retaining a subscription or changing its frames. */
+export async function readLiveGop(track: Moq.Track.Consumer): Promise<LiveGop> {
+	const subscriber = track.subscribe();
+	try {
+		const group = await subscriber.recvGroup();
+		if (!group) throw new Error("the fixture video has no published GOP");
+		try {
+			const frame = await group.readFrame();
+			if (!frame) throw new Error("the fixture GOP has no keyframe");
+			return { timestamp: frame.timestamp.asMillis() };
+		} finally {
+			group.close();
+		}
+	} finally {
+		subscriber.close();
+	}
+}
+
+/** True when the presented frame belongs to the sampled GOP or a newer one. */
+export function lateJoinStartsLive(gop: LiveGop, timestamp: number | undefined): boolean {
+	return timestamp !== undefined && timestamp >= gop.timestamp;
+}
+
 /** The camera publisher state mirrored onto its element for Playwright. */
 export type CaptureState = {
 	videoError?: string;
@@ -161,6 +192,8 @@ export type Sample = {
  * owns them rather than the driver reaching in.
  */
 export type InteropControl = {
+	/** Read the current video GOP while an existing viewer is still pulling media. */
+	liveGop(): Promise<LiveGop>;
 	/** Stop the fixture publisher, releasing its session. */
 	stop(): void;
 	/** Start (or restart) the fixture publisher on the same broadcast path. */

test/interop/clients/js/src/fixture.ts

@@ -17,7 +17,7 @@ import { Time } from "@moq/net";
 import * as Publish from "@moq/publish";
 import { Effect, Signal } from "@moq/signals";
 import type { Fault, FixtureState } from "./contract";
-import { KEYFRAME_INTERVAL_MS, OFFSET_STEPS, SAMPLE_RATE } from "./contract";
+import { KEYFRAME_INTERVAL_MS, OFFSET_STEPS, readLiveGop, SAMPLE_RATE } from "./contract";
 import * as Pattern from "./pattern";
 
 /** Cap the encoder rather than letting it track a bandwidth estimate, so runs are comparable. */
@@ -47,6 +47,7 @@ export class Fixture {
 	readonly host: HTMLElement;
 
 	readonly #signals = new Effect();
+	readonly #broadcast: Publish.Broadcast;
 	readonly #audio: AudioContext;
 	readonly #frameId = new Signal(-1);
 	readonly #audioState = new Signal<AudioContextState>("suspended");
@@ -95,6 +96,8 @@ export class Fixture {
 		});
 		this.#signals.cleanup(() => broadcast.close());
 
+		this.#broadcast = broadcast;
+
 		const video = new Publish.Video.Encoder("video", {
 			broadcast,
 			capture,
@@ -183,8 +186,8 @@ export class Fixture {
 		this.#signals.interval(schedule, (SCHEDULE_AHEAD * 1000) / 4);
 	}
 
-	// Paint and emit one frame per pattern tick. The counter comes from the audio clock rather than
-	// the timer, so a late or coalesced tick skips a frame instead of drifting the two apart.
+	// Paint and request capture per pattern tick. Capture and encoding can finish later. The counter
+	// comes from the audio clock, so a coalesced tick skips a frame instead of drifting the two apart.
 	#runPicture(ctx: CanvasRenderingContext2D, track: CanvasSource, start: number, fault: Fault): void {
 		let painted = -1;
 
@@ -202,14 +205,21 @@ export class Fixture {
 		};
 
 		// Twice the frame rate: a tick that lands between frames is a no-op, and one that lands late
-		// still emits the frame the clock is on.
+		// still requests capture of the frame the clock is on.
 		this.#timer = self.setInterval(tick, 1000 / Pattern.FPS / 2);
 		this.#signals.cleanup(() => {
 			if (this.#timer !== undefined) self.clearInterval(this.#timer);
 			this.#timer = undefined;
 		});
 	}
 
+	/** Sample the current published keyframe before the viewer closes its subscription. */
+	async liveGop() {
+		const broadcast = this.#broadcast.net.peek();
+		if (!broadcast) throw new Error("the fixture is not publishing");
+		return readLiveGop(broadcast.track("video"));
+	}
+
 	/** Stop publishing and release the capture, audio graph, and session. */
 	close(): void {
 		this.#signals.close();

test/interop/clients/js/src/setup.ts

@@ -55,6 +55,10 @@ if (role === "publish") {
 	// The driver stops and restarts the publisher in place to exercise a same-path republish, which
 	// has to reuse this page so the audio context keeps its user activation.
 	publish({
+		liveGop: () => {
+			if (!fixture) throw new Error("the fixture is stopped");
+			return fixture.liveGop();
+		},
 		stop: () => {
 			fixture?.close();
 			fixture = undefined;

test/justfile

@@ -24,7 +24,7 @@ harness:
     ./lib/harness.test.sh
     cd .. && bun install --frozen-lockfile
     cd interop/clients/js && bunx playwright install chromium
-    cd interop/clients/js && bun test ./harness.browser.ts
+    cd interop/clients/js && bun test ./harness.browser.ts ./src/contract.test.ts
 
 # Cross-language media interop test, built from this checkout. Stands up a
 # relay and runs the publisher x subscriber matrix. Default: rust only (fast