Convert sensitive wall clock usages to monotonic (#24864)

frigate/comms/webpush.py

@@ -7,6 +7,7 @@ import logging
 import os
 import queue
 import threading
+import time
 from collections.abc import Callable, Iterator
 from dataclasses import dataclass
 from multiprocessing.synchronize import Event as MpEvent
@@ -185,11 +186,13 @@ class WebPushClient(Communicator):
         }
         self.suspension_broadcaster: Callable[[str, Any, bool], None] | None = None
         self.config_lock = threading.Lock()
+        # monotonic times, so a backward wall clock step can't hold every
+        # notification in cooldown. -inf means none has been sent yet
         self.last_camera_notification_time: dict[str, float] = {
-            c.name: 0  # type: ignore[misc]
+            c.name: float("-inf")  # type: ignore[misc]
             for c in self.config.cameras.values()
         }
-        self.last_notification_time: float = 0
+        self.last_notification_time: float = float("-inf")
         self.user_cameras: dict[str, set[str]] = {}
         self.notification_queue: queue.Queue[PushNotification] = queue.Queue()
         self.notification_thread: threading.Thread | None = None
@@ -344,7 +347,7 @@ class WebPushClient(Communicator):
             if "add" in updates:
                 for camera in updates["add"]:
                     self.suspended_cameras[camera] = 0
-                    self.last_camera_notification_time[camera] = 0
+                    self.last_camera_notification_time[camera] = float("-inf")
 
                 self._refresh_user_cameras()
 
@@ -499,7 +502,7 @@ class WebPushClient(Communicator):
         return camera in allowed
 
     def _within_cooldown(self, camera: str) -> bool:
-        now = datetime.datetime.now().timestamp()
+        now = time.monotonic()
         if now - self.last_notification_time < self.config.notifications.cooldown:
             logger.debug(
                 f"Skipping notification for {camera} - in global cooldown period"
@@ -544,7 +547,7 @@ class WebPushClient(Communicator):
         camera_name: str = getattr(
             self.config.cameras[camera], "friendly_name", None
         ) or titlecase(camera.replace("_", " "))
-        current_time = datetime.datetime.now().timestamp()
+        current_time = time.monotonic()
 
         if self._within_cooldown(camera):
             return
@@ -657,7 +660,7 @@ class WebPushClient(Communicator):
         camera_name: str = getattr(
             self.config.cameras[camera], "friendly_name", None
         ) or titlecase(camera.replace("_", " "))
-        current_time = datetime.datetime.now().timestamp()
+        current_time = time.monotonic()
 
         if self._within_cooldown(camera):
             return

frigate/data_processing/real_time/custom_classification.py

@@ -3,6 +3,7 @@
 import datetime
 import logging
 import os
+import time
 from typing import Any
 
 import cv2
@@ -67,7 +68,8 @@ class CustomStateClassificationProcessor(DeferredRealtimeProcessorApi):
         else:
             self.inference_speed = None
 
-        self.last_run = datetime.datetime.now().timestamp()
+        # monotonic so a backward wall clock step can't pause classification
+        self.last_run = time.monotonic()
         self.__build_detector()
 
     def __build_detector(self) -> None:
@@ -215,11 +217,12 @@ class CustomStateClassificationProcessor(DeferredRealtimeProcessorApi):
         should_run = False
 
         now = datetime.datetime.now().timestamp()
+        mono_now = time.monotonic()
         if (
             self.model_config.state_config.interval
-            and now > self.last_run + self.model_config.state_config.interval
+            and mono_now > self.last_run + self.model_config.state_config.interval
         ):
-            self.last_run = now
+            self.last_run = mono_now
             should_run = True
 
         if (
@@ -228,8 +231,8 @@ class CustomStateClassificationProcessor(DeferredRealtimeProcessorApi):
             and any([box_overlaps(crop, mb) for mb in frame_data.get("motion", [])])
         ):
             # classification should run at most once per second
-            if now > self.last_run + 1:
-                self.last_run = now
+            if mono_now > self.last_run + 1:
+                self.last_run = mono_now
                 should_run = True
 
         # Shortcut: always run if we have a pending state verification to complete
@@ -237,9 +240,9 @@ class CustomStateClassificationProcessor(DeferredRealtimeProcessorApi):
             not should_run
             and camera in self.state_history
             and self.state_history[camera]["pending_state"] is not None
-            and now > self.last_run + 0.5
+            and mono_now > self.last_run + 0.5
         ):
