sfu: report forward scheduling delay next to forwarding latency (#5065)

* sfu: report forward scheduling delay next to forwarding latency

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* sfu: keep full forwarding transit when packing scheduling delay

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* sfu: saturate reported forwarding p90s instead of wrapping

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

go.mod

@@ -21,7 +21,7 @@ require (
 	github.com/jxskiss/base62 v1.1.0
 	github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731
 	github.com/livekit/mediatransportutil v0.0.0-20261014192331-dd8e31a0adc0
-	github.com/livekit/protocol v1.53.1-0.20261023175527-46332a036e25
+	github.com/livekit/protocol v1.53.1-0.20261026092902-78ad907b61e0
 	github.com/livekit/psrpc v0.8.2
 	github.com/mackerelio/go-osstat v0.2.8
 	github.com/magefile/mage v1.17.2

go.sum

@@ -166,8 +166,8 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT
 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ=
 github.com/livekit/mediatransportutil v0.0.0-20261014192331-dd8e31a0adc0 h1:BSLyHr3Lsl71bHGHsfiYNmbohugR2yoaf9HjAZeFJnc=
 github.com/livekit/mediatransportutil v0.0.0-20261014192331-dd8e31a0adc0/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU=
-github.com/livekit/protocol v1.53.1-0.20261023175527-46332a036e25 h1:v9UMGn8kGnrs0gja5795d8RBCQqKCVlXVeH14UDiuwc=
-github.com/livekit/protocol v1.53.1-0.20261023175527-46332a036e25/go.mod h1:h9P4KCc3K715NCgCrkkFsb8/RRY1kKqOS1EIHUNRlCk=
+github.com/livekit/protocol v1.53.1-0.20261026092902-78ad907b61e0 h1:YUmqxA8BFQMnJ8YLVuB0t9eAYDitz3icWkfk3Ua+4Wc=
+github.com/livekit/protocol v1.53.1-0.20261026092902-78ad907b61e0/go.mod h1:h9P4KCc3K715NCgCrkkFsb8/RRY1kKqOS1EIHUNRlCk=
 github.com/livekit/psrpc v0.8.2 h1:7tbkhBJoYlK5Csfkdv1XYbZjyAQeaEZgupvIXNCv7Dc=
 github.com/livekit/psrpc v0.8.2/go.mod h1:6KnU8vULoizX0Er9NNfeKJDV8ug87j3fuo19M0bhj6Q=
 github.com/livekit/webrtc-pion/v4 v4.2.18-warp.7 h1:I1UkbeJI5V6slqLmef0JoaGF8wPnFsKzzXGUzz4KDvs=

pkg/sfu/buffer/buffer_base.go

@@ -88,6 +88,8 @@ type ExtPacket struct {
 	AbsCaptureTimeExt    *act.AbsCaptureTime
 	IsOutOfOrder         bool
 	IsBuffered           bool
+	// the reader was waiting on an empty buffer when this packet arrived
+	ReaderWasWaiting bool
 
 	// backing storage for Packet when the buffer creates the ext packet,
 	// extension and CSRC slices point into the write input until ReadExtended re-points them into RawPacket
@@ -170,7 +172,8 @@ type BufferBase struct {
 
 	params BufferBaseParams
 
-	readCond *sync.Cond
+	readCond      *sync.Cond
+	readerWaiting bool
 
 	bucket               *bucket.Bucket[uint64, uint16]
 	lastBucketCapCheckAt int64
@@ -674,7 +677,9 @@ func (b *BufferBase) ReadExtended(buf []byte) (*ExtPacket, error) {
 			return patched, nil
 		}
 
+		b.readerWaiting = true
 		b.readCond.Wait()
+		b.readerWaiting = false
 	}
 }
 
@@ -947,6 +952,7 @@ func (b *BufferBase) HandleIncomingPacketLocked(
 	if ep == nil {
 		return 0, errors.New("could not get ext packet")
 	}
+	ep.ReaderWasWaiting = b.readerWaiting && b.extPackets.Len() == 0
 	b.extPackets.PushBack(ep)
 	b.readCond.Broadcast()
 

pkg/sfu/buffer/buffer_test.go

@@ -779,3 +779,54 @@ func TestGetPacketConcurrentWithWrite(t *testing.T) {
 		wg.Wait()
 	})
 }
+
+func TestReaderWasWaiting(t *testing.T) {
+	buff := NewBuffer(123, 1, 1)
+	buff.OnRtcpFeedback(func(_ []rtcp.Packet) {})
+	require.NoError(t, buff.Bind(webrtc.RTPParameters{Codecs: []webrtc.RTPCodecParameters{opusCodec}}, opusCodec.RTPCodecCapability, 0))
+	write := func(sn uint16) {
+		pkt := &rtp.Packet{
+			Header:  rtp.Header{Version: 2, PayloadType: 111, SequenceNumber: sn, Timestamp: uint32(sn) * 960, SSRC: 123},
+			Payload: make([]byte, 100),
+		}
+		b, err := pkt.Marshal()
+		require.NoError(t, err)
+		_, err = buff.Write(b)
+		require.NoError(t, err)
+	}
+	read := func() *ExtPacket {
+		extPkt, err := buff.ReadExtended(make([]byte, 1500))
+		require.NoError(t, err)
+		require.NotNil(t, extPkt)
+		return extPkt
+	}
+
+	// packets that arrive while nobody reads queue up, none found the reader waiting
+	write(1)
+	write(2)
+	require.False(t, read().ReaderWasWaiting)
+	require.False(t, read().ReaderWasWaiting)
+
+	// a reader parked on the empty buffer is found waiting by the next packet
+	got := make(chan bool, 1)
+	go func() {
+		extPkt, err := buff.ReadExtended(make([]byte, 1500))
+		got <- err == nil && extPkt != nil && extPkt.ReaderWasWaiting
+	}()
+	require.Eventually(t, func() bool {
+		buff.RLock()
+		defer buff.RUnlock()
+		return buff.readerWaiting
+	}, time.Second, time.Millisecond)
+	write(3)
+	require.True(t, <-got)
+
+	// two packets that arrive before the parked reader wakes: only the first found it waiting
+	buff.Lock()
+	buff.readerWaiting = true
+	buff.Unlock()
+	write(4)
+	write(5)
+	require.True(t, read().ReaderWasWaiting)
+	require.False(t, read().ReaderWasWaiting)
+}

pkg/sfu/forwardstats.go

@@ -1,6 +1,7 @@
 package sfu
 
 import (
+	"math"
 	"math/bits"
 	"sync"
 	"time"
@@ -31,7 +32,7 @@ const (
 	forwardSampleShardSel  = forwardSampleNumShards - 1
 )
 
-// forwardSampleShard is a ring of transit samples with multiple producers and a
+// forwardSampleShard is a ring of packed samples with multiple producers and a
 // single consumer. A producer reserves a slot, stores the value, then publishes
 // the slot's epoch (reserved index + 1). The consumer reads a slot only once its
 // epoch marks the value committed for that index.
@@ -42,7 +43,7 @@ type forwardSampleShard struct {
 	seq      [forwardSampleShardCap]atomic.Uint64 // per-slot publish epoch
 }
 
-// forwardSampleBuffer holds per-packet transit samples produced on the packet
+// forwardSampleBuffer holds per-packet samples produced on the packet
 // path and consumed by the background worker, which performs metric emission.
 type forwardSampleBuffer struct {
 	shards  [forwardSampleNumShards]forwardSampleShard
@@ -51,11 +52,11 @@ type forwardSampleBuffer struct {
 
 // push records a sample: reserve a slot, store the value, then publish the
 // slot's epoch. The shard is selected from arrival time bits.
-func (b *forwardSampleBuffer) push(arrival, transitNs int64) {
+func (b *forwardSampleBuffer) push(arrival, sample int64) {
 	sh := &b.shards[(uint64(arrival)>>6)&forwardSampleShardSel]
 	i := sh.writeIdx.Add(1) - 1
 	slot := i & forwardSampleShardMask
-	sh.ring[slot].Store(transitNs)
+	sh.ring[slot].Store(sample)
 	sh.seq[slot].Store(i + 1)
 }
 
@@ -67,7 +68,7 @@ func (b *forwardSampleBuffer) push(arrival, transitNs int64) {
 // draining stops and resumes from there on the next call, so no sample is read
 // stale or skipped. When producers get a shard's capacity ahead, or overwrite a
 // slot before it is read, the affected samples are counted as dropped.
-func (b *forwardSampleBuffer) drain(fn func(transitNs int64)) {
+func (b *forwardSampleBuffer) drain(fn func(sample int64)) {
 	for si := range b.shards {
 		sh := &b.shards[si]
 		w := sh.writeIdx.Load()
@@ -100,6 +101,34 @@ func (b *forwardSampleBuffer) takeDropped() uint64 {
 	return b.dropped.Swap(0)
 }
 
+// A sample packs transit (high 40 bits, in ns, up to ~18 minutes) and scheduling delay
+// (low 24 bits, in 256 ns units, up to ~4.29s) into one slot, so the second value costs
+// no extra atomic.
+const (
+	forwardSchedulingDelayBits  = 24
+	forwardSchedulingDelayShift = 8 // 256 ns units
+	forwardNoSchedulingDelay    = 1<<forwardSchedulingDelayBits - 1
+	forwardMaxTransitNs         = 1<<(64-forwardSchedulingDelayBits) - 1
+)
+
+// packForwardSample packs a sample, a negative schedulingDelayNs meaning none.
+func packForwardSample(transitNs, schedulingDelayNs int64) int64 {
+	delay := uint64(forwardNoSchedulingDelay)
+	if schedulingDelayNs >= 0 {
+		delay = min(uint64(schedulingDelayNs)>>forwardSchedulingDelayShift, forwardNoSchedulingDelay-1)
+	}
+	return int64(uint64(min(max(transitNs, 0), forwardMaxTransitNs))<<forwardSchedulingDelayBits | delay)
+}
+
+// unpackForwardSample returns the transit and scheduling delay, -1 when there is none.
+func unpackForwardSample(sample int64) (transitNs, schedulingDelayNs int64) {
+	transitNs = int64(uint64(sample) >> forwardSchedulingDelayBits)
+	if delay := uint64(sample) & forwardNoSchedulingDelay; delay != forwardNoSchedulingDelay {
+		return transitNs, int64(delay << forwardSchedulingDelayShift)
+	}
+	return transitNs, -1
+}
+
 // forwardSummary is a mergeable histogram of forwarding transit over an
 // interval. Each power-of-two octave [2^e, 2^(e+1)) us is split into
 // forwardHistSub linear sub-buckets, and each bucket keeps the observed
@@ -221,6 +250,19 @@ func (s forwardSummary) percentile(p float64) time.Duration {
 	return time.Duration(s.maxNs)
 }
 
+// forwardIntervalSummary holds one summary interval's transit and scheduling delay.
+type forwardIntervalSummary struct {
+	transit    forwardSummary
+	scheduling forwardSummary
+}
+
+func (s forwardIntervalSummary) merge(o forwardIntervalSummary) forwardIntervalSummary {
+	return forwardIntervalSummary{
+		transit:    s.transit.merge(o.transit),
+		scheduling: s.scheduling.merge(o.scheduling),
+	}
+}
+
 type ForwardStats struct {
 	samples forwardSampleBuffer
 
@@ -228,7 +270,7 @@ type ForwardStats struct {
 	// written by the background worker (flush) and read both by the worker
 	// (report) and by external callers (GetStats), so it is guarded by lock.
 	lock     sync.Mutex
-	ring     []forwardSummary
+	ring     []forwardIntervalSummary
 	ringHead int
 	ringLen  int
 
@@ -245,7 +287,7 @@ func NewForwardStats(summaryInterval, reportInterval, reportWindow time.Duration
 	}
 
 	s := &ForwardStats{
-		ring:            make([]forwardSummary, ringCap),
+		ring:            make([]forwardIntervalSummary, ringCap),
 		summaryInterval: summaryInterval,
 		reportInterval:  reportInterval,
 		closeCh:         make(chan struct{}),
@@ -259,8 +301,14 @@ func NewForwardStats(summaryInterval, reportInterval, reportWindow time.Duration
 // and returns the transit and whether it exceeds the high-latency threshold.
 // The sample is aggregated and emitted by the background worker.
 func (s *ForwardStats) Update(arrival, left int64) (int64, bool) {
+	return s.UpdateWithSchedulingDelay(arrival, left, -1)
+}
+
+// UpdateWithSchedulingDelay is Update that also records a scheduling delay when it is
+// not negative.
+func (s *ForwardStats) UpdateWithSchedulingDelay(arrival, left, schedulingDelay int64) (int64, bool) {
 	transit := left - arrival
-	s.samples.push(arrival, transit)
+	s.samples.push(arrival, packForwardSample(transit, schedulingDelay))
 	return transit, time.Duration(transit) > cHighForwardingLatency
 }
 
@@ -294,10 +342,14 @@ func (s *ForwardStats) run() {
 // histogram, and folds the interval summary into the window ring used for the
 // latency gauge.
 func (s *ForwardStats) flush() {
-	var summ forwardSummary
-	s.samples.drain(func(transitNs int64) {
+	var summ forwardIntervalSummary
+	s.samples.drain(func(sample int64) {
+		transitNs, schedulingDelayNs := unpackForwardSample(sample)
 		prometheus.RecordForwardLatencySample(transitNs)
-		summ.addSample(transitNs)
+		summ.transit.addSample(transitNs)
+		if schedulingDelayNs >= 0 {
+			summ.scheduling.addSample(schedulingDelayNs)
+		}
 	})
 
 	s.lock.Lock()
@@ -311,7 +363,7 @@ func (s *ForwardStats) flush() {
 
 // summarize merges the ring summaries covering the most recent window. A
 // window <= 0 (or >= the report window) covers the entire ring.
-func (s *ForwardStats) summarize(window time.Duration) forwardSummary {
+func (s *ForwardStats) summarize(window time.Duration) forwardIntervalSummary {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
@@ -327,7 +379,7 @@ func (s *ForwardStats) summarize(window time.Duration) forwardSummary {
 	}
 
 	// walk backwards from the most recent entry (ringHead-1) over n entries.
-	var w forwardSummary
+	var w forwardIntervalSummary
 	for i := 0; i < n; i++ {
 		idx := (s.ringHead - 1 - i + len(s.ring)) % len(s.ring)
 		w = w.merge(s.ring[idx])
@@ -341,11 +393,12 @@ func (s *ForwardStats) summarize(window time.Duration) forwardSummary {
 // that covers it). A duration <= 0, or one that meets/exceeds the report window,
 // covers the full window.
 func (s *ForwardStats) GetStats(duration time.Duration) time.Duration {
-	return s.summarize(duration).percentile(forwardLatencyPercentile)
+	return s.summarize(duration).transit.percentile(forwardLatencyPercentile)
 }
 
 func (s *ForwardStats) report() {
-	w := s.summarize(0)
+	ws := s.summarize(0)
+	w := ws.transit
 
 	p90 := w.percentile(forwardLatencyPercentile)
 	if dropped := s.samples.takeDropped(); dropped > 0 {
@@ -363,5 +416,11 @@ func (s *ForwardStats) report() {
 		)
 	}
 
-	prometheus.RecordForwardLatency(uint32(p90.Nanoseconds()))
+	prometheus.RecordForwardLatency(saturatingNs(p90))
+	prometheus.RecordForwardSchedulingDelay(saturatingNs(ws.scheduling.percentile(forwardLatencyPercentile)))
+}
+
+// saturatingNs converts to uint32 nanoseconds, saturating at ~4.29s instead of wrapping.
+func saturatingNs(d time.Duration) uint32 {
+	return uint32(min(max(d.Nanoseconds(), 0), math.MaxUint32))
 }

pkg/sfu/forwardstats_test.go

@@ -1,6 +1,7 @@
 package sfu
 
 import (
+	"math"
 	"sync"
 	"testing"
 	"time"
@@ -263,7 +264,7 @@ func TestForwardSampleBuffer_Concurrent(t *testing.T) {
 // ---------------------------------------------------------------------------
 
 func TestForwardStats_Update(t *testing.T) {
-	s := &ForwardStats{ring: make([]forwardSummary, 1)}
+	s := &ForwardStats{ring: make([]forwardIntervalSummary, 1)}
 
 	// below threshold
 	transit, isHigh := s.Update(1000, 1000+int64(5*time.Millisecond))
@@ -283,7 +284,7 @@ func TestForwardStats_Update(t *testing.T) {
 func TestForwardStats_Flush(t *testing.T) {
 	initPrometheus(t)
 
-	s := &ForwardStats{ring: make([]forwardSummary, 4)}
+	s := &ForwardStats{ring: make([]forwardIntervalSummary, 4)}
 
 	for i := 0; i < 10; i++ {
 		s.Update(0, int64((i+1)*1000)) // 1us..10us
@@ -292,7 +293,7 @@ func TestForwardStats_Flush(t *testing.T) {
 	s.flush()
 
 	require.Equal(t, 1, s.ringLen)
-	summ := s.ring[0]
+	summ := s.ring[0].transit
 	require.Equal(t, int64(10), summ.count)
 	require.Equal(t, int64(1000), summ.minNs)
 	require.Equal(t, int64(10000), summ.maxNs)
@@ -301,14 +302,69 @@ func TestForwardStats_Flush(t *testing.T) {
 	// a subsequent flush with no new samples appends an empty summary
 	s.flush()
 	require.Equal(t, 2, s.ringLen)
-	require.Equal(t, int64(0), s.ring[1].count)
+	require.Equal(t, int64(0), s.ring[1].transit.count)
+}
+
+func TestForwardSample_Pack(t *testing.T) {
+	transit, delay := unpackForwardSample(packForwardSample(250_000, 7_168))
+	require.Equal(t, int64(250_000), transit)
+	require.Equal(t, int64(7_168), delay)
+
+	// scheduling delay keeps 256 ns resolution
+	_, delay = unpackForwardSample(packForwardSample(250_000, 7_300))
+	require.Equal(t, int64(7_168), delay)
+
+	// no scheduling delay, and a zero delay is a sample, not "none"
+	_, delay = unpackForwardSample(packForwardSample(1_000, -1))
+	require.Equal(t, int64(-1), delay)
+	_, delay = unpackForwardSample(packForwardSample(1_000, 0))
+	require.Equal(t, int64(0), delay)
+
+	// a long stall keeps its full transit; the delay saturates and never reads as "none"
+	transit, delay = unpackForwardSample(packForwardSample(10*int64(time.Second), 10*int64(time.Second)))
+	require.Equal(t, 10*int64(time.Second), transit)
+	require.Equal(t, int64((forwardNoSchedulingDelay-1)<<forwardSchedulingDelayShift), delay)
+	transit, _ = unpackForwardSample(packForwardSample(math.MaxInt64, -1))
+	require.Equal(t, int64(forwardMaxTransitNs), transit)
+
+	// a negative transit (clock skew) clamps to zero
+	transit, _ = unpackForwardSample(packForwardSample(-5, -1))
+	require.Equal(t, int64(0), transit)
+}
+
+func TestSaturatingNs(t *testing.T) {
+	require.Equal(t, uint32(1_000_000), saturatingNs(time.Millisecond))
+	// a 5s p90 saturates instead of wrapping to ~705ms
+	require.Equal(t, uint32(math.MaxUint32), saturatingNs(5*time.Second))
+	require.Equal(t, uint32(0), saturatingNs(-time.Second))
+}
+
+func TestForwardStats_SchedulingDelay(t *testing.T) {
+	initPrometheus(t)
+
+	s := &ForwardStats{ring: make([]forwardIntervalSummary, 2)}
+
+	// a burst: only the first packet found the forwarder idle
+	s.UpdateWithSchedulingDelay(0, 300_000, 4_096)
+	for i := 0; i < 9; i++ {
+		s.Update(0, 300_000)
+	}
+	s.flush()
+
+	summ := s.ring[0]
+	require.Equal(t, int64(10), summ.transit.count)
+	require.Equal(t, int64(1), summ.scheduling.count)
+	require.Equal(t, 4096*time.Nanosecond, summ.scheduling.percentile(forwardLatencyPercentile))
+	require.Equal(t, 300*time.Microsecond, s.GetStats(0))
+
+	require.NotPanics(t, s.report)
 }
 
 func TestForwardStats_ReportWindow(t *testing.T) {
 	initPrometheus(t)
 
 	// window of 3 summary buckets
-	s := &ForwardStats{ring: make([]forwardSummary, 3)}
+	s := &ForwardStats{ring: make([]forwardIntervalSummary, 3)}
 
 	s.Update(0, 1000)
 	s.flush()
@@ -318,7 +374,7 @@ func TestForwardStats_ReportWindow(t *testing.T) {
 	// report merges the whole window without panicking and reflects both samples
 	var w forwardSummary
 	for i := 0; i < s.ringLen; i++ {
-		w = w.merge(s.ring[i])
+		w = w.merge(s.ring[i].transit)
 	}
 	require.Equal(t, int64(2), w.count)
 	require.Equal(t, int64(1000), w.minNs)
@@ -331,7 +387,7 @@ func TestForwardStats_GetStats(t *testing.T) {
 	initPrometheus(t)
 
 	// 5 buckets, each covering one 100ms summary interval.
-	s := &ForwardStats{ring: make([]forwardSummary, 5), summaryInterval: 100 * time.Millisecond}
+	s := &ForwardStats{ring: make([]forwardIntervalSummary, 5), summaryInterval: 100 * time.Millisecond}
 
 	// fold five 100ms buckets, one sample each, descending 5ms..1ms so trailing
 	// windows have distinct maxima.
@@ -395,7 +451,7 @@ func benchArrival(base, i int64) int64 {
 }
 
 func BenchmarkForwardStatsUpdate(b *testing.B) {
-	s := &ForwardStats{ring: make([]forwardSummary, 1)}
+	s := &ForwardStats{ring: make([]forwardIntervalSummary, 1)}
 
 	var gid atomic.Int64
 	b.RunParallel(func(pb *testing.PB) {

pkg/sfu/receiver_base.go

@@ -978,7 +978,13 @@ func (r *ReceiverBase) forwardRTP(
 		// serially, inflating the measured transit for the tail of the burst. That
 		// reflects loss recovery rather than steady-state forwarding health.
 		if writeCount > 0 && r.forwardStats != nil && !extPkt.IsBuffered && !extPkt.IsOutOfOrder {
-			if latency, isHigh := r.forwardStats.Update(extPkt.Arrival, mono.UnixNano()); isHigh {
+			// a packet that found the forwarder waiting measures only its wake-up; a queued
+			// one also waited behind earlier packets, so it carries no scheduling delay
+			schedulingDelay := int64(-1)
+			if extPkt.ReaderWasWaiting {
+				schedulingDelay = dequeuedAt - extPkt.Arrival
+			}
+			if latency, isHigh := r.forwardStats.UpdateWithSchedulingDelay(extPkt.Arrival, mono.UnixNano(), schedulingDelay); isHigh {
 				r.params.Logger.Debugw(
 					"high forwarding latency",
 					"latency", time.Duration(latency),

pkg/sfu/receiver_test.go

@@ -17,19 +17,23 @@ package sfu
 import (
 	"fmt"
 	"hash/fnv"
+	"io"
 	"math/rand"
 	"runtime"
 	"sync"
 	"testing"
 
 	"github.com/gammazero/workerpool"
+	"github.com/pion/rtp"
 	"github.com/pion/webrtc/v4"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/atomic"
 
+	"github.com/livekit/livekit-server/pkg/sfu/buffer"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
+	"github.com/livekit/protocol/utils/mono"
 )
 
 func TestWebRTCReceiver_OnCloseHandler(t *testing.T) {
@@ -128,6 +132,79 @@ func TestReceiverBaseAddDownTrackReturnsClosedWhenCloseWinsRace(t *testing.T) {
 	require.Empty(t, receiver.GetDownTracks())
 }
 
+// burstBuffer serves one packet per flag from ReadExtended, then io.EOF. A flag says
+// whether the packet found the reader waiting.
+type burstBuffer struct {
+	buffer.BufferProvider
+	readerWasWaiting []bool
+}
+
+func (b *burstBuffer) ReadExtended([]byte) (*buffer.ExtPacket, error) {
+	if len(b.readerWasWaiting) == 0 {
+		return nil, io.EOF
+	}
+	readerWasWaiting := b.readerWasWaiting[0]
+	b.readerWasWaiting = b.readerWasWaiting[1:]
+	return &buffer.ExtPacket{
+		Arrival:          mono.UnixNano(),
+		Packet:           &rtp.Packet{Header: rtp.Header{PayloadType: 111}},
+		ReaderWasWaiting: readerWasWaiting,
+	}, nil
+}
+
+type writeTestSender struct {
+	receiverCloseTestSender
+}
+
+func (s *writeTestSender) WriteRTP(*buffer.ExtPacket, int32) int32 {
+	return 1
+}
+
+func (s *writeTestSender) Close() {}
+
+func TestReceiverBaseSchedulingDelayOnlyWhenIdle(t *testing.T) {
+	receiver := NewReceiverBase(
+		ReceiverBaseParams{
+			TrackID:  "track",
+			StreamID: "stream",
+			Kind:     webrtc.RTPCodecTypeAudio,
+			Codec: webrtc.RTPCodecParameters{
+				RTPCodecCapability: webrtc.RTPCodecCapability{
+					MimeType:  webrtc.MimeTypeOpus,
+					ClockRate: 48000,
+					Channels:  2,
+				},
+				PayloadType: 111,
+			},
+			Logger: logger.GetLogger(),
+		},
+		&livekit.TrackInfo{
+			Sid:    "track",
+			Type:   livekit.TrackType_AUDIO,
+			Source: livekit.TrackSource_MICROPHONE,
+		},
+		ReceiverCodecStateNormal,
+	)
+	defer receiver.Close("test", false)
+
+	fs := &ForwardStats{ring: make([]forwardIntervalSummary, 1)}
+	receiver.SetForwardStats(fs)
+	require.NoError(t, receiver.AddDownTrack(&writeTestSender{receiverCloseTestSender{subscriberID: "subscriber"}}))
+
+	// a burst of five where only the first found the forwarder waiting, then one more that did
+	buff := &burstBuffer{readerWasWaiting: []bool{true, false, false, false, false, true}}
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	receiver.forwardRTP(0, buff, receiver.forwardersGeneration.Load(), &wg)
+
+	fs.flush()
+	summ := fs.ring[0]
+	require.Equal(t, int64(6), summ.transit.count)
+	// the burst's later packets queued behind earlier ones, so they carry no scheduling delay
+	require.Equal(t, int64(2), summ.scheduling.count)
+}
+
 func BenchmarkWriteRTP(b *testing.B) {
 	cases := []int{1, 2, 5, 10, 100, 250, 500}
 	workers := runtime.NumCPU()

pkg/telemetry/prometheus/node.go

@@ -174,6 +174,7 @@ func GetNodeStats(nodeStartedAt int64, prevStats []*livekit.NodeStats, rateInter
 		ParticipantRtcCanceled:     participantRTCCanceled.Load(),
 		ParticipantRtcActive:       participantRTCActive.Load(),
 		ForwardLatency:             forwardLatency.Load(),
+		ForwardSchedulingDelay:     forwardSchedulingDelay.Load(),
 		NumCpus:                    uint32(cpuStats.NumCPU()), // this will round down to the nearest integer
 		CpuLoad:                    float32(cpuStats.GetCPULoad()),
 		MemoryTotal:                memTotal,

pkg/telemetry/prometheus/packets.go

@@ -51,30 +51,32 @@ var (
 	participantRTCCanceled     atomic.Uint64
 	participantRTCActive       atomic.Uint64
 	forwardLatency             atomic.Uint32
+	forwardSchedulingDelay     atomic.Uint32
 
-	promPacketLabels          = []string{"direction", "transmission", "country"}
-	promPacketTotal           *prometheus.CounterVec
-	promPacketBytes           *prometheus.CounterVec
-	promFECLabels             = []string{"direction", "type"}
-	promFECPacketTotal        *prometheus.CounterVec
-	promFECPacketBytes        *prometheus.CounterVec
-	promRTCPLabels            = []string{"direction", "country"}
-	promStreamLabels          = []string{"direction", "source", "type", "country"}
-	promNackTotal             *prometheus.CounterVec
-	promRTXCappedTotal        *prometheus.CounterVec
-	promKeyFrameTotal         *prometheus.CounterVec
-	promPliTotal              *prometheus.CounterVec
-	promFirTotal              *prometheus.CounterVec
-	promPacketLossTotal       *prometheus.CounterVec
-	promPacketLoss            *prometheus.HistogramVec
-	promPacketOutOfOrderTotal *prometheus.CounterVec
-	promPacketOutOfOrder      *prometheus.HistogramVec
-	promJitter                *prometheus.HistogramVec
-	promRTT                   *prometheus.HistogramVec
-	promParticipantJoin       *prometheus.CounterVec
-	promConnections           *prometheus.GaugeVec
-	promForwardLatency        prometheus.Gauge
-	promForwardLatencyHist    prometheus.Histogram
+	promPacketLabels           = []string{"direction", "transmission", "country"}
+	promPacketTotal            *prometheus.CounterVec
+	promPacketBytes            *prometheus.CounterVec
+	promFECLabels              = []string{"direction", "type"}
+	promFECPacketTotal         *prometheus.CounterVec
+	promFECPacketBytes         *prometheus.CounterVec
+	promRTCPLabels             = []string{"direction", "country"}
+	promStreamLabels           = []string{"direction", "source", "type", "country"}
+	promNackTotal              *prometheus.CounterVec
+	promRTXCappedTotal         *prometheus.CounterVec
+	promKeyFrameTotal          *prometheus.CounterVec
+	promPliTotal               *prometheus.CounterVec
+	promFirTotal               *prometheus.CounterVec
+	promPacketLossTotal        *prometheus.CounterVec
+	promPacketLoss             *prometheus.HistogramVec
+	promPacketOutOfOrderTotal  *prometheus.CounterVec
+	promPacketOutOfOrder       *prometheus.HistogramVec
+	promJitter                 *prometheus.HistogramVec
+	promRTT                    *prometheus.HistogramVec
+	promParticipantJoin        *prometheus.CounterVec
+	promConnections            *prometheus.GaugeVec
+	promForwardLatency         prometheus.Gauge
+	promForwardSchedulingDelay prometheus.Gauge
+	promForwardLatencyHist     prometheus.Histogram
 )
 
 func initPacketStats(nodeID string, nodeType livekit.NodeType) {
@@ -193,6 +195,13 @@ func initPacketStats(nodeID string, nodeType livekit.NodeType) {
 		Name:        "latency",
 		ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()},
 	})
+	promForwardSchedulingDelay = prometheus.NewGauge(prometheus.GaugeOpts{
+		Namespace:   livekitNamespace,
+		Subsystem:   "forward",
+		Name:        "scheduling_delay",
+		Help:        "p90 time from arrival to dequeue, in ns, for packets that found their forwarder idle.",
+		ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()},
+	})
 	promForwardLatencyHist = prometheus.NewHistogram(prometheus.HistogramOpts{
 		Namespace:   livekitNamespace,
 		Subsystem:   "forward_latency",
@@ -231,6 +240,7 @@ func initPacketStats(nodeID string, nodeType livekit.NodeType) {
 	prometheus.MustRegister(promParticipantJoin)
 	prometheus.MustRegister(promConnections)
 	prometheus.MustRegister(promForwardLatency)
+	prometheus.MustRegister(promForwardSchedulingDelay)
 	prometheus.MustRegister(promForwardLatencyHist)
 }
 
@@ -442,3 +452,8 @@ func RecordForwardLatency(p90 uint32) {
 	forwardLatency.Store(p90)
 	promForwardLatency.Set(float64(p90))
 }
+
+func RecordForwardSchedulingDelay(p90 uint32) {
+	forwardSchedulingDelay.Store(p90)
+	promForwardSchedulingDelay.Set(float64(p90))
+}