From 27e715f038d98dd2a67f1a771a53e8df7407e95f Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Thu, 20 Aug 2026 02:35:57 +0800 Subject: [PATCH] test(media): align authoritative route evidence (#55) * test(media): observe authoritative SFU routes * test(media): align authoritative route evidence --- scripts/peer-assisted-benchmark.ts | 280 +++++++++++-- tests/peer-assisted-benchmark.test.ts | 545 +++++++++++++++++++++++++- 2 files changed, 786 insertions(+), 39 deletions(-) diff --git a/scripts/peer-assisted-benchmark.ts b/scripts/peer-assisted-benchmark.ts index 9da1f54..40cde80 100644 --- a/scripts/peer-assisted-benchmark.ts +++ b/scripts/peer-assisted-benchmark.ts @@ -6,6 +6,7 @@ import { basename, dirname, isAbsolute, join, resolve } from "node:path"; import { fileURLToPath, pathToFileURL } from "node:url"; import WebSocket from "ws"; +import type { ParticipantRouteAssignment } from "../src/shared/protocol"; import { createScreenerServer, type ScreenerServer, @@ -80,9 +81,10 @@ interface ConnectionObservation { error?: string; } -interface MediaAssignmentObservation { - parentPeerId: string | null; - childPeerIds: string[]; +interface ActiveRouteReadyObservation { + revision: number; + upstreamKind: ParticipantRouteAssignment["upstream"]["kind"]; + sfuPublicationGeneration: string | null; } interface PageObservation { @@ -95,7 +97,9 @@ interface PageObservation { authenticateSentAtEpochMs: number | null; signalingConnected: boolean; qualitySettings: QualitySettings | null; - assignment: MediaAssignmentObservation; + routeRevision: number | null; + routeAssignment: ParticipantRouteAssignment | null; + activeRouteReady: ActiveRouteReadyObservation[]; maxActiveOutboundMediaEdges: number; maxAssignedChildren: number; firstDecodedAtEpochMs: number | null; @@ -464,6 +468,69 @@ export function activeVideoEdgeCount( }).length; } +function hasAuthoritativeMediaUpstream(page: PageObservation): boolean { + return ( + page.routeRevision !== null && + (page.routeAssignment?.upstream.kind === "peer" || + page.routeAssignment?.upstream.kind === "sfu") + ); +} + +function routeParentPeerId(page: PageObservation): string | null { + return page.routeAssignment?.upstream.kind === "peer" + ? page.routeAssignment.upstream.peerId + : null; +} + +function routeChildPeerIds(page: PageObservation): readonly string[] { + return page.routeAssignment?.childPeerIds ?? []; +} + +function inspectSfuPublication(pages: readonly PageObservation[]) { + const sfuViewers = pages.filter( + (page) => + page.role === "viewer" && page.routeAssignment?.upstream.kind === "sfu", + ); + const publicationOwners = pages.filter( + (page) => + typeof page.routeAssignment?.sfuPublicationGeneration === "string", + ); + if (sfuViewers.length === 0) { + return { + observed: false, + rootCount: 0, + coherent: publicationOwners.length === 0, + }; + } + + if (sfuViewers.length > 2 || publicationOwners.length !== 1) { + return { observed: true, rootCount: sfuViewers.length, coherent: false }; + } + const host = publicationOwners[0]!; + const hostRevision = host.routeRevision; + const hostGeneration = host.routeAssignment?.sfuPublicationGeneration ?? null; + const hasMatchingActiveReady = (page: PageObservation): boolean => + page.activeRouteReady.some( + (ready) => + ready.revision === page.routeRevision && + ready.upstreamKind === page.routeAssignment?.upstream.kind && + ready.sfuPublicationGeneration === + page.routeAssignment?.sfuPublicationGeneration, + ); + const coherent = + host.role === "host" && + hostRevision !== null && + hostGeneration !== null && + pages.every((page) => page.routeRevision === hostRevision) && + hasMatchingActiveReady(host) && + sfuViewers.every( + (viewer) => + viewer.routeAssignment?.sfuPublicationGeneration === null && + hasMatchingActiveReady(viewer), + ); + return { observed: true, rootCount: sfuViewers.length, coherent }; +} + export function summarizeSamples(samples: TimedSample[], viewerCount: number) { let maxHostActiveMediaEdges = 0; let maxHostAssignedChildren = 0; @@ -479,7 +546,7 @@ export function summarizeSamples(samples: TimedSample[], viewerCount: number) { ); maxHostAssignedChildren = Math.max( maxHostAssignedChildren, - host.assignment.childPeerIds.length, + routeChildPeerIds(host).length, host.maxAssignedChildren, ); } @@ -494,6 +561,7 @@ export function summarizeSamples(samples: TimedSample[], viewerCount: number) { } } const finalPages = samples.at(-1)?.pages ?? []; + const sfuPublication = inspectSfuPublication(finalPages); const viewerContinuity = finalPages .filter((page) => page.role === "viewer") .map((page) => { @@ -529,7 +597,9 @@ export function summarizeSamples(samples: TimedSample[], viewerCount: number) { }); return { label: page.label, - assigned: page.assignment.parentPeerId !== null, + assigned: hasAuthoritativeMediaUpstream(page), + routeRevision: page.routeRevision, + upstreamKind: page.routeAssignment?.upstream.kind ?? null, activeUpstream: activeReceive !== undefined, statsClean, framesIncreased: @@ -539,6 +609,7 @@ export function summarizeSamples(samples: TimedSample[], viewerCount: number) { }; }); everyViewerDecoded = + sfuPublication.coherent && viewerContinuity.length === viewerCount && viewerContinuity.every( (viewer) => @@ -576,6 +647,9 @@ export function summarizeSamples(samples: TimedSample[], viewerCount: number) { maxHostAssignedChildren, maxRelayActiveMediaEdges, everyViewerDecoded, + sfuPublicationObserved: sfuPublication.observed, + sfuRootCount: sfuPublication.rootCount, + sfuPublicationCoherent: sfuPublication.coherent, viewerContinuity, firstFrames, maxFirstDecodedAfterAuthenticateMs: @@ -586,8 +660,12 @@ export function summarizeSamples(samples: TimedSample[], viewerCount: number) { viewerIndex: page.viewerIndex, peerId: page.peerId, qualitySettings: page.qualitySettings, - parentPeerId: page.assignment.parentPeerId, - childPeerIds: page.assignment.childPeerIds, + parentPeerId: routeParentPeerId(page), + childPeerIds: routeChildPeerIds(page), + routeRevision: page.routeRevision, + routeUpstream: page.routeAssignment?.upstream ?? null, + sfuPublicationGeneration: + page.routeAssignment?.sfuPublicationGeneration ?? null, })), finalEdges: finalPages.flatMap((page) => page.connections @@ -612,11 +690,25 @@ export function summarizeSamples(samples: TimedSample[], viewerCount: number) { }; } -function buildRunChecks( +export function buildRunChecks( summary: NonNullable, viewerCount: number, profileId: ProfileId, ): RunCheck[] { + const sfuConsistencyCheck: RunCheck = summary.sfuPublicationObserved + ? { + name: "sfu-route-consistency", + passed: summary.sfuPublicationCoherent, + actual: summary.sfuRootCount, + expected: + "1-2 active SFU roots, one Host publication, and one active revision across all participants", + } + : { + name: "no-orphan-sfu-publication", + passed: summary.sfuPublicationCoherent, + actual: summary.sfuPublicationCoherent, + expected: "no SFU publication generation when no SFU root is active", + }; return [ { name: "host-active-media-edges", @@ -642,6 +734,7 @@ function buildRunChecks( actual: summary.everyViewerDecoded, expected: `${viewerCount} viewers with a live upstream, clean stats, and increasing decoded frames`, }, + sfuConsistencyCheck, { name: "first-decoded-frame", passed: @@ -693,7 +786,9 @@ export function buildBenchmarkInitScript(options: { authenticateSentAtEpochMs: null, signalingConnected: false, qualitySettings: null, - assignment: { parentPeerId: null, childPeerIds: [] }, + routeRevision: null, + routeAssignment: null, + activeRouteReady: [], maxActiveOutboundMediaEdges: 0, maxAssignedChildren: 0, firstDecodedAtEpochMs: null, @@ -704,23 +799,103 @@ export function buildBenchmarkInitScript(options: { const descriptions = []; const accumulators = new WeakMap(); const statsModule = import("/src/client/webrtc/stats.ts"); + let signalingSocket = null; + let peerAssisted = false; + let transitionRevision = -1; + let transitionPhase = null; + let plannedRouteAssignment = null; - function cloneAssignment(value) { - if (!value || !Array.isArray(value.childPeerIds)) { - return { parentPeerId: null, childPeerIds: [] }; + function isOpaqueId(value) { + return typeof value === "string" && + value.length >= 8 && + value.length <= 128 && + /^[A-Za-z0-9_-]+$/.test(value); + } + + function cloneRouteAssignment(value) { + if (!value || typeof value !== "object" || Array.isArray(value)) return null; + const upstream = value.upstream; + if (!upstream || typeof upstream !== "object" || Array.isArray(upstream)) return null; + let clonedUpstream; + if (upstream.kind === "peer" && isOpaqueId(upstream.peerId)) { + clonedUpstream = { kind: "peer", peerId: upstream.peerId }; + } else if (upstream.kind === "none" || upstream.kind === "sfu") { + clonedUpstream = { kind: upstream.kind }; + } else { + return null; + } + if ( + !Array.isArray(value.childPeerIds) || + value.childPeerIds.length > 2 || + !value.childPeerIds.every(isOpaqueId) || + new Set(value.childPeerIds).size !== value.childPeerIds.length || + (value.sfuPublicationGeneration !== null && + !isOpaqueId(value.sfuPublicationGeneration)) + ) { + return null; } return { - parentPeerId: typeof value.parentPeerId === "string" ? value.parentPeerId : null, - childPeerIds: value.childPeerIds.filter((item) => typeof item === "string"), + upstream: clonedUpstream, + childPeerIds: [...value.childPeerIds], + sfuPublicationGeneration: value.sfuPublicationGeneration, }; } - function setAssignment(value) { - state.assignment = cloneAssignment(value); - state.maxAssignedChildren = Math.max( - state.maxAssignedChildren, - state.assignment.childPeerIds.length, - ); + function sameRouteAssignment(left, right) { + return left.upstream.kind === right.upstream.kind && + (left.upstream.kind !== "peer" || + (right.upstream.kind === "peer" && + left.upstream.peerId === right.upstream.peerId)) && + left.sfuPublicationGeneration === right.sfuPublicationGeneration && + left.childPeerIds.length === right.childPeerIds.length && + left.childPeerIds.every((peerId, index) => peerId === right.childPeerIds[index]); + } + + function resetRouteTransition() { + transitionRevision = -1; + transitionPhase = null; + plannedRouteAssignment = null; + state.routeRevision = null; + state.routeAssignment = null; + } + + function acceptRouteUpdate(revision, phase, value) { + const assignment = cloneRouteAssignment(value); + if ( + !Number.isSafeInteger(revision) || + revision < 0 || + (phase !== "prepare" && phase !== "active") || + !assignment + ) { + return false; + } + if (revision < transitionRevision) { + return false; + } + if (revision === transitionRevision) { + if ( + !plannedRouteAssignment || + !sameRouteAssignment(plannedRouteAssignment, assignment) || + (transitionPhase === "active" && phase === "prepare") + ) { + return false; + } + if (transitionPhase === phase) { + return true; + } + } + transitionRevision = revision; + transitionPhase = phase; + plannedRouteAssignment = assignment; + if (phase === "active") { + state.routeRevision = revision; + state.routeAssignment = cloneRouteAssignment(assignment); + state.maxAssignedChildren = Math.max( + state.maxAssignedChildren, + assignment.childPeerIds.length, + ); + } + return true; } function sdpKey(sdp) { @@ -745,24 +920,54 @@ export function buildBenchmarkInitScript(options: { try { message = JSON.parse(value); } catch { return; } if (!message || typeof message.type !== "string") return; if (direction === "out" && message.type === "authenticate") { - socket.__screenerBenchmarkSignal = true; + signalingSocket = socket; + peerAssisted = false; + state.activeRouteReady = []; state.role = message.role; state.roomId = message.roomId; state.authenticateSentAtEpochMs = Date.now(); + state.signalingConnected = false; + } else if (socket !== signalingSocket) { + return; } else if (direction === "out" && message.type === "set-quality-settings") { state.qualitySettings = message.qualitySettings; + } else if ( + direction === "out" && + message.type === "route-ready" && + message.phase === "active" && + peerAssisted && + Number.isSafeInteger(message.revision) && + message.revision === state.routeRevision && + state.routeAssignment + ) { + state.activeRouteReady.push({ + revision: message.revision, + upstreamKind: state.routeAssignment.upstream.kind, + sfuPublicationGeneration: + state.routeAssignment.sfuPublicationGeneration, + }); + if (state.activeRouteReady.length > 32) state.activeRouteReady.shift(); } if (direction === "in" && message.type === "authenticated") { - socket.__screenerBenchmarkSignal = true; state.peerId = message.peerId; state.authenticatedAtEpochMs = Date.now(); state.signalingConnected = true; + peerAssisted = message.mediaMode === "peer-assisted"; + resetRouteTransition(); if (message.mediaMode === "peer-assisted") { - setAssignment(message.mediaAssignment); + acceptRouteUpdate( + message.routeRevision, + "active", + message.routeAssignment, + ); state.qualitySettings = message.qualitySettings; } - } else if (direction === "in" && message.type === "media-assignment") { - setAssignment(message.mediaAssignment); + } else if ( + direction === "in" && + message.type === "route-update" && + peerAssisted + ) { + acceptRouteUpdate(message.revision, message.phase, message.assignment); } else if (direction === "in" && message.type === "quality-settings") { state.qualitySettings = message.qualitySettings; } @@ -775,14 +980,15 @@ export function buildBenchmarkInitScript(options: { const socket = Reflect.construct(Target, args, Target); const nativeSend = socket.send; socket.send = function(data) { + const result = nativeSend.call(socket, data); handleSignalMessage(data, "out", socket); - return nativeSend.call(socket, data); + return result; }; socket.addEventListener("message", (event) => { handleSignalMessage(event.data, "in", socket); }); socket.addEventListener("close", () => { - if (socket.__screenerBenchmarkSignal) state.signalingConnected = false; + if (socket === signalingSocket) state.signalingConnected = false; }); return socket; }, @@ -973,7 +1179,8 @@ export function buildBenchmarkInitScript(options: { function baseSnapshot() { return { ...state, - assignment: cloneAssignment(state.assignment), + routeAssignment: cloneRouteAssignment(state.routeAssignment), + activeRouteReady: state.activeRouteReady.map((ready) => ({ ...ready })), }; } @@ -1313,7 +1520,7 @@ async function waitForViewerMedia( const snapshot = await progressSample(cdp, page); if ( snapshot.peerId && - snapshot.assignment.parentPeerId && + hasAuthoritativeMediaUpstream(snapshot) && snapshot.connections.some( (connection) => connection.connectionState === "connected" && @@ -1592,7 +1799,7 @@ function descendantsOf( for (const page of pages) { if ( page.peerId && - page.assignment.parentPeerId === parent && + routeParentPeerId(page) === parent && !descendants.includes(page.peerId) ) { descendants.push(page.peerId); @@ -1617,8 +1824,8 @@ async function runRecovery( const relay = finalPages.find( (page) => page.role === "viewer" && - page.assignment.parentPeerId === host.peerId && - page.assignment.childPeerIds.length > 0, + routeParentPeerId(page) === host.peerId && + routeChildPeerIds(page).length > 0, ); if (!relay?.peerId) { return { triggered: true, error: "No first-level relay was available" }; @@ -1653,7 +1860,7 @@ async function runRecovery( ); maxHostAssignedChildren = Math.max( maxHostAssignedChildren, - currentHost.assignment.childPeerIds.length, + routeChildPeerIds(currentHost).length, ); } const affected = observations.filter((page) => @@ -1661,8 +1868,8 @@ async function runRecovery( ); const branchRoot = affected.find( (page) => - page.assignment.parentPeerId !== null && - page.assignment.parentPeerId !== relay.peerId, + routeParentPeerId(page) !== null && + routeParentPeerId(page) !== relay.peerId, ); if (!reassignedAtEpochMs && branchRoot) { reassignedAtEpochMs = Date.now(); @@ -1933,6 +2140,7 @@ export async function main(): Promise { "Synthetic canvas motion exercises real Chromium WebRTC but is not a game-capture quality claim.", "Headless runs are topology and transport evidence, not representative GPU or power evidence.", "CPU, GPU, NIC totals, glass-to-glass latency, generational visual quality, mobile browsers, and TURN require external or device-specific measurement.", + "The local runner does not start LiveKit; SFU consistency is reported only when an SFU route is actually observed.", "The harness emits raw gate fields and simple invariants; it does not implement a route score or runtime policy.", ], runs: [], diff --git a/tests/peer-assisted-benchmark.test.ts b/tests/peer-assisted-benchmark.test.ts index 8e0b36f..ac98c45 100644 --- a/tests/peer-assisted-benchmark.test.ts +++ b/tests/peer-assisted-benchmark.test.ts @@ -1,13 +1,19 @@ +import { runInNewContext } from "node:vm"; import { describe, expect, it } from "vitest"; import { activeVideoEdgeCount, + buildRunChecks, buildBenchmarkInitScript, everyViewerAdvanced, parseBenchmarkConfig, parseViewerCounts, summarizeSamples, } from "../scripts/peer-assisted-benchmark"; +import type { + MediaRouteUpstream, + ParticipantRouteAssignment, +} from "../src/shared/protocol"; const lowQualitySettings = { resolution: "720p", @@ -34,10 +40,23 @@ function page( qualitySettings: lowQualitySettings, maxActiveOutboundMediaEdges: sendEdges, maxAssignedChildren: sendEdges, - assignment: { - parentPeerId: role === "viewer" ? "parent-peer" : null, - childPeerIds: Array.from({ length: sendEdges }, (_, index) => `child-${index}`), + routeRevision: 1 as number | null, + routeAssignment: { + upstream: + role === "viewer" + ? ({ kind: "peer", peerId: "parent-peer" } as MediaRouteUpstream) + : ({ kind: "none" } as MediaRouteUpstream), + childPeerIds: Array.from( + { length: sendEdges }, + (_, index) => `child-peer-${index}`, + ), + sfuPublicationGeneration: null as string | null, }, + activeRouteReady: [] as Array<{ + revision: number; + upstreamKind: ParticipantRouteAssignment["upstream"]["kind"]; + sfuPublicationGeneration: string | null; + }>, firstDecodedAtEpochMs: role === "viewer" ? 1_500 : null, firstRenderedAtEpochMs: role === "viewer" ? 1_550 : null, renderedFrames: role === "viewer" ? 10 : 0, @@ -76,6 +95,117 @@ function page( }; } +class FakeWebSocket { + private readonly listeners = new Map< + string, + Array<(event: { data?: string }) => void> + >(); + + send(_data: string): void {} + + addEventListener( + type: string, + listener: (event: { data?: string }) => void, + ): void { + const listeners = this.listeners.get(type) ?? []; + listeners.push(listener); + this.listeners.set(type, listeners); + } + + emitMessage(message: unknown): void { + for (const listener of this.listeners.get("message") ?? []) { + listener({ data: JSON.stringify(message) }); + } + } + + emitClose(): void { + for (const listener of this.listeners.get("close") ?? []) { + listener({}); + } + } +} + +interface ObserverSnapshot { + peerId: string | null; + signalingConnected: boolean; + routeRevision: number | null; + routeAssignment: ParticipantRouteAssignment | null; + activeRouteReady: Array<{ + revision: number; + upstreamKind: ParticipantRouteAssignment["upstream"]["kind"]; + sfuPublicationGeneration: string | null; + }>; + maxAssignedChildren: number; +} + +function createObserverHarness() { + const context: Record = { + WebSocket: FakeWebSocket, + RTCPeerConnection: class FakePeerConnection {}, + navigator: { mediaDevices: {} }, + document: { querySelector: () => null }, + localStorage: { removeItem: () => undefined }, + performance: { timeOrigin: 0 }, + setInterval: () => 1, + clearInterval: () => undefined, + queueMicrotask, + }; + const source = buildBenchmarkInitScript({ + label: "host-1", + role: "host", + viewerIndex: null, + clearHostRoom: false, + width: 1280, + height: 720, + frameRate: 30, + }).replace( + 'const statsModule = import("/src/client/webrtc/stats.ts");', + "const statsModule = new Promise(() => undefined);", + ); + runInNewContext(source, context); + const WebSocketConstructor = context.WebSocket as new () => FakeWebSocket; + const api = context.__SCREENER_BENCHMARK__ as { + snapshot: () => ObserverSnapshot; + }; + return { + socket: () => new WebSocketConstructor(), + snapshot: () => structuredClone(api.snapshot()), + }; +} + +function authenticate(socket: FakeWebSocket): void { + socket.send( + JSON.stringify({ + type: "authenticate", + roomId: "1", + role: "host", + }), + ); +} + +function routeAssignment( + childPeerIds: string[], + generation: string | null = null, +): ParticipantRouteAssignment { + return { + upstream: { kind: "none" }, + childPeerIds, + sfuPublicationGeneration: generation, + }; +} + +function markActiveRouteReady(observation: ReturnType): void { + if (observation.routeRevision === null) { + throw new Error("An active route revision is required"); + } + observation.activeRouteReady.push({ + revision: observation.routeRevision, + upstreamKind: observation.routeAssignment.upstream.kind, + sfuPublicationGeneration: + observation.routeAssignment.sfuPublicationGeneration, + }); +} + describe("peer-assisted benchmark configuration", () => { it("uses the bounded 1/3/5/8 matrix by default", () => { expect(parseViewerCounts(undefined)).toEqual([1, 3, 5, 8]); @@ -194,6 +324,247 @@ describe("peer-assisted benchmark observations", () => { ).toBe(false); }); + it("accepts an authoritative SFU root and records its route generation", () => { + const initial = [ + page("host", "host", 1, 0), + page("viewer", "viewer-1", 0, 1), + ]; + initial[0]!.routeRevision = 7; + initial[0]!.routeAssignment.sfuPublicationGeneration = + "publication_generation_12345678"; + initial[1]!.routeRevision = 7; + initial[1]!.routeAssignment.upstream = { kind: "sfu" }; + markActiveRouteReady(initial[0]!); + markActiveRouteReady(initial[1]!); + + const final = structuredClone(initial); + final[1]!.connections[0]!.receiveTotals!.framesTotal = 20; + const summary = summarizeSamples( + [ + { atEpochMs: 2_000, elapsedMs: 0, pages: initial }, + { atEpochMs: 4_000, elapsedMs: 2_000, pages: final }, + ], + 1, + ); + + expect(summary.everyViewerDecoded).toBe(true); + expect(summary.viewerContinuity[0]).toMatchObject({ + assigned: true, + routeRevision: 7, + upstreamKind: "sfu", + }); + expect(summary.finalTopology).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + role: "host", + routeRevision: 7, + routeUpstream: { kind: "none" }, + sfuPublicationGeneration: "publication_generation_12345678", + }), + expect.objectContaining({ + role: "viewer", + parentPeerId: null, + routeRevision: 7, + routeUpstream: { kind: "sfu" }, + }), + ]), + ); + }); + + it("fails SFU continuity without one same-revision Host publication", () => { + function summarizeSfu( + hostGeneration: string | null, + hostRevision: number, + viewerRevision: number, + viewerGeneration: string | null = null, + ) { + const initial = [ + page("host", "host", 1, 0), + page("viewer", "viewer-1", 0, 1), + ]; + initial[0]!.routeRevision = hostRevision; + initial[0]!.routeAssignment.sfuPublicationGeneration = hostGeneration; + initial[1]!.routeRevision = viewerRevision; + initial[1]!.routeAssignment.upstream = { kind: "sfu" }; + initial[1]!.routeAssignment.sfuPublicationGeneration = viewerGeneration; + markActiveRouteReady(initial[0]!); + markActiveRouteReady(initial[1]!); + const final = structuredClone(initial); + final[1]!.connections[0]!.receiveTotals!.framesTotal = 20; + return summarizeSamples( + [ + { atEpochMs: 2_000, elapsedMs: 0, pages: initial }, + { atEpochMs: 4_000, elapsedMs: 2_000, pages: final }, + ], + 1, + ); + } + + const missing = summarizeSfu(null, 7, 7); + expect(missing.sfuPublicationCoherent).toBe(false); + expect(missing.everyViewerDecoded).toBe(false); + + const mismatchedRevision = summarizeSfu( + "publication_generation_12345678", + 7, + 8, + ); + expect(mismatchedRevision.sfuPublicationCoherent).toBe(false); + expect(mismatchedRevision.everyViewerDecoded).toBe(false); + + const mismatchedOwner = summarizeSfu( + "publication_generation_12345678", + 7, + 7, + "viewer_generation_12345678", + ); + expect(mismatchedOwner.sfuPublicationCoherent).toBe(false); + expect(mismatchedOwner.everyViewerDecoded).toBe(false); + }); + + it("requires every SFU-rooted participant to share the active revision", () => { + const initial = [ + page("host", "host", 1, 0), + page("viewer", "viewer-1", 1, 1), + page("viewer", "viewer-2", 0, 1), + ]; + initial[0]!.routeRevision = 7; + initial[0]!.routeAssignment.sfuPublicationGeneration = + "publication_generation_12345678"; + initial[1]!.routeRevision = 7; + initial[1]!.routeAssignment.upstream = { kind: "sfu" }; + initial[2]!.routeRevision = 6; + initial[2]!.routeAssignment.upstream = { + kind: "peer", + peerId: initial[1]!.peerId, + }; + markActiveRouteReady(initial[0]!); + markActiveRouteReady(initial[1]!); + + const final = structuredClone(initial); + final[1]!.connections.at(-1)!.receiveTotals!.framesTotal = 20; + final[2]!.connections.at(-1)!.receiveTotals!.framesTotal = 20; + const summary = summarizeSamples( + [ + { atEpochMs: 2_000, elapsedMs: 0, pages: initial }, + { atEpochMs: 4_000, elapsedMs: 2_000, pages: final }, + ], + 2, + ); + + expect(summary.sfuPublicationObserved).toBe(true); + expect(summary.sfuPublicationCoherent).toBe(false); + expect(summary.everyViewerDecoded).toBe(false); + }); + + it("does not treat old peer frames as active SFU media", () => { + const initial = [ + page("host", "host", 1, 0), + page("viewer", "viewer-1", 0, 1), + ]; + initial[0]!.routeRevision = 7; + initial[0]!.routeAssignment.sfuPublicationGeneration = + "publication_generation_12345678"; + initial[1]!.routeRevision = 7; + initial[1]!.routeAssignment.upstream = { kind: "sfu" }; + markActiveRouteReady(initial[0]!); + + const final = structuredClone(initial); + final[1]!.connections[0]!.receiveTotals!.framesTotal = 20; + const summary = summarizeSamples( + [ + { atEpochMs: 2_000, elapsedMs: 0, pages: initial }, + { atEpochMs: 4_000, elapsedMs: 2_000, pages: final }, + ], + 1, + ); + + expect(summary.sfuPublicationObserved).toBe(true); + expect(summary.sfuPublicationCoherent).toBe(false); + expect(summary.everyViewerDecoded).toBe(false); + }); + + it("reports only the no-orphan invariant when no SFU root is active", () => { + const initial = [ + page("host", "host", 1, 0), + page("viewer", "viewer-1", 0, 1), + ]; + const final = structuredClone(initial); + final[1]!.connections[0]!.receiveTotals!.framesTotal = 20; + const summary = summarizeSamples( + [ + { atEpochMs: 2_000, elapsedMs: 0, pages: initial }, + { atEpochMs: 4_000, elapsedMs: 2_000, pages: final }, + ], + 1, + ); + const checks = buildRunChecks(summary, 1, "720p30"); + + expect(summary.sfuPublicationObserved).toBe(false); + expect(checks.some((check) => check.name === "sfu-route-consistency")).toBe( + false, + ); + expect(checks).toContainEqual( + expect.objectContaining({ + name: "no-orphan-sfu-publication", + passed: true, + }), + ); + + const orphaned = structuredClone(final); + orphaned[0]!.routeAssignment.sfuPublicationGeneration = + "publication_generation_12345678"; + const orphanedSummary = summarizeSamples( + [ + { atEpochMs: 2_000, elapsedMs: 0, pages: initial }, + { atEpochMs: 4_000, elapsedMs: 2_000, pages: orphaned }, + ], + 1, + ); + expect(orphanedSummary.sfuPublicationObserved).toBe(false); + expect(orphanedSummary.sfuPublicationCoherent).toBe(false); + expect( + buildRunChecks(orphanedSummary, 1, "720p30").find( + (check) => check.name === "no-orphan-sfu-publication", + )?.passed, + ).toBe(false); + }); + + it("fails closed when decoded media has no authoritative upstream", () => { + const initial = [ + page("host", "host", 1, 0), + page("viewer", "viewer-1", 0, 1), + ]; + initial[1]!.routeAssignment.upstream = { kind: "none" }; + const final = structuredClone(initial); + final[1]!.connections[0]!.receiveTotals!.framesTotal = 20; + + expect( + summarizeSamples( + [ + { atEpochMs: 2_000, elapsedMs: 0, pages: initial }, + { atEpochMs: 4_000, elapsedMs: 2_000, pages: final }, + ], + 1, + ).everyViewerDecoded, + ).toBe(false); + + final[1]!.routeAssignment.upstream = { + kind: "peer", + peerId: "parent-peer", + }; + final[1]!.routeRevision = null; + expect( + summarizeSamples( + [ + { atEpochMs: 2_000, elapsedMs: 0, pages: initial }, + { atEpochMs: 4_000, elapsedMs: 2_000, pages: final }, + ], + 1, + ).everyViewerDecoded, + ).toBe(false); + }); + it("requires every viewer to decode and render after a quality change", () => { const before = [ page("host", "host", 2, 0), @@ -215,6 +586,174 @@ describe("peer-assisted benchmark observations", () => { expect(everyViewerAdvanced(before, after)).toBe(false); }); + it("observes only active authoritative route transitions", () => { + const observer = createObserverHarness(); + const owner = observer.socket(); + const outsider = observer.socket(); + const activeRoute = routeAssignment(["child_peer_0001"]); + const preparedRoute = routeAssignment( + ["child_peer_0001", "child_peer_0002"], + "publication_generation_12345678", + ); + const conflictingRoute = routeAssignment(["conflict_peer_01"]); + + outsider.emitMessage({ + type: "authenticated", + role: "host", + peerId: "outsider_peer_01", + mediaMode: "peer-assisted", + routeRevision: 99, + routeAssignment: conflictingRoute, + qualitySettings: lowQualitySettings, + }); + expect(observer.snapshot().peerId).toBeNull(); + + authenticate(owner); + owner.emitMessage({ + type: "authenticated", + role: "host", + peerId: "host_peer_0001", + mediaMode: "peer-assisted", + routeRevision: 4, + routeAssignment: activeRoute, + qualitySettings: lowQualitySettings, + }); + expect(observer.snapshot()).toMatchObject({ + routeRevision: 4, + routeAssignment: activeRoute, + maxAssignedChildren: 1, + }); + + owner.emitMessage({ + type: "route-update", + revision: 5, + phase: "prepare", + assignment: preparedRoute, + }); + expect(observer.snapshot()).toMatchObject({ + routeRevision: 4, + routeAssignment: activeRoute, + maxAssignedChildren: 1, + }); + + owner.emitMessage({ + type: "route-update", + revision: 3, + phase: "active", + assignment: preparedRoute, + }); + owner.emitMessage({ + type: "route-update", + revision: 5, + phase: "active", + assignment: conflictingRoute, + }); + expect(observer.snapshot().routeAssignment).toEqual(activeRoute); + + owner.emitMessage({ + type: "route-update", + revision: 5, + phase: "active", + assignment: preparedRoute, + }); + expect(observer.snapshot()).toMatchObject({ + routeRevision: 5, + routeAssignment: preparedRoute, + maxAssignedChildren: 2, + }); + + owner.send( + JSON.stringify({ type: "route-ready", revision: 5, phase: "active" }), + ); + outsider.send( + JSON.stringify({ type: "route-ready", revision: 5, phase: "active" }), + ); + expect(observer.snapshot().activeRouteReady).toEqual([ + { + revision: 5, + upstreamKind: "none", + sfuPublicationGeneration: "publication_generation_12345678", + }, + ]); + + owner.emitMessage({ + type: "route-update", + revision: 5, + phase: "prepare", + assignment: preparedRoute, + }); + owner.emitMessage({ + type: "route-update", + revision: 5, + phase: "active", + assignment: activeRoute, + }); + owner.emitMessage({ + type: "media-assignment", + mediaAssignment: { + parentPeerId: null, + childPeerIds: ["legacy_child_01"], + }, + }); + expect(observer.snapshot()).toMatchObject({ + routeRevision: 5, + routeAssignment: preparedRoute, + maxAssignedChildren: 2, + }); + }); + + it("clears hybrid state only from the current authenticated socket", () => { + const observer = createObserverHarness(); + const previous = observer.socket(); + const current = observer.socket(); + const assignment = routeAssignment(["child_peer_0001"]); + + authenticate(previous); + previous.emitMessage({ + type: "authenticated", + role: "host", + peerId: "host_peer_0001", + mediaMode: "peer-assisted", + routeRevision: 4, + routeAssignment: assignment, + qualitySettings: lowQualitySettings, + }); + authenticate(current); + expect(observer.snapshot()).toMatchObject({ + signalingConnected: false, + routeRevision: 4, + routeAssignment: assignment, + activeRouteReady: [], + }); + current.emitMessage({ + type: "authenticated", + role: "host", + peerId: "host_peer_0001", + viewerPeerIds: [], + }); + expect(observer.snapshot()).toMatchObject({ + signalingConnected: true, + routeRevision: null, + routeAssignment: null, + }); + + previous.emitMessage({ + type: "route-update", + revision: 6, + phase: "active", + assignment, + }); + previous.emitClose(); + expect(observer.snapshot()).toMatchObject({ + signalingConnected: true, + routeRevision: null, + routeAssignment: null, + }); + + current.emitClose(); + expect(observer.snapshot().signalingConnected).toBe(false); + }); + it("injects the requested deterministic capture dimensions", () => { const source = buildBenchmarkInitScript({ label: "host-1",