diff --git a/scripts/peer-assisted-benchmark.ts b/scripts/peer-assisted-benchmark.ts index 525143f..ab1be0f 100644 --- a/scripts/peer-assisted-benchmark.ts +++ b/scripts/peer-assisted-benchmark.ts @@ -14,6 +14,7 @@ import { loadConfig } from "../src/server/config"; import { QUALITY_PROFILES, QUALITY_RESOLUTIONS, + qualitySettingsEqual, type QualityProfileId, type QualitySettings, } from "../src/client/media/quality"; @@ -80,6 +81,7 @@ interface PageObservation { maxAssignedChildren: number; firstDecodedAtEpochMs: number | null; firstRenderedAtEpochMs: number | null; + renderedFrames: number; connections: ConnectionObservation[]; } @@ -677,6 +679,7 @@ export function buildBenchmarkInitScript(options: { maxAssignedChildren: 0, firstDecodedAtEpochMs: null, firstRenderedAtEpochMs: null, + renderedFrames: 0, }; const connections = []; const descriptions = []; @@ -859,6 +862,7 @@ export function buildBenchmarkInitScript(options: { const onFrame = (now) => { const at = performance.timeOrigin + now; if (state.firstRenderedAtEpochMs === null) state.firstRenderedAtEpochMs = at; + state.renderedFrames += 1; video.requestVideoFrameCallback(onFrame); }; video.requestVideoFrameCallback(onFrame); @@ -1382,7 +1386,10 @@ async function runQualityControlSmoke( initialSettings: QualitySettings, signal: AbortSignal, ): Promise<{ + settingsPropagated: boolean; peerConnectionsStable: boolean; + senderReadbacksMatched: boolean; + viewersAdvanced: boolean; }> { const hostPage = pages[0]; if (!hostPage) { @@ -1395,7 +1402,7 @@ async function runQualityControlSmoke( degradationPreference: "balanced", } as const satisfies QualitySettings; - const waitForSettings = async (settings: QualitySettings): Promise => { + const waitForSettings = async (settings: QualitySettings): Promise => { const serializedSettings = JSON.stringify(JSON.stringify(settings)); await Promise.all( pages.map((page) => @@ -1409,13 +1416,21 @@ async function runQualityControlSmoke( ), ), ); + const snapshots = await Promise.all( + pages.map((page) => quickSnapshot(cdp, page)), + ); + return snapshots.every( + (page) => + page.qualitySettings !== null && + qualitySettingsEqual(page.qualitySettings, settings), + ); }; - const sendingLabels = new Set( - baseline - .filter((page) => page.assignment.childPeerIds.length > 0) - .map((page) => page.label), + const activeSenderCounts = new Map( + baseline.map((page) => [page.label, activeVideoEdgeCount(page, "send")]), + ); + const sendingPages = pages.filter( + (page) => (activeSenderCounts.get(page.label) ?? 0) > 0, ); - const sendingPages = pages.filter((page) => sendingLabels.has(page.label)); await Promise.all( sendingPages.map((page) => evaluate( @@ -1430,37 +1445,94 @@ async function runQualityControlSmoke( ), ), ); - const waitForReadback = async (label: "平衡" | "清晰"): Promise => { - const expected = JSON.stringify(`${label} / ${label}`); - await Promise.all( - sendingPages.map((page) => - waitForPage( + const waitForReadback = async ( + label: "平衡" | "清晰", + ): Promise => { + const matched = await Promise.all( + sendingPages.map(async (page) => { + const expectedCount = activeSenderCounts.get(page.label) ?? 0; + const expected = JSON.stringify(`${label} / ${label}`); + const predicate = `(() => { + const values = Array.from(document.querySelectorAll('.metric')) + .filter((metric) => metric.querySelector('dt')?.textContent?.trim() === '请求 / 应用优先级') + .map((metric) => metric.querySelector('dd')?.textContent?.trim()); + return values.length === ${expectedCount} && values.every((value) => value === ${expected}); + })()`; + await waitForPage( cdp, page, - `Array.from(document.querySelectorAll('.metric')).some((metric) => - metric.querySelector('dt')?.textContent?.trim() === '请求 / 应用优先级' && - metric.querySelector('dd')?.textContent?.trim() === ${expected} - )`, + predicate, 10_000, `${label} sender parameter readback`, signal, - ), - ), + ); + return evaluate(cdp, page, `Boolean(${predicate})`); + }), + ); + return ( + sendingPages.length > 0 && + matched.length === sendingPages.length && + matched.every(Boolean) ); }; + const viewerPages = pages.slice(1); + const waitForViewerProgress = async (): Promise => { + for (const page of viewerPages) { + await cdp.call("Page.bringToFront", {}, page.sessionId); + const before = [await progressSample(cdp, page)]; + const deadline = Date.now() + 10_000; + let advanced = false; + while (Date.now() < deadline) { + const after = [await progressSample(cdp, page)]; + if (everyViewerAdvanced(before, after)) { + advanced = true; + break; + } + await delay(100, signal); + } + if (!advanced) { + return false; + } + } + return viewerPages.length > 0; + }; - await applyQualityPreference(cdp, hostPage, "平衡", signal); - await waitForSettings(balanced); - await waitForReadback("平衡"); - await applyQualityPreference(cdp, hostPage, "清晰优先", signal); - await waitForSettings(initialSettings); - await waitForReadback("清晰"); + const runStep = async ( + label: "平衡" | "清晰优先", + readbackLabel: "平衡" | "清晰", + settings: QualitySettings, + ): Promise<{ + settingsPropagated: boolean; + senderReadbacksMatched: boolean; + viewersAdvanced: boolean; + }> => { + await cdp.call("Page.bringToFront", {}, hostPage.sessionId); + await applyQualityPreference(cdp, hostPage, label, signal); + const settingsPropagated = await waitForSettings(settings); + const senderReadbacksMatched = await waitForReadback(readbackLabel); + const viewersAdvanced = await waitForViewerProgress(); + return { + settingsPropagated, + senderReadbacksMatched, + viewersAdvanced, + }; + }; + + const balancedResult = await runStep("平衡", "平衡", balanced); + const clarityResult = await runStep("清晰优先", "清晰", initialSettings); const final = await Promise.all(pages.map((page) => quickSnapshot(cdp, page))); return { + settingsPropagated: + balancedResult.settingsPropagated && clarityResult.settingsPropagated, peerConnectionsStable: peerConnectionFingerprint(final) === baselineFingerprint, + senderReadbacksMatched: + balancedResult.senderReadbacksMatched && + clarityResult.senderReadbacksMatched, + viewersAdvanced: + balancedResult.viewersAdvanced && clarityResult.viewersAdvanced, }; } @@ -1471,6 +1543,25 @@ function totalDecodedFrames(page: PageObservation): number { ); } +export function everyViewerAdvanced( + before: PageObservation[], + after: PageObservation[], +): boolean { + const baselineViewers = before.filter((page) => page.role === "viewer"); + const currentByLabel = new Map(after.map((page) => [page.label, page])); + return ( + baselineViewers.length > 0 && + baselineViewers.every((baseline) => { + const current = currentByLabel.get(baseline.label); + return ( + current !== undefined && + totalDecodedFrames(current) > totalDecodedFrames(baseline) && + current.renderedFrames > baseline.renderedFrames + ); + }) + ); +} + function descendantsOf( rootPeerId: string, pages: PageObservation[], @@ -1677,8 +1768,8 @@ async function runCase( ); checks.push({ name: "quality-control-propagation", - passed: true, - actual: true, + passed: qualityControl.settingsPropagated, + actual: qualityControl.settingsPropagated, expected: "balanced and clarity settings reach every participant", }); checks.push({ @@ -1689,9 +1780,15 @@ async function runCase( }); checks.push({ name: "quality-control-sender-readback", - passed: true, - actual: true, - expected: "host and active relay display both sender preference readbacks", + passed: qualityControl.senderReadbacksMatched, + actual: qualityControl.senderReadbacksMatched, + expected: "every baseline active video sender displays matching preference readback", + }); + checks.push({ + name: "quality-control-viewer-progress", + passed: qualityControl.viewersAdvanced, + actual: qualityControl.viewersAdvanced, + expected: "every viewer decodes and renders frames after both setting changes", }); } if (config.recoveryViewerCount === viewerCount) { diff --git a/src/client/media/host-sfu-route.ts b/src/client/media/host-sfu-route.ts index 02dc763..c76a154 100644 --- a/src/client/media/host-sfu-route.ts +++ b/src/client/media/host-sfu-route.ts @@ -5,7 +5,10 @@ import type { ServerMessage, } from "../../shared/protocol"; import { SfuPublisher, type SfuConnectionConfig } from "../sfu/publisher"; -import type { QualityProfile } from "./quality"; +import type { + QualityProfile, + VideoSenderParameterReadback, +} from "./quality"; import { MediaRouteTransition, type RouteOperationToken, @@ -20,6 +23,7 @@ interface HostPublisherTransport { replaceStream(stream: MediaStream): Promise; updateProfile(profile: QualityProfile): Promise; getQualityWarning?(): string | null; + getSenderParameters?(): VideoSenderParameterReadback | null; disconnect(): Promise; } @@ -224,6 +228,12 @@ export class HostSfuRoute { : null; } + getSenderParameters(): VideoSenderParameterReadback | null { + return this.active?.active + ? (this.active.publisher.getSenderParameters?.() ?? null) + : null; + } + replaceStream(stream: MediaStream): Promise { return this.active ? this.active.publisher.replaceStream(stream).catch(() => false) diff --git a/src/client/pages/HostPage.tsx b/src/client/pages/HostPage.tsx index 9f6e44e..6b9b470 100644 --- a/src/client/pages/HostPage.tsx +++ b/src/client/pages/HostPage.tsx @@ -371,12 +371,12 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { commitQuality(nextProfile); setDetails(captureDetails(activeStream)); - if (peerAssistedRef.current) { + const roomSettingsSent = + !peerAssistedRef.current || signalRef.current?.send({ type: "set-quality-settings", qualitySettings: nextProfile, - }); - } + }) === true; const activeSfuRoute = hostSfuRouteRef.current; const [results, sfuUpdated] = await Promise.all([ Promise.all( @@ -396,11 +396,18 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { hostSfuRouteRef.current === activeSfuRoute ? (activeSfuRoute?.getQualityWarning() ?? null) : null; + const connectionWarning = + failed > 0 || !sfuUpdated + ? "画质已切换,但部分观看连接未能应用新参数" + : null; + const syncWarning = roomSettingsSent + ? null + : "房间画质同步将在信令重连后继续"; + const warning = [sfuWarning ?? connectionWarning, syncWarning] + .filter((message): message is string => message !== null) + .join(";"); setNotice( - sfuWarning ?? - (failed > 0 || !sfuUpdated - ? "画质已切换,但部分观看连接未能应用新参数" - : `画质已切换为 ${qualitySettingsLabel(nextProfile)}`), + warning || `画质已切换为 ${qualitySettingsLabel(nextProfile)}`, ); } } catch (error) { diff --git a/src/client/sfu/publisher.ts b/src/client/sfu/publisher.ts index 5773272..42204db 100644 --- a/src/client/sfu/publisher.ts +++ b/src/client/sfu/publisher.ts @@ -42,6 +42,7 @@ export class SfuPublisher { private video: PublishedTrack | null = null; private audio: PublishedTrack | null = null; private profile: QualityProfile | null = null; + private senderParameters: VideoSenderParameterReadback | null = null; private qualityWarning: string | null = null; private state: PublisherState = "idle"; private generation = 0; @@ -122,6 +123,10 @@ export class SfuPublisher { if (!this.owns(room, generation)) { return false; } + const senderParameters = await configurePublishedVideo(video, profile); + if (!this.owns(room, generation)) { + return false; + } let audio: PublishedTrack | null = null; if (audioTrack) { @@ -139,7 +144,7 @@ export class SfuPublisher { this.video = video; this.audio = audio; this.profile = profile; - this.qualityWarning = null; + this.retainSenderParameters(senderParameters); this.state = "active"; return true; } catch (error) { @@ -187,6 +192,7 @@ export class SfuPublisher { this.video = null; this.audio = null; this.profile = null; + this.senderParameters = null; this.qualityWarning = null; this.state = "prepared"; return true; @@ -202,7 +208,8 @@ export class SfuPublisher { } const sdk = this.sdk; const previousVideo = this.video; - if (!sdk || !previousVideo) { + const profile = this.profile; + if (!sdk || !previousVideo || !profile) { throw new Error("SFU publisher has no active video publication"); } @@ -244,10 +251,18 @@ export class SfuPublisher { this.audio = nextAudio; } + const senderParameters = await configurePublishedVideo( + previousVideo, + profile, + ); + if (!this.owns(room, generation)) { + return false; + } previousVideo.rawTrack = nextVideoTrack; if (previousAudio && nextAudioTrack) { previousAudio.rawTrack = nextAudioTrack; } + this.retainSenderParameters(senderParameters); return true; } catch (error) { if (!this.owns(room, generation)) { @@ -276,6 +291,14 @@ export class SfuPublisher { if (!this.owns(room, generation)) { return false; } + const senderParameters = await configurePublishedVideo( + previousVideo, + profile, + ); + if (!this.owns(room, generation)) { + return false; + } + this.retainSenderParameters(senderParameters); } return false; } catch (rollbackError) { @@ -307,7 +330,7 @@ export class SfuPublisher { return false; } this.profile = profile; - this.qualityWarning = senderParameterWarning(readback); + this.retainSenderParameters(readback); return true; } catch (error) { if (!this.owns(room, generation)) { @@ -318,10 +341,11 @@ export class SfuPublisher { ? `应用 SFU 发送参数失败:${error.message}` : "应用 SFU 发送参数失败"; try { - await configurePublishedVideo(video, previousProfile); + const readback = await configurePublishedVideo(video, previousProfile); if (!this.owns(room, generation)) { return false; } + this.senderParameters = readback; this.qualityWarning = failureWarning; return false; } catch (rollbackError) { @@ -338,6 +362,10 @@ export class SfuPublisher { return this.qualityWarning; } + getSenderParameters(): VideoSenderParameterReadback | null { + return this.senderParameters; + } + async disconnect(): Promise { const room = this.invalidate(); if (room) { @@ -393,6 +421,7 @@ export class SfuPublisher { this.video = null; this.audio = null; this.profile = null; + this.senderParameters = null; this.qualityWarning = null; return room; } @@ -413,6 +442,13 @@ export class SfuPublisher { this.terminalNotified = true; this.events.onDisconnected?.(); } + + private retainSenderParameters( + readback: VideoSenderParameterReadback, + ): void { + this.senderParameters = readback; + this.qualityWarning = senderParameterWarning(readback); + } } async function publishTrack( diff --git a/tests/peer-assisted-benchmark.test.ts b/tests/peer-assisted-benchmark.test.ts index 6074bc8..8e0b36f 100644 --- a/tests/peer-assisted-benchmark.test.ts +++ b/tests/peer-assisted-benchmark.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it } from "vitest"; import { activeVideoEdgeCount, buildBenchmarkInitScript, + everyViewerAdvanced, parseBenchmarkConfig, parseViewerCounts, summarizeSamples, @@ -31,7 +32,6 @@ function page( authenticateSentAtEpochMs: 900, signalingConnected: true, qualitySettings: lowQualitySettings, - senderParameterApplications: [], maxActiveOutboundMediaEdges: sendEdges, maxAssignedChildren: sendEdges, assignment: { @@ -40,6 +40,7 @@ function page( }, firstDecodedAtEpochMs: role === "viewer" ? 1_500 : null, firstRenderedAtEpochMs: role === "viewer" ? 1_550 : null, + renderedFrames: role === "viewer" ? 10 : 0, connections: [ ...Array.from({ length: sendEdges }, (_, index) => ({ index, @@ -193,6 +194,27 @@ describe("peer-assisted benchmark observations", () => { ).toBe(false); }); + it("requires every viewer to decode and render after a quality change", () => { + const before = [ + page("host", "host", 2, 0), + page("viewer", "viewer-1", 1, 1), + page("viewer", "viewer-2", 0, 1), + ]; + const after = structuredClone(before); + for (const viewer of after.filter((entry) => entry.role === "viewer")) { + viewer.connections.at(-1)!.receiveTotals!.framesTotal += 1; + viewer.renderedFrames += 1; + } + expect(everyViewerAdvanced(before, after)).toBe(true); + + after[2]!.renderedFrames = before[2]!.renderedFrames; + expect(everyViewerAdvanced(before, after)).toBe(false); + after[2]!.renderedFrames += 1; + after[1]!.connections.at(-1)!.receiveTotals!.framesTotal = + before[1]!.connections.at(-1)!.receiveTotals!.framesTotal; + expect(everyViewerAdvanced(before, after)).toBe(false); + }); + it("injects the requested deterministic capture dimensions", () => { const source = buildBenchmarkInitScript({ label: "host-1", @@ -208,5 +230,6 @@ describe("peer-assisted benchmark observations", () => { expect(source).toContain("getDisplayMedia"); expect(source).toContain("collectConnectionMetrics"); expect(source).toContain("set-quality-settings"); + expect(source).toContain("renderedFrames"); }); }); diff --git a/tests/sfu-client.test.ts b/tests/sfu-client.test.ts index a837fa0..d9f45b7 100644 --- a/tests/sfu-client.test.ts +++ b/tests/sfu-client.test.ts @@ -18,6 +18,11 @@ const livekit = vi.hoisted(() => { readonly getParameters = vi.fn(() => this.parameters); readonly setParameters = vi.fn( async (parameters: RTCRtpSendParameters): Promise => { + const failure = state.nextSenderParameterError; + if (failure) { + state.nextSenderParameterError = null; + throw failure; + } this.parameters = parameters; }, ); @@ -124,9 +129,11 @@ const livekit = vi.hoisted(() => { const state: { rooms: FakeRoom[]; connectGate: Promise | null; + nextSenderParameterError: Error | null; } = { rooms: [], connectGate: null, + nextSenderParameterError: null, }; return { @@ -210,6 +217,7 @@ function deferred(): { beforeEach(() => { livekit.state.rooms.length = 0; livekit.state.connectGate = null; + livekit.state.nextSenderParameterError = null; vi.stubGlobal("MediaStream", FakeMediaStream); }); @@ -246,6 +254,7 @@ describe("SfuPublisher", () => { ).resolves.toBe(true); const room = livekit.state.rooms[0]; + const sender = room.localParticipant.publications[0].track.sender; expect(room.localParticipant.publishTrack).toHaveBeenNthCalledWith(1, video, { source: Track.Source.ScreenShare, simulcast: false, @@ -259,12 +268,43 @@ describe("SfuPublisher", () => { source: Track.Source.ScreenShareAudio, dtx: false, }); + expect(sender.setParameters).toHaveBeenCalledOnce(); + expect(publisher.getSenderParameters()).toEqual({ + requested: { + maxBitrate: 8_000_000, + maxFramerate: 60, + scaleResolutionDownBy: 1, + degradationPreference: "maintain-resolution", + }, + applied: { + maxBitrate: 8_000_000, + maxFramerate: 60, + scaleResolutionDownBy: 1, + degradationPreference: "maintain-resolution", + }, + mismatches: [], + }); await expect(publisher.deactivate()).resolves.toBe(true); expect(room.localParticipant.unpublishTrack).toHaveBeenCalledTimes(2); expect(room.disconnect).not.toHaveBeenCalled(); }); + it("fails closed when initial sender configuration is rejected", async () => { + const disconnected = vi.fn(); + const publisher = new SfuPublisher({ onDisconnected: disconnected }); + await publisher.connect(connection); + livekit.state.nextSenderParameterError = new Error("parameters rejected"); + + await expect( + publisher.activate(stream(track("video", "video-1")), qualityProfile), + ).rejects.toThrow("parameters rejected"); + + expect(livekit.state.rooms[0].disconnect).toHaveBeenCalledWith(false); + expect(disconnected).toHaveBeenCalledOnce(); + expect(publisher.getSenderParameters()).toBeNull(); + }); + it("updates the active sender profile without republishing", async () => { const publisher = new SfuPublisher(); await publisher.connect(connection); @@ -365,6 +405,44 @@ describe("SfuPublisher", () => { expect(localTrack.currentTrack).toBe(previousVideo); }); + it("reapplies and retains the current sender settings after replacing video", async () => { + const publisher = new SfuPublisher(); + await publisher.connect(connection); + await publisher.activate(stream(track("video", "video-1")), qualityProfile); + const localTrack = livekit.state.rooms[0].localParticipant.publications[0].track; + + await expect( + publisher.replaceStream(stream(track("video", "video-2"))), + ).resolves.toBe(true); + + expect(localTrack.sender.setParameters).toHaveBeenCalledTimes(2); + expect(publisher.getSenderParameters()?.requested).toMatchObject({ + maxBitrate: 8_000_000, + maxFramerate: 60, + degradationPreference: "maintain-resolution", + }); + }); + + it("restores sender settings when replacement configuration is rejected", async () => { + const previousVideo = track("video", "video-1"); + const publisher = new SfuPublisher(); + await publisher.connect(connection); + await publisher.activate(stream(previousVideo), qualityProfile); + const localTrack = livekit.state.rooms[0].localParticipant.publications[0].track; + localTrack.sender.setParameters.mockRejectedValueOnce( + new Error("replacement parameters rejected"), + ); + + await expect( + publisher.replaceStream(stream(track("video", "video-2"))), + ).resolves.toBe(false); + + expect(localTrack.replaceTrack).toHaveBeenCalledTimes(2); + expect(localTrack.currentTrack).toBe(previousVideo); + expect(localTrack.sender.setParameters).toHaveBeenCalledTimes(3); + expect(publisher.getSenderParameters()?.applied.maxBitrate).toBe(8_000_000); + }); + it("disconnects fail-closed when replacement rollback also fails", async () => { const disconnected = vi.fn(); const previousVideo = track("video", "video-1");