feat: bootstrap new subscribers from a cached video frame (GOP) instead of a PLI

When a down track is added, replay the publisher's most recent group-of-pictures
so the subscriber gets a decodable stream immediately, instead of triggering a
PLI that spikes the publisher uplink and re-sends a key frame to every subscriber.

The cache is opt-in per published track via EnableVideoCaching (with
VideoCachingMaxDuration bounding the cached GOP); when disabled the ingest and
bucket-sizing paths are byte-for-byte the original logic. Adds hit/miss and
forwarded-PLI counters for observability.

Squashed from branch duan/video-frame-cache-relay (video frame caching only;
the start-at-desired-quality work is already in master).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

pkg/rtc/mediatrack.go

@@ -99,6 +99,8 @@ type MediaTrackParams struct {
 	EnableRTPStreamRestartDetection  bool
 	UpdateTrackInfoByVideoSizeChange bool
 	ForceBackupCodecPolicySimulcast  bool
+	EnableVideoCaching               bool
+	VideoCachingMaxDuration          time.Duration
 }
 
 func NewMediaTrack(params MediaTrackParams, ti *livekit.TrackInfo) *MediaTrack {
@@ -407,6 +409,18 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe
 			return newCodec, false
 		}
 
+		receiverOpts := []sfu.ReceiverOpts{
+			sfu.WithPliThrottleConfig(t.params.PLIThrottleConfig),
+			sfu.WithAudioConfig(t.params.AudioConfig),
+			sfu.WithLoadBalanceThreshold(20),
+			sfu.WithForwardStats(t.params.ForwardStats),
+			sfu.WithEnableRTPStreamRestartDetection(t.params.EnableRTPStreamRestartDetection),
+		}
+		if t.params.EnableVideoCaching {
+			// VideoCachingMaxDuration <= 0 falls back to sfu.DefaultVideoFrameCacheMaxDuration in WithVideoFrameCache
+			receiverOpts = append(receiverOpts, sfu.WithVideoFrameCache(t.params.VideoCachingMaxDuration))
+		}
+
 		newWR := sfu.NewWebRTCReceiver(
 			receiver,
 			track,
@@ -414,11 +428,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe
 			LoggerWithCodecMime(t.params.Logger, mimeType),
 			t.params.OnRTCP,
 			t.params.VideoConfig.StreamTrackerManager,
-			sfu.WithPliThrottleConfig(t.params.PLIThrottleConfig),
-			sfu.WithAudioConfig(t.params.AudioConfig),
-			sfu.WithLoadBalanceThreshold(20),
-			sfu.WithForwardStats(t.params.ForwardStats),
-			sfu.WithEnableRTPStreamRestartDetection(t.params.EnableRTPStreamRestartDetection),
+			receiverOpts...,
 		)
 		newWR.OnCloseHandler(func() {
 			t.MediaTrackReceiver.SetClosing(false)

pkg/rtc/participant.go

@@ -226,6 +226,8 @@ type ParticipantParams struct {
 	ForceBackupCodecPolicySimulcast bool
 	DisableTransceiverReuseForE2EE  bool
 	EnableStartAtDesiredQuality     bool
+	EnableVideoCaching              bool
+	VideoCachingMaxDuration         time.Duration
 }
 
 type ParticipantImpl struct {
@@ -3382,6 +3384,8 @@ func (p *ParticipantImpl) addMediaTrack(signalCid string, ti *livekit.TrackInfo)
 		EnableRTPStreamRestartDetection:  p.params.EnableRTPStreamRestartDetection,
 		UpdateTrackInfoByVideoSizeChange: p.params.UseOneShotSignallingMode,
 		ForceBackupCodecPolicySimulcast:  p.params.ForceBackupCodecPolicySimulcast,
+		EnableVideoCaching:               p.params.EnableVideoCaching,
+		VideoCachingMaxDuration:          p.params.VideoCachingMaxDuration,
 	}, ti)
 
 	mt.OnSubscribedMaxQualityChange(p.onSubscribedMaxQualityChange)

pkg/sfu/buffer/buffer.go

@@ -22,6 +22,7 @@ import (
 	"github.com/pion/rtcp"
 	"github.com/pion/rtp"
 	"github.com/pion/webrtc/v4"
+	"go.uber.org/atomic"
 
 	sutils "github.com/livekit/livekit-server/pkg/utils"
 	"github.com/livekit/mediatransportutil/pkg/bucket"
@@ -72,6 +73,9 @@ type Buffer struct {
 
 	primaryBufferForRTX *Buffer
 	rtxPktBuf           []byte
+
+	// number of PLIs actually forwarded to the publisher (past the throttle gate, see sendPLI)
+	pliForwardedCount atomic.Uint32
 }
 
 func NewBuffer(ssrc uint32, maxVideoPkts, maxAudioPkts int) *Buffer {
@@ -335,7 +339,6 @@ func (b *Buffer) sendPLI() {
 		return
 	}
 
-	b.logger.Debugw("send pli", "mediaSSRC", ssrc)
 	pli := []rtcp.Packet{
 		&rtcp.PictureLossIndication{
 			SenderSSRC: ssrc,
@@ -344,10 +347,17 @@ func (b *Buffer) sendPLI() {
 	}
 
 	if cb := b.getOnRtcpFeedback(); cb != nil {
+		pliForwardedCount := b.pliForwardedCount.Inc()
 		cb(pli)
+		b.logger.Debugw("send pli", "mediaSSRC", ssrc, "pliForwardedCount", pliForwardedCount)
 	}
 }
 
+// PLIForwardedCount returns the number of PLIs actually forwarded to the publisher (past the throttle gate).
+func (b *Buffer) PLIForwardedCount() uint32 {
+	return b.pliForwardedCount.Load()
+}
+
 func (b *Buffer) calc(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime int64, isBuffered bool, isRTX bool) []rtcp.Packet {
 	b.BufferBase.HandleIncomingPacketLocked(
 		rawPkt,

pkg/sfu/buffer/buffer_base.go

@@ -89,10 +89,16 @@ type BufferProvider interface {
 	SetPaused(paused bool)
 
 	SendPLI(force bool)
+	// PLIForwardedCount returns the number of PLIs actually forwarded to the publisher (past the throttle gate).
+	PLIForwardedCount() uint32
 
 	ReadExtended(buf []byte) (*ExtPacket, error)
 	GetPacket(buf []byte, esn uint64) (int, error)
 
+	EnableVideoFrameCache(maxDuration time.Duration)
+	GetVideoFrameCache() ([]*ExtPacket, bool)
+	GetPacketsAfter(afterSN uint64) ([]*ExtPacket, bool)
+
 	GetAudioLevel() (float64, bool)
 	GetTemporalLayerFpsForSpatial(layer int32) []float32
 	GetStats() *livekit.RTPStats
@@ -200,6 +206,16 @@ type BufferBase struct {
 	ddExtID  uint8
 	ddParser *DependencyDescriptorParser
 
+	// video frame cache: mark the most recent key frame so the current group-of-pictures can be read
+	// back from the retransmit bucket (see EnableVideoFrameCache / GetVideoFrameCache). No packets are copied here - only
+	// the key-frame boundary is tracked.
+	videoFrameCacheEnabled     bool
+	videoFrameCacheMaxDuration time.Duration // optional bound on key-frame interval (0 = bucket-bounded only)
+	videoFrameCacheHasKeyFrame bool
+	videoFrameCacheKeyFrameSN  uint64 // ext sequence number of the current GOP's first key-frame packet
+	videoFrameCacheKeyFrameTS  uint64 // ext timestamp of the current GOP's key frame
+	videoFrameCacheLatestTS    uint64 // maximum ext timestamp seen in the current GOP (resets on a new key frame)
+
 	isPaused            bool
 	frameRateCalculator [DefaultMaxLayerSpatial + 1]FrameRateCalculator
 	frameRateCalculated bool
@@ -541,6 +557,8 @@ func (b *BufferBase) restartStreamLocked(reason string, isDetected bool) {
 	b.StopKeyFrameSeeder()
 	b.stopRTPStats("stream-restart")
 	b.flushExtPacketsLocked()
+	// the marked GOP references the pre-restart sequence-number base / evicted bucket contents
+	b.videoFrameCacheHasKeyFrame = false
 
 	// restart
 	b.snRangeMap = utils.NewRangeMap[uint64, uint64](100)
@@ -643,6 +661,170 @@ func (b *BufferBase) ReadExtended(buf []byte) (*ExtPacket, error) {
 	}
 }
 
+// EnableVideoFrameCache turns on video frame cache tracking for this (video) buffer: the most recent key frame is marked
+// so the current group-of-pictures can be read back from the retransmit bucket via GetVideoFrameCache. No
+// packets are copied - only the key-frame boundary is tracked. maxDuration bounds the served
+// key-frame interval AND drives the retransmit bucket to retain that much history (see
+// maybeGrowBucket), so the key frame is not evicted before it can be read; <= 0 keeps the default
+// ~1s retransmit window (and bounds the GOP to it). No-op for audio buffers.
+func (b *BufferBase) EnableVideoFrameCache(maxDuration time.Duration) {
+	b.Lock()
+	defer b.Unlock()
+
+	b.videoFrameCacheEnabled = true
+	b.videoFrameCacheMaxDuration = maxDuration
+	b.logger.Debugw("video frame cache enabled on buffer", "maxDuration", maxDuration)
+}
+
+// markVideoFrameCacheLocked records the key-frame boundary of the current GOP and tracks its span. Caller holds
+// the lock.
+func (b *BufferBase) markVideoFrameCacheLocked(ep *ExtPacket) {
+	if ep == nil || ep.Packet == nil || len(ep.Packet.Payload) == 0 {
+		return
+	}
+	if ep.IsKeyFrame && (!b.videoFrameCacheHasKeyFrame || ep.ExtTimestamp != b.videoFrameCacheKeyFrameTS) {
+		// a new key frame starts a new GOP; remember its first packet's sequence number and reset the
+		// span to the key frame so a stale packet from the previous GOP cannot stretch it
+		b.videoFrameCacheKeyFrameSN = ep.ExtSequenceNumber
+		b.videoFrameCacheKeyFrameTS = ep.ExtTimestamp
+		b.videoFrameCacheLatestTS = ep.ExtTimestamp
+		b.videoFrameCacheHasKeyFrame = true
+		b.logger.Debugw("video frame cache: marked key frame", "keyFrameSN", b.videoFrameCacheKeyFrameSN, "keyFrameTS", b.videoFrameCacheKeyFrameTS)
+		return
+	}
+	// track the maximum timestamp seen in the current GOP (not the last-written one) so an
+	// out-of-order, older packet arriving last cannot shrink the measured span and let GetVideoFrameCache serve
+	// more than videoFrameCacheMaxDuration. The head packet's timestamp is always <= videoFrameCacheLatestTS, so the
+	// duration gate in GetVideoFrameCache strictly bounds the served GOP.
+	if ep.ExtTimestamp > b.videoFrameCacheLatestTS {
+		b.videoFrameCacheLatestTS = ep.ExtTimestamp
+	}
+}
+
+// GetVideoFrameCache reads the packets of the current group-of-pictures (from the most recent key frame up to
+// the latest packet) back from the retransmit bucket, so a newly attached relay / down track can be
+// bootstrapped without requesting a fresh key frame (PLI). Returns (nil, false) when the cache is
+// disabled, no key frame has been marked, the key-frame interval exceeds the configured bound, or
+// the key frame is no longer retained in the bucket (the GOP length is ultimately bounded by the
+// bucket capacity). Lost packets within the GOP are skipped.
+//
+// The packets are returned as ExtPackets reconstructed from the bucket bytes so they can be replayed
+// through the normal forward path (WriteRTP): ExtSequenceNumber comes from the bucket key and
+// ExtTimestamp / IsKeyFrame are derived from the marked key frame. The dependency descriptor is not
+// reconstructed (SVC replay is not supported here). The returned packets are self-contained copies.
+func (b *BufferBase) GetVideoFrameCache() ([]*ExtPacket, bool) {
+	b.Lock()
+	defer b.Unlock()
+
+	if !b.videoFrameCacheEnabled || !b.videoFrameCacheHasKeyFrame || b.bucket == nil {
+		b.logger.Debugw(
+			"video frame cache miss: not ready",
+			"videoFrameCacheEnabled", b.videoFrameCacheEnabled,
+			"videoFrameCacheHasKeyFrame", b.videoFrameCacheHasKeyFrame,
+			"hasBucket", b.bucket != nil,
+		)
+		return nil, false
+	}
+
+	if b.videoFrameCacheMaxDuration > 0 && b.clockRate > 0 {
+		maxTicks := uint64(b.videoFrameCacheMaxDuration.Seconds() * float64(b.clockRate))
+		if b.videoFrameCacheLatestTS > b.videoFrameCacheKeyFrameTS+maxTicks {
+			// key-frame interval longer than the bound - too old to serve a complete replay
+			b.logger.Debugw(
+				"video frame cache miss: key-frame interval exceeds bound",
+				"keyFrameTS", b.videoFrameCacheKeyFrameTS,
+				"latestTS", b.videoFrameCacheLatestTS,
+				"spanTicks", b.videoFrameCacheLatestTS-b.videoFrameCacheKeyFrameTS,
+				"maxTicks", maxTicks,
+			)
+			return nil, false
+		}
+	}
+
+	headSN := uint64(b.bucket.HeadSequenceNumber())
+	if headSN < b.videoFrameCacheKeyFrameSN {
+		b.logger.Debugw("video frame cache miss: head behind key frame", "headSN", headSN, "keyFrameSN", b.videoFrameCacheKeyFrameSN)
+		return nil, false
+	}
+
+	pkts := b.reconstructPacketsLocked(b.videoFrameCacheKeyFrameSN, headSN)
+	// the key frame itself must be present (its first packet), otherwise the GOP cannot be served
+	if len(pkts) == 0 || pkts[0].ExtSequenceNumber != b.videoFrameCacheKeyFrameSN {
+		var firstSN uint64
+		if len(pkts) > 0 {
+			firstSN = pkts[0].ExtSequenceNumber
+		}
+		b.logger.Debugw(
+			"video frame cache miss: key frame evicted from bucket",
+			"keyFrameSN", b.videoFrameCacheKeyFrameSN,
+			"headSN", headSN,
+			"bucketCapacity", b.bucket.Capacity(),
+			"reconstructed", len(pkts),
+			"firstSN", firstSN,
+		)
+		return nil, false
+	}
+	return pkts, true
+}
+
+// GetPacketsAfter reads the packets newer than afterSN (exclusive) up to the current head from the
+// retransmit bucket, reconstructed as ExtPackets like GetVideoFrameCache. It is used to catch a video frame cache replay up to
+// the live forwarding point. Returns (nil, false) when the cache is disabled or nothing newer is
+// retained.
+func (b *BufferBase) GetPacketsAfter(afterSN uint64) ([]*ExtPacket, bool) {
+	b.Lock()
+	defer b.Unlock()
+
+	if !b.videoFrameCacheEnabled || !b.videoFrameCacheHasKeyFrame || b.bucket == nil {
+		return nil, false
+	}
+	headSN := uint64(b.bucket.HeadSequenceNumber())
+	if headSN <= afterSN {
+		return nil, false
+	}
+	pkts := b.reconstructPacketsLocked(afterSN+1, headSN)
+	if len(pkts) == 0 {
+		return nil, false
+	}
+	return pkts, true
+}
+
+// reconstructPacketsLocked builds self-contained ExtPackets for the sequence-number range
+// [fromSN, headSN] from the retransmit bucket. Lost packets are skipped. ExtTimestamp is
+// reconstructed relative to the marked key frame (a GOP spans well under one 32-bit timestamp wrap)
+// and IsKeyFrame flags packets at the key-frame timestamp. The dependency descriptor is not
+// reconstructed. Caller holds the lock.
+func (b *BufferBase) reconstructPacketsLocked(fromSN, headSN uint64) []*ExtPacket {
+	keyFrameRTPTS := uint32(b.videoFrameCacheKeyFrameTS)
+	var pkts []*ExtPacket
+	buf := make([]byte, bucket.RTPMaxPktSize)
+	for sn := fromSN; sn <= headSN; sn++ {
+		n, err := b.bucket.GetPacket(buf, sn)
+		if err != nil {
+			continue // lost packet, skip
+		}
+		// copy out of the reused read buffer so the parsed packet is self-contained
+		raw := make([]byte, n)
+		copy(raw, buf[:n])
+		p := &rtp.Packet{}
+		if err := p.Unmarshal(raw); err != nil {
+			continue
+		}
+
+		extTS := b.videoFrameCacheKeyFrameTS + uint64(p.Timestamp-keyFrameRTPTS)
+		pkts = append(pkts, &ExtPacket{
+			VideoLayer:        VideoLayer{Spatial: InvalidLayerSpatial, Temporal: InvalidLayerTemporal},
+			Arrival:           mono.UnixNano(),
+			ExtSequenceNumber: sn,
+			ExtTimestamp:      extTS,
+			Packet:            p,
+			IsKeyFrame:        extTS == b.videoFrameCacheKeyFrameTS,
+			RawPacket:         raw,
+		})
+	}
+	return pkts
+}
+
 func (b *BufferBase) SetPLIThrottle(duration int64) {
 	b.Lock()
 	defer b.Unlock()
@@ -909,6 +1091,9 @@ func (b *BufferBase) HandleIncomingPacketLocked(
 		return 0, errors.New("could not get ext packet")
 	}
 	b.extPackets.PushBack(ep)
+	if b.videoFrameCacheEnabled && b.codecType == webrtc.RTPCodecTypeVideo {
+		b.markVideoFrameCacheLocked(ep)
+	}
 	b.readCond.Broadcast()
 
 	if b.extPackets.Len() > b.bucket.Capacity() {
@@ -1224,6 +1409,25 @@ func (b *BufferBase) flushExtPacketsLocked() {
 	b.extPackets.Clear()
 }
 
+// bucketGrowTarget computes how many packets the retransmit bucket should retain (targetPkts) and
+// the cap that allows growing to it (effectiveMaxPkts), given the measured packets-per-second.
+//
+// Normally the target is ~1s of packets (the NACK / retransmit window), bounded by maxPkts. When
+// videoFrameCacheSizing is set (the video frame cache is enabled with a positive duration), the bucket must retain
+// the whole GOP duration plus ~0.5s margin so the key frame is not evicted before it is at most
+// videoFrameCacheMaxDuration old; the cap is raised to fit since the default maxPkts is only ~1s worth.
+func bucketGrowTarget(pps, maxPkts int, videoFrameCacheSizing bool, videoFrameCacheMaxDuration time.Duration) (targetPkts, effectiveMaxPkts int) {
+	targetPkts = pps
+	effectiveMaxPkts = maxPkts
+	if videoFrameCacheSizing && videoFrameCacheMaxDuration > 0 {
+		targetPkts = int(float64(pps)*videoFrameCacheMaxDuration.Seconds()) + pps/2
+		if targetPkts > effectiveMaxPkts {
+			effectiveMaxPkts = targetPkts
+		}
+	}
+	return
+}
+
 func (b *BufferBase) maybeGrowBucket(now int64) {
 	if now-b.lastBucketCapCheckAt < bucketCapCheckInterval {
 		return
@@ -1241,31 +1445,45 @@ func (b *BufferBase) maybeGrowBucket(now int64) {
 		if b.codecType == webrtc.RTPCodecTypeAudio {
 			maxPkts = b.params.MaxAudioPkts
 		}
+
+		// when the video frame cache is enabled the bucket must retain the whole configured GOP
+		// duration (the key frame must survive until GetVideoFrameCache reads it), which is more than the normal
+		// ~1s retransmit window. In that case the target / cap are computed below from pps; otherwise
+		// keep the original fast path.
+		videoFrameCacheSizing := b.codecType == webrtc.RTPCodecTypeVideo && b.videoFrameCacheEnabled && b.videoFrameCacheMaxDuration > 0
+		if cap >= maxPkts && !videoFrameCacheSizing {
+			return
+		}
+
+		deltaInfo := b.rtpStats.DeltaInfo(b.ppsSnapshotId)
+		if deltaInfo == nil {
+			return
+		}
+		duration := deltaInfo.EndTime.Sub(deltaInfo.StartTime)
+		if duration < 500*time.Millisecond {
+			return
+		}
+		pps := int(time.Duration(deltaInfo.Packets) * time.Second / duration)
+
+		targetPkts, maxPkts := bucketGrowTarget(pps, maxPkts, videoFrameCacheSizing, b.videoFrameCacheMaxDuration)
 		if cap >= maxPkts {
 			return
 		}
 
 		oldCap := cap
-		if deltaInfo := b.rtpStats.DeltaInfo(b.ppsSnapshotId); deltaInfo != nil {
-			duration := deltaInfo.EndTime.Sub(deltaInfo.StartTime)
-			if duration < 500*time.Millisecond {
-				return
-			}
-
-			pps := int(time.Duration(deltaInfo.Packets) * time.Second / duration)
-			for pps > cap && cap < maxPkts {
-				cap = b.bucket.Grow()
-			}
-			if cap > oldCap {
-				b.logger.Infow(
-					"grow bucket",
-					"from", oldCap,
-					"to", cap,
-					"pps", pps,
-					"deltaInfo", deltaInfo,
-					"rtpStats", b.rtpStats,
-				)
-			}
+		for targetPkts > cap && cap < maxPkts {
+			cap = b.bucket.Grow()
+		}
+		if cap > oldCap {
+			b.logger.Infow(
+				"grow bucket",
+				"from", oldCap,
+				"to", cap,
+				"pps", pps,
+				"targetPkts", targetPkts,
+				"deltaInfo", deltaInfo,
+				"rtpStats", b.rtpStats,
+			)
 		}
 	}()
 }

pkg/sfu/buffer/videoframecache_test.go

@@ -0,0 +1,233 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package buffer
+
+import (
+	"testing"
+	"time"
+
+	"github.com/pion/rtp"
+	"github.com/pion/webrtc/v4"
+	"github.com/stretchr/testify/require"
+
+	"github.com/livekit/mediatransportutil/pkg/bucket"
+	"github.com/livekit/protocol/logger"
+)
+
+func newVideoFrameCacheTestBuffer(maxDuration time.Duration) *BufferBase {
+	b := &BufferBase{
+		codecType: webrtc.RTPCodecTypeVideo,
+		clockRate: 90000,
+		logger:    logger.GetLogger(),
+	}
+	b.bucket = bucket.NewBucket[uint64, uint16](256, bucket.RTPMaxPktSize, bucket.RTPSeqNumOffset)
+	b.EnableVideoFrameCache(maxDuration)
+	return b
+}
+
+// addBucketPacket marshals an RTP packet and stores it in the bucket keyed by ext sequence number.
+func addBucketPacket(t *testing.T, b *BufferBase, extSN uint64, ts uint32) {
+	p := rtp.Packet{
+		Header:  rtp.Header{Version: 2, PayloadType: 96, SequenceNumber: uint16(extSN), Timestamp: ts, SSRC: 123},
+		Payload: []byte{1, 2, 3, 4},
+	}
+	raw, err := p.Marshal()
+	require.NoError(t, err)
+	_, err = b.bucket.AddPacketWithSequenceNumber(raw, extSN)
+	require.NoError(t, err)
+}
+
+func videoFrameCacheMarkPkt(extSN, extTS uint64, keyFrame bool) *ExtPacket {
+	return &ExtPacket{
+		ExtSequenceNumber: extSN,
+		ExtTimestamp:      extTS,
+		IsKeyFrame:        keyFrame,
+		Packet:            &rtp.Packet{Payload: []byte{1, 2, 3, 4}},
+	}
+}
+
+func TestVideoFrameCacheReadsFromBucket(t *testing.T) {
+	b := newVideoFrameCacheTestBuffer(0)
+
+	// store SN 100..104 in the bucket
+	for i := uint64(0); i < 5; i++ {
+		addBucketPacket(t, b, 100+i, uint32(1000*(i+1)))
+	}
+
+	// no key frame marked yet -> not available
+	_, ok := b.GetVideoFrameCache()
+	require.False(t, ok)
+
+	// mark: delta at 100, key frame at 101, deltas after
+	b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(100, 1000, false))
+	b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(101, 2000, true))
+	b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(102, 3000, false))
+	b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(103, 4000, false))
+	b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(104, 5000, false))
+
+	// GOP is [key frame .. head] = 101..104
+	pkts, ok := b.GetVideoFrameCache()
+	require.True(t, ok)
+	require.Len(t, pkts, 4)
+	require.Equal(t, uint16(101), pkts[0].Packet.SequenceNumber)
+	require.Equal(t, uint64(101), pkts[0].ExtSequenceNumber)
+	require.True(t, pkts[0].IsKeyFrame) // first packet is at the key-frame timestamp
+	require.False(t, pkts[1].IsKeyFrame)
+	require.Equal(t, uint16(104), pkts[3].Packet.SequenceNumber)
+
+	// a new key frame moves the boundary
+	addBucketPacket(t, b, 105, 6000)
+	b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(105, 6000, true))
+	pkts, ok = b.GetVideoFrameCache()
+	require.True(t, ok)
+	require.Len(t, pkts, 1)
+	require.Equal(t, uint16(105), pkts[0].Packet.SequenceNumber)
+	require.True(t, pkts[0].IsKeyFrame)
+}
+
+func TestVideoFrameCacheGetPacketsAfter(t *testing.T) {
+	b := newVideoFrameCacheTestBuffer(0)
+	for i := uint64(0); i < 5; i++ {
+		addBucketPacket(t, b, 100+i, uint32(1000*(i+1)))
+	}
+	b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(100, 1000, true))
+	for i := uint64(1); i < 5; i++ {
+		b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(100+i, 1000*(i+1), false))
+	}
+
+	// everything after 102 is 103, 104
+	more, ok := b.GetPacketsAfter(102)
+	require.True(t, ok)
+	require.Len(t, more, 2)
+	require.Equal(t, uint64(103), more[0].ExtSequenceNumber)
+	require.Equal(t, uint64(104), more[1].ExtSequenceNumber)
+
+	// caught up: nothing after the head
+	_, ok = b.GetPacketsAfter(104)
+	require.False(t, ok)
+}
+
+func TestVideoFrameCacheSkipsLostPackets(t *testing.T) {
+	b := newVideoFrameCacheTestBuffer(0)
+
+	// store the key frame and a later packet, leaving a gap at 101
+	addBucketPacket(t, b, 100, 1000)
+	addBucketPacket(t, b, 102, 3000)
+
+	b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(100, 1000, true))
+	b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(102, 3000, false))
+
+	pkts, ok := b.GetVideoFrameCache()
+	require.True(t, ok)
+	require.Len(t, pkts, 2) // the missing 101 is skipped
+	require.Equal(t, uint16(100), pkts[0].Packet.SequenceNumber)
+	require.Equal(t, uint16(102), pkts[1].Packet.SequenceNumber)
+}
+
+func TestVideoFrameCacheKeyFrameEvicted(t *testing.T) {
+	b := newVideoFrameCacheTestBuffer(0)
+
+	// mark a key frame at a sequence number that is not in the bucket
+	addBucketPacket(t, b, 200, 1000)
+	b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(100, 500, true)) // 100 was never stored / evicted
+	b.videoFrameCacheLatestTS = 1000
+
+	_, ok := b.GetVideoFrameCache()
+	require.False(t, ok)
+}
+
+func TestVideoFrameCacheDurationBound(t *testing.T) {
+	// 2s cap at 90kHz -> 180000 ticks
+	b := newVideoFrameCacheTestBuffer(2 * time.Second)
+	addBucketPacket(t, b, 100, 1000)
+
+	const kfTS = uint64(1_000_000)
+	b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(100, kfTS, true))
+
+	// within the bound
+	b.videoFrameCacheLatestTS = kfTS + 90000 // +1s
+	_, ok := b.GetVideoFrameCache()
+	require.True(t, ok)
+
+	// beyond the bound -> not served
+	b.videoFrameCacheLatestTS = kfTS + 180001 // > 2s
+	_, ok = b.GetVideoFrameCache()
+	require.False(t, ok)
+}
+
+func TestVideoFrameCacheSpanUsesMaxTimestamp(t *testing.T) {
+	b := newVideoFrameCacheTestBuffer(0)
+
+	// key frame at 1000, then a later packet at 1200 advances the span
+	b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(100, 1000, true))
+	b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(101, 1200, false))
+	require.Equal(t, uint64(1200), b.videoFrameCacheLatestTS)
+
+	// an out-of-order, older packet arriving last must not shrink the measured span
+	b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(102, 1100, false))
+	require.Equal(t, uint64(1200), b.videoFrameCacheLatestTS)
+
+	// a new key frame resets the span to itself (a stale packet cannot stretch the new GOP)
+	b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(103, 5000, true))
+	require.Equal(t, uint64(5000), b.videoFrameCacheLatestTS)
+	b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(104, 4000, false)) // stale, older than the new key frame
+	require.Equal(t, uint64(5000), b.videoFrameCacheLatestTS)
+}
+
+func TestBucketGrowTarget(t *testing.T) {
+	const pps = 600
+	const maxPkts = 500 // default ~1s cap
+
+	// video frame cache sizing off -> unchanged original behavior: target is ~1s (pps), cap stays maxPkts
+	target, cap := bucketGrowTarget(pps, maxPkts, false, 0)
+	require.Equal(t, pps, target)
+	require.Equal(t, maxPkts, cap)
+
+	// videoFrameCacheMaxDuration <= 0 is treated as not sizing even if the flag is set
+	target, cap = bucketGrowTarget(pps, maxPkts, true, 0)
+	require.Equal(t, pps, target)
+	require.Equal(t, maxPkts, cap)
+
+	// video frame cache sizing on, 2s: target is pps*2 + 0.5s margin, and the cap is raised to fit
+	target, cap = bucketGrowTarget(pps, maxPkts, true, 2*time.Second)
+	require.Equal(t, pps*2+pps/2, target) // 1500
+	require.Equal(t, target, cap)         // raised above the 500 default
+	require.Greater(t, cap, maxPkts)
+
+	// video frame cache sizing on but the target fits within maxPkts -> cap is not lowered
+	target, cap = bucketGrowTarget(100, 1000, true, 2*time.Second)
+	require.Equal(t, 100*2+100/2, target) // 250
+	require.Equal(t, 1000, cap)           // unchanged, target < maxPkts
+
+	// scales with duration
+	target1s, _ := bucketGrowTarget(pps, maxPkts, true, time.Second)
+	target3s, _ := bucketGrowTarget(pps, maxPkts, true, 3*time.Second)
+	require.Greater(t, target3s, target1s)
+}
+
+func TestVideoFrameCacheDisabled(t *testing.T) {
+	b := &BufferBase{codecType: webrtc.RTPCodecTypeVideo, clockRate: 90000, logger: logger.GetLogger()}
+	b.bucket = bucket.NewBucket[uint64, uint16](256, bucket.RTPMaxPktSize, bucket.RTPSeqNumOffset)
+	addBucketPacket(t, b, 100, 1000)
+	b.markVideoFrameCacheLocked(videoFrameCacheMarkPkt(100, 1000, true)) // marking is a no-op while disabled, but set fields anyway
+
+	_, ok := b.GetVideoFrameCache()
+	require.False(t, ok)
+
+	// audio buffers never enable the cache
+	a := &BufferBase{codecType: webrtc.RTPCodecTypeAudio, clockRate: 48000, logger: logger.GetLogger()}
+	a.EnableVideoFrameCache(0)
+	require.False(t, a.videoFrameCacheEnabled)
+}

