Apply start time correction in discrete steps (#748)

Author: milos-lk

pkg/synchronizer/track.go

@@ -34,6 +34,7 @@ import (
 const (
 	cStartTimeAdjustWindow    = 2 * time.Minute
 	cStartTimeAdjustThreshold = 5 * time.Second
+	cStartTimeAdjustStep      = 5 * time.Millisecond
 
 	cHighDriftLoggingThreshold = 20 * time.Millisecond
 )
@@ -90,6 +91,7 @@ type TrackSynchronizer struct {
 
 	propagationDelayEstimator *OWDEstimator
 	totalStartTimeAdjustment  time.Duration
+	startTimeAdjustResidual   time.Duration
 }
 
 func newTrackSynchronizer(s *Synchronizer, track TrackRemote) *TrackSynchronizer {
@@ -585,7 +587,7 @@ func (t *TrackSynchronizer) maybeAdjustStartTime(asr *augmentedSenderReport) int
 	if int32(samplesDiff) < 0 {
 		// out-of-order, pre-start, skip
 		t.logger.Debugw(
-			"no adjustment due to pre-staart report",
+			"no adjustment due to pre-start report",
 			"receivedSR", wrappedAugmentedSenderReportLogger{asr},
 			"state", t,
 			"nowTS", nowTS,
@@ -623,9 +625,8 @@ func (t *TrackSynchronizer) maybeAdjustStartTime(asr *augmentedSenderReport) int
 				getLoggingFields()...,
 			)
 		} else {
-			t.logger.Debugw("adjusting start time", getLoggingFields()...)
-			t.totalStartTimeAdjustment += time.Duration(startTimeNano - adjustedStartTimeNano)
-			t.startTime = time.Unix(0, adjustedStartTimeNano)
+			applied := t.applyQuantizedStartTimeAdvance(time.Duration(startTimeNano - adjustedStartTimeNano))
+			t.logger.Debugw("adjusting start time", append(getLoggingFields(), "applied", applied)...)
 		}
 	}
 
@@ -652,6 +653,25 @@ func (t *TrackSynchronizer) isPacketTooOld(packetTime time.Time) bool {
 	return t.oldPacketThreshold != 0 && mono.Now().Sub(packetTime) > t.oldPacketThreshold
 }
 
+func (t *TrackSynchronizer) applyQuantizedStartTimeAdvance(deltaTotal time.Duration) time.Duration {
+	// include any prior residual
+	deltaTotal += t.startTimeAdjustResidual
+
+	quanta := deltaTotal / cStartTimeAdjustStep
+	residual := deltaTotal % cStartTimeAdjustStep
+
+	if quanta > 0 {
+		applied := quanta * cStartTimeAdjustStep
+		t.startTime = t.startTime.Add(-applied)
+		t.totalStartTimeAdjustment += applied
+		t.startTimeAdjustResidual = residual
+		return applied
+	}
+
+	t.startTimeAdjustResidual = deltaTotal
+	return 0
+}
+
 func (t *TrackSynchronizer) MarshalLogObject(e zapcore.ObjectEncoder) error {
 	if t == nil {
 		return nil

pkg/synchronizer/track_test.go

@@ -0,0 +1,166 @@
+// Copyright 2025 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package synchronizer
+
+import (
+	"testing"
+	"time"
+
+	"github.com/pion/rtp"
+	"github.com/pion/webrtc/v4"
+	"github.com/stretchr/testify/require"
+
+	"github.com/livekit/media-sdk/jitter"
+	"github.com/livekit/protocol/logger"
+	"github.com/livekit/protocol/utils/mono"
+)
+
+// ---- test fakes & helpers ----
+
+type fakeTrack struct {
+	id   string
+	rate uint32
+	kind webrtc.RTPCodecType
+}
+
+func (f fakeTrack) ID() string { return f.id }
+func (f fakeTrack) Codec() webrtc.RTPCodecParameters {
+	return webrtc.RTPCodecParameters{RTPCodecCapability: webrtc.RTPCodecCapability{ClockRate: f.rate}}
+}
+func (f fakeTrack) Kind() webrtc.RTPCodecType { return f.kind }
+func (f fakeTrack) SSRC() webrtc.SSRC         { return 1234 }
+
+func newTSForTests(tc *testing.T, clockRate uint32, kind webrtc.RTPCodecType) *TrackSynchronizer {
+	t := &TrackSynchronizer{
+		sync:               nil, // construct directly to avoid depending on Synchronizer
+		track:              fakeTrack{id: "t", rate: clockRate, kind: kind},
+		logger:             logger.NewTestLogger(tc),
+		rtpConverter:       newRTPConverter(int64(clockRate)),
+		maxTsDiff:          200 * time.Millisecond,
+		maxDriftAdjustment: 5 * time.Millisecond,
+	}
+	// set a stable startTime well in the past to make time.Since(startTime) > 0
+	t.startTime = time.Now().Add(-150 * time.Millisecond)
+	t.initTime = t.startTime
+	// pick an arbitrary RTP base
+	t.startRTP = 1000
+	return t
+}
+
+// ---- tests ----
+
+func TestApplyQuantizedStartTimeAdvance_ExactQuanta(t *testing.T) {
+	ts := newTSForTests(t, 48000, webrtc.RTPCodecTypeAudio)
+	base := ts.startTime
+
+	// 25ms delta with 5ms step -> apply 25ms, residual 0
+	applied := ts.applyQuantizedStartTimeAdvance(25 * time.Millisecond)
+	require.Equal(t, 25*time.Millisecond, applied)
+	require.Equal(t, 25*time.Millisecond, base.Sub(ts.startTime))
+	require.Equal(t, time.Duration(0), ts.startTimeAdjustResidual)
+	require.Equal(t, 25*time.Millisecond, ts.totalStartTimeAdjustment)
+}
+
+func TestApplyQuantizedStartTimeAdvance_ResidualCarryAcrossCalls(t *testing.T) {
+	ts := newTSForTests(t, 48000, webrtc.RTPCodecTypeAudio)
+	base := ts.startTime
+
+	// First call: 3ms (<5ms step) -> apply 0, residual=3ms
+	applied1 := ts.applyQuantizedStartTimeAdvance(3 * time.Millisecond)
+	require.Equal(t, time.Duration(0), applied1)
+	require.Equal(t, 3*time.Millisecond, ts.startTimeAdjustResidual)
+	require.Equal(t, ts.startTime, base)
+	require.Equal(t, 3*time.Millisecond, ts.startTimeAdjustResidual)
+}
+
+func TestApplyQuantizedStartTimeAdvance_NoOpForZero(t *testing.T) {
+	ts := newTSForTests(t, 48000, webrtc.RTPCodecTypeAudio)
+	base := ts.startTime
+
+	applied := ts.applyQuantizedStartTimeAdvance(0)
+	require.Equal(t, time.Duration(0), applied)
+	require.Equal(t, ts.startTime, base)
+	require.Equal(t, time.Duration(0), ts.startTimeAdjustResidual)
+}
+
+func TestGetPTSWithoutRebase_Increasing(t *testing.T) {
+	clock := uint32(48000)
+	ts := newTSForTests(t, clock, webrtc.RTPCodecTypeAudio)
+
+	// Simulate accepting two frames in order: 20ms and then 20ms later
+	// Convert 20ms -> RTP ticks
+	rtp20ms := ts.rtpConverter.toRTP(20 * time.Millisecond)
+
+	now := time.Now()
+	// First packet initializes lastTS path
+	ts.lastTS = 0
+	ts.lastPTS = 0
+
+	adj1, err := ts.getPTSWithoutRebase(jitter.ExtPacket{
+		Packet:     &rtp.Packet{Header: rtp.Header{Timestamp: ts.startRTP + rtp20ms}},
+		ReceivedAt: now,
+	})
+	require.NoError(t, err)
+
+	adj2, err := ts.getPTSWithoutRebase(jitter.ExtPacket{
+		Packet:     &rtp.Packet{Header: rtp.Header{Timestamp: ts.startRTP + 2*rtp20ms}},
+		ReceivedAt: now,
+	})
+	require.NoError(t, err)
+
+	require.Greater(t, adj2, adj1)
+}
+
+func TestGetPTSWithRebase_PropelsForward(t *testing.T) {
+	clock := uint32(48000)
+	ts := newTSForTests(t, clock, webrtc.RTPCodecTypeAudio)
+	ts.rtcpSenderReportRebaseEnabled = true
+
+	ts.maxTsDiff = 30 * time.Millisecond
+
+	// 1) Seed: make adjusted ~500ms on the first packet.
+	ts.startTime = mono.Now().Add(-500 * time.Millisecond)
+	ts.currentPTSOffset = 0
+	ts.lastPTS = 0
+	ts.startRTP = 100000
+	ts.lastTS = ts.startRTP
+
+	rtp500ms := ts.rtpConverter.toRTP(500 * time.Millisecond)
+	rtp10ms := ts.rtpConverter.toRTP(10 * time.Millisecond)
+
+	// First packet (~500ms)
+	ts1 := ts.startRTP + rtp500ms
+	adj1, err := ts.getPTSWithRebase(jitter.ExtPacket{
+		Packet:     &rtp.Packet{Header: rtp.Header{Timestamp: ts1}},
+		ReceivedAt: time.Time{}, // not used for estimatedPTS here
+	})
+	require.NoError(t, err)
+	require.InDelta(t, 500*time.Millisecond, adj1, float64(20*time.Millisecond))
+
+	// 2) Simulate startTime shift LATER (closer to now) so next estimatedPTS is tiny (~5–10ms)
+	ts.startTime = mono.Now().Add(-5 * time.Millisecond)
+
+	// Second packet: +10ms RTP so ts != lastTS. After correction, adjusted will be tiny and < lastPTSAdjusted.
+	ts2 := ts1 + rtp10ms
+	prev := ts.lastPTSAdjusted      // ~500ms from first call
+	want := prev + time.Millisecond // propel to ~501ms
+
+	adj2, err := ts.getPTSWithRebase(jitter.ExtPacket{
+		Packet:     &rtp.Packet{Header: rtp.Header{Timestamp: ts2}},
+		ReceivedAt: time.Time{},
+	})
+	require.NoError(t, err)
+	require.Equal(t, want, adj2)
+}