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

@@ -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)]
 		}
 	}
 

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
 // -----------------------------------------------------------------------------

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,

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

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{},
+	}))
+}

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
 	}
 

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]

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,
 						)
 					}
 				}