Fix rfc8888 implementation

pkg/rfc8888/recorder.go

@@ -48,8 +48,12 @@ func (r *Recorder) BuildReport(now time.Time, maxSize int) *rtcp.CCFeedbackRepor
 		ReportTimestamp: ntp.ToNTP32(now),
 	}
 
-	maxReportBlocks := (maxSize - 12 - (8 * len(r.streams))) / 2
-	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))

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 := range streams {
+			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)
+		}
+	})
 }

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,
@@ -49,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{
@@ -60,7 +66,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

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 := range 2 * maxReportBlocks {
+			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)
+	})
+}