diff --git a/pkg/rfc8888/recorder.go b/pkg/rfc8888/recorder.go index 35a65cfc..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 := (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)) diff --git a/pkg/rfc8888/recorder_test.go b/pkg/rfc8888/recorder_test.go index 1f4bbb6d..e55d3da7 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 := 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) + } + }) } diff --git a/pkg/rfc8888/stream_log.go b/pkg/rfc8888/stream_log.go index d95deccd..d6b22973 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, @@ -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 diff --git a/pkg/rfc8888/stream_log_test.go b/pkg/rfc8888/stream_log_test.go index 2f28bc3d..103f916b 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 := 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) + }) +}