From cfb1f4b12907fa78d3dd6ad3763e8a90a5428a65 Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Mon, 31 Aug 2026 03:13:51 +0800 Subject: [PATCH] fix(route): stabilize bounded quality convergence (#351) --- .../0005-automatic-hybrid-media-routing.md | 13 +- docs/product/routing-transport.md | 14 +- docs/research/realtime-quality-adaptation.md | 27 +++ docs/verification-status.md | 5 +- src/client/sfu/publisher.ts | 4 +- src/client/webrtc/stats.ts | 32 +++ src/server/room-route-controller.ts | 215 ++++++++++++------ tests/room-route-controller.test.ts | 169 +++++++++++++- tests/sfu-client.test.ts | 57 ++++- 9 files changed, 451 insertions(+), 85 deletions(-) diff --git a/docs/adr/0005-automatic-hybrid-media-routing.md b/docs/adr/0005-automatic-hybrid-media-routing.md index 7e90ea9..25502d7 100644 --- a/docs/adr/0005-automatic-hybrid-media-routing.md +++ b/docs/adr/0005-automatic-hybrid-media-routing.md @@ -237,7 +237,11 @@ change clears that run. A newly committed availability or direct-convergence edge may establish a new run without first reporting healthy. Quality- and root-convergence commits remain disarmed until the new active identity reports a fresh healthy delta; an explicit sender reset clears the old latch and starts a -new sequence. A single native limitation interval remains diagnostic. +new sequence. A single native limitation interval remains diagnostic. Once a +Peer quality experiment starts, a later unknown current-sender window remains +inconclusive inside that bounded operation; it does not cancel the candidate. +Fresh healthy evidence, an explicit reset, or an edge/sender identity change +still ends the experiment. When the current sender remains persistently limited, the controller considers the shallowest affected child first and uses the existing deterministic candidate filters, ordering, cursor, reservations, and total @@ -262,7 +266,12 @@ candidate and advance the existing cursor without resetting the deadline. Unknown candidate-sender evidence waits within the existing deadline; a stale relative proof rejects that candidate and advances the existing cursor, while a degraded candidate sender cannot commit. None creates a score. Old-edge -recovery, authority change, or deadline aborts the experiment. A successful commit clears and rebaselines the affected subtree. +recovery, authority change, or deadline ends the experiment. A failed candidate +opportunity is consumed for that exact current sender sequence; an unrelated +route revision cannot reopen it. A newly eligible parent, a less disruptive +capacity transition, healthy recovery, or an edge/sender identity change is a +new opportunity. A successful commit clears and rebaselines the 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 parent-to-child sender moves only that child. If several senders on one parent diff --git a/docs/product/routing-transport.md b/docs/product/routing-transport.md index dd13b3c..66c4434 100644 --- a/docs/product/routing-transport.md +++ b/docs/product/routing-transport.md @@ -122,11 +122,15 @@ retained sender closes. A complete zero-frame current window counts as zero delivery only when an overlapping candidate actually decodes clean video; two zero-frame paths remain unknown. A missing or limited exact candidate sender cannot commit; an expired one-shot - relative proof rejects that candidate. When no different Peer parent is - eligible, the same operation may append one same-parent connection regeneration - candidate; it uses the same first-frame, overlap, and proof rules and is damped - by its exact post-commit sender identity. Ordinary quality moves remain - Peer-to-Peer. SFU may +relative proof rejects that candidate. An unknown current-sender window clears +the trigger run but does not cancel an already-started bounded Peer experiment; +healthy recovery, explicit reset, or identity change does. Failed candidates are +not retried for the same exact sender sequence merely because an unrelated route +fact changed, while a new parent or improved capacity transition remains a new +opportunity. When no different Peer parent is eligible, the same operation may +append one same-parent connection regeneration candidate; it uses the same +first-frame, overlap, and proof rules and is damped by its exact post-commit +sender identity. 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 diff --git a/docs/research/realtime-quality-adaptation.md b/docs/research/realtime-quality-adaptation.md index 3eca831..9741434 100644 --- a/docs/research/realtime-quality-adaptation.md +++ b/docs/research/realtime-quality-adaptation.md @@ -224,6 +224,27 @@ A LiveKit H.264 publication may use more than one hardware encoder for its representations. Reducing Host network copies to one publication does not by itself prove lower Host encode cost. +Production room 8489 separated an SFU delivery problem from route churn. SFU +roots showed synchronized freeze windows and frequent 1920-to-960 representation +changes before any quality-convergence operation existed, while a brief direct +Peer Viewer remained near 1080p and full cadence. Later, the Host publication +also reported sustained native bandwidth limitation. This excludes quality +probing as the initial trigger and does not support a Host-wide capture or encode +failure, but it does not isolate one SFU cause. + +Pinned LiveKit server 1.13.5 delays Dynacast layer downgrades by five seconds, +reenables demanded layers immediately, and then relies on keyframe acquisition; +its forwarder grace and individual PLI retry intervals are shorter than that +delay. Those mechanics +can explain several-second stalls during representation churn, but the retained +warning-level logs contain no subscribed-quality or publisher-layer transition, +so Dynacast disablement is not established as the cause of each freeze. SFU +publisher diagnostics therefore keep aggregate bitrate and native limitation +across all active representations while reporting dimensions and cadence from +the highest active representation. A future event can distinguish publisher +layer disablement from subscriber-only forwarding changes without altering the +LiveKit adaptation policy. + ## Host And Page Cost Boundaries Each ordinary Browser child is an independent PeerConnection and normally an @@ -370,6 +391,9 @@ percentages, and tuning loops are not adopted for this Browser product. - Codec cost and quality after stock BWE reaches steady state. - Mixed P2P/SFU public-network quality, long-running thermals, A/V sync, mobile lifecycle, and 20-Viewer resource admission. +- Controlled and production correlation of Host SFU active representations, + subscribed-quality changes, keyframe reacquisition, and Viewer freezes before + changing Dynacast policy or its delay. ## Primary Sources @@ -395,6 +419,9 @@ percentages, and tuning loops are not adopted for this Browser product. - [Chrome timer throttling](https://developer.chrome.com/blog/timer-throttling-in-chrome-88) - [Chrome Page Lifecycle](https://developer.chrome.com/docs/web-platform/page-lifecycle-api) - [LiveKit server forwarder](https://github.com/livekit/livekit/blob/v1.13.5/pkg/sfu/forwarder.go) +- [LiveKit server defaults](https://github.com/livekit/livekit/blob/v1.13.5/pkg/config/config.go) +- [LiveKit Dynacast manager](https://github.com/livekit/livekit/blob/v1.13.5/pkg/rtc/dynacast/dynacastmanagervideo.go) +- [LiveKit keyframe requests](https://github.com/livekit/livekit/blob/v1.13.5/pkg/sfu/downtrack.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) diff --git a/docs/verification-status.md b/docs/verification-status.md index baa71c1..5cf927e 100644 --- a/docs/verification-status.md +++ b/docs/verification-status.md @@ -46,7 +46,10 @@ content intent, startup-quality workaround, LiveKit-owned SFU adaptation, native edge convergence, and same-edge connection regeneration invariants. The 2026-08-30 production comparison confirms recovery on a fresh same-route sender, but does not establish its quality or resource benefit on heterogeneous public -networks. See [media quality](./product/media-quality.md) +networks. Room 8489 confirms synchronized SFU stalls and representation churn, +but retained logs do not prove whether Dynacast layer disablement, subscriber +forwarding, or keyframe reacquisition owned each freeze. See +[media quality](./product/media-quality.md) and [realtime quality research](./research/realtime-quality-adaptation.md). ## Browser Lifecycle diff --git a/src/client/sfu/publisher.ts b/src/client/sfu/publisher.ts index 2f08add..be697c5 100644 --- a/src/client/sfu/publisher.ts +++ b/src/client/sfu/publisher.ts @@ -33,6 +33,7 @@ import { collectNativeSenderQualityFromReport, createNativeSenderQualityAccumulator, createStatsAccumulator, + highestActiveVideoRid, maxEncodedVideoFrames, mergeStatsReports, type StatsAccumulator, @@ -1170,13 +1171,14 @@ export class SfuPublisher { identity.accumulator, { trackIdentifier: identity.videoTrack.id, + rid: highestActiveVideoRid(report, identity.videoTrack.id), audioTrackIdentifier: identity.audioTrack?.id ?? null, }, ); const metrics = { ...connectionMetrics, bitrateKbps: - connectionMetrics.bitrateKbps ?? nativeQuality.bitrateKbps, + nativeQuality.bitrateKbps ?? connectionMetrics.bitrateKbps, nativeEdgeQualityState: nativeQuality.nativeEdgeQualityState, ...(nativeQuality.nativeEdgeQualityState === "unknown" ? {} diff --git a/src/client/webrtc/stats.ts b/src/client/webrtc/stats.ts index 53fc09c..5f05adc 100644 --- a/src/client/webrtc/stats.ts +++ b/src/client/webrtc/stats.ts @@ -502,6 +502,38 @@ export function maxEncodedVideoFrames( return frames; } +export function highestActiveVideoRid( + report: RTCStatsReport, + trackIdentifier: string, +): string | null { + let selectedRid: string | null = null; + let selectedPixels = -1; + report.forEach((raw) => { + const record = raw as StatsRecord; + if ( + record.type !== "outbound-rtp" || + record.kind !== "video" || + record.isRemote === true || + record.active === false || + mediaTrackIdentifier(report, record, "send") !== trackIdentifier + ) { + return; + } + const rid = stringValue(record, "rid"); + const width = numberValue(record, "frameWidth"); + const height = numberValue(record, "frameHeight"); + if (!rid || width === null || height === null || width <= 0 || height <= 0) { + return; + } + const pixels = width * height; + if (pixels > selectedPixels) { + selectedRid = rid; + selectedPixels = pixels; + } + }); + return selectedRid; +} + function transportRecord( report: RTCStatsReport, media: StatsRecord | null, diff --git a/src/server/room-route-controller.ts b/src/server/room-route-controller.ts index 5f2cf08..99b7184 100644 --- a/src/server/room-route-controller.ts +++ b/src/server/room-route-controller.ts @@ -154,6 +154,7 @@ interface ChildOperation { deadlineAtMs: number; builtAtFactVersion: number; deferredParentPeerIds: string[]; + qualitySource?: SenderQualityIdentity; current?: Attempt; } @@ -263,14 +264,14 @@ interface SenderQualityObservation { parentSessionId: string; connectionId: string; senderIdentity: string; - state: "healthy" | "degraded"; + state: "unknown" | "healthy" | "degraded"; consecutiveDegradedWindows: number; lastSampleTimestampMs: number; lastAcceptedAtMs: number; - consumedAtFactVersion?: number; + consumedQualityCandidates: Map; } -interface SenderQualityRegenerationBlock { +interface SenderQualityIdentity { childSessionId: string; parentPeerId: string; parentSessionId: string; @@ -369,7 +370,7 @@ export class RoomRouteController { private readonly senderQualityBaselines = new Map(); private readonly senderQualityRegenerationBlocks = new Map< string, - SenderQualityRegenerationBlock + SenderQualityIdentity >(); private sfuPublisherQualityObservation?: SfuPublisherQualityObservation; private readonly qualityBaselinesPending = new Set(); @@ -1124,38 +1125,42 @@ export class RoomRouteController { ) { return rejected(); } + const previous = this.senderQualityObservations.get(input.childPeerId); + const sameConnectionIdentity = + previous?.childSessionId === child.sessionId && + previous.parentPeerId === input.parentPeerId && + previous.parentSessionId === input.parentSessionId && + previous.connectionId === input.connectionId; if (input.state === "unknown") { - if (this.senderQualityObservations.has(input.childPeerId)) { + if (previous && previous.state !== "unknown") { this.debug("sender-quality-reset", { parent: this.debugPeer(input.parentPeerId), child: this.debugPeer(input.childPeerId), }); } - this.senderQualityObservations.delete(input.childPeerId); - const released = - operation?.reason === "quality-convergence" && - operation.childPeerId === input.childPeerId - ? this.abortOperation(input.acceptedAtMs, "aborted") - : []; + if (previous && sameConnectionIdentity) { + this.senderQualityObservations.set(input.childPeerId, { + ...previous, + state: "unknown", + consecutiveDegradedWindows: 0, + lastAcceptedAtMs: input.acceptedAtMs, + }); + } else { + this.senderQualityObservations.delete(input.childPeerId); + } return { accepted: true, committed: false, failedPeerIds: [], activeRevision: this.revision, - released, + released: [], }; } if (!input.senderIdentity || input.sampleTimestampMs === null) { return rejected(); } - const previous = this.senderQualityObservations.get(input.childPeerId); - const sameIdentity = - previous?.childSessionId === child.sessionId && - previous.parentPeerId === input.parentPeerId && - previous.parentSessionId === input.parentSessionId && - previous.connectionId === input.connectionId; const sameSenderIdentity = - sameIdentity && previous?.senderIdentity === input.senderIdentity; + sameConnectionIdentity && previous?.senderIdentity === input.senderIdentity; if ( sameSenderIdentity && previous !== undefined && @@ -1192,8 +1197,6 @@ export class RoomRouteController { previous !== undefined && input.acceptedAtMs < previous.lastAcceptedAtMs + VIEWER_QUALITY_EVIDENCE_EXPIRY_MS; - const baselineRequired = - this.senderQualityBaselines.get(input.childPeerId) === true; if (input.state === "healthy") { this.senderQualityBaselines.delete(input.childPeerId); } @@ -1213,14 +1216,10 @@ export class RoomRouteController { consecutiveDegradedWindows, lastSampleTimestampMs: input.sampleTimestampMs, lastAcceptedAtMs: input.acceptedAtMs, - consumedAtFactVersion: - baselineRequired && input.state === "degraded" - ? this.factVersion - : sameSenderIdentity && - previous?.state === "degraded" && - input.state === "degraded" - ? previous.consumedAtFactVersion - : undefined, + consumedQualityCandidates: + sameSenderIdentity && previous && input.state !== "healthy" + ? previous.consumedQualityCandidates + : new Map(), }); if ( !sameFreshSenderIdentity || @@ -1702,12 +1701,6 @@ export class RoomRouteController { if (reason === "direct-convergence") { this.directContinuations.delete(childPeerId); } - if (reason === "quality-convergence") { - const observation = this.senderQualityObservations.get(childPeerId); - if (observation) { - observation.consumedAtFactVersion = this.factVersion; - } - } this.finishTiming( childPeerId, nowMs, @@ -1758,6 +1751,10 @@ export class RoomRouteController { deadlineAtMs: nowMs + this.options.operationTimeoutMs, builtAtFactVersion: this.factVersion, deferredParentPeerIds: [], + qualitySource: + reason === "quality-convergence" + ? this.currentQualityOperationSource(childPeerId) + : undefined, }; this.debug("operation-started", { child: this.debugPeer(childPeerId), @@ -2401,6 +2398,9 @@ export class RoomRouteController { if (directConvergence) { this.consumeDirectContinuationCandidate(operation); } + if (qualityConvergence) { + this.consumeQualityOperationCandidates(operation); + } if (operation.current) released.push(...reservationResources(operation.current.reservation)); const revisionAdvanced = Boolean(operation.current); if (operation.current) this.advanceActiveRevision(operation); @@ -2441,14 +2441,6 @@ export class RoomRouteController { factsChanged ? "stale" : "operation-deadline", ); } - if (qualityConvergence) { - const observation = this.senderQualityObservations.get( - operation.childPeerId, - ); - if (observation) { - observation.consumedAtFactVersion = this.factVersion; - } - } this.blockAndClear( operation, !backgroundConvergence && @@ -2518,6 +2510,9 @@ export class RoomRouteController { if (peerNoProgressExpired) { result.expired = true; } + if (guardFailed || peerNoProgressExpired) { + this.consumeQualityCandidateOpportunity(operation, plan); + } if (operation.reason === "direct-convergence") { this.consumeDirectContinuationCandidate(operation); } @@ -2540,7 +2535,6 @@ export class RoomRouteController { } if (operation.cursor < operation.candidates.length) return result; const sfuBootstrap = operation.reason === "sfu-bootstrap"; - const qualityConvergence = operation.reason === "quality-convergence"; const backgroundConvergence = this.isBackgroundConvergence( operation.reason, ); @@ -2584,14 +2578,6 @@ export class RoomRouteController { rejectionBucket, ); } - if (qualityConvergence) { - const observation = this.senderQualityObservations.get( - operation.childPeerId, - ); - if (observation) { - observation.consumedAtFactVersion = this.factVersion; - } - } this.blockAndClear( operation, !backgroundConvergence && @@ -2872,6 +2858,47 @@ export class RoomRouteController { ); } + private qualityCandidateOpportunityAvailable( + childPeerId: string, + plan: CandidatePlan, + ): boolean { + const consumed = this.senderQualityObservations + .get(childPeerId) + ?.consumedQualityCandidates.get( + this.candidateOpportunityBase(childPeerId, plan), + ); + return ( + consumed === undefined || + endpointTransitionRank(plan.endpointTransition) < consumed + ); + } + + private consumeQualityCandidateOpportunity( + operation: ChildOperation, + plan: CandidatePlan | undefined, + ): void { + if (operation.reason !== "quality-convergence" || !plan) { + return; + } + const observation = this.currentQualityOperationObservation(operation); + if (!observation) { + return; + } + consumeOpportunity( + observation.consumedQualityCandidates, + this.candidateOpportunityBase(operation.childPeerId, plan), + endpointTransitionRank(plan.endpointTransition), + ); + } + + private consumeQualityOperationCandidates( + operation: ChildOperation, + ): void { + for (const candidate of operation.candidates) { + this.consumeQualityCandidateOpportunity(operation, candidate); + } + } + private consumeActiveEdgeOpportunity( child: Participant, edge: CommittedEdge, @@ -2931,6 +2958,52 @@ export class RoomRouteController { : "unknown"; } + private currentQualityOperationSource( + childPeerId: string, + ): SenderQualityIdentity | undefined { + const child = this.participants.get(childPeerId); + const edge = this.upstreamByViewer.get(childPeerId); + const observation = this.senderQualityObservations.get(childPeerId); + if ( + !child?.sessionId || + edge?.kind !== "peer" || + !observation || + observation.childSessionId !== child.sessionId || + observation.childSessionId !== edge.childSessionId || + observation.parentPeerId !== edge.parentPeerId || + observation.parentSessionId !== edge.parentSessionId || + observation.connectionId !== edge.connectionId + ) { + return undefined; + } + return { + childSessionId: observation.childSessionId, + parentPeerId: observation.parentPeerId, + parentSessionId: observation.parentSessionId, + connectionId: observation.connectionId, + senderIdentity: observation.senderIdentity, + }; + } + + private currentQualityOperationObservation( + operation: ChildOperation, + ): SenderQualityObservation | undefined { + const source = operation.qualitySource; + const current = this.currentQualityOperationSource(operation.childPeerId); + if ( + !source || + !current || + source.childSessionId !== current.childSessionId || + source.parentPeerId !== current.parentPeerId || + source.parentSessionId !== current.parentSessionId || + source.connectionId !== current.connectionId || + source.senderIdentity !== current.senderIdentity + ) { + return undefined; + } + return this.senderQualityObservations.get(operation.childPeerId); + } + private senderQualityRegenerationBlocked( childPeerId: string, edge: Extract, { kind: "peer" }>, @@ -3096,7 +3169,10 @@ export class RoomRouteController { this.candidateValid(childPeerId, regenerationPlan) ? [regenerationPlan] : []; - return [...clearPeers, ...remainingPeers, ...sfu, ...regeneration]; + return [...clearPeers, ...remainingPeers, ...sfu, ...regeneration].filter( + (candidate) => + this.qualityCandidateOpportunityAvailable(childPeerId, candidate), + ); } private selectQualityChild(nowMs: number): string | undefined { @@ -3106,12 +3182,11 @@ export class RoomRouteController { return this.availableViewers() .filter(({ peerId }) => { const edge = this.upstreamByViewer.get(peerId); - const observation = this.senderQualityObservations.get(peerId); return Boolean( edge?.kind === "peer" && this.senderQualityBaselines.get(peerId) !== true && - observation?.consumedAtFactVersion !== this.factVersion && - this.senderQualityPersistentlyDegraded(peerId, edge, nowMs), + this.senderQualityPersistentlyDegraded(peerId, edge, nowMs) && + this.buildQualityCandidates(peerId, nowMs).length > 0, ); }) .sort( @@ -3255,21 +3330,29 @@ export class RoomRouteController { nowMs: number, ): boolean { const currentEdge = this.upstreamByViewer.get(operation.childPeerId); + const observation = this.currentQualityOperationObservation(operation); const tuple = operation.current?.tuple ?? operation.candidates[operation.cursor]?.tuple; if ( - !currentEdge || + currentEdge?.kind !== "peer" || !tuple || - !this.senderQualityPersistentlyDegraded( - operation.childPeerId, - currentEdge, - nowMs, - ) + !observation ) { return false; } - return tuple.kind === "peer" || this.hostFanoutNeedsSfuRelief(nowMs); + const state = this.senderQualityState( + operation.childPeerId, + currentEdge, + nowMs, + ); + if (state === "healthy") { + return false; + } + return ( + tuple.kind === "peer" || + (state === "degraded" && this.hostFanoutNeedsSfuRelief(nowMs)) + ); } private hostFanoutNeedsSfuRelief(nowMs: number): boolean { @@ -3914,8 +3997,8 @@ export class RoomRouteController { for (const childPeerId of this.childrenOf(producerPeerId)) { if (regeneratingChild && childPeerId === operation.childPeerId) { // A failed same-edge trial must not turn the retained sender into a - // global baseline block. The operation's consumed fact version - // already supplies one-shot retry damping. + // global baseline block. Its exact candidate opportunity owns retry + // damping while the retained sender identity remains current. continue; } this.senderQualityObservations.delete(childPeerId); @@ -4784,7 +4867,7 @@ function tupleKey(tuple: CandidateTuple): string { } function senderQualityRegenerationBlockMatches( - block: SenderQualityRegenerationBlock, + block: SenderQualityIdentity, childSessionId: string, input: SenderQualityEvidenceInput, ): boolean { diff --git a/tests/room-route-controller.test.ts b/tests/room-route-controller.test.ts index 9a392b4..9a658a8 100644 --- a/tests/room-route-controller.test.ts +++ b/tests/room-route-controller.test.ts @@ -758,6 +758,163 @@ describe("RoomRouteController", () => { expect(committed.released).toContain("a_overlap"); }); + it("keeps one quality experiment through inconclusive current windows", () => { + 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: 9, + }); + observePersistentDegraded(routes, A, "a_from_host", 10); + + const operation = routes.reconcile(14).operation!; + expect(operation.candidates.map((candidate) => candidate.tuple)).toEqual([ + { kind: "peer", parentPeerId: B, transport: "direct" }, + { + kind: "peer", + parentPeerId: HOST, + transport: "direct", + regenerate: true, + }, + ]); + beginCandidate(routes, { + nowMs: 15, + connectionId: "a_from_b_failed", + reservation: { kind: "direct" }, + }); + + const currentRevision = routes.snapshot().revision; + expect( + routes.observeSenderQualityEvidence({ + parentPeerId: HOST, + parentSessionId: "host_session", + childPeerId: A, + routeRevision: currentRevision, + connectionId: "a_from_host", + senderIdentity: null, + sampleTimestampMs: null, + state: "unknown", + acceptedAtMs: 16, + }).accepted, + ).toBe(true); + expect(routes.snapshot().operation).toMatchObject({ + reason: "quality-convergence", + cursor: 0, + }); + + const afterFirst = routes.operationExpired( + routes.snapshot().operation!.wakeAtMs, + ); + expect(afterFirst.accepted).toBe(true); + expect(routes.snapshot().operation).toMatchObject({ + reason: "quality-convergence", + cursor: 1, + }); + const regeneration = beginCandidate(routes, { + nowMs: 20, + connectionId: "a_regeneration_failed", + reservation: { kind: "direct", overlap: "a_regeneration_overlap" }, + }).operation!; + expect( + routes.candidateFailed( + { + childPeerId: A, + childSessionId: `${A}_session`, + revision: regeneration.current!.revision, + connectionId: "a_regeneration_failed", + }, + 21, + ).accepted, + ).toBe(true); + expect(routes.snapshot().operation).toBeUndefined(); + + for (const acceptedAtMs of [22, 23, 24]) { + expect( + observeSenderState(routes, { + childPeerId: A, + connectionId: "a_from_host", + state: "degraded", + acceptedAtMs, + }).accepted, + ).toBe(true); + } + expect(routes.reconcile(25).operation).toBeUndefined(); + routes.touchExternalFacts(); + expect(routes.reconcile(26).operation).toBeUndefined(); + + addViewer(routes, C, 2); + routes.hydrateEdge(C, peerEdge(B, "c_from_b")); + observeSenderState(routes, { + childPeerId: C, + connectionId: "c_from_b", + parentPeerId: B, + state: "healthy", + acceptedAtMs: 27, + }); + const newOpportunity = routes.reconcile(28).operation; + expect(newOpportunity).toMatchObject({ + childPeerId: A, + reason: "quality-convergence", + }); + expect(newOpportunity?.candidates.map((candidate) => candidate.tuple)).toEqual([ + { kind: "peer", parentPeerId: C, transport: "direct" }, + ]); + }); + + it("does not restart a quality candidate set after its total deadline", () => { + 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: 9, + }); + observePersistentDegraded(routes, A, "a_from_host", 10); + + const operation = routes.reconcile(14).operation!; + beginCandidate(routes, { + nowMs: 15, + connectionId: "a_from_b_pending", + reservation: { kind: "direct" }, + }); + expect(routes.operationExpired(operation.deadlineAtMs).accepted).toBe(true); + expect(routes.snapshot().operation).toBeUndefined(); + expect(routes.reconcile(operation.deadlineAtMs + 1).operation).toBeUndefined(); + routes.touchExternalFacts(); + expect(routes.reconcile(operation.deadlineAtMs + 2).operation).toBeUndefined(); + + observeSenderState(routes, { + childPeerId: A, + connectionId: "a_from_host", + state: "healthy", + acceptedAtMs: operation.deadlineAtMs + 3, + }); + for (const acceptedAtMs of [ + operation.deadlineAtMs + 4, + operation.deadlineAtMs + 5, + operation.deadlineAtMs + 6, + ]) { + observeSenderState(routes, { + childPeerId: A, + connectionId: "a_from_host", + state: "degraded", + acceptedAtMs, + }); + } + expect( + routes.reconcile(operation.deadlineAtMs + 7).operation?.reason, + ).toBe("quality-convergence"); + }); + it("keeps isolated limitation windows diagnostic", () => { const routes = controller(2, { qualityConvergenceEnabled: true }); addViewer(routes, A, 2); @@ -1895,18 +2052,14 @@ describe("RoomRouteController", () => { 105, ).accepted, ).toBe(true); + expect(routes.setEffectiveCapacity(B, `${B}_session`, 1)).toBe(true); observePersistentDegraded(routes, B, "b_from_host", 104); const operation = routes.reconcile(108).operation!; const childPeerId = operation.childPeerId; + expect(childPeerId).toBe(A); expect(operation.candidates.map((candidate) => candidate.tuple)).toEqual([ - { kind: "peer", parentPeerId: A, transport: "direct" }, + { kind: "peer", parentPeerId: B, transport: "direct" }, { kind: "sfu", publication: "create" }, - { - kind: "peer", - parentPeerId: HOST, - transport: "direct", - regenerate: true, - }, ]); const direct = beginCandidate(routes, { nowMs: 109, @@ -2014,9 +2167,11 @@ describe("RoomRouteController", () => { 105, ).accepted, ).toBe(true); + expect(routes.setEffectiveCapacity(B, `${B}_session`, 1)).toBe(true); observePersistentDegraded(routes, B, "b_from_host", 104); const operation = routes.reconcile(108).operation!; const childPeerId = operation.childPeerId; + expect(childPeerId).toBe(A); const peerAttempt = beginCandidate(routes, { nowMs: 109, connectionId: "quality_peer_candidate", diff --git a/tests/sfu-client.test.ts b/tests/sfu-client.test.ts index 05e2b32..4fbc1b8 100644 --- a/tests/sfu-client.test.ts +++ b/tests/sfu-client.test.ts @@ -429,15 +429,29 @@ function simulcastSenderReport( timestamp: number, lowBytesSent: number, highBytesSent: number, + highActive = true, ): RTCStatsReport { - const layer = (id: string, bytesSent: number, framesEncoded: number) => ({ + const layer = ( + id: string, + rid: string, + width: number, + height: number, + bytesSent: number, + framesEncoded: number, + active = true, + ) => ({ id, type: "outbound-rtp", timestamp, kind: "video", + rid, + active, mediaSourceId: "video-source", bytesSent, framesEncoded, + framesPerSecond: active ? 30 : 0, + frameWidth: width, + frameHeight: height, qualityLimitationReason: "none", qualityLimitationDurations: { none: timestamp / 1_000, @@ -447,8 +461,16 @@ function simulcastSenderReport( }, }); return statsReport([ - layer("video-low", lowBytesSent, timestamp / 40), - layer("video-high", highBytesSent, timestamp / 20), + layer("video-low", "q", 960, 540, lowBytesSent, timestamp / 40), + layer( + "video-high", + "f", + 1920, + 1080, + highBytesSent, + timestamp / 20, + highActive, + ), { id: "video-source", type: "media-source", @@ -547,6 +569,11 @@ describe("SfuPublisher", () => { await vi.advanceTimersByTimeAsync(2_000); expect(updates.at(-1)?.bitrateKbps).toBeNull(); + expect(updates.at(-1)).toMatchObject({ + rtpRid: "f", + frameWidth: 1920, + frameHeight: 1080, + }); expect( sfuPublisherQualityEvidenceFromMetrics( updates.at(-1)!, @@ -569,6 +596,30 @@ describe("SfuPublisher", () => { }); }); + it("reports the highest active SFU representation", async () => { + vi.useFakeTimers(); + const updates: Array = []; + const publisher = new SfuPublisher({ + onStats: (metrics) => updates.push(metrics), + }); + const video = track("video", "video-1"); + await publisher.connect(connection); + await publisher.activate(stream(video), qualityProfile); + const sender = livekit.state.rooms[0].localParticipant.publications[0].track + .sender; + sender.getStats.mockResolvedValueOnce( + simulcastSenderReport(video.id, 1_000, 100_000, 200_000, false), + ); + + await vi.advanceTimersByTimeAsync(2_000); + + expect(updates.at(-1)).toMatchObject({ + rtpRid: "q", + frameWidth: 960, + frameHeight: 540, + }); + }); + it("enables balanced SFU adaptation after startup frames", async () => { vi.useFakeTimers(); const publisher = new SfuPublisher();