From 30ffe44d60e4f693b7f5daf246690f6b8d761171 Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Sat, 22 Aug 2026 08:47:06 +0800 Subject: [PATCH] feat(routing): probe Host parents without breaking media (#183) --- src/client/media/host-provisional-child.ts | 264 +++++++++ src/client/media/host-sfu-route.ts | 8 +- src/client/pages/HostPage.tsx | 163 +++++- src/server/hybrid-media-router.ts | 154 ++++- tests/host-peer-tracks.test.ts | 258 ++++++++- tests/route-transition.test.ts | 195 ++++++- tests/server-signal.test.ts | 617 ++++++++++++++++++++- 7 files changed, 1624 insertions(+), 35 deletions(-) create mode 100644 src/client/media/host-provisional-child.ts diff --git a/src/client/media/host-provisional-child.ts b/src/client/media/host-provisional-child.ts new file mode 100644 index 0000000..6e6496e --- /dev/null +++ b/src/client/media/host-provisional-child.ts @@ -0,0 +1,264 @@ +import type { + IceConfig, + ParticipantRouteAssignment, + SignalPayload, +} from "../../shared/protocol"; +import { HostPeer } from "../webrtc/host-peer"; +import type { PeerSnapshot } from "../types"; +import type { QualityProfile } from "./quality"; + +export interface HostPreparedChildIdentity { + revision: number; + peerId: string; + failedConnectionId: string | null; +} + +interface HostProvisionalInput { + revision: number; + assignment: ParticipantRouteAssignment; + activeChildPeerIds: readonly string[]; + selectedPeerId: string | null; + maxMediaEdges: number; +} + +interface HostProvisionalPrepareInput extends HostProvisionalInput { + iceConfig: IceConfig; + stream: MediaStream; + profile: QualityProfile; +} + +interface HostProvisionalChildEvents { + sendSignal: (peerId: string, payload: SignalPayload) => boolean; + hasActivePeer?: (peerId: string) => boolean; + onPromotedStreamFailure?: (peer: HostPeer) => void; + onPromotedUpdate?: (peer: HostPeer, snapshot: PeerSnapshot) => void; +} + +export type HostPreparedChildResolution = + | { kind: "ordinary" } + | { kind: "promote"; peerId: string } + | { kind: "failed"; peerId: string; connectionId: string }; + +export type HostPreparedChildActivation = + | Exclude + | { kind: "promote"; peerId: string; peer: HostPeer }; + +interface PreparedHostChild extends HostPreparedChildIdentity { + peer: HostPeer; +} + +export class HostProvisionalChild { + private prepared: PreparedHostChild | null = null; + private promotedPeer: HostPeer | null = null; + private signalingPeer: HostPeer | null = null; + + constructor(private readonly events: HostProvisionalChildEvents) {} + + prepare(input: HostProvisionalPrepareInput): boolean { + const peerId = plannedHostProvisionalChild(input); + if ( + this.promotedPeer || + !peerId || + this.events.hasActivePeer?.(peerId) + ) { + this.discard(); + return false; + } + if ( + this.prepared?.revision === input.revision && + this.prepared.peerId === peerId + ) { + return this.prepared.failedConnectionId === null; + } + + this.discard(); + let peer: HostPeer; + peer = new HostPeer( + peerId, + input.iceConfig, + input.stream, + input.profile, + { + sendSignal: (targetPeerId, payload) => + this.signalingPeer === peer && targetPeerId === peerId + ? this.events.sendSignal(targetPeerId, payload) + : false, + onUpdate: (snapshot) => { + if (this.prepared?.peer === peer) { + if (snapshot.connectionState === "failed") { + this.fail(peer); + } + return; + } + if (this.promotedPeer === peer) { + this.events.onPromotedUpdate?.(peer, snapshot); + } + }, + }, + ); + this.signalingPeer = peer; + this.prepared = { + revision: input.revision, + peerId, + peer, + failedConnectionId: null, + }; + void peer + .start() + .catch(() => false) + .then((started) => { + if (!started && this.prepared?.peer === peer) { + this.fail(peer); + } + }); + return true; + } + + ownsSignal(fromPeerId: string, payload: SignalPayload): boolean { + const prepared = this.prepared; + return Boolean( + prepared && + prepared.failedConnectionId === null && + prepared.peerId === fromPeerId && + prepared.peer.connectionId === payload.connectionId, + ); + } + + acceptSignal(fromPeerId: string, payload: SignalPayload): boolean { + const prepared = this.prepared; + if (!this.ownsSignal(fromPeerId, payload) || !prepared) { + return false; + } + void prepared.peer.acceptSignal(payload); + return true; + } + + updateProfile(profile: QualityProfile): Promise { + const peer = this.livePeer(); + return peer ? peer.updateProfile(profile) : Promise.resolve(true); + } + + async replaceStream(stream: MediaStream): Promise { + const peer = this.livePeer(); + if (!peer) { + return true; + } + const replaced = await peer.replaceStream(stream).catch(() => false); + if (this.prepared?.peer !== peer) { + if (this.promotedPeer === peer && !replaced) { + this.events.onPromotedStreamFailure?.(peer); + } + return this.promotedPeer === peer ? replaced : true; + } + if (!replaced) { + this.fail(peer); + } + return replaced; + } + + activate(input: HostProvisionalInput): HostPreparedChildActivation { + const prepared = this.prepared; + const activation = resolveHostPreparedChildActivation({ + ...input, + prepared, + }); + if (activation.kind === "promote") { + if (prepared) { + this.prepared = null; + this.promotedPeer = prepared.peer; + return { ...activation, peer: prepared.peer }; + } + this.discard(); + return { kind: "ordinary" }; + } + if (activation.kind === "failed") { + this.prepared = null; + prepared?.peer.dispose(); + return activation; + } + this.discard(); + return activation; + } + + discard(): void { + const prepared = this.prepared; + this.prepared = null; + if (this.signalingPeer === prepared?.peer) { + this.signalingPeer = null; + } + prepared?.peer.dispose(); + } + + private livePeer(): HostPeer | null { + return this.prepared?.failedConnectionId === null + ? this.prepared.peer + : null; + } + + private fail(peer: HostPeer): void { + if (this.prepared?.peer !== peer) { + return; + } + this.prepared.failedConnectionId = peer.connectionId; + if (this.signalingPeer === peer) { + this.signalingPeer = null; + } + peer.dispose(); + } +} + +export function plannedHostProvisionalChild( + input: HostProvisionalInput, +): string | null { + const childPeerIds = [...new Set(input.assignment.childPeerIds)]; + const peerId = childPeerIds.find( + (candidate) => !input.activeChildPeerIds.includes(candidate), + ); + if ( + childPeerIds.length !== input.assignment.childPeerIds.length || + childPeerIds.length !== input.activeChildPeerIds.length + 1 || + input.activeChildPeerIds.some( + (candidate) => !childPeerIds.includes(candidate), + ) || + hostPhysicalMediaEdges(input.assignment, input.selectedPeerId) > + input.maxMediaEdges + ) { + return null; + } + return peerId ?? null; +} + +export function resolveHostPreparedChildActivation( + input: HostProvisionalInput & { + prepared: HostPreparedChildIdentity | null; + }, +): HostPreparedChildResolution { + const peerId = plannedHostProvisionalChild(input); + if ( + !peerId || + input.prepared?.revision !== input.revision || + input.prepared.peerId !== peerId + ) { + return { kind: "ordinary" }; + } + return input.prepared.failedConnectionId + ? { + kind: "failed", + peerId, + connectionId: input.prepared.failedConnectionId, + } + : { kind: "promote", peerId }; +} + +function hostPhysicalMediaEdges( + assignment: ParticipantRouteAssignment, + selectedPeerId: string | null, +): number { + const selectedOverlayEdges = + selectedPeerId && !assignment.childPeerIds.includes(selectedPeerId) ? 1 : 0; + return ( + assignment.childPeerIds.length + + (assignment.sfuPublicationGeneration ? 1 : 0) + + selectedOverlayEdges + ); +} diff --git a/src/client/media/host-sfu-route.ts b/src/client/media/host-sfu-route.ts index b499730..786d95c 100644 --- a/src/client/media/host-sfu-route.ts +++ b/src/client/media/host-sfu-route.ts @@ -302,14 +302,14 @@ export class HostSfuRoute { if ( !publicationGeneration || publicationGeneration !== message.publicationGeneration || - message.oldConnectionId !== publicationGeneration || - !this.lastConfig || - this.lastConfig.revision !== message.revision + message.oldConnectionId !== publicationGeneration ) { return false; } this.selectedEdgeTurn = message; - void this.acceptConfig(this.lastConfig); + if (this.lastConfig?.revision === message.revision) { + void this.acceptConfig(this.lastConfig); + } return true; } diff --git a/src/client/pages/HostPage.tsx b/src/client/pages/HostPage.tsx index ca131b5..43e20da 100644 --- a/src/client/pages/HostPage.tsx +++ b/src/client/pages/HostPage.tsx @@ -92,6 +92,9 @@ import { HostSfuRoute, type HostSfuPublisherSnapshot, } from "../media/host-sfu-route"; +import { + HostProvisionalChild, +} from "../media/host-provisional-child"; import { ParentEdgeQualityEvidenceReporter, } from "../media/parent-edge-quality-evidence"; @@ -126,12 +129,20 @@ type SelectedEdgeTurn = Extract< ServerMessage, { type: "selected-edge-turn"; edgeKind: "peer-selected" } >; +type HostRouteAssignment = Extract< + ServerMessage, + { type: "route-update" } +>["assignment"]; interface ActiveSelectedHostChild { peerId: string; connectionId: string; currentRouteRevision: number; pendingCarryRevision: number | null; } +interface FailedActiveHostChild { + revision: number; + peerId: string; +} interface CaptureDetails { resolution: string; @@ -269,6 +280,9 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { const peersRef = useRef(new Map()); const retiredConnectionsRef = useRef(new Map()); const selectedHostChildRef = useRef(null); + const hostProvisionalChildRef = useRef(null); + const failedActiveHostChildRef = useRef(null); + const activeHostChildPeerIdsRef = useRef([]); const hostPeerIdRef = useRef(null); const viewerQualityEvidenceRef = useRef( new Map(), @@ -405,6 +419,10 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { peersRef.current.clear(); retiredConnectionsRef.current.clear(); selectedHostChildRef.current = null; + hostProvisionalChildRef.current?.discard(); + hostProvisionalChildRef.current = null; + failedActiveHostChildRef.current = null; + activeHostChildPeerIdsRef.current = []; hostPeerIdRef.current = null; viewerQualityEvidenceTimersRef.current.forEach((timer) => window.clearTimeout(timer), @@ -518,6 +536,10 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { peersRef.current.clear(); retiredConnectionsRef.current.clear(); selectedHostChildRef.current = null; + hostProvisionalChildRef.current?.discard(); + hostProvisionalChildRef.current = null; + failedActiveHostChildRef.current = null; + activeHostChildPeerIdsRef.current = []; hostPeerIdRef.current = null; void hostSfuRouteRef.current?.disconnect(); hostSfuRouteRef.current = null; @@ -759,9 +781,14 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { const activeSfuRoute = hostSfuRouteRef.current; const [results, sfuUpdated] = await Promise.all([ Promise.all( - [...peersRef.current.values()].map((peer) => - peer.updateProfile(nextProfile), - ), + [ + ...[...peersRef.current.values()].map((peer) => + peer.updateProfile(nextProfile), + ), + ...(hostProvisionalChildRef.current + ? [hostProvisionalChildRef.current.updateProfile(nextProfile)] + : []), + ], ), activeSfuRoute?.updateProfile(nextProfile) ?? Promise.resolve(true), @@ -857,6 +884,99 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { }); } + function discardPreparedHostChild(): void { + const provisional = hostProvisionalChildRef.current; + hostProvisionalChildRef.current = null; + provisional?.discard(); + } + + function prepareHostChild( + revision: number, + assignment: HostRouteAssignment, + generation: number, + ): boolean { + const stream = streamRef.current; + const iceConfig = iceConfigRef.current; + const signal = signalRef.current; + if ( + !isCurrentGeneration(generation) || + !stream || + !iceConfig || + !signal + ) { + discardPreparedHostChild(); + return false; + } + hostProvisionalChildRef.current ??= new HostProvisionalChild({ + sendSignal: (targetPeerId, payload) => + isCurrentGeneration(generation) && signalRef.current === signal + ? signal.send({ type: "signal", targetPeerId, payload }) + : false, + hasActivePeer: (peerId) => peersRef.current.has(peerId), + onPromotedStreamFailure: (peer) => { + if ( + isCurrentGeneration(generation) && + peersRef.current.get(peer.peerId) === peer + ) { + removePeer(peer.peerId); + } + }, + onPromotedUpdate: (peer, snapshot) => { + if ( + isCurrentGeneration(generation) && + peersRef.current.get(peer.peerId) === peer + ) { + updatePeerSnapshot(snapshot); + } + }, + }); + return hostProvisionalChildRef.current.prepare({ + revision, + assignment, + activeChildPeerIds: activeHostChildPeerIdsRef.current, + selectedPeerId: selectedHostChildRef.current?.peerId ?? null, + maxMediaEdges: MAX_HOST_MEDIA_CHILDREN, + iceConfig, + stream, + profile: qualitySettingsRef.current, + }); + } + + function activatePreparedHostChild( + revision: number, + assignment: HostRouteAssignment, + ): void { + const activeChildPeerIds = activeHostChildPeerIdsRef.current; + const childPeerIds = [...new Set(assignment.childPeerIds)]; + const provisional = hostProvisionalChildRef.current; + const activation = provisional?.activate({ + revision, + assignment, + activeChildPeerIds, + selectedPeerId: selectedHostChildRef.current?.peerId ?? null, + maxMediaEdges: MAX_HOST_MEDIA_CHILDREN, + }) ?? { kind: "ordinary" as const }; + if (activation.kind === "promote") { + hostProvisionalChildRef.current = null; + failedActiveHostChildRef.current = null; + peersRef.current.set(activation.peerId, activation.peer); + updatePeerSnapshot(activation.peer.getSnapshot()); + } else if (activation.kind === "failed") { + hostProvisionalChildRef.current = null; + failedActiveHostChildRef.current = { + revision, + peerId: activation.peerId, + }; + return; + } else { + discardPreparedHostChild(); + if (failedActiveHostChildRef.current?.revision !== revision) { + failedActiveHostChildRef.current = null; + } + } + activeHostChildPeerIdsRef.current = childPeerIds; + } + async function startPeer( peerId: string, generation: number, @@ -1006,14 +1126,19 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { childPeerIds: string[], generation: number, ): void { + const failedPrepared = failedActiveHostChildRef.current; + const authoritativeChildPeerIds = + failedPrepared?.revision === activeRouteRevisionRef.current + ? childPeerIds.filter((peerId) => peerId !== failedPrepared.peerId) + : childPeerIds; const selected = selectedHostChildRef.current; const selectedIsCurrent = selected?.currentRouteRevision === activeRouteRevisionRef.current && peersRef.current.get(selected.peerId)?.connectionId === selected.connectionId; const effectiveChildPeerIds = selectedIsCurrent - ? [...new Set([...childPeerIds, selected.peerId])] - : childPeerIds; + ? [...new Set([...authoritativeChildPeerIds, selected.peerId])] + : authoritativeChildPeerIds; if (selected && !selectedIsCurrent) { selectedHostChildRef.current = null; } @@ -1041,6 +1166,8 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { return; } if (message.type === "authenticated" && message.role === "host") { + discardPreparedHostChild(); + failedActiveHostChildRef.current = null; const selected = selectedHostChildRef.current; if (selected) { removePeer(selected.peerId); @@ -1068,6 +1195,9 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { message.mediaMode === "peer-assisted" ) { activeRouteRevisionRef.current = message.routeRevision; + activeHostChildPeerIdsRef.current = [ + ...message.routeAssignment.childPeerIds, + ]; peerAssistedRef.current = true; signalRef.current?.send({ type: "set-quality-settings", @@ -1084,6 +1214,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { return; } activeRouteRevisionRef.current = 0; + activeHostChildPeerIdsRef.current = []; peerAssistedRef.current = false; clearHostSfuRoute(); const currentViewerIds = new Set(message.viewerPeerIds); @@ -1183,6 +1314,11 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { } if (message.type === "route-update") { if (peerAssistedRef.current) { + if (message.phase === "prepare") { + prepareHostChild(message.revision, message.assignment, generation); + } else { + activatePreparedHostChild(message.revision, message.assignment); + } if (message.phase === "active") { const selected = selectedHostChildRef.current; if (selected?.pendingCarryRevision === message.revision) { @@ -1220,6 +1356,11 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { } if (message.type === "media-assignment") { if (peerAssistedRef.current && !hostSfuRouteRef.current) { + discardPreparedHostChild(); + failedActiveHostChildRef.current = null; + activeHostChildPeerIdsRef.current = [ + ...message.mediaAssignment.childPeerIds, + ]; reconcilePeerAssistedChildren( message.mediaAssignment.childPeerIds, generation, @@ -1246,6 +1387,14 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { return; } if (message.type === "signal") { + if ( + hostProvisionalChildRef.current?.acceptSignal( + message.fromPeerId, + message.payload, + ) + ) { + return; + } void peersRef.current.get(message.fromPeerId)?.acceptSignal(message.payload); return; } @@ -1509,7 +1658,8 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { try { const activeSfuRoute = hostSfuRouteRef.current; - const [replacements, sfuReplaced] = await Promise.all([ + const provisional = hostProvisionalChildRef.current; + const [replacements, , sfuReplaced] = await Promise.all([ Promise.all( [...peersRef.current.entries()].map(async ([peerId, peer]) => { try { @@ -1523,6 +1673,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { } }), ), + provisional?.replaceStream(captured) ?? Promise.resolve(true), activeSfuRoute?.replaceStream(captured) ?? Promise.resolve(true), ]); diff --git a/src/server/hybrid-media-router.ts b/src/server/hybrid-media-router.ts index aeeeb90..f40705a 100644 --- a/src/server/hybrid-media-router.ts +++ b/src/server/hybrid-media-router.ts @@ -1228,6 +1228,12 @@ export class HybridMediaRouter { const ownsHealthyActiveSfu = healthyPending ? this.ownsActiveSfuForHealthyReselection(participant, healthyPending) : false; + const ownsQualityActiveSfu = healthyPending + ? this.ownsActiveOrdinarySfuForPeerQualityReselection( + participant, + healthyPending, + ) + : false; if ( message.phase === "active" && healthy?.activeRevision === message.revision && @@ -1261,8 +1267,31 @@ export class HybridMediaRouter { ) { this.abortPeerQualityReselectionForHardFailure(participant.roomId); this.activeSelectedSfuIngresses.delete(participant.roomId); + this.failBackToPeerBaseline(participant.roomId); return; } + if ( + message.phase === "prepare" && + message.connectionId === null && + healthyPending?.revision === message.revision && + ownsQualityActiveSfu + ) { + this.abortPeerQualityReselectionForHardFailure( + participant.roomId, + participant.role === "viewer" ? participant.peerId : undefined, + ); + const rollbackRevision = this.mediaRouteControllers + .get(participant.roomId) + ?.getActiveRoute().revision; + if (rollbackRevision === undefined) { + return; + } + message = { + ...message, + revision: rollbackRevision, + phase: "active", + }; + } if ( message.phase === "prepare" && healthyPending?.revision === message.revision && @@ -1893,6 +1922,11 @@ export class HybridMediaRouter { } if (currentPending) { this.abortPendingRoute(key); + } else if ( + currentIngress && + this.selectedSfuIngressAttemptIsCurrent(current) + ) { + this.failBackToPeerBaseline(key); } }, SELECTED_EDGE_TURN_ATTEMPT_TIMEOUT_MS); timer.unref(); @@ -1958,6 +1992,23 @@ export class HybridMediaRouter { }); } + private selectedSfuIngressAttemptIsCurrent( + attempt: SelectedSfuIngressAttempt, + ): boolean { + const active = this.mediaRouteControllers + .get(attempt.roomId) + ?.getActiveRoute(); + return ( + active?.revision === attempt.revision && + active.sfu.publicationGeneration === attempt.publicationGeneration && + active.assignments.get(attempt.hostPeerId)?.sfuPublicationGeneration === + attempt.publicationGeneration && + this.connectedPeer(attempt.roomId, attempt.hostPeerId)?.sessionId === + attempt.hostSessionId && + this.options.getShareGeneration(attempt.roomId) === attempt.shareGeneration + ); + } + private finishSelectedEdgeTurnFailure( participant: AuthenticatedRouteParticipant, message: Extract, @@ -2395,15 +2446,20 @@ export class HybridMediaRouter { } const candidateExclusions = new Set(excludedParentPeerIds); - candidateExclusions.add( - this.peerRelayTopology.getHostPeerId(roomId) ?? guard.parentPeerId, - ); + const hostPeerId = this.peerRelayTopology.getHostPeerId(roomId); for (const [peerId, assignment] of active.assignments) { + const isHost = peerId === hostPeerId; + const participant = isHost + ? this.options.roomStore.getConnectedHost(roomId) + : this.options.roomStore.getConnectedViewer(roomId, peerId); + const supportedParent = isHost + ? assignment.upstream.kind === "none" + : assignment.upstream.kind === "peer" || + assignment.upstream.kind === "sfu"; if ( - (assignment.upstream.kind !== "peer" && - assignment.upstream.kind !== "sfu") || - !this.options.roomStore.getConnectedViewer(roomId, peerId) || - assignment.childPeerIds.length + 1 > + !participant || + !supportedParent || + this.activeParentMediaEdges(roomId, peerId, assignment) + 1 > this.peerRelayTopology.getDownstreamCapacity(roomId, peerId) ) { candidateExclusions.add(peerId); @@ -2417,7 +2473,7 @@ export class HybridMediaRouter { MAX_PEER_RELAY_DEPTH, ); const newParent = newParentPeerId - ? this.options.roomStore.getConnectedViewer(roomId, newParentPeerId) + ? this.connectedPeer(roomId, newParentPeerId) : undefined; if (!newParentPeerId || !newParent) { return false; @@ -2996,14 +3052,24 @@ export class HybridMediaRouter { const parentAssignment = active?.assignments.get( probe?.newParentPeerId ?? "", ); + const hostPeerId = this.peerRelayTopology.getHostPeerId(roomId); + const supportedParent = + probe?.newParentPeerId === hostPeerId + ? parentAssignment?.upstream.kind === "none" + : parentAssignment?.upstream.kind === "peer" || + parentAssignment?.upstream.kind === "sfu"; return Boolean( probe && this.pendingRoutePreparations.get(roomId) === pending && active?.revision === probe.activeRevision && this.options.getShareGeneration(roomId) === probe.shareGeneration && - (parentAssignment?.upstream.kind === "peer" || - parentAssignment?.upstream.kind === "sfu") && - parentAssignment.childPeerIds.length + 1 <= + supportedParent && + parentAssignment && + this.activeParentMediaEdges( + roomId, + probe.newParentPeerId, + parentAssignment, + ) + 1 <= this.peerRelayTopology.getDownstreamCapacity( roomId, probe.newParentPeerId, @@ -3016,6 +3082,26 @@ export class HybridMediaRouter { ); } + private activeParentMediaEdges( + roomId: string, + parentPeerId: string, + assignment: ParticipantRouteAssignment, + ): number { + const selectedOverlayEdges = [...this.selectedPeerEdgeLeases.values()].some( + (lease) => + lease.roomId === roomId && + lease.parentPeerId === parentPeerId && + !assignment.childPeerIds.includes(lease.viewerPeerId), + ) + ? 1 + : 0; + return ( + assignment.childPeerIds.length + + (assignment.sfuPublicationGeneration ? 1 : 0) + + selectedOverlayEdges + ); + } + private ownsActiveSfuForHealthyReselection( participant: AuthenticatedRouteParticipant, pending: PendingRoutePreparation, @@ -3047,6 +3133,52 @@ export class HybridMediaRouter { ); } + private ownsActiveOrdinarySfuForPeerQualityReselection( + participant: AuthenticatedRouteParticipant, + pending: PendingRoutePreparation, + ): boolean { + const probe = pending.peerQualityReselection; + if ( + !probe || + participant.peerId !== probe.newParentPeerId || + pending.expectedSessionIds.get(participant.peerId) !== + participant.sessionId || + !this.peerQualityReselectionIsCurrent(participant.roomId, pending) + ) { + return false; + } + const active = this.mediaRouteControllers + .get(participant.roomId) + ?.getActiveRoute(); + const assignment = active?.assignments.get(participant.peerId); + if (!active?.sfu.publicationGeneration || !assignment) { + return false; + } + if (participant.role === "viewer") { + return ( + !this.selectedPeerEdgeLeases.has( + viewerConnectionKey(participant.roomId, participant.peerId), + ) && + assignment.upstream.kind === "sfu" && + active.sfu.rootPeerIds.includes(participant.peerId) + ); + } + const activeIngress = this.activeSelectedSfuIngresses.get( + participant.roomId, + ); + const hasCurrentSelectedIngress = + activeIngress?.hostPeerId === participant.peerId && + activeIngress.hostSessionId === participant.sessionId && + activeIngress.shareGeneration === probe.shareGeneration && + activeIngress.publicationGeneration === active.sfu.publicationGeneration; + return ( + !hasCurrentSelectedIngress && + participant.peerId === + this.peerRelayTopology.getHostPeerId(participant.roomId) && + assignment.sfuPublicationGeneration === active.sfu.publicationGeneration + ); + } + private abortPendingRoute( roomId: string, excludedPeerId?: string, diff --git a/tests/host-peer-tracks.test.ts b/tests/host-peer-tracks.test.ts index 4f7d0da..48e5d17 100644 --- a/tests/host-peer-tracks.test.ts +++ b/tests/host-peer-tracks.test.ts @@ -4,10 +4,14 @@ import { QUALITY_PROFILES, type QualityProfile, } from "../src/client/media/quality.ts"; +import { HostProvisionalChild } from "../src/client/media/host-provisional-child.ts"; import type { PeerSnapshot } from "../src/client/types.ts"; import { HostPeer } from "../src/client/webrtc/host-peer.ts"; import { ViewerRelay } from "../src/client/webrtc/viewer-relay.ts"; -import type { IceConfig } from "../src/shared/protocol.ts"; +import type { + IceConfig, + ParticipantRouteAssignment, +} from "../src/shared/protocol.ts"; const statsCallbacks: Array<() => void> = []; @@ -108,6 +112,8 @@ class FakePeerConnection { this.configurations.push(configuration); }); + readonly restartIce = vi.fn(); + addTransceiver( trackOrKind: MediaStreamTrack | string, init?: RTCRtpTransceiverInit, @@ -981,6 +987,256 @@ describe("HostPeer source replacement", () => { }); }); +function hostAssignment( + childPeerIds: string[], + publicationGeneration: string | null = null, +): ParticipantRouteAssignment { + return { + upstream: { kind: "none" }, + childPeerIds, + sfuPublicationGeneration: publicationGeneration, + }; +} + +function hostProvisionalInput( + revision: number, + childPeerIds: string[], + stream: MediaStream, + overrides: { + activeChildPeerIds?: string[]; + publicationGeneration?: string | null; + selectedPeerId?: string | null; + } = {}, +) { + return { + revision, + assignment: hostAssignment( + childPeerIds, + overrides.publicationGeneration ?? null, + ), + activeChildPeerIds: overrides.activeChildPeerIds ?? [], + selectedPeerId: overrides.selectedPeerId ?? null, + maxMediaEdges: 2, + iceConfig: { iceServers: [] }, + stream, + profile: QUALITY_PROFILES["720p30"], + }; +} + +describe("Host provisional child runtime ownership", () => { + it("accepts only the exact connection and promotes the same peer", async () => { + const signals: Array<{ peerId: string; connectionId: string }> = []; + const promotedUpdates = vi.fn(); + const owner = new HostProvisionalChild({ + sendSignal: (peerId, payload) => { + signals.push({ peerId, connectionId: payload.connectionId }); + return true; + }, + onPromotedUpdate: promotedUpdates, + }); + const stream = createStream(createTrack("video", "host-probe-video"), null); + const input = hostProvisionalInput(7, ["probe-child"], stream); + + expect(owner.prepare(input)).toBe(true); + await vi.waitFor(() => expect(signals).toHaveLength(1)); + const preparedConnection = FakePeerConnection.latest!; + const preparedConnectionId = signals[0]!.connectionId; + expect(FakePeerConnection.activeCount).toBe(1); + + expect(owner.acceptSignal("wrong-child", { + kind: "description", + connectionId: preparedConnectionId, + description: { type: "answer", sdp: "wrong-peer" }, + })).toBe(false); + expect(owner.acceptSignal("probe-child", { + kind: "description", + connectionId: "wrong-connection", + description: { type: "answer", sdp: "wrong-connection" }, + })).toBe(false); + expect(owner.acceptSignal("probe-child", { + kind: "candidate", + connectionId: preparedConnectionId, + candidate: { candidate: "host-probe-candidate" }, + })).toBe(true); + expect(owner.acceptSignal("probe-child", { + kind: "description", + connectionId: preparedConnectionId, + description: { type: "answer", sdp: "right-answer" }, + })).toBe(true); + await vi.waitFor(() => + expect(preparedConnection.remoteDescription?.sdp).toBe("right-answer"), + ); + expect(preparedConnection.addedIceCandidates).toEqual([ + { candidate: "host-probe-candidate" }, + ]); + + const activation = owner.activate(input); + expect(activation.kind).toBe("promote"); + if (activation.kind !== "promote") { + throw new Error("expected the prepared Host child to promote"); + } + expect(activation.peer.connectionId).toBe(preparedConnectionId); + expect(FakePeerConnection.latest).toBe(preparedConnection); + expect(FakePeerConnection.activeCount).toBe(1); + await expect(activation.peer.restartIce()).resolves.toBe(true); + expect(signals).toEqual([ + { peerId: "probe-child", connectionId: preparedConnectionId }, + { peerId: "probe-child", connectionId: preparedConnectionId }, + ]); + expect(promotedUpdates).toHaveBeenCalledWith( + activation.peer, + expect.objectContaining({ connectionId: preparedConnectionId }), + ); + activation.peer.dispose(); + }); + + it("disposes stale prepares on replacement, rollback, and auth reset", async () => { + const owner = new HostProvisionalChild({ sendSignal: () => true }); + const stream = createStream(createTrack("video", "host-cleanup-video"), null); + + expect(owner.prepare(hostProvisionalInput(7, ["first-probe"], stream))).toBe(true); + const first = FakePeerConnection.latest!; + expect(owner.prepare(hostProvisionalInput(8, ["second-probe"], stream))).toBe(true); + const second = FakePeerConnection.latest!; + expect(first.connectionState).toBe("closed"); + + expect(owner.activate({ + revision: 9, + assignment: hostAssignment([]), + activeChildPeerIds: [], + selectedPeerId: null, + maxMediaEdges: 2, + })).toEqual({ kind: "ordinary" }); + expect(second.connectionState).toBe("closed"); + + expect(owner.prepare(hostProvisionalInput(10, ["auth-probe"], stream))).toBe(true); + const authProbe = FakePeerConnection.latest!; + owner.discard(); + expect(authProbe.connectionState).toBe("closed"); + expect(FakePeerConnection.activeCount).toBe(0); + }); + + it("retains failed connection identity until authoritative activation", async () => { + const signals: Array<{ connectionId: string }> = []; + const owner = new HostProvisionalChild({ + sendSignal: (_peerId, payload) => { + signals.push({ connectionId: payload.connectionId }); + return false; + }, + }); + const stream = createStream(createTrack("video", "host-failed-video"), null); + const input = hostProvisionalInput(11, ["failed-probe"], stream); + + expect(owner.prepare(input)).toBe(true); + await vi.waitFor(() => expect(FakePeerConnection.activeCount).toBe(0)); + expect(signals).toHaveLength(1); + const instanceCount = FakePeerConnection.instances.length; + expect(owner.prepare(input)).toBe(false); + expect(FakePeerConnection.instances).toHaveLength(instanceCount); + + expect(owner.activate(input)).toEqual({ + kind: "failed", + peerId: "failed-probe", + connectionId: signals[0]!.connectionId, + }); + expect(FakePeerConnection.instances).toHaveLength(instanceCount); + }); + + it("replaces the prepared stream in place and tombstones a failed replacement", async () => { + const signals: Array<{ connectionId: string }> = []; + const owner = new HostProvisionalChild({ + sendSignal: (_peerId, payload) => { + signals.push({ connectionId: payload.connectionId }); + return true; + }, + }); + const initial = createStream(createTrack("video", "host-initial-video"), null); + const input = hostProvisionalInput(12, ["stream-probe"], initial); + expect(owner.prepare(input)).toBe(true); + await vi.waitFor(() => expect(signals).toHaveLength(1)); + const preparedConnection = FakePeerConnection.latest!; + const preparedConnectionId = signals[0]!.connectionId; + const instanceCount = FakePeerConnection.instances.length; + + const replacementVideo = createTrack("video", "host-replacement-video"); + await expect( + owner.replaceStream(createStream(replacementVideo, null)), + ).resolves.toBe(true); + expect(preparedConnection.senders[0]!.track).toBe(replacementVideo); + const activation = owner.activate(input); + expect(activation.kind).toBe("promote"); + if (activation.kind !== "promote") { + throw new Error("expected stream-updated Host probe to promote"); + } + expect(FakePeerConnection.instances).toHaveLength(instanceCount); + expect(FakePeerConnection.latest).toBe(preparedConnection); + expect(activation.peer.connectionId).toBe(preparedConnectionId); + activation.peer.dispose(); + + const failedSignals: Array<{ connectionId: string }> = []; + const failedOwner = new HostProvisionalChild({ + sendSignal: (_peerId, payload) => { + failedSignals.push({ connectionId: payload.connectionId }); + return true; + }, + }); + const failedInput = hostProvisionalInput( + 13, + ["failed-stream-probe"], + initial, + ); + expect(failedOwner.prepare(failedInput)).toBe(true); + await vi.waitFor(() => expect(failedSignals).toHaveLength(1)); + const failedConnection = FakePeerConnection.latest!; + const failedConnectionId = failedSignals[0]!.connectionId; + const failedInstanceCount = FakePeerConnection.instances.length; + failedConnection.senders[0]!.failNextReplace = true; + await expect( + failedOwner.replaceStream( + createStream(createTrack("video", "host-failed-replacement"), null), + ), + ).resolves.toBe(false); + expect(failedConnection.connectionState).toBe("closed"); + const failedActivation = failedOwner.activate(failedInput); + expect(failedActivation).toMatchObject({ + kind: "failed", + peerId: "failed-stream-probe", + connectionId: failedConnectionId, + }); + expect(FakePeerConnection.instances).toHaveLength(failedInstanceCount); + + const promotedFailure = vi.fn(); + const racingSignals: string[] = []; + const racingOwner = new HostProvisionalChild({ + sendSignal: (_peerId, payload) => { + racingSignals.push(payload.connectionId); + return true; + }, + onPromotedStreamFailure: promotedFailure, + }); + const racingInput = hostProvisionalInput( + 14, + ["racing-stream-probe"], + initial, + ); + expect(racingOwner.prepare(racingInput)).toBe(true); + await vi.waitFor(() => expect(racingSignals).toHaveLength(1)); + const racingConnection = FakePeerConnection.latest!; + racingConnection.senders[0]!.failNextReplace = true; + const racingReplacement = racingOwner.replaceStream( + createStream(createTrack("video", "host-racing-replacement"), null), + ); + const racingActivation = racingOwner.activate(racingInput); + expect(racingActivation.kind).toBe("promote"); + if (racingActivation.kind !== "promote") { + throw new Error("expected racing Host probe to promote"); + } + await expect(racingReplacement).resolves.toBe(false); + expect(promotedFailure).toHaveBeenCalledWith(racingActivation.peer); + racingActivation.peer.dispose(); + }); +}); + describe("ViewerRelay downstream ownership", () => { it("promotes the exact prepared child connection within the current Viewer cap", async () => { const signals: Array<{ peerId: string; connectionId: string }> = []; diff --git a/tests/route-transition.test.ts b/tests/route-transition.test.ts index 264de2a..f875b9e 100644 --- a/tests/route-transition.test.ts +++ b/tests/route-transition.test.ts @@ -9,6 +9,10 @@ import { reportActivePeerRouteFailure, } from "../src/client/media/route-transition.ts"; import { HostSfuRoute } from "../src/client/media/host-sfu-route.ts"; +import { + plannedHostProvisionalChild, + resolveHostPreparedChildActivation, +} from "../src/client/media/host-provisional-child.ts"; import { ViewerSfuRoute } from "../src/client/media/viewer-sfu-route.ts"; import { QUALITY_PROFILES } from "../src/client/media/quality.ts"; import { @@ -49,11 +53,14 @@ function sfuConfig(revision: number) { }; } -function hostSfuIngressGrant(newConnectionId = "selected-connection-new") { +function hostSfuIngressGrant( + newConnectionId = "selected-connection-new", + revision = 1, +) { return { type: "selected-edge-turn" as const, edgeKind: "host-sfu-ingress" as const, - revision: 1, + revision, hostPeerId: "host_12345678", publicationGeneration: "generation-a", oldConnectionId: "generation-a", @@ -89,6 +96,116 @@ function createFakePublisher(log: string[], label: string) { }; } +describe("Host provisional child ownership", () => { + it("admits only one strict child within the physical Host edge budget", () => { + expect( + plannedHostProvisionalChild({ + revision: 7, + assignment: hostAssignment(null, ["active-child", "probe-child"]), + activeChildPeerIds: ["active-child"], + selectedPeerId: null, + maxMediaEdges: 2, + }), + ).toBe("probe-child"); + expect( + plannedHostProvisionalChild({ + revision: 7, + assignment: hostAssignment("publication", ["probe-child"]), + activeChildPeerIds: [], + selectedPeerId: null, + maxMediaEdges: 2, + }), + ).toBe("probe-child"); + expect( + plannedHostProvisionalChild({ + revision: 7, + assignment: hostAssignment(null, ["probe-child"]), + activeChildPeerIds: [], + selectedPeerId: "selected-overlay", + maxMediaEdges: 2, + }), + ).toBe("probe-child"); + expect( + plannedHostProvisionalChild({ + revision: 7, + assignment: hostAssignment("publication", [ + "active-child", + "probe-child", + ]), + activeChildPeerIds: ["active-child"], + selectedPeerId: null, + maxMediaEdges: 2, + }), + ).toBeNull(); + expect( + plannedHostProvisionalChild({ + revision: 7, + assignment: hostAssignment(null, ["active-child", "probe-child"]), + activeChildPeerIds: ["active-child"], + selectedPeerId: "selected-overlay", + maxMediaEdges: 2, + }), + ).toBeNull(); + }); + + it("promotes the exact identity and preserves a failed connection tombstone", () => { + const input = { + revision: 7, + assignment: hostAssignment(null, ["active-child", "probe-child"]), + activeChildPeerIds: ["active-child"], + selectedPeerId: null, + maxMediaEdges: 2, + }; + expect( + resolveHostPreparedChildActivation({ + ...input, + prepared: { + revision: 7, + peerId: "probe-child", + failedConnectionId: null, + }, + }), + ).toEqual({ kind: "promote", peerId: "probe-child" }); + expect( + resolveHostPreparedChildActivation({ + ...input, + prepared: { + revision: 7, + peerId: "probe-child", + failedConnectionId: "failed-probe-connection", + }, + }), + ).toEqual({ + kind: "failed", + peerId: "probe-child", + connectionId: "failed-probe-connection", + }); + expect( + resolveHostPreparedChildActivation({ ...input, prepared: null }), + ).toEqual({ kind: "ordinary" }); + expect( + resolveHostPreparedChildActivation({ + ...input, + prepared: { + revision: 6, + peerId: "probe-child", + failedConnectionId: null, + }, + }), + ).toEqual({ kind: "ordinary" }); + expect( + resolveHostPreparedChildActivation({ + ...input, + prepared: { + revision: 7, + peerId: "other-child", + failedConnectionId: null, + }, + }), + ).toEqual({ kind: "ordinary" }); + }); +}); + interface FakeSubscriberEvents { onStream: (stream: MediaStream | null) => void; onVideoAvailability?: (available: boolean) => void; @@ -402,6 +519,80 @@ describe("HostSfuRoute", () => { }]); }); + it("accepts a forward-revision selected ingress before its config arrives", async () => { + const messages: ClientMessage[] = []; + const publishers: ReturnType[] = []; + const failures: Array<() => void> = []; + const route = new HostSfuRoute({ + getStream: () => ({}) as MediaStream, + getProfile: () => QUALITY_PROFILES["720p30"], + reconcileChildren: () => undefined, + send: (message) => { + messages.push(message); + return true; + }, + createPublisher: (onDisconnected) => { + const publisher = createFakePublisher( + [], + `publisher-${publishers.length + 1}`, + ); + publishers.push(publisher); + failures.push(onDisconnected); + return publisher; + }, + }); + const active = hostAssignment("generation-a"); + route.accept({ revision: 1, phase: "prepare", assignment: active }); + await route.acceptConfig(sfuConfig(1)); + await route.acceptAndWait({ + revision: 1, + phase: "active", + assignment: active, + }); + + messages.length = 0; + route.accept({ + revision: 2, + phase: "prepare", + assignment: hostAssignment("generation-a", ["viewer-root"]), + }); + failures[0](); + expect(messages).toContainEqual({ + type: "route-failed", + revision: 2, + phase: "prepare", + connectionId: null, + }); + + await route.acceptAndWait({ + revision: 3, + phase: "active", + assignment: active, + }); + expect(messages).toContainEqual({ type: "refresh-sfu", revision: 3 }); + expect( + route.startSelectedEdgeTurn( + hostSfuIngressGrant("selected-forward-connection", 3), + ), + ).toBe(true); + await route.acceptConfig(sfuConfig(3)); + + await vi.waitFor(() => expect(publishers).toHaveLength(2)); + expect(publishers[1]?.connect).toHaveBeenCalledWith({ + url: "wss://sfu.example.test", + token: "token-3", + rtcConfig: { + iceServers: [hostSfuIngressGrant().iceServer], + iceTransportPolicy: "relay", + }, + }); + await vi.waitFor(() => expect(messages).toContainEqual({ + type: "route-ready", + revision: 3, + phase: "active", + })); + }); + it("reports the selected ingress connection when it fails during peer prepare", async () => { const messages: ClientMessage[] = []; const publishers: ReturnType[] = []; diff --git a/tests/server-signal.test.ts b/tests/server-signal.test.ts index 2fc3761..a425cd8 100644 --- a/tests/server-signal.test.ts +++ b/tests/server-signal.test.ts @@ -319,6 +319,7 @@ async function startRelayRelativeFpsFixture( onSfuTokenIssue?: () => void; prepareTimeoutMs?: number; hostFirstCandidate?: boolean; + hostShareGeneration?: string; secondChild?: boolean; selectedEdgeTurn?: boolean; sfu?: boolean; @@ -339,7 +340,14 @@ async function startRelayRelativeFpsFixture( : await startHarness({ peerAssistedMedia: true, now }); const host = await openClient(harness.webSocketUrl); const hostAuth = peerAssisted( - await authenticate(host, harness.room, "host", `${prefix}-host`), + await authenticate( + host, + harness.room, + "host", + `${prefix}-host`, + 1, + options.hostShareGeneration, + ), ); const parent = await openClient(harness.webSocketUrl); const parentAuth = peerAssisted( @@ -570,6 +578,42 @@ async function confirmRelayChildSevere( } } +async function prepareHostParentQualityProbe( + fixture: Awaited>, + advanceWindow: () => void, + connectionId: string, +) { + fixture.alternate.socket.close(); + const [current] = await Promise.all([ + nextActiveRouteAfter(fixture.child, fixture.routeRevision), + nextActiveRouteAfter(fixture.parent, fixture.routeRevision), + nextActiveRouteAfter(fixture.host, fixture.routeRevision), + ]); + fixture.routeRevision = current.revision; + await confirmRelayChildSevere( + fixture, + fixture.child, + fixture.childConnectionId, + advanceWindow, + ); + const [viewerPrepare, hostPrepare] = await Promise.all([ + nextPreparedRoute(fixture.child), + nextPreparedRoute(fixture.host), + ]); + expect(viewerPrepare.assignment.upstream).toEqual({ + kind: "peer", + peerId: fixture.hostAuth.peerId, + }); + expect(hostPrepare.revision).toBe(viewerPrepare.revision); + await sendTestOffer( + fixture.host, + fixture.child, + fixture.childAuth.peerId, + connectionId, + ); + return { viewerPrepare, hostPrepare }; +} + async function reconnectRelayQualityEdges( fixture: Awaited>, suffix: string, @@ -1119,6 +1163,92 @@ async function activateSingleViewerSfu( return { ...direct, revision: viewerPrepare.revision }; } +async function prepareSfuHostParentQualityProbe( + harness: SignalHarness, + active: Awaited>, + prefix: string, + advanceWindow: () => void, +) { + active.viewer.socket.send(JSON.stringify({ + type: "relay-capacity", + downstreamEdges: 1, + })); + const hostFiller = await openClient(harness.webSocketUrl); + await authenticate( + hostFiller, + harness.room, + "viewer", + `${prefix}-filler`, + 0, + ); + const parent = await openClient(harness.webSocketUrl); + const parentAuth = peerAssisted(await authenticate( + parent, + harness.room, + "viewer", + `${prefix}-parent`, + 1, + )); + expect(parentAuth.routeAssignment.upstream).toEqual({ + kind: "peer", + peerId: active.viewerAuth.peerId, + }); + const child = await openClient(harness.webSocketUrl); + const childAuth = peerAssisted(await authenticate( + child, + harness.room, + "viewer", + `${prefix}-child`, + 0, + )); + expect(childAuth.routeAssignment.upstream).toEqual({ + kind: "peer", + peerId: parentAuth.peerId, + }); + hostFiller.socket.close(); + const current = await nextActiveRouteAfter(child, childAuth.routeRevision); + await Promise.all([ + nextActiveRouteRevision(active.host, current.revision), + nextActiveRouteRevision(active.viewer, current.revision), + nextActiveRouteRevision(parent, current.revision), + ]); + + await sendTestOffer( + active.viewer, + parent, + parentAuth.peerId, + `${prefix}_parent_connection`, + ); + const childConnectionId = `${prefix}_child_connection`; + await sendTestOffer(parent, child, childAuth.peerId, childConnectionId); + for (let sequence = 0; sequence < 3; sequence += 1) { + child.socket.send(JSON.stringify(viewerQualityEvidenceWithMetrics( + childConnectionId, + current.revision, + sequence, + { framesDecodedDelta: 0 }, + ))); + await correlateParentEdgeQualityEvidence(parent); + advanceWindow(); + } + const [childPrepare, hostPrepare] = await Promise.all([ + nextPreparedRoute(child), + nextPreparedRoute(active.host), + ]); + expect(childPrepare.assignment.upstream).toEqual({ + kind: "peer", + peerId: active.hostAuth.peerId, + }); + expect(hostPrepare.revision).toBe(childPrepare.revision); + return { + child, + childPrepare, + current, + hostPrepare, + parentAuth, + }; +} + async function activateTwoFailedViewerSfuRoots( webSocketUrl: string, room: CreatedRoom, @@ -3553,13 +3683,20 @@ describe("WebSocket signaling", () => { }); }); - it("probes an eligible peer Viewer rather than the Host", async () => { + it("uses an eligible Host as the provisional peer parent", async () => { let now = 120_000; const fixture = await startRelayRelativeFpsFixture( "quality_skip_host", () => now, - { alternateRelayCapacity: 1 }, + { alternateRelayCapacity: 0 }, ); + fixture.alternate.socket.close(); + const [current] = await Promise.all([ + nextActiveRouteAfter(fixture.child, fixture.routeRevision), + nextActiveRouteAfter(fixture.parent, fixture.routeRevision), + nextActiveRouteAfter(fixture.host, fixture.routeRevision), + ]); + fixture.routeRevision = current.revision; const advanceWindow = () => { now += 2_000; }; await confirmRelayChildSevere( fixture, @@ -3569,28 +3706,29 @@ describe("WebSocket signaling", () => { ); const moved = await completePreparedPeerMigration( fixture.child, - fixture.alternate, + fixture.host, fixture.childAuth.peerId, - fixture.alternateAuth.peerId, + fixture.hostAuth.peerId, fixture.routeRevision, - "quality_skip_host_probe", + "quality_host_probe", ); - expect(await nextActiveRouteRevision(fixture.host, moved.revision)).toMatchObject({ - phase: "active", - }); expect(moved.assignment.upstream).toEqual({ kind: "peer", - peerId: fixture.alternateAuth.peerId, + peerId: fixture.hostAuth.peerId, }); }); - it("keeps the old edge when no eligible Viewer alternate exists", async () => { + it("keeps the old edge when the Host upload budget is already full", async () => { let now = 125_000; const fixture = await startRelayRelativeFpsFixture( "quality_only_host", () => now, { alternateRelayCapacity: 0 }, ); + await Promise.all([ + nextActiveRouteRevision(fixture.host, fixture.routeRevision), + nextActiveRouteRevision(fixture.alternate, fixture.routeRevision), + ]); const advanceWindow = () => { now += 2_000; }; await confirmRelayChildSevere( fixture, @@ -3603,6 +3741,298 @@ describe("WebSocket signaling", () => { ); }); + it("rolls a Host-parent probe back only for its exact revision and connection", async () => { + let now = 126_000; + const fixture = await startRelayRelativeFpsFixture( + "quality_host_exact_failure", + () => now, + { alternateRelayCapacity: 0 }, + ); + const probeConnectionId = "quality_host_exact_failure_probe"; + const { viewerPrepare } = await prepareHostParentQualityProbe( + fixture, + () => { now += 2_000; }, + probeConnectionId, + ); + + for (const [revision, connectionId] of [ + [viewerPrepare.revision - 1, probeConnectionId], + [viewerPrepare.revision, `${probeConnectionId}-wrong`], + ] as const) { + fixture.child.socket.send(JSON.stringify({ + type: "route-failed", + revision, + phase: "prepare", + connectionId, + })); + } + await Promise.all([ + expect( + fixture.child.inbox.next("route-update", 40), + ).rejects.toThrow("Timed out"), + expect( + fixture.host.inbox.next("route-update", 40), + ).rejects.toThrow("Timed out"), + ]); + + fixture.child.socket.send(JSON.stringify({ + type: "route-failed", + revision: viewerPrepare.revision, + phase: "prepare", + connectionId: probeConnectionId, + })); + const [viewerRollback, hostRollback] = await Promise.all([ + nextActiveRouteAfter(fixture.child, viewerPrepare.revision), + nextActiveRouteAfter(fixture.host, viewerPrepare.revision), + ]); + expect(viewerRollback.assignment.upstream).toEqual({ + kind: "peer", + peerId: fixture.parentAuth.peerId, + }); + expect(hostRollback.revision).toBe(viewerRollback.revision); + await expect( + sendTestOffer( + fixture.parent, + fixture.child, + fixture.childAuth.peerId, + "quality_host_exact_failure_old_edge", + ), + ).resolves.toMatchObject({ fromPeerId: fixture.parentAuth.peerId }); + }); + + it("invalidates a Host-parent probe when the Host session is replaced", async () => { + let now = 127_000; + const shareGeneration = "quality_host_session_share"; + const fixture = await startRelayRelativeFpsFixture( + "quality_host_session", + () => now, + { alternateRelayCapacity: 0, hostShareGeneration: shareGeneration }, + ); + const { viewerPrepare } = await prepareHostParentQualityProbe( + fixture, + () => { now += 2_000; }, + "quality_host_session_probe", + ); + const oldClosed = new Promise((resolve) => + fixture.host.socket.once("close", (code) => resolve(code)), + ); + const replacement = await openClient(fixture.harness.webSocketUrl); + const replacementAuth = peerAssisted(await authenticate( + replacement, + fixture.harness.room, + "host", + "quality_host_session-host", + 1, + shareGeneration, + )); + expect(await oldClosed).toBe(4001); + expect(replacementAuth.peerId).toBe(fixture.hostAuth.peerId); + expect(replacementAuth.routeRevision).toBeGreaterThan(viewerPrepare.revision); + expect(replacementAuth.routeAssignment.childPeerIds).not.toContain( + fixture.childAuth.peerId, + ); + const rollback = await nextActiveRouteAfter( + fixture.child, + viewerPrepare.revision, + ); + expect(rollback.assignment.upstream).toEqual({ + kind: "peer", + peerId: fixture.parentAuth.peerId, + }); + + fixture.child.socket.send(JSON.stringify({ + type: "route-ready", + revision: viewerPrepare.revision, + phase: "prepare", + })); + await expect( + fixture.child.inbox.next("route-update", 40), + ).rejects.toThrow("Timed out"); + }); + + it("invalidates a Host-parent probe when a new share generation starts", async () => { + let now = 127_500; + const fixture = await startRelayRelativeFpsFixture( + "quality_host_share", + () => now, + { + alternateRelayCapacity: 0, + hostShareGeneration: "quality_host_share_a", + }, + ); + const { viewerPrepare } = await prepareHostParentQualityProbe( + fixture, + () => { now += 2_000; }, + "quality_host_share_probe", + ); + const oldClosed = new Promise((resolve) => + fixture.host.socket.once("close", (code) => resolve(code)), + ); + const replacement = await openClient(fixture.harness.webSocketUrl); + const replacementAuth = peerAssisted(await authenticate( + replacement, + fixture.harness.room, + "host", + "quality_host_share-host", + 1, + "quality_host_share_b", + )); + expect(await oldClosed).toBe(4001); + expect(replacementAuth.peerId).toBe(fixture.hostAuth.peerId); + expect(replacementAuth.routeAssignment.childPeerIds).not.toContain( + fixture.childAuth.peerId, + ); + await fixture.child.inbox.next("sharing-stopped"); + const nextShareRoute = await nextActiveRouteRevision( + fixture.child, + replacementAuth.routeRevision, + ); + expect(nextShareRoute.assignment.upstream).toEqual({ + kind: "peer", + peerId: fixture.parentAuth.peerId, + }); + + fixture.child.socket.send(JSON.stringify({ + type: "route-ready", + revision: viewerPrepare.revision, + phase: "prepare", + })); + fixture.child.socket.send(JSON.stringify({ + type: "route-failed", + revision: viewerPrepare.revision, + phase: "prepare", + connectionId: "quality_host_share_probe", + })); + await expect( + fixture.child.inbox.next("route-update", 40), + ).rejects.toThrow("Timed out"); + }); + + it("counts an answered Host selected overlay against a quality-probe slot", async () => { + let now = 128_000; + const harness = await startSfuHarness({ + now: () => now, + tokenIssuer: { + async issueToken({ peerId }) { return `token-${peerId}`; }, + }, + selectedEdgeTurn: true, + }); + const active = await activateSingleViewerSfu( + harness.webSocketUrl, + harness.room, + "quality-host-selected-budget", + ); + active.viewer.socket.send(JSON.stringify({ + type: "relay-capacity", + downstreamEdges: 1, + })); + const hostFiller = await openClient(harness.webSocketUrl); + const hostFillerAuth = peerAssisted(await authenticate( + hostFiller, + harness.room, + "viewer", + "quality-host-selected-budget-filler", + 0, + )); + expect(hostFillerAuth.routeAssignment.upstream).toEqual({ + kind: "peer", + peerId: active.hostAuth.peerId, + }); + const parent = await openClient(harness.webSocketUrl); + const parentAuth = peerAssisted(await authenticate( + parent, + harness.room, + "viewer", + "quality-host-selected-budget-parent", + 1, + )); + expect(parentAuth.routeAssignment.upstream).toEqual({ + kind: "peer", + peerId: active.viewerAuth.peerId, + }); + const child = await openClient(harness.webSocketUrl); + const childAuth = peerAssisted(await authenticate( + child, + harness.room, + "viewer", + "quality-host-selected-budget-child", + 0, + )); + expect(childAuth.routeAssignment.upstream).toEqual({ + kind: "peer", + peerId: parentAuth.peerId, + }); + + hostFiller.socket.close(); + const current = await nextActiveRouteAfter(child, childAuth.routeRevision); + await Promise.all([ + nextActiveRouteRevision(active.host, current.revision), + nextActiveRouteRevision(active.viewer, current.revision), + nextActiveRouteRevision(parent, current.revision), + ]); + const grant = await startPeerSelectedTurn( + active.viewer, + active.host, + current.revision, + ); + active.host.socket.send(JSON.stringify({ + type: "signal", + targetPeerId: active.viewerAuth.peerId, + payload: { + kind: "description", + connectionId: grant.newConnectionId, + description: { type: "offer", sdp: "v=0\r\n" }, + }, + })); + await active.viewer.inbox.next("signal"); + active.viewer.socket.send(JSON.stringify({ + type: "signal", + targetPeerId: active.hostAuth.peerId, + payload: { + kind: "description", + connectionId: grant.newConnectionId, + description: { type: "answer", sdp: "v=0\r\n" }, + }, + })); + await active.host.inbox.next("signal"); + + await sendTestOffer( + active.viewer, + parent, + parentAuth.peerId, + "quality_host_selected_budget_parent_connection", + ); + const childConnectionId = "quality_host_selected_budget_child_connection"; + await sendTestOffer(parent, child, childAuth.peerId, childConnectionId); + for (let sequence = 0; sequence < 3; sequence += 1) { + child.socket.send(JSON.stringify(viewerQualityEvidenceWithMetrics( + childConnectionId, + current.revision, + sequence, + { framesDecodedDelta: 0 }, + ))); + await correlateParentEdgeQualityEvidence(parent); + now += 2_000; + } + await expect(child.inbox.next("route-update", 40)).rejects.toThrow( + "Timed out", + ); + + active.host.socket.send(JSON.stringify({ + type: "signal", + targetPeerId: active.viewerAuth.peerId, + payload: { + kind: "candidate", + connectionId: grant.newConnectionId, + candidate: null, + }, + })); + await expect(active.viewer.inbox.next("signal")).resolves.toMatchObject({ + fromPeerId: active.hostAuth.peerId, + payload: { connectionId: grant.newConnectionId }, + }); + }); + it("uses a capable SFU root as the provisional peer parent", async () => { let now = 128_000; const harness = await startSfuHarness({ @@ -6374,6 +6804,171 @@ describe("WebSocket signaling", () => { }); }); + it("does not let a soft Host-parent probe swallow active SFU publication failure", async () => { + let now = 205_000; + const harness = await startSfuHarness({ + now: () => now, + tokenIssuer: { + async issueToken({ peerId }) { return `token-${peerId}`; }, + }, + }); + const active = await activateSingleViewerSfu( + harness.webSocketUrl, + harness.room, + "quality-host-sfu-loss", + ); + const { child, childPrepare, current, hostPrepare, parentAuth } = + await prepareSfuHostParentQualityProbe( + harness, + active, + "quality_host_sfu_loss", + () => { now += 2_000; }, + ); + + active.host.socket.send(JSON.stringify({ + type: "route-failed", + revision: current.revision, + phase: "prepare", + connectionId: null, + })); + await expect(child.inbox.next("route-update", 40)).rejects.toThrow( + "Timed out", + ); + + active.host.socket.send(JSON.stringify({ + type: "route-failed", + revision: hostPrepare.revision, + phase: "prepare", + connectionId: null, + })); + const [childBaseline, firstHostRecovery] = await Promise.all([ + nextActiveRouteAfter(child, childPrepare.revision), + nextActiveRouteAfter(active.host, hostPrepare.revision), + ]); + const hostBaseline = firstHostRecovery.assignment.sfuPublicationGeneration === null + ? firstHostRecovery + : await nextActiveRouteAfter(active.host, firstHostRecovery.revision); + expect(childBaseline.assignment.upstream).toEqual({ + kind: "peer", + peerId: parentAuth.peerId, + }); + expect(hostBaseline.assignment.sfuPublicationGeneration).toBeNull(); + }); + + it("bounds selected ingress after active Host SFU failure during a soft probe", async () => { + try { + let now = 208_000; + const harness = await startSfuHarness({ + now: () => now, + tokenIssuer: { + async issueToken({ peerId }) { return `token-${peerId}`; }, + }, + selectedEdgeTurn: true, + }); + const active = await activateSingleViewerSfu( + harness.webSocketUrl, + harness.room, + "quality-host-selected-timeout", + ); + const { child, childPrepare, hostPrepare, parentAuth } = + await prepareSfuHostParentQualityProbe( + harness, + active, + "quality_host_selected_timeout", + () => { now += 2_000; }, + ); + + vi.useFakeTimers(); + active.host.socket.send(JSON.stringify({ + type: "route-failed", + revision: hostPrepare.revision, + phase: "prepare", + connectionId: null, + })); + const [childRollback, hostRollback, grant] = await Promise.all([ + nextActiveRouteAfter(child, childPrepare.revision), + nextActiveRouteAfter(active.host, hostPrepare.revision), + active.host.inbox.next("selected-edge-turn"), + ]); + expect(childRollback.assignment.upstream).toEqual({ + kind: "peer", + peerId: parentAuth.peerId, + }); + expect(hostRollback.revision).toBe(hostPrepare.revision + 1); + expect(hostRollback.assignment.sfuPublicationGeneration).toBeTruthy(); + expect(grant).toMatchObject({ + edgeKind: "host-sfu-ingress", + revision: hostRollback.revision, + }); + + await vi.advanceTimersByTimeAsync(30_000); + await expect(active.host.inbox.next("error")).resolves.toMatchObject({ + code: "PEER_NOT_FOUND", + message: "Selected SFU relay ingress timed out", + }); + await expect( + nextActiveRouteAfter(active.host, hostRollback.revision), + ).resolves.toMatchObject({ + revision: hostRollback.revision + 1, + assignment: { sfuPublicationGeneration: null }, + }); + } finally { + vi.useRealTimers(); + } + }); + + it("does not let a soft Host-parent probe swallow active selected ingress failure", async () => { + let now = 210_000; + const harness = await startSfuHarness({ + now: () => now, + tokenIssuer: { + async issueToken({ peerId }) { return `token-${peerId}`; }, + }, + selectedEdgeTurn: true, + }); + const active = await activateSingleViewerSfu( + harness.webSocketUrl, + harness.room, + "quality-selected-host-loss", + ); + active.host.socket.send(JSON.stringify({ + type: "route-failed", + revision: active.revision, + phase: "active", + connectionId: null, + })); + const ingress = await active.host.inbox.next("selected-edge-turn"); + active.host.socket.send(JSON.stringify({ + type: "route-ready", + revision: active.revision, + phase: "active", + })); + + const { child, childPrepare, current, hostPrepare } = + await prepareSfuHostParentQualityProbe( + harness, + active, + "quality_selected_host", + () => { now += 2_000; }, + ); + + active.host.socket.send(JSON.stringify({ + type: "route-failed", + revision: current.revision, + phase: "active", + connectionId: ingress.newConnectionId, + })); + const [childBaseline, firstHostRecovery] = await Promise.all([ + nextActiveRouteAfter(child, childPrepare.revision), + nextActiveRouteAfter(active.host, hostPrepare.revision), + ]); + const hostBaseline = firstHostRecovery.assignment.sfuPublicationGeneration === null + ? firstHostRecovery + : await nextActiveRouteAfter(active.host, firstHostRecovery.revision); + expect(childBaseline.assignment.upstream.kind).toBe("peer"); + expect(hostBaseline.assignment.sfuPublicationGeneration).toBeNull(); + }); + it("keeps ordinary intent when topology abort fills the overlap slot", async () => { const reselections = vi.spyOn( HybridMediaRouter.prototype,