Log more details of RTP stats snap shots. (#3190)

* Log more details of RTP stats snap shots. Seeing cases of loss more than 100%. Logging snap shots to understand it better. * log message * use delta to update packets lost from RR * remove cast
2026-10-23 02:12:44 +00:00 · 2024-11-21 16:41:03 +05:30
1 parent 9f25603213
commit d5cc567140
@@ -140,15 +140,17 @@ var (
 // -------------------------------------------------------------------

 type DownTrackState struct {
-	RTPStats                   *rtpstats.RTPStatsSender
-	DeltaStatsSenderSnapshotId uint32
-	ForwarderState             *livekit.RTPForwarderState
+	RTPStats                    *rtpstats.RTPStatsSender
+	DeltaStatsSenderSnapshotId  uint32
+	ForwarderState              *livekit.RTPForwarderState
+	PlayoutDelayControllerState PlayoutDelayControllerState
 }

 func (d DownTrackState) MarshalLogObject(e zapcore.ObjectEncoder) error {
 	e.AddObject("RTPStats", d.RTPStats)
 	e.AddUint32("DeltaStatsSenderSnapshotId", d.DeltaStatsSenderSnapshotId)
 	e.AddObject("ForwarderState", logger.Proto(d.ForwarderState))
+	e.AddObject("PlayoutDelayControllerState", d.PlayoutDelayControllerState)
 	return nil
 }

@@ -1257,11 +1259,16 @@ func (d *DownTrack) MaxLayer() buffer.VideoLayer {
 }

 func (d *DownTrack) GetState() DownTrackState {
-	return DownTrackState{
+	dts := DownTrackState{
 		RTPStats:                   d.rtpStats,
 		DeltaStatsSenderSnapshotId: d.deltaStatsSenderSnapshotId,
 		ForwarderState:             d.forwarder.GetState(),
 	}
+
+	if d.playoutDelay != nil {
+		dts.PlayoutDelayControllerState = d.playoutDelay.GetState()
+	}
+	return dts
 }

 func (d *DownTrack) SeedState(state DownTrackState) {
@@ -1271,6 +1278,9 @@ func (d *DownTrack) SeedState(state DownTrackState) {
 	if state.RTPStats != nil {
 		d.rtpStats.Seed(state.RTPStats)
 		d.deltaStatsSenderSnapshotId = state.DeltaStatsSenderSnapshotId
+		if d.playoutDelay != nil {
+			d.playoutDelay.SeedState(state.PlayoutDelayControllerState)
+		}
 	}
 	d.forwarder.SeedState(state.ForwarderState)
 }
@@ -22,20 +22,25 @@ import (
 	pd "github.com/livekit/livekit-server/pkg/sfu/rtpextension/playoutdelay"
 	"github.com/livekit/livekit-server/pkg/sfu/rtpstats"
 	"github.com/livekit/protocol/logger"
+	"go.uber.org/zap/zapcore"
 )

+const (
+	jitterMultiToDelay      = 10
+	targetDelayLogThreshold = 500
+
+	// limit max delay change to make it smoother for a/v sync
+	maxDelayChangePerSec = 80
+)
+
+// ----------------------------------------------------
+
 type PlayoutDelayState int32

 const (
 	PlayoutDelayStateChanged PlayoutDelayState = iota
 	PlayoutDelaySending
 	PlayoutDelayAcked
-
-	jitterMultiToDelay      = 10
-	targetDelayLogThreshold = 500
-
-	// limit max delay change to make it smoother for a/v sync
-	maxDelayChangePerSec = 80
 )

 func (s PlayoutDelayState) String() string {
@@ -50,6 +55,19 @@ func (s PlayoutDelayState) String() string {
 	return "Unknown"
 }

+// ----------------------------------------------------
+
+type PlayoutDelayControllerState struct {
+	SenderSnapshotID uint32
+}
+
+func (p PlayoutDelayControllerState) MarshalLogObject(e zapcore.ObjectEncoder) error {
+	e.AddUint32("SenderSnapshotID", p.SenderSnapshotID)
+	return nil
+}
+
+// ----------------------------------------------------
+
 type PlayoutDelayController struct {
 	lock               sync.Mutex
 	state              atomic.Int32
@@ -60,7 +78,7 @@ type PlayoutDelayController struct {
 	sendingAtTime      time.Time
 	logger             logger.Logger
 	rtpStats           *rtpstats.RTPStatsSender
-	snapshotID         uint32
+	senderSnapshotID   uint32

 	highDelayCount atomic.Uint32
 }
@@ -73,24 +91,40 @@ func NewPlayoutDelayController(minDelay, maxDelay uint32, logger logger.Logger,
 		maxDelay = pd.PlayoutDelayMaxValue
 	}
 	c := &PlayoutDelayController{
-		currentDelay: minDelay,
-		minDelay:     minDelay,
-		maxDelay:     maxDelay,
-		logger:       logger,
-		rtpStats:     rtpStats,
-		snapshotID:   rtpStats.NewSenderSnapshotId(),
+		currentDelay:     minDelay,
+		minDelay:         minDelay,
+		maxDelay:         maxDelay,
+		logger:           logger,
+		rtpStats:         rtpStats,
+		senderSnapshotID: rtpStats.NewSenderSnapshotId(),
 	}
 	return c, c.createExtData()
 }

+func (c *PlayoutDelayController) GetState() PlayoutDelayControllerState {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	return PlayoutDelayControllerState{
+		SenderSnapshotID: c.senderSnapshotID,
+	}
+}
+
+func (c *PlayoutDelayController) SeedState(pdcs PlayoutDelayControllerState) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	c.senderSnapshotID = pdcs.SenderSnapshotID
+}
+
 func (c *PlayoutDelayController) SetJitter(jitter uint32) {
-	deltaInfo := c.rtpStats.DeltaInfoSender(c.snapshotID)
+	c.lock.Lock()
+	deltaInfoSender := c.rtpStats.DeltaInfoSender(c.senderSnapshotID)
 	var nackPercent uint32
-	if deltaInfo != nil && deltaInfo.Packets > 0 {
-		nackPercent = deltaInfo.Nacks * 100 / deltaInfo.Packets
+	if deltaInfoSender != nil && deltaInfoSender.Packets > 0 {
+		nackPercent = deltaInfoSender.Nacks * 100 / deltaInfoSender.Packets
 	}

-	c.lock.Lock()
 	targetDelay := jitter * jitterMultiToDelay
 	if nackPercent > 60 {
 		targetDelay += (nackPercent - 60) * 2
@@ -58,6 +58,36 @@ type RTPDeltaInfo struct {
 	Firs                 uint32
 }

+func (r *RTPDeltaInfo) MarshalLogObject(e zapcore.ObjectEncoder) error {
+	if r == nil {
+		return nil
+	}
+
+	e.AddTime("StartTime", r.StartTime)
+	e.AddTime("EndTime", r.EndTime)
+	e.AddUint32("Packets", r.Packets)
+	e.AddUint64("Bytes", r.Bytes)
+	e.AddUint64("HeaderBytes", r.HeaderBytes)
+	e.AddUint32("PacketsDuplicate", r.PacketsDuplicate)
+	e.AddUint64("BytesDuplicate", r.BytesDuplicate)
+	e.AddUint64("HeaderBytesDuplicate", r.HeaderBytesDuplicate)
+	e.AddUint32("PacketsPadding", r.PacketsPadding)
+	e.AddUint64("BytesPadding", r.BytesPadding)
+	e.AddUint64("HeaderBytesPadding", r.HeaderBytesPadding)
+	e.AddUint32("PacketsLost", r.PacketsLost)
+	e.AddUint32("PacketsMissing", r.PacketsMissing)
+	e.AddUint32("PacketsOutOfOrder", r.PacketsOutOfOrder)
+	e.AddUint32("Frames", r.Frames)
+	e.AddUint32("RttMax", r.RttMax)
+	e.AddFloat64("JitterMax", r.JitterMax)
+	e.AddUint32("Nacks", r.Nacks)
+	e.AddUint32("Plis", r.Plis)
+	e.AddUint32("Firs", r.Firs)
+	return nil
+}
+
+// -------------------------------------------------------
+
 type snapshot struct {
 	snapshotLite

@@ -80,6 +110,27 @@ type snapshot struct {
 	maxJitter float64
 }

+func (s *snapshot) MarshalLogObject(e zapcore.ObjectEncoder) error {
+	if s == nil {
+		return nil
+	}
+
+	e.AddObject("snapshotLite", &s.snapshotLite)
+	e.AddUint64("headerBytes", s.headerBytes)
+	e.AddUint64("packetsDuplicate", s.packetsDuplicate)
+	e.AddUint64("bytesDuplicate", s.bytesDuplicate)
+	e.AddUint64("headerBytesDuplicate", s.headerBytesDuplicate)
+	e.AddUint64("packetsPadding", s.packetsPadding)
+	e.AddUint64("bytesPadding", s.bytesPadding)
+	e.AddUint64("headerBytesPadding", s.headerBytesPadding)
+	e.AddUint32("frames", s.frames)
+	e.AddUint32("plis", s.plis)
+	e.AddUint32("firs", s.firs)
+	e.AddUint32("maxRtt", s.maxRtt)
+	e.AddFloat64("maxJitter", s.maxJitter)
+	return nil
+}
+
 // ------------------------------------------------------------------

 type wrappedRTPDriftLogger struct {
@@ -399,12 +450,11 @@ func (r *rtpStatsBase) deltaInfo(
 	}
 	if packetsExpected > cNumSequenceNumbers {
 		loggingFields = []interface{}{
-			"startSN", then.extStartSN,
-			"endSN", now.extStartSN,
-			"packetsExpected", packetsExpected,
-			"startTime", startTime,
-			"endTime", endTime,
+			"snapshotID", snapshotID,
+			"snapshotNow", now,
+			"snapshotThen", then,
 			"duration", endTime.Sub(startTime),
+			"packetsExpected", packetsExpected,
 		}
 		err = errors.New("too many packets expected in delta")
 		return
@@ -426,11 +476,13 @@ func (r *rtpStatsBase) deltaInfo(
 	packetsPadding := now.packetsPadding - then.packetsPadding
 	if packetsExpected < packetsPadding {
 		loggingFields = []interface{}{
+			"snapshotID", snapshotID,
+			"snapshotNow", now,
+			"snapshotThen", then,
+			"duration", endTime.Sub(startTime),
 			"packetsExpected", packetsExpected,
 			"packetsPadding", packetsPadding,
 			"packetsLost", packetsLost,
-			"startSequenceNumber", then.extStartSN,
-			"endSequenceNumber", now.extStartSN - 1,
 		}
 		err = errors.New("padding packets more than expected")
 		packetsExpected = 0
@@ -44,6 +44,23 @@ type RTPDeltaInfoLite struct {
 	Nacks             uint32
 }

+func (r *RTPDeltaInfoLite) MarshalLogObject(e zapcore.ObjectEncoder) error {
+	if r == nil {
+		return nil
+	}
+
+	e.AddTime("StartTime", r.StartTime)
+	e.AddTime("EndTime", r.EndTime)
+	e.AddUint32("Packets", r.Packets)
+	e.AddUint64("Bytes", r.Bytes)
+	e.AddUint32("PacketsLost", r.PacketsLost)
+	e.AddUint32("PacketsOutOfOrder", r.PacketsOutOfOrder)
+	e.AddUint32("Nacks", r.Nacks)
+	return nil
+}
+
+// -------------------------------------------------------
+
 type snapshotLite struct {
 	isValid bool

@@ -59,6 +76,21 @@ type snapshotLite struct {
 	nacks uint32
 }

+func (s *snapshotLite) MarshalLogObject(e zapcore.ObjectEncoder) error {
+	if s == nil {
+		return nil
+	}
+
+	e.AddBool("isValid", s.isValid)
+	e.AddTime("startTime", s.startTime)
+	e.AddUint64("extStartSN", s.extStartSN)
+	e.AddUint64("bytes", s.bytes)
+	e.AddUint64("packetsOutOfOrder", s.packetsOutOfOrder)
+	e.AddUint64("packetsLost", s.packetsLost)
+	e.AddUint32("nacks", s.nacks)
+	return nil
+}
+
 // ------------------------------------------------------------------

 type RTPStatsParams struct {
@@ -280,17 +312,16 @@ func (r *rtpStatsBaseLite) deltaInfoLite(
 	startTime := then.startTime
 	endTime := now.startTime

-	packetsExpected := now.extStartSN - then.extStartSN
+	packetsExpected := uint32(now.extStartSN - then.extStartSN)
 	if then.extStartSN > extHighestSN {
 		packetsExpected = 0
 	}
 	if packetsExpected > cNumSequenceNumbers {
 		loggingFields = []interface{}{
-			"startSN", then.extStartSN,
-			"endSN", now.extStartSN,
+			"snapshotLiteID", snapshotLiteID,
+			"snapshotLiteNow", now,
+			"snapshotLiteThen", then,
 			"packetsExpected", packetsExpected,
-			"startTime", startTime,
-			"endTime", endTime,
 			"duration", endTime.Sub(startTime).String(),
 		}
 		err = errors.New("too many packets expected in delta lite")
@@ -308,11 +339,22 @@ func (r *rtpStatsBaseLite) deltaInfoLite(
 	if int32(packetsLost) < 0 {
 		packetsLost = 0
 	}
+	if packetsLost > packetsExpected {
+		loggingFields = []interface{}{
+			"snapshotLiteID", snapshotLiteID,
+			"snapshotLiteNow", now,
+			"snapshotLiteThen", then,
+			"packetsExpected", packetsExpected,
+			"packetsLost", packetsLost,
+			"duration", endTime.Sub(startTime).String(),
+		}
+		err = errors.New("unexpected number of packets lost in delta lite")
+	}

 	deltaInfoLite = &RTPDeltaInfoLite{
 		StartTime:         startTime,
 		EndTime:           endTime,
-		Packets:           uint32(packetsExpected),
+		Packets:           packetsExpected,
 		Bytes:             now.bytes - then.bytes,
 		PacketsLost:       packetsLost,
 		PacketsOutOfOrder: uint32(now.packetsOutOfOrder - then.packetsOutOfOrder),
@@ -138,6 +138,30 @@ type senderSnapshot struct {
 	intervalStats intervalStats
 }

+func (s *senderSnapshot) MarshalLogObject(e zapcore.ObjectEncoder) error {
+	if s == nil {
+		return nil
+	}
+
+	e.AddBool("isValid", s.isValid)
+	e.AddTime("startTime", s.startTime)
+	e.AddUint64("extStartSN", s.extStartSN)
+	e.AddUint64("bytes", s.bytes)
+	e.AddUint64("headerBytes", s.headerBytes)
+	e.AddUint64("packetsPadding", s.packetsPadding)
+	e.AddUint64("bytesPadding", s.bytesPadding)
+	e.AddUint64("headerBytesPadding", s.headerBytesPadding)
+	e.AddUint64("packetsDuplicate", s.packetsDuplicate)
+	e.AddUint64("bytesDuplicate", s.bytesDuplicate)
+	e.AddUint64("headerBytesDuplicate", s.headerBytesDuplicate)
+	e.AddUint64("packetsOutOfOrder", s.packetsOutOfOrder)
+	e.AddUint64("packetsLostFeed", s.packetsLostFeed)
+	e.AddUint64("packetsLost", s.packetsLost)
+	e.AddUint32("frames", s.frames)
+	e.AddUint32("nacks", s.nacks)
+	return nil
+}
+
 // -------------------------------------------------------------------

 type rttMarker struct {
@@ -514,9 +538,12 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt
 		}
 	}

-	r.packetsLostFromRR = r.packetsLostFromRR&0xFFFF_FFFF_FF00_0000 + uint64(rr.TotalLost)
-	if ((rr.TotalLost-r.lastRR.TotalLost)&((1<<24)-1)) < (1<<23) && rr.TotalLost < r.lastRR.TotalLost {
-		r.packetsLostFromRR += (1 << 24)
+	lossDelta := (rr.TotalLost - r.lastRR.TotalLost) & ((1 << 24) - 1)
+	if lossDelta > 0 && lossDelta < (1<<23) {
+		r.packetsLostFromRR += uint64(lossDelta)
+		if rr.TotalLost < r.lastRR.TotalLost {
+			r.packetsLostFromRR += (1 << 24)
+		}
 	}

 	if isRttChanged {
@@ -563,6 +590,7 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt
 				"packetsInInterval", extReceivedRRSN-s.extLastRRSN,
 				"rtpStats", lockedRTPStatsSenderLogEncoder{r},
 			)
+			s.extLastRRSN = extReceivedRRSN
 			continue
 		}

@@ -768,11 +796,10 @@ func (r *RTPStatsSender) DeltaInfoSender(senderSnapshotID uint32) *RTPDeltaInfo
 	if packetsExpected > cNumSequenceNumbers {
 		r.logger.Warnw(
 			"too many packets expected in delta (sender)", nil,
-			"startSN", then.extStartSN,
-			"endSN", now.extStartSN,
+			"senderSnapshotID", senderSnapshotID,
+			"senderSnapshotNow", now,
+			"senderSnapshotThen", then,
 			"packetsExpected", packetsExpected,
-			"startTime", startTime,
-			"endTime", endTime,
 			"duration", endTime.Sub(startTime),
 			"rtpStats", lockedRTPStatsSenderLogEncoder{r},
 		)
@@ -793,15 +820,14 @@ func (r *RTPStatsSender) DeltaInfoSender(senderSnapshotID uint32) *RTPDeltaInfo
 	}
 	if packetsLost > packetsExpected {
 		r.logger.Warnw(
-			"unexpected number of packets lost",
-			fmt.Errorf(
-				"start: %d, end: %d, expected: %d, lost: report: %d, feed: %d",
-				then.extStartSN,
-				now.extStartSN,
-				packetsExpected,
-				packetsLost,
-				packetsLostFeed,
-			),
+			"unexpected number of packets lost", nil,
+			"senderSnapshotID", senderSnapshotID,
+			"senderSnapshotNow", now,
+			"senderSnapshotThen", then,
+			"packetsExpected", packetsExpected,
+			"packetsLost", packetsLost,
+			"packetsLostFeed", packetsLostFeed,
+			"duration", endTime.Sub(startTime),
 			"rtpStats", lockedRTPStatsSenderLogEncoder{r},
 		)
 		packetsLost = packetsExpected
@@ -906,7 +932,7 @@ func (r *RTPStatsSender) getSenderSnapshot(startTime time.Time, s *senderSnapsho
 		bytesDuplicate:       r.bytesDuplicate,
 		headerBytesDuplicate: r.headerBytesDuplicate,
 		packetsOutOfOrder:    s.packetsOutOfOrder + s.intervalStats.packetsOutOfOrder,
-		packetsLostFeed:      r.packetsLost,
+		packetsLostFeed:      s.packetsLost + s.intervalStats.packetsLost,
 		packetsLost:          r.packetsLostFromRR,
 		frames:               s.frames + s.intervalStats.frames,
 		nacks:                r.nacks,