diff --git a/frigate/api/motion_search.py b/frigate/api/motion_search.py index d8449bde98..fadb4adb4b 100644 --- a/frigate/api/motion_search.py +++ b/frigate/api/motion_search.py @@ -10,6 +10,8 @@ from pydantic import BaseModel, Field from frigate.api.auth import require_camera_access from frigate.api.defs.tags import Tags from frigate.jobs.motion_search import ( + MotionSearchCapacityError, + MotionSearchUserBusyError, cancel_motion_search_job, get_motion_search_job, start_motion_search_job, @@ -20,6 +22,9 @@ 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.""" @@ -138,18 +143,44 @@ 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, - ) + if body.end_time - body.start_time > MAX_SEARCH_RANGE_SECONDS: + return JSONResponse( + content={ + "success": False, + "message": "Time range must be 48 hours or less", + }, + status_code=400, + ) + + 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=request.headers.get("remote-user", "anonymous"), + ) + 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..a5f3d73090 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,10 @@ 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 @dataclass @@ -98,6 +98,7 @@ class MotionSearchJob(Job): min_area: float = 5.0 parallel: bool = False max_results: int = 25 + username: str = "" # Track progress total_frames_processed: int = 0 @@ -112,6 +113,7 @@ class MotionSearchJob(Job): def to_dict(self) -> dict[str, Any]: """Convert to dictionary for WebSocket transmission.""" d = asdict(self) + d.pop("username") if self.metrics: d["metrics"] = self.metrics.to_dict() return d @@ -833,11 +835,29 @@ 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 stop_all_motion_search_jobs() -> None: """Cancel all running motion search jobs for clean shutdown.""" with _jobs_lock: @@ -857,10 +877,12 @@ def start_motion_search_job( min_area: float = 5.0, parallel: bool = False, max_results: int = 25, + username: str = "", ) -> str: """Start a new motion search job. - Returns the job ID. + Returns the job ID. Raises MotionSearchUserBusyError when the user already + has an active job and MotionSearchCapacityError when MAX_ACTIVE_JOBS are active. """ job = MotionSearchJob( camera=camera_name, @@ -871,14 +893,23 @@ def start_motion_search_job( min_area=min_area, parallel=parallel, max_results=max_results, + username=username, ) cancel_event = threading.Event() with _jobs_lock: - _motion_search_jobs[job.id] = (job, cancel_event) + _evict_finished_jobs(datetime.now().timestamp()) - set_current_job(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 j.end_time is None] + if any(j.username == username for j in active): + raise MotionSearchUserBusyError() + if len(active) >= MAX_ACTIVE_JOBS: + raise MotionSearchCapacityError() + + _motion_search_jobs[job.id] = (job, cancel_event) runner = MotionSearchRunner(job, config, cancel_event) runner.start() @@ -904,10 +935,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: 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..380cefd8c8 --- /dev/null +++ b/frigate/test/http_api/test_http_motion_search.py @@ -0,0 +1,77 @@ +"""Tests for motion search job limits.""" + +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") + 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"] + motion_search.get_motion_search_job(job_id).end_time = 1.0 + self.assertEqual(self.start(client, "alice").status_code, 200) + + 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_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"] + 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_not_broadcast(self): + job = motion_search.MotionSearchJob(username="alice") + self.assertNotIn("username", job.to_dict())