Rate-limit the NTP-clamp and SR-outlier warn storms

pkg/synchronizer/participantclock.go

@@ -31,6 +31,7 @@ type ParticipantClock struct {
 	participantID string
 	tracks        map[string]*NtpEstimator
 	seenFirstSR   map[string]bool
+	outliers      map[string]int
 }
 
 // NewParticipantClock creates a new ParticipantClock.
@@ -40,6 +41,7 @@ func NewParticipantClock(l logger.Logger, participantID string) *ParticipantCloc
 		participantID: participantID,
 		tracks:        make(map[string]*NtpEstimator),
 		seenFirstSR:   make(map[string]bool),
+		outliers:      make(map[string]int),
 	}
 }
 
@@ -59,23 +61,43 @@ func (pc *ParticipantClock) OnSenderReport(trackID string, clockRate uint32, ntp
 	result := est.OnSenderReport(ntpTime, rtpTimestamp, receivedAt)
 	switch result {
 	case SROutlier:
+		pc.outliers[trackID]++
 		if pc.logger != nil {
-			pc.logger.Warnw("sender report rejected as outlier", nil,
-				"trackID", trackID,
-				"rtpTimestamp", rtpTimestamp,
-				"ntpTime", ntpTime,
-			)
+			if pc.outliers[trackID] == 1 {
+				pc.logger.Warnw("sender report rejected as outlier", nil,
+					"trackID", trackID,
+					"rtpTimestamp", rtpTimestamp,
+					"ntpTime", ntpTime,
+				)
+			} else {
+				pc.logger.Debugw("sender report rejected as outlier (suppressed)",
+					"trackID", trackID,
+					"outliersTotal", pc.outliers[trackID],
+				)
+			}
 		}
 	case SRRebuilt:
 		if pc.logger != nil {
 			pc.logger.Infow("NTP estimator rebuilt after persistent outliers",
 				"participantID", pc.participantID,
 				"trackID", trackID,
+				"outliersTotal", pc.outliers[trackID],
 			)
 		}
+		delete(pc.outliers, trackID)
 		// A rebuild starts a new regression — the next SR is effectively a first.
 		delete(pc.seenFirstSR, trackID)
 	case SRAccepted:
+		if n := pc.outliers[trackID]; n > 0 {
+			if pc.logger != nil {
+				pc.logger.Infow("sender report accepted after outliers",
+					"participantID", pc.participantID,
+					"trackID", trackID,
+					"rejected", n,
+				)
+			}
+			delete(pc.outliers, trackID)
+		}
 		if !pc.seenFirstSR[trackID] {
 			pc.seenFirstSR[trackID] = true
 			if pc.logger != nil {
@@ -120,6 +142,7 @@ func (pc *ParticipantClock) ResetTrack(trackID string) {
 
 	if est, ok := pc.tracks[trackID]; ok {
 		est.Reset()
+		delete(pc.outliers, trackID)
 	}
 }
 
@@ -130,6 +153,7 @@ func (pc *ParticipantClock) RemoveTrack(trackID string) {
 
 	delete(pc.tracks, trackID)
 	delete(pc.seenFirstSR, trackID)
+	delete(pc.outliers, trackID)
 }
 
 // HasTrack returns true if the participant has a track with the given ID.

pkg/synchronizer/syncenginetrack.go

@@ -87,6 +87,10 @@ type syncEngineTrack struct {
 	hasLastNtpPTS   bool          // lastNtpPTS holds a baseline from the current NTP regression (cleared when NTP becomes unavailable or on RTP discontinuity)
 	ntpCorrection   time.Duration // smoothing correction for SR-induced NTP jumps
 
+	clampedPackets int
+	clampStartPTS  time.Duration
+	clampMaxDiff   time.Duration
+
 	// pipeline time feedback
 	lastTimelyPacket time.Time
 
@@ -274,6 +278,8 @@ func (st *syncEngineTrack) GetPTS(pkt jitter.ExtPacket) (time.Duration, error) {
 	}
 
 	if !ntpNowReady {
+		st.endClampEpisodeLocked(wallPTS)
+
 		// NTP is not the source for this packet — either it was never ready,
 		// the estimator was rebuilt internally (persistent outliers in the
 		// timeline's NtpEstimator), the discontinuity branch above just reset
@@ -333,14 +339,18 @@ func (st *syncEngineTrack) GetPTS(pkt jitter.ExtPacket) (time.Duration, error) {
 		clamped := false
 		diff := pts - wallPTS
 		if diff > ntpTrustThreshold || diff < -ntpTrustThreshold {
-			st.logger.Warnw("NTP PTS exceeds trust threshold, clamping to wall clock", nil,
-				"rawNtpPTS", rawNtpPTS,
-				"ntpCorrection", st.ntpCorrection,
-				"wallPTS", wallPTS,
-				"diff", diff,
-			)
+			if st.noteClampLocked(diff, wallPTS) {
+				st.logger.Warnw("NTP PTS exceeds trust threshold, clamping to wall clock", nil,
+					"rawNtpPTS", rawNtpPTS,
+					"ntpCorrection", st.ntpCorrection,
+					"wallPTS", wallPTS,
+					"diff", diff,
+				)
+			}
 			pts = wallPTS
 			clamped = true
+		} else {
+			st.endClampEpisodeLocked(wallPTS)
 		}
 
 		// On first successful NTP PTS that is NOT clamped, compute the
@@ -469,6 +479,42 @@ func (st *syncEngineTrack) GetPTS(pkt jitter.ExtPacket) (time.Duration, error) {
 	return pts, nil
 }
 
+func (st *syncEngineTrack) noteClampLocked(diff, wallPTS time.Duration) bool {
+	st.clampedPackets++
+	if st.clampedPackets == 1 {
+		st.clampStartPTS = wallPTS
+		st.clampMaxDiff = diff
+		return true
+	}
+
+	mag, curMax := diff, st.clampMaxDiff
+	if mag < 0 {
+		mag = -mag
+	}
+	if curMax < 0 {
+		curMax = -curMax
+	}
+	if mag > curMax {
+		st.clampMaxDiff = diff
+	}
+	return false
+}
+
+func (st *syncEngineTrack) endClampEpisodeLocked(wallPTS time.Duration) {
+	if st.clampedPackets == 0 {
+		return
+	}
+
+	st.logger.Infow("NTP clamping episode ended",
+		"clampedPackets", st.clampedPackets,
+		"episodeSpan", wallPTS-st.clampStartPTS,
+		"maxDiff", st.clampMaxDiff,
+	)
+	st.clampedPackets = 0
+	st.clampStartPTS = 0
+	st.clampMaxDiff = 0
+}
+
 func (st *syncEngineTrack) wallClockPTS(pkt jitter.ExtPacket) (slewed, unslewed time.Duration) {
 	return st.wallClockPTSForRTPLocked(pkt.Timestamp, pkt.ReceivedAt)
 }
@@ -558,6 +604,8 @@ func (st *syncEngineTrack) Close() {
 // would require holding st.mu across the user-supplied callback, which is
 // an unacceptable constraint on what the callback may do.
 func (st *syncEngineTrack) closeLocked() {
+	st.endClampEpisodeLocked(st.lastPTSAdjusted)
+
 	st.closed = true
 	// Clear the field too: OnRTCP invocations that arrive AFTER close and
 	// somehow pass the closed check would see nil. Also helps GC release