From a949fbaf159d2f8441718aff10a5bf14d34b0a6d Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Wed, 19 Aug 2026 20:39:56 +0800 Subject: [PATCH] feat(quality): add authenticated viewer evidence --- docs/project-memory.md | 6 +- docs/research/realtime-quality-adaptation.md | 27 +- docs/status.md | 13 +- src/client/media/viewer-quality-evidence.ts | 250 ++++++++++++++ src/client/pages/HostPage.tsx | 128 ++++++- src/client/pages/ViewerPage.tsx | 102 ++++++ src/client/types.ts | 10 + src/client/webrtc/stats.ts | 163 +++------ src/server/hybrid-media-router.ts | 32 +- src/server/signaling.ts | 188 +++++++++- src/shared/protocol.ts | 106 ++++++ src/shared/video-codec-evidence.ts | 166 +++++++++ tests/client-state.test.ts | 113 +++++- tests/protocol.test.ts | 131 +++++++ tests/server-signal.test.ts | 343 +++++++++++++++++++ tests/viewer-quality-evidence.test.ts | 247 +++++++++++++ 16 files changed, 1862 insertions(+), 163 deletions(-) create mode 100644 src/client/media/viewer-quality-evidence.ts create mode 100644 src/shared/video-codec-evidence.ts create mode 100644 tests/viewer-quality-evidence.test.ts diff --git a/docs/project-memory.md b/docs/project-memory.md index adbe22a..57c11cb 100644 --- a/docs/project-memory.md +++ b/docs/project-memory.md @@ -25,7 +25,7 @@ Last updated: 2026-08-19 - Keep the current failure-only controller default-off until config migration and exact-room gates pass. Preserve break-before-make, sticky healthy subtrees, mobile leaves, and PR #20 prewarm; public transport/load remains open. - Local reparenting is a later candidate: start with unassigned relay admission rescue; do not block current work. - Flagship media defaults to UDP; HTTPS/WSS remains TLS/TCP. Current production still requires coturn UDP/TCP. Future TURN absence is normal, partial config fails, and any media-TCP/port compatibility mode is chosen only by canary. -- Treat settings as ceilings and deployed degradation as unclassified. Use local Host A+B to compare exact `769de201f7cc` against current `main` for Host-only recovery, game FPS/load, preview cost, actual codec/encoder and one-variable rebuilds; then add authenticated C and test standard simulcast/LiveKit/SVC before custom `LOW`. Never use UA or a composite score. +- Treat settings as ceilings and deployed degradation as unclassified. Use correlated Host A+B/Viewer C to compare exact `769de201f7cc` against current `main` for Host-only recovery, game FPS/load, preview cost, actual codec/encoder and one-variable rebuilds; then test standard simulcast/LiveKit/SVC before custom `LOW`. Never use UA or a composite score. - Production reports poor film audio and self-echo when system capture includes voice software. Diagnose audio A/B/C and sync; Web cannot isolate arbitrary processes and `maxBitrate` is not quality-up. A Windows 11 native candidate defaults to game-process-tree audio and never widens silently; Windows 10 remains unresolved/unsupported. See `docs/research/browser-screen-audio-quality.md`. - Do not add scene detection, dynamic-FPS control, or forced AV1 until stats and target hardware prove the need. Codec acceptance requires actual negotiation, power-efficient candidate evidence, interval encode cost, game FPS, CPU/GPU and sender count; Discord's native capture/hardware tuning is comparison evidence, not proof of server re-encoding or a reusable preset. - Keep room policy deployment-driven: public/password-only rooms are random and temporary; password plus SQLite enables sequential persistent rooms and reusable stopped links. @@ -45,9 +45,9 @@ Last updated: 2026-08-19 - Draft native ladder #16/#18/#22/#23/#25/#28 has one bounded Chrome 720p30 loop: one WebCodecs object feeds two Pion/browser legs, applies an experiment-only minimum stock-GCC target, and recovers one isolated loss. It does not prove one hardware encode. - Stock Pion GCC plus negotiated RTX is a no-go. The no-RTX candidate can distort loss statistics and stays outside product code until audio, heterogeneous estimates, broader loss, TURN/reconnect, and browser-diversity gates pass; do not add custom congestion control to bypass them. - Chrome 151 short synthetic `1/3/5/8` runs kept host edges at two, relay edges at one, and all viewers decoding; a three-viewer relay close recovered in 5.32 seconds. This proves topology/control only, not quality, load, endurance, or silent partitions. -- The current quality stack uses one strict, non-persistent `QualitySettings` value with bounded ceilings and clarity/balanced/fluid preference. Sender readback and limitation warnings remain manual diagnostics. ADR-0007's automatic per-path `HIGH`/`FALLBACK` and unique on-demand shared `LOW` are accepted mainline behavior, but the C report, state controller, and `LOW` runtime are unimplemented. +- The current quality stack uses one strict, non-persistent `QualitySettings` value with bounded ceilings and clarity/balanced/fluid preference. Sender readback and limitation warnings remain manual diagnostics. ADR-0007's automatic per-path `HIGH`/`FALLBACK` and unique on-demand shared `LOW` are accepted mainline behavior, but the state controller and `LOW` runtime are unimplemented. - Chrome 151 synthetic 720p30 propagated balanced/clarity to one host and three viewers without changing peer identities; all viewers kept decoding and fanout stayed host two/relay one. This proves control continuity only. -- Local Host A+B aligns current capture settings with one unique outbound RTP/remote/transport pair per tick, rebases deltas on ambiguity, identity change, or reset, and rejects stale source generations. Codec evidence is nullable and transport-bound; no defaults or profile capability are inferred, raw fmtp/SDP/stats are not retained, and multiple RTP/encodings remain unknown. Authenticated C, automatic control, and `LOW` remain absent. +- Local Host A+B aligns capture with one unique outbound RTP/remote/transport pair and rejects ambiguous or stale generations. Authenticated Viewer C now adds a nullable, sanitized two-second receive window of at most 2 KiB for each current ordinary or peer-assisted P2P hop; the server derives identity and current connection/revision, and the parent correlates only matching local B. It retains no raw fmtp/SDP/stats, performs no media action, and fails closed for SFU-fed roots. Automatic control and `LOW` remain absent. - Release `769de201f7cc` passed CI/hygiene/typecheck, 22 files/274 tests, both builds, zero production dependency vulnerabilities, atomic activation, public health/access, exact bundle, and service/log checks. ## Provisional Quality Targets diff --git a/docs/research/realtime-quality-adaptation.md b/docs/research/realtime-quality-adaptation.md index d68724a..b7bdd52 100644 --- a/docs/research/realtime-quality-adaptation.md +++ b/docs/research/realtime-quality-adaptation.md @@ -164,20 +164,28 @@ This is a diagnostic classification, not a weighted health score. Missing or reset evidence remains unknown and rebases the interval. The smallest implementation sequence is local A+B correlation in one host -sampling tick, including the interval, media/stat identity, and valid deltas. -Only after that is trustworthy should a minimal authenticated C report carry -the receive/decode and derived negotiation signals needed by the two-state +sampling tick, followed by a minimal authenticated C report carrying the +receive/decode and derived negotiation signals needed by the two-state predicate. It never carries raw SDP, raw stats, candidate addresses, or raw -device/network identifiers. Opaque server-issued path and connection-generation -IDs provide authorization and correlation; a general remote stats stream or +device/network identifiers. Server-authoritative path and connection generations +provide authorization and correlation; a general remote stats stream or telemetry pipeline is unnecessary. -The repository now implements the local host A+B foundation: same-tick capture +The repository implements the local host A+B foundation: same-tick capture settings plus one uniquely matched outbound RTP sample, explicit sample/media identity and adjacent deltas, `remoteId` linkage, and the selected path reached through that RTP stream's transport. Source replacement blocks sampling and invalidates in-flight generations. +Authenticated Viewer C is also implemented for each current ordinary or +peer-assisted P2P hop. A viewer sends one nullable, sanitized aggregate window +on the existing two-second stats cadence, bounded to 2 KiB; the server derives +the current viewer, parent, connection, and route revision, while the parent +accepts only the matching local generation and expires it. The report contains +no raw SDP, stats, candidates, addresses, device identifiers, or room identity, +is neither stored nor used for media or routing action, and fails closed for an +SFU-fed root until a real SFU last-hop B/generation exists. + Codec evidence follows only that outbound RTP object's `codecId`, and the referenced `RTCCodecStats` must use the same transport. The local diagnostics expose nullable, bounded MIME plus a normalized profile token and an ASCII allowlist of @@ -195,9 +203,10 @@ The same unique outbound object supplies nullable current configured `scalabilityMode`. Sender parameter readback exposes an applied mode only for one unambiguous encoding. Current quality settings do not request a mode, so the requested value remains null and a browser-reported default is not called -a mismatch; multiple encodings remain unknown. This still does not implement -an authenticated C report, two-state controller, or on-demand `LOW` runtime, -and browser support remains subject to the controlled matrix. +a mismatch; multiple encodings remain unknown. Inbound stats provide no current +standard `scalabilityMode` source, so C does not carry a null-only placeholder. +The two-state controller and on-demand `LOW` runtime remain unimplemented, and +browser support remains subject to the controlled matrix. Official W3C text checked 2026-08-19 defines names ending in `Id` as stats-object references. In particular, outbound [`mediaSourceId`](https://www.w3.org/TR/webrtc-stats/#dom-rtcoutboundrtpstreamstats-mediasourceid) diff --git a/docs/status.md b/docs/status.md index b4a7d25..45c3eb4 100644 --- a/docs/status.md +++ b/docs/status.md @@ -17,7 +17,7 @@ unconfigured, so production remains one host connection per viewer and P2P/TURN. - SFU normally feeds one or two roots. Downstream cap is two, browser relay one; exceptional central edges need a separate cap. Production routing is off. - Configured fallback adds a non-secret standby URL and token-free DNS/TLS warmup; absent configuration adds no field, import, request, participant, or edge. - Each viewer advertises relay capacity 0/1; detected mobile/iPad clients are leaves. Withdrawal does not move a healthy edge. Browser relays stay one-child and host two-child. -- Production has memory-only video settings and manual readback. Automatic `HIGH`/`FALLBACK` plus one shared on-demand `LOW` is accepted but not deployed; host A+B exists, while C/actions do not. Browser audio is request/presence-only. +- Production has memory-only video settings and manual readback. Automatic `HIGH`/`FALLBACK` plus one shared on-demand `LOW` is accepted but not deployed; local A+B and authenticated P2P Viewer C are read-only evidence. Browser audio is request/presence-only. ## Verified Evidence @@ -26,7 +26,7 @@ unconfigured, so production remains one host connection per viewer and P2P/TURN. - Synthetic Chrome 151 720p30 propagated balanced/clarity to one host and three viewers without peer changes; decoding and 2/1 fanout continued. This proves control continuity only. - Chrome 151/LiveKit localhost A/B measured cold active/render at 1.481/2.257 seconds versus standby 0.200/0.320 with host edges at two. It includes SDK/prewarm and is not public-network evidence. - Atomic activation passed health/access, exact bundle, SQLite preservation, services, and logs; the old release is rollback-ready. Responsive 320/375/390 px checks passed without live media. -- Host A+B aligns capture/outbound identity, deltas, path, and nullable transport-bound codec/`scalabilityMode`; stale or ambiguous data stays unknown and no raw media metadata is retained. Review found no P1/P2; 22 files/299 tests and both builds passed. +- Host A+B aligns capture/outbound identity, deltas, path, and nullable transport-bound codec/`scalabilityMode`. Authenticated Viewer C is limited to the current ordinary or peer-assisted P2P hop and its connection/revision; two-second windows are at most 2 KiB, sanitized, generation-bound, and read-only. Stale, ambiguous, and SFU-fed evidence fails closed; no raw media metadata is retained. - HTTPS/WSS, access and room/WebSocket auth, renewal, public STUN, and authenticated TURN/UDP/TCP relay-only paths pass; TURN/TLS is off. - Draft #16/#18/#22/#23/#25/#28 passed isolated experiments only. ADR-0006 reached host setup/one encoder output, then timed out before viewer 1 decoded/rendered. @@ -37,7 +37,7 @@ unconfigured, so production remains one host connection per viewer and P2P/TURN. - Silent partitions can wait 30 to 60 seconds for heartbeat detection before the default 5-second grace; this remains unverified. - Real audio and heterogeneous clients remain unverified. Production reports poor film audio and voice-call self-echo under system capture; there is no app audio ceiling or Web process isolation. Diagnose A/B/C and sync, then gate Windows 11 game-process audio; Windows 10 stays unresolved without system fallback. - Production `769de201f7cc` reportedly sustains native `qualityLimitationReason=bandwidth` and severe blur with one capable LAN/direct viewer; only Host refresh restores quality. Within the five-second grace, stable-clientId Viewer refresh retains its `peerId`/`HostPeer`, while Host refresh rebuilds peers/capture; churn persistence is also reported. This prioritizes but does not prove sender/PC/GCC/capture generation over network or lifecycle causes. -- ADR-0007 remains incomplete. Host A+B codec fields exist, but browser support and the reported case are unverified; authenticated C, the controller, and `LOW` are absent. Weak paths must share one `LOW`; fail closed protects `HIGH` only exceptionally, and an unreliable supported cohort fails acceptance. +- ADR-0007 remains incomplete. A+B/C browser support and the reported case are unverified; the controller, `LOW`, and an authenticated SFU last-hop C generation are absent. Weak paths must share one `LOW`; fail closed protects `HIGH` only exceptionally, and an unreliable supported cohort fails acceptance. - Browser relays re-encode. ADR-0006 is no-go-unclassified: downstream checkpoints and two-edge/FIFO/TURN gates are absent. Spikes prove one WebCodecs object, not hardware; GCC+RTX is no-go and no-RTX weakens stats. - Browser fanout is host two/viewer one; any accepted endpoint relay stays capped at two downstream edges. - ADR-0005 accepts SFU-root fallback, but code is failure-only/TURN-required; UDP-first config, edge grants, bounded failure, cost/trust, and public canary are absent. @@ -46,14 +46,13 @@ unconfigured, so production remains one host connection per viewer and P2P/TURN. ## Next Milestone -Use Host A+B to compare exact production and current `main` on one wired/direct +Use correlated Host A+B/Viewer C to compare exact production and current `main` on one wired/direct game fixture. Record actual codec/encoder, game FPS, CPU/GPU, preview on/off and A+B while separately rebuilding one `HostPeer`, replacing capture, and fully -refreshing Host; retain the matching Viewer-local sample until authenticated C -lands. Only a proven stuck generation justifies guarded recovery; never +refreshing Host. Only a proven stuck generation justifies guarded recovery; never reconnect periodically, force AV1, or raise ceilings blindly. -After authenticated video C, test simulcast/LiveKit/SVC in order and stop at the +After that reproduction, test simulcast/LiveKit/SVC in order and stop at the first fit before custom `LOW`, independently of ADR-0006. Diagnose audio A/B/C, sync, and voice-source leakage in parallel when it does not displace that P0; `maxBitrate` is not a quality fix. WebRTC/LiveKit owns congestion/layers; the app diff --git a/src/client/media/viewer-quality-evidence.ts b/src/client/media/viewer-quality-evidence.ts new file mode 100644 index 0000000..2f7973d --- /dev/null +++ b/src/client/media/viewer-quality-evidence.ts @@ -0,0 +1,250 @@ +import { + MAX_MEDIA_ROUTE_REVISION, + MAX_VIEWER_QUALITY_EVIDENCE_BYTES, + VIEWER_QUALITY_EVIDENCE_INTERVAL_MS, + viewerQualityEvidenceMessageSchema, + viewerQualityEvidenceMetricsSchema, + type ClientMessage, + type ServerMessage, + type ViewerQualityEvidenceMetrics, +} from "../../shared/protocol"; +import { + EMPTY_METRICS, + type ConnectionMetrics, + type PeerSnapshot, +} from "../types"; + +type ViewerQualityEvidence = Extract< + ServerMessage, + { type: "viewer-quality-evidence" } +>; + +type ViewerQualityEvidenceWindow = Pick< + Extract, + "windowMs" | "metrics" +>; + +function boundedNumber(value: number | null, maximum: number): number | null { + return value !== null && Number.isFinite(value) && value >= 0 && value <= maximum + ? value + : null; +} + +function boundedInteger(value: number | null, maximum: number): number | null { + return value !== null && Number.isInteger(value) && value >= 0 && value <= maximum + ? value + : null; +} + +function boundedString( + value: string | null, + maximumLength: number, + pattern: RegExp, +): string | null { + return value !== null && + value.length <= maximumLength && + pattern.test(value) + ? value + : null; +} + +export function qualityEvidenceWindowFromMetrics( + metrics: ConnectionMetrics, +): ViewerQualityEvidenceWindow | null { + if (metrics.sampleWindowMs === null || !Number.isFinite(metrics.sampleWindowMs)) { + return null; + } + const windowMs = Math.round(metrics.sampleWindowMs); + if (windowMs < 1_000 || windowMs > 5_000) { + return null; + } + + const boundedWidth = boundedInteger(metrics.frameWidth, 16_384); + const boundedHeight = boundedInteger(metrics.frameHeight, 16_384); + const width = boundedWidth !== null && boundedWidth >= 1 ? boundedWidth : null; + const height = + boundedHeight !== null && boundedHeight >= 1 ? boundedHeight : null; + const dimensions = + 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), + bitrateKbps: boundedNumber(metrics.bitrateKbps, 100_000), + packetsReceivedDelta: boundedInteger( + metrics.intervalPacketsReceived, + 1_000_000, + ), + packetsLostDelta: boundedInteger(metrics.intervalPacketsLost, 1_000_000), + jitterMs: boundedNumber(metrics.jitterMs, 60_000), + framesDecodedDelta: boundedInteger( + metrics.intervalFramesDecoded, + 10_000, + ), + framesDroppedDelta: boundedInteger( + metrics.intervalFramesDropped, + 10_000, + ), + decodeMsPerFrame: boundedNumber(metrics.intervalDecodeMs, 60_000), + freezeCountDelta: boundedInteger(metrics.intervalFreezeCount, 10_000), + freezeDurationMsDelta: freezeDuration, + codec: boundedString( + metrics.codec, + 64, + /^video\/[A-Za-z0-9.+-]{1,32}$/i, + ), + codecProfile: boundedString( + metrics.codecProfile, + 64, + /^[a-z0-9-]+=[a-z0-9]+$/, + ), + codecParameters: boundedString( + metrics.codecParameters, + 128, + /^[a-z0-9-]+=[a-z0-9]+(?:; [a-z0-9-]+=[a-z0-9]+)*$/, + ), + }; + const parsed = viewerQualityEvidenceMetricsSchema.safeParse(candidate); + return parsed.success ? { windowMs, metrics: parsed.data } : null; +} + +export class ViewerQualityEvidenceReporter { + private connectionId: string | null = null; + private routeRevision: number | null = null; + private sequence = 0; + private lastSentAtMs: number | null = null; + private lastSampleTimestampMs: number | null = null; + + constructor( + private readonly send: (message: ClientMessage) => boolean, + private readonly now: () => number = Date.now, + ) {} + + offer(snapshot: PeerSnapshot, routeRevision: number): boolean { + if ( + !Number.isSafeInteger(routeRevision) || + routeRevision < 0 || + routeRevision > MAX_MEDIA_ROUTE_REVISION + ) { + return false; + } + const window = qualityEvidenceWindowFromMetrics(snapshot.metrics); + const sampleTimestampMs = snapshot.metrics.sampleTimestampMs; + if ( + !window || + sampleTimestampMs === null || + !Number.isFinite(sampleTimestampMs) || + sampleTimestampMs < 0 + ) { + return false; + } + if (this.connectionId !== snapshot.connectionId) { + this.connectionId = snapshot.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; + } + const now = this.now(); + if ( + !Number.isFinite(now) || + (this.lastSentAtMs !== null && + now - this.lastSentAtMs < VIEWER_QUALITY_EVIDENCE_INTERVAL_MS) + ) { + return false; + } + + const message = { + type: "viewer-quality-evidence" as const, + guard: { + connectionId: snapshot.connectionId, + routeRevision, + }, + sequence: this.sequence, + ...window, + }; + const parsed = viewerQualityEvidenceMessageSchema.safeParse(message); + if ( + !parsed.success || + new TextEncoder().encode(JSON.stringify(parsed.data)).byteLength > + MAX_VIEWER_QUALITY_EVIDENCE_BYTES || + !this.send(parsed.data) + ) { + return false; + } + + this.sequence += 1; + this.lastSentAtMs = now; + this.lastSampleTimestampMs = sampleTimestampMs; + return true; + } + + reset(): void { + this.connectionId = null; + this.routeRevision = null; + this.sequence = 0; + this.lastSentAtMs = null; + this.lastSampleTimestampMs = null; + } +} + +export function qualityEvidenceMatchesSnapshot( + evidence: ViewerQualityEvidence, + snapshot: PeerSnapshot | null, +): boolean { + return ( + snapshot !== null && + snapshot.peerId === evidence.viewerPeerId && + snapshot.connectionId === evidence.guard.connectionId + ); +} + +export function metricsFromQualityEvidence( + evidence: ViewerQualityEvidence, +): ConnectionMetrics { + const metrics = evidence.metrics; + return { + ...EMPTY_METRICS, + sampleWindowMs: evidence.windowMs, + framesPerSecond: metrics.framesPerSecond, + frameWidth: metrics.width, + frameHeight: metrics.height, + resolution: + metrics.width !== null && metrics.height !== null + ? `${metrics.width}x${metrics.height}` + : null, + bitrateKbps: metrics.bitrateKbps, + packetsLost: metrics.packetsLostDelta, + intervalPacketsReceived: metrics.packetsReceivedDelta, + intervalPacketsLost: metrics.packetsLostDelta, + jitterMs: metrics.jitterMs, + framesDropped: metrics.framesDroppedDelta, + intervalFramesDecoded: metrics.framesDecodedDelta, + intervalFramesDropped: metrics.framesDroppedDelta, + intervalDecodeMs: metrics.decodeMsPerFrame, + intervalFreezeCount: metrics.freezeCountDelta, + intervalFreezeDurationMs: metrics.freezeDurationMsDelta, + codec: metrics.codec, + codecProfile: metrics.codecProfile, + codecParameters: metrics.codecParameters, + }; +} diff --git a/src/client/pages/HostPage.tsx b/src/client/pages/HostPage.tsx index ad8e9a7..7150390 100644 --- a/src/client/pages/HostPage.tsx +++ b/src/client/pages/HostPage.tsx @@ -12,6 +12,7 @@ import { import { useEffect, useMemo, useRef, useState } from "react"; import { DEFAULT_QUALITY_SETTINGS, + VIEWER_QUALITY_EVIDENCE_EXPIRY_MS, type CreateRoomResponse, type IceConfig, type ServerMessage, @@ -52,6 +53,7 @@ import { } from "../media/quality"; import { HostSfuRoute } from "../media/host-sfu-route"; import { SfuStandbyPrewarmer } from "../media/sfu-standby-prewarmer"; +import { metricsFromQualityEvidence } from "../media/viewer-quality-evidence"; import type { PeerSnapshot, SignalConnectionState, @@ -65,6 +67,11 @@ import { sourceSwitchNotice } from "./host-page-notices"; type HostPhase = "idle" | "starting" | "live" | "ended" | "error"; +type ViewerQualityEvidence = Extract< + ServerMessage, + { type: "viewer-quality-evidence" } +>; + interface CaptureDetails { resolution: string; frameRate: number | null; @@ -142,6 +149,9 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { const [peerSnapshots, setPeerSnapshots] = useState>( () => new Map(), ); + const [viewerQualityEvidence, setViewerQualityEvidence] = useState< + Map + >(() => new Map()); const [notice, setNotice] = useState(null); const [copied, setCopied] = useState(false); const [switchingSource, setSwitchingSource] = useState(false); @@ -154,7 +164,14 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { const signalRef = useRef(null); const iceConfigRef = useRef(null); const peersRef = useRef(new Map()); + const viewerQualityEvidenceRef = useRef( + new Map(), + ); + const viewerQualityEvidenceTimersRef = useRef( + new Map(), + ); const peerAssistedRef = useRef(false); + const activeRouteRevisionRef = useRef(0); const generationRef = useRef(0); const activeGenerationRef = useRef(null); const sourceSwitchRef = useRef(null); @@ -193,6 +210,12 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { signalRef.current?.stop(); peersRef.current.forEach((peer) => peer.dispose()); peersRef.current.clear(); + viewerQualityEvidenceTimersRef.current.forEach((timer) => + window.clearTimeout(timer), + ); + viewerQualityEvidenceTimersRef.current.clear(); + viewerQualityEvidenceRef.current.clear(); + activeRouteRevisionRef.current = 0; void hostSfuRouteRef.current?.disconnect(); hostSfuRouteRef.current = null; sfuStandbyPrewarmerRef.current?.dispose(); @@ -301,6 +324,13 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { setRelayAvailable(false); setMaxViewers(null); setPeerSnapshots(new Map()); + viewerQualityEvidenceTimersRef.current.forEach((timer) => + window.clearTimeout(timer), + ); + viewerQualityEvidenceTimersRef.current.clear(); + viewerQualityEvidenceRef.current = new Map(); + setViewerQualityEvidence(new Map()); + activeRouteRevisionRef.current = 0; setSignalStatus("offline"); setSwitchingSource(false); setChangingQuality(false); @@ -327,6 +357,13 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { } function updatePeerSnapshot(snapshot: PeerSnapshot): void { + const evidence = viewerQualityEvidenceRef.current.get(snapshot.peerId); + if ( + evidence && + evidence.guard.connectionId !== snapshot.connectionId + ) { + clearViewerQualityEvidence(snapshot.peerId); + } setPeerSnapshots((current) => { const next = new Map(current); next.set(snapshot.peerId, snapshot); @@ -334,6 +371,59 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { }); } + function clearViewerQualityEvidence(peerId: string): void { + const timer = viewerQualityEvidenceTimersRef.current.get(peerId); + if (timer !== undefined) { + window.clearTimeout(timer); + viewerQualityEvidenceTimersRef.current.delete(peerId); + } + if (!viewerQualityEvidenceRef.current.has(peerId)) { + return; + } + const next = new Map(viewerQualityEvidenceRef.current); + next.delete(peerId); + viewerQualityEvidenceRef.current = next; + setViewerQualityEvidence(next); + } + + function clearAllViewerQualityEvidence(): void { + viewerQualityEvidenceTimersRef.current.forEach((timer) => + window.clearTimeout(timer), + ); + viewerQualityEvidenceTimersRef.current.clear(); + viewerQualityEvidenceRef.current = new Map(); + setViewerQualityEvidence(new Map()); + } + + function acceptViewerQualityEvidence(evidence: ViewerQualityEvidence): void { + const peer = peersRef.current.get(evidence.viewerPeerId); + if ( + !peer || + peer.connectionId !== evidence.guard.connectionId || + evidence.guard.routeRevision !== activeRouteRevisionRef.current + ) { + return; + } + const next = new Map(viewerQualityEvidenceRef.current); + next.set(evidence.viewerPeerId, evidence); + viewerQualityEvidenceRef.current = next; + setViewerQualityEvidence(next); + const previousTimer = viewerQualityEvidenceTimersRef.current.get( + evidence.viewerPeerId, + ); + if (previousTimer !== undefined) { + window.clearTimeout(previousTimer); + } + const timer = window.setTimeout(() => { + if ( + viewerQualityEvidenceRef.current.get(evidence.viewerPeerId) === evidence + ) { + clearViewerQualityEvidence(evidence.viewerPeerId); + } + }, VIEWER_QUALITY_EVIDENCE_EXPIRY_MS); + viewerQualityEvidenceTimersRef.current.set(evidence.viewerPeerId, timer); + } + function watchCaptureEnd( captured: MediaStream, generation: number, @@ -471,6 +561,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { } function removePeer(peerId: string): void { + clearViewerQualityEvidence(peerId); peersRef.current.get(peerId)?.dispose(); peersRef.current.delete(peerId); setPeerSnapshots((current) => { @@ -625,6 +716,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { return; } if (message.type === "authenticated" && message.role === "host") { + clearAllViewerQualityEvidence(); setSfuStandbyUrl( "sfuStandbyUrl" in message ? message.sfuStandbyUrl : null, ); @@ -633,6 +725,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { "mediaMode" in message && message.mediaMode === "peer-assisted" ) { + activeRouteRevisionRef.current = message.routeRevision; peerAssistedRef.current = true; signalRef.current?.send({ type: "set-quality-settings", @@ -648,6 +741,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { .then(() => showHostSfuQualityWarning(route, generation)); return; } + activeRouteRevisionRef.current = 0; peerAssistedRef.current = false; clearHostSfuRoute(); const currentViewerIds = new Set(message.viewerPeerIds); @@ -660,6 +754,13 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { } if (message.type === "route-update") { if (peerAssistedRef.current) { + if ( + message.phase === "active" && + message.revision !== activeRouteRevisionRef.current + ) { + activeRouteRevisionRef.current = message.revision; + clearAllViewerQualityEvidence(); + } const route = ensureHostSfuRoute(generation); void route .acceptAndWait(message) @@ -667,6 +768,10 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { } return; } + if (message.type === "viewer-quality-evidence") { + acceptViewerQualityEvidence(message); + return; + } if (message.type === "sfu-config") { if (peerAssistedRef.current) { const route = ensureHostSfuRoute(generation); @@ -1432,11 +1537,24 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { )} {showConnectionDetails && ( - + <> + + {viewerQualityEvidence.get(viewer.peerId) && ( + <> +

