sfu: cap retransmissions per down track (#5054)

* sfu: cap retransmissions per down track (rtc.rtx config, off by default)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* sfu: rtx budget review fixes (sequencer undo, sent bytes, newest first, rtx_budget)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* sfu: rtx budget second review (order unchanged when off, keep NACK window, metric in down track)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* sfu: rtx budget sorts NACKs newest first by extended sequence number

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

pkg/config/config.go

@@ -155,6 +155,8 @@ type RTCConfig struct {
 	// Throttle periods for pli/fir rtcp packets
 	PLIThrottle sfu.PLIThrottleConfig `yaml:"pli_throttle,omitempty"`
 
+	RTXBudget sfu.RTXBudgetConfig `yaml:"rtx_budget,omitempty"`
+
 	CongestionControl CongestionControlConfig `yaml:"congestion_control,omitempty"`
 
 	FlexFEC FlexFECConfig `yaml:"flexfec,omitempty"`
@@ -597,6 +599,7 @@ var DefaultConfig = Config{
 		PacketBufferSizeVideo:             500,
 		PacketBufferSizeAudio:             200,
 		PLIThrottle:                       sfu.DefaultPLIThrottleConfig,
+		RTXBudget:                         sfu.DefaultRTXBudgetConfig,
 		DatachannelDataTrackTargetLatency: 100 * time.Millisecond,
 		CongestionControl: CongestionControlConfig{
 			Enabled:                   true,

pkg/rtc/config.go

@@ -19,6 +19,7 @@ import (
 	"github.com/pion/webrtc/v4"
 
 	"github.com/livekit/livekit-server/pkg/config"
+	"github.com/livekit/livekit-server/pkg/sfu"
 	"github.com/livekit/livekit-server/pkg/sfu/buffer"
 	act "github.com/livekit/livekit-server/pkg/sfu/rtpextension/abscapturetime"
 	dd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/dependencydescriptor"
@@ -60,6 +61,7 @@ type DirectionConfig struct {
 	RTPHeaderExtension RTPHeaderExtensionConfig
 	RTCPFeedback       RTCPFeedbackConfig
 	FlexFEC            FlexFECDirectionConfig
+	RTXBudget          sfu.RTXBudgetConfig
 }
 
 // FlexFECDirectionConfig enables flexfec-03 for a transport direction.
@@ -93,6 +95,9 @@ func NewWebRTCConfig(conf *config.Config) (*WebRTCConfig, error) {
 		rtcConf.PacketBufferSizeAudio = rtcConf.PacketBufferSize
 	}
 
+	subscriberConfig := getSubscriberConfig(rtcConf.CongestionControl.UseSendSideBWEInterceptor || rtcConf.CongestionControl.UseSendSideBWE)
+	subscriberConfig.RTXBudget = rtcConf.RTXBudget
+
 	return &WebRTCConfig{
 		WebRTCConfig: *webRTCConfig,
 		Receiver: ReceiverConfig{
@@ -100,7 +105,7 @@ func NewWebRTCConfig(conf *config.Config) (*WebRTCConfig, error) {
 			PacketBufferSizeAudio: rtcConf.PacketBufferSizeAudio,
 		},
 		Publisher:  getPublisherConfig(false, flexFEC),
-		Subscriber: getSubscriberConfig(rtcConf.CongestionControl.UseSendSideBWEInterceptor || rtcConf.CongestionControl.UseSendSideBWE),
+		Subscriber: subscriberConfig,
 		flexFEC:    flexFEC,
 	}, nil
 }
@@ -110,7 +115,9 @@ func (c *WebRTCConfig) UpdatePublisherConfig(consolidated bool) {
 }
 
 func (c *WebRTCConfig) UpdateSubscriberConfig(ccConf config.CongestionControlConfig) {
+	rtxBudget := c.Subscriber.RTXBudget
 	c.Subscriber = getSubscriberConfig(ccConf.UseSendSideBWEInterceptor || ccConf.UseSendSideBWE)
+	c.Subscriber.RTXBudget = rtxBudget
 }
 
 func (c *WebRTCConfig) SetBufferFactory(factory *buffer.Factory) {

pkg/rtc/subscribedtrack.go

@@ -157,6 +157,7 @@ func NewSubscribedTrack(params SubscribedTrackParams) (*SubscribedTrack, error)
 		DisableSenderReportPassThrough: params.Subscriber.GetDisableSenderReportPassThrough(),
 		SupportsCodecChange:            params.Subscriber.SupportsCodecChange(),
 		EnableStartAtDesiredQuality:    params.EnableStartAtDesiredQuality,
+		RTXBudget:                      params.SubscriberConfig.RTXBudget,
 		Listener:                       s,
 	})
 	if err != nil {

pkg/sfu/downtrack.go

@@ -15,12 +15,14 @@
 package sfu
 
 import (
+	"cmp"
 	"encoding/binary"
 	"errors"
 	"fmt"
 	"io"
 	"math"
 	"math/rand"
+	"slices"
 	"strings"
 	"sync"
 	"time"
@@ -49,6 +51,7 @@ import (
 	pd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/playoutdelay"
 	"github.com/livekit/livekit-server/pkg/sfu/rtpstats"
 	"github.com/livekit/livekit-server/pkg/sfu/utils"
+	"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
 )
 
 // -------------------------------------------------------------------
@@ -246,6 +249,7 @@ type DownTrackParams struct {
 	SupportsCodecChange            bool
 	StripPacketTrailer             bool
 	EnableStartAtDesiredQuality    bool
+	RTXBudget                      RTXBudgetConfig
 	Listener                       DownTrackListener
 }
 
@@ -267,6 +271,7 @@ type DownTrack struct {
 	payloadTypeRTX    atomic.Uint32
 	sequencer         *sequencer
 	rtxSequenceNumber atomic.Uint64
+	rtxBudget         *rtxBudget
 
 	receiverLock sync.RWMutex
 	receiver     TrackReceiver
@@ -379,6 +384,7 @@ func NewDownTrack(params DownTrackParams) (*DownTrack, error) {
 		retransmitCh:        make(chan struct{}, 1),
 		createdAt:           time.Now().UnixNano(),
 		receiver:            params.Receiver,
+		rtxBudget:           newRTXBudget(params.RTXBudget.forKind(kind)),
 	}
 	d.lastUnmutedAt.Store(time.Now())
 
@@ -1184,6 +1190,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) int32 {
 
 	headerSize := hdr.MarshalSize()
 	pacerPacket.HeaderSize = headerSize
+	d.rtxBudget.onMediaSent(headerSize + len(payload))
 	d.rtpStats.Update(
 		extPkt.Arrival,
 		tp.rtp.extSequenceNumber,
@@ -2471,13 +2478,28 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
 	nackAcks := uint32(0)
 	nackMisses := uint32(0)
 	numRepeatedNACKs := uint32(0)
+	var capped []extPacketMeta
 	epms := d.sequencer.getExtPacketMetas(filtered)
+	if d.rtxBudget != nil {
+		// newest first: when the budget runs short, the oldest packets are the least likely
+		// to still arrive in time for the receiver's jitter buffer. A batch can hold several
+		// NACK messages, so sort instead of relying on their order.
+		slices.SortFunc(epms, func(a, b extPacketMeta) int {
+			return cmp.Compare(b.extSequenceNumber, a.extSequenceNumber)
+		})
+	}
 	for i := range epms {
 		epm := &epms[i]
 		if disallowedLayers[epm.layer] {
 			continue
 		}
 
+		// over budget: skip without reading the publisher's buffer
+		if len(capped) != 0 || !d.rtxBudget.allow() {
+			capped = append(capped, *epm)
+			continue
+		}
+
 		nackAcks++
 
 		pktBuff := *src
@@ -2494,7 +2516,15 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
 			numRepeatedNACKs++
 		}
 
-		d.retransmitPacket(epm, pktBuff[:n], false)
+		if sent, err := d.retransmitPacket(epm, pktBuff[:n], false); err == nil {
+			d.rtxBudget.spend(sent)
+		}
+	}
+	if len(capped) != 0 {
+		// a capped packet was not sent, so it does not count toward the sequencer's NACK limit
+		d.sequencer.unmarkNacked(capped)
+		d.rtxBudget.numCapped.Add(uint64(len(capped)))
+		prometheus.RecordRTXCapped(d.kind.String(), uint32(len(capped)))
 	}
 
 	d.totalRepeatedNACKs.Add(numRepeatedNACKs)

pkg/sfu/downtrack_downstream_integration_test.go

@@ -495,6 +495,161 @@ func TestDownTrackRetransmitsPacketsAsIs(t *testing.T) {
 	}
 }
 
+// TestDownTrackCapsRetransmissions forwards media, NACKs all of it, and checks that an
+// uncapped down track retransmits everything while a capped one retransmits only what its
+// budget allows, newest first, without reading the capped packets from the buffer. Capped
+// packets NACKed again keep failing while the budget is empty, past maxAck, and are
+// retransmitted once media refills it. Repeat NACKs are spaced past the sequencer's
+// duplicate window.
+func TestDownTrackCapsRetransmissions(t *testing.T) {
+	const numPackets = 20
+
+	setup := func(t *testing.T) (*downTrackHarness, *capturingPacer, []byte) {
+		h := vnettest.NewHosts(t)
+		factory := buffer.NewFactoryOfBufferFactory(500, 500).CreateBufferFactory()
+		cp := &capturingPacer{inner: pacer.NewPassThrough(logger.GetLogger(), newNullBWE())}
+		t.Cleanup(cp.Stop)
+
+		// every NACKed packet is read back from the receiver as this source packet
+		source, err := (&rtp.Packet{
+			Header:  rtp.Header{Version: 2, PayloadType: 111, SSRC: 0x44444444},
+			Payload: distinctivePayload(7, 600),
+		}).Marshal()
+		require.NoError(t, err)
+		dh := newBoundDownTrack(t, h, factory, opusCodecParams, cp, vnettest.MediaEngineConfig{}, func(r *sfufakes.FakeTrackReceiver) {
+			r.ReadRTPCalls(func(buf []byte, _ uint8, _ uint64) (int, error) {
+				return copy(buf, source), nil
+			})
+		})
+		return dh, cp, source
+	}
+
+	// extended, so the sequence number can wrap
+	nextSN := uint64(30000)
+	nextTS := uint32(1000)
+	writeMedia := func(t *testing.T, dh *downTrackHarness, payloadSize int) []uint16 {
+		var sns []uint16
+		for i := 0; i < numPackets; i++ {
+			nextTS += 960
+			ep, err := testutils.GetTestExtPacket(&testutils.TestExtPacketParams{
+				SequenceNumber: uint16(nextSN),
+				SNCycles:       int(nextSN >> 16),
+				Timestamp:      nextTS,
+				SSRC:           0x44444444,
+				PayloadType:    111,
+				PayloadSize:    payloadSize,
+			})
+			require.NoError(t, err)
+			require.Equal(t, int32(1), dh.dt.WriteRTP(ep, 0), "packet %d should be forwarded", nextSN)
+			sns = append(sns, uint16(nextSN))
+			nextSN++
+		}
+		return sns
+	}
+	sendNACK := func(t *testing.T, dh *downTrackHarness, sns []uint16) {
+		nack := rtcp.TransportLayerNack{MediaSSRC: dh.dt.SSRC()}
+		for _, sn := range sns {
+			nack.Nacks = append(nack.Nacks, rtcp.NackPair{PacketID: sn})
+		}
+		b, err := nack.Marshal()
+		require.NoError(t, err)
+		dh.dt.HandleRTCPForTest(b)
+	}
+	rtxSeqs := func(cp *capturingPacer) []uint16 {
+		var sns []uint16
+		for _, p := range cp.rtxPackets() {
+			sns = append(sns, p.hdr.SequenceNumber)
+		}
+		return sns
+	}
+
+	t.Run("uncapped", func(t *testing.T) {
+		dh, cp, _ := setup(t)
+		sns := writeMedia(t, dh, 40)
+		sendNACK(t, dh, sns)
+
+		require.Eventually(t, func() bool {
+			return len(cp.rtxPackets()) == numPackets
+		}, 5*time.Second, 20*time.Millisecond, "expected every packet to be retransmitted")
+		require.Equal(t, sns, rtxSeqs(cp), "without a budget the order is unchanged")
+	})
+
+	t.Run("capped", func(t *testing.T) {
+		dh, cp, source := setup(t)
+		// two and a half retransmissions: the third overdraws the budget, a fourth is refused
+		dh.dt.SetRTXBudgetForTest(RTXBudgetLimit{MaxRatio: 0.5, BurstBytes: 5 * len(source) / 2})
+
+		sns := writeMedia(t, dh, 40)
+		sendNACK(t, dh, sns)
+
+		// the capped count is added after the batch, so the retransmissions are all enqueued by then
+		require.Eventually(t, func() bool {
+			return dh.dt.RTXCappedForTest() == numPackets-3
+		}, 5*time.Second, 20*time.Millisecond, "expected the batch to be capped")
+		require.ElementsMatch(t, sns[numPackets-3:], rtxSeqs(cp), "the newest packets go first")
+		require.Equal(t, 3, dh.receiver.ReadRTPCallCount(), "a capped retransmission must not read the publisher's buffer")
+
+		// with the budget empty, the capped packets fail again on every NACK, past maxAck
+		capped := sns[:numPackets-3]
+		for i := 0; i < 4; i++ {
+			time.Sleep(NACKIgnoreWindowForTest + 20*time.Millisecond)
+			sendNACK(t, dh, capped)
+			want := uint64((i + 2) * len(capped))
+			require.Eventually(t, func() bool {
+				return dh.dt.RTXCappedForTest() == want
+			}, 5*time.Second, 20*time.Millisecond, "expected NACK %d to be capped", i+2)
+		}
+
+		// media refills the budget, and the earlier capped packets are retransmitted
+		writeMedia(t, dh, 600)
+		time.Sleep(NACKIgnoreWindowForTest + 20*time.Millisecond)
+		sendNACK(t, dh, capped)
+		require.Eventually(t, func() bool {
+			return len(cp.rtxPackets()) == 6
+		}, 5*time.Second, 20*time.Millisecond, "expected capped packets to be retransmitted after the refill")
+		require.ElementsMatch(t, append(sns[numPackets-3:], capped[len(capped)-3:]...), rtxSeqs(cp))
+	})
+
+	// the budget floor of 1500 bytes sends three 612-byte packets: they must be the three
+	// newest of the NACK, whatever its order
+	newestOf := func(t *testing.T, start uint64, nackIndexes []int, newestIndexes []int) {
+		dh, cp, _ := setup(t)
+		dh.dt.SetRTXBudgetForTest(RTXBudgetLimit{MaxRatio: 0.5})
+
+		nextSN = start
+		sns := writeMedia(t, dh, 40)
+		require.Eventually(t, func() bool {
+			return dh.capture.count() >= numPackets
+		}, 10*time.Second, 20*time.Millisecond, "expected the media to arrive")
+		// past the window the sequencer keeps after a packet is sent
+		time.Sleep(NACKIgnoreWindowForTest + 20*time.Millisecond)
+
+		var nacked []uint16
+		for _, i := range nackIndexes {
+			nacked = append(nacked, sns[i])
+		}
+		sendNACK(t, dh, nacked)
+
+		var want []uint16
+		for _, i := range newestIndexes {
+			want = append(want, sns[i])
+		}
+		require.Eventually(t, func() bool {
+			return dh.dt.RTXCappedForTest() == uint64(len(nacked)-len(want))
+		}, 5*time.Second, 20*time.Millisecond, "expected all but the newest to be capped")
+		require.Equal(t, want, rtxSeqs(cp))
+	}
+
+	t.Run("newest first for a NACK in mixed order", func(t *testing.T) {
+		newestOf(t, 31000, []int{10, 19, 0, 15, 5}, []int{19, 15, 10})
+	})
+
+	t.Run("newest first across the sequence number wrap", func(t *testing.T) {
+		// packets 65530..65535 then 0..13: three before the wrap and three after it, out of order
+		newestOf(t, 65530, []int{4, 8, 1, 6, 9, 2}, []int{9, 8, 6})
+	})
+}
+
 // TestDownTrackRetransmitsPacketsViaRTX exercises the retransmit path when an RTX
 // stream is negotiated. Each retransmitted packet must be emitted on the RTX SSRC
 // with the RTX payload type, carry the original media sequence number as the OSN

pkg/sfu/export_test.go

@@ -84,6 +84,19 @@ func (d *DownTrack) RetransmitForTest(
 	return d.retransmitPacket(&epm, sourcePkt, false)
 }
 
+// NACKIgnoreWindowForTest is the longest time the sequencer ignores a repeat NACK for a packet.
+const NACKIgnoreWindowForTest = ignoreRetransmission * time.Millisecond
+
+// RTXCappedForTest returns the NACKed packets the budget has refused.
+func (d *DownTrack) RTXCappedForTest() uint64 {
+	return d.rtxBudget.numCapped.Load()
+}
+
+// SetRTXBudgetForTest replaces the down track's retransmission budget.
+func (d *DownTrack) SetRTXBudgetForTest(config RTXBudgetLimit) {
+	d.rtxBudget = newRTXBudget(config)
+}
+
 const KeyFrameIntervalMaxForTest = keyFrameIntervalMax * time.Millisecond
 
 func (d *DownTrack) KeyFrameRequesterWakeupsForTest() uint32 {

pkg/sfu/rtxbudget.go

@@ -0,0 +1,113 @@
+// 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 sfu
+
+import (
+	"sync"
+
+	"github.com/pion/webrtc/v4"
+	"go.uber.org/atomic"
+)
+
+// RTXBudgetConfig caps the bytes a down track retransmits in response to NACKs.
+type RTXBudgetConfig struct {
+	Audio RTXBudgetLimit `yaml:"audio,omitempty"`
+	Video RTXBudgetLimit `yaml:"video,omitempty"`
+}
+
+type RTXBudgetLimit struct {
+	// retransmitted bytes allowed, as a fraction of the media bytes sent; 0 turns the cap off
+	MaxRatio float64 `yaml:"max_ratio,omitempty"`
+	// retransmitted bytes allowed at once, above the steady ratio; at least one full packet
+	BurstBytes int `yaml:"burst_bytes,omitempty"`
+}
+
+// the budget never holds less than one full packet
+const minRTXBurstBytes = 1500
+
+// The bursts cover about 250 ms of total loss: a 2 Mbps video stream and a 128 kbps audio stream.
+var DefaultRTXBudgetConfig = RTXBudgetConfig{
+	Audio: RTXBudgetLimit{BurstBytes: 4_000},
+	Video: RTXBudgetLimit{BurstBytes: 64_000},
+}
+
+func (c RTXBudgetConfig) forKind(kind webrtc.RTPCodecType) RTXBudgetLimit {
+	if kind == webrtc.RTPCodecTypeAudio {
+		return c.Audio
+	}
+	return c.Video
+}
+
+// rtxBudget is a token bucket filled by media bytes sent and drained by retransmitted bytes.
+// A retransmission is allowed while the bucket is above zero and may overdraw it, so the
+// check needs no packet size and runs before the packet is read from the publisher's buffer.
+// A nil budget allows every retransmission.
+type rtxBudget struct {
+	ratio float64
+	burst float64
+
+	mediaBytes atomic.Uint64
+
+	lock           sync.Mutex
+	tokens         float64
+	mediaBytesSeen uint64
+
+	// NACKed packets not retransmitted because the budget was empty
+	numCapped atomic.Uint64
+}
+
+func newRTXBudget(config RTXBudgetLimit) *rtxBudget {
+	if config.MaxRatio <= 0 {
+		return nil
+	}
+	burst := float64(max(config.BurstBytes, minRTXBurstBytes))
+	return &rtxBudget{
+		ratio:  config.MaxRatio,
+		burst:  burst,
+		tokens: burst,
+	}
+}
+
+func (b *rtxBudget) onMediaSent(size int) {
+	if b == nil {
+		return
+	}
+	b.mediaBytes.Add(uint64(size))
+}
+
+// allow reports whether the budget has room for another retransmission.
+func (b *rtxBudget) allow() bool {
+	if b == nil {
+		return true
+	}
+
+	b.lock.Lock()
+	defer b.lock.Unlock()
+
+	mediaBytes := b.mediaBytes.Load()
+	b.tokens = min(b.burst, b.tokens+b.ratio*float64(mediaBytes-b.mediaBytesSeen))
+	b.mediaBytesSeen = mediaBytes
+	return b.tokens > 0
+}
+
+func (b *rtxBudget) spend(size int) {
+	if b == nil {
+		return
+	}
+
+	b.lock.Lock()
+	b.tokens -= float64(size)
+	b.lock.Unlock()
+}

pkg/sfu/rtxbudget_test.go

@@ -0,0 +1,76 @@
+// 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 sfu
+
+import (
+	"testing"
+
+	"github.com/pion/webrtc/v4"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRTXBudget(t *testing.T) {
+	t.Run("no cap without a ratio", func(t *testing.T) {
+		b := newRTXBudget(RTXBudgetLimit{BurstBytes: 100})
+		require.Nil(t, b)
+		require.True(t, b.allow())
+		b.spend(1_000_000)
+		b.onMediaSent(10)
+		require.True(t, b.allow())
+	})
+
+	t.Run("starts with the burst", func(t *testing.T) {
+		b := newRTXBudget(RTXBudgetLimit{MaxRatio: 0.3, BurstBytes: 3000})
+		require.True(t, b.allow())
+		b.spend(2999)
+		require.True(t, b.allow())
+		b.spend(1)
+		require.False(t, b.allow())
+	})
+
+	t.Run("burst holds at least one full packet", func(t *testing.T) {
+		b := newRTXBudget(RTXBudgetLimit{MaxRatio: 0.3})
+		b.spend(minRTXBurstBytes - 1)
+		require.True(t, b.allow())
+	})
+
+	t.Run("an overdraw is paid back from media sent at the ratio", func(t *testing.T) {
+		b := newRTXBudget(RTXBudgetLimit{MaxRatio: 0.5, BurstBytes: 2000})
+		b.spend(2900) // overdrawn by 900
+		require.False(t, b.allow())
+
+		b.onMediaSent(1800) // pays back exactly 900
+		require.False(t, b.allow())
+		b.onMediaSent(2)
+		require.True(t, b.allow())
+	})
+
+	t.Run("refill stops at the burst", func(t *testing.T) {
+		b := newRTXBudget(RTXBudgetLimit{MaxRatio: 0.5, BurstBytes: 2000})
+		b.onMediaSent(1_000_000)
+		require.True(t, b.allow()) // refills, but only up to the burst
+		b.spend(2000)
+		require.False(t, b.allow())
+	})
+}
+
+func TestRTXBudgetConfigForKind(t *testing.T) {
+	c := RTXBudgetConfig{
+		Audio: RTXBudgetLimit{MaxRatio: 0.1},
+		Video: RTXBudgetLimit{MaxRatio: 0.3},
+	}
+	require.Equal(t, 0.1, c.forKind(webrtc.RTPCodecTypeAudio).MaxRatio)
+	require.Equal(t, 0.3, c.forKind(webrtc.RTPCodecTypeVideo).MaxRatio)
+}

pkg/sfu/sequencer.go

@@ -101,6 +101,7 @@ type extPacketMeta struct {
 	packetMeta
 	extSequenceNumber uint64
 	extTimestamp      uint64
+	slot              int
 }
 
 func (epm extPacketMeta) MarshalLogObject(e zapcore.ObjectEncoder) error {
@@ -397,6 +398,7 @@ func (s *sequencer) getExtPacketMetas(seqNo []uint16) []extPacketMeta {
 				packetMeta:        *meta,
 				extSequenceNumber: extSN,
 				extTimestamp:      extTS,
+				slot:              int(slot),
 			}
 			epm.codecBytesSlice = append([]byte{}, meta.codecBytesSlice...)
 			epm.ddBytesSlice = append([]byte{}, meta.ddBytesSlice...)
@@ -407,6 +409,23 @@ func (s *sequencer) getExtPacketMetas(seqNo []uint16) []extPacketMeta {
 	return extPacketMetas
 }
 
+// unmarkNacked reverts the NACK count getExtPacketMetas added for packets that were not
+// retransmitted, so they do not run into maxAck. lastNack stays, so a repeat NACK inside the
+// duplicate window is still ignored.
+func (s *sequencer) unmarkNacked(epms []extPacketMeta) {
+	s.Lock()
+	defer s.Unlock()
+
+	for i := range epms {
+		meta := &s.meta[epms[i].slot]
+		// skip if the slot holds a newer packet, or a newer NACK marked it again
+		if meta.targetSeqNo != epms[i].targetSeqNo || meta.lastNack != epms[i].lastNack || meta.nacked == 0 {
+			continue
+		}
+		meta.nacked--
+	}
+}
+
 func (s *sequencer) lookupExtPacketMeta(extSN uint64) *extPacketMeta {
 	s.Lock()
 	defer s.Unlock()

pkg/sfu/sequencer_test.go

@@ -78,6 +78,46 @@ func Test_sequencer(t *testing.T) {
 	require.Equal(t, 1, len(m))
 }
 
+func Test_sequencer_unmarkNacked(t *testing.T) {
+	newPushed := func(n uint64) *sequencer {
+		seq := newSequencer(100, false, logger.GetLogger())
+		for i := uint64(1); i <= n; i++ {
+			seq.push(time.Now().UnixNano(), i, i, 123, true, 0, nil, 0, nil, nil, 0)
+		}
+		time.Sleep((ignoreRetransmission + 10) * time.Millisecond)
+		return seq
+	}
+
+	t.Run("an unmarked packet is served again past maxAck, but not inside the duplicate window", func(t *testing.T) {
+		seq := newPushed(50)
+		for i := 0; i < maxAck+2; i++ {
+			res := seq.getExtPacketMetas([]uint16{40})
+			require.Len(t, res, 1, "NACK %d", i)
+			seq.unmarkNacked(res)
+
+			require.Empty(t, seq.getExtPacketMetas([]uint16{40}), "a repeat NACK inside the window is ignored")
+			time.Sleep((ignoreRetransmission + 10) * time.Millisecond)
+		}
+	})
+
+	t.Run("a reused slot is left alone", func(t *testing.T) {
+		seq := newPushed(50)
+		res := seq.getExtPacketMetas([]uint16{40})
+		require.Len(t, res, 1)
+
+		// wrap the ring so slot of 40 holds a newer packet that was NACKed once
+		for i := uint64(51); i <= 140; i++ {
+			seq.push(time.Now().UnixNano(), i, i, 123, true, 0, nil, 0, nil, nil, 0)
+		}
+		time.Sleep((ignoreRetransmission + 10) * time.Millisecond)
+		require.Len(t, seq.getExtPacketMetas([]uint16{140}), 1)
+		require.EqualValues(t, 140, seq.meta[res[0].slot].targetSeqNo)
+
+		seq.unmarkNacked(res)
+		require.EqualValues(t, 1, seq.meta[res[0].slot].nacked)
+	})
+}
+
 func Test_sequencer_flush(t *testing.T) {
 	seq := newSequencer(500, false, logger.GetLogger())
 	off := uint16(15)

pkg/telemetry/prometheus/packets.go

@@ -61,6 +61,7 @@ var (
 	promRTCPLabels            = []string{"direction", "country"}
 	promStreamLabels          = []string{"direction", "source", "type", "country"}
 	promNackTotal             *prometheus.CounterVec
+	promRTXCappedTotal        *prometheus.CounterVec
 	promPliTotal              *prometheus.CounterVec
 	promFirTotal              *prometheus.CounterVec
 	promPacketLossTotal       *prometheus.CounterVec
@@ -106,6 +107,13 @@ func initPacketStats(nodeID string, nodeType livekit.NodeType) {
 		Name:        "total",
 		ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()},
 	}, promRTCPLabels)
+	promRTXCappedTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace:   livekitNamespace,
+		Subsystem:   "rtx_capped",
+		Name:        "total",
+		Help:        "NACKed packets not retransmitted because the down track was over its retransmission budget.",
+		ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()},
+	}, []string{"kind"})
 	promPliTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
 		Namespace:   livekitNamespace,
 		Subsystem:   "pli",
@@ -202,6 +210,7 @@ func initPacketStats(nodeID string, nodeType livekit.NodeType) {
 	prometheus.MustRegister(promFECPacketTotal)
 	prometheus.MustRegister(promFECPacketBytes)
 	prometheus.MustRegister(promNackTotal)
+	prometheus.MustRegister(promRTXCappedTotal)
 	prometheus.MustRegister(promPliTotal)
 	prometheus.MustRegister(promFirTotal)
 	prometheus.MustRegister(promPacketLossTotal)
@@ -272,6 +281,13 @@ func IncrementBytes(country string, direction Direction, count uint64, retransmi
 	}
 }
 
+func RecordRTXCapped(kind string, packets uint32) {
+	if promRTXCappedTotal == nil {
+		return
+	}
+	promRTXCappedTotal.WithLabelValues(kind).Add(float64(packets))
+}
+
 func IncrementRTCP(country string, direction Direction, nack, pli, fir uint32) {
 	if nack > 0 {
 		promNackTotal.WithLabelValues(string(direction), country).Add(float64(nack))