Merge pull request #339 from kerberos-io/fix/idr-loss-recovery

fix/idr-loss-recovery

.github/workflows/pr-build.yml

@@ -10,7 +10,7 @@ jobs:
   build:
     runs-on: ${{ matrix.runner }}
     permissions:
-      contents: write
+      contents: read
     strategy:
       matrix:
         include:
@@ -21,11 +21,6 @@ jobs:
     steps:
       - name: Checkout
         uses: actions/checkout@v4
-      - name: Login to DockerHub
-        uses: docker/login-action@v3
-        with:
-          username: ${{ secrets.DOCKER_USERNAME }}
-          password: ${{ secrets.DOCKER_PASSWORD }}
       - name: Run Build
         run: |
           docker build --build-arg VERSION=$(echo "${{ github.event.pull_request.head.sha }}" | cut -c1-7) -t ${{ matrix.architecture }} .

machinery/go.mod

@@ -25,7 +25,7 @@ require (
 	github.com/gorilla/websocket v1.5.3
 	github.com/kellydunn/golang-geo v0.7.0
 	github.com/kerberos-io/joy4 v1.0.64
-	github.com/kerberos-io/onvif v1.2.2
+	github.com/kerberos-io/onvif v1.2.3
 	github.com/minio/minio-go/v6 v6.0.57
 	github.com/moq-dev/moq-go v0.5.7
 	github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646

machinery/go.sum

@@ -776,8 +776,8 @@ github.com/kellydunn/golang-geo v0.7.0 h1:A5j0/BvNgGwY6Yb6inXQxzYwlPHc6WVZR+Mrar
 github.com/kellydunn/golang-geo v0.7.0/go.mod h1:YYlQPJ+DPEzrHx8kT3oPHC/NjyvCCXE+IuKGKdrjrcU=
 github.com/kerberos-io/joy4 v1.0.64 h1:gTUSotHSOhp9mNqEecgq88tQHvpj7TjmrvPUsPm0idg=
 github.com/kerberos-io/joy4 v1.0.64/go.mod h1:nZp4AjvKvTOXRrmDyAIOw+Da+JA5OcSo/JundGfOlFU=
-github.com/kerberos-io/onvif v1.2.2 h1:QnxITps7xvAVD2abWRsa3+p9QexjKESQQldUFMx/mYA=
-github.com/kerberos-io/onvif v1.2.2/go.mod h1:XSgEQXmEDjUQTbdXvsaRJt6Az8YPGj7L+j5iXKEGijU=
+github.com/kerberos-io/onvif v1.2.3 h1:RDnDvLBd92VCgFOrlbGavzCVN9lQA/qs9novdNFIAOU=
+github.com/kerberos-io/onvif v1.2.3/go.mod h1:XSgEQXmEDjUQTbdXvsaRJt6Az8YPGj7L+j5iXKEGijU=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/asmfmt v1.3.2/go.mod h1:AG8TuvYojzulgDAMCnYn50l/5QV3Bs/tp6j0HLHbNSE=

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/cloud/cloud.go

@@ -541,13 +541,23 @@ func (state *heartbeatONVIFState) fetchEvents(device *goonvif.Device) ([]byte, b
 		}
 	}
 
-	if !initialStateSubscriptionCreated && state.loopPullPoint == "" {
+	// Fall back to querying the relay outputs and digital inputs directly when
+	// no subscription could be created, and also when the subscriptions
+	// produced no I/O events: some devices accept the subscription but reject
+	// the pulls, or never publish the initial I/O state.
+	noSubscription := !initialStateSubscriptionCreated && state.loopPullPoint == ""
+	if noSubscription || heartbeatEventsListEmpty(onvifEventsList) {
 		return state.fallbackEvents(device), operationFailed
 	}
 
 	return onvifEventsList, operationFailed
 }
 
