feat(quality): implement strict v13 quality shadow (#291)

This commit is contained in:
AAA authored and GitHub committed 2026-08-27 02:57:38 +08:00
1 parent 5a081fb119
commit 501df27bce
22 files changed
+1467 -203

No files matched your search

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