观看端接收

+ + + )} + )} {viewer.error &&

{viewer.error}

} diff --git a/src/client/pages/ViewerPage.tsx b/src/client/pages/ViewerPage.tsx index 7b9f1d4..d6d5e87 100644 --- a/src/client/pages/ViewerPage.tsx +++ b/src/client/pages/ViewerPage.tsx @@ -9,6 +9,7 @@ import { import { useEffect, useMemo, useRef, useState } from "react"; import { DEFAULT_QUALITY_SETTINGS, + VIEWER_QUALITY_EVIDENCE_EXPIRY_MS, type IceConfig, type MediaAssignment, type ServerMessage, @@ -28,6 +29,11 @@ import { SignalingClient } from "../lib/signaling"; import type { QualitySettings } from "../media/quality"; import { relayCapacityMessageForBrowser } from "../media/relay-capability"; import { SfuStandbyPrewarmer } from "../media/sfu-standby-prewarmer"; +import { + metricsFromQualityEvidence, + qualityEvidenceMatchesSnapshot, + ViewerQualityEvidenceReporter, +} from "../media/viewer-quality-evidence"; import { ViewerMessageAuthority } from "../media/viewer-message-authority"; import { ViewerSfuRoute } from "../media/viewer-sfu-route"; import type { @@ -48,6 +54,11 @@ interface ViewerPageProps { onAuthorizationRequired: () => void; } +type ViewerQualityEvidence = Extract< + ServerMessage, + { type: "viewer-quality-evidence" } +>; + export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps) { const forceRelay = useMemo( () => new URLSearchParams(window.location.search).get("relay") === "1", @@ -61,6 +72,8 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps) const [remoteStream, setRemoteStream] = useState(null); const [peerSnapshot, setPeerSnapshot] = useState(null); const [relaySnapshot, setRelaySnapshot] = useState(null); + const [relayChildEvidence, setRelayChildEvidence] = + useState(null); const [playbackBlocked, setPlaybackBlocked] = useState(false); const [muted, setMuted] = useState(false); const [showConnectionDetails, setShowConnectionDetails] = useState(false); @@ -83,6 +96,8 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps) let currentIceConfig: IceConfig | null = null; let currentHostOnline = false; let peerAssisted = false; + let currentPeerId: string | null = null; + let currentRouteRevision = 0; let currentQualitySettings: QualitySettings = DEFAULT_QUALITY_SETTINGS; let currentAssignment: MediaAssignment = { parentPeerId: null, @@ -96,6 +111,8 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps) > = []; const messageAuthority = new ViewerMessageAuthority(); let sfuStandbyPrewarmer: SfuStandbyPrewarmer | null = null; + let relayChildEvidenceCurrent: ViewerQualityEvidence | null = null; + let relayChildEvidenceTimer: number | null = null; function setSfuStandbyUrl(url: string | null | undefined): void { if (!url) { @@ -150,6 +167,37 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps) }, }, ); + const qualityEvidenceReporter = new ViewerQualityEvidenceReporter( + (message) => active && signal.send(message), + ); + + function clearRelayChildEvidence(): void { + relayChildEvidenceCurrent = null; + setRelayChildEvidence(null); + if (relayChildEvidenceTimer !== null) { + window.clearTimeout(relayChildEvidenceTimer); + relayChildEvidenceTimer = null; + } + } + + function acceptRelayChildEvidence(evidence: ViewerQualityEvidence): void { + if ( + !peerAssisted || + evidence.parentPeerId !== currentPeerId || + evidence.guard.routeRevision !== currentRouteRevision || + !qualityEvidenceMatchesSnapshot(evidence, viewerRelay?.getSnapshot() ?? null) + ) { + return; + } + clearRelayChildEvidence(); + relayChildEvidenceCurrent = evidence; + setRelayChildEvidence(evidence); + relayChildEvidenceTimer = window.setTimeout(() => { + if (relayChildEvidenceCurrent === evidence) { + clearRelayChildEvidence(); + } + }, VIEWER_QUALITY_EVIDENCE_EXPIRY_MS); + } function ensureViewerRelay(): ViewerRelay | null { if (viewerRelay) { @@ -175,6 +223,15 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps) onUpdate: (snapshot) => { if (active) { setRelaySnapshot(snapshot); + if ( + relayChildEvidenceCurrent && + !qualityEvidenceMatchesSnapshot( + relayChildEvidenceCurrent, + snapshot, + ) + ) { + clearRelayChildEvidence(); + } } }, }, @@ -301,6 +358,7 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps) } function clearUpstreamState(): void { + qualityEvidenceReporter.reset(); peerRef.current?.dispose(); peerRef.current = null; setRemoteStream(null); @@ -319,12 +377,16 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps) MAX_VIEWER_MEDIA_CHILDREN, ); const previousParentId = currentAssignment.parentPeerId; + const previousChildId = currentAssignment.childPeerIds[0] ?? null; currentAssignment = nextAssignment; if (previousParentId !== nextAssignment.parentPeerId) { clearUpstreamState(); setStatusText("正在恢复连接"); } + if (previousChildId !== (nextAssignment.childPeerIds[0] ?? null)) { + clearRelayChildEvidence(); + } ensureViewerRelay()?.setChild(nextAssignment.childPeerIds[0] ?? null); } @@ -362,6 +424,7 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps) }, onUpdate: (snapshot) => { if (active) { + qualityEvidenceReporter.offer(snapshot, currentRouteRevision); if ( !currentHostOnline && snapshot.connectionState === "failed" @@ -395,6 +458,8 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps) authorityToken: number, ): Promise { if (message.type === "authenticated") { + clearRelayChildEvidence(); + currentPeerId = message.peerId; setSfuStandbyUrl( "sfuStandbyUrl" in message ? message.sfuStandbyUrl : null, ); @@ -408,6 +473,14 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps) currentAssignment = { parentPeerId: null, childPeerIds: [] }; } peerAssisted = nextPeerAssisted; + const nextRouteRevision = + nextPeerAssisted && "routeRevision" in message + ? message.routeRevision + : 0; + if (nextRouteRevision !== currentRouteRevision) { + clearRelayChildEvidence(); + } + currentRouteRevision = nextRouteRevision; const relayCapacity = relayCapacityMessageForBrowser(nextPeerAssisted); if (relayCapacity) { signal.send(relayCapacity); @@ -467,10 +540,20 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps) } if (message.type === "route-update") { if (peerAssisted) { + if (message.phase === "active") { + if (message.revision !== currentRouteRevision) { + clearRelayChildEvidence(); + } + currentRouteRevision = message.revision; + } ensureViewerSfuRoute().accept(message); } return; } + if (message.type === "viewer-quality-evidence") { + acceptRelayChildEvidence(message); + return; + } if (message.type === "sfu-config") { if (peerAssisted) { await ensureViewerSfuRoute().acceptConfig(message); @@ -591,6 +674,13 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps) signal.start(); return () => { active = false; + currentPeerId = null; + qualityEvidenceReporter.reset(); + relayChildEvidenceCurrent = null; + if (relayChildEvidenceTimer !== null) { + window.clearTimeout(relayChildEvidenceTimer); + relayChildEvidenceTimer = null; + } sfuStandbyPrewarmer?.dispose(); signal.stop(); void viewerSfuRoute?.disconnect(); @@ -797,6 +887,18 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps) /> )} + {showConnectionDetails && relayChildEvidence && ( +
+

下游接收数据

+ +
+ )} ); diff --git a/src/client/types.ts b/src/client/types.ts index 5ac4efa..2e72678 100644 --- a/src/client/types.ts +++ b/src/client/types.ts @@ -26,10 +26,15 @@ export interface ConnectionMetrics { bitrateKbps: number | null; availableOutgoingKbps: number | null; framesPerSecond: number | null; + frameWidth: number | null; + frameHeight: number | null; resolution: string | null; packetsLost: number | null; + intervalPacketsReceived: number | null; + intervalPacketsLost: number | null; jitterMs: number | null; framesDropped: number | null; + intervalFramesDecoded: number | null; intervalFramesDropped: number | null; intervalFreezeCount: number | null; intervalFreezeDurationMs: number | null; @@ -77,10 +82,15 @@ export const EMPTY_METRICS: ConnectionMetrics = { bitrateKbps: null, availableOutgoingKbps: null, framesPerSecond: null, + frameWidth: null, + frameHeight: null, resolution: null, packetsLost: null, + intervalPacketsReceived: null, + intervalPacketsLost: null, jitterMs: null, framesDropped: null, + intervalFramesDecoded: null, intervalFramesDropped: null, intervalFreezeCount: null, intervalFreezeDurationMs: null, diff --git a/src/client/webrtc/stats.ts b/src/client/webrtc/stats.ts index 475540e..95d38c6 100644 --- a/src/client/webrtc/stats.ts +++ b/src/client/webrtc/stats.ts @@ -2,6 +2,7 @@ import { EMPTY_METRICS, type ConnectionMetrics, } from "../types"; +import { deriveVideoCodecEvidence } from "../../shared/video-codec-evidence"; type StatsRecord = Record & { id: string; @@ -18,6 +19,8 @@ export interface StatsAccumulator { timestamp: number | null; previousTotalEncodeTime: number | null; previousTotalDecodeTime: number | null; + previousPacketsReceived: number | null; + previousPacketsLost: number | null; previousFramesDropped: number | null; previousFreezeCount: number | null; previousTotalFreezesDuration: number | null; @@ -39,6 +42,8 @@ export function createStatsAccumulator(): StatsAccumulator { timestamp: null, previousTotalEncodeTime: null, previousTotalDecodeTime: null, + previousPacketsReceived: null, + previousPacketsLost: null, previousFramesDropped: null, previousFreezeCount: null, previousTotalFreezesDuration: null, @@ -200,56 +205,6 @@ function linkedRemoteInbound( : null; } -interface CodecEvidence { - codec: string | null; - profile: string | null; - parameters: string | null; -} - -type CodecParameterRule = { - profileKey: string | null; - parameterKeys: readonly string[]; - validators: Readonly>; -}; - -const CODEC_PARAMETER_RULES: Readonly> = { - "video/h264": { - profileKey: "profile-level-id", - parameterKeys: ["packetization-mode", "level-asymmetry-allowed"], - validators: { - "profile-level-id": /^[0-9a-f]{6}$/i, - "packetization-mode": /^[0-2]$/, - "level-asymmetry-allowed": /^[01]$/, - }, - }, - "video/vp9": { - profileKey: "profile-id", - parameterKeys: ["max-fr", "max-fs"], - validators: { - "profile-id": /^[0-3]$/, - "max-fr": /^[1-9][0-9]{0,9}$/, - "max-fs": /^[1-9][0-9]{0,9}$/, - }, - }, - "video/vp8": { - profileKey: null, - parameterKeys: ["max-fr", "max-fs"], - validators: { - "max-fr": /^[1-9][0-9]{0,9}$/, - "max-fs": /^[1-9][0-9]{0,9}$/, - }, - }, - "video/av1": { - profileKey: "profile", - parameterKeys: ["level-idx", "tier"], - validators: { - profile: /^[0-2]$/, - "level-idx": /^(?:[0-9]|[12][0-9]|3[01])$/, - tier: /^[01]$/, - }, - }, -}; - function linkedVideoCodec( report: RTCStatsReport, media: StatsRecord | null, @@ -266,62 +221,12 @@ function linkedVideoCodec( : null; } -function deriveCodecEvidence(codec: StatsRecord | null): CodecEvidence { +function deriveCodecEvidence(codec: StatsRecord | null) { const mimeType = stringValue(codec, "mimeType"); - if (!mimeType) { - return { codec: null, profile: null, parameters: null }; - } - const rule = CODEC_PARAMETER_RULES[mimeType.toLowerCase()]; - const fmtp = stringValue(codec, "sdpFmtpLine"); - if (!rule || !fmtp || fmtp.length > 2_048) { - return { codec: mimeType, profile: null, parameters: null }; - } - - const segments = fmtp.split(";"); - if (segments.length > 32) { - return { codec: mimeType, profile: null, parameters: null }; - } - const values = new Map(); - const seenKeys = new Set(); - for (const segment of segments) { - const separator = segment.indexOf("="); - if (separator < 1) { - continue; - } - const key = segment.slice(0, separator).trim().toLowerCase(); - const validator = rule.validators[key]; - if (!validator) { - continue; - } - if (seenKeys.has(key)) { - values.delete(key); - continue; - } - seenKeys.add(key); - const value = segment.slice(separator + 1).trim().toLowerCase(); - if (!validator.test(value)) { - continue; - } - values.set(key, value); - } - - const profileValue = rule.profileKey - ? values.get(rule.profileKey) - : undefined; - const parameters = rule.parameterKeys - .flatMap((key) => { - const value = values.get(key); - return value ? [`${key}=${value}`] : []; - }) - .join("; "); - return { - codec: mimeType, - profile: - rule.profileKey && profileValue - ? `${rule.profileKey}=${profileValue}` - : null, - parameters: parameters || null, - }; + return deriveVideoCodecEvidence( + mimeType, + stringValue(codec, "sdpFmtpLine"), + ); } function scalabilityModeValue(media: StatsRecord | null): string | null { @@ -383,6 +288,11 @@ export async function collectConnectionMetrics( const timestamp = numberValue(media, "timestamp"); const framesEncoded = numberValue(media, "framesEncoded"); const framesDecoded = numberValue(media, "framesDecoded"); + const packetsReceived = numberValue(media, "packetsReceived"); + const packetsLost = numberValue( + direction === "send" ? remoteInbound : media, + "packetsLost", + ); const totalEncodeTime = numberValue(media, "totalEncodeTime"); const totalDecodeTime = numberValue(media, "totalDecodeTime"); const framesDropped = numberValue(media, "framesDropped"); @@ -441,6 +351,30 @@ export async function collectConnectionMetrics( sampleWindowMs !== null ? previous.frames : null, ) : null; + const intervalPacketsReceived = + direction === "receive" + ? intervalDelta( + packetsReceived, + previous.previousPacketsReceived, + sampleWindowMs !== null, + ) + : null; + const intervalPacketsLost = + direction === "receive" + ? intervalDelta( + packetsLost, + previous.previousPacketsLost, + sampleWindowMs !== null, + ) + : null; + const intervalFramesDecoded = + direction === "receive" + ? intervalDelta( + framesDecoded, + previous.frames, + sampleWindowMs !== null, + ) + : null; const intervalFramesDropped = intervalDelta( framesDropped, previous.previousFramesDropped, @@ -474,6 +408,8 @@ export async function collectConnectionMetrics( previous.timestamp = timestamp; previous.previousTotalEncodeTime = totalEncodeTime; previous.previousTotalDecodeTime = totalDecodeTime; + previous.previousPacketsReceived = packetsReceived; + previous.previousPacketsLost = packetsLost; previous.previousFramesDropped = framesDropped; previous.previousFreezeCount = freezeCount; previous.previousTotalFreezesDuration = totalFreezesDuration; @@ -481,14 +417,7 @@ export async function collectConnectionMetrics( previous.previousRetransmittedBytes = retransmittedBytes; const linkedCodec = linkedVideoCodec(report, media, transport); - const codecEvidence = - direction === "send" - ? deriveCodecEvidence(linkedCodec) - : { - codec: stringValue(linkedCodec, "mimeType"), - profile: null, - parameters: null, - }; + const codecEvidence = deriveCodecEvidence(linkedCodec); const width = numberValue(media, "frameWidth"); const height = numberValue(media, "frameHeight"); const iceProtocol = @@ -523,15 +452,19 @@ export async function collectConnectionMetrics( ? numberValue(pair, "availableOutgoingBitrate")! / 1_000 : null, framesPerSecond: numberValue(media, "framesPerSecond") ?? derivedFps, + frameWidth: width, + frameHeight: height, resolution: width !== null && height !== null ? `${width}x${height}` : null, - packetsLost: - numberValue(direction === "send" ? remoteInbound : media, "packetsLost"), + packetsLost, + intervalPacketsReceived, + intervalPacketsLost, jitterMs: numberValue(direction === "send" ? remoteInbound : media, "jitter") !== null ? numberValue(direction === "send" ? remoteInbound : media, "jitter")! * 1_000 : null, framesDropped, + intervalFramesDecoded, intervalFramesDropped, intervalFreezeCount, intervalFreezeDurationMs: diff --git a/src/server/hybrid-media-router.ts b/src/server/hybrid-media-router.ts index e6fe652..941c4bf 100644 --- a/src/server/hybrid-media-router.ts +++ b/src/server/hybrid-media-router.ts @@ -207,17 +207,35 @@ export class HybridMediaRouter { parentPeerId: string, childPeerId: string, ): boolean { + return ( + this.resolveActivePeerEdge(roomId, childPeerId)?.parentPeerId === + parentPeerId + ); + } + + resolveActivePeerEdge( + roomId: string, + childPeerId: string, + ): { revision: number; parentPeerId: string } | undefined { const controller = this.mediaRouteControllers.get(roomId); if (!controller) { - return false; + return undefined; } - const assignments = controller.getActiveRoute().assignments; + const active = controller.getActiveRoute(); + const assignments = active.assignments; const child = assignments.get(childPeerId); - return ( - child?.upstream.kind === "peer" && - child.upstream.peerId === parentPeerId && - assignments.get(parentPeerId)?.childPeerIds.includes(childPeerId) === true - ); + if ( + child?.upstream.kind !== "peer" || + assignments + .get(child.upstream.peerId) + ?.childPeerIds.includes(childPeerId) !== true + ) { + return undefined; + } + return { + revision: active.revision, + parentPeerId: child.upstream.peerId, + }; } setViewerRelayCapacity( diff --git a/src/server/signaling.ts b/src/server/signaling.ts index 41edf99..ef84c99 100644 --- a/src/server/signaling.ts +++ b/src/server/signaling.ts @@ -7,13 +7,19 @@ import WebSocket, { WebSocketServer } from "ws"; import { DEFAULT_QUALITY_SETTINGS, MAX_SIGNAL_BYTES, + MAX_VIEWER_QUALITY_EVIDENCE_BYTES, + VIEWER_QUALITY_EVIDENCE_INTERVAL_MS, decodeClientMessage, type ClientMessage, type QualitySettings, type Role, type ServerMessage, } from "../shared/protocol.js"; -import { RoomStore, RoomStoreError } from "./room-store.js"; +import { + RoomStore, + RoomStoreError, + type ConnectedPeer, +} from "./room-store.js"; import { HybridMediaRouter, type SfuFallbackOptions, @@ -41,6 +47,16 @@ interface SocketState { authenticated?: AuthenticatedSession; } +interface ViewerQualityEvidenceGate { + viewerSessionId: string; + parentSessionId: string; + parentPeerId: string; + connectionId: string; + routeRevision: number; + sequence: number; + acceptedAtMs: number; +} + export interface SignalingOptions { server: HttpServer; roomStore: RoomStore; @@ -70,6 +86,10 @@ export class SignalingServer { private readonly socketsBySessionId = new Map(); private readonly viewerGraceTimers = new Map(); private readonly connectionIdsByViewer = new Map(); + private readonly viewerQualityEvidenceGates = new Map< + string, + ViewerQualityEvidenceGate + >(); private readonly qualitySettingsByRoom = new Map(); private readonly shareGenerationsByRoom = new Map(); private readonly hybridMediaRouter?: HybridMediaRouter; @@ -122,9 +142,7 @@ export class SignalingServer { viewerConnectionKey(roomId, viewerPeerId), ), deleteConnectionId: (roomId, viewerPeerId) => - this.connectionIdsByViewer.delete( - viewerConnectionKey(roomId, viewerPeerId), - ), + this.deleteViewerConnectionId(roomId, viewerPeerId), }); } @@ -185,6 +203,7 @@ export class SignalingServer { } this.viewerGraceTimers.clear(); this.hybridMediaRouter?.close(); + this.viewerQualityEvidenceGates.clear(); this.shareGenerationsByRoom.clear(); this.options.server.off("upgrade", this.upgradeHandler); @@ -242,6 +261,13 @@ export class SignalingServer { this.rejectInvalidMessage(socket); return; } + if ( + message.type === "viewer-quality-evidence" && + Buffer.byteLength(encoded, "utf8") > MAX_VIEWER_QUALITY_EVIDENCE_BYTES + ) { + this.rejectInvalidMessage(socket); + return; + } const state = this.socketStates.get(socket); if (!state) { @@ -339,6 +365,11 @@ export class SignalingServer { : Date.parse(participant.expiresAt), shareGeneration, }; + if (participant.role === "viewer") { + this.viewerQualityEvidenceGates.delete( + viewerConnectionKey(participant.roomId, participant.peerId), + ); + } this.clearViewerGrace(participant.roomId, participant.peerId); const routeParticipant = { roomId: participant.roomId, @@ -570,6 +601,9 @@ export class SignalingServer { message.revision, ); return; + case "viewer-quality-evidence": + this.handleViewerQualityEvidence(socket, authenticated, message); + return; case "stop-sharing": case "close-room": if (authenticated.role !== "host") { @@ -613,6 +647,98 @@ export class SignalingServer { } } + private handleViewerQualityEvidence( + socket: WebSocket, + source: AuthenticatedSession, + message: Extract, + ): void { + if (source.role !== "viewer") { + return; + } + const viewerState = this.socketStates.get(socket); + if (!viewerState || viewerState.authenticated !== source) { + return; + } + + const connectionId = this.connectionIdsByViewer.get( + viewerConnectionKey(source.roomId, source.peerId), + ); + if (!connectionId || message.guard.connectionId !== connectionId) { + return; + } + + let routeRevision = 0; + let parent: ConnectedPeer | undefined; + if (this.isPeerAssistedRoom(source.roomId)) { + const edge = this.hybridMediaRouter?.resolveActivePeerEdge( + source.roomId, + source.peerId, + ); + if (!edge) { + return; + } + routeRevision = edge.revision; + parent = this.connectedPeer(source.roomId, edge.parentPeerId); + } else { + parent = this.options.roomStore.getConnectedHost(source.roomId); + } + if ( + !parent || + message.guard.routeRevision !== routeRevision || + parent.peerId === source.peerId + ) { + return; + } + + const gateKey = viewerConnectionKey(source.roomId, source.peerId); + const previous = this.viewerQualityEvidenceGates.get(gateKey); + const sameGeneration = + previous?.viewerSessionId === viewerState.sessionId && + previous.parentSessionId === parent.sessionId && + previous.parentPeerId === parent.peerId && + previous.connectionId === connectionId && + previous.routeRevision === routeRevision; + const now = this.now(); + if ( + sameGeneration && + previous && + (message.sequence <= previous.sequence || + now - previous.acceptedAtMs < VIEWER_QUALITY_EVIDENCE_INTERVAL_MS) + ) { + return; + } + + const forwarded = { + type: "viewer-quality-evidence" as const, + viewerPeerId: source.peerId, + parentPeerId: parent.peerId, + guard: { + connectionId, + routeRevision, + }, + 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(parent.sessionId, encoded) + ) { + return; + } + this.viewerQualityEvidenceGates.set(gateKey, { + viewerSessionId: viewerState.sessionId, + parentSessionId: parent.sessionId, + parentPeerId: parent.peerId, + connectionId, + routeRevision, + sequence: message.sequence, + acceptedAtMs: now, + }); + } + private routeSignal( sourceSocket: WebSocket, source: AuthenticatedSession, @@ -640,8 +766,9 @@ export class SignalingServer { return; } if (description?.type === "offer") { - this.connectionIdsByViewer.set( - viewerConnectionKey(source.roomId, viewer.peerId), + this.setViewerConnectionId( + source.roomId, + viewer.peerId, message.payload.connectionId, ); } @@ -713,8 +840,9 @@ export class SignalingServer { } if (description?.type === "offer") { - this.connectionIdsByViewer.set( - viewerConnectionKey(source.roomId, targetPeerId), + this.setViewerConnectionId( + source.roomId, + targetPeerId, message.payload.connectionId, ); } @@ -819,8 +947,9 @@ export class SignalingServer { disconnected.peerId, ) ) { - this.connectionIdsByViewer.delete( - viewerConnectionKey(disconnected.roomId, disconnected.peerId), + this.deleteViewerConnectionId( + disconnected.roomId, + disconnected.peerId, ); if (this.isPeerAssistedRoom(disconnected.roomId)) { this.hybridMediaRouter!.removeViewer( @@ -977,6 +1106,19 @@ export class SignalingServer { } } + private sendEncodedToSession(sessionId: string, encoded: string): boolean { + const socket = this.socketsBySessionId.get(sessionId); + if (!socket || socket.readyState !== WebSocket.OPEN) { + return false; + } + if (socket.bufferedAmount > MAX_BUFFERED_SIGNAL_BYTES) { + socket.terminate(); + return false; + } + socket.send(encoded); + return true; + } + private send(socket: WebSocket, message: ServerMessage): void { if (socket.readyState !== WebSocket.OPEN) { return; @@ -1023,6 +1165,32 @@ export class SignalingServer { this.connectionIdsByViewer.delete(key); } } + for (const key of this.viewerQualityEvidenceGates.keys()) { + if (key.startsWith(prefix)) { + this.viewerQualityEvidenceGates.delete(key); + } + } + } + + private setViewerConnectionId( + roomId: string, + viewerPeerId: string, + connectionId: string, + ): void { + const key = viewerConnectionKey(roomId, viewerPeerId); + if (this.connectionIdsByViewer.get(key) !== connectionId) { + this.viewerQualityEvidenceGates.delete(key); + } + this.connectionIdsByViewer.set(key, connectionId); + } + + private deleteViewerConnectionId( + roomId: string, + viewerPeerId: string, + ): void { + const key = viewerConnectionKey(roomId, viewerPeerId); + this.connectionIdsByViewer.delete(key); + this.viewerQualityEvidenceGates.delete(key); } } diff --git a/src/shared/protocol.ts b/src/shared/protocol.ts index 1c9db32..15c94ff 100644 --- a/src/shared/protocol.ts +++ b/src/shared/protocol.ts @@ -1,10 +1,15 @@ import { z } from "zod"; +import { isCanonicalVideoCodecEvidence } from "./video-codec-evidence.js"; + export const MAX_VIEWERS_PER_ROOM_LIMIT = 16; export const MAX_SIGNAL_BYTES = 64 * 1024; export const ROOM_CODE_LENGTH = 12; export const MAX_MEDIA_ROUTE_REVISION = Number.MAX_SAFE_INTEGER; export const MAX_SFU_TOKEN_LENGTH = 8 * 1024; +export const MAX_VIEWER_QUALITY_EVIDENCE_BYTES = 2 * 1024; +export const VIEWER_QUALITY_EVIDENCE_INTERVAL_MS = 2_000; +export const VIEWER_QUALITY_EVIDENCE_EXPIRY_MS = 5_000; const opaqueIdSchema = z .string() @@ -181,6 +186,94 @@ export type ParticipantRouteAssignment = z.infer< typeof participantRouteAssignmentSchema >; +const nullableEvidenceNumber = (maximum: number) => + z.number().finite().min(0).max(maximum).nullable(); + +const nullableEvidenceInteger = (maximum: number) => + z.number().int().min(0).max(maximum).nullable(); + +export const viewerQualityEvidenceMetricsSchema = z + .object({ + width: z.number().int().min(1).max(16_384).nullable(), + height: z.number().int().min(1).max(16_384).nullable(), + framesPerSecond: nullableEvidenceNumber(240), + bitrateKbps: nullableEvidenceNumber(100_000), + packetsReceivedDelta: nullableEvidenceInteger(1_000_000), + packetsLostDelta: nullableEvidenceInteger(1_000_000), + jitterMs: nullableEvidenceNumber(60_000), + framesDecodedDelta: nullableEvidenceInteger(10_000), + framesDroppedDelta: nullableEvidenceInteger(10_000), + decodeMsPerFrame: nullableEvidenceNumber(60_000), + freezeCountDelta: nullableEvidenceInteger(10_000), + freezeDurationMsDelta: nullableEvidenceNumber(5_000), + codec: z + .string() + .max(64) + .regex(/^video\/[A-Za-z0-9.+-]{1,32}$/i) + .nullable(), + codecProfile: z + .string() + .max(64) + .regex(/^[a-z0-9-]+=[a-z0-9]+$/) + .nullable(), + codecParameters: z + .string() + .max(128) + .regex(/^[a-z0-9-]+=[a-z0-9]+(?:; [a-z0-9-]+=[a-z0-9]+)*$/) + .nullable(), + }) + .strict() + .refine( + (metrics) => + (metrics.width === null && metrics.height === null) || + (metrics.width !== null && metrics.height !== null), + { message: "Viewer quality dimensions must be present together" }, + ) + .refine(isCanonicalVideoCodecEvidence, { + message: "Viewer codec evidence is not canonical", + path: ["codecParameters"], + }); +export type ViewerQualityEvidenceMetrics = z.infer< + typeof viewerQualityEvidenceMetricsSchema +>; + +const viewerQualityEvidenceGuardSchema = z + .object({ + connectionId: opaqueIdSchema, + routeRevision: mediaRouteRevisionSchema, + }) + .strict(); + +const viewerQualityEvidenceWindowShape = { + sequence: z.number().int().min(0).max(Number.MAX_SAFE_INTEGER), + windowMs: z.number().int().min(1_000).max(5_000), + 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", + }); +export type ViewerQualityEvidenceMessage = z.infer< + typeof viewerQualityEvidenceMessageSchema +>; + const authenticateMessageSchema = z.discriminatedUnion("role", [ z .object({ @@ -253,6 +346,7 @@ export const clientMessageSchema = z.union([ revision: mediaRouteRevisionSchema, }) .strict(), + viewerQualityEvidenceMessageSchema, z .object({ type: z.literal("stop-sharing"), @@ -361,6 +455,18 @@ export const serverMessageSchema = z.union([ qualitySettings: qualitySettingsSchema, }) .strict(), + z + .object({ + type: z.literal("viewer-quality-evidence"), + viewerPeerId: opaqueIdSchema, + parentPeerId: opaqueIdSchema, + guard: viewerQualityEvidenceGuardSchema, + ...viewerQualityEvidenceWindowShape, + }) + .strict() + .refine(freezeFitsEvidenceWindow, { + message: "Viewer freeze duration exceeds its evidence window", + }), z .object({ type: z.literal("ice-config"), diff --git a/src/shared/video-codec-evidence.ts b/src/shared/video-codec-evidence.ts new file mode 100644 index 0000000..81b596e --- /dev/null +++ b/src/shared/video-codec-evidence.ts @@ -0,0 +1,166 @@ +export interface VideoCodecEvidence { + codec: string | null; + profile: string | null; + parameters: string | null; +} + +interface CodecParameterRule { + profileKey: string | null; + parameterKeys: readonly string[]; + validators: Readonly>; +} + +const VIDEO_CODEC_PARAMETER_RULES: Readonly< + Record +> = { + "video/h264": { + profileKey: "profile-level-id", + parameterKeys: ["packetization-mode", "level-asymmetry-allowed"], + validators: { + "profile-level-id": /^[0-9a-f]{6}$/, + "packetization-mode": /^[0-2]$/, + "level-asymmetry-allowed": /^[01]$/, + }, + }, + "video/vp9": { + profileKey: "profile-id", + parameterKeys: ["max-fr", "max-fs"], + validators: { + "profile-id": /^[0-3]$/, + "max-fr": /^[1-9][0-9]{0,9}$/, + "max-fs": /^[1-9][0-9]{0,9}$/, + }, + }, + "video/vp8": { + profileKey: null, + parameterKeys: ["max-fr", "max-fs"], + validators: { + "max-fr": /^[1-9][0-9]{0,9}$/, + "max-fs": /^[1-9][0-9]{0,9}$/, + }, + }, + "video/av1": { + profileKey: "profile", + parameterKeys: ["level-idx", "tier"], + validators: { + profile: /^[0-2]$/, + "level-idx": /^(?:[0-9]|[12][0-9]|3[01])$/, + tier: /^[01]$/, + }, + }, +}; + +const VIDEO_MIME_TYPE = /^video\/[A-Za-z0-9.+-]{1,32}$/i; + +export function deriveVideoCodecEvidence( + codec: string | null, + rawParameters: string | null, +): VideoCodecEvidence { + if (!codec || !VIDEO_MIME_TYPE.test(codec)) { + return { codec: null, profile: null, parameters: null }; + } + const rule = VIDEO_CODEC_PARAMETER_RULES[codec.toLowerCase()]; + if (!rule || !rawParameters || rawParameters.length > 2_048) { + return { codec, profile: null, parameters: null }; + } + + const segments = rawParameters.split(";"); + if (segments.length > 32) { + return { codec, profile: null, parameters: null }; + } + const values = new Map(); + const seenKeys = new Set(); + for (const segment of segments) { + const separator = segment.indexOf("="); + if (separator < 1) { + continue; + } + const key = segment.slice(0, separator).trim().toLowerCase(); + const validator = rule.validators[key]; + if (!validator) { + continue; + } + if (seenKeys.has(key)) { + values.delete(key); + continue; + } + seenKeys.add(key); + const value = segment.slice(separator + 1).trim().toLowerCase(); + if (validator.test(value)) { + values.set(key, value); + } + } + + const profileValue = rule.profileKey + ? values.get(rule.profileKey) + : undefined; + const parameters = rule.parameterKeys + .flatMap((key) => { + const value = values.get(key); + return value ? [`${key}=${value}`] : []; + }) + .join("; "); + return { + codec, + profile: + rule.profileKey && profileValue + ? `${rule.profileKey}=${profileValue}` + : null, + parameters: parameters || null, + }; +} + +export function isCanonicalVideoCodecEvidence(value: { + codec: string | null; + codecProfile: string | null; + codecParameters: string | null; +}): boolean { + const rule = value.codec + ? VIDEO_CODEC_PARAMETER_RULES[value.codec.toLowerCase()] + : undefined; + if (!rule) { + return value.codecProfile === null && value.codecParameters === null; + } + return ( + canonicalValuesMatch( + value.codecProfile, + rule.profileKey ? [rule.profileKey] : [], + rule.validators, + ) && + canonicalValuesMatch( + value.codecParameters, + rule.parameterKeys, + rule.validators, + ) + ); +} + +function canonicalValuesMatch( + encoded: string | null, + orderedKeys: readonly string[], + validators: Readonly>, +): boolean { + if (encoded === null) { + return true; + } + const segments = encoded.split("; "); + if (segments.length > orderedKeys.length) { + return false; + } + let previousKeyIndex = -1; + for (const segment of segments) { + const separator = segment.indexOf("="); + const key = separator < 1 ? "" : segment.slice(0, separator); + const keyIndex = orderedKeys.indexOf(key); + const candidate = separator < 1 ? "" : segment.slice(separator + 1); + if ( + keyIndex <= previousKeyIndex || + keyIndex < 0 || + !validators[key]?.test(candidate) + ) { + return false; + } + previousKeyIndex = keyIndex; + } + return true; +} diff --git a/tests/client-state.test.ts b/tests/client-state.test.ts index f5333cc..d6a0963 100644 --- a/tests/client-state.test.ts +++ b/tests/client-state.test.ts @@ -17,6 +17,7 @@ import { shouldReconnectSignaling, SignalingClient, } from "../src/client/lib/signaling.ts"; +import { qualityEvidenceWindowFromMetrics } from "../src/client/media/viewer-quality-evidence.ts"; import { createStatsAccumulator, collectConnectionMetrics } from "../src/client/webrtc/stats.ts"; afterEach(() => { @@ -574,18 +575,27 @@ describe("WebRTC stats parsing", () => { getStats: async () => report, } as unknown as RTCPeerConnection; - await expect( - collectConnectionMetrics( - connection, - "send", - createStatsAccumulator(), - ), - ).resolves.toMatchObject({ + const metrics = await collectConnectionMetrics( + connection, + "send", + createStatsAccumulator(), + ); + expect(metrics).toMatchObject({ codec: mimeType, codecProfile: profile, codecParameters: parameters, scalabilityMode: null, }); + expect( + qualityEvidenceWindowFromMetrics({ + ...metrics, + sampleWindowMs: 2_000, + })?.metrics, + ).toMatchObject({ + codec: mimeType, + codecProfile: profile, + codecParameters: parameters, + }); }, ); @@ -996,4 +1006,93 @@ describe("WebRTC stats parsing", () => { intervalRetransmittedBytes: 800, }); }); + + it("derives a bounded inbound C window from adjacent RTP counters", async () => { + const inbound = ( + timestamp: number, + packetsReceived: number, + packetsLost: number, + framesDecoded: number, + totalDecodeTime: number, + ) => + new Map([ + [ + "transport", + { id: "transport", type: "transport", timestamp }, + ], + [ + "codec", + { + id: "codec", + type: "codec", + timestamp, + transportId: "transport", + mimeType: "video/H264", + sdpFmtpLine: + "profile-level-id=42e01f; packetization-mode=1; " + + "sprop-parameter-sets=must-not-leave-stats", + }, + ], + [ + "inbound", + { + id: "inbound", + type: "inbound-rtp", + timestamp, + kind: "video", + ssrc: 101, + transportId: "transport", + codecId: "codec", + bytesReceived: timestamp * 10, + packetsReceived, + packetsLost, + framesDecoded, + framesDropped: 3, + freezeCount: 1, + totalFreezesDuration: 0.25, + totalDecodeTime, + frameWidth: 1_920, + frameHeight: 1_080, + scalabilityMode: "L3T3_KEY", + }, + ], + ]) as unknown as RTCStatsReport; + const reports = [ + inbound(1_000, 1_000, 10, 60, 0.12), + inbound(3_000, 2_500, 12, 180, 0.42), + inbound(5_000, 4_000, 11, 300, 0.72), + ]; + const connection = { + getStats: async () => reports.shift()!, + } as unknown as RTCPeerConnection; + const accumulator = createStatsAccumulator(); + + await collectConnectionMetrics(connection, "receive", accumulator); + const stable = await collectConnectionMetrics( + connection, + "receive", + accumulator, + ); + const correctedLoss = await collectConnectionMetrics( + connection, + "receive", + accumulator, + ); + + expect(stable).toMatchObject({ + sampleWindowMs: 2_000, + frameWidth: 1_920, + frameHeight: 1_080, + intervalPacketsReceived: 1_500, + intervalPacketsLost: 2, + intervalFramesDecoded: 120, + intervalDecodeMs: 2.5, + codec: "video/H264", + codecProfile: "profile-level-id=42e01f", + codecParameters: "packetization-mode=1", + scalabilityMode: null, + }); + expect(JSON.stringify(stable)).not.toContain("sprop-parameter-sets"); + expect(correctedLoss.intervalPacketsLost).toBeNull(); + }); }); diff --git a/tests/protocol.test.ts b/tests/protocol.test.ts index 15845e7..e262ba2 100644 --- a/tests/protocol.test.ts +++ b/tests/protocol.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it } from "vitest"; import { MAX_MEDIA_ROUTE_REVISION, MAX_SFU_TOKEN_LENGTH, + MAX_VIEWER_QUALITY_EVIDENCE_BYTES, MAX_VIEWERS_PER_ROOM_LIMIT, clientMessageSchema, decodeClientMessage, @@ -19,6 +20,34 @@ const qualitySettings = { degradationPreference: "maintain-resolution", } as const; +const qualityEvidence = { + type: "viewer-quality-evidence", + guard: { + connectionId: "connection_12345678", + routeRevision: 0, + }, + sequence: 0, + windowMs: 2_000, + metrics: { + width: 1_920, + height: 1_080, + framesPerSecond: 59.8, + bitrateKbps: 7_500, + packetsReceivedDelta: 1_500, + packetsLostDelta: 2, + jitterMs: 3.5, + framesDecodedDelta: 120, + framesDroppedDelta: 1, + decodeMsPerFrame: 2.4, + freezeCountDelta: 0, + freezeDurationMsDelta: 0, + codec: "video/H264", + codecProfile: "profile-level-id=42e01f", + codecParameters: + "packetization-mode=1; level-asymmetry-allowed=1", + }, +} as const; + describe("client signaling protocol", () => { it("accepts a bounded authentication message", () => { expect( @@ -220,6 +249,87 @@ describe("client signaling protocol", () => { ).toBe(false); }); + it("accepts only strict, bounded viewer quality evidence", () => { + expect(clientMessageSchema.safeParse(qualityEvidence).success).toBe(true); + expect( + Buffer.byteLength(JSON.stringify(qualityEvidence), "utf8"), + ).toBeLessThanOrEqual(MAX_VIEWER_QUALITY_EVIDENCE_BYTES); + + for (const invalid of [ + { ...qualityEvidence, roomId }, + { + ...qualityEvidence, + guard: { ...qualityEvidence.guard, peerId: "viewer_12345678" }, + }, + { ...qualityEvidence, windowMs: 999 }, + { ...qualityEvidence, sequence: -1 }, + { + ...qualityEvidence, + metrics: { ...qualityEvidence.metrics, width: null }, + }, + { + ...qualityEvidence, + metrics: { ...qualityEvidence.metrics, bitrateKbps: Number.POSITIVE_INFINITY }, + }, + { + ...qualityEvidence, + metrics: { ...qualityEvidence.metrics, freezeDurationMsDelta: 2_001 }, + }, + { + ...qualityEvidence, + metrics: { + ...qualityEvidence.metrics, + codecParameters: "sprop-parameter-sets=deadbeef", + }, + }, + { + ...qualityEvidence, + metrics: { + ...qualityEvidence.metrics, + codecProfile: "device-id=deadbeef", + }, + }, + { + ...qualityEvidence, + metrics: { + ...qualityEvidence.metrics, + scalabilityMode: "DEVICE_ABC123", + }, + }, + { + ...qualityEvidence, + metrics: { + ...qualityEvidence.metrics, + decoderImplementation: "device-specific-decoder", + }, + }, + ]) { + expect(clientMessageSchema.safeParse(invalid).success).toBe(false); + } + }); + + it.each([ + ["video/VP8", null, "max-fr=60; max-fs=8160"], + ["video/VP9", "profile-id=2", "max-fs=8160"], + ["video/AV1", "profile=1", "level-idx=31; tier=0"], + ["video/unknown", null, null], + ])( + "accepts canonical viewer codec evidence for %s", + (codec, codecProfile, codecParameters) => { + expect( + clientMessageSchema.safeParse({ + ...qualityEvidence, + metrics: { + ...qualityEvidence.metrics, + codec, + codecProfile, + codecParameters, + }, + }).success, + ).toBe(true); + }, + ); + it("accepts bounded route acknowledgements, failures, and SFU refreshes", () => { expect( clientMessageSchema.safeParse({ @@ -500,6 +610,27 @@ describe("server signaling protocol", () => { } }); + it("accepts only canonical server-derived viewer evidence envelopes", () => { + const forwarded = { + ...qualityEvidence, + viewerPeerId: "viewer_12345678", + parentPeerId: "host_12345678", + }; + expect(serverMessageSchema.safeParse(forwarded).success).toBe(true); + expect( + Buffer.byteLength(JSON.stringify(forwarded), "utf8"), + ).toBeLessThanOrEqual(MAX_VIEWER_QUALITY_EVIDENCE_BYTES); + expect( + serverMessageSchema.safeParse({ ...forwarded, roomId }).success, + ).toBe(false); + expect( + serverMessageSchema.safeParse({ + ...qualityEvidence, + viewerPeerId: "viewer_12345678", + }).success, + ).toBe(false); + }); + it.each([0, 1.5, MAX_VIEWERS_PER_ROOM_LIMIT + 1])( "rejects viewer limit %s outside the protocol boundary", (maxViewers) => { diff --git a/tests/server-signal.test.ts b/tests/server-signal.test.ts index 64d8313..24c7743 100644 --- a/tests/server-signal.test.ts +++ b/tests/server-signal.test.ts @@ -6,6 +6,7 @@ import WebSocket from "ws"; import { MAX_SIGNAL_BYTES, + MAX_VIEWER_QUALITY_EVIDENCE_BYTES, decodeServerMessage, type Role, type ServerMessage, @@ -161,6 +162,7 @@ async function startHarness( persistent?: boolean; peerAssistedMedia?: boolean; peerAssistedPrimaryRoomOnly?: boolean; + now?: () => number; } = {}, ): Promise { const config = testConfig(); @@ -182,6 +184,7 @@ async function startHarness( config, roomStore, serveFrontend: false, + now: overrides.now, authenticationTimeoutMs: overrides.authenticationTimeoutMs ?? 500, viewerDisconnectGraceMs: overrides.viewerDisconnectGraceMs ?? 50, heartbeatIntervalMs: 60_000, @@ -199,6 +202,38 @@ async function startHarness( }; } +function viewerQualityEvidence( + connectionId: string, + routeRevision: number, + sequence: number, + bitrateKbps = 7_500, +) { + return { + type: "viewer-quality-evidence", + guard: { connectionId, routeRevision }, + sequence, + windowMs: 2_000, + metrics: { + width: 1_920, + height: 1_080, + framesPerSecond: 60, + bitrateKbps, + packetsReceivedDelta: 1_500, + packetsLostDelta: 2, + jitterMs: 3.5, + framesDecodedDelta: 120, + framesDroppedDelta: 1, + decodeMsPerFrame: 2.4, + freezeCountDelta: 0, + freezeDurationMsDelta: 0, + codec: "video/H264", + codecProfile: "profile-level-id=42e01f", + codecParameters: + "packetization-mode=1; level-asymmetry-allowed=1", + }, + } as const; +} + async function startSfuHarness(options: { tokenIssuer: SfuTokenIssuer; prepareTimeoutMs?: number; @@ -1119,6 +1154,314 @@ describe("WebSocket signaling", () => { expect(harness.roomStore.size).toBe(0); }); + it("forwards only current, bounded ordinary-P2P viewer C evidence", async () => { + let now = Date.now(); + const harness = await startHarness({ now: () => now }); + const host = await openClient(harness.webSocketUrl); + const hostAuth = await authenticate( + host, + harness.room, + "host", + "quality-host-client", + ); + const viewer = await openClient(harness.webSocketUrl); + const viewerAuth = await authenticate( + viewer, + harness.room, + "viewer", + "quality-viewer-client", + ); + await host.inbox.next("peer-joined"); + + host.socket.send( + JSON.stringify({ + type: "signal", + targetPeerId: viewerAuth.peerId, + payload: { + kind: "description", + connectionId: "quality_connection_first", + description: { type: "offer", sdp: "v=0\r\n" }, + }, + }), + ); + await viewer.inbox.next("signal"); + + viewer.socket.send( + JSON.stringify(viewerQualityEvidence("quality_connection_first", 0, 0)), + ); + expect(await host.inbox.next("viewer-quality-evidence")).toEqual({ + ...viewerQualityEvidence("quality_connection_first", 0, 0), + viewerPeerId: viewerAuth.peerId, + parentPeerId: hostAuth.peerId, + }); + + viewer.socket.send( + JSON.stringify(viewerQualityEvidence("quality_connection_first", 0, 1)), + ); + viewer.socket.send( + JSON.stringify( + viewerQualityEvidence("quality_connection_first", 0, 1, 7_000), + ), + ); + await host.inbox.expectNone(30); + + now += 2_000; + viewer.socket.send( + JSON.stringify( + viewerQualityEvidence("quality_connection_first", 0, 1, 7_000), + ), + ); + expect( + (await host.inbox.next("viewer-quality-evidence")).metrics.bitrateKbps, + ).toBe(7_000); + + now += 2_000; + viewer.socket.send( + JSON.stringify( + viewerQualityEvidence("quality_connection_first", 0, 2, 7_000), + ), + ); + expect( + (await host.inbox.next("viewer-quality-evidence")).metrics.bitrateKbps, + ).toBe(7_000); + + viewer.socket.send( + JSON.stringify( + viewerQualityEvidence("quality_connection_stale", 0, 3, 6_500), + ), + ); + viewer.socket.send( + JSON.stringify( + viewerQualityEvidence("quality_connection_first", 1, 3, 6_500), + ), + ); + await host.inbox.expectNone(30); + + host.socket.send( + JSON.stringify({ + type: "signal", + targetPeerId: viewerAuth.peerId, + payload: { + kind: "description", + connectionId: "quality_connection_second", + description: { type: "offer", sdp: "v=0\r\n" }, + }, + }), + ); + await viewer.inbox.next("signal"); + viewer.socket.send( + JSON.stringify( + viewerQualityEvidence("quality_connection_first", 0, 3, 6_000), + ), + ); + await host.inbox.expectNone(30); + viewer.socket.send( + JSON.stringify( + viewerQualityEvidence("quality_connection_second", 0, 0, 6_000), + ), + ); + expect( + (await host.inbox.next("viewer-quality-evidence")).guard.connectionId, + ).toBe("quality_connection_second"); + + host.socket.send( + JSON.stringify( + viewerQualityEvidence("quality_connection_second", 0, 1, 5_500), + ), + ); + await host.inbox.expectNone(30); + + const replacementViewer = await openClient(harness.webSocketUrl); + const replacementAuth = await authenticate( + replacementViewer, + harness.room, + "viewer", + "quality-viewer-client", + ); + expect(replacementAuth).toMatchObject({ + peerId: viewerAuth.peerId, + connectionId: "quality_connection_second", + }); + expect(await host.inbox.next("peer-joined")).toMatchObject({ + peerId: viewerAuth.peerId, + }); + replacementViewer.socket.send( + JSON.stringify( + viewerQualityEvidence("quality_connection_second", 0, 0, 5_000), + ), + ); + expect(await host.inbox.next("viewer-quality-evidence")).toMatchObject({ + viewerPeerId: viewerAuth.peerId, + guard: { connectionId: "quality_connection_second" }, + sequence: 0, + metrics: { bitrateKbps: 5_000 }, + }); + }); + + it("derives peer-assisted C routing from the active assignment", async () => { + const harness = await startHarness({ peerAssistedMedia: true }); + const host = await openClient(harness.webSocketUrl); + const hostAuth = peerAssisted( + await authenticate(host, harness.room, "host", "peer-quality-host"), + ); + host.inbox.ignore("route-update"); + host.inbox.ignore("media-assignment"); + const viewer = await openClient(harness.webSocketUrl); + const viewerAuth = peerAssisted( + await authenticate( + viewer, + harness.room, + "viewer", + "peer-quality-viewer", + ), + ); + expect(viewerAuth.routeAssignment.upstream).toEqual({ + kind: "peer", + peerId: hostAuth.peerId, + }); + + host.socket.send( + JSON.stringify({ + type: "signal", + targetPeerId: viewerAuth.peerId, + payload: { + kind: "description", + connectionId: "peer_quality_connection", + description: { type: "offer", sdp: "v=0\r\n" }, + }, + }), + ); + await viewer.inbox.next("signal"); + viewer.socket.send( + JSON.stringify( + viewerQualityEvidence( + "peer_quality_connection", + viewerAuth.routeRevision, + 0, + ), + ), + ); + expect(await host.inbox.next("viewer-quality-evidence")).toMatchObject({ + viewerPeerId: viewerAuth.peerId, + parentPeerId: hostAuth.peerId, + guard: { + connectionId: "peer_quality_connection", + routeRevision: viewerAuth.routeRevision, + }, + }); + + viewer.socket.send( + JSON.stringify( + viewerQualityEvidence( + "peer_quality_connection", + viewerAuth.routeRevision + 1, + 1, + 7_000, + ), + ), + ); + await host.inbox.expectNone(30); + + viewer.inbox.ignore("route-update"); + viewer.inbox.ignore("media-assignment"); + const secondViewer = await openClient(harness.webSocketUrl); + await authenticate( + secondViewer, + harness.room, + "viewer", + "peer-quality-viewer-two", + ); + const relayChild = await openClient(harness.webSocketUrl); + const relayChildAuth = peerAssisted( + await authenticate( + relayChild, + harness.room, + "viewer", + "peer-quality-relay-child", + ), + ); + expect(relayChildAuth.routeAssignment.upstream).toEqual({ + kind: "peer", + peerId: viewerAuth.peerId, + }); + + viewer.socket.send( + JSON.stringify( + viewerQualityEvidence( + "peer_quality_connection", + relayChildAuth.routeRevision, + 1, + 6_500, + ), + ), + ); + expect(await host.inbox.next("viewer-quality-evidence")).toMatchObject({ + viewerPeerId: viewerAuth.peerId, + guard: { + connectionId: "peer_quality_connection", + routeRevision: relayChildAuth.routeRevision, + }, + sequence: 1, + }); + + viewer.socket.send( + JSON.stringify({ + type: "signal", + targetPeerId: relayChildAuth.peerId, + payload: { + kind: "description", + connectionId: "relay_child_quality_connection", + description: { type: "offer", sdp: "v=0\r\n" }, + }, + }), + ); + await relayChild.inbox.next("signal"); + relayChild.socket.send( + JSON.stringify( + viewerQualityEvidence( + "relay_child_quality_connection", + relayChildAuth.routeRevision, + 0, + ), + ), + ); + expect(await viewer.inbox.next("viewer-quality-evidence")).toMatchObject({ + viewerPeerId: relayChildAuth.peerId, + parentPeerId: viewerAuth.peerId, + guard: { + connectionId: "relay_child_quality_connection", + routeRevision: relayChildAuth.routeRevision, + }, + }); + await host.inbox.expectNone(30); + }); + + it("rejects viewer C evidence above the byte cap", async () => { + const harness = await startHarness(); + const viewer = await openClient(harness.webSocketUrl); + await authenticate( + viewer, + harness.room, + "viewer", + "oversized-quality-viewer", + ); + const closed = new Promise((resolve) => + viewer.socket.once("close", (code) => resolve(code)), + ); + const valid = JSON.stringify( + viewerQualityEvidence("oversized_quality_connection", 0, 0), + ); + + viewer.socket.send( + `${valid}${" ".repeat(MAX_VIEWER_QUALITY_EVIDENCE_BYTES + 1)}`, + ); + + expect(await viewer.inbox.next("error")).toMatchObject({ + code: "INVALID_MESSAGE", + }); + expect(await closed).toBe(1008); + }); + it("routes offer and answer only between the host and the targeted viewer", async () => { const harness = await startHarness(); const secondRoom = harness.roomStore.createRoom(); diff --git a/tests/viewer-quality-evidence.test.ts b/tests/viewer-quality-evidence.test.ts new file mode 100644 index 0000000..ad42da5 --- /dev/null +++ b/tests/viewer-quality-evidence.test.ts @@ -0,0 +1,247 @@ +import { describe, expect, it } from "vitest"; + +import type { ClientMessage, ServerMessage } from "../src/shared/protocol.ts"; +import { + metricsFromQualityEvidence, + qualityEvidenceMatchesSnapshot, + qualityEvidenceWindowFromMetrics, + ViewerQualityEvidenceReporter, +} from "../src/client/media/viewer-quality-evidence.ts"; +import { + EMPTY_METRICS, + type ConnectionMetrics, + type PeerSnapshot, +} from "../src/client/types.ts"; + +function receiveMetrics( + overrides: Partial = {}, +): ConnectionMetrics { + return { + ...EMPTY_METRICS, + sampleTimestampMs: 10_000, + sampleWindowMs: 2_000, + framesPerSecond: 60, + frameWidth: 1_920, + frameHeight: 1_080, + resolution: "1920x1080", + bitrateKbps: 7_500, + intervalPacketsReceived: 1_500, + intervalPacketsLost: 2, + jitterMs: 3.5, + intervalFramesDecoded: 120, + intervalFramesDropped: 1, + intervalDecodeMs: 2.4, + intervalFreezeCount: 0, + intervalFreezeDurationMs: 0, + codec: "video/H264", + codecProfile: "profile-level-id=42e01f", + codecParameters: + "packetization-mode=1; level-asymmetry-allowed=1", + scalabilityMode: "L3T3_KEY", + ...overrides, + }; +} + +function snapshot( + connectionId: string, + metrics = receiveMetrics(), +): PeerSnapshot { + return { + peerId: "host_12345678", + connectionId, + connectionState: "connected", + iceConnectionState: "connected", + metrics, + error: null, + }; +} + +describe("viewer quality evidence", () => { + it("normalizes only bounded adjacent receive evidence", () => { + expect(qualityEvidenceWindowFromMetrics(receiveMetrics())).toEqual({ + windowMs: 2_000, + metrics: { + width: 1_920, + height: 1_080, + framesPerSecond: 60, + bitrateKbps: 7_500, + packetsReceivedDelta: 1_500, + packetsLostDelta: 2, + jitterMs: 3.5, + framesDecodedDelta: 120, + framesDroppedDelta: 1, + decodeMsPerFrame: 2.4, + freezeCountDelta: 0, + freezeDurationMsDelta: 0, + codec: "video/H264", + codecProfile: "profile-level-id=42e01f", + codecParameters: + "packetization-mode=1; level-asymmetry-allowed=1", + }, + }); + + expect( + qualityEvidenceWindowFromMetrics( + receiveMetrics({ + sampleWindowMs: 9_000, + rtpStatsId: "must-not-leave-the-client", + }), + ), + ).toBeNull(); + const bounded = qualityEvidenceWindowFromMetrics( + receiveMetrics({ + frameWidth: 0, + frameHeight: 0, + bitrateKbps: 200_000, + intervalPacketsLost: -1, + codecParameters: "raw fmtp; secret=value/with/slash", + }), + ); + expect(bounded?.metrics).toMatchObject({ + width: null, + height: null, + bitrateKbps: null, + packetsLostDelta: null, + codecParameters: null, + }); + expect(JSON.stringify(bounded)).not.toContain("rtpStatsId"); + expect(JSON.stringify(bounded)).not.toContain("must-not-leave-the-client"); + expect(JSON.stringify(bounded)).not.toContain("scalabilityMode"); + expect(JSON.stringify(bounded)).not.toContain("L3T3_KEY"); + }); + + it("rate-limits, deduplicates, and resets sequence per connection", () => { + let now = 10_000; + const sent: ClientMessage[] = []; + const reporter = new ViewerQualityEvidenceReporter( + (message) => { + sent.push(message); + return true; + }, + () => now, + ); + 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( + 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 }), + ), + 0, + ), + ).toBe(true); + expect( + reporter.offer( + snapshot( + first.connectionId, + receiveMetrics({ sampleTimestampMs: 12_000, bitrateKbps: 6_000 }), + ), + 2, + ), + ).toBe(true); + expect( + reporter.offer(snapshot("connection_second_12345678"), 2), + ).toBe(true); + now += 2_000; + expect( + reporter.offer( + snapshot( + "connection_second_12345678", + receiveMetrics({ sampleTimestampMs: 14_000 }), + ), + 2, + ), + ).toBe(true); + + expect( + sent.map((message) => + message.type === "viewer-quality-evidence" + ? { + connectionId: message.guard.connectionId, + revision: message.guard.routeRevision, + sequence: message.sequence, + } + : null, + ), + ).toEqual([ + { + connectionId: "connection_first_12345678", + revision: 0, + sequence: 0, + }, + { + connectionId: "connection_first_12345678", + revision: 0, + sequence: 1, + }, + { + connectionId: "connection_first_12345678", + revision: 2, + sequence: 2, + }, + { + connectionId: "connection_second_12345678", + revision: 2, + sequence: 0, + }, + { + connectionId: "connection_second_12345678", + revision: 2, + sequence: 1, + }, + ]); + }); + + it("pairs and renders only the current parent connection", () => { + const evidence = { + type: "viewer-quality-evidence", + viewerPeerId: "viewer_12345678", + parentPeerId: "host_12345678", + guard: { + connectionId: "connection_12345678", + routeRevision: 0, + }, + sequence: 3, + ...qualityEvidenceWindowFromMetrics(receiveMetrics())!, + } as Extract; + const current = { + ...snapshot(evidence.guard.connectionId), + peerId: evidence.viewerPeerId, + }; + + expect(qualityEvidenceMatchesSnapshot(evidence, current)).toBe(true); + expect( + qualityEvidenceMatchesSnapshot(evidence, { + ...current, + connectionId: "connection_replaced_12345678", + }), + ).toBe(false); + expect(metricsFromQualityEvidence(evidence)).toMatchObject({ + sampleWindowMs: 2_000, + frameWidth: 1_920, + frameHeight: 1_080, + resolution: "1920x1080", + bitrateKbps: 7_500, + packetsLost: 2, + framesDropped: 1, + intervalFramesDecoded: 120, + intervalFramesDropped: 1, + intervalFreezeCount: 0, + codec: "video/H264", + }); + }); +});