From 4733e864bc9e5d5356c01aef47737972d73b1779 Mon Sep 17 00:00:00 2001
From: Raja Subramanian <raja.gobi@tutanota.com>
Date: Thu, 18 Jul 2024 11:02:52 +0530
Subject: [PATCH] Rollover sequence number when time stamp is moving forward.
 (#2876)

* Rollover sequence number when time stamp is moving forward.

Seeing large gaps in sequence number due to potential network issues.
In that gap, the sequence number could roll over.

Using packet time jumps to figure out if a roll over could have happened
and force roll over the sequence number to ensure that it does not flow
backwards.

* fix test
---
 pkg/sfu/buffer/rtpstats_receiver.go      | 61 ++++++++++++++++--------
 pkg/sfu/buffer/rtpstats_receiver_test.go |  2 +-
 2 files changed, 41 insertions(+), 22 deletions(-)

diff --git a/pkg/sfu/buffer/rtpstats_receiver.go b/pkg/sfu/buffer/rtpstats_receiver.go
index 11f2d4a34..84da6ccae 100644
--- a/pkg/sfu/buffer/rtpstats_receiver.go
+++ b/pkg/sfu/buffer/rtpstats_receiver.go
@@ -167,6 +167,24 @@ func (r *RTPStatsReceiver) Update(
 	var resSN utils.WrapAroundUpdateResult[uint64]
 	var gapSN int64
 	var resTS utils.WrapAroundUpdateResult[uint64]
+
+	getLoggingFields := func() []interface{} {
+		return []interface{}{
+			"resSN", resSN,
+			"gapSN", gapSN,
+			"resTS", resTS,
+			"gapTS", int64(resTS.ExtendedVal - resTS.PreExtendedHighest),
+			"packetTime", time.Unix(0, packetTime).String(),
+			"sequenceNumber", sequenceNumber,
+			"timestamp", timestamp,
+			"marker", marker,
+			"hdrSize", hdrSize,
+			"payloadSize", payloadSize,
+			"paddingSize", paddingSize,
+			"rtpStats", lockedRTPStatsReceiverLogEncoder{r},
+		}
+	}
+
 	if !r.initialized {
 		if payloadSize == 0 {
 			// do not start on a padding only packet
@@ -199,32 +217,33 @@ func (r *RTPStatsReceiver) Update(
 			flowState.IsNotHandled = true
 			return
 		}
-
 		gapSN = int64(resSN.ExtendedVal - resSN.PreExtendedHighest)
-		if gapSN <= 0 {
-			resTS = r.timestamp.Update(timestamp)
-		} else {
-			resTS = r.timestamp.Rollover(timestamp, r.getTSRolloverCount(packetTime-r.highestTime))
+
+		resTS = r.timestamp.Rollover(timestamp, r.getTSRolloverCount(packetTime-r.highestTime))
+		if resTS.IsUnhandled {
+			flowState.IsNotHandled = true
+			return
+		}
+		gapTS := int64(resTS.ExtendedVal - resTS.PreExtendedHighest)
+
+		// it is possible that sequence number has rolled over too
+		if gapSN < 0 && gapTS > 0 {
+			// not possible to know how many cycles of sequence number roll over could have happened,
+			// use 1 to ensure that it at least does not go backwards
+			resSN = r.sequenceNumber.Rollover(sequenceNumber, 1)
+			if resSN.IsUnhandled {
+				flowState.IsNotHandled = true
+				return
+			}
+
+			r.logger.Warnw(
+				"forcing sequence number rollover", nil,
+				getLoggingFields()...,
+			)
 		}
 	}
 
 	pktSize := uint64(hdrSize + payloadSize + paddingSize)
-	getLoggingFields := func() []interface{} {
-		return []interface{}{
-			"resSN", resSN,
-			"gapSN", gapSN,
-			"resTS", resTS,
-			"gapTS", int64(resTS.ExtendedVal - resTS.PreExtendedHighest),
-			"packetTime", time.Unix(0, packetTime).String(),
-			"sequenceNumber", sequenceNumber,
-			"timestamp", timestamp,
-			"marker", marker,
-			"hdrSize", hdrSize,
-			"payloadSize", payloadSize,
-			"paddingSize", paddingSize,
-			"rtpStats", lockedRTPStatsReceiverLogEncoder{r},
-		}
-	}
 	if gapSN <= 0 { // duplicate OR out-of-order
 		if gapSN != 0 {
 			r.packetsOutOfOrder++
diff --git a/pkg/sfu/buffer/rtpstats_receiver_test.go b/pkg/sfu/buffer/rtpstats_receiver_test.go
index 075578dd8..eb9ed72bf 100644
--- a/pkg/sfu/buffer/rtpstats_receiver_test.go
+++ b/pkg/sfu/buffer/rtpstats_receiver_test.go
@@ -190,7 +190,7 @@ func Test_RTPStatsReceiver_Update(t *testing.T) {
 	require.Equal(t, uint64(9), r.packetsLost)
 
 	// out-of-order should decrement number of lost packets
-	packet = getPacket(sequenceNumber-6, timestamp-45000, 1000)
+	packet = getPacket(sequenceNumber-6, timestamp-18000, 1000)
 	flowState = r.Update(
 		time.Now().UnixNano(),
 		packet.Header.SequenceNumber,