From 57b2a12415834541b1294ba4cf8e405c38e9296b Mon Sep 17 00:00:00 2001 From: 5ur3 Date: Mon, 6 Jul 2026 21:08:25 +0300 Subject: [PATCH 1/3] Fix rfc8888 implementation --- pkg/rfc8888/recorder.go | 2 +- pkg/rfc8888/recorder_test.go | 22 +++++++++++++++++++++ pkg/rfc8888/stream_log.go | 13 ++++++++++++- pkg/rfc8888/stream_log_test.go | 35 ++++++++++++++++++++++++++++++++++ 4 files changed, 70 insertions(+), 2 deletions(-) diff --git a/pkg/rfc8888/recorder.go b/pkg/rfc8888/recorder.go index 35a65cfc..1a735459 100644 --- a/pkg/rfc8888/recorder.go +++ b/pkg/rfc8888/recorder.go @@ -48,7 +48,7 @@ func (r *Recorder) BuildReport(now time.Time, maxSize int) *rtcp.CCFeedbackRepor ReportTimestamp: ntp.ToNTP32(now), } - maxReportBlocks := (maxSize - 12 - (8 * len(r.streams))) / 2 + maxReportBlocks := max((maxSize-12-(8*len(r.streams)))/2, 0) maxReportBlocksPerStream := maxReportBlocks / len(r.streams) for _, log := range r.streams { diff --git a/pkg/rfc8888/recorder_test.go b/pkg/rfc8888/recorder_test.go index 1f4bbb6d..7eb9fbc5 100644 --- a/pkg/rfc8888/recorder_test.go +++ b/pkg/rfc8888/recorder_test.go @@ -188,4 +188,26 @@ func TestRecorder(t *testing.T) { } assert.Less(t, blocks*2, maxSize) }) + + t.Run("non-positive per-stream budget does not panic", func(t *testing.T) { + recorder := NewRecorder() + now := time.Time{} + maxSize := 1200 + + streams := 250 + for i := 0; i < streams; i++ { + ssrc := uint32(i + 1) + recorder.AddPacket(now, ssrc, 0, 0) + recorder.AddPacket(now, ssrc, 1, 0) + } + + var report *rtcp.CCFeedbackReport + assert.NotPanics(t, func() { + report = recorder.BuildReport(now, maxSize) + }) + assert.Len(t, report.ReportBlocks, streams) + for _, block := range report.ReportBlocks { + assert.Empty(t, block.MetricBlocks) + } + }) } diff --git a/pkg/rfc8888/stream_log.go b/pkg/rfc8888/stream_log.go index d95deccd..5563c60e 100644 --- a/pkg/rfc8888/stream_log.go +++ b/pkg/rfc8888/stream_log.go @@ -38,6 +38,10 @@ func (l *streamLog) add(ts time.Time, sequenceNumber uint16, ecn uint8) { l.init = true l.nextSequenceNumberToReport = unwrappedSequenceNumber } + // Drop late/duplicate packets below the report pointer: metricsAfter never reads below it, so they would leak. + if unwrappedSequenceNumber < l.nextSequenceNumberToReport { + return + } l.log[unwrappedSequenceNumber] = &packetReport{ arrivalTime: ts, ecn: ecn, @@ -60,7 +64,14 @@ func (l *streamLog) metricsAfter(reference time.Time, maxReportBlocks int64) rtc numReports := l.lastSequenceNumberReceived - l.nextSequenceNumberToReport + 1 if numReports > maxReportBlocks { numReports = maxReportBlocks - l.nextSequenceNumberToReport = l.lastSequenceNumberReceived - maxReportBlocks + 1 + newNext := l.lastSequenceNumberReceived - maxReportBlocks + 1 + // Reclaim entries we advance past: metricsAfter never reads below the pointer, so they would leak. + for seq := range l.log { + if seq < newNext { + delete(l.log, seq) + } + } + l.nextSequenceNumberToReport = newNext } metricBlocks := make([]rtcp.CCFeedbackMetricBlock, numReports) offset := l.nextSequenceNumberToReport diff --git a/pkg/rfc8888/stream_log_test.go b/pkg/rfc8888/stream_log_test.go index 2f28bc3d..a9f27ef9 100644 --- a/pkg/rfc8888/stream_log_test.go +++ b/pkg/rfc8888/stream_log_test.go @@ -582,3 +582,38 @@ func TestRemoveOldestPackets(t *testing.T) { assert.Equal(t, uint16(2), metrics.BeginSequence) assert.Lenf(t, metrics.MetricBlocks, 16384, "%v != %v", len(metrics.MetricBlocks), 16384) } + +func TestStreamLogDoesNotLeakBelowReportPointer(t *testing.T) { + const maxReportBlocks = 4 + base := time.Time{} + + t.Run("cap on throughput", func(t *testing.T) { + sl := newStreamLog(0) + for i := 0; i < 2*maxReportBlocks; i++ { + sl.add(base, uint16(i), 0) //nolint:gosec // G115 + } + sl.metricsAfter(base.Add(time.Second), maxReportBlocks) + assert.Empty(t, sl.log) + }) + + t.Run("cap grown by packet loss", func(t *testing.T) { + sl := newStreamLog(0) + sl.add(base, 0, 0) + for i := 2; i < 2*maxReportBlocks; i++ { // seq 1 is never received + sl.add(base, uint16(i), 0) //nolint:gosec // G115 + } + sl.metricsAfter(base.Add(time.Second), maxReportBlocks) + assert.Empty(t, sl.log) + }) + + t.Run("late packet below the pointer", func(t *testing.T) { + sl := newStreamLog(0) + sl.add(base, 0, 0) + sl.add(base, 1, 0) + sl.add(base, 2, 0) + sl.metricsAfter(base.Add(time.Second), maxReportBlocks) + + sl.add(base, 1, 0) // duplicate of an already-reported sequence + assert.Empty(t, sl.log) + }) +} From 85e3a91692a0127a905ae093255d787a80360db6 Mon Sep 17 00:00:00 2001 From: 5ur3 Date: Mon, 6 Jul 2026 21:21:54 +0300 Subject: [PATCH 2/3] Fix linter issues --- pkg/rfc8888/recorder_test.go | 2 +- pkg/rfc8888/stream_log.go | 2 ++ pkg/rfc8888/stream_log_test.go | 2 +- 3 files changed, 4 insertions(+), 2 deletions(-) diff --git a/pkg/rfc8888/recorder_test.go b/pkg/rfc8888/recorder_test.go index 7eb9fbc5..e55d3da7 100644 --- a/pkg/rfc8888/recorder_test.go +++ b/pkg/rfc8888/recorder_test.go @@ -195,7 +195,7 @@ func TestRecorder(t *testing.T) { maxSize := 1200 streams := 250 - for i := 0; i < streams; i++ { + for i := range streams { ssrc := uint32(i + 1) recorder.AddPacket(now, ssrc, 0, 0) recorder.AddPacket(now, ssrc, 1, 0) diff --git a/pkg/rfc8888/stream_log.go b/pkg/rfc8888/stream_log.go index 5563c60e..d6b22973 100644 --- a/pkg/rfc8888/stream_log.go +++ b/pkg/rfc8888/stream_log.go @@ -53,6 +53,8 @@ func (l *streamLog) add(ts time.Time, sequenceNumber uint16, ecn uint8) { // metricsAfter iterates over all packets order of their sequence number. // Packets are removed until the first loss is detected. +// +//nolint:cyclop func (l *streamLog) metricsAfter(reference time.Time, maxReportBlocks int64) rtcp.CCFeedbackReportBlock { if len(l.log) == 0 { return rtcp.CCFeedbackReportBlock{ diff --git a/pkg/rfc8888/stream_log_test.go b/pkg/rfc8888/stream_log_test.go index a9f27ef9..103f916b 100644 --- a/pkg/rfc8888/stream_log_test.go +++ b/pkg/rfc8888/stream_log_test.go @@ -589,7 +589,7 @@ func TestStreamLogDoesNotLeakBelowReportPointer(t *testing.T) { t.Run("cap on throughput", func(t *testing.T) { sl := newStreamLog(0) - for i := 0; i < 2*maxReportBlocks; i++ { + for i := range 2 * maxReportBlocks { sl.add(base, uint16(i), 0) //nolint:gosec // G115 } sl.metricsAfter(base.Add(time.Second), maxReportBlocks) From b5a088234e8a7ef5e1611b917e6c9c65be3c29c0 Mon Sep 17 00:00:00 2001 From: 5ur3 Date: Mon, 6 Jul 2026 21:40:50 +0300 Subject: [PATCH 3/3] Fix possible division by zero --- pkg/rfc8888/recorder.go | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/pkg/rfc8888/recorder.go b/pkg/rfc8888/recorder.go index 1a735459..c1438515 100644 --- a/pkg/rfc8888/recorder.go +++ b/pkg/rfc8888/recorder.go @@ -48,8 +48,12 @@ func (r *Recorder) BuildReport(now time.Time, maxSize int) *rtcp.CCFeedbackRepor ReportTimestamp: ntp.ToNTP32(now), } - maxReportBlocks := max((maxSize-12-(8*len(r.streams)))/2, 0) - maxReportBlocksPerStream := maxReportBlocks / len(r.streams) + streamCount := len(r.streams) + if streamCount == 0 { + return report + } + maxReportBlocks := max((maxSize-12-(8*streamCount))/2, 0) + maxReportBlocksPerStream := maxReportBlocks / streamCount for _, log := range r.streams { block := log.metricsAfter(now, int64(maxReportBlocksPerStream))