diff --git a/pkg/synchronizer/ntpestimator.go b/pkg/synchronizer/ntpestimator.go index 5370ac4e..8cd32228 100644 --- a/pkg/synchronizer/ntpestimator.go +++ b/pkg/synchronizer/ntpestimator.go @@ -115,15 +115,15 @@ func NewNtpEstimator(clockRate uint32) *NtpEstimator { } } -// Reset clears all state, returning the estimator to its initial condition. -// Used when a stream discontinuity is detected (e.g., stream restart with a new -// RTP offset) and the old regression is no longer valid. +// Reset clears all state including OWD. Called via SessionTimeline.ResetTrack for true stream discontinuities where every prior sample is stale. func (e *NtpEstimator) Reset() { e.mu.Lock() defer e.mu.Unlock() e.resetLocked() + e.owdEstimator = latency.NewOWDEstimator(latency.OWDEstimatorParamsDefault) } +// resetLocked clears the regression only; OWD is preserved because outlier-triggered rebuilds don't imply a network change and re-converging OWD would thrash the tempo controller. func (e *NtpEstimator) resetLocked() { e.samples = [maxSRSamples]srSample{} e.sampleLen = 0 @@ -137,10 +137,6 @@ func (e *NtpEstimator) resetLocked() { e.residStd = 0 e.ready = false e.consecutiveOutliers = 0 - // Reset OWD: a sender NTP step that triggered the regression rebuild also - // invalidates the previously-measured clock offset, so the estimator must - // re-converge from the new sender state. - e.owdEstimator = latency.NewOWDEstimator(latency.OWDEstimatorParamsDefault) } // SRResult indicates the outcome of processing a sender report. diff --git a/pkg/synchronizer/syncenginetrack.go b/pkg/synchronizer/syncenginetrack.go index 70bdb0b2..ead76bd6 100644 --- a/pkg/synchronizer/syncenginetrack.go +++ b/pkg/synchronizer/syncenginetrack.go @@ -52,6 +52,9 @@ const ( // deadbandThreshold is the minimum |correction| before slew smoothing kicks in. deadbandThreshold = 5 * time.Millisecond + + // wallSlewAlpha bounds publisher sample-clock skew: steady-state |drift| = R·δ·(1-α)/α ≈ 10ms for R=20ms, δ=±500ppm, α=0.01. + wallSlewAlpha = 0.01 ) // syncEngineTrack implements TrackSync for a single track within a SyncEngine. @@ -425,6 +428,8 @@ func (st *syncEngineTrack) GetPTS(pkt jitter.ExtPacket) (time.Duration, error) { "deadline", deadline, "behindBy", limit-oldPTS, ) + // Reset timeliness clock so a lingering transient deficit doesn't fire force-correction on every subsequent packet, collapsing them into the same newPTS at the mixer. + st.lastTimelyPacket = time.Now() } } else { st.lastTimelyPacket = time.Now() @@ -489,43 +494,33 @@ func (st *syncEngineTrack) wallClockPTSForRTP(ts uint32, receivedAt time.Time) ( // wallClockPTSForRTPLocked is the lock-held implementation of wall-clock PTS // computation. Caller must hold st.mu. func (st *syncEngineTrack) wallClockPTSForRTPLocked(ts uint32, receivedAt time.Time) time.Duration { - // Defense in depth: if receivedAt is the zero time.Time, wallElapsed - // below would be a ~63-billion-second negative duration, the sanity - // threshold would reject rtpDerived, and the function would return 0 - // regardless of sessionOffset — locking the trust-threshold clamp onto - // a bogus baseline. initializeLocked already substitutes time.Now() for - // zero ReceivedAt on the first packet; do the same here in case a later - // caller (e.g., OnRTCP using sr.RTPTime + an unset receivedAt) misses it. + // A zero receivedAt would make wallElapsed hugely negative and lock every downstream clamp onto a bogus baseline. if receivedAt.IsZero() { receivedAt = time.Now() } - // Wall-clock elapsed since this track started, plus session offset wallElapsed := receivedAt.Sub(st.startTime) + st.sessionOffset + if wallElapsed < 0 { + wallElapsed = 0 + } + + if !st.initialized { + return wallElapsed + } - // If we have processed at least one packet, use RTP delta from the previous - // pure-wall-clock baseline for more precision (immune to receivedAt jitter). // uint32 subtraction wraps for backward RTP timestamps; the sanity check // below catches the resulting huge positive delta and falls back to wall clock. - if st.initialized { - rtpDelta := ts - st.lastTS - rtpDerived := st.lastWallPTS + st.converter.ToDuration(rtpDelta) - - // Sanity check: if RTP-derived PTS diverges from wall-clock by > 5s, use wall clock. - diff := rtpDerived - wallElapsed - if diff < 0 { - diff = -diff - } - if diff <= wallClockSanityThreshold { - return rtpDerived - } - } + rtpDelta := ts - st.lastTS + rtpDerived := st.lastWallPTS + st.converter.ToDuration(rtpDelta) + diff := rtpDerived - wallElapsed - // Use wall-clock elapsed, ensuring non-negative. - if wallElapsed < 0 { - wallElapsed = 0 + if diff > wallClockSanityThreshold || diff < -wallClockSanityThreshold { + return wallElapsed } - return wallElapsed + + // Slew rtpDerived toward wallElapsed each packet so publisher sample-clock skew stops accumulating across the sanity band; see wallSlewAlpha for the steady-state math. + absorbed := time.Duration(float64(diff) * wallSlewAlpha) + return rtpDerived - absorbed } // OnSenderReport implements TrackSync. It stores a callback invoked on sender reports.