okooooooo/Piik
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(client): simplify route candidate proof

Browse Source
This commit is contained in:
AAA committed 2026-08-23 17:17:04 +08:00
1 parent e2cd2aaab1
commit cd534c7e68
9 files changed
+946 -3625
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/client/media/host-provisional-child.ts
+115 -126
View File
@@ -1,6 +1,7 @@
import type {
IceConfig,
ParticipantRouteAssignment,
ServerMessage,
SignalPayload,
} from "../../shared/protocol";
import { countEndpointMediaCopies } from "../../shared/media-copy-accounting";
@@ -8,68 +9,15 @@ 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;
}
export interface ActiveSelectedHostChild {
peerId: string;
connectionId: string;
currentRouteRevision: number;
pendingCarryRevision: number | null;
}
export function hostSelectedChildForConnection(
children: ReadonlyMap<string, ActiveSelectedHostChild>,
peerId: string,
connectionId: string,
): ActiveSelectedHostChild | undefined {
const child = children.get(peerId);
return child?.connectionId === connectionId ? child : undefined;
}
export function carryHostSelectedChild(
children: Map<string, ActiveSelectedHostChild>,
peerId: string,
connectionId: string,
revision: number,
): boolean {
const child = hostSelectedChildForConnection(
children,
peerId,
connectionId,
);
if (!child || revision < child.currentRouteRevision) {
return false;
}
child.currentRouteRevision = revision;
child.pendingCarryRevision = revision;
return true;
}
export function acceptHostSelectedChildrenRevision(
children: Map<string, ActiveSelectedHostChild>,
revision: number,
): string[] {
const removedPeerIds: string[] = [];
for (const [peerId, child] of children) {
if (child.pendingCarryRevision === revision) {
child.pendingCarryRevision = null;
} else {
children.delete(peerId);
removedPeerIds.push(peerId);
}
}
return removedPeerIds;
}
type SelectedEdgeTurn = Extract<
ServerMessage,
{ type: "selected-edge-turn"; edgeKind: "peer-selected" }
>;
interface HostProvisionalInput {
revision: number;
assignment: ParticipantRouteAssignment;
activeChildPeerIds: readonly string[];
selectedPeerIds: Iterable<string>;
maxMediaEdges: number;
}
@@ -81,53 +29,104 @@ interface HostProvisionalPrepareInput extends HostProvisionalInput {
interface HostProvisionalChildEvents {
sendSignal: (peerId: string, payload: SignalPayload) => boolean;
hasActivePeer?: (peerId: string) => boolean;
activeConnectionId?: (peerId: string) => string | null;
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: "ordinary" }
| { kind: "promote"; peerId: string; peer: HostPeer };
interface PreparedHostChild extends HostPreparedChildIdentity {
interface PreparedHostChild {
revision: number;
peerId: string;
peer: HostPeer;
failed: boolean;
replacesConnectionId: string | null;
}
export class HostProvisionalChild {
private prepared: PreparedHostChild | null = null;
private preparedRevision: number | null = null;
private plannedChildPeerIds: string[] = [];
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)
) {
const planned = validPlannedHostChildren(input);
if (this.promotedPeer || !planned) {
this.discard();
return false;
}
if (
this.prepared?.revision === input.revision &&
this.prepared.peerId === peerId
this.preparedRevision === input.revision &&
samePeerIds(this.plannedChildPeerIds, planned)
) {
return this.prepared.failedConnectionId === null;
return this.prepared ? !this.prepared.failed : true;
}
const peerId = planned.find(
(candidate) => !input.activeChildPeerIds.includes(candidate),
);
this.discard();
this.preparedRevision = input.revision;
this.plannedChildPeerIds = planned;
if (!peerId) {
return true;
}
if (this.events.activeConnectionId?.(peerId)) {
this.discard();
return false;
}
this.startPrepared(input.revision, peerId, input);
return true;
}
prepareSelectedTurn(
message: SelectedEdgeTurn,
input: Pick<HostProvisionalPrepareInput, "iceConfig" | "stream" | "profile">,
now = Date.now(),
): boolean {
if (
this.preparedRevision !== message.revision ||
!this.plannedChildPeerIds.includes(message.viewerPeerId) ||
Date.parse(message.expiresAt) <= now
) {
return false;
}
const prepared = this.prepared;
if (prepared && prepared.peerId !== message.viewerPeerId) {
return false;
}
if (prepared?.peer.connectionId === message.newConnectionId) {
return !prepared.failed;
}
const oldConnectionId =
prepared?.peer.connectionId ??
this.events.activeConnectionId?.(message.viewerPeerId);
if (oldConnectionId !== message.oldConnectionId) {
return false;
}
this.discardPeer();
this.startPrepared(message.revision, message.viewerPeerId, input, message);
return true;
}
private startPrepared(
revision: number,
peerId: string,
input: Pick<HostProvisionalPrepareInput, "iceConfig" | "stream" | "profile">,
selectedTurn?: SelectedEdgeTurn,
): void {
let peer: HostPeer;
peer = new HostPeer(
peerId,
input.iceConfig,
selectedTurn
? { iceServers: [selectedTurn.iceServer] }
: input.iceConfig,
input.stream,
input.profile,
{
@@ -147,13 +146,16 @@ export class HostProvisionalChild {
}
},
},
selectedTurn !== undefined,
selectedTurn?.newConnectionId,
);
this.signalingPeer = peer;
this.prepared = {
revision: input.revision,
revision,
peerId,
peer,
failedConnectionId: null,
failed: false,
replacesConnectionId: selectedTurn?.oldConnectionId ?? null,
};
void peer
.start()
@@ -163,14 +165,13 @@ export class HostProvisionalChild {
this.fail(peer);
}
});
return true;
}
ownsSignal(fromPeerId: string, payload: SignalPayload): boolean {
const prepared = this.prepared;
return Boolean(
prepared &&
prepared.failedConnectionId === null &&
!prepared.failed &&
prepared.peerId === fromPeerId &&
prepared.peer.connectionId === payload.connectionId,
);
@@ -210,29 +211,35 @@ export class HostProvisionalChild {
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") {
const currentConnectionId = prepared
? this.events.activeConnectionId?.(prepared.peerId) ?? null
: null;
if (
prepared &&
!prepared.failed &&
this.preparedRevision === input.revision &&
prepared.revision === input.revision &&
input.assignment.childPeerIds.includes(prepared.peerId) &&
(!input.activeChildPeerIds.includes(prepared.peerId) ||
prepared.replacesConnectionId === currentConnectionId)
) {
this.prepared = null;
prepared?.peer.dispose();
return activation;
this.preparedRevision = null;
this.plannedChildPeerIds = [];
this.promotedPeer = prepared.peer;
return { kind: "promote", peerId: prepared.peerId, peer: prepared.peer };
}
this.discard();
return activation;
return { kind: "ordinary" };
}
discard(): void {
this.preparedRevision = null;
this.plannedChildPeerIds = [];
this.discardPeer();
}
private discardPeer(): void {
const prepared = this.prepared;
this.prepared = null;
if (this.signalingPeer === prepared?.peer) {
@@ -242,7 +249,7 @@ export class HostProvisionalChild {
}
private livePeer(): HostPeer | null {
return this.prepared?.failedConnectionId === null
return this.prepared && !this.prepared.failed
? this.prepared.peer
: null;
}
@@ -251,7 +258,7 @@ export class HostProvisionalChild {
if (this.prepared?.peer !== peer) {
return;
}
this.prepared.failedConnectionId = peer.connectionId;
this.prepared.failed = true;
if (this.signalingPeer === peer) {
this.signalingPeer = null;
}
@@ -259,48 +266,30 @@ export class HostProvisionalChild {
}
}
export function plannedHostProvisionalChild(
function validPlannedHostChildren(
input: HostProvisionalInput,
): string | null {
): 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 ||
childPeerIds.length < input.activeChildPeerIds.length ||
childPeerIds.length > input.activeChildPeerIds.length + 1 ||
input.activeChildPeerIds.some(
(candidate) => !childPeerIds.includes(candidate),
) ||
countEndpointMediaCopies({
childPeerIds: input.assignment.childPeerIds,
publicationGeneration: input.assignment.sfuPublicationGeneration,
selectedChildPeerIds: input.selectedPeerIds,
}) > input.maxMediaEdges
) {
return null;
}
return peerId ?? null;
return childPeerIds;
}
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 samePeerIds(left: readonly string[], right: readonly string[]): boolean {
return (
left.length === right.length &&
left.every((peerId, index) => peerId === right[index])
);
}
src/client/media/host-sfu-route.ts
+122 -104
View File
@@ -40,7 +40,6 @@ interface HostPublisherSlot {
connected: boolean;
active: boolean;
failed: boolean;
acknowledgeActive: boolean;
connectionId: string;
selectedEdgeTurn: boolean;
}
@@ -71,6 +70,7 @@ export class HostSfuRoute {
private transitionTail: Promise<void> = Promise.resolve();
private resyncGeneration = 0;
private resyncing = false;
private paused = false;
private closed = false;
private lastConfig: Extract<ServerMessage, { type: "sfu-config" }> | null =
null;
@@ -81,7 +81,7 @@ export class HostSfuRoute {
constructor(private readonly events: HostSfuRouteEvents) {}
accept(update: RouteUpdateInput, acknowledge = true): RouteUpdateResult {
accept(update: RouteUpdateInput): RouteUpdateResult {
if (this.closed) {
return "stale";
}
@@ -90,6 +90,10 @@ export class HostSfuRoute {
if (result === "stale") {
return result;
}
if (this.paused) {
this.clearPending();
return result;
}
if (previousRevision !== update.revision) {
this.recovery = null;
this.lastFailureStage = null;
@@ -116,16 +120,15 @@ export class HostSfuRoute {
const token = this.route.token();
if (token && !this.resyncing) {
void this.queueActivation(token, acknowledge);
void this.queueActivation(token);
}
return result;
}
async acceptAndWait(
update: RouteUpdateInput,
acknowledge = true,
): Promise<RouteUpdateResult> {
const result = this.accept(update, acknowledge);
const result = this.accept(update);
await this.transitionTail;
return result;
}
@@ -134,7 +137,7 @@ export class HostSfuRoute {
update: RouteUpdateInput,
): Promise<RouteUpdateResult> {
this.lastFailureStage = null;
const result = this.accept(update, false);
const result = this.accept(update);
if (result !== "stale" || this.closed) {
await this.transitionTail;
return result;
@@ -176,7 +179,7 @@ export class HostSfuRoute {
this.resyncing = false;
const token = this.route.token();
if (token && this.route.getPhase() === "active") {
await this.queueActivation(token, false);
await this.queueActivation(token);
}
return accepted;
}
@@ -184,7 +187,7 @@ export class HostSfuRoute {
async acceptConfig(
message: Extract<ServerMessage, { type: "sfu-config" }>,
): Promise<void> {
if (this.closed || !this.route.acceptsConfig(message.revision)) {
if (this.closed || this.paused || !this.route.acceptsConfig(message.revision)) {
return;
}
const token = this.route.token();
@@ -198,10 +201,8 @@ export class HostSfuRoute {
if (this.active?.publicationGeneration === publicationGeneration) {
this.active.revision = message.revision;
if (phase === "prepare" && this.active.connected) {
this.ready(message.revision, "prepare");
} else if (phase === "active" && this.active.active) {
await this.queueActivation(token, true);
if (phase === "active" && this.active.active) {
await this.queueActivation(token);
}
return;
}
@@ -210,9 +211,6 @@ export class HostSfuRoute {
this.pending.publicationGeneration === publicationGeneration &&
!this.pending.failed
) {
if (this.pending.connected && phase === "prepare") {
this.ready(message.revision, "prepare");
}
return;
}
@@ -241,7 +239,6 @@ export class HostSfuRoute {
connected: false,
active: false,
failed: false,
acknowledgeActive: false,
connectionId: selectedEdgeTurn?.newConnectionId ?? publicationGeneration,
selectedEdgeTurn: selectedEdgeTurn !== null,
};
@@ -271,10 +268,11 @@ export class HostSfuRoute {
return;
}
slot.connected = true;
if (phase === "prepare") {
this.ready(message.revision, "prepare");
} else if (!this.resyncing) {
await this.queueActivation(token, true);
if (!(await this.preparePublisher(slot))) {
return;
}
if (phase === "active" && !this.resyncing) {
await this.queueActivation(token);
}
} catch {
await disconnectPublisher(publisher);
@@ -282,6 +280,14 @@ export class HostSfuRoute {
}
}
setPaused(paused: boolean): void {
this.paused = paused;
if (paused) {
this.clearPending();
this.selectedEdgeTurn = null;
}
}
/** Apply one controller-selected TURN grant to the next SFU publisher PC. */
startSelectedEdgeTurn(
message: Extract<
@@ -292,20 +298,33 @@ export class HostSfuRoute {
if (
this.closed ||
!this.route.acceptsConfig(message.revision) ||
this.route.getPhase() !== "prepare" ||
Date.parse(message.expiresAt) <= Date.now()
) {
return false;
}
const assignment = this.route.getPlannedAssignment();
const publicationGeneration = assignment?.sfuPublicationGeneration;
const pending = this.pending;
if (
!publicationGeneration ||
publicationGeneration !== message.publicationGeneration ||
message.oldConnectionId !== publicationGeneration ||
this.active?.publicationGeneration === publicationGeneration
(pending !== null &&
(pending.revision !== message.revision ||
pending.publicationGeneration !== publicationGeneration))
) {
return false;
}
if (pending?.connectionId === message.newConnectionId) {
return !pending.failed;
}
if (
(pending?.connectionId ?? publicationGeneration) !==
message.oldConnectionId
) {
return false;
}
this.clearPending();
this.selectedEdgeTurn = message;
if (this.lastConfig?.revision === message.revision) {
void this.acceptConfig(this.lastConfig);
@@ -314,9 +333,14 @@ export class HostSfuRoute {
}
updateProfile(profile: QualityProfile): Promise<boolean> {
return this.active?.active
? this.active.publisher.updateProfile(profile).catch(() => false)
: Promise.resolve(true);
const slots = this.publishingSlots();
return slots.length === 0
? Promise.resolve(true)
: Promise.all(
slots.map((slot) =>
slot.publisher.updateProfile(profile).catch(() => false),
),
).then((results) => results.every(Boolean));
}
getQualityWarning(): string | null {
@@ -335,9 +359,21 @@ export class HostSfuRoute {
}
replaceStream(stream: MediaStream): Promise<boolean> {
return this.active
? this.active.publisher.replaceStream(stream).catch(() => false)
: Promise.resolve(true);
const slots = this.publishingSlots();
return slots.length === 0
? Promise.resolve(true)
: Promise.all(
slots.map((slot) =>
slot.publisher.replaceStream(stream).catch(() => false),
),
).then((results) => results.every(Boolean));
}
private publishingSlots(): HostPublisherSlot[] {
return [...new Set([this.active, this.pending])].filter(
(slot): slot is HostPublisherSlot =>
slot !== null && slot.active && !slot.failed,
);
}
async failActivePublisher(): Promise<void> {
@@ -382,17 +418,14 @@ export class HostSfuRoute {
});
}
private queueActivation(
token: RouteOperationToken,
acknowledge: boolean,
): Promise<void> {
private queueActivation(token: RouteOperationToken): Promise<void> {
return this.queueTransition(async () => {
if (this.closed || this.resyncing || !this.route.owns(token, "active")) {
return;
}
const assignment = this.route.getActiveAssignment();
if (assignment) {
await this.activate(assignment, token, acknowledge);
await this.activate(assignment, token);
}
});
}
@@ -406,7 +439,6 @@ export class HostSfuRoute {
private async activate(
assignment: ParticipantRouteAssignment,
token: RouteOperationToken,
acknowledge: boolean,
): Promise<void> {
const publicationGeneration = assignment.sfuPublicationGeneration;
if (publicationGeneration === null) {
@@ -416,7 +448,7 @@ export class HostSfuRoute {
return;
}
this.events.reconcileChildren(assignment.childPeerIds);
this.commitMedia(token, acknowledge);
this.commitMedia(token);
return;
}
@@ -429,9 +461,7 @@ export class HostSfuRoute {
this.active.revision = token.revision;
this.clearPending();
if (this.active.active) {
this.commitMedia(token, acknowledge);
} else {
this.active.acknowledgeActive ||= acknowledge;
this.commitMedia(token);
}
return;
}
@@ -441,89 +471,84 @@ export class HostSfuRoute {
!pending ||
pending.revision !== token.revision ||
pending.publicationGeneration !== publicationGeneration ||
!pending.connected
!pending.connected ||
!pending.active
) {
await this.retireActive();
if (acknowledge && this.route.owns(token, "active")) {
if (this.route.owns(token, "active")) {
this.requestRecovery(token.revision);
}
return;
}
await this.retireActive(pending);
if (!this.route.owns(token, "active")) {
this.clearPending(pending);
return;
}
this.pending = null;
const previous = this.active;
this.active = pending;
pending.acknowledgeActive ||= acknowledge;
const publishedStream = this.events.getStream();
const publishedProfile = this.events.getProfile();
if (!publishedStream) {
if (!this.route.markMediaActive(token)) {
this.active = previous;
await disconnectPublisher(pending.publisher);
this.handleFailure(pending);
return;
}
try {
const activated = await pending.publisher.activate(
publishedStream,
publishedProfile,
);
const activeToken = this.currentActiveToken(pending);
if (!activated || !activeToken) {
if (this.active === pending) {
this.active = null;
}
await disconnectPublisher(pending.publisher);
if (!activated && activeToken) {
this.handleFailure(pending);
}
return;
this.lastFailureStage = null;
this.recovery = null;
if (previous && previous !== pending) {
if (previous.active) {
await previous.publisher.deactivate().catch(() => false);
}
pending.active = true;
await disconnectPublisher(previous.publisher);
}
}
private async preparePublisher(slot: HostPublisherSlot): Promise<boolean> {
const stream = this.events.getStream();
const profile = this.events.getProfile();
if (!stream) {
this.handleFailure(slot);
return false;
}
try {
if (!(await slot.publisher.activate(stream, profile))) {
this.handleFailure(slot);
await disconnectPublisher(slot.publisher);
return false;
}
if (this.pending !== slot || !this.ownsPublisherSlot(slot)) {
await disconnectPublisher(slot.publisher);
return false;
}
slot.active = true;
const latestStream = this.events.getStream();
if (
latestStream &&
latestStream !== publishedStream &&
!(await pending.publisher.replaceStream(latestStream))
latestStream !== stream &&
!(await slot.publisher.replaceStream(latestStream))
) {
await disconnectPublisher(pending.publisher);
this.handleFailure(pending);
return;
this.handleFailure(slot);
await disconnectPublisher(slot.publisher);
return false;
}
const latestProfile = this.events.getProfile();
if (
latestProfile !== publishedProfile &&
!(await pending.publisher.updateProfile(latestProfile))
latestProfile !== profile &&
!(await slot.publisher.updateProfile(latestProfile))
) {
await disconnectPublisher(pending.publisher);
this.handleFailure(pending);
return;
this.handleFailure(slot);
await disconnectPublisher(slot.publisher);
return false;
}
return this.pending === slot && this.ownsPublisherSlot(slot);
} catch {
await disconnectPublisher(pending.publisher);
this.handleFailure(pending);
return;
this.handleFailure(slot);
await disconnectPublisher(slot.publisher);
return false;
}
}
const activeToken = this.currentActiveToken(pending);
if (!activeToken) {
if (this.active === pending) {
this.active = null;
}
await disconnectPublisher(pending.publisher);
return;
}
if (!this.route.owns(token, "active")) {
pending.acknowledgeActive = false;
return;
}
const acknowledgeActive = pending.acknowledgeActive;
pending.acknowledgeActive = false;
this.commitMedia(activeToken, acknowledgeActive);
private ownsPublisherSlot(slot: HostPublisherSlot): boolean {
return (
this.route.getRevision() === slot.revision &&
this.route.getPlannedAssignment()?.sfuPublicationGeneration ===
slot.publicationGeneration
);
}
private currentActiveToken(
@@ -542,15 +567,12 @@ export class HostSfuRoute {
return token;
}
private commitMedia(token: RouteOperationToken, acknowledge: boolean): void {
private commitMedia(token: RouteOperationToken): void {
if (!this.route.markMediaActive(token)) {
return;
}
this.lastFailureStage = null;
this.recovery = null;
if (acknowledge) {
this.ready(token.revision, "active");
}
}
private clearPending(expected?: HostPublisherSlot): void {
@@ -648,10 +670,6 @@ export class HostSfuRoute {
});
}
private ready(revision: number, phase: MediaRoutePhase): void {
this.events.send({ type: "route-ready", revision, phase });
}
private handlePublisherStats(
slot: HostPublisherSlot,
metrics: ConnectionMetrics | null,
src/client/media/route-transition.ts
+12
View File
@@ -16,6 +16,18 @@ export interface RouteOperationToken {
}
export type RouteUpdateResult = "accepted" | "duplicate" | "stale";
export type PeerSignalOwner = "pending" | "active";
export function exactPeerSignalOwner(
connectionId: string,
pendingConnectionId: string | null,
activeConnectionId: string | null,
): PeerSignalOwner | null {
if (connectionId === pendingConnectionId) {
return "pending";
}
return connectionId === activeConnectionId ? "active" : null;
}
export class MediaRouteTransition {
private revision = -1;
src/client/media/viewer-sfu-route.ts
+110 -116
View File
@@ -28,6 +28,9 @@ interface ViewerSubscriberSlot {
connected: boolean;
activated: boolean;
mediaAvailable: boolean;
stream: MediaStream | null;
decodedFrame: boolean;
readySent: boolean;
failed: boolean;
activationToken: RouteOperationToken | null;
}
@@ -56,7 +59,6 @@ interface ViewerSfuRouteEvents {
onSfuVideoAvailability?: (available: boolean) => void;
onSfuUpdate?: (metrics: ConnectionMetrics | null) => void;
onSfuState?: (state: "connected" | "reconnecting") => void;
onHealthySfu?: (revision: number) => void;
send: (message: ClientMessage) => boolean;
createSubscriber?: (
events: {
@@ -74,12 +76,11 @@ export class ViewerSfuRoute {
private pending: ViewerSubscriberSlot | null = null;
private active: ViewerSubscriberSlot | null = null;
private recovery: { revision: number; refreshed: boolean } | null = null;
private peerProbeRevision: number | null = null;
private failedPeerProbe: { revision: number; parentPeerId: string; connectionId: string } | null = null;
private healthySfuWindows: number | null = null;
private pendingPeerRevision: number | null = null;
private transitionTail: Promise<void> = Promise.resolve();
private resyncGeneration = 0;
private resyncing = false;
private paused = false;
private closed = false;
constructor(private readonly events: ViewerSfuRouteEvents) {}
@@ -93,26 +94,43 @@ export class ViewerSfuRoute {
if (result === "stale") {
return result;
}
const rearmHealthySfu =
result === "duplicate" &&
update.phase === "active" &&
update.assignment.upstream.kind === "sfu";
if (this.paused) {
this.discardPending();
return result;
}
if (result === "duplicate") {
return result;
}
if (previousRevision !== update.revision) {
this.recovery = null;
this.peerProbeRevision = null;
this.failedPeerProbe = null;
this.healthySfuWindows = null;
if (
update.phase !== "prepare" &&
this.pendingPeerRevision !== null
) {
this.events.preparePeer?.(null);
this.pendingPeerRevision = null;
}
}
if (update.phase === "prepare") {
this.events.prepareChild?.(update.assignment.childPeerIds, update.revision);
const mediaUpstream = this.route.getMediaAssignment()?.upstream;
if (
const needsPeerCandidate =
update.assignment.upstream.kind === "peer" &&
(mediaUpstream?.kind !== "peer" ||
mediaUpstream.peerId !== update.assignment.upstream.peerId)
) {
this.peerProbeRevision = update.revision;
mediaUpstream.peerId !== update.assignment.upstream.peerId);
const replacedPeerCandidate = this.pendingPeerRevision !== null;
if (replacedPeerCandidate) {
this.pendingPeerRevision = null;
this.events.preparePeer?.(null);
}
if (needsPeerCandidate && update.assignment.upstream.kind === "peer") {
this.pendingPeerRevision = update.revision;
this.events.preparePeer?.(update.assignment, update.revision);
} else if (
update.assignment.upstream.kind !== "peer" &&
!replacedPeerCandidate
) {
this.events.preparePeer?.(null);
}
if (this.pending?.revision !== update.revision) {
this.clearPending();
@@ -127,26 +145,52 @@ export class ViewerSfuRoute {
if (!token) {
return result;
}
if (this.peerProbeRevision !== update.revision) {
if (this.pendingPeerRevision !== update.revision) {
this.preparePeer(update.assignment);
}
if (this.resyncing) {
return result;
}
void this.queueActiveRoute(token, acknowledge);
if (rearmHealthySfu) {
this.armHealthySfuReselection(update.revision);
}
return result;
}
beginSelectedPeerCandidate(revision: number, parentPeerId: string): boolean {
const assignment = this.route.getPlannedAssignment();
if (
this.paused ||
this.route.getRevision() !== revision ||
this.route.getPhase() !== "prepare" ||
assignment?.upstream.kind !== "peer" ||
assignment.upstream.peerId !== parentPeerId
) {
return false;
}
this.pendingPeerRevision = revision;
return true;
}
setPaused(paused: boolean): void {
this.paused = paused;
if (paused) {
this.discardPending();
}
}
private discardPending(): void {
this.pendingPeerRevision = null;
this.events.preparePeer?.(null);
this.events.prepareChild?.(null);
this.clearPending();
}
async resyncAuthoritative(
update: RouteUpdateInput,
): Promise<RouteUpdateResult> {
if (this.closed) {
return "stale";
}
this.peerProbeRevision = this.failedPeerProbe = null;
this.pendingPeerRevision = null;
if (!this.active && !this.pending) {
const result = this.accept(update, false);
if (result !== "stale") {
@@ -158,7 +202,6 @@ export class ViewerSfuRoute {
this.resyncing = true;
this.route.reset();
this.recovery = null;
this.healthySfuWindows = null;
this.events.prepareChild?.(null);
const pending = this.pending;
const active = this.active;
@@ -204,7 +247,7 @@ export class ViewerSfuRoute {
async acceptConfig(
message: Extract<ServerMessage, { type: "sfu-config" }>,
): Promise<void> {
if (this.closed || !this.route.acceptsConfig(message.revision)) {
if (this.closed || this.paused || !this.route.acceptsConfig(message.revision)) {
return;
}
const token = this.route.token();
@@ -215,15 +258,9 @@ export class ViewerSfuRoute {
}
if (this.active?.revision === message.revision && this.active.activated) {
if (phase === "prepare") {
this.ready(message.revision, "prepare");
}
return;
}
if (this.pending?.revision === message.revision && !this.pending.failed) {
if (this.pending.connected && phase === "prepare") {
this.ready(message.revision, "prepare");
}
return;
}
@@ -239,29 +276,30 @@ export class ViewerSfuRoute {
this.handleVideoAvailability(slot, available);
},
onStats: (metrics: ConnectionMetrics) => {
if (this.active === slot && !slot.failed) {
this.events.onSfuUpdate?.(metrics);
if (this.healthySfuWindows !== null && this.healthySfuWindows >= 0) {
if (this.pending === slot && !slot.failed) {
if ((metrics.intervalFramesDecoded ?? 0) > 0) {
slot.decodedFrame = true;
if (
(metrics.intervalPacketsReceived ?? 0) <= 0 ||
(metrics.intervalFramesDecoded ?? 0) <= 0
phase === "prepare" &&
this.route.getRevision() === slot.revision &&
this.route.getPhase() === "prepare" &&
!slot.readySent
) {
this.healthySfuWindows = null;
} else if (++this.healthySfuWindows === 2) {
this.healthySfuWindows = null;
this.events.onHealthySfu?.(slot.revision);
slot.readySent = this.events.send({
type: "route-ready",
revision: slot.revision,
phase: "prepare",
});
}
void this.queueTransition(() => this.promotePendingSfu(slot));
}
} else if (this.active === slot && !slot.failed) {
this.events.onSfuUpdate?.(metrics);
}
},
onState: (state: "connected" | "reconnecting") => {
if (this.active === slot && !slot.failed) {
this.events.onSfuState?.(state);
if (state === "reconnecting") {
this.healthySfuWindows = -1;
} else if (this.healthySfuWindows === -1) {
this.healthySfuWindows = 0;
}
}
},
onDisconnected: () => this.handleFailure(slot),
@@ -275,6 +313,9 @@ export class ViewerSfuRoute {
connected: false,
activated: false,
mediaAvailable: false,
stream: null,
decodedFrame: false,
readySent: false,
failed: false,
activationToken: null,
};
@@ -296,9 +337,13 @@ export class ViewerSfuRoute {
return;
}
slot.connected = true;
if (phase === "prepare") {
this.ready(message.revision, "prepare");
} else if (!this.resyncing) {
if (!subscriber.activate()) {
this.handleFailure(slot);
await disconnectSubscriber(subscriber);
return;
}
slot.activated = true;
if (phase === "active" && !this.resyncing) {
await this.queueActiveRoute(token, true);
}
} catch {
@@ -316,25 +361,6 @@ export class ViewerSfuRoute {
);
}
reportPeerProbeFailure(parentPeerId: string, connectionId: string, ready: boolean): boolean {
const revision = this.peerProbeRevision, phase = this.route.getPhase();
const upstream = this.route.getPlannedAssignment()?.upstream;
if (revision === null || phase === null ||
upstream?.kind !== "peer" || upstream.peerId !== parentPeerId) return false;
if (ready) this.failedPeerProbe = { revision, parentPeerId, connectionId };
this.events.send({ type: "route-failed", revision, phase, connectionId });
return true;
}
armHealthySfuReselection(revision: number): void {
if (
this.route.getRevision() === revision &&
this.route.getPlannedAssignment()?.upstream.kind === "sfu"
) {
this.healthySfuWindows = 0;
}
}
async disconnect(): Promise<void> {
if (this.closed) {
return;
@@ -344,9 +370,7 @@ export class ViewerSfuRoute {
this.resyncing = false;
this.route.reset();
this.recovery = null;
this.peerProbeRevision = null;
this.failedPeerProbe = null;
this.healthySfuWindows = null;
this.pendingPeerRevision = null;
this.events.preparePeer?.(null);
this.events.prepareChild?.(null);
await this.queueTransition(async () => {
@@ -399,31 +423,22 @@ export class ViewerSfuRoute {
) {
this.clearPending();
this.active.revision = token.revision;
this.commitMedia(token, acknowledge);
this.commitMedia(token);
return;
}
await this.activateSfu(token, acknowledge);
return;
}
const probeRevision = this.peerProbeRevision;
const failedProbe = this.failedPeerProbe;
if (failedProbe?.revision === token.revision &&
assignment.upstream.kind === "peer" &&
assignment.upstream.peerId === failedProbe.parentPeerId) {
this.failedPeerProbe = null;
this.events.send({ type: "route-failed", revision: token.revision,
phase: "active", connectionId: failedProbe.connectionId });
return;
}
const pendingPeerRevision = this.pendingPeerRevision;
const promoted =
probeRevision === token.revision
pendingPeerRevision === token.revision
? await this.events.activatePeer(assignment, token.revision)
: false;
if (probeRevision === token.revision && promoted !== true) {
if (pendingPeerRevision === token.revision && promoted !== true) {
return;
}
this.peerProbeRevision = null;
this.pendingPeerRevision = null;
this.clearPending();
if (this.active) {
await this.retireActive();
@@ -438,7 +453,7 @@ export class ViewerSfuRoute {
return;
}
this.activateChildren(assignment.childPeerIds, token.revision);
this.commitMedia(token, acknowledge);
this.commitMedia(token);
});
}
@@ -471,66 +486,57 @@ export class ViewerSfuRoute {
return;
}
pending.activationToken = token;
try {
if (!pending.subscriber.activate()) {
await disconnectSubscriber(pending.subscriber);
this.handleFailure(pending);
return;
}
pending.activated = true;
} catch {
await disconnectSubscriber(pending.subscriber);
this.handleFailure(pending);
}
await this.promotePendingSfu(pending);
}
private handleStream(slot: ViewerSubscriberSlot, stream: MediaStream): void {
if (this.closed || slot.failed || !slot.activated) {
return;
}
slot.stream = stream;
if (this.active === slot) {
const assignment = this.route.getMediaAssignment();
if (assignment?.upstream.kind === "sfu") {
const recovered = !slot.mediaAvailable;
slot.mediaAvailable = true;
this.events.onSfuStream(stream, assignment, false);
if (recovered) {
this.ready(slot.revision, "active");
}
}
return;
}
void this.queueTransition(() => this.promotePendingSfu(slot));
}
private async promotePendingSfu(slot: ViewerSubscriberSlot): Promise<void> {
const token = slot.activationToken;
const assignment = this.route.getActiveAssignment();
if (
this.pending !== slot ||
!slot.stream ||
!slot.decodedFrame ||
!token ||
!this.route.owns(token, "active") ||
assignment?.upstream.kind !== "sfu"
) {
return;
}
this.pending = null;
const previous = this.active;
this.active = slot;
if (!this.route.markMediaActive(token)) {
this.active = previous;
void disconnectSubscriber(slot.subscriber);
await disconnectSubscriber(slot.subscriber);
return;
}
this.recovery = null;
slot.mediaAvailable = true;
this.events.onSfuStream(stream, assignment, true);
this.events.onSfuStream(slot.stream, assignment, true);
if (previous && previous !== slot) {
try {
previous.subscriber.deactivate();
} catch {
// Disconnect remains the fail-closed cleanup path.
}
void disconnectSubscriber(previous.subscriber);
await disconnectSubscriber(previous.subscriber);
}
this.ready(token.revision, "active");
}
private handleVideoAvailability(
@@ -552,9 +558,7 @@ export class ViewerSfuRoute {
if (!changed) {
return;
}
if (available) {
this.ready(slot.revision, "active");
} else {
if (!available) {
this.events.send({
type: "route-media-unavailable",
revision: slot.revision,
@@ -562,18 +566,11 @@ export class ViewerSfuRoute {
}
}
private commitMedia(token: RouteOperationToken, acknowledge: boolean): void {
private commitMedia(token: RouteOperationToken): void {
if (!this.route.markMediaActive(token)) {
return;
}
this.recovery = null;
const assignment = this.route.getMediaAssignment();
if (
acknowledge &&
(assignment?.upstream.kind !== "sfu" || this.active?.mediaAvailable === true)
) {
this.ready(token.revision, "active");
}
}
private clearPending(): void {
@@ -658,9 +655,6 @@ export class ViewerSfuRoute {
});
}
private ready(revision: number, phase: MediaRoutePhase): void {
this.events.send({ type: "route-ready", revision, phase });
}
}
async function disconnectSubscriber(
src/client/pages/HostPage.tsx
+21 -133
View File
@@ -92,11 +92,7 @@ import {
type HostSfuPublisherSnapshot,
} from "../media/host-sfu-route";
import {
acceptHostSelectedChildrenRevision,
carryHostSelectedChild,
hostSelectedChildForConnection,
HostProvisionalChild,
type ActiveSelectedHostChild,
} from "../media/host-provisional-child";
import {
ParentEdgeQualityEvidenceReporter,
@@ -134,18 +130,10 @@ type ViewerQualityEvidence = Extract<
ServerMessage,
{ type: "viewer-quality-evidence" }
>;
type SelectedEdgeTurn = Extract<
ServerMessage,
{ type: "selected-edge-turn"; edgeKind: "peer-selected" }
>;
type HostRouteAssignment = Extract<
ServerMessage,
{ type: "route-update" }
>["assignment"];
interface FailedActiveHostChild {
revision: number;
peerId: string;
}
interface CaptureDetails {
resolution: string;
@@ -282,11 +270,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
const iceConfigRef = useRef<IceConfig | null>(null);
const peersRef = useRef(new Map<string, HostPeer>());
const retiredConnectionsRef = useRef(new Map<string, string>());
const selectedHostChildrenRef = useRef(
new Map<string, ActiveSelectedHostChild>(),
);
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(
@@ -424,10 +408,8 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
peersRef.current.forEach((peer) => peer.dispose());
peersRef.current.clear();
retiredConnectionsRef.current.clear();
selectedHostChildrenRef.current.clear();
hostProvisionalChildRef.current?.discard();
hostProvisionalChildRef.current = null;
failedActiveHostChildRef.current = null;
activeHostChildPeerIdsRef.current = [];
hostPeerIdRef.current = null;
viewerQualityEvidenceTimersRef.current.forEach((timer) =>
@@ -541,10 +523,8 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
peersRef.current.forEach((peer) => peer.dispose());
peersRef.current.clear();
retiredConnectionsRef.current.clear();
selectedHostChildrenRef.current.clear();
hostProvisionalChildRef.current?.discard();
hostProvisionalChildRef.current = null;
failedActiveHostChildRef.current = null;
activeHostChildPeerIdsRef.current = [];
hostPeerIdRef.current = null;
void hostSfuRouteRef.current?.disconnect();
@@ -878,6 +858,10 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
}
sharingPausedRef.current = nextPaused;
setSharingPaused(nextPaused);
hostSfuRouteRef.current?.setPaused(nextPaused);
if (nextPaused) {
discardPreparedHostChild();
}
signalRef.current?.setSharingPaused(nextPaused);
setNotice(nextPaused ? "音视频分享已暂停" : "音视频分享已恢复");
}
@@ -908,7 +892,6 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
retiredConnectionsRef.current.set(peerId, peer.connectionId);
peer.dispose();
}
selectedHostChildrenRef.current.delete(peerId);
peersRef.current.delete(peerId);
setPeerSnapshots((current) => {
const next = new Map(current);
@@ -945,7 +928,8 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
isCurrentGeneration(generation) && signalRef.current === signal
? signal.send({ type: "signal", targetPeerId, payload })
: false,
hasActivePeer: (peerId) => peersRef.current.has(peerId),
activeConnectionId: (peerId) =>
peersRef.current.get(peerId)?.connectionId ?? null,
onPromotedStreamFailure: (peer) => {
if (
isCurrentGeneration(generation) &&
@@ -967,7 +951,6 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
revision,
assignment,
activeChildPeerIds: activeHostChildPeerIdsRef.current,
selectedPeerIds: selectedHostChildrenRef.current.keys(),
maxMediaEdges: MAX_ENDPOINT_MEDIA_CHILDREN,
iceConfig,
stream,
@@ -986,26 +969,16 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
revision,
assignment,
activeChildPeerIds,
selectedPeerIds: selectedHostChildrenRef.current.keys(),
maxMediaEdges: MAX_ENDPOINT_MEDIA_CHILDREN,
}) ?? { kind: "ordinary" as const };
if (activation.kind === "promote") {
hostProvisionalChildRef.current = null;
failedActiveHostChildRef.current = null;
const previous = peersRef.current.get(activation.peerId);
peersRef.current.set(activation.peerId, activation.peer);
previous?.dispose();
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;
}
@@ -1014,7 +987,6 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
peerId: string,
generation: number,
attempt = 0,
selectedTurn?: SelectedEdgeTurn,
): Promise<void> {
if (!isCurrentGeneration(generation)) {
return;
@@ -1023,7 +995,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
// negotiation whose offer or answer may have been lost with the WebSocket.
const existing = peersRef.current.get(peerId);
if (existing) {
if (!selectedTurn && existing.isConnected()) {
if (existing.isConnected()) {
return;
}
removePeer(peerId);
@@ -1035,32 +1007,10 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
return;
}
const reportSelectedTurnFailure = (connectionId: string): void => {
const selected = hostSelectedChildForConnection(
selectedHostChildrenRef.current,
peerId,
connectionId,
);
if (
!selectedTurn ||
!selected ||
!isCurrentGeneration(generation) ||
signalRef.current !== signal
) {
return;
}
signal.send({
type: "route-failed",
revision: selected.currentRouteRevision,
phase: "active",
connectionId,
});
};
let peer: HostPeer;
peer = new HostPeer(
peerId,
selectedTurn ? { iceServers: [selectedTurn.iceServer] } : iceConfig,
iceConfig,
activeStream,
qualitySettingsRef.current,
{
@@ -1073,17 +1023,10 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
isCurrentGeneration(generation) &&
peersRef.current.get(peerId) === peer
) {
if (selectedTurn && snapshot.connectionState === "failed") {
reportSelectedTurnFailure(peer.connectionId);
removePeer(peerId);
return;
}
updatePeerSnapshot(snapshot);
}
},
},
selectedTurn !== undefined,
selectedTurn?.newConnectionId,
);
peersRef.current.set(peerId, peer);
let started: boolean;
@@ -1091,7 +1034,6 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
started = await peer.start();
} catch (error) {
if (peersRef.current.get(peerId) === peer) {
reportSelectedTurnFailure(peer.connectionId);
removePeer(peerId);
}
throw error;
@@ -1100,9 +1042,8 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
return;
}
reportSelectedTurnFailure(peer.connectionId);
removePeer(peerId);
if (selectedTurn || !isCurrentGeneration(generation) || attempt >= 1) {
if (!isCurrentGeneration(generation) || attempt >= 1) {
return;
}
window.setTimeout(() => {
@@ -1162,28 +1103,9 @@ 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 selectedPeerIds: string[] = [];
for (const [peerId, selected] of selectedHostChildrenRef.current) {
const selectedIsCurrent =
selected.currentRouteRevision === activeRouteRevisionRef.current &&
peersRef.current.get(peerId)?.connectionId === selected.connectionId;
if (selectedIsCurrent) {
selectedPeerIds.push(peerId);
} else {
selectedHostChildrenRef.current.delete(peerId);
}
}
const effectiveChildPeerIds = [
...new Set([...authoritativeChildPeerIds, ...selectedPeerIds]),
];
reconcileBoundedMediaChildren(
peersRef.current.keys(),
effectiveChildPeerIds,
childPeerIds,
MAX_ENDPOINT_MEDIA_CHILDREN,
removePeer,
(peerId) => {
@@ -1206,10 +1128,6 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
}
if (message.type === "authenticated" && message.role === "host") {
discardPreparedHostChild();
failedActiveHostChildRef.current = null;
for (const peerId of [...selectedHostChildrenRef.current.keys()]) {
removePeer(peerId);
}
hostPeerIdRef.current = message.peerId;
setViewerAccessUpdating(false);
clearAllViewerQualityEvidence();
@@ -1275,40 +1193,19 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
}
return;
}
const current = peersRef.current.get(message.viewerPeerId);
const iceConfig = iceConfigRef.current;
const activeStream = streamRef.current;
if (
peerAssistedRef.current &&
message.parentPeerId === hostPeerIdRef.current &&
current?.connectionId === message.newConnectionId &&
carryHostSelectedChild(
selectedHostChildrenRef.current,
message.viewerPeerId,
message.newConnectionId,
message.revision,
)
iceConfig &&
activeStream
) {
return;
}
const oldConnectionId =
current?.connectionId ??
retiredConnectionsRef.current.get(message.viewerPeerId);
if (
peerAssistedRef.current &&
message.parentPeerId === hostPeerIdRef.current &&
message.revision === activeRouteRevisionRef.current &&
Date.parse(message.expiresAt) > Date.now() &&
oldConnectionId === message.oldConnectionId
) {
if (current) {
removePeer(message.viewerPeerId);
}
selectedHostChildrenRef.current.set(message.viewerPeerId, {
peerId: message.viewerPeerId,
connectionId: message.newConnectionId,
currentRouteRevision: message.revision,
pendingCarryRevision: null,
hostProvisionalChildRef.current?.prepareSelectedTurn(message, {
iceConfig,
stream: activeStream,
profile: qualitySettingsRef.current,
});
void startPeer(message.viewerPeerId, generation, 0, message);
}
return;
}
@@ -1357,14 +1254,6 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
} else {
activatePreparedHostChild(message.revision, message.assignment);
}
if (message.phase === "active") {
for (const peerId of acceptHostSelectedChildrenRevision(
selectedHostChildrenRef.current,
message.revision,
)) {
removePeer(peerId);
}
}
if (
message.phase === "active" &&
message.revision !== activeRouteRevisionRef.current
@@ -1392,10 +1281,9 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
}
return;
}
if (message.type === "media-assignment") {
if (message.type === "media-assignment") {
if (peerAssistedRef.current && !hostSfuRouteRef.current) {
discardPreparedHostChild();
failedActiveHostChildRef.current = null;
activeHostChildPeerIdsRef.current = [
...message.mediaAssignment.childPeerIds,
];
src/client/pages/ViewerPage.tsx
+166 -211
View File
@@ -60,6 +60,7 @@ import {
ViewerQualityEvidenceReporter,
} from "../media/viewer-quality-evidence";
import { ViewerMessageAuthority } from "../media/viewer-message-authority";
import { exactPeerSignalOwner } from "../media/route-transition";
import { ViewerSfuRoute } from "../media/viewer-sfu-route";
import {
applyViewerVolume,
@@ -75,7 +76,6 @@ import type {
import {
limitMediaAssignment,
MAX_ENDPOINT_MEDIA_CHILDREN,
retainSelectedMediaParent,
viewerRestartMessage,
viewerSignalMessage,
} from "../webrtc/media-assignment";
@@ -95,20 +95,17 @@ type SelectedEdgeTurn = Extract<
ServerMessage,
{ type: "selected-edge-turn"; edgeKind: "peer-selected" }
>;
interface ActiveSelectedEdgeTurn {
grant: SelectedEdgeTurn;
currentRouteRevision: number;
pendingCarryRevision: number | null;
}
interface SfuUpstreamState {
connectionState: "connected" | "reconnecting";
metrics: ConnectionMetrics | null;
}
interface PeerProbe {
interface PendingPeerRoute {
revision: number;
parentPeerId: string;
peer: ViewerPeer | null;
ready: boolean;
readySent: boolean;
selectedTurn: SelectedEdgeTurn | null;
candidateConnectionId: string | null;
stream: MediaStream | null;
snapshot: PeerSnapshot | null;
}
@@ -272,12 +269,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
let preparedParentSignals: Array<
Extract<ServerMessage, { type: "signal" }>
> = [];
let peerProbe: PeerProbe | null = null;
let retiredUpstream: {
parentPeerId: string;
connectionId: string;
} | null = null;
let selectedEdgeTurn: ActiveSelectedEdgeTurn | null = null;
let pendingPeer: PendingPeerRoute | null = null;
const messageAuthority = new ViewerMessageAuthority();
let sfuStandbyPrewarmer: SfuStandbyPrewarmer | null = null;
let relayChildEvidenceCurrent: ViewerQualityEvidencePresentation | null =
@@ -438,20 +430,6 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
payload,
})
: false,
onSelectedEdgeFailed: (
_childPeerId,
connectionId,
revision,
) => {
if (active && peerAssisted) {
signal.send({
type: "route-failed",
revision,
phase: "active",
connectionId,
});
}
},
onUpdate: (snapshot) => {
if (active) {
setRelaySnapshot(snapshot);
@@ -501,35 +479,31 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
}
}
function discardPeerProbe(): void {
const probe = peerProbe;
peerProbe = null;
function discardPendingPeer(): void {
const probe = pendingPeer;
pendingPeer = null;
prepareParent(null);
probe?.peer?.dispose();
}
function peerProbeHasCurrentMedia(
probe: PeerProbe,
): probe is PeerProbe & { peer: ViewerPeer; stream: MediaStream; snapshot: PeerSnapshot } {
const { snapshot, stream } = probe;
function pendingPeerHasDecodedFrame(
probe: PendingPeerRoute,
): probe is PendingPeerRoute & { peer: ViewerPeer; snapshot: PeerSnapshot } {
const { snapshot } = probe;
return (
snapshot !== null &&
stream !== null &&
snapshot.peerId === probe.parentPeerId &&
probe.peer?.hasConnectionId(snapshot.connectionId) === true &&
snapshot.connectionState === "connected" &&
(snapshot.metrics.intervalPacketsReceived ?? 0) > 0 &&
(snapshot.metrics.intervalFramesDecoded ?? 0) > 0 &&
stream.getVideoTracks().some((track) => track.readyState === "live")
(snapshot.metrics.intervalFramesDecoded ?? 0) > 0
);
}
function provePeerProbe(): void {
const probe = peerProbe;
if (!probe || probe.ready || !peerProbeHasCurrentMedia(probe)) {
function provePendingPeer(): void {
const probe = pendingPeer;
if (!probe || probe.readySent || !pendingPeerHasDecodedFrame(probe)) {
return;
}
probe.ready = signal.send({
probe.readySent = signal.send({
type: "route-ready",
revision: probe.revision,
phase: "prepare",
@@ -547,28 +521,22 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
return;
}
if (revision !== undefined) {
const probe = peerProbe;
const probe = pendingPeer;
if (
!probe ||
!probe.ready ||
!probe.readySent ||
probe.revision !== revision ||
assignment.upstream.kind !== "peer" ||
assignment.upstream.peerId !== probe.parentPeerId ||
!peerProbeHasCurrentMedia(probe)
!probe.peer ||
!probe.stream ||
!probe.snapshot
) {
if (probe?.ready && probe.snapshot) {
signal.send({
type: "route-failed",
revision,
phase: "active",
connectionId: probe.snapshot.connectionId,
});
}
discardPeerProbe();
discardPendingPeer();
return false;
}
const previousPeer = peerRef.current;
peerProbe = null;
pendingPeer = null;
prepareParent(null);
peerRef.current = probe.peer;
previousPeer?.dispose();
@@ -601,13 +569,15 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
if (!active || viewerSfuRoute !== route) {
return;
}
discardPeerProbe();
discardPendingPeer();
if (revision !== undefined && assignment?.upstream.kind === "peer") {
peerProbe = {
pendingPeer = {
revision,
parentPeerId: assignment.upstream.peerId,
peer: null,
ready: false,
readySent: false,
selectedTurn: null,
candidateConnectionId: null,
stream: null,
snapshot: null,
};
@@ -666,8 +636,6 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
);
}
},
onHealthySfu: (revision) =>
signal.send({ type: "sfu-reselection-ready", revision }),
onSfuVideoAvailability: (available) => {
if (!active || viewerSfuRoute !== route || available) {
return;
@@ -697,7 +665,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
);
setStatusText("正在播放");
if (initialVideoStream) {
if (!peerProbe) {
if (!pendingPeer) {
peerRef.current?.dispose();
peerRef.current = null;
setPeerSnapshot(null);
@@ -713,7 +681,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
function clearViewerSfuRoute(): void {
const route = viewerSfuRoute;
discardPeerProbe();
discardPendingPeer();
viewerSfuRoute = null;
setSfuUpstream(null);
void route?.disconnect();
@@ -751,7 +719,6 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
function clearUpstreamState(): void {
qualityEvidenceReporter.reset();
const peer = peerRef.current;
retiredUpstream = peer?.getConnectionIdentity() ?? retiredUpstream;
peer?.dispose();
peerRef.current = null;
setRemoteStream(null);
@@ -781,49 +748,66 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
reconcileRelayChildren(previousChildPeerIds);
}
function ensurePeerProbe(): ViewerPeer | null {
const probe = peerProbe;
function ensurePendingPeerRoute(): ViewerPeer | null {
const probe = pendingPeer;
if (!probe || !currentIceConfig) return null;
if (probe.peer) return probe.peer;
const peer: ViewerPeer = new ViewerPeer(currentIceConfig, {
sendSignal: (targetPeerId, payload) =>
signal.send(viewerSignalMessage(peerAssisted, targetPeerId, payload)),
sendRestartRequest: () => false,
onStream: (stream) => {
if (peerProbe === probe) {
probe.stream = stream;
provePeerProbe();
} else if (active && peerRef.current === peer) {
setRemoteStream(stream);
ensureViewerRelay()?.setStream(stream);
setStatusText("正在播放");
}
const peer: ViewerPeer = new ViewerPeer(
probe.selectedTurn
? { iceServers: [probe.selectedTurn.iceServer] }
: currentIceConfig,
{
sendSignal: (targetPeerId, payload) =>
signal.send(viewerSignalMessage(peerAssisted, targetPeerId, payload)),
sendRestartRequest: () => false,
onStream: (stream) => {
if (pendingPeer === probe) {
probe.stream = stream;
provePendingPeer();
} else if (active && peerRef.current === peer) {
setRemoteStream(stream);
ensureViewerRelay()?.setStream(stream);
setStatusText("正在播放");
}
},
onUpdate: (snapshot) => {
if (pendingPeer === probe) {
probe.snapshot = snapshot;
probe.candidateConnectionId = snapshot.connectionId;
provePendingPeer();
} else if (active && peerRef.current === peer) {
qualityEvidenceReporter.offer(snapshot, currentRouteRevision);
setPeerSnapshot(snapshot);
setStatusText(
snapshot.connectionState === "connected"
? "已连接"
: "正在恢复连接",
);
}
},
onRecoveryExhausted: (parentPeerId, connectionId): boolean => {
if (pendingPeer === probe && peer.hasConnectionId(connectionId)) {
signal.send({
type: "route-failed",
revision: probe.revision,
phase: "prepare",
connectionId,
});
probe.candidateConnectionId = connectionId;
probe.peer = null;
probe.stream = null;
probe.snapshot = null;
probe.readySent = false;
peer.dispose();
return true;
}
return peerRef.current === peer && viewerSfuRoute
? viewerSfuRoute.reportPeerFailure(parentPeerId, connectionId)
: true;
},
},
onUpdate: (snapshot) => {
if (peerProbe === probe) {
probe.snapshot = snapshot;
provePeerProbe();
} else if (active && peerRef.current === peer) {
qualityEvidenceReporter.offer(snapshot, currentRouteRevision);
setPeerSnapshot(snapshot);
setStatusText(
snapshot.connectionState === "connected"
? "已连接"
: "正在恢复连接",
);
}
},
onRecoveryExhausted: (parentPeerId, connectionId): boolean => {
if (peerProbe === probe && peer.hasConnectionId(connectionId)) {
viewerSfuRoute?.reportPeerProbeFailure(parentPeerId, connectionId, probe.ready);
discardPeerProbe();
return true;
}
return peerRef.current === peer && viewerSfuRoute
? viewerSfuRoute.reportPeerFailure(parentPeerId, connectionId)
: true;
},
});
probe.selectedTurn !== null,
);
probe.peer = peer;
return peer;
}
@@ -836,9 +820,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
return null;
}
const peer = new ViewerPeer(
selectedEdgeTurn
? { iceServers: [selectedEdgeTurn.grant.iceServer] }
: currentIceConfig,
currentIceConfig,
{
sendSignal: (targetPeerId, payload) =>
signal.send(
@@ -885,21 +867,6 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
}
},
onRecoveryExhausted: (parentPeerId, connectionId) => {
if (
selectedEdgeTurn?.grant.parentPeerId === parentPeerId &&
selectedEdgeTurn.grant.newConnectionId === connectionId
) {
signal.send({
type: "route-failed",
revision: selectedEdgeTurn.currentRouteRevision,
phase: "active",
connectionId,
});
selectedEdgeTurn = null;
clearUpstreamState();
setStatusText("无法建立媒体连接");
return true;
}
if (viewerSfuRoute) {
return viewerSfuRoute.reportPeerFailure(parentPeerId, connectionId);
}
@@ -907,7 +874,6 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
return true;
},
},
selectedEdgeTurn !== null,
);
peerRef.current = peer;
return peer;
@@ -918,12 +884,6 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
authorityToken: number,
): Promise<void> {
if (message.type === "authenticated") {
const hadSelectedUpstream = selectedEdgeTurn !== null;
selectedEdgeTurn = null;
if (hadSelectedUpstream) {
clearUpstreamState();
}
viewerRelay?.clearSelectedEdgeTurn();
viewerAuthenticated = true;
setAccessState("ready");
setViewerPasswordDraft("");
@@ -950,7 +910,6 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
? message.routeRevision
: 0;
if (nextRouteRevision !== currentRouteRevision) {
viewerRelay?.clearSelectedEdgeTurn();
clearRelayChildEvidence();
}
currentRouteRevision = nextRouteRevision;
@@ -970,11 +929,14 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
currentIceConfig = message.iceConfig;
currentHostOnline = message.hostOnline;
setHostOnline(message.hostOnline);
setHostPaused(message.hostPaused ?? false);
const sharingPaused = message.hostPaused ?? false;
setHostPaused(sharingPaused);
if (nextPeerAssisted && "qualitySettings" in message) {
currentQualitySettings = message.qualitySettings;
void viewerRelay?.updateProfile(currentQualitySettings);
await ensureViewerSfuRoute().resyncAuthoritative(
const route = ensureViewerSfuRoute();
route.setPaused(sharingPaused);
await route.resyncAuthoritative(
{
revision: message.routeRevision,
phase: "active",
@@ -984,9 +946,6 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
if (!active || !messageAuthority.owns(authorityToken)) {
return;
}
if (message.routeAssignment.upstream.kind === "sfu") {
ensureViewerSfuRoute().armHealthySfuReselection(message.routeRevision);
}
}
if (
!message.hostOnline &&
@@ -1031,54 +990,61 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
return;
}
if (message.parentPeerId === currentPeerId) {
ensureViewerRelay()?.startSelectedEdgeTurn(
ensureViewerRelay()?.prepareSelectedEdgeTurn(
message,
currentPeerId,
currentRouteRevision,
message.revision,
);
return;
}
let probe = pendingPeer;
if (
selectedEdgeTurn?.grant.parentPeerId === message.parentPeerId &&
selectedEdgeTurn.grant.viewerPeerId === message.viewerPeerId &&
selectedEdgeTurn.grant.newConnectionId === message.newConnectionId &&
message.revision >= selectedEdgeTurn.currentRouteRevision
) {
selectedEdgeTurn.currentRouteRevision = message.revision;
selectedEdgeTurn.pendingCarryRevision = message.revision;
return;
}
if (
message.revision !== currentRouteRevision ||
Date.parse(message.expiresAt) <= Date.now()
!viewerSfuRoute?.beginSelectedPeerCandidate(
message.revision,
message.parentPeerId,
)
) {
return;
}
const currentIdentity =
peerRef.current?.getConnectionIdentity() ?? retiredUpstream;
if (!probe) {
probe = pendingPeer = {
revision: message.revision,
parentPeerId: message.parentPeerId,
peer: null,
readySent: false,
selectedTurn: null,
candidateConnectionId: null,
stream: null,
snapshot: null,
};
prepareParent(message.parentPeerId);
}
if (probe.selectedTurn?.newConnectionId === message.newConnectionId) {
return;
}
const activeIdentity = peerRef.current?.getConnectionIdentity();
const oldConnectionId =
probe?.candidateConnectionId ??
probe?.peer?.getConnectionIdentity()?.connectionId ??
(activeIdentity?.parentPeerId === message.parentPeerId
? activeIdentity.connectionId
: null);
if (
message.viewerPeerId !== currentPeerId ||
currentIdentity?.parentPeerId !== message.parentPeerId ||
currentIdentity.connectionId !== message.oldConnectionId
probe?.revision !== message.revision ||
probe.parentPeerId !== message.parentPeerId ||
Date.parse(message.expiresAt) <= Date.now() ||
oldConnectionId !== message.oldConnectionId
) {
return;
}
selectedEdgeTurn = {
grant: message,
currentRouteRevision: message.revision,
pendingCarryRevision: null,
};
clearViewerSfuRoute();
clearUpstreamState();
acceptAssignedRoute(message.revision, {
kind: "peer",
peerId: message.parentPeerId,
});
currentAssignment = {
parentPeerId: message.parentPeerId,
childPeerIds: currentAssignment.childPeerIds,
};
setStatusText("正在恢复连接");
probe.peer?.dispose();
probe.peer = null;
probe.readySent = false;
probe.selectedTurn = message;
probe.candidateConnectionId = null;
probe.stream = null;
probe.snapshot = null;
return;
}
if (message.type === "route-update") {
@@ -1087,39 +1053,10 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
message.phase === "active" &&
message.revision >= currentRouteRevision
) {
viewerRelay?.acceptActiveRevision(message.revision);
if (selectedEdgeTurn) {
if (
selectedEdgeTurn.pendingCarryRevision === message.revision
) {
selectedEdgeTurn.pendingCarryRevision = null;
} else {
selectedEdgeTurn = null;
clearUpstreamState();
}
}
if (message.revision !== currentRouteRevision) {
clearRelayChildEvidence();
}
currentRouteRevision = message.revision;
if (selectedEdgeTurn) {
applyMediaAssignment(
{
parentPeerId: selectedEdgeTurn.grant.parentPeerId,
childPeerIds: message.assignment.childPeerIds,
},
true,
);
acceptAssignedRoute(
message.revision,
{
kind: "peer",
peerId: selectedEdgeTurn.grant.parentPeerId,
},
"active",
);
return;
}
}
const result = ensureViewerSfuRoute().accept(message);
if (result !== "stale") {
@@ -1144,15 +1081,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
}
if (message.type === "media-assignment") {
if (peerAssisted && !viewerSfuRoute) {
applyMediaAssignment(
selectedEdgeTurn
? retainSelectedMediaParent(
message.mediaAssignment,
selectedEdgeTurn.grant.parentPeerId,
)
: message.mediaAssignment,
selectedEdgeTurn !== null,
);
applyMediaAssignment(message.mediaAssignment);
}
return;
}
@@ -1164,8 +1093,34 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
return;
}
if (message.type === "signal") {
if (peerAssisted && peerProbe?.parentPeerId === message.fromPeerId) {
await ensurePeerProbe()?.acceptSignal(message.fromPeerId, message.payload);
if (peerAssisted && pendingPeer?.parentPeerId === message.fromPeerId) {
if (currentAssignment.parentPeerId !== message.fromPeerId) {
await ensurePendingPeerRoute()?.acceptSignal(
message.fromPeerId,
message.payload,
);
return;
}
const activeIdentity = peerRef.current?.getConnectionIdentity();
const owner = exactPeerSignalOwner(
message.payload.connectionId,
pendingPeer.selectedTurn?.newConnectionId ??
pendingPeer.candidateConnectionId,
activeIdentity?.parentPeerId === message.fromPeerId
? activeIdentity.connectionId
: null,
);
if (owner === "pending") {
await ensurePendingPeerRoute()?.acceptSignal(
message.fromPeerId,
message.payload,
);
} else if (owner === "active") {
await peerRef.current?.acceptSignal(
message.fromPeerId,
message.payload,
);
}
return;
}
if (
@@ -1218,6 +1173,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
currentHostOnline = message.online;
setHostOnline(message.online);
setHostPaused(message.paused);
viewerSfuRoute?.setPaused(message.paused);
if (!message.online && !peerRef.current?.isConnected()) {
setStatusText("等待开始分享");
} else if (message.online && !peerRef.current?.isConnected()) {
@@ -1230,7 +1186,6 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) {
return;
}
if (message.type === "sharing-stopped") {
selectedEdgeTurn = null;
setSfuStandbyUrl(null);
setAssignedRoute(null);
currentHostOnline = false;
src/client/webrtc/viewer-relay.ts
+89 -192
View File
@@ -14,41 +14,27 @@ import { MAX_ENDPOINT_MEDIA_CHILDREN } from "./media-assignment";
interface ViewerRelayEvents {
sendSignal: (peerId: string, payload: SignalPayload) => boolean;
onUpdate?: (snapshot: PeerSnapshot | null) => void;
onSelectedEdgeFailed?: (
childPeerId: string,
connectionId: string,
revision: number,
) => void;
}
type SelectedEdgeTurn = Extract<
ServerMessage,
{ type: "selected-edge-turn"; edgeKind: "peer-selected" }
>;
interface SelectedChildConnection {
peerId: string;
connectionId: string;
revision: number;
pendingCarryRevision: number | null;
}
interface PreparedChild {
revision: number;
childPeerId: string;
peer: HostPeer;
failed: boolean;
replacesConnectionId: string | null;
}
type FailedPreparedChild = Omit<PreparedChild, "peer"> & { connectionId: string };
export class ViewerRelay {
private childPeerIds: string[] = [];
private stream: MediaStream | null = null;
private readonly peers = new Map<string, HostPeer>();
private readonly snapshots = new Map<string, PeerSnapshot>();
private readonly retiredConnections = new Map<string, string>();
private readonly selectedChildConnections = new Map<
string,
SelectedChildConnection
>();
private preparedChild: PreparedChild | null = null;
private failedPreparedChild: FailedPreparedChild | null = null;
private preparedRevision: number | null = null;
private plannedChildPeerIds: string[] = [];
private syncQueue = Promise.resolve();
private disposed = false;
@@ -68,142 +54,104 @@ export class ViewerRelay {
: null;
}
startSelectedEdgeTurn(
prepareSelectedEdgeTurn(
message: SelectedEdgeTurn,
parentPeerId: string,
routeRevision: number,
now = Date.now(),
): boolean {
const stream = this.stream;
const currentPeer = this.peers.get(message.viewerPeerId);
const connectionId =
currentPeer?.connectionId ??
this.retiredConnections.get(message.viewerPeerId);
const selected = this.selectedChildConnections.get(message.viewerPeerId);
if (
selected?.peerId === message.viewerPeerId &&
selected.connectionId === message.newConnectionId &&
currentPeer?.connectionId === selected.connectionId &&
message.parentPeerId === parentPeerId &&
message.revision >= selected.revision
) {
selected.revision = message.revision;
selected.pendingCarryRevision = message.revision;
return true;
}
if (
this.disposed ||
!stream ||
message.parentPeerId !== parentPeerId ||
message.revision !== routeRevision ||
Date.parse(message.expiresAt) <= now ||
connectionId !== message.oldConnectionId ||
(!this.childPeerIds.includes(message.viewerPeerId) &&
this.childPeerIds.length >= MAX_ENDPOINT_MEDIA_CHILDREN)
this.preparedRevision !== routeRevision ||
!this.plannedChildPeerIds.includes(message.viewerPeerId)
) {
return false;
}
this.disposePeer(message.viewerPeerId);
if (!this.childPeerIds.includes(message.viewerPeerId)) {
this.childPeerIds.push(message.viewerPeerId);
const prepared = this.preparedChild;
if (prepared && prepared.childPeerId !== message.viewerPeerId) {
return false;
}
this.selectedChildConnections.set(message.viewerPeerId, {
peerId: message.viewerPeerId,
connectionId: message.newConnectionId,
revision: message.revision,
pendingCarryRevision: null,
});
this.startPeer(message.viewerPeerId, stream, 0, message);
if (prepared?.peer.connectionId === message.newConnectionId) {
return !prepared.failed;
}
const oldConnectionId =
prepared?.peer.connectionId ??
this.peers.get(message.viewerPeerId)?.connectionId;
if (oldConnectionId !== message.oldConnectionId) {
return false;
}
this.discardPreparedPeer();
this.startPreparedChild(message.revision, message.viewerPeerId, stream, message);
return true;
}
clearSelectedEdgeTurn(): void {
for (const peerId of [...this.selectedChildConnections.keys()]) {
this.clearSelectedChildConnection(peerId);
}
}
acceptActiveRevision(revision: number): void {
for (const [peerId, selected] of [...this.selectedChildConnections]) {
if (selected.pendingCarryRevision === revision) {
selected.pendingCarryRevision = null;
} else {
this.clearSelectedChildConnection(peerId);
}
}
}
prepareChild(
revision: number,
plannedChildPeerIds: readonly string[],
): boolean {
const stream = this.stream;
const planned = [...new Set(plannedChildPeerIds)];
const childPeerId = planned.find(
(peerId) => !this.childPeerIds.includes(peerId),
);
if (
this.disposed ||
!stream ||
planned.length !== plannedChildPeerIds.length ||
planned.length !== this.childPeerIds.length + 1 ||
planned.length < this.childPeerIds.length ||
planned.length > this.childPeerIds.length + 1 ||
planned.length > MAX_ENDPOINT_MEDIA_CHILDREN ||
this.childPeerIds.some((peerId) => !planned.includes(peerId)) ||
!childPeerId ||
this.peers.has(childPeerId)
this.childPeerIds.some((peerId) => !planned.includes(peerId))
) {
this.discardPreparedChild();
return false;
}
if (
this.preparedChild?.revision === revision &&
this.preparedChild.childPeerId === childPeerId
this.preparedRevision === revision &&
samePeerIds(this.plannedChildPeerIds, planned)
) {
return true;
}
if (
this.failedPreparedChild?.revision === revision &&
this.failedPreparedChild.childPeerId === childPeerId
) {
return false;
return this.preparedChild ? !this.preparedChild.failed : true;
}
this.discardPreparedChild();
const peer = this.createPeer(childPeerId, stream);
this.preparedChild = { revision, childPeerId, peer };
void peer
.start()
.catch(() => false)
.then((started) => {
if (!started && this.preparedChild?.peer === peer) {
this.failPreparedChild(peer);
}
});
this.preparedRevision = revision;
this.plannedChildPeerIds = planned;
const childPeerId = planned.find(
(peerId) => !this.childPeerIds.includes(peerId),
);
if (!childPeerId) {
return true;
}
this.startPreparedChild(revision, childPeerId, stream);
return true;
}
activateChildren(revision: number, childPeerIds: readonly string[]): void {
const prepared = this.preparedChild;
const failed = this.failedPreparedChild;
const extendsActive =
childPeerIds.length === this.childPeerIds.length + 1 &&
this.childPeerIds.every((peerId) => childPeerIds.includes(peerId));
const replacedPeer = prepared
? this.peers.get(prepared.childPeerId) ?? null
: null;
const replacesActive =
replacedPeer?.connectionId === prepared?.replacesConnectionId;
let promotedPeerId: string | null = null;
if (
prepared?.revision === revision &&
!prepared.failed &&
childPeerIds.includes(prepared.childPeerId) &&
extendsActive
(extendsActive || replacesActive)
) {
this.preparedChild = null;
this.preparedRevision = null;
this.plannedChildPeerIds = [];
this.peers.set(prepared.childPeerId, prepared.peer);
replacedPeer?.dispose();
this.snapshots.set(prepared.childPeerId, prepared.peer.getSnapshot());
promotedPeerId = prepared.childPeerId;
} else if (
failed?.revision === revision &&
childPeerIds.includes(failed.childPeerId) &&
extendsActive
) {
return;
} else {
this.discardPreparedChild();
}
@@ -211,20 +159,19 @@ export class ViewerRelay {
}
discardPreparedChild(): void {
this.preparedRevision = null;
this.plannedChildPeerIds = [];
this.discardPreparedPeer();
}
private discardPreparedPeer(): void {
const prepared = this.preparedChild;
this.preparedChild = null;
this.failedPreparedChild = null;
prepared?.peer.dispose();
}
setChildren(childPeerIds: readonly string[]): void {
const failedChildPeerId = this.failedPreparedChild?.childPeerId;
this.reconcileChildren(
failedChildPeerId
? childPeerIds.filter((peerId) => peerId !== failedChildPeerId)
: childPeerIds,
null,
);
this.reconcileChildren(childPeerIds, null);
}
private reconcileChildren(
@@ -238,20 +185,6 @@ export class ViewerRelay {
0,
MAX_ENDPOINT_MEDIA_CHILDREN,
);
for (const selectedPeerId of this.selectedChildConnections.keys()) {
if (!nextChildPeerIds.includes(selectedPeerId)) {
if (nextChildPeerIds.length >= MAX_ENDPOINT_MEDIA_CHILDREN) {
const removableIndex = nextChildPeerIds.findLastIndex(
(peerId) => !this.selectedChildConnections.has(peerId),
);
if (removableIndex < 0) {
continue;
}
nextChildPeerIds.splice(removableIndex, 1);
}
nextChildPeerIds.push(selectedPeerId);
}
}
for (const childPeerId of this.childPeerIds) {
if (!nextChildPeerIds.includes(childPeerId)) {
this.disposePeer(childPeerId);
@@ -260,9 +193,6 @@ export class ViewerRelay {
this.childPeerIds = nextChildPeerIds;
for (const childPeerId of this.childPeerIds) {
const peer = this.peers.get(childPeerId);
if (this.selectedChildConnections.has(childPeerId)) {
continue;
}
if (
this.stream &&
childPeerId !== promotedPeerId &&
@@ -310,6 +240,7 @@ export class ViewerRelay {
): number {
const prepared = this.preparedChild;
return prepared?.childPeerId === fromPeerId &&
!prepared.failed &&
prepared.peer.connectionId === connectionId
? prepared.revision
: activeRevision;
@@ -323,6 +254,7 @@ export class ViewerRelay {
const prepared = this.preparedChild;
if (
prepared?.revision === routeRevision &&
!prepared.failed &&
prepared.childPeerId === fromPeerId &&
prepared.peer.connectionId === payload.connectionId
) {
@@ -330,14 +262,10 @@ export class ViewerRelay {
return true;
}
const peer = this.peers.get(fromPeerId);
const selected = this.selectedChildConnections.get(fromPeerId);
if (
!this.childPeerIds.includes(fromPeerId) ||
!peer ||
peer.connectionId !== payload.connectionId ||
(selected?.peerId === fromPeerId &&
(selected.connectionId !== payload.connectionId ||
selected.revision !== routeRevision))
peer.connectionId !== payload.connectionId
) {
return false;
}
@@ -386,7 +314,6 @@ export class ViewerRelay {
stop(): void {
this.stream = null;
this.discardPreparedChild();
this.clearSelectedEdgeTurn();
this.disposePeers();
}
@@ -397,11 +324,34 @@ export class ViewerRelay {
this.disposed = true;
this.stream = null;
this.discardPreparedChild();
this.selectedChildConnections.clear();
this.childPeerIds = [];
this.disposePeers();
}
private startPreparedChild(
revision: number,
childPeerId: string,
stream: MediaStream,
selectedTurn?: SelectedEdgeTurn,
): void {
const peer = this.createPeer(childPeerId, stream, selectedTurn);
this.preparedChild = {
revision,
childPeerId,
peer,
failed: false,
replacesConnectionId: selectedTurn?.oldConnectionId ?? null,
};
void peer
.start()
.catch(() => false)
.then((started) => {
if (!started && this.preparedChild?.peer === peer) {
this.failPreparedChild(peer);
}
});
}
private async syncStream(
stream: MediaStream,
replacePrepared: boolean,
@@ -484,7 +434,6 @@ export class ViewerRelay {
childPeerId: string,
stream: MediaStream,
attempt = 0,
selectedTurn?: SelectedEdgeTurn,
): void {
if (
this.disposed ||
@@ -495,7 +444,7 @@ export class ViewerRelay {
return;
}
const peer = this.createPeer(childPeerId, stream, selectedTurn);
const peer = this.createPeer(childPeerId, stream);
this.peers.set(childPeerId, peer);
void peer
.start()
@@ -504,10 +453,6 @@ export class ViewerRelay {
if (started || this.peers.get(childPeerId) !== peer) {
return;
}
if (selectedTurn) {
this.failSelectedChild(childPeerId, peer.connectionId);
return;
}
this.disposePeer(childPeerId);
if (
attempt < 1 &&
@@ -557,10 +502,6 @@ export class ViewerRelay {
snapshot.peerId === childPeerId &&
snapshot.connectionId === peer.connectionId
) {
if (selectedTurn && snapshot.connectionState === "failed") {
this.failSelectedChild(childPeerId, peer.connectionId);
return;
}
this.snapshots.set(childPeerId, {
...snapshot,
metrics: { ...snapshot.metrics },
@@ -575,34 +516,12 @@ export class ViewerRelay {
return peer;
}
private failSelectedChild(childPeerId: string, connectionId: string): void {
const selected = this.selectedChildConnections.get(childPeerId);
if (
selected?.peerId !== childPeerId ||
selected.connectionId !== connectionId
) {
return;
}
this.events.onSelectedEdgeFailed?.(
childPeerId,
connectionId,
selected.revision,
);
this.disposePeer(childPeerId);
}
private failPreparedChild(peer: HostPeer): void {
const prepared = this.preparedChild;
if (prepared?.peer !== peer) {
return;
}
this.preparedChild = null;
const { revision, childPeerId } = prepared;
this.failedPreparedChild = {
revision,
childPeerId,
connectionId: peer.connectionId,
};
prepared.failed = true;
peer.dispose();
}
@@ -614,39 +533,17 @@ export class ViewerRelay {
private disposePeer(childPeerId: string): void {
const peer = this.peers.get(childPeerId);
if (
peer &&
this.selectedChildConnections.get(childPeerId)?.connectionId ===
peer.connectionId
) {
this.selectedChildConnections.delete(childPeerId);
this.childPeerIds = this.childPeerIds.filter(
(peerId) => peerId !== childPeerId,
);
}
if (peer) {
this.retiredConnections.delete(childPeerId);
this.retiredConnections.set(childPeerId, peer.connectionId);
while (this.retiredConnections.size > MAX_ENDPOINT_MEDIA_CHILDREN) {
const oldestPeerId = this.retiredConnections.keys().next().value;
if (typeof oldestPeerId === "string") {
this.retiredConnections.delete(oldestPeerId);
}
}
}
this.peers.delete(childPeerId);
this.snapshots.delete(childPeerId);
this.events.onUpdate?.(this.getSnapshot());
peer?.dispose();
}
private clearSelectedChildConnection(childPeerId: string): void {
if (!this.selectedChildConnections.delete(childPeerId)) {
return;
}
this.childPeerIds = this.childPeerIds.filter(
(peerId) => peerId !== childPeerId,
);
this.disposePeer(childPeerId);
}
}
function samePeerIds(left: readonly string[], right: readonly string[]): boolean {
return (
left.length === right.length &&
left.every((peerId, index) => peerId === right[index])
);
}
tests/host-peer-tracks.test.ts
+93 -567
View File
@@ -1005,7 +1005,6 @@ function hostProvisionalInput(
overrides: {
activeChildPeerIds?: string[];
publicationGeneration?: string | null;
selectedPeerIds?: readonly string[];
} = {},
) {
return {
@@ -1015,7 +1014,6 @@ function hostProvisionalInput(
overrides.publicationGeneration ?? null,
),
activeChildPeerIds: overrides.activeChildPeerIds ?? [],
selectedPeerIds: overrides.selectedPeerIds ?? [],
maxMediaEdges: 2,
iceConfig: { iceServers: [] },
stream,
@@ -1104,7 +1102,6 @@ describe("Host provisional child runtime ownership", () => {
revision: 9,
assignment: hostAssignment([]),
activeChildPeerIds: [],
selectedPeerIds: [],
maxMediaEdges: 2,
})).toEqual({ kind: "ordinary" });
expect(second.connectionState).toBe("closed");
@@ -1116,124 +1113,70 @@ describe("Host provisional child runtime ownership", () => {
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);
it("updates the prepared stream before promotion", async () => {
const owner = new HostProvisionalChild({ sendSignal: () => true });
const initial = createStream(createTrack("video", "initial-video"), null);
const replacement = createTrack("video", "replacement-video");
const input = hostProvisionalInput(12, ["stream-child"], initial);
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);
const connection = FakePeerConnection.latest!;
await expect(
owner.replaceStream(createStream(replacement, null)),
).resolves.toBe(true);
expect(connection.senders[0]!.track).toBe(replacement);
expect(owner.activate(input)).toEqual({
kind: "failed",
peerId: "failed-probe",
connectionId: signals[0]!.connectionId,
const activation = owner.activate({
revision: input.revision,
assignment: input.assignment,
activeChildPeerIds: [],
maxMediaEdges: 2,
});
expect(FakePeerConnection.instances).toHaveLength(instanceCount);
expect(activation.kind).toBe("promote");
if (activation.kind === "promote") {
activation.peer.dispose();
}
});
it("replaces the prepared stream in place and tombstones a failed replacement", async () => {
const signals: Array<{ connectionId: string }> = [];
it("replaces an active child with selected TURN only on active promotion", async () => {
const signals: Array<{ peerId: string; connectionId: string }> = [];
const activeConnection = { id: "active-connection" };
const owner = new HostProvisionalChild({
sendSignal: (_peerId, payload) => {
signals.push({ connectionId: payload.connectionId });
sendSignal: (peerId, payload) => {
signals.push({ peerId, connectionId: payload.connectionId });
return true;
},
activeConnectionId: () => activeConnection.id,
});
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(
const input = hostProvisionalInput(
14,
["racing-stream-probe"],
initial,
["selected-child"],
createStream(createTrack("video", "selected-video"), null),
{ activeChildPeerIds: ["selected-child"] },
);
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");
expect(owner.prepare(input)).toBe(true);
expect(FakePeerConnection.latest).toBeNull();
expect(owner.prepareSelectedTurn({
...selectedEdgeTurn(
"selected-child",
activeConnection.id,
"selected-connection",
),
revision: 14,
}, input)).toBe(true);
await vi.waitFor(() => expect(signals).toHaveLength(1));
expect(signals[0]?.connectionId).toBe("selected-connection");
const activation = owner.activate({
revision: 14,
assignment: input.assignment,
activeChildPeerIds: ["selected-child"],
maxMediaEdges: 2,
});
expect(activation.kind).toBe("promote");
if (activation.kind === "promote") {
activation.peer.dispose();
}
await expect(racingReplacement).resolves.toBe(false);
expect(promotedFailure).toHaveBeenCalledWith(racingActivation.peer);
racingActivation.peer.dispose();
});
});
@@ -1341,43 +1284,6 @@ describe("ViewerRelay downstream ownership", () => {
relay.dispose();
});
it("fails closed when the committed prepared connection already failed", async () => {
const signals: Array<{ peerId: string; connectionId: string }> = [];
let rejectSignalsFor: string | null = null;
const relay = new ViewerRelay(
{ iceServers: [] },
QUALITY_PROFILES["720p30"],
{
sendSignal: (peerId, payload) => {
signals.push({ peerId, connectionId: payload.connectionId });
return peerId !== rejectSignalsFor;
},
},
);
relay.setStream(createStream(createTrack("video", "failed-active-video"), null));
rejectSignalsFor = "failed-child";
expect(relay.prepareChild(7, ["failed-child"])).toBe(true);
await vi.waitFor(() => expect(signals).toHaveLength(1));
const failedConnectionId = signals[0]!.connectionId;
const failedConnection = FakePeerConnection.latest!;
await vi.waitFor(() => expect(failedConnection.connectionState).toBe("closed"));
const instanceCount = FakePeerConnection.instances.length;
expect(relay.prepareChild(7, ["failed-child"])).toBe(false);
expect(FakePeerConnection.instances).toHaveLength(instanceCount);
relay.setChildren(["failed-child"]);
relay.activateChildren(7, ["failed-child"]);
expect(FakePeerConnection.instances).toHaveLength(instanceCount);
expect(FakePeerConnection.activeCount).toBe(0);
relay.setChildren(["failed-child"]);
relay.activateChildren(7, ["failed-child"]);
expect(FakePeerConnection.instances).toHaveLength(instanceCount);
expect(signals).toHaveLength(1);
expect(signals[0]!.connectionId).toBe(failedConnectionId);
relay.dispose();
});
it("keeps the prepared connection when its stream replacement succeeds", async () => {
const signals: Array<{ peerId: string; connectionId: string }> = [];
const relay = new ViewerRelay(
@@ -1412,37 +1318,6 @@ describe("ViewerRelay downstream ownership", () => {
relay.dispose();
});
it("retains failed replacement identity across repeated active updates", async () => {
const signals: Array<{ peerId: string; connectionId: string }> = [];
const relay = new ViewerRelay(
{ iceServers: [] },
QUALITY_PROFILES["720p30"],
{
sendSignal: (peerId, payload) => {
signals.push({ peerId, connectionId: payload.connectionId });
return true;
},
},
);
relay.setStream(createStream(createTrack("video", "initial-video"), null));
expect(relay.prepareChild(7, ["prepared-child"])).toBe(true);
await vi.waitFor(() => expect(signals).toHaveLength(1));
const preparedConnection = FakePeerConnection.latest!;
const preparedConnectionId = signals[0]!.connectionId;
preparedConnection.senders[0]!.failNextReplace = true;
const instanceCount = FakePeerConnection.instances.length;
relay.setStream(createStream(createTrack("video", "failed-video"), null));
await vi.waitFor(() => expect(preparedConnection.connectionState).toBe("closed"));
relay.activateChildren(7, ["prepared-child"]);
relay.activateChildren(7, ["prepared-child"]);
expect(FakePeerConnection.instances).toHaveLength(instanceCount);
expect(FakePeerConnection.activeCount).toBe(0);
expect(signals).toHaveLength(1);
expect(signals[0]!.connectionId).toBe(preparedConnectionId);
relay.dispose();
});
it("admits provisional children until endpoint cap three is full", async () => {
const relay = new ViewerRelay(
{ iceServers: [] },
@@ -1470,419 +1345,70 @@ describe("ViewerRelay downstream ownership", () => {
relay.dispose();
});
it("clamps a malicious fourth child to three physical endpoint copies", async () => {
const targets: string[] = [];
it("replaces the exact prepared child with selected TURN in the same slot", async () => {
const signals: Array<{ peerId: string; connectionId: string }> = [];
const relay = new ViewerRelay(
{ iceServers: [] },
QUALITY_PROFILES["1080p60"],
QUALITY_PROFILES["720p30"],
{
sendSignal: (peerId) => {
targets.push(peerId);
sendSignal: (peerId, payload) => {
signals.push({ peerId, connectionId: payload.connectionId });
return true;
},
},
);
const firstVideo = createTrack("video", "first-video");
const firstAudio = createTrack("audio", "first-audio");
relay.setChildren(["child-a", "child-b", "child-c", "ignored-fourth"]);
relay.setStream(createStream(firstVideo, firstAudio));
await vi.waitFor(() =>
expect(targets).toEqual(["child-a", "child-b", "child-c"]),
);
const [childAConnection, childBConnection, childCConnection] =
FakePeerConnection.instances;
expect(childAConnection).toBeDefined();
expect(childBConnection).toBeDefined();
expect(childCConnection).toBeDefined();
childAConnection!.connectionState = "connected";
childBConnection!.connectionState = "connected";
childCConnection!.connectionState = "connected";
const retiredConnectionId = relay.getSnapshot("child-a")!.connectionId;
const nextVideo = createTrack("video", "next-video");
const nextAudio = createTrack("audio", "next-audio");
relay.setStream(createStream(nextVideo, nextAudio));
await vi.waitFor(() => {
expect(childAConnection!.senders[0]?.track).toBe(nextVideo);
expect(childAConnection!.senders[1]?.track).toBe(nextAudio);
});
relay.setChildren(["child-b", "child-c", "child-d"]);
await vi.waitFor(() => expect(targets).toContain("child-d"));
expect(childAConnection!.connectionState).toBe("closed");
FakePeerConnection.latest!.connectionState = "connected";
expect(targets).toEqual(["child-a", "child-b", "child-c", "child-d"]);
expect(FakePeerConnection.activeCount).toBe(3);
expect(FakePeerConnection.peakActiveCount).toBe(3);
expect(targets).not.toContain("ignored-fourth");
await relay.acceptSignal("stale-child", {
kind: "description",
connectionId: "stale-connection",
description: { type: "answer", sdp: "stale-answer" },
}, 7);
await relay.recover("stale-child", "stale-connection", true);
await relay.recover("child-b", "wrong-connection", true);
expect(FakePeerConnection.activeCount).toBe(3);
const childBConnectionId = relay.getSnapshot("child-b")!.connectionId;
await relay.recover("child-b", childBConnectionId, true);
await vi.waitFor(() =>
expect(FakePeerConnection.latest).not.toBe(childBConnection),
);
expect(FakePeerConnection.activeCount).toBe(3);
expect(FakePeerConnection.peakActiveCount).toBe(3);
expect(
relay.startSelectedEdgeTurn(
selectedEdgeTurn(
"child-a",
retiredConnectionId,
"selected-connection-overflow",
),
"selected-parent",
7,
),
).toBe(false);
expect(FakePeerConnection.activeCount).toBe(3);
expect(FakePeerConnection.peakActiveCount).toBe(3);
relay.dispose();
});
it("rebuilds a retired child as one relay-only selected edge", async () => {
const sendSignal = vi.fn(() => true);
const relay = new ViewerRelay(
{ iceServers: [] },
QUALITY_PROFILES["720p30"],
{ sendSignal },
);
relay.setChildren(["selected-child"]);
relay.setStream(createStream(createTrack("video", "selected-video"), null));
await vi.waitFor(() => expect(sendSignal).toHaveBeenCalledOnce());
const oldConnectionId = relay.getSnapshot()!.connectionId;
FakePeerConnection.latest!.connectionState = "connected";
relay.setChildren([]);
sendSignal.mockClear();
const selectedGrant = selectedEdgeTurn(
"selected-child",
oldConnectionId,
"selected-connection-new",
);
expect(
relay.startSelectedEdgeTurn(selectedGrant, "wrong-parent", 7),
).toBe(false);
expect(
relay.startSelectedEdgeTurn(selectedGrant, "selected-parent", 6),
).toBe(false);
expect(
relay.startSelectedEdgeTurn(
selectedGrant,
"selected-parent",
7,
Date.parse(selectedGrant.expiresAt),
),
).toBe(false);
expect(
relay.startSelectedEdgeTurn(
selectedGrant,
"selected-parent",
7,
),
).toBe(true);
await vi.waitFor(() => expect(sendSignal).toHaveBeenCalledOnce());
expect(sendSignal).toHaveBeenCalledWith(
"selected-child",
expect.objectContaining({
kind: "description",
connectionId: "selected-connection-new",
description: expect.objectContaining({ type: "offer" }),
}),
);
expect(FakePeerConnection.latest?.configurations).toEqual([{
iceServers: [selectedTurnIceServer],
iceTransportPolicy: "relay",
}]);
const selectedConnection = FakePeerConnection.latest!;
expect(
relay.startSelectedEdgeTurn(
{ ...selectedGrant, revision: 8 },
"selected-parent",
7,
),
).toBe(true);
expect(FakePeerConnection.latest).toBe(selectedConnection);
expect(sendSignal).toHaveBeenCalledOnce();
relay.acceptActiveRevision(8);
relay.setChildren([]);
expect(selectedConnection.connectionState).not.toBe("closed");
expect(relay.getSnapshot("selected-child")?.connectionId).toBe(
"selected-connection-new",
);
expect(FakePeerConnection.activeCount).toBe(1);
expect(FakePeerConnection.peakActiveCount).toBe(1);
const pendingCandidate = { candidate: "candidate-before-answer" };
await expect(
relay.acceptSignal("selected-child", {
kind: "candidate",
connectionId: "selected-connection-new",
candidate: pendingCandidate,
}, 8),
).resolves.toBe(true);
expect(selectedConnection.addedIceCandidates).toEqual([]);
await expect(
relay.acceptSignal("selected-child", {
kind: "description",
connectionId: "selected-connection-new",
description: { type: "answer", sdp: "selected-answer" },
}, 8),
).resolves.toBe(true);
expect(selectedConnection.remoteDescription?.type).toBe("answer");
expect(selectedConnection.addedIceCandidates).toEqual([pendingCandidate]);
await expect(
relay.acceptSignal("selected-child", {
kind: "candidate",
connectionId: "selected-connection-new",
candidate: null,
}, 8),
).resolves.toBe(true);
expect(selectedConnection.addedIceCandidates).toEqual([
pendingCandidate,
null,
]);
await expect(
relay.acceptSignal("selected-child", {
kind: "candidate",
connectionId: "selected-connection-new",
candidate: null,
}, 7),
).resolves.toBe(false);
await expect(
relay.acceptSignal("selected-child", {
kind: "candidate",
connectionId: oldConnectionId,
candidate: null,
}, 8),
).resolves.toBe(false);
expect(
relay.startSelectedEdgeTurn(
{
...selectedEdgeTurn(
"selected-child",
"selected-connection-new",
"selected-connection-replacement",
),
revision: 9,
},
"selected-parent",
9,
),
).toBe(true);
await vi.waitFor(() => expect(sendSignal).toHaveBeenCalledTimes(2));
expect(selectedConnection.connectionState).toBe("closed");
const replacementConnection = FakePeerConnection.latest!;
expect(replacementConnection.connectionState).not.toBe("closed");
expect(FakePeerConnection.activeCount).toBe(1);
expect(FakePeerConnection.peakActiveCount).toBe(1);
relay.acceptActiveRevision(9);
expect(replacementConnection.connectionState).toBe("closed");
await expect(
relay.acceptSignal("selected-child", {
kind: "candidate",
connectionId: "selected-connection-replacement",
candidate: null,
}, 9),
).resolves.toBe(false);
expect(FakePeerConnection.activeCount).toBe(0);
expect(FakePeerConnection.peakActiveCount).toBe(1);
relay.dispose();
});
it("keeps concurrent selected children isolated across carry and failure", async () => {
const sendSignal = vi.fn(() => true);
const onSelectedEdgeFailed = vi.fn();
const relay = new ViewerRelay(
{ iceServers: [] },
QUALITY_PROFILES["720p30"],
{ sendSignal, onSelectedEdgeFailed },
);
relay.setChildren(["selected-a", "selected-b"]);
relay.setStream(createStream(createTrack("video", "selected-video"), null));
await vi.waitFor(() => expect(sendSignal).toHaveBeenCalledTimes(2));
const oldConnectionA = relay.getSnapshot("selected-a")!.connectionId;
const oldConnectionB = relay.getSnapshot("selected-b")!.connectionId;
relay.setChildren([]);
expect(FakePeerConnection.activeCount).toBe(0);
sendSignal.mockClear();
const grantA = selectedEdgeTurn(
"selected-a",
oldConnectionA,
"selected-connection-a",
);
const grantB = selectedEdgeTurn(
"selected-b",
oldConnectionB,
"selected-connection-b",
);
expect(relay.startSelectedEdgeTurn(grantA, "selected-parent", 7)).toBe(true);
expect(relay.startSelectedEdgeTurn(grantB, "selected-parent", 7)).toBe(true);
await vi.waitFor(() => expect(sendSignal).toHaveBeenCalledTimes(2));
const [selectedConnectionA, selectedConnectionB] =
FakePeerConnection.instances.slice(-2);
expect(selectedConnectionA).toBeDefined();
expect(selectedConnectionB).toBeDefined();
expect(FakePeerConnection.activeCount).toBe(2);
await vi.waitFor(() => expect(signals).toHaveLength(1));
const activeConnection = FakePeerConnection.latest!;
const activeConnectionId = signals[0]!.connectionId;
activeConnection.connectionState = "connected";
expect(relay.prepareChild(7, ["selected-child"])).toBe(true);
expect(FakePeerConnection.latest).toBe(activeConnection);
expect(
relay.startSelectedEdgeTurn(
{ ...grantA, revision: 8 },
"selected-parent",
7,
),
).toBe(true);
expect(
relay.startSelectedEdgeTurn(
{ ...grantB, revision: 8 },
"selected-parent",
7,
),
).toBe(true);
relay.acceptActiveRevision(8);
relay.setChildren([]);
expect(selectedConnectionA!.connectionState).not.toBe("closed");
expect(selectedConnectionB!.connectionState).not.toBe("closed");
expect(FakePeerConnection.activeCount).toBe(2);
await expect(
relay.acceptSignal(
"selected-a",
{
kind: "description",
connectionId: "selected-connection-a",
description: { type: "answer", sdp: "selected-answer-a" },
},
8,
),
).resolves.toBe(true);
await expect(
relay.acceptSignal(
"selected-b",
{
kind: "description",
connectionId: "selected-connection-b",
description: { type: "answer", sdp: "selected-answer-b" },
},
8,
),
).resolves.toBe(true);
expect(selectedConnectionA!.remoteDescription?.sdp).toBe(
"selected-answer-a",
);
expect(selectedConnectionB!.remoteDescription?.sdp).toBe(
"selected-answer-b",
);
FakePeerConnection.offersFailing = 1;
expect(
relay.startSelectedEdgeTurn(
{
...selectedEdgeTurn(
"selected-a",
"selected-connection-a",
"selected-connection-a-failed",
),
revision: 8,
},
"selected-parent",
8,
),
).toBe(true);
await vi.waitFor(() =>
expect(onSelectedEdgeFailed).toHaveBeenCalledWith(
"selected-a",
"selected-connection-a-failed",
8,
),
);
expect(selectedConnectionA!.connectionState).toBe("closed");
expect(selectedConnectionB!.connectionState).not.toBe("closed");
expect(relay.getSnapshot("selected-a")).toBeNull();
expect(relay.getSnapshot("selected-b")?.connectionId).toBe(
"selected-connection-b",
);
expect(FakePeerConnection.activeCount).toBe(1);
await expect(
relay.acceptSignal(
"selected-a",
{
kind: "candidate",
connectionId: "selected-connection-a-failed",
candidate: null,
},
8,
),
).resolves.toBe(false);
await expect(
relay.acceptSignal(
"selected-b",
{
kind: "candidate",
connectionId: "selected-connection-b",
candidate: null,
},
8,
),
).resolves.toBe(true);
expect(
relay.startSelectedEdgeTurn(
{ ...grantB, revision: 9 },
"selected-parent",
8,
),
).toBe(true);
relay.acceptActiveRevision(9);
expect(selectedConnectionB!.connectionState).not.toBe("closed");
relay.clearSelectedEdgeTurn();
expect(selectedConnectionB!.connectionState).toBe("closed");
expect(FakePeerConnection.activeCount).toBe(0);
relay.dispose();
});
it("reports a selected child whose relay-only offer fails", async () => {
const onSelectedEdgeFailed = vi.fn();
const relay = new ViewerRelay(
{ iceServers: [] },
QUALITY_PROFILES["720p30"],
{ sendSignal: () => true, onSelectedEdgeFailed },
);
relay.setChildren(["selected-child"]);
relay.setStream(createStream(createTrack("video", "selected-video"), null));
await vi.waitFor(() => expect(relay.getSnapshot()).not.toBeNull());
const oldConnectionId = relay.getSnapshot()!.connectionId;
relay.setChildren([]);
FakePeerConnection.offersFailing = 1;
expect(
relay.startSelectedEdgeTurn(
relay.prepareSelectedEdgeTurn(
selectedEdgeTurn(
"selected-child",
oldConnectionId,
"selected-failed-connection",
activeConnectionId,
"selected-connection",
),
"selected-parent",
7,
),
).toBe(true);
await vi.waitFor(() =>
expect(onSelectedEdgeFailed).toHaveBeenCalledWith(
await vi.waitFor(() => expect(signals).toHaveLength(2));
const selectedConnection = FakePeerConnection.latest!;
expect(activeConnection.connectionState).toBe("connected");
expect(selectedConnection.configurations).toEqual([
{
iceServers: [selectedTurnIceServer],
iceTransportPolicy: "relay",
},
]);
expect(signals[1]).toEqual({
peerId: "selected-child",
connectionId: "selected-connection",
});
await expect(
relay.acceptSignal(
"selected-child",
"selected-failed-connection",
{
kind: "description",
connectionId: "selected-connection",
description: { type: "answer", sdp: "selected-answer" },
},
7,
),
).resolves.toBe(true);
selectedConnection.connectionState = "connected";
relay.activateChildren(7, ["selected-child"]);
expect(activeConnection.connectionState).toBe("closed");
expect(relay.getSnapshot("selected-child")?.connectionId).toBe(
"selected-connection",
);
expect(relay.getSnapshot("selected-child")).toBeNull();
expect(FakePeerConnection.latest?.connectionState).toBe("closed");
expect(FakePeerConnection.activeCount).toBe(1);
relay.dispose();
});
tests/route-transition.test.ts
+218 -2176
View File
File diff suppressed because it is too large. Load diff