From ad175408c046f11b0c739ed1bdf6e2ec9947bba2 Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Thu, 27 Aug 2026 20:21:55 +0800 Subject: [PATCH] fix(route): guard SFU quality convergence (#301) Fixes the production-proven P2P-to-SFU quality misrouting. Local full checks and independent P0/P1 review passed; the GitHub job did not start any steps. --- .env.example | 2 + .../0005-automatic-hybrid-media-routing.md | 57 +- docs/product/routing-transport.md | 18 +- docs/project-memory.md | 19 +- docs/reference/configuration.md | 5 + docs/research/realtime-quality-adaptation.md | 19 + docs/status.md | 15 +- docs/todo.md | 5 +- src/client/media/route-transition.ts | 3 +- src/client/media/sender-quality-evidence.ts | 51 +- src/client/media/sfu-quality-probe.ts | 132 ++++ src/client/media/viewer-sfu-route.ts | 74 +- src/client/pages/ViewerPage.tsx | 10 + src/client/webrtc/host-peer.ts | 10 +- src/server/hybrid-media-router.ts | 3 + src/server/room-route-controller.ts | 347 ++++++++- src/shared/protocol.ts | 30 +- tests/host-peer-tracks.test.ts | 2 + tests/hybrid-media-router.test.ts | 35 +- tests/protocol.test.ts | 13 +- tests/room-route-controller.test.ts | 672 ++++++++++++------ tests/route-transition.test.ts | 100 ++- tests/sender-quality-evidence.test.ts | 50 +- tests/sfu-quality-probe.test.ts | 94 +++ 24 files changed, 1432 insertions(+), 334 deletions(-) create mode 100644 src/client/media/sfu-quality-probe.ts create mode 100644 tests/sfu-quality-probe.test.ts diff --git a/.env.example b/.env.example index a6fed96..f581711 100644 --- a/.env.example +++ b/.env.example @@ -23,6 +23,8 @@ PEER_ASSISTED_MEDIA=false # One steady outbound media-copy capacity for every non-server endpoint. # A peer child or Host SFU publication consumes one copy. Valid values: 1, 2, 3. ENDPOINT_MEDIA_COPY_CAPACITY=2 +# During pre-release route canaries, this enables sanitized room/ordinal events. +# NODE_DEBUG=screener-route # Optional automatic SFU fallback. Set PEER_ASSISTED_MEDIA=true and supply all # four values together; leaving them empty keeps LiveKit completely dormant. diff --git a/docs/adr/0005-automatic-hybrid-media-routing.md b/docs/adr/0005-automatic-hybrid-media-routing.md index dd5c432..76e3d7b 100644 --- a/docs/adr/0005-automatic-hybrid-media-routing.md +++ b/docs/adr/0005-automatic-hybrid-media-routing.md @@ -1,6 +1,6 @@ # ADR-0005: Automatic Hybrid Media Routing -- Status: accepted and deployed; native-edge convergence is opt-in per share +- Status: accepted; deployed convergence correction in progress - Date: 2026-08-20 - Last updated: 2026-08-27 @@ -190,18 +190,24 @@ same-identity delta in `none` is healthy; one in `bandwidth` or `cpu` is degraded; `other`, missing, reset, hidden, or stale evidence is unknown. SFU publication and subscription state remains owned by LiveKit plus the exact Viewer's continuing decoded progress. Screener does not combine loss, RTT, -jitter, bitrate, FPS, resolution, or freezes into another score. +jitter, bitrate, FPS, resolution, or freezes into a weighted route score. -When the current source path remains degraded, the controller considers the -shallowest affected child first and uses the existing deterministic candidate -filters, ordering, cursor, reservations, and total deadline. Candidate parents -must have a healthy source path and available steady capacity. Only one -candidate runs at a time; the old route keeps playing. First decoded frame proves -candidate usability, and a complete healthy native edge delta proves the new -path has escaped the old limitation. If the old edge recovers, the candidate is -degraded or unknown, authority changes, or the deadline expires, the candidate -is aborted and the graph stays unchanged. A successful commit clears and -rebaselines quality state for that edge and its affected subtree. +Routing degradation requires three consecutive complete degraded deltas for the +same sender identity. Healthy, unknown, stale, reset, source change, or identity +change clears that run. This is the same persistence semantic used by the Host +quality warning; a single native limitation interval remains diagnostic. + +When the current source path remains persistently degraded, the controller +considers the shallowest affected child first and uses the existing +deterministic candidate filters, ordering, cursor, reservations, and total +deadline. Candidate Peer parents must have a healthy source path and available +steady capacity. Only one candidate runs at a time; the old route keeps playing. +First decoded frame proves candidate usability, and a complete healthy native +edge delta proves the new path has escaped the old limitation. If the old edge +recovers, the candidate is degraded or unknown, authority changes, or the +deadline expires, the candidate is aborted and the graph stays unchanged. A +successful commit clears and rebaselines quality state for that edge and its +affected subtree. The same rule supplies both active parent change and relay abdication. A bad relay ingress reparents that relay while retaining its subtree. A bad exact @@ -211,13 +217,26 @@ again; relief can cancel the remaining work without a parent score or explicit capacity penalty. The Host capture/source is a separate fact. A stalled source cannot be repaired -by topology. When capture is progressing but Host-origin edges remain degraded, -healthy Peer candidates stay first. If no Peer candidate can establish a healthy -edge, the same operation may create or reuse the one bounded SFU publication as -the suffix. Host-to-SFU publication health and the first SFU Viewer's decoded -progress must both hold before commit. That SFU-fed Viewer then remains an -ordinary Peer parent, allowing `SFU -> Viewer -> Peer` distribution without -turning every Viewer into an SFU subscription. +by topology. One degraded Host-origin edge may move only to a healthy Peer. SFU +becomes a quality suffix only when every current Host-origin Peer edge, with a +minimum of two, independently remains persistently degraded. This bounded +condition identifies possible Host fanout pressure without making SFU an +ordinary candidate. Healthy Peer candidates still run first; if none succeeds, +the same operation may create or reuse the single Host publication. The exact +Host-to-SFU publication generation must supply three consecutive healthy sender +windows before commit. The old Peer route and candidate SFU subscription overlap +while the same Viewer gathers three complete receiver windows. Each candidate +window must decode without a freeze or pause and must +not regress delivered pixel area, rounded frame rate, or bitrate against a fresh +old-route window. This strict partial order has no weights or tradeoff score; an +unknown, incomparable, or worse candidate never commits. One Viewer moves, then +all edge state is re-observed before another Host-relief move. SFU availability +fallback remains independent. An SFU-fed Viewer remains an ordinary Peer parent, +allowing `SFU -> Viewer -> Peer` distribution without turning every Viewer into +an SFU subscription. The shared publication is only resource topology: its +Host-to-SFU ingress proof and each exact SFU-to-Viewer subscription proof remain +separate. A successful Viewer canary cannot authorize any other Viewer, and a +new publication generation invalidates all prior ingress authority. After commit, the new active identity starts from unknown and must establish a fresh healthy delta before a later degradation can trigger another move. There diff --git a/docs/product/routing-transport.md b/docs/product/routing-transport.md index 827316b..c9454bd 100644 --- a/docs/product/routing-transport.md +++ b/docs/product/routing-transport.md @@ -91,11 +91,19 @@ credentials cannot recreate off-ledger media. ADR-0005 owns that lifecycle. WebRTC and LiveKit own ICE, consent, congestion control, bitrate, frame rate, resolution, retransmission, reconnect, and SFU layer selection. When the per-share convergence gate is enabled, Screener uses their exact categorical -edge state only to move one degraded local edge through the existing serial -operation; disabling the gate leaves that state diagnostic. Screener has no -weighted route score, parent-wide quality inference, all-pairs probe, periodic -rebalance, persistent parent blacklist, NAT classification, or independent -depth cap. +edge state only to move one persistently degraded local edge through the +existing serial operation. Ordinary quality moves remain Peer-to-Peer. SFU may +join quality work only as Host fanout relief when every current Host-origin Peer +edge, with at least two such edges, is persistently degraded; one Viewer moves +before the controller observes the new topology again. During that overlap, the +same Viewer must prove the SFU candidate does not regress delivered resolution, +frame rate, or bitrate across persistent complete windows and produces no +freeze or pause; otherwise the old Peer route stays. The Host publication +ingress and every Viewer subscription are separate quality paths. One Viewer's +SFU proof never authorizes another Viewer. Disabling the gate leaves quality +state diagnostic. Screener has no weighted route score, general parent-wide +prediction, all-pairs probe, periodic rebalance, persistent parent blacklist, +NAT classification, or independent depth cap. Direct P2P exposes endpoint network addresses to the trusted peer. SFU media is encrypted hop-by-hop with DTLS-SRTP but terminates at the SFU; the product does diff --git a/docs/project-memory.md b/docs/project-memory.md index 4067158..ea36bb2 100644 --- a/docs/project-memory.md +++ b/docs/project-memory.md @@ -38,10 +38,11 @@ implementation and routine UI detail. one serial child operation. First decoded frame commits availability work; accepted quality work adds native edge proof before commit in that operation. - WebRTC and LiveKit own congestion control, media adaptation, reconnect, and SFU - layers. Current source implements opt-in, event-driven local topology - convergence from their native edge state, with bounded SFU only after healthy - Peer escape fails. No weighted score, all-pairs probe, parent-wide prediction, - or periodic rebalance is accepted. + layers. Current source uses persistent native edge state for opt-in local P2P + convergence. SFU quality work is limited to multi-root Host fanout relief and + each Viewer must prove its own candidate non-regression before commit. No + weighted score, all-pairs probe, general parent-wide prediction, or periodic + rebalance is accepted. - Browser video uses `motion`; each share chooses H.264 with VP8 fallback through an actual sender probe unless the Host explicitly selects VP8 or H264. Screen audio uses 64/128/192 kbps ceilings with 128 default. @@ -51,10 +52,12 @@ implementation and routine UI detail. ## Current Snapshot -Canonical source and production use the strict `screener-v14` Browser/server -contract with the opt-in native-edge convergence controller, optional SQLite -room authority, and the H.264/VP8 sender gate. [Status](./status.md) is the sole -owner of exact source, artifact, deployment, milestone, and blocker identity. +Canonical source uses the strict `screener-v15` Browser/server contract with the +corrected opt-in native-edge convergence controller. Production remains on +`screener-v14` until the correction release completes. Both retain optional +SQLite room authority and the H.264/VP8 sender gate. [Status](./status.md) is the +sole owner of exact source, artifact, deployment, milestone, and blocker +identity. Current work is owned by the [TODO ledger](./todo.md). Open physical evidence is owned by [verification status](./verification-status.md). Environment and initial diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index d1cdbe9..8949880 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -39,6 +39,11 @@ Removed access, room TTL, endpoint-tier, room-rollout, and TURN variables fail startup even when blank. The private deployment is upgraded atomically; there are no compatibility aliases or dual configuration readers. +During pre-release route canaries, standard Node `NODE_DEBUG=screener-route` +enables sanitized room and participant-ordinal events. It records route reasons, +candidates, revisions, quality states and commit/failure outcomes, but not raw +Peer IDs, SDP, ICE candidates, tokens or media credentials. + ## Public And Private Ports | Port | Scope | Owner | diff --git a/docs/research/realtime-quality-adaptation.md b/docs/research/realtime-quality-adaptation.md index ff6d39d..2d6110d 100644 --- a/docs/research/realtime-quality-adaptation.md +++ b/docs/research/realtime-quality-adaptation.md @@ -209,6 +209,24 @@ One report is still not a future guarantee, so a candidate must carry real media while the working route remains and prove a complete healthy delta before commit. Failure remains inconclusive without cutting the old route. +Production observation on 2026-08-27 showed that one two-second degraded delta +could move an otherwise usable Host Peer edge to an already-active SFU whose +Viewer experience was worse. The SFU proof established healthy publication +ingress and decoded progress, not superiority over the old path. Active routing +therefore needs the existing persistent-limitation semantic, and SFU quality use +must be limited to multi-edge Host fanout relief rather than ordinary edge +replacement. + +LiveKit exposes one participant-level `ConnectionQualityInfo` with no direction +field, while stream state only identifies an SFU-paused subscription. Neither +proves that a candidate SFU path is visually better than a retained P2P path. +The bounded canary therefore compares the same Viewer's simultaneous inbound +WebRTC windows by strict non-regression of delivered dimensions, frame rate, and +bitrate, with decoded progress and no freeze or pause. It does not combine them +into a scalar score or infer unconnected path quality. The single SFU +publication does not imply one quality node: Host ingress is generation-scoped, +and every Viewer subscription needs its own candidate proof. + SFU has asymmetric evidence. The Browser publication proves shared Host-to-SFU ingress, while exact SFU-to-Viewer sending and layer selection live inside LiveKit. The accepted model therefore keeps SFU as a bounded suffix: LiveKit @@ -250,6 +268,7 @@ percentages, and tuning loops are not adopted for this Browser product. - [LiveKit screen-share encodings](https://github.com/livekit/client-sdk-js/blob/v2.22.0/src/room/participant/publishUtils.ts) - [LiveKit Dynacast and simulcast](https://docs.livekit.io/transport/media/advanced/) - [LiveKit server forwarder](https://github.com/livekit/livekit/blob/v1.13.5/pkg/sfu/forwarder.go) +- [LiveKit connection-quality protocol](https://github.com/livekit/protocol/blob/main/protobufs/livekit_rtc.proto) - [Screego codec ordering](https://github.com/screego/server/blob/v1.12.4/ui/src/useRoom.ts) - [Screego VP9 regression](https://github.com/screego/server/pull/132) - [Media Capabilities](https://www.w3.org/TR/media-capabilities/) diff --git a/docs/status.md b/docs/status.md index 5699f26..3a4339f 100644 --- a/docs/status.md +++ b/docs/status.md @@ -45,7 +45,7 @@ This is the current execution index. Git history owns completed timelines; ## Current Source -- Canonical source uses the strict `screener-v14` Browser/server contract. +- Canonical source uses the strict `screener-v15` Browser/server contract. Optional SQLite stable authority, non-expiring local preferred code, atomic room replacement and the Host codec selector remain implemented; production selects stable storage. Graceful restart, crash, timeout and network loss use @@ -72,9 +72,10 @@ This is the current execution index. Git history owns completed timelines; ## Current Milestone -Native-edge local convergence is implemented and deployed behind its default-off -per-share policy. Automated route invariants and release postflight pass; -representative public-network benefit remains a physical evidence boundary. +Native-edge local convergence is deployed behind its default-off per-share +policy. Production evidence found that one limitation window could move a usable +Host Peer edge to a worse SFU path. The active correction requires persistent +degradation and reserves SFU quality work for multi-edge Host fanout relief. ## Active Boundaries @@ -91,4 +92,8 @@ representative public-network benefit remains a physical evidence boundary. ## Current Hold -No source or deployment P0/P1 is open. Broader product evidence remains parked. +The opt-in quality path has a P1 misrouting hold until the persistence and SFU +eligibility correction is deployed. Availability routing remains operational; +Hosts should leave topology optimization disabled meanwhile. Sanitized route +event logging is enabled for the canary. Broader product evidence remains +parked. diff --git a/docs/todo.md b/docs/todo.md index abff0ee..2d8979b 100644 --- a/docs/todo.md +++ b/docs/todo.md @@ -7,7 +7,10 @@ topics are not implementation authority. ## Now -No implementation task is currently authorized. +1. **Correct native-edge convergence.** Require persistent degraded evidence, + keep ordinary quality moves P2P-only, reserve SFU quality work for multi-edge + Host fanout relief, require same-Viewer SFU non-regression before commit, and + retain sanitized route-event evidence for the canary. ## Parked Repository Work diff --git a/src/client/media/route-transition.ts b/src/client/media/route-transition.ts index 6935e7f..c786665 100644 --- a/src/client/media/route-transition.ts +++ b/src/client/media/route-transition.ts @@ -151,7 +151,8 @@ function sameCandidate( return ( left?.childPeerId === right.childPeerId && left.connectionId === right.connectionId && - left.transport === right.transport + left.transport === right.transport && + left.qualityProbe === right.qualityProbe ); } diff --git a/src/client/media/sender-quality-evidence.ts b/src/client/media/sender-quality-evidence.ts index b51c4d8..3cefdd2 100644 --- a/src/client/media/sender-quality-evidence.ts +++ b/src/client/media/sender-quality-evidence.ts @@ -11,6 +11,8 @@ const connectionGenerations = new Map(); const publicationGenerations = new Map(); const unknownConnections = new Set(); const unknownPublications = new Set(); +const connectionSampleTimestamps = new Map(); +const publicationSampleTimestamps = new Map(); export function invalidateSenderQualityEvidence(): void { senderEvidenceGeneration += 1; @@ -18,6 +20,8 @@ export function invalidateSenderQualityEvidence(): void { publicationGenerations.clear(); unknownConnections.clear(); unknownPublications.clear(); + connectionSampleTimestamps.clear(); + publicationSampleTimestamps.clear(); } function ownsCurrentEvidenceGeneration( @@ -46,6 +50,7 @@ function nativeQualityState( (metrics.nativeEdgeQualityState !== "healthy" && metrics.nativeEdgeQualityState !== "degraded") || metrics.sampleWindowMs === null || + metrics.sampleTimestampMs === null || metrics.intervalFramesEncoded === null || metrics.intervalFramesEncoded <= 0 ) { @@ -63,6 +68,7 @@ export function senderQualityEvidenceFromSnapshot( routeRevision: number, ): Extract | null { const evidence = nativeQualityState(snapshot.metrics); + const rawSampleTimestampMs = snapshot.metrics.sampleTimestampMs; const hasIdentity = snapshot.metrics.rtpStatsId !== null && snapshot.metrics.trackIdentifier !== null; @@ -73,6 +79,15 @@ export function senderQualityEvidenceFromSnapshot( ) && hasIdentity ? evidence : "unknown"; + const sampleTimestampMs = + state === "unknown" ? null : snapshot.metrics.sampleTimestampMs; + if ( + sampleTimestampMs !== null && + (connectionSampleTimestamps.get(snapshot.connectionId) ?? -1) >= + sampleTimestampMs + ) { + return null; + } const parsed = senderQualityEvidenceMessageSchema.safeParse({ type: "sender-quality-evidence", childPeerId: snapshot.peerId, @@ -83,17 +98,28 @@ export function senderQualityEvidenceFromSnapshot( state === "unknown" ? null : snapshot.metrics.trackIdentifier, + sampleTimestampMs, routeRevision, state, }); + if (!parsed.success) { + return null; + } if (state === "unknown") { if (!emitUnknownOnce(unknownConnections, snapshot.connectionId)) { return null; } + if (rawSampleTimestampMs !== null) { + connectionSampleTimestamps.set( + snapshot.connectionId, + rawSampleTimestampMs, + ); + } } else { unknownConnections.delete(snapshot.connectionId); + connectionSampleTimestamps.set(snapshot.connectionId, sampleTimestampMs!); } - return parsed.success ? parsed.data : null; + return parsed.data; } export function sfuPublisherQualityEvidenceFromMetrics( @@ -105,24 +131,45 @@ export function sfuPublisherQualityEvidenceFromMetrics( { type: "sfu-publisher-quality-evidence" } > | null { const evidence = nativeQualityState(metrics); + const rawSampleTimestampMs = metrics.sampleTimestampMs; const state = ownsCurrentEvidenceGeneration( publicationGenerations, publicationGeneration, ) ? evidence : "unknown"; + const sampleTimestampMs = + state === "unknown" ? null : metrics.sampleTimestampMs; + if ( + sampleTimestampMs !== null && + (publicationSampleTimestamps.get(publicationGeneration) ?? -1) >= + sampleTimestampMs + ) { + return null; + } const parsed = sfuPublisherQualityEvidenceMessageSchema.safeParse({ type: "sfu-publisher-quality-evidence", routeRevision, publicationGeneration, state, + sampleTimestampMs, }); + if (!parsed.success) { + return null; + } if (state === "unknown") { if (!emitUnknownOnce(unknownPublications, publicationGeneration)) { return null; } + if (rawSampleTimestampMs !== null) { + publicationSampleTimestamps.set( + publicationGeneration, + rawSampleTimestampMs, + ); + } } else { unknownPublications.delete(publicationGeneration); + publicationSampleTimestamps.set(publicationGeneration, sampleTimestampMs!); } - return parsed.success ? parsed.data : null; + return parsed.data; } diff --git a/src/client/media/sfu-quality-probe.ts b/src/client/media/sfu-quality-probe.ts new file mode 100644 index 0000000..05f4e3d --- /dev/null +++ b/src/client/media/sfu-quality-probe.ts @@ -0,0 +1,132 @@ +import { + PERSISTENT_NATIVE_EDGE_DEGRADED_WINDOWS, + VIEWER_QUALITY_EVIDENCE_EXPIRY_MS, +} from "../../shared/protocol"; +import type { ConnectionMetrics } from "../types"; + +interface DeliveredVideoSample { + timestampMs: number; + pixels: number; + framesPerSecond: number; + bitrateKbps: number; +} + +function deliveredVideoSample( + metrics: ConnectionMetrics, +): DeliveredVideoSample | null { + const { + sampleTimestampMs, + sampleWindowMs, + frameWidth, + frameHeight, + framesPerSecond, + bitrateKbps, + intervalFramesDecoded, + } = metrics; + if ( + sampleTimestampMs === null || + sampleWindowMs === null || + frameWidth === null || + frameHeight === null || + framesPerSecond === null || + bitrateKbps === null || + intervalFramesDecoded === null || + !Number.isFinite(sampleTimestampMs) || + !Number.isFinite(sampleWindowMs) || + !Number.isFinite(frameWidth) || + !Number.isFinite(frameHeight) || + !Number.isFinite(framesPerSecond) || + !Number.isFinite(bitrateKbps) || + frameWidth <= 0 || + frameHeight <= 0 || + sampleWindowMs < 1_000 || + sampleWindowMs > 5_000 || + framesPerSecond <= 0 || + bitrateKbps <= 0 || + intervalFramesDecoded <= 0 + ) { + return null; + } + return { + timestampMs: sampleTimestampMs, + pixels: frameWidth * frameHeight, + framesPerSecond: Math.round(framesPerSecond), + bitrateKbps, + }; +} + +export function sfuCandidateDoesNotRegress( + current: ConnectionMetrics, + candidate: ConnectionMetrics, +): boolean | null { + const currentVideo = deliveredVideoSample(current); + const candidateVideo = deliveredVideoSample(candidate); + if ( + !currentVideo || + !candidateVideo || + Math.abs(candidateVideo.timestampMs - currentVideo.timestampMs) > + VIEWER_QUALITY_EVIDENCE_EXPIRY_MS || + candidate.intervalFreezeCount === null || + candidate.intervalFreezeDurationMs === null || + candidate.intervalPauseCount === null || + candidate.intervalPauseDurationMs === null + ) { + return null; + } + return ( + candidate.intervalFreezeCount === 0 && + candidate.intervalFreezeDurationMs === 0 && + candidate.intervalPauseCount === 0 && + candidate.intervalPauseDurationMs === 0 && + candidateVideo.pixels >= currentVideo.pixels && + candidateVideo.framesPerSecond >= currentVideo.framesPerSecond && + candidateVideo.bitrateKbps >= currentVideo.bitrateKbps + ); +} + +export class SfuQualityProbe { + private lastCandidateTimestampMs: number | null = null; + private lastCurrentTimestampMs: number | null = null; + private consecutiveNonRegressingWindows = 0; + + observe( + current: ConnectionMetrics | null, + candidate: ConnectionMetrics, + ): boolean { + if ( + candidate.sampleTimestampMs === null || + candidate.sampleTimestampMs === this.lastCandidateTimestampMs + ) { + return false; + } + this.lastCandidateTimestampMs = candidate.sampleTimestampMs; + if ( + !current || + current.sampleTimestampMs === null || + (this.lastCurrentTimestampMs !== null && + current.sampleTimestampMs <= this.lastCurrentTimestampMs) + ) { + this.consecutiveNonRegressingWindows = 0; + return false; + } + this.lastCurrentTimestampMs = current.sampleTimestampMs; + const nonRegressing = sfuCandidateDoesNotRegress(current, candidate); + if (nonRegressing === null) { + this.consecutiveNonRegressingWindows = 0; + return false; + } + this.consecutiveNonRegressingWindows = nonRegressing + ? this.consecutiveNonRegressingWindows + 1 + : 0; + return ( + this.consecutiveNonRegressingWindows >= + PERSISTENT_NATIVE_EDGE_DEGRADED_WINDOWS + ); + } + + reset(): void { + this.lastCandidateTimestampMs = null; + this.lastCurrentTimestampMs = null; + this.consecutiveNonRegressingWindows = 0; + } +} diff --git a/src/client/media/viewer-sfu-route.ts b/src/client/media/viewer-sfu-route.ts index b92621a..bc45d4c 100644 --- a/src/client/media/viewer-sfu-route.ts +++ b/src/client/media/viewer-sfu-route.ts @@ -8,6 +8,7 @@ import type { import { SfuSubscriber } from "../sfu/subscriber"; import type { ConnectionMetrics } from "../types"; import type { SfuConnectionConfig } from "../sfu/publisher"; +import { SfuQualityProbe } from "./sfu-quality-probe"; import { MediaRouteTransition, reportActivePeerRouteFailure, @@ -35,6 +36,8 @@ interface ViewerSubscriberSlot { stream: MediaStream | null; decodedFrame: boolean; readySent: boolean; + qualityProbe: SfuQualityProbe | null; + qualityApproved: boolean; failed: boolean; activationToken: RouteOperationToken | null; } @@ -68,6 +71,8 @@ interface ViewerSfuRouteEvents { ) => void; onSfuVideoAvailability?: (available: boolean, revision: number) => void; onSfuUpdate?: (metrics: ConnectionMetrics | null, revision: number) => void; + currentPeerMetrics?: () => ConnectionMetrics | null; + qualityProbeEligible?: () => boolean; onSfuDecodedFrameSample?: ( framesDecodedDelta: number | null, revision: number, @@ -207,11 +212,24 @@ export class ViewerSfuRoute { setPaused(paused: boolean): void { this.paused = paused; + if (paused) { + this.pending?.qualityProbe?.reset(); + if (this.pending) { + this.pending.qualityApproved = false; + } + } if (paused) { this.discardPending(); } } + resetQualityProbe(): void { + this.pending?.qualityProbe?.reset(); + if (this.pending) { + this.pending.qualityApproved = false; + } + } + reconnectActive(): boolean { const revision = this.route.getRevision(); const assignment = this.route.getActiveAssignment(); @@ -379,6 +397,23 @@ export class ViewerSfuRoute { onStats: (metrics: ConnectionMetrics) => { if (this.active === slot && !slot.failed) { this.events.onSfuUpdate?.(metrics, slot.revision); + } else if ( + this.pending === slot && + !slot.failed && + slot.qualityProbe + ) { + if (this.events.qualityProbeEligible?.() === false) { + slot.qualityProbe.reset(); + slot.qualityApproved = false; + } else if ( + slot.qualityProbe.observe( + this.events.currentPeerMetrics?.() ?? null, + metrics, + ) + ) { + slot.qualityApproved = true; + this.sendPendingReady(slot); + } } }, onDecodedFrameSample: (framesDecodedDelta: number | null) => { @@ -400,6 +435,7 @@ export class ViewerSfuRoute { const subscriber = this.events.createSubscriber?.(subscriberEvents) ?? new SfuSubscriber(subscriberEvents); + const preparedCandidate = this.route.getPreparedCandidate(); slot = { revision: message.revision, publicationGeneration, @@ -410,6 +446,12 @@ export class ViewerSfuRoute { stream: null, decodedFrame: false, readySent: false, + qualityProbe: + preparedCandidate?.transport === "sfu" && + preparedCandidate.qualityProbe + ? new SfuQualityProbe() + : null, + qualityApproved: false, failed: false, activationToken: null, }; @@ -653,15 +695,33 @@ export class ViewerSfuRoute { return true; } slot.decodedFrame = true; - if (phase === "prepare" && !slot.readySent) { - slot.readySent = this.events.send({ - type: "route-ready", - revision: slot.revision, - phase: "prepare", - }); + if (phase === "prepare") { + this.sendPendingReady(slot); + return slot.qualityProbe !== null || slot.readySent; } void this.queueTransition(() => this.promotePendingSfu(slot)); - return phase !== "prepare" || slot.readySent; + return true; + } + + private sendPendingReady(slot: ViewerSubscriberSlot): void { + if ( + this.pending !== slot || + slot.failed || + slot.readySent || + !slot.decodedFrame || + (slot.qualityProbe !== null && !slot.qualityApproved) || + (slot.qualityProbe !== null && + this.events.qualityProbeEligible?.() === false) || + this.route.getPhase() !== "prepare" || + this.route.getRevision() !== slot.revision + ) { + return; + } + slot.readySent = this.events.send({ + type: "route-ready", + revision: slot.revision, + phase: "prepare", + }); } private async promotePendingSfu(slot: ViewerSubscriberSlot): Promise { diff --git a/src/client/pages/ViewerPage.tsx b/src/client/pages/ViewerPage.tsx index 7d9f3c1..de94097 100644 --- a/src/client/pages/ViewerPage.tsx +++ b/src/client/pages/ViewerPage.tsx @@ -420,6 +420,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { let currentRouteRevision = 0; let currentRouteAssignment: ParticipantRouteAssignment | null = null; let currentRouteConnectionId: string | null = null; + let activePeerMetrics: ConnectionMetrics | null = null; let pendingRouteConnection: { revision: number; connectionId: string; @@ -456,6 +457,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { invalidatePresentedMedia(); const newlySuspended = !pageSuspended; pageSuspended = true; + viewerSfuRoute?.resetQualityProbe(); syncDecodedFrameStallPause(); if (newlySuspended) { invalidateSenderQualityEvidence(); @@ -610,6 +612,9 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { currentRouteAssignment = assignment; currentRouteConnectionId = assignment.upstream.kind === "none" ? null : connectionId; + if (assignment.upstream.kind !== "peer") { + activePeerMetrics = null; + } pendingRouteConnection = null; } @@ -980,6 +985,8 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { ); } }, + currentPeerMetrics: () => activePeerMetrics, + qualityProbeEligible: () => !pageSuspended, onSfuState: (state, revision) => { if (active && viewerSfuRoute === route) { const connected = state === "connected"; @@ -1101,6 +1108,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { const peer = peerRef.current; peer?.dispose(); peerRef.current = null; + activePeerMetrics = null; setPeerSnapshot(null); if (clearMedia) { clearRemoteMedia(); @@ -1159,6 +1167,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { probe.snapshot = snapshot; provePendingPeer(); } else if (active && peerRef.current === peer) { + activePeerMetrics = snapshot.metrics; observeActiveDecodedFrames( "peer", `${probe.parentPeerId}:${snapshot.connectionId}`, @@ -1245,6 +1254,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { }, onUpdate: (snapshot) => { if (active) { + activePeerMetrics = snapshot.metrics; observeActiveDecodedFrames( "peer", `${snapshot.peerId}:${snapshot.connectionId}`, diff --git a/src/client/webrtc/host-peer.ts b/src/client/webrtc/host-peer.ts index b11b98b..5f96333 100644 --- a/src/client/webrtc/host-peer.ts +++ b/src/client/webrtc/host-peer.ts @@ -1,4 +1,8 @@ -import type { IceConfig, SignalPayload } from "../../shared/protocol"; +import { + PERSISTENT_NATIVE_EDGE_DEGRADED_WINDOWS, + type IceConfig, + type SignalPayload, +} from "../../shared/protocol"; import { createOpaqueId } from "../lib/opaque-id"; import { audioSenderParameterWarning, @@ -671,7 +675,9 @@ export class HostPeer { } private persistentLimitationWarning(): string | null { - if (this.limitationSamples < 3) { + if ( + this.limitationSamples < PERSISTENT_NATIVE_EDGE_DEGRADED_WINDOWS + ) { return null; } switch (this.limitationReason) { diff --git a/src/server/hybrid-media-router.ts b/src/server/hybrid-media-router.ts index 5acc4dd..376f979 100644 --- a/src/server/hybrid-media-router.ts +++ b/src/server/hybrid-media-router.ts @@ -299,6 +299,7 @@ export class HybridMediaRouter { message.rtpStatsId && message.trackIdentifier ? `${message.rtpStatsId}\u0000${message.trackIdentifier}` : null, + sampleTimestampMs: message.sampleTimestampMs, state: message.state, acceptedAtMs: this.now(), }; @@ -351,6 +352,7 @@ export class HybridMediaRouter { publicationGeneration: message.publicationGeneration, routeRevision: message.routeRevision, state: message.state, + sampleTimestampMs: message.sampleTimestampMs, acceptedAtMs: this.now(), }; const result = controller.observeSfuPublisherQualityEvidence( @@ -1951,6 +1953,7 @@ function preparedRouteCandidate( childPeerId: operation.childPeerId, connectionId, transport: tuple.kind === "peer" ? tuple.transport : "sfu", + qualityProbe: operation.reason === "quality-convergence", }; } diff --git a/src/server/room-route-controller.ts b/src/server/room-route-controller.ts index 6ac55d8..b635719 100644 --- a/src/server/room-route-controller.ts +++ b/src/server/room-route-controller.ts @@ -3,6 +3,7 @@ import { debuglog } from "node:util"; import { MAX_MEDIA_ROUTE_REVISION, + PERSISTENT_NATIVE_EDGE_DEGRADED_WINDOWS, VIEWER_QUALITY_EVIDENCE_EXPIRY_MS, type RouteDemandReason, type RouteDiagnosticFinalRoute, @@ -126,6 +127,9 @@ interface Attempt { mediaReady: boolean; senderQualityState?: "healthy" | "degraded"; senderQualityAcceptedAtMs?: number; + senderQualityConsecutiveHealthyWindows?: number; + senderQualitySampleTimestampMs?: number; + senderQualityIdentity?: string; } interface ChildOperation { @@ -252,6 +256,8 @@ interface SenderQualityObservation { connectionId: string; senderIdentity: string; state: "healthy" | "degraded"; + consecutiveDegradedWindows: number; + lastSampleTimestampMs: number; lastAcceptedAtMs: number; consumedAtFactVersion?: number; } @@ -263,6 +269,7 @@ export interface SenderQualityEvidenceInput { routeRevision: number; connectionId: string; senderIdentity: string | null; + sampleTimestampMs: number | null; state: "unknown" | "healthy" | "degraded"; acceptedAtMs: number; } @@ -279,6 +286,8 @@ interface SfuPublisherQualityObservation { hostSessionId: string; publicationGeneration: string; state: "healthy" | "degraded"; + consecutiveHealthyWindows: number; + lastSampleTimestampMs: number; lastAcceptedAtMs: number; } @@ -288,6 +297,7 @@ export interface SfuPublisherQualityEvidenceInput { publicationGeneration: string; routeRevision: number; state: "unknown" | "healthy" | "degraded"; + sampleTimestampMs: number | null; acceptedAtMs: number; } @@ -941,6 +951,7 @@ export class RoomRouteController { input.state, input.acceptedAtMs, input.senderIdentity, + input.sampleTimestampMs, commitReservation, ); } @@ -959,6 +970,12 @@ export class RoomRouteController { return rejected(); } if (input.state === "unknown") { + if (this.senderQualityObservations.has(input.childPeerId)) { + this.debug("sender-quality-reset", { + parent: this.debugPeer(input.parentPeerId), + child: this.debugPeer(input.childPeerId), + }); + } this.senderQualityObservations.delete(input.childPeerId); this.senderQualityBaselines.set( input.childPeerId, @@ -977,7 +994,7 @@ export class RoomRouteController { released, }; } - if (!input.senderIdentity) { + if (!input.senderIdentity || input.sampleTimestampMs === null) { return rejected(); } const previous = this.senderQualityObservations.get(input.childPeerId); @@ -988,6 +1005,24 @@ export class RoomRouteController { previous.connectionId === input.connectionId; const sameSenderIdentity = sameIdentity && previous?.senderIdentity === input.senderIdentity; + if ( + sameSenderIdentity && + previous !== undefined && + input.sampleTimestampMs <= previous.lastSampleTimestampMs + ) { + return { + accepted: true, + committed: false, + failedPeerIds: [], + activeRevision: this.revision, + released: [], + }; + } + const sameFreshSenderIdentity = + sameSenderIdentity && + previous !== undefined && + input.acceptedAtMs < + previous.lastAcceptedAtMs + VIEWER_QUALITY_EVIDENCE_EXPIRY_MS; const senderIdentityChanged = sameIdentity && previous?.senderIdentity !== input.senderIdentity; const wasFreshHealthy = @@ -1003,6 +1038,12 @@ export class RoomRouteController { if (input.state === "healthy") { this.senderQualityBaselines.delete(input.childPeerId); } + const consecutiveDegradedWindows = + input.state === "degraded" + ? sameFreshSenderIdentity && previous.state === "degraded" + ? safeAdd(previous.consecutiveDegradedWindows, 1) + : 1 + : 0; this.senderQualityObservations.set(input.childPeerId, { childSessionId: child.sessionId, parentPeerId: input.parentPeerId, @@ -1010,6 +1051,8 @@ export class RoomRouteController { connectionId: input.connectionId, senderIdentity: input.senderIdentity, state: input.state, + consecutiveDegradedWindows, + lastSampleTimestampMs: input.sampleTimestampMs, lastAcceptedAtMs: input.acceptedAtMs, consumedAtFactVersion: (baselineRequired || senderIdentityChanged) && @@ -1021,6 +1064,19 @@ export class RoomRouteController { ? previous.consumedAtFactVersion : undefined, }); + if ( + !sameFreshSenderIdentity || + previous.state !== input.state || + consecutiveDegradedWindows === + PERSISTENT_NATIVE_EDGE_DEGRADED_WINDOWS + ) { + this.debug("sender-quality-observed", { + parent: this.debugPeer(input.parentPeerId), + child: this.debugPeer(input.childPeerId), + state: input.state, + consecutiveDegradedWindows, + }); + } if ( input.state === "healthy" && !wasFreshHealthy && @@ -1089,6 +1145,7 @@ export class RoomRouteController { input.state, input.acceptedAtMs, input.publicationGeneration, + input.sampleTimestampMs, commitReservation, ); } @@ -1102,6 +1159,9 @@ export class RoomRouteController { return rejected(); } if (input.state === "unknown") { + if (this.sfuPublisherQualityObservation) { + this.debug("sfu-publisher-quality-reset", {}); + } this.sfuPublisherQualityObservation = undefined; return { accepted: true, @@ -1111,22 +1171,102 @@ export class RoomRouteController { released: [], }; } + if (input.sampleTimestampMs === null) { + return rejected(); + } const previous = this.sfuPublisherQualityObservation; - const wasFreshHealthy = + if ( + previous?.hostSessionId === input.hostSessionId && + previous.publicationGeneration === input.publicationGeneration && + input.sampleTimestampMs <= previous.lastSampleTimestampMs + ) { + return { + accepted: true, + committed: false, + failedPeerIds: [], + activeRevision: this.revision, + released: [], + }; + } + const sameFreshPublication = previous?.hostSessionId === input.hostSessionId && previous.publicationGeneration === input.publicationGeneration && - previous.state === "healthy" && input.acceptedAtMs < previous.lastAcceptedAtMs + VIEWER_QUALITY_EVIDENCE_EXPIRY_MS; + const consecutiveHealthyWindows = + input.state === "healthy" + ? sameFreshPublication && previous.state === "healthy" + ? safeAdd(previous.consecutiveHealthyWindows, 1) + : 1 + : 0; + const wasPersistentlyHealthy = + sameFreshPublication && + previous.state === "healthy" && + previous.consecutiveHealthyWindows >= + PERSISTENT_NATIVE_EDGE_DEGRADED_WINDOWS; this.sfuPublisherQualityObservation = { hostSessionId: input.hostSessionId, publicationGeneration: input.publicationGeneration, state: input.state, + consecutiveHealthyWindows, + lastSampleTimestampMs: input.sampleTimestampMs, lastAcceptedAtMs: input.acceptedAtMs, }; - if (input.state === "healthy" && !wasFreshHealthy) { + if ( + previous?.hostSessionId !== input.hostSessionId || + previous.publicationGeneration !== input.publicationGeneration || + previous.state !== input.state || + consecutiveHealthyWindows === + PERSISTENT_NATIVE_EDGE_DEGRADED_WINDOWS + ) { + this.debug("sfu-publisher-quality-observed", { + state: input.state, + consecutiveHealthyWindows, + }); + } + if ( + input.state === "healthy" && + consecutiveHealthyWindows >= + PERSISTENT_NATIVE_EDGE_DEGRADED_WINDOWS && + !wasPersistentlyHealthy + ) { this.touchFacts(); } + if ( + operation?.reason === "quality-convergence" && + attempt?.tuple.kind === "sfu" && + attempt.tuple.publication === "reuse" && + attempt.hostSessionId === input.hostSessionId && + attempt.publicationGeneration === input.publicationGeneration && + attempt.mediaReady + ) { + const guard = { + childPeerId: operation.childPeerId, + childSessionId: operation.childSessionId, + revision: attempt.revision, + connectionId: attempt.connectionId, + }; + const settled = + input.state === "degraded" + ? this.candidateFailed(guard, input.acceptedAtMs) + : consecutiveHealthyWindows >= + PERSISTENT_NATIVE_EDGE_DEGRADED_WINDOWS + ? this.candidateReady( + guard, + input.acceptedAtMs, + commitReservation, + ) + : null; + if (settled) { + return { + accepted: settled.accepted, + committed: settled.committed === true, + failedPeerIds: settled.failedPeerIds, + activeRevision: settled.activeRevision, + released: settled.released, + }; + } + } return { accepted: true, committed: false, @@ -1142,13 +1282,23 @@ export class RoomRouteController { state: "unknown" | "healthy" | "degraded", acceptedAtMs: number, senderIdentity: string | null, + sampleTimestampMs: number | null, commitReservation: ( reservation: CandidateReservation, ) => boolean, ): SenderQualityEvidenceResult { + if (attempt.senderQualityState !== state) { + this.debug("candidate-quality-observed", { + child: this.debugPeer(operation.childPeerId), + candidate: this.debugTuple(attempt.tuple), + state, + }); + } if (state === "unknown") { attempt.senderQualityState = undefined; attempt.senderQualityAcceptedAtMs = undefined; + attempt.senderQualityConsecutiveHealthyWindows = undefined; + attempt.senderQualitySampleTimestampMs = undefined; return { accepted: true, committed: false, @@ -1157,7 +1307,7 @@ export class RoomRouteController { released: [], }; } - if (!senderIdentity) { + if (!senderIdentity || sampleTimestampMs === null) { return { accepted: false, committed: false, @@ -1166,8 +1316,60 @@ export class RoomRouteController { released: [], }; } + if ( + attempt.senderQualityIdentity !== undefined && + attempt.senderQualityIdentity !== senderIdentity + ) { + const failed = this.candidateFailed( + { + childPeerId: operation.childPeerId, + childSessionId: operation.childSessionId, + revision: attempt.revision, + connectionId: attempt.connectionId, + }, + acceptedAtMs, + ); + return { + accepted: failed.accepted, + committed: failed.committed === true, + failedPeerIds: failed.failedPeerIds, + activeRevision: failed.activeRevision, + released: failed.released, + }; + } + if ( + attempt.senderQualitySampleTimestampMs !== undefined && + sampleTimestampMs <= attempt.senderQualitySampleTimestampMs + ) { + return { + accepted: true, + committed: false, + failedPeerIds: [], + activeRevision: this.revision, + released: [], + }; + } + const consecutiveHealthyWindows = + attempt.tuple.kind === "sfu" && state === "healthy" + ? attempt.senderQualityState === "healthy" && + attempt.senderQualityAcceptedAtMs !== undefined && + acceptedAtMs < + attempt.senderQualityAcceptedAtMs + + VIEWER_QUALITY_EVIDENCE_EXPIRY_MS + ? safeAdd( + attempt.senderQualityConsecutiveHealthyWindows ?? 0, + 1, + ) + : 1 + : state === "healthy" + ? 1 + : 0; attempt.senderQualityState = state; attempt.senderQualityAcceptedAtMs = acceptedAtMs; + attempt.senderQualityConsecutiveHealthyWindows = + consecutiveHealthyWindows; + attempt.senderQualitySampleTimestampMs = sampleTimestampMs; + attempt.senderQualityIdentity = senderIdentity; const guard = { childPeerId: operation.childPeerId, childSessionId: operation.childSessionId, @@ -1177,7 +1379,10 @@ export class RoomRouteController { const settled = state === "degraded" ? this.candidateFailed(guard, acceptedAtMs) - : attempt.mediaReady + : attempt.mediaReady && + (attempt.tuple.kind !== "sfu" || + consecutiveHealthyWindows >= + PERSISTENT_NATIVE_EDGE_DEGRADED_WINDOWS) ? this.candidateReady(guard, acceptedAtMs, commitReservation) : null; return { @@ -1470,16 +1675,6 @@ export class RoomRouteController { if (tuple.kind === "sfu" && tuple.publication !== "reuse" && !input.publicationConnectionId) { throw new Error("SFU publication candidate needs an ingress connection identity"); } - const initialSfuQuality = - publicationGeneration && - this.sfuPublisherQualityObservation?.publicationGeneration === - publicationGeneration && - this.sfuPublisherQualityObservation.hostSessionId === hostSessionId && - input.nowMs < - this.sfuPublisherQualityObservation.lastAcceptedAtMs + - VIEWER_QUALITY_EVIDENCE_EXPIRY_MS - ? this.sfuPublisherQualityObservation - : undefined; operation.current = { tuple, revision: this.allocateRevision(), @@ -1495,12 +1690,6 @@ export class RoomRouteController { reservation: input.reservation, transportConnected: false, mediaReady: false, - ...(tuple.kind === "sfu" && initialSfuQuality - ? { - senderQualityState: initialSfuQuality.state, - senderQualityAcceptedAtMs: initialSfuQuality.lastAcceptedAtMs, - } - : {}), }; this.debug("candidate-started", { child: this.debugPeer(operation.childPeerId), @@ -1694,7 +1883,20 @@ export class RoomRouteController { ], }; } - if (attempt.senderQualityState === undefined) { + const candidateQualityState = + attempt.tuple.kind === "sfu" && attempt.tuple.publication === "reuse" + ? this.sfuPublisherQualityState( + attempt.publicationGeneration!, + nowMs, + ) + : attempt.senderQualityState ?? "unknown"; + const candidateQualityReady = + candidateQualityState === "healthy" && + (attempt.tuple.kind !== "sfu" || + attempt.tuple.publication === "reuse" || + (attempt.senderQualityConsecutiveHealthyWindows ?? 0) >= + PERSISTENT_NATIVE_EDGE_DEGRADED_WINDOWS); + if (!candidateQualityReady && candidateQualityState === "unknown") { return { accepted: true, committed: false, @@ -1704,10 +1906,12 @@ export class RoomRouteController { }; } if ( - attempt.senderQualityAcceptedAtMs === undefined || - nowMs >= - attempt.senderQualityAcceptedAtMs + - VIEWER_QUALITY_EVIDENCE_EXPIRY_MS || + ((attempt.tuple.kind !== "sfu" || + attempt.tuple.publication !== "reuse") && + (attempt.senderQualityAcceptedAtMs === undefined || + nowMs >= + attempt.senderQualityAcceptedAtMs + + VIEWER_QUALITY_EVIDENCE_EXPIRY_MS)) || (attempt.tuple.kind === "peer" && !this.qualitySourcePathHealthy( attempt.tuple.parentPeerId, @@ -1716,7 +1920,7 @@ export class RoomRouteController { ) { return this.candidateFailed(guard, nowMs); } - if (attempt.senderQualityState === "degraded") { + if (candidateQualityState === "degraded") { return this.candidateFailed(guard, nowMs); } } @@ -1916,6 +2120,13 @@ export class RoomRouteController { this.operation = undefined; return result; } + if ( + operation.reason === "quality-convergence" && + !this.qualityOperationStillEligible(operation, nowMs) + ) { + released.push(...this.abortOperation(nowMs, "aborted")); + return result; + } if (nowMs >= operation.deadlineAtMs) { const directConvergence = operation.reason === "direct-convergence"; const sfuBootstrap = operation.reason === "sfu-bootstrap"; @@ -2017,7 +2228,18 @@ export class RoomRouteController { if (attempt) { const guardFailed = failedGuard && this.guardMatches(failedGuard, operation, attempt); const plan = operation.candidates[operation.cursor]; - if (!guardFailed && plan && this.candidateValid(operation.childPeerId, plan, attempt)) return result; + const qualityPeerSourceInvalid = + operation.reason === "quality-convergence" && + attempt.tuple.kind === "peer" && + !this.qualitySourcePathHealthy(attempt.tuple.parentPeerId, nowMs); + if ( + !guardFailed && + !qualityPeerSourceInvalid && + plan && + this.candidateValid(operation.childPeerId, plan, attempt) + ) { + return result; + } if (!guardFailed) { this.noteRejection(operation.demandPeerId, "stale"); } @@ -2319,7 +2541,11 @@ export class RoomRouteController { observation.publicationGeneration === publicationGeneration && observation.hostSessionId === this.hostPublication?.hostSessionId && nowMs < observation.lastAcceptedAtMs + VIEWER_QUALITY_EVIDENCE_EXPIRY_MS - ? observation.state + ? observation.state === "healthy" && + observation.consecutiveHealthyWindows < + PERSISTENT_NATIVE_EDGE_DEGRADED_WINDOWS + ? "unknown" + : observation.state : "unknown"; } @@ -2404,7 +2630,8 @@ export class RoomRouteController { (candidate) => candidate.tuple.kind === "peer", ); const sfu = - current.parentPeerId === this.options.hostPeerId + current.parentPeerId === this.options.hostPeerId && + this.hostFanoutNeedsSfuRelief(nowMs) ? candidates.filter((candidate) => candidate.tuple.kind === "sfu") : []; return [...peers, ...sfu]; @@ -2421,7 +2648,7 @@ export class RoomRouteController { return Boolean( edge?.kind === "peer" && observation?.consumedAtFactVersion !== this.factVersion && - this.senderQualityState(peerId, edge, nowMs) === "degraded", + this.senderQualityPersistentlyDegraded(peerId, edge, nowMs), ); }) .sort( @@ -2431,6 +2658,62 @@ export class RoomRouteController { )[0]?.peerId; } + private senderQualityPersistentlyDegraded( + childPeerId: string, + edge: CommittedEdge, + nowMs: number, + ): boolean { + const observation = this.senderQualityObservations.get(childPeerId); + return ( + this.senderQualityState(childPeerId, edge, nowMs) === "degraded" && + observation !== undefined && + observation.consecutiveDegradedWindows >= + PERSISTENT_NATIVE_EDGE_DEGRADED_WINDOWS + ); + } + + private qualityOperationStillEligible( + operation: ChildOperation, + nowMs: number, + ): boolean { + const currentEdge = this.upstreamByViewer.get(operation.childPeerId); + const tuple = + operation.current?.tuple ?? + operation.candidates[operation.cursor]?.tuple; + if ( + !currentEdge || + !tuple || + !this.senderQualityPersistentlyDegraded( + operation.childPeerId, + currentEdge, + nowMs, + ) + ) { + return false; + } + return tuple.kind === "peer" || this.hostFanoutNeedsSfuRelief(nowMs); + } + + private hostFanoutNeedsSfuRelief(nowMs: number): boolean { + const hostPeerEdges = [...this.upstreamByViewer].filter( + (entry): entry is [string, Extract, { kind: "peer" }>] => { + const edge = entry[1]; + return ( + edge.kind === "peer" && + edge.parentPeerId === this.options.hostPeerId && + edge.usable && + edge.physicalActive + ); + }, + ); + return ( + hostPeerEdges.length >= 2 && + hostPeerEdges.every(([childPeerId, edge]) => + this.senderQualityPersistentlyDegraded(childPeerId, edge, nowMs), + ) + ); + } + private selectNextChild(): string | undefined { const viewers = this.availableViewers(); const departedParent = viewers.find(({ peerId }) => { diff --git a/src/shared/protocol.ts b/src/shared/protocol.ts index f19bc85..aca9fbe 100644 --- a/src/shared/protocol.ts +++ b/src/shared/protocol.ts @@ -6,7 +6,7 @@ import { isCanonicalVideoCodecEvidence } from "./video-codec-evidence.js"; export const MAX_VIEWERS_PER_ROOM_LIMIT = 20; export const MAX_PARTICIPANTS_PER_ROOM_LIMIT = MAX_VIEWERS_PER_ROOM_LIMIT + 1; export const MAX_SIGNAL_BYTES = 64 * 1024; -export const SIGNALING_PROTOCOL = "screener-v14"; +export const SIGNALING_PROTOCOL = "screener-v15"; export const SIGNAL_CLOSE_CODES = { serviceRestart: 1012, sessionReplaced: 4001, @@ -21,6 +21,7 @@ export const MAX_ICE_SERVER_URLS = 8; 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; +export const PERSISTENT_NATIVE_EDGE_DEGRADED_WINDOWS = 3; export const MAX_DISPLAY_NAME_CODE_POINTS = 24; export const DEFAULT_VIEWER_DISPLAY_NAME = "访客"; export const DEFAULT_HOST_DISPLAY_NAME_PREFIX = "分享者"; @@ -328,6 +329,7 @@ export const preparedRouteCandidateSchema = z childPeerId: opaqueIdSchema, connectionId: opaqueIdSchema, transport: z.enum(["direct", "sfu"]), + qualityProbe: z.boolean(), }) .strict(); export type PreparedRouteCandidate = z.infer< @@ -662,6 +664,12 @@ export const senderQualityEvidenceMessageSchema = z connectionId: opaqueIdSchema, rtpStatsId: z.string().min(1).max(256).nullable(), trackIdentifier: z.string().min(1).max(256).nullable(), + sampleTimestampMs: z + .number() + .finite() + .min(0) + .max(Number.MAX_SAFE_INTEGER) + .nullable(), routeRevision: mediaRouteRevisionSchema, state: z.enum(["unknown", "healthy", "degraded"]), }) @@ -670,7 +678,8 @@ export const senderQualityEvidenceMessageSchema = z if ( (message.state !== "unknown" && (message.rtpStatsId === null || - message.trackIdentifier === null)) + message.trackIdentifier === null || + message.sampleTimestampMs === null)) ) { context.addIssue({ code: "custom", @@ -686,8 +695,23 @@ export const sfuPublisherQualityEvidenceMessageSchema = z publicationGeneration: opaqueIdSchema, routeRevision: mediaRouteRevisionSchema, state: z.enum(["unknown", "healthy", "degraded"]), + sampleTimestampMs: z + .number() + .finite() + .min(0) + .max(Number.MAX_SAFE_INTEGER) + .nullable(), }) - .strict(); + .strict() + .superRefine((message, context) => { + if (message.state !== "unknown" && message.sampleTimestampMs === null) { + context.addIssue({ + code: "custom", + message: "Known SFU publisher quality needs an exact sample", + path: ["sampleTimestampMs"], + }); + } + }); const authenticateMessageSchema = z.discriminatedUnion("role", [ z diff --git a/tests/host-peer-tracks.test.ts b/tests/host-peer-tracks.test.ts index 1fd9756..d9032a2 100644 --- a/tests/host-peer-tracks.test.ts +++ b/tests/host-peer-tracks.test.ts @@ -1468,6 +1468,7 @@ function hostProvisionalInput( childPeerId: childPeerIds.at(-1) ?? "candidate-child", connectionId: `candidate-connection-${revision}`, transport: "direct" as const, + qualityProbe: false, }, assignment: hostAssignment( childPeerIds, @@ -1609,6 +1610,7 @@ describe("ViewerRelay downstream ownership", () => { childPeerId, connectionId: `relay-candidate-${revision}`, transport: "direct" as const, + qualityProbe: false, }); it("keeps a completed H264 decision across source replacement", async () => { diff --git a/tests/hybrid-media-router.test.ts b/tests/hybrid-media-router.test.ts index 1823a3d..8cc5d92 100644 --- a/tests/hybrid-media-router.test.ts +++ b/tests/hybrid-media-router.test.ts @@ -222,6 +222,7 @@ describe("HybridMediaRouter v9 runtime", () => { complete(router, first); await vi.waitFor(() => expect(preparedFor(sent, first.sessionId)).toBeDefined()); const firstPrepare = preparedFor(sent, first.sessionId)!; + expect(firstPrepare.candidate.qualityProbe).toBe(false); router.handleRouteReady(first, { type: "route-ready", revision: firstPrepare.revision, @@ -249,6 +250,7 @@ describe("HybridMediaRouter v9 runtime", () => { connectionId: secondEdge.connectionId, rtpStatsId: "second-rtp", trackIdentifier: "track", + sampleTimestampMs: 100, routeRevision: secondEdge.revision, state: "healthy", }), @@ -260,21 +262,25 @@ describe("HybridMediaRouter v9 runtime", () => { connectionId: firstEdge.connectionId, rtpStatsId: "first-rtp", trackIdentifier: "track", + sampleTimestampMs: 100, routeRevision: firstEdge.revision, state: "healthy", }), ).toBe(true); - expect( - router.observeSenderQualityEvidence(host, { - type: "sender-quality-evidence", - childPeerId: first.peerId, - connectionId: firstEdge.connectionId, - rtpStatsId: "first-rtp", - trackIdentifier: "track", - routeRevision: firstEdge.revision, - state: "degraded", - }), - ).toBe(true); + for (let window = 0; window < 3; window += 1) { + expect( + router.observeSenderQualityEvidence(host, { + type: "sender-quality-evidence", + childPeerId: first.peerId, + connectionId: firstEdge.connectionId, + rtpStatsId: "first-rtp", + trackIdentifier: "track", + sampleTimestampMs: 101 + window, + routeRevision: firstEdge.revision, + state: "degraded", + }), + ).toBe(true); + } await vi.waitFor(() => expect(preparedFor(sent, first.sessionId)?.revision).toBeGreaterThan( @@ -282,6 +288,7 @@ describe("HybridMediaRouter v9 runtime", () => { ), ); const qualityPrepare = preparedFor(sent, first.sessionId)!; + expect(qualityPrepare.candidate.qualityProbe).toBe(true); router.handleRouteReady(first, { type: "route-ready", revision: qualityPrepare.revision, @@ -298,6 +305,7 @@ describe("HybridMediaRouter v9 runtime", () => { connectionId: qualityPrepare.candidate.connectionId, rtpStatsId: "candidate-rtp", trackIdentifier: "track", + sampleTimestampMs: 200, routeRevision: qualityPrepare.revision, state: "healthy", }), @@ -379,7 +387,9 @@ describe("HybridMediaRouter v9 runtime", () => { phase: "prepare", }); const active = router.resolveActiveViewerMediaEdge(room.roomId, viewer.peerId)!; - for (const state of ["healthy", "degraded"] as const) { + for (const [index, state] of ( + ["healthy", "degraded"] as const + ).entries()) { expect( router.observeSenderQualityEvidence(host, { type: "sender-quality-evidence", @@ -387,6 +397,7 @@ describe("HybridMediaRouter v9 runtime", () => { connectionId: active.connectionId, rtpStatsId: "viewer-rtp", trackIdentifier: "track", + sampleTimestampMs: 100 + index, routeRevision: active.revision, state, }), diff --git a/tests/protocol.test.ts b/tests/protocol.test.ts index 6ae3b7a..447a06c 100644 --- a/tests/protocol.test.ts +++ b/tests/protocol.test.ts @@ -97,7 +97,7 @@ describe("client signaling protocol", () => { "utf8", ); - expect(SIGNALING_PROTOCOL).toBe("screener-v14"); + expect(SIGNALING_PROTOCOL).toBe("screener-v15"); expect(nativeWire).toMatch(/signalingProtocol\s*=\s*"screener-v6"/); }); @@ -869,6 +869,7 @@ describe("client signaling protocol", () => { connectionId: "connection_12345678", rtpStatsId: "rtp-stats-1", trackIdentifier: "track-1", + sampleTimestampMs: 2_000, routeRevision: 3, state: "degraded", } as const; @@ -880,12 +881,19 @@ describe("client signaling protocol", () => { rtpStatsId: null, }).success, ).toBe(false); + expect( + clientMessageSchema.safeParse({ + ...sender, + sampleTimestampMs: null, + }).success, + ).toBe(false); expect( clientMessageSchema.safeParse({ type: "sfu-publisher-quality-evidence", publicationGeneration: "publication_generation_12345678", routeRevision: 4, state: "healthy", + sampleTimestampMs: 2_000, }).success, ).toBe(true); }); @@ -1483,6 +1491,7 @@ describe("server signaling protocol", () => { childPeerId: "child_12345678", connectionId: "connection_12345678", transport: "direct", + qualityProbe: false, }, }).success, ).toBe(true); @@ -1496,6 +1505,7 @@ describe("server signaling protocol", () => { childPeerId: "child_12345678", connectionId: "connection_12345678", transport: "sfu", + qualityProbe: true, }, }).success, ).toBe(true); @@ -1509,6 +1519,7 @@ describe("server signaling protocol", () => { childPeerId: "child_12345678", connectionId: "connection_12345678", transport: "direct", + qualityProbe: false, unexpected: null, }, }).success, diff --git a/tests/room-route-controller.test.ts b/tests/room-route-controller.test.ts index 8b29498..1b43e0c 100644 --- a/tests/room-route-controller.test.ts +++ b/tests/room-route-controller.test.ts @@ -129,6 +129,82 @@ function peerEdge( }; } +function observeSenderState( + routes: RoomRouteController, + input: { + childPeerId: string; + connectionId: string; + state: "healthy" | "degraded"; + acceptedAtMs: number; + parentPeerId?: string; + }, +) { + const parentPeerId = input.parentPeerId ?? HOST; + return routes.observeSenderQualityEvidence({ + parentPeerId, + parentSessionId: + parentPeerId === HOST ? "host_session" : `${parentPeerId}_session`, + childPeerId: input.childPeerId, + routeRevision: routes.snapshot().revision, + connectionId: input.connectionId, + senderIdentity: `${input.connectionId}-rtp\u0000track`, + sampleTimestampMs: input.acceptedAtMs, + state: input.state, + acceptedAtMs: input.acceptedAtMs, + }); +} + +function observePersistentDegraded( + routes: RoomRouteController, + childPeerId: string, + connectionId: string, + acceptedAtMs: number, + parentPeerId = HOST, +): void { + expect( + observeSenderState(routes, { + childPeerId, + connectionId, + parentPeerId, + state: "healthy", + acceptedAtMs, + }).accepted, + ).toBe(true); + for (let offset = 1; offset <= 3; offset += 1) { + expect( + observeSenderState(routes, { + childPeerId, + connectionId, + parentPeerId, + state: "degraded", + acceptedAtMs: acceptedAtMs + offset, + }).accepted, + ).toBe(true); + } +} + +function observePersistentSfuHealthy( + routes: RoomRouteController, + publicationGeneration: string, + routeRevision: number, + acceptedAtMs: number, +) { + let result; + for (let offset = 0; offset < 3; offset += 1) { + result = routes.observeSfuPublisherQualityEvidence({ + hostPeerId: HOST, + hostSessionId: "host_session", + publicationGeneration, + routeRevision, + state: "healthy", + sampleTimestampMs: acceptedAtMs + offset, + acceptedAtMs: acceptedAtMs + offset, + }); + expect(result.accepted).toBe(true); + } + return result!; +} + type RouteQualityMetrics = Pick< ViewerQualityEvidenceMetrics, | "framesDecodedDelta" @@ -152,52 +228,117 @@ function qualityMetrics( } describe("RoomRouteController", () => { + it("keeps isolated limitation windows diagnostic", () => { + const routes = controller(2, { qualityConvergenceEnabled: true }); + addViewer(routes, A, 2); + addViewer(routes, B, 2); + routes.hydrateEdge(A, peerEdge(HOST, "a_from_host")); + routes.hydrateEdge(B, peerEdge(HOST, "b_from_host")); + observeSenderState(routes, { + childPeerId: B, + connectionId: "b_from_host", + state: "healthy", + acceptedAtMs: 99, + }); + observeSenderState(routes, { + childPeerId: A, + connectionId: "a_from_host", + state: "healthy", + acceptedAtMs: 99, + }); + observeSenderState(routes, { + childPeerId: A, + connectionId: "a_from_host", + state: "degraded", + acceptedAtMs: 100, + }); + expect( + routes.observeSenderQualityEvidence({ + parentPeerId: HOST, + parentSessionId: "host_session", + childPeerId: A, + routeRevision: routes.snapshot().revision, + connectionId: "a_from_host", + senderIdentity: "a_from_host-rtp\u0000track", + sampleTimestampMs: 100, + state: "degraded", + acceptedAtMs: 101, + }).accepted, + ).toBe(true); + expect(routes.reconcile(101).operation).toBeUndefined(); + for (const acceptedAtMs of [102]) { + observeSenderState(routes, { + childPeerId: A, + connectionId: "a_from_host", + state: "degraded", + acceptedAtMs, + }); + expect(routes.reconcile(acceptedAtMs).operation).toBeUndefined(); + } + observeSenderState(routes, { + childPeerId: A, + connectionId: "a_from_host", + state: "degraded", + acceptedAtMs: 103, + }); + expect(routes.reconcile(104).operation).toMatchObject({ + childPeerId: A, + reason: "quality-convergence", + }); + }); + + it("never treats one degraded Host edge as SFU fanout pressure", () => { + const routes = controller(2, { + sfuEnabled: true, + qualityConvergenceEnabled: true, + }); + addViewer(routes, A, 1); + addViewer(routes, B, 0); + routes.hydrateEdge(A, peerEdge(HOST, "a_from_host")); + routes.hydrateHostPublication( + "publication_generation_12345678", + "publication_resource", + ); + routes.hydrateEdge(B, { + kind: "sfu", + publicationGeneration: "publication_generation_12345678", + transport: "sfu", + connectionId: "b_from_sfu", + usable: true, + physicalActive: true, + resource: "b_subscription", + }); + observePersistentDegraded(routes, A, "a_from_host", 99); + observePersistentSfuHealthy( + routes, + "publication_generation_12345678", + routes.snapshot().revision, + 103, + ); + expect(routes.reconcile(104).operation).toBeUndefined(); + expect(routes.snapshot().upstreamByViewer.get(A)).toMatchObject({ + kind: "peer", + parentPeerId: HOST, + }); + }); + it("keeps the old edge until one native-healthy quality candidate commits", () => { const routes = controller(2, { qualityConvergenceEnabled: true }); addViewer(routes, A, 2); addViewer(routes, B, 2); routes.hydrateEdge(A, peerEdge(HOST, "a_from_host")); routes.hydrateEdge(B, peerEdge(HOST, "b_from_host")); - const activeRevision = routes.snapshot().revision; - expect( - routes.observeSenderQualityEvidence({ - parentPeerId: HOST, - parentSessionId: "host_session", + observeSenderState(routes, { childPeerId: B, - routeRevision: activeRevision, connectionId: "b_from_host", - senderIdentity: "b-rtp\u0000b-track", state: "healthy", acceptedAtMs: 100, }).accepted, ).toBe(true); - expect( - routes.observeSenderQualityEvidence({ - parentPeerId: HOST, - parentSessionId: "host_session", - childPeerId: A, - routeRevision: activeRevision, - connectionId: "a_from_host", - senderIdentity: "a-rtp\u0000a-track", - state: "healthy", - acceptedAtMs: 99, - }).accepted, - ).toBe(true); - expect( - routes.observeSenderQualityEvidence({ - parentPeerId: HOST, - parentSessionId: "host_session", - childPeerId: A, - routeRevision: activeRevision, - connectionId: "a_from_host", - senderIdentity: "a-rtp\u0000a-track", - state: "degraded", - acceptedAtMs: 100, - }).accepted, - ).toBe(true); + observePersistentDegraded(routes, A, "a_from_host", 99); - const operation = routes.reconcile(101).operation!; + const operation = routes.reconcile(103).operation!; expect(operation.reason).toBe("quality-convergence"); expect(operation.candidates[0]?.tuple).toEqual({ kind: "peer", @@ -205,7 +346,7 @@ describe("RoomRouteController", () => { transport: "direct", }); const begin = beginCandidate(routes, { - nowMs: 102, + nowMs: 104, connectionId: "a_from_b", reservation: { kind: "direct" }, }); @@ -216,7 +357,7 @@ describe("RoomRouteController", () => { revision: current.revision, connectionId: "a_from_b", }; - expect(routes.candidateReady(guard, 103)).toMatchObject({ + expect(routes.candidateReady(guard, 105)).toMatchObject({ accepted: true, committed: false, }); @@ -234,8 +375,9 @@ describe("RoomRouteController", () => { routeRevision: current.revision, connectionId: "a_from_b", senderIdentity: "a-from-b-rtp\u0000a-track", + sampleTimestampMs: 106, state: "healthy", - acceptedAtMs: 104, + acceptedAtMs: 106, }).committed, ).toBe(true); expect(routes.snapshot().upstreamByViewer.get(A)).toMatchObject({ @@ -253,14 +395,17 @@ describe("RoomRouteController", () => { routeRevision: migratedRevision, connectionId: "a_from_b", senderIdentity: "a-from-b-rtp\u0000a-track", + sampleTimestampMs: at, state, acceptedAtMs: at, }); - expect(observeMigrated("degraded", 105).accepted).toBe(true); - expect(routes.reconcile(106).operation).toBeUndefined(); - expect(observeMigrated("healthy", 107).accepted).toBe(true); - expect(observeMigrated("degraded", 108).accepted).toBe(true); - expect(routes.reconcile(109).operation?.reason).toBe( + expect(observeMigrated("degraded", 107).accepted).toBe(true); + expect(routes.reconcile(108).operation).toBeUndefined(); + expect(observeMigrated("healthy", 109).accepted).toBe(true); + expect(observeMigrated("degraded", 110).accepted).toBe(true); + expect(observeMigrated("degraded", 111).accepted).toBe(true); + expect(observeMigrated("degraded", 112).accepted).toBe(true); + expect(routes.reconcile(113).operation?.reason).toBe( "quality-convergence", ); }); @@ -356,28 +501,18 @@ describe("RoomRouteController", () => { addViewer(routes, B, 2); routes.hydrateEdge(A, peerEdge(HOST, "a_from_host")); routes.hydrateEdge(B, peerEdge(HOST, "b_from_host")); - const revision = routes.snapshot().revision; - for (const [childPeerId, connectionId, state, acceptedAtMs] of [ - [B, "b_from_host", "healthy", 100], - [A, "a_from_host", "healthy", 99], - [A, "a_from_host", "degraded", 100], - ] as const) { - expect( - routes.observeSenderQualityEvidence({ - parentPeerId: HOST, - parentSessionId: "host_session", - childPeerId, - routeRevision: revision, - connectionId, - senderIdentity: `${connectionId}-rtp\u0000track`, - state, - acceptedAtMs, - }).accepted, - ).toBe(true); - } - routes.reconcile(101); + expect( + observeSenderState(routes, { + childPeerId: B, + connectionId: "b_from_host", + state: "healthy", + acceptedAtMs: 100, + }).accepted, + ).toBe(true); + observePersistentDegraded(routes, A, "a_from_host", 99); + routes.reconcile(103); const begin = beginCandidate(routes, { - nowMs: 102, + nowMs: 104, connectionId: "a_from_b", reservation: { kind: "direct" }, }); @@ -390,8 +525,9 @@ describe("RoomRouteController", () => { routeRevision: current.revision, connectionId: "a_from_b", senderIdentity: "a-from-b-rtp\u0000track", + sampleTimestampMs: 105, state: "healthy", - acceptedAtMs: 103, + acceptedAtMs: 105, }).committed, ).toBe(false); expect( @@ -402,61 +538,86 @@ describe("RoomRouteController", () => { revision: current.revision, connectionId: "a_from_b", }, - 104, + 106, ).committed, ).toBe(true); }); + it("rejects a candidate whose sender identity changes during proof", () => { + const routes = controller(2, { qualityConvergenceEnabled: true }); + addViewer(routes, A, 2); + addViewer(routes, B, 2); + routes.hydrateEdge(A, peerEdge(HOST, "a_from_host")); + routes.hydrateEdge(B, peerEdge(HOST, "b_from_host")); + observeSenderState(routes, { + childPeerId: B, + connectionId: "b_from_host", + state: "healthy", + acceptedAtMs: 99, + }); + observePersistentDegraded(routes, A, "a_from_host", 99); + routes.reconcile(103); + const begin = beginCandidate(routes, { + nowMs: 104, + connectionId: "a_from_b", + reservation: { kind: "direct" }, + }); + const current = begin.operation!.current!; + const candidateEvidence = (senderIdentity: string, sampleTimestampMs: number) => + routes.observeSenderQualityEvidence({ + parentPeerId: B, + parentSessionId: `${B}_session`, + childPeerId: A, + routeRevision: current.revision, + connectionId: "a_from_b", + senderIdentity, + sampleTimestampMs, + state: "healthy", + acceptedAtMs: sampleTimestampMs, + }); + expect(candidateEvidence("candidate-rtp-1\u0000track", 105).accepted).toBe( + true, + ); + expect(candidateEvidence("candidate-rtp-2\u0000track", 106).committed).toBe( + false, + ); + expect(routes.snapshot().upstreamByViewer.get(A)).toMatchObject({ + kind: "peer", + parentPeerId: HOST, + connectionId: "a_from_host", + }); + }); + it("preempts quality work for a Viewer without media", () => { const routes = controller(2, { qualityConvergenceEnabled: true }); addViewer(routes, A, 2); addViewer(routes, B, 2); routes.hydrateEdge(A, peerEdge(HOST, "a_from_host")); routes.hydrateEdge(B, peerEdge(HOST, "b_from_host")); - const revision = routes.snapshot().revision; - for (const [childPeerId, connectionId, state, acceptedAtMs] of [ - [B, "b_from_host", "healthy", 100], - [A, "a_from_host", "healthy", 99], - [A, "a_from_host", "degraded", 100], - ] as const) { - routes.observeSenderQualityEvidence({ - parentPeerId: HOST, - parentSessionId: "host_session", - childPeerId, - routeRevision: revision, - connectionId, - senderIdentity: `${connectionId}-rtp\u0000track`, - state, - acceptedAtMs, - }); - } - expect(routes.reconcile(101).operation?.reason).toBe( + observeSenderState(routes, { + childPeerId: B, + connectionId: "b_from_host", + state: "healthy", + acceptedAtMs: 100, + }); + observePersistentDegraded(routes, A, "a_from_host", 99); + expect(routes.reconcile(103).operation?.reason).toBe( "quality-convergence", ); - addViewer(routes, C, 0, 102); - expect(routes.reconcile(103).operation).toMatchObject({ + addViewer(routes, C, 0, 104); + expect(routes.reconcile(105).operation).toMatchObject({ childPeerId: C, reason: "join", }); - commitCurrent(routes, 104, "c_join_connection"); - const resumedRevision = routes.snapshot().revision; - for (const [childPeerId, connectionId, state, acceptedAtMs] of [ - [B, "b_from_host", "healthy", 105], - [A, "a_from_host", "healthy", 105], - [A, "a_from_host", "degraded", 106], - ] as const) { - routes.observeSenderQualityEvidence({ - parentPeerId: HOST, - parentSessionId: "host_session", - childPeerId, - routeRevision: resumedRevision, - connectionId, - senderIdentity: `${connectionId}-rtp\u0000track`, - state, - acceptedAtMs, - }); - } - expect(routes.reconcile(107).operation?.reason).toBe( + commitCurrent(routes, 106, "c_join_connection"); + observeSenderState(routes, { + childPeerId: B, + connectionId: "b_from_host", + state: "healthy", + acceptedAtMs: 107, + }); + observePersistentDegraded(routes, A, "a_from_host", 107); + expect(routes.reconcile(111).operation?.reason).toBe( "quality-convergence", ); expect(routes.snapshot().upstreamByViewer.get(A)).toMatchObject({ @@ -471,36 +632,22 @@ describe("RoomRouteController", () => { addViewer(routes, B, 2); routes.hydrateEdge(A, peerEdge(HOST, "a_from_host")); routes.hydrateEdge(B, peerEdge(HOST, "b_from_host")); - const revision = routes.snapshot().revision; - for (const [state, acceptedAtMs] of [ - ["healthy", 99], - ["degraded", 100], - ] as const) { - routes.observeSenderQualityEvidence({ - parentPeerId: HOST, - parentSessionId: "host_session", - childPeerId: A, - routeRevision: revision, - connectionId: "a_from_host", - senderIdentity: "a-rtp\u0000track", - state, - acceptedAtMs, - }); - } - expect(routes.reconcile(101).operation).toBeUndefined(); + observePersistentDegraded(routes, A, "a_from_host", 99); + expect(routes.reconcile(103).operation).toBeUndefined(); expect( routes.observeSenderQualityEvidence({ parentPeerId: HOST, parentSessionId: "host_session", childPeerId: B, - routeRevision: revision, + routeRevision: routes.snapshot().revision, connectionId: "b_from_host", senderIdentity: "b-rtp\u0000track", + sampleTimestampMs: 104, state: "healthy", - acceptedAtMs: 102, + acceptedAtMs: 104, }).accepted, ).toBe(true); - expect(routes.reconcile(103).operation).toMatchObject({ + expect(routes.reconcile(105).operation).toMatchObject({ childPeerId: A, reason: "quality-convergence", }); @@ -525,6 +672,7 @@ describe("RoomRouteController", () => { routeRevision: routes.snapshot().revision, connectionId, senderIdentity: `${connectionId}-rtp\u0000track`, + sampleTimestampMs: acceptedAtMs, state, acceptedAtMs, }); @@ -537,9 +685,15 @@ describe("RoomRouteController", () => { expect(observeCurrent(A, "a_from_host", "degraded", 100).accepted).toBe( true, ); - routes.reconcile(101); + expect(observeCurrent(A, "a_from_host", "degraded", 101).accepted).toBe( + true, + ); + expect(observeCurrent(A, "a_from_host", "degraded", 102).accepted).toBe( + true, + ); + routes.reconcile(103); let begin = beginCandidate(routes, { - nowMs: 102, + nowMs: 104, connectionId: "a_from_b_stale", reservation: { kind: "direct" }, }); @@ -552,6 +706,7 @@ describe("RoomRouteController", () => { routeRevision: current.revision, connectionId: "a_from_b_stale", senderIdentity: "a-stale-rtp\u0000a-track", + sampleTimestampMs: 104, state: "healthy", acceptedAtMs: 104, }).accepted, @@ -580,12 +735,18 @@ describe("RoomRouteController", () => { expect(observeCurrent(A, "a_from_host", "degraded", 5_201).accepted).toBe( true, ); - expect(observeCurrent(B, "b_from_host", "healthy", 5_201).accepted).toBe( + expect(observeCurrent(A, "a_from_host", "degraded", 5_202).accepted).toBe( true, ); - routes.reconcile(5_202); + expect(observeCurrent(A, "a_from_host", "degraded", 5_203).accepted).toBe( + true, + ); + expect(observeCurrent(B, "b_from_host", "healthy", 5_203).accepted).toBe( + true, + ); + routes.reconcile(5_204); begin = beginCandidate(routes, { - nowMs: 5_203, + nowMs: 5_205, connectionId: "a_from_b_bad_source", reservation: { kind: "direct" }, }); @@ -598,11 +759,12 @@ describe("RoomRouteController", () => { routeRevision: current.revision, connectionId: "a_from_b_bad_source", senderIdentity: "a-bad-source-rtp\u0000a-track", + sampleTimestampMs: 5_206, state: "healthy", - acceptedAtMs: 5_204, + acceptedAtMs: 5_206, }).accepted, ).toBe(true); - expect(observeCurrent(B, "b_from_host", "degraded", 5_205).accepted).toBe( + expect(observeCurrent(B, "b_from_host", "degraded", 5_207).accepted).toBe( true, ); expect( @@ -613,7 +775,7 @@ describe("RoomRouteController", () => { revision: current.revision, connectionId: "a_from_b_bad_source", }, - 5_206, + 5_208, ).committed, ).not.toBe(true); expect(routes.snapshot().upstreamByViewer.get(A)).toMatchObject({ @@ -627,24 +789,14 @@ describe("RoomRouteController", () => { addViewer(routes, B, 2); routes.hydrateEdge(A, peerEdge(HOST, "a_from_host")); routes.hydrateEdge(B, peerEdge(HOST, "b_from_host")); - const revision = routes.snapshot().revision; - for (const [childPeerId, connectionId, state, acceptedAtMs] of [ - [B, "b_from_host", "healthy", 0], - [A, "a_from_host", "healthy", 0], - [A, "a_from_host", "degraded", 1], - ] as const) { - routes.observeSenderQualityEvidence({ - parentPeerId: HOST, - parentSessionId: "host_session", - childPeerId, - routeRevision: revision, - connectionId, - senderIdentity: `${connectionId}-rtp\u0000track`, - state, - acceptedAtMs, - }); - } - expect(routes.reconcile(2).operation?.reason).toBe("quality-convergence"); + observeSenderState(routes, { + childPeerId: B, + connectionId: "b_from_host", + state: "healthy", + acceptedAtMs: 0, + }); + observePersistentDegraded(routes, A, "a_from_host", 0); + expect(routes.reconcile(4).operation?.reason).toBe("quality-convergence"); const begin = beginCandidate(routes, { nowMs: 5_001, connectionId: "expired_candidate", @@ -655,44 +807,24 @@ describe("RoomRouteController", () => { expect(routes.snapshot().operation).toBeUndefined(); }); - it("uses bounded SFU only after a degraded Host edge has no Peer escape", () => { - const routes = controller(1, { + it("uses bounded SFU only when every Host root is persistently degraded", () => { + const routes = controller(2, { sfuEnabled: true, qualityConvergenceEnabled: true, }); addViewer(routes, A, 1); + addViewer(routes, B, 1); routes.hydrateEdge(A, peerEdge(HOST, "a_from_host")); - const activeRevision = routes.snapshot().revision; - expect( - routes.observeSenderQualityEvidence({ - parentPeerId: HOST, - parentSessionId: "host_session", - childPeerId: A, - routeRevision: activeRevision, - connectionId: "a_from_host", - senderIdentity: "a-rtp\u0000a-track", - state: "healthy", - acceptedAtMs: 99, - }).accepted, - ).toBe(true); - expect( - routes.observeSenderQualityEvidence({ - parentPeerId: HOST, - parentSessionId: "host_session", - childPeerId: A, - routeRevision: activeRevision, - connectionId: "a_from_host", - senderIdentity: "a-rtp\u0000a-track", - state: "degraded", - acceptedAtMs: 100, - }).accepted, - ).toBe(true); - const operation = routes.reconcile(101).operation!; + routes.hydrateEdge(B, peerEdge(HOST, "b_from_host")); + observePersistentDegraded(routes, A, "a_from_host", 99); + expect(routes.reconcile(103).operation).toBeUndefined(); + observePersistentDegraded(routes, B, "b_from_host", 104); + const operation = routes.reconcile(108).operation!; expect(operation.candidates.map((candidate) => candidate.tuple)).toEqual([ { kind: "sfu", publication: "create" }, ]); const begin = beginCandidate(routes, { - nowMs: 102, + nowMs: 109, connectionId: "a_from_sfu", reservation: { kind: "sfu-create", @@ -713,18 +845,16 @@ describe("RoomRouteController", () => { revision: current.revision, connectionId: "a_from_sfu", }, - 103, + 110, ), ).toMatchObject({ accepted: true, committed: false }); expect( - routes.observeSfuPublisherQualityEvidence({ - hostPeerId: HOST, - hostSessionId: "host_session", - publicationGeneration: "publication_generation_12345678", - routeRevision: current.revision, - state: "healthy", - acceptedAtMs: 104, - }).committed, + observePersistentSfuHealthy( + routes, + "publication_generation_12345678", + current.revision, + 111, + ).committed, ).toBe(true); expect(routes.snapshot().upstreamByViewer.get(A)).toMatchObject({ kind: "sfu", @@ -732,47 +862,141 @@ describe("RoomRouteController", () => { }); }); - it("uses SFU reuse only after its active publication becomes healthy", () => { + it("abandons Host relief when another Host root recovers during the canary", () => { const routes = controller(2, { sfuEnabled: true, qualityConvergenceEnabled: true, }); addViewer(routes, A, 1); + addViewer(routes, B, 1); routes.hydrateEdge(A, peerEdge(HOST, "a_from_host")); + routes.hydrateEdge(B, peerEdge(HOST, "b_from_host")); + observePersistentDegraded(routes, A, "a_from_host", 99); + observePersistentDegraded(routes, B, "b_from_host", 99); + routes.reconcile(103); + const begin = beginCandidate(routes, { + nowMs: 104, + connectionId: "a_from_sfu", + reservation: { + kind: "sfu-create", + edge: "sfu-edge", + publication: "sfu-publication", + overlap: "host-overlap", + }, + publicationGeneration: "publication_generation_12345678", + publicationConnectionId: "publication_connection_12345678", + hostSessionId: "host_session", + }); + const current = begin.operation!.current!; + expect( + observeSenderState(routes, { + childPeerId: B, + connectionId: "b_from_host", + state: "healthy", + acceptedAtMs: 105, + }).accepted, + ).toBe(true); + expect( + routes.candidateReady( + { + childPeerId: A, + childSessionId: `${A}_session`, + revision: current.revision, + connectionId: "a_from_sfu", + }, + 106, + ).committed, + ).not.toBe(true); + expect(routes.snapshot().operation).toBeUndefined(); + expect(routes.snapshot().upstreamByViewer.get(A)).toMatchObject({ + kind: "peer", + parentPeerId: HOST, + connectionId: "a_from_host", + }); + }); + + it("uses SFU reuse only for multi-edge Host relief with healthy ingress", () => { + const routes = controller(3, { + sfuEnabled: true, + qualityConvergenceEnabled: true, + }); + addViewer(routes, A, 1); + addViewer(routes, B, 1); + routes.hydrateEdge(A, peerEdge(HOST, "a_from_host")); + routes.hydrateEdge(B, peerEdge(HOST, "b_from_host")); routes.hydrateHostPublication( "publication_generation_12345678", "publication_resource", ); const revision = routes.snapshot().revision; - for (const [state, acceptedAtMs] of [ - ["healthy", 99], - ["degraded", 100], - ] as const) { - routes.observeSenderQualityEvidence({ - parentPeerId: HOST, - parentSessionId: "host_session", - childPeerId: A, - routeRevision: revision, - connectionId: "a_from_host", - senderIdentity: "a-rtp\u0000track", - state, - acceptedAtMs, - }); - } - expect(routes.reconcile(101).operation).toBeUndefined(); + observePersistentDegraded(routes, A, "a_from_host", 99); + observePersistentDegraded(routes, B, "b_from_host", 99); + expect(routes.reconcile(103).operation).toBeUndefined(); + observePersistentSfuHealthy( + routes, + "publication_generation_12345678", + revision, + 104, + ); + expect(routes.reconcile(105).operation?.candidates[0]?.tuple).toEqual({ + kind: "sfu", + publication: "reuse", + }); + const begin = beginCandidate(routes, { + nowMs: 106, + connectionId: "a_from_sfu_reuse", + reservation: { kind: "sfu-reuse", edge: "a_subscription" }, + }); + const current = begin.operation!.current!; expect( routes.observeSfuPublisherQualityEvidence({ hostPeerId: HOST, hostSessionId: "host_session", publicationGeneration: "publication_generation_12345678", routeRevision: revision, - state: "healthy", - acceptedAtMs: 102, + state: "unknown", + sampleTimestampMs: null, + acceptedAtMs: 107, }).accepted, ).toBe(true); - expect(routes.reconcile(103).operation?.candidates[0]?.tuple).toEqual({ + expect( + routes.candidateReady( + { + childPeerId: A, + childSessionId: `${A}_session`, + revision: current.revision, + connectionId: "a_from_sfu_reuse", + }, + 108, + ).committed, + ).not.toBe(true); + for (const acceptedAtMs of [2_100, 4_100, 6_100]) { + observeSenderState(routes, { + childPeerId: A, + connectionId: "a_from_host", + state: "degraded", + acceptedAtMs, + }); + observeSenderState(routes, { + childPeerId: B, + connectionId: "b_from_host", + state: "degraded", + acceptedAtMs, + }); + const settled = routes.observeSfuPublisherQualityEvidence({ + hostPeerId: HOST, + hostSessionId: "host_session", + publicationGeneration: "publication_generation_12345678", + routeRevision: revision, + state: "healthy", + sampleTimestampMs: acceptedAtMs, + acceptedAtMs, + }); + expect(settled.committed).toBe(acceptedAtMs === 6_100); + } + expect(routes.snapshot().upstreamByViewer.get(A)).toMatchObject({ kind: "sfu", - publication: "reuse", + connectionId: "a_from_sfu_reuse", }); }); @@ -798,16 +1022,12 @@ describe("RoomRouteController", () => { resource: "b_subscription", }); const revision = routes.snapshot().revision; - expect( - routes.observeSfuPublisherQualityEvidence({ - hostPeerId: HOST, - hostSessionId: "host_session", - publicationGeneration: "publication_generation_12345678", - routeRevision: revision, - state: "healthy", - acceptedAtMs: 100, - }).accepted, - ).toBe(true); + observePersistentSfuHealthy( + routes, + "publication_generation_12345678", + revision, + 100, + ); expect( routes.observeQualityEvidence({ childPeerId: B, @@ -826,22 +1046,8 @@ describe("RoomRouteController", () => { acceptedAtMs: 100, }), ).toBe("accepted"); - for (const [state, acceptedAtMs] of [ - ["healthy", 99], - ["degraded", 100], - ] as const) { - routes.observeSenderQualityEvidence({ - parentPeerId: HOST, - parentSessionId: "host_session", - childPeerId: A, - routeRevision: revision, - connectionId: "a_from_host", - senderIdentity: "a-rtp\u0000track", - state, - acceptedAtMs, - }); - } - expect(routes.reconcile(101).operation?.candidates[0]?.tuple).toEqual({ + observePersistentDegraded(routes, A, "a_from_host", 99); + expect(routes.reconcile(103).operation?.candidates[0]?.tuple).toEqual({ kind: "peer", parentPeerId: B, transport: "direct", diff --git a/tests/route-transition.test.ts b/tests/route-transition.test.ts index dfbd531..056c9ea 100644 --- a/tests/route-transition.test.ts +++ b/tests/route-transition.test.ts @@ -13,7 +13,10 @@ import { reportActivePeerRouteFailure, } from "../src/client/media/route-transition.ts"; import { ViewerSfuRoute } from "../src/client/media/viewer-sfu-route.ts"; -import type { ConnectionMetrics } from "../src/client/types.ts"; +import { + EMPTY_METRICS, + type ConnectionMetrics, +} from "../src/client/types.ts"; const peerAssignment = ( parentPeerId: string, @@ -88,10 +91,12 @@ const candidate = ( childPeerId = "viewer_12345678", transport: "direct" | "sfu" = "direct", connectionId = `candidate_${revision}_12345678`, + qualityProbe = false, ): PreparedRouteCandidate => ({ childPeerId, connectionId, transport, + qualityProbe, }); function createFakeSubscriber(events: SubscriberEvents, log: string[], label: string) { @@ -121,6 +126,27 @@ function createFakeSubscriber(events: SubscriberEvents, log: string[], label: st }; } +function receiveMetrics( + timestampMs: number, + overrides: Partial = {}, +): ConnectionMetrics { + return { + ...EMPTY_METRICS, + sampleTimestampMs: timestampMs, + sampleWindowMs: 2_000, + frameWidth: 1920, + frameHeight: 1080, + framesPerSecond: 60, + bitrateKbps: 5_000, + intervalFramesDecoded: 120, + intervalFreezeCount: 0, + intervalFreezeDurationMs: 0, + intervalPauseCount: 0, + intervalPauseDurationMs: 0, + ...overrides, + }; +} + describe("minimal route transition contracts", () => { it("rebinds Viewer route ownership before a post-restart prepare", async () => { const prepared: Array<{ parentPeerId: string | null; revision?: number }> = []; @@ -567,6 +593,78 @@ describe("minimal route transition contracts", () => { expect(subscribers[0]?.disconnect).toHaveBeenCalledOnce(); }); + it("keeps a quality SFU candidate pending until three non-regressing windows", async () => { + const messages: ClientMessage[] = []; + const subscribers: ReturnType[] = []; + let currentMetrics = receiveMetrics(1_000); + let qualityProbeEligible = true; + const route = new ViewerSfuRoute("viewer_12345678", { + activatePeer: () => true, + reconcileSfuChildren: () => undefined, + onSfuStream: () => undefined, + currentPeerMetrics: () => currentMetrics, + qualityProbeEligible: () => qualityProbeEligible, + send: (message) => { + messages.push(message); + return true; + }, + createSubscriber: (events) => { + const subscriber = createFakeSubscriber( + events, + [], + `subscriber-${subscribers.length + 1}`, + ); + subscribers.push(subscriber); + return subscriber; + }, + }); + + route.accept({ + revision: 1, + phase: "active", + assignment: peerAssignment("parent_12345678"), + }); + route.accept({ + revision: 2, + phase: "prepare", + assignment: viewerSfuAssignment(), + candidate: candidate( + 2, + "viewer_12345678", + "sfu", + "candidate_2_12345678", + true, + ), + }); + await route.acceptConfig(sfuConfig(2)); + const subscriber = subscribers[0]!; + subscriber.events.onStream({} as MediaStream); + subscriber.events.onFirstDecodedFrame(); + expect(messages.filter((message) => message.type === "route-ready")).toEqual([]); + + for (const timestampMs of [1_100, 3_100]) { + currentMetrics = receiveMetrics(timestampMs - 100); + subscriber.events.onStats(receiveMetrics(timestampMs)); + } + expect(messages.filter((message) => message.type === "route-ready")).toEqual([]); + + qualityProbeEligible = false; + for (const timestampMs of [5_100, 7_100, 9_100]) { + currentMetrics = receiveMetrics(timestampMs - 100); + subscriber.events.onStats(receiveMetrics(timestampMs)); + } + expect(messages.filter((message) => message.type === "route-ready")).toEqual([]); + + qualityProbeEligible = true; + for (const timestampMs of [11_100, 13_100, 15_100]) { + currentMetrics = receiveMetrics(timestampMs - 100); + subscriber.events.onStats(receiveMetrics(timestampMs)); + } + expect(messages.filter((message) => message.type === "route-ready")).toEqual([ + { type: "route-ready", revision: 2, phase: "prepare" }, + ]); + }); + it("reconnects only the active SFU subscriber on the current route", async () => { const messages: ClientMessage[] = []; const streams: MediaStream[] = []; diff --git a/tests/sender-quality-evidence.test.ts b/tests/sender-quality-evidence.test.ts index 8428fd2..819aa19 100644 --- a/tests/sender-quality-evidence.test.ts +++ b/tests/sender-quality-evidence.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it } from "vitest"; import { invalidateSenderQualityEvidence, senderQualityEvidenceFromSnapshot, + sfuPublisherQualityEvidenceFromMetrics, } from "../src/client/media/sender-quality-evidence.ts"; import type { PeerSnapshot } from "../src/client/types.ts"; import { @@ -166,16 +167,61 @@ describe("native sender quality evidence", () => { expect(senderQualityEvidenceFromSnapshot(snapshot, 7)).toMatchObject({ state: "unknown", }); - expect(senderQualityEvidenceFromSnapshot(snapshot, 7)).toMatchObject({ + expect(senderQualityEvidenceFromSnapshot(snapshot, 7)).toBeNull(); + const nextSnapshot = { + ...snapshot, + metrics: { + ...snapshot.metrics, + sampleTimestampMs: 2_000, + }, + }; + expect(senderQualityEvidenceFromSnapshot(nextSnapshot, 7)).toMatchObject({ type: "sender-quality-evidence", childPeerId: snapshot.peerId, connectionId: snapshot.connectionId, routeRevision: 7, state: "healthy", + sampleTimestampMs: 2_000, }); invalidateSenderQualityEvidence(); - expect(senderQualityEvidenceFromSnapshot(snapshot, 7)).toMatchObject({ + expect(senderQualityEvidenceFromSnapshot(nextSnapshot, 7)).toMatchObject({ state: "unknown", }); + expect(senderQualityEvidenceFromSnapshot(nextSnapshot, 7)).toBeNull(); + }); + + it("waits for a new SFU publisher stats window after rebaseline", () => { + invalidateSenderQualityEvidence(); + const metrics = { + ...collectConnectionMetricsFromReport( + report(1_000, "none", { + none: 1, + bandwidth: 0, + cpu: 0, + other: 0, + }), + "send", + createStatsAccumulator(), + ), + sampleTimestampMs: 1_000, + sampleWindowMs: 2_000, + intervalFramesEncoded: 100, + qualityLimitationReason: "none", + nativeEdgeQualityState: "healthy" as const, + }; + const generation = "publication_generation_12345678"; + expect( + sfuPublisherQualityEvidenceFromMetrics(metrics, 3, generation), + ).toMatchObject({ state: "unknown" }); + expect( + sfuPublisherQualityEvidenceFromMetrics(metrics, 3, generation), + ).toBeNull(); + expect( + sfuPublisherQualityEvidenceFromMetrics( + { ...metrics, sampleTimestampMs: 3_000 }, + 3, + generation, + ), + ).toMatchObject({ state: "healthy", sampleTimestampMs: 3_000 }); }); }); diff --git a/tests/sfu-quality-probe.test.ts b/tests/sfu-quality-probe.test.ts new file mode 100644 index 0000000..3d31aca --- /dev/null +++ b/tests/sfu-quality-probe.test.ts @@ -0,0 +1,94 @@ +import { describe, expect, it } from "vitest"; + +import { + SfuQualityProbe, + sfuCandidateDoesNotRegress, +} from "../src/client/media/sfu-quality-probe.ts"; +import { + EMPTY_METRICS, + type ConnectionMetrics, +} from "../src/client/types.ts"; + +function metrics( + timestampMs: number, + overrides: Partial = {}, +): ConnectionMetrics { + return { + ...EMPTY_METRICS, + sampleTimestampMs: timestampMs, + sampleWindowMs: 2_000, + frameWidth: 1920, + frameHeight: 1080, + framesPerSecond: 60, + bitrateKbps: 5_000, + intervalFramesDecoded: 120, + intervalFreezeCount: 0, + intervalFreezeDurationMs: 0, + intervalPauseCount: 0, + intervalPauseDurationMs: 0, + ...overrides, + }; +} + +describe("SFU quality probe", () => { + it("accepts only complete same-Viewer non-regressing windows", () => { + const current = metrics(2_000); + expect(sfuCandidateDoesNotRegress(current, metrics(2_100))).toBe(true); + expect( + sfuCandidateDoesNotRegress( + current, + metrics(2_100, { bitrateKbps: 4_999 }), + ), + ).toBe(false); + expect( + sfuCandidateDoesNotRegress( + current, + metrics(2_100, { frameHeight: 720 }), + ), + ).toBe(false); + expect( + sfuCandidateDoesNotRegress( + current, + metrics(2_100, { intervalFreezeCount: 1 }), + ), + ).toBe(false); + expect( + sfuCandidateDoesNotRegress( + current, + metrics(8_000), + ), + ).toBeNull(); + }); + + it("requires three consecutive windows and resets after a regression", () => { + const probe = new SfuQualityProbe(); + expect(probe.observe(metrics(1_000), metrics(1_100))).toBe(false); + expect(probe.observe(metrics(3_000), metrics(3_100))).toBe(false); + expect( + probe.observe(metrics(5_000), metrics(5_100, { framesPerSecond: 30 })), + ).toBe(false); + expect(probe.observe(metrics(7_000), metrics(7_100))).toBe(false); + expect(probe.observe(metrics(9_000), metrics(9_100))).toBe(false); + expect(probe.observe(metrics(11_000), metrics(11_100))).toBe(true); + }); + + it("resets when either side lacks a new comparable window", () => { + const probe = new SfuQualityProbe(); + expect(probe.observe(metrics(1_000), metrics(1_100))).toBe(false); + expect(probe.observe(metrics(3_000), metrics(3_100))).toBe(false); + expect(probe.observe(metrics(3_000), metrics(5_100))).toBe(false); + expect(probe.observe(metrics(7_000), metrics(7_100))).toBe(false); + expect(probe.observe(metrics(9_000), metrics(9_100))).toBe(false); + expect(probe.observe(metrics(11_000), metrics(11_100))).toBe(true); + + probe.reset(); + expect(probe.observe(metrics(13_000), metrics(13_100))).toBe(false); + expect( + probe.observe( + metrics(15_000), + metrics(15_100, { intervalFreezeDurationMs: null }), + ), + ).toBe(false); + expect(probe.observe(metrics(17_000), metrics(17_100))).toBe(false); + }); +});