diff --git a/src/client/lib/signaling.ts b/src/client/lib/signaling.ts index d9ea38b..34b4a77 100644 --- a/src/client/lib/signaling.ts +++ b/src/client/lib/signaling.ts @@ -288,8 +288,8 @@ export class SignalingClient { this.clearAuthenticationTimer(); if (this.terminalMessage) { const terminalMessage = this.terminalMessage; - this.terminalMessage = null; if (this.send(terminalMessage)) { + this.terminalMessage = null; this.stop(); return; } diff --git a/src/client/media/viewer-quality-evidence.ts b/src/client/media/viewer-quality-evidence.ts index 29ba924..8bdcc79 100644 --- a/src/client/media/viewer-quality-evidence.ts +++ b/src/client/media/viewer-quality-evidence.ts @@ -460,6 +460,19 @@ export function freshViewerQualityEvidence( return presentation?.fresh ? presentation.evidence : null; } +export function retainPresentViewerQualityEvidence( + presentations: Map, + presentPeerIds: ReadonlySet, +): Map { + let retained: Map | null = null; + for (const peerId of presentations.keys()) { + if (presentPeerIds.has(peerId)) continue; + retained ??= new Map(presentations); + retained.delete(peerId); + } + return retained ?? presentations; +} + export function classifyHostViewerQualityEvidence( evidence: ViewerQualityEvidence, hostPeerId: string | null, diff --git a/src/client/pages/HostPage.tsx b/src/client/pages/HostPage.tsx index 5a9087f..7bb53da 100644 --- a/src/client/pages/HostPage.tsx +++ b/src/client/pages/HostPage.tsx @@ -134,6 +134,7 @@ import { qualityEvidenceUpstreamMatches, reconcileViewerQualityEvidencePresentation, refreshViewerQualityEvidencePresentation, + retainPresentViewerQualityEvidence, type ViewerQualityEvidencePresentation, } from "../media/viewer-quality-evidence"; import type { @@ -436,6 +437,9 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { >(() => new Map()); const [noticeValue, setNoticeValue] = useState(null); const [noticeComic, setNoticeComic] = useState(null); + const [hostSfuQualityWarning, setHostSfuQualityWarning] = useState< + string | null + >(null); function setNotice(value: string | null, comic: ComicKind | null = null): void { setNoticeValue(value ? { kind: "text", text: value } : null); setNoticeComic(comic); @@ -729,6 +733,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { function clearHostSfuRoute(): void { const route = hostSfuRouteRef.current; hostSfuRouteRef.current = null; + setHostSfuQualityWarning(null); sfuStandbyPrewarmerRef.current?.setUrl(null); void route?.disconnect(); } @@ -742,19 +747,19 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { sfuStandbyPrewarmerRef.current.setUrl(url); } - function showHostSfuQualityWarning( + function syncHostSfuQualityWarning( route: HostSfuRoute, generation: number, - ): void { + ): string | null { if ( - isCurrentGeneration(generation) && - hostSfuRouteRef.current === route + !isCurrentGeneration(generation) || + hostSfuRouteRef.current !== route ) { - const warning = route.getQualityWarning(); - if (warning) { - setNotice(warning, "route-failed"); - } + return null; } + const warning = route.getQualityWarning(); + setHostSfuQualityWarning(warning); + return warning; } function disposeResources(notifyServer: boolean): void { @@ -765,6 +770,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { codecProbeAbortRef.current = null; videoCodecRef.current = VP8_ONLY_VIDEO_CODEC; setResolvedVideoCodec(null); + setHostSfuQualityWarning(null); const signal = signalRef.current; if (signal) { if (notifyServer) { @@ -1032,6 +1038,30 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { commitViewerQualityEvidence(peerId, null); } + function retainViewerQualityEvidenceForPresence( + entries: readonly ParticipantPresenceEntry[], + ): void { + const presentPeerIds = new Set( + entries + .filter((entry) => entry.role === "viewer") + .map((entry) => entry.peerId), + ); + const current = viewerQualityEvidenceRef.current; + const retained = retainPresentViewerQualityEvidence( + current, + presentPeerIds, + ); + if (retained === current) return; + for (const peerId of current.keys()) { + if (retained.has(peerId)) continue; + const timer = viewerQualityEvidenceTimersRef.current.get(peerId); + if (timer !== undefined) window.clearTimeout(timer); + viewerQualityEvidenceTimersRef.current.delete(peerId); + } + viewerQualityEvidenceRef.current = retained; + setViewerQualityEvidence(retained); + } + function acceptViewerQualityEvidence(evidence: ViewerQualityEvidence): void { const peer = peersRef.current.get(evidence.viewerPeerId); const directSnapshot = peer?.getSnapshot() ?? null; @@ -1211,14 +1241,13 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { ) { const failed = results.filter((updated) => !updated).length; const sfuWarning = - hostSfuRouteRef.current === activeSfuRoute - ? (activeSfuRoute?.getQualityWarning() ?? null) + activeSfuRoute && hostSfuRouteRef.current === activeSfuRoute + ? syncHostSfuQualityWarning(activeSfuRoute, generation) : null; const connectionWarning = failed > 0 || !sfuUpdated ? say("host.notice.partialApply") : null; - const warning = sfuWarning ?? connectionWarning; const successNotice = audioChanged && !videoChanged ? say("host.notice.audioSet", { label: say(AUDIO_QUALITY_CAPTIONS[resolveScreenAudioQuality(nextProfile.screenAudioQuality)]) }) @@ -1227,7 +1256,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { : say("host.notice.qualitySet", { label: qualitySettingsLabel(nextProfile), }); - setNotice(warning || successNotice); + setNotice(connectionWarning ?? (sfuWarning ? null : successNotice)); } } catch (error) { if ( @@ -1676,7 +1705,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { if (reauthenticated && hostSfuRouteRef.current === route) { await route.updateProfile(currentQualitySettings); } - showHostSfuQualityWarning(route, generation); + syncHostSfuQualityWarning(route, generation); }); return; } @@ -1687,6 +1716,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { return; } if (message.type === "viewer-presence") { + retainViewerQualityEvidenceForPresence(message.viewers); setParticipantPresence(message.viewers); return; } @@ -1732,7 +1762,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { ) { activeRouteRevisionRef.current = message.revision; } - showHostSfuQualityWarning(route, generation); + syncHostSfuQualityWarning(route, generation); } return; } @@ -1745,7 +1775,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { const route = ensureHostSfuRoute(generation); void route .acceptConfig(message) - .then(() => showHostSfuQualityWarning(route, generation)); + .then(() => syncHostSfuQualityWarning(route, generation)); } return; } @@ -2183,7 +2213,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { } const sfuWarning = activeSfuRoute && hostSfuRouteRef.current === activeSfuRoute - ? activeSfuRoute.getQualityWarning() + ? syncHostSfuQualityWarning(activeSfuRoute, generation) : null; previousStream.getTracks().forEach((track) => track.stop()); if (retiringStreamRef.current === previousStream) { @@ -2216,13 +2246,18 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { isCurrentGeneration(generation) && sourceSwitchRef.current === token ) { + const sourceNotice = + sfuReplaced && sfuWarning && failedPeerIds.length === 0 + ? null + : sourceSwitchNotice({ + failedPeerCount: failedPeerIds.length, + sfuReplaced, + sfuWarning: null, + }); setNotice( - sourceSwitchNotice({ - failedPeerCount: failedPeerIds.length, - sfuReplaced, - sfuWarning, - }), - sfuWarning || !sfuReplaced || failedPeerIds.length > 0 + sourceNotice, + sourceNotice && + (sfuWarning || !sfuReplaced || failedPeerIds.length > 0) ? "warning" : null, ); @@ -2897,6 +2932,14 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { } /> ) : null} + {hostSfuQualityWarning && + hostSfuQualityWarning !== noticeText ? ( + + ) : null} {noticeText && (vis || noticeText !== phaseLine) ? ( (); private readonly viewerGraceTimers = new Map(); private readonly connectionIdsByViewer = new Map(); + private readonly ordinaryReadyConnectionIdsByViewer = new Map< + string, + string + >(); private readonly viewerQualityEvidenceGates = new Map< string, ViewerQualityEvidenceGate @@ -271,6 +275,7 @@ export class SignalingServer { this.shareGenerationsByRoom.clear(); this.pausedShareGenerationsByRoom.clear(); this.ordinaryActiveHostChildrenByRoom.clear(); + this.ordinaryReadyConnectionIdsByViewer.clear(); await closed; clearTimeout(forceCloseTimer); if (routeCloseError) { @@ -1462,6 +1467,18 @@ export class SignalingServer { fromPeerId: source.peerId, payload: message.payload, }); + const connectionKey = viewerConnectionKey(source.roomId, source.peerId); + if ( + description?.type === "answer" && + this.connectionIdsByViewer.get(connectionKey) === + message.payload.connectionId + ) { + this.ordinaryReadyConnectionIdsByViewer.set( + connectionKey, + message.payload.connectionId, + ); + this.sendViewerPresence(source.roomId); + } } private routePeerAssistedSignal( @@ -1934,13 +1951,19 @@ export class SignalingServer { viewer.peerId, ) : undefined; - const mediaReady = - activeEdge !== undefined && - upstream.kind !== "none" && - (activeEdge.upstream.kind === "sfu" - ? upstream.kind === "sfu" - : upstream.kind === "peer" && - activeEdge.upstream.peerId === upstream.peerId); + const connectionKey = viewerConnectionKey(roomId, viewer.peerId); + const mediaReady = this.isHybridMediaEnabled() + ? activeEdge !== undefined && + upstream.kind !== "none" && + (activeEdge.upstream.kind === "sfu" + ? upstream.kind === "sfu" + : upstream.kind === "peer" && + activeEdge.upstream.peerId === upstream.peerId) + : upstream.kind === "peer" && + this.connectionIdsByViewer.has(connectionKey) && + this.ordinaryReadyConnectionIdsByViewer.get( + connectionKey, + ) === this.connectionIdsByViewer.get(connectionKey); viewers.push({ role: "viewer", peerId: viewer.peerId, @@ -2081,6 +2104,11 @@ export class SignalingServer { this.connectionIdsByViewer.delete(key); } } + for (const key of this.ordinaryReadyConnectionIdsByViewer.keys()) { + if (key.startsWith(prefix)) { + this.ordinaryReadyConnectionIdsByViewer.delete(key); + } + } for (const key of this.viewerQualityEvidenceGates.keys()) { if (key.startsWith(prefix)) { this.viewerQualityEvidenceGates.delete(key); @@ -2100,6 +2128,7 @@ export class SignalingServer { ): void { const key = viewerConnectionKey(roomId, viewerPeerId); if (this.connectionIdsByViewer.get(key) !== connectionId) { + this.ordinaryReadyConnectionIdsByViewer.delete(key); this.viewerQualityEvidenceGates.delete(key); this.viewerQualityEvidenceAttemptAtMs.delete(key); } @@ -2112,6 +2141,7 @@ export class SignalingServer { ): void { const key = viewerConnectionKey(roomId, viewerPeerId); this.connectionIdsByViewer.delete(key); + this.ordinaryReadyConnectionIdsByViewer.delete(key); this.viewerQualityEvidenceGates.delete(key); this.viewerQualityEvidenceAttemptAtMs.delete(key); } diff --git a/tests/client-state.test.ts b/tests/client-state.test.ts index 1ed120e..0aaf6f1 100644 --- a/tests/client-state.test.ts +++ b/tests/client-state.test.ts @@ -814,6 +814,92 @@ describe("client signaling recovery policy", () => { signal.stop(); }); + it("retains a terminal command until an authenticated socket accepts it", () => { + vi.useFakeTimers(); + const sockets: FakeWebSocket[] = []; + class FakeWebSocket extends EventTarget { + static readonly OPEN = 1; + static readonly CLOSING = 2; + readyState = FakeWebSocket.OPEN; + readonly send = vi.fn(); + readonly close = vi.fn(() => { + this.readyState = FakeWebSocket.CLOSING; + }); + + constructor(readonly url: string) { + super(); + sockets.push(this); + } + } + vi.stubGlobal("WebSocket", FakeWebSocket); + vi.stubGlobal("window", { + location: new URL("https://share.test/r/1234"), + setTimeout: globalThis.setTimeout, + clearTimeout: globalThis.clearTimeout, + }); + const signal = new SignalingClient( + { + roomId: "1234", + role: "viewer", + clientId: "viewer-client", + }, + { + onMessage: () => undefined, + onStatus: () => undefined, + onTerminated: () => undefined, + onAccessRequired: () => undefined, + }, + ); + const receiveAuthenticated = (socket: FakeWebSocket) => { + const event = new Event("message"); + Object.defineProperty(event, "data", { + value: JSON.stringify({ + type: "authenticated", + protocol: SIGNALING_PROTOCOL, + role: "viewer", + peerId: "viewer_12345678", + roomExpiresAt: null, + maxViewers: 8, + endpointMediaCopyCapacity: 2, + hostOnline: true, + connectionId: null, + viewerPeerIds: [], + iceConfig: { iceServers: [] }, + codeEntryPolicy: "open", + viewerAuthorizationGeneration: "viewer_generation_12345678", + }), + }); + socket.dispatchEvent(event); + }; + + signal.start(); + signal.sendThenStop({ type: "abandon-room" }); + sockets[0]!.dispatchEvent(new Event("open")); + sockets[0]!.readyState = FakeWebSocket.CLOSING; + receiveAuthenticated(sockets[0]!); + expect(sockets[0]!.send).toHaveBeenCalledOnce(); + + const closed = new Event("close"); + Object.defineProperties(closed, { + code: { value: 1006 }, + reason: { value: "network interrupted" }, + }); + sockets[0]!.dispatchEvent(closed); + vi.advanceTimersByTime(750); + sockets[1]!.dispatchEvent(new Event("open")); + receiveAuthenticated(sockets[1]!); + + expect( + sockets[1]!.send.mock.calls.map(([value]) => + JSON.parse(String(value)), + ), + ).toEqual([ + expect.objectContaining({ type: "authenticate" }), + { type: "abandon-room" }, + ]); + expect(sockets[1]!.close).toHaveBeenCalledWith(1000, "client closed"); + }); + it("classifies a close-only authentication failure without claiming replacement", () => { const sockets: FakeWebSocket[] = []; class FakeWebSocket extends EventTarget { diff --git a/tests/server-signal.test.ts b/tests/server-signal.test.ts index 3d836ee..b5d1bb7 100644 --- a/tests/server-signal.test.ts +++ b/tests/server-signal.test.ts @@ -2216,6 +2216,100 @@ describe("WebSocket signaling", () => { }); }); + it("marks an ordinary Host edge ready after its exact SDP handshake", async () => { + const harness = await startHarness({ peerAssistedMedia: false }); + const host = await openClient(harness.webSocketUrl); + const hostAuth = await authenticate( + host, + harness.room, + "host", + "ordinary-ready-host", + 1, + undefined, + { viewerPresence: true }, + ); + await host.inbox.next("viewer-presence"); + const viewer = await openClient(harness.webSocketUrl); + const viewerAuth = await authenticate( + viewer, + harness.room, + "viewer", + "ordinary-ready-viewer", + 1, + undefined, + { viewerPresence: true }, + ); + expect((await host.inbox.next("peer-joined")).peerId).toBe( + viewerAuth.peerId, + ); + const assigned = await nextViewerPresenceMatching( + host, + (message) => + viewerPresenceEntries(message).some( + (entry) => entry.peerId === viewerAuth.peerId, + ), + ); + expect(viewerPresenceEntries(assigned)[0]).not.toHaveProperty("mediaReady"); + + const connectionId = "ordinary_ready_connection_12345678"; + host.socket.send( + JSON.stringify({ + type: "signal", + targetPeerId: viewerAuth.peerId, + payload: { + kind: "description", + connectionId, + description: { type: "offer", sdp: "v=0\r\n" }, + }, + }), + ); + expect(await viewer.inbox.next("signal")).toMatchObject({ + fromPeerId: hostAuth.peerId, + payload: { connectionId, description: { type: "offer" } }, + }); + host.socket.send( + JSON.stringify({ + type: "set-display-name", + displayName: "等待应答", + }), + ); + const offered = await nextViewerPresenceMatching( + host, + (message) => + viewerPresenceEntries(message).some( + (entry) => entry.peerId === viewerAuth.peerId, + ), + ); + expect(viewerPresenceEntries(offered)[0]).not.toHaveProperty("mediaReady"); + viewer.socket.send( + JSON.stringify({ + type: "signal", + payload: { + kind: "description", + connectionId, + description: { type: "answer", sdp: "v=0\r\n" }, + }, + }), + ); + expect(await host.inbox.next("signal")).toMatchObject({ + fromPeerId: viewerAuth.peerId, + payload: { connectionId, description: { type: "answer" } }, + }); + const ready = await nextViewerPresenceMatching( + host, + (message) => + viewerPresenceEntries(message).some( + (entry) => + entry.peerId === viewerAuth.peerId && entry.mediaReady === true, + ), + ); + expect(viewerPresenceEntries(ready)[0]).toMatchObject({ + peerId: viewerAuth.peerId, + upstream: { kind: "peer", peerId: hostAuth.peerId }, + mediaReady: true, + }); + }); + it("binds new-share quality and preserves active-share quality across reconnect", async () => { const harness = await startHarness({ peerAssistedMedia: true }); const initialSettings: QualitySettings = { diff --git a/tests/viewer-quality-evidence.test.ts b/tests/viewer-quality-evidence.test.ts index df0d41b..8e0c21a 100644 --- a/tests/viewer-quality-evidence.test.ts +++ b/tests/viewer-quality-evidence.test.ts @@ -16,6 +16,7 @@ import { qualityEvidenceWindowFromMetrics, reconcileViewerQualityEvidencePresentation, refreshViewerQualityEvidencePresentation, + retainPresentViewerQualityEvidence, type ViewerQualityEvidence, ViewerQualityEvidenceReporter, } from "../src/client/media/viewer-quality-evidence.ts"; @@ -809,4 +810,34 @@ describe("viewer quality evidence", () => { ).toBeNull(); }); + it("retains evidence only for Viewers in the authoritative presence", () => { + const first = presentViewerQualityEvidence( + null, + serverEvidence({ viewerPeerId: "viewer_first_12345678" }), + 0, + ); + const second = presentViewerQualityEvidence( + null, + serverEvidence({ viewerPeerId: "viewer_second_12345678" }), + 0, + ); + const presentations = new Map([ + [first.evidence.viewerPeerId, first], + [second.evidence.viewerPeerId, second], + ]); + + expect( + retainPresentViewerQualityEvidence( + presentations, + new Set(presentations.keys()), + ), + ).toBe(presentations); + const retained = retainPresentViewerQualityEvidence( + presentations, + new Set([second.evidence.viewerPeerId]), + ); + expect([...retained.keys()]).toEqual([second.evidence.viewerPeerId]); + expect(presentations.size).toBe(2); + }); + });