Fix duration inflation caused by late-subscribing tracks (#859)

End() previously computed endedAt as wallclock + max PTS offset, which included the late-subscription base offset (e.g. +44s for a track that joined 44s late), inflating reported duration far beyond actual recording time. This replaces the offset-based computation with the actual max adjusted PTS reached by any track, making both endedAt and the drain ceiling directly reflect the media timeline.

Author: milos-lk

pkg/synchronizer/participant.go

@@ -43,21 +43,21 @@ func (p *participantSynchronizer) onSenderReport(pkt *rtcp.SenderReport) {
 	}
 }
 
-func (p *participantSynchronizer) getMaxOffset() time.Duration {
+func (p *participantSynchronizer) getMaxPTSAdjusted() time.Duration {
 	p.Lock()
 	defer p.Unlock()
 
-	var maxOffset time.Duration
+	var maxPTS time.Duration
 	for _, t := range p.tracks {
 		t.Lock()
-		o := max(t.currentPTSOffset, t.desiredPTSOffset)
+		pts := t.lastPTSAdjusted
 		t.Unlock()
 
-		if o > maxOffset {
-			maxOffset = o
+		if pts > maxPTS {
+			maxPTS = pts
 		}
 	}
-	return maxOffset
+	return maxPTS
 }
 
 func (p *participantSynchronizer) drain(maxPTS time.Duration) {

pkg/synchronizer/synchronizer.go

@@ -273,20 +273,20 @@ func (s *Synchronizer) OnRTCP(packet rtcp.Packet) {
 }
 
 func (s *Synchronizer) End() {
-	endTime := time.Now()
-
 	s.Lock()
 	defer s.Unlock()
 
-	// find the earliest time we can stop all tracks
-	var maxOffset time.Duration
+	// maxPTS is the drain ceiling: the maximum adjusted PTS after which tracks
+	// return EOF. Use the furthest adjusted PTS any track has actually reached
+	// so that all tracks can drain up to the same point in the output timeline.
+	var maxPTS time.Duration
 	for _, p := range s.psByIdentity {
-		if m := p.getMaxOffset(); m > maxOffset {
-			maxOffset = m
+		if m := p.getMaxPTSAdjusted(); m > maxPTS {
+			maxPTS = m
 		}
 	}
-	s.endedAt = endTime.Add(maxOffset).UnixNano()
-	maxPTS := time.Duration(s.endedAt - s.startedAt)
+
+	s.endedAt = s.startedAt + int64(maxPTS)
 
 	// drain all
 	for _, p := range s.psByIdentity {

pkg/synchronizer/synchronizer_test.go

@@ -65,6 +65,46 @@ func fakeVideo90k(ssrc uint32) *synchronizerfakes.FakeTrackRemote {
 
 // ---------- tests ----------
 
+// Regression: a late-subscribing track must not inflate the reported duration.
+func TestEnd_LateTrack_DoesNotInflateDuration(t *testing.T) {
+	s := synchronizer.NewSynchronizerWithOptions()
+
+	audio := fakeAudio48k(0xA001)
+	ts1 := s.AddTrack(audio, "p1")
+	ts1.Initialize(pkt(1000).Packet)
+	_, _ = ts1.GetPTS(pkt(1000))
+
+	// Push ~2s of audio (100 packets at 20ms)
+	step := uint32(48000 * 20 / 1000) // 960 ticks
+	cur := uint32(1000)
+	for i := 0; i < 100; i++ {
+		cur += step
+		_, _ = ts1.GetPTS(pkt(cur))
+	}
+
+	// Late track subscribes 3s after the first
+	time.Sleep(3 * time.Second)
+
+	late := fakeAudio48k(0xA002)
+	late.IDReturns("audio-late")
+	ts2 := s.AddTrack(late, "p2")
+	ts2.Initialize(pkt(5000).Packet)
+	_, _ = ts2.GetPTS(pkt(5000))
+
+	// Push a few packets on late track
+	cur2 := uint32(5000)
+	for i := 0; i < 10; i++ {
+		cur2 += step
+		_, _ = ts2.GetPTS(pkt(cur2))
+	}
+
+	s.End()
+
+	duration := time.Duration(s.GetEndedAt() - s.GetStartedAt())
+	require.Less(t, duration, 5*time.Second,
+		"duration must not be inflated by late track's base offset; got %v", duration)
+}
+
 // Initialize + same timestamp returns previous adjusted value (near zero right after init)
 func TestInitialize_AndSameTimestamp(t *testing.T) {
 	s := synchronizer.NewSynchronizerWithOptions(synchronizer.WithMaxTsDiff(50 * time.Millisecond))