Initialize should be called with the packet reception time (#762)

Adding private function that does almost the same thing as the existing initialze, with the diff that it takes what it used to be now from the packet reception time. Main reason is that initialize might be delayed (until stable sequence of packets arrive), if time.Now is used on the first packet which was gated for a while - it will induce an offset.
Using packet reception time might have as a result that some other track initialized the syncrhonizer main start point to a time t1 that comes after a time of reception of this track first packet. This will in turn make base offset negative - and potentially first few PTSs returned by GetPTS.
For now these negative PTS could be ignored at the caller site before pushing a packet to the gst pipeline. Ideally - these are not ignored but a new offset is generated for the given source with a base time that would reflect the negative offset and pushed to the gst appsource.

Author: milos-lk

pkg/synchronizer/track.go

@@ -139,7 +139,7 @@ func (t *TrackSynchronizer) OnSenderReport(f func(drift time.Duration)) {
 func (t *TrackSynchronizer) PrimeForStart(pkt jitter.ExtPacket) ([]jitter.ExtPacket, int, bool) {
 	if t.initialized || t.startGate == nil {
 		if !t.initialized {
-			t.Initialize(pkt.Packet)
+			t.initialize(pkt)
 		}
 		return []jitter.ExtPacket{pkt}, 0, true
 	}
@@ -154,21 +154,34 @@ func (t *TrackSynchronizer) PrimeForStart(pkt jitter.ExtPacket) ([]jitter.ExtPac
 	}
 
 	if !t.initialized {
-		t.Initialize(ready[0].Packet)
+		t.initialize(ready[0])
 	}
 
 	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.initialize(jitter.ExtPacket{
+		Packet:     pkt,
+		ReceivedAt: mono.Now(),
+	})
+}
+
+func (t *TrackSynchronizer) initialize(extPkt jitter.ExtPacket) {
+	receivedAt := extPkt.ReceivedAt
+	if receivedAt.IsZero() {
+		receivedAt = mono.Now()
+	}
+
 	t.Lock()
 	synchronizer := t.sync
 	t.Unlock()
-	startedAt := now.UnixNano()
+
+	startedAt := receivedAt.UnixNano()
+	firstStartedAt := startedAt
 	if synchronizer != nil {
-		startedAt = synchronizer.getOrSetStartedAt(now.UnixNano())
+		firstStartedAt = synchronizer.getOrSetStartedAt(startedAt)
 	}
 
 	t.Lock()
@@ -177,12 +190,12 @@ func (t *TrackSynchronizer) Initialize(pkt *rtp.Packet) {
 		return
 	}
 
-	t.currentPTSOffset = time.Duration(now.UnixNano() - startedAt)
+	t.currentPTSOffset = time.Duration(startedAt - firstStartedAt)
 	t.desiredPTSOffset = t.currentPTSOffset
 	t.basePTSOffset = t.desiredPTSOffset
 
-	t.startTime = now
-	t.startRTP = pkt.Timestamp
+	t.startTime = receivedAt
+	t.startRTP = extPkt.Packet.Timestamp
 	t.lastPTS = 0
 	t.lastPTSAdjusted = t.currentPTSOffset
 	t.initialized = true

pkg/synchronizer/track_test.go

@@ -122,6 +122,63 @@ func TestGetPTSWithoutRebase_Increasing(t *testing.T) {
 	require.Greater(t, adj2, adj1)
 }
 
+func TestGetPTSWithoutRebase_NegativeAdjustedPTS(t *testing.T) {
+	sync := NewSynchronizerWithOptions()
+	base := time.Now()
+	futureStart := base.Add(500 * time.Millisecond)
+	sync.getOrSetStartedAt(futureStart.UnixNano())
+
+	track := fakeTrack{id: "neg", rate: 90000, kind: webrtc.RTPCodecTypeVideo}
+	ts := newTrackSynchronizer(sync, track)
+	ts.logger = logger.NewTestLogger(t)
+
+	firstReceivedAt := base
+	firstPacket := jitter.ExtPacket{
+		Packet: &rtp.Packet{
+			Header:  rtp.Header{Timestamp: 90000, SequenceNumber: 1},
+			Payload: []byte{0x01},
+		},
+		ReceivedAt: firstReceivedAt,
+	}
+
+	ts.initialize(firstPacket)
+	require.Less(t, ts.currentPTSOffset, time.Duration(0), "expected negative PTS offset when synchronizer start is later")
+
+	stepTS := ts.rtpConverter.toRTP(10 * time.Millisecond)
+	secondPacket := jitter.ExtPacket{
+		Packet: &rtp.Packet{
+			Header:  rtp.Header{Timestamp: firstPacket.Packet.Timestamp + stepTS, SequenceNumber: 2},
+			Payload: []byte{0x02},
+		},
+		ReceivedAt: firstReceivedAt.Add(10 * time.Millisecond),
+	}
+
+	adjusted, err := ts.GetPTS(secondPacket)
+	require.NoError(t, err)
+	require.Less(t, adjusted, time.Duration(0), "expected negative adjusted PTS")
+}
+
+func TestInitializeUsesBufferedPacketTiming(t *testing.T) {
+	sync := NewSynchronizerWithOptions()
+	syncStart := mono.Now().Add(-250 * time.Millisecond)
+	sync.getOrSetStartedAt(syncStart.UnixNano())
+
+	track := fakeTrack{id: "init", rate: 90000, kind: webrtc.RTPCodecTypeVideo}
+	ts := newTrackSynchronizer(sync, track)
+	ts.logger = logger.NewTestLogger(t)
+
+	receivedAt := time.Now().Add(-50 * time.Millisecond)
+	packet := &rtp.Packet{Header: rtp.Header{Timestamp: 42}}
+	extPkt := jitter.ExtPacket{Packet: packet, ReceivedAt: receivedAt}
+
+	ts.initialize(extPkt)
+
+	require.True(t, ts.initialized, "track should initialize")
+	require.Equal(t, receivedAt, ts.startTime, "startTime should match buffered arrival time")
+	require.Equal(t, packet.Timestamp, ts.startRTP, "startRTP should come from packet timestamp")
+	require.InDelta(t, receivedAt.Sub(syncStart), ts.currentPTSOffset, float64(5*time.Millisecond))
+}
+
 func TestGetPTSWithRebase_PropelsForward(t *testing.T) {
 	clock := uint32(48000)
 	ts := newTSForTests(t, clock, webrtc.RTPCodecTypeAudio)