drop motion search stall detection and range cap

frigate/api/motion_search.py

@@ -23,9 +23,6 @@ logger = logging.getLogger(__name__)
 
 router = APIRouter(tags=[Tags.motion_search])
 
-# Covers the UI's day windows (25 h on a DST fall-back day) with room to spare.
-MAX_SEARCH_RANGE_SECONDS = 48 * 3600
-
 
 class MotionSearchRequest(BaseModel):
     """Request body for motion search."""
@@ -158,15 +155,6 @@ async def start_motion_search(
             status_code=400,
         )
 
-    if body.end_time - body.start_time > MAX_SEARCH_RANGE_SECONDS:
-        return JSONResponse(
-            content={
-                "success": False,
-                "message": f"Time range must be {MAX_SEARCH_RANGE_SECONDS // 3600} hours or less",
-            },
-            status_code=400,
-        )
-
     try:
         job_id = start_motion_search_job(
             config=config,

frigate/jobs/motion_search.py

@@ -20,7 +20,6 @@ from frigate.jobs.job import Job
 from frigate.jobs.motion_search_batch import (
     build_segment_time_map,
     coalesce_runs,
-    keyframes_cover_stream,
     stream_time_to_absolute,
 )
 from frigate.jobs.motion_search_decode import (
@@ -54,9 +53,6 @@ FALLBACK_SAMPLE_FPS = 2.0
 MAX_ACTIVE_JOBS = 3
 # Finished jobs stay pollable for this long before eviction (seconds).
 FINISHED_JOB_TTL = 600.0
-# Missing stream time beyond this marks a run as truncated, e.g. a segment the
-# HLS demuxer skipped after a read timeout.
-MAX_STREAM_HOLE_SECONDS = 5.0
 # A cancelled job stops holding a slot after this long, even if its thread is wedged.
 CANCEL_GRACE_SECONDS = 60.0
 
@@ -654,44 +650,37 @@ class MotionSearchRunner(threading.Thread):
             kf_pts = probe_vod_keyframe_pts(
                 self.ffprobe_path, vod_url, self._should_stop
             )
-            # Probe and decode fetch the stream separately, so equal counts only
+            frames = list(
-            # pair frames with timestamps when the probe saw the whole stream.
+                iter_vod_frames(
-            if keyframes_cover_stream(
+                    self.ffmpeg_path,
-                kf_pts, _stream_duration(time_map), MAX_STREAM_HOLE_SECONDS
+                    vod_url,
-            ):
+                    self.scaled[0],
-                frames = list(
+                    self.scaled[1],
-                    iter_vod_frames(
+                    self.channels,
-                        self.ffmpeg_path,
+                    self.decode_args,
-                        vod_url,
+                    self.crop,
-                        self.scaled[0],
+                    self.scaled,
-                        self.scaled[1],
+                    True,
-                        self.channels,
+                    self._should_stop,
-                        self.decode_args,
+                    skip_nonkey=True,
-                        self.crop,
+                    fps_rate=None,
-                        self.scaled,
-                        True,
-                        self._should_stop,
-                        skip_nonkey=True,
-                        fps_rate=None,
-                    )
                 )
-                if len(frames) == len(kf_pts):
+            )
-                    abs_times = [stream_time_to_absolute(time_map, p) for p in kf_pts]
+            if kf_pts and len(frames) == len(kf_pts):
-                    indexed = list(enumerate(frames))
+                abs_times = [stream_time_to_absolute(time_map, p) for p in kf_pts]
+                indexed = list(enumerate(frames))
 
-                    def _ts_kf(i: int) -> float:
+                def _ts_kf(i: int) -> float:
-                        return abs_times[i]
+                    return abs_times[i]
 
-                    results = self._detect_with_progress(indexed, _ts_kf)
+                results = self._detect_with_progress(indexed, _ts_kf)
-                    return results, len(frames)
+                return results, len(frames)
 
-                logger.debug(
+            logger.debug(
-                    "Keyframe count mismatch (%d decoded vs %d probed), using cadence",
+                "Keyframe count mismatch (%d decoded vs %d probed), using cadence",
-                    len(frames),
+                len(frames),
-                    len(kf_pts),
+                len(kf_pts),
-                )
+            )
-            else:
-                logger.debug("Keyframe probe has a gap, using cadence")
 
         return self._process_run_cadence(vod_url, time_map)
 
@@ -715,15 +704,6 @@ class MotionSearchRunner(threading.Thread):
                 fps_rate=self.fps_rate,
             )
         )
-        expected = _stream_duration(time_map) * self.fps_rate
-        if (
-            not self._should_stop()
-            and expected - len(frames) > MAX_STREAM_HOLE_SECONDS * self.fps_rate
-        ):
-            raise RuntimeError(
-                f"VOD decode returned {len(frames)} of ~{expected:.0f} frames"
-            )
-
         indexed = list(enumerate(frames))
 
         def _ts_fps(i: int) -> float:
@@ -886,11 +866,6 @@ def _evict_finished_jobs(now: float) -> None:
         del _motion_search_jobs[job_id]
 
 
-def _stream_duration(time_map: list[tuple[float, float, float]]) -> float:
-    offset, _, duration = time_map[-1]
-    return offset + duration
-
-
 def _is_active(job: MotionSearchJob, now: float) -> bool:
     """True while the job's thread may still be running and holding a slot."""
     if job.end_time is not None:
@@ -943,10 +918,9 @@ def start_motion_search_job(
         now = datetime.now().timestamp()
         _evict_finished_jobs(now)
 
-        # A cancelled job keeps its thread until the runner sets end_time, so
+        # A cancelled job keeps its thread until the runner sets end_time, so it
-        # count threads rather than statuses.
+        # still counts toward the global cap. Its user may start a new search.
         active = [j for j, _ in _motion_search_jobs.values() if _is_active(j, now)]
-        # The UI forgets a cancelled job at once, so it only counts globally.
         if username is not None and any(
             j.username == username and j.status != JobStatusTypesEnum.cancelled
             for j in active

frigate/jobs/motion_search_batch.py

@@ -73,14 +73,3 @@ def stream_time_to_absolute(
         idx = 0
     stream_offset, abs_start, _duration = time_map[idx]
     return abs_start + (stream_time - stream_offset)
-
-
-def keyframes_cover_stream(pts: list[float], duration: float, max_hole: float) -> bool:
-    """True if sorted keyframe ``pts`` span 0..``duration`` with no gap over ``max_hole``.
-
-    A segment skipped or cut short by a read timeout leaves such a gap.
-    """
-    if not pts:
-        return False
-    edges = [0.0, *pts, duration]
-    return max(b - a for a, b in zip(edges, edges[1:])) <= max_hole

frigate/jobs/motion_search_decode.py

@@ -107,16 +107,7 @@ def keyframe_sampling_eligible(
     return max(gaps) <= max_gap
 
 
-# rw_timeout (microseconds) ends a stalled VOD read instead of blocking forever.
+VOD_PROTOCOL_ARGS = ["-protocol_whitelist", "pipe,file,http,tcp"]
-# The HLS demuxer logs this at warning level when a segment read fails or times out.
-HLS_SEGMENT_SKIPPED = b"failed too many times, skipping"
-
-VOD_PROTOCOL_ARGS = [
-    "-protocol_whitelist",
-    "pipe,file,http,tcp",
-    "-rw_timeout",
-    "30000000",
-]
 
 
 def build_vod_decode_command(
@@ -153,7 +144,7 @@ def build_vod_decode_command(
         filters.append(f"scale={sw}:{sh}")
 
     pix_fmt = "gray" if gray else "bgr24"
-    cmd = [ffmpeg_path, "-hide_banner", "-loglevel", "warning"]
+    cmd = [ffmpeg_path, "-hide_banner", "-loglevel", "error"]
     if skip_nonkey:
         cmd += ["-skip_frame", "nokey"]
     cmd += [*decode_args, *VOD_PROTOCOL_ARGS, "-i", vod_url, "-an"]
@@ -215,13 +206,8 @@ def _run_vod_decode(
             yield frame
     finally:
         terminate_ffmpeg_stream(proc)
-        stderr_file.seek(0)
-        segment_skipped = HLS_SEGMENT_SKIPPED in stderr_file.read()
         stderr_file.close()
 
-    if segment_skipped and not should_stop():
-        raise RuntimeError("VOD decode skipped a segment that failed to load")
-
     if count == 0 and software_retry and not should_stop():
         logger.warning("Hardware VOD decode produced no frames, retrying in software")
         yield from _run_vod_decode(
@@ -355,7 +341,6 @@ def probe_video_dimensions(
         ffprobe_path,
         "-v",
         "error",
-        *VOD_PROTOCOL_ARGS,
         "-select_streams",
         "v:0",
         "-show_entries",

frigate/test/http_api/test_http_motion_search.py

@@ -97,14 +97,6 @@ class TestHttpMotionSearch(BaseTestHttp):
                 self.assertEqual(self.start(client, f"user{i}").status_code, 200)
             self.assertEqual(self.start(client, "late").status_code, 429)
 
-    def test_over_long_range_rejected(self):
-        body = {**BODY, "end_time": BODY["start_time"] + 48 * 3600 + 1}
-        with AuthTestClient(self.app) as client:
-            resp = self.start(client, "alice", body)
-
-        self.assertEqual(resp.status_code, 400)
-        self.assertEqual(motion_search._motion_search_jobs, {})
-
     def test_finished_jobs_evicted_after_ttl(self):
         with AuthTestClient(self.app) as client:
             old_id = self.start(client, "alice").json()["job_id"]

frigate/test/test_motion_search_batch.py

@@ -6,7 +6,6 @@ from dataclasses import dataclass
 from frigate.jobs.motion_search_batch import (
     build_segment_time_map,
     coalesce_runs,
-    keyframes_cover_stream,
     stream_time_to_absolute,
 )
 
@@ -57,17 +56,3 @@ class TestTimestampMapping(unittest.TestCase):
         run = [_Seg("a", 1000.0, 1010.0)]
         time_map = build_segment_time_map(run)
         self.assertAlmostEqual(stream_time_to_absolute(time_map, 9.9), 1009.9)
-
-
-class TestKeyframesCoverStream(unittest.TestCase):
-    def test_dense_keyframes_cover(self):
-        self.assertTrue(keyframes_cover_stream([0.5, 2.5, 4.5, 6.5, 8.5], 10.0, 5.0))
-
-    def test_gap_from_skipped_segment(self):
-        self.assertFalse(keyframes_cover_stream([0.5, 2.5, 12.5, 14.5], 20.0, 5.0))
-
-    def test_truncated_tail(self):
-        self.assertFalse(keyframes_cover_stream([0.5, 2.5, 4.5], 20.0, 5.0))
-
-    def test_empty(self):
-        self.assertFalse(keyframes_cover_stream([], 10.0, 5.0))

frigate/test/test_motion_search_decode.py

@@ -1,6 +1,5 @@
 """Tests for the motion search hardware-accelerated decode helpers."""
 
-import io
 import subprocess as sp
 import unittest
 from types import SimpleNamespace
@@ -9,7 +8,6 @@ from unittest import mock
 from frigate.jobs.motion_search_decode import (
     KEYFRAME_MAX_GAP_SECONDS,
     build_vod_decode_command,
-    iter_vod_frames,
     keyframe_sampling_eligible,
     probe_video_dimensions,
     probe_vod_keyframe_pts,
@@ -104,7 +102,6 @@ class TestVodDecodeCommand(unittest.TestCase):
         joined = " ".join(cmd)
         self.assertIn("-skip_frame nokey", joined)
         self.assertIn("-protocol_whitelist pipe,file,http,tcp", joined)
-        self.assertLess(cmd.index("-rw_timeout"), cmd.index("-i"))
         self.assertIn(f"-i {self.URL}", joined)
         self.assertIn("crop=100:80:10:20", joined)
         self.assertIn("scale=50:40", joined)
@@ -143,44 +140,6 @@ class TestVodDecodeCommand(unittest.TestCase):
         self.assertIn("format=nv12", joined)
 
 
-class TestVodDecodeSkippedSegment(unittest.TestCase):
-    def _decode(self, stderr_output: bytes) -> list:
-        def popen(cmd, stdout, stderr):
-            stderr.write(stderr_output)
-            return mock.Mock(stdout=io.BytesIO(b"\0" * 8))
-
-        with (
-            mock.patch("frigate.jobs.motion_search_decode.sp.Popen", side_effect=popen),
-            mock.patch("frigate.jobs.motion_search_decode.terminate_ffmpeg_stream"),
-        ):
-            return list(
-                iter_vod_frames(
-                    "ffmpeg",
-                    "http://x/index.m3u8",
-                    2,
-                    2,
-                    1,
-                    [],
-                    None,
-                    None,
-                    True,
-                    lambda: False,
-                    skip_nonkey=False,
-                    fps_rate=2.0,
-                )
-            )
-
-    def test_clean_decode_returns_frames(self):
-        self.assertEqual(len(self._decode(b"")), 2)
-
-    def test_skipped_segment_raises(self):
-        stderr = (
-            b"[hls @ 0x1] Segment 1 of playlist 0 failed too many times, skipping\n"
-        )
-        with self.assertRaises(RuntimeError):
-            self._decode(stderr)
-
-
 class TestProbeVodKeyframePts(unittest.TestCase):
     def _proc(self, stdout="", returncode=0):
         proc = mock.Mock(returncode=returncode)
@@ -197,13 +156,11 @@ class TestProbeVodKeyframePts(unittest.TestCase):
         with mock.patch(
             "frigate.jobs.motion_search_decode.sp.Popen",
             return_value=self._proc(sample),
-        ) as popen:
+        ):
             pts = probe_vod_keyframe_pts(
                 "ffprobe", "http://x/index.m3u8", lambda: False
             )
         self.assertEqual(pts, [0.0, 2.0])
-        cmd = popen.call_args.args[0]
-        self.assertLess(cmd.index("-rw_timeout"), cmd.index("-i"))
 
     def test_returns_empty_on_failure(self):
         with mock.patch(
@@ -237,10 +194,9 @@ class TestProbeVideoDimensions(unittest.TestCase):
         completed = mock.Mock(stdout=sample, returncode=0)
         with mock.patch(
             "frigate.jobs.motion_search_decode.sp.run", return_value=completed
-        ) as run:
+        ):
             dims = probe_video_dimensions("ffprobe", "/tmp/a.mp4")
         assert dims is not None
-        self.assertIn("-rw_timeout", run.call_args.args[0])
         width, height, fps = dims
         self.assertEqual((width, height), (1920, 1080))
         self.assertAlmostEqual(fps, 29.97, places=2)

frigate/test/test_motion_search_runner.py

@@ -1,85 +0,0 @@
-"""Tests for motion search run decoding and truncation checks."""
-
-import threading
-import unittest
-from types import SimpleNamespace
-from unittest import mock
-
-import numpy as np
-
-from frigate.jobs.motion_search import MotionSearchJob, MotionSearchRunner
-
-# Two contiguous 10 s segments, so the run's stream is 20 s long.
-RUN = [
-    SimpleNamespace(start_time=1000.0, end_time=1010.0),
-    SimpleNamespace(start_time=1010.0, end_time=1020.0),
-]
-FRAME = np.zeros((2, 2), dtype=np.uint8)
-
-
-class TestProcessRun(unittest.TestCase):
-    def setUp(self):
-        requestor = mock.patch("frigate.jobs.motion_search.InterProcessRequestor")
-        requestor.start()
-        self.addCleanup(requestor.stop)
-
-        self.runner = MotionSearchRunner(
-            MotionSearchJob(camera="cam"), mock.Mock(), threading.Event()
-        )
-        self.runner.fps_rate = 2.0
-        self.runner._detect_with_progress = mock.Mock(return_value=[])
-
-    def _process(self, kf_pts, keyframe_frames, cadence_frames):
-        def frames(*args, skip_nonkey, fps_rate):
-            count = keyframe_frames if skip_nonkey else cadence_frames
-            return iter([FRAME] * count)
-
-        with (
-            mock.patch(
-                "frigate.jobs.motion_search.probe_vod_keyframe_pts",
-                return_value=kf_pts,
-            ),
-            mock.patch(
-                "frigate.jobs.motion_search.iter_vod_frames", side_effect=frames
-            ) as decode,
-        ):
-            result = self.runner._process_run(RUN)
-        return result, [c.kwargs["skip_nonkey"] for c in decode.call_args_list]
-
-    def test_complete_keyframe_probe_pairs_timestamps(self):
-        pts = [0.5, 2.5, 4.5, 6.5, 8.5, 10.5, 12.5, 14.5, 16.5, 18.5]
-        (_, frames), modes = self._process(pts, len(pts), 0)
-
-        self.assertEqual(frames, len(pts))
-        self.assertEqual(modes, [True])
-        ts_fn = self.runner._detect_with_progress.call_args.args[1]
-        self.assertAlmostEqual(ts_fn(5), 1010.5)
-
-    def test_keyframe_probe_gap_falls_back_to_cadence(self):
-        # Probe lost the second segment; an equal decode count must not be trusted.
-        pts = [0.5, 2.5, 4.5, 6.5, 8.5]
-        (_, frames), modes = self._process(pts, len(pts), 40)
-
-        self.assertEqual(modes, [False])
-        self.assertEqual(frames, 40)
-
-    def test_cadence_shortfall_raises(self):
-        self.runner.use_keyframe = False
-        with self.assertRaises(RuntimeError):
-            self._process([], 0, 20)
-
-    def test_zero_frames_raises(self):
-        self.runner.use_keyframe = False
-        with self.assertRaises(RuntimeError):
-            self._process([], 0, 0)
-
-    def test_small_cadence_drift_is_tolerated(self):
-        self.runner.use_keyframe = False
-        (_, frames), _ = self._process([], 0, 38)
-        self.assertEqual(frames, 38)
-
-    def test_shortfall_after_stop_does_not_raise(self):
-        self.runner.use_keyframe = False
-        self.runner.cancel_event.set()
-        (_, frames), _ = self._process([], 0, 3)
-        self.assertEqual(frames, 3)