diff --git a/docs/todo.md b/docs/todo.md index 6183f15..df57cd6 100644 --- a/docs/todo.md +++ b/docs/todo.md @@ -1,6 +1,6 @@ # Current TODO Ledger -Last reviewed: 2026-08-30 +Last reviewed: 2026-08-31 Only **Now** is executable. Observations, old branches, experiments, and parked topics are not implementation authority. @@ -59,6 +59,11 @@ until their stated boundary is met. connection-scoped classifier with bounded false rejection. Do not hard-skip Peer candidates from a participant-wide NAT label or inject predicted ports into production ICE without a new accepted decision and a measured gate. +14. **Unresolved route-state ownership claims.** Reopen lower-revision + reauthentication, active SFU failure during an unrelated prepare, and + multi-child relay-evidence ownership only from an exact current-wire + reproduction. Do not add a second revision namespace, parallel failure + state, or generalized evidence map from static possibility alone. ## Decision Needed diff --git a/src/client/lib/signaling.ts b/src/client/lib/signaling.ts index 82afc29..d9ea38b 100644 --- a/src/client/lib/signaling.ts +++ b/src/client/lib/signaling.ts @@ -83,8 +83,6 @@ export class SignalingClient { private terminalTimer: number | null = null; private terminalMessage: ClientMessage | null = null; private socketGeneration = 0; - private hostOnline = false; - private authoritativeRoute = false; private challengeSequence = 0; private watchdogTimer: number | null = null; private watchdogDeadlineMs = 0; @@ -297,8 +295,16 @@ export class SignalingClient { } } this.events.onStatus("connected"); + this.refreshSignalingWatchdog(); + } + if ( + message.type === "host-status" && + this.identity.role === "host" && + message.online && + !message.paused + ) { + this.identity.sharingPaused = false; } - this.updateAuthoritativeActivity(message); this.events.onMessage(message); if (message.type === "authenticated") { this.reconcileHostQualityIntent(message); @@ -314,8 +320,6 @@ export class SignalingClient { } this.socket = null; this.authenticated = false; - this.hostOnline = false; - this.authoritativeRoute = false; this.clearSignalingWatchdog(); this.clearAuthenticationTimer(); if (!this.stopped && shouldReconnectSignaling(event.code)) { @@ -422,40 +426,6 @@ export class SignalingClient { } } - private updateAuthoritativeActivity(message: ServerMessage): void { - if (message.type === "authenticated") { - this.hostOnline = message.hostOnline; - this.authoritativeRoute = - this.identity.role === "host" || - !("mediaMode" in message) || - message.routeAssignment.upstream.kind !== "none"; - } else if (message.type === "host-status") { - this.hostOnline = message.online; - if ( - this.identity.role === "host" && - message.online && - !message.paused - ) { - this.identity.sharingPaused = false; - } - } else if ( - message.type === "route-update" && - message.phase === "active" && - this.identity.role === "viewer" - ) { - this.authoritativeRoute = message.assignment.upstream.kind !== "none"; - } else if ( - message.type === "sharing-stopped" || - message.type === "room-closed" - ) { - this.hostOnline = false; - this.authoritativeRoute = false; - } else { - return; - } - this.refreshSignalingWatchdog(); - } - private refreshSignalingWatchdog(): void { if (!this.isWatchdogEligible()) { this.clearSignalingWatchdog(); @@ -543,8 +513,6 @@ export class SignalingClient { } this.socket = null; this.authenticated = false; - this.hostOnline = false; - this.authoritativeRoute = false; this.clearAuthenticationTimer(); this.clearSignalingWatchdog(); this.events.onStatus("reconnecting"); @@ -597,8 +565,6 @@ export class SignalingClient { private isWatchdogEligible(): boolean { return ( this.authenticated && - this.hostOnline && - this.authoritativeRoute && this.isDocumentVisible() && this.socket?.readyState === WebSocket.OPEN ); diff --git a/src/client/media/host-sfu-route.ts b/src/client/media/host-sfu-route.ts index fd4176a..e064459 100644 --- a/src/client/media/host-sfu-route.ts +++ b/src/client/media/host-sfu-route.ts @@ -115,9 +115,6 @@ export class HostSfuRoute { } if (update.assignment.sfuPublicationGeneration === null) { this.clearPending(); - if (this.route.getMediaAssignment()?.sfuPublicationGeneration) { - this.events.reconcileChildren(update.assignment.childPeerIds); - } } return result; } diff --git a/src/client/media/route-transition.ts b/src/client/media/route-transition.ts index c786665..744b1ee 100644 --- a/src/client/media/route-transition.ts +++ b/src/client/media/route-transition.ts @@ -190,6 +190,7 @@ function sameAssignment( left: ParticipantRouteAssignment, right: ParticipantRouteAssignment, ): boolean { + const rightChildren = new Set(right.childPeerIds); return ( left.upstream.kind === right.upstream.kind && (left.upstream.kind !== "peer" || @@ -197,7 +198,7 @@ function sameAssignment( left.upstream.peerId === right.upstream.peerId)) && left.sfuPublicationGeneration === right.sfuPublicationGeneration && left.childPeerIds.length === right.childPeerIds.length && - left.childPeerIds.every((peerId, index) => peerId === right.childPeerIds[index]) + left.childPeerIds.every((peerId) => rightChildren.has(peerId)) ); } diff --git a/src/client/media/video-frame-proof.ts b/src/client/media/video-frame-proof.ts index 36f58d6..7c395ae 100644 --- a/src/client/media/video-frame-proof.ts +++ b/src/client/media/video-frame-proof.ts @@ -47,12 +47,11 @@ export function observeCompositedVideoFrame( return; } const currentDecodedFrames = decodedFrameCount(target); - if ( - (baselineDecodedFrames !== null && - currentDecodedFrames !== null && - currentDecodedFrames > baselineDecodedFrames) || - video.currentTime > baselineCurrentTime - ) { + const decodedFrameAdvanced = + baselineDecodedFrames !== null && currentDecodedFrames !== null + ? currentDecodedFrames > baselineDecodedFrames + : null; + if (decodedFrameAdvanced ?? (video.currentTime > baselineCurrentTime)) { finish(); } } diff --git a/src/client/media/viewer-presentation.ts b/src/client/media/viewer-presentation.ts index 438ce26..fb948d8 100644 --- a/src/client/media/viewer-presentation.ts +++ b/src/client/media/viewer-presentation.ts @@ -268,11 +268,7 @@ export function reduceViewerPresentation( connection: action.connection, }; case "media-bound": { - if ( - (state.media && action.generation <= state.media.generation) || - (state.routeStatus?.revision === action.revision && - state.routeStatus.state === "failed") - ) { + if (state.media && action.generation <= state.media.generation) { return state; } return { @@ -283,7 +279,7 @@ export function reduceViewerPresentation( proofEpoch: 0, framePresented: false, }, - retainedFrame: state.retainedFrame || hasCurrentFrame(state), + retainedFrame: false, autoplayBlockedGeneration: null, playbackFailedGeneration: null, }; diff --git a/src/client/media/viewer-sfu-route.ts b/src/client/media/viewer-sfu-route.ts index bd92963..48d5d55 100644 --- a/src/client/media/viewer-sfu-route.ts +++ b/src/client/media/viewer-sfu-route.ts @@ -37,7 +37,6 @@ interface ViewerSubscriberSlot { subscriber: ViewerSubscriberTransport; connected: boolean; activated: boolean; - mediaAvailable: boolean; stream: MediaStream | null; decodedFrame: boolean; readySent: boolean; @@ -126,9 +125,10 @@ export class ViewerSfuRoute { } const previousRevision = this.route.getRevision(); const previousMediaAssignment = this.route.getMediaAssignment(); - const sameCommittedMedia = - update.phase === "active" && - sameUpstream(previousMediaAssignment, update.assignment); + const sameCommittedMedia = sameUpstream( + previousMediaAssignment, + update.assignment, + ); const continuingRecovery = previousRevision !== update.revision && sameCommittedMedia ? this.manualReconnectRevision !== null @@ -243,7 +243,10 @@ export class ViewerSfuRoute { return result; } const transition = this.queueActiveRoute(token, acknowledge, this.paused); - if (continuingRecovery === "manual" && this.pending === null) { + if ( + this.manualReconnectRevision === update.revision && + this.pending === null + ) { void transition.then(() => { if ( this.closed || @@ -521,7 +524,6 @@ export class ViewerSfuRoute { subscriber, connected: false, activated: false, - mediaAvailable: false, stream: null, decodedFrame: false, readySent: false, @@ -771,7 +773,6 @@ export class ViewerSfuRoute { if (this.active === slot) { const assignment = this.route.getMediaAssignment(); if (assignment?.upstream.kind === "sfu") { - slot.mediaAvailable = true; this.events.onSfuStream(stream, assignment, false, slot.revision); } return; @@ -854,7 +855,6 @@ export class ViewerSfuRoute { if (manualReconnect) { this.manualReconnectRevision = null; } - slot.mediaAvailable = true; this.events.onSfuStream(slot.stream, assignment, true, slot.revision); if (previous && previous !== slot) { previous.failed = true; @@ -883,10 +883,7 @@ export class ViewerSfuRoute { ) { return; } - const changed = slot.mediaAvailable !== available; - slot.mediaAvailable = available; this.events.onSfuVideoAvailability?.(available, slot.revision); - if (!changed || this.paused) return; } private commitMedia(token: RouteOperationToken): void { @@ -1023,8 +1020,8 @@ export class ViewerSfuRoute { this.recovery = { revision, refreshed: false }; } if (!this.recovery.refreshed) { - this.recovery.refreshed = true; if (this.events.send({ type: "refresh-sfu", revision })) { + this.recovery.refreshed = true; return; } } diff --git a/src/client/pages/ViewerPage.tsx b/src/client/pages/ViewerPage.tsx index 4c95c11..ca2ef19 100644 --- a/src/client/pages/ViewerPage.tsx +++ b/src/client/pages/ViewerPage.tsx @@ -730,6 +730,17 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { !video.ended, ); } + function committedTransportConnected(): boolean { + if (currentRouteAssignment?.upstream.kind === "peer") { + return ( + peerRef.current?.isConnected() === true && + peerRef.current.isRecovering() === false + ); + } + return currentRouteAssignment?.upstream.kind === "sfu" + ? sfuTransportConnected + : false; + } function offerPeerQualityEvidence( snapshot: PeerSnapshot, peer: ViewerPeer, @@ -1172,7 +1183,8 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { } }, currentPeerMetrics: () => activePeerMetrics, - qualityProbeEligible: () => !pageSuspended, + qualityProbeEligible: () => + qualityPresentationEligible(committedTransportConnected()), onSfuState: (state, revision) => { if (active && viewerSfuRoute === route) { const connected = state === "connected"; @@ -1193,7 +1205,13 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { } }, onSfuVideoAvailability: (available, revision) => { - if (!active || viewerSfuRoute !== route || available) { + if (!active || viewerSfuRoute !== route) { + return; + } + if (available) { + setSfuUpstream((current) => + current ? { ...current, connectionState: "connected" } : current, + ); return; } invalidatePresentedMedia(); @@ -1408,7 +1426,9 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { dispatchPresentation({ type: "connection", revision: currentRouteRevision, - connection: connectionFact(snapshot.connectionState), + connection: peer.isRecovering() + ? "reconnecting" + : connectionFact(snapshot.connectionState), }); } }, @@ -1931,12 +1951,6 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { return; } if (message.type === "error") { - if (message.code === "PEER_NOT_FOUND" && !currentHostOnline) { - clearPeerState(); - clearHostPresence(false); - dispatchPresentation({ type: "host", host: "offline" }); - return; - } if ( [ "AUTH_REQUIRED", diff --git a/src/client/sfu/publisher.ts b/src/client/sfu/publisher.ts index e4ace66..2f08add 100644 --- a/src/client/sfu/publisher.ts +++ b/src/client/sfu/publisher.ts @@ -122,6 +122,8 @@ export class SfuPublisher { private statsTimer: ReturnType | null = null; private startupVideoProfilePending = false; private paused = false; + private replacementVideoTrack: MediaStreamTrack | null = null; + private replacementAudioTrack: MediaStreamTrack | null = null; constructor(private readonly events: PublisherEvents = {}) {} @@ -372,6 +374,9 @@ export class SfuPublisher { return false; } const nextAudioTrack = stream.getAudioTracks()[0] ?? null; + this.replacementVideoTrack = nextVideoTrack; + this.replacementAudioTrack = nextAudioTrack; + this.applyPausedState(nextVideoTrack, nextAudioTrack); const previousAudio = this.audio; const previousVideoTrack = previousVideo.rawTrack; let retainedNextVideoTrack = false; @@ -444,6 +449,7 @@ export class SfuPublisher { if (!videoConfiguration || !this.owns(room, generation)) { return false; } + this.applyPausedState(nextVideoTrack, nextAudioTrack); previousVideo.rawTrack = nextVideoTrack; previousVideo.sourceTrack = nextVideoSource; retainedNextVideoTrack = true; @@ -521,6 +527,10 @@ export class SfuPublisher { if (!retainedNextVideoTrack) { nextVideoTrack.stop(); } + if (this.replacementVideoTrack === nextVideoTrack) { + this.replacementVideoTrack = null; + this.replacementAudioTrack = null; + } } }); } @@ -774,12 +784,19 @@ export class SfuPublisher { setPaused(paused: boolean): void { this.paused = paused; - if (this.video) { - this.video.rawTrack.enabled = !paused; - } - if (this.audio) { - this.audio.rawTrack.enabled = !paused; - } + this.applyPausedState( + this.video?.rawTrack ?? null, + this.audio?.rawTrack ?? null, + ); + this.applyPausedState(this.replacementVideoTrack, this.replacementAudioTrack); + } + + private applyPausedState( + videoTrack: MediaStreamTrack | null, + audioTrack: MediaStreamTrack | null, + ): void { + if (videoTrack) videoTrack.enabled = !this.paused; + if (audioTrack) audioTrack.enabled = !this.paused; } private reapplyAudioAfterReconnect(room: Room, generation: number): void { diff --git a/src/client/webrtc/host-peer.ts b/src/client/webrtc/host-peer.ts index 9bf0449..8df790f 100644 --- a/src/client/webrtc/host-peer.ts +++ b/src/client/webrtc/host-peer.ts @@ -56,6 +56,8 @@ export class HostPeer { private readonly pendingCandidates: SignalCandidate[] = []; private statsAccumulator = createStatsAccumulator(); private senderVideoTrack: MediaStreamTrack | null; + private replacementVideoTrack: MediaStreamTrack | null = null; + private replacementAudioTrack: MediaStreamTrack | null = null; private videoSender: RTCRtpSender | null = null; private audioSender: RTCRtpSender | null = null; private statsTimer: number | null = null; @@ -79,6 +81,7 @@ export class HostPeer { private startupVideoProfilePending: boolean; private connectedOnce = false; private awaitingReconnect = false; + private paused = false; private snapshot: PeerSnapshot; constructor( @@ -93,6 +96,7 @@ export class HostPeer { ) { this.connectionId = connectionId; const sourceVideoTrack = stream.getVideoTracks()[0] ?? null; + this.paused = sourceVideoTrack?.enabled === false; this.senderVideoTrack = sourceVideoTrack ? cloneSenderVideoTrack(sourceVideoTrack) : null; @@ -177,6 +181,9 @@ export class HostPeer { const nextVideoTrack = cloneSenderVideoTrack(nextSourceVideoTrack); let retainedNextVideoTrack = false; const nextAudioTrack = nextStream.getAudioTracks()[0] ?? null; + this.replacementVideoTrack = nextVideoTrack; + this.replacementAudioTrack = nextAudioTrack; + this.applyPausedState(nextVideoTrack, nextAudioTrack); const previousAudioTrack = audioSender.track; this.statsSamplingBlocked = true; this.statsAccumulator = createStatsAccumulator(); @@ -200,6 +207,7 @@ export class HostPeer { if (this.disposed) { return false; } + this.applyPausedState(nextVideoTrack, nextAudioTrack); this.senderVideoTrack = nextVideoTrack; this.stream = nextStream; retainedNextVideoTrack = true; @@ -223,6 +231,10 @@ export class HostPeer { if (!retainedNextVideoTrack) { nextVideoTrack.stop(); } + if (this.replacementVideoTrack === nextVideoTrack) { + this.replacementVideoTrack = null; + this.replacementAudioTrack = null; + } this.statsAccumulator = createStatsAccumulator(); this.statsSamplingBlocked = false; } @@ -238,13 +250,17 @@ export class HostPeer { } setPaused(paused: boolean): void { - if (this.senderVideoTrack) { - this.senderVideoTrack.enabled = !paused; - } - const audioTrack = this.audioSender?.track; - if (audioTrack) { - audioTrack.enabled = !paused; - } + this.paused = paused; + this.applyPausedState(this.senderVideoTrack, this.audioSender?.track ?? null); + this.applyPausedState(this.replacementVideoTrack, this.replacementAudioTrack); + } + + private applyPausedState( + videoTrack: MediaStreamTrack | null, + audioTrack: MediaStreamTrack | null, + ): void { + if (videoTrack) videoTrack.enabled = !this.paused; + if (audioTrack) audioTrack.enabled = !this.paused; } private updateOwnedProfile( diff --git a/src/server/hybrid-media-router.ts b/src/server/hybrid-media-router.ts index 459815e..53ed6ab 100644 --- a/src/server/hybrid-media-router.ts +++ b/src/server/hybrid-media-router.ts @@ -1104,11 +1104,17 @@ export class HybridMediaRouter { private async pumpRoom(roomId: string, room: RoomRuntime): Promise { const controller = room.controller; if (!controller) return; + let broadcastRevision = controller.snapshot().revision; + const broadcastRevisionChange = (): void => { + const revision = controller.snapshot().revision; + if (revision === broadcastRevision) return; + this.broadcastActive(roomId, room); + broadcastRevision = revision; + }; for (;;) { - const beforeRevision = controller.snapshot().revision; const reconciled = controller.reconcile(this.now()); this.releaseResources(reconciled.released); - if (controller.snapshot().revision !== beforeRevision) this.broadcastActive(roomId, room); + broadcastRevisionChange(); this.sendRouteFailures( roomId, reconciled.failedPeerIds, @@ -1203,7 +1209,6 @@ export class HybridMediaRouter { let beginGuard = guard; let currentOperation = controller.snapshot().operation; if (plan.endpointTransition.kind === "bounded-gap") { - const beforeGap = controller.snapshot().revision; const gap = controller.retireCurrentCandidateProducer(guard, this.now()); this.releaseResources(gap.released); this.sendRouteFailures( @@ -1215,7 +1220,7 @@ export class HybridMediaRouter { this.releaseReservation(preparation.prepared.reservation); continue; } - if (controller.snapshot().revision !== beforeGap) this.broadcastActive(roomId, room); + broadcastRevisionChange(); currentOperation = controller.snapshot().operation; const currentPlan = currentOperation?.candidates[currentOperation.cursor]; if (!currentOperation || !currentPlan) { @@ -1313,17 +1318,18 @@ export class HybridMediaRouter { ...publication.resource.fence, viewerPeerId: operation.childPeerId, }; - if (!(await this.reserveSfuSubscription(fence))) { - return { kind: "denied", rejectionBucket: "sfu-admission" }; - } const currentEdge = snapshot.upstreamByViewer.get(operation.childPeerId); - const edge = + const borrowedEdge = currentEdge?.kind === "sfu" && currentEdge.publicationGeneration === publication.generation && currentEdge.resource.kind === "sfu-subscription" && !currentEdge.resource.released ? currentEdge.resource - : subscriptionResource(fence); + : null; + if (!borrowedEdge && !(await this.reserveSfuSubscription(fence))) { + return { kind: "denied", rejectionBucket: "sfu-admission" }; + } + const edge = borrowedEdge ?? subscriptionResource(fence); try { const token = await fallback.tokenIssuer.issueToken({ roomId, @@ -1340,6 +1346,7 @@ export class HybridMediaRouter { reservation: { kind: "sfu-reuse", edge, + ...(borrowedEdge ? { borrowed: true as const } : {}), ...(overlap ? { overlap } : {}), }, publicationGeneration: publication.generation, @@ -1352,7 +1359,7 @@ export class HybridMediaRouter { }, }; } catch { - this.releaseSubscription(fence); + if (!borrowedEdge) this.releaseResource(edge); return { kind: "denied", rejectionBucket: "candidate-failed" }; } } @@ -2273,7 +2280,12 @@ function reservationResources( reservation: CandidateReservation, ): RouteResource[] { const resources: RouteResource[] = []; - if ("edge" in reservation) resources.push(reservation.edge); + if ( + "edge" in reservation && + (reservation.kind !== "sfu-reuse" || !reservation.borrowed) + ) { + resources.push(reservation.edge); + } if ("publication" in reservation) resources.push(reservation.publication); if ("overlap" in reservation && reservation.overlap) { resources.push(reservation.overlap); diff --git a/src/server/room-route-controller.ts b/src/server/room-route-controller.ts index e1a80f3..5f2cf08 100644 --- a/src/server/room-route-controller.ts +++ b/src/server/room-route-controller.ts @@ -25,7 +25,7 @@ export type CandidateTuple = export type CandidateReservation = | { kind: "direct"; overlap?: Resource } - | { kind: "sfu-reuse"; edge: Resource; overlap?: Resource } + | { kind: "sfu-reuse"; edge: Resource; borrowed?: true; overlap?: Resource } | { kind: "sfu-create"; edge: Resource; publication: Resource; overlap?: Resource }; export type EndpointRetirement = @@ -1668,8 +1668,7 @@ export class RoomRouteController { ? "quality-convergence" : rootConvergence ? "root-convergence" - : (this.routeTimings.get(childPeerId)?.reason ?? - this.routeDemandReason(childPeerId)); + : this.routeDemandReason(childPeerId); const demandPeerId = bootstrap?.demandPeerId ?? childPeerId; const demand = this.participants.get(demandPeerId)!; this.ensureDemand(demandPeerId, nowMs, reason); @@ -1977,7 +1976,8 @@ export class RoomRouteController { transport: edge.transport, }; } - this.upstreamByViewer.delete(retirement.childPeerId); + edge.usable = false; + edge.physicalActive = false; } else { const publication = this.hostPublication; if (!publication || !publication.physicalActive || @@ -4860,7 +4860,12 @@ function endpointTransitionEquals(left: EndpointTransition, right: EndpointTrans function reservationResources(reservation: CandidateReservation): Resource[] { const resources: Resource[] = []; - if ("edge" in reservation) resources.push(reservation.edge); + if ( + "edge" in reservation && + (reservation.kind !== "sfu-reuse" || !reservation.borrowed) + ) { + resources.push(reservation.edge); + } if ("publication" in reservation) resources.push(reservation.publication); if ("overlap" in reservation && reservation.overlap !== undefined) resources.push(reservation.overlap); return resources; diff --git a/src/server/signaling.ts b/src/server/signaling.ts index b7c14bc..42811d6 100644 --- a/src/server/signaling.ts +++ b/src/server/signaling.ts @@ -243,6 +243,21 @@ export class SignalingServer { clearTimeout(timer); } this.viewerGraceTimers.clear(); + for (const state of this.socketStates.values()) { + clearTimeout(state.authenticationTimer); + } + const closed = new Promise((resolve) => { + this.webSocketServer.close(() => resolve()); + }); + for (const socket of this.webSocketServer.clients) { + socket.close(SIGNAL_CLOSE_CODES.serviceRestart, "Service restart"); + } + const forceCloseTimer = setTimeout(() => { + for (const socket of this.webSocketServer.clients) { + socket.terminate(); + } + }, SERVICE_RESTART_CLOSE_GRACE_MS); + forceCloseTimer.unref(); let routeCloseError: unknown; try { await this.hybridMediaRouter?.close(); @@ -256,18 +271,6 @@ export class SignalingServer { this.shareGenerationsByRoom.clear(); this.pausedShareGenerationsByRoom.clear(); this.ordinaryActiveHostChildrenByRoom.clear(); - const closed = new Promise((resolve) => { - this.webSocketServer.close(() => resolve()); - }); - for (const socket of this.webSocketServer.clients) { - socket.close(SIGNAL_CLOSE_CODES.serviceRestart, "Service restart"); - } - const forceCloseTimer = setTimeout(() => { - for (const socket of this.webSocketServer.clients) { - socket.terminate(); - } - }, SERVICE_RESTART_CLOSE_GRACE_MS); - forceCloseTimer.unref(); await closed; clearTimeout(forceCloseTimer); if (routeCloseError) { @@ -1140,6 +1143,7 @@ export class SignalingServer { } } else { const host = this.options.roomStore.getConnectedHost(roomId); + const promotedPeerIds = this.fillOrdinaryHostChildren(roomId); if (host) { for (const viewer of update.revokedViewers) { this.sendToSession(host.sessionId, { @@ -1147,6 +1151,12 @@ export class SignalingServer { peerId: viewer.peerId, }); } + for (const peerId of promotedPeerIds) { + this.sendToSession(host.sessionId, { + type: "peer-joined", + peerId, + }); + } } } @@ -1679,6 +1689,7 @@ export class SignalingServer { this.qualitySettingsByRoom.delete(roomId); this.routePolicyByRoom.delete(roomId); this.pausedShareGenerationsByRoom.delete(roomId); + this.ordinaryActiveHostChildrenByRoom.delete(roomId); this.clearRoomConnectionIds(roomId); if (this.isHybridMediaEnabled()) { this.hybridMediaRouter!.stopRoom(roomId); diff --git a/tests/client-state.test.ts b/tests/client-state.test.ts index 6cc4412..1ed120e 100644 --- a/tests/client-state.test.ts +++ b/tests/client-state.test.ts @@ -1337,7 +1337,13 @@ describe("client signaling recovery policy", () => { routePolicy: DEFAULT_ROUTE_POLICY, }); advance(5_000); - expect(sockets[0]!.send).toHaveBeenCalledTimes(1); + expect(sockets[0]!.send).toHaveBeenCalledTimes(2); + const waiting = JSON.parse(String(sockets[0]!.send.mock.calls.at(-1)![0])); + expect(waiting.type).toBe("signaling-challenge"); + receive({ + type: "signaling-challenge-response", + sequence: waiting.sequence, + }); receive({ type: "route-update", @@ -1357,7 +1363,7 @@ describe("client signaling recovery policy", () => { documentTarget.dispatchEvent(new Event("visibilitychange")); advance(20_000); expect(sockets).toHaveLength(1); - expect(sockets[0]!.send).toHaveBeenCalledTimes(2); + expect(sockets[0]!.send).toHaveBeenCalledTimes(3); visibilityState = "visible"; documentTarget.dispatchEvent(new Event("visibilitychange")); @@ -1369,9 +1375,9 @@ describe("client signaling recovery policy", () => { }); now += 7_000; vi.advanceTimersByTime(5_000); - expect(sockets[0]!.send).toHaveBeenCalledTimes(3); - advance(5_000); expect(sockets[0]!.send).toHaveBeenCalledTimes(4); + advance(5_000); + expect(sockets[0]!.send).toHaveBeenCalledTimes(5); expect(onMessage).toHaveBeenCalledTimes(2); signal.stop(); }); diff --git a/tests/host-peer-tracks.test.ts b/tests/host-peer-tracks.test.ts index d3511fa..110223f 100644 --- a/tests/host-peer-tracks.test.ts +++ b/tests/host-peer-tracks.test.ts @@ -796,6 +796,38 @@ describe("HostPeer source replacement", () => { expect(sourceVideo.stop).not.toHaveBeenCalled(); }); + it("applies pause authority to a source replacement while it is in flight", async () => { + const peer = createPeer( + createStream( + createTrack("video", "old-video"), + createTrack("audio", "old-audio"), + ), + ); + await expect(peer.start()).resolves.toBe(true); + const connection = FakePeerConnection.latest!; + const videoSender = connection.senders[0]!; + const nextVideo = createTrack("video", "next-video"); + const nextAudio = createTrack("audio", "next-audio"); + videoSender.deferReplaceCall = 1; + + const replacing = peer.replaceStream(createStream(nextVideo, nextAudio)); + await vi.waitFor(() => + expect(videoSender.replaceTrack).toHaveBeenCalledOnce(), + ); + const replacementVideo = videoSender.replaceTrack.mock.calls[0]![0]!; + peer.setPaused(true); + expect(replacementVideo.enabled).toBe(false); + expect(nextAudio.enabled).toBe(false); + + videoSender.releaseDeferredReplaceTrack(); + await expect(replacing).resolves.toBe(true); + expect(videoSender.track?.enabled).toBe(false); + expect(connection.senders[1]!.track?.enabled).toBe(false); + peer.setPaused(false); + expect(videoSender.track?.enabled).toBe(true); + expect(connection.senders[1]!.track?.enabled).toBe(true); + }); + it("fills a pre-negotiated audio sender that started without a track", async () => { const peer = createPeer( createStream(createTrack("video", "old-video"), null), diff --git a/tests/route-transition.test.ts b/tests/route-transition.test.ts index e166264..cd914a2 100644 --- a/tests/route-transition.test.ts +++ b/tests/route-transition.test.ts @@ -227,6 +227,29 @@ describe("minimal route transition contracts", () => { ).toBe("stale"); }); + it("treats child assignment order as presentation-only", () => { + const route = new MediaRouteTransition(); + route.accept({ + revision: 1, + phase: "active", + assignment: peerAssignment("parent", ["first", "second"]), + }); + route.accept({ + revision: 2, + phase: "prepare", + assignment: peerAssignment("parent", ["second", "first"]), + candidate: candidate(2, "second"), + }); + + expect( + route.accept({ + revision: 2, + phase: "active", + assignment: peerAssignment("parent", ["first", "second"]), + }), + ).toBe("accepted"); + }); + it("keeps duplicate peer prepare idempotent and releases it before takeover", async () => { const prepared: Array<{ parentPeerId: string | null; revision?: number }> = []; @@ -394,6 +417,40 @@ describe("minimal route transition contracts", () => { expect(publishers[1]?.disconnect).not.toHaveBeenCalled(); }); + it("changes Host children only after active authority commits", async () => { + const reconciledChildren: string[][] = []; + const route = new HostSfuRoute({ + getStream: () => ({}) as MediaStream, + getProfile: () => QUALITY_PROFILES["720p30"], + getVideoCodec: () => "vp8", + reconcileChildren: (children) => reconciledChildren.push(children), + send: () => true, + createPublisher: () => createFakePublisher([], "publisher"), + }); + await route.acceptAndWait({ + revision: 1, + phase: "active", + assignment: hostAssignment("publication", ["committed-child"]), + }); + await route.acceptConfig(sfuConfig(1)); + reconciledChildren.length = 0; + + route.accept({ + revision: 2, + phase: "prepare", + assignment: hostAssignment(null, ["candidate-child"]), + candidate: candidate(2, "candidate-child"), + }); + expect(reconciledChildren).toEqual([]); + + await route.acceptAndWait({ + revision: 2, + phase: "active", + assignment: hostAssignment(null, ["candidate-child"]), + }); + expect(reconciledChildren).toEqual([["candidate-child"]]); + }); + it("reports and retires an exact active SFU source replacement failure", async () => { const messages: ClientMessage[] = []; const publishers: ReturnType[] = []; @@ -1201,6 +1258,96 @@ describe("minimal route transition contracts", () => { await route.disconnect(); }); + it("retargets manual SFU recovery through an unrelated prepare", async () => { + const messages: ClientMessage[] = []; + const streams: MediaStream[] = []; + const subscribers: ReturnType[] = []; + const route = new ViewerSfuRoute("viewer_12345678", { + activatePeer: () => true, + reconcileSfuChildren: () => undefined, + onSfuStream: (stream) => streams.push(stream), + send: (message) => { + messages.push(message); + return true; + }, + createSubscriber: (events) => { + const subscriber = createFakeSubscriber(events, [], "active"); + subscribers.push(subscriber); + return subscriber; + }, + }); + const assignment = viewerSfuAssignment(["relay-child"]); + route.accept({ revision: 7, phase: "active", assignment }); + await route.acceptConfig(sfuConfig(7)); + subscribers[0]!.events.onStream({} as MediaStream); + subscribers[0]!.events.onFirstDecodedFrame(); + await vi.waitFor(() => expect(streams).toHaveLength(1)); + expect(route.reconnectActive()).toBe(true); + + route.accept({ + revision: 8, + phase: "prepare", + assignment, + candidate: candidate(8, "relay-child"), + }); + route.accept({ revision: 8, phase: "active", assignment }); + + await vi.waitFor(() => + expect( + messages.filter((message) => message.type === "refresh-sfu"), + ).toEqual([ + { type: "refresh-sfu", revision: 7 }, + { type: "refresh-sfu", revision: 8 }, + ]), + ); + }); + + it("does not spend SFU recovery until refresh signaling is sent", async () => { + const messages: ClientMessage[] = []; + const streams: MediaStream[] = []; + let acceptsRefresh = false; + const subscribers: ReturnType[] = []; + const route = new ViewerSfuRoute("viewer_12345678", { + activatePeer: () => true, + reconcileSfuChildren: () => undefined, + onSfuStream: (stream) => streams.push(stream), + send: (message) => { + messages.push(message); + return message.type !== "refresh-sfu" || acceptsRefresh; + }, + createSubscriber: (events) => { + const subscriber = createFakeSubscriber(events, [], "active"); + subscribers.push(subscriber); + return subscriber; + }, + }); + route.accept({ + revision: 7, + phase: "active", + assignment: viewerSfuAssignment(), + }); + await route.acceptConfig(sfuConfig(7)); + subscribers[0]!.events.onStream({} as MediaStream); + subscribers[0]!.events.onFirstDecodedFrame(); + await vi.waitFor(() => expect(streams).toHaveLength(1)); + subscribers[0]!.events.onDisconnected(); + acceptsRefresh = true; + + route.accept({ + revision: 8, + phase: "active", + assignment: viewerSfuAssignment(), + }); + await vi.waitFor(() => + expect( + messages.filter((message) => message.type === "refresh-sfu"), + ).toEqual([ + { type: "refresh-sfu", revision: 7 }, + { type: "refresh-sfu", revision: 8 }, + ]), + ); + }); + it("keeps an in-flight SFU recovery across an unrelated room revision", async () => { let releaseRecovery!: (connected: boolean) => void; const recoveryConnected = new Promise((resolve) => { diff --git a/tests/server-signal.test.ts b/tests/server-signal.test.ts index eada95d..3d836ee 100644 --- a/tests/server-signal.test.ts +++ b/tests/server-signal.test.ts @@ -1437,8 +1437,11 @@ describe("WebSocket signaling", () => { await host.inbox.expectNone(30); }); - it("rotates a grant without disturbing code-admitted Viewer media", async () => { - const harness = await startHarness(); + it("rotates a grant and promotes the waiting code-admitted Viewer", async () => { + const harness = await startHarness({ + endpointMediaCopyCapacity: 1, + maxViewersPerRoom: 2, + }); const host = await openClient(harness.webSocketUrl); const hostAuth = await authenticate( host, @@ -1476,10 +1479,10 @@ describe("WebSocket signaling", () => { undefined, { codeOnly: true, displayName: "房间号观众" }, ); - await host.inbox.next("peer-joined"); expect( viewerPresenceEntries(await host.inbox.next("viewer-presence")), ).toHaveLength(2); + expect((await host.inbox.next("peer-waiting")).peerId).toBe(codeAuth.peerId); const grantClosed = new Promise((resolve) => grantViewer.socket.once("close", (code) => resolve(code)), @@ -1488,6 +1491,7 @@ describe("WebSocket signaling", () => { await grantViewer.inbox.next("viewer-grant-revoked"); expect(await grantClosed).toBe(4004); expect((await host.inbox.next("peer-left")).peerId).toBe(grantAuth.peerId); + expect((await host.inbox.next("peer-joined")).peerId).toBe(codeAuth.peerId); const remaining = viewerPresenceEntries( await host.inbox.next("viewer-presence"), diff --git a/tests/sfu-client.test.ts b/tests/sfu-client.test.ts index 950b0ec..05e2b32 100644 --- a/tests/sfu-client.test.ts +++ b/tests/sfu-client.test.ts @@ -1616,6 +1616,44 @@ describe("SfuPublisher", () => { expect(previousPublishedVideo.stop).not.toHaveBeenCalled(); }); + it("applies pause authority to an SFU source replacement while it is in flight", async () => { + const publisher = new SfuPublisher(); + const previousAudio = track("audio", "audio-1"); + const nextVideo = track("video", "video-2"); + const nextAudio = track("audio", "audio-2"); + await publisher.connect(connection); + await publisher.activate( + stream(track("video", "video-1"), previousAudio), + qualityProfile, + ); + const room = livekit.state.rooms[0]; + const videoPublication = room.localParticipant.publications[0].track; + const replaceGate = deferred(); + videoPublication.replaceTrack.mockImplementationOnce( + async (replacement: MediaStreamTrack) => { + await replaceGate.promise; + videoPublication.currentTrack = replacement; + videoPublication.sender.track = replacement; + }, + ); + + const replacing = publisher.replaceStream(stream(nextVideo, nextAudio)); + await vi.waitFor(() => + expect(videoPublication.replaceTrack).toHaveBeenCalledOnce(), + ); + const replacementVideo = videoPublication.replaceTrack.mock.calls[0]![0]!; + publisher.setPaused(true); + expect(replacementVideo.enabled).toBe(false); + expect(nextAudio.enabled).toBe(false); + + replaceGate.resolve(); + await expect(replacing).resolves.toBe(true); + expect(videoPublication.currentTrack.enabled).toBe(false); + expect(room.localParticipant.publications[1].track.currentTrack.enabled).toBe( + false, + ); + }); + it("uses the screen audio preset when a replacement adds audio", async () => { const publisher = new SfuPublisher(); const audio = track("audio", "audio-2"); diff --git a/tests/video-frame-proof.test.ts b/tests/video-frame-proof.test.ts index a9a40b1..bb5d050 100644 --- a/tests/video-frame-proof.test.ts +++ b/tests/video-frame-proof.test.ts @@ -10,7 +10,7 @@ interface FakeVideo extends EventTarget { readyState: number; currentTime: number; decodedFrames: number; - getVideoPlaybackQuality: () => { totalVideoFrames: number }; + getVideoPlaybackQuality?: () => { totalVideoFrames: number }; requestVideoFrameCallback?: (callback: () => void) => number; cancelVideoFrameCallback?: (handle: number) => void; } @@ -108,4 +108,39 @@ describe("composited video frame proof", () => { expect(onFrame).toHaveBeenCalledOnce(); cancel(); }); + + it("does not substitute media time for an available decoded-frame count", () => { + const stream = {} as MediaStream; + const video = fakeVideo(stream); + const onFrame = vi.fn(); + const cancel = observeCompositedVideoFrame( + video as unknown as HTMLVideoElement, + stream, + onFrame, + ); + + video.readyState = 2; + video.currentTime = 1; + video.dispatchEvent(new Event("timeupdate")); + expect(onFrame).not.toHaveBeenCalled(); + cancel(); + }); + + it("uses media time only when decoded-frame count is unavailable", () => { + const stream = {} as MediaStream; + const video = fakeVideo(stream); + delete video.getVideoPlaybackQuality; + const onFrame = vi.fn(); + const cancel = observeCompositedVideoFrame( + video as unknown as HTMLVideoElement, + stream, + onFrame, + ); + + video.readyState = 2; + video.currentTime = 1; + video.dispatchEvent(new Event("timeupdate")); + expect(onFrame).toHaveBeenCalledOnce(); + cancel(); + }); }); diff --git a/tests/viewer-presentation.test.ts b/tests/viewer-presentation.test.ts index 56035ad..47f7a4a 100644 --- a/tests/viewer-presentation.test.ts +++ b/tests/viewer-presentation.test.ts @@ -640,13 +640,28 @@ describe("Viewer presentation reducer", () => { revision: 4, }); expect(lateFrame).toBe(failed); + const rebound = reduceViewerPresentation(failed, { + type: "media-bound", + generation: 4, + revision: 4, + }); + expect(rebound).toMatchObject({ + media: { generation: 4, framePresented: false }, + retainedFrame: false, + routeStatus: { revision: 4, state: "failed" }, + }); expect( - reduceViewerPresentation(failed, { - type: "media-bound", + reduceViewerPresentation(rebound, { + type: "frame-presented", generation: 4, + proofEpoch: 0, revision: 4, }), - ).toBe(failed); + ).toMatchObject({ + media: { generation: 4, framePresented: true }, + routeStatus: null, + connection: "connected", + }); const restored = reduceViewerPresentation(failed, { type: "route", revision: 4, @@ -663,18 +678,6 @@ describe("Viewer presentation reducer", () => { }).media?.framePresented, ).toBe(true); - const localFailure = reduceViewerPresentation(playing, { - type: "route-status", - revision: 4, - state: "failed", - }); - expect( - reduceViewerPresentation(localFailure, { - type: "media-bound", - generation: 4, - revision: 4, - }), - ).toBe(localFailure); }); it("honors terminal access, Host pause, and typed route status priority", () => {