fix: prevent segment timestamps from regressing on out-of-order delivery

ZMQ segment update messages can arrive out of order when the subscriber
queue has a backlog. Previously, each message unconditionally overwrote
latest_valid_segment_time, latest_invalid_segment_time, and
latest_cache_segment_time with the payload timestamp, so a delayed
message carrying an older timestamp could move the value backwards.

The stale-recording restart logic in run() compares these timestamps
against the current wall clock to decide whether to restart the ffmpeg
record process. A regressed timestamp makes the watchdog believe no
segment has been received for longer than 120 s, triggering a spurious
restart of the record process.

Fix by replacing the direct assignment with max() so each field only
ever advances. The inner while loop is extracted into a new private
method _drain_segment_updates() to allow the behaviour to be unit
tested independently of the full run() lifecycle.

Adds frigate/test/test_camera_watchdog.py with six tests covering
in-order delivery, out-of-order delivery, cache segment resets, and
filtering of messages belonging to other cameras.

Co-authored-by: Claude <claude@anthropic.com>

frigate/test/test_camera_watchdog.py

@@ -0,0 +1,106 @@
+import logging
+import unittest
+from unittest.mock import MagicMock
+
+from frigate.comms.recordings_updater import RecordingsDataTypeEnum
+from frigate.video.ffmpeg import CameraWatchdog
+
+
+def _make_watchdog(camera_name: str = "front_door") -> CameraWatchdog:
+    """Create a CameraWatchdog with only the attributes needed by _process_segment_updates."""
+    watchdog = object.__new__(CameraWatchdog)
+    watchdog.config = MagicMock()
+    watchdog.config.name = camera_name
+    watchdog.logger = logging.getLogger("test.watchdog")
+    watchdog.latest_valid_segment_time = 0.0
+    watchdog.latest_invalid_segment_time = 0.0
+    watchdog.latest_cache_segment_time = 0.0
+    watchdog.segment_subscriber = MagicMock()
+    return watchdog
+
+
+def _topic(type_enum: RecordingsDataTypeEnum) -> str:
+    return f"recordings/{type_enum.value}"
+
+
+def _msg(topic: RecordingsDataTypeEnum, camera: str, t: float):
+    return (_topic(topic), (camera, t, None))
+
+
+class TestProcessSegmentUpdates(unittest.TestCase):
+    def test_in_order_valid_segments(self):
+        """Segments arriving in order advance latest_valid_segment_time to the last value."""
+        watchdog = _make_watchdog()
+        watchdog.segment_subscriber.check_for_update.side_effect = [
+            _msg(RecordingsDataTypeEnum.valid, "front_door", 10.0),
+            _msg(RecordingsDataTypeEnum.valid, "front_door", 20.0),
+            _msg(RecordingsDataTypeEnum.valid, "front_door", 30.0),
+            (None, None),
+        ]
+        watchdog._drain_segment_updates()
+        self.assertEqual(watchdog.latest_valid_segment_time, 30.0)
+
+    def test_out_of_order_valid_segments(self):
+        """Out-of-order valid segments still result in the maximum timestamp."""
+        watchdog = _make_watchdog()
+        watchdog.segment_subscriber.check_for_update.side_effect = [
+            _msg(RecordingsDataTypeEnum.valid, "front_door", 20.0),
+            _msg(RecordingsDataTypeEnum.valid, "front_door", 10.0),
+            _msg(RecordingsDataTypeEnum.valid, "front_door", 30.0),
+            (None, None),
+        ]
+        watchdog._drain_segment_updates()
+        self.assertEqual(watchdog.latest_valid_segment_time, 30.0)
+
+    def test_out_of_order_cache_segments(self):
+        """Out-of-order cache (latest) segments still result in the maximum timestamp."""
+        watchdog = _make_watchdog()
+        watchdog.segment_subscriber.check_for_update.side_effect = [
+            _msg(RecordingsDataTypeEnum.latest, "front_door", 100.0),
+            _msg(RecordingsDataTypeEnum.latest, "front_door", 50.0),
+            _msg(RecordingsDataTypeEnum.latest, "front_door", 200.0),
+            (None, None),
+        ]
+        watchdog._drain_segment_updates()
+        self.assertEqual(watchdog.latest_cache_segment_time, 200.0)
+
+    def test_cache_none_resets_to_zero(self):
+        """A None cache segment time resets latest_cache_segment_time to 0."""
+        watchdog = _make_watchdog()
+        watchdog.segment_subscriber.check_for_update.side_effect = [
+            _msg(RecordingsDataTypeEnum.latest, "front_door", 100.0),
+            (_topic(RecordingsDataTypeEnum.latest), ("front_door", None, None)),
+            (None, None),
+        ]
+        watchdog._drain_segment_updates()
+        self.assertEqual(watchdog.latest_cache_segment_time, 0.0)
+
+    def test_messages_for_other_camera_ignored(self):
+        """Segment messages for a different camera do not update timestamps."""
+        watchdog = _make_watchdog(camera_name="front_door")
+        watchdog.segment_subscriber.check_for_update.side_effect = [
+            _msg(RecordingsDataTypeEnum.valid, "back_yard", 999.0),
+            _msg(RecordingsDataTypeEnum.invalid, "back_yard", 999.0),
+            _msg(RecordingsDataTypeEnum.latest, "back_yard", 999.0),
+            (None, None),
+        ]
+        watchdog._drain_segment_updates()
+        self.assertEqual(watchdog.latest_valid_segment_time, 0.0)
+        self.assertEqual(watchdog.latest_invalid_segment_time, 0.0)
+        self.assertEqual(watchdog.latest_cache_segment_time, 0.0)
+
+    def test_mixed_types_tracked_independently(self):
+        """Valid and cache segment times are each tracked independently."""
+        watchdog = _make_watchdog()
+        watchdog.segment_subscriber.check_for_update.side_effect = [
+            _msg(RecordingsDataTypeEnum.valid, "front_door", 10.0),
+            _msg(RecordingsDataTypeEnum.latest, "front_door", 30.0),
+            (None, None),
+        ]
+        watchdog._drain_segment_updates()
+        self.assertEqual(watchdog.latest_valid_segment_time, 10.0)
+        self.assertEqual(watchdog.latest_cache_segment_time, 30.0)
+
+
+if __name__ == "__main__":
+    unittest.main()

