From 7c16ca6a0cdfe9172eaa8ffbdd445aa54e480997 Mon Sep 17 00:00:00 2001
From: Raja Subramanian <raja.gobi@tutanota.com>
Date: Sun, 4 Feb 2024 11:12:22 +0530
Subject: [PATCH] Log feed Sender Report to better understand forwarded sender
 report (#2443)

anomalies.
---
 pkg/sfu/buffer/rtpstats_sender.go | 26 +++++++++++++++++++++++++-
 pkg/sfu/downtrack.go              |  2 +-
 pkg/sfu/streamtrackermanager.go   |  9 +++++++++
 3 files changed, 35 insertions(+), 2 deletions(-)

diff --git a/pkg/sfu/buffer/rtpstats_sender.go b/pkg/sfu/buffer/rtpstats_sender.go
index 618341e75..98b25744a 100644
--- a/pkg/sfu/buffer/rtpstats_sender.go
+++ b/pkg/sfu/buffer/rtpstats_sender.go
@@ -160,6 +160,9 @@ type RTPStatsSender struct {
 	clockSkewCount              int
 	outOfOrderSenderReportCount int
 	metadataCacheOverflowCount  int
+
+	srFeedFirst  *RTCPSenderReportData
+	srFeedNewest *RTCPSenderReportData
 }
 
 func NewRTPStatsSender(params RTPStatsParams) *RTPStatsSender {
@@ -198,6 +201,15 @@ func (r *RTPStatsSender) Seed(from *RTPStatsSender) {
 	r.nextSenderSnapshotID = from.nextSenderSnapshotID
 	r.senderSnapshots = make([]senderSnapshot, cap(from.senderSnapshots))
 	copy(r.senderSnapshots, from.senderSnapshots)
+
+	if from.srFeedFirst != nil {
+		srFeedFirst := *from.srFeedFirst
+		r.srFeedFirst = &srFeedFirst
+	}
+	if from.srFeedNewest != nil {
+		srFeedNewest := *from.srFeedNewest
+		r.srFeedNewest = &srFeedNewest
+	}
 }
 
 func (r *RTPStatsSender) NewSnapshotId() uint32 {
@@ -571,10 +583,16 @@ func (r *RTPStatsSender) LastReceiverReportTime() time.Time {
 	return r.lastRRTime
 }
 
-func (r *RTPStatsSender) MaybeAdjustFirstPacketTime(ts uint32) {
+func (r *RTPStatsSender) MaybeAdjustFirstPacketTime(srData *RTCPSenderReportData, ts uint32) {
 	r.lock.Lock()
 	defer r.lock.Unlock()
 
+	srDataCopy := *srData
+	if r.srFeedFirst == nil {
+		r.srFeedFirst = &srDataCopy
+	}
+	r.srFeedNewest = &srDataCopy
+
 	r.maybeAdjustFirstPacketTime(ts, uint32(r.extStartTS))
 }
 
@@ -637,8 +655,11 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, calculatedClockRate ui
 			if r.clockSkewCount%10 == 0 {
 				r.logger.Infow(
 					"sending sender report, clock skew",
+					"first", r.srFirst.ToString(),
 					"last", r.srNewest.ToString(),
 					"curr", srData.ToString(),
+					"firstFeed", r.srFeedFirst.ToString(),
+					"lastFeed", r.srFeedNewest.ToString(),
 					"timeNow", time.Now().String(),
 					"extStartTS", r.extStartTS,
 					"extHighestTS", r.extHighestTS,
@@ -674,8 +695,11 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, calculatedClockRate ui
 		if r.outOfOrderSenderReportCount%10 == 0 {
 			r.logger.Infow(
 				"sending sender report, out-of-order, repairing",
+				"first", r.srFirst.ToString(),
 				"last", r.srNewest.ToString(),
 				"curr", srData.ToString(),
+				"firstFeed", r.srFeedFirst.ToString(),
+				"lastFeed", r.srFeedNewest.ToString(),
 				"timeNow", time.Now().String(),
 				"extStartTS", r.extStartTS,
 				"extHighestTS", r.extHighestTS,
diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go
index f12859142..470e11eed 100644
--- a/pkg/sfu/downtrack.go
+++ b/pkg/sfu/downtrack.go
@@ -1914,7 +1914,7 @@ func (d *DownTrack) sendSilentFrameOnMuteForOpus() {
 
 func (d *DownTrack) HandleRTCPSenderReportData(_payloadType webrtc.PayloadType, isSVC bool, layer int32, srData *buffer.RTCPSenderReportData) error {
 	if (layer == d.forwarder.GetReferenceLayerSpatial() || (layer == 0 && isSVC)) && srData != nil {
-		d.rtpStats.MaybeAdjustFirstPacketTime(srData.RTPTimestamp + uint32(d.forwarder.GetReferenceTimestampOffset()))
+		d.rtpStats.MaybeAdjustFirstPacketTime(srData, srData.RTPTimestamp+uint32(d.forwarder.GetReferenceTimestampOffset()))
 	}
 	return nil
 }
diff --git a/pkg/sfu/streamtrackermanager.go b/pkg/sfu/streamtrackermanager.go
index a934b9e93..2817ef31f 100644
--- a/pkg/sfu/streamtrackermanager.go
+++ b/pkg/sfu/streamtrackermanager.go
@@ -572,6 +572,15 @@ func (s *StreamTrackerManager) updateLayerOffsetLocked(ref, other int32) {
 		// Rationale: higher layers could be paused for extended periods of time
 		// due to adaptive stream/dynacast or publisher constraints like CPU/bandwidth.
 		// The check is to avoid using very old reports.
+		s.logger.Infow(
+			"skipping offset update",
+			"ref", ref,
+			"refNTP", srRef.NTPTimestamp.Time().String(),
+			"refRTP", srRef.RTPTimestamp,
+			"other", other,
+			"otherNTP", srOther.NTPTimestamp.Time().String(),
+			"otherRTP", srOther.RTPTimestamp,
+		)
 		return
 	}
 	rtpDiff := ntpDiff.Nanoseconds() * int64(s.clockRate) / 1e9