fix(route): fix route and media state ownership (#347)

This commit is contained in:
AAA authored and GitHub committed 2026-08-31 01:45:43 +08:00
1 parent 6daf288af8
commit e47d06fcdd
20 files changed
+446 -145

No files matched your search

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