From 1dc9b8fc5cda07a7d892a430e4a9144765171425 Mon Sep 17 00:00:00 2001
From: Raja Subramanian <raja.gobi@tutanota.com>
Date: Fri, 7 Nov 2025 21:46:14 +0530
Subject: [PATCH] Use buffered indicator to exclude from forwarding latency. 
 (#4062)

* Debug high forwarding latency missing.

* log highest

* log condition

* update log

* log

* log

* change log

* Track start up delay.

Digging into forwarding latency, there are a few things
1. Seems to be caused due to forwarding packets queued before bind. They
   would be in the queue till bind. There are two ways it is showing up
   a. Bind itself is delayed and releasing queued packets causes the
      high forwarding latency.
   b. There is a significant gap between bind and first packet being
      pulled off the queue to be forwarded, in one example 100ms.

(a) is understandable if the signalling delays things. Can drop these
packets without forwarding or indicate in the packet that it is a queued
packet and drop it from forwarding latency calculation. Dropping is
probably better as down stream components like egress will see a burst
in these situations.

(b) looks like go scheduling latency? Unsure.

Logging more to understand this better.

* log start

* Use buffered indicator to exclude from forwarding latency.

Buffered packets live the queue for a while before Bind releases them.
They have high(ish) queuing latency and not true representation of
forwarding latency.
---
 pkg/sfu/buffer/buffer.go | 18 +++++++++++-------
 pkg/sfu/receiver.go      |  2 +-
 2 files changed, 12 insertions(+), 8 deletions(-)

diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go
index 4ceb8430c..929ddabe9 100644
--- a/pkg/sfu/buffer/buffer.go
+++ b/pkg/sfu/buffer/buffer.go
@@ -74,6 +74,7 @@ type ExtPacket struct {
 	DependencyDescriptor *ExtDependencyDescriptor
 	AbsCaptureTimeExt    *act.AbsCaptureTime
 	IsOutOfOrder         bool
+	IsBuffered           bool
 }
 
 // VideoSize represents video resolution
@@ -323,9 +324,11 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
 		}
 	}
 
-	b.logger.Debugw("releasing queued packets on bind")
+	if len(b.pPackets) != 0 {
+		b.logger.Debugw("releasing queued packets on bind")
+	}
 	for _, pp := range b.pPackets {
-		b.calc(pp.packet, nil, pp.arrivalTime, false)
+		b.calc(pp.packet, nil, pp.arrivalTime, false, true)
 	}
 	b.pPackets = nil
 	b.bound = true
@@ -436,7 +439,7 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) {
 		return
 	}
 
-	b.calc(pkt, &rtpPacket, now, false)
+	b.calc(pkt, &rtpPacket, now, false, false)
 	b.readCond.Broadcast()
 	b.Unlock()
 	return
@@ -493,7 +496,7 @@ func (b *Buffer) writeRTX(rtxPkt *rtp.Packet, arrivalTime int64) {
 		return
 	}
 
-	b.calc(b.rtxPktBuf[:n], &repairedPkt, arrivalTime, true)
+	b.calc(b.rtxPktBuf[:n], &repairedPkt, arrivalTime, true, false)
 	b.readCond.Broadcast()
 }
 
@@ -624,7 +627,7 @@ func (b *Buffer) SetRTT(rtt uint32) {
 	}
 }
 
-func (b *Buffer) calc(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime int64, isRTX bool) {
+func (b *Buffer) calc(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime int64, isRTX bool, isBuffered bool) {
 	defer func() {
 		b.doNACKs()
 
@@ -727,7 +730,7 @@ func (b *Buffer) calc(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime int64, i
 		return
 	}
 
-	ep := b.getExtPacket(rtpPacket, arrivalTime, flowState)
+	ep := b.getExtPacket(rtpPacket, arrivalTime, isBuffered, flowState)
 	if ep == nil {
 		return
 	}
@@ -899,7 +902,7 @@ func (b *Buffer) processHeaderExtensions(p *rtp.Packet, arrivalTime int64, isRTX
 	}
 }
 
-func (b *Buffer) getExtPacket(rtpPacket *rtp.Packet, arrivalTime int64, flowState rtpstats.RTPFlowState) *ExtPacket {
+func (b *Buffer) getExtPacket(rtpPacket *rtp.Packet, arrivalTime int64, isBuffered bool, flowState rtpstats.RTPFlowState) *ExtPacket {
 	ep := &ExtPacket{
 		Arrival:           arrivalTime,
 		ExtSequenceNumber: flowState.ExtSequenceNumber,
@@ -910,6 +913,7 @@ func (b *Buffer) getExtPacket(rtpPacket *rtp.Packet, arrivalTime int64, flowStat
 			Temporal: InvalidLayerTemporal,
 		},
 		IsOutOfOrder: flowState.IsOutOfOrder,
+		IsBuffered:   isBuffered,
 	}
 
 	if len(rtpPacket.Payload) == 0 {
diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go
index c03110031..0b46f299f 100644
--- a/pkg/sfu/receiver.go
+++ b/pkg/sfu/receiver.go
@@ -830,7 +830,7 @@ func (w *WebRTCReceiver) forwardRTP(layer int32, buff *buffer.Buffer) {
 		}
 
 		// track delay/jitter
-		if writeCount.Load() > 0 && w.forwardStats != nil {
+		if writeCount.Load() > 0 && w.forwardStats != nil && !pkt.IsBuffered {
 			if latency, isHigh := w.forwardStats.Update(pkt.Arrival, mono.UnixNano()); isHigh {
 				w.logger.Infow(
 					"high forwarding latency",