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

fix(media): complete quality control verification

Browse Source
This commit is contained in:
AAA committed 2026-08-19 14:24:23 +08:00
1 parent e649bc636d
commit 5bffd5a9f6
6 files changed
+292 -41
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

scripts/peer-assisted-benchmark.ts
+125 -28
View File
@@ -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<void> => {
const waitForSettings = async (settings: QualitySettings): Promise<boolean> => {
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<void> => {
const expected = JSON.stringify(`${label} / ${label}`);
await Promise.all(
sendingPages.map((page) =>
waitForPage(
const waitForReadback = async (
label: "平衡" | "清晰",
): Promise<boolean> => {
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<boolean>(cdp, page, `Boolean(${predicate})`);
}),
);
return (
sendingPages.length > 0 &&
matched.length === sendingPages.length &&
matched.every(Boolean)
);
};
const viewerPages = pages.slice(1);
const waitForViewerProgress = async (): Promise<boolean> => {
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) {
src/client/media/host-sfu-route.ts
+11 -1
View File
@@ -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<boolean>;
updateProfile(profile: QualityProfile): Promise<boolean>;
getQualityWarning?(): string | null;
getSenderParameters?(): VideoSenderParameterReadback | null;
disconnect(): Promise<void>;
}
@@ -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<boolean> {
return this.active
? this.active.publisher.replaceStream(stream).catch(() => false)
src/client/pages/HostPage.tsx
+14 -7
View File
@@ -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) {
src/client/sfu/publisher.ts
+40 -4
View File
@@ -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<void> {
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(
tests/peer-assisted-benchmark.test.ts
+24 -1
View File
@@ -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");
});
});
tests/sfu-client.test.ts
+78
View File
@@ -18,6 +18,11 @@ const livekit = vi.hoisted(() => {
readonly getParameters = vi.fn(() => this.parameters);
readonly setParameters = vi.fn(
async (parameters: RTCRtpSendParameters): Promise<void> => {
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<void> | 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");