diff --git a/docs/adr/0005-automatic-hybrid-media-routing.md b/docs/adr/0005-automatic-hybrid-media-routing.md index f4f1b69..4b85aa4 100644 --- a/docs/adr/0005-automatic-hybrid-media-routing.md +++ b/docs/adr/0005-automatic-hybrid-media-routing.md @@ -181,6 +181,12 @@ the LiveKit namespace to Screener and complete startup ownership/cleanup before accepting traffic; [self-hosting operations](../operations/self-hosting.md) owns that procedure. +Retiring a publication generation releases every physical SFU handle. An old +SFU root that still carries Peer descendants remains only as an inactive graph +anchor until those descendants are reassigned; a childless old root is removed +immediately. This preserves one source-reachable transition graph without +keeping old media active. + Only confirmed LiveKit media-participant absence retires an abandoned Host publication. A Host signaling disconnect keeps the generation charged while the exact LiveKit Host participant remains; one bounded control-plane check releases @@ -240,6 +246,12 @@ time and the old route keeps playing. The same Viewer compares fresh, overlapping old and candidate receive windows. A P2P candidate commits only after first decoded frame and three consecutive windows with no freeze/pause, no lower pixel area or rounded FPS, and a strict improvement in at least one of those dimensions. +An otherwise complete current-route window with zero decoded frames is compared +as zero delivered pixels, FPS, and bitrate. It can approve only a temporally +overlapping candidate window that actually decoded video and satisfies the same +freeze, pause, and P2P/SFU partial-order checks. If both routes decode nothing, +the source may be stalled, so the window remains unknown and cannot justify a +move. Bitrate does not rank P2P candidates because codec and content phase make it non-monotonic. Three consecutive comparable non-improving windows reject the candidate and advance the existing cursor without resetting the deadline. A diff --git a/docs/product/routing-transport.md b/docs/product/routing-transport.md index 9882685..f013fc1 100644 --- a/docs/product/routing-transport.md +++ b/docs/product/routing-transport.md @@ -113,9 +113,11 @@ per-share convergence gate is enabled, Screener uses an exact persistent native sender limitation only to trigger one measured experiment through the existing serial operation. The same Viewer must then prove a P2P candidate strictly improves delivered pixel area or rounded FPS without regressing either dimension -or producing a freeze/pause across three fresh overlapping windows. A persistently -limited exact candidate sender cannot commit. Ordinary quality moves remain -Peer-to-Peer. SFU may +or producing a freeze/pause across three fresh overlapping windows. 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 +persistently limited exact candidate sender cannot commit. 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/src/client/lib/signaling.ts b/src/client/lib/signaling.ts index dcf5962..82afc29 100644 --- a/src/client/lib/signaling.ts +++ b/src/client/lib/signaling.ts @@ -89,6 +89,7 @@ export class SignalingClient { private watchdogTimer: number | null = null; private watchdogDeadlineMs = 0; private pendingChallenge: PendingSignalingChallenge | null = null; + private previousChallenge: PendingSignalingChallenge | null = null; private visibilityListenerAttached = false; private hostQualityIntent: HostQualityIntent | null = null; @@ -507,6 +508,9 @@ export class SignalingClient { : this.challengeSequence + 1; const sequence = this.challengeSequence; socket.send(JSON.stringify({ type: "signaling-challenge", sequence })); + this.previousChallenge = this.pendingChallenge?.generation === generation + ? this.pendingChallenge + : null; this.pendingChallenge = { generation, sequence, confirm }; this.armSignalingWatchdog(SIGNALING_CHALLENGE_TIMEOUT_MS); } @@ -516,10 +520,14 @@ export class SignalingClient { generation: number, ): void { const pending = this.pendingChallenge; + const previous = this.previousChallenge; if ( - !pending || - pending.generation !== generation || - pending.sequence !== sequence || + ![pending, previous].some( + (challenge) => + challenge !== null && + challenge.generation === generation && + challenge.sequence === sequence, + ) || generation !== this.socketGeneration ) { return; @@ -553,6 +561,7 @@ export class SignalingClient { private clearSignalingWatchdog(): void { this.pendingChallenge = null; + this.previousChallenge = null; if (this.watchdogTimer !== null) { window.clearTimeout(this.watchdogTimer); this.watchdogTimer = null; diff --git a/src/client/media/candidate-quality-probe.ts b/src/client/media/candidate-quality-probe.ts index a3ce7ce..971b657 100644 --- a/src/client/media/candidate-quality-probe.ts +++ b/src/client/media/candidate-quality-probe.ts @@ -9,6 +9,30 @@ interface DeliveredVideoSample { bitrateKbps: number | null; } +function deliveredNothingSample( + metrics: ConnectionMetrics, +): DeliveredVideoSample | null { + const { sampleTimestampMs, sampleWindowMs, intervalFramesDecoded } = metrics; + if ( + sampleTimestampMs === null || + sampleWindowMs === null || + !Number.isFinite(sampleTimestampMs) || + !Number.isFinite(sampleWindowMs) || + sampleWindowMs < 1_000 || + sampleWindowMs > 5_000 || + intervalFramesDecoded !== 0 + ) { + return null; + } + return { + timestampMs: sampleTimestampMs, + windowMs: sampleWindowMs, + pixels: 0, + framesPerSecond: 0, + bitrateKbps: 0, + }; +} + function deliveredVideoSample( metrics: ConnectionMetrics, ): DeliveredVideoSample | null { @@ -60,7 +84,9 @@ function comparableSamples( current: ConnectionMetrics, candidate: ConnectionMetrics, ): [DeliveredVideoSample, DeliveredVideoSample] | null { - const currentVideo = deliveredVideoSample(current); + // A dead incumbent is comparable only when the candidate actually delivers. + const currentVideo = + deliveredVideoSample(current) ?? deliveredNothingSample(current); const candidateVideo = deliveredVideoSample(candidate); if ( !currentVideo || @@ -139,16 +165,29 @@ class ConsecutiveCandidateQualityProbe { current: ConnectionMetrics | null, candidate: ConnectionMetrics, ): CandidateQualityProbeResult { + const candidateTimestampMs = candidate.sampleTimestampMs; if ( - candidate.sampleTimestampMs === null || - candidate.sampleTimestampMs === this.lastCandidateTimestampMs + candidateTimestampMs === null || + !Number.isFinite(candidateTimestampMs) ) { + this.resetRuns(); return "pending"; } - this.lastCandidateTimestampMs = candidate.sampleTimestampMs; + if (candidateTimestampMs === this.lastCandidateTimestampMs) { + return "pending"; + } + if ( + this.lastCandidateTimestampMs !== null && + candidateTimestampMs < this.lastCandidateTimestampMs + ) { + this.resetRuns(); + return "pending"; + } + this.lastCandidateTimestampMs = candidateTimestampMs; if ( !current || current.sampleTimestampMs === null || + !Number.isFinite(current.sampleTimestampMs) || (this.lastCurrentTimestampMs !== null && current.sampleTimestampMs <= this.lastCurrentTimestampMs) ) { diff --git a/src/client/media/host-sfu-route.ts b/src/client/media/host-sfu-route.ts index c6fef9b..6110450 100644 --- a/src/client/media/host-sfu-route.ts +++ b/src/client/media/host-sfu-route.ts @@ -234,11 +234,11 @@ export class HostSfuRoute { let slot: HostPublisherSlot; const publisher = this.events.createPublisher?.( - () => this.handleFailure(slot), + () => this.failPublisherSlot(slot), (metrics) => this.handlePublisherStats(slot, metrics), ) ?? new SfuPublisher({ - onDisconnected: () => this.handleFailure(slot), + onDisconnected: () => this.failPublisherSlot(slot), onStats: (metrics) => this.handlePublisherStats(slot, metrics), }); slot = { @@ -258,11 +258,20 @@ export class HostSfuRoute { if (!connected) { if (this.pending === slot && this.route.owns(token)) { this.handleFailure(slot); + } else { + if (this.pending === slot) { + this.pending = null; + } + slot.failed = true; } await disconnectPublisher(publisher); return; } if (this.pending !== slot || !this.route.owns(token)) { + if (this.pending === slot) { + this.pending = null; + } + slot.failed = true; await disconnectPublisher(publisher); return; } @@ -273,6 +282,7 @@ export class HostSfuRoute { if (this.pending === slot) { this.pending = null; } + slot.failed = true; await disconnectPublisher(publisher); return; } @@ -285,8 +295,8 @@ export class HostSfuRoute { await this.queueActivation(token); } } catch { + this.failPublisherSlot(slot); await disconnectPublisher(publisher); - this.handleFailure(slot); } } @@ -319,15 +329,39 @@ export class HostSfuRoute { : null; } - replaceStream(stream: MediaStream): Promise { + async replaceStream(stream: MediaStream): Promise { const slots = this.publishingSlots(); - return slots.length === 0 - ? Promise.resolve(true) - : Promise.all( - slots.map((slot) => - slot.publisher.replaceStream(stream).catch(() => false), - ), - ).then((results) => results.every(Boolean)); + const initialActive = this.active; + if (slots.length === 0) { + return true; + } + + const results = await Promise.all( + slots.map((slot) => slot.publisher.replaceStream(stream).catch(() => false)), + ); + let replaced = true; + await this.queueTransition(async () => { + const currentActive = this.active; + if (currentActive) { + const activeIndex = slots.indexOf(currentActive); + replaced = activeIndex < 0 || results[activeIndex] === true; + } else if (initialActive) { + const activeIndex = slots.indexOf(initialActive); + const stillPlanned = + this.route.getPlannedAssignment()?.sfuPublicationGeneration === + initialActive.publicationGeneration; + replaced = + activeIndex < 0 || results[activeIndex] === true || !stillPlanned; + } + + for (const [index, slot] of slots.entries()) { + if (results[index] === false) { + this.failPublisherSlot(slot); + await disconnectPublisher(slot.publisher); + } + } + }); + return replaced; } private publishingSlots(): HostPublisherSlot[] { @@ -337,18 +371,6 @@ export class HostSfuRoute { ); } - async failActivePublisher(): Promise { - await this.queueTransition(async () => { - const slot = this.active; - if (!slot) { - return; - } - this.active = null; - await disconnectPublisher(slot.publisher); - this.handleFailure(slot); - }); - } - async disconnect(): Promise { if (this.closed) { return; @@ -363,6 +385,12 @@ export class HostSfuRoute { const active = this.active; this.pending = null; this.active = null; + if (pending) { + pending.failed = true; + } + if (active) { + active.failed = true; + } if (active?.active) { await active.publisher.deactivate().catch(() => false); } @@ -455,12 +483,14 @@ export class HostSfuRoute { this.active = pending; if (!this.route.markMediaActive(token)) { this.active = previous; + pending.failed = true; await disconnectPublisher(pending.publisher); return; } this.lastFailureStage = null; this.recovery = null; if (previous && previous !== pending) { + previous.failed = true; if (previous.active) { await previous.publisher.deactivate().catch(() => false); } @@ -473,12 +503,12 @@ export class HostSfuRoute { const profile = this.events.getProfile(); const videoCodec = this.events.getVideoCodec(); if (!stream) { - this.handleFailure(slot); + this.failPublisherSlot(slot); return false; } try { if (!(await slot.publisher.activate(stream, profile, videoCodec))) { - this.handleFailure(slot); + this.failPublisherSlot(slot); await disconnectPublisher(slot.publisher); return false; } @@ -493,7 +523,7 @@ export class HostSfuRoute { latestStream !== stream && !(await slot.publisher.replaceStream(latestStream)) ) { - this.handleFailure(slot); + this.failPublisherSlot(slot); await disconnectPublisher(slot.publisher); return false; } @@ -502,13 +532,13 @@ export class HostSfuRoute { !qualitySettingsEqual(latestProfile, profile) && !(await slot.publisher.updateProfile(latestProfile)) ) { - this.handleFailure(slot); + this.failPublisherSlot(slot); await disconnectPublisher(slot.publisher); return false; } return this.pending === slot && this.ownsPublisherSlot(slot); } catch { - this.handleFailure(slot); + this.failPublisherSlot(slot); await disconnectPublisher(slot.publisher); return false; } @@ -536,6 +566,7 @@ export class HostSfuRoute { } const pending = this.pending; this.pending = null; + pending.failed = true; void disconnectPublisher(pending.publisher); } @@ -545,6 +576,7 @@ export class HostSfuRoute { } const active = this.active; this.active = null; + active.failed = true; await this.disconnectRetiredPublisher(active); } @@ -559,6 +591,7 @@ export class HostSfuRoute { } const active = this.active; this.active = null; + active.failed = true; return active; } @@ -577,15 +610,16 @@ export class HostSfuRoute { } const wasActive = this.active === slot; const wasPending = this.pending === slot; + if (!wasActive && !wasPending) { + slot.failed = true; + return; + } const assignment = this.route.getPlannedAssignment(); const revision = this.route.getRevision(); const phase = this.route.getPhase(); const matchesPlannedRoute = revision === slot.revision && assignment?.sfuPublicationGeneration === slot.publicationGeneration; - if (!wasActive && !wasPending && !matchesPlannedRoute) { - return; - } this.lastFailureStage = slot.publisher.getFailureStage?.() ?? "transport"; slot.failed = true; if (wasPending) { @@ -610,6 +644,14 @@ export class HostSfuRoute { this.requestRecovery(revision); } + private failPublisherSlot(slot: HostPublisherSlot): void { + if (this.pending !== slot && this.active !== slot) { + slot.failed = true; + return; + } + this.handleFailure(slot); + } + private requestRecovery(revision: number): void { if (!this.recovery || this.recovery.revision !== revision) { this.recovery = { revision, refreshed: false }; diff --git a/src/client/media/viewer-sfu-route.ts b/src/client/media/viewer-sfu-route.ts index 3476e11..7c67c1c 100644 --- a/src/client/media/viewer-sfu-route.ts +++ b/src/client/media/viewer-sfu-route.ts @@ -502,7 +502,7 @@ export class ViewerSfuRoute { this.events.onSfuState?.(state, slot.revision); } }, - onDisconnected: () => this.handleFailure(slot), + onDisconnected: () => this.failSubscriberSlot(slot), }; const subscriber = this.events.createSubscriber?.(subscriberEvents) ?? @@ -540,19 +540,28 @@ export class ViewerSfuRoute { if (this.pending === slot && currentToken) { slot.revision = currentToken.revision; this.handleFailure(slot); + } else { + if (this.pending === slot) { + this.pending = null; + } + slot.failed = true; } await disconnectSubscriber(subscriber); return; } const currentToken = this.currentPendingToken(slot, token); if (this.pending !== slot || !currentToken) { + if (this.pending === slot) { + this.pending = null; + } + slot.failed = true; await disconnectSubscriber(subscriber); return; } slot.revision = currentToken.revision; slot.connected = true; if (!subscriber.activate()) { - this.handleFailure(slot); + this.failSubscriberSlot(slot); await disconnectSubscriber(subscriber); return; } @@ -564,8 +573,8 @@ export class ViewerSfuRoute { await this.queueActiveRoute(currentToken, true); } } catch { + this.failSubscriberSlot(slot); await disconnectSubscriber(subscriber); - this.handleFailure(slot); } } @@ -596,6 +605,12 @@ export class ViewerSfuRoute { const active = this.active; this.pending = null; this.active = null; + if (pending) { + pending.failed = true; + } + if (active) { + active.failed = true; + } this.events.onSfuUpdate?.( null, active?.revision ?? pending?.revision ?? this.route.getRevision() ?? 0, @@ -826,6 +841,7 @@ export class ViewerSfuRoute { this.active = slot; if (!this.route.markMediaActive(token)) { this.active = previous; + slot.failed = true; await disconnectSubscriber(slot.subscriber); return; } @@ -836,6 +852,7 @@ export class ViewerSfuRoute { slot.mediaAvailable = true; this.events.onSfuStream(slot.stream, assignment, true, slot.revision); if (previous && previous !== slot) { + previous.failed = true; try { previous.subscriber.deactivate(); } catch { @@ -880,6 +897,7 @@ export class ViewerSfuRoute { } const pending = this.pending; this.pending = null; + pending.failed = true; void disconnectSubscriber(pending.subscriber); } @@ -889,6 +907,7 @@ export class ViewerSfuRoute { } const active = this.active; this.active = null; + active.failed = true; this.events.onSfuUpdate?.(null, active.revision); await this.disconnectRetiredSubscriber(active); } @@ -908,6 +927,7 @@ export class ViewerSfuRoute { } const active = this.active; this.active = null; + active.failed = true; this.events.onSfuUpdate?.(null, active.revision); return active; } @@ -930,8 +950,13 @@ export class ViewerSfuRoute { return; } const wasActive = this.active === slot; + const wasPending = this.pending === slot; + if (!wasActive && !wasPending) { + slot.failed = true; + return; + } const failedManualReconnect = - this.pending === slot && + wasPending && this.manualReconnectRevision === slot.revision && this.active?.publicationGeneration === slot.publicationGeneration && this.active.activated && @@ -980,6 +1005,14 @@ export class ViewerSfuRoute { this.requestRecovery(revision); } + private failSubscriberSlot(slot: ViewerSubscriberSlot): void { + if (this.pending !== slot && this.active !== slot) { + slot.failed = true; + return; + } + this.handleFailure(slot); + } + private requestRecovery(revision: number): void { if (!this.recovery || this.recovery.revision !== revision) { this.recovery = { revision, refreshed: false }; diff --git a/src/client/pages/HostPage.tsx b/src/client/pages/HostPage.tsx index a21d213..d0e8393 100644 --- a/src/client/pages/HostPage.tsx +++ b/src/client/pages/HostPage.tsx @@ -2146,14 +2146,6 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { activeSfuRoute && hostSfuRouteRef.current === activeSfuRoute ? activeSfuRoute.getQualityWarning() : null; - if ( - !sfuReplaced && - activeSfuRoute && - hostSfuRouteRef.current === activeSfuRoute - ) { - await activeSfuRoute.failActivePublisher(); - } - previousStream.getTracks().forEach((track) => track.stop()); if (retiringStreamRef.current === previousStream) { retiringStreamRef.current = null; diff --git a/src/client/pages/ViewerPage.tsx b/src/client/pages/ViewerPage.tsx index fceb459..4e30d17 100644 --- a/src/client/pages/ViewerPage.tsx +++ b/src/client/pages/ViewerPage.tsx @@ -94,10 +94,10 @@ import type { PeerSnapshot, } from "../types"; import { + createOwnedViewerRestartSender, limitMediaAssignment, MAX_ENDPOINT_MEDIA_CHILDREN, type MediaAssignment, - viewerRestartMessage, viewerSignalMessage, } from "../webrtc/media-assignment"; import { ViewerPeer } from "../webrtc/viewer-peer"; @@ -1243,7 +1243,20 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { { sendSignal: (targetPeerId, payload) => signal.send(viewerSignalMessage(peerAssisted, targetPeerId, payload)), - sendRestartRequest: () => false, + sendRestartRequest: createOwnedViewerRestartSender( + peerAssisted, + (targetPeerId, connectionId) => { + const identity = peer.getConnectionIdentity(); + return ( + active && + pendingPeer !== probe && + peerRef.current === peer && + identity?.parentPeerId === targetPeerId && + identity.connectionId === connectionId + ); + }, + (message) => signal.send(message), + ), onStream: (stream) => { if (pendingPeer === probe) { probe.stream = stream; @@ -1357,15 +1370,19 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { signal.send( viewerSignalMessage(peerAssisted, targetPeerId, payload), ), - sendRestartRequest: (targetPeerId, connectionId, rebuild) => - signal.send( - viewerRestartMessage( - peerAssisted, - targetPeerId, - connectionId, - rebuild, - ), - ), + sendRestartRequest: createOwnedViewerRestartSender( + peerAssisted, + (targetPeerId, connectionId) => { + const identity = peer.getConnectionIdentity(); + return ( + active && + peerRef.current === peer && + identity?.parentPeerId === targetPeerId && + identity.connectionId === connectionId + ); + }, + (message) => signal.send(message), + ), onStream: (nextStream) => { if (active) { bindRemoteStream(nextStream, currentRouteRevision); diff --git a/src/client/sfu/publisher.ts b/src/client/sfu/publisher.ts index bcda39f..d5cde2d 100644 --- a/src/client/sfu/publisher.ts +++ b/src/client/sfu/publisher.ts @@ -136,6 +136,7 @@ export class SfuPublisher { } const room = new sdk.Room({ + disconnectOnPageLeave: false, dynacast: true, stopLocalTrackOnUnpublish: false, }); @@ -441,6 +442,7 @@ export class SfuPublisher { if (!this.owns(room, generation)) { return false; } + this.failureStage = "source"; const failureWarning = say("host.fail.sfuSwitch"); // A rejected publish/unpublish may have changed server state without diff --git a/src/client/sfu/subscriber.ts b/src/client/sfu/subscriber.ts index 3db175a..dc6a654 100644 --- a/src/client/sfu/subscriber.ts +++ b/src/client/sfu/subscriber.ts @@ -278,7 +278,10 @@ export class SfuSubscriber { } } this.desiredTrackSids.clear(); - this.clearMedia(true); + // LiveKit full reconnect removes participants before it emits + // Reconnecting. Keep any route-owned proof intent until the route is + // deactivated or the subscriber terminates. + this.clearMedia(true, true); } }, ); @@ -290,6 +293,7 @@ export class SfuSubscriber { room.on(sdk.RoomEvent.Reconnected, () => { if (this.owns(room, generation) && this.state === "active") { this.reconcileHostSubscriptions(room, sdk, generation); + this.startDecodedFrameProof(); this.events.onState?.("connected"); void this.updateStats(); } @@ -625,10 +629,14 @@ export class SfuSubscriber { this.audioStatsInFlight = null; } - private clearMedia(notify: boolean): void { + private clearMedia(notify: boolean, preserveDecodedFrameProof = false): void { const hadStream = this.streamEmitted; const hadVideo = this.video !== null; - this.stopDecodedFrameProof(); + if (preserveDecodedFrameProof) { + this.cancelDecodedFrameObserver(); + } else { + this.stopDecodedFrameProof(); + } if (this.state !== "active") { this.stopStats(); } diff --git a/src/client/webrtc/host-peer.ts b/src/client/webrtc/host-peer.ts index 57986ea..c378bea 100644 --- a/src/client/webrtc/host-peer.ts +++ b/src/client/webrtc/host-peer.ts @@ -74,6 +74,8 @@ export class HostPeer { private senderMutationTail: Promise = Promise.resolve(); private negotiationTail: Promise = Promise.resolve(); private startupVideoProfilePending: boolean; + private connectedOnce = false; + private awaitingReconnect = false; private snapshot: PeerSnapshot; constructor( @@ -360,6 +362,20 @@ export class HostPeer { }); }); this.connection.addEventListener("connectionstatechange", () => { + const state = this.connection.connectionState; + if (state === "connected") { + const replayProfile = this.awaitingReconnect; + this.connectedOnce = true; + this.awaitingReconnect = false; + if (replayProfile) { + this.appliedVideoProfile = null; + this.appliedAudioQuality = null; + this.appliedAudioSenderParameters = null; + void this.updateProfile(this.desiredProfile); + } + } else if (this.connectedOnce && state !== "closed") { + this.awaitingReconnect = true; + } this.emit(); }); this.connection.addEventListener("iceconnectionstatechange", () => this.emit()); diff --git a/src/client/webrtc/media-assignment.ts b/src/client/webrtc/media-assignment.ts index 5bb1dbf..8424244 100644 --- a/src/client/webrtc/media-assignment.ts +++ b/src/client/webrtc/media-assignment.ts @@ -79,3 +79,26 @@ export function viewerRestartMessage( } : { type: "restart-request", connectionId, rebuild }; } + +export function createOwnedViewerRestartSender( + peerAssisted: boolean, + ownsActivePeer: (targetPeerId: string, connectionId: string) => boolean, + send: ( + message: Extract, + ) => boolean, +): ( + targetPeerId: string, + connectionId: string, + rebuild: boolean, +) => boolean { + return (targetPeerId, connectionId, rebuild) => + ownsActivePeer(targetPeerId, connectionId) && + send( + viewerRestartMessage( + peerAssisted, + targetPeerId, + connectionId, + rebuild, + ), + ); +} diff --git a/src/server/hybrid-media-router.ts b/src/server/hybrid-media-router.ts index ee20f53..70443f6 100644 --- a/src/server/hybrid-media-router.ts +++ b/src/server/hybrid-media-router.ts @@ -77,6 +77,8 @@ type SfuDrainTask = SfuDrainTarget & { interface HostOfflineCheck { hostPeerId: string; fence: SfuResourceFence; + hostSessionId: string; + connectionId: string; timer: NodeJS.Timeout; } @@ -216,6 +218,21 @@ export class HybridMediaRouter { connectParticipant(input: AuthenticatedRouteParticipant): HybridAuthenticationState { const room = this.room(input.roomId); if (input.role === "host") { + if (room.controller && room.hostPeerId !== input.peerId) { + this.clearDeadline(room); + for (const viewerPeerId of this.options.roomStore.getViewerPeerIds( + input.roomId, + )) { + this.options.deleteConnectionId(input.roomId, viewerPeerId); + } + this.releaseResources( + room.controller.rebindHostIdentity( + input.peerId, + input.sessionId, + this.now(), + ), + ); + } room.hostPeerId = input.peerId; this.cancelHostOfflineCheck(input.roomId); if (!room.controller) this.createController(input.roomId, room, input); @@ -1181,6 +1198,7 @@ export class HybridMediaRouter { route: this.debugTuple(roomId, plan.tuple), cursor: operation.cursor, }); + let beginGuard = guard; let currentOperation = controller.snapshot().operation; if (plan.endpointTransition.kind === "bounded-gap") { const beforeGap = controller.snapshot().revision; @@ -1197,14 +1215,15 @@ export class HybridMediaRouter { } if (controller.snapshot().revision !== beforeGap) this.broadcastActive(roomId, room); currentOperation = controller.snapshot().operation; - } - const currentPlan = currentOperation?.candidates[currentOperation.cursor]; - if (!currentOperation || !currentPlan) { - this.releaseReservation(preparation.prepared.reservation); - continue; + const currentPlan = currentOperation?.candidates[currentOperation.cursor]; + if (!currentOperation || !currentPlan) { + this.releaseReservation(preparation.prepared.reservation); + continue; + } + beginGuard = cursorGuard(currentOperation, currentPlan); } const begun = controller.beginCurrentCandidate({ - guard: cursorGuard(currentOperation, currentPlan), + guard: beginGuard, nowMs: this.now(), connectionId: preparation.prepared.connectionId, reservation: preparation.prepared.reservation, @@ -1841,23 +1860,34 @@ export class HybridMediaRouter { }); } - private activeSfuFence(roomId: string): SfuResourceFence | null { + private activeHostPublication( + roomId: string, + ): Omit | null { const publication = this.rooms.get(roomId)?.controller?.snapshot().hostPublication; - return publication?.physicalActive && publication.resource.kind === "sfu-publication" - ? publication.resource.fence + return publication?.physicalActive && + publication.resource.kind === "sfu-publication" + ? { + fence: publication.resource.fence, + hostSessionId: publication.hostSessionId, + connectionId: publication.connectionId, + } : null; } private scheduleHostOfflineCheck(roomId: string, hostPeerId: string): void { this.cancelHostOfflineCheck(roomId); const fallback = this.options.sfuFallback; - const fence = this.activeSfuFence(roomId); - if (!fallback || !fence || this.closing) return; + const publication = this.activeHostPublication(roomId); + if (!fallback || !publication || this.closing) return; const timer = setTimeout(() => { void this.checkHostOffline(roomId).catch(() => undefined); }, fallback.hostOfflineCheckMs ?? DEFAULT_HOST_OFFLINE_CHECK_MS); timer.unref(); - this.hostOfflineChecks.set(roomId, { hostPeerId, fence, timer }); + this.hostOfflineChecks.set(roomId, { + hostPeerId, + ...publication, + timer, + }); } private async checkHostOffline(roomId: string): Promise { @@ -1872,6 +1902,7 @@ export class HybridMediaRouter { try { exists = await fallback.roomControl.hostParticipantExists(check.fence); } catch { + if (this.hostOfflineChecks.get(roomId) !== check) return; this.hostOfflineChecks.delete(roomId); this.scheduleHostOfflineCheck(roomId, check.hostPeerId); return; @@ -1887,6 +1918,22 @@ export class HybridMediaRouter { const snapshot = controller?.snapshot(); const publication = snapshot?.hostPublication; if (!room || !controller || !snapshot || !publication) return; + if ( + publication.resource.kind !== "sfu-publication" || + publication.resource.fence.roomId !== check.fence.roomId || + publication.resource.fence.shareGeneration !== + check.fence.shareGeneration || + publication.resource.fence.publicationGeneration !== + check.fence.publicationGeneration || + publication.generation !== check.fence.publicationGeneration || + publication.hostSessionId !== check.hostSessionId || + publication.connectionId !== check.connectionId || + !publication.physicalActive + ) { + this.scheduleHostOfflineCheck(roomId, check.hostPeerId); + return; + } + const beforeRevision = snapshot.revision; controller.invalidateHostPublication({ hostSessionId: publication.hostSessionId, routeRevision: snapshot.revision, @@ -1902,6 +1949,19 @@ export class HybridMediaRouter { connectionId: publication.connectionId, }), ); + const afterRetirement = controller.snapshot().hostPublication; + if ( + afterRetirement?.generation === publication.generation && + afterRetirement.connectionId === publication.connectionId && + afterRetirement.physicalActive + ) { + if (controller.snapshot().revision !== beforeRevision) { + this.broadcastActive(roomId, room); + } + this.scheduleHostOfflineCheck(roomId, check.hostPeerId); + this.requestPump(roomId); + return; + } this.broadcastActive(roomId, room); this.requestPump(roomId); } diff --git a/src/server/room-route-controller.ts b/src/server/room-route-controller.ts index 88984c1..8b4416a 100644 --- a/src/server/room-route-controller.ts +++ b/src/server/room-route-controller.ts @@ -355,6 +355,7 @@ export class RoomRouteController { private sfuPublisherQualityObservation?: SfuPublisherQualityObservation; private readonly qualityBaselinesPending = new Set(); private readonly directContinuations = new Map(); + private readonly retiringPublicationGenerations = new Set(); private sfuBootstrapIntent?: SfuBootstrapIntent; private readonly consumedSfuBootstrapOpportunities = new Map< string, @@ -546,6 +547,78 @@ export class RoomRouteController { return released; } + rebindHostIdentity( + peerId: string, + sessionId: string, + nowMs?: number, + ): readonly Resource[] { + const previousPeerId = this.options.hostPeerId; + if (peerId === previousPeerId) { + return this.upsertParticipant({ + peerId, + role: "host", + sessionId, + effectiveDownstreamCapacity: this.options.endpointMediaCopyCapacity, + }, nowMs); + } + if (this.participants.has(peerId)) { + throw new Error("Route Host identity collides with another participant"); + } + const host = this.participants.get(previousPeerId); + if (!host || host.role !== "host") { + throw new Error("Route Host identity is unavailable"); + } + + const released = new Set( + this.operation ? this.abortOperation(nowMs, "stale") : [], + ); + for (const edge of this.upstreamByViewer.values()) { + if (edge.kind === "sfu" && edge.physicalActive) { + released.add(edge.resource); + } + } + if (this.hostPublication?.physicalActive) { + released.add(this.hostPublication.resource); + } + this.upstreamByViewer.clear(); + this.hostPublication = null; + this.routeTimings.clear(); + this.qualityObservations.clear(); + this.senderQualityObservations.clear(); + this.senderQualityBaselines.clear(); + this.qualityBaselinesPending.clear(); + this.sfuPublisherQualityObservation = undefined; + this.directContinuations.clear(); + this.retiringPublicationGenerations.clear(); + this.sfuBootstrapIntent = undefined; + this.consumedSfuBootstrapOpportunities.clear(); + this.rootConvergenceRootPeerId = undefined; + for (const participant of this.participants.values()) { + participant.availabilityExhausted = false; + participant.bootstrapFailureReported = undefined; + participant.sfuFirstAtNextRoute = undefined; + participant.consumedCandidateOpportunities.clear(); + if ( + participant.role === "viewer" && + participant.sessionId !== null && + nowMs !== undefined + ) { + this.recordDemand(participant.peerId, nowMs, "edge-unavailable"); + } + } + + this.participants.delete(previousPeerId); + host.peerId = peerId; + host.sessionId = sessionId; + host.departureConfirmed = false; + this.participants.set(peerId, host); + this.options.hostPeerId = peerId; + this.revision = this.allocateRevision(); + this.touchFacts(); + this.assertGraph(); + return [...released]; + } + disconnectSession(peerId: string, sessionId: string): boolean { const participant = this.participants.get(peerId); if (!participant || participant.sessionId !== sessionId) return false; @@ -771,14 +844,20 @@ export class RoomRouteController { (edge.kind === "peer" && edge.parentSessionId !== guard.parentSessionId)) { return []; } - edge.physicalActive = false; - edge.usable = false; + const released: Resource[] = []; + if (edge.kind === "sfu") { + this.retireSfuEdge(guard.childPeerId, edge, released); + this.pruneRetiringSfuAnchors(); + } else { + edge.physicalActive = false; + edge.usable = false; + } this.clearQualityForParticipant(guard.childPeerId); child.availabilityExhausted = false; this.revision = this.allocateRevision(); if (this.operation) this.operation.baseRevision = this.revision; this.touchFacts(); - return edge.transport === "direct" ? [] : [edge.resource]; + return released; } retireHostPublication(guard: { @@ -793,13 +872,24 @@ export class RoomRouteController { publication.hostSessionId !== guard.hostSessionId || publication.generation !== guard.generation || publication.connectionId !== guard.connectionId) return []; + const released: Resource[] = []; + for (const [viewerPeerId, edge] of this.upstreamByViewer) { + if ( + edge.kind === "sfu" && + edge.publicationGeneration === publication.generation + ) { + this.retireSfuEdge(viewerPeerId, edge, released); + } + } publication.physicalActive = false; publication.usable = false; this.clearSfuQuality(); this.revision = this.allocateRevision(); if (this.operation) this.operation.baseRevision = this.revision; this.touchFacts(); - return [publication.resource]; + released.push(publication.resource); + this.pruneRetiringSfuAnchors(); + return released; } setPaused(paused: boolean, nowMs?: number): readonly Resource[] { @@ -1020,10 +1110,6 @@ export class RoomRouteController { }); } this.senderQualityObservations.delete(input.childPeerId); - this.senderQualityBaselines.set( - input.childPeerId, - this.senderQualityBaselines.get(input.childPeerId) ?? true, - ); const released = operation?.reason === "quality-convergence" && operation.childPeerId === input.childPeerId @@ -1968,7 +2054,7 @@ export class RoomRouteController { attempt.tuple.publication === "reuse" || (attempt.senderQualityConsecutiveHealthyWindows ?? 0) >= PERSISTENT_NATIVE_EDGE_DEGRADED_WINDOWS); - if (!candidateQualityReady && candidateQualityState === "unknown") { + if (!candidateQualityReady && candidateQualityState !== "degraded") { return { accepted: true, committed: false, @@ -2032,7 +2118,9 @@ export class RoomRouteController { ? [...this.upstreamByViewer] .filter( ([childPeerId, edge]) => - childPeerId !== operation.childPeerId && edge.kind === "sfu", + childPeerId !== operation.childPeerId && + edge.kind === "sfu" && + edge.physicalActive, ) .map(([childPeerId]) => childPeerId) : []; @@ -2167,6 +2255,7 @@ export class RoomRouteController { this.participants.clear(); this.routeTimings.clear(); this.directContinuations.clear(); + this.retiringPublicationGenerations.clear(); this.sfuBootstrapIntent = undefined; this.consumedSfuBootstrapOpportunities.clear(); this.rootConvergenceRootPeerId = undefined; @@ -2536,6 +2625,7 @@ export class RoomRouteController { this.senderQualityBaselines.set(operation.childPeerId, false); this.revision = attempt.revision; this.operation = undefined; + this.pruneRetiringSfuAnchors(); this.touchFacts(); if (createsHostRoot) { this.stageRootConvergence(operation.childPeerId); @@ -3595,8 +3685,11 @@ export class RoomRouteController { for (const participant of this.participants.values()) { if (participant.role !== "viewer" || !participant.departureConfirmed || this.childrenOf(participant.peerId).length > 0) continue; const edge = this.upstreamByViewer.get(participant.peerId); - if (edge && edge.transport !== "direct" && edge.physicalActive) released.push(edge.resource); - this.upstreamByViewer.delete(participant.peerId); + if (edge?.kind === "sfu") { + this.retireSfuEdge(participant.peerId, edge, released, false); + } else { + this.upstreamByViewer.delete(participant.peerId); + } this.clearQualityForParticipant(participant.peerId); this.participants.delete(participant.peerId); this.routeTimings.delete(participant.peerId); @@ -3614,6 +3707,7 @@ export class RoomRouteController { } this.revision = this.allocateRevision(); this.touchFacts(); + this.pruneRetiringSfuAnchors(); } return removed; } @@ -3695,9 +3789,13 @@ export class RoomRouteController { private retireInvalidOperationEdge(childPeerId: string, released: Resource[]): boolean { const edge = this.upstreamByViewer.get(childPeerId); if (!edge || !this.edgeRequiresMove(childPeerId, edge)) return false; - if (edge.kind === "sfu" && edge.physicalActive) released.push(edge.resource); - edge.physicalActive = false; - edge.usable = false; + if (edge.kind === "sfu") { + this.retireSfuEdge(childPeerId, edge, released); + this.pruneRetiringSfuAnchors(); + } else { + edge.physicalActive = false; + edge.usable = false; + } if (edge.kind === "sfu" && !this.hasSfuSubscribers() && this.hostPublication) { if (this.hostPublication.physicalActive) released.push(this.hostPublication.resource); this.hostPublication.physicalActive = false; @@ -4313,24 +4411,83 @@ export class RoomRouteController { const released: Resource[] = []; for (const [viewerPeerId, edge] of this.upstreamByViewer) { if (edge.kind !== "sfu" || edge.publicationGeneration !== generation) continue; - if (edge.physicalActive) released.push(edge.resource); - this.clearQualityForParticipant(viewerPeerId); - this.upstreamByViewer.delete(viewerPeerId); + this.retireSfuEdge(viewerPeerId, edge, released); } if (this.hostPublication?.generation === generation) { if (this.hostPublication.physicalActive) released.push(this.hostPublication.resource); this.hostPublication = null; } + this.pruneRetiringSfuAnchors(); return released; } + private retireSfuEdge( + viewerPeerId: string, + edge: Extract, { kind: "sfu" }>, + released: Resource[], + retainAnchor = this.childrenOf(viewerPeerId).length > 0, + ): void { + if (edge.physicalActive) released.push(edge.resource); + this.clearQualityForParticipant(viewerPeerId); + if (retainAnchor) { + edge.usable = false; + edge.physicalActive = false; + this.retiringPublicationGenerations.add(edge.publicationGeneration); + } else { + this.upstreamByViewer.delete(viewerPeerId); + } + } + + private pruneRetiringSfuAnchors(): void { + let removed = true; + while (removed) { + removed = false; + for (const [viewerPeerId, edge] of this.upstreamByViewer) { + if ( + edge.kind === "sfu" && + !edge.physicalActive && + this.retiringPublicationGenerations.has( + edge.publicationGeneration, + ) && + this.childrenOf(viewerPeerId).length === 0 + ) { + this.upstreamByViewer.delete(viewerPeerId); + removed = true; + } + } + } + for (const generation of this.retiringPublicationGenerations) { + if (![...this.upstreamByViewer.values()].some( + (edge) => + edge.kind === "sfu" && + !edge.physicalActive && + edge.publicationGeneration === generation, + )) { + this.retiringPublicationGenerations.delete(generation); + } + } + } + private assertGraph(): void { const host = this.participants.get(this.options.hostPeerId); if (!host || host.role !== "host") throw new Error("Route Host is missing"); for (const [child, edge] of this.upstreamByViewer) { this.assertViewer(child); if (edge.kind === "peer" && !this.participants.has(edge.parentPeerId)) throw new Error("Route parent is missing"); - if (edge.kind === "sfu" && this.hostPublication?.generation !== edge.publicationGeneration) throw new Error("SFU publication is stale"); + if ( + edge.kind === "sfu" && + edge.physicalActive && + this.hostPublication?.generation !== edge.publicationGeneration + ) { + throw new Error("SFU publication is stale"); + } + if ( + edge.kind === "sfu" && + !edge.physicalActive && + !this.retiringPublicationGenerations.has(edge.publicationGeneration) + ) { + throw new Error("Retiring SFU publication is unknown"); + } const seen = new Set(); let current = child; while (current !== this.options.hostPeerId) { @@ -4338,13 +4495,34 @@ export class RoomRouteController { seen.add(current); const currentEdge = this.upstreamByViewer.get(current); if (!currentEdge) throw new Error("Route is not source-reachable"); - if (currentEdge.kind === "sfu") break; + if (currentEdge.kind === "sfu") { + if ( + currentEdge.physicalActive + ? this.hostPublication?.generation !== currentEdge.publicationGeneration + : !this.retiringPublicationGenerations.has( + currentEdge.publicationGeneration, + ) + ) { + throw new Error("Route terminates at an unknown SFU publication"); + } + break; + } current = currentEdge.parentPeerId; } } for (const participant of this.participants.values()) if (this.usedSlots(participant.peerId) > this.options.endpointMediaCopyCapacity) { throw new Error("Committed route exceeds endpoint capacity"); } + for (const generation of this.retiringPublicationGenerations) { + if (![...this.upstreamByViewer.values()].some( + (edge) => + edge.kind === "sfu" && + !edge.physicalActive && + edge.publicationGeneration === generation, + )) { + throw new Error("Retiring SFU publication has no anchor"); + } + } } private assertViewer(peerId: string): void { diff --git a/tests/candidate-quality-probe.test.ts b/tests/candidate-quality-probe.test.ts index 85a58e4..24c2ebb 100644 --- a/tests/candidate-quality-probe.test.ts +++ b/tests/candidate-quality-probe.test.ts @@ -233,3 +233,107 @@ describe("SFU quality probe", () => { expect(probe.observe(metrics(17_000), metrics(17_100))).toBe("pending"); }); }); + +describe("zero-frame current route", () => { + const zeroFrames = (timestampMs: number) => + metrics(timestampMs, { intervalFramesDecoded: 0 }); + const degraded = (timestampMs: number) => + metrics(timestampMs, { + frameWidth: 320, + frameHeight: 180, + framesPerSecond: 10, + }); + + it("approves clean delivery on both candidate route types", () => { + for (const probe of [new P2pQualityProbe(), new SfuQualityProbe()]) { + expect(probe.observe(zeroFrames(1_000), metrics(1_100))).toBe("pending"); + expect(probe.observe(zeroFrames(3_000), metrics(3_100))).toBe("pending"); + expect(probe.observe(zeroFrames(5_000), metrics(5_100))).toBe("approved"); + } + }); + + it("continues a real improvement when the current route then stops decoding", () => { + const probe = new P2pQualityProbe(); + expect(probe.observe(degraded(1_000), metrics(1_100))).toBe("pending"); + expect(probe.observe(zeroFrames(3_000), metrics(3_100))).toBe("pending"); + expect(probe.observe(zeroFrames(5_000), metrics(5_100))).toBe("approved"); + }); + + it("never approves when both routes stop decoding", () => { + for (const probe of [new P2pQualityProbe(), new SfuQualityProbe()]) { + for (let index = 0; index < 15; index += 1) { + const timestampMs = 1_000 + index * 2_000; + expect( + probe.observe(zeroFrames(timestampMs), zeroFrames(timestampMs + 100)), + ).toBe("pending"); + } + } + }); + + it("rejects a candidate that decodes only with freezes", () => { + for (const probe of [new P2pQualityProbe(), new SfuQualityProbe()]) { + for (const timestampMs of [1_000, 3_000]) { + expect( + probe.observe( + zeroFrames(timestampMs), + metrics(timestampMs + 100, { intervalFreezeCount: 1 }), + ), + ).toBe("pending"); + } + expect( + probe.observe( + zeroFrames(5_000), + metrics(5_100, { intervalFreezeCount: 1 }), + ), + ).toBe("rejected"); + } + }); + + it("keeps incomplete and non-overlapping candidates unknown", () => { + const incomplete = new SfuQualityProbe(); + for (const timestampMs of [1_000, 3_000, 5_000]) { + expect( + incomplete.observe( + zeroFrames(timestampMs), + metrics(timestampMs + 100, { bitrateKbps: null }), + ), + ).toBe("pending"); + } + + const nonOverlapping = new P2pQualityProbe(); + for (const timestampMs of [1_000, 4_000, 7_000]) { + expect( + nonOverlapping.observe( + zeroFrames(timestampMs), + metrics(timestampMs + 2_100), + ), + ).toBe("pending"); + } + }); + + it("clears accumulated evidence when the current window is missing", () => { + const probe = new P2pQualityProbe(); + expect(probe.observe(degraded(1_000), metrics(1_100))).toBe("pending"); + expect(probe.observe(degraded(3_000), metrics(3_100))).toBe("pending"); + expect(probe.observe(null, metrics(5_100))).toBe("pending"); + expect(probe.observe(degraded(7_000), metrics(7_100))).toBe("pending"); + expect(probe.observe(degraded(9_000), metrics(9_100))).toBe("pending"); + expect(probe.observe(degraded(11_000), metrics(11_100))).toBe("approved"); + }); + + it("ignores a duplicate timestamp but clears a late candidate window", () => { + const probe = new P2pQualityProbe(); + expect(probe.observe(degraded(1_000), metrics(1_100))).toBe("pending"); + expect(probe.observe(degraded(3_000), metrics(3_100))).toBe("pending"); + expect(probe.observe(degraded(5_000), metrics(3_100))).toBe("pending"); + expect(probe.observe(degraded(7_000), metrics(7_100))).toBe("approved"); + + probe.reset(); + expect(probe.observe(degraded(9_000), metrics(9_100))).toBe("pending"); + expect(probe.observe(degraded(11_000), metrics(11_100))).toBe("pending"); + expect(probe.observe(degraded(13_000), metrics(10_100))).toBe("pending"); + expect(probe.observe(degraded(15_000), metrics(15_100))).toBe("pending"); + expect(probe.observe(degraded(17_000), metrics(17_100))).toBe("pending"); + expect(probe.observe(degraded(19_000), metrics(19_100))).toBe("approved"); + }); +}); diff --git a/tests/client-state.test.ts b/tests/client-state.test.ts index d0d6d57..e8ea68f 100644 --- a/tests/client-state.test.ts +++ b/tests/client-state.test.ts @@ -1208,9 +1208,9 @@ describe("client signaling recovery policy", () => { }); advance(2_000); - expect(sockets).toHaveLength(2); - expect(sockets[0]!.close).toHaveBeenCalledWith(4002, "signaling timeout"); - expect(statuses.at(-1)).toBe("reconnecting"); + expect(sockets).toHaveLength(1); + expect(sockets[0]!.close).not.toHaveBeenCalled(); + expect(statuses.at(-1)).toBe("connected"); expect(onTerminated).not.toHaveBeenCalled(); expect( sockets[0]!.send.mock.calls @@ -1218,27 +1218,18 @@ describe("client signaling recovery policy", () => { .includes("route-failed"), ).toBe(false); + advance(3_000); + const next = JSON.parse(String(sockets[0]!.send.mock.calls.at(-1)![0])); + expect(next).toEqual({ type: "signaling-challenge", sequence: 3 }); receive(sockets[0]!, { type: "signaling-challenge-response", - sequence: confirm.sequence, - }); - const oldClose = new Event("close"); - Object.defineProperties(oldClose, { - code: { value: 4002 }, - reason: { value: "signaling timeout" }, - }); - sockets[0]!.dispatchEvent(oldClose); - authenticate(sockets[1]!); - advance(5_000); - const healthy = JSON.parse(String(sockets[1]!.send.mock.calls.at(-1)![0])); - receive(sockets[1]!, { - type: "signaling-challenge-response", - sequence: healthy.sequence, + sequence: next.sequence, }); advance(2_000); - expect(sockets).toHaveLength(2); - expect(onMessage).toHaveBeenCalledTimes(2); + expect(sockets).toHaveLength(1); + expect(confirm.sequence).toBe(2); + expect(onMessage).toHaveBeenCalledTimes(1); signal.stop(); }); diff --git a/tests/host-peer-tracks.test.ts b/tests/host-peer-tracks.test.ts index 89a6dd0..248d29b 100644 --- a/tests/host-peer-tracks.test.ts +++ b/tests/host-peer-tracks.test.ts @@ -958,6 +958,38 @@ describe("HostPeer source replacement", () => { expect(videoSender.appliedMaxBitrates).toEqual([3_000_000, 8_000_000]); }); + it("replays the desired profile only after a connected peer reconnects", async () => { + const peer = createPeer( + createStream( + createTrack("video", "video"), + createTrack("audio", "audio"), + ), + ); + + await expect(peer.start()).resolves.toBe(true); + await acceptPeerAnswer(peer); + const connection = FakePeerConnection.latest!; + const videoSender = connection.senders[0]!; + const audioSender = connection.senders[1]!; + const videoCalls = videoSender.setParameters.mock.calls.length; + const audioCalls = audioSender.setParameters.mock.calls.length; + + connection.dispatchEvent(new Event("connectionstatechange")); + await Promise.resolve(); + expect(videoSender.setParameters).toHaveBeenCalledTimes(videoCalls); + expect(audioSender.setParameters).toHaveBeenCalledTimes(audioCalls); + + connection.connectionState = "disconnected"; + connection.dispatchEvent(new Event("connectionstatechange")); + connection.connectionState = "connected"; + connection.dispatchEvent(new Event("connectionstatechange")); + + await vi.waitFor(() => + expect(videoSender.setParameters).toHaveBeenCalledTimes(videoCalls + 1), + ); + expect(audioSender.setParameters).toHaveBeenCalledTimes(audioCalls + 1); + }); + it("continues queued profile updates after initial configuration rejects", async () => { const updates: PeerSnapshot[] = []; const peer = createPeer( diff --git a/tests/hybrid-media-router.test.ts b/tests/hybrid-media-router.test.ts index ea1f7f7..a7813a2 100644 --- a/tests/hybrid-media-router.test.ts +++ b/tests/hybrid-media-router.test.ts @@ -33,19 +33,21 @@ function connectHost( store: RoomStore, room: CreatedRoom, routePolicy?: RoutePolicy, + clientId = "host_client_12345678", + sessionId = "host_session_12345678", ) { const connected = store.connectParticipant({ roomId: room.roomId, role: "host", token: room.hostToken, - clientId: "host_client_12345678", - sessionId: "host_session_12345678", + clientId, + sessionId, }); return { roomId: room.roomId, role: "host" as const, peerId: connected.peerId, - sessionId: "host_session_12345678", + sessionId, ...(routePolicy ? { routePolicy } : {}), }; } @@ -120,12 +122,24 @@ function deferred(): { return { promise, resolve, reject }; } +function deferredValue(): { + promise: Promise; + resolve: (value: T) => void; +} { + let resolve!: (value: T) => void; + const promise = new Promise((settle) => { + resolve = settle; + }); + return { promise, resolve }; +} + function harness( endpointMediaCopyCapacity: 1 | 2 | 3, withSfu = false, prepareTimeoutMs?: number, sfuIngressCapacity = 2, drainRetryMs?: number, + hostOfflineCheckMs?: number, ) { const store = createStore(); const sent = new Map(); @@ -150,6 +164,7 @@ function harness( roomControl, ...(prepareTimeoutMs ? { prepareTimeoutMs } : {}), ...(drainRetryMs ? { drainRetryMs } : {}), + ...(hostOfflineCheckMs ? { hostOfflineCheckMs } : {}), tokenIssuer: { async issueToken({ peerId }) { if (nextTokenIssueError) { @@ -183,6 +198,7 @@ function harness( return { store, sent, + connections, admission, roomControl, onRoutesChanged, @@ -513,6 +529,70 @@ describe("HybridMediaRouter v9 runtime", () => { } }); + it("rebinds a replacement Host and clears offline Viewer media ownership", async () => { + const { store, sent, connections, router } = harness(2); + try { + const room = await store.createRoom(); + const previousHost = connectHost(store, room); + complete(router, previousHost); + const viewer = connectViewer(store, room, "host-rebind-offline-viewer"); + complete(router, viewer); + const prepared = await vi.waitFor(() => { + const message = preparedFor(sent, viewer.sessionId); + expect(message).toBeDefined(); + return message!; + }); + router.handleRouteReady(viewer, { + type: "route-ready", + revision: prepared.revision, + phase: "prepare", + }); + router.setViewerRelayCapacity(viewer, 1); + await vi.waitFor(() => + expect(connections.has(`${room.roomId}:${viewer.peerId}`)).toBe(true), + ); + + store.disconnectParticipant(room.roomId, viewer.peerId, viewer.sessionId); + router.disconnectParticipant(room.roomId, viewer.peerId, viewer.sessionId); + store.disconnectParticipant( + room.roomId, + previousHost.peerId, + previousHost.sessionId, + ); + router.disconnectParticipant( + room.roomId, + previousHost.peerId, + previousHost.sessionId, + ); + + const resumedHost = connectHost( + store, + room, + undefined, + "replacement-host-client", + "replacement-host-session", + ); + expect(resumedHost.peerId).not.toBe(previousHost.peerId); + complete(router, resumedHost); + expect(connections.has(`${room.roomId}:${viewer.peerId}`)).toBe(false); + + const reconnectedViewer = connectViewer( + store, + room, + "host-rebind-offline-viewer", + ); + expect(reconnectedViewer.peerId).toBe(viewer.peerId); + router.connectParticipant(reconnectedViewer); + expect( + router + .routeDiagnosticSnapshot(room.roomId) + .children.find((child) => child.effectiveCapacity === 1), + ).toBeDefined(); + } finally { + await router.close(); + } + }); + it("accepts active failure for a direct connection adopted during rebuild", async () => { const { store, sent, router } = harness(2); try { @@ -1704,6 +1784,206 @@ describe("HybridMediaRouter v9 runtime", () => { await router.close(); }); + it("discards an SFU preparation whose cursor becomes stale while awaiting LiveKit", async () => { + const { store, sent, admission, roomControl, router } = harness(1, true); + try { + const room = await store.createRoom(); + const host = connectHost(store, room); + complete(router, host); + const root = connectViewer(store, room, "prepare-stale-root"); + complete(router, root); + const direct = await vi.waitFor(() => { + const prepared = preparedFor(sent, root.sessionId); + expect(prepared?.candidate.transport).toBe("direct"); + return prepared!; + }); + router.handleRouteReady(root, { + type: "route-ready", + revision: direct.revision, + phase: "prepare", + }); + const gate = deferred(); + roomControl!.createBarrier = gate.promise; + router.setViewerRelayCapacity(root, 0); + const waiting = connectViewer(store, room, "prepare-stale-waiting"); + complete(router, waiting); + await vi.waitFor(() => + expect(admission?.usage().ingress).toBe(1), + ); + store.disconnectParticipant(room.roomId, root.peerId, root.sessionId); + router.disconnectParticipant(room.roomId, root.peerId, root.sessionId); + gate.resolve(); + await vi.waitFor(() => + expect(admission?.usage()).toEqual({ ingress: 0, egress: 0 }), + ); + expect(router.routeDiagnosticSnapshot(room.roomId).operation).toBeNull(); + } finally { + await router.close(); + } + }); + + it("reschedules the regular Host-offline check after a read error", async () => { + const { store, sent, admission, roomControl, router } = harness( + 1, + true, + undefined, + 2, + undefined, + 10, + ); + try { + const room = await store.createRoom(); + const { host } = await establishSfuRoom(store, sent, router, room); + const retry = deferredValue(); + const hostCheck = vi + .spyOn(roomControl!, "hostParticipantExists") + .mockRejectedValueOnce(new Error("temporary LiveKit read failure")) + .mockReturnValueOnce(retry.promise); + store.disconnectParticipant(room.roomId, host.peerId, host.sessionId); + router.disconnectParticipant(room.roomId, host.peerId, host.sessionId); + + await vi.waitFor(() => expect(hostCheck).toHaveBeenCalledTimes(2)); + expect(admission?.usage()).toEqual({ ingress: 1, egress: 2 }); + } finally { + await router.close(); + } + }); + + it("rechecks Host-offline retirement after a live route operation", async () => { + const { store, sent, admission, roomControl, router } = harness( + 1, + true, + undefined, + 2, + undefined, + 10, + ); + try { + const room = await store.createRoom(); + const { host, first, second } = await establishSfuRoom( + store, + sent, + router, + room, + ); + router.setViewerRelayCapacity(second, 1); + const active = router.resolveActiveViewerMediaEdge( + room.roomId, + first.peerId, + )!; + router.handleRouteFailed(first, { + type: "route-failed", + revision: active.revision, + phase: "active", + connectionId: active.connectionId, + }); + await vi.waitFor(() => { + const prepared = preparedFor(sent, first.sessionId); + expect(prepared?.revision).toBeGreaterThan(active.revision); + expect(prepared?.candidate.transport).toBe("direct"); + return prepared!; + }); + const retry = deferredValue(); + const hostCheck = vi + .spyOn(roomControl!, "hostParticipantExists") + .mockResolvedValueOnce(false) + .mockReturnValueOnce(retry.promise); + store.disconnectParticipant(room.roomId, host.peerId, host.sessionId); + router.disconnectParticipant(room.roomId, host.peerId, host.sessionId); + + await vi.waitFor(() => expect(hostCheck).toHaveBeenCalledTimes(2)); + expect( + router.resolveActiveViewerMediaEdge(room.roomId, first.peerId), + ).toBeUndefined(); + retry.resolve(false); + await vi.waitFor(() => + expect(admission?.usage()).toEqual({ ingress: 0, egress: 0 }), + ); + expect(hostCheck).toHaveBeenCalledTimes(2); + } finally { + await router.close(); + } + }); + + it("ignores an old Host-offline result after a new SFU generation commits", async () => { + const { store, sent, admission, roomControl, router } = harness( + 1, + true, + undefined, + 2, + undefined, + 10, + ); + try { + const room = await store.createRoom(); + const { host, first, second } = await establishSfuRoom( + store, + sent, + router, + room, + ); + const oldRevision = Math.max( + router.resolveActiveViewerMediaEdge(room.roomId, first.peerId)!.revision, + router.resolveActiveViewerMediaEdge(room.roomId, second.peerId)!.revision, + ); + const delayed = deferredValue(); + const hostCheck = vi + .spyOn(roomControl!, "hostParticipantExists") + .mockReturnValueOnce(delayed.promise); + store.disconnectParticipant(room.roomId, host.peerId, host.sessionId); + router.disconnectParticipant(room.roomId, host.peerId, host.sessionId); + await vi.waitFor(() => expect(hostCheck).toHaveBeenCalledTimes(1)); + + const replacementHost = connectHost( + store, + room, + undefined, + "new-physical-host-client", + "new-physical-host-session", + ); + complete(router, replacementHost); + const handled = new Set(); + for (let index = 0; index < 3; index += 1) { + const next = await vi.waitFor(() => { + const prepared = [first, second] + .map((viewer) => ({ + viewer, + prepared: preparedFor(sent, viewer.sessionId), + })) + .find( + ({ prepared }) => + prepared && + prepared.revision > oldRevision && + !handled.has(prepared.candidate.connectionId), + ); + expect(prepared).toBeDefined(); + return prepared!; + }); + handled.add(next.prepared!.candidate.connectionId); + router.handleRouteReady(next.viewer, { + type: "route-ready", + revision: next.prepared!.revision, + phase: "prepare", + }); + } + await vi.waitFor(() => + expect(admission?.usage()).toEqual({ ingress: 1, egress: 2 }), + ); + const currentFence = roomControl!.created.at(-1)!; + expect(currentFence.publicationGeneration).not.toBe( + roomControl!.created[0]!.publicationGeneration, + ); + + delayed.resolve(false); + await Promise.resolve(); + await Promise.resolve(); + expect(admission?.usage()).toEqual({ ingress: 1, egress: 2 }); + expect(roomControl!.deleted).not.toContainEqual(currentFence); + } finally { + await router.close(); + } + }); + it("resolves exact direct and peer-relayed Viewer evidence sources", async () => { const { store, sent, router } = harness(1); try { diff --git a/tests/media-assignment.test.ts b/tests/media-assignment.test.ts index 139bf63..449d7b6 100644 --- a/tests/media-assignment.test.ts +++ b/tests/media-assignment.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest"; import type { SignalPayload } from "../src/shared/protocol.ts"; import { + createOwnedViewerRestartSender, limitMediaAssignment, MAX_ENDPOINT_MEDIA_CHILDREN, reconcileBoundedMediaChildren, @@ -116,4 +117,41 @@ describe("peer-assisted client assignment", () => { rebuild: false, }); }); + + it("allows a pending peer restart closure only after exact promotion", () => { + let active = false; + const messages: unknown[] = []; + const sendRestart = createOwnedViewerRestartSender( + true, + (targetPeerId, connectionId) => + active && + targetPeerId === "parent_12345678" && + connectionId === "connection_12345678", + (message) => { + messages.push(message); + return true; + }, + ); + + expect( + sendRestart("parent_12345678", "connection_12345678", false), + ).toBe(false); + expect(messages).toEqual([]); + + active = true; + expect( + sendRestart("parent_12345678", "connection_12345678", true), + ).toBe(true); + expect(messages).toEqual([ + { + type: "restart-request", + targetPeerId: "parent_12345678", + connectionId: "connection_12345678", + rebuild: true, + }, + ]); + expect(sendRestart("other-parent", "connection_12345678", true)).toBe( + false, + ); + }); }); diff --git a/tests/room-route-controller.test.ts b/tests/room-route-controller.test.ts index d0fa9aa..e93f184 100644 --- a/tests/room-route-controller.test.ts +++ b/tests/room-route-controller.test.ts @@ -388,6 +388,60 @@ describe("RoomRouteController", () => { expect(routes.diagnosticParticipantLabel(C)).not.toBe(label); }); + it("clears an unknown sender run without rearming its healthy baseline", () => { + 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")); + for (const childPeerId of [A, B]) { + expect( + observeSenderState(routes, { + childPeerId, + connectionId: `${childPeerId === A ? "a" : "b"}_from_host`, + state: "healthy", + acceptedAtMs: 0, + }).accepted, + ).toBe(true); + } + expect( + observeSenderState(routes, { + childPeerId: A, + connectionId: "a_from_host", + state: "degraded", + acceptedAtMs: 1, + }).accepted, + ).toBe(true); + expect( + routes.observeSenderQualityEvidence({ + parentPeerId: HOST, + parentSessionId: "host_session", + childPeerId: A, + routeRevision: routes.snapshot().revision, + connectionId: "a_from_host", + senderIdentity: null, + sampleTimestampMs: null, + state: "unknown", + acceptedAtMs: 2, + }).accepted, + ).toBe(true); + + for (const acceptedAtMs of [3, 4, 5]) { + expect( + observeSenderState(routes, { + childPeerId: A, + connectionId: "a_from_host", + state: "degraded", + acceptedAtMs, + }).accepted, + ).toBe(true); + } + expect(routes.reconcile(6).operation).toMatchObject({ + childPeerId: A, + reason: "quality-convergence", + }); + }); + it("keeps isolated limitation windows diagnostic", () => { const routes = controller(2, { qualityConvergenceEnabled: true }); addViewer(routes, A, 2); @@ -1307,28 +1361,119 @@ describe("RoomRouteController", () => { hostSessionId: "host_session", }); const current = begin.operation!.current!; + const guard: CandidateGuard = { + childPeerId, + childSessionId: `${childPeerId}_session`, + revision: current.revision, + connectionId: "quality_sfu_candidate", + }; + expect( + routes.candidateReady(guard, 112), + ).toMatchObject({ accepted: true, committed: false }); + for (const acceptedAtMs of [113, 114]) { + expect( + routes.observeSfuPublisherQualityEvidence({ + hostPeerId: HOST, + hostSessionId: "host_session", + publicationGeneration: "publication_generation_12345678", + routeRevision: current.revision, + state: "healthy", + sampleTimestampMs: acceptedAtMs, + acceptedAtMs, + }), + ).toMatchObject({ accepted: true, committed: false }); + expect(routes.candidateReady(guard, acceptedAtMs)).toMatchObject({ + accepted: true, + committed: false, + }); + } + expect( + routes.observeSfuPublisherQualityEvidence({ + hostPeerId: HOST, + hostSessionId: "host_session", + publicationGeneration: "publication_generation_12345678", + routeRevision: current.revision, + state: "healthy", + sampleTimestampMs: 115, + acceptedAtMs: 115, + }).committed, + ).toBe(true); + expect(routes.snapshot().upstreamByViewer.get(childPeerId)).toMatchObject({ + kind: "sfu", + connectionId: "quality_sfu_candidate", + }); + }); + + it("rejects an SFU quality candidate on degraded publisher evidence", () => { + const routes = controller(2, { + sfuEnabled: true, + qualityConvergenceEnabled: true, + }); + addViewer(routes, A, 1); + addViewer(routes, B, 0); + routes.hydrateEdge(A, peerEdge(HOST, "a_from_host")); + routes.hydrateEdge(B, peerEdge(HOST, "b_from_host")); + observePersistentDegraded(routes, A, "a_from_host", 99); + expect(routes.reconcile(103).operation).toBeUndefined(); + observePersistentDegraded(routes, B, "b_from_host", 104); + const operation = routes.reconcile(108).operation!; + const childPeerId = operation.childPeerId; + const peerAttempt = beginCandidate(routes, { + nowMs: 109, + connectionId: "quality_peer_candidate", + reservation: { kind: "direct" }, + }).operation!; + expect( + routes.candidateFailed( + { + childPeerId, + childSessionId: `${childPeerId}_session`, + revision: peerAttempt.current!.revision, + connectionId: "quality_peer_candidate", + }, + 110, + ).accepted, + ).toBe(true); + const sfuAttempt = beginCandidate(routes, { + nowMs: 111, + connectionId: "quality_sfu_degraded", + reservation: { + kind: "sfu-create", + edge: "sfu-edge", + publication: "sfu-publication", + overlap: "host-overlap", + }, + publicationGeneration: "publication_generation_degraded", + publicationConnectionId: "publication_connection_degraded", + hostSessionId: "host_session", + }).operation!; + const current = sfuAttempt.current!; expect( routes.candidateReady( { childPeerId, childSessionId: `${childPeerId}_session`, revision: current.revision, - connectionId: "quality_sfu_candidate", + connectionId: current.connectionId, }, 112, ), ).toMatchObject({ accepted: true, committed: false }); expect( - observePersistentSfuHealthy( - routes, - "publication_generation_12345678", - current.revision, - 113, - ).committed, - ).toBe(true); + routes.observeSfuPublisherQualityEvidence({ + hostPeerId: HOST, + hostSessionId: "host_session", + publicationGeneration: "publication_generation_degraded", + routeRevision: current.revision, + state: "degraded", + sampleTimestampMs: 113, + acceptedAtMs: 113, + }), + ).toMatchObject({ accepted: true, committed: false }); + expect(routes.snapshot().operation).toBeUndefined(); expect(routes.snapshot().upstreamByViewer.get(childPeerId)).toMatchObject({ - kind: "sfu", - connectionId: "quality_sfu_candidate", + kind: "peer", + parentPeerId: HOST, }); }); @@ -3229,12 +3374,18 @@ describe("RoomRouteController", () => { }); - it("replaces one publication and makes every old subscriber waiting", () => { + it("keeps only old SFU roots that still anchor peer descendants", () => { const routes = controller(1, { sfuEnabled: true }); addViewer(routes, A, 0); - addViewer(routes, B, 0); + addViewer(routes, B, 1); + addViewer(routes, C, 0); + addViewer(routes, D, 0); routes.hydrateHostPublication("publication_1", "publication_resource_1"); - for (const [peerId, resource] of [[A, "subscription_a"], [B, "subscription_b"]] as const) { + for (const [peerId, resource] of [ + [A, "subscription_a"], + [B, "subscription_b"], + [D, "subscription_d"], + ] as const) { routes.hydrateEdge(peerId, { kind: "sfu", publicationGeneration: "publication_1", @@ -3245,6 +3396,7 @@ describe("RoomRouteController", () => { resource, }); } + routes.hydrateEdge(C, peerEdge(B, "c_from_b")); expect(routes.invalidateHostPublication({ hostSessionId: "host_session", routeRevision: 0, @@ -3276,17 +3428,27 @@ describe("RoomRouteController", () => { connectionId: "a_sfu_2", }, 22); expect(settled.accepted).toBe(true); - expect(settled.released).toHaveLength(4); expect(settled.released).toEqual(expect.arrayContaining([ "publication_overlap", "subscription_a", "subscription_b", + "subscription_d", "publication_resource_1", ])); expect(routes.snapshot().upstreamByViewer.get(A)).toMatchObject({ publicationGeneration: "publication_2", }); - expect(routes.snapshot().upstreamByViewer.has(B)).toBe(false); + expect(routes.snapshot().upstreamByViewer.get(B)).toMatchObject({ + kind: "sfu", + publicationGeneration: "publication_1", + usable: false, + physicalActive: false, + }); + expect(routes.snapshot().upstreamByViewer.get(C)).toMatchObject({ + kind: "peer", + parentPeerId: B, + }); + expect(routes.snapshot().upstreamByViewer.has(D)).toBe(false); expect(routes.routeDiagnosticSnapshot(22).children[1]).toMatchObject({ demandAgeMs: 12, finalRoute: "waiting", diff --git a/tests/route-transition.test.ts b/tests/route-transition.test.ts index 765569a..165d604 100644 --- a/tests/route-transition.test.ts +++ b/tests/route-transition.test.ts @@ -393,6 +393,144 @@ describe("minimal route transition contracts", () => { expect(publishers[1]?.disconnect).not.toHaveBeenCalled(); }); + it("reports and retires an exact active SFU source replacement failure", async () => { + const messages: ClientMessage[] = []; + const publishers: ReturnType[] = []; + const route = new HostSfuRoute({ + getStream: () => ({}) as MediaStream, + getProfile: () => QUALITY_PROFILES["720p30"], + getVideoCodec: () => "vp8", + reconcileChildren: () => undefined, + send: (message) => { + messages.push(message); + return true; + }, + createPublisher: () => { + const publisher = createFakePublisher([], "publisher-active"); + publishers.push(publisher); + return publisher; + }, + }); + + await route.acceptAndWait({ + revision: 1, + phase: "active", + assignment: hostAssignment("publication-active"), + }); + await route.acceptConfig(sfuConfig(1)); + publishers[0]!.replaceStream.mockResolvedValue(false); + + await expect(route.replaceStream({} as MediaStream)).resolves.toBe(false); + expect(publishers[0]!.disconnect).toHaveBeenCalledOnce(); + expect(messages).toContainEqual({ type: "refresh-sfu", revision: 1 }); + }); + + it("retires a failed pending source without invalidating healthy active SFU", async () => { + const messages: ClientMessage[] = []; + const publishers: ReturnType[] = []; + const route = new HostSfuRoute({ + getStream: () => ({}) as MediaStream, + getProfile: () => QUALITY_PROFILES["720p30"], + getVideoCodec: () => "vp8", + reconcileChildren: () => undefined, + send: (message) => { + messages.push(message); + return true; + }, + createPublisher: () => { + const publisher = createFakePublisher( + [], + `publisher-${publishers.length + 1}`, + ); + publishers.push(publisher); + return publisher; + }, + }); + + await route.acceptAndWait({ + revision: 1, + phase: "active", + assignment: hostAssignment("publication-active"), + }); + await route.acceptConfig(sfuConfig(1)); + route.accept({ + revision: 2, + phase: "prepare", + assignment: hostAssignment("publication-pending"), + candidate: candidate(2, "viewer_12345678", "sfu"), + }); + await route.acceptConfig(sfuConfig(2)); + publishers[1]!.replaceStream.mockResolvedValue(false); + + await expect(route.replaceStream({} as MediaStream)).resolves.toBe(true); + expect(publishers[0]!.disconnect).not.toHaveBeenCalled(); + expect(publishers[1]!.disconnect).toHaveBeenCalledOnce(); + expect(messages).toContainEqual({ + type: "route-failed", + revision: 2, + phase: "prepare", + connectionId: null, + }); + }); + + it("reports success when a pending SFU promotion supersedes an old failure", async () => { + const publishers: ReturnType[] = []; + const route = new HostSfuRoute({ + getStream: () => ({}) as MediaStream, + getProfile: () => QUALITY_PROFILES["720p30"], + getVideoCodec: () => "vp8", + reconcileChildren: () => undefined, + send: () => true, + createPublisher: () => { + const publisher = createFakePublisher( + [], + `publisher-${publishers.length + 1}`, + ); + publishers.push(publisher); + return publisher; + }, + }); + + await route.acceptAndWait({ + revision: 1, + phase: "active", + assignment: hostAssignment("publication-old"), + }); + await route.acceptConfig(sfuConfig(1)); + route.accept({ + revision: 2, + phase: "prepare", + assignment: hostAssignment("publication-new"), + candidate: candidate(2, "viewer_12345678", "sfu"), + }); + await route.acceptConfig(sfuConfig(2)); + + let resolveOldReplacement!: (replaced: boolean) => void; + publishers[0]!.replaceStream.mockImplementation( + () => + new Promise((resolve) => { + resolveOldReplacement = resolve; + }), + ); + const oldReplacementCalls = publishers[0]!.replaceStream.mock.calls.length; + const replacing = route.replaceStream({} as MediaStream); + await vi.waitFor(() => + expect(publishers[0]!.replaceStream).toHaveBeenCalledTimes( + oldReplacementCalls + 1, + ), + ); + + await route.acceptAndWait({ + revision: 2, + phase: "active", + assignment: hostAssignment("publication-new"), + }); + resolveOldReplacement(false); + + await expect(replacing).resolves.toBe(true); + expect(publishers[1]!.disconnect).not.toHaveBeenCalled(); + }); + it("reconciles paused Host active truth by exact SFU publication", async () => { @@ -510,6 +648,147 @@ describe("minimal route transition contracts", () => { expect(publishers[1]?.disconnect).toHaveBeenCalled(); }); + it("ignores a retired publisher failure after same-generation resync", async () => { + let releaseOldConnect!: () => void; + const oldConnectGate = new Promise((resolve) => { + releaseOldConnect = resolve; + }); + const messages: ClientMessage[] = []; + const publishers: ReturnType[] = []; + const route = new HostSfuRoute({ + getStream: () => ({}) as MediaStream, + getProfile: () => QUALITY_PROFILES["720p30"], + getVideoCodec: () => "vp8", + reconcileChildren: () => undefined, + send: (message) => { + messages.push(message); + return true; + }, + createPublisher: () => { + const publisher = createFakePublisher( + [], + `publisher-${publishers.length + 1}`, + ); + if (publishers.length === 1) { + publisher.connect.mockImplementation(async () => { + await oldConnectGate; + throw new Error("retired publisher"); + }); + } + publishers.push(publisher); + return publisher; + }, + }); + + await route.acceptAndWait({ + revision: 1, + phase: "active", + assignment: hostAssignment("publication-old"), + }); + await route.acceptConfig(sfuConfig(1)); + route.accept({ + revision: 2, + phase: "prepare", + assignment: hostAssignment("publication-new"), + candidate: candidate(2, "viewer_12345678", "sfu"), + }); + const stale = route.acceptConfig(sfuConfig(2)); + await vi.waitFor(() => expect(publishers[1]?.connect).toHaveBeenCalledOnce()); + + await route.resyncAuthoritative({ + revision: 1, + phase: "active", + assignment: hostAssignment("publication-old"), + }); + route.accept({ + revision: 2, + phase: "prepare", + assignment: hostAssignment("publication-new"), + candidate: candidate(2, "viewer_12345678", "sfu"), + }); + const current = route.acceptConfig(sfuConfig(2)); + await vi.waitFor(() => expect(publishers).toHaveLength(3)); + releaseOldConnect(); + await Promise.all([stale, current]); + + expect(messages.filter((message) => message.type === "route-failed")).toEqual( + [], + ); + expect(publishers[2]?.activate).toHaveBeenCalledOnce(); + }); + + it("ignores a retired subscriber failure after same-generation resync", async () => { + let releaseOldConnect!: () => void; + const oldConnectGate = new Promise((resolve) => { + releaseOldConnect = resolve; + }); + const messages: ClientMessage[] = []; + const subscribers: ReturnType[] = []; + const route = new ViewerSfuRoute("viewer_12345678", { + activatePeer: () => true, + reconcileSfuChildren: () => undefined, + onSfuStream: () => undefined, + send: (message) => { + messages.push(message); + return true; + }, + createSubscriber: (events) => { + const subscriber = createFakeSubscriber( + events, + [], + `subscriber-${subscribers.length + 1}`, + ); + if (subscribers.length === 1) { + subscriber.connect.mockImplementation(async () => { + await oldConnectGate; + throw new Error("retired subscriber"); + }); + } + subscribers.push(subscriber); + return subscriber; + }, + }); + + route.accept({ + revision: 1, + phase: "active", + assignment: viewerSfuAssignment([], "publication-old"), + }); + await route.acceptConfig(sfuConfig(1)); + route.accept({ + revision: 2, + phase: "prepare", + assignment: viewerSfuAssignment([], "publication-new"), + candidate: candidate(2, "viewer_12345678", "sfu"), + }); + const stale = route.acceptConfig(sfuConfig(2)); + await vi.waitFor(() => expect(subscribers[1]?.connect).toHaveBeenCalledOnce()); + + await route.resyncAuthoritative( + { + revision: 1, + phase: "active", + assignment: viewerSfuAssignment([], "publication-old"), + }, + "viewer_12345678", + ); + route.accept({ + revision: 2, + phase: "prepare", + assignment: viewerSfuAssignment([], "publication-new"), + candidate: candidate(2, "viewer_12345678", "sfu"), + }); + const current = route.acceptConfig(sfuConfig(2)); + await vi.waitFor(() => expect(subscribers).toHaveLength(3)); + releaseOldConnect(); + await Promise.all([stale, current]); + + expect(messages.filter((message) => message.type === "route-failed")).toEqual( + [], + ); + expect(subscribers[2]?.activate).toHaveBeenCalledOnce(); + }); + it("uses the subscriber first decoded frame as the only SFU prepare ready", async () => { const log: string[] = []; diff --git a/tests/sfu-client.test.ts b/tests/sfu-client.test.ts index e9590cc..0da72a8 100644 --- a/tests/sfu-client.test.ts +++ b/tests/sfu-client.test.ts @@ -720,6 +720,7 @@ describe("SfuPublisher", () => { const room = livekit.state.rooms[0]; expect(room.options).toEqual({ + disconnectOnPageLeave: false, dynacast: true, stopLocalTrackOnUnpublish: false, }); @@ -1656,6 +1657,73 @@ describe("SfuSubscriber", () => { expect(proofs).toEqual([8]); }); + it("retains pending frame proof through LiveKit full reconnect order", async () => { + vi.useFakeTimers(); + const proofs = vi.fn(() => true); + const subscriber = new SfuSubscriber({ + onStream: vi.fn(), + onFirstDecodedFrame: proofs, + }); + await subscriber.connect(connection); + const room = livekit.state.rooms[0]; + const oldHost = new livekit.FakeRemoteParticipant("host"); + const oldPublication = new livekit.FakeRemotePublication( + "host-video-before-reconnect", + Track.Source.ScreenShare, + ); + oldHost.add(oldPublication); + room.remoteParticipants.set("host", oldHost); + expect(subscriber.activate()).toBe(true); + subscriber.armDecodedFrameProof(); + + let releaseOldRead!: () => void; + const oldRead = new Promise((resolve) => { + releaseOldRead = resolve; + }); + room.emit( + RoomEvent.TrackSubscribed, + remoteTrack(track("video", "video-before-reconnect"), async () => { + await oldRead; + return statsReport([]); + }), + oldPublication, + oldHost, + ); + + room.remoteParticipants.delete("host"); + room.emit(RoomEvent.ParticipantDisconnected, oldHost); + room.emit(RoomEvent.Reconnecting); + + const newHost = new livekit.FakeRemoteParticipant("host"); + const newPublication = new livekit.FakeRemotePublication( + "host-video-after-reconnect", + Track.Source.ScreenShare, + ); + newHost.add(newPublication); + room.remoteParticipants.set("host", newHost); + room.emit(RoomEvent.Reconnected); + room.emit( + RoomEvent.TrackSubscribed, + remoteTrack(track("video", "video-after-reconnect"), async () => + statsReport([ + { + id: "video-after-reconnect", + type: "inbound-rtp", + timestamp: 1, + kind: "video", + framesDecoded: 1, + }, + ]), + ), + newPublication, + newHost, + ); + + await vi.waitFor(() => expect(proofs).toHaveBeenCalledOnce()); + releaseOldRead(); + expect(subscriber.deactivate()).toBe(true); + }); + it("reconciles a Host publication announced while connect is pending", async () => { const gate = deferred(); livekit.state.connectGate = gate.promise;