Keep track of expected RTP time stamp and control drift. (#1681)

* Keep track of expected RTP time stamp and control drift.

- Use monotonic clock in RTCP Sender Report and packet times
- Keep the time stamp close to expected time stamp on layer/SSRC
  switches

* clean up

* fix test compile

* more test compile failures

pkg/sfu/buffer/buffer.go

@@ -27,17 +27,17 @@ import (
 )
 
 const (
-	ReportDelta = 1e9
+	ReportDelta = time.Second
 )
 
 type pendingPacket struct {
-	arrivalTime int64
+	arrivalTime time.Time
 	packet      []byte
 }
 
 type ExtPacket struct {
 	VideoLayer
-	Arrival              int64
+	Arrival              time.Time
 	Packet               *rtp.Packet
 	Payload              interface{}
 	KeyFrame             bool
@@ -58,7 +58,7 @@ type Buffer struct {
 	closeOnce     sync.Once
 	mediaSSRC     uint32
 	clockRate     uint32
-	lastReport    int64
+	lastReport    time.Time
 	twccExt       uint8
 	audioLevelExt uint8
 	bound         bool
@@ -163,7 +163,7 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
 	b.deltaStatsSnapshotId = b.rtpStats.NewSnapshotId()
 
 	b.clockRate = codec.ClockRate
-	b.lastReport = time.Now().UnixNano()
+	b.lastReport = time.Now()
 	b.mime = strings.ToLower(codec.MimeType)
 
 	for _, ext := range params.HeaderExtensions {
@@ -260,12 +260,12 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) {
 		copy(packet, pkt)
 		b.pPackets = append(b.pPackets, pendingPacket{
 			packet:      packet,
-			arrivalTime: time.Now().UnixNano(),
+			arrivalTime: time.Now(),
 		})
 		return
 	}
 
-	b.calc(pkt, time.Now().UnixNano())
+	b.calc(pkt, time.Now())
 	return
 }
 
@@ -392,7 +392,7 @@ func (b *Buffer) SetRTT(rtt uint32) {
 	}
 }
 
-func (b *Buffer) calc(pkt []byte, arrivalTime int64) {
+func (b *Buffer) calc(pkt []byte, arrivalTime time.Time) {
 	pktBuf, err := b.bucket.AddPacket(pkt)
 	if err != nil {
 		//
@@ -486,7 +486,7 @@ func (b *Buffer) doFpsCalc(ep *ExtPacket) {
 	}
 }
 
-func (b *Buffer) updateStreamState(p *rtp.Packet, arrivalTime int64) {
+func (b *Buffer) updateStreamState(p *rtp.Packet, arrivalTime time.Time) {
 	flowState := b.rtpStats.Update(&p.Header, len(p.Payload), int(p.PaddingSize), arrivalTime)
 
 	if b.nacker != nil {
@@ -500,12 +500,12 @@ func (b *Buffer) updateStreamState(p *rtp.Packet, arrivalTime int64) {
 	}
 }
 
-func (b *Buffer) processHeaderExtensions(p *rtp.Packet, arrivalTime int64) {
+func (b *Buffer) processHeaderExtensions(p *rtp.Packet, arrivalTime time.Time) {
 	// submit to TWCC even if it is a padding only packet. Clients use padding only packets as probes
 	// for bandwidth estimation
 	if b.twcc != nil && b.twccExt != 0 {
 		if ext := p.GetExtension(b.twccExt); ext != nil {
-			b.twcc.Push(binary.BigEndian.Uint16(ext[0:2]), arrivalTime, p.Marker)
+			b.twcc.Push(binary.BigEndian.Uint16(ext[0:2]), arrivalTime.UnixNano(), p.Marker)
 		}
 	}
 
@@ -530,7 +530,7 @@ func (b *Buffer) processHeaderExtensions(p *rtp.Packet, arrivalTime int64) {
 	}
 }
 
-func (b *Buffer) getExtPacket(rtpPacket *rtp.Packet, arrivalTime int64) *ExtPacket {
+func (b *Buffer) getExtPacket(rtpPacket *rtp.Packet, arrivalTime time.Time) *ExtPacket {
 	ep := &ExtPacket{
 		Packet:  rtpPacket,
 		Arrival: arrivalTime,
@@ -615,8 +615,8 @@ func (b *Buffer) doNACKs() {
 	}
 }
 
-func (b *Buffer) doReports(arrivalTime int64) {
-	timeDiff := arrivalTime - b.lastReport
+func (b *Buffer) doReports(arrivalTime time.Time) {
+	timeDiff := arrivalTime.Sub(b.lastReport)
 	if timeDiff < ReportDelta {
 		return
 	}

pkg/sfu/buffer/rtpstats.go

@@ -1,6 +1,7 @@
 package buffer
 
 import (
+	"errors"
 	"fmt"
 	"math"
 	"sync"
@@ -123,11 +124,15 @@ type RTPStats struct {
 	lastRRTime             time.Time
 	lastRR                 rtcp.ReceptionReport
 
-	highestTS   uint32
-	tsCycles    uint32
-	highestTime int64
+	extStartTS uint64
+	highestTS  uint32
+	tsCycles   uint32
 
-	lastTransit uint32
+	firstTime   time.Time
+	highestTime time.Time
+
+	lastTransit   uint32
+	lastJitterRTP uint32
 
 	bytes                uint64
 	headerBytes          uint64
@@ -180,6 +185,8 @@ type RTPStats struct {
 	firstSenderReportRTP     uint32
 	firstFeedSenderReportNTP mediatransportutil.NtpTime
 	firstFeedSenderReportRTP uint32
+	lastSRTime               time.Time
+	lastSRNTP                mediatransportutil.NtpTime
 
 	nextSnapshotId uint32
 	snapshots      map[uint32]*Snapshot
@@ -206,7 +213,7 @@ func (r *RTPStats) Seed(from *RTPStats) {
 	r.resyncOnNextPacket = from.resyncOnNextPacket
 
 	r.startTime = from.startTime
-	// do not clone endTime as a non-zero endTime indiacates an ended object
+	// do not clone endTime as a non-zero endTime indicates an ended object
 
 	r.extStartSN = from.extStartSN
 	r.highestSN = from.highestSN
@@ -216,11 +223,15 @@ func (r *RTPStats) Seed(from *RTPStats) {
 	r.lastRRTime = from.lastRRTime
 	r.lastRR = from.lastRR
 
+	r.extStartTS = from.extStartTS
 	r.highestTS = from.highestTS
 	r.tsCycles = from.tsCycles
+
+	r.firstTime = from.firstTime
 	r.highestTime = from.highestTime
 
 	r.lastTransit = from.lastTransit
+	r.lastJitterRTP = from.lastJitterRTP
 
 	r.bytes = from.bytes
 	r.headerBytes = from.headerBytes
@@ -280,6 +291,8 @@ func (r *RTPStats) Seed(from *RTPStats) {
 	r.firstSenderReportRTP = from.firstSenderReportRTP
 	r.firstFeedSenderReportNTP = from.firstFeedSenderReportNTP
 	r.firstFeedSenderReportRTP = from.firstFeedSenderReportRTP
+	r.lastSRTime = from.lastSRTime
+	r.lastSRNTP = from.lastSRNTP
 
 	r.nextSnapshotId = from.nextSnapshotId
 	for id, ss := range from.snapshots {
@@ -324,7 +337,7 @@ func (r *RTPStats) IsActive() bool {
 	return r.initialized && r.endTime.IsZero()
 }
 
-func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, packetTime int64) (flowState RTPFlowState) {
+func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, packetTime time.Time) (flowState RTPFlowState) {
 	r.lock.Lock()
 	defer r.lock.Unlock()
 
@@ -338,14 +351,17 @@ func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, pa
 
 		r.startTime = time.Now()
 
-		r.highestSN = rtph.SequenceNumber - 1
-		r.highestTS = rtph.Timestamp
-		r.highestTime = packetTime
-
 		r.extStartSN = uint32(rtph.SequenceNumber)
+		r.highestSN = rtph.SequenceNumber - 1
 		r.cycles = 0
+
+		r.extStartTS = uint64(rtph.Timestamp)
+		r.highestTS = rtph.Timestamp
 		r.tsCycles = 0
 
+		r.firstTime = packetTime
+		r.highestTime = packetTime
+
 		first = true
 
 		// initialize snapshots if any
@@ -378,7 +394,7 @@ func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, pa
 		}
 
 		// adjust start to account for out-of-order packets before a cycle completes
-		if !r.maybeAdjustStartSN(rtph, packetTime, pktSize, hdrSize, payloadSize) {
+		if !r.maybeAdjustStartSN(rtph, pktSize, hdrSize, payloadSize) {
 			if !r.isSnInfoLost(rtph.SequenceNumber) {
 				r.bytesDuplicate += pktSize
 				r.headerBytesDuplicate += hdrSize
@@ -447,7 +463,7 @@ func (r *RTPStats) ResyncOnNextPacket() {
 	r.resyncOnNextPacket = true
 }
 
-func (r *RTPStats) maybeAdjustStartSN(rtph *rtp.Header, packetTime int64, pktSize uint64, hdrSize uint64, payloadSize int) bool {
+func (r *RTPStats) maybeAdjustStartSN(rtph *rtp.Header, pktSize uint64, hdrSize uint64, payloadSize int) bool {
 	if (r.getExtHighestSN() - r.extStartSN + 1) >= (NumSequenceNumbers / 2) {
 		return false
 	}
@@ -501,7 +517,7 @@ func (r *RTPStats) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt uint32
 		return
 	}
 
-	rtt, err := mediatransportutil.GetRttMsFromReceiverReportOnly(&rr)
+	rtt, err := mediatransportutil.GetRttMs(&rr, r.lastSRNTP, r.lastSRTime)
 	if err == nil {
 		isRttChanged = rtt != r.rtt
 	} else {
@@ -795,6 +811,32 @@ func (r *RTPStats) GetRtcpSenderReportDataExt() *RTCPSenderReportDataExt {
 	}
 }
 
+func (r *RTPStats) GetExpectedRTPTimestamp(at time.Time) (uint32, error) {
+	r.lock.RLock()
+	defer r.lock.RUnlock()
+
+	if !r.initialized {
+		return 0, errors.New("uninitilaized")
+	}
+
+	timeDiff := at.Sub(r.firstTime)
+	rtpDiff := timeDiff.Nanoseconds() * int64(r.params.ClockRate) / 1e9
+	expectedExtRTP := r.extStartTS + uint64(rtpDiff)
+	r.logger.Debugw(
+		"expected RTP timestamp",
+		"firstTime", r.firstTime.String(),
+		"checkAt", at.String(),
+		"timeDiff", timeDiff,
+		"firstRTP", r.extStartTS,
+		"rtpDiff", rtpDiff,
+		"expectedExtRTP", expectedExtRTP,
+		"expectedRTP", uint32(expectedExtRTP),
+		"highestTS", r.highestTS,
+		"highestTime", r.highestTime.String(),
+	)
+	return uint32(expectedExtRTP), nil
+}
+
 func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, srDataExt *RTCPSenderReportDataExt) *rtcp.SenderReport {
 	r.lock.Lock()
 	defer r.lock.Unlock()
@@ -808,28 +850,27 @@ func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, srDataExt *RTCPSenderReportD
 		return nil
 	}
 
-	// NTP timestamp in sender report from publisher side could have a different base,
-	// i. e. although it should be wall clock at time of send, have observed instances of older timer.
-	// It is not possible to accurately calculate current time in the NTP time base of the publisher side.
-	// So, using a smoothed version of one way delay for use in sender reports.
-	now := time.Now()
+	// construct current time based on monotonic clock
+	timeSinceFirst := time.Since(r.firstTime)
+	now := r.firstTime.Add(timeSinceFirst)
 	nowNTP := mediatransportutil.ToNtpTime(now)
-	nowRTP := r.highestTS
 
-	isUsingSmoothed := true
-	smoothedLocalTimeOfLatestSenderReportNTP := srDataExt.SenderReportData.NTPTimestamp.Time().Add(srDataExt.SmoothedOWD)
-	if smoothedLocalTimeOfLatestSenderReportNTP.After(now) {
-		isUsingSmoothed = false
-		r.logger.Debugw("smoothed time of NTP is ahead",
-			"now", now,
-			"smoothed", smoothedLocalTimeOfLatestSenderReportNTP,
-			"diff", smoothedLocalTimeOfLatestSenderReportNTP.Sub(now),
+	expectedExtRTP := r.extStartTS + uint64(timeSinceFirst.Nanoseconds()*int64(r.params.ClockRate)/1e9)
+	if getExtTS(r.highestTS, r.tsCycles) > expectedExtRTP || now.Before(r.highestTime) {
+		r.logger.Debugw(
+			"anachronous sender report",
+			"firstTime", r.firstTime.String(),
+			"currentTime", now.String(),
+			"timSinceFirst", timeSinceFirst,
+			"extStartTS", r.extStartTS,
+			"highestExtRTP", getExtTS(r.highestTS, r.tsCycles),
+			"expectedExtRTP", expectedExtRTP,
 		)
-		nowRTP += uint32(now.Sub(time.Unix(0, r.highestTime)).Milliseconds() * int64(r.params.ClockRate) / 1000)
-	} else {
-		nowRTP = srDataExt.SenderReportData.RTPTimestamp + uint32(now.Sub(smoothedLocalTimeOfLatestSenderReportNTP).Milliseconds()*int64(r.params.ClockRate)/1000)
 	}
 
+	timeSinceHighest := time.Since(r.highestTime)
+	nowRTP := r.highestTS + uint32(timeSinceHighest.Nanoseconds()*int64(r.params.ClockRate)/1e9)
+
 	// TODO-REMOVE-AFTER-DEBUG
 	if r.firstSenderReportNTP == 0 {
 		r.firstSenderReportNTP = nowNTP
@@ -838,57 +879,50 @@ func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, srDataExt *RTCPSenderReportD
 		r.firstFeedSenderReportNTP = srDataExt.SenderReportData.NTPTimestamp
 		r.firstFeedSenderReportRTP = srDataExt.SenderReportData.RTPTimestamp
 	} else {
-		highestTime := time.Unix(0, r.highestTime)
 		ntpTime := nowNTP.Time()
 
-		ntpDiffLocal := ntpTime.Sub(highestTime)
+		ntpDiffLocal := ntpTime.Sub(r.highestTime)
 		rtpDiffLocal := int32(nowRTP - r.highestTS)
 		rtpOffsetLocal := int32(nowRTP - r.highestTS - uint32(ntpDiffLocal.Nanoseconds()*int64(r.params.ClockRate)/1e9))
 
-		ntpDiffSmoothed := ntpTime.Sub(smoothedLocalTimeOfLatestSenderReportNTP)
-		rtpDiffSmoothed := int32(nowRTP - srDataExt.SenderReportData.RTPTimestamp)
-		rtpOffsetSmoothed := int32(nowRTP - srDataExt.SenderReportData.RTPTimestamp - uint32(ntpDiffSmoothed.Nanoseconds()*int64(r.params.ClockRate)/1e9))
-
 		timeSinceFirst := nowNTP.Time().Sub(r.firstSenderReportNTP.Time())
 		rtpDiffSinceFirst := getExtTS(nowRTP, r.tsCycles) - getExtTS(r.firstSenderReportRTP, 0)
 		drift := int64(uint64(timeSinceFirst.Nanoseconds()*int64(r.params.ClockRate)/1e9) - rtpDiffSinceFirst)
-		driftTime := (float64(drift) * 1000) / float64(r.params.ClockRate)
+		driftMs := (float64(drift) * 1000) / float64(r.params.ClockRate)
 
 		feedTimeSinceFirst := srDataExt.SenderReportData.NTPTimestamp.Time().Sub(r.firstFeedSenderReportNTP.Time())
 		// using tsCycles for extending feed time stamp too
 		feedRtpDiffSinceFirst := getExtTS(srDataExt.SenderReportData.RTPTimestamp, r.tsCycles) - getExtTS(r.firstFeedSenderReportRTP, 0)
 		feedDrift := int64(uint64(feedTimeSinceFirst.Nanoseconds()*int64(r.params.ClockRate)/1e9) - feedRtpDiffSinceFirst)
-		feedDriftTime := (float64(feedDrift) * 1000) / float64(r.params.ClockRate)
+		feedDriftMs := (float64(feedDrift) * 1000) / float64(r.params.ClockRate)
 
 		r.logger.Debugw(
 			"sending sender report",
 			"highestTS", r.highestTS,
-			"highestTime", highestTime,
-			"smoothedTime", smoothedLocalTimeOfLatestSenderReportNTP,
+			"highestTime", r.highestTime.String(),
 			"reportTS", nowRTP,
-			"reportTime", ntpTime,
+			"reportTime", ntpTime.String(),
 			"rtpDiffLocal", rtpDiffLocal,
 			"ntpDiffLocal", ntpDiffLocal,
 			"rtpOffsetLocal", rtpOffsetLocal,
-			"rtpDiffSmoothed", rtpDiffSmoothed,
-			"ntpDiffSmoothed", ntpDiffSmoothed,
-			"rtpOffsetSmoothed", rtpOffsetSmoothed,
 			"timeSinceFirst", timeSinceFirst,
 			"rtpDiffSinceFirst", rtpDiffSinceFirst,
 			"drift", drift,
-			"driftTime(ms)", driftTime,
-			"smoothed", isUsingSmoothed,
+			"driftMs", driftMs,
 			"feedRTP", srDataExt.SenderReportData.RTPTimestamp,
-			"feedNTP", srDataExt.SenderReportData.NTPTimestamp.Time(),
-			"feedArrival", srDataExt.SenderReportData.ArrivalTime,
+			"feedNTP", srDataExt.SenderReportData.NTPTimestamp.Time().String(),
+			"feedArrival", srDataExt.SenderReportData.ArrivalTime.String(),
 			"smoothedOWD", srDataExt.SmoothedOWD,
 			"feedTimeSinceFirst", feedTimeSinceFirst,
 			"feedRtpDiffSinceFirst", feedRtpDiffSinceFirst,
 			"feedDrift", feedDrift,
-			"feedDriftTime(ms)", feedDriftTime,
+			"feedDriftMs", feedDriftMs,
 		)
 	}
 
+	r.lastSRTime = now
+	r.lastSRNTP = nowNTP
+
 	return &rtcp.SenderReport{
 		SSRC:        ssrc,
 		NTPTime:     uint64(nowNTP),
@@ -1400,8 +1434,20 @@ func (r *RTPStats) getIntervalStats(startInclusive uint16, endExclusive uint16)
 	return
 }
 
-func (r *RTPStats) updateJitter(rtph *rtp.Header, packetTime int64) {
-	packetTimeRTP := uint32(packetTime / 1e6 * int64(r.params.ClockRate/1e3))
+func (r *RTPStats) updateJitter(rtph *rtp.Header, packetTime time.Time) {
+	// Do not update jitter on multiple packets of same frame.
+	// All packets of a frame have the same time stamp.
+	// NOTE: This does not protect against using more than one packet of the same frame
+	//       if packets arrive out-of-order. For example,
+	//          p1f1 -> p1f2 -> p2f1
+	//       In this case, p2f1 (packet 2, frame 1) will still be used in jitter calculation
+	//       although it is the second packet of a frame because of out-of-order receival.
+	if r.lastJitterRTP == rtph.Timestamp {
+		return
+	}
+
+	timeSinceFirst := packetTime.Sub(r.firstTime)
+	packetTimeRTP := uint32(timeSinceFirst.Nanoseconds() * int64(r.params.ClockRate) / 1e9)
 	transit := packetTimeRTP - rtph.Timestamp
 
 	if r.lastTransit != 0 {
@@ -1422,6 +1468,7 @@ func (r *RTPStats) updateJitter(rtph *rtp.Header, packetTime int64) {
 	}
 
 	r.lastTransit = transit
+	r.lastJitterRTP = rtph.Timestamp
 }
 
 func (r *RTPStats) updateGapHistogram(gap int) {

pkg/sfu/buffer/rtpstats_test.go

@@ -43,7 +43,7 @@ func TestRTPStats(t *testing.T) {
 		timestamp += uint32(now.Sub(lastFrameTime).Seconds() * float64(clockRate))
 		for i := 0; i < packetsPerFrame; i++ {
 			packet := getPacket(sequenceNumber, timestamp, packetSize)
-			r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano())
+			r.Update(&packet.Header, len(packet.Payload), 0, time.Now())
 			if (sequenceNumber % 100) == 0 {
 				jump := uint16(rand.Float64() * 120.0)
 				sequenceNumber += jump
@@ -70,7 +70,7 @@ func TestRTPStats_Update(t *testing.T) {
 	sequenceNumber := uint16(rand.Float64() * float64(1<<16))
 	timestamp := uint32(rand.Float64() * float64(1<<32))
 	packet := getPacket(sequenceNumber, timestamp, 1000)
-	flowState := r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano())
+	flowState := r.Update(&packet.Header, len(packet.Payload), 0, time.Now())
 	require.False(t, flowState.HasLoss)
 	require.True(t, r.initialized)
 	require.Equal(t, sequenceNumber, r.highestSN)
@@ -80,14 +80,14 @@ func TestRTPStats_Update(t *testing.T) {
 	sequenceNumber++
 	timestamp += 3000
 	packet = getPacket(sequenceNumber, timestamp, 1000)
-	flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano())
+	flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now())
 	require.False(t, flowState.HasLoss)
 	require.Equal(t, sequenceNumber, r.highestSN)
 	require.Equal(t, timestamp, r.highestTS)
 
 	// out-of-order
 	packet = getPacket(sequenceNumber-10, timestamp-30000, 1000)
-	flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano())
+	flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now())
 	require.False(t, flowState.HasLoss)
 	require.Equal(t, sequenceNumber, r.highestSN)
 	require.Equal(t, timestamp, r.highestTS)
@@ -96,7 +96,7 @@ func TestRTPStats_Update(t *testing.T) {
 
 	// duplicate
 	packet = getPacket(sequenceNumber-10, timestamp-30000, 1000)
-	flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano())
+	flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now())
 	require.False(t, flowState.HasLoss)
 	require.Equal(t, sequenceNumber, r.highestSN)
 	require.Equal(t, timestamp, r.highestTS)
@@ -107,7 +107,7 @@ func TestRTPStats_Update(t *testing.T) {
 	sequenceNumber += 10
 	timestamp += 30000
 	packet = getPacket(sequenceNumber, timestamp, 1000)
-	flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano())
+	flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now())
 	require.True(t, flowState.HasLoss)
 	require.Equal(t, sequenceNumber-9, flowState.LossStartInclusive)
 	require.Equal(t, sequenceNumber, flowState.LossEndExclusive)
@@ -115,7 +115,7 @@ func TestRTPStats_Update(t *testing.T) {
 
 	// out-of-order should decrement number of lost packets
 	packet = getPacket(sequenceNumber-15, timestamp-45000, 1000)
-	flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now().UnixNano())
+	flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now())
 	require.False(t, flowState.HasLoss)
 	require.Equal(t, sequenceNumber, r.highestSN)
 	require.Equal(t, timestamp, r.highestTS)

pkg/sfu/downtrack.go

@@ -276,7 +276,12 @@ func NewDownTrack(
 		kind:           kind,
 		codec:          codecs[0].RTPCodecCapability,
 	}
-	d.forwarder = NewForwarder(d.kind, d.logger, d.receiver.GetReferenceLayerRTPTimestamp)
+	d.forwarder = NewForwarder(
+		d.kind,
+		d.logger,
+		d.receiver.GetReferenceLayerRTPTimestamp,
+		d.getExpectedRTPTimestamp,
+	)
 	d.forwarder.OnParkedLayerExpired(func() {
 		if sal := d.getStreamAllocatorListener(); sal != nil {
 			sal.OnSubscriptionChanged(d)
@@ -638,7 +643,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error {
 		}
 	}
 
-	d.rtpStats.Update(hdr, len(payload), 0, time.Now().UnixNano())
+	d.rtpStats.Update(hdr, len(payload), 0, extPkt.Arrival)
 	return nil
 }
 
@@ -717,7 +722,7 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool) int {
 		}
 
 		if !paddingOnMute {
-			d.rtpStats.Update(&hdr, 0, len(payload), time.Now().UnixNano())
+			d.rtpStats.Update(&hdr, 0, len(payload), time.Now())
 		}
 
 		//
@@ -1214,7 +1219,7 @@ func (d *DownTrack) writeOpusBlankFrame(hdr *rtp.Header, frameEndNeeded bool) (i
 
 	_, err := d.writeStream.WriteRTP(hdr, payload)
 	if err == nil {
-		d.rtpStats.Update(hdr, len(payload), 0, time.Now().UnixNano())
+		d.rtpStats.Update(hdr, len(payload), 0, time.Now())
 	}
 	return hdr.MarshalSize() + len(payload), err
 }
@@ -1233,7 +1238,7 @@ func (d *DownTrack) writeOpusRedBlankFrame(hdr *rtp.Header, frameEndNeeded bool)
 
 	_, err := d.writeStream.WriteRTP(hdr, payload)
 	if err == nil {
-		d.rtpStats.Update(hdr, len(payload), 0, time.Now().UnixNano())
+		d.rtpStats.Update(hdr, len(payload), 0, time.Now())
 	}
 	return hdr.MarshalSize() + len(payload), err
 }
@@ -1254,7 +1259,7 @@ func (d *DownTrack) writeVP8BlankFrame(hdr *rtp.Header, frameEndNeeded bool) (in
 
 	_, err = d.writeStream.WriteRTP(hdr, payload)
 	if err == nil {
-		d.rtpStats.Update(hdr, len(payload), 0, time.Now().UnixNano())
+		d.rtpStats.Update(hdr, len(payload), 0, time.Now())
 	}
 	return hdr.MarshalSize() + len(payload), err
 }
@@ -1276,7 +1281,7 @@ func (d *DownTrack) writeH264BlankFrame(hdr *rtp.Header, frameEndNeeded bool) (i
 	payload := buf[:offset]
 	_, err := d.writeStream.WriteRTP(hdr, payload)
 	if err == nil {
-		d.rtpStats.Update(hdr, len(payload), 0, time.Now().UnixNano())
+		d.rtpStats.Update(hdr, len(payload), 0, time.Now())
 	}
 	return hdr.MarshalSize() + offset, err
 }
@@ -1499,7 +1504,7 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) {
 			d.streamAllocatorBytesCounter.Add(uint32(pkt.Header.MarshalSize() + len(payload)))
 			d.bytesRetransmitted.Add(uint32(pkt.Header.MarshalSize() + len(payload)))
 
-			d.rtpStats.Update(&pkt.Header, len(payload), 0, time.Now().UnixNano())
+			d.rtpStats.Update(&pkt.Header, len(payload), 0, time.Now())
 		}
 	}
 
@@ -1624,6 +1629,10 @@ func (d *DownTrack) DebugInfo() map[string]interface{} {
 	}
 }
 
+func (d *DownTrack) getExpectedRTPTimestamp(at time.Time) (uint32, error) {
+	return d.rtpStats.GetExpectedRTPTimestamp(at)
+}
+
 func (d *DownTrack) GetConnectionScoreAndQuality() (float32, livekit.ConnectionQuality) {
 	return d.connectionStats.GetScoreAndQuality()
 }

pkg/sfu/forwarder.go

@@ -158,6 +158,7 @@ type Forwarder struct {
 	kind                          webrtc.RTPCodecType
 	logger                        logger.Logger
 	getReferenceLayerRTPTimestamp func(ts uint32, layer int32, referenceLayer int32) (uint32, error)
+	getExpectedRTPTimestamp       func(at time.Time) (uint32, error)
 
 	muted    bool
 	pubMuted bool
@@ -185,11 +186,13 @@ func NewForwarder(
 	kind webrtc.RTPCodecType,
 	logger logger.Logger,
 	getReferenceLayerRTPTimestamp func(ts uint32, layer int32, referenceLayer int32) (uint32, error),
+	getExpectedRTPTimestamp func(at time.Time) (uint32, error),
 ) *Forwarder {
 	f := &Forwarder{
 		kind:                          kind,
 		logger:                        logger,
 		getReferenceLayerRTPTimestamp: getReferenceLayerRTPTimestamp,
+		getExpectedRTPTimestamp:       getExpectedRTPTimestamp,
 		referenceLayerSpatial:         buffer.InvalidLayerSpatial,
 		lastAllocation:                VideoAllocationDefault,
 		rtpMunger:                     NewRTPMunger(logger),
@@ -1446,26 +1449,46 @@ func (f *Forwarder) getTranslationParamsCommon(extPkt *buffer.ExtPacket, layer i
 
 			// Compute how much time passed between the old RTP extPkt
 			// and the current packet, and fix timestamp on source change
-			td := uint32(1)
+			//
+			// There are three time stamps to consider here
+			//   1. lastTS -> time stamp of last sent packet
+			//   2. refTS -> time stamp of this packet (after munging) calculated using feed's RTCP sender report
+			//   3. expectedTS -> time stamp of this packet (after munging) calculated using this stream's RTCP sender report
+			// Ideally, refTS and expectedTS should be very close and lastTS should be before both of those.
+			// But, cases like muting/unmuting, clock vagaries make them not satisfy those conditions always.
+			//
+			// There are 6 orderings to consider (considering only inequalities). Resolve them using following rules
+			//   1. Timestamp has to move forward
+			//   2. Keep next time stamp close to expected
+			lastTS := f.rtpMunger.GetLast().LastTS
+			refTS := lastTS
+			expectedTS := lastTS
+			switchingAt := time.Now()
 			if f.getReferenceLayerRTPTimestamp != nil {
-				refTS, err := f.getReferenceLayerRTPTimestamp(extPkt.Packet.Timestamp, layer, f.referenceLayerSpatial)
+				ts, err := f.getReferenceLayerRTPTimestamp(extPkt.Packet.Timestamp, layer, f.referenceLayerSpatial)
 				if err == nil {
-					last := f.rtpMunger.GetLast()
-					td = refTS - last.LastTS
-					if td == 0 || td > (1<<31) {
-						f.logger.Debugw("reference timestamp out-of-order, using default", "lastTS", last.LastTS, "refTS", refTS, "td", int32(td))
-						td = 1
-					} else if td > uint32(0.5*float32(f.codec.ClockRate)) {
-						// log jumps greater than 0.5 seconds
-						f.logger.Debugw("reference timestamp too far ahead", "lastTS", last.LastTS, "refTS", refTS, "td", td)
-					}
-					f.logger.Debugw("reference timestamp on switch", "lastTS", last.LastTS, "refTS", refTS, "td", int32(td), "switchingAt", time.Now())
-				} else {
-					f.logger.Debugw("reference timestamp get error, using default", "error", err)
+					refTS = ts
 				}
 			}
+			if f.getExpectedRTPTimestamp != nil {
+				ts, err := f.getExpectedRTPTimestamp(switchingAt)
+				if err == nil {
+					expectedTS = ts
+				}
+			}
+			nextTS, explain := getNextTimestamp(lastTS, refTS, expectedTS)
+			f.logger.Debugw(
+				"next timestamp on switch",
+				"switchingAt", switchingAt.String(),
+				"lastTS", lastTS,
+				"refTS", refTS,
+				"expectedTS", expectedTS,
+				"nextTS", nextTS,
+				"jump", nextTS-lastTS,
+				"explanation", explain,
+			)
 
-			f.rtpMunger.UpdateSnTsOffsets(extPkt, 1, td)
+			f.rtpMunger.UpdateSnTsOffsets(extPkt, 1, nextTS-lastTS)
 			f.codecMunger.UpdateOffsets(extPkt)
 		}
 
@@ -1734,3 +1757,40 @@ done:
 
 	return float64(distance) / float64(maxSeenLayer.Temporal+1)
 }
+
+func getNextTimestamp(lastTS uint32, refTS uint32, expectedTS uint32) (uint32, string) {
+	isInOrder := func(val1, val2 uint32) bool {
+		diff := val1 - val2
+		return diff != 0 && diff < (1<<31)
+	}
+
+	rl := isInOrder(refTS, lastTS)
+	el := isInOrder(expectedTS, lastTS)
+	er := isInOrder(expectedTS, refTS)
+
+	nextTS := lastTS + 1
+	explain := "l = r = e"
+
+	switch {
+	case rl && el && er: // lastTS < refTS < expectedTS
+		nextTS = uint32(float64(refTS) + 0.95*float64(expectedTS-refTS))
+		explain = fmt.Sprintf("l < r < e, %d, %d", refTS-lastTS, expectedTS-refTS)
+	case rl && el && !er: // lastTS < expectedTS < refTS
+		nextTS = uint32(float64(expectedTS) + 0.5*float64(refTS-expectedTS))
+		explain = fmt.Sprintf("l < e < r, %d, %d", expectedTS-lastTS, refTS-expectedTS)
+	case !rl && el && er: // refTS < lastTS < expectedTS
+		nextTS = uint32(float64(lastTS) + 0.5*float64(expectedTS-lastTS))
+		explain = fmt.Sprintf("r < l < e, %d, %d", lastTS-refTS, expectedTS-lastTS)
+	case !rl && !el && er: // refTS < expectedTS < lastTS
+		nextTS = lastTS + 1
+		explain = fmt.Sprintf("r < e < l, %d, %d", expectedTS-refTS, lastTS-expectedTS)
+	case rl && !el && !er: // expectedTS < lastTS < refTS
+		nextTS = uint32(float64(lastTS) + 0.5*float64(refTS-lastTS))
+		explain = fmt.Sprintf("e < l < r, %d, %d", lastTS-expectedTS, refTS-lastTS)
+	case !rl && !el && !er: // expectedTS < refTS < lastTS
+		nextTS = lastTS + 1
+		explain = fmt.Sprintf("e < r < l, %d, %d", refTS-expectedTS, lastTS-refTS)
+	}
+
+	return nextTS, explain
+}

pkg/sfu/forwarder_test.go

@@ -18,7 +18,7 @@ func disable(f *Forwarder) {
 }
 
 func newForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType) *Forwarder {
-	f := NewForwarder(kind, logger.GetLogger(), nil)
+	f := NewForwarder(kind, logger.GetLogger(), nil, nil)
 	f.DetermineCodec(codec, nil)
 	return f
 }

pkg/sfu/streamtrackermanager.go

@@ -529,8 +529,9 @@ func (s *StreamTrackerManager) GetReferenceLayerRTPTimestamp(ts uint32, layer in
 	// NOTE: It is possible that reference layer has stopped (due to dynacast/adaptive streaming OR publisher
 	// constraints). It should be okay even if the layer has stopped for a long time when using modulo arithmetic for
 	// RTP time stamp (uint32 arithmetic).
-	ntpDiff := float64(int64(srRef.SenderReportData.NTPTimestamp-srLayer.SenderReportData.NTPTimestamp)) / float64(1<<32)
-	normalizedTS := srLayer.SenderReportData.RTPTimestamp + uint32(ntpDiff*float64(s.clockRate))
+	ntpDiff := srRef.SenderReportData.NTPTimestamp.Time().Sub(srLayer.SenderReportData.NTPTimestamp.Time())
+	rtpDiff := ntpDiff.Nanoseconds() * int64(s.clockRate) / 1e9
+	normalizedTS := srLayer.SenderReportData.RTPTimestamp + uint32(rtpDiff)
 
 	// now that both RTP timestamps correspond to roughly the same NTP time,
 	// the diff between them is the offset in RTP timestamp units between layer and referenceLayer.

pkg/sfu/testutils/data.go

@@ -1,6 +1,8 @@
 package testutils
 
 import (
+	"time"
+
 	"github.com/pion/rtp"
 	"github.com/pion/webrtc/v3"
 
@@ -18,7 +20,7 @@ type TestExtPacketParams struct {
 	SSRC           uint32
 	PayloadSize    int
 	PaddingSize    byte
-	ArrivalTime    int64
+	ArrivalTime    time.Time
 	VideoLayer     buffer.VideoLayer
 }