From 47324e9c2efd252a86db5f50f1f41c180f567132 Mon Sep 17 00:00:00 2001 From: David Chen Date: Wed, 14 Oct 2026 21:54:09 -0700 Subject: [PATCH] PR fixes for FlexFEC implementation (#4898) * do not NACK on fec recovered packet * refactor common code into unlockAndInvokeFECRecovery * add FEC stats to RTPStats * add log for when fecdecoder unavailable after creation * rename markAsFECRepair to setFECRepairBufferLimit for clarity * refactor packet generation for tests into util * fix test race in CI * move the unlocks back inline for readability * update RTPstats * Fix FEC write return after merging RTX feedback changes * Use shared RTP packet helper in merged FEC NACK test --- pkg/sfu/buffer/buffer.go | 26 +++++++-- pkg/sfu/buffer/buffer_fec_test.go | 69 +++++++++-------------- pkg/sfu/buffer/factory.go | 4 +- pkg/sfu/rtpstats/rtpstats_base.go | 86 ++++++++++++++++++++++++++++- pkg/sfu/rtpstats/rtpstats_test.go | 46 +++++++++++++++ pkg/sfu/testutils/rtptest/packet.go | 46 +++++++++++++++ 6 files changed, 224 insertions(+), 53 deletions(-) create mode 100644 pkg/sfu/testutils/rtptest/packet.go diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 5f8b8609f..9ab4efcbf 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -452,7 +452,7 @@ func (b *Buffer) SetPrimaryBufferForFEC(primaryBuffer *Buffer) { } } -func (b *Buffer) markAsFECRepair() { +func (b *Buffer) setFECRepairBufferLimit() { b.Lock() defer b.Unlock() @@ -564,6 +564,7 @@ func (b *Buffer) writeFEC(fecPkt *rtp.Packet, arrivalTime int64) { b.fecSSRC = fecPkt.SSRC b.maybeCreateFECDecoderLocked() if b.fecDecoder == nil { + b.logger.Infow("flexfec decoder unavailable after creation attempt", "fecSSRC", b.fecSSRC, "mediaSSRC", b.BufferBase.SSRC(), "payloadType", b.fecPayloadType) b.Unlock() return } @@ -583,6 +584,20 @@ func (b *Buffer) feedFECLocked( ) (fecRecoveryDelta, func(received int, recovered int, discarded int, bytesReceived int)) { statsBefore := b.fecDecoder.Stats() recovered := b.fecDecoder.DecodeFEC(pkt) + statsAfter := b.fecDecoder.Stats() + + fecPackets := statsAfter.FECPacketsReceived - statsBefore.FECPacketsReceived + fecBytes := statsAfter.FECBytesReceived - statsBefore.FECBytesReceived + fecPacketsDiscarded := statsAfter.FECPacketsDiscarded - statsBefore.FECPacketsDiscarded + fecPacketsRecovered := statsAfter.PacketsRecovered - statsBefore.PacketsRecovered + if b.rtpStats != nil { + b.rtpStats.UpdateFEC( + fecPackets, + fecBytes, + fecPacketsDiscarded, + fecPacketsRecovered, + ) + } if len(recovered) > 0 && b.fecPktBuf == nil { b.fecPktBuf = make([]byte, bucket.RTPMaxPktSize) @@ -603,12 +618,11 @@ func (b *Buffer) feedFECLocked( } if cb := b.onFECRecovery; cb != nil { - statsAfter := b.fecDecoder.Stats() return fecRecoveryDelta{ - received: int(statsAfter.FECPacketsReceived - statsBefore.FECPacketsReceived), - recovered: len(recovered), - discarded: int(statsAfter.FECPacketsDiscarded - statsBefore.FECPacketsDiscarded), - bytesReceived: int(statsAfter.FECBytesReceived - statsBefore.FECBytesReceived), + received: int(fecPackets), + recovered: int(fecPacketsRecovered), + discarded: int(fecPacketsDiscarded), + bytesReceived: int(fecBytes), }, cb } diff --git a/pkg/sfu/buffer/buffer_fec_test.go b/pkg/sfu/buffer/buffer_fec_test.go index 1e5d1546c..364f73a11 100644 --- a/pkg/sfu/buffer/buffer_fec_test.go +++ b/pkg/sfu/buffer/buffer_fec_test.go @@ -18,7 +18,6 @@ import ( "crypto/aes" "crypto/cipher" "encoding/binary" - "math/rand" "sync" "testing" "time" @@ -31,6 +30,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/livekit/livekit-server/pkg/sfu/testutils/rtptest" "github.com/livekit/mediatransportutil/pkg/twcc" ) @@ -49,34 +49,6 @@ var flexFECCodec = webrtc.RTPCodecParameters{ PayloadType: webrtc.PayloadType(fecTestFECPT), } -func fecTestMediaPackets(t *testing.T, baseSN uint16, count int) []rtp.Packet { - t.Helper() - rng := rand.New(rand.NewSource(int64(baseSN))) - pkts := make([]rtp.Packet, 0, count) - for i := 0; i < count; i++ { - payload := make([]byte, 50+rng.Intn(200)) - rng.Read(payload) - // valid VP8 payload descriptor (S=1, no extensions) so the video - // packet processing in the buffer accepts the packet - payload[0] = 0x10 - sn := baseSN + uint16(i) - pkts = append(pkts, rtp.Packet{ - Header: rtp.Header{ - Version: 2, - PayloadType: uint8(vp8Codec.PayloadType), - SequenceNumber: sn, - // derive timestamp from the sequence number so windows - // generated separately stay monotonic - Timestamp: 90000 + 3000*uint32(sn), - SSRC: fecTestMediaSSRC, - Marker: i == count-1, - }, - Payload: payload, - }) - } - return pkts -} - func bindFECTestBuffer(t *testing.T, buff *Buffer) { t.Helper() buff.codecType = webrtc.RTPCodecTypeVideo @@ -195,7 +167,7 @@ func TestBufferFECRecoversDroppedPacket(t *testing.T) { receivedDelta += received }) - media := fecTestMediaPackets(t, 100, 10) + media := rtptest.GenerateVP8Packets(100, 10, uint8(vp8Codec.PayloadType), fecTestMediaSSRC) encoder := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC) fecPackets := encoder.EncodeFec(media, 2) require.NotEmpty(t, fecPackets) @@ -218,6 +190,17 @@ func TestBufferFECRecoversDroppedPacket(t *testing.T) { assert.Equal(t, 1, recoveredDelta) assert.Equal(t, len(fecPackets), receivedDelta) + fecPayloadBytes := 0 + for i := range fecPackets { + fecPayloadBytes += len(fecPackets[i].Payload) + } + rtpStats := primary.GetStats() + require.NotNil(t, rtpStats) + assert.EqualValues(t, len(fecPackets), rtpStats.FecPackets) + assert.EqualValues(t, fecPayloadBytes, rtpStats.FecBytes) + assert.EqualValues(t, 0, rtpStats.FecPacketsDiscarded) + assert.EqualValues(t, 1, rtpStats.FecPacketsRecovered) + // the 9 received packets flow through the ext packet pipeline, the // recovered one fills the bucket like an RTX repair extSNBySN := readExtSequenceNumbers(t, primary, len(media)-1) @@ -237,7 +220,7 @@ func TestBufferFECRecoversEncryptedPayload(t *testing.T) { aead, err := cipher.NewGCM(block) require.NoError(t, err) - media := fecTestMediaPackets(t, 120, 5) + media := rtptest.GenerateVP8Packets(120, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC) plaintexts := make([][]byte, len(media)) for i := range media { plaintexts[i] = append([]byte("encrypted-video-frame-"), byte(i)) @@ -293,7 +276,7 @@ func TestBufferFECRecoveryCallbackCanReenterBuffer(t *testing.T) { callbackDone <- struct{}{} }) - media := fecTestMediaPackets(t, 150, 5) + media := rtptest.GenerateVP8Packets(150, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC) for i := range media { writePacket(t, primary, &media[i]) } @@ -332,7 +315,7 @@ func TestBufferFECPairAfterPackets(t *testing.T) { bindFECTestBuffer(t, primary) encoder := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC) - media := fecTestMediaPackets(t, 200, 10) + media := rtptest.GenerateVP8Packets(200, 10, uint8(vp8Codec.PayloadType), fecTestMediaSSRC) fecPackets := encoder.EncodeFec(media, 2) require.NotEmpty(t, fecPackets) @@ -361,7 +344,7 @@ func TestBufferFECPairAfterPackets(t *testing.T) { assert.EqualValues(t, 1, stats.PacketsRecovered) // the next window recovers normally - media2 := fecTestMediaPackets(t, 210, 10) + media2 := rtptest.GenerateVP8Packets(210, 10, uint8(vp8Codec.PayloadType), fecTestMediaSSRC) fecPackets2 := encoder.EncodeFec(media2, 2) require.NotEmpty(t, fecPackets2) @@ -393,7 +376,7 @@ func TestBufferFECCoupledBeforeBuffersExist(t *testing.T) { fecBuff := factory.GetOrNew(packetio.RTPBufferPacket, fecTestFECSSRC).(*Buffer) bindFECTestBuffer(t, primary) - media := fecTestMediaPackets(t, 300, 5) + media := rtptest.GenerateVP8Packets(300, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC) encoder := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC) fecPackets := encoder.EncodeFec(media, 1) require.NotEmpty(t, fecPackets) @@ -424,7 +407,7 @@ func TestBufferFECSequenceNumberWrap(t *testing.T) { fecBuff := factory.GetOrNew(packetio.RTPBufferPacket, fecTestFECSSRC).(*Buffer) bindFECTestBuffer(t, primary) - media := fecTestMediaPackets(t, 65533, 5) + media := rtptest.GenerateVP8Packets(65533, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC) for i := range media { media[i].Timestamp = 90000 + 3000*uint32(i) } @@ -454,7 +437,7 @@ func TestBufferFECRecoveryAfterPaddingRemoval(t *testing.T) { fecBuff := factory.GetOrNew(packetio.RTPBufferPacket, fecTestFECSSRC).(*Buffer) bindFECTestBuffer(t, primary) - media := fecTestMediaPackets(t, 800, 5) + media := rtptest.GenerateVP8Packets(800, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC) fecPackets := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC).EncodeFec(media, 1) require.Len(t, fecPackets, 1) @@ -523,7 +506,7 @@ func TestBufferFECIgnoresUnexpectedPayloadType(t *testing.T) { Codecs: []webrtc.RTPCodecParameters{vp8Codec}, }, vp8Codec.RTPCodecCapability, 0)) - media := fecTestMediaPackets(t, 400, 5) + media := rtptest.GenerateVP8Packets(400, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC) encoder := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC) fecPackets := encoder.EncodeFec(media, 1) require.NotEmpty(t, fecPackets) @@ -576,7 +559,7 @@ func TestBufferFECNACKSuppression(t *testing.T) { factory.SetFECPair(fecTestFECSSRC, fecTestMediaSSRC) bindFECTestBuffer(t, primary) - media := fecTestMediaPackets(t, 700, 10) + media := rtptest.GenerateVP8Packets(700, 10, uint8(vp8Codec.PayloadType), fecTestMediaSSRC) encoder := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC) fecPackets := encoder.EncodeFec(media, 2) require.NotEmpty(t, fecPackets) @@ -631,7 +614,7 @@ func TestBufferFECNACKNotSentForRecoveredPacket(t *testing.T) { // single FEC packet protecting the whole window; dropping two of the // protected packets keeps the window unrecoverable until one arrives - media := fecTestMediaPackets(t, 100, 5) // 100..104 + media := rtptest.GenerateVP8Packets(100, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC) // 100..104 fecPackets := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC).EncodeFec(media, 1) require.Len(t, fecPackets, 1) @@ -640,9 +623,9 @@ func TestBufferFECNACKNotSentForRecoveredPacket(t *testing.T) { writePacket(t, primary, &media[4]) // 104 -> gap 102, 103 // an unrelated gap outside the FEC-protected window that FEC cannot recover - extra := fecTestMediaPackets(t, 105, 3) // 105..107 - writePacket(t, primary, &extra[0]) // 105 - writePacket(t, primary, &extra[2]) // 107 -> gap 106 + extra := rtptest.GenerateVP8Packets(105, 3, uint8(vp8Codec.PayloadType), fecTestMediaSSRC) // 105..107 + writePacket(t, primary, &extra[0]) // 105 + writePacket(t, primary, &extra[2]) // 107 -> gap 106 const unrecoverableSN = uint16(106) // FEC packet retained: two protected packets (102, 103) still missing diff --git a/pkg/sfu/buffer/factory.go b/pkg/sfu/buffer/factory.go index aaed2c561..fd7561e7f 100644 --- a/pkg/sfu/buffer/factory.go +++ b/pkg/sfu/buffer/factory.go @@ -93,7 +93,7 @@ func (f *Factory) GetOrNew(packetType packetio.BufferPacketType, ssrc uint32) io } for fec, base := range f.fecPair { if fec == ssrc { - buffer.markAsFECRepair() + buffer.setFECRepairBufferLimit() baseBuffer, ok := f.rtpBuffers[base] if ok { buffer.SetPrimaryBufferForFEC(baseBuffer) @@ -175,7 +175,7 @@ func (f *Factory) SetFECPair(fec, base uint32) { } f.Unlock() if fecBuffer != nil && baseBuffer == nil { - fecBuffer.markAsFECRepair() + fecBuffer.setFECRepairBufferLimit() } if fecBuffer != nil && baseBuffer != nil { fecBuffer.SetPrimaryBufferForFEC(baseBuffer) diff --git a/pkg/sfu/rtpstats/rtpstats_base.go b/pkg/sfu/rtpstats/rtpstats_base.go index fdd92e89e..25697edd4 100644 --- a/pkg/sfu/rtpstats/rtpstats_base.go +++ b/pkg/sfu/rtpstats/rtpstats_base.go @@ -48,6 +48,10 @@ type RTPDeltaInfo struct { PacketsPadding uint32 BytesPadding uint64 HeaderBytesPadding uint64 + FecPackets uint32 + FecBytes uint64 + FecPacketsDiscarded uint32 + FecPacketsRecovered uint32 PacketsLost uint32 PacketsMissing uint32 PacketsOutOfOrder uint32 @@ -76,6 +80,10 @@ func (r *RTPDeltaInfo) MarshalLogObject(e zapcore.ObjectEncoder) error { e.AddUint32("PacketsPadding", r.PacketsPadding) e.AddUint64("BytesPadding", r.BytesPadding) e.AddUint64("HeaderBytesPadding", r.HeaderBytesPadding) + e.AddUint32("FecPackets", r.FecPackets) + e.AddUint64("FecBytes", r.FecBytes) + e.AddUint32("FecPacketsDiscarded", r.FecPacketsDiscarded) + e.AddUint32("FecPacketsRecovered", r.FecPacketsRecovered) e.AddUint32("PacketsLost", r.PacketsLost) e.AddUint32("PacketsMissing", r.PacketsMissing) e.AddUint32("PacketsOutOfOrder", r.PacketsOutOfOrder) @@ -104,6 +112,11 @@ type snapshot struct { bytesPadding uint64 headerBytesPadding uint64 + fecPackets uint64 + fecBytes uint64 + fecPacketsDiscarded uint64 + fecPacketsRecovered uint64 + frames uint32 plis uint32 @@ -126,6 +139,10 @@ func (s *snapshot) MarshalLogObject(e zapcore.ObjectEncoder) error { e.AddUint64("packetsPadding", s.packetsPadding) e.AddUint64("bytesPadding", s.bytesPadding) e.AddUint64("headerBytesPadding", s.headerBytesPadding) + e.AddUint64("fecPackets", s.fecPackets) + e.AddUint64("fecBytes", s.fecBytes) + e.AddUint64("fecPacketsDiscarded", s.fecPacketsDiscarded) + e.AddUint64("fecPacketsRecovered", s.fecPacketsRecovered) e.AddUint32("frames", s.frames) e.AddUint32("plis", s.plis) e.AddUint32("firs", s.firs) @@ -222,6 +239,11 @@ type rtpStatsBase struct { bytesPadding uint64 headerBytesPadding uint64 + fecPackets uint64 + fecBytes uint64 + fecPacketsDiscarded uint64 + fecPacketsRecovered uint64 + frames uint32 jitter float64 @@ -285,6 +307,11 @@ func (r *rtpStatsBase) seed(from *rtpStatsBase) bool { r.bytesPadding = from.bytesPadding r.headerBytesPadding = from.headerBytesPadding + r.fecPackets = from.fecPackets + r.fecBytes = from.fecBytes + r.fecPacketsDiscarded = from.fecPacketsDiscarded + r.fecPacketsRecovered = from.fecPacketsRecovered + r.frames = from.frames r.jitter = from.jitter @@ -324,6 +351,25 @@ func (r *rtpStatsBase) newSnapshotID(extStartSN uint64) uint32 { return id } +func (r *rtpStatsBase) UpdateFEC( + fecPackets uint64, + fecBytes uint64, + fecPacketsDiscarded uint64, + fecPacketsRecovered uint64, +) { + r.lock.Lock() + defer r.lock.Unlock() + + if r.endTime != 0 { + return + } + + r.fecPackets += fecPackets + r.fecBytes += fecBytes + r.fecPacketsDiscarded += fecPacketsDiscarded + r.fecPacketsRecovered += fecPacketsRecovered +} + func (r *rtpStatsBase) UpdateFir(firCount uint32) { r.lock.Lock() defer r.lock.Unlock() @@ -503,8 +549,12 @@ func (r *rtpStatsBase) deltaInfo( } if packetsExpected == 0 { deltaInfo = &RTPDeltaInfo{ - StartTime: time.Unix(0, startTime), - EndTime: time.Unix(0, endTime), + StartTime: time.Unix(0, startTime), + EndTime: time.Unix(0, endTime), + FecPackets: uint32(now.fecPackets - then.fecPackets), + FecBytes: now.fecBytes - then.fecBytes, + FecPacketsDiscarded: uint32(now.fecPacketsDiscarded - then.fecPacketsDiscarded), + FecPacketsRecovered: uint32(now.fecPacketsRecovered - then.fecPacketsRecovered), } return } @@ -544,6 +594,10 @@ func (r *rtpStatsBase) deltaInfo( PacketsPadding: uint32(packetsPadding), BytesPadding: now.bytesPadding - then.bytesPadding, HeaderBytesPadding: now.headerBytesPadding - then.headerBytesPadding, + FecPackets: uint32(now.fecPackets - then.fecPackets), + FecBytes: now.fecBytes - then.fecBytes, + FecPacketsDiscarded: uint32(now.fecPacketsDiscarded - then.fecPacketsDiscarded), + FecPacketsRecovered: uint32(now.fecPacketsRecovered - then.fecPacketsRecovered), PacketsLost: packetsLost, PacketsOutOfOrder: uint32(now.packetsOutOfOrder - then.packetsOutOfOrder), Frames: now.frames - then.frames, @@ -588,6 +642,11 @@ func (r *rtpStatsBase) marshalLogObject( e.AddFloat64("bitratePadding", float64(r.bytesPadding)*8.0/elapsedSeconds) e.AddUint64("headerBytesPadding", r.headerBytesPadding) + e.AddUint64("fecPackets", r.fecPackets) + e.AddUint64("fecBytes", r.fecBytes) + e.AddUint64("fecPacketsDiscarded", r.fecPacketsDiscarded) + e.AddUint64("fecPacketsRecovered", r.fecPacketsRecovered) + e.AddUint32("frames", r.frames) e.AddFloat64("frameRate", float64(r.frames)/elapsedSeconds) @@ -638,6 +697,11 @@ func (r *rtpStatsBase) toProto( p.BitratePadding = float64(r.bytesPadding) * 8.0 / p.Duration p.HeaderBytesPadding = r.headerBytesPadding + p.FecPackets = uint32(r.fecPackets) + p.FecBytes = r.fecBytes + p.FecPacketsDiscarded = uint32(r.fecPacketsDiscarded) + p.FecPacketsRecovered = uint32(r.fecPacketsRecovered) + p.Frames = r.frames p.FrameRate = float64(r.frames) / p.Duration @@ -813,6 +877,10 @@ func (r *rtpStatsBase) getSnapshot(startTime int64, extStartSN uint64) snapshot packetsPadding: r.packetsPadding, bytesPadding: r.bytesPadding, headerBytesPadding: r.headerBytesPadding, + fecPackets: r.fecPackets, + fecBytes: r.fecBytes, + fecPacketsDiscarded: r.fecPacketsDiscarded, + fecPacketsRecovered: r.fecPacketsRecovered, frames: r.frames, plis: r.plis, firs: r.firs, @@ -853,6 +921,11 @@ func AggregateRTPDeltaInfo(deltaInfoList []*RTPDeltaInfo) *RTPDeltaInfo { bytesPadding := uint64(0) headerBytesPadding := uint64(0) + fecPackets := uint32(0) + fecBytes := uint64(0) + fecPacketsDiscarded := uint32(0) + fecPacketsRecovered := uint32(0) + packetsLost := uint32(0) packetsMissing := uint32(0) packetsOutOfOrder := uint32(0) @@ -891,6 +964,11 @@ func AggregateRTPDeltaInfo(deltaInfoList []*RTPDeltaInfo) *RTPDeltaInfo { bytesPadding += deltaInfo.BytesPadding headerBytesPadding += deltaInfo.HeaderBytesPadding + fecPackets += deltaInfo.FecPackets + fecBytes += deltaInfo.FecBytes + fecPacketsDiscarded += deltaInfo.FecPacketsDiscarded + fecPacketsRecovered += deltaInfo.FecPacketsRecovered + packetsLost += deltaInfo.PacketsLost packetsMissing += deltaInfo.PacketsMissing packetsOutOfOrder += deltaInfo.PacketsOutOfOrder @@ -925,6 +1003,10 @@ func AggregateRTPDeltaInfo(deltaInfoList []*RTPDeltaInfo) *RTPDeltaInfo { PacketsPadding: packetsPadding, BytesPadding: bytesPadding, HeaderBytesPadding: headerBytesPadding, + FecPackets: fecPackets, + FecBytes: fecBytes, + FecPacketsDiscarded: fecPacketsDiscarded, + FecPacketsRecovered: fecPacketsRecovered, PacketsLost: packetsLost, PacketsMissing: packetsMissing, PacketsOutOfOrder: packetsOutOfOrder, diff --git a/pkg/sfu/rtpstats/rtpstats_test.go b/pkg/sfu/rtpstats/rtpstats_test.go index 9039ad773..752c749a2 100644 --- a/pkg/sfu/rtpstats/rtpstats_test.go +++ b/pkg/sfu/rtpstats/rtpstats_test.go @@ -223,6 +223,52 @@ func Test_RTPStatsReceiver_Update(t *testing.T) { r.Stop() } +func Test_RTPStatsReceiver_UpdateFEC(t *testing.T) { + r := NewRTPStatsReceiver(RTPStatsParams{}) + r.SetClockRate(90000) + snapshotID := r.NewSnapshotId() + + packet := getPacket(100, 3000, 1000) + r.Update( + time.Now().UnixNano(), + packet.SequenceNumber, + packet.Timestamp, + packet.Marker, + packet.Header.MarshalSize(), + len(packet.Payload), + 0, + ) + r.UpdateFEC(3, 900, 1, 2) + + stats := r.ToProto() + require.EqualValues(t, 3, stats.FecPackets) + require.EqualValues(t, 900, stats.FecBytes) + require.EqualValues(t, 1, stats.FecPacketsDiscarded) + require.EqualValues(t, 2, stats.FecPacketsRecovered) + + delta := r.DeltaInfo(snapshotID) + require.EqualValues(t, 3, delta.FecPackets) + require.EqualValues(t, 900, delta.FecBytes) + require.EqualValues(t, 1, delta.FecPacketsDiscarded) + require.EqualValues(t, 2, delta.FecPacketsRecovered) + + // FEC-only intervals are reported even if no primary packet arrived. + r.UpdateFEC(2, 500, 1, 1) + delta = r.DeltaInfo(snapshotID) + require.EqualValues(t, 2, delta.FecPackets) + require.EqualValues(t, 500, delta.FecBytes) + require.EqualValues(t, 1, delta.FecPacketsDiscarded) + require.EqualValues(t, 1, delta.FecPacketsRecovered) + + r.Stop() + r.UpdateFEC(1, 100, 1, 1) + stats = r.ToProto() + require.EqualValues(t, 5, stats.FecPackets) + require.EqualValues(t, 1400, stats.FecBytes) + require.EqualValues(t, 2, stats.FecPacketsDiscarded) + require.EqualValues(t, 3, stats.FecPacketsRecovered) +} + func Test_RTPStatsReceiver_Restart(t *testing.T) { clockRate := uint32(90000) r := NewRTPStatsReceiver(RTPStatsParams{}) diff --git a/pkg/sfu/testutils/rtptest/packet.go b/pkg/sfu/testutils/rtptest/packet.go new file mode 100644 index 000000000..9be018254 --- /dev/null +++ b/pkg/sfu/testutils/rtptest/packet.go @@ -0,0 +1,46 @@ +// 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 rtptest + +import ( + "math/rand" + + "github.com/pion/rtp" +) + +// GenerateVP8Packets returns deterministic RTP packets with valid VP8 payload descriptors. +func GenerateVP8Packets(baseSN uint16, count int, payloadType uint8, ssrc uint32) []rtp.Packet { + rng := rand.New(rand.NewSource(int64(baseSN))) + packets := make([]rtp.Packet, 0, count) + for i := 0; i < count; i++ { + payload := make([]byte, 50+rng.Intn(200)) + rng.Read(payload) + payload[0] = 0x10 + + sequenceNumber := baseSN + uint16(i) + packets = append(packets, rtp.Packet{ + Header: rtp.Header{ + Version: 2, + PayloadType: payloadType, + SequenceNumber: sequenceNumber, + Timestamp: 90000 + 3000*uint32(sequenceNumber), + SSRC: ssrc, + Marker: i == count-1, + }, + Payload: payload, + }) + } + return packets +}