From abe467f8504775fa30190e149935d52cb10bc464 Mon Sep 17 00:00:00 2001 From: David Colburn Date: Fri, 14 Aug 2026 21:28:27 -0400 Subject: [PATCH 1/3] Rate-limit the NTP-clamp and SR-outlier warn storms Both warn sites emit once per episode instead of once per packet, cutting sync-engine warn volume by orders of magnitude without losing the signal that the condition occurred. --- pkg/synchronizer/participantclock.go | 30 +++++++++++--- pkg/synchronizer/syncenginetrack.go | 60 +++++++++++++++++++++++++--- 2 files changed, 79 insertions(+), 11 deletions(-) diff --git a/pkg/synchronizer/participantclock.go b/pkg/synchronizer/participantclock.go index 542b3097..8c2964d8 100644 --- a/pkg/synchronizer/participantclock.go +++ b/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,18 +61,27 @@ 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], ) } // A rebuild starts a new regression — the next SR is effectively a first. @@ -128,8 +139,17 @@ func (pc *ParticipantClock) RemoveTrack(trackID string) { pc.mu.Lock() defer pc.mu.Unlock() + if n := pc.outliers[trackID]; n > 1 && pc.logger != nil { + pc.logger.Infow("track sender report outlier summary", + "participantID", pc.participantID, + "trackID", trackID, + "outliersTotal", n, + ) + } + 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. diff --git a/pkg/synchronizer/syncenginetrack.go b/pkg/synchronizer/syncenginetrack.go index e54988e5..a8257c52 100644 --- a/pkg/synchronizer/syncenginetrack.go +++ b/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 From 66e5c7cfb2e12e133a67d89c69b59982dad0b6ad Mon Sep 17 00:00:00 2001 From: David Colburn Date: Fri, 14 Aug 2026 21:47:56 -0400 Subject: [PATCH 2/3] log outliers on transition to or from accepted --- pkg/synchronizer/participantclock.go | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/pkg/synchronizer/participantclock.go b/pkg/synchronizer/participantclock.go index 8c2964d8..d7553e44 100644 --- a/pkg/synchronizer/participantclock.go +++ b/pkg/synchronizer/participantclock.go @@ -84,9 +84,20 @@ func (pc *ParticipantClock) OnSenderReport(trackID string, clockRate uint32, ntp "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 { @@ -139,14 +150,6 @@ func (pc *ParticipantClock) RemoveTrack(trackID string) { pc.mu.Lock() defer pc.mu.Unlock() - if n := pc.outliers[trackID]; n > 1 && pc.logger != nil { - pc.logger.Infow("track sender report outlier summary", - "participantID", pc.participantID, - "trackID", trackID, - "outliersTotal", n, - ) - } - delete(pc.tracks, trackID) delete(pc.seenFirstSR, trackID) delete(pc.outliers, trackID) From 133de0b36af1f71841b070534266616a4bc8bb76 Mon Sep 17 00:00:00 2001 From: David Colburn Date: Fri, 14 Aug 2026 22:15:36 -0400 Subject: [PATCH 3/3] reset outlier count on reset --- pkg/synchronizer/participantclock.go | 1 + 1 file changed, 1 insertion(+) diff --git a/pkg/synchronizer/participantclock.go b/pkg/synchronizer/participantclock.go index d7553e44..e6350955 100644 --- a/pkg/synchronizer/participantclock.go +++ b/pkg/synchronizer/participantclock.go @@ -142,6 +142,7 @@ func (pc *ParticipantClock) ResetTrack(trackID string) { if est, ok := pc.tracks[trackID]; ok { est.Reset() + delete(pc.outliers, trackID) } }