Adding burst estimator based start gate for initializing track synchronizer (#759)

* Adding burst estimator based start gate for initializing track synchronizer

RTP packets arriving at a receiver site might come in a burst because of reconnect sending older packets through a NACK + retransmission. Anchoring track start time during phases of bursts would result in setting wrong start time for the track and breaking inter-track syncrhonization.
Adding burst esitmator gate before init which will check between packets RTP and reception time and would allow init only when pace stabilizes (no burst is onging).

Author: milos-lk

pkg/synchronizer/start_gate.go

@@ -0,0 +1,218 @@
+package synchronizer
+
+import (
+	"strings"
+	"time"
+
+	"github.com/livekit/media-sdk/jitter"
+	"github.com/livekit/protocol/logger"
+	"github.com/pion/webrtc/v4"
+)
+
+// startGate is a lightweight buffer that decides when a track should be
+// considered "live". Callers push packets into the gate until it returns a
+// slice of packets that can be used to initialize downstream synchronisation.
+// The second return value exposes how many packets were discarded while the
+// gate was waiting for stability, and the third return value indicates whether
+// the gate has finished its job.
+type startGate interface {
+	Push(pkt jitter.ExtPacket) ([]jitter.ExtPacket, int, bool)
+}
+
+// burstEstimatorGate implements startGate using the burst-estimation logic. It
+// buffers packets until their arrival cadence matches the RTP timestamp spacing
+// closely enough to assume we are caught up with the realtime stream. Minimum
+// arrival spacing is derived from a fraction of the expected RTP interval
+type burstEstimatorGate struct {
+	clockRate         uint32
+	maxSkewFactor     float64
+	minArrivalFactor  float64
+	scoreTarget       int
+	maxStableDuration time.Duration
+	flushAfter        time.Duration
+	firstArrival      time.Time
+	logger            logger.Logger
+
+	score       int
+	lastTS      uint32
+	lastArrival time.Time
+	hasLast     bool
+	done        bool
+	buffer      []jitter.ExtPacket
+	maxBuffer   int
+
+	// stats
+	totalDropped int
+}
+
+func newStartGate(clockRate uint32, kind webrtc.RTPCodecType, logger logger.Logger) startGate {
+	be := &burstEstimatorGate{
+		clockRate:         clockRate,
+		maxSkewFactor:     0.3,
+		minArrivalFactor:  0.2,
+		scoreTarget:       5,
+		maxBuffer:         1000, // high bitrate key frames can span hundreds of packets
+		maxStableDuration: time.Second,
+		flushAfter:        2 * time.Second,
+		logger:            logger,
+	}
+
+	if kind == webrtc.RTPCodecTypeAudio {
+		be.maxBuffer = 200
+	}
+
+	return be
+}
+
+// Push feeds one packet into the gate. Once pacing stabilizes it returns the
+// buffered packets that should be used to initialize the track synchronizer,
+// along with the number of packets dropped while waiting and a done flag.
+func (b *burstEstimatorGate) Push(pkt jitter.ExtPacket) ([]jitter.ExtPacket, int, bool) {
+	if b.done {
+		return nil, 0, true
+	}
+
+	elapsed := time.Duration(0)
+	if !b.firstArrival.IsZero() {
+		elapsed = time.Since(b.firstArrival)
+	} else {
+		b.firstArrival = time.Now()
+	}
+	if b.flushAfter > 0 && elapsed > b.flushAfter {
+		ready := append(b.buffer, pkt)
+		b.buffer = nil
+		b.done = true
+		b.logCompletion("flush_after_timeout")
+		return ready, 0, true
+	}
+
+	if len(pkt.Payload) == 0 {
+		return nil, 0, false
+	}
+
+	if !b.hasLast {
+		b.lastTS = pkt.Timestamp
+		b.lastArrival = pkt.ReceivedAt
+		b.hasLast = true
+		b.buffer = append(b.buffer[:0], pkt)
+		return nil, 0, false
+	}
+
+	signedTsDelta := int32(pkt.Timestamp - b.lastTS)
+	arrivalDelta := pkt.ReceivedAt.Sub(b.lastArrival)
+
+	if signedTsDelta < 0 {
+		// ignore out-of-order packets while continuing to wait for stability
+		return nil, 0, false
+	}
+
+	if signedTsDelta == 0 {
+		// multiple packets with the same timestamp (e.g. key frame)
+		b.buffer = append(b.buffer, pkt)
+		if dropped := b.enforceWindow(); dropped > 0 {
+			return nil, dropped, false
+		}
+		return nil, 0, false
+	}
+
+	b.lastTS = pkt.Timestamp
+	b.lastArrival = pkt.ReceivedAt
+
+	tsDuration := b.timestampToDuration(uint32(signedTsDelta))
+
+	minArrival := time.Duration(float64(tsDuration) * b.minArrivalFactor)
+
+	if arrivalDelta < minArrival {
+		dropped := b.restartSequence(pkt)
+		return nil, dropped, false
+	}
+
+	skew := arrivalDelta - tsDuration
+	if skew < 0 {
+		skew = -skew
+	}
+
+	maxSkew := time.Duration(float64(tsDuration) * b.maxSkewFactor)
+
+	if skew > maxSkew {
+		dropped := b.restartSequence(pkt)
+		return nil, dropped, false
+	}
+
+	b.buffer = append(b.buffer, pkt)
+	if dropped := b.enforceWindow(); dropped > 0 {
+		return nil, dropped, false
+	}
+
+	b.score++
+
+	reachedScore := b.score >= b.scoreTarget
+	reachedDuration := b.totalBufferedDuration() >= b.maxStableDuration
+	if reachedScore || reachedDuration {
+		b.done = true
+		ready := b.buffer
+		b.buffer = nil
+		reasons := make([]string, 0, 2)
+		if reachedScore {
+			reasons = append(reasons, "score_target")
+		}
+		if reachedDuration {
+			reasons = append(reasons, "max_stable_duration")
+		}
+		reason := strings.Join(reasons, ",")
+		b.logCompletion(reason)
+		return ready, 0, true
+	}
+
+	return nil, 0, false
+}
+
+func (b *burstEstimatorGate) timestampToDuration(delta uint32) time.Duration {
+	if b.clockRate == 0 {
+		return 0
+	}
+	return time.Duration(int64(delta) * int64(time.Second) / int64(b.clockRate))
+}
+
+func (b *burstEstimatorGate) enforceWindow() int {
+	if b.maxBuffer == 0 || len(b.buffer) <= b.maxBuffer {
+		return 0
+	}
+	dropped := len(b.buffer) - b.maxBuffer
+	b.buffer = b.buffer[dropped:]
+	b.score = 0
+	b.totalDropped += dropped
+	return dropped
+}
+
+func (b *burstEstimatorGate) restartSequence(seed jitter.ExtPacket) int {
+	dropped := len(b.buffer)
+	b.buffer = b.buffer[:0]
+	b.score = 0
+	b.buffer = append(b.buffer, seed)
+	b.totalDropped += dropped
+	return dropped
+}
+
+func (b *burstEstimatorGate) totalBufferedDuration() time.Duration {
+	if len(b.buffer) < 2 {
+		return 0
+	}
+	first := b.buffer[0].Timestamp
+	last := b.buffer[len(b.buffer)-1].Timestamp
+	return b.timestampToDuration(last - first)
+}
+
+func (b *burstEstimatorGate) logCompletion(reason string) {
+	if b.logger == nil {
+		return
+	}
+
+	b.logger.Infow(
+		"start gate sequence completed",
+		"reason", reason,
+		"score", b.score,
+		"elapsed", time.Since(b.firstArrival),
+		"droppedPackets", b.totalDropped,
+	)
+}

pkg/synchronizer/synchronizer.go

@@ -39,6 +39,7 @@ type SynchronizerConfig struct {
 	PreJitterBufferReceiveTimeEnabled bool
 	RTCPSenderReportRebaseEnabled     bool
 	OldPacketThreshold                time.Duration
+	EnableStartGate                   bool
 
 	OnStarted func()
 }
@@ -108,6 +109,13 @@ func WithOldPacketThreshold(oldPacketThreshold time.Duration) SynchronizerOption
 	}
 }
 
