feat(quality): add authenticated viewer evidence

This commit is contained in:
AAA committed 2026-08-19 20:39:56 +08:00
1 parent 32415fb117
commit a949fbaf15
16 files changed
+1862 -163

No files matched your search

+3 -3
View File
@@ -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
+18 -9
View File
@@ -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)
+6 -7
View File
@@ -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
+250
View File
@@ -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<ClientMessage, { type: "viewer-quality-evidence" }>,
"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,
};
}
+123 -5
View File
@@ -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<Map<string, PeerSnapshot>>(
() => new Map(),
);
const [viewerQualityEvidence, setViewerQualityEvidence] = useState<
Map<string, ViewerQualityEvidence>
>(() => new Map());
const [notice, setNotice] = useState<string | null>(null);
const [copied, setCopied] = useState(false);
const [switchingSource, setSwitchingSource] = useState(false);
@@ -154,7 +164,14 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
const signalRef = useRef<SignalingClient | null>(null);
const iceConfigRef = useRef<IceConfig | null>(null);
const peersRef = useRef(new Map<string, HostPeer>());
const viewerQualityEvidenceRef = useRef(
new Map<string, ViewerQualityEvidence>(),
);
const viewerQualityEvidenceTimersRef = useRef(
new Map<string, number>(),
);
const peerAssistedRef = useRef(false);
const activeRouteRevisionRef = useRef(0);
const generationRef = useRef(0);
const activeGenerationRef = useRef<number | null>(null);
const sourceSwitchRef = useRef<object | null>(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 = {}) {
)}
</div>
{showConnectionDetails && (
<StatsGrid
metrics={viewer.metrics}
direction="send"
senderParameters={viewer.senderParameters}
/>
<>
<StatsGrid
metrics={viewer.metrics}
direction="send"
senderParameters={viewer.senderParameters}
/>
{viewerQualityEvidence.get(viewer.peerId) && (
<>
<p className="section-meta">观看端接收</p>
<StatsGrid
metrics={metricsFromQualityEvidence(
viewerQualityEvidence.get(viewer.peerId)!,
)}
direction="receive"
/>
</>
)}
</>
)}
{viewer.error && <p className="inline-error">{viewer.error}</p>}
</article>
+102
View File
@@ -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<MediaStream | null>(null);
const [peerSnapshot, setPeerSnapshot] = useState<PeerSnapshot | null>(null);
const [relaySnapshot, setRelaySnapshot] = useState<PeerSnapshot | null>(null);
const [relayChildEvidence, setRelayChildEvidence] =
useState<ViewerQualityEvidence | null>(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<void> {
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)
/>
</section>
)}
{showConnectionDetails && relayChildEvidence && (
<section
className="viewer-stats"
aria-labelledby="relay-child-stats-heading"
>
<h2 id="relay-child-stats-heading">下游接收数据</h2>
<StatsGrid
metrics={metricsFromQualityEvidence(relayChildEvidence)}
direction="receive"
/>
</section>
)}
</main>
</div>
);
+10
View File
@@ -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,
+48 -115
View File
@@ -2,6 +2,7 @@ import {
EMPTY_METRICS,
type ConnectionMetrics,
} from "../types";
import { deriveVideoCodecEvidence } from "../../shared/video-codec-evidence";
type StatsRecord = Record<string, unknown> & {
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<Record<string, RegExp>>;
};
const CODEC_PARAMETER_RULES: Readonly<Record<string, CodecParameterRule>> = {
"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<string, string>();
const seenKeys = new Set<string>();
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:
+25 -7
View File
@@ -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(
+178 -10
View File
@@ -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<string, WebSocket>();
private readonly viewerGraceTimers = new Map<string, NodeJS.Timeout>();
private readonly connectionIdsByViewer = new Map<string, string>();
private readonly viewerQualityEvidenceGates = new Map<
string,
ViewerQualityEvidenceGate
>();
private readonly qualitySettingsByRoom = new Map<string, QualitySettings>();
private readonly shareGenerationsByRoom = new Map<string, string>();
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<ClientMessage, { type: "viewer-quality-evidence" }>,
): 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);
}
}
+106
View File
@@ -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"),
+166
View File
@@ -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<Record<string, RegExp>>;
}
const VIDEO_CODEC_PARAMETER_RULES: Readonly<
Record<string, CodecParameterRule>
> = {
"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<string, string>();
const seenKeys = new Set<string>();
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<Record<string, RegExp>>,
): 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;
}
+106 -7
View File
@@ -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<string, unknown>([
[
"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();
});
});
+131
View File
@@ -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) => {
+343
View File
@@ -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<SignalHarness> {
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<number>((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();
+247
View File
@@ -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> = {},
): 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<ServerMessage, { type: "viewer-quality-evidence" }>;
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",
});
});
});