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

feat(media): add uplink attribution diagnostics (#319)

Browse Source
This commit is contained in:
AAA authored and GitHub committed 2026-08-29 00:27:21 +08:00
1 parent b129b3dbc0
commit 982094a7fa
6 files changed
+467 -8
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/client/sfu/publisher.ts
+2
View File
@@ -1079,6 +1079,8 @@ export class SfuPublisher {
);
const metrics = {
...connectionMetrics,
bitrateKbps:
connectionMetrics.bitrateKbps ?? nativeQuality.bitrateKbps,
nativeEdgeQualityState: nativeQuality.nativeEdgeQualityState,
...(nativeQuality.nativeEdgeQualityState === "unknown"
? {}
src/client/webrtc/stats.ts
+23 -4
View File
@@ -61,8 +61,9 @@ export interface StatsMediaSelector {
interface NativeSenderQualityPrevious {
timestamp: number;
durations: QualityLimitationDurations;
framesEncoded: number;
durations: QualityLimitationDurations | null;
framesEncoded: number | null;
bytesSent: number | null;
}
export interface NativeSenderQualityAccumulator {
@@ -83,6 +84,7 @@ export function collectNativeSenderQualityFromReport(
| "qualityLimitationReason"
| "sampleWindowMs"
| "intervalFramesEncoded"
| "bitrateKbps"
> {
const records: StatsRecord[] = [];
report.forEach((raw) => {
@@ -106,6 +108,7 @@ export function collectNativeSenderQualityFromReport(
const durations = qualityLimitationDurationsValue(record);
const reason = stringValue(record, "qualityLimitationReason");
const framesEncoded = numberValue(record, "framesEncoded");
const bytesSent = numberValue(record, "bytesSent");
const previous = accumulator.previousByStatsId.get(record.id);
const windowMs =
timestamp !== null && previous && timestamp > previous.timestamp
@@ -120,20 +123,34 @@ export function collectNativeSenderQualityFromReport(
const intervalFramesEncoded =
previous &&
framesEncoded !== null &&
previous.framesEncoded !== null &&
framesEncoded >= previous.framesEncoded
? framesEncoded - previous.framesEncoded
: null;
if (timestamp !== null && durations && framesEncoded !== null) {
const bitrateKbps =
previous &&
windowMs !== null &&
bytesSent !== null &&
previous.bytesSent !== null &&
bytesSent >= previous.bytesSent
? ((bytesSent - previous.bytesSent) * 8) / (windowMs / 1_000) / 1_000
: null;
if (timestamp !== null) {
accumulator.previousByStatsId.set(record.id, {
timestamp,
durations,
framesEncoded,
bytesSent,
});
} else {
accumulator.previousByStatsId.delete(record.id);
}
return { state, reason, windowMs, intervalFramesEncoded };
return { state, reason, windowMs, intervalFramesEncoded, bitrateKbps };
});
const bitrateKbps =
samples.length > 0 && samples.every((sample) => sample.bitrateKbps !== null)
? samples.reduce((total, sample) => total + sample.bitrateKbps!, 0)
: null;
if (
samples.length === 0 ||
samples.some(
@@ -148,6 +165,7 @@ export function collectNativeSenderQualityFromReport(
qualityLimitationReason: null,
sampleWindowMs: null,
intervalFramesEncoded: null,
bitrateKbps,
};
}
const degraded = samples.filter((sample) => sample.state === "degraded");
@@ -164,6 +182,7 @@ export function collectNativeSenderQualityFromReport(
(total, sample) => total + sample.intervalFramesEncoded!,
0,
),
bitrateKbps,
};
}
src/server/hybrid-media-router.ts
+147 -3
View File
@@ -107,6 +107,32 @@ type PrepareResult =
}
| { kind: "ready"; prepared: PreparedCandidate };
type QualitySampleReference =
| {
kind: "peer";
childPeerId: string;
routeRevision: number;
connectionId: string;
}
| {
kind: "sfu";
routeRevision: number;
publicationGeneration: string;
hostSessionId: string;
};
interface QualityCopyContext {
committedCopies: number;
candidateReservedCopies: number;
possibleCopies: number;
endpointCapacity: number;
operationReason: OperationSnapshot["reason"] | null;
candidateTransition:
| CandidatePlan["endpointTransition"]["kind"]
| null;
sampleRole: "active" | "candidate" | "unknown";
}
export type ActiveViewerMediaEdge =
| {
revision: number;
@@ -296,7 +322,25 @@ export class HybridMediaRouter {
if (!room || !controller) {
return false;
}
const before = controller.snapshot().revision;
const snapshot = controller.snapshot();
const before = snapshot.revision;
const copyContext = this.qualityCopyContext(
snapshot,
room.hostPeerId ?? "",
participant.peerId,
participant.role === "host"
? this.options.endpointMediaCopyCapacity
: Math.min(
room.advertisedCapacityByViewer.get(participant.peerId) ?? 0,
this.options.endpointMediaCopyCapacity,
),
{
kind: "peer",
childPeerId: message.childPeerId,
routeRevision: message.routeRevision,
connectionId: message.connectionId,
},
);
const input: SenderQualityEvidenceInput = {
parentPeerId: participant.peerId,
parentSessionId: participant.sessionId,
@@ -319,6 +363,8 @@ export class HybridMediaRouter {
this.debug(participant.roomId, "sender-quality-evidence", {
parent: this.debugPeer(participant.roomId, participant.peerId),
child: this.debugPeer(participant.roomId, message.childPeerId),
routeRevision: message.routeRevision,
...copyContext,
state: message.state,
reason: message.diagnostics.reason,
framesPerSecond: message.diagnostics.framesPerSecond,
@@ -370,10 +416,23 @@ export class HybridMediaRouter {
if (!room || !controller || participant.role !== "host") {
return false;
}
const operation = controller.snapshot().operation;
const snapshot = controller.snapshot();
const operation = snapshot.operation;
const candidateChildPeerId = operation?.childPeerId;
const candidateConnectionId = operation?.current?.connectionId;
const before = controller.snapshot().revision;
const before = snapshot.revision;
const copyContext = this.qualityCopyContext(
snapshot,
participant.peerId,
participant.peerId,
this.options.endpointMediaCopyCapacity,
{
kind: "sfu",
routeRevision: message.routeRevision,
publicationGeneration: message.publicationGeneration,
hostSessionId: participant.sessionId,
},
);
const input: SfuPublisherQualityEvidenceInput = {
hostPeerId: participant.peerId,
hostSessionId: participant.sessionId,
@@ -389,6 +448,8 @@ export class HybridMediaRouter {
);
if (result.accepted) {
this.debug(participant.roomId, "sfu-publisher-quality-evidence", {
routeRevision: message.routeRevision,
...copyContext,
state: message.state,
reason: message.diagnostics.reason,
framesPerSecond: message.diagnostics.framesPerSecond,
@@ -1951,6 +2012,89 @@ export class HybridMediaRouter {
);
}
private qualityCopyContext(
snapshot: RouteSnapshot<RouteResource>,
hostPeerId: string,
observedPeerId: string,
endpointCapacity: number,
sample: QualitySampleReference,
): QualityCopyContext {
let committedCopies = 0;
for (const edge of snapshot.upstreamByViewer.values()) {
if (
edge.kind === "peer" &&
edge.parentPeerId === observedPeerId &&
edge.physicalActive
) {
committedCopies += 1;
}
}
if (
observedPeerId === hostPeerId &&
snapshot.hostPublication?.physicalActive
) {
committedCopies += 1;
}
const operation = snapshot.operation;
const current = operation?.current;
const currentPlan = current
? operation.candidates[operation.cursor]
: undefined;
const candidateReservedCopies =
current?.tuple.kind === "peer" &&
current.tuple.parentPeerId === observedPeerId
? 1
: current?.tuple.kind === "sfu" &&
observedPeerId === hostPeerId &&
current.tuple.publication !== "reuse"
? 1
: 0;
const candidateSample =
sample.kind === "peer"
? current?.tuple.kind === "peer" &&
operation?.childPeerId === sample.childPeerId &&
current.tuple.parentPeerId === observedPeerId &&
current.revision === sample.routeRevision &&
current.connectionId === sample.connectionId
: current?.tuple.kind === "sfu" &&
observedPeerId === hostPeerId &&
current.revision === sample.routeRevision;
const activeEdge =
sample.kind === "peer"
? snapshot.upstreamByViewer.get(sample.childPeerId)
: undefined;
const activeSample =
sample.kind === "peer"
? sample.routeRevision === snapshot.revision &&
activeEdge?.kind === "peer" &&
activeEdge.physicalActive &&
activeEdge.usable &&
activeEdge.parentPeerId === observedPeerId &&
activeEdge.connectionId === sample.connectionId
: sample.routeRevision === snapshot.revision &&
snapshot.hostPublication?.physicalActive === true &&
snapshot.hostPublication.usable &&
snapshot.hostPublication.generation ===
sample.publicationGeneration &&
snapshot.hostPublication.hostSessionId === sample.hostSessionId;
return {
committedCopies,
candidateReservedCopies,
possibleCopies: committedCopies + candidateReservedCopies,
endpointCapacity,
operationReason: operation?.reason ?? null,
candidateTransition: currentPlan?.endpointTransition.kind ?? null,
sampleRole: candidateSample
? "candidate"
: activeSample
? "active"
: "unknown",
};
}
private debugTuple(roomId: string, tuple: CandidateTuple): string {
return tuple.kind === "peer"
? `p2p:${this.debugPeer(roomId, tuple.parentPeerId)}`
tests/client-state.test.ts
+67
View File
@@ -47,6 +47,8 @@ import { qualityEvidenceWindowFromMetrics } from "../src/client/media/viewer-qua
import {
collectConnectionMetrics,
collectConnectionMetricsFromReport,
collectNativeSenderQualityFromReport,
createNativeSenderQualityAccumulator,
createStatsAccumulator,
} from "../src/client/webrtc/stats.ts";
@@ -1508,6 +1510,71 @@ describe("client signaling recovery policy", () => {
});
describe("WebRTC stats parsing", () => {
it("sums sender layers and rebases incomplete bitrate windows", () => {
const accumulator = createNativeSenderQualityAccumulator();
const sample = (
timestamp: number,
bytesByLayer: Readonly<Record<string, number>>,
): number | null => {
const records: Array<[string, Record<string, unknown>]> = [
[
"source",
{
id: "source",
type: "media-source",
timestamp,
trackIdentifier: "track",
},
],
];
for (const [id, bytesSent] of Object.entries(bytesByLayer)) {
records.push([
id,
{
id,
type: "outbound-rtp",
kind: "video",
timestamp,
mediaSourceId: "source",
bytesSent,
framesEncoded: timestamp / 20,
qualityLimitationReason: "none",
qualityLimitationDurations: {
none: timestamp / 1_000,
bandwidth: 0,
cpu: 0,
other: 0,
},
},
]);
}
return collectNativeSenderQualityFromReport(
new Map(records) as unknown as RTCStatsReport,
"track",
accumulator,
).bitrateKbps;
};
expect(sample(1_000, { low: 10_000, high: 20_000 })).toBeNull();
expect(sample(3_000, { low: 110_000, high: 220_000 })).toBe(1_200);
expect(
sample(5_000, { low: 210_000, high: 420_000, mid: 50_000 }),
).toBeNull();
expect(
sample(7_000, { low: 310_000, high: 620_000, mid: 150_000 }),
).toBe(1_600);
expect(
sample(9_000, { low: 10_000, high: 820_000, mid: 250_000 }),
).toBeNull();
expect(
sample(11_000, {
low: 110_000,
high: 1_020_000,
mid: 350_000,
}),
).toBe(1_600);
});
it("binds media, remote evidence, transport, and pair without guessing", async () => {
const entry = (
id: string,
tests/hybrid-media-router.test.ts
+169 -1
View File
@@ -234,6 +234,123 @@ async function establishSfuRoom(
}
describe("HybridMediaRouter v9 runtime", () => {
it("projects exact endpoint copy context from the current snapshot", async () => {
const { router } = harness(2);
const internal = router as unknown as {
qualityCopyContext(
snapshot: unknown,
hostPeerId: string,
observedPeerId: string,
endpointCapacity: number,
sample: unknown,
): Record<string, unknown>;
};
const edge = (parentPeerId: string, connectionId: string) => ({
kind: "peer", childSessionId: "child-session", parentPeerId,
parentSessionId: "parent-session", transport: "direct", connectionId,
usable: true, physicalActive: true,
});
const operation = (tuple: unknown, transition: unknown) => ({
childPeerId: "candidate", childSessionId: "candidate-session",
demandPeerId: "candidate", reason: "quality-convergence",
baseRevision: 7, factVersion: 1, candidates: [{ tuple, endpointTransition: transition }],
cursor: 0, deadlineAtMs: 20_000, wakeAtMs: 20_000,
current: { tuple, revision: 8, connectionId: "candidate-connection" },
});
const snapshot = (
upstreamByViewer: Map<string, unknown>,
op?: unknown,
hostPublication: unknown = null,
) => ({
revision: 7, paused: false, factVersion: 1, upstreamByViewer,
hostPublication, ...(op ? { operation: op } : {}),
});
const project = (
state: unknown,
observedPeerId: string,
endpointCapacity: number,
sample: unknown,
) => internal.qualityCopyContext(
state, "host", observedPeerId, endpointCapacity, sample,
);
const hostEdges = new Map<string, unknown>([
["a", edge("host", "a-connection")],
["b", edge("host", "b-connection")],
]);
const peerCandidate = {
kind: "peer", parentPeerId: "host", transport: "direct",
};
const sfuSample = {
kind: "sfu", routeRevision: 8,
publicationGeneration: "publication", hostSessionId: "host-session",
};
const publication = {
generation: "publication", hostSessionId: "host-session",
connectionId: "publication-connection", usable: true,
physicalActive: true, resource: {},
};
try {
expect(project(
snapshot(hostEdges, operation(peerCandidate, {
kind: "overlap", producerPeerId: "host",
})),
"host", 2,
{ kind: "peer", childPeerId: "candidate", routeRevision: 8,
connectionId: "candidate-connection" },
)).toMatchObject({
committedCopies: 2, candidateReservedCopies: 1, possibleCopies: 3,
endpointCapacity: 2, operationReason: "quality-convergence",
candidateTransition: "overlap", sampleRole: "candidate",
});
const oneHostEdge = new Map<string, unknown>([
["a", edge("host", "a-connection")],
]);
const sfuCases = [
{
state: snapshot(hostEdges, operation(
{ kind: "sfu", publication: "create" },
{ kind: "overlap", producerPeerId: "host" },
)),
expected: { committedCopies: 2, candidateReservedCopies: 1,
possibleCopies: 3, candidateTransition: "overlap" },
},
{
state: snapshot(oneHostEdge, operation(
{ kind: "sfu", publication: "reuse" }, { kind: "none" },
), publication),
expected: { committedCopies: 2, candidateReservedCopies: 0,
possibleCopies: 2, candidateTransition: "none" },
},
];
for (const testCase of sfuCases) {
expect(project(testCase.state, "host", 2, sfuSample)).toMatchObject({
...testCase.expected, sampleRole: "candidate",
});
}
const relayEdges = new Map<string, unknown>([
["a", edge("relay", "relay-a")],
["b", edge("relay", "relay-b")],
]);
const relaySample = {
kind: "peer", childPeerId: "a", routeRevision: 7,
connectionId: "relay-a",
};
expect(project(snapshot(relayEdges), "relay", 3, relaySample)).toMatchObject({
committedCopies: 2, candidateReservedCopies: 0, possibleCopies: 2,
endpointCapacity: 3, sampleRole: "active",
});
expect(project(
snapshot(relayEdges), "relay", 3,
{ ...relaySample, routeRevision: 6 },
).sampleRole).toBe("unknown");
} finally {
await router.close();
}
});
it("keeps the controller diagnostic label until a departed relay is pruned", async () => {
const { store, sent, router } = harness(2);
try {
@@ -487,12 +604,22 @@ describe("HybridMediaRouter v9 runtime", () => {
revision: secondPrepare.revision,
phase: "prepare",
});
router.setViewerRelayCapacity(second, 2);
router.setViewerRelayCapacity(second, 3);
const firstEdge = router.resolveActiveViewerMediaEdge(room.roomId, first.peerId)!;
const secondEdge = router.resolveActiveViewerMediaEdge(room.roomId, second.peerId)!;
expect(firstEdge.upstream).toEqual({ kind: "peer", peerId: host.peerId });
expect(secondEdge.upstream).toEqual({ kind: "peer", peerId: host.peerId });
const debug = vi.spyOn(
router as unknown as {
debug(
roomId: string,
event: string,
details: Record<string, unknown>,
): void;
},
"debug",
);
expect(
router.observeSenderQualityEvidence(host, {
type: "sender-quality-evidence",
@@ -506,6 +633,20 @@ describe("HybridMediaRouter v9 runtime", () => {
diagnostics: senderDiagnostics("healthy"),
}),
).toBe(true);
expect(debug).toHaveBeenCalledWith(
room.roomId,
"sender-quality-evidence",
expect.objectContaining({
routeRevision: secondEdge.revision,
committedCopies: 2,
candidateReservedCopies: 0,
possibleCopies: 2,
endpointCapacity: 2,
operationReason: null,
candidateTransition: null,
sampleRole: "active",
}),
);
expect(
router.observeSenderQualityEvidence(host, {
type: "sender-quality-evidence",
@@ -542,6 +683,33 @@ describe("HybridMediaRouter v9 runtime", () => {
);
const qualityPrepare = preparedFor(sent, first.sessionId)!;
expect(qualityPrepare.candidate.qualityProbe).toBe(true);
expect(
router.observeSenderQualityEvidence(second, {
type: "sender-quality-evidence",
childPeerId: first.peerId,
connectionId: qualityPrepare.candidate.connectionId,
rtpStatsId: "candidate-rtp",
trackIdentifier: "track",
sampleTimestampMs: 200,
routeRevision: qualityPrepare.revision,
state: "healthy",
diagnostics: senderDiagnostics("healthy"),
}),
).toBe(true);
expect(debug).toHaveBeenCalledWith(
room.roomId,
"sender-quality-evidence",
expect.objectContaining({
routeRevision: qualityPrepare.revision,
committedCopies: 0,
candidateReservedCopies: 1,
possibleCopies: 1,
endpointCapacity: 2,
operationReason: "quality-convergence",
candidateTransition: "none",
sampleRole: "candidate",
}),
);
router.handleRouteReady(first, {
type: "route-ready",
revision: qualityPrepare.revision,
tests/sfu-client.test.ts
+59
View File
@@ -413,6 +413,40 @@ function senderReport(
]);
}
function simulcastSenderReport(
trackId: string,
timestamp: number,
lowBytesSent: number,
highBytesSent: number,
): RTCStatsReport {
const layer = (id: string, bytesSent: number, framesEncoded: number) => ({
id,
type: "outbound-rtp",
timestamp,
kind: "video",
mediaSourceId: "video-source",
bytesSent,
framesEncoded,
qualityLimitationReason: "none",
qualityLimitationDurations: {
none: timestamp / 1_000,
bandwidth: 0,
cpu: 0,
other: 0,
},
});
return statsReport([
layer("video-low", lowBytesSent, timestamp / 40),
layer("video-high", highBytesSent, timestamp / 20),
{
id: "video-source",
type: "media-source",
timestamp,
trackIdentifier: trackId,
},
]);
}
function audioSenderReport(
trackId: string,
timestamp: number,
@@ -481,6 +515,31 @@ afterEach(() => {
});
describe("SfuPublisher", () => {
it("reports the aggregate bitrate of active simulcast layers", async () => {
vi.useFakeTimers();
const updates: Array<ConnectionMetrics | null> = [];
const publisher = new SfuPublisher({
onStats: (metrics) => updates.push(metrics),
});
const video = track("video", "video-1");
await publisher.connect(connection);
await publisher.activate(stream(video), qualityProfile);
const sender = livekit.state.rooms[0].localParticipant.publications[0].track
.sender;
sender.getStats
.mockResolvedValueOnce(
simulcastSenderReport(video.id, 1_000, 100_000, 200_000),
)
.mockResolvedValueOnce(
simulcastSenderReport(video.id, 3_000, 300_000, 600_000),
);
await vi.advanceTimersByTimeAsync(2_000);
expect(updates.at(-1)?.bitrateKbps).toBeNull();
await vi.advanceTimersByTimeAsync(2_000);
expect(updates.at(-1)?.bitrateKbps).toBe(2_400);
});
it("enables balanced SFU adaptation after startup frames", async () => {
vi.useFakeTimers();
const publisher = new SfuPublisher();