From 2b4397bcba108b66241e4b2197c6bc7fb2d6bc3c Mon Sep 17 00:00:00 2001 From: Nicolas Mowen Date: Mon, 2 Nov 2026 15:33:55 -0700 Subject: [PATCH] Convert sensitive wall clock usages to monotonic (#24864) --- frigate/comms/webpush.py | 15 +++-- .../real_time/custom_classification.py | 17 ++--- frigate/jobs/vlm_watch.py | 13 ++-- frigate/object_detection/base.py | 6 +- frigate/output/birdseye.py | 9 +-- frigate/stats/util.py | 10 ++- frigate/test/test_wall_clock_step.py | 62 +++++++++++++++++++ frigate/test/test_watchdog_notices.py | 4 +- frigate/util/file.py | 6 +- frigate/util/rknn_converter.py | 4 +- frigate/video/ffmpeg.py | 33 +++++++--- frigate/watchdog.py | 6 +- 12 files changed, 140 insertions(+), 45 deletions(-) create mode 100644 frigate/test/test_wall_clock_step.py diff --git a/frigate/comms/webpush.py b/frigate/comms/webpush.py index cdac3587b7..5235a0b72c 100644 --- a/frigate/comms/webpush.py +++ b/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 diff --git a/frigate/data_processing/real_time/custom_classification.py b/frigate/data_processing/real_time/custom_classification.py index 4083f2a160..94b7ebeb68 100644 --- a/frigate/data_processing/real_time/custom_classification.py +++ b/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)" diff --git a/frigate/jobs/vlm_watch.py b/frigate/jobs/vlm_watch.py index 1e6a542c21..a50c940c97 100644 --- a/frigate/jobs/vlm_watch.py +++ b/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 diff --git a/frigate/object_detection/base.py b/frigate/object_detection/base.py index 49cfc9b3d0..de2e8eeaee 100644 --- a/frigate/object_detection/base.py +++ b/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 diff --git a/frigate/output/birdseye.py b/frigate/output/birdseye.py index 44799db0f5..fe46626afb 100644 --- a/frigate/output/birdseye.py +++ b/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 diff --git a/frigate/stats/util.py b/frigate/stats/util.py index 6d26d16a62..d610e2b25d 100644 --- a/frigate/stats/util.py +++ b/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, diff --git a/frigate/test/test_wall_clock_step.py b/frigate/test/test_wall_clock_step.py new file mode 100644 index 0000000000..c2253d25be --- /dev/null +++ b/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() diff --git a/frigate/test/test_watchdog_notices.py b/frigate/test/test_watchdog_notices.py index f1cc19c830..2d49846cd2 100644 --- a/frigate/test/test_watchdog_notices.py +++ b/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() diff --git a/frigate/util/file.py b/frigate/util/file.py index 9cb83127b1..68d71b2746 100644 --- a/frigate/util/file.py +++ b/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 diff --git a/frigate/util/rknn_converter.py b/frigate/util/rknn_converter.py index a23c5712d1..ee4ab829dc 100644 --- a/frigate/util/rknn_converter.py +++ b/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}") diff --git a/frigate/video/ffmpeg.py b/frigate/video/ffmpeg.py index 9cf5dcd4e1..65a3ee5d3a 100644 --- a/frigate/video/ffmpeg.py +++ b/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 diff --git a/frigate/watchdog.py b/frigate/watchdog.py index 9dfa4f5581..9e945d4c9a 100644 --- a/frigate/watchdog.py +++ b/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