migrate fec into buffer_base

pkg/rtc/mediatrack.go

@@ -37,7 +37,6 @@ import (
 	"github.com/livekit/livekit-server/pkg/sfu/connectionquality"
 	"github.com/livekit/livekit-server/pkg/sfu/interceptor"
 	"github.com/livekit/livekit-server/pkg/telemetry"
-	"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
 	util "github.com/livekit/mediatransportutil"
 )
 
@@ -582,10 +581,6 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe
 
 	buff.OnNotifyRTX(t.MediaTrackReceiver.setLayerRtxInfo)
 
-	buff.OnFECRecovery(func(received int, recovered int, discarded int, bytesReceived int) {
-		prometheus.RecordFECUpstream(received, recovered, discarded, uint64(bytesReceived))
-	})
-
 	// if subscriber request fps before fps calculated, update them after fps updated.
 	buff.OnFpsChanged(func() {
 		t.MediaTrackSubscriptions.UpdateVideoLayers()

pkg/sfu/buffer/buffer.go

@@ -23,7 +23,6 @@ import (
 	"github.com/pion/rtp"
 	"github.com/pion/webrtc/v4"
 
-	"github.com/livekit/livekit-server/pkg/sfu/flexfec"
 	sutils "github.com/livekit/livekit-server/pkg/utils"
 	"github.com/livekit/mediatransportutil/pkg/bucket"
 	"github.com/livekit/mediatransportutil/pkg/twcc"
@@ -42,10 +41,7 @@ const (
 	maxPendingFECRepairPackets = 100
 )
 
-var (
-	errInvalidCodec           = errors.New("invalid codec")
-	errFECMediaPacketNotFound = errors.New("fec media packet not found")
-)
+var errInvalidCodec = errors.New("invalid codec")
 
 var _ BufferProvider = (*Buffer)(nil)
 
@@ -54,21 +50,6 @@ type pendingPacket struct {
 	packet      []byte
 }
 
-type fecRecoveryDelta struct {
-	received      int
-	recovered     int
-	discarded     int
-	bytesReceived int
-}
-
-func (d fecRecoveryDelta) invoke(callback func(received int, recovered int, discarded int, bytesReceived int)) {
-	if callback != nil && (d.recovered > 0 || d.received > 0 || d.discarded > 0) {
-		callback(d.received, d.recovered, d.discarded, d.bytesReceived)
-	}
-}
-
-type cbFECRecovery func(received int, recovered int, discarded int, bytesReceived int)
-
 // Buffer contains all packets
 type Buffer struct {
 	*BufferBase
@@ -96,10 +77,6 @@ type Buffer struct {
 
 	primaryBufferForFEC *Buffer
 	isFECRepair         bool
-	fecSSRC             uint32
-	fecDecoder          *flexfec.Decoder
-	fecPktBuf           []byte
-	onFECRecovery       cbFECRecovery
 
 	streamInfoProbe       *StreamInfoProbe
 	warnedPendingOverflow bool
@@ -123,13 +100,12 @@ type StreamInfoProbe struct {
 func NewBuffer(ssrc uint32, maxVideoPkts, maxAudioPkts int) *Buffer {
 	b := &Buffer{}
 	b.BufferBase = NewBufferBase(BufferBaseParams{
-		SSRC:                  ssrc,
-		MaxVideoPkts:          maxVideoPkts,
-		MaxAudioPkts:          maxAudioPkts,
-		LoggerComponents:      []string{sutils.ComponentPub, sutils.ComponentSFU},
-		SendPLI:               b.sendPLI,
-		OnStreamRestartLocked: b.recreateFECDecoderLocked,
-		IsReportingEnabled:    true,
+		SSRC:               ssrc,
+		MaxVideoPkts:       maxVideoPkts,
+		MaxAudioPkts:       maxAudioPkts,
+		LoggerComponents:   []string{sutils.ComponentPub, sutils.ComponentSFU},
+		SendPLI:            b.sendPLI,
+		IsReportingEnabled: true,
 	})
 	return b
 }
@@ -175,10 +151,9 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
 	rtcpPackets := b.getRTCPPackets(now)
 
 	b.isBound = true
-	b.maybeCreateFECDecoderLocked()
 	b.Unlock()
 
-	b.packetPostProcess(rtcpPackets, fecRecoveryDelta{}, nil)
+	b.packetPostProcess(rtcpPackets)
 	return nil
 }
 
@@ -282,20 +257,10 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) {
 
 	b.ingestLocked(pkt, rtpPacket, now, false, false)
 
-	var fecDelta fecRecoveryDelta
-	var onFECRecovery cbFECRecovery
-	if b.fecDecoder != nil {
-		// feed media into the FEC decoder, a media arrival can complete a
-		// 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)
-	}
-
 	rtcpPackets := b.getRTCPPackets(now)
 	b.Unlock()
 
-	b.packetPostProcess(rtcpPackets, fecDelta, onFECRecovery)
+	b.packetPostProcess(rtcpPackets)
 	return
 }
 
@@ -415,16 +380,10 @@ func (b *Buffer) writeRTX(rtxPkt *rtp.Packet, arrivalTime int64) {
 
 	b.ingestLocked(b.rtxPktBuf[:n], &repairedPkt, arrivalTime, false, true)
 
-	var fecDelta fecRecoveryDelta
-	var onFECRecovery cbFECRecovery
-	if b.fecDecoder != nil {
-		fecDelta, onFECRecovery = b.feedFECLocked(&repairedPkt, arrivalTime)
-	}
-
 	rtcpPackets := b.getRTCPPackets(arrivalTime)
 	b.Unlock()
 
-	b.packetPostProcess(rtcpPackets, fecDelta, onFECRecovery)
+	b.packetPostProcess(rtcpPackets)
 }
 
 func (b *Buffer) SetPrimaryBufferForFEC(primaryBuffer *Buffer) {
@@ -437,7 +396,7 @@ func (b *Buffer) SetPrimaryBufferForFEC(primaryBuffer *Buffer) {
 
 	// Let the primary know the repair stream SSRC so its decoder is ready
 	// before the first FEC packet arrives.
-	primaryBuffer.setFECSSRC(ssrc)
+	primaryBuffer.SetFECSSRC(ssrc)
 
 	for _, pp := range pkts {
 		var rtpPacket rtp.Packet
@@ -471,84 +430,8 @@ func (b *Buffer) setFECRepairBufferLimit() {
 	b.pPackets = retained
 }
 
-func (b *Buffer) setFECSSRC(ssrc uint32) {
-	b.Lock()
-	b.fecSSRC = ssrc
-	b.maybeCreateFECDecoderLocked()
-	b.Unlock()
-}
-
-// maybeCreateFECDecoderLocked creates the FEC decoder as soon as the repair
-// stream SSRC is known and the buffer is bound with a negotiated flexfec
-// payload type. Protected media is read from the primary RTP packet bucket.
-func (b *Buffer) maybeCreateFECDecoderLocked() {
-	if b.fecDecoder != nil || b.fecSSRC == 0 || !b.isBound || b.fecPayloadType == 0 {
-		return
-	}
-
-	b.fecDecoder = flexfec.NewDecoder(b.fecSSRC, b.BufferBase.SSRC(), b.getFECMediaPacketLocked, b.logger)
-	b.logger.Debugw("flexfec decoder created", "fecSSRC", b.fecSSRC, "mediaSSRC", b.BufferBase.SSRC())
-}
-
-// recreateFECDecoderLocked drops FlexFEC state from the previous sequence
-// space. Stream restart resyncs the packet bucket and sequence map, so repair
-// packets retained by the old decoder would recover against the wrong media
-// or be treated as duplicates of the new repair stream.
-func (b *Buffer) recreateFECDecoderLocked() {
-	if b.fecDecoder == nil {
-		return
-	}
-
-	b.fecDecoder = nil
-	b.maybeCreateFECDecoderLocked()
-}
-
-func (b *Buffer) getFECMediaPacketLocked(sequenceNumber uint16, dst []byte) (int, error) {
-	if b.bucket == nil || b.rtpStats == nil {
-		return 0, errFECMediaPacketNotFound
-	}
-
-	// FlexFEC masks use the publisher's sequence-number space. BufferBase
-	// removes padding-only packets from the downstream space, so resolve the
-	// original extended sequence number and apply the same adjustment used
-	// when the packet was inserted into the bucket.
-	highestSequenceNumber := b.rtpStats.ExtendedHighestSequenceNumber()
-	extendedSequenceNumber := int64(highestSequenceNumber) + int64(int16(sequenceNumber-uint16(highestSequenceNumber)))
-	if extendedSequenceNumber < 0 {
-		return 0, errFECMediaPacketNotFound
-	}
-
-	extendedSN := uint64(extendedSequenceNumber)
-	sequenceNumberAdjustment, err := b.snRangeMap.GetValue(extendedSN)
-	if err != nil || sequenceNumberAdjustment > extendedSN {
-		return 0, errFECMediaPacketNotFound
-	}
-
-	return b.bucket.GetPacket(dst, extendedSN-sequenceNumberAdjustment)
-}
-
-// OnFECRecovery is called with counter deltas whenever FEC packets are
-// processed since the previous callback: FEC packets received, media packets
-// recovered, FEC packets discarded and FEC bytes received.
-func (b *Buffer) OnFECRecovery(fn cbFECRecovery) {
-	b.Lock()
-	b.onFECRecovery = fn
-	b.Unlock()
-}
-
-// FECDecoderStats returns cumulative FlexFEC decode counters of the buffer.
-func (b *Buffer) FECDecoderStats() flexfec.DecoderStats {
-	b.RLock()
-	defer b.RUnlock()
-
-	if b.fecDecoder == nil {
-		return flexfec.DecoderStats{}
-	}
-	return b.fecDecoder.Stats()
-}
-
-// writeFEC handles a packet of the coupled FlexFEC repair stream, recovered
-// media packets are injected into the regular packet pipeline.
+// writeFEC accounts for transport feedback on the coupled repair stream.
+// BufferBase handles decoding and injecting recovered media packets.
 func (b *Buffer) writeFEC(fecPkt *rtp.Packet, arrivalTime int64) {
 	b.Lock()
 	if !b.isBound {
@@ -565,83 +448,14 @@ func (b *Buffer) writeFEC(fecPkt *rtp.Packet, arrivalTime int64) {
 		}
 	}
 
-	if b.fecPayloadType == 0 || fecPkt.PayloadType != b.fecPayloadType {
-		b.logger.Debugw("unexpected fec payload type", "expected", b.fecPayloadType, "actual", fecPkt.PayloadType)
+	if !b.BufferBase.HandleFECPacketLocked(fecPkt, arrivalTime) {
 		b.Unlock()
 		return
 	}
-
-	if b.fecDecoder == nil {
-		// normally created when the pair is declared, fall back to the
-		// observed repair stream SSRC
-		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
-		}
-	}
-
-	fecDelta, onFECRecovery := b.feedFECLocked(fecPkt, arrivalTime)
 	rtcpPackets := b.getRTCPPackets(arrivalTime)
 	b.Unlock()
 
-	b.packetPostProcess(rtcpPackets, fecDelta, onFECRecovery)
-}
-
-// feedFECLocked runs a media or FEC packet through the FEC decoder and
-// injects recovered packets into the packet pipeline. Must be called with the
-// buffer lock held and a non-nil decoder.
-func (b *Buffer) feedFECLocked(
-	pkt *rtp.Packet,
-	arrivalTime int64,
-) (fecRecoveryDelta, cbFECRecovery) {
-	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)
-	}
-	for _, rp := range recovered {
-		n, err := rp.MarshalTo(b.fecPktBuf)
-		if err != nil {
-			b.logger.Warnw("could not marshal fec recovered packet", err, "ssrc", b.BufferBase.SSRC(), "sn", rp.SequenceNumber)
-			continue
-		}
-
-		// 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. 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 {
-		return fecRecoveryDelta{
-			received:      int(fecPackets),
-			recovered:     int(fecPacketsRecovered),
-			discarded:     int(fecPacketsDiscarded),
-			bytesReceived: int(fecBytes),
-		}, cb
-	}
-
-	return fecRecoveryDelta{}, nil
+	b.packetPostProcess(rtcpPackets)
 }
 
 func (b *Buffer) Read(buff []byte) (n int, err error) {
@@ -738,13 +552,7 @@ func (b *Buffer) ingestLocked(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime
 // `rtcpPackets` should be RTCP feedback from the current NACK state, which
 // already reflects any FEC recovery performed during this write.
 // This is to ensure that packets recovered via FEC is not NACKed.
-func (b *Buffer) packetPostProcess(
-	rtcpPackets []rtcp.Packet,
-	fecDelta fecRecoveryDelta,
-	onFECRecovery cbFECRecovery,
-) {
-	fecDelta.invoke(onFECRecovery)
-
+func (b *Buffer) packetPostProcess(rtcpPackets []rtcp.Packet) {
 	if len(rtcpPackets) != 0 {
 		if cb := b.getOnRtcpFeedback(); cb != nil {
 			cb(rtcpPackets)

pkg/sfu/buffer/buffer_base.go

@@ -32,10 +32,12 @@ import (
 	"go.uber.org/atomic"
 
 	"github.com/livekit/livekit-server/pkg/sfu/audio"
+	"github.com/livekit/livekit-server/pkg/sfu/flexfec"
 	act "github.com/livekit/livekit-server/pkg/sfu/rtpextension/abscapturetime"
 	dd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/dependencydescriptor"
 	"github.com/livekit/livekit-server/pkg/sfu/rtpstats"
 	"github.com/livekit/livekit-server/pkg/sfu/utils"
+	"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
 	"github.com/livekit/mediatransportutil/pkg/bucket"
 	"github.com/livekit/mediatransportutil/pkg/codec"
 	"github.com/livekit/mediatransportutil/pkg/nack"
@@ -46,6 +48,8 @@ import (
 )
 
 var (
+	errFECMediaPacketNotFound = errors.New("fec media packet not found")
+
 	ExtPacketFactory = &sync.Pool{
 		New: func() any {
 			return &ExtPacket{}
@@ -152,15 +156,14 @@ const (
 )
 
 type BufferBaseParams struct {
-	SSRC                  uint32
-	MaxVideoPkts          int
-	MaxAudioPkts          int
-	LoggerComponents      []string
-	SendPLI               func()
-	OnStreamRestartLocked func()
-	IsReportingEnabled    bool
-	IsOOBSequenceNumber   bool
-	IsDDRestartEnabled    bool
+	SSRC                uint32
+	MaxVideoPkts        int
+	MaxAudioPkts        int
+	LoggerComponents    []string
+	SendPLI             func()
+	IsReportingEnabled  bool
+	IsOOBSequenceNumber bool
+	IsDDRestartEnabled  bool
 }
 
 type BufferBase struct {
@@ -188,6 +191,9 @@ type BufferBase struct {
 	payloadType    uint8
 	rtxPayloadType uint8
 	fecPayloadType uint8
+	fecSSRC        uint32
+	fecDecoder     *flexfec.Decoder
+	fecPktBuf      []byte
 
 	snRangeMap *utils.RangeMap[uint64, uint64]
 
@@ -427,6 +433,7 @@ func (b *BufferBase) BindLocked(rtpParameters webrtc.RTPParameters, codec webrtc
 	}
 
 	b.StartKeyFrameSeeder()
+	b.maybeCreateFECDecoderLocked()
 
 	return nil
 }
@@ -600,9 +607,7 @@ func (b *BufferBase) restartStreamLocked(reason string, isDetected bool) {
 
 	b.StartKeyFrameSeeder()
 
-	if f := b.params.OnStreamRestartLocked; f != nil {
-		f()
-	}
+	b.recreateFECDecoderLocked()
 
 	if isDetected {
 		b.isRestartPending = true
@@ -773,6 +778,32 @@ func (b *BufferBase) HandleIncomingPacketLocked(
 		}
 	}
 
+	sequenceNumber := rtpPacket.SequenceNumber
+	extSequenceNumber, err := b.handleIncomingPacketLocked(
+		rawPkt, rtpPacket, arrivalTime, isBuffered, isRTX, skippedSeqs, oobSequenceNumber,
+	)
+	if b.fecDecoder != nil {
+		// Media can complete a retained FEC window. Recover before callers
+		// generate feedback, using the publisher's sequence number because
+		// ingestion adjusts it when padding has been removed.
+		mediaPacket := *rtpPacket
+		mediaPacket.SequenceNumber = sequenceNumber
+		b.feedFECLocked(&mediaPacket, arrivalTime)
+	}
+	return extSequenceNumber, err
+}
+
+// handleIncomingPacketLocked ingests media without re-entering the decoder.
+// Recovered packets use this path because DecodeFEC already chains recovery.
+func (b *BufferBase) handleIncomingPacketLocked(
+	rawPkt []byte,
+	rtpPacket *rtp.Packet,
+	arrivalTime int64,
+	isBuffered bool,
+	isRTX bool,
+	skippedSeqs []uint16,
+	oobSequenceNumber uint16,
+) (uint64, error) {
 	b.processAudioSsrcLevelHeaderExtension(rtpPacket, arrivalTime)
 
 	if len(skippedSeqs) > 0 {
@@ -959,6 +990,154 @@ func (b *BufferBase) HandleIncomingPacketLocked(
 	return ep.ExtSequenceNumber, nil
 }
 
+// SetFECSSRC identifies the repair stream protecting this buffer. It can be
+// called before or after Bind; decoding starts once FlexFEC is negotiated.
+func (b *BufferBase) SetFECSSRC(ssrc uint32) {
+	b.Lock()
+	b.fecSSRC = ssrc
+	b.maybeCreateFECDecoderLocked()
+	b.Unlock()
+}
+
+// maybeCreateFECDecoderLocked creates the FEC decoder as soon as the repair
+// stream SSRC is known and the buffer is bound with a negotiated flexfec
+// payload type. Protected media is read from the primary RTP packet bucket.
+func (b *BufferBase) maybeCreateFECDecoderLocked() {
+	if b.fecDecoder != nil || b.fecSSRC == 0 || b.bucket == nil || b.fecPayloadType == 0 {
+		return
+	}
+
+	b.fecDecoder = flexfec.NewDecoder(b.fecSSRC, b.SSRC(), b.getFECMediaPacketLocked, b.logger)
+	b.logger.Debugw("flexfec decoder created", "fecSSRC", b.fecSSRC, "mediaSSRC", b.SSRC())
+}
+
+// recreateFECDecoderLocked drops FlexFEC state from the previous sequence
+// space. Stream restart resyncs the packet bucket and sequence map, so repair
+// packets retained by the old decoder would recover against the wrong media
+// or be treated as duplicates of the new repair stream.
+func (b *BufferBase) recreateFECDecoderLocked() {
+	if b.fecDecoder == nil {
+		return
+	}
+
+	b.fecDecoder = nil
+	b.maybeCreateFECDecoderLocked()
+}
+
+func (b *BufferBase) getFECMediaPacketLocked(sequenceNumber uint16, dst []byte) (int, error) {
+	if b.bucket == nil || b.rtpStats == nil {
+		return 0, errFECMediaPacketNotFound
+	}
+
+	// FlexFEC masks use the publisher's sequence-number space. BufferBase
+	// removes padding-only packets from the downstream space, so resolve the
+	// original extended sequence number and apply the same adjustment used
+	// when the packet was inserted into the bucket.
+	highestSequenceNumber := b.rtpStats.ExtendedHighestSequenceNumber()
+	extendedSequenceNumber := int64(highestSequenceNumber) + int64(int16(sequenceNumber-uint16(highestSequenceNumber)))
+	if extendedSequenceNumber < 0 {
+		return 0, errFECMediaPacketNotFound
+	}
+
+	extendedSN := uint64(extendedSequenceNumber)
+	sequenceNumberAdjustment, err := b.snRangeMap.GetValue(extendedSN)
+	if err != nil || sequenceNumberAdjustment > extendedSN {
+		return 0, errFECMediaPacketNotFound
+	}
+
+	return b.bucket.GetPacket(dst, extendedSN-sequenceNumberAdjustment)
+}
+
+// FECDecoderStats returns cumulative FlexFEC decode counters of the buffer.
+func (b *BufferBase) FECDecoderStats() flexfec.DecoderStats {
+	b.RLock()
+	defer b.RUnlock()
+
+	if b.fecDecoder == nil {
+		return flexfec.DecoderStats{}
+	}
+	return b.fecDecoder.Stats()
+}
+
+// HandleFECPacket decodes a packet from the coupled FlexFEC repair stream
+// and injects recovered media into the regular packet pipeline. It reports
+// whether the packet was passed to the decoder.
+func (b *BufferBase) HandleFECPacket(fecPkt *rtp.Packet, arrivalTime int64) bool {
+	b.Lock()
+	defer b.Unlock()
+
+	return b.HandleFECPacketLocked(fecPkt, arrivalTime)
+}
+
+// HandleFECPacketLocked requires the buffer lock to be held. Recovery updates
+// NACK state before callers generate feedback.
+func (b *BufferBase) HandleFECPacketLocked(fecPkt *rtp.Packet, arrivalTime int64) bool {
+	if b.isClosed.Load() || b.bucket == nil {
+		return false
+	}
+
+	if b.fecPayloadType == 0 || fecPkt.PayloadType != b.fecPayloadType {
+		b.logger.Debugw("unexpected fec payload type", "expected", b.fecPayloadType, "actual", fecPkt.PayloadType)
+		return false
+	}
+
+	if b.fecDecoder == nil {
+		// normally created when the pair is declared, fall back to the
+		// observed repair stream SSRC
+		b.fecSSRC = fecPkt.SSRC
+		b.maybeCreateFECDecoderLocked()
+		if b.fecDecoder == nil {
+			b.logger.Infow("flexfec decoder unavailable after creation attempt", "fecSSRC", b.fecSSRC, "mediaSSRC", b.SSRC(), "payloadType", b.fecPayloadType)
+			return false
+		}
+	}
+
+	b.feedFECLocked(fecPkt, arrivalTime)
+	return true
+}
+
+// feedFECLocked runs a media or FEC packet through the FEC decoder and
+// injects recovered packets into the packet pipeline. Must be called with the
+// buffer lock held and a non-nil decoder.
+func (b *BufferBase) feedFECLocked(pkt *rtp.Packet, arrivalTime int64) {
+	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,
+		)
+	}
+
+	prometheus.RecordFECUpstream(int(fecPackets), int(fecPacketsRecovered), int(fecPacketsDiscarded), fecBytes)
+
+	if len(recovered) > 0 && b.fecPktBuf == nil {
+		b.fecPktBuf = make([]byte, bucket.RTPMaxPktSize)
+	}
+	for _, rp := range recovered {
+		n, err := rp.MarshalTo(b.fecPktBuf)
+		if err != nil {
+			b.logger.Warnw("could not marshal fec recovered packet", err, "ssrc", b.SSRC(), "sn", rp.SequenceNumber)
+			continue
+		}
+
+		// 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. RTCP is generated once by the caller
+		// after all recovered packets have updated NACK state.
+		b.handleIncomingPacketLocked(b.fecPktBuf[:n], rp, arrivalTime, false, true, nil, 0)
+	}
+}
+
 func (b *BufferBase) updateNACKState(sequenceNumber uint16, flowState rtpstats.RTPFlowState) {
 	if b.nacker == nil {
 		return

pkg/sfu/buffer/buffer_fec_test.go

@@ -27,11 +27,15 @@ import (
 	"github.com/pion/rtp"
 	"github.com/pion/transport/v4/packetio"
 	"github.com/pion/webrtc/v4"
+	promclient "github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/livekit/livekit-server/pkg/sfu/testutils/rtptest"
+	"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
 	"github.com/livekit/mediatransportutil/pkg/twcc"
+	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/protocol/utils/mono"
 )
 
 const (
@@ -49,14 +53,43 @@ var flexFECCodec = webrtc.RTPCodecParameters{
 	PayloadType: webrtc.PayloadType(fecTestFECPT),
 }
 
-func bindFECTestBuffer(t *testing.T, buff *Buffer) {
+func bindFECTestBuffer(t *testing.T, buff interface {
+	Bind(webrtc.RTPParameters, webrtc.RTPCodecCapability, int) error
+}) {
 	t.Helper()
-	buff.codecType = webrtc.RTPCodecTypeVideo
+	require.NoError(t, prometheus.Init("test", livekit.NodeType_SERVER))
 	require.NoError(t, buff.Bind(webrtc.RTPParameters{
 		Codecs: []webrtc.RTPCodecParameters{vp8Codec, flexFECCodec},
 	}, vp8Codec.RTPCodecCapability, 0))
 }
 
+func fecUpstreamCounters(t *testing.T) map[string]float64 {
+	t.Helper()
+	families, err := promclient.DefaultGatherer.Gather()
+	require.NoError(t, err)
+	counters := make(map[string]float64)
+	for _, family := range families {
+		if family.GetName() != "livekit_fec_total" && family.GetName() != "livekit_fec_bytes" {
+			continue
+		}
+		for _, metric := range family.Metric {
+			var direction, kind string
+			for _, label := range metric.Label {
+				switch label.GetName() {
+				case "direction":
+					direction = label.GetValue()
+				case "type":
+					kind = label.GetValue()
+				}
+			}
+			if direction == "incoming" {
+				counters[family.GetName()+"/"+kind] += metric.GetCounter().GetValue()
+			}
+		}
+	}
+	return counters
+}
+
 func writePacket(t *testing.T, buff *Buffer, pkt *rtp.Packet) {
 	t.Helper()
 	raw, err := pkt.Marshal()
@@ -160,12 +193,7 @@ func TestBufferFECRecoversDroppedPacket(t *testing.T) {
 	factory.SetFECPair(fecTestFECSSRC, fecTestMediaSSRC)
 
 	bindFECTestBuffer(t, primary)
-
-	var recoveredDelta, receivedDelta int
-	primary.OnFECRecovery(func(received int, recovered int, discarded int, bytesReceived int) {
-		recoveredDelta += recovered
-		receivedDelta += received
-	})
+	metricsBefore := fecUpstreamCounters(t)
 
 	media := rtptest.GenerateVP8Packets(100, 10, uint8(vp8Codec.PayloadType), fecTestMediaSSRC)
 	encoder := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC)
@@ -187,8 +215,6 @@ func TestBufferFECRecoversDroppedPacket(t *testing.T) {
 	assert.EqualValues(t, len(fecPackets), stats.FECPacketsReceived)
 	assert.EqualValues(t, 1, stats.PacketsRecovered)
 	assert.EqualValues(t, 0, stats.FECPacketsDiscarded)
-	assert.Equal(t, 1, recoveredDelta)
-	assert.Equal(t, len(fecPackets), receivedDelta)
 
 	fecPayloadBytes := 0
 	for i := range fecPackets {
@@ -200,6 +226,10 @@ func TestBufferFECRecoversDroppedPacket(t *testing.T) {
 	assert.EqualValues(t, fecPayloadBytes, rtpStats.FecBytes)
 	assert.EqualValues(t, 0, rtpStats.FecPacketsDiscarded)
 	assert.EqualValues(t, 1, rtpStats.FecPacketsRecovered)
+	metricsAfter := fecUpstreamCounters(t)
+	assert.EqualValues(t, len(fecPackets), metricsAfter["livekit_fec_total/received"]-metricsBefore["livekit_fec_total/received"])
+	assert.EqualValues(t, 1, metricsAfter["livekit_fec_total/recovered"]-metricsBefore["livekit_fec_total/recovered"])
+	assert.EqualValues(t, fecPayloadBytes, metricsAfter["livekit_fec_bytes/received"]-metricsBefore["livekit_fec_bytes/received"])
 
 	// the 9 received packets flow through the ext packet pipeline, the
 	// recovered one fills the bucket like an RTX repair
@@ -262,46 +292,77 @@ func TestBufferFECRecoversEncryptedPayload(t *testing.T) {
 	assert.Equal(t, plaintexts[droppedIdx], decrypted)
 }
 
-func TestBufferFECRecoveryCallbackCanReenterBuffer(t *testing.T) {
-	factory := NewFactoryOfBufferFactory(500, 200).CreateBufferFactory()
+func TestBufferBaseFECRecovery(t *testing.T) {
+	for _, test := range []struct {
+		name        string
+		pairOnBind  bool
+		lateMedia   bool
+		lateRTX     bool
+		parsePacket bool
+	}{
+		{name: "paired before bind", pairOnBind: true},
+		{name: "paired after bind"},
+		{name: "late media", lateMedia: true, parsePacket: true},
+		{name: "late RTX", lateMedia: true, lateRTX: true},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			base := NewBufferBase(BufferBaseParams{
+				SSRC:         fecTestMediaSSRC,
+				MaxVideoPkts: 500,
+				MaxAudioPkts: 200,
+			})
+			t.Cleanup(func() { base.CloseWithReason("test") })
+			if test.pairOnBind {
+				base.SetFECSSRC(fecTestFECSSRC)
+				require.Nil(t, base.fecDecoder)
+			}
+			bindFECTestBuffer(t, base)
+			if !test.pairOnBind {
+				base.SetFECSSRC(fecTestFECSSRC)
+			}
+			require.NotNil(t, base.fecDecoder)
 
-	primary := factory.GetOrNew(packetio.RTPBufferPacket, fecTestMediaSSRC).(*Buffer)
-	fecBuff := factory.GetOrNew(packetio.RTPBufferPacket, fecTestFECSSRC).(*Buffer)
-	factory.SetFECPair(fecTestFECSSRC, fecTestMediaSSRC)
-	bindFECTestBuffer(t, primary)
+			media := rtptest.GenerateVP8Packets(150, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC)
+			fecPackets := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC).EncodeFec(media, 1)
+			require.Len(t, fecPackets, 1)
 
-	callbackDone := make(chan struct{}, 1)
-	primary.OnFECRecovery(func(received int, recovered int, discarded int, bytesReceived int) {
-		primary.FECDecoderStats()
-		callbackDone <- struct{}{}
-	})
+			ingest := func(pkt *rtp.Packet, isRTX bool) {
+				raw, err := pkt.Marshal()
+				require.NoError(t, err)
+				parsed := pkt
+				if test.parsePacket {
+					parsed = nil
+				}
+				_, err = base.HandleIncomingPacket(raw, parsed, mono.UnixNano(), false, isRTX, nil, 0)
+				require.NoError(t, err)
+			}
+			for i := range media {
+				if i != 2 && (i != 3 || !test.lateMedia) {
+					ingest(&media[i], false)
+				}
+			}
+			base.HandleFECPacket(&fecPackets[0], mono.UnixNano())
+			if test.lateMedia {
+				require.Zero(t, base.FECDecoderStats().PacketsRecovered)
+				ingest(&media[3], test.lateRTX)
+			}
 
-	media := rtptest.GenerateVP8Packets(150, 5, uint8(vp8Codec.PayloadType), fecTestMediaSSRC)
-	for i := range media {
-		writePacket(t, primary, &media[i])
-	}
-	fecPackets := pionflexfec.NewFlexEncoder03(fecTestFECPT, fecTestFECSSRC).EncodeFec(media, 1)
-	require.NotEmpty(t, fecPackets)
+			stats := base.GetStats()
+			require.EqualValues(t, 1, stats.FecPackets)
+			require.EqualValues(t, len(fecPackets[0].Payload), stats.FecBytes)
+			require.EqualValues(t, 1, stats.FecPacketsRecovered)
+			require.EqualValues(t, 1, base.FECDecoderStats().PacketsRecovered)
+			require.Empty(t, base.GetNACKPairs())
 
-	rawFEC, err := fecPackets[0].Marshal()
-	require.NoError(t, err)
-	writeDone := make(chan error, 1)
-	go func() {
-		_, err := fecBuff.Write(rawFEC)
-		writeDone <- err
-	}()
-
-	select {
-	case <-callbackDone:
-	case <-time.After(time.Second):
-		t.Fatal("FEC recovery callback deadlocked while re-entering the buffer")
-	}
-
-	select {
-	case err := <-writeDone:
-		require.NoError(t, err)
-	case <-time.After(time.Second):
-		t.Fatal("FEC packet write did not return after callback")
+			var raw [1500]byte
+			n, err := base.GetPacket(raw[:], uint64(media[2].SequenceNumber))
+			require.NoError(t, err)
+			var recovered rtp.Packet
+			require.NoError(t, recovered.Unmarshal(raw[:n]))
+			assert.Equal(t, media[2].Header, recovered.Header)
+			assert.Equal(t, media[2].Payload, recovered.Payload)
+			assert.Equal(t, len(media), base.extPackets.Len())
+		})
 	}
 }