feat(routing): probe Host parents without breaking media (#183)
This commit is contained in:
AAA authored and GitHub committed
1 parent
9080e68601
commit
30ffe44d60
7 files changed
+1624
-35
No files matched your search
@@ -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<HostPreparedChildResolution, { kind: "promote" }>
|
||||
| { 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<boolean> {
|
||||
const peer = this.livePeer();
|
||||
return peer ? peer.updateProfile(profile) : Promise.resolve(true);
|
||||
}
|
||||
|
||||
async replaceStream(stream: MediaStream): Promise<boolean> {
|
||||
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
|
||||
);
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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<string, HostPeer>());
|
||||
const retiredConnectionsRef = useRef(new Map<string, string>());
|
||||
const selectedHostChildRef = useRef<ActiveSelectedHostChild | null>(null);
|
||||
const hostProvisionalChildRef = useRef<HostProvisionalChild | null>(null);
|
||||
const failedActiveHostChildRef = useRef<FailedActiveHostChild | null>(null);
|
||||
const activeHostChildPeerIdsRef = useRef<string[]>([]);
|
||||
const hostPeerIdRef = useRef<string | null>(null);
|
||||
const viewerQualityEvidenceRef = useRef(
|
||||
new Map<string, ViewerQualityEvidence>(),
|
||||
@@ -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),
|
||||
]);
|
||||
|
||||
|
||||
@@ -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<ClientMessage, { type: "route-failed" }>,
|
||||
@@ -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,
|
||||
|
||||
@@ -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 }> = [];
|
||||
|
||||
@@ -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<typeof createFakePublisher>[] = [];
|
||||
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<typeof createFakePublisher>[] = [];
|
||||
|
||||
+606
-11
@@ -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<ReturnType<typeof startRelayRelativeFpsFixture>>,
|
||||
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<ReturnType<typeof startRelayRelativeFpsFixture>>,
|
||||
suffix: string,
|
||||
@@ -1119,6 +1163,92 @@ async function activateSingleViewerSfu(
|
||||
return { ...direct, revision: viewerPrepare.revision };
|
||||
}
|
||||
|
||||
async function prepareSfuHostParentQualityProbe(
|
||||
harness: SignalHarness,
|
||||
active: Awaited<ReturnType<typeof activateSingleViewerSfu>>,
|
||||
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<number>((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<number>((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,
|
||||
|
||||
Reference in new issue
Block a user