From 099e03fdf8838cda332ca16b086b74ced24040ac Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Mon, 12 Oct 2026 12:08:49 +0530 Subject: [PATCH] sfu: forward the dependency descriptor for VP8 simulcast (#4987) * sfu: forward the dependency descriptor for VP8 simulcast libwebrtc sends VP8 with a one byte payload descriptor when it sends the dependency descriptor, so a VP8 subscriber gets neither picture IDs nor the descriptor and finds frame references by sequence number alone. Under loss with temporal filtering it stalls until a key frame. DDForwarder rewrites the publisher's descriptor for each forwarded packet: frame numbers continue across simulcast layer switches and stream restarts, the structure goes out with the current layer's key frame, and the active decode targets follow the forwarded temporal layer. Co-Authored-By: Claude Opus 5.5 * sfu: keep VP8 descriptor frame numbers from going back Rebase frame numbers only when a new stream would go back past the last number sent: on a layer switch, a stream restart (also when it starts with a delta frame) and a new forwarder on the same downtrack, which now carries over the last number sent. Co-Authored-By: Claude Opus 5.5 * sfu: set VP8 descriptor decode targets from the packet's layer The active decode targets use the temporal layer the packet is forwarded at, not the selector's next layer on the last packet before a down switch. They also go out on key frames, combined with the publisher's own mask. Co-Authored-By: Claude Opus 5.5 * sfu: carry the current active decode targets on every DD packet The parser puts the stream's current mask on every packet, so a mask change on a dropped packet still reaches the VP8 descriptor forwarder, which no longer tracks the publisher's mask itself. Co-Authored-By: Claude Opus 5.5 * sfu: take the first DD active decode targets mask at any sequence number A publisher whose first packet has RTP sequence number 0 never had its key frame's mask applied, so every packet reported no active decode targets. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- pkg/sfu/buffer/dependencydescriptorparser.go | 13 +- .../buffer/dependencydescriptorparser_test.go | 64 ++++- pkg/sfu/forwarder.go | 20 ++ pkg/sfu/forwarder_test.go | 75 ++++++ pkg/sfu/testutils/data.go | 2 + pkg/sfu/videolayerselector/ddforwarder.go | 141 +++++++++++ .../videolayerselector/ddforwarder_test.go | 234 ++++++++++++++++++ 7 files changed, 541 insertions(+), 8 deletions(-) create mode 100644 pkg/sfu/videolayerselector/ddforwarder.go create mode 100644 pkg/sfu/videolayerselector/ddforwarder_test.go diff --git a/pkg/sfu/buffer/dependencydescriptorparser.go b/pkg/sfu/buffer/dependencydescriptorparser.go index 343eafc24..332c38a28 100644 --- a/pkg/sfu/buffer/dependencydescriptorparser.go +++ b/pkg/sfu/buffer/dependencydescriptorparser.go @@ -69,6 +69,7 @@ type DependencyDescriptorParser struct { // repeating the same structure (e. g. screen content), valid late/retransmitted // frames would be dropped as "earlier than current structure". structureChangeExtFrameNum uint64 + hasActiveDecodeTargets bool // the first mask is taken whatever its sequence number activeDecodeTargetsExtSeq uint64 activeDecodeTargetsMask uint32 frameChecker *FrameIntegrityChecker @@ -99,8 +100,11 @@ type ExtDependencyDescriptor struct { DecodeTargets []DependencyDescriptorDecodeTarget StructureUpdated bool ActiveDecodeTargetsUpdated bool - Integrity bool - ExtFrameNum uint64 + // the stream's current active decode targets, on every packet, also when the mask + // changed on a packet the subscriber does not get + ActiveDecodeTargetsMask uint32 + Integrity bool + ExtFrameNum uint64 // the frame number of the keyframe which the current frame depends on ExtKeyFrameNum uint64 @@ -262,7 +266,8 @@ func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescr // so don't need to notify max layer change here. } - if mask := ddVal.ActiveDecodeTargetsBitmask; mask != nil && extSeq > r.activeDecodeTargetsExtSeq { + if mask := ddVal.ActiveDecodeTargetsBitmask; mask != nil && (!r.hasActiveDecodeTargets || extSeq > r.activeDecodeTargetsExtSeq) { + r.hasActiveDecodeTargets = true r.activeDecodeTargetsExtSeq = extSeq if *mask != r.activeDecodeTargetsMask { r.activeDecodeTargetsMask = *mask @@ -285,6 +290,7 @@ func (r *DependencyDescriptorParser) Parse(pkt *rtp.Packet) (*ExtDependencyDescr extDD.DecodeTargets = r.decodeTargets extDD.ExtKeyFrameNum = r.structureExtFrameNum + extDD.ActiveDecodeTargetsMask = r.activeDecodeTargetsMask return extDD, videoLayer, nil } @@ -294,6 +300,7 @@ func (r *DependencyDescriptorParser) restart() { r.structure = nil r.structureExtFrameNum = 0 r.structureChangeExtFrameNum = 0 + r.hasActiveDecodeTargets = false r.activeDecodeTargetsExtSeq = 0 r.activeDecodeTargetsMask = 0 r.decodeTargets = r.decodeTargets[:0] diff --git a/pkg/sfu/buffer/dependencydescriptorparser_test.go b/pkg/sfu/buffer/dependencydescriptorparser_test.go index 7a09b013c..370c53023 100644 --- a/pkg/sfu/buffer/dependencydescriptorparser_test.go +++ b/pkg/sfu/buffer/dependencydescriptorparser_test.go @@ -75,19 +75,34 @@ func (f *ddTestFeeder) deltaFrame(frameNumber uint16) (*ExtDependencyDescriptor, return f.feed(frameNumber, f.structure.Templates[1], nil) } +// deltaFrameWithMask sends a delta frame that changes the active decode targets +func (f *ddTestFeeder) deltaFrameWithMask(frameNumber uint16, activeDecodeTargets uint32) (*ExtDependencyDescriptor, error) { + return f.feedWithMask(frameNumber, f.structure.Templates[1], nil, &activeDecodeTargets) +} + func (f *ddTestFeeder) feed( frameNumber uint16, template *dd.FrameDependencyTemplate, attachedStructure *dd.FrameDependencyStructure, +) (*ExtDependencyDescriptor, error) { + return f.feedWithMask(frameNumber, template, attachedStructure, nil) +} + +func (f *ddTestFeeder) feedWithMask( + frameNumber uint16, + template *dd.FrameDependencyTemplate, + attachedStructure *dd.FrameDependencyStructure, + activeDecodeTargets *uint32, ) (*ExtDependencyDescriptor, error) { f.t.Helper() ddVal := &dd.DependencyDescriptor{ - FirstPacketInFrame: true, - LastPacketInFrame: true, - FrameNumber: frameNumber, - FrameDependencies: template, - AttachedStructure: attachedStructure, + FirstPacketInFrame: true, + LastPacketInFrame: true, + FrameNumber: frameNumber, + FrameDependencies: template, + AttachedStructure: attachedStructure, + ActiveDecodeTargetsBitmask: activeDecodeTargets, } buf, err := (&dd.DependencyDescriptorExtension{Descriptor: ddVal, Structure: f.structure}).Marshal() require.NoError(f.t, err) @@ -187,6 +202,45 @@ func TestDependencyDescriptorParserExtKeyFrameNum(t *testing.T) { require.EqualValues(t, 4, extDD.ExtKeyFrameNum) } +// every packet carries the stream's current active decode targets, also after the mask +// changed on a packet the forwarder dropped +func TestDependencyDescriptorParserActiveDecodeTargetsMask(t *testing.T) { + f := newDDTestFeeder(t) + structure := newL1T1Structure(0) + + extDD, err := f.keyFrame(0, structure) + require.NoError(t, err) + require.EqualValues(t, 1, extDD.ActiveDecodeTargetsMask) + + extDD, err = f.deltaFrameWithMask(1, 0) + require.NoError(t, err) + require.EqualValues(t, 0, extDD.ActiveDecodeTargetsMask) + + extDD, err = f.deltaFrame(2) + require.NoError(t, err) + require.EqualValues(t, 0, extDD.ActiveDecodeTargetsMask) + + extDD, err = f.keyFrame(3, structure) + require.NoError(t, err) + require.EqualValues(t, 1, extDD.ActiveDecodeTargetsMask) +} + +// a first key frame at RTP sequence number 0 still makes its structure's decode targets +// active +func TestDependencyDescriptorParserActiveDecodeTargetsMaskAtSequenceZero(t *testing.T) { + f := newDDTestFeeder(t) + f.seq = 0xffff // the first packet goes out with sequence number 0 + structure := newL1T1Structure(0) + + extDD, err := f.keyFrame(0, structure) + require.NoError(t, err) + require.EqualValues(t, 1, extDD.ActiveDecodeTargetsMask) + + extDD, err = f.deltaFrame(1) + require.NoError(t, err) + require.EqualValues(t, 1, extDD.ActiveDecodeTargetsMask) +} + // a pooled ExtDependencyDescriptor carries the storage for its descriptor, so parsing a delta // frame must not allocate and a reused descriptor must not show the previous packet's values func TestDependencyDescriptorParserNoAlloc(t *testing.T) { diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index f0eed8824..ef5b0548a 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -281,6 +281,11 @@ type Forwarder struct { rtpMunger *RTPMunger vls videolayerselector.VideoLayerSelector + // set for VP8 with the dependency descriptor, forwards the descriptor to the subscriber + ddForwarder *videolayerselector.DDForwarder + // temporal layer of the packet being translated, the selector's current layer is + // already the next one on the last packet before a down switch + packetTemporal int32 codecMunger codecmunger.CodecMunger } @@ -413,11 +418,15 @@ func (f *Forwarder) DetermineCodec(codec webrtc.RTPCodecCapability, extensions [ return false } + prevDDForwarder := f.ddForwarder + f.ddForwarder = nil switch f.mime { case mime.MimeTypeVP8: f.isDDAvailable = ddAvailable(extensions) if f.isDDAvailable { f.codecMunger = codecmunger.NewVP8DD(f.logger) + // the downtrack keeps its SSRC, so the frame numbers continue + f.ddForwarder = videolayerselector.NewDDForwarderFromOther(prevDDForwarder, f.logger) } else { f.codecMunger = codecmunger.NewVP8FromOther(f.codecMunger, f.logger) } @@ -2193,6 +2202,7 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in // never carries the descriptor, so it does not count. f.logger.Infow("turning off dependency descriptor for VP8 temporal layers", "layer", layer) f.isDDAvailable = false + f.ddForwarder = nil f.codecMunger = codecmunger.NewVP8FromOther(f.codecMunger, f.logger) if f.started { // the munger takes its picture ID and TL0PICIDX from the first packet it sees @@ -2261,6 +2271,7 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in tp.ddBytes, tp.ddBytesLen, tp.ddBytesSpill = result.DDBytes, result.DDBytesLen, result.DDBytesSpill tp.marker = result.RTPMarker tp.isEndOfLayerFrame = f.isEndOfLayerFrame(extPkt) + f.packetTemporal = f.vls.GetCurrent().Temporal starting, err := f.getTranslationParamsCommon(extPkt, layer, &tp) tp.isStarting = starting @@ -2291,12 +2302,21 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in return tp, nil } + if f.ddForwarder != nil { + // after translateCodecHeader, which picks the packet's temporal layer + var ddResult videolayerselector.VideoLayerSelectorResult + if f.ddForwarder.Forward(extPkt, f.packetTemporal, &ddResult) { + tp.ddBytes, tp.ddBytesLen, tp.ddBytesSpill = ddResult.DDBytes, ddResult.DDBytesLen, ddResult.DDBytesSpill + } + } + return tp, nil } func (f *Forwarder) translateCodecHeader(extPkt *buffer.ExtPacket, tp *TranslationParams) error { // codec specific forwarding check and any needed packet munging tl := f.vls.SelectTemporal(extPkt) + f.packetTemporal = tl inputSize, codecBytes, codecBytesLen, err := f.codecMunger.UpdateAndGet( extPkt, tp.rtp.snOrdering == SequenceNumberOrderingOutOfOrder, diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index 6d82db34e..36f2e92b7 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -15,6 +15,7 @@ package sfu import ( + "encoding/hex" "testing" "github.com/pion/rtp/codecs" @@ -2114,6 +2115,80 @@ func TestForwarderVP8DDWithPictureID(t *testing.T) { require.True(t, f.isDDAvailable, "temporal layers still switch with the dependency descriptor") } +func TestForwarderVP8ForwardsDependencyDescriptor(t *testing.T) { + // an L3T3 key frame descriptor with its dependency structure, the capture the + // dependency descriptor package tests with + raw, err := hex.DecodeString("c1017280081485214eafffaaaa863cf0430c10c302afc0aaa0063c00430010c002a000a80006000040001d954926e082b04a0941b820ac1282503157f974000ca864330e222222eca8655304224230eca877530077004200ef008601df010d") + require.NoError(t, err) + keyFrameDescriptor := dd.DependencyDescriptor{} + _, err = (&dd.DependencyDescriptorExtension{Descriptor: &keyFrameDescriptor}).Unmarshal(raw) + require.NoError(t, err) + require.NotNil(t, keyFrameDescriptor.AttachedStructure) + + packet := func(sn uint16, frameNum uint64, isKeyFrame bool) *buffer.ExtPacket { + extPkt, err := testutils.GetTestExtPacketVP8DD( + &testutils.TestExtPacketParams{SequenceNumber: sn, Timestamp: 0xabcdef + uint32(frameNum)*3000, SSRC: 0x12345678, PayloadSize: 20, Marker: true, IsKeyFrame: isKeyFrame}, + &testutils.TestDDParams{FirstPacketInFrame: true, ExtFrameNum: frameNum, DTIs: "SSS", DecodeTargets: testutils.TestVP8DecodeTargets}, + ) + require.NoError(t, err) + descriptor := keyFrameDescriptor + descriptor.FrameNumber = uint16(frameNum) + if !isKeyFrame { + descriptor.AttachedStructure = nil + descriptor.ActiveDecodeTargetsBitmask = nil + } + extPkt.DependencyDescriptor.Descriptor = &descriptor + return extPkt + } + + f := NewForwarder(webrtc.RTPCodecTypeVideo, logger.GetLogger(), true, true, true, nil) + f.DetermineCodec(testutils.TestVP8Codec, []webrtc.RTPHeaderExtensionParameter{{URI: dd.ExtensionURI}}, livekit.VideoLayer_MODE_UNUSED) + f.vls.SetCurrent(buffer.VideoLayer{Spatial: 0, Temporal: 0}) + f.vls.SetTarget(buffer.VideoLayer{Spatial: 0, Temporal: 0}) + + // the key frame carries the structure, too big to hold inline + tp, err := f.GetTranslationParams(packet(10, 100, true), 0) + require.NoError(t, err) + require.False(t, tp.shouldDrop) + require.NotEmpty(t, tp.ddBytesSpill) + + tp, err = f.GetTranslationParams(packet(11, 101, false), 0) + require.NoError(t, err) + require.False(t, tp.shouldDrop) + require.NotEmpty(t, tp.ddInline()) + + // a restarted upstream receiver sets the codec again, the frame numbers continue on + // the same downtrack + f.DetermineCodec(testutils.TestVP8Codec, []webrtc.RTPHeaderExtensionParameter{{URI: dd.ExtensionURI}}, livekit.VideoLayer_MODE_UNUSED) + tp, err = f.GetTranslationParams(packet(12, 50, true), 0) + require.NoError(t, err) + require.False(t, tp.shouldDrop) + out := dd.DependencyDescriptor{} + _, err = (&dd.DependencyDescriptorExtension{Descriptor: &out, Structure: keyFrameDescriptor.AttachedStructure}).Unmarshal(tp.ddBytesSpill) + require.NoError(t, err) + require.Equal(t, uint16(102), out.FrameNumber) + + // the last packet of a TL2 frame that starts a down switch to TL0 is forwarded at TL2, + // its descriptor names the TL2 decode target + f.vls.SetCurrent(buffer.VideoLayer{Spatial: 0, Temporal: 2}) + f.vls.SetTarget(buffer.VideoLayer{Spatial: 0, Temporal: 0}) + tl2 := packet(13, 51, false) + tl2.Temporal = 2 + tp, err = f.GetTranslationParams(tl2, 0) + require.NoError(t, err) + require.False(t, tp.shouldDrop) + require.Equal(t, int32(0), f.vls.GetCurrent().Temporal, "the switch takes effect after this packet") + out = dd.DependencyDescriptor{} + _, err = (&dd.DependencyDescriptorExtension{Descriptor: &out, Structure: keyFrameDescriptor.AttachedStructure}).Unmarshal(tp.ddInline()) + require.NoError(t, err) + require.Equal(t, uint32(1<<0|1<<1|1<<2), *out.ActiveDecodeTargetsBitmask) + + // without the extension the forwarder writes no descriptor + f = NewForwarder(webrtc.RTPCodecTypeVideo, logger.GetLogger(), true, true, true, nil) + f.DetermineCodec(testutils.TestVP8Codec, nil, livekit.VideoLayer_MODE_UNUSED) + require.Nil(t, f.ddForwarder) +} + func TestForwarderPadDroppedSlot(t *testing.T) { f := NewForwarder(webrtc.RTPCodecTypeVideo, logger.GetLogger(), true, true, true, nil) f.DetermineCodec(testutils.TestVP8Codec, []webrtc.RTPHeaderExtensionParameter{{URI: dd.ExtensionURI}}, livekit.VideoLayer_MODE_UNUSED) diff --git a/pkg/sfu/testutils/data.go b/pkg/sfu/testutils/data.go index 1056e4119..6ac6a9f2b 100644 --- a/pkg/sfu/testutils/data.go +++ b/pkg/sfu/testutils/data.go @@ -143,6 +143,8 @@ func GetTestExtPacketVP8DD(params *TestExtPacketParams, ddParams *TestDDParams) }, DecodeTargets: ddParams.DecodeTargets, ExtFrameNum: ddParams.ExtFrameNum, + // all decode targets active, as the parser reports after a structure + ActiveDecodeTargetsMask: 1< d.extLastFrameNum { + d.hasLast = true + d.extLastFrameNum = extOutFrameNum + } + + // tell the subscriber which decode targets to expect: the publisher's active ones up to + // the forwarded temporal layer. On a key frame too, where the structure alone would make + // all of them active. + d.activeDecodeTargetsBitmask = 0 + for _, dt := range ddwdt.DecodeTargets { + if dt.Layer.Temporal <= temporal { + d.activeDecodeTargetsBitmask |= 1 << dt.Target + } + } + d.activeDecodeTargetsBitmask &= ddwdt.ActiveDecodeTargetsMask + d.descriptor = *descriptor + d.descriptor.FrameNumber = uint16(extOutFrameNum) + d.descriptor.ActiveDecodeTargetsBitmask = &d.activeDecodeTargetsBitmask + + if err := result.marshalDependencyDescriptorExtension(&dd.DependencyDescriptorExtension{ + Descriptor: &d.descriptor, + Structure: d.structure, + }); err != nil { + d.logger.Warnw("could not marshal dependency descriptor", err, "extFrameNum", extFrameNum) + return false + } + return true +} + +// rebase moves the frame numbers up when a new stream would go back past the last one sent +func (d *DDForwarder) rebase(extFrameNum uint64) { + if d.hasLast && extFrameNum+d.frameNumOffset <= d.extLastFrameNum { + d.frameNumOffset = d.extLastFrameNum + 1 - extFrameNum + } +} diff --git a/pkg/sfu/videolayerselector/ddforwarder_test.go b/pkg/sfu/videolayerselector/ddforwarder_test.go new file mode 100644 index 000000000..cc4a47a4a --- /dev/null +++ b/pkg/sfu/videolayerselector/ddforwarder_test.go @@ -0,0 +1,234 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package videolayerselector + +import ( + "encoding/hex" + "testing" + + "github.com/pion/rtp" + "github.com/stretchr/testify/require" + + "github.com/livekit/livekit-server/pkg/sfu/buffer" + dd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/dependencydescriptor" + "github.com/livekit/protocol/logger" +) + +// ddForwarderTest builds packets with the L3T3 fixture descriptor and decodes what a +// DDForwarder writes +type ddForwarderTest struct { + t *testing.T + keyFrameDescriptor dd.DependencyDescriptor + structure *dd.FrameDependencyStructure + decodeTargets []buffer.DependencyDescriptorDecodeTarget +} + +func newDDForwarderTest(t *testing.T) *ddForwarderTest { + raw, err := hex.DecodeString(dependencyDescriptorFixture) + require.NoError(t, err) + keyFrameDescriptor := dd.DependencyDescriptor{} + _, err = (&dd.DependencyDescriptorExtension{Descriptor: &keyFrameDescriptor}).Unmarshal(raw) + require.NoError(t, err) + require.NotNil(t, keyFrameDescriptor.AttachedStructure) + + return &ddForwarderTest{ + t: t, + keyFrameDescriptor: keyFrameDescriptor, + structure: keyFrameDescriptor.AttachedStructure, + decodeTargets: []buffer.DependencyDescriptorDecodeTarget{ + {Target: 2, Layer: buffer.VideoLayer{Temporal: 2}}, + {Target: 1, Layer: buffer.VideoLayer{Temporal: 1}}, + {Target: 0, Layer: buffer.VideoLayer{Temporal: 0}}, + }, + } +} + +func (d *ddForwarderTest) packet(ssrc uint32, extFrameNum uint64, isKeyFrame bool, restartGeneration int) *buffer.ExtPacket { + descriptor := d.keyFrameDescriptor + descriptor.FrameNumber = uint16(extFrameNum) + if !isKeyFrame { + descriptor.AttachedStructure = nil + descriptor.ActiveDecodeTargetsBitmask = nil + } + return &buffer.ExtPacket{ + Packet: &rtp.Packet{Header: rtp.Header{SSRC: ssrc}}, + IsKeyFrame: isKeyFrame, + DependencyDescriptor: &buffer.ExtDependencyDescriptor{ + Descriptor: &descriptor, + DecodeTargets: d.decodeTargets, + ExtFrameNum: extFrameNum, + RestartGeneration: restartGeneration, + // all decode targets active, as the parser reports after a structure + ActiveDecodeTargetsMask: 1<