From b08f67f61aeea922cc698d002012b28b4050f75c Mon Sep 17 00:00:00 2001 From: David Chen Date: Tue, 22 Sep 2026 10:17:06 -0700 Subject: [PATCH] do not NACK on fec recovered packet --- pkg/sfu/buffer/buffer.go | 60 +++++++++++++++++-------- pkg/sfu/buffer/buffer_fec_test.go | 74 +++++++++++++++++++++++++++++++ 2 files changed, 116 insertions(+), 18 deletions(-) diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index b77f24642..4a3368aa5 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -279,22 +279,17 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) { return } - rtcpPackets := b.calc(pkt, &rtpPacket, now, false, false) + b.ingestLocked(pkt, &rtpPacket, now, false, false) var fecDelta fecRecoveryDelta var onFECRecovery func(received int, recovered int, discarded int, bytesReceived int) if b.fecDecoder != nil { // feed media into the FEC decoder, a media arrival can complete a - // previously unrecoverable FEC window + // previously unrecoverable FEC window. Recovered packets update NACK + // state before feedback is generated below, so a packet repaired by + // FEC is not needlessly NACKed. fecDelta, onFECRecovery = b.feedFECLocked(&rtpPacket, now) } - b.Unlock() - fecDelta.invoke(onFECRecovery) - - if len(rtcpPackets) != 0 { - if cb := b.getOnRtcpFeedback(); cb != nil { - cb(rtcpPackets) - } - } + b.finishWriteLocked(now, fecDelta, onFECRecovery) return } @@ -412,14 +407,13 @@ func (b *Buffer) writeRTX(rtxPkt *rtp.Packet, arrivalTime int64) { return } - b.calc(b.rtxPktBuf[:n], &repairedPkt, arrivalTime, false, true) + b.ingestLocked(b.rtxPktBuf[:n], &repairedPkt, arrivalTime, false, true) var fecDelta fecRecoveryDelta var onFECRecovery func(received int, recovered int, discarded int, bytesReceived int) if b.fecDecoder != nil { fecDelta, onFECRecovery = b.feedFECLocked(&repairedPkt, arrivalTime) } - b.Unlock() - fecDelta.invoke(onFECRecovery) + b.finishWriteLocked(arrivalTime, fecDelta, onFECRecovery) } func (b *Buffer) SetPrimaryBufferForFEC(primaryBuffer *Buffer) { @@ -565,8 +559,7 @@ func (b *Buffer) writeFEC(fecPkt *rtp.Packet, arrivalTime int64) { } fecDelta, onFECRecovery := b.feedFECLocked(fecPkt, arrivalTime) - b.Unlock() - fecDelta.invoke(onFECRecovery) + b.finishWriteLocked(arrivalTime, fecDelta, onFECRecovery) } // feedFECLocked runs a media or FEC packet through the FEC decoder and @@ -592,8 +585,9 @@ func (b *Buffer) feedFECLocked( // recovered packets flow through the regular pipeline: they are // forwarded downstream and stop NACKs for the lost sequence numbers. // They do not re-enter the decoder because chained recovery already - // completed within DecodeFEC. - b.calc(b.fecPktBuf[:n], rp, arrivalTime, false, true) + // completed within DecodeFEC. RTCP is generated once by the caller + // after all recovered packets have updated NACK state. + b.ingestLocked(b.fecPktBuf[:n], rp, arrivalTime, false, true) } if cb := b.onFECRecovery; cb != nil { @@ -683,7 +677,11 @@ func (b *Buffer) sendPLI() { } } -func (b *Buffer) calc(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime int64, isBuffered bool, isRTX bool) []rtcp.Packet { +// ingestLocked runs a packet through the incoming packet pipeline, updating RTP +// stats, NACK state and the packet bucket. It intentionally does not generate +// RTCP so that FEC recovery can update NACK state before feedback is generated +// once. Must be called with the buffer lock held. +func (b *Buffer) ingestLocked(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime int64, isBuffered bool, isRTX bool) { b.BufferBase.HandleIncomingPacketLocked( rawPkt, rtpPacket, @@ -693,10 +691,36 @@ func (b *Buffer) calc(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime int64, i nil, 0, ) +} +func (b *Buffer) calc(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime int64, isBuffered bool, isRTX bool) []rtcp.Packet { + b.ingestLocked(rawPkt, rtpPacket, arrivalTime, isBuffered, isRTX) return b.getRTCPPackets(arrivalTime) } +// finishWriteLocked generates RTCP feedback from the current NACK state, which +// already reflects any FEC recovery performed during this write, releases the +// buffer lock, then invokes the FEC recovery callback and delivers RTCP +// feedback. Generating RTCP after recovery ensures sequence numbers repaired by +// FEC are not needlessly NACKed. Must be called with the buffer lock held; it +// releases the lock before returning. +func (b *Buffer) finishWriteLocked( + arrivalTime int64, + fecDelta fecRecoveryDelta, + onFECRecovery func(received int, recovered int, discarded int, bytesReceived int), +) { + rtcpPackets := b.getRTCPPackets(arrivalTime) + b.Unlock() + + fecDelta.invoke(onFECRecovery) + + if len(rtcpPackets) != 0 { + if cb := b.getOnRtcpFeedback(); cb != nil { + cb(rtcpPackets) + } + } +} + func (b *Buffer) getRTCPPackets(arrivalTime int64) []rtcp.Packet { var pkts []rtcp.Packet if nackPkt := b.getNACKPacket(); nackPkt != nil { diff --git a/pkg/sfu/buffer/buffer_fec_test.go b/pkg/sfu/buffer/buffer_fec_test.go index 876e58dbb..1e5d1546c 100644 --- a/pkg/sfu/buffer/buffer_fec_test.go +++ b/pkg/sfu/buffer/buffer_fec_test.go @@ -19,10 +19,12 @@ import ( "crypto/cipher" "encoding/binary" "math/rand" + "sync" "testing" "time" pionflexfec "github.com/pion/interceptor/pkg/flexfec" + "github.com/pion/rtcp" "github.com/pion/rtp" "github.com/pion/transport/v4/packetio" "github.com/pion/webrtc/v4" @@ -598,3 +600,75 @@ func TestBufferFECNACKSuppression(t *testing.T) { require.Empty(t, primary.nacker.Nacks(), "NACK for recovered packet not suppressed") } + +func TestBufferFECNACKNotSentForRecoveredPacket(t *testing.T) { + // A late media arrival can complete a retained FEC window. RTCP feedback + // must be generated after FEC recovery so the sequence number repaired by + // FEC is not NACKed, while an unrelated unrecoverable gap is still NACKed. + factory := NewFactoryOfBufferFactory(500, 200).CreateBufferFactory() + + primary := factory.GetOrNew(packetio.RTPBufferPacket, fecTestMediaSSRC).(*Buffer) + fecBuff := factory.GetOrNew(packetio.RTPBufferPacket, fecTestFECSSRC).(*Buffer) + factory.SetFECPair(fecTestFECSSRC, fecTestMediaSSRC) + bindFECTestBuffer(t, primary) + + var mu sync.Mutex + nackedSNs := make(map[uint16]bool) + primary.OnRtcpFeedback(func(fb []rtcp.Packet) { + mu.Lock() + defer mu.Unlock() + for _, pkt := range fb { + if nack, ok := pkt.(*rtcp.TransportLayerNack); ok { + for _, pair := range nack.Nacks { + pair.Range(func(sn uint16) bool { + nackedSNs[sn] = true + return true + }) + } + } + } + }) + + // 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 + fecPackets := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC).EncodeFec(media, 1) + require.Len(t, fecPackets, 1) + + writePacket(t, primary, &media[0]) // 100 + writePacket(t, primary, &media[1]) // 101 + 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 + const unrecoverableSN = uint16(106) + + // FEC packet retained: two protected packets (102, 103) still missing + writePacket(t, fecBuff, &fecPackets[0]) + require.EqualValues(t, 0, primary.FECDecoderStats().PacketsRecovered) + require.Len(t, primary.nacker.Nacks(), 3, "expected queued NACKs for 102, 103, 106") + + // let the queued NACKs age past the minimum NACK interval so they would be + // sent on the next RTCP generation, and ignore any feedback emitted during + // setup + time.Sleep(30 * time.Millisecond) + mu.Lock() + nackedSNs = make(map[uint16]bool) + mu.Unlock() + + // late arrival of 102 leaves only 103 missing, recovered in the same write + writePacket(t, primary, &media[2]) // 102 + require.EqualValues(t, 1, primary.FECDecoderStats().PacketsRecovered) + + // only the unrecoverable gap should remain queued; 102 arrived and 103 was + // recovered, both cleared from the NACK queue + require.Len(t, primary.nacker.Nacks(), 1) + + mu.Lock() + defer mu.Unlock() + assert.True(t, nackedSNs[unrecoverableSN], "expected NACK for unrecoverable gap %d", unrecoverableSN) + assert.False(t, nackedSNs[media[2].SequenceNumber], "arrived packet %d must not be NACKed", media[2].SequenceNumber) + assert.False(t, nackedSNs[media[3].SequenceNumber], "FEC-recovered packet %d must not be NACKed", media[3].SequenceNumber) +}