From 5655670ae69e175a5957aff7b079d344517ac5fc Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Tue, 18 Aug 2026 18:56:33 +0800 Subject: [PATCH] Strip packet trailers from every VP9 layer frame VP9 SVC sends each spatial layer as its own encoded frame, so a picture carries one trailer per layer, but only the top layer's last packet has the RTP marker bit set. Fix https://github.com/livekit/egress/issues/1347 --- pkg/sfu/downtrack.go | 20 ++++-- .../downtrack_downstream_integration_test.go | 63 ++++++++++++++++++- pkg/sfu/export_test.go | 10 +-- pkg/sfu/forwarder.go | 18 ++++++ pkg/sfu/forwarder_test.go | 34 ++++++++++ pkg/sfu/packettrailer/packet_trailer.go | 8 ++- pkg/sfu/sequencer.go | 7 +++ pkg/sfu/sequencer_test.go | 19 +++--- 8 files changed, 159 insertions(+), 20 deletions(-) diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index b0326f4a9..1b993158c 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -1059,9 +1059,10 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) int32 { } payload = payload[:len(tp.codecBytes)+n] + trailerStripped := 0 if d.params.StripPacketTrailer { - if strip := packettrailer.StripTrailer(payload, tp.marker); strip > 0 { - payload = payload[:len(payload)-strip] + if trailerStripped = packettrailer.StripTrailer(payload, tp.marker || tp.isEndOfLayerFrame); trailerStripped > 0 { + payload = payload[:len(payload)-trailerStripped] } } @@ -1132,6 +1133,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) int32 { tp.incomingHeaderSize, tp.ddBytes, actBytes, + trailerStripped, ) } @@ -2243,9 +2245,17 @@ func (d *DownTrack) retransmitPacket(epm *extPacketMeta, sourcePkt []byte, isPro payload = payload[:rtxOffset+int(epm.numCodecBytesOut)+len(pkt.Payload)-int(epm.numCodecBytesIn)] } - if d.params.StripPacketTrailer { - if strip := packettrailer.StripTrailer(payload[rtxOffset:], epm.marker); strip > 0 { - payload = payload[:len(payload)-strip] + // replay the strip done on the original transmission to keep the retransmitted + // payload byte identical to it + if epm.trailerStripped != 0 { + if int(epm.trailerStripped) > len(payload)-rtxOffset { + d.params.Logger.Warnw( + "recorded packet trailer size overflows payload", errPayloadOverflow, + "trailerStripped", epm.trailerStripped, + "payloadSize", len(payload)-rtxOffset, + ) + } else { + payload = payload[:len(payload)-int(epm.trailerStripped)] } } diff --git a/pkg/sfu/downtrack_downstream_integration_test.go b/pkg/sfu/downtrack_downstream_integration_test.go index 2d52d35ca..c711efdbf 100644 --- a/pkg/sfu/downtrack_downstream_integration_test.go +++ b/pkg/sfu/downtrack_downstream_integration_test.go @@ -66,6 +66,7 @@ import ( "github.com/livekit/livekit-server/pkg/sfu/bwe/sendsidebwe" "github.com/livekit/livekit-server/pkg/sfu/ccutils" "github.com/livekit/livekit-server/pkg/sfu/pacer" + "github.com/livekit/livekit-server/pkg/sfu/packettrailer" "github.com/livekit/livekit-server/pkg/sfu/sfufakes" "github.com/livekit/livekit-server/pkg/sfu/streamallocator" "github.com/livekit/livekit-server/pkg/sfu/testutils" @@ -557,6 +558,7 @@ func TestDownTrackRetransmitsPacketsAsIs(t *testing.T) { uint64(ts), 0, raw, + 0, ) require.NoError(t, err) targetSN++ @@ -635,7 +637,7 @@ func TestDownTrackRetransmitsPacketsViaRTX(t *testing.T) { ts := uint32(700000 + i*3000) wants[targetSN] = want{payload: payload, ts: ts} - _, err = dh.dt.RetransmitForTest(uint64(src.SequenceNumber), targetSN, ts, uint64(ts), 0, raw) + _, err = dh.dt.RetransmitForTest(uint64(src.SequenceNumber), targetSN, ts, uint64(ts), 0, raw, 0) require.NoError(t, err) targetSN++ } @@ -673,6 +675,65 @@ func TestDownTrackRetransmitsPacketsViaRTX(t *testing.T) { } } +// TestDownTrackReplaysPacketTrailerStripOnRetransmit verifies that a packet whose +// packet trailer was stripped when it was forwarded is retransmitted with the same +// bytes removed, i. e. the retransmitted media payload is byte identical to the +// original transmission. +func TestDownTrackReplaysPacketTrailerStripOnRetransmit(t *testing.T) { + h := buildVNet(t) + factory := buffer.NewFactoryOfBufferFactory(500, 500).CreateBufferFactory() + cp := &capturingPacer{inner: pacer.NewPassThrough(logger.GetLogger(), newNullBWE())} + t.Cleanup(cp.Stop) + + dh := newBoundDownTrack(t, h, factory, vp8CodecParams, cp, mediaEngineConfig{video: true}) + require.NotZero(t, dh.dt.SSRCRTX(), "RTX ssrc should be negotiated") + + video := distinctivePayload(11, 40) + trailer := lktsTrailer() + src := &rtp.Packet{ + Header: rtp.Header{ + Version: 2, + PayloadType: 96, + SequenceNumber: 3000, + Timestamp: 270000, + SSRC: 0x44444444, + }, + Payload: append(append([]byte{}, video...), trailer...), + } + raw, err := src.Marshal() + require.NoError(t, err) + + osn := uint16(60000) + _, err = dh.dt.RetransmitForTest( + uint64(src.SequenceNumber), + osn, + src.Timestamp, + uint64(src.Timestamp), + 0, + raw, + uint8(len(trailer)), + ) + require.NoError(t, err) + + require.Eventually(t, func() bool { + return len(cp.rtxPackets()) >= 1 + }, 5*time.Second, 20*time.Millisecond, "expected RTX packet to be emitted") + + pr := cp.rtxPackets()[0] + require.EqualValues(t, osn, binary.BigEndian.Uint16(pr.payload[0:2])) + require.Equal(t, video, pr.payload[2:], "retransmitted payload must not carry the packet trailer") +} + +// lktsTrailer builds a 15-byte LKTS packet trailer carrying a user timestamp TLV. +func lktsTrailer() []byte { + trailer := []byte{0x01 ^ 0xFF, 8 ^ 0xFF} + for i := 0; i < 8; i++ { + trailer = append(trailer, byte(i)^0xFF) + } + trailer = append(trailer, 15^0xFF) + return append(trailer, packettrailer.Magic[:]...) +} + // ----------------------------------------------------------------------------- // WritePaddingRTP: padding-only packets // ----------------------------------------------------------------------------- diff --git a/pkg/sfu/export_test.go b/pkg/sfu/export_test.go index da31e2b75..449289e20 100644 --- a/pkg/sfu/export_test.go +++ b/pkg/sfu/export_test.go @@ -56,13 +56,15 @@ func (d *DownTrack) RetransmitForTest( extTimestamp uint64, layer int8, sourcePkt []byte, + trailerStripped uint8, ) (int, error) { epm := extPacketMeta{ packetMeta: packetMeta{ - sourceSeqNo: sourceSeqNo, - targetSeqNo: targetSeqNo, - timestamp: timestamp, - layer: layer, + sourceSeqNo: sourceSeqNo, + targetSeqNo: targetSeqNo, + timestamp: timestamp, + layer: layer, + trailerStripped: trailerStripped, }, extSequenceNumber: uint64(targetSeqNo), extTimestamp: extTimestamp, diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 527d5165f..6d9069151 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -23,6 +23,7 @@ import ( "time" "github.com/pion/rtp" + "github.com/pion/rtp/codecs" "github.com/pion/webrtc/v4" "go.uber.org/zap/zapcore" @@ -202,6 +203,8 @@ type TranslationParams struct { incomingHeaderSize int codecBytes []byte marker bool + // end of the svc spatial layer frame + isEndOfLayerFrame bool } // ------------------------------------------------------------------- @@ -2125,6 +2128,20 @@ func (f *Forwarder) getTranslationParamsAudio(extPkt *buffer.ExtPacket, layer in return tp, err } +// should be called with lock held +func (f *Forwarder) isEndOfLayerFrame(extPkt *buffer.ExtPacket) bool { + if extPkt.DependencyDescriptor != nil { + return extPkt.DependencyDescriptor.Descriptor != nil && extPkt.DependencyDescriptor.Descriptor.LastPacketInFrame + } + + if f.mime == mime.MimeTypeVP9 { + vp9, ok := extPkt.Payload.(codecs.VP9Packet) + return ok && vp9.E + } + + return false +} + // should be called with lock held func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer int32) (TranslationParams, error) { tp := TranslationParams{} @@ -2169,6 +2186,7 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in tp.isSwitching = result.IsSwitching tp.ddBytes = result.DependencyDescriptorExtension tp.marker = result.RTPMarker + tp.isEndOfLayerFrame = f.isEndOfLayerFrame(extPkt) starting, err := f.getTranslationParamsCommon(extPkt, layer, &tp) tp.isStarting = starting diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index ff83d61a6..b9d8170f8 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -17,14 +17,17 @@ package sfu import ( "testing" + "github.com/pion/rtp/codecs" "github.com/pion/webrtc/v4" "github.com/stretchr/testify/require" "github.com/livekit/mediatransportutil/pkg/codec" + "github.com/livekit/protocol/codecs/mime" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/livekit/livekit-server/pkg/sfu/buffer" + dd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/dependencydescriptor" "github.com/livekit/livekit-server/pkg/sfu/testutils" ) @@ -2215,3 +2218,34 @@ func TestGetRefLayerRTPTimestampBounds(t *testing.T) { require.Error(t, err) // unavailable sender report, not invalid layer require.Contains(t, err.Error(), "unavailable") } + +// TestForwarderIsEndOfLayerFrame checks the end-of-layer-frame detection used to +// locate packet trailers, which VP9 SVC can carry at the end of any spatial layer +// frame and not just at the end of a picture. +func TestForwarderIsEndOfLayerFrame(t *testing.T) { + vp9Codec := webrtc.RTPCodecCapability{MimeType: mime.MimeTypeVP9.String(), ClockRate: 90000} + + f := newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo) + require.False(t, f.isEndOfLayerFrame(&buffer.ExtPacket{Payload: codecs.VP9Packet{E: true}})) + + f = newForwarder(vp9Codec, webrtc.RTPCodecTypeVideo) + require.True(t, f.isEndOfLayerFrame(&buffer.ExtPacket{Payload: codecs.VP9Packet{E: true}})) + require.False(t, f.isEndOfLayerFrame(&buffer.ExtPacket{Payload: codecs.VP9Packet{E: false}})) + require.False(t, f.isEndOfLayerFrame(&buffer.ExtPacket{})) + + require.True(t, f.isEndOfLayerFrame(&buffer.ExtPacket{ + DependencyDescriptor: &buffer.ExtDependencyDescriptor{ + Descriptor: &dd.DependencyDescriptor{LastPacketInFrame: true}, + }, + Payload: codecs.VP9Packet{E: false}, + })) + require.False(t, f.isEndOfLayerFrame(&buffer.ExtPacket{ + DependencyDescriptor: &buffer.ExtDependencyDescriptor{ + Descriptor: &dd.DependencyDescriptor{LastPacketInFrame: false}, + }, + Payload: codecs.VP9Packet{E: true}, + })) + require.False(t, f.isEndOfLayerFrame(&buffer.ExtPacket{ + DependencyDescriptor: &buffer.ExtDependencyDescriptor{}, + })) +} diff --git a/pkg/sfu/packettrailer/packet_trailer.go b/pkg/sfu/packettrailer/packet_trailer.go index 3b58f6681..d86a4cddc 100644 --- a/pkg/sfu/packettrailer/packet_trailer.go +++ b/pkg/sfu/packettrailer/packet_trailer.go @@ -25,9 +25,11 @@ const ( // StripTrailer returns the number of bytes to strip from the end of an RTP // payload if it contains an LKTS trailer. The trailer is located by checking // for the "LKTS" magic suffix and then reading the XORed trailer_len byte -// immediately before it. Returns 0 if absent or ineligible. -func StripTrailer(payload []byte, marker bool) int { - if !marker || len(payload) < envelopeSize { +// immediately before it. isEndOfFrame must be set only for packets ending an +// encoded frame, i. e. where a trailer could have been appended. Returns 0 if +// absent or ineligible. +func StripTrailer(payload []byte, isEndOfFrame bool) int { + if !isEndOfFrame || len(payload) < envelopeSize { return 0 } diff --git a/pkg/sfu/sequencer.go b/pkg/sfu/sequencer.go index 0507120c1..f737ccbb5 100644 --- a/pkg/sfu/sequencer.go +++ b/pkg/sfu/sequencer.go @@ -66,6 +66,8 @@ type packetMeta struct { ddBytesSlice []byte // abs-capture-time of packet actBytes []byte + // number of packet trailer bytes stripped when the packet was forwarded + trailerStripped uint8 } func (pm packetMeta) MarshalLogObject(e zapcore.ObjectEncoder) error { @@ -87,6 +89,9 @@ func (pm packetMeta) MarshalLogObject(e zapcore.ObjectEncoder) error { if len(pm.actBytes) != 0 { e.AddInt("actBytes", len(pm.actBytes)) } + if pm.trailerStripped != 0 { + e.AddUint8("trailerStripped", pm.trailerStripped) + } return nil } @@ -178,6 +183,7 @@ func (s *sequencer) push( numCodecBytesIn int, ddBytes []byte, actBytes []byte, + trailerStripped int, ) { s.Lock() defer s.Unlock() @@ -246,6 +252,7 @@ func (s *sequencer) push( marker: marker, layer: layer, numCodecBytesIn: uint8(numCodecBytesIn), + trailerStripped: uint8(trailerStripped), lastNack: s.getRefTime(packetTime), // delay retransmissions after the original transmission } pm := &s.meta[slot] diff --git a/pkg/sfu/sequencer_test.go b/pkg/sfu/sequencer_test.go index be91af8be..4d769e096 100644 --- a/pkg/sfu/sequencer_test.go +++ b/pkg/sfu/sequencer_test.go @@ -29,11 +29,11 @@ func Test_sequencer(t *testing.T) { off := uint16(15) for i := uint64(1); i < 518; i++ { - seq.push(time.Now().UnixNano(), i, i+uint64(off), 123, true, 2, nil, 0, nil, nil) + seq.push(time.Now().UnixNano(), i, i+uint64(off), 123, true, 2, nil, 0, nil, nil, 0) } // send the last two out-of-order - seq.push(time.Now().UnixNano(), 519, 519+uint64(off), 123, false, 2, nil, 0, nil, nil) - seq.push(time.Now().UnixNano(), 518, 518+uint64(off), 123, true, 2, nil, 0, nil, nil) + seq.push(time.Now().UnixNano(), 519, 519+uint64(off), 123, false, 2, nil, 0, nil, nil, 0) + seq.push(time.Now().UnixNano(), 518, 518+uint64(off), 123, true, 2, nil, 0, nil, nil, 0) req := []uint16{57, 58, 62, 63, 513, 514, 515, 516, 517} res := seq.getExtPacketMetas(req) @@ -63,14 +63,14 @@ func Test_sequencer(t *testing.T) { require.Equal(t, val.extTimestamp, uint64(123)) } - seq.push(time.Now().UnixNano(), 521, 521+uint64(off), 123, true, 1, nil, 0, nil, nil) + seq.push(time.Now().UnixNano(), 521, 521+uint64(off), 123, true, 1, nil, 0, nil, nil, 0) m := seq.getExtPacketMetas([]uint16{521 + off}) require.Equal(t, 0, len(m)) time.Sleep((ignoreRetransmission + 10) * time.Millisecond) m = seq.getExtPacketMetas([]uint16{521 + off}) require.Equal(t, 1, len(m)) - seq.push(time.Now().UnixNano(), 505, 505+uint64(off), 123, false, 1, nil, 0, nil, nil) + seq.push(time.Now().UnixNano(), 505, 505+uint64(off), 123, false, 1, nil, 0, nil, nil, 0) m = seq.getExtPacketMetas([]uint16{505 + off}) require.Equal(t, 0, len(m)) time.Sleep((ignoreRetransmission + 10) * time.Millisecond) @@ -83,7 +83,7 @@ func Test_sequencer_flush(t *testing.T) { off := uint16(15) for i := uint64(1); i < 100; i++ { - seq.push(time.Now().UnixNano(), i, i+uint64(off), 123, true, 2, nil, 0, nil, nil) + seq.push(time.Now().UnixNano(), i, i+uint64(off), 123, true, 2, nil, 0, nil, nil, 0) } preFlush := []uint16{57 + off, 58 + off} @@ -96,7 +96,7 @@ func Test_sequencer_flush(t *testing.T) { // the sequencer re-initializes on the next push and works normally for new packets for i := uint64(200); i < 210; i++ { - seq.push(time.Now().UnixNano(), i, i+uint64(off), 456, true, 3, nil, 0, nil, nil) + seq.push(time.Now().UnixNano(), i, i+uint64(off), 456, true, 3, nil, 0, nil, nil, 0) } postFlush := []uint16{205 + off} require.Equal(t, 0, len(seq.getExtPacketMetas(postFlush))) // not enough time elapsed yet @@ -200,6 +200,7 @@ func Test_sequencer_getNACKSeqNo_exclusion(t *testing.T) { len(tt.fields.codecBytesOversized), tt.fields.ddBytesOversized, tt.fields.actBytesOdd, + 0, ) } else { if i.seqNo%2 == 0 { @@ -214,6 +215,7 @@ func Test_sequencer_getNACKSeqNo_exclusion(t *testing.T) { tt.fields.numCodecBytesInEven, tt.fields.ddBytesEven, tt.fields.actBytesEven, + 0, ) } else { n.push( @@ -227,6 +229,7 @@ func Test_sequencer_getNACKSeqNo_exclusion(t *testing.T) { tt.fields.numCodecBytesInOdd, tt.fields.ddBytesOdd, tt.fields.actBytesOdd, + 0, ) } } @@ -354,6 +357,7 @@ func Test_sequencer_getNACKSeqNo_no_exclusion(t *testing.T) { tt.fields.numCodecBytesInEven, tt.fields.ddBytesEven, tt.fields.actBytesEven, + 0, ) } else { n.push( @@ -367,6 +371,7 @@ func Test_sequencer_getNACKSeqNo_no_exclusion(t *testing.T) { tt.fields.numCodecBytesInOdd, tt.fields.ddBytesOdd, tt.fields.actBytesOdd, + 0, ) } }