fix(capture): honor permitted H264 frame gaps

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

machinery/src/capture/gortsplib.go

@@ -32,6 +32,7 @@ import (
 	"github.com/bluenviron/gortsplib/v5/pkg/format/rtplpcm"
 	"github.com/bluenviron/gortsplib/v5/pkg/format/rtpmpeg4audio"
 	"github.com/bluenviron/gortsplib/v5/pkg/format/rtpsimpleaudio"
+	"github.com/bluenviron/mediacommon/pkg/bits"
 	"github.com/bluenviron/mediacommon/pkg/codecs/h264"
 	"github.com/bluenviron/mediacommon/pkg/codecs/h265"
 	"github.com/bluenviron/mediacommon/pkg/codecs/mpeg4audio"
@@ -159,13 +160,10 @@ type Golibrtsp struct {
 }
 
 // streamHealth instruments the RTSP read path. gortsplib delivers every RTP
-// packet on a single read goroutine; queue.WritePacket() is synchronous, so if
-// a downstream consumer (recording, muxing, WebRTC) is slow or the process is
-// CPU-starved, WritePacket() blocks, the TCP socket is not drained, and the
-// camera advances RTP sequence numbers -> "RTP packets lost". This type makes
-// the two failure modes distinguishable:
-//   - large writeMax / writeAvg  => downstream back-pressure (our side).
-//   - large gapMax with fast writes => upstream network / camera stall.
+// packet on a single read goroutine, so prolonged queue lock contention or
+// process starvation can delay socket reads. Queue consumers process packets
+// after releasing the lock; a slow recorder therefore falls behind its cursor
+// instead of directly blocking WritePacket.
 type streamHealth struct {
 	mu          sync.Mutex
 	windowStart time.Time
@@ -178,6 +176,152 @@ type streamHealth struct {
 	decodeErrs  int64
 }
 
+type videoReferenceState struct {
+	haveSequence       bool
+	lastSequence       uint16
+	valid              bool
+	droppedAccessUnits uint64
+}
+
+type h264FrameNumTracker struct {
+	sps          *h264.SPS
+	haveFrameNum bool
+	lastFrameNum uint32
+}
+
+const maxH264SliceHeaderBytes = 64
+
+func (t *h264FrameNumTracker) setSPS(sps *h264.SPS) {
+	if sps == nil {
+		return
+	}
+	if t.sps == nil ||
+		t.sps.Log2MaxFrameNumMinus4 != sps.Log2MaxFrameNumMinus4 ||
+		t.sps.SeparateColourPlaneFlag != sps.SeparateColourPlaneFlag ||
+		t.sps.FrameMbsOnlyFlag != sps.FrameMbsOnlyFlag {
+		t.haveFrameNum = false
+	}
+	copy := *sps
+	t.sps = &copy
+}
+
+func (t *h264FrameNumTracker) observe(au [][]byte) (uint32, uint32, bool, bool) {
+	if t.sps == nil || !t.sps.FrameMbsOnlyFlag {
+		return 0, 0, false, false
+	}
+
+	for _, nalu := range au {
+		if len(nalu) == 0 {
+			continue
+		}
+		naluType := h264.NALUType(nalu[0] & 0x1F)
+		if naluType != h264.NALUTypeIDR && naluType != h264.NALUTypeNonIDR {
+			continue
+		}
+
+		frameNum, err := h264SliceFrameNum(nalu, t.sps)
+		if err != nil {
+			return 0, 0, false, false
+		}
+		if naluType == h264.NALUTypeIDR {
+			t.haveFrameNum = true
+			t.lastFrameNum = frameNum
+			return frameNum, frameNum, false, true
+		}
+
+		// Non-reference pictures do not advance the reference-picture frame_num
+		// sequence and therefore cannot prove that a reference picture is missing.
+		if (nalu[0]>>5)&0x03 == 0 {
+			return frameNum, frameNum, false, true
+		}
+		if !t.haveFrameNum {
+			t.haveFrameNum = true
+			t.lastFrameNum = frameNum
+			return frameNum, frameNum, false, true
+		}
+
+		frameNumBits := t.sps.Log2MaxFrameNumMinus4 + 4
+		if frameNumBits > 16 {
+			return 0, 0, false, false
+		}
+		maxFrameNum := uint32(1) << frameNumBits
+		expected := (t.lastFrameNum + 1) & (maxFrameNum - 1)
+		// When the SPS permits frame_num gaps, the decoder is required to
+		// synthesize non-existing pictures and the discontinuity is not evidence
+		// of transport loss. RTP/depacketizer errors still invalidate references.
+		gap := frameNum != expected && !t.sps.GapsInFrameNumValueAllowedFlag
+		t.lastFrameNum = frameNum
+		return expected, frameNum, gap, true
+	}
+
+	return 0, 0, false, false
+}
+
+func h264SliceFrameNum(nalu []byte, sps *h264.SPS) (uint32, error) {
+	if len(nalu) < 2 || sps == nil {
+		return 0, errors.New("H264 slice or SPS is missing")
+	}
+	frameNumBits := sps.Log2MaxFrameNumMinus4 + 4
+	if frameNumBits > 16 {
+		return 0, fmt.Errorf("invalid H264 frame_num width: %d", frameNumBits)
+	}
+	header := nalu[1:]
+	if len(header) > maxH264SliceHeaderBytes {
+		header = header[:maxH264SliceHeaderBytes]
+	}
+	header = h264.EmulationPreventionRemove(header)
+	pos := 0
+
+	for range 3 {
+		if _, err := bits.ReadGolombUnsigned(header, &pos); err != nil {
+			return 0, fmt.Errorf("read H264 slice header: %w", err)
+		}
+	}
+	if sps.SeparateColourPlaneFlag {
+		if _, err := bits.ReadBits(header, &pos, 2); err != nil {
+			return 0, fmt.Errorf("read H264 colour plane: %w", err)
+		}
+	}
+	frameNum, err := bits.ReadBits(header, &pos, int(frameNumBits))
+	if err != nil {
+		return 0, fmt.Errorf("read H264 frame_num: %w", err)
+	}
+	return uint32(frameNum), nil
+}
+
+func (s *videoReferenceState) observeSequence(sequence uint16) (uint16, bool) {
+	expected := sequence
+	gap := false
+	if s.haveSequence {
+		expected = s.lastSequence + 1
+		gap = sequence != expected
+		if gap {
+			s.valid = false
+		}
+	}
+	s.haveSequence = true
+	s.lastSequence = sequence
+	return expected, gap
+}
+
+func (s *videoReferenceState) invalidate() {
+	s.valid = false
+}
+
+func (s *videoReferenceState) accept(randomAccess bool) (bool, uint64) {
+	if randomAccess {
+		dropped := s.droppedAccessUnits
+		s.droppedAccessUnits = 0
+		s.valid = true
+		return true, dropped
+	}
+	if !s.valid {
+		s.droppedAccessUnits++
+		return false, 0
+	}
+	return true, 0
+}
+
 const (
 	streamHealthWindow    = 10 * time.Second
 	streamHealthWriteWarn = 150 * time.Millisecond
@@ -190,9 +334,9 @@ func newStreamHealth() *streamHealth {
 }
 
 // observePacket records one processed video frame: the wall-clock gap since the
-// previous frame (arrival cadence) and how long WritePacket() blocked
-// (back-pressure). It emits an immediate warning when either side stalls and a
-// periodic summary every streamHealthWindow.
+// previous frame and how long WritePacket spent acquiring and updating the
+// queue. It emits an immediate warning when either side stalls and a periodic
+// summary every streamHealthWindow.
 func (h *streamHealth) observePacket(streamType string, writeDur time.Duration) {
 	if h == nil {
 		return
@@ -264,9 +408,9 @@ func (h *streamHealth) observePacket(streamType string, writeDur time.Duration)
 	}
 }
 
-// observeLost is invoked by gortsplib when RTP sequence numbers skip. On a TCP
-// transport this means the sender (camera) dropped packets because we were not
-// reading fast enough, not loss on the wire.
+// observeLost is invoked by gortsplib when RTP sequence numbers skip. With the
+// configured TCP transport, the sender or an upstream RTSP component omitted
+// those packets; sender-side pressure is one possible cause.
 func (h *streamHealth) observeLost(streamType string, lost uint64) {
 	if h == nil {
 		return
@@ -887,6 +1031,14 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 	// called when a video RTP packet arrives for H264
 	var filteredAU [][]byte
 	if g.VideoH264Media != nil && g.VideoH264Forma != nil {
+		referenceState := &videoReferenceState{}
+		frameNumTracker := &h264FrameNumTracker{}
+		if len(g.VideoH264Forma.SPS) > 0 {
+			var sps h264.SPS
+			if sps.Unmarshal(g.VideoH264Forma.SPS) == nil {
+				frameNumTracker.setSPS(&sps)
+			}
+		}
 
 		// Extracts DTS from the bitstream to support B-frame H264 streams.
 		// Created once per stream (tracks reorder state across access units).
@@ -902,6 +1054,16 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 			default:
 			}
 
+			if expected, gap := referenceState.observeSequence(rtppkt.SequenceNumber); gap {
+				log.WithFields(log.Fields{
+					"component":                "capture",
+					"event":                    "video_rtp_sequence_gap",
+					"expected_sequence_number": expected,
+					"received_sequence_number": rtppkt.SequenceNumber,
+					"stream":                   streamType,
+				}).Warn("RTP sequence gap invalidated video reference chain")
+			}
+
 			if len(rtppkt.Payload) > 0 {
 
 				// decode timestamps — validate each call separately
@@ -917,6 +1079,9 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 				// this is a keyframe.
 				au, errDecode := g.VideoH264Decoder.Decode(rtppkt)
 				if errDecode != nil {
+					if errDecode != rtph264.ErrMorePacketsNeeded {
+						referenceState.invalidate()
+					}
 					if errDecode != rtph264.ErrNonStartingPacketAndNoPrevious && errDecode != rtph264.ErrMorePacketsNeeded {
 						log.Error("capture.golibrtsp.Start(): " + errDecode.Error())
 					}
@@ -961,6 +1126,7 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 						if errSPS == nil {
 							// Debug SPS information
 							g.debugSPSInfo(&sps, streamType)
+							frameNumTracker.setSPS(&sps)
 
 							// Get width
 							g.Streams[g.VideoH264Index].Width = sps.Width()
@@ -1022,6 +1188,39 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 					return
 				}
 
+				if expected, received, gap, ok := frameNumTracker.observe(au); ok && gap {
+					referenceState.invalidate()
+					log.WithFields(log.Fields{
+						"component":          "capture",
+						"event":              "h264_frame_num_gap",
+						"expected_frame_num": expected,
+						"received_frame_num": received,
+						"stream":             streamType,
+					}).Warn("H264 frame_num gap invalidated video reference chain")
+				}
+
+				accepted, recoveredDropped := referenceState.accept(idrPresent)
+				if !accepted {
+					if referenceState.droppedAccessUnits == 1 {
+						log.WithFields(log.Fields{
+							"codec":     "H264",
+							"component": "capture",
+							"event":     "dependent_access_units_dropped",
+							"stream":    streamType,
+						}).Warn("Dropping dependent video frames until the next IDR")
+					}
+					return
+				}
+				if recoveredDropped > 0 {
+					log.WithFields(log.Fields{
+						"codec":                "H264",
+						"component":            "capture",
+						"dropped_access_units": recoveredDropped,
+						"event":                "video_reference_chain_recovered",
+						"stream":               streamType,
+					}).Info("Video reference chain recovered at IDR")
+				}
+
 				if idrPresent {
 					log.WithFields(log.Fields{
 						"component":  "capture",
@@ -1165,6 +1364,7 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 
 	// called when a video RTP packet arrives for H265
 	if g.VideoH265Media != nil && g.VideoH265Forma != nil {
+		referenceState := &videoReferenceState{}
 
 		// Extracts DTS from the bitstream to support B-frame H265 streams.
 		// Created once per stream (tracks reorder state across access units).
@@ -1180,6 +1380,16 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 			default:
 			}
 
+			if expected, gap := referenceState.observeSequence(rtppkt.SequenceNumber); gap {
+				log.WithFields(log.Fields{
+					"component":                "capture",
+					"event":                    "video_rtp_sequence_gap",
+					"expected_sequence_number": expected,
+					"received_sequence_number": rtppkt.SequenceNumber,
+					"stream":                   streamType,
+				}).Warn("RTP sequence gap invalidated video reference chain")
+			}
+
 			if len(rtppkt.Payload) > 0 {
 
 				// decode timestamps — validate each call separately
@@ -1195,6 +1405,9 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 				// this is a keyframe.
 				au, errDecode := g.VideoH265Decoder.Decode(rtppkt)
 				if errDecode != nil {
+					if errDecode != rtph265.ErrMorePacketsNeeded {
+						referenceState.invalidate()
+					}
 					if errDecode != rtph265.ErrNonStartingPacketAndNoPrevious && errDecode != rtph265.ErrMorePacketsNeeded {
 						log.Error("capture.golibrtsp.Start(): " + errDecode.Error())
 					}
@@ -1244,6 +1457,28 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
 					return
 				}
 
+				accepted, recoveredDropped := referenceState.accept(isRandomAccess)
+				if !accepted {
+					if referenceState.droppedAccessUnits == 1 {
+						log.WithFields(log.Fields{
+							"codec":     "H265",
+							"component": "capture",
+							"event":     "dependent_access_units_dropped",
+							"stream":    streamType,
+						}).Warn("Dropping dependent video frames until the next random-access frame")
+					}
+					return
+				}
+				if recoveredDropped > 0 {
+					log.WithFields(log.Fields{
+						"codec":                "H265",
+						"component":            "capture",
+						"dropped_access_units": recoveredDropped,
+						"event":                "video_reference_chain_recovered",
+						"stream":               streamType,
+					}).Info("Video reference chain recovered at random-access frame")
+				}
+
 				// add VPS, SPS and PPS before random access access unit
 				if isRandomAccess {
 					au = append([][]byte{

machinery/src/capture/gortsplib_test.go

@@ -11,6 +11,8 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
+
+	"github.com/bluenviron/mediacommon/pkg/codecs/h264"
 )
 
 func TestPreRecordingGOPCount(t *testing.T) {
@@ -134,3 +136,114 @@ func TestRTSPSTLSConfig(t *testing.T) {
 		}
 	})
 }
+
+func TestVideoReferenceStateDropsDependentFramesUntilRandomAccess(t *testing.T) {
+	state := &videoReferenceState{}
+
+	accepted, recovered := state.accept(false)
+	if accepted || recovered != 0 || state.droppedAccessUnits != 1 {
+		t.Fatalf("initial dependent frame = (accepted=%t, recovered=%d, dropped=%d), want (false, 0, 1)",
+			accepted, recovered, state.droppedAccessUnits)
+	}
+
+	accepted, recovered = state.accept(true)
+	if !accepted || recovered != 1 || !state.valid {
+		t.Fatalf("random-access recovery = (accepted=%t, recovered=%d, valid=%t), want (true, 1, true)",
+			accepted, recovered, state.valid)
+	}
+
+	if expected, gap := state.observeSequence(100); gap || expected != 100 {
+		t.Fatalf("first sequence observation = (expected=%d, gap=%t), want (100, false)", expected, gap)
+	}
+	if expected, gap := state.observeSequence(102); !gap || expected != 101 || state.valid {
+		t.Fatalf("sequence gap = (expected=%d, gap=%t, valid=%t), want (101, true, false)",
+			expected, gap, state.valid)
+	}
+
+	accepted, recovered = state.accept(false)
+	if accepted || recovered != 0 || state.droppedAccessUnits != 1 {
+		t.Fatalf("post-loss dependent frame = (accepted=%t, recovered=%d, dropped=%d), want (false, 0, 1)",
+			accepted, recovered, state.droppedAccessUnits)
+	}
+
+	accepted, recovered = state.accept(true)
+	if !accepted || recovered != 1 || !state.valid {
+		t.Fatalf("post-loss recovery = (accepted=%t, recovered=%d, valid=%t), want (true, 1, true)",
+			accepted, recovered, state.valid)
+	}
+}
+
+func TestVideoReferenceStateAcceptsSequenceWraparound(t *testing.T) {
+	state := &videoReferenceState{valid: true}
+
+	if _, gap := state.observeSequence(^uint16(0)); gap {
+		t.Fatal("first sequence observation reported a gap")
+	}
+	if expected, gap := state.observeSequence(0); gap || expected != 0 || !state.valid {
+		t.Fatalf("sequence wraparound = (expected=%d, gap=%t, valid=%t), want (0, false, true)",
+			expected, gap, state.valid)
+	}
+}
+
+func TestH264FrameNumTrackerDetectsMissingIDR(t *testing.T) {
+	tracker := &h264FrameNumTracker{}
+	tracker.setSPS(&h264.SPS{
+		Log2MaxFrameNumMinus4: 1,
+		FrameMbsOnlyFlag:      true,
+	})
+
+	if _, _, gap, ok := tracker.observe([][]byte{testH264Slice(h264.NALUTypeIDR, 0)}); !ok || gap {
+		t.Fatalf("initial IDR observation = (ok=%t, gap=%t), want (true, false)", ok, gap)
+	}
+	for frameNum := uint32(1); frameNum <= 24; frameNum++ {
+		if expected, received, gap, ok := tracker.observe([][]byte{testH264Slice(h264.NALUTypeNonIDR, frameNum)}); !ok || gap {
+			t.Fatalf("frame_num %d observation = (expected=%d, received=%d, ok=%t, gap=%t), want no gap",
+				frameNum, expected, received, ok, gap)
+		}
+	}
+
+	expected, received, gap, ok := tracker.observe([][]byte{testH264Slice(h264.NALUTypeNonIDR, 1)})
+	if !ok || !gap || expected != 25 || received != 1 {
+		t.Fatalf("missing IDR transition = (expected=%d, received=%d, ok=%t, gap=%t), want (25, 1, true, true)",
+			expected, received, ok, gap)
+	}
+}
+
+func TestH264FrameNumTrackerResetsAtIDR(t *testing.T) {
+	tracker := &h264FrameNumTracker{}
+	tracker.setSPS(&h264.SPS{
+		Log2MaxFrameNumMinus4: 1,
+		FrameMbsOnlyFlag:      true,
+	})
+
+	tracker.observe([][]byte{testH264Slice(h264.NALUTypeIDR, 0)})
+	tracker.observe([][]byte{testH264Slice(h264.NALUTypeNonIDR, 1)})
+	if expected, received, gap, ok := tracker.observe([][]byte{testH264Slice(h264.NALUTypeIDR, 0)}); !ok || gap {
+		t.Fatalf("IDR reset = (expected=%d, received=%d, ok=%t, gap=%t), want no gap",
+			expected, received, ok, gap)
+	}
+	if expected, received, gap, ok := tracker.observe([][]byte{testH264Slice(h264.NALUTypeNonIDR, 1)}); !ok || gap || expected != 1 || received != 1 {
+		t.Fatalf("post-IDR frame = (expected=%d, received=%d, ok=%t, gap=%t), want (1, 1, true, false)",
+			expected, received, ok, gap)
+	}
+}
+
+func TestH264FrameNumTrackerHonorsPermittedGaps(t *testing.T) {
+	tracker := &h264FrameNumTracker{}
+	tracker.setSPS(&h264.SPS{
+		Log2MaxFrameNumMinus4:          1,
+		FrameMbsOnlyFlag:               true,
+		GapsInFrameNumValueAllowedFlag: true,
+	})
+
+	tracker.observe([][]byte{testH264Slice(h264.NALUTypeIDR, 0)})
+	expected, received, gap, ok := tracker.observe([][]byte{testH264Slice(h264.NALUTypeNonIDR, 2)})
+	if !ok || gap || expected != 1 || received != 2 {
+		t.Fatalf("permitted frame_num gap = (expected=%d, received=%d, ok=%t, gap=%t), want (1, 2, true, false)",
+			expected, received, ok, gap)
+	}
+}
+
+func testH264Slice(naluType h264.NALUType, frameNum uint32) []byte {
+	return []byte{0x60 | byte(naluType), 0xE0 | byte(frameNum&0x1F)}
+}

machinery/src/packets/queue.go

@@ -40,6 +40,9 @@ func NewQueue() *Queue {
 }
 
 func (self *Queue) SetMaxGopCount(n int) {
+	if n < 1 {
+		n = 1
+	}
 	self.lock.Lock()
 	self.maxgopcount = n
 	self.lock.Unlock()
@@ -97,20 +100,31 @@ func (self *Queue) WritePacket(pkt Packet) (err error) {
 	self.lock.Lock()
 
 	self.buf.Push(pkt)
-	if pkt.Idx == int8(self.videoidx) && pkt.IsKeyFrame {
+	isVideoKeyframe := pkt.Idx == int8(self.videoidx) && pkt.IsKeyFrame
+	if isVideoKeyframe {
 		self.curgopcount++
-	}
 
-	for self.curgopcount >= self.maxgopcount && self.buf.Count > 1 {
-		pkt := self.buf.Pop()
-		if pkt.Idx == int8(self.videoidx) && pkt.IsKeyFrame {
-			self.curgopcount--
-		}
-		if self.curgopcount < self.maxgopcount {
-			break
+		for self.curgopcount > self.maxgopcount && self.buf.Count > 0 {
+			for self.buf.Count > 0 {
+				oldest := self.buf.Pop()
+				if oldest.Idx == int8(self.videoidx) && oldest.IsKeyFrame {
+					self.curgopcount--
+					break
+				}
+			}
+
+			// Discard the dependent frames from the evicted GOP as well. Keeping
+			// them would let an overrun cursor resume on a P-frame whose reference
+			// keyframe is no longer in the queue.
+			for self.buf.Count > 0 {
+				oldest := self.buf.Get(self.buf.Head)
+				if oldest.Idx == int8(self.videoidx) && oldest.IsKeyFrame {
+					break
+				}
+				self.buf.Pop()
+			}
 		}
 	}
-	//println("shrink", self.curgopcount, self.maxgopcount, self.buf.Head, self.buf.Tail, "count", self.buf.Count, "size", self.buf.Size)
 
 	self.cond.Broadcast()
 

machinery/src/packets/queue_test.go

@@ -107,3 +107,102 @@ func TestQueueCursorReadPacketContextDoesNotAllocateWhenPacketIsAvailable(t *tes
 		t.Fatalf("ReadPacketContext() allocations = %.2f, want 0", allocations)
 	}
 }
+
+func TestQueueEvictionKeepsKeyframeAtHead(t *testing.T) {
+	queue := NewQueue()
+	if err := queue.WriteHeader([]Stream{{Index: 0, IsVideo: true}}); err != nil {
+		t.Fatal(err)
+	}
+	queue.SetMaxGopCount(2)
+
+	for _, packet := range []Packet{
+		{Idx: 0, IsVideo: true, IsKeyFrame: true, Data: []byte("IDR-0")},
+		{Idx: 1, IsAudio: true, Data: []byte("audio-0")},
+		{Idx: 0, IsVideo: true, Data: []byte("P-0")},
+		{Idx: 0, IsVideo: true, IsKeyFrame: true, Data: []byte("IDR-1")},
+		{Idx: 0, IsVideo: true, Data: []byte("P-1")},
+		{Idx: 0, IsVideo: true, IsKeyFrame: true, Data: []byte("IDR-2")},
+	} {
+		if err := queue.WritePacket(packet); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	oldest, err := queue.Oldest().ReadPacket()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !oldest.IsKeyFrame || string(oldest.Data) != "IDR-1" {
+		t.Fatalf("oldest packet = %q (keyframe=%t), want IDR-1 keyframe", oldest.Data, oldest.IsKeyFrame)
+	}
+}
+
+func TestQueueWithSingleGOPRetainsCurrentKeyframe(t *testing.T) {
+	queue := NewQueue()
+	if err := queue.WriteHeader([]Stream{{Index: 0, IsVideo: true}}); err != nil {
+		t.Fatal(err)
+	}
+	queue.SetMaxGopCount(1)
+
+	for _, packet := range []Packet{
+		{Idx: 0, IsVideo: true, IsKeyFrame: true, Data: []byte("IDR-0")},
+		{Idx: 0, IsVideo: true, Data: []byte("P-0")},
+		{Idx: 0, IsVideo: true, IsKeyFrame: true, Data: []byte("IDR-1")},
+	} {
+		if err := queue.WritePacket(packet); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	oldest, err := queue.Oldest().ReadPacket()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !oldest.IsKeyFrame || string(oldest.Data) != "IDR-1" {
+		t.Fatalf("oldest packet = %q (keyframe=%t), want IDR-1 keyframe", oldest.Data, oldest.IsKeyFrame)
+	}
+}
+
+func TestLaggingQueueCursorResumesAtKeyframe(t *testing.T) {
+	queue := NewQueue()
+	if err := queue.WriteHeader([]Stream{{Index: 0, IsVideo: true}}); err != nil {
+		t.Fatal(err)
+	}
+	queue.SetMaxGopCount(2)
+
+	for _, packet := range []Packet{
+		{Idx: 0, IsVideo: true, IsKeyFrame: true, Data: []byte("IDR-0")},
+		{Idx: 0, IsVideo: true, Data: []byte("P-0")},
+		{Idx: 0, IsVideo: true, IsKeyFrame: true, Data: []byte("IDR-1")},
+	} {
+		if err := queue.WritePacket(packet); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	cursor := queue.Oldest()
+	first, err := cursor.ReadPacket()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(first.Data) != "IDR-0" {
+		t.Fatalf("first packet = %q, want IDR-0", first.Data)
+	}
+
+	for _, packet := range []Packet{
+		{Idx: 0, IsVideo: true, Data: []byte("P-1")},
+		{Idx: 0, IsVideo: true, IsKeyFrame: true, Data: []byte("IDR-2")},
+	} {
+		if err := queue.WritePacket(packet); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	resumed, err := cursor.ReadPacket()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !resumed.IsKeyFrame || string(resumed.Data) != "IDR-1" {
+		t.Fatalf("resumed packet = %q (keyframe=%t), want IDR-1 keyframe", resumed.Data, resumed.IsKeyFrame)
+	}
+}