From 992ee15437b6f41cd31aaffe7864ac7da219281c Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Thu, 27 Aug 2026 22:59:12 +0800 Subject: [PATCH] fix(route): fix committed route state and bounded convergence (#303) --- .../0005-automatic-hybrid-media-routing.md | 12 + docs/product/presentation-lifecycle.md | 3 + docs/product/routing-transport.md | 12 + docs/project-memory.md | 10 +- docs/research/realtime-quality-adaptation.md | 14 + docs/status.md | 15 +- docs/todo.md | 4 +- src/client/media/sender-quality-evidence.ts | 10 + src/client/pages/HostPage.tsx | 8 +- src/server/hybrid-media-router.ts | 126 ++++---- src/server/room-route-controller.ts | 305 ++++++++++++++---- src/server/signaling.ts | 88 +---- src/shared/protocol.ts | 45 ++- tests/hybrid-media-router.test.ts | 44 ++- tests/protocol.test.ts | 17 +- tests/room-route-controller.test.ts | 96 ++++++ tests/server-signal.test.ts | 3 + 17 files changed, 568 insertions(+), 244 deletions(-) diff --git a/docs/adr/0005-automatic-hybrid-media-routing.md b/docs/adr/0005-automatic-hybrid-media-routing.md index 7cc9b46..a7a2a1f 100644 --- a/docs/adr/0005-automatic-hybrid-media-routing.md +++ b/docs/adr/0005-automatic-hybrid-media-routing.md @@ -76,6 +76,13 @@ There is no independent depth cap. Room admission and acyclicity bound the graph shallowest-first is a preference, not a periodic balancing mandate. IP address, geography, NAT guess, UA, and scalar quality score do not select a parent. +One structural exception is event-driven Host-root capacity convergence. After a +new Host-root edge commits, if that root has zero direct children and another +Host root has at least two, one deterministic direct child may prepare against the new +root. A fresh healthy candidate sender proof is required before make-before-break +commit. Success moves one branch toward even fanout; failure consumes that one-shot intent. No timer, +continuous rebalance or general load score is introduced. + A failed exact tuple is consumed only for its current operation. A later external fact may make it eligible again; there is no persistent parent blacklist. @@ -119,6 +126,11 @@ Peer candidates converge behind working media, one at a time and round-robin across SFU Viewers. New join or repair work preempts this background convergence. Exhausting direct candidates simply keeps the working SFU route. +The five-second no-transport-progress window is candidate-relative across every +Peer operation. A new candidate receives its own window within the unchanged +total operation deadline; exact transport progress retains it until media proof +or that deadline. + If the Host is full and no publication exists, one bounded SFU-bootstrap intent owns the original waiting demand and a finite cursor of eligible Host-direct children. A carrier must create the publication through a normal overlap; it may diff --git a/docs/product/presentation-lifecycle.md b/docs/product/presentation-lifecycle.md index 1224b3c..0c19e1a 100644 --- a/docs/product/presentation-lifecycle.md +++ b/docs/product/presentation-lifecycle.md @@ -43,6 +43,9 @@ must continuously show a truthful accessible connection stage rather than a black screen, ICE-connected state, or unproved `playing` event. - A proved current frame remains visible behind non-terminal recovery state. +- Host roster connection state follows the server's committed physical media + path. A pending candidate or temporarily stale quality sample may annotate + optimization, but cannot downgrade an active Viewer to routing. - Host pause retains the current frame and waits for resume. - Host signaling loss alone does not invalidate media that is still healthy. - Exact media failure or terminal route failure invalidates current-frame proof diff --git a/docs/product/routing-transport.md b/docs/product/routing-transport.md index c9454bd..90d5a8a 100644 --- a/docs/product/routing-transport.md +++ b/docs/product/routing-transport.md @@ -67,6 +67,18 @@ without discarding its subtree; departure and capacity changes move only affecte children. ADR-0005 owns filtering, ordering, cursor, revision, reservation, and rollback mechanics. +Every unconnected Peer candidate gets the same five-second no-progress window. +Transport-connected progress retains it through the operation deadline. This +applies to join, recovery, direct convergence, quality convergence and structural +convergence; candidate order never resets the total deadline. + +When a newly committed Host-root Viewer exposes unused downstream capacity while +another Host root has at least two direct children, the same background operation may +move one of those children to the new root. The candidate must prove a healthy +native sender edge before commit. This one-shot event-driven move reduces root +fanout skew; it is not periodic balancing and does not move a healthy branch +without admitted overlap. + Framework reconnect runs before route reassignment. Manual media reconnect also rebuilds only the current P2P parent or current SFU subscription; it does not perform quality selection or choose another route. diff --git a/docs/project-memory.md b/docs/project-memory.md index 9bae02e..58ed962 100644 --- a/docs/project-memory.md +++ b/docs/project-memory.md @@ -52,11 +52,11 @@ implementation and routine UI detail. ## Current Snapshot -Canonical source and production use the strict `screener-v15` Browser/server -contract with the corrected 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-v16` Browser/server contract with +committed media readiness, bounded candidate-relative progress and one-shot +Host-root convergence. Production remains on `screener-v15` until the current +correction release completes. Both retain optional SQLite room authority and the +H.264/VP8 sender gate. [Status](./status.md) owns exact identity. Current work is owned by the [TODO ledger](./todo.md). Open physical evidence is owned by [verification status](./verification-status.md). Environment and initial diff --git a/docs/research/realtime-quality-adaptation.md b/docs/research/realtime-quality-adaptation.md index 2d6110d..4a1d84b 100644 --- a/docs/research/realtime-quality-adaptation.md +++ b/docs/research/realtime-quality-adaptation.md @@ -227,6 +227,20 @@ 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. +The 2026-08-27 room-6020 canary observed 21 successful Peer candidates, all +within two seconds, and 19 failed candidates with a 15-second median. Reusing the +existing five-second no-progress window for each background Peer candidate +therefore removes measured queue tail without adding a new threshold; candidates +with transport progress still retain the total deadline. The same run showed +that native sender health can recover while a Viewer still reports low delivered +FPS, so receiver-quality routing remains an evidence problem rather than grounds +for an uncalibrated FPS threshold. + +Canary logs use stable room-scoped anonymous ordinals. They may record native +limitation reason and bounded sender/receiver FPS and bitrate already collected +by the application, but never display names, raw Peer IDs, SDP, ICE candidates, +tokens or media credentials. + 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 diff --git a/docs/status.md b/docs/status.md index d36745d..0849beb 100644 --- a/docs/status.md +++ b/docs/status.md @@ -45,7 +45,7 @@ This is the current execution index. Git history owns completed timelines; ## Current Source -- Canonical source uses the strict `screener-v15` Browser/server contract. +- Canonical source uses the strict `screener-v16` 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,10 +72,10 @@ This is the current execution index. Git history owns completed timelines; ## Current Milestone -Native-edge local convergence is deployed behind its default-off per-share -policy. Persistent exact sender windows drive ordinary P2P movement. SFU quality -work is limited to multi-root Host fanout relief and requires independent -Host-ingress and same-Viewer candidate proof before one move. +The 6020 canary correction is implemented in source and awaits deployment. It +keeps Host roster state on committed media, bounds each no-progress Peer +candidate at five seconds, adds one-shot healthy Host-root fanout convergence, +and stabilizes anonymized canary identity and metrics. ## Active Boundaries @@ -92,5 +92,6 @@ Host-ingress and same-Viewer candidate proof before one move. ## Current Hold -No source or deployment P0/P1 is open. Sanitized route-event logging remains -enabled for the pre-release canary. Broader product evidence remains parked. +Production `screener-v15` can falsely show committed Viewers as routing and can +spend 15 seconds on background candidates. `screener-v16` is the active P1 +correction. Sanitized route logging remains enabled. diff --git a/docs/todo.md b/docs/todo.md index abff0ee..b17afba 100644 --- a/docs/todo.md +++ b/docs/todo.md @@ -7,7 +7,9 @@ topics are not implementation authority. ## Now -No implementation task is currently authorized. +1. **Release the 6020 canary follow-up.** Preserve committed Host roster state, + reuse the five-second candidate progress gate, perform one-shot healthy root + fanout convergence, and retain stable bounded canary evidence. ## Parked Repository Work diff --git a/src/client/media/sender-quality-evidence.ts b/src/client/media/sender-quality-evidence.ts index 3cefdd2..8cdd541 100644 --- a/src/client/media/sender-quality-evidence.ts +++ b/src/client/media/sender-quality-evidence.ts @@ -101,6 +101,11 @@ export function senderQualityEvidenceFromSnapshot( sampleTimestampMs, routeRevision, state, + diagnostics: { + reason: state === "unknown" ? null : snapshot.metrics.qualityLimitationReason, + framesPerSecond: snapshot.metrics.framesPerSecond, + bitrateKbps: snapshot.metrics.bitrateKbps, + }, }); if (!parsed.success) { return null; @@ -153,6 +158,11 @@ export function sfuPublisherQualityEvidenceFromMetrics( publicationGeneration, state, sampleTimestampMs, + diagnostics: { + reason: state === "unknown" ? null : metrics.qualityLimitationReason, + framesPerSecond: metrics.framesPerSecond, + bitrateKbps: metrics.bitrateKbps, + }, }); if (!parsed.success) { return null; diff --git a/src/client/pages/HostPage.tsx b/src/client/pages/HostPage.tsx index 47be873..3c21ad2 100644 --- a/src/client/pages/HostPage.tsx +++ b/src/client/pages/HostPage.tsx @@ -2951,17 +2951,15 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { const hasCurrentQualityEvidence = hasMatchingQualityEvidence && qualityPresentation?.fresh === true; - const hasCurrentSfuEvidence = - viewer.upstream.kind === "sfu" && - viewer.sfuMediaReady === true; + const hasCommittedMedia = viewer.mediaReady === true; const hasCurrentRouteEvidence = hasPeerRouteEvidence(snapshot) || hasCurrentQualityEvidence || - hasCurrentSfuEvidence; + hasCommittedMedia; const hasCurrentConnectionEvidence = snapshot?.connectionState === "connected" || hasCurrentQualityEvidence || - hasCurrentSfuEvidence; + hasCommittedMedia; const viewerState = hasCurrentConnectionEvidence ? "connected" diff --git a/src/server/hybrid-media-router.ts b/src/server/hybrid-media-router.ts index 376f979..0a968c4 100644 --- a/src/server/hybrid-media-router.ts +++ b/src/server/hybrid-media-router.ts @@ -86,6 +86,8 @@ interface RoomRuntime { pump?: Promise; deadlineTimer?: NodeJS.Timeout; sfuRefreshesInFlight: Set; + debugOrdinalByPeerId: Map; + nextDebugOrdinal: number; } interface PreparedCandidate { @@ -117,13 +119,6 @@ export type ActiveViewerMediaEdge = upstream: { kind: "sfu" }; }; -export interface ActiveSfuViewerMediaState { - viewerPeerId: string; - viewerSessionId: string; - revision: number; - sfuPublicationGeneration: string; -} - export interface SfuFallbackOptions { url: string; tokenIssuer: SfuTokenIssuer; @@ -143,10 +138,7 @@ export interface HybridMediaRouterOptions { setConnectionId: (roomId: string, viewerPeerId: string, connectionId: string) => void; deleteConnectionId: (roomId: string, viewerPeerId: string) => void; getShareGeneration: (roomId: string) => string | undefined; - onViewerMediaSnapshot?: ( - roomId: string, - viewers: readonly ActiveSfuViewerMediaState[], - ) => void; + onRoutesChanged?: (roomId: string) => void; now?: () => number; } @@ -206,6 +198,7 @@ export class HybridMediaRouter { room.advertisedCapacityByViewer.get(input.peerId) ?? 0, ); } + this.debugPeer(input.roomId, input.peerId); if (room.controller) { this.releaseResources( room.controller.upsertParticipant({ @@ -273,10 +266,26 @@ export class HybridMediaRouter { input: RouteQualityEvidenceInput & { roomId: string }, ): RouteQualityEvidenceResult { const { roomId, ...evidence } = input; - return ( + const result = this.rooms.get(roomId)?.controller?.observeQualityEvidence(evidence) ?? - "rejected" - ); + "rejected"; + if (result !== "rejected") { + this.debug(roomId, "viewer-quality-evidence", { + viewer: this.debugPeer(roomId, input.childPeerId), + upstream: + input.upstream.kind === "peer" + ? `p2p:${this.debugPeer(roomId, input.upstream.peerId)}` + : "sfu", + framesPerSecond: input.metrics.framesPerSecond, + bitrateKbps: input.metrics.bitrateKbps, + width: input.metrics.width, + height: input.metrics.height, + framesDecoded: input.metrics.framesDecodedDelta, + freezes: input.metrics.freezeCountDelta, + pauses: input.metrics.pauseCountDelta, + }); + } + return result; } observeSenderQualityEvidence( @@ -307,6 +316,16 @@ export class HybridMediaRouter { input, (reservation) => this.commitReservation(reservation), ); + if (result.accepted) { + this.debug(participant.roomId, "sender-quality-evidence", { + parent: this.debugPeer(participant.roomId, participant.peerId), + child: this.debugPeer(participant.roomId, message.childPeerId), + state: message.state, + reason: message.diagnostics.reason, + framesPerSecond: message.diagnostics.framesPerSecond, + bitrateKbps: message.diagnostics.bitrateKbps, + }); + } this.releaseResources(result.released); if (result.committed) { this.resourceWaiters.delete(participant.roomId); @@ -359,6 +378,14 @@ export class HybridMediaRouter { input, (reservation) => this.commitReservation(reservation), ); + if (result.accepted) { + this.debug(participant.roomId, "sfu-publisher-quality-evidence", { + state: message.state, + reason: message.diagnostics.reason, + framesPerSecond: message.diagnostics.framesPerSecond, + bitrateKbps: message.diagnostics.bitrateKbps, + }); + } this.releaseResources(result.released); if ( result.committed && @@ -460,24 +487,6 @@ export class HybridMediaRouter { : { kind: "none" }; } - viewerSfuMediaStateIsCurrent( - roomId: string, - viewerPeerId: string, - revision: number, - publicationGeneration: string, - ): boolean { - const snapshot = this.rooms.get(roomId)?.controller?.snapshot(); - const edge = snapshot?.upstreamByViewer.get(viewerPeerId); - return Boolean( - snapshot?.revision === revision && - edge?.kind === "sfu" && - edge.physicalActive && - edge.publicationGeneration === publicationGeneration && - snapshot.hostPublication?.physicalActive && - snapshot.hostPublication.generation === publicationGeneration, - ); - } - peerSignalAuthorization(input: { roomId: string; sourcePeerId: string; @@ -834,7 +843,7 @@ export class HybridMediaRouter { ...(edge.kind === "peer" ? { parentSessionId: edge.parentSessionId } : {}), }, this.now()); if (invalidated && edge.usable && snapshot.paused) { - this.publishViewerMediaState(participant.roomId, controller.snapshot()); + this.options.onRoutesChanged?.(participant.roomId); } this.requestPump(participant.roomId); } @@ -905,6 +914,8 @@ export class HybridMediaRouter { advertisedCapacityByViewer: new Map(), requested: false, sfuRefreshesInFlight: new Set(), + debugOrdinalByPeerId: new Map(), + nextDebugOrdinal: 0, }; this.rooms.set(roomId, room); } @@ -1011,7 +1022,10 @@ export class HybridMediaRouter { rejectionBucket: preparation.rejectionBucket, cursor: operation.cursor, }); - if (operation.reason === "quality-convergence") { + if ( + operation.reason === "quality-convergence" || + operation.reason === "root-convergence" + ) { const skipped = controller.skipCurrentCandidate( guard, this.now(), @@ -1506,30 +1520,7 @@ export class HybridMediaRouter { } } } - this.publishViewerMediaState(roomId, snapshot); - } - - private publishViewerMediaState( - roomId: string, - snapshot: RouteSnapshot, - ): void { - const viewers: ActiveSfuViewerMediaState[] = []; - for (const viewer of this.options.roomStore.getConnectedViewers(roomId)) { - const edge = snapshot.upstreamByViewer.get(viewer.peerId); - if ( - edge?.kind === "sfu" && - edge.physicalActive && - this.pathIsPhysical(roomId, snapshot, viewer.peerId) - ) { - viewers.push({ - viewerPeerId: viewer.peerId, - viewerSessionId: viewer.sessionId, - revision: snapshot.revision, - sfuPublicationGeneration: edge.publicationGeneration, - }); - } - } - this.options.onViewerMediaSnapshot?.(roomId, viewers); + this.options.onRoutesChanged?.(roomId); } private pathIsPhysical( @@ -1863,11 +1854,16 @@ export class HybridMediaRouter { } private debugPeer(roomId: string, peerId: string): string { - if (peerId === this.rooms.get(roomId)?.hostPeerId) return "host"; - const viewerIndex = this.options.roomStore - .getViewerPeerIds(roomId) - .indexOf(peerId); - return viewerIndex >= 0 ? `viewer-${viewerIndex + 1}` : "viewer-unknown"; + const room = this.rooms.get(roomId); + if (peerId === room?.hostPeerId) return "host"; + if (!room) return "viewer-unknown"; + let ordinal = room.debugOrdinalByPeerId.get(peerId); + if (ordinal === undefined) { + ordinal = room.nextDebugOrdinal; + room.nextDebugOrdinal += 1; + room.debugOrdinalByPeerId.set(peerId, ordinal); + } + return `viewer-${ordinal}`; } private debugTuple(roomId: string, tuple: CandidateTuple): string { @@ -1953,7 +1949,9 @@ function preparedRouteCandidate( childPeerId: operation.childPeerId, connectionId, transport: tuple.kind === "peer" ? tuple.transport : "sfu", - qualityProbe: operation.reason === "quality-convergence", + qualityProbe: + operation.reason === "quality-convergence" || + operation.reason === "root-convergence", }; } diff --git a/src/server/room-route-controller.ts b/src/server/room-route-controller.ts index b635719..693cfb5 100644 --- a/src/server/room-route-controller.ts +++ b/src/server/room-route-controller.ts @@ -123,6 +123,7 @@ interface Attempt { publicationGeneration?: string; publicationConnectionId?: string; reservation: CandidateReservation; + startedAtMs: number; transportConnected: boolean; mediaReady: boolean; senderQualityState?: "healthy" | "degraded"; @@ -316,7 +317,13 @@ export interface RouteQualityEvidenceInput { | "freezeDurationMsDelta" | "pauseCountDelta" | "pauseDurationMsDelta" - >; + > & + Partial< + Pick< + ViewerQualityEvidenceMetrics, + "width" | "height" | "framesPerSecond" | "bitrateKbps" + > + >; acceptedAtMs: number; } @@ -350,6 +357,7 @@ export class RoomRouteController { private readonly directContinuations = new Map(); private sfuBootstrapIntent?: SfuBootstrapIntent; private sfuBootstrapExhaustedAtFactVersion?: number; + private rootConvergenceRootPeerId?: string; constructor(private readonly options: ControllerOptions) { assertEndpointMediaCopyCapacity(options.endpointMediaCopyCapacity); @@ -936,7 +944,7 @@ export class RoomRouteController { const operation = this.operation; const attempt = operation?.current; if ( - operation?.reason === "quality-convergence" && + this.operationRequiresHealthyCandidate(operation) && attempt?.tuple.kind === "peer" && operation.childPeerId === input.childPeerId && operation.childSessionId === child.sessionId && @@ -1418,7 +1426,7 @@ export class RoomRouteController { const operation = this.operation; const attempt = operation?.current; const ownsCandidate = - operation?.reason === "quality-convergence" && + this.operationRequiresHealthyCandidate(operation) && attempt && (attempt.tuple.kind === "peer" ? attempt.tuple.parentPeerId === parentPeerId && @@ -1434,19 +1442,11 @@ export class RoomRouteController { released.push(...validation.released); const failedPeerIds = failedPeerIdsFrom(validation); if ( - this.operation?.reason === "direct-convergence" && + this.operation && + this.isBackgroundConvergence(this.operation.reason) && (this.hasSfuBootstrapWork() || this.selectNextChild()) ) { - this.debug("direct-convergence-preempted", { - child: this.debugPeer(this.operation.childPeerId), - }); - released.push(...this.abortOperation(nowMs, "aborted")); - } - if ( - this.operation?.reason === "quality-convergence" && - (this.hasSfuBootstrapWork() || this.selectNextChild()) - ) { - this.debug("quality-convergence-preempted", { + this.debug(`${this.operation.reason}-preempted`, { child: this.debugPeer(this.operation.childPeerId), }); released.push(...this.abortOperation(nowMs, "aborted")); @@ -1473,11 +1473,16 @@ export class RoomRouteController { routeChildPeerId || bootstrap || continuation ? undefined : this.selectQualityChild(nowMs); + const rootConvergence = + routeChildPeerId || bootstrap || continuation || qualityChildPeerId + ? undefined + : this.takeRootConvergenceIntent(); const childPeerId = routeChildPeerId ?? bootstrap?.carrierPeerId ?? continuation?.childPeerId ?? - qualityChildPeerId; + qualityChildPeerId ?? + rootConvergence?.childPeerId; if (!childPeerId) return { removedPeerIds, failedPeerIds, released }; const child = this.participants.get(childPeerId)!; const reason: RouteDemandReason = continuation @@ -1486,8 +1491,10 @@ export class RoomRouteController { ? "sfu-bootstrap" : qualityChildPeerId ? "quality-convergence" - : (this.routeTimings.get(childPeerId)?.reason ?? - this.routeDemandReason(childPeerId)); + : rootConvergence + ? "root-convergence" + : (this.routeTimings.get(childPeerId)?.reason ?? + this.routeDemandReason(childPeerId)); const demandPeerId = bootstrap?.demandPeerId ?? childPeerId; const demand = this.participants.get(demandPeerId)!; this.ensureDemand(demandPeerId, nowMs, reason); @@ -1495,7 +1502,9 @@ export class RoomRouteController { ? [continuation.plan] : qualityChildPeerId ? this.buildQualityCandidates(childPeerId, nowMs) - : this.buildCandidates(childPeerId, bootstrap !== undefined); + : rootConvergence + ? [rootConvergence.plan] + : this.buildCandidates(childPeerId, bootstrap !== undefined); if (!bootstrap && !continuation && demand.sfuFirstAtNextRoute) { const sfuIndex = candidates.findIndex( (candidate) => candidate.tuple.kind === "sfu", @@ -1515,14 +1524,15 @@ export class RoomRouteController { child: this.debugPeer(childPeerId), reason, }); - if (reason === "direct-convergence") { - this.directContinuations.delete(childPeerId); - continue; - } - if (reason === "quality-convergence") { - const observation = this.senderQualityObservations.get(childPeerId); - if (observation) { - observation.consumedAtFactVersion = this.factVersion; + if (this.isBackgroundConvergence(reason)) { + 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, @@ -1621,8 +1631,9 @@ export class RoomRouteController { }; } if ( - operation.reason === "quality-convergence" && + this.operationRequiresHealthyCandidate(operation) && plan.tuple.kind === "peer" && + operation.reason === "quality-convergence" && !this.qualitySourcePathHealthy(plan.tuple.parentPeerId, input.nowMs) ) { this.noteRejection(operation.demandPeerId, "stale"); @@ -1688,6 +1699,7 @@ export class RoomRouteController { publicationGeneration, publicationConnectionId: input.publicationConnectionId, reservation: input.reservation, + startedAtMs: input.nowMs, transportConnected: false, mediaReady: false, }; @@ -1861,16 +1873,19 @@ export class RoomRouteController { released: validation.released, }; } - if (operation.reason === "quality-convergence") { + if (this.operationRequiresHealthyCandidate(operation)) { attempt.mediaReady = true; const currentEdge = this.upstreamByViewer.get(operation.childPeerId); if ( - !currentEdge || - this.senderQualityState( - operation.childPeerId, - currentEdge, - nowMs, - ) !== "degraded" + (operation.reason === "quality-convergence" && + (!currentEdge || + this.senderQualityState( + operation.childPeerId, + currentEdge, + nowMs, + ) !== "degraded")) || + (operation.reason === "root-convergence" && + !this.rootConvergenceOperationStillEligible(operation)) ) { return { accepted: true, @@ -1912,7 +1927,8 @@ export class RoomRouteController { nowMs >= attempt.senderQualityAcceptedAtMs + VIEWER_QUALITY_EVIDENCE_EXPIRY_MS)) || - (attempt.tuple.kind === "peer" && + (operation.reason === "quality-convergence" && + attempt.tuple.kind === "peer" && !this.qualitySourcePathHealthy( attempt.tuple.parentPeerId, nowMs, @@ -2044,10 +2060,16 @@ export class RoomRouteController { operationExpired(nowMs: number): SettleResult { if (this.operation) { - this.debug("operation-deadline", { + const current = this.operation.current; + const noProgress = + current?.tuple.kind === "peer" && + !current.transportConnected && + nowMs >= current.startedAtMs + this.directHeadStartMs() && + nowMs < this.operation.deadlineAtMs; + this.debug(noProgress ? "candidate-no-progress" : "operation-deadline", { child: this.debugPeer(this.operation.childPeerId), - candidate: this.operation.current - ? this.debugTuple(this.operation.current.tuple) + candidate: current + ? this.debugTuple(current.tuple) : null, cursor: this.operation.cursor, candidateCount: this.operation.candidates.length, @@ -2084,6 +2106,7 @@ export class RoomRouteController { this.directContinuations.clear(); this.sfuBootstrapIntent = undefined; this.sfuBootstrapExhaustedAtFactVersion = undefined; + this.rootConvergenceRootPeerId = undefined; this.paused = true; return [...resources]; } @@ -2127,10 +2150,20 @@ export class RoomRouteController { released.push(...this.abortOperation(nowMs, "aborted")); return result; } + if ( + operation.reason === "root-convergence" && + !this.rootConvergenceOperationStillEligible(operation) + ) { + released.push(...this.abortOperation(nowMs, "aborted")); + return result; + } if (nowMs >= operation.deadlineAtMs) { const directConvergence = operation.reason === "direct-convergence"; const sfuBootstrap = operation.reason === "sfu-bootstrap"; const qualityConvergence = operation.reason === "quality-convergence"; + const backgroundConvergence = this.isBackgroundConvergence( + operation.reason, + ); if (directConvergence) { this.consumeDirectContinuationCandidate(operation); } @@ -2146,15 +2179,13 @@ export class RoomRouteController { ) : undefined; const bootstrapAvailable = - !directConvergence && + !backgroundConvergence && !sfuBootstrap && - !qualityConvergence && !factsChanged && this.bootstrapCandidateAvailable(operation.demandPeerId); const exhausted = sfuBootstrap ? failedBootstrapDemand !== undefined - : !directConvergence && - !qualityConvergence && + : !backgroundConvergence && !factsChanged && !bootstrapAvailable; if (bootstrapAvailable) this.stageSfuBootstrap(operation); @@ -2162,7 +2193,7 @@ export class RoomRouteController { this.finishTiming( operation.demandPeerId, nowMs, - directConvergence || qualityConvergence + backgroundConvergence ? this.currentFinalRoute(operation.demandPeerId) : exhausted ? "failed" @@ -2180,9 +2211,8 @@ export class RoomRouteController { } this.blockAndClear( operation, - !directConvergence && + !backgroundConvergence && !sfuBootstrap && - !qualityConvergence && !factsChanged, released, revisionAdvanced, @@ -2193,7 +2223,7 @@ export class RoomRouteController { return { ...result, exhausted: - directConvergence || qualityConvergence ? undefined : exhausted, + backgroundConvergence ? undefined : exhausted, exhaustedChildPeerId: failedBootstrapDemand ?? (exhausted ? operation.demandPeerId : undefined), expired: true, @@ -2228,12 +2258,17 @@ export class RoomRouteController { if (attempt) { const guardFailed = failedGuard && this.guardMatches(failedGuard, operation, attempt); const plan = operation.candidates[operation.cursor]; + const peerNoProgressExpired = + attempt.tuple.kind === "peer" && + !attempt.transportConnected && + nowMs >= attempt.startedAtMs + this.directHeadStartMs(); const qualityPeerSourceInvalid = operation.reason === "quality-convergence" && attempt.tuple.kind === "peer" && !this.qualitySourcePathHealthy(attempt.tuple.parentPeerId, nowMs); if ( !guardFailed && + !peerNoProgressExpired && !qualityPeerSourceInvalid && plan && this.candidateValid(operation.childPeerId, plan, attempt) @@ -2241,7 +2276,13 @@ export class RoomRouteController { return result; } if (!guardFailed) { - this.noteRejection(operation.demandPeerId, "stale"); + this.noteRejection( + operation.demandPeerId, + peerNoProgressExpired ? "first-frame-timeout" : "stale", + ); + } + if (peerNoProgressExpired) { + result.expired = true; } if (operation.reason === "direct-convergence") { this.consumeDirectContinuationCandidate(operation); @@ -2264,9 +2305,11 @@ export class RoomRouteController { this.clearCandidateTiming(operation.demandPeerId); } if (operation.cursor < operation.candidates.length) return result; - const directConvergence = operation.reason === "direct-convergence"; const sfuBootstrap = operation.reason === "sfu-bootstrap"; const qualityConvergence = operation.reason === "quality-convergence"; + const backgroundConvergence = this.isBackgroundConvergence( + operation.reason, + ); const factsChanged = operation.builtAtFactVersion !== this.factVersion; const rejectionBucket = factsChanged ? "stale" @@ -2283,15 +2326,13 @@ export class RoomRouteController { ) : undefined; const bootstrapAvailable = - !directConvergence && + !backgroundConvergence && !sfuBootstrap && - !qualityConvergence && !factsChanged && this.bootstrapCandidateAvailable(operation.demandPeerId); const exhausted = sfuBootstrap ? failedBootstrapDemand !== undefined - : !directConvergence && - !qualityConvergence && + : !backgroundConvergence && !factsChanged && !bootstrapAvailable; if (bootstrapAvailable) this.stageSfuBootstrap(operation); @@ -2299,7 +2340,7 @@ export class RoomRouteController { this.finishTiming( operation.demandPeerId, nowMs, - directConvergence || qualityConvergence + backgroundConvergence ? this.currentFinalRoute(operation.demandPeerId) : exhausted ? "failed" @@ -2317,15 +2358,14 @@ export class RoomRouteController { } this.blockAndClear( operation, - !directConvergence && + !backgroundConvergence && !sfuBootstrap && - !qualityConvergence && !factsChanged, released, this.revision !== activeRevisionAtStart, ); result.exhausted = - directConvergence || qualityConvergence ? undefined : exhausted; + backgroundConvergence ? undefined : exhausted; if (failedBootstrapDemand) { result.exhaustedChildPeerId = failedBootstrapDemand; } else if (exhausted) { @@ -2339,6 +2379,10 @@ export class RoomRouteController { private commitAttempt(operation: ChildOperation, attempt: Attempt): Resource[] { const beforeResources = new Set(this.committedResources()); const old = this.upstreamByViewer.get(operation.childPeerId); + const createsHostRoot = + attempt.tuple.kind === "peer" && + attempt.tuple.parentPeerId === this.options.hostPeerId && + !(old?.kind === "peer" && old.parentPeerId === this.options.hostPeerId); this.clearQualityForParticipant(operation.childPeerId); if (attempt.tuple.kind === "sfu" && attempt.tuple.publication === "replace") { @@ -2429,6 +2473,9 @@ export class RoomRouteController { this.revision = attempt.revision; this.operation = undefined; this.touchFacts(); + if (createsHostRoot) { + this.stageRootConvergence(operation.childPeerId); + } const participant = this.participants.get(operation.childPeerId); if (participant) { participant.blockedAtFactVersion = undefined; @@ -2658,6 +2705,121 @@ export class RoomRouteController { )[0]?.peerId; } + private stageRootConvergence(rootPeerId: string): void { + if (this.options.qualityConvergenceEnabled !== true) return; + const convergence = this.rootConvergencePlan(rootPeerId); + if (!convergence) return; + this.rootConvergenceRootPeerId = rootPeerId; + this.debug("root-convergence-staged", { + root: this.debugPeer(rootPeerId), + donor: this.debugPeer(convergence.donorPeerId), + child: this.debugPeer(convergence.childPeerId), + }); + } + + private takeRootConvergenceIntent(): { + childPeerId: string; + plan: CandidatePlan; + } | undefined { + const rootPeerId = this.rootConvergenceRootPeerId; + this.rootConvergenceRootPeerId = undefined; + if (!rootPeerId || this.options.qualityConvergenceEnabled !== true) { + return undefined; + } + const convergence = this.rootConvergencePlan(rootPeerId); + return convergence + ? { childPeerId: convergence.childPeerId, plan: convergence.plan } + : undefined; + } + + private rootConvergencePlan(rootPeerId: string): { + donorPeerId: string; + childPeerId: string; + plan: CandidatePlan; + } | undefined { + const root = this.participants.get(rootPeerId); + if ( + !root?.sessionId || + root.departureConfirmed || + !this.isActiveHostRoot(rootPeerId) || + this.activeDirectChildren(rootPeerId).length !== 0 + ) { + return undefined; + } + const donorPeerId = this.childrenOf(this.options.hostPeerId) + .filter( + (peerId) => + peerId !== rootPeerId && + this.isActiveHostRoot(peerId) && + this.activeDirectChildren(peerId).length >= 2, + ) + .sort( + (left, right) => + this.activeDirectChildren(right).length - + this.activeDirectChildren(left).length || + compareParticipant( + this.participants.get(left)!, + this.participants.get(right)!, + ), + )[0]; + if (!donorPeerId) return undefined; + const childPeerId = this.activeDirectChildren(donorPeerId).sort( + (left, right) => + compareParticipant( + this.participants.get(right)!, + this.participants.get(left)!, + ), + )[0]; + if (!childPeerId) return undefined; + const plan = this.planCandidate(childPeerId, { + kind: "peer", + parentPeerId: rootPeerId, + transport: "direct", + }); + return plan && + plan.endpointTransition.kind !== "bounded-gap" && + this.candidateValid(childPeerId, plan) + ? { donorPeerId, childPeerId, plan } + : undefined; + } + + private rootConvergenceOperationStillEligible( + operation: ChildOperation, + ): boolean { + if (operation.reason !== "root-convergence") return true; + const tuple = + operation.current?.tuple ?? operation.candidates[operation.cursor]?.tuple; + const current = this.upstreamByViewer.get(operation.childPeerId); + return Boolean( + tuple?.kind === "peer" && + current?.kind === "peer" && + current.usable && + current.physicalActive && + current.parentPeerId !== tuple.parentPeerId && + this.isActiveHostRoot(tuple.parentPeerId) && + this.isActiveHostRoot(current.parentPeerId) && + this.activeDirectChildren(tuple.parentPeerId).length === 0 && + this.activeDirectChildren(current.parentPeerId).length >= 2, + ); + } + + private isActiveHostRoot(peerId: string): boolean { + const edge = this.upstreamByViewer.get(peerId); + return Boolean( + edge?.kind === "peer" && + edge.parentPeerId === this.options.hostPeerId && + edge.usable && + edge.physicalActive, + ); + } + + private activeDirectChildren(parentPeerId: string): string[] { + return this.childrenOf(parentPeerId).filter((childPeerId) => { + const edge = this.upstreamByViewer.get(childPeerId); + return edge?.kind === "peer" && edge.usable && edge.physicalActive; + }); + } + private senderQualityPersistentlyDegraded( childPeerId: string, edge: CommittedEdge, @@ -2790,6 +2952,23 @@ export class RoomRouteController { ); } + private isBackgroundConvergence(reason: RouteDemandReason): boolean { + return ( + reason === "direct-convergence" || + reason === "quality-convergence" || + reason === "root-convergence" + ); + } + + private operationRequiresHealthyCandidate( + operation: ChildOperation | undefined, + ): operation is ChildOperation { + return ( + operation?.reason === "quality-convergence" || + operation?.reason === "root-convergence" + ); + } + private consumeDirectContinuationCandidate( operation: ChildOperation, ): void { @@ -3421,6 +3600,15 @@ export class RoomRouteController { ) { return this.directHeadStartDeadlineAt(operation); } + if ( + operation.current?.tuple.kind === "peer" && + !operation.current.transportConnected + ) { + return Math.min( + operation.deadlineAtMs, + operation.current.startedAtMs + this.directHeadStartMs(), + ); + } return operation.deadlineAtMs; } @@ -3434,10 +3622,7 @@ export class RoomRouteController { } private foregroundSfuIndex(operation: ChildOperation): number { - if ( - operation.reason === "direct-convergence" || - operation.reason === "quality-convergence" - ) { + if (this.isBackgroundConvergence(operation.reason)) { return -1; } const index = operation.candidates.findIndex( diff --git a/src/server/signaling.ts b/src/server/signaling.ts index 25a1291..5ec5446 100644 --- a/src/server/signaling.ts +++ b/src/server/signaling.ts @@ -36,7 +36,6 @@ import { } from "./room-store.js"; import { HybridMediaRouter, - type ActiveSfuViewerMediaState, type SfuFallbackOptions, } from "./hybrid-media-router.js"; import { createIceConfig, type IceConfigOptions } from "./ice.js"; @@ -79,12 +78,6 @@ interface ViewerQualityEvidenceGate { sequence: number; } -interface ViewerMediaReadyState { - sessionId: string; - revision: number; - sfuPublicationGeneration: string; -} - interface SenderQualityRateWindow { startedAtMs: number; count: number; @@ -129,10 +122,6 @@ export class SignalingServer { string, SenderQualityRateWindow >(); - private readonly viewerMediaReadyByRoom = new Map< - string, - Map - >(); private readonly qualitySettingsByRoom = new Map(); private readonly routePolicyByRoom = new Map(); private readonly shareGenerationsByRoom = new Map(); @@ -191,8 +180,7 @@ export class SignalingServer { deleteConnectionId: (roomId, viewerPeerId) => this.deleteViewerConnectionId(roomId, viewerPeerId), getShareGeneration: (roomId) => this.shareGenerationsByRoom.get(roomId), - onViewerMediaSnapshot: (roomId, viewers) => - this.replaceViewerMediaState(roomId, viewers), + onRoutesChanged: (roomId) => this.sendViewerPresence(roomId), now: this.now, }); } @@ -265,7 +253,6 @@ export class SignalingServer { this.viewerQualityEvidenceGates.clear(); this.viewerQualityEvidenceAttemptAtMs.clear(); this.senderQualityRateBySession.clear(); - this.viewerMediaReadyByRoom.clear(); this.routePolicyByRoom.clear(); this.shareGenerationsByRoom.clear(); this.pausedShareGenerationsByRoom.clear(); @@ -1906,20 +1893,6 @@ export class SignalingServer { } const connectedViewers = this.options.roomStore.getConnectedViewers(roomId); - const currentSessions = new Map( - connectedViewers.map((viewer) => [viewer.peerId, viewer.sessionId]), - ); - const readyByViewer = this.viewerMediaReadyByRoom.get(roomId); - if (readyByViewer) { - for (const [viewerPeerId, ready] of readyByViewer) { - if (currentSessions.get(viewerPeerId) !== ready.sessionId) { - readyByViewer.delete(viewerPeerId); - } - } - if (readyByViewer.size === 0) { - this.viewerMediaReadyByRoom.delete(roomId); - } - } const viewers: ParticipantPresenceEntry[] = []; const host = this.options.roomStore.getConnectedHost(roomId); const hostSocket = host @@ -1962,31 +1935,27 @@ export class SignalingServer { : host && this.isOrdinaryHostChild(roomId, viewer.peerId) ? { kind: "peer" as const, peerId: host.peerId } : { kind: "none" as const }; - const ready = readyByViewer?.get(viewer.peerId); - const sfuMediaReady = - upstream.kind === "sfu" && - ready?.sessionId === viewer.sessionId && - this.hybridMediaRouter!.viewerSfuMediaStateIsCurrent( - roomId, - viewer.peerId, - ready.revision, - ready.sfuPublicationGeneration, - ); - if (ready && !sfuMediaReady) { - readyByViewer?.delete(viewer.peerId); - } + const activeEdge = this.isHybridMediaEnabled() + ? this.hybridMediaRouter!.resolveActiveViewerMediaEdge( + roomId, + viewer.peerId, + ) + : undefined; + const mediaReady = + activeEdge !== undefined && + upstream.kind !== "none" && + (activeEdge.upstream.kind === "sfu" + ? upstream.kind === "sfu" + : upstream.kind === "peer" && + activeEdge.upstream.peerId === upstream.peerId); viewers.push({ role: "viewer", peerId: viewer.peerId, displayName: viewerState.displayName, upstream, - ...(sfuMediaReady ? { sfuMediaReady: true as const } : {}), + ...(mediaReady ? { mediaReady: true as const } : {}), }); } - if (readyByViewer?.size === 0) { - this.viewerMediaReadyByRoom.delete(roomId); - } - const message = { type: "viewer-presence" as const, viewers }; const recipients = new Set(); if (hostSocket && hostState?.viewerPresence) { @@ -2010,26 +1979,6 @@ export class SignalingServer { recipients.forEach((socket) => this.send(socket, message)); } - private replaceViewerMediaState( - roomId: string, - viewers: readonly ActiveSfuViewerMediaState[], - ): void { - if (viewers.length > 0) { - const readyByViewer = new Map(); - for (const viewer of viewers) { - readyByViewer.set(viewer.viewerPeerId, { - sessionId: viewer.viewerSessionId, - revision: viewer.revision, - sfuPublicationGeneration: viewer.sfuPublicationGeneration, - }); - } - this.viewerMediaReadyByRoom.set(roomId, readyByViewer); - } else { - this.viewerMediaReadyByRoom.delete(roomId); - } - this.sendViewerPresence(roomId); - } - private sendToSession(sessionId: string, message: ServerMessage): void { const socket = this.socketsBySessionId.get(sessionId); if (socket) { @@ -2115,12 +2064,6 @@ export class SignalingServer { this.clearViewerGrace(roomId, peerId); this.deleteViewerConnectionId(roomId, peerId); - const readyByViewer = this.viewerMediaReadyByRoom.get(roomId); - readyByViewer?.delete(peerId); - if (readyByViewer?.size === 0) { - this.viewerMediaReadyByRoom.delete(roomId); - } - const ordinaryChildren = this.ordinaryActiveHostChildrenByRoom.get(roomId); ordinaryChildren?.delete(peerId); if (ordinaryChildren?.size === 0) { @@ -2139,7 +2082,6 @@ export class SignalingServer { } private clearRoomConnectionIds(roomId: string): void { - this.viewerMediaReadyByRoom.delete(roomId); const prefix = `${roomId}:`; for (const key of this.connectionIdsByViewer.keys()) { if (key.startsWith(prefix)) { diff --git a/src/shared/protocol.ts b/src/shared/protocol.ts index aca9fbe..185ea18 100644 --- a/src/shared/protocol.ts +++ b/src/shared/protocol.ts @@ -6,7 +6,7 @@ import { isCanonicalVideoCodecEvidence } from "./video-codec-evidence.js"; export const MAX_VIEWERS_PER_ROOM_LIMIT = 20; export const MAX_PARTICIPANTS_PER_ROOM_LIMIT = MAX_VIEWERS_PER_ROOM_LIMIT + 1; export const MAX_SIGNAL_BYTES = 64 * 1024; -export const SIGNALING_PROTOCOL = "screener-v15"; +export const SIGNALING_PROTOCOL = "screener-v16"; export const SIGNAL_CLOSE_CODES = { serviceRestart: 1012, sessionReplaced: 4001, @@ -104,7 +104,7 @@ export const participantPresenceEntrySchema = z.discriminatedUnion("role", [ peerId: opaqueIdSchema, displayName: displayNameSchema, upstream: mediaRouteUpstreamSchema, - sfuMediaReady: z.literal(true).optional(), + mediaReady: z.literal(true).optional(), }) .strict(), ]); @@ -118,7 +118,7 @@ export const viewerPresenceEntrySchema = z peerId: opaqueIdSchema, displayName: displayNameSchema, upstream: mediaRouteUpstreamSchema, - sfuMediaReady: z.literal(true).optional(), + mediaReady: z.literal(true).optional(), }) .strict(); export type ViewerPresenceEntry = z.infer; @@ -363,6 +363,7 @@ export const routeDemandReasonSchema = z.enum([ "sfu-bootstrap", "direct-convergence", "quality-convergence", + "root-convergence", ]); export type RouteDemandReason = z.infer; @@ -649,6 +650,14 @@ const viewerQualityEvidenceWindowShape = { metrics: viewerQualityEvidenceMetricsSchema, }; +const senderQualityDiagnosticsSchema = z + .object({ + reason: z.enum(["none", "bandwidth", "cpu"]).nullable(), + framesPerSecond: nullableEvidenceNumber(240), + bitrateKbps: nullableEvidenceNumber(100_000), + }) + .strict(); + export const viewerQualityEvidenceMessageSchema = z .object({ type: z.literal("viewer-quality-evidence"), @@ -672,6 +681,7 @@ export const senderQualityEvidenceMessageSchema = z .nullable(), routeRevision: mediaRouteRevisionSchema, state: z.enum(["unknown", "healthy", "degraded"]), + diagnostics: senderQualityDiagnosticsSchema, }) .strict() .superRefine((message, context) => { @@ -687,6 +697,20 @@ export const senderQualityEvidenceMessageSchema = z path: ["rtpStatsId"], }); } + const reason = message.diagnostics.reason; + if ( + (message.state === "healthy" && reason !== "none") || + (message.state === "degraded" && + reason !== "bandwidth" && + reason !== "cpu") || + (message.state === "unknown" && reason !== null) + ) { + context.addIssue({ + code: "custom", + message: "Sender quality reason must match its native state", + path: ["diagnostics", "reason"], + }); + } }); export const sfuPublisherQualityEvidenceMessageSchema = z @@ -701,6 +725,7 @@ export const sfuPublisherQualityEvidenceMessageSchema = z .min(0) .max(Number.MAX_SAFE_INTEGER) .nullable(), + diagnostics: senderQualityDiagnosticsSchema, }) .strict() .superRefine((message, context) => { @@ -711,6 +736,20 @@ export const sfuPublisherQualityEvidenceMessageSchema = z path: ["sampleTimestampMs"], }); } + const reason = message.diagnostics.reason; + if ( + (message.state === "healthy" && reason !== "none") || + (message.state === "degraded" && + reason !== "bandwidth" && + reason !== "cpu") || + (message.state === "unknown" && reason !== null) + ) { + context.addIssue({ + code: "custom", + message: "SFU publisher reason must match its native state", + path: ["diagnostics", "reason"], + }); + } }); const authenticateMessageSchema = z.discriminatedUnion("role", [ diff --git a/tests/hybrid-media-router.test.ts b/tests/hybrid-media-router.test.ts index 8cc5d92..9241b91 100644 --- a/tests/hybrid-media-router.test.ts +++ b/tests/hybrid-media-router.test.ts @@ -12,6 +12,15 @@ import { FakeSfuRoomControl } from "./fake-sfu-room-control.ts"; const SHARE_GENERATION = "share_generation_12345678"; +function senderDiagnostics(state: "unknown" | "healthy" | "degraded") { + return { + reason: + state === "unknown" ? null : state === "healthy" ? "none" : "bandwidth", + framesPerSecond: null, + bitrateKbps: null, + } as const; +} + function createStore(maxViewersPerRoom = 20) { return new RoomStore({ leaseMs: 60_000, @@ -113,7 +122,7 @@ function harness( }) : undefined; const roomControl = withSfu ? new FakeSfuRoomControl() : undefined; - const onViewerMediaSnapshot = vi.fn(); + const onRoutesChanged = vi.fn(); let nextTokenIssueError: Error | null = null; const router = new HybridMediaRouter({ roomStore: store, @@ -153,14 +162,14 @@ function harness( connections.delete(`${roomId}:${viewerPeerId}`); }, getShareGeneration: () => SHARE_GENERATION, - onViewerMediaSnapshot, + onRoutesChanged, }); return { store, sent, admission, roomControl, - onViewerMediaSnapshot, + onRoutesChanged, failNextTokenIssue() { nextTokenIssueError = new Error("token issue failed"); }, @@ -253,6 +262,7 @@ describe("HybridMediaRouter v9 runtime", () => { sampleTimestampMs: 100, routeRevision: secondEdge.revision, state: "healthy", + diagnostics: senderDiagnostics("healthy"), }), ).toBe(true); expect( @@ -265,6 +275,7 @@ describe("HybridMediaRouter v9 runtime", () => { sampleTimestampMs: 100, routeRevision: firstEdge.revision, state: "healthy", + diagnostics: senderDiagnostics("healthy"), }), ).toBe(true); for (let window = 0; window < 3; window += 1) { @@ -278,6 +289,7 @@ describe("HybridMediaRouter v9 runtime", () => { sampleTimestampMs: 101 + window, routeRevision: firstEdge.revision, state: "degraded", + diagnostics: senderDiagnostics("degraded"), }), ).toBe(true); } @@ -308,6 +320,7 @@ describe("HybridMediaRouter v9 runtime", () => { sampleTimestampMs: 200, routeRevision: qualityPrepare.revision, state: "healthy", + diagnostics: senderDiagnostics("healthy"), }), ).toBe(true); await vi.waitFor(() => @@ -400,6 +413,7 @@ describe("HybridMediaRouter v9 runtime", () => { sampleTimestampMs: 100 + index, routeRevision: active.revision, state, + diagnostics: senderDiagnostics(state), }), ).toBe(true); } @@ -427,11 +441,11 @@ describe("HybridMediaRouter v9 runtime", () => { }); - it("reports active SFU viewers as one authoritative snapshot", async () => { - const { store, sent, router, onViewerMediaSnapshot } = harness(1, true); + it("notifies the presence owner after the committed graph changes", async () => { + const { store, sent, router, onRoutesChanged } = harness(1, true); try { const room = await store.createRoom(); - const { first, second } = await establishSfuRoom( + await establishSfuRoom( store, sent, router, @@ -439,23 +453,7 @@ describe("HybridMediaRouter v9 runtime", () => { ); await vi.waitFor(() => { - const [reportedRoomId, viewers] = - onViewerMediaSnapshot.mock.calls.at(-1) ?? []; - expect(reportedRoomId).toBe(room.roomId); - expect( - new Set( - (viewers ?? []).map( - (viewer: { viewerPeerId: string }) => viewer.viewerPeerId, - ), - ), - ).toEqual(new Set([first.peerId, second.peerId])); - expect( - new Set( - (viewers ?? []).map( - (viewer: { revision: number }) => viewer.revision, - ), - ).size, - ).toBe(1); + expect(onRoutesChanged.mock.calls.at(-1)).toEqual([room.roomId]); }); } finally { await router.close(); diff --git a/tests/protocol.test.ts b/tests/protocol.test.ts index 447a06c..896619b 100644 --- a/tests/protocol.test.ts +++ b/tests/protocol.test.ts @@ -97,7 +97,7 @@ describe("client signaling protocol", () => { "utf8", ); - expect(SIGNALING_PROTOCOL).toBe("screener-v15"); + expect(SIGNALING_PROTOCOL).toBe("screener-v16"); expect(nativeWire).toMatch(/signalingProtocol\s*=\s*"screener-v6"/); }); @@ -872,6 +872,11 @@ describe("client signaling protocol", () => { sampleTimestampMs: 2_000, routeRevision: 3, state: "degraded", + diagnostics: { + reason: "bandwidth", + framesPerSecond: 24, + bitrateKbps: 1_500, + }, } as const; expect(clientMessageSchema.safeParse(sender).success).toBe(true); expect( @@ -879,6 +884,7 @@ describe("client signaling protocol", () => { ...sender, state: "healthy", rtpStatsId: null, + diagnostics: { ...sender.diagnostics, reason: "none" }, }).success, ).toBe(false); expect( @@ -894,6 +900,11 @@ describe("client signaling protocol", () => { routeRevision: 4, state: "healthy", sampleTimestampMs: 2_000, + diagnostics: { + reason: "none", + framesPerSecond: 30, + bitrateKbps: 2_000, + }, }).success, ).toBe(true); }); @@ -1307,7 +1318,7 @@ describe("server signaling protocol", () => { { ...viewer, upstream: { kind: "sfu" }, - sfuMediaReady: true, + mediaReady: true, }, ], }).success, @@ -1315,7 +1326,7 @@ describe("server signaling protocol", () => { expect( serverMessageSchema.safeParse({ type: "viewer-presence", - viewers: [{ ...viewer, sfuMediaReady: false }], + viewers: [{ ...viewer, mediaReady: false }], }).success, ).toBe(false); expect( diff --git a/tests/room-route-controller.test.ts b/tests/room-route-controller.test.ts index 1b43e0c..5c4a53c 100644 --- a/tests/room-route-controller.test.ts +++ b/tests/room-route-controller.test.ts @@ -24,6 +24,7 @@ function controller( options: { sfuEnabled?: boolean; qualityConvergenceEnabled?: boolean; + operationTimeoutMs?: number; } = {}, ) { const routes = new RoomRouteController({ @@ -410,6 +411,101 @@ describe("RoomRouteController", () => { ); }); + it("uses one healthy serial operation to distribute a newly committed Host root", () => { + const routes = controller(2, { qualityConvergenceEnabled: true }); + addViewer(routes, B, 2); + addViewer(routes, C, 0); + addViewer(routes, D, 0); + routes.hydrateEdge(B, peerEdge(HOST, "b_from_host")); + routes.hydrateEdge(C, peerEdge(B, "c_from_b")); + routes.hydrateEdge(D, peerEdge(B, "d_from_b")); + addViewer(routes, A, 2, 10); + + const joined = routes.reconcile(10).operation!; + expect(joined.candidates[0]?.tuple).toEqual({ + kind: "peer", + parentPeerId: HOST, + transport: "direct", + }); + commitCurrent(routes, 11, "a_from_host"); + + const convergence = routes.reconcile(13).operation!; + expect(convergence).toMatchObject({ reason: "root-convergence" }); + expect(convergence.candidates).toHaveLength(1); + expect(convergence.candidates[0]?.tuple).toEqual({ + kind: "peer", + parentPeerId: A, + transport: "direct", + }); + const oldParent = ( + routes.snapshot().upstreamByViewer.get(convergence.childPeerId) as { + parentPeerId: string; + } + ).parentPeerId; + expect(oldParent).toBe(B); + + const prepared = beginCandidate(routes, { + nowMs: 14, + connectionId: "root_convergence_candidate", + reservation: { kind: "direct" }, + }).operation!; + const guard = { + childPeerId: convergence.childPeerId, + childSessionId: convergence.childSessionId, + revision: prepared.current!.revision, + connectionId: "root_convergence_candidate", + }; + expect(routes.candidateReady(guard, 15)).toMatchObject({ + accepted: true, + committed: false, + }); + expect( + routes.observeSenderQualityEvidence({ + parentPeerId: A, + parentSessionId: `${A}_session`, + childPeerId: convergence.childPeerId, + routeRevision: prepared.current!.revision, + connectionId: "root_convergence_candidate", + senderIdentity: "root-rtp\u0000track", + sampleTimestampMs: 16, + state: "healthy", + acceptedAtMs: 16, + }).committed, + ).toBe(true); + expect( + routes.snapshot().upstreamByViewer.get(convergence.childPeerId), + ).toMatchObject({ parentPeerId: A }); + expect(routes.reconcile(17).operation).toBeUndefined(); + }); + + it("gives each unconnected Peer candidate its own bounded progress window", () => { + const routes = controller(2, { operationTimeoutMs: 20_000 }); + addViewer(routes, B, 2); + addViewer(routes, C, 2); + routes.hydrateEdge(B, peerEdge(HOST, "b_from_host")); + routes.hydrateEdge(C, peerEdge(HOST, "c_from_host")); + addViewer(routes, A, 0, 0); + + const operation = routes.reconcile(0).operation!; + beginCandidate(routes, { + nowMs: 100, + connectionId: "first_silent_candidate", + reservation: { kind: "direct" }, + }); + expect(routes.snapshot().operation?.wakeAtMs).toBe(5_100); + expect(routes.operationExpired(5_100).accepted).toBe(true); + + beginCandidate(routes, { + nowMs: 5_200, + connectionId: "second_silent_candidate", + reservation: { kind: "direct" }, + }); + expect(routes.snapshot().operation?.wakeAtMs).toBe(10_200); + expect(routes.snapshot().operation?.deadlineAtMs).toBe( + operation.deadlineAtMs, + ); + }); + it("keeps only fresh exact-edge quality shadow aggregates", () => { const routes = controller(2); addViewer(routes, A, 1, 0); diff --git a/tests/server-signal.test.ts b/tests/server-signal.test.ts index 675be26..a71fd7b 100644 --- a/tests/server-signal.test.ts +++ b/tests/server-signal.test.ts @@ -905,6 +905,9 @@ describe("WebSocket signaling", () => { viewer.upstream.peerId === hostAuth.peerId, ), ).toHaveLength(2); + expect( + viewerPresenceEntries(full).every((viewer) => viewer.mediaReady === true), + ).toBe(true); const relay = viewerPresenceEntries(full).find( (viewer) => viewer.upstream.kind === "peer" &&