Cache RTCP sender report in forwarder state. (#2994)

* Cache RTCP sender report in forwarder state.

To be used in migration.

TODO: need to check more places to operate pure in unix nano rather than
converting.

* match name

pkg/sfu/buffer/buffer.go

@@ -35,7 +35,6 @@ import (
 	dd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/dependencydescriptor"
 	"github.com/livekit/livekit-server/pkg/sfu/utils"
 	sutils "github.com/livekit/livekit-server/pkg/utils"
-	"github.com/livekit/mediatransportutil"
 	"github.com/livekit/mediatransportutil/pkg/bucket"
 	"github.com/livekit/mediatransportutil/pkg/nack"
 	"github.com/livekit/mediatransportutil/pkg/twcc"
@@ -935,12 +934,12 @@ func (b *Buffer) buildReceptionReport() *rtcp.ReceptionReport {
 
 func (b *Buffer) SetSenderReportData(rtpTime uint32, ntpTime uint64, packets uint32, octets uint32) {
 	b.RLock()
-	srData := &RTCPSenderReportData{
-		RTPTimestamp: rtpTime,
-		NTPTimestamp: mediatransportutil.NtpTime(ntpTime),
-		At:           b.getMonotonicNow(),
+	srData := &livekit.RTCPSenderReportState{
+		RtpTimestamp: rtpTime,
+		NtpTimestamp: ntpTime,
+		At:           b.getMonotonicNowUnixNano(),
 		Packets:      packets,
-		Octets:       octets,
+		Octets:       uint64(octets),
 	}
 
 	didSet := false
@@ -956,7 +955,7 @@ func (b *Buffer) SetSenderReportData(rtpTime uint32, ntpTime uint64, packets uin
 	}
 }
 
-func (b *Buffer) GetSenderReportData() *RTCPSenderReportData {
+func (b *Buffer) GetSenderReportData() *livekit.RTCPSenderReportState {
 	b.RLock()
 	defer b.RUnlock()
 
@@ -1126,10 +1125,6 @@ func (b *Buffer) getMonotonicNowUnixNano() int64 {
 	return b.baseTime.Add(time.Since(b.baseTime)).UnixNano()
 }
 
-func (b *Buffer) getMonotonicNow() time.Time {
-	return b.baseTime.Add(time.Since(b.baseTime))
-}
-
 // ---------------------------------------------------------------
 
 // SVC-TODO: Have to use more conditions to differentiate between

pkg/sfu/buffer/rtpstats_base.go

@@ -21,6 +21,7 @@ import (
 	"time"
 
 	"go.uber.org/zap/zapcore"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/livekit/mediatransportutil"
@@ -112,53 +113,56 @@ type snapshot struct {
 
 // ------------------------------------------------------------------
 
-type RTCPSenderReportData struct {
-	RTPTimestamp    uint32
-	RTPTimestampExt uint64
-	NTPTimestamp    mediatransportutil.NtpTime
-	At              time.Time
-	AtAdjusted      time.Time
-	Packets         uint32
-	Octets          uint32
+type wrappedRTPDriftLogger struct {
+	*livekit.RTPDrift
 }
 
-func (r *RTCPSenderReportData) PropagationDelay(passThrough bool) time.Duration {
+func (w wrappedRTPDriftLogger) MarshalLogObject(e zapcore.ObjectEncoder) error {
+	rd := w.RTPDrift
+	if rd == nil {
+		return nil
+	}
+
+	e.AddTime("StartTime", rd.StartTime.AsTime())
+	e.AddTime("EndTime", rd.EndTime.AsTime())
+	e.AddFloat64("Duration", rd.Duration)
+	e.AddUint64("StartTimestamp", rd.StartTimestamp)
+	e.AddUint64("EndTimestamp", rd.EndTimestamp)
+	e.AddUint64("RtpClockTicks", rd.RtpClockTicks)
+	e.AddInt64("DriftSamples", rd.DriftSamples)
+	e.AddFloat64("DriftMs", rd.DriftMs)
+	e.AddFloat64("ClockRate", rd.ClockRate)
+	return nil
+}
+
+// ------------------------------------------------------------------
+
+type WrappedRTCPSenderReportStateLogger struct {
+	*livekit.RTCPSenderReportState
+}
+
+func (w WrappedRTCPSenderReportStateLogger) MarshalLogObject(e zapcore.ObjectEncoder) error {
+	rsrs := w.RTCPSenderReportState
+	if rsrs == nil {
+		return nil
+	}
+
+	e.AddUint32("RtpTimestamp", rsrs.RtpTimestamp)
+	e.AddUint64("RtpTimestampExt", rsrs.RtpTimestampExt)
+	e.AddTime("NtpTimestamp", mediatransportutil.NtpTime(rsrs.NtpTimestamp).Time())
+	e.AddTime("At", time.Unix(0, rsrs.At))
+	e.AddTime("AtAdjusted", time.Unix(0, rsrs.AtAdjusted))
+	e.AddUint32("Packets", rsrs.Packets)
+	e.AddUint64("Octets", rsrs.Octets)
+	return nil
+}
+
+func RTCPSenderReportPropagationDelay(rsrs *livekit.RTCPSenderReportState, passThrough bool) time.Duration {
 	if passThrough {
 		return 0
 	}
 
-	return r.AtAdjusted.Sub(r.NTPTimestamp.Time())
-}
-
-func (r *RTCPSenderReportData) ToString() string {
-	if r == nil {
-		return ""
-	}
-
-	return fmt.Sprintf("ntp: %s, rtp: %d, extRtp: %d, at: %s, atAdj: %s, p: %d, o: %d",
-		r.NTPTimestamp.Time().String(),
-		r.RTPTimestamp,
-		r.RTPTimestampExt,
-		r.At.String(),
-		r.AtAdjusted.String(),
-		r.Packets,
-		r.Octets,
-	)
-}
-
-func (r *RTCPSenderReportData) MarshalLogObject(e zapcore.ObjectEncoder) error {
-	if r == nil {
-		return nil
-	}
-
-	e.AddTime("NTPTimestamp", r.NTPTimestamp.Time())
-	e.AddUint32("RTPTimestamp", r.RTPTimestamp)
-	e.AddUint64("RTPTimestampExt", r.RTPTimestampExt)
-	e.AddTime("At", r.At)
-	e.AddTime("AtAdjusted", r.AtAdjusted)
-	e.AddUint32("Packets", r.Packets)
-	e.AddUint32("Octets", r.Octets)
-	return nil
+	return time.Unix(0, rsrs.AtAdjusted).Sub(mediatransportutil.NtpTime(rsrs.NtpTimestamp).Time())
 }
 
 // ------------------------------------------------------------------
@@ -226,8 +230,8 @@ type rtpStatsBase struct {
 	rtt    uint32
 	maxRtt uint32
 
-	srFirst  *RTCPSenderReportData
-	srNewest *RTCPSenderReportData
+	srFirst  *livekit.RTCPSenderReportState
+	srNewest *livekit.RTCPSenderReportState
 
 	nextSnapshotID uint32
 	snapshots      []snapshot
@@ -298,18 +302,8 @@ func (r *rtpStatsBase) seed(from *rtpStatsBase) bool {
 	r.rtt = from.rtt
 	r.maxRtt = from.maxRtt
 
-	if from.srFirst != nil {
-		srFirst := *from.srFirst
-		r.srFirst = &srFirst
-	} else {
-		r.srFirst = nil
-	}
-	if from.srNewest != nil {
-		srNewest := *from.srNewest
-		r.srNewest = &srNewest
-	} else {
-		r.srNewest = nil
-	}
+	r.srFirst = proto.Clone(from.srFirst).(*livekit.RTCPSenderReportState)
+	r.srNewest = proto.Clone(from.srNewest).(*livekit.RTCPSenderReportState)
 
 	r.nextSnapshotID = from.nextSnapshotID
 	r.snapshots = make([]snapshot, cap(from.snapshots))
@@ -510,7 +504,7 @@ func (r *rtpStatsBase) GetRtt() uint32 {
 	return r.rtt
 }
 
-func (r *rtpStatsBase) maybeAdjustFirstPacketTime(srData *RTCPSenderReportData, tsOffset uint64, extStartTS uint64) (err error, loggingFields []interface{}) {
+func (r *rtpStatsBase) maybeAdjustFirstPacketTime(srData *livekit.RTCPSenderReportState, tsOffset uint64, extStartTS uint64) (err error, loggingFields []interface{}) {
 	if time.Since(r.startTime) > cFirstPacketTimeAdjustWindow {
 		return
 	}
@@ -521,8 +515,8 @@ func (r *rtpStatsBase) maybeAdjustFirstPacketTime(srData *RTCPSenderReportData,
 	// abnormal delay (maybe due to pacing or maybe due to queuing
 	// in some network element along the way), push back first time
 	// to an earlier instance.
-	timeSinceReceive := time.Since(srData.AtAdjusted)
-	extNowTS := srData.RTPTimestampExt - tsOffset + uint64(timeSinceReceive.Nanoseconds()*int64(r.params.ClockRate)/1e9)
+	timeSinceReceive := time.Since(time.Unix(0, srData.AtAdjusted))
+	extNowTS := srData.RtpTimestampExt - tsOffset + uint64(timeSinceReceive.Nanoseconds()*int64(r.params.ClockRate)/1e9)
 	samplesDiff := int64(extNowTS - extStartTS)
 	if samplesDiff < 0 {
 		// out-of-order, skip
@@ -543,7 +537,7 @@ func (r *rtpStatsBase) maybeAdjustFirstPacketTime(srData *RTCPSenderReportData,
 			"adjustment", time.Duration(r.firstTime - firstTime).String(),
 			"extNowTS", extNowTS,
 			"extStartTS", extStartTS,
-			"srData", srData,
+			"srData", WrappedRTCPSenderReportStateLogger{srData},
 			"tsOffset", tsOffset,
 			"timeSinceReceive", timeSinceReceive.String(),
 			"timeSinceFirst", timeSinceFirst.String(),
@@ -730,8 +724,8 @@ func (r *rtpStatsBase) MarshalLogObject(e zapcore.ObjectEncoder) error {
 	e.AddUint32("rtt", r.rtt)
 	e.AddUint32("maxRtt", r.maxRtt)
 
-	e.AddObject("srFirst", r.srFirst)
-	e.AddObject("srNewest", r.srNewest)
+	e.AddObject("srFirst", WrappedRTCPSenderReportStateLogger{r.srFirst})
+	e.AddObject("srNewest", WrappedRTCPSenderReportStateLogger{r.srNewest})
 	return nil
 }
 
@@ -800,7 +794,12 @@ func (r *rtpStatsBase) toString(
 	str += ", rtt(ms):"
 	str += fmt.Sprintf("%d|%d", p.RttCurrent, p.RttMax)
 
-	str += fmt.Sprintf(", pd: %s, nrd: %s, rrd: %s", RTPDriftToString(p.PacketDrift), RTPDriftToString(p.ReportDrift), RTPDriftToString(p.RebasedReportDrift))
+	str += fmt.Sprintf(", pd: %s, nrd: %s, rxrd: %s, rbrd: %s",
+		RTPDriftToString(p.PacketDrift),
+		RTPDriftToString(p.NtpReportDrift),
+		RTPDriftToString(p.ReceivedReportDrift),
+		RTPDriftToString(p.RebasedReportDrift),
+	)
 	return str
 }
 
@@ -841,7 +840,7 @@ func (r *rtpStatsBase) toProto(
 	jitterTime := jitter / float64(r.params.ClockRate) * 1e6
 	maxJitterTime := maxJitter / float64(r.params.ClockRate) * 1e6
 
-	packetDrift, ntpReportDrift, rebasedReportDrift := r.getDrift(extStartTS, extHighestTS)
+	packetDrift, ntpReportDrift, receivedReportDrift, rebasedReportDrift := r.getDrift(extStartTS, extHighestTS)
 
 	p := &livekit.RTPStats{
 		StartTime:            timestamppb.New(r.startTime),
@@ -885,8 +884,9 @@ func (r *rtpStatsBase) toProto(
 		RttCurrent:           r.rtt,
 		RttMax:               r.maxRtt,
 		PacketDrift:          packetDrift,
-		ReportDrift:          ntpReportDrift,
+		NtpReportDrift:       ntpReportDrift,
 		RebasedReportDrift:   rebasedReportDrift,
+		ReceivedReportDrift:  receivedReportDrift,
 	}
 
 	gapsPresent := false
@@ -968,7 +968,7 @@ func (r *rtpStatsBase) getAndResetSnapshot(snapshotID uint32, extStartSN uint64,
 	return &then, &now
 }
 
-func (r *rtpStatsBase) getDrift(extStartTS, extHighestTS uint64) (packetDrift *livekit.RTPDrift, ntpReportDrift *livekit.RTPDrift, rebasedReportDrift *livekit.RTPDrift) {
+func (r *rtpStatsBase) getDrift(extStartTS, extHighestTS uint64) (packetDrift *livekit.RTPDrift, ntpReportDrift *livekit.RTPDrift, receivedReportDrift *livekit.RTPDrift, rebasedReportDrift *livekit.RTPDrift) {
 	if r.firstTime != 0 {
 		elapsed := r.highestTime - r.firstTime
 		rtpClockTicks := extHighestTS - extStartTS
@@ -989,18 +989,18 @@ func (r *rtpStatsBase) getDrift(extStartTS, extHighestTS uint64) (packetDrift *l
 		}
 	}
 
-	if r.srFirst != nil && r.srNewest != nil && r.srFirst.RTPTimestamp != r.srNewest.RTPTimestamp {
-		rtpClockTicks := r.srNewest.RTPTimestampExt - r.srFirst.RTPTimestampExt
+	if r.srFirst != nil && r.srNewest != nil && r.srFirst.RtpTimestamp != r.srNewest.RtpTimestamp {
+		rtpClockTicks := r.srNewest.RtpTimestampExt - r.srFirst.RtpTimestampExt
 
-		elapsed := r.srNewest.NTPTimestamp.Time().Sub(r.srFirst.NTPTimestamp.Time())
+		elapsed := mediatransportutil.NtpTime(r.srNewest.NtpTimestamp).Time().Sub(mediatransportutil.NtpTime(r.srFirst.NtpTimestamp).Time())
 		if elapsed.Seconds() > 0.0 {
 			driftSamples := int64(rtpClockTicks - uint64(elapsed.Nanoseconds()*int64(r.params.ClockRate)/1e9))
 			ntpReportDrift = &livekit.RTPDrift{
-				StartTime:      timestamppb.New(r.srFirst.NTPTimestamp.Time()),
-				EndTime:        timestamppb.New(r.srNewest.NTPTimestamp.Time()),
+				StartTime:      timestamppb.New(mediatransportutil.NtpTime(r.srFirst.NtpTimestamp).Time()),
+				EndTime:        timestamppb.New(mediatransportutil.NtpTime(r.srNewest.NtpTimestamp).Time()),
 				Duration:       elapsed.Seconds(),
-				StartTimestamp: r.srFirst.RTPTimestampExt,
-				EndTimestamp:   r.srNewest.RTPTimestampExt,
+				StartTimestamp: r.srFirst.RtpTimestampExt,
+				EndTimestamp:   r.srNewest.RtpTimestampExt,
 				RtpClockTicks:  rtpClockTicks,
 				DriftSamples:   driftSamples,
 				DriftMs:        (float64(driftSamples) * 1000) / float64(r.params.ClockRate),
@@ -1008,15 +1008,31 @@ func (r *rtpStatsBase) getDrift(extStartTS, extHighestTS uint64) (packetDrift *l
 			}
 		}
 
-		elapsed = r.srNewest.AtAdjusted.Sub(r.srFirst.AtAdjusted)
+		elapsed = time.Duration(r.srNewest.At - r.srFirst.At)
+		if elapsed.Seconds() > 0.0 {
+			driftSamples := int64(rtpClockTicks - uint64(elapsed.Nanoseconds()*int64(r.params.ClockRate)/1e9))
+			receivedReportDrift = &livekit.RTPDrift{
+				StartTime:      timestamppb.New(time.Unix(0, r.srFirst.At)),
+				EndTime:        timestamppb.New(time.Unix(0, r.srNewest.At)),
+				Duration:       elapsed.Seconds(),
+				StartTimestamp: r.srFirst.RtpTimestampExt,
+				EndTimestamp:   r.srNewest.RtpTimestampExt,
+				RtpClockTicks:  rtpClockTicks,
+				DriftSamples:   driftSamples,
+				DriftMs:        (float64(driftSamples) * 1000) / float64(r.params.ClockRate),
+				ClockRate:      float64(rtpClockTicks) / elapsed.Seconds(),
+			}
+		}
+
+		elapsed = time.Duration(r.srNewest.AtAdjusted - r.srFirst.AtAdjusted)
 		if elapsed.Seconds() > 0.0 {
 			driftSamples := int64(rtpClockTicks - uint64(elapsed.Nanoseconds()*int64(r.params.ClockRate)/1e9))
 			rebasedReportDrift = &livekit.RTPDrift{
-				StartTime:      timestamppb.New(r.srFirst.AtAdjusted),
-				EndTime:        timestamppb.New(r.srNewest.AtAdjusted),
+				StartTime:      timestamppb.New(time.Unix(0, r.srFirst.AtAdjusted)),
+				EndTime:        timestamppb.New(time.Unix(0, r.srNewest.AtAdjusted)),
 				Duration:       elapsed.Seconds(),
-				StartTimestamp: r.srFirst.RTPTimestampExt,
-				EndTimestamp:   r.srNewest.RTPTimestampExt,
+				StartTimestamp: r.srFirst.RtpTimestampExt,
+				EndTimestamp:   r.srNewest.RtpTimestampExt,
 				RtpClockTicks:  rtpClockTicks,
 				DriftSamples:   driftSamples,
 				DriftMs:        (float64(driftSamples) * 1000) / float64(r.params.ClockRate),

pkg/sfu/buffer/rtpstats_receiver.go

@@ -21,8 +21,10 @@ import (
 
 	"github.com/pion/rtcp"
 	"go.uber.org/zap/zapcore"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/livekit/livekit-server/pkg/sfu/utils"
+	"github.com/livekit/mediatransportutil"
 	"github.com/livekit/protocol/livekit"
 	protoutils "github.com/livekit/protocol/utils"
 )
@@ -372,23 +374,23 @@ func (r *RTPStatsReceiver) Update(
 	return
 }
 
-func (r *RTPStatsReceiver) getExtendedSenderReport(srData *RTCPSenderReportData) *RTCPSenderReportData {
+func (r *RTPStatsReceiver) getExtendedSenderReport(srData *livekit.RTCPSenderReportState) *livekit.RTCPSenderReportState {
 	tsCycles := uint64(0)
 	if r.srNewest != nil {
 		// use time since last sender report to ensure long gaps where the time stamp might
 		// jump more than half the range
-		timeSinceLastReport := srData.NTPTimestamp.Time().Sub(r.srNewest.NTPTimestamp.Time())
-		expectedRTPTimestampExt := r.srNewest.RTPTimestampExt + uint64(timeSinceLastReport.Nanoseconds()*int64(r.params.ClockRate)/1e9)
+		timeSinceLastReport := mediatransportutil.NtpTime(srData.NtpTimestamp).Time().Sub(mediatransportutil.NtpTime(r.srNewest.NtpTimestamp).Time())
+		expectedRTPTimestampExt := r.srNewest.RtpTimestampExt + uint64(timeSinceLastReport.Nanoseconds()*int64(r.params.ClockRate)/1e9)
 		lbound := expectedRTPTimestampExt - uint64(cReportSlack*float64(r.params.ClockRate))
 		ubound := expectedRTPTimestampExt + uint64(cReportSlack*float64(r.params.ClockRate))
-		isInRange := (srData.RTPTimestamp-uint32(lbound) < (1 << 31)) && (uint32(ubound)-srData.RTPTimestamp < (1 << 31))
+		isInRange := (srData.RtpTimestamp-uint32(lbound) < (1 << 31)) && (uint32(ubound)-srData.RtpTimestamp < (1 << 31))
 		if isInRange {
 			lbTSCycles := lbound & 0xFFFF_FFFF_0000_0000
 			ubTSCycles := ubound & 0xFFFF_FFFF_0000_0000
 			if lbTSCycles == ubTSCycles {
 				tsCycles = lbTSCycles
 			} else {
-				if srData.RTPTimestamp < (1 << 31) {
+				if srData.RtpTimestamp < (1 << 31) {
 					// rolled over
 					tsCycles = ubTSCycles
 				} else {
@@ -398,26 +400,26 @@ func (r *RTPStatsReceiver) getExtendedSenderReport(srData *RTCPSenderReportData)
 		} else {
 			// ideally this method should not be required, but there are clients
 			// negotiating one clock rate, but actually send media at a different rate.
-			tsCycles = r.srNewest.RTPTimestampExt & 0xFFFF_FFFF_0000_0000
-			if (srData.RTPTimestamp-r.srNewest.RTPTimestamp) < (1<<31) && srData.RTPTimestamp < r.srNewest.RTPTimestamp {
+			tsCycles = r.srNewest.RtpTimestampExt & 0xFFFF_FFFF_0000_0000
+			if (srData.RtpTimestamp-r.srNewest.RtpTimestamp) < (1<<31) && srData.RtpTimestamp < r.srNewest.RtpTimestamp {
 				tsCycles += (1 << 32)
 			}
 
 			if tsCycles >= (1 << 32) {
-				if (srData.RTPTimestamp-r.srNewest.RTPTimestamp) >= (1<<31) && srData.RTPTimestamp > r.srNewest.RTPTimestamp {
+				if (srData.RtpTimestamp-r.srNewest.RtpTimestamp) >= (1<<31) && srData.RtpTimestamp > r.srNewest.RtpTimestamp {
 					tsCycles -= (1 << 32)
 				}
 			}
 		}
 	}
 
-	srDataExt := *srData
-	srDataExt.RTPTimestampExt = uint64(srDataExt.RTPTimestamp) + tsCycles
-	return &srDataExt
+	srDataExt := proto.Clone(srData).(*livekit.RTCPSenderReportState)
+	srDataExt.RtpTimestampExt = uint64(srDataExt.RtpTimestamp) + tsCycles
+	return srDataExt
 }
 
-func (r *RTPStatsReceiver) checkOutOfOrderSenderReport(srData *RTCPSenderReportData) bool {
-	if r.srNewest != nil && srData.RTPTimestampExt < r.srNewest.RTPTimestampExt {
+func (r *RTPStatsReceiver) checkOutOfOrderSenderReport(srData *livekit.RTCPSenderReportState) bool {
+	if r.srNewest != nil && srData.RtpTimestampExt < r.srNewest.RtpTimestampExt {
 		// This can happen when a track is replaced with a null and then restored -
 		// i. e. muting replacing with null and unmute restoring the original track.
 		// Or it could be due bad report generation.
@@ -426,7 +428,7 @@ func (r *RTPStatsReceiver) checkOutOfOrderSenderReport(srData *RTCPSenderReportD
 		if (r.outOfOrderSenderReportCount-1)%10 == 0 {
 			r.logger.Infow(
 				"received sender report, out-of-order, skipping",
-				"current", srData,
+				"current", WrappedRTCPSenderReportStateLogger{srData},
 				"count", r.outOfOrderSenderReportCount,
 				"rtpStats", lockedRTPStatsReceiverLogEncoder{r},
 			)
@@ -437,17 +439,17 @@ func (r *RTPStatsReceiver) checkOutOfOrderSenderReport(srData *RTCPSenderReportD
 	return false
 }
 
-func (r *RTPStatsReceiver) checkRTPClockSkewForSenderReport(srData *RTCPSenderReportData) {
+func (r *RTPStatsReceiver) checkRTPClockSkewForSenderReport(srData *livekit.RTCPSenderReportState) {
 	if r.srNewest == nil {
 		return
 	}
 
-	timeSinceLast := srData.NTPTimestamp.Time().Sub(r.srNewest.NTPTimestamp.Time()).Seconds()
-	rtpDiffSinceLast := srData.RTPTimestampExt - r.srNewest.RTPTimestampExt
+	timeSinceLast := time.Duration(srData.NtpTimestamp - r.srNewest.NtpTimestamp).Seconds()
+	rtpDiffSinceLast := srData.RtpTimestampExt - r.srNewest.RtpTimestampExt
 	calculatedClockRateFromLast := float64(rtpDiffSinceLast) / timeSinceLast
 
-	timeSinceFirst := srData.NTPTimestamp.Time().Sub(r.srFirst.NTPTimestamp.Time()).Seconds()
-	rtpDiffSinceFirst := srData.RTPTimestampExt - r.srFirst.RTPTimestampExt
+	timeSinceFirst := time.Duration(srData.NtpTimestamp - r.srFirst.NtpTimestamp).Seconds()
+	rtpDiffSinceFirst := srData.RtpTimestampExt - r.srFirst.RtpTimestampExt
 	calculatedClockRateFromFirst := float64(rtpDiffSinceFirst) / timeSinceFirst
 
 	if (timeSinceLast > 0.2 && math.Abs(float64(r.params.ClockRate)-calculatedClockRateFromLast) > 0.2*float64(r.params.ClockRate)) ||
@@ -456,7 +458,7 @@ func (r *RTPStatsReceiver) checkRTPClockSkewForSenderReport(srData *RTCPSenderRe
 		if (r.clockSkewCount-1)%100 == 0 {
 			r.logger.Infow(
 				"received sender report, clock skew",
-				"current", srData,
+				"current", WrappedRTCPSenderReportStateLogger{srData},
 				"timeSinceFirst", timeSinceFirst,
 				"rtpDiffSinceFirst", rtpDiffSinceFirst,
 				"calculatedFirst", calculatedClockRateFromFirst,
@@ -470,13 +472,13 @@ func (r *RTPStatsReceiver) checkRTPClockSkewForSenderReport(srData *RTCPSenderRe
 	}
 }
 
-func (r *RTPStatsReceiver) checkRTPClockSkewAgainstMediaPathForSenderReport(srData *RTCPSenderReportData) {
+func (r *RTPStatsReceiver) checkRTPClockSkewAgainstMediaPathForSenderReport(srData *livekit.RTCPSenderReportState) {
 	if r.highestTime == 0 {
 		return
 	}
 
-	timeSinceSR := time.Since(srData.AtAdjusted)
-	extNowTSSR := srData.RTPTimestampExt + uint64(timeSinceSR.Nanoseconds()*int64(r.params.ClockRate)/1e9)
+	timeSinceSR := time.Since(time.Unix(0, srData.AtAdjusted))
+	extNowTSSR := srData.RtpTimestampExt + uint64(timeSinceSR.Nanoseconds()*int64(r.params.ClockRate)/1e9)
 
 	timeSinceHighest := time.Since(time.Unix(0, r.highestTime))
 	extNowTSHighest := r.timestamp.GetExtendedHighest() + uint64(timeSinceHighest.Nanoseconds()*int64(r.params.ClockRate)/1e9)
@@ -492,7 +494,7 @@ func (r *RTPStatsReceiver) checkRTPClockSkewAgainstMediaPathForSenderReport(srDa
 		if (r.clockSkewMediaPathCount-1)%100 == 0 {
 			r.logger.Infow(
 				"received sender report, clock skew against media path",
-				"current", srData,
+				"current", WrappedRTCPSenderReportStateLogger{srData},
 				"timeSinceSR", timeSinceSR,
 				"extNowTSSR", extNowTSSR,
 				"timeSinceHighest", timeSinceHighest,
@@ -508,9 +510,9 @@ func (r *RTPStatsReceiver) checkRTPClockSkewAgainstMediaPathForSenderReport(srDa
 	}
 }
 
-func (r *RTPStatsReceiver) updatePropagationDelayAndRecordSenderReport(srData *RTCPSenderReportData) {
-	senderClockTime := srData.NTPTimestamp.Time()
-	estimatedPropagationDelay, stepChange := r.propagationDelayEstimator.Update(senderClockTime, srData.At)
+func (r *RTPStatsReceiver) updatePropagationDelayAndRecordSenderReport(srData *livekit.RTCPSenderReportState) {
+	senderClockTime := mediatransportutil.NtpTime(srData.NtpTimestamp).Time()
+	estimatedPropagationDelay, stepChange := r.propagationDelayEstimator.Update(senderClockTime, time.Unix(0, srData.At))
 	if stepChange {
 		r.logger.Debugw(
 			"propagation delay step change",
@@ -523,11 +525,11 @@ func (r *RTPStatsReceiver) updatePropagationDelayAndRecordSenderReport(srData *R
 		r.srFirst = srData
 	}
 	// adjust receive time to estimated propagation delay
-	srData.AtAdjusted = senderClockTime.Add(estimatedPropagationDelay)
+	srData.AtAdjusted = senderClockTime.Add(estimatedPropagationDelay).UnixNano()
 	r.srNewest = srData
 }
 
-func (r *RTPStatsReceiver) SetRtcpSenderReportData(srData *RTCPSenderReportData) bool {
+func (r *RTPStatsReceiver) SetRtcpSenderReportData(srData *livekit.RTCPSenderReportState) bool {
 	r.lock.Lock()
 	defer r.lock.Unlock()
 
@@ -536,7 +538,7 @@ func (r *RTPStatsReceiver) SetRtcpSenderReportData(srData *RTCPSenderReportData)
 	}
 
 	// prevent against extreme case of anachronous sender reports
-	if r.srNewest != nil && r.srNewest.NTPTimestamp > srData.NTPTimestamp {
+	if r.srNewest != nil && r.srNewest.NtpTimestamp > srData.NtpTimestamp {
 		r.logger.Infow(
 			"received sender report, anachronous, dropping",
 			"current", srData,
@@ -561,16 +563,11 @@ func (r *RTPStatsReceiver) SetRtcpSenderReportData(srData *RTCPSenderReportData)
 	return true
 }
 
-func (r *RTPStatsReceiver) GetRtcpSenderReportData() *RTCPSenderReportData {
+func (r *RTPStatsReceiver) GetRtcpSenderReportData() *livekit.RTCPSenderReportState {
 	r.lock.RLock()
 	defer r.lock.RUnlock()
 
-	if r.srNewest == nil {
-		return nil
-	}
-
-	srNewestCopy := *r.srNewest
-	return &srNewestCopy
+	return proto.Clone(r.srNewest).(*livekit.RTCPSenderReportState)
 }
 
 func (r *RTPStatsReceiver) LastSenderReportTime() time.Time {
@@ -578,7 +575,7 @@ func (r *RTPStatsReceiver) LastSenderReportTime() time.Time {
 	defer r.lock.RUnlock()
 
 	if r.srNewest != nil {
-		return r.srNewest.At
+		return time.Unix(0, r.srNewest.At)
 	}
 
 	return time.Time{}
@@ -625,9 +622,9 @@ func (r *RTPStatsReceiver) GetRtcpReceptionReport(ssrc uint32, proxyFracLost uin
 	lastSR := uint32(0)
 	dlsr := uint32(0)
 	if r.srNewest != nil {
-		lastSR = uint32(r.srNewest.NTPTimestamp >> 16)
-		if !r.srNewest.At.IsZero() {
-			delayUS := time.Since(r.srNewest.At).Microseconds()
+		lastSR = uint32(r.srNewest.NtpTimestamp >> 16)
+		if r.srNewest.At != 0 {
+			delayUS := time.Since(time.Unix(0, r.srNewest.At)).Microseconds()
 			dlsr = uint32(delayUS * 65536 / 1e6)
 		}
 	}
@@ -719,10 +716,17 @@ func (r lockedRTPStatsReceiverLogEncoder) MarshalLogObject(e zapcore.ObjectEncod
 
 	e.AddUint64("extStartSN", r.sequenceNumber.GetExtendedStart())
 	e.AddUint64("extHighestSN", r.sequenceNumber.GetExtendedHighest())
-	e.AddUint64("extStartTS", r.timestamp.GetExtendedStart())
-	e.AddUint64("extHighestTS", r.timestamp.GetExtendedHighest())
+	extStartTS, extHighestTS := r.timestamp.GetExtendedStart(), r.timestamp.GetExtendedHighest()
+	e.AddUint64("extStartTS", extStartTS)
+	e.AddUint64("extHighestTS", extHighestTS)
 
 	e.AddObject("propagationDelayEstimator", r.propagationDelayEstimator)
+
+	packetDrift, ntpReportDrift, receivedReportDrift, rebasedReportDrift := r.getDrift(extStartTS, extHighestTS)
+	e.AddObject("packetDrift", wrappedRTPDriftLogger{packetDrift})
+	e.AddObject("ntpReportDrift", wrappedRTPDriftLogger{ntpReportDrift})
+	e.AddObject("receivedReportDrift", wrappedRTPDriftLogger{receivedReportDrift})
+	e.AddObject("rebasedReportDrift", wrappedRTPDriftLogger{rebasedReportDrift})
 	return nil
 }
 

pkg/sfu/buffer/rtpstats_sender.go

@@ -483,7 +483,7 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt
 
 	if r.srNewest != nil {
 		var err error
-		rtt, err = mediatransportutil.GetRttMs(&rr, r.srNewest.NTPTimestamp, r.srNewest.At)
+		rtt, err = mediatransportutil.GetRttMs(&rr, mediatransportutil.NtpTime(r.srNewest.NtpTimestamp), time.Unix(0, r.srNewest.At))
 		if err == nil {
 			isRttChanged = rtt != r.rtt
 		} else {
@@ -584,7 +584,7 @@ func (r *RTPStatsSender) LastReceiverReportTime() time.Time {
 	return r.lastRRTime
 }
 
-func (r *RTPStatsSender) MaybeAdjustFirstPacketTime(publisherSRData *RTCPSenderReportData, tsOffset uint64) {
+func (r *RTPStatsSender) MaybeAdjustFirstPacketTime(publisherSRData *livekit.RTCPSenderReportState, tsOffset uint64) {
 	r.lock.Lock()
 	defer r.lock.Unlock()
 
@@ -612,7 +612,7 @@ func (r *RTPStatsSender) GetExpectedRTPTimestamp(at time.Time) (expectedTSExt ui
 	return
 }
 
-func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *RTCPSenderReportData, tsOffset uint64, passThrough bool) *rtcp.SenderReport {
+func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livekit.RTCPSenderReportState, tsOffset uint64, passThrough bool) *rtcp.SenderReport {
 	r.lock.Lock()
 	defer r.lock.Unlock()
 
@@ -620,26 +620,26 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *RTCPS
 		return nil
 	}
 
-	timeSincePublisherSRAdjusted := time.Since(publisherSRData.AtAdjusted)
-	now := publisherSRData.AtAdjusted.Add(timeSincePublisherSRAdjusted)
+	timeSincePublisherSRAdjusted := time.Since(time.Unix(0, publisherSRData.AtAdjusted))
+	now := publisherSRData.AtAdjusted + timeSincePublisherSRAdjusted.Nanoseconds()
 	var (
 		nowNTP    mediatransportutil.NtpTime
 		nowRTPExt uint64
 	)
 	if passThrough {
-		nowNTP = publisherSRData.NTPTimestamp
-		nowRTPExt = publisherSRData.RTPTimestampExt - tsOffset
+		nowNTP = mediatransportutil.NtpTime(publisherSRData.NtpTimestamp)
+		nowRTPExt = publisherSRData.RtpTimestampExt - tsOffset
 	} else {
-		nowNTP = mediatransportutil.ToNtpTime(now)
-		nowRTPExt = publisherSRData.RTPTimestampExt - tsOffset + uint64(timeSincePublisherSRAdjusted.Nanoseconds()*int64(r.params.ClockRate)/1e9)
+		nowNTP = mediatransportutil.ToNtpTime(time.Unix(0, now))
+		nowRTPExt = publisherSRData.RtpTimestampExt - tsOffset + uint64(timeSincePublisherSRAdjusted.Nanoseconds()*int64(r.params.ClockRate)/1e9)
 	}
 
 	packetCount := uint32(r.getTotalPacketsPrimary(r.extStartSN, r.extHighestSN) + r.packetsDuplicate + r.packetsPadding)
-	octetCount := uint32(r.bytes + r.bytesDuplicate + r.bytesPadding)
-	srData := &RTCPSenderReportData{
-		NTPTimestamp:    nowNTP,
-		RTPTimestamp:    uint32(nowRTPExt),
-		RTPTimestampExt: nowRTPExt,
+	octetCount := r.bytes + r.bytesDuplicate + r.bytesPadding
+	srData := &livekit.RTCPSenderReportState{
+		NtpTimestamp:    uint64(nowNTP),
+		RtpTimestamp:    uint32(nowRTPExt),
+		RtpTimestampExt: nowRTPExt,
 		At:              now,
 		AtAdjusted:      now,
 		Packets:         packetCount,
@@ -652,18 +652,18 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *RTCPS
 			"feed", publisherSRData,
 			"tsOffset", tsOffset,
 			"timeNow", time.Now().String(),
-			"now", now.String(),
-			"timeSinceHighest", now.Sub(time.Unix(0, r.highestTime)).String(),
-			"timeSinceFirst", now.Sub(time.Unix(0, r.firstTime)).String(),
+			"now", time.Unix(0, now).String(),
+			"timeSinceHighest", time.Unix(0, now).Sub(time.Unix(0, r.highestTime)).String(),
+			"timeSinceFirst", time.Unix(0, now).Sub(time.Unix(0, r.firstTime)).String(),
 			"timeSincePublisherSRAdjusted", timeSincePublisherSRAdjusted.String(),
-			"timeSincePublisherSR", time.Since(publisherSRData.At).String(),
+			"timeSincePublisherSR", time.Since(time.Unix(0, publisherSRData.At)).String(),
 			"nowRTPExt", nowRTPExt,
 			"rtpStats", lockedRTPStatsSenderLogEncoder{r},
 		}
 	}
-	if r.srNewest != nil && nowRTPExt >= r.srNewest.RTPTimestampExt {
-		timeSinceLastReport := nowNTP.Time().Sub(r.srNewest.NTPTimestamp.Time())
-		rtpDiffSinceLastReport := nowRTPExt - r.srNewest.RTPTimestampExt
+	if r.srNewest != nil && nowRTPExt >= r.srNewest.RtpTimestampExt {
+		timeSinceLastReport := nowNTP.Time().Sub(mediatransportutil.NtpTime(r.srNewest.NtpTimestamp).Time())
+		rtpDiffSinceLastReport := nowRTPExt - r.srNewest.RtpTimestampExt
 		windowClockRate := float64(rtpDiffSinceLastReport) / timeSinceLastReport.Seconds()
 		if timeSinceLastReport.Seconds() > 0.2 && math.Abs(float64(r.params.ClockRate)-windowClockRate) > 0.2*float64(r.params.ClockRate) {
 			r.clockSkewCount++
@@ -680,7 +680,7 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *RTCPS
 		}
 	}
 
-	if r.srNewest != nil && nowRTPExt < r.srNewest.RTPTimestampExt {
+	if r.srNewest != nil && nowRTPExt < r.srNewest.RtpTimestampExt {
 		// If report being generated is behind the last report, skip it.
 		// Should not happen.
 		r.logger.Infow("sending sender report, out-of-order, skipping", getFields()...)
@@ -697,7 +697,7 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *RTCPS
 		NTPTime:     uint64(nowNTP),
 		RTPTime:     uint32(nowRTPExt),
 		PacketCount: packetCount,
-		OctetCount:  octetCount,
+		OctetCount:  uint32(octetCount),
 	}
 }
 
@@ -1025,5 +1025,11 @@ func (r lockedRTPStatsSenderLogEncoder) MarshalLogObject(e zapcore.ObjectEncoder
 	e.AddUint64("packetsLostFromRR", r.packetsLostFromRR)
 	e.AddFloat64("jitterFromRR", r.jitterFromRR)
 	e.AddFloat64("maxJitterFromRR", r.maxJitterFromRR)
+
+	packetDrift, ntpReportDrift, receivedReportDrift, rebasedReportDrift := r.getDrift(r.extStartTS, r.extHighestTS)
+	e.AddObject("packetDrift", wrappedRTPDriftLogger{packetDrift})
+	e.AddObject("ntpReportDrift", wrappedRTPDriftLogger{ntpReportDrift})
+	e.AddObject("receivedReportDrift", wrappedRTPDriftLogger{receivedReportDrift})
+	e.AddObject("rebasedReportDrift", wrappedRTPDriftLogger{rebasedReportDrift})
 	return nil
 }