Align downstream FEC grouping with frame boundaries

Generate repairs at each frame marker, splitting large frames into blocks of at most 48 packets. Match upstream Q8 repair-count rounding and minimum-one protection, and lazily reuse packet buffers.

Reserve measured RTP repair overhead during layer allocation so small frames have enough headroom. Cover sparse frames, large-frame tails, buffer reuse, real Pion wire recovery, and live preset changes.

pkg/sfu/downtrack.go

@@ -273,6 +273,7 @@ type DownTrack struct {
 	fecLock              sync.Mutex
 	fecEncoder           atomic.Pointer[flexfec.Encoder]
 	fecProtectionPercent atomic.Uint32
+	fecNotifiedOverhead  atomic.Uint32
 	fecState             flexfec.EncoderState
 
 	receiverLock sync.RWMutex

pkg/sfu/downtrack_fec.go

@@ -35,9 +35,10 @@ func (d *DownTrack) bindFEC(t webrtc.TrackLocalContext) {
 	}
 	for _, codec := range d.negotiatedCodecParameters {
 		if strings.EqualFold(codec.MimeType, webrtc.MimeTypeFlexFEC03) {
-			encoder := flexfec.NewEncoder(uint8(codec.PayloadType), uint32(t.SSRCForwardErrorCorrection()), d.params.OnFECSent)
+			encoder := flexfec.NewEncoder(uint8(codec.PayloadType), uint32(t.SSRCForwardErrorCorrection()), d.onFECSent)
 			encoder.SeedState(d.fecState)
 			encoder.SetProtectionPercent(d.fecProtectionPercent.Load())
+			d.fecNotifiedOverhead.Store(encoder.OverheadPercent())
 			d.fecEncoder.Store(encoder)
 			return
 		}
@@ -79,12 +80,27 @@ func (d *DownTrack) seedFECState(state flexfec.EncoderState) {
 	}
 }
 
-// Include the nominal repair overhead in every allocator input, so layer
-// selection leaves room for repair traffic. The pacer/BWE account actual bytes.
+func (d *DownTrack) onFECSent(packets int, bytes int) {
+	if d.params.OnFECSent != nil {
+		d.params.OnFECSent(packets, bytes)
+	}
+	if encoder := d.fecEncoder.Load(); encoder != nil {
+		percent := encoder.OverheadPercent()
+		if d.fecNotifiedOverhead.Swap(percent) != percent {
+			if listener := d.getStreamAllocatorListener(); listener != nil {
+				listener.OnSubscriptionChanged(d)
+			}
+		}
+	}
+}
+
+// Reserve the greater of configured and measured repair overhead in every
+// allocator input. Small frames can require much more than the nominal preset.
+// The pacer/BWE account actual bytes, including all repair headers.
 func (d *DownTrack) getLayeredBitrateWithFEC() ([]int32, Bitrates) {
 	layers, bitrates := d.Receiver().GetLayeredBitrate()
-	if d.fecEncoder.Load() != nil {
-		percent := int64(d.fecProtectionPercent.Load())
+	if encoder := d.fecEncoder.Load(); encoder != nil {
+		percent := int64(encoder.OverheadPercent())
 		if percent == 0 {
 			return layers, bitrates
 		}
@@ -120,6 +136,7 @@ func (d *DownTrack) SetFECProtection(level livekit.FECProtection) {
 	encoder := d.fecEncoder.Load()
 	if encoder != nil {
 		encoder.SetProtectionPercent(percent)
+		d.fecNotifiedOverhead.Store(percent)
 	}
 	d.fecLock.Unlock()
 	if encoder != nil {

pkg/sfu/downtrack_fec_integration_test.go

@@ -62,16 +62,20 @@ func TestDownTrackFlexFECOnWire(t *testing.T) {
 		pacing                         string
 		level                          *livekit.FECProtection
 		protectionPercent              uint32
+		framePackets, frameRepairs     int
 	}{
-		{"pass through", true, true, false, "pass", livekit.FECProtection_FEC_MEDIUM.Enum(), 25},
-		{"queued", true, true, false, "queue", livekit.FECProtection_FEC_MEDIUM.Enum(), 25},
-		{"low", true, true, false, "queue", livekit.FECProtection_FEC_LOW.Enum(), 15},
-		{"high", true, true, false, "leaky", livekit.FECProtection_FEC_HIGH.Enum(), 35},
-		{"none", true, true, false, "pass", livekit.FECProtection_FEC_NONE.Enum(), 0},
-		{"default none", true, true, false, "pass", nil, 0},
-		{"paced encrypted", true, true, true, "leaky", livekit.FECProtection_FEC_MEDIUM.Enum(), 25},
-		{"subscriber declines", true, false, false, "pass", livekit.FECProtection_FEC_HIGH.Enum(), 35},
-		{"disabled", false, true, false, "pass", livekit.FECProtection_FEC_HIGH.Enum(), 35},
+		{"pass through", true, true, false, "pass", livekit.FECProtection_FEC_MEDIUM.Enum(), 25, 20, 5},
+		{"queued", true, true, false, "queue", livekit.FECProtection_FEC_MEDIUM.Enum(), 25, 20, 5},
+		{"low", true, true, false, "queue", livekit.FECProtection_FEC_LOW.Enum(), 15, 20, 3},
+		{"high", true, true, false, "leaky", livekit.FECProtection_FEC_HIGH.Enum(), 35, 20, 7},
+		{"none", true, true, false, "pass", livekit.FECProtection_FEC_NONE.Enum(), 0, 20, 0},
+		{"default none", true, true, false, "pass", nil, 0, 20, 0},
+		{"paced encrypted", true, true, true, "leaky", livekit.FECProtection_FEC_MEDIUM.Enum(), 25, 20, 5},
+		{"subscriber declines", true, false, false, "pass", livekit.FECProtection_FEC_HIGH.Enum(), 35, 20, 0},
+		{"disabled", false, true, false, "pass", livekit.FECProtection_FEC_HIGH.Enum(), 35, 20, 0},
+		{"single packet frames", true, true, false, "pass", livekit.FECProtection_FEC_LOW.Enum(), 15, 1, 1},
+		{"small frames", true, true, false, "queue", livekit.FECProtection_FEC_LOW.Enum(), 15, 2, 1},
+		{"large frames", true, true, false, "leaky", livekit.FECProtection_FEC_HIGH.Enum(), 35, 99, 35},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			h := vnettest.NewHosts(t)
@@ -150,13 +154,15 @@ func TestDownTrackFlexFECOnWire(t *testing.T) {
 				require.Zero(t, fecSSRC)
 			}
 
-			const mediaCount = 20
+			const frameCount = 2
+			mediaCount := frameCount * tc.framePackets
 			sendMedia := func(start, count int) {
 				for i := start; i < start+count; i++ {
+					frameIndex := i / tc.framePackets
 					ep, err := testutils.GetTestExtPacketVP8(&testutils.TestExtPacketParams{
-						SequenceNumber: uint16(6000 + i), Timestamp: uint32(270000 + i*3000), SSRC: 0x44444444,
-						PayloadType: 96, PayloadSize: 100 + i, IsKeyFrame: true, Marker: true,
-					}, &codec.VP8{FirstByte: 0x10, S: true, PictureID: uint16(i + 1), IsKeyFrame: true})
+						SequenceNumber: uint16(6000 + i), Timestamp: uint32(270000 + frameIndex*3000), SSRC: 0x44444444,
+						PayloadType: 96, PayloadSize: 100 + i, IsKeyFrame: true, Marker: i%tc.framePackets == tc.framePackets-1,
+					}, &codec.VP8{FirstByte: 0x10, S: i%tc.framePackets == 0, PictureID: uint16(frameIndex + 1), IsKeyFrame: true})
 					require.NoError(t, err)
 					ep.Packet.Payload = distinctivePayload(byte(i), 100+i)
 					require.EqualValues(t, 1, dt.WriteRTP(ep, 0))
@@ -166,7 +172,7 @@ func TestDownTrackFlexFECOnWire(t *testing.T) {
 			sendMedia(0, mediaCount)
 			expected := mediaCount
 			if negotiated {
-				expected += mediaCount * int(tc.protectionPercent) / 100
+				expected += frameCount * tc.frameRepairs
 			}
 			require.Eventually(t, func() bool { return capture.count() >= expected }, 10*time.Second, 10*time.Millisecond)
 			require.Equal(t, expected, capture.count())
@@ -187,16 +193,15 @@ func TestDownTrackFlexFECOnWire(t *testing.T) {
 				require.Empty(t, repair)
 				return
 			}
-			require.Len(t, repair, mediaCount*int(tc.protectionPercent)/100)
+			require.Len(t, repair, frameCount*tc.frameRepairs)
 			for i := 1; i < len(repair); i++ {
 				require.Equal(t, repair[i-1].SequenceNumber+1, repair[i].SequenceNumber)
 			}
-			// Drop media in the second and third groups (protected by every enabled
-			// preset) and recover from the actual wire
+			// Drop media from the first and last frames and recover from the actual wire
 			// representation, including translated SSRC/PT/sequence and payload.
 			stored := map[uint16][]byte{}
 			for i, packet := range media {
-				if i != 7 && i != 12 {
+				if i != 0 && i != mediaCount-1 {
 					stored[packet.SequenceNumber], _ = packet.Marshal()
 				}
 			}
@@ -217,7 +222,7 @@ func TestDownTrackFlexFECOnWire(t *testing.T) {
 				}
 			}
 			require.Len(t, recoveredPackets, 2)
-			for _, lost := range []*rtp.Packet{media[7], media[12]} {
+			for _, lost := range []*rtp.Packet{media[0], media[mediaCount-1]} {
 				want, _ := lost.Marshal()
 				require.Equal(t, want, recoveredPackets[lost.SequenceNumber])
 			}
@@ -226,18 +231,18 @@ func TestDownTrackFlexFECOnWire(t *testing.T) {
 				sequence := repair[len(repair)-1].SequenceNumber
 				start := mediaCount
 				for _, update := range []struct {
-					level   livekit.FECProtection
-					percent int
+					level           livekit.FECProtection
+					repairsPerFrame int
 				}{
-					{livekit.FECProtection_FEC_HIGH, 35},
+					{livekit.FECProtection_FEC_HIGH, 7},
 					{livekit.FECProtection_FEC_NONE, 0},
-					{livekit.FECProtection_FEC_LOW, 15},
+					{livekit.FECProtection_FEC_LOW, 3},
 				} {
 					before := capture.count()
 					dt.SetFECProtection(update.level)
 					sendMedia(start, mediaCount)
 					start += mediaCount
-					numRepair := mediaCount * update.percent / 100
+					numRepair := frameCount * update.repairsPerFrame
 					require.Eventually(t, func() bool { return capture.count() >= before+mediaCount+numRepair }, 5*time.Second, 10*time.Millisecond)
 					require.Equal(t, before+mediaCount+numRepair, capture.count())
 					for _, packet := range capture.all()[before:] {
@@ -281,7 +286,7 @@ func TestDownTrackFlexFECOnWire(t *testing.T) {
 				dt.ForceForwardLayerForTest(buffer.VideoLayer{Spatial: 0, Temporal: 0})
 				before := capture.count()
 				sendMedia(start, mediaCount)
-				require.Eventually(t, func() bool { return capture.count() >= before+25 }, 5*time.Second, 10*time.Millisecond)
+				require.Eventually(t, func() bool { return capture.count() >= before+mediaCount+frameCount*5 }, 5*time.Second, 10*time.Millisecond)
 				repairs := 0
 				for _, packet := range capture.all()[before:] {
 					if packet.SSRC == fecSSRC {
@@ -290,7 +295,7 @@ func TestDownTrackFlexFECOnWire(t *testing.T) {
 						repairs++
 					}
 				}
-				require.Equal(t, 5, repairs)
+				require.Equal(t, frameCount*5, repairs)
 			}
 		})
 	}

pkg/sfu/downtrack_fec_test.go

@@ -57,10 +57,11 @@ func TestDownTrackFECNegotiationAndLifecycle(t *testing.T) {
 			require.NotNil(t, old)
 			p := rtp.Header{Version: 2, SSRC: 123, PayloadType: 96}
 			var lastSequenceNumber uint16
-			for i := range flexfec.MediaPacketsPerGroup {
+			for i := range 5 {
 				p.SequenceNumber++
+				p.Marker = i == 4
 				repair := old.Encode(&p, []byte{1, 2, 3})
-				if i == flexfec.MediaPacketsPerGroup-1 {
+				if i == 4 {
 					require.Len(t, repair, 1)
 					require.EqualValues(t, 456, repair[0].SSRC)
 					require.EqualValues(t, 118, repair[0].PayloadType, "use negotiated PT")
@@ -69,8 +70,9 @@ func TestDownTrackFECNegotiationAndLifecycle(t *testing.T) {
 			}
 			d.bindFEC(c)
 			require.NotSame(t, old, d.fecEncoder.Load())
-			for range flexfec.MediaPacketsPerGroup {
+			for i := range 5 {
 				p.SequenceNumber++
+				p.Marker = i == 4
 				require.Empty(t, old.Encode(&p, []byte{1, 2, 3}), "old queued writes cannot generate repair")
 				for _, repair := range d.fecEncoder.Load().Encode(&p, []byte{1, 2, 3}) {
 					require.Equal(t, lastSequenceNumber+1, repair.SequenceNumber, "reusing an SSRC must preserve repair sequencing")
@@ -89,7 +91,9 @@ func TestDownTrackFECBandwidthReservation(t *testing.T) {
 	_, got := d.getLayeredBitrateWithFEC()
 	require.Equal(t, rates, got)
 	d.SetFECProtection(livekit.FECProtection_FEC_MEDIUM)
-	d.fecEncoder.Store(flexfec.NewEncoder(115, 456, nil))
+	encoder := flexfec.NewEncoder(115, 456, nil)
+	encoder.SetProtectionPercent(25)
+	d.fecEncoder.Store(encoder)
 	layers, got := d.getLayeredBitrateWithFEC()
 	require.Equal(t, []int32{0, 1}, layers)
 	require.EqualValues(t, 125_000, got[0][0])
@@ -105,6 +109,41 @@ type fecTrackReceiver struct {
 
 func (r *fecTrackReceiver) GetLayeredBitrate() ([]int32, Bitrates) { return []int32{0, 1}, r.rates }
 
+type fecAllocatorListener struct {
+	DownTrackStreamAllocatorListener
+	changes int
+}
+
+func (l *fecAllocatorListener) OnSubscriptionChanged(*DownTrack) { l.changes++ }
+
+func TestDownTrackFECMeasuredBandwidthReservation(t *testing.T) {
+	listener := &fecAllocatorListener{}
+	d := &DownTrack{
+		params: DownTrackParams{EnableFlexFEC: true}, kind: webrtc.RTPCodecTypeVideo,
+		receiver:                  &fecTrackReceiver{rates: Bitrates{{100_000}}},
+		negotiatedCodecParameters: []webrtc.RTPCodecParameters{{RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeFlexFEC03}, PayloadType: 115}},
+		streamAllocatorListener:   listener,
+	}
+	d.SetFECProtection(livekit.FECProtection_FEC_LOW)
+	d.bindFEC(fecTrackContext{ssrcFEC: 456})
+	e := d.fecEncoder.Load()
+	p := rtp.Header{Version: 2, SSRC: 123, PayloadType: 96, Marker: true}
+	for range 2 {
+		p.SequenceNumber++
+		repairs := e.Encode(&p, make([]byte, 100))
+		require.Len(t, repairs, 1)
+		e.RecordSent(1, len(repairs[0].Payload), repairs[0].MarshalSize())
+	}
+	_, rates := d.getLayeredBitrateWithFEC()
+	require.EqualValues(t, 218_000, rates[0][0], "reserve actual small-frame overhead, including RTP/FEC headers")
+	require.Equal(t, 1, listener.changes, "reallocate on an overhead change, not on every repair")
+	d.SetFECProtection(livekit.FECProtection_FEC_NONE)
+	_, rates = d.getLayeredBitrateWithFEC()
+	require.EqualValues(t, 100_000, rates[0][0])
+	require.Equal(t, 2, listener.changes)
+	d.closeFEC()
+}
+
 func TestDownTrackFECProtectionChanges(t *testing.T) {
 	d := &DownTrack{
 		params: DownTrackParams{EnableFlexFEC: true}, kind: webrtc.RTPCodecTypeVideo,
@@ -121,8 +160,9 @@ func TestDownTrackFECProtectionChanges(t *testing.T) {
 	p := rtp.Header{Version: 2, SSRC: 123, PayloadType: 96}
 	send := func() int {
 		count := 0
-		for range 20 {
+		for i := range 20 {
 			p.SequenceNumber++
+			p.Marker = i == 19
 			count += len(d.fecEncoder.Load().Encode(&p, []byte{1, 2, 3}))
 		}
 		return count

pkg/sfu/flexfec/encoder.go

@@ -21,6 +21,7 @@ import (
 
 	pionflexfec "github.com/pion/interceptor/pkg/flexfec"
 	"github.com/pion/rtp"
+	"go.uber.org/atomic"
 	"go.uber.org/zap/zapcore"
 
 	"github.com/livekit/protocol/utils/mono"
@@ -30,12 +31,14 @@ const (
 	// DefaultProtectionPercent is the repair-to-media packet ratio when unspecified.
 	DefaultProtectionPercent uint32 = 0
 	MaxProtectionPercent     uint32 = 100
-	// MediaPacketsPerGroup bounds both retained packet memory and encoding work.
-	MediaPacketsPerGroup = 5
-	maxEncoderGroupAge   = 200 * time.Millisecond
-	// Reserve the FEC header and up to 16 bytes of outer RTP extensions (AST/TWCC).
-	// Groups of at most 15 packets use only the first FlexFEC packet mask.
-	maxEncoderMediaPacketSize = maxMediaPacketSize - pionflexfec.BaseFec03HeaderSize - 16
+	// MaxMediaPacketsPerGroup bounds memory and work for large frames. Like
+	// libwebrtc, use at most 48 media packets in a protection block.
+	MaxMediaPacketsPerGroup = 48
+	maxEncoderGroupAge      = 200 * time.Millisecond
+	overheadUpdateInterval  = time.Second
+	// Reserve all three FEC masks and up to 16 bytes of outer RTP extensions
+	// (AST/TWCC). A 48-packet group can require the third mask.
+	maxEncoderMediaPacketSize = maxMediaPacketSize - pionflexfec.BaseFec03HeaderSize - 12 - 16
 )
 
 // EncoderState preserves the repair sequence when a sender SSRC is reused.
@@ -52,8 +55,8 @@ func (s EncoderState) MarshalLogObject(e zapcore.ObjectEncoder) error {
 }
 
 type encoderMediaGroup struct {
-	packets [MediaPacketsPerGroup]rtp.Packet
-	storage [MediaPacketsPerGroup][maxEncoderMediaPacketSize]byte
+	packets [MaxMediaPacketsPerGroup]rtp.Packet
+	storage [MaxMediaPacketsPerGroup]*[maxEncoderMediaPacketSize]byte
 }
 
 // Encoder batches final, outgoing video packets for Pion's FlexFEC-03 encoder.
@@ -71,7 +74,12 @@ type Encoder struct {
 	closed            bool
 	onSent            func(packets int, bytes int)
 	protectionPercent uint32
-	repairCredit      uint32
+	protectionFactor  uint32
+
+	overheadPercent   atomic.Uint32
+	overheadUpdatedAt int64
+	sentMediaBytes    uint64
+	sentRepairBytes   uint64
 }
 
 func NewEncoder(payloadType uint8, ssrc uint32, onSent func(packets int, bytes int)) *Encoder {
@@ -100,12 +108,13 @@ func (e *Encoder) SeedState(state EncoderState) {
 	}
 	e.sequenceNumber = state.NextSequenceNumber
 	e.count = 0
-	e.repairCredit = 0
 }
 
-// Encode returns independently owned repair packets for complete groups only.
-// Gaps (including skipped padding), codec switches and stale groups start a new
-// group, since Pion requires consecutive sequence numbers. RTP wrap is valid.
+// Encode returns independently owned repairs at the end of a video frame (the
+// RTP marker), splitting large frames at MaxMediaPacketsPerGroup. Timestamp
+// changes discard an incomplete previous frame. Gaps, codec switches and stale
+// groups also reset the block; Pion requires consecutive sequence numbers.
+// Media is never delayed. No timer or worker is needed for sparse frames.
 // Call after the media write, including any interceptor header modifications.
 func (e *Encoder) Encode(header *rtp.Header, payload []byte) []rtp.Packet {
 	e.mu.Lock()
@@ -116,6 +125,7 @@ func (e *Encoder) Encode(header *rtp.Header, payload []byte) []rtp.Packet {
 
 	now := mono.UnixNano()
 	size := header.MarshalSize() + len(payload)
+	e.sentMediaBytes += uint64(size)
 	if header.Padding || len(payload) == 0 || size > maxEncoderMediaPacketSize {
 		e.count = 0
 		return nil
@@ -129,6 +139,7 @@ func (e *Encoder) Encode(header *rtp.Header, payload []byte) []rtp.Packet {
 		previous := &e.media.packets[e.count-1]
 		if header.SequenceNumber != previous.SequenceNumber+1 ||
 			header.SSRC != previous.SSRC || header.PayloadType != previous.PayloadType ||
+			header.Timestamp != previous.Timestamp ||
 			now-e.startedAt > int64(maxEncoderGroupAge) {
 			e.count = 0
 		}
@@ -137,6 +148,10 @@ func (e *Encoder) Encode(header *rtp.Header, payload []byte) []rtp.Packet {
 		e.startedAt = now
 	}
 
+	if e.media.storage[e.count] == nil {
+		// Retain only the packet buffers this stream has actually needed.
+		e.media.storage[e.count] = new([maxEncoderMediaPacketSize]byte)
+	}
 	raw := e.media.storage[e.count][:size]
 	n, err := header.MarshalTo(raw)
 	if err != nil {
@@ -149,21 +164,17 @@ func (e *Encoder) Encode(header *rtp.Header, payload []byte) []rtp.Packet {
 		return nil
 	}
 	e.count++
-	if e.count < MediaPacketsPerGroup {
+	if !header.Marker && e.count < MaxMediaPacketsPerGroup {
 		return nil
 	}
 
+	count := e.count
 	e.count = 0
-	// Carry fractional packets between groups rather than rounding each group
-	// up (which would turn e.g. 30% into 40%). Storage and latency stay bounded
-	// at five media packets, independent of the configured percentage.
-	e.repairCredit += e.protectionPercent * MediaPacketsPerGroup
-	numRepair := e.repairCredit / 100
-	e.repairCredit %= 100
-	if numRepair == 0 {
-		return nil
-	}
-	repair := e.encoder.EncodeFec(e.media.packets[:], numRepair)
+	// Match libwebrtc's ForwardErrorCorrection::NumFecPackets: round the Q8
+	// protection factor, with at least one repair for every protected block.
+	// In particular, a one-packet frame is protected immediately at every preset.
+	numRepair := max(uint32(1), (uint32(count)*e.protectionFactor+128)>>8)
+	repair := e.encoder.EncodeFec(e.media.packets[:count], numRepair)
 	for i := range repair {
 		// Pion's encoder uses a constant timestamp and a fixed initial sequence number.
 		// Use this stream's media clock and a randomized, continuous repair sequence.
@@ -175,24 +186,54 @@ func (e *Encoder) Encode(header *rtp.Header, payload []byte) []rtp.Packet {
 }
 
 // SetProtectionPercent updates the packet ratio without changing the repair
-// SSRC or sequence number. A new value discards partial groups and old credit.
+// SSRC or sequence number. A new value discards partial groups and measurements.
 // Values above 100 are capped; zero disables encoding before any packet copies.
 func (e *Encoder) SetProtectionPercent(percent uint32) {
 	percent = min(percent, MaxProtectionPercent)
 	e.mu.Lock()
 	if e.protectionPercent != percent {
 		e.protectionPercent = percent
+		// Publish-track presets use the same conversion to libwebrtc's Q8 rate.
+		e.protectionFactor = percent * 255 / 100
 		e.count = 0
-		e.repairCredit = 0
+		e.overheadPercent.Store(percent)
+		e.overheadUpdatedAt = 0
+		e.sentMediaBytes = 0
+		e.sentRepairBytes = 0
 	}
 	e.mu.Unlock()
 }
 
-// RecordSent reports successful repair writes, with RTP payload bytes like the
-// upstream counters. Callbacks run outside the encoder lock.
-func (e *Encoder) RecordSent(packets int, bytes int) {
-	if e.onSent != nil && packets != 0 {
-		e.onSent(packets, bytes)
+// OverheadPercent reserves at least the configured rate, using measured RTP
+// bytes when frame rounding or repair headers increase the actual overhead.
+// Reads by the allocator do not contend with encoding.
+func (e *Encoder) OverheadPercent() uint32 {
+	return e.overheadPercent.Load()
+}
+
+// RecordSent accounts successful repairs, including headers for the allocator
+// and payload bytes for the upstream-compatible telemetry callback. Refresh the
+// estimate on the first repair and at most once per second thereafter.
+// Callbacks run outside the encoder lock.
+func (e *Encoder) RecordSent(packets int, payloadBytes int, rtpBytes int) {
+	if packets == 0 {
+		return
+	}
+	e.mu.Lock()
+	if !e.closed && e.protectionPercent != 0 {
+		e.sentRepairBytes += uint64(rtpBytes)
+		now := mono.UnixNano()
+		if e.sentMediaBytes != 0 && (e.overheadUpdatedAt == 0 || now-e.overheadUpdatedAt >= int64(overheadUpdateInterval)) {
+			percent := uint32((e.sentRepairBytes*100 + e.sentMediaBytes - 1) / e.sentMediaBytes)
+			e.overheadPercent.Store(max(e.protectionPercent, percent))
+			e.overheadUpdatedAt = now
+			e.sentMediaBytes = 0
+			e.sentRepairBytes = 0
+		}
+	}
+	e.mu.Unlock()
+	if e.onSent != nil {
+		e.onSent(packets, payloadBytes)
 	}
 }
 

pkg/sfu/flexfec/encoder_test.go

@@ -25,15 +25,74 @@ import (
 	"github.com/livekit/protocol/logger"
 )
 
+func makeEncoderFrame(t *testing.T, base uint16, count int) []rtp.Packet {
+	t.Helper()
+	packets := makeMediaPackets(t, base, count)
+	for i := range packets {
+		packets[i].Timestamp = uint32(base) * 3000
+	}
+	return packets
+}
+
 func TestEncoderDefaultsToNoProtection(t *testing.T) {
-	encoder := NewEncoder(testFECPT, testFECSSRC, nil)
-	p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT}
+	e := NewEncoder(testFECPT, testFECSSRC, nil)
+	p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT, Marker: true}
 	for range 100 {
 		p.SequenceNumber++
-		require.Empty(t, encoder.Encode(&p, []byte{1, 2, 3}))
+		require.Empty(t, e.Encode(&p, []byte{1, 2, 3}))
+	}
+	require.Zero(t, e.count)
+	require.Nil(t, e.media, "negotiation alone must not allocate packet storage")
+}
+
+func TestEncoderFrameProtectionCounts(t *testing.T) {
+	// Expected counts from upstream's 38/63/89 Q8 presets, including minimum-one
+	// protection and rounding cases where a nominal percentage alone differs.
+	for _, tc := range []struct{ packets, low, medium, high int }{
+		{1, 1, 1, 1}, {2, 1, 1, 1}, {3, 1, 1, 1}, {5, 1, 1, 2},
+		{6, 1, 1, 2}, {8, 1, 2, 3}, {20, 3, 5, 7}, {48, 7, 12, 17},
+	} {
+		for _, level := range []struct {
+			percent uint32
+			want    int
+		}{
+			{0, 0}, {15, tc.low}, {25, tc.medium}, {35, tc.high}, {100, tc.packets}, {101, tc.packets},
+		} {
+			t.Run(fmt.Sprintf("packets_%d/percent_%d", tc.packets, level.percent), func(t *testing.T) {
+				e := NewEncoder(testFECPT, testFECSSRC, nil)
+				e.SetProtectionPercent(level.percent)
+				for i, p := range makeEncoderFrame(t, 100, tc.packets) {
+					repairs := e.Encode(&p.Header, p.Payload)
+					if i == tc.packets-1 {
+						require.Len(t, repairs, level.want)
+						for _, repair := range repairs {
+							require.Equal(t, p.Timestamp, repair.Timestamp)
+						}
+					} else {
+						require.Empty(t, repairs, "do not close the group before the frame ends")
+					}
+				}
+				require.Zero(t, e.count)
+			})
+		}
+	}
+}
+
+func TestEncoderSparseFrames(t *testing.T) {
+	e := NewEncoder(testFECPT, testFECSSRC, nil)
+	e.SetProtectionPercent(15)
+	// No wall-clock wait is necessary: complete frames must not depend on the
+	// previous frame's group age or on the arrival of a later frame.
+	for i := range 3 {
+		e.startedAt -= int64(10 * maxEncoderGroupAge)
+		p := makeEncoderFrame(t, uint16(100+i), 1)[0]
+		repairs := e.Encode(&p.Header, p.Payload)
+		require.Len(t, repairs, 1, "even the first isolated frame needs immediate protection")
+		decoder := newTestDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger())
+		recovered := decoder.DecodeFEC(&repairs[0])
+		require.Len(t, recovered, 1)
+		requirePacketEqual(t, &p, recovered[0])
 	}
-	require.Zero(t, encoder.count, "disabled protection must not retain media")
-	require.Nil(t, encoder.media, "negotiation alone must not allocate packet storage")
 }
 
 func TestEncoderState(t *testing.T) {
@@ -41,14 +100,10 @@ func TestEncoderState(t *testing.T) {
 	e.SetProtectionPercent(20)
 	e.SeedState(EncoderState{SSRC: testFECSSRC, NextSequenceNumber: 65535})
 	for _, sn := range []uint16{65535, 0} {
-		media := makeMediaPackets(t, 100, MediaPacketsPerGroup)
-		for i := range media {
-			repair := e.Encode(&media[i].Header, media[i].Payload)
-			if i == MediaPacketsPerGroup-1 {
-				require.Len(t, repair, 1)
-				require.Equal(t, sn, repair[0].SequenceNumber)
-			}
-		}
+		p := makeEncoderFrame(t, 100, 1)[0]
+		repair := e.Encode(&p.Header, p.Payload)
+		require.Len(t, repair, 1)
+		require.Equal(t, sn, repair[0].SequenceNumber)
 		e.Close()
 		state := e.GetState()
 		require.Equal(t, sn+1, state.NextSequenceNumber)
@@ -63,10 +118,10 @@ func TestEncoderState(t *testing.T) {
 
 func TestEncoderRecoveryWithReusedMemory(t *testing.T) {
 	for _, base := range []uint16{100, 65533} {
-		media := makeMediaPackets(t, base, 2*MediaPacketsPerGroup)
-		encoder := NewEncoder(testFECPT, testFECSSRC, nil)
-		encoder.SetProtectionPercent(20)
-		encoder.sequenceNumber = 65535
+		media := append(makeEncoderFrame(t, base, 5), makeEncoderFrame(t, base+5, 5)...)
+		e := NewEncoder(testFECPT, testFECSSRC, nil)
+		e.SetProtectionPercent(20)
+		e.sequenceNumber = 65535
 		decoder := newTestDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger())
 		var repair []rtp.Packet
 		for i := range media {
@@ -74,8 +129,7 @@ func TestEncoderRecoveryWithReusedMemory(t *testing.T) {
 			require.NoError(t, media[i].SetExtension(3, []byte{byte(i), 2, 3}))
 			require.NoError(t, media[i].SetExtension(5, []byte{7, byte(i)}))
 			packet := media[i].Clone()
-			repair = append(repair, encoder.Encode(&packet.Header, packet.Payload)...)
-			// The caller owns and immediately recycles every part of the input.
+			repair = append(repair, e.Encode(&packet.Header, packet.Payload)...)
 			clear(packet.Payload)
 			clear(packet.CSRC)
 			clear(packet.GetExtension(3))
@@ -88,28 +142,23 @@ func TestEncoderRecoveryWithReusedMemory(t *testing.T) {
 		require.Equal(t, uint16(65535), repair[0].SequenceNumber)
 		require.Zero(t, repair[1].SequenceNumber)
 		for i := range repair {
-			require.Equal(t, media[(i+1)*MediaPacketsPerGroup-1].Timestamp, repair[i].Timestamp)
 			recovered := decoder.DecodeFEC(&repair[i])
 			require.Len(t, recovered, 1)
-			expected, err := media[i*MediaPacketsPerGroup+2].Marshal()
-			require.NoError(t, err)
-			actual, err := recovered[0].Marshal()
-			require.NoError(t, err)
-			require.Equal(t, expected, actual)
+			requirePacketEqual(t, &media[i*5+2], recovered[0])
 		}
 	}
 }
 
 func TestEncoderGroupBoundaries(t *testing.T) {
-	for _, name := range []string{"gap", "duplicate", "out of order", "codec", "ssrc", "stale", "oversized", "padding", "empty"} {
+	for _, name := range []string{"gap", "duplicate", "out of order", "codec", "ssrc", "timestamp", "stale", "oversized", "padding", "empty"} {
 		t.Run(name, func(t *testing.T) {
 			e := NewEncoder(testFECPT, testFECSSRC, nil)
 			e.SetProtectionPercent(20)
-			media := makeMediaPackets(t, 100, 10)
-			for i := 0; i < 4; i++ {
+			media := makeEncoderFrame(t, 100, 5)
+			for i := 0; i < 3; i++ {
 				require.Empty(t, e.Encode(&media[i].Header, media[i].Payload))
 			}
-			p := &media[4]
+			p := &media[3]
 			switch name {
 			case "gap":
 				p.SequenceNumber++
@@ -121,6 +170,8 @@ func TestEncoderGroupBoundaries(t *testing.T) {
 				p.PayloadType++
 			case "ssrc":
 				p.SSRC++
+			case "timestamp":
+				p.Timestamp++
 			case "stale":
 				e.startedAt -= int64(maxEncoderGroupAge)
 			case "oversized":
@@ -130,56 +181,75 @@ func TestEncoderGroupBoundaries(t *testing.T) {
 			case "empty":
 				p.Payload = nil
 			}
-			require.Empty(t, e.Encode(&p.Header, p.Payload), "must not protect a discontinuous group")
-			fresh := makeMediaPackets(t, 200, MediaPacketsPerGroup)
-			for i := range fresh {
-				out := e.Encode(&fresh[i].Header, fresh[i].Payload)
-				if i == MediaPacketsPerGroup-1 {
-					require.Len(t, out, 1, "resume protection after the discontinuity")
-				} else {
-					require.Empty(t, out)
-				}
-			}
+			require.Empty(t, e.Encode(&p.Header, p.Payload))
+			require.LessOrEqual(t, e.count, 1, "discard the previous partial block")
+			fresh := makeEncoderFrame(t, 200, 1)[0]
+			repairs := e.Encode(&fresh.Header, fresh.Payload)
+			require.Len(t, repairs, 1)
+			decoder := newTestDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger())
+			recovered := decoder.DecodeFEC(&repairs[0])
+			require.Len(t, recovered, 1, "a missing marker must not mix consecutive frames")
+			requirePacketEqual(t, &fresh, recovered[0])
 		})
 	}
 }
 
+func TestEncoderLargeFrame(t *testing.T) {
+	e := NewEncoder(testFECPT, testFECSSRC, nil)
+	e.SetProtectionPercent(35)
+	media := makeEncoderFrame(t, 65510, 2*MaxMediaPacketsPerGroup+3)
+	decoder := newTestDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger())
+	var repair []rtp.Packet
+	for i := range media {
+		out := e.Encode(&media[i].Header, media[i].Payload)
+		switch i {
+		case 47, 95:
+			require.Len(t, out, 17)
+		case 98:
+			require.Len(t, out, 1, "protect the tail of the large frame")
+		default:
+			require.Empty(t, out)
+		}
+		require.Less(t, e.count, MaxMediaPacketsPerGroup)
+		repair = append(repair, out...)
+		if i != 47 && i != 95 && i != 98 {
+			decoder.DecodeFEC(&media[i])
+		}
+	}
+	var recovered []*rtp.Packet
+	for i := range repair {
+		recovered = append(recovered, decoder.DecodeFEC(&repair[i])...)
+	}
+	require.Len(t, recovered, 3)
+	for i, lost := range []int{47, 95, 98} {
+		requirePacketEqual(t, &media[lost], recovered[i])
+	}
+}
+
 func TestEncoderSizeLimitAndClose(t *testing.T) {
 	e := NewEncoder(testFECPT, testFECSSRC, nil)
-	e.SetProtectionPercent(20)
-	p := &rtp.Packet{Header: rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT}, Payload: make([]byte, maxEncoderMediaPacketSize-12)}
-	for i := range MediaPacketsPerGroup {
+	e.SetProtectionPercent(1) // one repair covering all three packet masks
+	p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT}
+	payload := make([]byte, maxEncoderMediaPacketSize-12)
+	for i := range MaxMediaPacketsPerGroup {
 		p.SequenceNumber++
-		out := e.Encode(&p.Header, p.Payload)
-		if i == MediaPacketsPerGroup-1 {
+		out := e.Encode(&p, payload)
+		if i == MaxMediaPacketsPerGroup-1 {
 			require.Len(t, out, 1)
-			require.NoError(t, out[0].SetExtension(3, []byte{1, 2, 3}))
-			require.NoError(t, out[0].SetExtension(5, []byte{1, 2}))
+			out[0].Extension = true
+			out[0].ExtensionProfile = 0x1000 // RFC 8285 two-byte extensions
+			require.NoError(t, out[0].SetExtension(20, []byte{1, 2, 3}))
+			require.NoError(t, out[0].SetExtension(22, []byte{1, 2}))
 			require.LessOrEqual(t, out[0].MarshalSize(), maxMediaPacketSize)
 		}
 	}
 	e.Close()
-	for range 2 * MediaPacketsPerGroup {
+	p.Marker = true
+	for range 10 {
 		p.SequenceNumber++
-		require.Empty(t, e.Encode(&p.Header, p.Payload))
+		require.Empty(t, e.Encode(&p, payload))
 	}
-}
-
-func TestEncoderConcurrentClose(t *testing.T) {
-	e := NewEncoder(testFECPT, testFECSSRC, nil)
-	e.SetProtectionPercent(20)
-	var wg sync.WaitGroup
-	for n := range 4 {
-		wg.Go(func() {
-			p := rtp.Header{Version: 2, SSRC: testMediaSSRC, SequenceNumber: uint16(n * 1000)}
-			for range 100 {
-				e.Encode(&p, []byte{1, 2, 3})
-				p.SequenceNumber++
-			}
-		})
-	}
-	e.Close()
-	wg.Wait()
+	require.Nil(t, e.media)
 }
 
 func TestEncoderAccumulationReusesStorage(t *testing.T) {
@@ -188,9 +258,12 @@ func TestEncoderAccumulationReusesStorage(t *testing.T) {
 	p := rtp.Header{Version: 2, SSRC: testMediaSSRC}
 	require.NoError(t, p.SetExtension(3, []byte{1, 2, 3}))
 	payload := make([]byte, 1200)
+	require.Empty(t, e.Encode(&p, payload))
+	require.NotNil(t, e.media.storage[0])
+	require.Nil(t, e.media.storage[1], "small frames do not allocate large-frame payload storage")
 	allocs := testing.AllocsPerRun(100, func() {
 		e.count = 0
-		for range MediaPacketsPerGroup - 1 {
+		for range MaxMediaPacketsPerGroup - 1 {
 			p.SequenceNumber++
 			e.Encode(&p, payload)
 		}
@@ -198,124 +271,119 @@ func TestEncoderAccumulationReusesStorage(t *testing.T) {
 	require.Zero(t, allocs)
 }
 
-func BenchmarkEncoder(b *testing.B) {
-	for _, percent := range []uint32{0, 15, 25, 35} {
-		b.Run(fmt.Sprintf("percent_%d", percent), func(b *testing.B) {
-			e := NewEncoder(testFECPT, testFECSSRC, nil)
-			e.SetProtectionPercent(percent)
-			p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT}
-			_ = p.SetExtension(3, []byte{1, 2, 3})
-			payload := make([]byte, 1200)
-			// Warm the reusable packet slots and Pion's coverage cache.
-			for range 2 * MediaPacketsPerGroup {
-				p.SequenceNumber++
-				e.Encode(&p, payload)
-			}
-			b.ReportAllocs()
-			b.SetBytes(int64(MediaPacketsPerGroup * len(payload)))
-			b.ResetTimer()
-			for b.Loop() {
-				for range MediaPacketsPerGroup {
-					p.SequenceNumber++
-					e.Encode(&p, payload)
-				}
-			}
-		})
-	}
-}
-
-func TestEncoderProtectionPercent(t *testing.T) {
-	for percent := uint32(0); percent <= 101; percent++ {
-		t.Run(fmt.Sprint(percent), func(t *testing.T) {
-			encoder := NewEncoder(testFECPT, testFECSSRC, nil)
-			encoder.SetProtectionPercent(percent)
-			p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT}
-			var count int
-			var previous uint16
-			for range 100 {
-				p.SequenceNumber++
-				repairs := encoder.Encode(&p, []byte{1, 2, 3})
-				require.LessOrEqual(t, len(repairs), MediaPacketsPerGroup)
-				for _, repair := range repairs {
-					if count > 0 {
-						require.Equal(t, previous+1, repair.SequenceNumber)
-					}
-					count++
-					previous = repair.SequenceNumber
-				}
-			}
-			require.EqualValues(t, min(percent, MaxProtectionPercent), count)
-		})
-	}
-}
-
 func TestEncoderProtectionPercentChanges(t *testing.T) {
-	encoder := NewEncoder(testFECPT, testFECSSRC, nil)
-	encoder.SetProtectionPercent(20)
-	p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT}
-	send := func(count int) []rtp.Packet {
-		var result []rtp.Packet
-		for range count {
-			p.SequenceNumber++
-			result = append(result, encoder.Encode(&p, []byte{1, 2, 3})...)
-		}
-		return result
-	}
-	first := send(MediaPacketsPerGroup)
+	e := NewEncoder(testFECPT, testFECSSRC, nil)
+	e.SetProtectionPercent(20)
+	p := makeEncoderFrame(t, 100, 1)[0]
+	first := e.Encode(&p.Header, p.Payload)
 	require.Len(t, first, 1)
-	require.Empty(t, send(4))
-	encoder.SetProtectionPercent(40)
-	require.Empty(t, send(4), "discard the old partial group")
-	repairs := send(1)
-	require.Len(t, repairs, 2)
+	p.Marker = false
+	for range 3 {
+		p.SequenceNumber++
+		require.Empty(t, e.Encode(&p.Header, p.Payload))
+	}
+	e.SetProtectionPercent(40)
+	p.SequenceNumber++
+	p.Marker = true
+	repairs := e.Encode(&p.Header, p.Payload)
+	require.Len(t, repairs, 1, "discard the old partial frame on a setting change")
 	require.Equal(t, first[0].SequenceNumber+1, repairs[0].SequenceNumber)
-	encoder.SetProtectionPercent(0)
-	require.Empty(t, send(100))
-	require.Zero(t, encoder.count)
-	encoder.SetProtectionPercent(10)
-	require.Empty(t, send(5))
-	encoder.SetProtectionPercent(10) // unchanged settings must retain fractional credit
-	last := send(5)
-	require.Len(t, last, 1)
-	require.Equal(t, repairs[1].SequenceNumber+1, last[0].SequenceNumber)
+	e.SetProtectionPercent(0)
+	require.Empty(t, e.Encode(&p.Header, p.Payload))
+	require.Zero(t, e.OverheadPercent())
+	e.SetProtectionPercent(40)
+	p.Marker = false
+	for range 4 {
+		p.SequenceNumber++
+		require.Empty(t, e.Encode(&p.Header, p.Payload))
+	}
+	e.SetProtectionPercent(40)
+	p.SequenceNumber++
+	p.Marker = true
+	last := e.Encode(&p.Header, p.Payload)
+	require.Len(t, last, 2, "unchanged settings retain the partial frame")
+	require.Equal(t, repairs[0].SequenceNumber+1, last[0].SequenceNumber)
 }
 
-func TestEncoderMultipleRepairRecovery(t *testing.T) {
-	encoder := NewEncoder(testFECPT, testFECSSRC, nil)
-	encoder.SetProtectionPercent(60)
-	media := makeMediaPackets(t, 100, MediaPacketsPerGroup)
-	decoder := newTestDecoder(testFECSSRC, testMediaSSRC, logger.GetLogger())
-	var repair []rtp.Packet
-	for i := range media {
-		repair = append(repair, encoder.Encode(&media[i].Header, media[i].Payload)...)
-		if i >= 3 {
-			decoder.DecodeFEC(&media[i])
-		}
-	}
-	require.Len(t, repair, 3)
-	for i := range repair {
-		recovered := decoder.DecodeFEC(&repair[i])
-		require.Len(t, recovered, 1)
-		requirePacketEqual(t, &media[i], recovered[0])
+func TestEncoderMeasuredOverhead(t *testing.T) {
+	var e *Encoder
+	callbacks := 0
+	e = NewEncoder(testFECPT, testFECSSRC, func(int, int) {
+		e.GetState() // callbacks must be outside the encoder lock
+		callbacks++
+	})
+	e.SetProtectionPercent(15)
+	require.EqualValues(t, 15, e.OverheadPercent())
+	p := rtp.Header{Version: 2, SSRC: testMediaSSRC, Marker: true}
+	payload := make([]byte, 100)
+	repairs := e.Encode(&p, payload)
+	require.Len(t, repairs, 1)
+	e.RecordSent(1, len(repairs[0].Payload), repairs[0].MarshalSize())
+	require.EqualValues(t, 118, e.OverheadPercent(), "132 repair bytes / 112 media bytes, rounded up")
+	p.Marker = false
+	for i := range 20 {
+		p.SequenceNumber++
+		p.Marker = i == 19
+		repairs = e.Encode(&p, payload)
 	}
+	require.Len(t, repairs, 3)
+	e.overheadUpdatedAt -= int64(overheadUpdateInterval)
+	e.RecordSent(3, 3*len(repairs[0].Payload), 3*repairs[0].MarshalSize())
+	require.EqualValues(t, 19, e.OverheadPercent(), "adapt when frames become larger, including the longer FEC masks")
+	require.Equal(t, 2, callbacks)
+	e.SetProtectionPercent(25)
+	require.EqualValues(t, 25, e.OverheadPercent(), "clear stale measurements on a preset change")
 }
 
-func TestEncoderConcurrentProtectionChanges(t *testing.T) {
-	encoder := NewEncoder(testFECPT, testFECSSRC, nil)
-	encoder.SetProtectionPercent(20)
+func TestEncoderConcurrentProtectionChangesAndClose(t *testing.T) {
+	e := NewEncoder(testFECPT, testFECSSRC, nil)
+	e.SetProtectionPercent(20)
 	var wg sync.WaitGroup
 	wg.Go(func() {
 		for i := range 1000 {
-			encoder.SetProtectionPercent(uint32(i % 101))
+			e.SetProtectionPercent(uint32(i % 101))
+			e.OverheadPercent()
 		}
 	})
 	wg.Go(func() {
-		p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT}
+		p := rtp.Header{Version: 2, SSRC: testMediaSSRC, Marker: true}
 		for range 1000 {
 			p.SequenceNumber++
-			encoder.Encode(&p, []byte{1, 2, 3})
+			repair := e.Encode(&p, []byte{1, 2, 3})
+			for _, packet := range repair {
+				e.RecordSent(1, len(packet.Payload), packet.MarshalSize())
+			}
 		}
 	})
+	wg.Go(e.Close)
 	wg.Wait()
-	encoder.Close()
+}
+
+func BenchmarkEncoder(b *testing.B) {
+	for _, packets := range []int{1, 5, 20, MaxMediaPacketsPerGroup} {
+		for _, percent := range []uint32{0, 15, 25, 35} {
+			b.Run(fmt.Sprintf("packets_%d/percent_%d", packets, percent), func(b *testing.B) {
+				e := NewEncoder(testFECPT, testFECSSRC, nil)
+				e.SetProtectionPercent(percent)
+				p := rtp.Header{Version: 2, SSRC: testMediaSSRC, PayloadType: testMediaPT}
+				_ = p.SetExtension(3, []byte{1, 2, 3})
+				payload := make([]byte, 1200)
+				send := func() {
+					p.Timestamp += 3000
+					for i := range packets {
+						p.SequenceNumber++
+						p.Marker = i == packets-1
+						e.Encode(&p, payload)
+					}
+				}
+				send()
+				b.ReportAllocs()
+				b.SetBytes(int64(packets * len(payload)))
+				b.ResetTimer()
+				for b.Loop() {
+					send()
+				}
+			})
+		}
+	}
 }

pkg/sfu/pacer/base.go

@@ -87,7 +87,7 @@ func (b *Base) SendPacket(p *Packet) (int, error) {
 
 	if p.FEC != nil && written > 0 && !p.IsRTX && !p.IsProbe {
 		repair := p.FEC.Encode(p.Header, p.Payload)
-		sent, payloadBytes := 0, 0
+		sent, payloadBytes, rtpBytes := 0, 0, 0
 		for i := range repair {
 			packet := &repair[i]
 			fec := PacketFactory.Get().(*Packet)
@@ -114,9 +114,10 @@ func (b *Base) SendPacket(p *Packet) (int, error) {
 			if fecErr == nil && n > 0 {
 				sent++
 				payloadBytes += len(packet.Payload)
+				rtpBytes += packet.MarshalSize()
 			}
 		}
-		p.FEC.RecordSent(sent, payloadBytes)
+		p.FEC.RecordSent(sent, payloadBytes, rtpBytes)
 	}
 	return written, nil
 }

pkg/sfu/pacer/base_fec_test.go

@@ -72,9 +72,9 @@ func TestPacerFECFinalHeadersAndAccounting(t *testing.T) {
 	encoder := flexfec.NewEncoder(115, 456, func(n int, bytes int) { sent += n; payloadBytes += bytes })
 	encoder.SetProtectionPercent(20)
 	total := 0
-	for i := range flexfec.MediaPacketsPerGroup {
+	for i := range 5 {
 		p := PacketFactory.Get().(*Packet)
-		h := &rtp.Header{Version: 2, SSRC: 123, PayloadType: 96, SequenceNumber: uint16(i), Timestamp: uint32(i * 3000)}
+		h := &rtp.Header{Version: 2, SSRC: 123, PayloadType: 96, SequenceNumber: uint16(i), Timestamp: 3000, Marker: i == 4}
 		_ = h.SetExtension(3, []byte{0, 0, 0})
 		_ = h.SetExtension(5, []byte{0, 0})
 		_ = h.SetExtension(7, []byte{0, 0})
@@ -122,7 +122,7 @@ func TestPacerFECSkipsRTXProbesAndFailedWrites(t *testing.T) {
 			e.SetProtectionPercent(20)
 			for i := range 10 {
 				p := PacketFactory.Get().(*Packet)
-				*p = Packet{Header: &rtp.Header{Version: 2, SSRC: 123, SequenceNumber: uint16(i)}, HeaderSize: 12, Payload: []byte{1}, FEC: e, WriteStream: w, IsRTX: mode == "rtx", IsProbe: mode == "probe"}
+				*p = Packet{Header: &rtp.Header{Version: 2, SSRC: 123, SequenceNumber: uint16(i), Marker: true}, HeaderSize: 12, Payload: []byte{1}, FEC: e, WriteStream: w, IsRTX: mode == "rtx", IsProbe: mode == "probe"}
 				n, err := b.SendPacket(p)
 				if w.failMedia {
 					require.ErrorIs(t, err, io.ErrClosedPipe)