diff --git a/pkg/config/config.go b/pkg/config/config.go index bbc64ca43..19a42518b 100644 --- a/pkg/config/config.go +++ b/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, diff --git a/pkg/rtc/config.go b/pkg/rtc/config.go index bcdd664bc..ae3cb0c41 100644 --- a/pkg/rtc/config.go +++ b/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) { diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index 1771767d6..fcaac3deb 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/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 { diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index efb10a3d0..2e3598356 100644 --- a/pkg/sfu/downtrack.go +++ b/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) diff --git a/pkg/sfu/downtrack_downstream_integration_test.go b/pkg/sfu/downtrack_downstream_integration_test.go index e16ecd765..36a798fe1 100644 --- a/pkg/sfu/downtrack_downstream_integration_test.go +++ b/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 diff --git a/pkg/sfu/export_test.go b/pkg/sfu/export_test.go index 8baa42ade..dabceed75 100644 --- a/pkg/sfu/export_test.go +++ b/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 { diff --git a/pkg/sfu/rtxbudget.go b/pkg/sfu/rtxbudget.go new file mode 100644 index 000000000..3b540f556 --- /dev/null +++ b/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() +} diff --git a/pkg/sfu/rtxbudget_test.go b/pkg/sfu/rtxbudget_test.go new file mode 100644 index 000000000..2e8163663 --- /dev/null +++ b/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) +} diff --git a/pkg/sfu/sequencer.go b/pkg/sfu/sequencer.go index 091189d0c..258e36bed 100644 --- a/pkg/sfu/sequencer.go +++ b/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() diff --git a/pkg/sfu/sequencer_test.go b/pkg/sfu/sequencer_test.go index dd314a870..e12a70444 100644 --- a/pkg/sfu/sequencer_test.go +++ b/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) diff --git a/pkg/telemetry/prometheus/packets.go b/pkg/telemetry/prometheus/packets.go index ed8646e16..fad8ce769 100644 --- a/pkg/telemetry/prometheus/packets.go +++ b/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))