pkg/sfu/receiver.go

@@ -93,6 +93,23 @@ func WithForwardStats(forwardStats *ForwardStats) ReceiverOpts {
 	}
 }
 
+// DefaultVideoFrameCacheMaxDuration bounds the cached GOP (and sizes the retransmit bucket) when the video
+// frame cache is enabled via WithVideoFrameCache.
+const DefaultVideoFrameCacheMaxDuration = 1 * time.Second
+
+// WithVideoFrameCache enables the video frame cache on the receiver so a newly added down track is
+// bootstrapped by replaying the cached GOP instead of triggering a PLI. maxDuration bounds the
+// cached video frame (<= 0 uses DefaultVideoFrameCacheMaxDuration). No-op for audio receivers.
+func WithVideoFrameCache(maxDuration time.Duration) ReceiverOpts {
+	return func(w *WebRTCReceiver) *WebRTCReceiver {
+		if maxDuration <= 0 {
+			maxDuration = DefaultVideoFrameCacheMaxDuration
+		}
+		w.ReceiverBase.EnableVideoFrameCache(maxDuration)
+		return w
+	}
+}
+
 // NewWebRTCReceiver creates a new webrtc track receiver
 func NewWebRTCReceiver(
 	receiver *webrtc.RTPReceiver,

pkg/sfu/receiver_base.go

@@ -215,6 +215,16 @@ type ReceiverBase struct {
 	videoSizes         [buffer.DefaultMaxLayerSpatial + 1]codec.VideoSize
 	onVideoSizeChanged func()
 
+	// video frame cache: when enabled, buffers retain the current frame cache and a newly added down track is
+	// bootstrapped by replaying it (see EnableVideoFrameCache / replayVideoFrameCache)
+	videoFrameCacheEnabled     bool
+	videoFrameCacheMaxDuration time.Duration
+
+	// counts of subscribers bootstrapped from the video frame cache (hit) vs. those that fell back to a PLI
+	// because no usable video frame was cached (miss); see replayVideoFrameCache
+	videoFrameCacheHitCount  atomic.Uint32
+	videoFrameCacheMissCount atomic.Uint32
+
 	rtt uint32
 
 	streamTrackerManager *StreamTrackerManager
@@ -560,11 +570,166 @@ func (r *ReceiverBase) AddDownTrack(track TrackSender) error {
 	track.UpTrackMaxPublishedLayerChange(r.streamTrackerManager.GetMaxPublishedLayer())
 	track.UpTrackMaxTemporalLayerSeenChange(r.streamTrackerManager.GetMaxTemporalLayerSeen())
 
+	if r.videoFrameCacheEnabled && r.Kind() == webrtc.RTPCodecTypeVideo {
+		// Bootstrap the new track from the cached video frame before joining the live broadcast: replay the
+		// cached video frame (paced) and catch up to the live forwarding point, then Store so the live feed takes
+		// over seamlessly. Held out of the broadcast during replay to avoid interleaving replayed and
+		// live packets.
+		go func() {
+			r.replayVideoFrameCache(track)
+			if r.IsClosed() || track.IsClosed() {
+				return
+			}
+			r.downTrackSpreader.Store(track)
+
+			r.params.Logger.Debugw(
+				"downtrack added after cached video frame replay",
+				"subscriberID", track.SubscriberID(),
+				"videoFrameCacheHitCount", r.videoFrameCacheHitCount.Load(),
+				"videoFrameCacheMissCount", r.videoFrameCacheMissCount.Load(),
+				"pliForwardedCount", r.pliForwardedCount(),
+			)
+		}()
+		return nil
+	}
+
 	r.downTrackSpreader.Store(track)
 	r.params.Logger.Debugw("downtrack added", "subscriberID", track.SubscriberID())
 	return nil
 }
 
+// EnableVideoFrameCache turns on the video frame cache for this receiver: each (current and future) buffer
+// retains the current cached frame, and a newly added down track is bootstrapped from it (see replayVideoFrameCache).
+// maxDuration bounds the cached video frame and sizes the retransmit bucket accordingly. No-op for audio.
+func (r *ReceiverBase) EnableVideoFrameCache(maxDuration time.Duration) {
+	if r.Kind() != webrtc.RTPCodecTypeVideo {
+		return
+	}
+
+	r.bufferMu.Lock()
+	r.videoFrameCacheEnabled = true
+	r.videoFrameCacheMaxDuration = maxDuration
+	buffers := r.buffers
+	r.bufferMu.Unlock()
+
+	for _, buff := range buffers {
+		if buff != nil {
+			buff.EnableVideoFrameCache(maxDuration)
+		}
+	}
+}
+
+// videoFrameCacheReplayMaxCatchupRounds bounds how many catch-up passes a video frame cache replay makes before giving up and
+// joining the live feed (replaying at ~2x real time converges quickly; this guards against a
+// pathological case where live keeps outrunning the replay).
+const videoFrameCacheReplayMaxCatchupRounds = 8
+
+// replayVideoFrameCache bootstraps a freshly added down track by replaying the publisher's cached GOP, then the
+// packets accumulated since, until it catches up to the live forwarding point. Packets are paced at
+// half the (estimated) video frame interval so the replay runs at ~2x real time and converges on
+// live. It writes directly to the track (which is not yet in the live broadcast), so there is no
+// interleaving with live packets.
+//
+// It replays the highest spatial layer that currently has a cached GOP: a subscriber requesting full
+// quality has its forwarder targeting the top layer, and lower layers may not even be flowing (e.g.
+// paused by dynacast), so layer 0 often has no GOP.
+func (r *ReceiverBase) replayVideoFrameCache(track TrackSender) {
+	var (
+		buff  buffer.BufferProvider
+		pkts  []*buffer.ExtPacket
+		layer = buffer.InvalidLayerSpatial
+	)
+	for l := int32(buffer.DefaultMaxLayerSpatial); l >= 0; l-- {
+		b, _ := r.getBuffer(l)
+		if b == nil {
+			continue
+		}
+		if got, ok := b.GetVideoFrameCache(); ok && len(got) > 0 {
+			buff, pkts, layer = b, got, l
+			break
+		}
+	}
+
+	if layer == buffer.InvalidLayerSpatial {
+		// no usable GOP cached on any layer - the down track falls back to requesting a key frame (PLI)
+		missCount := r.videoFrameCacheMissCount.Inc()
+		r.params.Logger.Debugw(
+			"subscriber bootstrap: video frame cache miss, falling back to PLI",
+			"subscriberID", track.SubscriberID(),
+			"videoFrameCacheEnabled", r.videoFrameCacheEnabled,
+			"videoFrameCacheHitCount", r.videoFrameCacheHitCount.Load(),
+			"videoFrameCacheMissCount", missCount,
+		)
+		return
+	}
+
+	hitCount := r.videoFrameCacheHitCount.Inc()
+	half := r.videoFrameCacheReplayFrameInterval(layer) / 2
+	r.params.Logger.Debugw(
+		"subscriber bootstrap: video frame cache hit, replaying",
+		"subscriberID", track.SubscriberID(),
+		"layer", layer,
+		"packets", len(pkts),
+		"frameIntervalHalf", half,
+		"videoFrameCacheHitCount", hitCount,
+		"videoFrameCacheMissCount", r.videoFrameCacheMissCount.Load(),
+	)
+
+	var lastSN uint64
+	write := func(eps []*buffer.ExtPacket) bool {
+		var lastTS uint64
+		for i, ep := range eps {
+			if r.IsClosed() || track.IsClosed() {
+				return false
+			}
+			// pace at frame boundaries (a new timestamp starts a new frame)
+			if i > 0 && ep.ExtTimestamp != lastTS && half > 0 {
+				time.Sleep(half)
+			}
+			lastTS = ep.ExtTimestamp
+			rp := *ep
+			rp.Arrival = mono.UnixNano()
+			track.WriteRTP(&rp, layer)
+			lastSN = ep.ExtSequenceNumber
+		}
+		return true
+	}
+
+	if !write(pkts) {
+		return
+	}
+
+	// catch up to live: replay the packets that arrived during the replay, repeating (each round is
+	// shorter since the replay outruns live) until nothing newer remains.
+	for round := 0; round < videoFrameCacheReplayMaxCatchupRounds; round++ {
+		if r.IsClosed() || track.IsClosed() {
+			return
+		}
+		more, ok := buff.GetPacketsAfter(lastSN)
+		if !ok || len(more) == 0 {
+			break
+		}
+		if !write(more) {
+			return
+		}
+	}
+}
+
+// videoFrameCacheReplayFrameInterval estimates the video frame interval for the spatial layer, used to pace a
+// video frame cache replay. Falls back to 30fps when the frame rate is not yet known.
+func (r *ReceiverBase) videoFrameCacheReplayFrameInterval(layer int32) time.Duration {
+	var fps float32
+	for _, f := range r.GetTemporalLayerFpsForSpatial(layer) {
+		if f > fps {
+			fps = f
+		}
+	}
+	if fps <= 0 {
+		fps = 30
+	}
+	return time.Duration(float64(time.Second) / float64(fps))
+}
+
 func (r *ReceiverBase) DeleteDownTrack(subscriberID livekit.ParticipantID) {
 	r.downTrackSpreader.Free(subscriberID)
 	r.params.Logger.Debugw("downtrack deleted", "subscriberID", subscriberID)
@@ -679,6 +844,21 @@ func (r *ReceiverBase) SendPLI(layer int32, force bool) {
 	buff.SendPLI(force)
 }
 
+// pliForwardedCount returns the total number of PLIs actually forwarded to the publisher across all
+// spatial-layer buffers (past each buffer's throttle gate).
+func (r *ReceiverBase) pliForwardedCount() uint32 {
+	r.bufferMu.RLock()
+	defer r.bufferMu.RUnlock()
+
+	var total uint32
+	for _, buff := range r.buffers {
+		if buff != nil {
+			total += buff.PLIForwardedCount()
+		}
+	}
+	return total
+}
+
 func (r *ReceiverBase) getBuffer(layer int32) (buffer.BufferProvider, int32) {
 	r.bufferMu.RLock()
 	defer r.bufferMu.RUnlock()
@@ -790,6 +970,9 @@ func (r *ReceiverBase) setupBuffer(buff buffer.BufferProvider, layer int32, rtt
 	if r.Kind() == webrtc.RTPCodecTypeVideo && layer == 0 {
 		buff.OnCodecChange(r.handleCodecChange)
 	}
+	if r.videoFrameCacheEnabled && r.Kind() == webrtc.RTPCodecTypeVideo {
+		buff.EnableVideoFrameCache(r.videoFrameCacheMaxDuration)
+	}
 	buff.OnStreamRestart(func(reason string) {
 		r.restartInternal(reason, true)
 	})