+// WithStartGate enabled will buffer incoming packets until pacing stabilizes before initializing tracks
+func WithStartGate() SynchronizerOption {
+	return func(config *SynchronizerConfig) {
+		config.EnableStartGate = true
+	}
+}
+
 // WithOnStarted sets the callback to be called when the synchronizer starts
 func WithOnStarted(onStarted func()) SynchronizerOption {
 	return func(config *SynchronizerConfig) {

pkg/synchronizer/track.go

@@ -56,6 +56,7 @@ type TrackSynchronizer struct {
 	track  TrackRemote
 	logger logger.Logger
 	*rtpConverter
+	startGate startGate
 
 	// config
 	maxTsDiff                         time.Duration // maximum acceptable difference between RTP packets
@@ -94,6 +95,7 @@ type TrackSynchronizer struct {
 	propagationDelayEstimator *OWDEstimator
 	totalStartTimeAdjustment  time.Duration
 	startTimeAdjustResidual   time.Duration
+	initialized               bool
 
 	stats stats
 }
@@ -115,6 +117,10 @@ func newTrackSynchronizer(s *Synchronizer, track TrackRemote) *TrackSynchronizer
 		propagationDelayEstimator:         NewOWDEstimator(OWDEstimatorParamsDefault),
 	}
 
+	if s.config.EnableStartGate {
+		t.startGate = newStartGate(track.Codec().ClockRate, track.Kind(), t.logger)
+	}
+
 	return t
 }
 
@@ -125,19 +131,56 @@ func (t *TrackSynchronizer) OnSenderReport(f func(drift time.Duration)) {
 	t.onSR = f
 }
 
+// PrimeForStart buffers incoming packets until pacing stabilizes, initializing the
+// track synchronizer automatically once a suitable sequence has been observed.
+// It returns the packets that should be forwarded, the number of packets
+// dropped while waiting, and a boolean indicating whether the track is ready to
+// process samples. After the gate finishes, there is no need to call the API again.
+func (t *TrackSynchronizer) PrimeForStart(pkt jitter.ExtPacket) ([]jitter.ExtPacket, int, bool) {
+	if !t.initialized && len(pkt.Payload) == 0 {
+		// skip dummy packets until the track is initialized
+		return nil, 0, false
+	}
+
+	if t.initialized || t.startGate == nil {
+		if !t.initialized {
+			t.Initialize(pkt.Packet)
+		}
+		return []jitter.ExtPacket{pkt}, 0, true
+	}
+
+	ready, dropped, done := t.startGate.Push(pkt)
+	if !done {
+		return nil, dropped, false
+	}
+
+	if len(ready) == 0 {
+		ready = []jitter.ExtPacket{pkt}
+	}
+
+	if !t.initialized {
+		t.Initialize(ready[0].Packet)
+	}
+
+	return ready, dropped, true
+}
+
 // Initialize should be called as soon as the first packet is received
 func (t *TrackSynchronizer) Initialize(pkt *rtp.Packet) {
 	now := mono.Now()
 	t.Lock()
 	synchronizer := t.sync
 	t.Unlock()
-	if synchronizer == nil {
-		return
+	startedAt := now.UnixNano()
+	if synchronizer != nil {
+		startedAt = synchronizer.getOrSetStartedAt(now.UnixNano())
 	}
-	startedAt := synchronizer.getOrSetStartedAt(now.UnixNano())
 
 	t.Lock()
 	defer t.Unlock()
+	if t.initialized {
+		return
+	}
 
 	t.currentPTSOffset = time.Duration(now.UnixNano() - startedAt)
 	t.desiredPTSOffset = t.currentPTSOffset
@@ -147,6 +190,7 @@ func (t *TrackSynchronizer) Initialize(pkt *rtp.Packet) {
 	t.startRTP = pkt.Timestamp
 	t.lastPTS = 0
 	t.lastPTSAdjusted = t.currentPTSOffset
+	t.initialized = true
 	t.logger.Infow(
 		"initialized track synchronizer",
 		"state", t,

pkg/synchronizer/track_test.go

@@ -205,6 +205,58 @@ func TestShouldAdjustPTS_Deadband_AboveAdjusts(t *testing.T) {
 	require.True(t, ts.shouldAdjustPTS(), "delta > step should allow adjustment")
 }
 
+func TestPrimeForStartWithStartGate(t *testing.T) {
+	clock := uint32(90000)
+	ts := newTSForTests(t, clock, webrtc.RTPCodecTypeVideo)
+	ts.startGate = newStartGate(clock, webrtc.RTPCodecTypeVideo, ts.logger)
+	ts.sync = NewSynchronizerWithOptions()
+
+	stepDur := 20 * time.Millisecond
+	step := ts.rtpConverter.toRTP(stepDur)
+	baseTS := ts.startRTP
+	base := time.Now()
+
+	for i := 0; i < 5; i++ {
+		timestamp := baseTS + uint32(i+1)*step
+		pkt := jitter.ExtPacket{
+			Packet:     &rtp.Packet{Header: rtp.Header{Timestamp: timestamp}, Payload: []byte{0x01}},
+			ReceivedAt: base.Add(time.Duration(i+1) * stepDur),
+		}
+		ready, dropped, done := ts.PrimeForStart(pkt)
+		require.False(t, done)
+		require.Zero(t, dropped)
+		require.Nil(t, ready)
+	}
+
+	finalPkt := jitter.ExtPacket{
+		Packet:     &rtp.Packet{Header: rtp.Header{Timestamp: baseTS + uint32(6*step)}, Payload: []byte{0x02}},
+		ReceivedAt: base.Add(6 * stepDur),
+	}
+	ready, dropped, done := ts.PrimeForStart(finalPkt)
+	require.True(t, done, "gate should be done after final packet")
+	require.Zero(t, dropped, "no packets should be dropped")
+	require.NotEmpty(t, ready, "ready should have at least the final packet")
+	require.True(t, ts.initialized, "track should be initialized")
+}
+
+func TestPrimeForStartWithoutStartGate(t *testing.T) {
+	clock := uint32(48000)
+	ts := newTSForTests(t, clock, webrtc.RTPCodecTypeAudio)
+	ts.startGate = nil
+	ts.sync = NewSynchronizerWithOptions()
+
+	pkt := jitter.ExtPacket{
+		Packet:     &rtp.Packet{Header: rtp.Header{Timestamp: ts.startRTP + 1234}, Payload: []byte{0x01}},
+		ReceivedAt: time.Now(),
+	}
+
+	ready, dropped, done := ts.PrimeForStart(pkt)
+	require.True(t, done, "gate should be done after packet")
+	require.Zero(t, dropped, "no packets should be dropped")
+	require.Len(t, ready, 1, "ready should have the packet")
+	require.True(t, ts.initialized, "track should be initialized")
+}
+
 func TestShouldAdjustPTS_Deadband_NegativeDelta_Suppresses(t *testing.T) {
 	clock := uint32(48000)
 	ts := newTSForTests(t, clock, webrtc.RTPCodecTypeVideo)