From 126c6a8eb8b94feabf8741578e9106be41999f42 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 2 Oct 2026 07:32:21 -0700 Subject: [PATCH 1/2] chore: claim media late-join quest From 70321a0d143bae2d64f4c83fbffc4d9125e0bab3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 2 Oct 2026 07:46:35 -0700 Subject: [PATCH 2/2] test(media): measure late join against the published GOP Co-Authored-By: GPT-6 --- test/interop/README.md | 5 ++- test/interop/clients/js/harness.ts | 13 ++++--- test/interop/clients/js/media.ts | 20 +++++------ test/interop/clients/js/src/contract.test.ts | 37 +++++++++++++++++++- test/interop/clients/js/src/contract.ts | 33 +++++++++++++++++ test/interop/clients/js/src/fixture.ts | 18 +++++++--- test/interop/clients/js/src/setup.ts | 4 +++ test/justfile | 2 +- 8 files changed, 108 insertions(+), 24 deletions(-) diff --git a/test/interop/README.md b/test/interop/README.md index 8d862ba42e..35f8a1efa9 100644 --- a/test/interop/README.md +++ b/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. diff --git a/test/interop/clients/js/harness.ts b/test/interop/clients/js/harness.ts index bd59af1602..9b79020537 100644 --- a/test/interop/clients/js/harness.ts +++ b/test/interop/clients/js/harness.ts @@ -256,15 +256,18 @@ export async function readFixtureState(page: Page): Promise { } /** Invoke one of the page's {@link InteropControl} commands. */ -export async function command(page: Page, name: keyof InteropControl): Promise { - await page.evaluate( +export async function command( + page: Page, + name: K, +): Promise>> { + return (await page.evaluate( ([key, fn]) => { - const control = (window as unknown as Record void> | undefined>)[key]; + const control = (window as unknown as Record 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>; } /** Read the page's live resource counts, which outlive the player element. */ diff --git a/test/interop/clients/js/media.ts b/test/interop/clients/js/media.ts index f4e9577d3a..d73259a50e 100644 --- a/test/interop/clients/js/media.ts +++ b/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"); } diff --git a/test/interop/clients/js/src/contract.test.ts b/test/interop/clients/js/src/contract.test.ts index b36189ed58..755ffa9ce3 100644 --- a/test/interop/clients/js/src/contract.test.ts +++ b/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(); + } +}); diff --git a/test/interop/clients/js/src/contract.ts b/test/interop/clients/js/src/contract.ts index 00c8803e94..431c8e44e7 100644 --- a/test/interop/clients/js/src/contract.ts +++ b/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 { + 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; /** Stop the fixture publisher, releasing its session. */ stop(): void; /** Start (or restart) the fixture publisher on the same broadcast path. */ diff --git a/test/interop/clients/js/src/fixture.ts b/test/interop/clients/js/src/fixture.ts index 17f6faef55..a3d3ff1a78 100644 --- a/test/interop/clients/js/src/fixture.ts +++ b/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("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,7 +205,7 @@ 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); @@ -210,6 +213,13 @@ export class Fixture { }); } + /** 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(); diff --git a/test/interop/clients/js/src/setup.ts b/test/interop/clients/js/src/setup.ts index aeb4177abb..1cd5275a10 100644 --- a/test/interop/clients/js/src/setup.ts +++ b/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; diff --git a/test/justfile b/test/justfile index 1be6619e2c..fecc03f330 100644 --- a/test/justfile +++ b/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