+func heartbeatEventsListEmpty(eventsList []byte) bool {
+	trimmed := strings.TrimSpace(string(eventsList))
+	return trimmed == "" || trimmed == "[]" || trimmed == "null"
+}
+
 func getHeartbeatONVIFPayload(camera models.IPCamera, state *heartbeatONVIFState) heartbeatONVIFPayload {
 	if state == nil {
 		state = newHeartbeatONVIFState()

machinery/src/cloud/cloud_test.go

@@ -15,6 +15,7 @@ import (
 	goonvif "github.com/kerberos-io/onvif"
 	goonvifdevice "github.com/kerberos-io/onvif/device"
 	goonvifptz "github.com/kerberos-io/onvif/ptz"
+	goonvifxsd "github.com/kerberos-io/onvif/xsd/onvif"
 )
 
 func TestHeartbeatFailureLogOmitsHubResponseBody(t *testing.T) {
@@ -590,3 +591,60 @@ func countString(values []string, want string) int {
 	}
 	return count
 }
+
+// Some devices accept the subscription but reject the pulls, or never publish
+// the initial I/O state. The relay outputs and digital inputs must then be
+// queried directly instead of reporting no I/O at all.
+func TestHeartbeatONVIFPayloadFallsBackToDirectIOWhenSubscriptionsReturnNoEvents(t *testing.T) {
+	restoreHeartbeatONVIFStubs(t)
+
+	device := newTestONVIFDevice()
+	camera := models.IPCamera{
+		ONVIFXAddr:    "http://camera/onvif",
+		ONVIFUsername: "operator",
+		ONVIFPassword: "secret",
+	}
+
+	state := newHeartbeatONVIFState()
+	state.cameraConfiguration = camera
+	state.cameraKey = heartbeatONVIFCameraKey(camera)
+	state.device = device
+	state.staticLoaded = true
+	state.staticPayload.enabled = "true"
+
+	heartbeatCreatePullPointSubscription = func(*goonvif.Device) (string, error) {
+		return "pull-point", nil
+	}
+	heartbeatGetEventMessages = func(*goonvif.Device, string) ([]agentonvif.ONVIFEvents, error) {
+		return nil, errors.New("pull rejected")
+	}
+	heartbeatUnsubscribePullPoint = func(*goonvif.Device, string) error {
+		return nil
+	}
+	var relayCalls, inputCalls int
+	heartbeatGetRelayOutputs = func(*goonvif.Device) (goonvifdevice.GetRelayOutputsResponse, error) {
+		relayCalls++
+		return goonvifdevice.GetRelayOutputsResponse{RelayOutputs: []goonvifxsd.RelayOutput{{DeviceEntity: goonvifxsd.DeviceEntity{Token: "relay0"}}}}, nil
+	}
+	heartbeatGetDigitalInputs = func(*goonvif.Device) (goonvifdevice.GetDigitalInputsResponse, error) {
+		inputCalls++
+		return goonvifdevice.GetDigitalInputsResponse{DigitalInputs: []goonvifxsd.DigitalInput{{Token: "di0"}}}, nil
+	}
+
+	payload := getHeartbeatONVIFPayload(camera, state)
+
+	if relayCalls != 1 || inputCalls != 1 {
+		t.Fatalf("direct I/O queries: relay=%d inputs=%d, want 1 each", relayCalls, inputCalls)
+	}
+	var events []agentonvif.ONVIFEvents
+	if err := json.Unmarshal(payload.eventsList, &events); err != nil {
+		t.Fatalf("payload.eventsList is not valid JSON: %v", err)
+	}
+	got := map[string]string{}
+	for _, event := range events {
+		got[event.Key] = event.Type
+	}
+	if got["relay0"] != "output" || got["di0"] != "input" || len(got) != 2 {
+		t.Fatalf("payload.eventsList = %s, want relay0 output and di0 input", payload.eventsList)
+	}
+}

machinery/src/onvif/main.go

@@ -1108,6 +1108,23 @@ func CreatePullPointSubscription(dev *onvif.Device) (string, error) {
 	return pullPointAdress, err
 }
 
+const (
+	wsAddressingNamespace = "http://www.w3.org/2005/08/addressing"
+	pullMessagesAction    = "http://www.onvif.org/ver10/events/wsdl/PullPointSubscription/PullMessagesRequest"
+	unsubscribeAction     = "http://docs.oasis-open.org/wsn/bw-2/SubscriptionManager/UnsubscribeRequest"
+)
+
+// subscriptionAddressingHeader builds the WS-Addressing Action and To headers
+// for a request sent to a pull-point subscription. Some devices identify the
+// subscription from wsa:To rather than the request URL, and reject requests
+// without it as NotAuthorized.
+func subscriptionAddressingHeader(subscriptionAddress string, action string) string {
+	var address bytes.Buffer
+	_ = xml.EscapeText(&address, []byte(subscriptionAddress))
+	return `<wsa:Action xmlns:wsa="` + wsAddressingNamespace + `">` + action + `</wsa:Action>` +
+		`<wsa:To xmlns:wsa="` + wsAddressingNamespace + `">` + address.String() + `</wsa:To>`
+}
+
 func UnsubscribePullPoint(dev *onvif.Device, pullPointAddress string) error {
 
 	// Unsubscribe from the device
@@ -1121,7 +1138,7 @@ func UnsubscribePullPoint(dev *onvif.Device, pullPointAddress string) error {
 		return err
 	}
 
-	res, err := dev.SendSoap(pullPointAddress, string(requestBody))
+	res, err := dev.SendSoapWithHeader(pullPointAddress, string(requestBody), subscriptionAddressingHeader(pullPointAddress, unsubscribeAction))
 	if err != nil {
 		log.WithError(err).WithFields(log.Fields{
 			"component": "onvif",
@@ -1203,7 +1220,7 @@ func GetEventMessages(dev *onvif.Device, pullPointAddress string) ([]ONVIFEvents
 				log.Error("onvif.main.GetEventMessages(pullMessages): " + err.Error())
 				return eventsArray, err
 			}
-			res, err := dev.SendSoap(string(subscriptionURI), string(requestBody))
+			res, err := dev.SendSoapWithHeader(string(subscriptionURI), string(requestBody), subscriptionAddressingHeader(subscriptionURI, pullMessagesAction))
 			if err != nil {
 				log.Error("onvif.main.GetEventMessages(pullMessages): " + err.Error())
 				return eventsArray, err

machinery/src/onvif/subscription_test.go

@@ -0,0 +1,88 @@
+package onvif
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/kerberos-io/onvif"
+)
+
+// newSubscriptionTestDevice starts a fake ONVIF device and returns a connected
+// device plus the request bodies it received on the subscription endpoint.
+func newSubscriptionTestDevice(t *testing.T, response string) (*onvif.Device, string, func() []string) {
+	t.Helper()
+
+	var mu sync.Mutex
+	var subscriptionBodies []string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, _ := io.ReadAll(r.Body)
+		if r.URL.Path == "/onvif/subscription" {
+			mu.Lock()
+			subscriptionBodies = append(subscriptionBodies, string(body))
+			mu.Unlock()
+			_, _ = w.Write([]byte(response))
+			return
+		}
+		_, _ = w.Write([]byte(`<Envelope><Body/></Envelope>`))
+	}))
+	t.Cleanup(srv.Close)
+
+	dev, err := onvif.NewDevice(onvif.DeviceParams{Xaddr: strings.TrimPrefix(srv.URL, "http://")})
+	if err != nil {
+		t.Fatalf("NewDevice() error = %v", err)
+	}
+	subscriptionAddress := srv.URL + "/onvif/subscription?subscription=7&idx=2"
+	return dev, subscriptionAddress, func() []string {
+		mu.Lock()
+		defer mu.Unlock()
+		return append([]string(nil), subscriptionBodies...)
+	}
+}
+
+func assertSubscriptionAddressing(t *testing.T, body string, action string) {
+	t.Helper()
+
+	for _, want := range []string{
+		`<wsa:Action xmlns:wsa="http://www.w3.org/2005/08/addressing">` + action + `</wsa:Action>`,
+		`<wsa:To xmlns:wsa="http://www.w3.org/2005/08/addressing">`,
+		`/onvif/subscription?subscription=7&amp;idx=2</wsa:To>`,
+	} {
+		if !strings.Contains(body, want) {
+			t.Fatalf("request body does not contain %q:\n%s", want, body)
+		}
+	}
+}
+
+// Some devices identify a pull-point subscription by wsa:To and reject
+// PullMessages without it as NotAuthorized.
+func TestGetEventMessagesSendsSubscriptionAddressingHeaders(t *testing.T) {
+	dev, subscriptionAddress, bodies := newSubscriptionTestDevice(t, `<Envelope><Body><PullMessagesResponse/></Body></Envelope>`)
+
+	if _, err := GetEventMessages(dev, subscriptionAddress); err != nil {
+		t.Fatalf("GetEventMessages() error = %v", err)
+	}
+
+	received := bodies()
+	if len(received) != 1 {
+		t.Fatalf("subscription requests = %d, want 1", len(received))
+	}
+	assertSubscriptionAddressing(t, received[0], pullMessagesAction)
+}
+
+func TestUnsubscribePullPointSendsSubscriptionAddressingHeaders(t *testing.T) {
+	dev, subscriptionAddress, bodies := newSubscriptionTestDevice(t, `<Envelope><Body><UnsubscribeResponse/></Body></Envelope>`)
+
+	if err := UnsubscribePullPoint(dev, subscriptionAddress); err != nil {
+		t.Fatalf("UnsubscribePullPoint() error = %v", err)
+	}
+
+	received := bodies()
+	if len(received) != 1 {
+		t.Fatalf("subscription requests = %d, want 1", len(received))
+	}
+	assertSubscriptionAddressing(t, received[0], unsubscribeAction)
+}

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)
+	}
+}