From 99606e82c18b0e5deb89254fcd678079aa0bceff Mon Sep 17 00:00:00 2001 From: Luis Miranda <161006+luuuis@users.noreply.github.com> Date: Tue, 28 Apr 2026 13:09:57 +0100 Subject: [PATCH] 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 --- frigate/test/test_camera_watchdog.py | 106 +++++++++++++++++++++++++++ frigate/video/ffmpeg.py | 68 +++++++++-------- 2 files changed, 145 insertions(+), 29 deletions(-) create mode 100644 frigate/test/test_camera_watchdog.py diff --git a/frigate/test/test_camera_watchdog.py b/frigate/test/test_camera_watchdog.py new file mode 100644 index 0000000000..ad05e155f1 --- /dev/null +++ b/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() diff --git a/frigate/video/ffmpeg.py b/frigate/video/ffmpeg.py index d30dc3b188..dfd3467e2a 100644 --- a/frigate/video/ffmpeg.py +++ b/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: - 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 + self._drain_segment_updates() now = datetime.now().timestamp()