From baa8bcab4ac5e19189711fc72e5689491cd8860c Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Wed, 19 Aug 2026 02:55:42 +0800 Subject: [PATCH] feat: synchronize peer-assisted quality profiles --- src/client/media/quality.ts | 16 +++- src/client/pages/HostPage.tsx | 30 ++++--- src/client/pages/ViewerPage.tsx | 32 +++++-- src/client/webrtc/host-peer.ts | 140 +++++++++++++++++------------- src/client/webrtc/viewer-relay.ts | 30 ++++++- src/server/signaling.ts | 33 +++++++ src/shared/protocol.ts | 21 +++++ tests/host-peer-tracks.test.ts | 117 ++++++++++++++++++++++--- tests/protocol.test.ts | 55 ++++++++++++ tests/server-signal.test.ts | 101 +++++++++++++++++++++ 10 files changed, 482 insertions(+), 93 deletions(-) diff --git a/src/client/media/quality.ts b/src/client/media/quality.ts index 2a05058..4a65dbf 100644 --- a/src/client/media/quality.ts +++ b/src/client/media/quality.ts @@ -1,3 +1,7 @@ +import type { QualityProfileId } from "../../shared/protocol"; + +export type { QualityProfileId } from "../../shared/protocol"; + export const QUALITY_PROFILES = { "1080p60": { label: "1080p 60", @@ -20,9 +24,17 @@ export const QUALITY_PROFILES = { frameRate: 30, maxBitrate: 3_000_000, }, -} as const; +} as const satisfies Record< + QualityProfileId, + { + label: string; + width: number; + height: number; + frameRate: number; + maxBitrate: number; + } +>; -export type QualityProfileId = keyof typeof QUALITY_PROFILES; export type QualityProfile = (typeof QUALITY_PROFILES)[QualityProfileId]; function captureConstraints(profile: QualityProfile): MediaTrackConstraints { diff --git a/src/client/pages/HostPage.tsx b/src/client/pages/HostPage.tsx index 8dc38c0..bcae8a9 100644 --- a/src/client/pages/HostPage.tsx +++ b/src/client/pages/HostPage.tsx @@ -10,10 +10,11 @@ import { Users, } from "lucide-react"; import { useEffect, useMemo, useRef, useState } from "react"; -import type { - CreateRoomResponse, - IceConfig, - ServerMessage, +import { + DEFAULT_QUALITY_PROFILE_ID, + type CreateRoomResponse, + type IceConfig, + type ServerMessage, } from "../../shared/protocol"; import { AppHeader } from "../components/AppHeader"; import { @@ -109,7 +110,9 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { () => new URLSearchParams(window.location.search).get("relay") === "1", [], ); - const [qualityId, setQualityId] = useState("1080p60"); + const [qualityId, setQualityId] = useState( + DEFAULT_QUALITY_PROFILE_ID, + ); const [phase, setPhase] = useState("idle"); const [signalStatus, setSignalStatus] = useState("offline"); @@ -137,7 +140,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { const activeGenerationRef = useRef(null); const sourceSwitchRef = useRef(null); const qualityChangeRef = useRef(null); - const qualityIdRef = useRef("1080p60"); + const qualityIdRef = useRef(DEFAULT_QUALITY_PROFILE_ID); const picturePausedRef = useRef(false); const retiringStreamRef = useRef(null); @@ -302,6 +305,12 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { commitQuality(nextId); setDetails(captureDetails(activeStream)); + if (peerAssistedRef.current) { + signalRef.current?.send({ + type: "set-quality-profile", + qualityProfileId: nextId, + }); + } const results = await Promise.all( [...peersRef.current.values()].map((peer) => peer.updateProfile(profile)), ); @@ -507,6 +516,10 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { message.mediaMode === "peer-assisted" ) { peerAssistedRef.current = true; + signalRef.current?.send({ + type: "set-quality-profile", + qualityProfileId: qualityIdRef.current, + }); reconcilePeerAssistedChildren( message.mediaAssignment.childPeerIds, generation, @@ -812,10 +825,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { return { peerId, peer, - replaced: await peer.replaceStream( - captured, - QUALITY_PROFILES[qualityIdRef.current], - ), + replaced: await peer.replaceStream(captured), }; } catch { return { peerId, peer, replaced: false }; diff --git a/src/client/pages/ViewerPage.tsx b/src/client/pages/ViewerPage.tsx index ff22b08..aaa903d 100644 --- a/src/client/pages/ViewerPage.tsx +++ b/src/client/pages/ViewerPage.tsx @@ -7,10 +7,11 @@ import { VolumeX, } from "lucide-react"; import { useEffect, useMemo, useRef, useState } from "react"; -import type { - IceConfig, - MediaAssignment, - ServerMessage, +import { + DEFAULT_QUALITY_PROFILE_ID, + type IceConfig, + type MediaAssignment, + type ServerMessage, } from "../../shared/protocol"; import { AppHeader } from "../components/AppHeader"; import { @@ -22,7 +23,10 @@ import { import { StatsGrid } from "../components/StatsGrid"; import { getStableClientId } from "../lib/session"; import { SignalingClient } from "../lib/signaling"; -import { QUALITY_PROFILES } from "../media/quality"; +import { + QUALITY_PROFILES, + type QualityProfileId, +} from "../media/quality"; import type { PeerSnapshot, SignalConnectionState, @@ -65,6 +69,7 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps) let currentIceConfig: IceConfig | null = null; let currentHostOnline = false; let peerAssisted = false; + let currentQualityProfileId: QualityProfileId = DEFAULT_QUALITY_PROFILE_ID; let currentAssignment: MediaAssignment = { parentPeerId: null, childPeerIds: [], @@ -118,7 +123,7 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps) } viewerRelay = new ViewerRelay( currentIceConfig, - QUALITY_PROFILES["1080p60"], + QUALITY_PROFILES[currentQualityProfileId], { sendSignal: (targetPeerId, payload) => active && @@ -249,7 +254,11 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps) currentHostOnline = message.hostOnline; setRelayAvailable(message.iceConfig.relayAvailable); setHostOnline(message.hostOnline); - if (nextPeerAssisted) { + if (nextPeerAssisted && "qualityProfileId" in message) { + currentQualityProfileId = message.qualityProfileId; + void viewerRelay?.updateProfile( + QUALITY_PROFILES[currentQualityProfileId], + ); applyMediaAssignment(message.mediaAssignment); } if ( @@ -295,6 +304,15 @@ export function ViewerPage({ roomId, onAuthorizationRequired }: ViewerPageProps) } return; } + if (message.type === "quality-profile") { + if (peerAssisted) { + currentQualityProfileId = message.qualityProfileId; + void viewerRelay?.updateProfile( + QUALITY_PROFILES[currentQualityProfileId], + ); + } + return; + } if (message.type === "signal") { if ( peerAssisted && diff --git a/src/client/webrtc/host-peer.ts b/src/client/webrtc/host-peer.ts index 6ab4c12..60f83c6 100644 --- a/src/client/webrtc/host-peer.ts +++ b/src/client/webrtc/host-peer.ts @@ -33,13 +33,14 @@ export class HostPeer { private statsTimer: number | null = null; private disposed = false; private negotiating = false; + private senderMutationTail: Promise = Promise.resolve(); private snapshot: PeerSnapshot; constructor( readonly peerId: string, iceConfig: IceConfig, private stream: MediaStream, - private profile: QualityProfile, + private desiredProfile: QualityProfile, private readonly events: HostPeerEvents, private readonly forceRelay = false, ) { @@ -74,7 +75,12 @@ export class HostPeer { streams: [this.stream], }).sender; try { - await configureVideoSender(this.videoSender, this.profile); + await this.enqueueSenderMutation(async () => { + if (this.disposed || !this.videoSender) { + return; + } + await configureVideoSender(this.videoSender, this.desiredProfile); + }); } catch (error) { console.warn("Browser rejected preferred sender parameters", error); } @@ -88,74 +94,79 @@ export class HostPeer { return true; } - async replaceStream( - nextStream: MediaStream, - profile: QualityProfile, - ): Promise { + async replaceStream(nextStream: MediaStream): Promise { const nextVideoTrack = nextStream.getVideoTracks()[0]; - if ( - this.disposed || - !this.videoSender || - !this.audioSender || - !nextVideoTrack - ) { + if (this.disposed || !nextVideoTrack) { return false; } - const nextAudioTrack = nextStream.getAudioTracks()[0] ?? null; - const previousVideoTrack = this.videoSender.track; - const previousAudioTrack = this.audioSender.track; + return this.enqueueSenderMutation(async () => { + const videoSender = this.videoSender; + const audioSender = this.audioSender; + if (this.disposed || !videoSender || !audioSender) { + return false; + } - try { - await this.videoSender.replaceTrack(nextVideoTrack); - await this.audioSender.replaceTrack(nextAudioTrack); - } catch (error) { - await Promise.allSettled([ - this.videoSender.replaceTrack(previousVideoTrack), - this.audioSender.replaceTrack(previousAudioTrack), - ]); - this.setError(error, "切换共享源失败"); - return false; - } + const nextAudioTrack = nextStream.getAudioTracks()[0] ?? null; + const previousVideoTrack = videoSender.track; + const previousAudioTrack = audioSender.track; - if (this.disposed) { - return false; - } - this.stream = nextStream; - this.profile = profile; - this.statsAccumulator.bytes = null; - this.statsAccumulator.frames = null; - this.statsAccumulator.timestamp = null; - try { - await configureVideoSender(this.videoSender, this.profile); - } catch (error) { - console.warn("Browser rejected preferred sender parameters", error); - } - this.snapshot = { ...this.snapshot, error: null }; - this.emit(); - return true; + try { + await videoSender.replaceTrack(nextVideoTrack); + await audioSender.replaceTrack(nextAudioTrack); + } catch (error) { + await Promise.allSettled([ + videoSender.replaceTrack(previousVideoTrack), + audioSender.replaceTrack(previousAudioTrack), + ]); + this.setError(error, "切换共享源失败"); + return false; + } + + if (this.disposed) { + return false; + } + this.stream = nextStream; + this.statsAccumulator.bytes = null; + this.statsAccumulator.frames = null; + this.statsAccumulator.timestamp = null; + try { + await configureVideoSender(videoSender, this.desiredProfile); + } catch (error) { + console.warn("Browser rejected preferred sender parameters", error); + } + this.snapshot = { ...this.snapshot, error: null }; + this.emit(); + return true; + }); } - async updateProfile(profile: QualityProfile): Promise { - if (this.disposed || !this.videoSender) { - return false; - } - try { - await configureVideoSender(this.videoSender, profile); - } catch (error) { - this.setError(error, "调整画质失败"); - return false; - } + updateProfile(profile: QualityProfile): Promise { if (this.disposed) { - return false; + return Promise.resolve(false); } - this.profile = profile; - this.statsAccumulator.bytes = null; - this.statsAccumulator.frames = null; - this.statsAccumulator.timestamp = null; - this.snapshot = { ...this.snapshot, error: null }; - this.emit(); - return true; + this.desiredProfile = profile; + return this.enqueueSenderMutation(async () => { + const videoSender = this.videoSender; + if (this.disposed || !videoSender) { + return false; + } + try { + await configureVideoSender(videoSender, this.desiredProfile); + } catch (error) { + this.setError(error, "调整画质失败"); + return false; + } + if (this.disposed) { + return false; + } + this.statsAccumulator.bytes = null; + this.statsAccumulator.frames = null; + this.statsAccumulator.timestamp = null; + this.snapshot = { ...this.snapshot, error: null }; + this.emit(); + return true; + }); } async acceptSignal(payload: SignalPayload): Promise { @@ -242,6 +253,15 @@ export class HostPeer { this.connection.addEventListener("iceconnectionstatechange", () => this.emit()); } + private enqueueSenderMutation(operation: () => Promise): Promise { + const result = this.senderMutationTail.then(operation); + this.senderMutationTail = result.then( + () => undefined, + () => undefined, + ); + return result; + } + private async createOffer(restart: boolean): Promise { if (this.disposed) { return false; diff --git a/src/client/webrtc/viewer-relay.ts b/src/client/webrtc/viewer-relay.ts index 8da1223..9a8c0a8 100644 --- a/src/client/webrtc/viewer-relay.ts +++ b/src/client/webrtc/viewer-relay.ts @@ -18,7 +18,7 @@ export class ViewerRelay { constructor( private iceConfig: IceConfig, - private readonly profile: QualityProfile, + private desiredProfile: QualityProfile, private readonly events: ViewerRelayEvents, private readonly forceRelay = false, ) {} @@ -61,6 +61,19 @@ export class ViewerRelay { .catch(() => undefined); } + updateProfile(profile: QualityProfile): Promise { + if (this.disposed) { + return Promise.resolve(false); + } + this.desiredProfile = profile; + const result = this.syncQueue.then(() => this.syncProfile(profile)); + this.syncQueue = result.then( + () => undefined, + () => undefined, + ); + return result.catch(() => false); + } + async acceptSignal( fromPeerId: string, payload: SignalPayload, @@ -134,7 +147,7 @@ export class ViewerRelay { let replaced = false; try { - replaced = await peer.replaceStream(stream, this.profile); + replaced = await peer.replaceStream(stream); } catch { // Rebuilding below is safer than leaving a stopped upstream track. } @@ -151,6 +164,17 @@ export class ViewerRelay { this.startPeer(childPeerId, stream); } + private async syncProfile(profile: QualityProfile): Promise { + if (this.disposed || this.desiredProfile !== profile) { + return false; + } + const peer = this.peer; + if (!peer) { + return true; + } + return peer.updateProfile(this.desiredProfile); + } + private startPeer( childPeerId: string, stream: MediaStream, @@ -170,7 +194,7 @@ export class ViewerRelay { childPeerId, this.iceConfig, stream, - this.profile, + this.desiredProfile, { sendSignal: (targetPeerId, payload) => !this.disposed && diff --git a/src/server/signaling.ts b/src/server/signaling.ts index 9ef3ecb..5951a24 100644 --- a/src/server/signaling.ts +++ b/src/server/signaling.ts @@ -5,9 +5,11 @@ import type { Duplex } from "node:stream"; import WebSocket, { WebSocketServer } from "ws"; import { + DEFAULT_QUALITY_PROFILE_ID, MAX_SIGNAL_BYTES, decodeClientMessage, type ClientMessage, + type QualityProfileId, type Role, type ServerMessage, } from "../shared/protocol.js"; @@ -65,6 +67,10 @@ export class SignalingServer { private readonly socketsBySessionId = new Map(); private readonly viewerGraceTimers = new Map(); private readonly connectionIdsByViewer = new Map(); + private readonly qualityProfileIdsByRoom = new Map< + string, + QualityProfileId + >(); private readonly peerRelayTopology = new PeerRelayTopology(); private readonly now: () => number; private readonly authenticationTimeoutMs: number; @@ -348,6 +354,9 @@ export class SignalingServer { ...authenticatedMessage, mediaMode: "peer-assisted", mediaAssignment, + qualityProfileId: + this.qualityProfileIdsByRoom.get(participant.roomId) ?? + DEFAULT_QUALITY_PROFILE_ID, }); } else { this.send(socket, authenticatedMessage); @@ -446,6 +455,28 @@ export class SignalingServer { ), }); return; + case "set-quality-profile": + if (!this.options.peerAssistedMedia || authenticated.role !== "host") { + this.sendError( + socket, + "FORBIDDEN", + "Only a peer-assisted host may set the quality profile", + ); + return; + } + this.qualityProfileIdsByRoom.set( + authenticated.roomId, + message.qualityProfileId, + ); + for (const viewer of this.options.roomStore.getConnectedViewers( + authenticated.roomId, + )) { + this.sendToSession(viewer.sessionId, { + type: "quality-profile", + qualityProfileId: message.qualityProfileId, + }); + } + return; case "stop-sharing": case "close-room": if (authenticated.role !== "host") { @@ -729,6 +760,7 @@ export class SignalingServer { this.clearRoomGraceTimers(roomId); this.clearRoomConnectionIds(roomId); this.peerRelayTopology.deleteRoom(roomId); + this.qualityProfileIdsByRoom.delete(roomId); for (const sessionId of abandoned.sessionIds) { const socket = this.socketsBySessionId.get(sessionId); if (!socket) { @@ -744,6 +776,7 @@ export class SignalingServer { this.clearRoomGraceTimers(expired.roomId); this.clearRoomConnectionIds(expired.roomId); this.peerRelayTopology.deleteRoom(expired.roomId); + this.qualityProfileIdsByRoom.delete(expired.roomId); for (const sessionId of expired.sessionIds) { const socket = this.socketsBySessionId.get(sessionId); if (!socket) { diff --git a/src/shared/protocol.ts b/src/shared/protocol.ts index 6ef862d..77f52bd 100644 --- a/src/shared/protocol.ts +++ b/src/shared/protocol.ts @@ -25,6 +25,14 @@ export const roomCodeSchema = z export const roleSchema = z.enum(["host", "viewer"]); export type Role = z.infer; +export const qualityProfileIdSchema = z.enum([ + "1080p60", + "1080p30", + "720p30", +]); +export type QualityProfileId = z.infer; +export const DEFAULT_QUALITY_PROFILE_ID: QualityProfileId = "1080p60"; + const iceServerSchema = z .object({ urls: z.union([ @@ -125,6 +133,12 @@ export const clientMessageSchema = z.union([ }) .strict(), z.object({ type: z.literal("refresh-ice") }).strict(), + z + .object({ + type: z.literal("set-quality-profile"), + qualityProfileId: qualityProfileIdSchema, + }) + .strict(), z.object({ type: z.literal("stop-sharing") }).strict(), // Kept as a compatibility alias while previously deployed clients age out. z.object({ type: z.literal("close-room") }).strict(), @@ -163,6 +177,7 @@ const authenticatedMessageSchema = z.union([ ...authenticatedMessageShape, mediaMode: z.literal("peer-assisted"), mediaAssignment: mediaAssignmentSchema, + qualityProfileId: qualityProfileIdSchema, }) .strict(), ]); @@ -202,6 +217,12 @@ export const serverMessageSchema = z.union([ mediaAssignment: mediaAssignmentSchema, }) .strict(), + z + .object({ + type: z.literal("quality-profile"), + qualityProfileId: qualityProfileIdSchema, + }) + .strict(), z .object({ type: z.literal("ice-config"), diff --git a/tests/host-peer-tracks.test.ts b/tests/host-peer-tracks.test.ts index 36720bd..0c39392 100644 --- a/tests/host-peer-tracks.test.ts +++ b/tests/host-peer-tracks.test.ts @@ -9,12 +9,22 @@ const statsCallbacks: Array<() => void> = []; class FakeSender { failNextReplace = false; failNextSetParameters = false; + deferNextSetParameters = false; + readonly appliedMaxBitrates: Array = []; + private releaseSetParameters: (() => void) | null = null; readonly setParameters = vi.fn( - async (_parameters: RTCRtpSendParameters) => { + async (parameters: RTCRtpSendParameters) => { if (this.failNextSetParameters) { this.failNextSetParameters = false; throw new Error("setParameters failed"); } + if (this.deferNextSetParameters) { + this.deferNextSetParameters = false; + await new Promise((resolve) => { + this.releaseSetParameters = resolve; + }); + } + this.appliedMaxBitrates.push(parameters.encodings[0]?.maxBitrate); }, ); readonly replaceTrack = vi.fn(async (track: MediaStreamTrack | null) => { @@ -30,6 +40,11 @@ class FakeSender { getParameters(): RTCRtpSendParameters { return { encodings: [{}] } as RTCRtpSendParameters; } + + releaseDeferredSetParameters(): void { + this.releaseSetParameters?.(); + this.releaseSetParameters = null; + } } class FakePeerConnection { @@ -246,10 +261,7 @@ describe("HostPeer source replacement", () => { const nextVideo = createTrack("video", "next-video"); const nextAudio = createTrack("audio", "next-audio"); await expect( - peer.replaceStream( - createStream(nextVideo, nextAudio), - QUALITY_PROFILES["1080p60"], - ), + peer.replaceStream(createStream(nextVideo, nextAudio)), ).resolves.toBe(true); expect(connection.senders[0]?.track).toBe(nextVideo); @@ -272,7 +284,6 @@ describe("HostPeer source replacement", () => { await expect( peer.replaceStream( createStream(createTrack("video", "next-video"), nextAudio), - QUALITY_PROFILES["1080p30"], ), ).resolves.toBe(true); expect(connection.senders[1]?.track).toBe(nextAudio); @@ -291,7 +302,6 @@ describe("HostPeer source replacement", () => { await expect( peer.replaceStream( createStream(createTrack("video", "next-video"), null), - QUALITY_PROFILES["720p30"], ), ).resolves.toBe(true); @@ -320,6 +330,55 @@ describe("HostPeer source replacement", () => { }); }); + it("serializes initial sender configuration with a live profile update", async () => { + const peer = createPeer( + createStream( + createTrack("video", "video"), + createTrack("audio", "audio"), + ), + ); + + const starting = peer.start(); + const videoSender = FakePeerConnection.latest!.senders[0]!; + videoSender.deferNextSetParameters = true; + await vi.waitFor(() => + expect(videoSender.setParameters).toHaveBeenCalledTimes(1), + ); + + const updating = peer.updateProfile(QUALITY_PROFILES["1080p60"]); + await Promise.resolve(); + expect(videoSender.setParameters).toHaveBeenCalledTimes(1); + + videoSender.releaseDeferredSetParameters(); + await expect(starting).resolves.toBe(true); + await expect(updating).resolves.toBe(true); + expect(videoSender.appliedMaxBitrates).toEqual([3_000_000, 8_000_000]); + }); + + it("continues queued profile updates after initial configuration rejects", async () => { + const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined); + try { + const peer = createPeer( + createStream( + createTrack("video", "video"), + createTrack("audio", "audio"), + ), + ); + + const starting = peer.start(); + const videoSender = FakePeerConnection.latest!.senders[0]!; + videoSender.failNextSetParameters = true; + const updating = peer.updateProfile(QUALITY_PROFILES["1080p60"]); + + await expect(starting).resolves.toBe(true); + await expect(updating).resolves.toBe(true); + expect(videoSender.setParameters).toHaveBeenCalledTimes(2); + expect(videoSender.appliedMaxBitrates).toEqual([8_000_000]); + } finally { + warn.mockRestore(); + } + }); + it("can retry the selected quality after a sender update fails", async () => { const peer = createPeer( createStream( @@ -359,10 +418,7 @@ describe("HostPeer source replacement", () => { const nextVideo = createTrack("video", "next-video"); const nextAudio = createTrack("audio", "next-audio"); await expect( - peer.replaceStream( - createStream(nextVideo, nextAudio), - QUALITY_PROFILES["1080p60"], - ), + peer.replaceStream(createStream(nextVideo, nextAudio)), ).resolves.toBe(false); expect(videoSender.replaceTrack).toHaveBeenNthCalledWith(1, nextVideo); @@ -630,4 +686,43 @@ describe("ViewerRelay downstream ownership", () => { expect(connection.connectionState).toBe("closed"); await vi.waitFor(() => expect(targets.at(-1)).toBe("child-peer-two")); }); + + it("applies the latest profile to the current and future child", async () => { + const relay = new ViewerRelay( + { iceServers: [], expiresAt: null, relayAvailable: false }, + QUALITY_PROFILES["1080p60"], + { sendSignal: () => true }, + ); + relay.setChild("first-profile-child"); + relay.setStream(createStream(createTrack("video", "profile-video"), null)); + await vi.waitFor(() => + expect( + FakePeerConnection.latest?.senders[0]?.setParameters, + ).toHaveBeenCalled(), + ); + const firstConnection = FakePeerConnection.latest!; + + await expect( + relay.updateProfile(QUALITY_PROFILES["720p30"]), + ).resolves.toBe(true); + expect( + firstConnection.senders[0]?.setParameters.mock.calls.at(-1)?.[0], + ).toMatchObject({ + encodings: [{ maxBitrate: 3_000_000, maxFramerate: 30 }], + }); + + relay.setChild("second-profile-child"); + await vi.waitFor(() => + expect(FakePeerConnection.latest).not.toBe(firstConnection), + ); + const secondConnection = FakePeerConnection.latest!; + await vi.waitFor(() => + expect(secondConnection.senders[0]?.setParameters).toHaveBeenCalled(), + ); + expect( + secondConnection.senders[0]?.setParameters.mock.calls[0]?.[0], + ).toMatchObject({ + encodings: [{ maxBitrate: 3_000_000, maxFramerate: 30 }], + }); + }); }); diff --git a/tests/protocol.test.ts b/tests/protocol.test.ts index f91c1e8..d8f605c 100644 --- a/tests/protocol.test.ts +++ b/tests/protocol.test.ts @@ -132,6 +132,28 @@ describe("client signaling protocol", () => { true, ); }); + + it("accepts only a bounded quality profile update", () => { + expect( + clientMessageSchema.safeParse({ + type: "set-quality-profile", + qualityProfileId: "1080p30", + }).success, + ).toBe(true); + expect( + clientMessageSchema.safeParse({ + type: "set-quality-profile", + qualityProfileId: "1440p60", + }).success, + ).toBe(false); + expect( + clientMessageSchema.safeParse({ + type: "set-quality-profile", + qualityProfileId: "1080p30", + bitrate: 5_000_000, + }).success, + ).toBe(false); + }); }); describe("server signaling protocol", () => { @@ -187,6 +209,7 @@ describe("server signaling protocol", () => { parentPeerId: null, childPeerIds: ["viewer_12345678", "viewer_87654321"], }, + qualityProfileId: "1080p60", }; expect(serverMessageSchema.safeParse(peerAssisted).success).toBe(true); @@ -200,6 +223,14 @@ describe("server signaling protocol", () => { serverMessageSchema.safeParse({ ...authenticatedMessage(8), mediaMode: "peer-assisted", + qualityProfileId: "1080p60", + }).success, + ).toBe(false); + expect( + serverMessageSchema.safeParse({ + ...authenticatedMessage(8), + mediaMode: "peer-assisted", + mediaAssignment: peerAssisted.mediaAssignment, }).success, ).toBe(false); expect( @@ -215,6 +246,30 @@ describe("server signaling protocol", () => { }, }).success, ).toBe(false); + expect( + serverMessageSchema.safeParse({ + ...peerAssisted, + qualityProfileId: "1440p60", + }).success, + ).toBe(false); + expect( + serverMessageSchema.safeParse({ + ...authenticatedMessage(8), + qualityProfileId: "1080p60", + }).success, + ).toBe(false); + expect( + serverMessageSchema.safeParse({ + type: "quality-profile", + qualityProfileId: "720p30", + }).success, + ).toBe(true); + expect( + serverMessageSchema.safeParse({ + type: "quality-profile", + qualityProfileId: "1440p60", + }).success, + ).toBe(false); }); it.each([0, 1.5, MAX_VIEWERS_PER_ROOM_LIMIT + 1])( diff --git a/tests/server-signal.test.ts b/tests/server-signal.test.ts index 6f98c04..0604b0b 100644 --- a/tests/server-signal.test.ts +++ b/tests/server-signal.test.ts @@ -328,6 +328,94 @@ describe("WebSocket signaling", () => { expect(authenticated.hostOnline).toBe(false); }); + it("synchronizes a bounded quality profile across a peer-assisted room", async () => { + const harness = await startHarness({ peerAssistedMedia: true }); + const host = await openClient(harness.webSocketUrl); + const hostAuth = peerAssisted( + await authenticate(host, harness.room, "host", "quality-host"), + ); + expect(hostAuth.qualityProfileId).toBe("1080p60"); + + const viewer = await openClient(harness.webSocketUrl); + const viewerAuth = peerAssisted( + await authenticate(viewer, harness.room, "viewer", "quality-viewer"), + ); + expect(viewerAuth.qualityProfileId).toBe("1080p60"); + await host.inbox.next("media-assignment"); + + host.socket.send( + JSON.stringify({ + type: "set-quality-profile", + qualityProfileId: "1080p30", + }), + ); + expect(await viewer.inbox.next("quality-profile")).toEqual({ + type: "quality-profile", + qualityProfileId: "1080p30", + }); + + const lateViewer = await openClient(harness.webSocketUrl); + const lateViewerAuth = peerAssisted( + await authenticate( + lateViewer, + harness.room, + "viewer", + "quality-viewer-late", + ), + ); + expect(lateViewerAuth.qualityProfileId).toBe("1080p30"); + await host.inbox.next("media-assignment"); + + viewer.socket.send( + JSON.stringify({ + type: "set-quality-profile", + qualityProfileId: "720p30", + }), + ); + expect((await viewer.inbox.next("error")).code).toBe("FORBIDDEN"); + await lateViewer.inbox.expectNone(30); + }); + + it("keeps the P2P authenticated wire unchanged and forbids profile updates", async () => { + const harness = await startHarness(); + const host = await openClient(harness.webSocketUrl); + const hostAuth = await authenticate( + host, + harness.room, + "host", + "p2p-quality-host", + ); + expect("mediaMode" in hostAuth).toBe(false); + expect("qualityProfileId" in hostAuth).toBe(false); + + const viewer = await openClient(harness.webSocketUrl); + const viewerAuth = await authenticate( + viewer, + harness.room, + "viewer", + "p2p-quality-viewer", + ); + expect("mediaMode" in viewerAuth).toBe(false); + expect("qualityProfileId" in viewerAuth).toBe(false); + await host.inbox.next("peer-joined"); + + host.socket.send( + JSON.stringify({ + type: "set-quality-profile", + qualityProfileId: "1080p30", + }), + ); + expect((await host.inbox.next("error")).code).toBe("FORBIDDEN"); + viewer.socket.send( + JSON.stringify({ + type: "set-quality-profile", + qualityProfileId: "720p30", + }), + ); + expect((await viewer.inbox.next("error")).code).toBe("FORBIDDEN"); + await viewer.inbox.expectNone(30); + }); + it("routes offer and answer only between the host and the targeted viewer", async () => { const harness = await startHarness(); const secondRoom = harness.roomStore.createRoom(); @@ -804,6 +892,17 @@ describe("WebSocket signaling", () => { ); await host.inbox.next("media-assignment"); + host.socket.send( + JSON.stringify({ + type: "set-quality-profile", + qualityProfileId: "720p30", + }), + ); + expect(await viewer.inbox.next("quality-profile")).toEqual({ + type: "quality-profile", + qualityProfileId: "720p30", + }); + host.socket.send( JSON.stringify({ type: "signal", @@ -844,6 +943,7 @@ describe("WebSocket signaling", () => { parentPeerId: hostAuth.peerId, childPeerIds: [], }, + qualityProfileId: "720p30", }); const resumedHost = await openClient(harness.webSocketUrl); @@ -861,6 +961,7 @@ describe("WebSocket signaling", () => { parentPeerId: null, childPeerIds: [viewerAuth.peerId], }, + qualityProfileId: "720p30", }); expect(await reconnectedViewer.inbox.next("host-status")).toMatchObject({ online: true,