-            self.last_run = now
+            self.last_run = mono_now
             should_run = True
             logger.debug(
                 f"Running verification check for pending state: {self.state_history[camera]['pending_state']} ({self.state_history[camera]['consecutive_count']}/3)"

frigate/jobs/vlm_watch.py

@@ -80,11 +80,12 @@ class VLMWatchRunner(threading.Thread):
         self._broadcast_status()
         self.conversation = [{"role": "system", "content": self._build_system_prompt()}]
 
-        max_end_time = self.job.start_time + self.job.max_duration_minutes * 60
+        # monotonic so a wall clock step can't end the job early or let it overrun
+        max_end_time = time.monotonic() + self.job.max_duration_minutes * 60
 
         try:
             while not self.cancel_event.is_set():
-                if time.time() > max_end_time:
+                if time.monotonic() > max_end_time:
                     logger.debug(
                         "VLM watch job %s timed out after %d minutes",
                         self.job.id,
@@ -236,18 +237,18 @@ class VLMWatchRunner(threading.Thread):
         queue backup, but the loop only exits once a match has been seen
         *and* the cooldown period has elapsed.
         """
-        now = time.time()
+        now = time.monotonic()
         deadline = now + max_wait
         use_cooldown = not self.job.zones
         earliest_wake = now + _DETECTION_COOLDOWN_WITHOUT_ZONE if use_cooldown else 0
         triggered = False
 
         while not self.cancel_event.is_set():
-            remaining = deadline - time.time()
+            remaining = deadline - time.monotonic()
             if remaining <= 0:
                 break
 
-            if triggered and time.time() >= earliest_wake:
+            if triggered and time.monotonic() >= earliest_wake:
                 break
 
             result = self.detection_subscriber.check_for_update(
@@ -287,7 +288,7 @@ class VLMWatchRunner(threading.Thread):
                         "VLM watch job %s: detection match on %s, draining for %.0fs",
                         self.job.id,
                         self.job.camera,
-                        max(0, earliest_wake - time.time()),
+                        max(0, earliest_wake - time.monotonic()),
                     )
                     triggered = True
 

frigate/object_detection/base.py

@@ -1,4 +1,3 @@
-import datetime
 import logging
 import queue
 import threading
@@ -169,9 +168,10 @@ class DetectorRunner(FrigateProcess):
                 logger.warning(f"Failed to get frame {connection_id} from SHM")
                 continue
 
-            # detect and send the output
-            self.start_time.value = datetime.datetime.now().timestamp()
+            # detect and send the output. start_time is monotonic so the watchdog
+            # doesn't mistake a wall clock step for a stuck detector
             mono_start = time.monotonic()
+            self.start_time.value = mono_start
             detections = object_detector.detect_raw(input_frame)
             duration = time.monotonic() - mono_start
 

frigate/output/birdseye.py

@@ -1,6 +1,5 @@
 """Handle outputting birdseye frames via jsmpeg and go2rtc."""
 
-import datetime
 import glob
 import logging
 import math
@@ -8,6 +7,7 @@ import os
 import queue
 import subprocess as sp
 import threading
+import time
 import traceback
 from dataclasses import dataclass
 from multiprocessing.synchronize import Event as MpEvent
@@ -492,7 +492,7 @@ class BirdsEyeFrameManager:
         max_cameras = self.config.birdseye.layout.max_cameras
         max_camera_refresh = False
         if max_cameras:
-            now = datetime.datetime.now().timestamp()
+            now = time.monotonic()
 
             if (
                 len(self.active_cameras) == max_cameras
@@ -800,7 +800,8 @@ class BirdsEyeFrameManager:
 
         camera_state["live_active"] = self.camera_live_active(modes, activity)
 
-        now = datetime.datetime.now().timestamp()
+        # monotonic so a backward wall clock step can't freeze the output
+        now = time.monotonic()
 
         # limit output to 10 fps
         if not force_update and (now - self.last_output_time) < 1 / 10:
@@ -945,7 +946,7 @@ class Birdseye:
                 coordinates = self.birdseye_manager.get_camera_coordinates()
                 self.requestor.send_data(UPDATE_BIRDSEYE_LAYOUT, coordinates)
         if self._idle_interval:
-            now = datetime.datetime.now().timestamp()
+            now = time.monotonic()
             is_idle = len(self.birdseye_manager.camera_layout) == 0
             if (
                 is_idle

frigate/stats/util.py

@@ -106,6 +106,14 @@ def stats_init(
     return stats_tracking
 
 
+def _monotonic_to_wall(timestamp: float) -> float:
+    """Convert a monotonic timestamp to wall clock time, keeping 0 as unset."""
+    if timestamp <= 0:
+        return 0.0
+
+    return time.time() - (time.monotonic() - timestamp)
+
+
 def get_detector_stats(
     stats_tracking: StatsTrackingTypes,
 ) -> dict[str, dict[str, Any]]:
@@ -128,7 +136,7 @@ def get_detector_stats(
             "inference_speed": round(detector.avg_inference_speed.value * 1000, 2),  # type: ignore[attr-defined]
             # issue https://github.com/python/typeshed/issues/8799
             # from mypy 0.981 onwards
-            "detection_start": detector.detection_start.value,  # type: ignore[attr-defined]
+            "detection_start": _monotonic_to_wall(detector.detection_start.value),  # type: ignore[attr-defined]
             # issue https://github.com/python/typeshed/issues/8799
             # from mypy 0.981 onwards
             "pid": pid,

frigate/test/test_wall_clock_step.py

@@ -0,0 +1,62 @@
+"""Tests that watchdogs are not fooled by wall clock steps, such as an NTP sync at boot."""
+
+import time
+import unittest
+from unittest.mock import MagicMock, patch
+
+from frigate.stats.util import _monotonic_to_wall
+from frigate.video.ffmpeg import CameraWatchdog
+from frigate.watchdog import FrigateWatchdog
+
+
+class TestDetectorWatchdogClock(unittest.TestCase):
+    def test_inference_that_just_started_is_not_stuck(self):
+        detector = MagicMock()
+        detector.detection_start.value = time.monotonic()
+        detector.detect_process.is_alive.return_value = True
+        watchdog = FrigateWatchdog({"ov": detector}, MagicMock())
+
+        with patch("frigate.watchdog.raise_notice") as raise_notice:
+            watchdog._check_detectors()
+
+        detector.start_or_restart.assert_not_called()
+        raise_notice.assert_not_called()
+
+
+class TestDetectionStartStats(unittest.TestCase):
+    def test_idle_detector_reports_zero(self):
+        self.assertEqual(_monotonic_to_wall(0.0), 0.0)
+
+    def test_running_inference_reports_wall_clock_start(self):
+        start = _monotonic_to_wall(time.monotonic() - 5)
+        self.assertAlmostEqual(start, time.time() - 5, delta=0.5)
+
+
+class TestCameraNoFramesCheck(unittest.TestCase):
+    def setUp(self):
+        self.watchdog = CameraWatchdog.__new__(CameraWatchdog)
+        self.watchdog.capture_thread = MagicMock()
+        self.watchdog.capture_thread.current_frame.value = 1000.0
+        self.watchdog.last_frame_value = 0.0
+        self.watchdog.last_frame_change = 0.0
+
+    def test_new_frame_read_is_not_stalled(self):
+        self.assertFalse(self.watchdog._no_frames_received(50.0))
+
+    def test_wall_clock_step_is_not_stalled(self):
+        self.watchdog._no_frames_received(50.0)
+
+        # the capture thread stamps frames with wall clock time, which just
+        # jumped 44 minutes forward
+        self.watchdog.capture_thread.current_frame.value = 1000.0 + 2640
+        self.assertFalse(self.watchdog._no_frames_received(51.0))
+
+    def test_unchanged_frame_for_timeout_is_stalled(self):
+        self.watchdog._no_frames_received(50.0)
+
+        self.assertFalse(self.watchdog._no_frames_received(70.0))
+        self.assertTrue(self.watchdog._no_frames_received(70.5))
+
+
+if __name__ == "__main__":
+    unittest.main()

frigate/test/test_watchdog_notices.py

@@ -18,10 +18,10 @@ class TestDetectorStuckNotice(unittest.TestCase):
         watchdog = FrigateWatchdog({"ov": detector}, MagicMock())
 
         with (
-            patch("frigate.watchdog.datetime") as mock_datetime,
+            patch("frigate.watchdog.time") as mock_time,
             patch("frigate.watchdog.raise_notice") as raise_notice,
         ):
-            mock_datetime.datetime.now.return_value.timestamp.return_value = 100.0
+            mock_time.monotonic.return_value = 100.0
             watchdog._check_detectors()
 
         detector.start_or_restart.assert_called_once()

frigate/util/file.py

@@ -429,8 +429,8 @@ class FileLock:
         try:
             self._fd = os.open(self.lock_path, os.O_CREAT | os.O_RDWR)
 
-            start_time = time.time()
-            while time.time() - start_time < timeout:
+            start_time = time.monotonic()
+            while time.monotonic() - start_time < timeout:
                 try:
                     fcntl.flock(self._fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
                     self._acquired = True
@@ -438,7 +438,7 @@ class FileLock:
                     return True
                 except OSError:
                     # Lock is held by another process
-                    if time.time() - start_time >= timeout:
+                    if time.monotonic() - start_time >= timeout:
                         logger.warning(f"Timeout waiting for lock: {self.lock_path}")
                         os.close(self._fd)
                         self._fd = None

frigate/util/rknn_converter.py

@@ -292,10 +292,10 @@ def wait_for_conversion_completion(
     Returns:
         True if RKNN model appears, False if timeout
     """
-    start_time = time.time()
+    start_time = time.monotonic()
     lock = FileLock(lock_file_path, stale_timeout=600)
 
-    while time.time() - start_time < timeout:
+    while time.monotonic() - start_time < timeout:
         # Check if RKNN model appeared
         if rknn_path.exists():
             logger.info(f"RKNN model appeared: {rknn_path}")

frigate/video/ffmpeg.py

@@ -171,6 +171,10 @@ class CameraWatchdog(threading.Thread):
         self.stop_event = stop_event
         self.sleeptime = self.config.ffmpeg.retry_interval
         self.reconnect_timestamps = deque()
+        # current_frame is wall clock because it doubles as the frame time, so
+        # track the monotonic time it last changed for the no frames check
+        self.last_frame_value = 0.0
+        self.last_frame_change = 0.0
         self.stalls = stalls
         self.reconnects = reconnects
         self.detection_frame = detection_frame
@@ -293,6 +297,17 @@ class CameraWatchdog(threading.Thread):
 
         return None
 
+    def _no_frames_received(self, now: float, timeout: float = 20) -> bool:
+        """Return True if the capture thread hasn't started a new frame read in timeout seconds."""
+        frame_value = self.capture_thread.current_frame.value
+
+        if frame_value != self.last_frame_value:
+            self.last_frame_value = frame_value
+            self.last_frame_change = now
+            return False
+
+        return now - self.last_frame_change > timeout
+
     def _recorded_streams(self, roles: list[Any]) -> list[str]:
         """Record stream types the given roles cover that are currently recording."""
         return [
@@ -360,7 +375,7 @@ class CameraWatchdog(threading.Thread):
                 self.record_enable_time = datetime.now().astimezone(UTC)
 
         time.sleep(self.sleeptime)
-        last_restart_time = datetime.now().timestamp()
+        last_restart_time = time.monotonic()
 
         # 1 second watchdog loop
         while not self.stop_event.wait(1):
@@ -376,7 +391,7 @@ class CameraWatchdog(threading.Thread):
                 self.start_all_ffmpeg()
                 self._reset_segment_times()
                 self.record_enable_time = datetime.now().astimezone(UTC)
-                last_restart_time = datetime.now().timestamp()
+                last_restart_time = time.monotonic()
                 continue
 
             enabled = self.config.enabled
@@ -394,7 +409,7 @@ class CameraWatchdog(threading.Thread):
                     self.record_enable_time = None
 
                     # update camera status
-                    now = datetime.now().timestamp()
+                    now = time.monotonic()
                     self.detect_status.send("disabled", now)
                     self.record_status[STREAM_TYPE_MAIN].send("disabled", now)
                     # cameras without a sub stream never get a record_sub topic
@@ -413,7 +428,7 @@ class CameraWatchdog(threading.Thread):
                     self.start_all_ffmpeg()
                     self._reset_segment_times()
                     self.record_enable_time = datetime.now().astimezone(UTC)
-                    last_restart_time = datetime.now().timestamp()
+                    last_restart_time = time.monotonic()
                 self.was_record_enabled_in_config = record_enabled_in_config
                 continue
 
@@ -429,7 +444,7 @@ class CameraWatchdog(threading.Thread):
                     self.start_all_ffmpeg()
                     self._reset_segment_times()
                     self.record_enable_time = datetime.now().astimezone(UTC)
-                    last_restart_time = datetime.now().timestamp()
+                    last_restart_time = time.monotonic()
                 self.was_record_sub_enabled = record_sub_enabled
                 continue
 
@@ -465,7 +480,8 @@ class CameraWatchdog(threading.Thread):
                             segment_time if segment_time is not None else 0
                         )
 
-            now = datetime.now().timestamp()
+            # monotonic so a wall clock step can't trigger or block restarts
+            now = time.monotonic()
 
             # Check if enough time has passed to allow ffmpeg restart (backoff pacing)
             time_since_last_restart = now - last_restart_time
@@ -495,7 +511,7 @@ class CameraWatchdog(threading.Thread):
                     )
                     self.reset_capture_thread()
                     last_restart_time = now
-            elif now - self.capture_thread.current_frame.value > 20:
+            elif self._no_frames_received(now):
                 self.detect_status.send("offline", now)
                 self.camera_fps.value = 0
                 if can_restart:
@@ -589,7 +605,8 @@ class CameraWatchdog(threading.Thread):
                     self.reset_capture_thread()
                     last_restart_time = now
 
-            # Prune expired reconnect timestamps
+            # Prune expired reconnect timestamps. wall clock from here on,
+            # since processed_ts below is a frame time
             now = datetime.now().timestamp()
             while (
                 self.reconnect_timestamps and self.reconnect_timestamps[0] < now - 3600

frigate/watchdog.py

@@ -1,4 +1,3 @@
-import datetime
 import logging
 import threading
 import time
@@ -86,7 +85,7 @@ class FrigateWatchdog(threading.Thread):
             exitcode,
         )
 
-        now = datetime.datetime.now().timestamp()
+        now = time.monotonic()
 
         if entry.is_restarting_too_fast(now):
             logger.error(
@@ -113,7 +112,8 @@ class FrigateWatchdog(threading.Thread):
             logger.exception("Failed to restart %s", entry.name)
 
     def _check_detectors(self) -> None:
-        now = datetime.datetime.now().timestamp()
+        # detection_start is set from the monotonic clock
+        now = time.monotonic()
 
         for name, detector in self.detectors.items():
             detection_start = detector.detection_start.value