DownTrack: arm key frame requester timer only while a layer lock is pending (#4980)

The key frame requester re-armed its timer on every loop, so every video down
track woke up every 200 ms to 1 s for its whole life, even when the layer was
locked and there was nothing to do.

Re-arm the timer only while the requested layer is not locked, or while the
lock can change without an event: an acquisition grace is armed, or layers are
selected with the dependency descriptor, which can lose the lock on the packet
path. Post the event on resync, mute, max published layer change and packet
path layer switch, since those can change the lock without an allocation.

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

pkg/sfu/downtrack.go

@@ -341,6 +341,7 @@ type DownTrack struct {
 	keyFrameRequesterChMu     sync.RWMutex
 	keyFrameRequesterCh       chan struct{}
 	keyFrameRequesterChClosed bool
+	keyFrameRequesterWakeups  atomic.Uint32
 
 	retransmitChMu     sync.RWMutex
 	retransmitCh       chan struct{}
@@ -956,18 +957,20 @@ func (d *DownTrack) keyFrameRequester() {
 	defer timer.Stop()
 
 	for !d.IsClosed() {
-		timer.Reset(getInterval())
-
 		select {
 		case _, more := <-d.keyFrameRequesterCh:
 			if !more {
 				return
 			}
 			if !timer.Stop() {
-				<-timer.C
+				select {
+				case <-timer.C:
+				default:
+				}
 			}
 		case <-timer.C:
 		}
+		d.keyFrameRequesterWakeups.Inc()
 
 		locked, layer := d.forwarder.CheckSync()
 		if !locked && layer != buffer.InvalidLayerSpatial && d.writable.Load() {
@@ -984,6 +987,13 @@ func (d *DownTrack) keyFrameRequester() {
 				sal.OnAvailableLayersChanged(d)
 			}
 		}
+
+		// Re-arm only while there is something to wait for: a requested layer that is not locked yet,
+		// or a lock state that can change without an event. Once locked, sleep until the
+		// next event so a locked track does not wake up every interval.
+		if (!locked && layer != buffer.InvalidLayerSpatial) || d.forwarder.NeedsPeriodicSyncCheck() {
+			timer.Reset(getInterval())
+		}
 	}
 }
 
@@ -1197,6 +1207,8 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) int32 {
 
 	if tp.isSwitching {
 		d.postMaxLayerNotifierEvent("switching")
+		// the current layer moved on the packet path, so the layer lock may have changed
+		d.postKeyFrameRequestEvent()
 	}
 
 	if tp.isResuming {
@@ -1443,6 +1455,8 @@ func (d *DownTrack) handleMute(muted bool, changed bool) {
 	// and that could turn on/off layers on publisher side.
 	//
 	d.postMaxLayerNotifierEvent("mute")
+	// mute resyncs the forwarder, so the layer lock has to be re-acquired on unmute
+	d.postKeyFrameRequestEvent()
 
 	if sal := d.getStreamAllocatorListener(); sal != nil {
 		sal.OnSubscriptionChanged(d)
@@ -1659,6 +1673,8 @@ func (d *DownTrack) UpTrackBitrateAvailabilityChange() {
 
 func (d *DownTrack) UpTrackMaxPublishedLayerChange(maxPublishedLayer int32) {
 	if d.forwarder.SetMaxPublishedLayer(maxPublishedLayer) {
+		// may arm the acquisition grace, which the key frame requester has to expire
+		d.postKeyFrameRequestEvent()
 		if sal := d.getStreamAllocatorListener(); sal != nil {
 			sal.OnMaxPublishedSpatialChanged(d)
 		}
@@ -1804,6 +1820,7 @@ func (d *DownTrack) Pause() VideoAllocation {
 func (d *DownTrack) Resync() {
 	d.forwarder.Resync()
 	d.flushSequencer()
+	d.postKeyFrameRequestEvent()
 }
 
 // flushSequencer discards recorded packet metadata on a stream restart so that NACK

pkg/sfu/downtrack_downstream_integration_test.go

@@ -919,3 +919,67 @@ func TestDownTrackSendsProbePackets(t *testing.T) {
 		})
 	}
 }
+
+// -----------------------------------------------------------------------------
+// keyFrameRequester: PLI only while a layer lock is pending
+// -----------------------------------------------------------------------------
+
+// TestDownTrackKeyFrameRequesterArmsOnlyWhilePending verifies that the key frame
+// requester keeps sending PLIs while the target layer is not locked, and goes quiet
+// once a key frame on the target layer locks it (no periodic wake-ups while locked).
+func TestDownTrackKeyFrameRequesterArmsOnlyWhilePending(t *testing.T) {
+	h := vnettest.NewHosts(t)
+	factory := buffer.NewFactoryOfBufferFactory(500, 500).CreateBufferFactory()
+	p := pacer.NewPassThrough(logger.GetLogger(), newNullBWE())
+	t.Cleanup(p.Stop)
+
+	dh := newBoundDownTrack(t, h, factory, vp8CodecParams, p, vnettest.MediaEngineConfig{Video: true})
+
+	// allocate a target layer (spatial 0) without any media, so the lock is pending
+	var brs Bitrates
+	brs[0][0] = 100_000
+	dh.receiver.GetLayeredBitrateReturns([]int32{0}, brs)
+	dh.dt.UpTrackMaxPublishedLayerChange(0)
+	dh.dt.SetMaxSpatialLayer(0)
+	alloc := dh.dt.AllocateOptimal(false, false)
+	require.Equal(t, int32(0), alloc.TargetLayer.Spatial, "expected target layer to be allocated")
+
+	// unlocked: PLIs must keep coming at the key frame interval
+	require.Eventually(t, func() bool {
+		return dh.receiver.SendPLICallCount() >= 3
+	}, 5*time.Second, 20*time.Millisecond, "expected repeated PLIs while layer lock is pending")
+
+	// a key frame on the target layer locks it
+	vp8 := &codec.VP8{
+		FirstByte:  0x10,
+		S:          true,
+		PictureID:  1,
+		IsKeyFrame: true,
+	}
+	ep, err := testutils.GetTestExtPacketVP8(&testutils.TestExtPacketParams{
+		SequenceNumber: 5000,
+		Timestamp:      180000,
+		SSRC:           0x33333333,
+		PayloadType:    96,
+		PayloadSize:    50,
+		IsKeyFrame:     true,
+		Marker:         true,
+	}, vp8)
+	require.NoError(t, err)
+	ep.Packet.Payload = distinctivePayload(0x40, 50)
+	require.Equal(t, int32(1), dh.dt.WriteRTP(ep, 0), "keyframe should forward")
+
+	// let an already armed interval fire, then the requester must stay asleep
+	time.Sleep(2 * KeyFrameIntervalMaxForTest)
+	before := dh.receiver.SendPLICallCount()
+	wakeups := dh.dt.KeyFrameRequesterWakeupsForTest()
+	time.Sleep(2 * KeyFrameIntervalMaxForTest)
+	require.Equal(t, wakeups, dh.dt.KeyFrameRequesterWakeupsForTest(), "key frame requester must not wake up once the layer is locked")
+	require.Equal(t, before, dh.receiver.SendPLICallCount(), "no PLI expected once the layer is locked")
+
+	// a resync drops the lock and must wake the requester again
+	dh.dt.Resync()
+	require.Eventually(t, func() bool {
+		return dh.receiver.SendPLICallCount() > before
+	}, 5*time.Second, 20*time.Millisecond, "expected PLI after resync")
+}

pkg/sfu/export_test.go

@@ -14,7 +14,11 @@
 
 package sfu
 
-import "github.com/livekit/livekit-server/pkg/sfu/buffer"
+import (
+	"time"
+
+	"github.com/livekit/livekit-server/pkg/sfu/buffer"
+)
 
 // Test-only seams that expose DownTrack internals to the external sfu_test package.
 // This file is only compiled into the test binary (its name ends in _test.go), so it
@@ -71,3 +75,9 @@ func (d *DownTrack) RetransmitForTest(
 	}
 	return d.retransmitPacket(&epm, sourcePkt, false)
 }
+
+const KeyFrameIntervalMaxForTest = keyFrameIntervalMax * time.Millisecond
+
+func (d *DownTrack) KeyFrameRequesterWakeupsForTest() uint32 {
+	return d.keyFrameRequesterWakeups.Load()
+}

pkg/sfu/forwarder.go

@@ -360,6 +360,21 @@ func (f *Forwarder) MaybeExpireAcquireGrace() bool {
 	return !f.vls.GetCurrent().IsValid()
 }
 
+// NeedsPeriodicSyncCheck reports whether the layer lock state can change without an event, so the
+// key frame requester has to keep polling it: an initial-acquisition grace deadline is armed
+// (MaybeExpireAcquireGrace may still fire), or layers are selected with the dependency descriptor,
+// which can lose the lock on the packet path (key frame mismatch, broken frame chain).
+func (f *Forwarder) NeedsPeriodicSyncCheck() bool {
+	f.lock.RLock()
+	defer f.lock.RUnlock()
+
+	if f.acquireDeadline != 0 {
+		return true
+	}
+	_, isDD := f.vls.(*videolayerselector.DependencyDescriptor)
+	return isDD
+}
+
 func (f *Forwarder) SetMaxTemporalLayerSeen(maxTemporalLayerSeen int32) bool {
 	f.lock.Lock()
 	defer f.lock.Unlock()

pkg/sfu/forwarder_test.go

@@ -2502,3 +2502,22 @@ func TestPacerScratchFitsInlineDependencyDescriptor(t *testing.T) {
 	require.Len(t, p.HoldExtension(make([]byte, dd.MaxInlineExtensionSize)), dd.MaxInlineExtensionSize)
 	require.Nil(t, p.HoldExtension(make([]byte, dd.MaxInlineExtensionSize+1)))
 }
+
+func TestForwarderNeedsPeriodicSyncCheck(t *testing.T) {
+	f := newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo)
+	require.False(t, f.NeedsPeriodicSyncCheck())
+
+	// VP8 with the dependency descriptor uses it only for temporal layers
+	f.DetermineCodec(testutils.TestVP8Codec, []webrtc.RTPHeaderExtensionParameter{{URI: dd.ExtensionURI}}, livekit.VideoLayer_MODE_UNUSED)
+	require.False(t, f.NeedsPeriodicSyncCheck())
+
+	// a new max published layer before any layer is forwarded arms the acquisition grace
+	require.True(t, f.SetMaxPublishedLayer(1))
+	require.True(t, f.NeedsPeriodicSyncCheck())
+
+	// spatial layers selected with the dependency descriptor
+	f = newForwarder(testutils.TestVP8Codec, webrtc.RTPCodecTypeVideo)
+	av1 := webrtc.RTPCodecCapability{MimeType: mime.MimeTypeAV1.String(), ClockRate: 90000}
+	f.DetermineCodec(av1, []webrtc.RTPHeaderExtensionParameter{{URI: dd.ExtensionURI}}, livekit.VideoLayer_MODE_UNUSED)
+	require.True(t, f.NeedsPeriodicSyncCheck())
+}