From 982094a7fa111816dd6ddb11ae50bdfeedd2dd7c Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Sat, 29 Aug 2026 00:27:21 +0800 Subject: [PATCH] feat(media): add uplink attribution diagnostics (#319) --- src/client/sfu/publisher.ts | 2 + src/client/webrtc/stats.ts | 27 ++++- src/server/hybrid-media-router.ts | 150 +++++++++++++++++++++++++- tests/client-state.test.ts | 67 ++++++++++++ tests/hybrid-media-router.test.ts | 170 +++++++++++++++++++++++++++++- tests/sfu-client.test.ts | 59 +++++++++++ 6 files changed, 467 insertions(+), 8 deletions(-) diff --git a/src/client/sfu/publisher.ts b/src/client/sfu/publisher.ts index 635d354..f686fdf 100644 --- a/src/client/sfu/publisher.ts +++ b/src/client/sfu/publisher.ts @@ -1079,6 +1079,8 @@ export class SfuPublisher { ); const metrics = { ...connectionMetrics, + bitrateKbps: + connectionMetrics.bitrateKbps ?? nativeQuality.bitrateKbps, nativeEdgeQualityState: nativeQuality.nativeEdgeQualityState, ...(nativeQuality.nativeEdgeQualityState === "unknown" ? {} diff --git a/src/client/webrtc/stats.ts b/src/client/webrtc/stats.ts index 34df359..3008199 100644 --- a/src/client/webrtc/stats.ts +++ b/src/client/webrtc/stats.ts @@ -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, }; } diff --git a/src/server/hybrid-media-router.ts b/src/server/hybrid-media-router.ts index dcc7d24..ee20f53 100644 --- a/src/server/hybrid-media-router.ts +++ b/src/server/hybrid-media-router.ts @@ -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, + 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)}` diff --git a/tests/client-state.test.ts b/tests/client-state.test.ts index 0c7908a..d0d6d57 100644 --- a/tests/client-state.test.ts +++ b/tests/client-state.test.ts @@ -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>, + ): number | null => { + const records: Array<[string, Record]> = [ + [ + "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, diff --git a/tests/hybrid-media-router.test.ts b/tests/hybrid-media-router.test.ts index 9975b96..ea1f7f7 100644 --- a/tests/hybrid-media-router.test.ts +++ b/tests/hybrid-media-router.test.ts @@ -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; + }; + 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, + 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([ + ["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([ + ["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([ + ["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, + ): 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, diff --git a/tests/sfu-client.test.ts b/tests/sfu-client.test.ts index 6f378e5..e9590cc 100644 --- a/tests/sfu-client.test.ts +++ b/tests/sfu-client.test.ts @@ -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 = []; + 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();