fix(route): stabilize bounded quality convergence (#351)
This commit is contained in:
1 parent
56e19139e6
commit
cfb1f4b129
9 files changed
+451
-85
No files matched your search
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"
|
||||
? {}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -154,6 +154,7 @@ interface ChildOperation<Resource> {
|
||||
deadlineAtMs: number;
|
||||
builtAtFactVersion: number;
|
||||
deferredParentPeerIds: string[];
|
||||
qualitySource?: SenderQualityIdentity;
|
||||
current?: Attempt<Resource>;
|
||||
}
|
||||
|
||||
@@ -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<string, number>;
|
||||
}
|
||||
|
||||
interface SenderQualityRegenerationBlock {
|
||||
interface SenderQualityIdentity {
|
||||
childSessionId: string;
|
||||
parentPeerId: string;
|
||||
parentSessionId: string;
|
||||
@@ -369,7 +370,7 @@ export class RoomRouteController<Resource = unknown> {
|
||||
private readonly senderQualityBaselines = new Map<string, boolean>();
|
||||
private readonly senderQualityRegenerationBlocks = new Map<
|
||||
string,
|
||||
SenderQualityRegenerationBlock
|
||||
SenderQualityIdentity
|
||||
>();
|
||||
private sfuPublisherQualityObservation?: SfuPublisherQualityObservation;
|
||||
private readonly qualityBaselinesPending = new Set<string>();
|
||||
@@ -1124,38 +1125,42 @@ export class RoomRouteController<Resource = unknown> {
|
||||
) {
|
||||
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<Resource = unknown> {
|
||||
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<Resource = unknown> {
|
||||
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<Resource = unknown> {
|
||||
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<Resource = unknown> {
|
||||
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<Resource = unknown> {
|
||||
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<Resource = unknown> {
|
||||
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<Resource = unknown> {
|
||||
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<Resource = unknown> {
|
||||
}
|
||||
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<Resource = unknown> {
|
||||
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<Resource = unknown> {
|
||||
);
|
||||
}
|
||||
|
||||
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<Resource>,
|
||||
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<Resource>,
|
||||
): void {
|
||||
for (const candidate of operation.candidates) {
|
||||
this.consumeQualityCandidateOpportunity(operation, candidate);
|
||||
}
|
||||
}
|
||||
|
||||
private consumeActiveEdgeOpportunity(
|
||||
child: Participant,
|
||||
edge: CommittedEdge<Resource>,
|
||||
@@ -2931,6 +2958,52 @@ export class RoomRouteController<Resource = unknown> {
|
||||
: "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<Resource>,
|
||||
): 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<CommittedEdge<Resource>, { kind: "peer" }>,
|
||||
@@ -3096,7 +3169,10 @@ export class RoomRouteController<Resource = unknown> {
|
||||
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<Resource = unknown> {
|
||||
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<Resource = unknown> {
|
||||
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<Resource = unknown> {
|
||||
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 {
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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<ConnectionMetrics | null> = [];
|
||||
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();
|
||||
|
||||
Reference in new issue
Block a user