From d4e50b633f750ecdbca2b556f78692d4bcbe314f Mon Sep 17 00:00:00 2001
From: Raja Subramanian <raja.gobi@tutanota.com>
Date: Thu, 20 Jun 2024 10:52:12 +0530
Subject: [PATCH] Do not log warns on duplicate. (#2807)

With RTX, some clients use very old packets for probing. Check for
duplicate before logging warning about old packet/negative sequence
number jump.

Also, double the history so that duplicate tracking is better. Adds
about 1/2 KB per RTP stream.
---
 pkg/sfu/buffer/buffer.go            | 28 +++++++++++++++-------------
 pkg/sfu/buffer/rtpstats_receiver.go | 22 +++++++++++-----------
 2 files changed, 26 insertions(+), 24 deletions(-)

diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go
index 89f776608..829ee1794 100644
--- a/pkg/sfu/buffer/buffer.go
+++ b/pkg/sfu/buffer/buffer.go
@@ -611,12 +611,23 @@ func (b *Buffer) calc(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime time.Tim
 	rtpPacket.Header.SequenceNumber = uint16(flowState.ExtSequenceNumber)
 	_, err = b.bucket.AddPacketWithSequenceNumber(rawPkt, rtpPacket.Header.SequenceNumber)
 	if err != nil {
-		if errors.Is(err, bucket.ErrPacketTooOld) {
-			packetTooOldCount := b.packetTooOldCount.Inc()
-			if (packetTooOldCount-1)%100 == 0 {
+		if !flowState.IsDuplicate {
+			if errors.Is(err, bucket.ErrPacketTooOld) {
+				packetTooOldCount := b.packetTooOldCount.Inc()
+				if (packetTooOldCount-1)%100 == 0 {
+					b.logger.Warnw(
+						"could not add packet to bucket", err,
+						"count", packetTooOldCount,
+						"flowState", &flowState,
+						"snAdjustment", snAdjustment,
+						"incomingSequenceNumber", flowState.ExtSequenceNumber+snAdjustment,
+						"rtpStats", b.rtpStats,
+						"snRangeMap", b.snRangeMap,
+					)
+				}
+			} else if err != bucket.ErrRTXPacket {
 				b.logger.Warnw(
 					"could not add packet to bucket", err,
-					"count", packetTooOldCount,
 					"flowState", &flowState,
 					"snAdjustment", snAdjustment,
 					"incomingSequenceNumber", flowState.ExtSequenceNumber+snAdjustment,
@@ -624,15 +635,6 @@ func (b *Buffer) calc(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime time.Tim
 					"snRangeMap", b.snRangeMap,
 				)
 			}
-		} else if err != bucket.ErrRTXPacket {
-			b.logger.Warnw(
-				"could not add packet to bucket", err,
-				"flowState", &flowState,
-				"snAdjustment", snAdjustment,
-				"incomingSequenceNumber", flowState.ExtSequenceNumber+snAdjustment,
-				"rtpStats", b.rtpStats,
-				"snRangeMap", b.snRangeMap,
-			)
 		}
 		return
 	}
diff --git a/pkg/sfu/buffer/rtpstats_receiver.go b/pkg/sfu/buffer/rtpstats_receiver.go
index 94952ef1c..6cbc9fd4c 100644
--- a/pkg/sfu/buffer/rtpstats_receiver.go
+++ b/pkg/sfu/buffer/rtpstats_receiver.go
@@ -28,7 +28,7 @@ import (
 )
 
 const (
-	cHistorySize = 4096
+	cHistorySize = 8192
 
 	// RTCP Sender Reports are re-based to SFU time base so that all subscriber side
 	// can have the same time base (i. e. SFU time base). To convert publisher side
@@ -219,16 +219,6 @@ func (r *RTPStatsReceiver) Update(
 		}
 	}
 	if gapSN <= 0 { // duplicate OR out-of-order
-		if -gapSN >= cSequenceNumberLargeJumpThreshold {
-			if r.largeJumpNegativeCount%100 == 0 {
-				r.logger.Warnw(
-					"large sequence number gap negative", nil,
-					append(getLoggingFields(), "count", r.largeJumpNegativeCount)...,
-				)
-			}
-			r.largeJumpNegativeCount++
-		}
-
 		if gapSN != 0 {
 			r.packetsOutOfOrder++
 		}
@@ -248,6 +238,16 @@ func (r *RTPStatsReceiver) Update(
 		flowState.IsOutOfOrder = true
 		flowState.ExtSequenceNumber = resSN.ExtendedVal
 		flowState.ExtTimestamp = resTS.ExtendedVal
+
+		if !flowState.IsDuplicate && -gapSN >= cSequenceNumberLargeJumpThreshold {
+			if r.largeJumpNegativeCount%100 == 0 {
+				r.logger.Warnw(
+					"large sequence number gap negative", nil,
+					append(getLoggingFields(), "count", r.largeJumpNegativeCount)...,
+				)
+			}
+			r.largeJumpNegativeCount++
+		}
 	} else { // in-order
 		if gapSN >= cSequenceNumberLargeJumpThreshold || resTS.ExtendedVal < resTS.PreExtendedHighest {
 			if r.largeJumpCount%100 == 0 {