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.
This commit is contained in:
AAA authored and GitHub committed 2026-08-27 20:21:55 +08:00
1 parent b39097fd27
commit ad175408c0
24 files changed
+1432 -334

No files matched your search

+2
View File
@@ -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.
+38 -19
View File
@@ -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
+13 -5
View File
@@ -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
+11 -8
View File
@@ -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
+5
View File
@@ -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 |
@@ -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/)
+10 -5
View File
@@ -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.
+4 -1
View File
@@ -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
+2 -1
View File
@@ -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
);
}
+49 -2
View File
@@ -11,6 +11,8 @@ const connectionGenerations = new Map<string, number>();
const publicationGenerations = new Map<string, number>();
const unknownConnections = new Set<string>();
const unknownPublications = new Set<string>();
const connectionSampleTimestamps = new Map<string, number>();
const publicationSampleTimestamps = new Map<string, number>();
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<ClientMessage, { type: "sender-quality-evidence" }> | 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;
}
+132
View File
@@ -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;
}
}
+67 -7
View File
@@ -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<void> {
+10
View File
@@ -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}`,
+8 -2
View File
@@ -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) {
+3
View File
@@ -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",
};
}
+315 -32
View File
@@ -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<Resource> {
mediaReady: boolean;
senderQualityState?: "healthy" | "degraded";
senderQualityAcceptedAtMs?: number;
senderQualityConsecutiveHealthyWindows?: number;
senderQualitySampleTimestampMs?: number;
senderQualityIdentity?: string;
}
interface ChildOperation<Resource> {
@@ -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<Resource = unknown> {
input.state,
input.acceptedAtMs,
input.senderIdentity,
input.sampleTimestampMs,
commitReservation,
);
}
@@ -959,6 +970,12 @@ export class RoomRouteController<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
? 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<Resource = unknown> {
input.state,
input.acceptedAtMs,
input.publicationGeneration,
input.sampleTimestampMs,
commitReservation,
);
}
@@ -1102,6 +1159,9 @@ export class RoomRouteController<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
state: "unknown" | "healthy" | "degraded",
acceptedAtMs: number,
senderIdentity: string | null,
sampleTimestampMs: number | null,
commitReservation: (
reservation: CandidateReservation<Resource>,
) => boolean,
): SenderQualityEvidenceResult<Resource> {
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<Resource = unknown> {
released: [],
};
}
if (!senderIdentity) {
if (!senderIdentity || sampleTimestampMs === null) {
return {
accepted: false,
committed: false,
@@ -1166,8 +1316,60 @@ export class RoomRouteController<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
],
};
}
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<Resource = unknown> {
};
}
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<Resource = unknown> {
) {
return this.candidateFailed(guard, nowMs);
}
if (attempt.senderQualityState === "degraded") {
if (candidateQualityState === "degraded") {
return this.candidateFailed(guard, nowMs);
}
}
@@ -1916,6 +2120,13 @@ export class RoomRouteController<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
(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<Resource = unknown> {
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<Resource = unknown> {
)[0]?.peerId;
}
private senderQualityPersistentlyDegraded(
childPeerId: string,
edge: CommittedEdge<Resource>,
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<Resource>,
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<CommittedEdge<Resource>, { 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 }) => {
+27 -3
View File
@@ -6,7 +6,7 @@ import { isCanonicalVideoCodecEvidence } from "./video-codec-evidence.js";
export const MAX_VIEWERS_PER_ROOM_LIMIT = 20;
export const MAX_PARTICIPANTS_PER_ROOM_LIMIT = MAX_VIEWERS_PER_ROOM_LIMIT + 1;
export const MAX_SIGNAL_BYTES = 64 * 1024;
export const SIGNALING_PROTOCOL = "screener-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
+2
View File
@@ -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 () => {
+23 -12
View File
@@ -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,
}),
+12 -1
View File
@@ -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,
+439 -233
View File
@@ -129,6 +129,82 @@ function peerEdge(
};
}
function observeSenderState(
routes: RoomRouteController<string>,
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<string>,
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<string>,
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",
+99 -1
View File
@@ -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> = {},
): 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<typeof createFakeSubscriber>[] = [];
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[] = [];
+48 -2
View File
@@ -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 });
});
});
+94
View File
@@ -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> = {},
): 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);
});
});