diff --git a/frigate/jobs/motion_search.py b/frigate/jobs/motion_search.py index 234f27e821..75fc86aff5 100644 --- a/frigate/jobs/motion_search.py +++ b/frigate/jobs/motion_search.py @@ -20,6 +20,7 @@ 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 ( @@ -53,6 +54,9 @@ 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 @@ -585,7 +589,7 @@ class MotionSearchRunner(threading.Thread): # Decide keyframe vs fixed-cadence sampling once from the first run's GOP # (keyframe structure is a per-camera constant). - first_pts = probe_vod_keyframe_pts(ffprobe_path, first_url) + first_pts = probe_vod_keyframe_pts(ffprobe_path, first_url, self._should_stop) self.use_keyframe = keyframe_sampling_eligible(first_pts) logger.debug( @@ -647,38 +651,47 @@ class MotionSearchRunner(threading.Thread): time_map = build_segment_time_map(run) if self.use_keyframe: - kf_pts = probe_vod_keyframe_pts(self.ffprobe_path, vod_url) - frames = list( - iter_vod_frames( - self.ffmpeg_path, - vod_url, - self.scaled[0], - self.scaled[1], - self.channels, - self.decode_args, - self.crop, - self.scaled, - True, - self._should_stop, - skip_nonkey=True, - fps_rate=None, + 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 + # pair frames with timestamps when the probe saw the whole stream. + if keyframes_cover_stream( + kf_pts, _stream_duration(time_map), MAX_STREAM_HOLE_SECONDS + ): + frames = list( + iter_vod_frames( + self.ffmpeg_path, + vod_url, + self.scaled[0], + self.scaled[1], + self.channels, + self.decode_args, + self.crop, + self.scaled, + True, + self._should_stop, + skip_nonkey=True, + fps_rate=None, + ) ) - ) - if kf_pts and len(frames) == len(kf_pts): - abs_times = [stream_time_to_absolute(time_map, p) for p in kf_pts] - indexed = list(enumerate(frames)) + if len(frames) == len(kf_pts): + abs_times = [stream_time_to_absolute(time_map, p) for p in kf_pts] + indexed = list(enumerate(frames)) - def _ts_kf(i: int) -> float: - return abs_times[i] + def _ts_kf(i: int) -> float: + return abs_times[i] - results = self._detect_with_progress(indexed, _ts_kf) - return results, len(frames) + results = self._detect_with_progress(indexed, _ts_kf) + return results, len(frames) - logger.debug( - "Keyframe count mismatch (%d decoded vs %d probed), using cadence", - len(frames), - len(kf_pts), - ) + logger.debug( + "Keyframe count mismatch (%d decoded vs %d probed), using cadence", + len(frames), + len(kf_pts), + ) + else: + logger.debug("Keyframe probe has a gap, using cadence") return self._process_run_cadence(vod_url, time_map) @@ -702,6 +715,15 @@ 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: @@ -864,6 +886,11 @@ 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: @@ -919,7 +946,11 @@ def start_motion_search_job( # A cancelled job keeps its thread until the runner sets end_time, so # count threads rather than statuses. active = [j for j, _ in _motion_search_jobs.values() if _is_active(j, now)] - if username is not None and any(j.username == username for j in active): + # 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 + ): raise MotionSearchUserBusyError() if len(active) >= MAX_ACTIVE_JOBS: raise MotionSearchCapacityError() diff --git a/frigate/jobs/motion_search_batch.py b/frigate/jobs/motion_search_batch.py index f916da5e0a..66799b3d93 100644 --- a/frigate/jobs/motion_search_batch.py +++ b/frigate/jobs/motion_search_batch.py @@ -73,3 +73,14 @@ 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 diff --git a/frigate/jobs/motion_search_decode.py b/frigate/jobs/motion_search_decode.py index 376cdcb935..ed7d1fd445 100644 --- a/frigate/jobs/motion_search_decode.py +++ b/frigate/jobs/motion_search_decode.py @@ -10,6 +10,7 @@ import json import logging import subprocess as sp import tempfile +import time from collections.abc import Callable, Generator from typing import IO @@ -107,6 +108,9 @@ def keyframe_sampling_eligible( # rw_timeout (microseconds) ends a stalled VOD read instead of blocking forever. +# 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", @@ -149,7 +153,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", "error"] + cmd = [ffmpeg_path, "-hide_banner", "-loglevel", "warning"] if skip_nonkey: cmd += ["-skip_frame", "nokey"] cmd += [*decode_args, *VOD_PROTOCOL_ARGS, "-i", vod_url, "-an"] @@ -211,8 +215,13 @@ 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( @@ -270,11 +279,16 @@ def iter_vod_frames( ) -def probe_vod_keyframe_pts(ffprobe_path: str, vod_url: str) -> list[float]: +KEYFRAME_PROBE_TIMEOUT = 120.0 + + +def probe_vod_keyframe_pts( + ffprobe_path: str, vod_url: str, should_stop: Callable[[], bool] +) -> list[float]: """Return keyframe presentation timestamps (VOD stream time) in order. Reads packet flags via ffprobe over the VOD URL (no decode). Returns [] on - any failure so the caller can fall back. + any failure, timeout, or stop request so the caller can fall back. """ cmd = [ ffprobe_path, @@ -292,16 +306,27 @@ def probe_vod_keyframe_pts(ffprobe_path: str, vod_url: str) -> list[float]: "json", ] try: - completed = sp.run(cmd, capture_output=True, text=True, timeout=120) - except (OSError, sp.SubprocessError): + proc = sp.Popen(cmd, stdout=sp.PIPE, stderr=sp.DEVNULL, text=True) + except OSError: logger.warning("ffprobe failed for VOD keyframe probe") return [] - if completed.returncode != 0 or not completed.stdout: + deadline = time.monotonic() + KEYFRAME_PROBE_TIMEOUT + while True: + try: + stdout, _ = proc.communicate(timeout=0.5) + break + except sp.TimeoutExpired: + if should_stop() or time.monotonic() > deadline: + proc.kill() + proc.communicate() + return [] + + if proc.returncode != 0 or not stdout: return [] try: - packets = json.loads(completed.stdout).get("packets", []) + packets = json.loads(stdout).get("packets", []) except json.JSONDecodeError: return [] @@ -330,6 +355,7 @@ def probe_video_dimensions( ffprobe_path, "-v", "error", + *VOD_PROTOCOL_ARGS, "-select_streams", "v:0", "-show_entries", diff --git a/frigate/test/http_api/test_http_motion_search.py b/frigate/test/http_api/test_http_motion_search.py index aeaf19ce6b..003a2d5801 100644 --- a/frigate/test/http_api/test_http_motion_search.py +++ b/frigate/test/http_api/test_http_motion_search.py @@ -44,16 +44,21 @@ class TestHttpMotionSearch(BaseTestHttp): self.assertEqual(self.start(client, "alice").status_code, 200) self.assertIsNotNone(motion_search.get_motion_search_job(job_id)) - def test_cancelled_job_holds_slot_until_grace_expires(self): + def test_cancelled_job_frees_user_but_holds_global_slot_until_grace(self): with AuthTestClient(self.app) as client: job_id = self.start(client, "alice").json()["job_id"] + with patch("frigate.jobs.motion_search.InterProcessRequestor"): + self.assertTrue(motion_search.cancel_motion_search_job(job_id)) job = motion_search.get_motion_search_job(job_id) - job.status = "cancelled" - job.cancelled_at = datetime.now().timestamp() - self.assertEqual(self.start(client, "alice").status_code, 409) + self.assertIsNotNone(job.cancelled_at) + self.assertIsNone(job.end_time) + + self.assertEqual(self.start(client, "alice").status_code, 200) + self.assertEqual(self.start(client, "bob").status_code, 200) + self.assertEqual(self.start(client, "carol").status_code, 429) job.cancelled_at -= motion_search.CANCEL_GRACE_SECONDS + 1 - self.assertEqual(self.start(client, "alice").status_code, 200) + self.assertEqual(self.start(client, "carol").status_code, 200) def test_start_failure_releases_slot(self): self.runner.return_value.start.side_effect = RuntimeError("boom") @@ -78,6 +83,14 @@ class TestHttpMotionSearch(BaseTestHttp): self.assertEqual(self.start(client, "viewer").status_code, 200) self.assertEqual(self.start(client, "viewer").status_code, 200) + def test_proxy_user_header_keeps_per_user_limit_without_auth(self): + self.minimal_config["auth"] = {"enabled": False} + self.minimal_config["proxy"] = {"header_map": {"user": "x-forwarded-user"}} + app = self.create_app() + with AuthTestClient(app) as client: + self.assertEqual(self.start(client, "alice").status_code, 200) + self.assertEqual(self.start(client, "alice").status_code, 409) + def test_global_cap_enforced(self): with AuthTestClient(self.app) as client: for i in range(motion_search.MAX_ACTIVE_JOBS): @@ -109,6 +122,8 @@ class TestHttpMotionSearch(BaseTestHttp): self.assertIsNotNone(motion_search.get_motion_search_job(recent_id)) self.assertIsNotNone(motion_search.get_motion_search_job(running_id)) - def test_username_not_broadcast(self): - job = motion_search.MotionSearchJob(username="alice") - self.assertNotIn("username", job.to_dict()) + def test_username_and_cancel_time_not_broadcast(self): + job = motion_search.MotionSearchJob(username="alice", cancelled_at=1.0) + payload = job.to_dict() + self.assertNotIn("username", payload) + self.assertNotIn("cancelled_at", payload) diff --git a/frigate/test/test_motion_search_batch.py b/frigate/test/test_motion_search_batch.py index 0e7fcdfd8a..9c619034d4 100644 --- a/frigate/test/test_motion_search_batch.py +++ b/frigate/test/test_motion_search_batch.py @@ -6,6 +6,7 @@ from dataclasses import dataclass from frigate.jobs.motion_search_batch import ( build_segment_time_map, coalesce_runs, + keyframes_cover_stream, stream_time_to_absolute, ) @@ -56,3 +57,17 @@ 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)) diff --git a/frigate/test/test_motion_search_decode.py b/frigate/test/test_motion_search_decode.py index df2ff5cd13..113bbd18a2 100644 --- a/frigate/test/test_motion_search_decode.py +++ b/frigate/test/test_motion_search_decode.py @@ -1,5 +1,7 @@ """Tests for the motion search hardware-accelerated decode helpers.""" +import io +import subprocess as sp import unittest from types import SimpleNamespace from unittest import mock @@ -7,6 +9,7 @@ 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, @@ -140,7 +143,50 @@ 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) + proc.communicate.return_value = (stdout, None) + return proc + def test_parses_keyframe_packets(self): sample = ( '{"packets":[' @@ -148,19 +194,39 @@ class TestProbeVodKeyframePts(unittest.TestCase): '{"pts_time":"1.000000","flags":"___"},' '{"pts_time":"2.000000","flags":"K__"}]}' ) - completed = mock.Mock(stdout=sample, returncode=0) with mock.patch( - "frigate.jobs.motion_search_decode.sp.run", return_value=completed - ): - pts = probe_vod_keyframe_pts("ffprobe", "http://x/index.m3u8") + "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( - "frigate.jobs.motion_search_decode.sp.run", + "frigate.jobs.motion_search_decode.sp.Popen", side_effect=OSError("boom"), ): - self.assertEqual(probe_vod_keyframe_pts("ffprobe", "http://x"), []) + self.assertEqual( + probe_vod_keyframe_pts("ffprobe", "http://x", lambda: False), [] + ) + + def test_stop_request_kills_probe(self): + proc = self._proc() + proc.communicate.side_effect = [ + sp.TimeoutExpired("ffprobe", 0.5), + ("", None), + ] + with mock.patch( + "frigate.jobs.motion_search_decode.sp.Popen", return_value=proc + ): + self.assertEqual( + probe_vod_keyframe_pts("ffprobe", "http://x", lambda: True), [] + ) + proc.kill.assert_called_once() class TestProbeVideoDimensions(unittest.TestCase): @@ -171,9 +237,10 @@ 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) diff --git a/frigate/test/test_motion_search_runner.py b/frigate/test/test_motion_search_runner.py new file mode 100644 index 0000000000..08bc0d578f --- /dev/null +++ b/frigate/test/test_motion_search_runner.py @@ -0,0 +1,85 @@ +"""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)