From 501df27bce627f9cd82b3798da4e9aeecef7d433 Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Thu, 27 Aug 2026 02:57:38 +0800 Subject: [PATCH] feat(quality): implement strict v13 quality shadow (#291) --- .../0005-automatic-hybrid-media-routing.md | 16 +- docs/project-memory.md | 16 +- docs/research/realtime-quality-adaptation.md | 11 +- docs/status.md | 13 +- docs/todo.md | 18 +- docs/verification-status.md | 9 +- docs/方案设计.md | 2 +- src/client/media/viewer-presentation.ts | 67 +++-- src/client/media/viewer-quality-evidence.ts | 102 +++++-- src/client/pages/ViewerPage.tsx | 167 ++++++++++- src/client/types.ts | 4 + src/client/webrtc/stats.ts | 27 ++ src/server/hybrid-media-router.ts | 12 + src/server/room-route-controller.ts | 281 ++++++++++++++++++ src/server/signaling.ts | 80 +++-- src/shared/protocol.ts | 64 ++-- tests/client-state.test.ts | 111 ++++++- tests/protocol.test.ts | 106 ++++++- tests/room-route-controller.test.ts | 190 ++++++++++++ tests/server-signal.test.ts | 84 +++++- tests/viewer-presentation.test.ts | 80 ++++- tests/viewer-quality-evidence.test.ts | 210 +++++++++++-- 22 files changed, 1467 insertions(+), 203 deletions(-) diff --git a/docs/adr/0005-automatic-hybrid-media-routing.md b/docs/adr/0005-automatic-hybrid-media-routing.md index dcef0a8..f8b58c3 100644 --- a/docs/adr/0005-automatic-hybrid-media-routing.md +++ b/docs/adr/0005-automatic-hybrid-media-routing.md @@ -1,8 +1,8 @@ # ADR-0005: Automatic Hybrid Media Routing - Status: Base routing is deployed on strict v12. The evidence-only quality - shadow and atomic strict-v13 upgrade are accepted but not yet implemented or - deployed. Exact current identity is owned by + shadow is implemented in canonical source on strict v13 but not yet deployed. + Exact current identity is owned by [status](../status.md). - Date: 2026-08-20 - Last updated: 2026-08-27 @@ -361,8 +361,8 @@ current-candidate-start, first-decoded-frame, and final-outcome timing. A new demand overwrites that child's record; authoritative share stop/replacement, confirmed departure, and room deletion remove it. -An authenticated Host acceptance harness may request one snapshot. The accepted -v12 product UI exposes no diagnostic download button. The response +An authenticated Host acceptance harness may request one snapshot. The current +product UI exposes no diagnostic download button. The response reads the current graph, current operation, and latest records; the server does not push a periodic snapshot stream, start a timer, retain an event ring, log or persist the result, or create a second graph. Snapshot-local ordinals express @@ -424,10 +424,10 @@ deployment-wide. ## Current Source And Deployment Boundary -Production and canonical application/runtime source currently use strict -`screener-v12`; the accepted quality-shadow integration atomically replaces it -with strict `screener-v13`. Exact current source -and deployment identity is indexed by [status](../status.md). +Canonical application/runtime source uses the evidence-only strict +`screener-v13` quality shadow; production remains strict `screener-v12` until +its atomic application release. Exact current source and deployment identity is +indexed by [status](../status.md). Browser SFU PCs use empty external ICE-server lists. The controlled exact- candidate rollback/SFU commit and active-SFU cadence gates are closed. The operation owner remains only `route`; broader heterogeneous-network and SFU diff --git a/docs/project-memory.md b/docs/project-memory.md index b8d52fa..613f9e9 100644 --- a/docs/project-memory.md +++ b/docs/project-memory.md @@ -1,6 +1,6 @@ # Project Memory -Last updated: 2026-08-26 +Last updated: 2026-08-27 ## Current Product Truth @@ -120,13 +120,14 @@ Last updated: 2026-08-26 ## Current Source And Production -- Production and canonical source use the same strict `screener-v12` - application with adaptive H.264/VP8 selection, local Host codec override and +- Canonical source uses the evidence-only strict `screener-v13` quality shadow; + production remains strict `screener-v12` until the atomic application release. + Both retain adaptive H.264/VP8 selection, the local Host codec override and optional SQLite room authority. Exact identities are indexed by [status](./status.md). -- The accepted quality-shadow integration atomically replaces v12 with strict - `screener-v13`; old Browser pages must fail before room authority. It adds no - active route decision until production shadow evidence is accepted. +- Strict v13 rejects old Browser pages before room authority. Its bounded + current-edge aggregate adds no active route decision until annotated + production shadow evidence establishes an accepted policy. - Production uses SQLite-backed random rooms, 20-Viewer admission, endpoint cap `2`, fixed 9,000-publication ingress and `9000 * 20` subscription egress admission, STUN UDP 3478, LiveKit media UDP 7882, and Web TCP 80/443. Node 8787 and @@ -137,7 +138,8 @@ Last updated: 2026-08-26 ## Current Priority -1. Complete the evidence-only quality-route shadow on strict `screener-v13`. +1. Deploy the evidence-only strict `screener-v13` quality shadow and collect + annotated production samples. 2. Complete the mobile Viewer lifecycle matrix and representative real-network route/media acceptance. diff --git a/docs/research/realtime-quality-adaptation.md b/docs/research/realtime-quality-adaptation.md index fe6156c..bee0066 100644 --- a/docs/research/realtime-quality-adaptation.md +++ b/docs/research/realtime-quality-adaptation.md @@ -2,8 +2,8 @@ - Research date: 2026-08-26 - Scope: Browser game-screen capture, encoding, P2P forwarding, and LiveKit SFU -- Status: current Browser policy accepted; real-game, weak-device, and - heterogeneous-network quality remain open +- Status: strict-v13 evidence-only quality shadow implemented in canonical + source; production sampling and active policy remain open ## Current Conclusion @@ -354,9 +354,8 @@ page refresh also rebinds the same stable participant and committed graph. Only actual recovery exhaustion enters the normal route-failure operation and tries other eligible P2P parents before SFU. -Quality-driven relay abdication or active parent switching would require -counterfactual alternative-path evidence, bounded probing, and a separate -ADR-0005 decision. It is not implemented or authorized by current diagnostics. +Quality-driven relay abdication or active parent switching requires evidence +that current diagnostics do not yet provide. No active policy is implemented. The reopened quality-selection research retains that information boundary. Current-route WebRTC stats can prove user-visible degradation but cannot prove @@ -374,7 +373,7 @@ quality ceiling as a delivered floor requires a separate product SLO. Relay ingress degradation reparents that relay while retaining its subtree; one child's degradation moves only that child. -The accepted first stage is shadow evaluation only. Viewer evidence adds the +The implemented first stage is shadow evaluation only. Viewer evidence adds the standard cumulative-renderer deltas `freezeCount`, `totalFreezesDuration`, `pauseCount` and `totalPausesDuration` and requires continuing `framesDecoded` progress. W3C already diff --git a/docs/status.md b/docs/status.md index 785767b..e28e352 100644 --- a/docs/status.md +++ b/docs/status.md @@ -42,7 +42,8 @@ This is the current execution index. Git history owns completed timelines; ## Current Source -- Canonical `main` contains the same application tree deployed from `267e073`. +- Canonical `main` uses the evidence-only strict `screener-v13` quality shadow; + production remains the strict-v12 application tree deployed from `267e073`. Optional SQLite stable authority, non-expiring local preferred code, atomic room replacement and the Host codec selector are implemented; production selects stable storage. Graceful restart, crash, timeout and network loss use @@ -57,6 +58,10 @@ This is the current execution index. Git history owns completed timelines; quality score, layer selector, periodic rebalancing, or quality-driven parent change. P2P and SFU recover their current route before actual failure enters normal reassignment. +- Current Viewer freeze/pause evidence requires decoded progress and exact + foreground presentation authority. The controller retains one bounded, + fresh current-edge aggregate per child for the Host-only on-demand snapshot; + it does not change route facts, capacity, candidates, SFU usage or reconcile. - Viewer hidden/freeze/pagehide suppress decoded-stall routing authority; visible/resume/pageshow rebaseline and rearm current-frame proof. SFU pagehide no longer triggers LiveKit's automatic disconnect. Native Viewer @@ -64,10 +69,8 @@ This is the current execution index. Git history owns completed timelines; ## Current Milestone -1. Implement presentation-authoritative freeze/pause evidence with decoded - progress and the controller-owned shadow aggregate; collect production - samples without active route changes. This integration atomically replaces - current v12 with accepted strict `screener-v13`. +1. Atomically deploy strict `screener-v13`, then collect annotated production + freeze/pause samples without active route changes. 2. Finish the Android Chrome and iOS Safari Viewer lifecycle matrix. 3. Finish representative public-network direct, peer-relay, SFU, recovery, Pause/Resume, screen-audio, real-game A/V, and all-UDP-blocked acceptance diff --git a/docs/todo.md b/docs/todo.md index 363aa83..c153f32 100644 --- a/docs/todo.md +++ b/docs/todo.md @@ -1,22 +1,18 @@ # Current TODO Ledger -Last reviewed: 2026-08-26 +Last reviewed: 2026-08-27 Only **Now** is executable. A branch, old experiment, observation, or accepted later topic is not implementation authority by itself. ## Now -1. **Quality-route evidence and shadow calibration.** Make current Viewer - quality evidence presentation-authoritative with standard freeze/pause - deltas, continuing decoded progress, exact presentation epochs and baseline - fencing. Keep one bounded - current-epoch aggregate per exact child in the existing room controller and - expose it only through the Host acceptance snapshot. It must not alter facts, - capacity, routes or SFU usage. Collect annotated production samples before - accepting an active trigger, probation, restore or cooldown policy. Ship it - only as the atomic strict `screener-v13` Browser/server contract; v12 gets no - compatibility path. +1. **Quality-route evidence and shadow calibration.** Atomically deploy the + implemented strict `screener-v13` Browser/server quality shadow; v12 gets no + compatibility path. Collect annotated production samples from its bounded + current-edge freeze/pause aggregates before accepting an active trigger, + probation, restore or cooldown policy. Shadow evidence must not alter facts, + capacity, routes or SFU usage. 2. **Mobile Viewer lifecycle.** Run Android Chrome and iOS Safari matrices for autoplay gesture, foreground/background audio, foreground video recovery, lock/page reclamation, rotation, Wi-Fi/cellular migration, and assigned-relay diff --git a/docs/verification-status.md b/docs/verification-status.md index 78df91e..70ad682 100644 --- a/docs/verification-status.md +++ b/docs/verification-status.md @@ -64,8 +64,9 @@ may be interpreted. Git and pull requests own routine completed checks. ## Current Source Evidence -- Canonical source contains the same application tree deployed from `267e073`: - the adaptive H.264/VP8 sender decision with `contentHint = "motion"` reapplies the +- Canonical source uses strict `screener-v13`; production remains the strict-v12 + application tree deployed from `267e073`. The adaptive H.264/VP8 sender + decision with `contentHint = "motion"` reapplies the selected video profile after answer negotiation, leaves SFU representation construction to pinned LiveKit, keeps Dynacast/send-side BWE, disables AdaptiveStream, and disables SFU audio RED. @@ -76,6 +77,10 @@ may be interpreted. Git and pull requests own routine completed checks. authority, hidden-page stall suppression, retryable failure reporting, native controls, and same-route manual reconnect are current source behavior. Mobile physical evidence remains open. +- The evidence-only quality shadow accepts only exact foreground presentation + windows with decoded progress, keeps one bounded fresh aggregate per current + child, and exposes it through the Host-only anonymous route snapshot. It does + not change route facts, capacity, candidate order, SFU admission or reconcile. ## Open Physical Gates diff --git a/docs/方案设计.md b/docs/方案设计.md index fcf3192..4657e4c 100644 --- a/docs/方案设计.md +++ b/docs/方案设计.md @@ -2,7 +2,7 @@ ## 2026-08-18 14:45 -状态：production 与 canonical source 当前使用严格 `screener-v12`；已接受的 quality-shadow 集成会把 Browser/server 原子升级为唯一 `screener-v13`，在 authority 前拒绝旧页面，不保留 v12 parser。当前已实现每分享代次的 H.264 sender gate 与 VP8 fallback、本地 pre-share `VP8 | 自动 | H264` selector、可选 SQLite stable room authority、无期限 preferred code 和换房控制；这些不新增 codec wire、自定义 SFU layer list或应用媒体质量 controller。Video 使用 `contentHint = "motion"`，audio 使用 `music`。SFU 由 pinned LiveKit 默认 representations、Dynacast、server send-side BWE 和 disabled AdaptiveStream 拥有质量适配；screen audio 使用 64/128/192 kbps ceiling、stereo、DTX off 与 RED off。精确执行状态由 `docs/status.md` 持有。 +状态：canonical source 已实现唯一 strict `screener-v13` 的 evidence-only quality shadow，production 在原子应用发布前仍使用 strict `screener-v12`；旧页面在 authority 前拒绝，不保留 v12 parser。当前已实现每分享代次的 H.264 sender gate 与 VP8 fallback、本地 pre-share `VP8 | 自动 | H264` selector、可选 SQLite stable room authority、无期限 preferred code 和换房控制；这些不新增 codec wire、自定义 SFU layer list或应用媒体质量 controller。Video 使用 `contentHint = "motion"`，audio 使用 `music`。SFU 由 pinned LiveKit 默认 representations、Dynacast、server send-side BWE 和 disabled AdaptiveStream 拥有质量适配；screen audio 使用 64/128/192 kbps ceiling、stereo、DTX off 与 RED off。精确执行状态由 `docs/status.md` 持有。 ## 目标与边界 diff --git a/src/client/media/viewer-presentation.ts b/src/client/media/viewer-presentation.ts index 8534539..7b8d222 100644 --- a/src/client/media/viewer-presentation.ts +++ b/src/client/media/viewer-presentation.ts @@ -116,6 +116,7 @@ export type ViewerPresentationAction = } | { type: "media-bound"; generation: number; revision: number } | { type: "frame-presented"; generation: number; revision: number } + | { type: "frame-proof-reset"; generation: number; revision: number } | { type: "autoplay-blocked"; generation: number; revision: number } | { type: "autoplay-cleared"; generation: number } | { type: "playback-failed"; generation: number; revision: number } @@ -210,8 +211,7 @@ export function reduceViewerPresentation( kind: action.kind, }, routeStatus: - state.routeStatus?.revision === action.revision && - state.routeStatus.state === "waiting" + state.routeStatus?.revision === action.revision ? state.routeStatus : null, media: preserveMedia @@ -288,7 +288,11 @@ export function reduceViewerPresentation( case "media-bound": { if ( (state.revision !== null && action.revision < state.revision) || - (state.media && action.generation <= state.media.generation) + (state.media && action.generation <= state.media.generation) || + (state.routeStatus?.revision === action.revision && + state.routeStatus.state === "failed") || + (state.revision === action.revision && + state.failure === "ROUTE_EXHAUSTED") ) { return state; } @@ -315,7 +319,9 @@ export function reduceViewerPresentation( !state.media || state.media.generation !== action.generation || state.media.revision !== action.revision || - state.revision !== action.revision + state.revision !== action.revision || + state.routeStatus?.state === "failed" || + state.failure === "ROUTE_EXHAUSTED" ) { return state; } @@ -326,10 +332,22 @@ export function reduceViewerPresentation( retainedFrame: false, routeStatus: null, failure: - state.failure === "PLAYBACK_FAILED" || - state.failure === "ROUTE_EXHAUSTED" - ? null - : state.failure, + state.failure === "PLAYBACK_FAILED" ? null : state.failure, + }; + case "frame-proof-reset": + if ( + !state.media || + state.media.generation !== action.generation || + state.media.revision !== action.revision || + state.revision !== action.revision || + !state.media.framePresented + ) { + return state; + } + return { + ...state, + media: { ...state.media, framePresented: false }, + retainedFrame: true, }; case "autoplay-blocked": if ( @@ -540,6 +558,15 @@ export function deriveViewerPresentation( }; } + if (state.routeStatus?.state === "failed") { + return presentation( + "route-failed", + "没有可用的媒体线路", + frameOverlay, + state, + ); + } + switch (state.failure) { case "ROUTE_EXHAUSTED": return presentation( @@ -607,6 +634,18 @@ export function deriveViewerPresentation( state, ); } + if ( + state.signal === "reconnecting" || + state.connection === "reconnecting" || + state.connection === "failed" + ) { + return presentation( + "recovering", + "正在恢复连接", + frameOverlay, + state, + ); + } if (state.media && state.media.revision === state.revision) { return presentation( "receiving", @@ -623,18 +662,6 @@ export function deriveViewerPresentation( state, ); } - if ( - state.signal === "reconnecting" || - state.connection === "reconnecting" || - state.connection === "failed" - ) { - return presentation( - "recovering", - "正在恢复连接", - frameOverlay, - state, - ); - } if (state.host === "stopped" || state.host === "unknown") { return presentation( "waiting-host", diff --git a/src/client/media/viewer-quality-evidence.ts b/src/client/media/viewer-quality-evidence.ts index 5eb963c..2ac4a87 100644 --- a/src/client/media/viewer-quality-evidence.ts +++ b/src/client/media/viewer-quality-evidence.ts @@ -96,10 +96,6 @@ export function qualityEvidenceWindowFromMetrics( width !== null && height !== null ? { width, height } : { width: null, height: null }; - const freezeDuration = boundedNumber( - metrics.intervalFreezeDurationMs, - windowMs, - ); const candidate: ViewerQualityEvidenceMetrics = { ...dimensions, framesPerSecond: boundedNumber(metrics.framesPerSecond, 240), @@ -120,8 +116,22 @@ export function qualityEvidenceWindowFromMetrics( 10_000, ), decodeMsPerFrame: boundedNumber(metrics.intervalDecodeMs, 60_000), - freezeCountDelta: boundedInteger(metrics.intervalFreezeCount, 10_000), - freezeDurationMsDelta: freezeDuration, + freezeCountDelta: boundedInteger( + metrics.intervalFreezeCount, + Number.MAX_SAFE_INTEGER, + ), + freezeDurationMsDelta: boundedNumber( + metrics.intervalFreezeDurationMs, + Number.MAX_SAFE_INTEGER, + ), + pauseCountDelta: boundedInteger( + metrics.intervalPauseCount, + Number.MAX_SAFE_INTEGER, + ), + pauseDurationMsDelta: boundedNumber( + metrics.intervalPauseDurationMs, + Number.MAX_SAFE_INTEGER, + ), codec: boundedString( metrics.codec, 64, @@ -175,7 +185,9 @@ export function qualityEvidenceWindowFromMetrics( export class ViewerQualityEvidenceReporter { private connectionId: string | null = null; - private routeRevision: number | null = null; + private presentationEpoch = 0; + private presentationEligible = false; + private baselinePending = true; private sequence = 0; private lastSentAtMs: number | null = null; private lastSampleTimestampMs: number | null = null; @@ -185,11 +197,16 @@ export class ViewerQualityEvidenceReporter { private readonly now: () => number = Date.now, ) {} - offer(snapshot: PeerSnapshot, routeRevision: number): boolean { + offer( + snapshot: PeerSnapshot, + routeRevision: number, + presentationEligible: boolean, + ): boolean { return this.offerMetrics( snapshot.connectionId, snapshot.metrics, routeRevision, + presentationEligible, ); } @@ -197,6 +214,7 @@ export class ViewerQualityEvidenceReporter { connectionId: string, metrics: ConnectionMetrics, routeRevision: number, + presentationEligible: boolean, ): boolean { if ( !Number.isSafeInteger(routeRevision) || @@ -205,8 +223,14 @@ export class ViewerQualityEvidenceReporter { ) { return false; } - const window = qualityEvidenceWindowFromMetrics(metrics); const sampleTimestampMs = metrics.sampleTimestampMs; + if (this.connectionId !== connectionId) { + this.connectionId = connectionId; + this.sequence = 0; + this.lastSampleTimestampMs = null; + this.baselinePending = true; + } + const window = qualityEvidenceWindowFromMetrics(metrics); if ( !window || sampleTimestampMs === null || @@ -215,27 +239,36 @@ export class ViewerQualityEvidenceReporter { ) { return false; } - if (this.connectionId !== connectionId) { - this.connectionId = connectionId; - this.routeRevision = routeRevision; - this.sequence = 0; - this.lastSentAtMs = null; - this.lastSampleTimestampMs = null; - } else if (this.routeRevision !== routeRevision) { - this.routeRevision = routeRevision; - this.lastSentAtMs = null; - this.lastSampleTimestampMs = null; - } - if (this.sequence > Number.MAX_SAFE_INTEGER) { - return false; - } - if ( this.lastSampleTimestampMs !== null && sampleTimestampMs <= this.lastSampleTimestampMs ) { return false; } + this.lastSampleTimestampMs = sampleTimestampMs; + if (!presentationEligible) { + if (this.presentationEligible) { + this.invalidatePresentation(); + } + return false; + } + if (!this.presentationEligible) { + this.presentationEligible = true; + this.baselinePending = true; + } + const decodedProgress = + metrics.intervalFramesDecoded !== null && + metrics.intervalFramesDecoded > 0; + if (!decodedProgress) { + return false; + } + if (this.baselinePending) { + this.baselinePending = false; + return false; + } + if (this.sequence > Number.MAX_SAFE_INTEGER) { + return false; + } const now = this.now(); if ( !Number.isFinite(now) || @@ -250,6 +283,7 @@ export class ViewerQualityEvidenceReporter { guard: { connectionId, routeRevision, + presentationEpoch: this.presentationEpoch, }, sequence: this.sequence, ...window, @@ -266,13 +300,26 @@ export class ViewerQualityEvidenceReporter { this.sequence += 1; this.lastSentAtMs = now; - this.lastSampleTimestampMs = sampleTimestampMs; return true; } + invalidatePresentation(): void { + if (!this.presentationEligible) { + return; + } + if (this.presentationEpoch < Number.MAX_SAFE_INTEGER) { + this.presentationEpoch += 1; + } + this.presentationEligible = false; + this.baselinePending = true; + this.sequence = 0; + } + reset(): void { this.connectionId = null; - this.routeRevision = null; + this.presentationEpoch = 0; + this.presentationEligible = false; + this.baselinePending = true; this.sequence = 0; this.lastSentAtMs = null; this.lastSampleTimestampMs = null; @@ -286,7 +333,8 @@ function sameViewerQualityEvidenceIdentity( return ( previous.viewerPeerId === next.viewerPeerId && qualityEvidenceUpstreamMatches(previous, next.upstream) && - previous.guard.connectionId === next.guard.connectionId + previous.guard.connectionId === next.guard.connectionId && + previous.guard.presentationEpoch === next.guard.presentationEpoch ); } diff --git a/src/client/pages/ViewerPage.tsx b/src/client/pages/ViewerPage.tsx index 5e53844..0b9fd05 100644 --- a/src/client/pages/ViewerPage.tsx +++ b/src/client/pages/ViewerPage.tsx @@ -63,6 +63,7 @@ import { deriveViewerPresentation, reduceViewerPresentation, viewerFailureFromServerCode, + type ViewerPresentationAction, type ViewerRouteKind, type ViewerStage, } from "../media/viewer-presentation"; @@ -155,7 +156,7 @@ function StageOverlayIcon({ stage }: { stage: ViewerStage }) { } export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { - const [presentationState, dispatchPresentation] = useReducer( + const [presentationState, dispatchPresentationState] = useReducer( reduceViewerPresentation, INITIAL_VIEWER_PRESENTATION_STATE, ); @@ -234,6 +235,18 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { const peerRef = useRef(null); const viewerSfuRouteRef = useRef(null); const signalRef = useRef(null); + const presentationStateRef = useRef(presentationState); + presentationStateRef.current = presentationState; + function dispatchPresentation(action: ViewerPresentationAction): void { + presentationStateRef.current = reduceViewerPresentation( + presentationStateRef.current, + action, + ); + dispatchPresentationState(action); + } + const qualityEvidenceReporterRef = + useRef(null); + const qualityFrameProofGenerationRef = useRef(null); const displayNameRef = useRef(displayName); const remoteMediaRef = useRef(null); const mediaGenerationRef = useRef(0); @@ -279,6 +292,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { videoTrackKey, generation: ++mediaGenerationRef.current, }; + invalidateQualityPresentation(); remoteMediaRef.current = next; setRemoteMedia(next); dispatchPresentation({ @@ -289,22 +303,62 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { } function clearRemoteMedia(): void { + invalidateQualityPresentation(); remoteMediaRef.current = null; setRemoteMedia(null); dispatchPresentation({ type: "media-cleared" }); } + function resetCurrentFrameProof(): void { + const binding = remoteMediaRef.current; + if (!binding) { + return; + } + dispatchPresentation({ + type: "frame-proof-reset", + generation: binding.generation, + revision: binding.revision, + }); + } + + function invalidateQualityPresentation(): void { + qualityEvidenceReporterRef.current?.invalidatePresentation(); + qualityFrameProofGenerationRef.current = null; + } + + function invalidatePresentedMedia(): void { + invalidateQualityPresentation(); + resetCurrentFrameProof(); + } + + function mediaBindingIsCurrent(binding: RemoteMediaBinding): boolean { + const current = remoteMediaRef.current; + return ( + current?.generation === binding.generation && + current.revision === binding.revision && + current.stream === binding.stream + ); + } + function attemptPlayback( video: HTMLVideoElement, binding: RemoteMediaBinding, ): void { void video.play().then( - () => + () => { + if (!mediaBindingIsCurrent(binding)) { + return; + } dispatchPresentation({ type: "autoplay-cleared", generation: binding.generation, - }), + }); + }, (error: unknown) => { + if (!mediaBindingIsCurrent(binding)) { + return; + } + invalidatePresentedMedia(); if (isAutoplayPolicyRejection(error)) { dispatchPresentation({ type: "autoplay-blocked", @@ -384,16 +438,19 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { let relayChildEvidenceCurrent: ViewerQualityEvidencePresentation | null = null; let relayChildEvidenceTimer: number | null = null; + let sfuTransportConnected = false; let pageSuspended = document.visibilityState !== "visible"; const syncDecodedFrameStallPause = (): void => { decodedFrameStall.setPaused(currentHostPaused || pageSuspended); }; const suspendForPageLifecycle = (): void => { + invalidatePresentedMedia(); pageSuspended = true; syncDecodedFrameStallPause(); }; const recoverFromPageLifecycle = (): void => { + invalidatePresentedMedia(); pageSuspended = document.visibilityState !== "visible"; syncDecodedFrameStallPause(); if (pageSuspended) return; @@ -485,6 +542,50 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { const qualityEvidenceReporter = new ViewerQualityEvidenceReporter( (message) => active && signal.send(message), ); + qualityEvidenceReporterRef.current = qualityEvidenceReporter; + function qualityPresentationEligible( + connectionConnected: boolean, + ): boolean { + const state = presentationStateRef.current; + const binding = remoteMediaRef.current; + const video = videoRef.current; + return Boolean( + connectionConnected && + !pageSuspended && + document.visibilityState === "visible" && + !currentHostPaused && + currentRouteAssignment !== null && + currentRouteAssignment.upstream.kind !== "none" && + currentRouteConnectionId !== null && + binding && + binding.revision === currentRouteRevision && + qualityFrameProofGenerationRef.current === binding.generation && + state.routeStatus?.state !== "failed" && + state.failure !== "ROUTE_EXHAUSTED" && + video && + !video.paused && + !video.ended, + ); + } + function offerPeerQualityEvidence( + snapshot: PeerSnapshot, + peer: ViewerPeer, + ): void { + const connectionHealthy = + snapshot.connectionState === "connected" && !peer.isRecovering(); + const qualityEligible = + qualityPresentationEligible(connectionHealthy); + if (!connectionHealthy) { + invalidatePresentedMedia(); + } else if (!qualityEligible) { + invalidateQualityPresentation(); + } + qualityEvidenceReporter.offer( + snapshot, + currentRouteRevision, + qualityEligible, + ); + } function activateRouteIdentity( revision: number, assignment: ParticipantRouteAssignment, @@ -799,6 +900,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { return; } currentAssignment = { parentPeerId: null, childPeerIds: [] }; + sfuTransportConnected = false; setSfuUpstream(null); clearPeerState(); viewerRelay?.setChildren([]); @@ -838,6 +940,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { currentRouteConnectionId, metrics, revision, + qualityPresentationEligible(sfuTransportConnected), ); } } @@ -848,6 +951,13 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { }, onSfuState: (state, revision) => { if (active && viewerSfuRoute === route) { + const connected = state === "connected"; + if (sfuTransportConnected !== connected) { + sfuTransportConnected = connected; + if (!connected) { + invalidatePresentedMedia(); + } + } setSfuUpstream((current) => current ? { ...current, connectionState: state } : null, ); @@ -862,6 +972,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { if (!active || viewerSfuRoute !== route || available) { return; } + invalidatePresentedMedia(); setSfuUpstream((current) => current ? { ...current, connectionState: "reconnecting" } : current, ); @@ -889,6 +1000,9 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { const relay = ensureViewerRelay(); relay?.setStream(nextStream); bindRemoteStream(nextStream, revision); + if (initialVideoStream) { + sfuTransportConnected = true; + } setSfuUpstream( (current) => current ?? { connectionState: "connected", metrics: null }, @@ -913,6 +1027,8 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { const route = viewerSfuRoute; discardPendingPeer(); viewerSfuRoute = null; + sfuTransportConnected = false; + invalidatePresentedMedia(); if (viewerSfuRouteRef.current === route) { viewerSfuRouteRef.current = null; } @@ -1018,7 +1134,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { snapshot.metrics.intervalFramesDecoded, snapshot.connectionId, ); - qualityEvidenceReporter.offer(snapshot, currentRouteRevision); + offerPeerQualityEvidence(snapshot, peer); setPeerSnapshot(snapshot); dispatchPresentation({ type: "connection", @@ -1104,7 +1220,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { snapshot.metrics.intervalFramesDecoded, snapshot.connectionId, ); - qualityEvidenceReporter.offer(snapshot, currentRouteRevision); + offerPeerQualityEvidence(snapshot, peer); if ( !currentHostOnline && (snapshot.connectionState === "failed" || @@ -1131,12 +1247,12 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { if (viewerSfuRoute) { return viewerSfuRoute.reportPeerFailure(parentPeerId, connectionId); } + invalidateQualityPresentation(); dispatchPresentation({ type: "failure", failure: "ROUTE_EXHAUSTED", revision: currentRouteRevision, }); - setFrameProofEpoch((current) => current + 1); return true; }, }, @@ -1209,6 +1325,9 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { currentHostOnline = message.hostOnline; setHostOnline(message.hostOnline); const sharingPaused = message.hostPaused ?? false; + if (currentHostPaused !== sharingPaused && sharingPaused) { + invalidatePresentedMedia(); + } currentHostPaused = sharingPaused; syncDecodedFrameStallPause(); dispatchPresentation({ @@ -1342,14 +1461,14 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { return; } if (message.type === "route-status") { + if (message.state === "failed") { + invalidateQualityPresentation(); + } dispatchPresentation({ type: "route-status", revision: message.revision, state: message.state, }); - if (message.state === "failed") { - setFrameProofEpoch((current) => current + 1); - } return; } if (message.type === "viewer-quality-evidence") { @@ -1452,6 +1571,9 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { } if (message.type === "host-status") { currentHostOnline = message.online; + if (currentHostPaused !== message.paused && message.paused) { + invalidatePresentedMedia(); + } currentHostPaused = message.paused; syncDecodedFrameStallPause(); setHostOnline(message.online); @@ -1471,6 +1593,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { return; } if (message.type === "sharing-stopped") { + invalidatePresentedMedia(); currentRouteAssignment = null; currentRouteConnectionId = null; pendingRouteConnection = null; @@ -1586,6 +1709,9 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { window.removeEventListener("pageshow", recoverFromPageLifecycle); currentPeerId = null; qualityEvidenceReporter.reset(); + if (qualityEvidenceReporterRef.current === qualityEvidenceReporter) { + qualityEvidenceReporterRef.current = null; + } clearRelayChildEvidence(); sfuStandbyPrewarmer?.dispose(); signal.stop(); @@ -1656,13 +1782,18 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { if (!video || !remoteMedia) { return; } - return observeCompositedVideoFrame(video, remoteMedia.stream, () => + const binding = remoteMedia; + return observeCompositedVideoFrame(video, binding.stream, () => { + if (!mediaBindingIsCurrent(binding)) { + return; + } + qualityFrameProofGenerationRef.current = binding.generation; dispatchPresentation({ type: "frame-presented", - generation: remoteMedia.generation, - revision: remoteMedia.revision, - }), - ); + generation: binding.generation, + revision: binding.revision, + }); + }); }, [frameProofEpoch, presentationState.connection, remoteMedia]); function retryConnection(): void { @@ -1921,10 +2052,14 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { autoPlay controls={ presentation.overlay === "none" || - presentation.stage === "needs-play" + presentation.stage === "needs-play" || + (presentation.stage === "receiving" && + presentation.hasRetainedFrame) } playsInline onPlay={() => { + invalidateQualityPresentation(); + setFrameProofEpoch((current) => current + 1); const binding = remoteMediaRef.current; if (binding) { dispatchPresentation({ @@ -1933,6 +2068,8 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { }); } }} + onPause={invalidateQualityPresentation} + onEnded={invalidateQualityPresentation} /> {presentation.overlay === "blocking" && (
diff --git a/src/client/types.ts b/src/client/types.ts index 920aec4..e5e521c 100644 --- a/src/client/types.ts +++ b/src/client/types.ts @@ -45,6 +45,8 @@ export interface ConnectionMetrics { intervalFramesDropped: number | null; intervalFreezeCount: number | null; intervalFreezeDurationMs: number | null; + intervalPauseCount: number | null; + intervalPauseDurationMs: number | null; intervalRetransmittedPackets: number | null; intervalRetransmittedBytes: number | null; codec: string | null; @@ -125,6 +127,8 @@ export const EMPTY_METRICS: ConnectionMetrics = { intervalFramesDropped: null, intervalFreezeCount: null, intervalFreezeDurationMs: null, + intervalPauseCount: null, + intervalPauseDurationMs: null, intervalRetransmittedPackets: null, intervalRetransmittedBytes: null, codec: null, diff --git a/src/client/webrtc/stats.ts b/src/client/webrtc/stats.ts index c682880..4134216 100644 --- a/src/client/webrtc/stats.ts +++ b/src/client/webrtc/stats.ts @@ -26,6 +26,8 @@ export interface StatsAccumulator { previousFramesDropped: number | null; previousFreezeCount: number | null; previousTotalFreezesDuration: number | null; + previousPauseCount: number | null; + previousTotalPausesDuration: number | null; previousRetransmittedPackets: number | null; previousRetransmittedBytes: number | null; previousVideoJitterBufferDelay: number | null; @@ -92,6 +94,8 @@ export function createStatsAccumulator(): StatsAccumulator { previousFramesDropped: null, previousFreezeCount: null, previousTotalFreezesDuration: null, + previousPauseCount: null, + previousTotalPausesDuration: null, previousRetransmittedPackets: null, previousRetransmittedBytes: null, previousVideoJitterBufferDelay: null, @@ -493,6 +497,8 @@ export function collectConnectionMetricsFromReport( const framesDropped = numberValue(media, "framesDropped"); const freezeCount = numberValue(media, "freezeCount"); const totalFreezesDuration = numberValue(media, "totalFreezesDuration"); + const pauseCount = numberValue(media, "pauseCount"); + const totalPausesDuration = numberValue(media, "totalPausesDuration"); const retransmittedPackets = numberValue( media, direction === "send" @@ -609,6 +615,22 @@ export function collectConnectionMetricsFromReport( previous.previousTotalFreezesDuration, sampleWindowMs !== null, ); + const intervalPauseCount = + direction === "receive" + ? intervalDelta( + pauseCount, + previous.previousPauseCount, + sampleWindowMs !== null, + ) + : null; + const pauseDurationDelta = + direction === "receive" + ? intervalDelta( + totalPausesDuration, + previous.previousTotalPausesDuration, + sampleWindowMs !== null, + ) + : null; const intervalRetransmittedPackets = intervalDelta( retransmittedPackets, previous.previousRetransmittedPackets, @@ -643,6 +665,8 @@ export function collectConnectionMetricsFromReport( previous.previousFramesDropped = framesDropped; previous.previousFreezeCount = freezeCount; previous.previousTotalFreezesDuration = totalFreezesDuration; + previous.previousPauseCount = pauseCount; + previous.previousTotalPausesDuration = totalPausesDuration; previous.previousRetransmittedPackets = retransmittedPackets; previous.previousRetransmittedBytes = retransmittedBytes; previous.previousVideoJitterBufferDelay = videoJitterBufferDelay; @@ -840,6 +864,9 @@ export function collectConnectionMetricsFromReport( intervalFreezeCount, intervalFreezeDurationMs: freezeDurationDelta === null ? null : freezeDurationDelta * 1_000, + intervalPauseCount, + intervalPauseDurationMs: + pauseDurationDelta === null ? null : pauseDurationDelta * 1_000, intervalRetransmittedPackets, intervalRetransmittedBytes, codec: codecEvidence.codec, diff --git a/src/server/hybrid-media-router.ts b/src/server/hybrid-media-router.ts index 1b7259d..f8fa1e9 100644 --- a/src/server/hybrid-media-router.ts +++ b/src/server/hybrid-media-router.ts @@ -19,6 +19,8 @@ import { type CandidateReservation, type CandidateTuple, type OperationSnapshot, + type RouteQualityEvidenceInput, + type RouteQualityEvidenceResult, type RouteSnapshot, } from "./room-route-controller.js"; import type { RoomStore } from "./room-store.js"; @@ -263,6 +265,16 @@ export class HybridMediaRouter { ); } + observeQualityEvidence( + input: RouteQualityEvidenceInput & { roomId: string }, + ): RouteQualityEvidenceResult { + const { roomId, ...evidence } = input; + return ( + this.rooms.get(roomId)?.controller?.observeQualityEvidence(evidence) ?? + "rejected" + ); + } + isActivePeerParentOf( roomId: string, parentPeerId: string, diff --git a/src/server/room-route-controller.ts b/src/server/room-route-controller.ts index 059f458..537d44b 100644 --- a/src/server/room-route-controller.ts +++ b/src/server/room-route-controller.ts @@ -3,10 +3,12 @@ import { debuglog } from "node:util"; import { MAX_MEDIA_ROUTE_REVISION, + VIEWER_QUALITY_EVIDENCE_EXPIRY_MS, type RouteDemandReason, type RouteDiagnosticFinalRoute, type RouteDiagnosticRejectionBucket, type RouteDiagnosticSnapshot, + type ViewerQualityEvidenceMetrics, } from "../shared/protocol.js"; import { assertEndpointMediaCopyCapacity } from "../shared/media-copy-accounting.js"; @@ -221,6 +223,46 @@ interface RouteTimingRecord { rejectionBucket: RouteDiagnosticRejectionBucket; } +interface RouteQualityObservation { + childSessionId: string; + upstreamKind: "peer" | "sfu"; + upstreamPeerId: string | null; + connectionId: string; + presentationEpoch: number; + eligibleWindows: number; + eligibleDurationMs: number; + freezeWindows: number; + freezeCount: number; + freezeDurationMs: number; + pauseCount: number; + pauseDurationMs: number; + lastAcceptedAtMs: number; +} + +export interface RouteQualityEvidenceInput { + childPeerId: string; + childSessionId: string; + routeRevision: number; + connectionId: string; + upstream: { kind: "peer"; peerId: string } | { kind: "sfu" }; + presentationEpoch: number; + windowMs: number; + metrics: Pick< + ViewerQualityEvidenceMetrics, + | "framesDecodedDelta" + | "freezeCountDelta" + | "freezeDurationMsDelta" + | "pauseCountDelta" + | "pauseDurationMsDelta" + >; + acceptedAtMs: number; +} + +export type RouteQualityEvidenceResult = + | "rejected" + | "accepted" + | "observed"; + export class RoomRouteController { private readonly participants = new Map(); private readonly upstreamByViewer = new Map>(); @@ -232,6 +274,11 @@ export class RoomRouteController { private nextJoinOrder = 0; private paused = false; private readonly routeTimings = new Map(); + private readonly qualityObservations = new Map< + string, + RouteQualityObservation + >(); + private readonly qualityBaselinesPending = new Set(); private readonly directContinuations = new Map(); private sfuBootstrapIntent?: SfuBootstrapIntent; private sfuBootstrapExhaustedAtFactVersion?: number; @@ -311,6 +358,7 @@ export class RoomRouteController { : "waiting" : finalRoute, rejectionBucket: record?.rejectionBucket ?? "none", + quality: this.diagnosticQuality(viewer.peerId, edge, nowMs), }; }, ); @@ -348,6 +396,7 @@ export class RoomRouteController { let changed = false; const previousSessionId = current.sessionId; if (previousSessionId !== input.sessionId) { + this.clearQualityForParticipant(input.peerId); const revisionBefore = this.revision; if (this.operationUsesParticipantSession(input.peerId)) { released.push(...this.abortOperation(nowMs, "stale")); @@ -413,6 +462,7 @@ export class RoomRouteController { const participant = this.participants.get(peerId); if (!participant || participant.sessionId !== sessionId) return false; participant.sessionId = null; + this.clearQualityForParticipant(peerId); this.directContinuations.delete(peerId); this.clearSfuBootstrapForDemand(peerId); this.touchFacts(); @@ -429,6 +479,7 @@ export class RoomRouteController { if (participant.departureConfirmed && participant.sessionId === null && participant.effectiveDownstreamCapacity === 0) return true; participant.sessionId = null; + this.clearQualityForParticipant(peerId); participant.departureConfirmed = true; participant.effectiveDownstreamCapacity = 0; this.directContinuations.delete(peerId); @@ -521,6 +572,7 @@ export class RoomRouteController { return false; } if (edge.usable) { + this.clearQualityForParticipant(guard.childPeerId); edge.usable = false; this.directContinuations.delete(guard.childPeerId); child.blockedAtFactVersion = undefined; @@ -548,6 +600,7 @@ export class RoomRouteController { publication.generation !== guard.generation || publication.connectionId !== guard.connectionId) return false; if (publication.usable) { + this.clearSfuQuality(); publication.usable = false; this.directContinuations.clear(); this.sfuBootstrapIntent = undefined; @@ -582,6 +635,7 @@ export class RoomRouteController { return false; } if (input.newConnectionId === edge.connectionId) return true; + this.clearQualityForParticipant(input.childPeerId); edge.connectionId = input.newConnectionId; const child = this.participants.get(input.childPeerId); if (child) { @@ -603,6 +657,7 @@ export class RoomRouteController { } edge.physicalActive = false; edge.usable = false; + this.clearQualityForParticipant(guard.childPeerId); child.blockedAtFactVersion = undefined; this.revision = this.allocateRevision(); if (this.operation) this.operation.baseRevision = this.revision; @@ -624,6 +679,7 @@ export class RoomRouteController { publication.connectionId !== guard.connectionId) return []; publication.physicalActive = false; publication.usable = false; + this.clearSfuQuality(); this.revision = this.allocateRevision(); if (this.operation) this.operation.baseRevision = this.revision; this.touchFacts(); @@ -633,10 +689,126 @@ export class RoomRouteController { setPaused(paused: boolean, nowMs?: number): readonly Resource[] { if (this.paused === paused) return []; this.paused = paused; + if (paused) { + this.qualityObservations.clear(); + this.qualityBaselinesPending.clear(); + } this.touchFacts(); return paused ? this.abortOperation(nowMs, "aborted") : []; } + observeQualityEvidence( + input: RouteQualityEvidenceInput, + ): RouteQualityEvidenceResult { + if ( + this.paused || + input.routeRevision !== this.revision || + !Number.isSafeInteger(input.presentationEpoch) || + input.presentationEpoch < 0 || + !Number.isSafeInteger(input.windowMs) || + input.windowMs <= 0 || + !Number.isSafeInteger(input.acceptedAtMs) || + input.acceptedAtMs < 0 + ) { + return "rejected"; + } + const child = this.participants.get(input.childPeerId); + const edge = this.upstreamByViewer.get(input.childPeerId); + if ( + !child || + child.sessionId !== input.childSessionId || + !edge || + !edge.usable || + !edge.physicalActive || + edge.childSessionId !== input.childSessionId || + edge.connectionId !== input.connectionId || + !this.qualityUpstreamMatches(edge, input.upstream) || + !this.sourceUsableForQuality(edge) + ) { + return "rejected"; + } + + const upstreamPeerId = + input.upstream.kind === "peer" ? input.upstream.peerId : null; + let observation = this.qualityObservations.get(input.childPeerId); + const sameEdgeObservation = + observation?.childSessionId === input.childSessionId && + observation.upstreamKind === input.upstream.kind && + observation.upstreamPeerId === upstreamPeerId && + observation.connectionId === input.connectionId; + if ( + sameEdgeObservation && + observation && + input.presentationEpoch < observation.presentationEpoch + ) { + return "rejected"; + } + if ( + !sameEdgeObservation || + !observation || + observation.presentationEpoch !== input.presentationEpoch + ) { + observation = { + childSessionId: input.childSessionId, + upstreamKind: input.upstream.kind, + upstreamPeerId, + connectionId: input.connectionId, + presentationEpoch: input.presentationEpoch, + eligibleWindows: 0, + eligibleDurationMs: 0, + freezeWindows: 0, + freezeCount: 0, + freezeDurationMs: 0, + pauseCount: 0, + pauseDurationMs: 0, + lastAcceptedAtMs: input.acceptedAtMs, + }; + this.qualityObservations.set(input.childPeerId, observation); + } + + const metrics = input.metrics; + if ( + metrics.framesDecodedDelta === null || + metrics.framesDecodedDelta <= 0 || + metrics.freezeCountDelta === null || + metrics.freezeDurationMsDelta === null || + metrics.pauseCountDelta === null || + metrics.pauseDurationMsDelta === null + ) { + return "accepted"; + } + observation.lastAcceptedAtMs = input.acceptedAtMs; + if (this.qualityBaselinesPending.delete(input.childPeerId)) { + return "accepted"; + } + + observation.eligibleWindows = safeAdd(observation.eligibleWindows, 1); + observation.eligibleDurationMs = safeAdd( + observation.eligibleDurationMs, + input.windowMs, + ); + if (metrics.freezeCountDelta > 0) { + observation.freezeWindows = safeAdd(observation.freezeWindows, 1); + } + observation.freezeCount = safeAdd( + observation.freezeCount, + metrics.freezeCountDelta, + ); + observation.freezeDurationMs = safeAdd( + observation.freezeDurationMs, + metrics.freezeDurationMsDelta, + ); + observation.pauseCount = safeAdd( + observation.pauseCount, + metrics.pauseCountDelta, + ); + observation.pauseDurationMs = safeAdd( + observation.pauseDurationMs, + metrics.pauseDurationMsDelta, + ); + return "observed"; + } + reconcile(nowMs: number): ReconcileResult { const released: Resource[] = []; const validation = this.validateOrAdvance(nowMs); @@ -1164,6 +1336,8 @@ export class RoomRouteController { for (const resource of this.committedResources()) resources.add(resource); this.operation = undefined; this.upstreamByViewer.clear(); + this.qualityObservations.clear(); + this.qualityBaselinesPending.clear(); this.hostPublication = null; this.participants.clear(); this.routeTimings.clear(); @@ -1373,6 +1547,7 @@ export class RoomRouteController { private commitAttempt(operation: ChildOperation, attempt: Attempt): Resource[] { const beforeResources = new Set(this.committedResources()); const old = this.upstreamByViewer.get(operation.childPeerId); + this.clearQualityForParticipant(operation.childPeerId); if (attempt.tuple.kind === "sfu" && attempt.tuple.publication === "replace") { this.removePublicationGeneration(this.hostPublication!.generation); @@ -2017,6 +2192,7 @@ export class RoomRouteController { const edge = this.upstreamByViewer.get(participant.peerId); if (edge && edge.transport !== "direct" && edge.physicalActive) released.push(edge.resource); this.upstreamByViewer.delete(participant.peerId); + this.clearQualityForParticipant(participant.peerId); this.participants.delete(participant.peerId); this.routeTimings.delete(participant.peerId); this.directContinuations.delete(participant.peerId); @@ -2027,6 +2203,7 @@ export class RoomRouteController { } if (removed.length > 0) { if (!this.hasSfuSubscribers() && this.hostPublication) { + this.clearSfuQuality(); if (this.hostPublication.physicalActive) released.push(this.hostPublication.resource); this.hostPublication = null; } @@ -2408,6 +2585,99 @@ export class RoomRouteController { : { kind: "viewer", ordinal }; } + private diagnosticQuality( + childPeerId: string, + edge: CommittedEdge | undefined, + nowMs: number, + ): RouteDiagnosticSnapshot["children"][number]["quality"] { + const observation = this.qualityObservations.get(childPeerId); + const participant = this.participants.get(childPeerId); + if ( + this.paused || + !observation || + observation.eligibleWindows === 0 || + nowMs >= + observation.lastAcceptedAtMs + VIEWER_QUALITY_EVIDENCE_EXPIRY_MS || + !participant?.sessionId || + participant.sessionId !== observation.childSessionId || + !edge || + !edge.usable || + !edge.physicalActive || + edge.connectionId !== observation.connectionId || + !this.qualityObservationMatchesEdge(observation, edge) || + !this.sourceUsableForQuality(edge) + ) { + return null; + } + return { + eligibleWindows: observation.eligibleWindows, + eligibleDurationMs: observation.eligibleDurationMs, + freezeWindows: observation.freezeWindows, + freezeCount: observation.freezeCount, + freezeDurationMs: observation.freezeDurationMs, + pauseCount: observation.pauseCount, + pauseDurationMs: observation.pauseDurationMs, + }; + } + + private qualityUpstreamMatches( + edge: CommittedEdge, + upstream: RouteQualityEvidenceInput["upstream"], + ): boolean { + return edge.kind === "peer" + ? upstream.kind === "peer" && upstream.peerId === edge.parentPeerId + : upstream.kind === "sfu"; + } + + private qualityObservationMatchesEdge( + observation: RouteQualityObservation, + edge: CommittedEdge, + ): boolean { + return edge.kind === "peer" + ? observation.upstreamKind === "peer" && + observation.upstreamPeerId === edge.parentPeerId + : observation.upstreamKind === "sfu" && + observation.upstreamPeerId === null; + } + + private sourceUsableForQuality(edge: CommittedEdge): boolean { + if (edge.kind === "peer") { + return ( + edge.parentSessionId === + this.participants.get(edge.parentPeerId)?.sessionId && + this.sourceUsable(edge.parentPeerId) + ); + } + return Boolean( + this.hostPublication?.usable && + this.hostPublication.physicalActive && + this.hostPublication.generation === edge.publicationGeneration, + ); + } + + private clearQualityForParticipant(peerId: string): void { + const affected = + peerId === this.options.hostPeerId + ? new Set(this.upstreamByViewer.keys()) + : this.descendantsOf(peerId).add(peerId); + affected.forEach((childPeerId) => { + this.qualityObservations.delete(childPeerId); + if (this.upstreamByViewer.has(childPeerId)) { + this.qualityBaselinesPending.add(childPeerId); + } else { + this.qualityBaselinesPending.delete(childPeerId); + } + }); + } + + private clearSfuQuality(): void { + for (const [childPeerId, edge] of this.upstreamByViewer) { + if (edge.kind === "sfu") { + this.clearQualityForParticipant(childPeerId); + } + } + } + private recordDemand( childPeerId: string, nowMs: number, @@ -2550,6 +2820,7 @@ export class RoomRouteController { for (const [viewerPeerId, edge] of this.upstreamByViewer) { if (edge.kind !== "sfu" || edge.publicationGeneration !== generation) continue; if (edge.physicalActive) released.push(edge.resource); + this.clearQualityForParticipant(viewerPeerId); this.upstreamByViewer.delete(viewerPeerId); } if (this.hostPublication?.generation === generation) { @@ -2658,6 +2929,16 @@ function elapsedMs(startedAtMs: number, endedAtMs: number): number | null { return Number.isSafeInteger(elapsed) && elapsed >= 0 ? elapsed : null; } +function safeAdd(total: number, delta: number): number { + const rounded = Math.round(delta); + if (!Number.isSafeInteger(rounded) || rounded < 0) { + return total; + } + return rounded > Number.MAX_SAFE_INTEGER - total + ? Number.MAX_SAFE_INTEGER + : total + rounded; +} + function tupleKey(tuple: CandidateTuple): string { return tuple.kind === "peer" ? `peer:${tuple.parentPeerId}` : `sfu:${tuple.publication}`; } diff --git a/src/server/signaling.ts b/src/server/signaling.ts index db8d7d1..b950652 100644 --- a/src/server/signaling.ts +++ b/src/server/signaling.ts @@ -70,13 +70,11 @@ interface SocketState { interface ViewerQualityEvidenceGate { viewerSessionId: string; - recipientSessionId: string; upstreamKind: "peer" | "sfu"; upstreamPeerId: string | null; connectionId: string; - routeRevision: number; + presentationEpoch: number; sequence: number; - acceptedAtMs: number; } interface ViewerMediaReadyState { @@ -119,6 +117,7 @@ export class SignalingServer { string, ViewerQualityEvidenceGate >(); + private readonly viewerQualityEvidenceAttemptAtMs = new Map(); private readonly viewerMediaReadyByRoom = new Map< string, Map @@ -252,6 +251,7 @@ export class SignalingServer { routeCloseError = error; } this.viewerQualityEvidenceGates.clear(); + this.viewerQualityEvidenceAttemptAtMs.clear(); this.viewerMediaReadyByRoom.clear(); this.shareGenerationsByRoom.clear(); this.pausedShareGenerationsByRoom.clear(); @@ -608,9 +608,12 @@ export class SignalingServer { viewerPresence: message.viewerPresence === true, }; if (participant.role === "viewer") { - this.viewerQualityEvidenceGates.delete( - viewerConnectionKey(participant.roomId, participant.peerId), + const viewerKey = viewerConnectionKey( + participant.roomId, + participant.peerId, ); + this.viewerQualityEvidenceGates.delete(viewerKey); + this.viewerQualityEvidenceAttemptAtMs.delete(viewerKey); } this.clearViewerGrace(participant.roomId, participant.peerId); const routeParticipant = { @@ -1139,9 +1142,19 @@ export class SignalingServer { return; } - const connectionId = this.connectionIdsByViewer.get( - viewerConnectionKey(source.roomId, source.peerId), - ); + const gateKey = viewerConnectionKey(source.roomId, source.peerId); + const now = this.now(); + const previousAttemptAtMs = + this.viewerQualityEvidenceAttemptAtMs.get(gateKey); + if ( + previousAttemptAtMs !== undefined && + now - previousAttemptAtMs < VIEWER_QUALITY_EVIDENCE_INTERVAL_MS + ) { + return; + } + this.viewerQualityEvidenceAttemptAtMs.set(gateKey, now); + + const connectionId = this.connectionIdsByViewer.get(gateKey); if (!connectionId || message.guard.connectionId !== connectionId) { return; } @@ -1179,22 +1192,19 @@ export class SignalingServer { return; } - const gateKey = viewerConnectionKey(source.roomId, source.peerId); const previous = this.viewerQualityEvidenceGates.get(gateKey); - const sameGeneration = + const sameEdge = previous?.viewerSessionId === viewerState.sessionId && - previous.recipientSessionId === recipient.sessionId && previous.upstreamKind === upstream.kind && previous.upstreamPeerId === (upstream.kind === "peer" ? upstream.peerId : null) && - previous.connectionId === connectionId && - previous.routeRevision === routeRevision; - const now = this.now(); + previous.connectionId === connectionId; if ( - sameGeneration && + sameEdge && previous && - (message.sequence <= previous.sequence || - now - previous.acceptedAtMs < VIEWER_QUALITY_EVIDENCE_INTERVAL_MS) + (message.guard.presentationEpoch < previous.presentationEpoch || + (message.guard.presentationEpoch === previous.presentationEpoch && + message.sequence <= previous.sequence)) ) { return; } @@ -1209,29 +1219,42 @@ export class SignalingServer { guard: { connectionId, routeRevision, + presentationEpoch: message.guard.presentationEpoch, }, sequence: message.sequence, windowMs: message.windowMs, metrics: message.metrics, }; const encoded = JSON.stringify(forwarded); - if ( - Buffer.byteLength(encoded, "utf8") > - MAX_VIEWER_QUALITY_EVIDENCE_BYTES || - !this.sendEncodedToSession(recipient.sessionId, encoded) - ) { + if (Buffer.byteLength(encoded, "utf8") > MAX_VIEWER_QUALITY_EVIDENCE_BYTES) { + return; + } + const qualityResult = this.isHybridMediaEnabled() + ? (this.hybridMediaRouter?.observeQualityEvidence({ + roomId: source.roomId, + childPeerId: source.peerId, + childSessionId: viewerState.sessionId, + routeRevision, + connectionId, + upstream, + presentationEpoch: message.guard.presentationEpoch, + windowMs: message.windowMs, + metrics: message.metrics, + acceptedAtMs: now, + }) ?? "rejected") + : "accepted"; + if (qualityResult === "rejected") { return; } this.viewerQualityEvidenceGates.set(gateKey, { viewerSessionId: viewerState.sessionId, - recipientSessionId: recipient.sessionId, upstreamKind: upstream.kind, upstreamPeerId: upstream.kind === "peer" ? upstream.peerId : null, connectionId, - routeRevision, + presentationEpoch: message.guard.presentationEpoch, sequence: message.sequence, - acceptedAtMs: now, }); + this.sendEncodedToSession(recipient.sessionId, encoded); if (!host || host.sessionId === recipient.sessionId) { return; } @@ -1976,6 +1999,11 @@ export class SignalingServer { this.viewerQualityEvidenceGates.delete(key); } } + for (const key of this.viewerQualityEvidenceAttemptAtMs.keys()) { + if (key.startsWith(prefix)) { + this.viewerQualityEvidenceAttemptAtMs.delete(key); + } + } } private setViewerConnectionId( @@ -1986,6 +2014,7 @@ export class SignalingServer { const key = viewerConnectionKey(roomId, viewerPeerId); if (this.connectionIdsByViewer.get(key) !== connectionId) { this.viewerQualityEvidenceGates.delete(key); + this.viewerQualityEvidenceAttemptAtMs.delete(key); } this.connectionIdsByViewer.set(key, connectionId); } @@ -1997,6 +2026,7 @@ export class SignalingServer { const key = viewerConnectionKey(roomId, viewerPeerId); this.connectionIdsByViewer.delete(key); this.viewerQualityEvidenceGates.delete(key); + this.viewerQualityEvidenceAttemptAtMs.delete(key); } } diff --git a/src/shared/protocol.ts b/src/shared/protocol.ts index 57deb0f..e52d02c 100644 --- a/src/shared/protocol.ts +++ b/src/shared/protocol.ts @@ -6,7 +6,7 @@ import { isCanonicalVideoCodecEvidence } from "./video-codec-evidence.js"; export const MAX_VIEWERS_PER_ROOM_LIMIT = 20; export const MAX_PARTICIPANTS_PER_ROOM_LIMIT = MAX_VIEWERS_PER_ROOM_LIMIT + 1; export const MAX_SIGNAL_BYTES = 64 * 1024; -export const SIGNALING_PROTOCOL = "screener-v12"; +export const SIGNALING_PROTOCOL = "screener-v13"; export const SIGNAL_CLOSE_CODES = { serviceRestart: 1012, sessionReplaced: 4001, @@ -397,6 +397,22 @@ const routeDiagnosticParentSchema = z.discriminatedUnion("kind", [ }) .strict(), ]); +const routeDiagnosticQualityValueSchema = z + .number() + .int() + .min(0) + .max(Number.MAX_SAFE_INTEGER); +const routeDiagnosticQualitySchema = z + .object({ + eligibleWindows: routeDiagnosticQualityValueSchema, + eligibleDurationMs: routeDiagnosticQualityValueSchema, + freezeWindows: routeDiagnosticQualityValueSchema, + freezeCount: routeDiagnosticQualityValueSchema, + freezeDurationMs: routeDiagnosticQualityValueSchema, + pauseCount: routeDiagnosticQualityValueSchema, + pauseDurationMs: routeDiagnosticQualityValueSchema, + }) + .strict(); const routeDiagnosticChildSchema = z .object({ ordinal: routeDiagnosticOrdinalSchema, @@ -418,6 +434,7 @@ const routeDiagnosticChildSchema = z finalMs: routeDiagnosticDurationSchema, finalRoute: routeDiagnosticFinalRouteSchema, rejectionBucket: routeDiagnosticRejectionBucketSchema, + quality: routeDiagnosticQualitySchema.nullable(), }) .strict(); const routeDiagnosticOperationSchema = z @@ -524,6 +541,20 @@ const nullableSignedEvidenceNumber = (absoluteMaximum: number) => const nullableEvidenceInteger = (maximum: number) => z.number().int().min(0).max(maximum).nullable(); +const nullableSafeEvidenceNumber = z + .number() + .finite() + .min(0) + .max(Number.MAX_SAFE_INTEGER) + .nullable(); + +const nullableSafeEvidenceInteger = z + .number() + .int() + .min(0) + .max(Number.MAX_SAFE_INTEGER) + .nullable(); + export const viewerQualityEvidenceMetricsSchema = z .object({ width: z.number().int().min(1).max(16_384).nullable(), @@ -537,8 +568,10 @@ export const viewerQualityEvidenceMetricsSchema = z framesDecodedDelta: nullableEvidenceInteger(10_000), framesDroppedDelta: nullableEvidenceInteger(10_000), decodeMsPerFrame: nullableEvidenceNumber(60_000), - freezeCountDelta: nullableEvidenceInteger(10_000), - freezeDurationMsDelta: nullableEvidenceNumber(5_000), + freezeCountDelta: nullableSafeEvidenceInteger, + freezeDurationMsDelta: nullableSafeEvidenceNumber, + pauseCountDelta: nullableSafeEvidenceInteger, + pauseDurationMsDelta: nullableSafeEvidenceNumber, codec: z .string() .max(64) @@ -587,6 +620,11 @@ const viewerQualityEvidenceGuardSchema = z .object({ connectionId: opaqueIdSchema, routeRevision: mediaRouteRevisionSchema, + presentationEpoch: z + .number() + .int() + .min(0) + .max(Number.MAX_SAFE_INTEGER), }) .strict(); @@ -596,26 +634,13 @@ const viewerQualityEvidenceWindowShape = { metrics: viewerQualityEvidenceMetricsSchema, }; -function freezeFitsEvidenceWindow(value: { - windowMs: number; - metrics: ViewerQualityEvidenceMetrics; -}): boolean { - return ( - value.metrics.freezeDurationMsDelta === null || - value.metrics.freezeDurationMsDelta <= value.windowMs - ); -} - export const viewerQualityEvidenceMessageSchema = z .object({ type: z.literal("viewer-quality-evidence"), guard: viewerQualityEvidenceGuardSchema, ...viewerQualityEvidenceWindowShape, }) - .strict() - .refine(freezeFitsEvidenceWindow, { - message: "Viewer freeze duration exceeds its evidence window", - }); + .strict(); const authenticateMessageSchema = z.discriminatedUnion("role", [ z @@ -937,10 +962,7 @@ export const serverMessageSchema = z.union([ guard: viewerQualityEvidenceGuardSchema, ...viewerQualityEvidenceWindowShape, }) - .strict() - .refine(freezeFitsEvidenceWindow, { - message: "Viewer freeze duration exceeds its evidence window", - }), + .strict(), z .object({ type: z.literal("host-status"), diff --git a/tests/client-state.test.ts b/tests/client-state.test.ts index 7c63706..efcd204 100644 --- a/tests/client-state.test.ts +++ b/tests/client-state.test.ts @@ -1,6 +1,9 @@ import { afterEach, describe, expect, it, vi } from "vitest"; -import type { QualitySettings } from "../src/shared/protocol.ts"; +import { + SIGNALING_PROTOCOL, + type QualitySettings, +} from "../src/shared/protocol.ts"; import { ApiError, @@ -42,6 +45,7 @@ import { import { qualityEvidenceWindowFromMetrics } from "../src/client/media/viewer-quality-evidence.ts"; import { collectConnectionMetrics, + collectConnectionMetricsFromReport, createStatsAccumulator, } from "../src/client/webrtc/stats.ts"; @@ -887,7 +891,7 @@ describe("client signaling recovery policy", () => { socket.dispatchEvent(new Event("open")); receive(socket, { type: "authenticated", - protocol: "screener-v12", + protocol: "screener-v13", role: "host", peerId: "host_12345678", roomExpiresAt: null, @@ -923,7 +927,7 @@ describe("client signaling recovery policy", () => { }); receive(sockets[1]!, { type: "authenticated", - protocol: "screener-v12", + protocol: "screener-v13", role: "host", peerId: "host_12345678", roomExpiresAt: null, @@ -1035,7 +1039,7 @@ describe("client signaling recovery policy", () => { Object.defineProperty(event, "data", { value: JSON.stringify({ type: "authenticated", - protocol: "screener-v12", + protocol: SIGNALING_PROTOCOL, role: "host", peerId: "host_12345678", roomExpiresAt: null, @@ -1162,7 +1166,7 @@ describe("client signaling recovery policy", () => { socket.dispatchEvent(new Event("open")); receive(socket, { type: "authenticated", - protocol: "screener-v12", + protocol: "screener-v13", role: "host", peerId: "host_12345678", roomExpiresAt: null, @@ -1288,7 +1292,7 @@ describe("client signaling recovery policy", () => { sockets[0]!.dispatchEvent(new Event("open")); receive({ type: "authenticated", - protocol: "screener-v12", + protocol: "screener-v13", role: "viewer", peerId: "viewer_12345678", roomExpiresAt: null, @@ -1415,7 +1419,7 @@ describe("client signaling recovery policy", () => { JSON.parse(String(sockets[0]!.send.mock.calls[0]![0])), ).toMatchObject({ type: "authenticate", - protocol: "screener-v12", + protocol: "screener-v13", }); const message = new Event("message"); Object.defineProperty(message, "data", { value: payload }); @@ -2120,6 +2124,8 @@ describe("WebRTC stats parsing", () => { framesDropped: number, freezeCount: number, totalFreezesDuration: number, + pauseCount: number, + totalPausesDuration: number, retransmittedPacketsReceived: number, retransmittedBytesReceived: number, ) => @@ -2137,18 +2143,20 @@ describe("WebRTC stats parsing", () => { framesDropped, freezeCount, totalFreezesDuration, + pauseCount, + totalPausesDuration, retransmittedPacketsReceived, retransmittedBytesReceived, }, ], ]) as unknown as RTCStatsReport; const reports = [ - inbound(1_000, 101, 2, 1, 0.25, 3, 300), - inbound(2_000, 101, 5, 2, 0.75, 7, 900), - inbound(3_000, 202, 20, 8, 4, 30, 4_000), - inbound(4_000, 202, 22, 9, 4.25, 33, 4_600), - inbound(5_000, 202, 1, 0, 0.25, 2, 200), - inbound(6_000, 202, 4, 1, 0.5, 6, 1_000), + inbound(1_000, 101, 2, 1, 0.25, 0, 0, 3, 300), + inbound(2_000, 101, 5, 2, 2.75, 1, 5.5, 7, 900), + inbound(3_000, 202, 20, 8, 4, 3, 16, 30, 4_000), + inbound(4_000, 202, 22, 9, 6.25, 4, 21.5, 33, 4_600), + inbound(5_000, 202, 1, 0, 0.25, 0, 0.25, 2, 200), + inbound(6_000, 202, 4, 1, 2.5, 1, 6.25, 6, 1_000), ]; const connection = { getStats: async () => reports.shift()!, @@ -2168,6 +2176,8 @@ describe("WebRTC stats parsing", () => { intervalFramesDropped: null, intervalFreezeCount: null, intervalFreezeDurationMs: null, + intervalPauseCount: null, + intervalPauseDurationMs: null, intervalRetransmittedPackets: null, intervalRetransmittedBytes: null, }); @@ -2175,7 +2185,9 @@ describe("WebRTC stats parsing", () => { sampleWindowMs: 1_000, intervalFramesDropped: 3, intervalFreezeCount: 1, - intervalFreezeDurationMs: 500, + intervalFreezeDurationMs: 2_500, + intervalPauseCount: 1, + intervalPauseDurationMs: 5_500, intervalRetransmittedPackets: 4, intervalRetransmittedBytes: 600, }); @@ -2185,6 +2197,8 @@ describe("WebRTC stats parsing", () => { intervalFramesDropped: null, intervalFreezeCount: null, intervalFreezeDurationMs: null, + intervalPauseCount: null, + intervalPauseDurationMs: null, intervalRetransmittedPackets: null, intervalRetransmittedBytes: null, }); @@ -2192,7 +2206,9 @@ describe("WebRTC stats parsing", () => { sampleWindowMs: 1_000, intervalFramesDropped: 2, intervalFreezeCount: 1, - intervalFreezeDurationMs: 250, + intervalFreezeDurationMs: 2_250, + intervalPauseCount: 1, + intervalPauseDurationMs: 5_500, intervalRetransmittedPackets: 3, intervalRetransmittedBytes: 600, }); @@ -2201,6 +2217,8 @@ describe("WebRTC stats parsing", () => { intervalFramesDropped: null, intervalFreezeCount: null, intervalFreezeDurationMs: null, + intervalPauseCount: null, + intervalPauseDurationMs: null, intervalRetransmittedPackets: null, intervalRetransmittedBytes: null, }); @@ -2208,12 +2226,73 @@ describe("WebRTC stats parsing", () => { sampleWindowMs: 1_000, intervalFramesDropped: 3, intervalFreezeCount: 1, - intervalFreezeDurationMs: 250, + intervalFreezeDurationMs: 2_250, + intervalPauseCount: 1, + intervalPauseDurationMs: 6_000, intervalRetransmittedPackets: 4, intervalRetransmittedBytes: 800, }); }); + it("keeps unavailable pause counters unknown and receive-only", async () => { + const report = ( + type: "inbound-rtp" | "outbound-rtp", + timestamp: number, + ) => + new Map([ + [ + "video", + { + id: "video", + type, + timestamp, + kind: "video", + ssrc: 101, + ...(type === "inbound-rtp" + ? { bytesReceived: timestamp, framesDecoded: timestamp / 100 } + : { + bytesSent: timestamp, + framesEncoded: timestamp / 100, + pauseCount: timestamp / 1_000, + totalPausesDuration: timestamp / 1_000, + }), + }, + ], + ]) as unknown as RTCStatsReport; + + const receiveAccumulator = createStatsAccumulator(); + collectConnectionMetricsFromReport( + report("inbound-rtp", 1_000), + "receive", + receiveAccumulator, + ); + const receive = collectConnectionMetricsFromReport( + report("inbound-rtp", 2_000), + "receive", + receiveAccumulator, + ); + expect(receive).toMatchObject({ + intervalPauseCount: null, + intervalPauseDurationMs: null, + }); + + const sendAccumulator = createStatsAccumulator(); + collectConnectionMetricsFromReport( + report("outbound-rtp", 1_000), + "send", + sendAccumulator, + ); + const send = collectConnectionMetricsFromReport( + report("outbound-rtp", 2_000), + "send", + sendAccumulator, + ); + expect(send).toMatchObject({ + intervalPauseCount: null, + intervalPauseDurationMs: null, + }); + }); + it("derives a bounded inbound C window from adjacent RTP counters", async () => { const inbound = ( timestamp: number, diff --git a/tests/protocol.test.ts b/tests/protocol.test.ts index 1a9fa63..3efdfc3 100644 --- a/tests/protocol.test.ts +++ b/tests/protocol.test.ts @@ -50,6 +50,7 @@ const qualityEvidence = { guard: { connectionId: "connection_12345678", routeRevision: 0, + presentationEpoch: 0, }, sequence: 0, windowMs: 2_000, @@ -67,6 +68,8 @@ const qualityEvidence = { decodeMsPerFrame: 2.4, freezeCountDelta: 0, freezeDurationMsDelta: 0, + pauseCountDelta: 0, + pauseDurationMsDelta: 0, codec: "video/H264", codecProfile: "profile-level-id=42e01f", codecParameters: @@ -93,7 +96,7 @@ describe("client signaling protocol", () => { "utf8", ); - expect(SIGNALING_PROTOCOL).toBe("screener-v12"); + expect(SIGNALING_PROTOCOL).toBe("screener-v13"); expect(nativeWire).toMatch(/signalingProtocol\s*=\s*"screener-v6"/); }); @@ -754,6 +757,28 @@ describe("client signaling protocol", () => { it("accepts only strict, bounded viewer quality evidence", () => { expect(clientMessageSchema.safeParse(qualityEvidence).success).toBe(true); + expect( + clientMessageSchema.safeParse({ + ...qualityEvidence, + metrics: { + ...qualityEvidence.metrics, + freezeDurationMsDelta: 12_000, + pauseDurationMsDelta: 6_000, + }, + }).success, + ).toBe(true); + expect( + clientMessageSchema.safeParse({ + ...qualityEvidence, + metrics: { + ...qualityEvidence.metrics, + freezeCountDelta: null, + freezeDurationMsDelta: null, + pauseCountDelta: null, + pauseDurationMsDelta: null, + }, + }).success, + ).toBe(true); expect( Buffer.byteLength(JSON.stringify(qualityEvidence), "utf8"), ).toBeLessThanOrEqual(MAX_VIEWER_QUALITY_EVIDENCE_BYTES); @@ -776,7 +801,32 @@ describe("client signaling protocol", () => { }, { ...qualityEvidence, - metrics: { ...qualityEvidence.metrics, freezeDurationMsDelta: 2_001 }, + metrics: { ...qualityEvidence.metrics, freezeCountDelta: -1 }, + }, + { + ...qualityEvidence, + metrics: { + ...qualityEvidence.metrics, + freezeCountDelta: Number.MAX_SAFE_INTEGER + 1, + }, + }, + { + ...qualityEvidence, + metrics: { + ...qualityEvidence.metrics, + freezeDurationMsDelta: Number.MAX_SAFE_INTEGER + 1, + }, + }, + { + ...qualityEvidence, + metrics: { ...qualityEvidence.metrics, pauseCountDelta: -1 }, + }, + { + ...qualityEvidence, + metrics: { + ...qualityEvidence.metrics, + pauseDurationMsDelta: Number.MAX_SAFE_INTEGER + 1, + }, }, { ...qualityEvidence, @@ -947,7 +997,7 @@ describe("server signaling protocol", () => { } }); - it("accepts only the privacy-safe two-field route snapshot payload", () => { + it("accepts only the privacy-safe route snapshot payload", () => { const message = { type: "route-diagnostic-snapshot", snapshot: { @@ -964,6 +1014,15 @@ describe("server signaling protocol", () => { finalMs: 80, finalRoute: "direct", rejectionBucket: "none", + quality: { + eligibleWindows: 3, + eligibleDurationMs: 6_000, + freezeWindows: 1, + freezeCount: 1, + freezeDurationMs: 250, + pauseCount: 0, + pauseDurationMs: 0, + }, }, { ordinal: 2, @@ -977,6 +1036,7 @@ describe("server signaling protocol", () => { finalMs: null, finalRoute: "waiting", rejectionBucket: "sfu-admission", + quality: null, }, ], operation: { @@ -1006,6 +1066,36 @@ describe("server signaling protocol", () => { ], }, }, + { + ...message, + snapshot: { + ...message.snapshot, + children: [ + { + ...message.snapshot.children[0], + quality: { + ...message.snapshot.children[0].quality, + presentationEpoch: 1, + }, + }, + ], + }, + }, + { + ...message, + snapshot: { + ...message.snapshot, + children: [ + { + ...message.snapshot.children[0], + quality: { + ...message.snapshot.children[0].quality, + freezeCount: -1, + }, + }, + ], + }, + }, { ...message, snapshot: { @@ -1474,6 +1564,16 @@ describe("server signaling protocol", () => { upstream: { kind: "peer", peerId: "host_12345678" }, }; expect(serverMessageSchema.safeParse(forwarded).success).toBe(true); + expect( + serverMessageSchema.safeParse({ + ...forwarded, + metrics: { + ...forwarded.metrics, + freezeDurationMsDelta: 12_000, + pauseDurationMsDelta: 6_000, + }, + }).success, + ).toBe(true); expect( serverMessageSchema.safeParse({ ...forwarded, diff --git a/tests/room-route-controller.test.ts b/tests/room-route-controller.test.ts index 61e6581..432a498 100644 --- a/tests/room-route-controller.test.ts +++ b/tests/room-route-controller.test.ts @@ -1,5 +1,7 @@ import { describe, expect, it } from "vitest"; +import type { ViewerQualityEvidenceMetrics } from "../src/shared/protocol.ts"; + import { RoomRouteController, type CandidateGuard, @@ -124,7 +126,194 @@ function peerEdge( }; } +type RouteQualityMetrics = Pick< + ViewerQualityEvidenceMetrics, + | "framesDecodedDelta" + | "freezeCountDelta" + | "freezeDurationMsDelta" + | "pauseCountDelta" + | "pauseDurationMsDelta" +>; + +function qualityMetrics( + overrides: Partial = {}, +): RouteQualityMetrics { + return { + framesDecodedDelta: 120, + freezeCountDelta: 1, + freezeDurationMsDelta: 250, + pauseCountDelta: 0, + pauseDurationMsDelta: 0, + ...overrides, + }; +} + describe("RoomRouteController", () => { + it("keeps only fresh exact-edge quality shadow aggregates", () => { + const routes = controller(2); + addViewer(routes, A, 1, 0); + routes.reconcile(10); + commitCurrent(routes, 20, "a_quality_connection"); + const active = routes.snapshot(); + + expect( + routes.observeQualityEvidence({ + childPeerId: A, + childSessionId: `${A}_session`, + routeRevision: active.revision, + connectionId: "a_quality_connection", + upstream: { kind: "peer", peerId: HOST }, + presentationEpoch: 1, + windowMs: 2_000, + metrics: qualityMetrics(), + acceptedAtMs: 1_000, + }), + ).toBe("observed"); + expect(routes.snapshot()).toMatchObject({ + revision: active.revision, + factVersion: active.factVersion, + }); + addViewer(routes, B, 0, 1_100); + routes.reconcile(1_200); + commitCurrent(routes, 1_210, "b_quality_connection"); + expect(routes.routeDiagnosticSnapshot(1_300).children[0]?.quality).toEqual({ + eligibleWindows: 1, + eligibleDurationMs: 2_000, + freezeWindows: 1, + freezeCount: 1, + freezeDurationMs: 250, + pauseCount: 0, + pauseDurationMs: 0, + }); + expect(routes.routeDiagnosticSnapshot(5_999).children[0]?.quality).not.toBeNull(); + expect(routes.routeDiagnosticSnapshot(6_000).children[0]?.quality).toBeNull(); + + const nextRevision = routes.snapshot().revision; + expect( + routes.observeQualityEvidence({ + childPeerId: A, + childSessionId: `${A}_session`, + routeRevision: nextRevision, + connectionId: "a_quality_connection", + upstream: { kind: "peer", peerId: HOST }, + presentationEpoch: 2, + windowMs: 2_000, + metrics: qualityMetrics({ framesDecodedDelta: null }), + acceptedAtMs: 7_000, + }), + ).toBe("accepted"); + expect(routes.routeDiagnosticSnapshot(7_100).children[0]?.quality).toBeNull(); + expect( + routes.observeQualityEvidence({ + childPeerId: A, + childSessionId: `${A}_session`, + routeRevision: nextRevision, + connectionId: "a_quality_connection", + upstream: { kind: "peer", peerId: HOST }, + presentationEpoch: 2, + windowMs: 2_000, + metrics: qualityMetrics(), + acceptedAtMs: 7_100, + }), + ).toBe("observed"); + expect( + routes.observeQualityEvidence({ + childPeerId: A, + childSessionId: `${A}_session`, + routeRevision: nextRevision, + connectionId: "a_quality_connection", + upstream: { kind: "peer", peerId: HOST }, + presentationEpoch: 1, + windowMs: 2_000, + metrics: qualityMetrics(), + acceptedAtMs: 7_200, + }), + ).toBe("rejected"); + + routes.setPaused(true, 7_300); + expect(routes.routeDiagnosticSnapshot(7_301).children[0]?.quality).toBeNull(); + }); + + it("clears quality shadow for a relay subtree when its source changes", () => { + const routes = controller(2); + addViewer(routes, A, 2); + addViewer(routes, B, 2); + addViewer(routes, C, 0); + routes.hydrateEdge(A, peerEdge(HOST, "a_from_host")); + routes.hydrateEdge(B, peerEdge(A, "b_from_a")); + routes.hydrateEdge(C, peerEdge(B, "c_from_b")); + const revision = routes.snapshot().revision; + + for (const [childPeerId, parentPeerId, connectionId] of [ + [B, A, "b_from_a"], + [C, B, "c_from_b"], + ] as const) { + expect( + routes.observeQualityEvidence({ + childPeerId, + childSessionId: `${childPeerId}_session`, + routeRevision: revision, + connectionId, + upstream: { kind: "peer", peerId: parentPeerId }, + presentationEpoch: 1, + windowMs: 2_000, + metrics: qualityMetrics(), + acceptedAtMs: 1_000, + }), + ).toBe("observed"); + } + expect( + routes + .routeDiagnosticSnapshot(1_001) + .children.filter((child) => child.quality !== null), + ).toHaveLength(2); + + expect( + routes.adoptDirectConnection({ + childPeerId: A, + childSessionId: `${A}_session`, + parentSessionId: "host_session", + routeRevision: revision, + connectionId: "a_from_host", + newConnectionId: "a_from_host_recovered", + }), + ).toBe(true); + expect( + routes + .routeDiagnosticSnapshot(1_002) + .children.every((child) => child.quality === null), + ).toBe(true); + + for (const [acceptedAtMs, expected] of [ + [1_100, "accepted"], + [3_100, "observed"], + ] as const) { + for (const [childPeerId, parentPeerId, connectionId] of [ + [B, A, "b_from_a"], + [C, B, "c_from_b"], + ] as const) { + expect( + routes.observeQualityEvidence({ + childPeerId, + childSessionId: `${childPeerId}_session`, + routeRevision: revision, + connectionId, + upstream: { kind: "peer", peerId: parentPeerId }, + presentationEpoch: 1, + windowMs: 2_000, + metrics: qualityMetrics(), + acceptedAtMs, + }), + ).toBe(expected); + } + } + expect( + routes + .routeDiagnosticSnapshot(3_101) + .children.filter((child) => child.quality !== null), + ).toHaveLength(2); + }); + it("retains one latest timing sample for a 20-Viewer burst", () => { const routes = controller(2); const viewerPeerIds = Array.from( @@ -172,6 +361,7 @@ describe("RoomRouteController", () => { finalMs: null, finalRoute: "waiting", rejectionBucket: "none", + quality: null, }, ], operation: null, diff --git a/tests/server-signal.test.ts b/tests/server-signal.test.ts index dc38387..93046e2 100644 --- a/tests/server-signal.test.ts +++ b/tests/server-signal.test.ts @@ -406,10 +406,11 @@ function viewerQualityEvidenceMessage( connectionId: string, routeRevision: number, sequence = 0, + presentationEpoch = 0, ) { return { type: "viewer-quality-evidence" as const, - guard: { connectionId, routeRevision }, + guard: { connectionId, routeRevision, presentationEpoch }, sequence, windowMs: 2_000, metrics: { @@ -426,6 +427,8 @@ function viewerQualityEvidenceMessage( decodeMsPerFrame: 2.4, freezeCountDelta: 0, freezeDurationMsDelta: 0, + pauseCountDelta: 0, + pauseDurationMsDelta: 0, codec: "video/VP8", codecProfile: null, codecParameters: null, @@ -896,7 +899,8 @@ describe("WebSocket signaling", () => { }); it("forwards exact direct Viewer receive evidence to the Host", async () => { - const harness = await startHarness(); + let now = 10_000; + const harness = await startHarness({ now: () => now }); const host = await openClient(harness.webSocketUrl); const hostAuth = await authenticate( host, @@ -914,6 +918,7 @@ describe("WebSocket signaling", () => { "viewer", "evidence-direct-viewer", ); + host.inbox.ignore("viewer-presence"); await host.inbox.next("peer-joined"); const connectionId = "evidence_direct_connection_12345678"; host.socket.send( @@ -937,12 +942,44 @@ describe("WebSocket signaling", () => { upstream: { kind: "peer", peerId: hostAuth.peerId }, guard: { connectionId, routeRevision: 0 }, }); + + viewer.socket.send( + JSON.stringify(viewerQualityEvidenceMessage(connectionId, 0, 0, 1)), + ); + await host.inbox.expectNone(40); + now += 2_000; + viewer.socket.send( + JSON.stringify(viewerQualityEvidenceMessage(connectionId, 0, 7, 1)), + ); + expect(await host.inbox.next("viewer-quality-evidence")).toMatchObject({ + guard: { connectionId, routeRevision: 0, presentationEpoch: 1 }, + sequence: 7, + }); + now += 2_000; + viewer.socket.send( + JSON.stringify(viewerQualityEvidenceMessage(connectionId, 0, 1, 0)), + ); + await host.inbox.expectNone(40); + viewer.socket.send( + JSON.stringify(viewerQualityEvidenceMessage(connectionId, 0, 0, 2)), + ); + await host.inbox.expectNone(40); + now += 2_000; + viewer.socket.send( + JSON.stringify(viewerQualityEvidenceMessage(connectionId, 0, 9, 2)), + ); + expect(await host.inbox.next("viewer-quality-evidence")).toMatchObject({ + guard: { connectionId, routeRevision: 0, presentationEpoch: 2 }, + sequence: 9, + }); }); it("forwards a relayed child receive report to its exact parent and Host", async () => { + let now = 10_000; const harness = await startHarness({ peerAssistedMedia: true, endpointMediaCopyCapacity: 1, + now: () => now, }); const host = await openClient(harness.webSocketUrl); await authenticate( @@ -1020,6 +1057,45 @@ describe("WebSocket signaling", () => { expect(await host.inbox.next("viewer-quality-evidence")).toMatchObject( expected, ); + host.socket.send(JSON.stringify({ type: "request-route-diagnostic" })); + const diagnostic = await host.inbox.next("route-diagnostic-snapshot"); + expect( + diagnostic.snapshot.children.find((entry) => entry.quality !== null) + ?.quality, + ).toEqual({ + eligibleWindows: 1, + eligibleDurationMs: 2_000, + freezeWindows: 0, + freezeCount: 0, + freezeDurationMs: 0, + pauseCount: 0, + pauseDurationMs: 0, + }); + + now += 2_000; + const incomplete = viewerQualityEvidenceMessage( + childPrepare.candidate.connectionId, + childPrepare.revision, + 1, + ); + child.socket.send( + JSON.stringify({ + ...incomplete, + metrics: { ...incomplete.metrics, pauseCountDelta: null }, + }), + ); + expect(await parent.inbox.next("viewer-quality-evidence")).toMatchObject({ + metrics: { pauseCountDelta: null }, + }); + expect(await host.inbox.next("viewer-quality-evidence")).toMatchObject({ + metrics: { pauseCountDelta: null }, + }); + host.socket.send(JSON.stringify({ type: "request-route-diagnostic" })); + const unchanged = await host.inbox.next("route-diagnostic-snapshot"); + expect( + unchanged.snapshot.children.find((entry) => entry.quality !== null) + ?.quality?.eligibleWindows, + ).toBe(1); }); it("shares an opted-in Host name with Host and Viewer roster subscribers", async () => { @@ -2908,7 +2984,7 @@ describe("WebSocket signaling", () => { expect(await closeCode).toBe(1009); }); - it("terminates stale v10 before authentication", async () => { + it("terminates stale v12 before authentication", async () => { const harness = await startHarness(); const invalidClient = await openClient(harness.webSocketUrl); const closed = new Promise<{ code: number; reason: string }>((resolve) => @@ -2920,7 +2996,7 @@ describe("WebSocket signaling", () => { invalidClient.socket.send( JSON.stringify({ type: "authenticate", - protocol: "screener-v10", + protocol: "screener-v12", roomId: harness.room.roomId, role: "viewer", clientId: "invalid-client", diff --git a/tests/viewer-presentation.test.ts b/tests/viewer-presentation.test.ts index f1354b1..c9005e1 100644 --- a/tests/viewer-presentation.test.ts +++ b/tests/viewer-presentation.test.ts @@ -370,6 +370,50 @@ describe("Viewer presentation reducer", () => { }); }); + it("requires a new composited frame after presentation proof resets", () => { + const playing = apply( + { type: "access", access: "ready" }, + { type: "signal", signal: "connected" }, + { type: "host", host: "online" }, + { type: "route", revision: 3, phase: "active", kind: "p2p" }, + { type: "media-bound", generation: 2, revision: 3 }, + { type: "frame-presented", generation: 2, revision: 3 }, + ); + const reset = reduceViewerPresentation(playing, { + type: "frame-proof-reset", + generation: 2, + revision: 3, + }); + + expect(deriveViewerPresentation(reset)).toMatchObject({ + hasCurrentFrame: false, + hasRetainedFrame: true, + }); + expect( + deriveViewerPresentation( + reduceViewerPresentation(reset, { + type: "connection", + revision: 3, + connection: "reconnecting", + }), + ).stage, + ).toBe("recovering"); + expect( + reduceViewerPresentation(reset, { + type: "frame-proof-reset", + generation: 1, + revision: 3, + }), + ).toBe(reset); + expect( + reduceViewerPresentation(reset, { + type: "frame-presented", + generation: 2, + revision: 3, + }).media?.framePresented, + ).toBe(true); + }); + it("demotes the current frame when an exact route terminally fails", () => { const playing = apply( { type: "access", access: "ready" }, @@ -394,18 +438,40 @@ describe("Viewer presentation reducer", () => { failureCode: "ROUTE_EXHAUSTED", }); - const stillPlaying = reduceViewerPresentation(failed, { + const lateFrame = reduceViewerPresentation(failed, { type: "frame-presented", generation: 3, revision: 4, }); - expect(deriveViewerPresentation(stillPlaying)).toMatchObject({ - stage: "playing", - overlay: "none", - hasCurrentFrame: true, - hasRetainedFrame: false, - failureCode: null, + expect(lateFrame).toBe(failed); + expect( + reduceViewerPresentation(failed, { + type: "media-bound", + generation: 4, + revision: 4, + }), + ).toBe(failed); + expect( + reduceViewerPresentation(failed, { + type: "route", + revision: 4, + phase: "active", + kind: "sfu", + }).routeStatus, + ).toEqual({ revision: 4, state: "failed" }); + + const localFailure = reduceViewerPresentation(playing, { + type: "failure", + failure: "ROUTE_EXHAUSTED", + revision: 4, }); + expect( + reduceViewerPresentation(localFailure, { + type: "media-bound", + generation: 4, + revision: 4, + }), + ).toBe(localFailure); }); it("honors terminal access, Host pause, and typed route status priority", () => { diff --git a/tests/viewer-quality-evidence.test.ts b/tests/viewer-quality-evidence.test.ts index 8a4b2b4..ee2b5ab 100644 --- a/tests/viewer-quality-evidence.test.ts +++ b/tests/viewer-quality-evidence.test.ts @@ -55,6 +55,8 @@ function receiveMetrics( intervalDecodeMs: 2.4, intervalFreezeCount: 0, intervalFreezeDurationMs: 0, + intervalPauseCount: 0, + intervalPauseDurationMs: 0, codec: "video/H264", codecProfile: "profile-level-id=42e01f", codecParameters: @@ -84,6 +86,7 @@ function serverEvidence( upstream?: { kind: "peer"; peerId: string } | { kind: "sfu" }; connectionId?: string; routeRevision?: number; + presentationEpoch?: number; sequence?: number; metrics?: Partial; } = {}, @@ -99,6 +102,7 @@ function serverEvidence( guard: { connectionId: overrides.connectionId ?? "connection_12345678", routeRevision: overrides.routeRevision ?? 0, + presentationEpoch: overrides.presentationEpoch ?? 0, }, sequence: overrides.sequence ?? 0, windowMs: window.windowMs, @@ -124,6 +128,8 @@ describe("viewer quality evidence", () => { decodeMsPerFrame: 2.4, freezeCountDelta: 0, freezeDurationMsDelta: 0, + pauseCountDelta: 0, + pauseDurationMsDelta: 0, codec: "video/H264", codecProfile: "profile-level-id=42e01f", codecParameters: @@ -189,7 +195,7 @@ describe("viewer quality evidence", () => { expect(JSON.stringify(bounded)).not.toContain("203.0.113.20"); }); - it("rate-limits, deduplicates, and resets sequence per connection", () => { + it("baselines each presentation epoch and keeps rate limits across it", () => { let now = 10_000; const sent: ClientMessage[] = []; const reporter = new ViewerQualityEvidenceReporter( @@ -201,48 +207,102 @@ describe("viewer quality evidence", () => { ); const first = snapshot("connection_first_12345678"); - expect(reporter.offer(first, 0)).toBe(true); - expect(reporter.offer(first, 0)).toBe(false); - now += 1_000; + expect(reporter.offer(first, 0, true)).toBe(false); + expect(reporter.offer(first, 0, true)).toBe(false); + now += 2_000; expect( reporter.offer( snapshot( first.connectionId, - receiveMetrics({ sampleTimestampMs: 11_000, bitrateKbps: 7_000 }), - ), - 0, - ), - ).toBe(false); - now += 1_000; - expect( - reporter.offer( - snapshot( - first.connectionId, - receiveMetrics({ sampleTimestampMs: 12_000, bitrateKbps: 6_500 }), + receiveMetrics({ sampleTimestampMs: 12_000, bitrateKbps: 7_000 }), ), 0, + true, ), ).toBe(true); + now += 2_000; expect( reporter.offer( snapshot( first.connectionId, - receiveMetrics({ sampleTimestampMs: 12_000, bitrateKbps: 6_000 }), + receiveMetrics({ sampleTimestampMs: 14_000, bitrateKbps: 6_500 }), ), 2, + true, ), ).toBe(true); expect( - reporter.offer(snapshot("connection_second_12345678"), 2), + reporter.offer( + snapshot( + first.connectionId, + receiveMetrics({ sampleTimestampMs: 14_000, bitrateKbps: 6_000 }), + ), + 2, + true, + ), + ).toBe(false); + expect( + reporter.offer( + snapshot( + "connection_second_12345678", + receiveMetrics({ sampleTimestampMs: 16_000 }), + ), + 2, + true, + ), + ).toBe(false); + now += 2_000; + expect( + reporter.offer( + snapshot( + "connection_second_12345678", + receiveMetrics({ sampleTimestampMs: 18_000 }), + ), + 2, + true, + ), ).toBe(true); now += 2_000; expect( reporter.offer( snapshot( "connection_second_12345678", - receiveMetrics({ sampleTimestampMs: 14_000 }), + receiveMetrics({ sampleTimestampMs: 20_000 }), ), 2, + false, + ), + ).toBe(false); + expect( + reporter.offer( + snapshot( + "connection_second_12345678", + receiveMetrics({ sampleTimestampMs: 20_000 }), + ), + 2, + true, + ), + ).toBe(false); + now += 2_000; + expect( + reporter.offer( + snapshot( + "connection_second_12345678", + receiveMetrics({ sampleTimestampMs: 22_000 }), + ), + 2, + true, + ), + ).toBe(false); + now += 2_000; + expect( + reporter.offer( + snapshot( + "connection_second_12345678", + receiveMetrics({ sampleTimestampMs: 24_000 }), + ), + 2, + true, ), ).toBe(true); @@ -252,6 +312,7 @@ describe("viewer quality evidence", () => { ? { connectionId: message.guard.connectionId, revision: message.guard.routeRevision, + epoch: message.guard.presentationEpoch, sequence: message.sequence, } : null, @@ -260,27 +321,26 @@ describe("viewer quality evidence", () => { { connectionId: "connection_first_12345678", revision: 0, + epoch: 0, sequence: 0, }, - { - connectionId: "connection_first_12345678", - revision: 0, - sequence: 1, - }, { connectionId: "connection_first_12345678", revision: 2, - sequence: 2, + epoch: 0, + sequence: 1, }, { connectionId: "connection_second_12345678", revision: 2, + epoch: 0, sequence: 0, }, { connectionId: "connection_second_12345678", revision: 2, - sequence: 1, + epoch: 1, + sequence: 0, }, ]); }); @@ -297,6 +357,15 @@ describe("viewer quality evidence", () => { "sfu_connection_12345678", receiveMetrics(), 3, + true, + ), + ).toBe(false); + expect( + reporter.offerMetrics( + "sfu_connection_12345678", + receiveMetrics({ sampleTimestampMs: 12_000 }), + 3, + true, ), ).toBe(true); expect(sent).toEqual([ @@ -305,12 +374,87 @@ describe("viewer quality evidence", () => { guard: { connectionId: "sfu_connection_12345678", routeRevision: 3, + presentationEpoch: 0, }, sequence: 0, }), ]); }); + it("reports a recovered freeze only after decoded-frame progress resumes", () => { + let now = 10_000; + const sent: ClientMessage[] = []; + const reporter = new ViewerQualityEvidenceReporter( + (message) => { + sent.push(message); + return true; + }, + () => now, + ); + const connectionId = "decoded_progress_connection_12345678"; + + expect(reporter.offer(snapshot(connectionId), 4, true)).toBe(false); + now += 2_000; + expect( + reporter.offer( + snapshot( + connectionId, + receiveMetrics({ sampleTimestampMs: 12_000 }), + ), + 4, + true, + ), + ).toBe(true); + now += 2_000; + expect( + reporter.offer( + snapshot( + connectionId, + receiveMetrics({ + sampleTimestampMs: 14_000, + intervalFramesDecoded: 0, + intervalFreezeCount: 1, + intervalFreezeDurationMs: 2_000, + }), + ), + 4, + true, + ), + ).toBe(false); + now += 2_000; + expect( + reporter.offer( + snapshot( + connectionId, + receiveMetrics({ + sampleTimestampMs: 16_000, + intervalFreezeCount: 1, + intervalFreezeDurationMs: 2_000, + }), + ), + 4, + true, + ), + ).toBe(true); + + expect( + sent.flatMap((message) => + message.type === "viewer-quality-evidence" + ? [ + { + epoch: message.guard.presentationEpoch, + sequence: message.sequence, + freezeDurationMs: message.metrics.freezeDurationMsDelta, + }, + ] + : [], + ), + ).toEqual([ + { epoch: 0, sequence: 0, freezeDurationMs: 0 }, + { epoch: 0, sequence: 1, freezeDurationMs: 2_000 }, + ]); + }); + it("pairs and renders only the current upstream connection", () => { const evidence = { type: "viewer-quality-evidence", @@ -319,6 +463,7 @@ describe("viewer quality evidence", () => { guard: { connectionId: "connection_12345678", routeRevision: 0, + presentationEpoch: 0, }, sequence: 3, ...qualityEvidenceWindowFromMetrics(receiveMetrics())!, @@ -525,6 +670,21 @@ describe("viewer quality evidence", () => { ); expect(changedRoute.evidence.metrics.codec).toBe("video/H264"); + const changedPresentation = presentViewerQualityEvidence( + first, + serverEvidence({ + sequence: 0, + presentationEpoch: 1, + metrics: { + codec: null, + codecProfile: null, + codecParameters: null, + }, + }), + 2_000, + ); + expect(changedPresentation.evidence.metrics.codec).toBeNull(); + const changedConnection = presentViewerQualityEvidence( first, serverEvidence({