frigate/video/ffmpeg.py

@@ -202,6 +202,44 @@ class CameraWatchdog(threading.Thread):
         self._check_config_updates()
         return self.config.enabled
 
+    def _drain_segment_updates(self) -> None:
+        """Drain the segment subscriber queue and update latest segment timestamps."""
+        while True:
+            update = self.segment_subscriber.check_for_update(timeout=0)
+
+            if update == (None, None):
+                break
+
+            raw_topic, payload = update
+            if raw_topic and payload:
+                topic = str(raw_topic)
+                camera, segment_time, _ = payload
+
+                if camera != self.config.name:
+                    continue
+
+                if topic.endswith(RecordingsDataTypeEnum.valid.value):
+                    self.logger.debug(
+                        f"Latest valid recording segment time on {camera}: {segment_time}"
+                    )
+                    self.latest_valid_segment_time = max(
+                        self.latest_valid_segment_time, segment_time
+                    )
+                elif topic.endswith(RecordingsDataTypeEnum.invalid.value):
+                    self.logger.warning(
+                        f"Invalid recording segment detected for {camera} at {segment_time}"
+                    )
+                    self.latest_invalid_segment_time = max(
+                        self.latest_invalid_segment_time, segment_time
+                    )
+                elif topic.endswith(RecordingsDataTypeEnum.latest.value):
+                    if segment_time is not None:
+                        self.latest_cache_segment_time = max(
+                            self.latest_cache_segment_time, segment_time
+                        )
+                    else:
+                        self.latest_cache_segment_time = 0
+
     def reset_capture_thread(
         self, terminate: bool = True, drain_output: bool = True
     ) -> None:
@@ -303,35 +341,7 @@ class CameraWatchdog(threading.Thread):
             if not enabled:
                 continue
 
-            while True:
+            self._drain_segment_updates()
-                update = self.segment_subscriber.check_for_update(timeout=0)
-
-                if update == (None, None):
-                    break
-
-                raw_topic, payload = update
-                if raw_topic and payload:
-                    topic = str(raw_topic)
-                    camera, segment_time, _ = payload
-
-                    if camera != self.config.name:
-                        continue
-
-                    if topic.endswith(RecordingsDataTypeEnum.valid.value):
-                        self.logger.debug(
-                            f"Latest valid recording segment time on {camera}: {segment_time}"
-                        )
-                        self.latest_valid_segment_time = segment_time
-                    elif topic.endswith(RecordingsDataTypeEnum.invalid.value):
-                        self.logger.warning(
-                            f"Invalid recording segment detected for {camera} at {segment_time}"
-                        )
-                        self.latest_invalid_segment_time = segment_time
-                    elif topic.endswith(RecordingsDataTypeEnum.latest.value):
-                        if segment_time is not None:
-                            self.latest_cache_segment_time = segment_time
-                        else:
-                            self.latest_cache_segment_time = 0
 
             now = datetime.now().timestamp()