diff --git a/frigate/api/motion_search.py b/frigate/api/motion_search.py index d8449bde98..15f692047e 100644 --- a/frigate/api/motion_search.py +++ b/frigate/api/motion_search.py @@ -9,7 +9,10 @@ from pydantic import BaseModel, Field from frigate.api.auth import require_camera_access from frigate.api.defs.tags import Tags +from frigate.config import FrigateConfig from frigate.jobs.motion_search import ( + MotionSearchCapacityError, + MotionSearchUserBusyError, cancel_motion_search_job, get_motion_search_job, start_motion_search_job, @@ -95,6 +98,20 @@ class MotionSearchStatusResponse(BaseModel): progress: float | None = None +def _per_user_identity(request: Request, config: FrigateConfig) -> str | None: + """Return the requesting user, or None when callers share one identity. + + Internal-port requests are all "anonymous", and with auth disabled and no + proxy user header every browser is "viewer". + """ + username = request.headers.get("remote-user") + if not username or username == "anonymous": + return None + if config.auth.enabled or config.proxy.header_map.user: + return username + return None + + @router.post( "/{camera_name}/search/motion", response_model=MotionSearchStartResponse, @@ -138,18 +155,35 @@ async def start_motion_search( status_code=400, ) - # Start the job using the jobs module - job_id = start_motion_search_job( - config=config, - camera_name=camera_name, - start_time=body.start_time, - end_time=body.end_time, - polygon_points=body.polygon_points, - threshold=body.threshold, - min_area=body.min_area, - parallel=body.parallel, - max_results=body.max_results, - ) + try: + job_id = start_motion_search_job( + config=config, + camera_name=camera_name, + start_time=body.start_time, + end_time=body.end_time, + polygon_points=body.polygon_points, + threshold=body.threshold, + min_area=body.min_area, + parallel=body.parallel, + max_results=body.max_results, + username=_per_user_identity(request, config), + ) + except MotionSearchUserBusyError: + return JSONResponse( + content={ + "success": False, + "message": "A motion search is already running for this user", + }, + status_code=409, + ) + except MotionSearchCapacityError: + return JSONResponse( + content={ + "success": False, + "message": "Too many motion searches are running", + }, + status_code=429, + ) return JSONResponse( content={ diff --git a/frigate/jobs/motion_search.py b/frigate/jobs/motion_search.py index da33f276b8..2d8617972b 100644 --- a/frigate/jobs/motion_search.py +++ b/frigate/jobs/motion_search.py @@ -8,7 +8,7 @@ from collections.abc import Callable, Generator, Iterable from concurrent.futures import Future, ThreadPoolExecutor, as_completed from dataclasses import asdict, dataclass, field from datetime import datetime -from typing import Any, cast +from typing import Any import cv2 import numpy as np @@ -17,10 +17,6 @@ from frigate.comms.inter_process import InterProcessRequestor from frigate.config import FrigateConfig from frigate.const import STREAM_TYPE_MAIN, UPDATE_JOB_STATE from frigate.jobs.job import Job -from frigate.jobs.manager import ( - get_job_by_id, - set_current_job, -) from frigate.jobs.motion_search_batch import ( build_segment_time_map, coalesce_runs, @@ -53,6 +49,12 @@ PROGRESS_BROADCAST_INTERVAL = 1.0 # Output frame rate for the fixed-cadence fallback used on long-GOP cameras # (where keyframe sampling is too sparse). Keyframe cameras ignore this. FALLBACK_SAMPLE_FPS = 2.0 +# Each parallel job runs up to 4 decode workers, so this bounds total ffmpeg load. +MAX_ACTIVE_JOBS = 3 +# Finished jobs stay pollable for this long before eviction (seconds). +FINISHED_JOB_TTL = 600.0 +# A cancelled job stops holding a slot after this long, even if its thread is wedged. +CANCEL_GRACE_SECONDS = 60.0 @dataclass @@ -98,6 +100,8 @@ class MotionSearchJob(Job): min_area: float = 5.0 parallel: bool = False max_results: int = 25 + username: str | None = None + cancelled_at: float | None = None # Track progress total_frames_processed: int = 0 @@ -112,6 +116,8 @@ class MotionSearchJob(Job): def to_dict(self) -> dict[str, Any]: """Convert to dictionary for WebSocket transmission.""" d = asdict(self) + d.pop("username") + d.pop("cancelled_at") if self.metrics: d["metrics"] = self.metrics.to_dict() return d @@ -579,7 +585,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( @@ -641,7 +647,9 @@ 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) + kf_pts = probe_vod_keyframe_pts( + self.ffprobe_path, vod_url, self._should_stop + ) frames = list( iter_vod_frames( self.ffmpeg_path, @@ -777,13 +785,15 @@ class MotionSearchRunner(threading.Thread): run_results, frames = completed.pop(next_idx) if self._merge_run(runs[next_idx], run_results, frames, state): self.internal_stop_event.set() - for pending in futures: - pending.cancel() break next_idx += 1 if self.internal_stop_event.is_set(): break + + # Queued runs would otherwise still execute when the pool exits. + for pending in futures: + pending.cancel() else: for run in runs: if self._should_stop(): @@ -833,11 +843,36 @@ class MotionSearchRunner(threading.Thread): return deduplicated -# Module-level state for managing per-camera jobs +class MotionSearchUserBusyError(Exception): + """The user already has an active motion search job.""" + + +class MotionSearchCapacityError(Exception): + """The global limit of active motion search jobs is reached.""" + + _motion_search_jobs: dict[str, tuple[MotionSearchJob, threading.Event]] = {} _jobs_lock = threading.Lock() +def _evict_finished_jobs(now: float) -> None: + """Drop jobs that finished more than FINISHED_JOB_TTL ago. Caller holds the lock.""" + expired = [ + job_id + for job_id, (job, _) in _motion_search_jobs.items() + if job.end_time is not None and now - job.end_time > FINISHED_JOB_TTL + ] + for job_id in expired: + del _motion_search_jobs[job_id] + + +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: + return False + return job.cancelled_at is None or now - job.cancelled_at < CANCEL_GRACE_SECONDS + + def stop_all_motion_search_jobs() -> None: """Cancel all running motion search jobs for clean shutdown.""" with _jobs_lock: @@ -857,10 +892,13 @@ def start_motion_search_job( min_area: float = 5.0, parallel: bool = False, max_results: int = 25, + username: str | None = None, ) -> str: """Start a new motion search job. - Returns the job ID. + Returns the job ID. Raises MotionSearchUserBusyError when ``username`` already + has an active job and MotionSearchCapacityError when MAX_ACTIVE_JOBS are active. + Pass ``username=None`` when callers share an identity to skip the per-user check. """ job = MotionSearchJob( camera=camera_name, @@ -871,17 +909,38 @@ def start_motion_search_job( min_area=min_area, parallel=parallel, max_results=max_results, + username=username, ) cancel_event = threading.Event() with _jobs_lock: + now = datetime.now().timestamp() + _evict_finished_jobs(now) + + # A cancelled job keeps its thread until the runner sets end_time, so it + # 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)] + 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() + _motion_search_jobs[job.id] = (job, cancel_event) - set_current_job(job) - - runner = MotionSearchRunner(job, config, cancel_event) - runner.start() + runner: MotionSearchRunner | None = None + try: + runner = MotionSearchRunner(job, config, cancel_event) + runner.start() + except Exception: + if runner is not None: + runner.requestor.stop() + with _jobs_lock: + del _motion_search_jobs[job.id] + raise logger.debug( "Started motion search job %s for camera %s: " @@ -904,10 +963,7 @@ def get_motion_search_job(job_id: str) -> MotionSearchJob | None: """Get a motion search job by ID.""" with _jobs_lock: job_entry = _motion_search_jobs.get(job_id) - if job_entry: - return job_entry[0] - # Check completed jobs via manager - return cast(MotionSearchJob | None, get_job_by_id("motion_search", job_id)) + return job_entry[0] if job_entry else None def cancel_motion_search_job(job_id: str) -> bool: @@ -928,6 +984,7 @@ def cancel_motion_search_job(job_id: str) -> bool: cancel_event.set() job.status = JobStatusTypesEnum.cancelled + job.cancelled_at = datetime.now().timestamp() job_payload = job.to_dict() logger.info("Cancelled motion search job %s", job_id) diff --git a/frigate/jobs/motion_search_decode.py b/frigate/jobs/motion_search_decode.py index 3b6e0cf4cc..ba56ae2d44 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 @@ -264,11 +265,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, @@ -286,16 +292,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 [] diff --git a/frigate/test/http_api/test_http_motion_search.py b/frigate/test/http_api/test_http_motion_search.py new file mode 100644 index 0000000000..35a0211679 --- /dev/null +++ b/frigate/test/http_api/test_http_motion_search.py @@ -0,0 +1,121 @@ +"""Tests for motion search job limits.""" + +from datetime import datetime +from unittest.mock import patch + +from frigate.jobs import motion_search +from frigate.models import Event, Recordings, ReviewSegment +from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp + +URL = "/front_door/search/motion" +BODY = { + "start_time": 1000, + "end_time": 2000, + "polygon_points": [[0, 0], [1, 0], [1, 1]], +} + + +class TestHttpMotionSearch(BaseTestHttp): + def setUp(self): + super().setUp([Event, Recordings, ReviewSegment]) + self.app = self.create_app() + + motion_search._motion_search_jobs.clear() + self.addCleanup(motion_search._motion_search_jobs.clear) + + runner_patch = patch("frigate.jobs.motion_search.MotionSearchRunner") + self.runner = runner_patch.start() + self.addCleanup(runner_patch.stop) + + def start(self, client: AuthTestClient, user: str, body: dict = BODY): + return client.post(URL, json=body, headers={"remote-user": user}) + + def test_second_job_for_same_user_rejected(self): + with AuthTestClient(self.app) as client: + self.assertEqual(self.start(client, "alice").status_code, 200) + self.assertEqual(self.start(client, "alice").status_code, 409) + self.assertEqual(self.start(client, "bob").status_code, 200) + + def test_user_can_start_again_after_job_finishes(self): + with AuthTestClient(self.app) as client: + job_id = self.start(client, "alice").json()["job_id"] + job = motion_search.get_motion_search_job(job_id) + job.end_time = datetime.now().timestamp() + self.assertEqual(self.start(client, "alice").status_code, 200) + self.assertIsNotNone(motion_search.get_motion_search_job(job_id)) + + 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) + 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, "carol").status_code, 200) + + def test_start_failure_releases_slot(self): + self.runner.return_value.start.side_effect = RuntimeError("boom") + with AuthTestClient(self.app) as client: + with self.assertRaises(RuntimeError): + self.start(client, "alice") + self.assertEqual(motion_search._motion_search_jobs, {}) + self.runner.return_value.requestor.stop.assert_called_once() + + self.runner.return_value.start.side_effect = None + self.assertEqual(self.start(client, "alice").status_code, 200) + + def test_shared_identity_skips_per_user_limit(self): + with AuthTestClient(self.app) as client: + self.assertEqual(self.start(client, "anonymous").status_code, 200) + self.assertEqual(self.start(client, "anonymous").status_code, 200) + + motion_search._motion_search_jobs.clear() + self.minimal_config["auth"] = {"enabled": False} + app = self.create_app() + with AuthTestClient(app) as client: + 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): + self.assertEqual(self.start(client, f"user{i}").status_code, 200) + self.assertEqual(self.start(client, "late").status_code, 429) + + def test_finished_jobs_evicted_after_ttl(self): + with AuthTestClient(self.app) as client: + old_id = self.start(client, "alice").json()["job_id"] + recent_id = self.start(client, "bob").json()["job_id"] + running_id = self.start(client, "carol").json()["job_id"] + + now = 10_000.0 + motion_search.get_motion_search_job(old_id).end_time = ( + now - motion_search.FINISHED_JOB_TTL - 1 + ) + motion_search.get_motion_search_job(recent_id).end_time = now - 1 + motion_search._evict_finished_jobs(now) + + self.assertIsNone(motion_search.get_motion_search_job(old_id)) + self.assertIsNotNone(motion_search.get_motion_search_job(recent_id)) + self.assertIsNotNone(motion_search.get_motion_search_job(running_id)) + + 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_decode.py b/frigate/test/test_motion_search_decode.py index fda0837662..75e29e4351 100644 --- a/frigate/test/test_motion_search_decode.py +++ b/frigate/test/test_motion_search_decode.py @@ -1,5 +1,6 @@ """Tests for the motion search hardware-accelerated decode helpers.""" +import subprocess as sp import unittest from types import SimpleNamespace from unittest import mock @@ -140,6 +141,11 @@ class TestVodDecodeCommand(unittest.TestCase): 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":[' @@ -147,19 +153,37 @@ 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 + "frigate.jobs.motion_search_decode.sp.Popen", + return_value=self._proc(sample), ): - pts = probe_vod_keyframe_pts("ffprobe", "http://x/index.m3u8") + pts = probe_vod_keyframe_pts( + "ffprobe", "http://x/index.m3u8", lambda: False + ) self.assertEqual(pts, [0.0, 2.0]) 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): diff --git a/web/public/locales/en/views/motionSearch.json b/web/public/locales/en/views/motionSearch.json index 8358bfb75d..e2eccf2df8 100644 --- a/web/public/locales/en/views/motionSearch.json +++ b/web/public/locales/en/views/motionSearch.json @@ -56,6 +56,8 @@ "noROI": "Please draw a region of interest", "noTimeRange": "Please select a time range", "invalidTimeRange": "End time must be after start time", + "alreadyRunning": "You already have a motion search running. Wait for it to finish or cancel it.", + "tooManySearches": "Too many motion searches are running. Try again when one finishes.", "searchFailed": "Search failed: {{message}}", "polygonTooSmall": "Polygon must have at least 3 points", "unknown": "Unknown error" diff --git a/web/src/views/motion-search/MotionSearchView.tsx b/web/src/views/motion-search/MotionSearchView.tsx index 01c3632444..c9e4fba92b 100644 --- a/web/src/views/motion-search/MotionSearchView.tsx +++ b/web/src/views/motion-search/MotionSearchView.tsx @@ -892,6 +892,20 @@ export default function MotionSearchView({ setIsSearching(false); } } catch (error) { + const status = axios.isAxiosError(error) + ? error.response?.status + : undefined; + + if (status === 409 || status === 429) { + toast.error( + t( + status === 409 ? "errors.alreadyRunning" : "errors.tooManySearches", + ), + ); + setIsSearching(false); + return; + } + let errorMessage = t("errors.unknown"); if (axios.isAxiosError<{ message?: string; detail?: string }>(error)) {