feat: add configurable ntpTrustThreshold option

pkg/synchronizer/syncengine.go

@@ -29,6 +29,13 @@ import (
 const (
 	// defaultOldPacketThreshold is the default age after which packets are dropped.
 	defaultOldPacketThreshold = 500 * time.Millisecond
+
+	// defaultNtpTrustThreshold is the maximum allowed divergence between NTP-derived PTS
+	// and wall-clock PTS. If NTP disagrees with wall clock by more than this,
+	// the NTP data is suspect (bad SRs, clock jumps, nonsensical timing) and
+	// we clamp to wall clock. This prevents bad publishers from dragging PTS far
+	// from reality.
+	defaultNtpTrustThreshold = 500 * time.Millisecond
 )
 
 // SyncEngineOption configures a SyncEngine.
@@ -56,6 +63,14 @@ func WithSyncEngineOldPacketThreshold(d time.Duration) SyncEngineOption {
 	}
 }
 
+// WithSyncEngineNtpTrustThreshold sets the max divergence between NTP-derived
+// PTS and wall-clock PTS before clamping to wall clock.
+func WithSyncEngineNtpTrustThreshold(d time.Duration) SyncEngineOption {
+	return func(e *SyncEngine) {
+		e.ntpTrustThreshold = d
+	}
+}
+
 // WithSyncEngineMediaRunningTime sets the initial media running time provider and max delay.
 // If a track's PTS falls behind the deadline by more than maxDelay for >10s, PTS is force-corrected.
 func WithSyncEngineMediaRunningTime(mediaRunningTime func() (time.Duration, bool), maxDelay time.Duration) SyncEngineOption {
@@ -106,7 +121,8 @@ type SyncEngine struct {
 	logger                logger.Logger
 	enableStartGate       bool
 	oldPacketThreshold    time.Duration
-	audioDriftCompensated bool // audio drift handled externally (e.g., tempo controller)
+	ntpTrustThreshold     time.Duration // max NTP-vs-wall divergence before clamping to wall clock
+	audioDriftCompensated bool          // audio drift handled externally (e.g., tempo controller)
 	onStarted             func()
 
 	mediaRunningTime         func() (time.Duration, bool)
@@ -120,6 +136,7 @@ func NewSyncEngine(opts ...SyncEngineOption) *SyncEngine {
 		tracks:             make(map[uint32]*syncEngineTrack),
 		trackIDs:           make(map[string]*syncEngineTrack),
 		oldPacketThreshold: defaultOldPacketThreshold,
+		ntpTrustThreshold:  defaultNtpTrustThreshold,
 	}
 	for _, opt := range opts {
 		opt(e)

pkg/synchronizer/syncenginetrack.go

@@ -36,13 +36,6 @@ const (
 	// and wall-clock PTS before falling back to wall clock in wallClockPTS().
 	wallClockSanityThreshold = 5 * time.Second
 
-	// ntpTrustThreshold is the maximum allowed divergence between NTP-derived PTS
-	// and wall-clock PTS. If NTP disagrees with wall clock by more than this,
-	// the NTP data is suspect (bad SRs, clock jumps, nonsensical timing) and
-	// we clamp to wall clock. This prevents bad publishers from dragging PTS far
-	// from reality.
-	ntpTrustThreshold = 500 * time.Millisecond
-
 	// maxTimelyPacketAge is how long a track can be behind the pipeline deadline
 	// before its PTS is force-corrected forward.
 	maxTimelyPacketAge = 10 * time.Second
@@ -338,7 +331,7 @@ func (st *syncEngineTrack) GetPTS(pkt jitter.ExtPacket) (time.Duration, error) {
 		// Clamp corrected PTS to within trust threshold of wall clock.
 		clamped := false
 		diff := pts - wallPTS
-		if diff > ntpTrustThreshold || diff < -ntpTrustThreshold {
+		if diff > st.engine.ntpTrustThreshold || diff < -st.engine.ntpTrustThreshold {
 			if st.noteClampLocked(diff, wallPTS) {
 				st.logger.Warnw("NTP PTS exceeds trust threshold, clamping to wall clock", nil,
 					"rawNtpPTS", rawNtpPTS,