diff --git a/frigate/api/motion_search.py b/frigate/api/motion_search.py index fadb4adb4b..313f303990 100644 --- a/frigate/api/motion_search.py +++ b/frigate/api/motion_search.py @@ -9,6 +9,7 @@ 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, @@ -100,6 +101,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, @@ -147,7 +162,7 @@ async def start_motion_search( return JSONResponse( content={ "success": False, - "message": "Time range must be 48 hours or less", + "message": f"Time range must be {MAX_SEARCH_RANGE_SECONDS // 3600} hours or less", }, status_code=400, ) @@ -163,7 +178,7 @@ async def start_motion_search( min_area=body.min_area, parallel=body.parallel, max_results=body.max_results, - username=request.headers.get("remote-user", "anonymous"), + username=_per_user_identity(request, config), ) except MotionSearchUserBusyError: return JSONResponse( diff --git a/frigate/jobs/motion_search.py b/frigate/jobs/motion_search.py index cd4c4fd515..234f27e821 100644 --- a/frigate/jobs/motion_search.py +++ b/frigate/jobs/motion_search.py @@ -53,6 +53,8 @@ FALLBACK_SAMPLE_FPS = 2.0 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,7 +100,8 @@ class MotionSearchJob(Job): min_area: float = 5.0 parallel: bool = False max_results: int = 25 - username: str = "" + username: str | None = None + cancelled_at: float | None = None # Track progress total_frames_processed: int = 0 @@ -114,6 +117,7 @@ class MotionSearchJob(Job): """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 @@ -779,13 +783,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(): @@ -858,6 +864,13 @@ def _evict_finished_jobs(now: float) -> None: 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: @@ -877,12 +890,13 @@ def start_motion_search_job( min_area: float = 5.0, parallel: bool = False, max_results: int = 25, - username: str = "", + username: str | None = None, ) -> str: """Start a new motion search job. - Returns the job ID. Raises MotionSearchUserBusyError when the user already + 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, @@ -899,21 +913,26 @@ def start_motion_search_job( cancel_event = threading.Event() with _jobs_lock: - _evict_finished_jobs(datetime.now().timestamp()) + now = datetime.now().timestamp() + _evict_finished_jobs(now) # 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): + 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): raise MotionSearchUserBusyError() if len(active) >= MAX_ACTIVE_JOBS: raise MotionSearchCapacityError() _motion_search_jobs[job.id] = (job, cancel_event) + runner: MotionSearchRunner | None = None try: - MotionSearchRunner(job, config, cancel_event).start() + 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 @@ -960,6 +979,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..376cdcb935 100644 --- a/frigate/jobs/motion_search_decode.py +++ b/frigate/jobs/motion_search_decode.py @@ -106,7 +106,13 @@ def keyframe_sampling_eligible( return max(gaps) <= max_gap -VOD_PROTOCOL_ARGS = ["-protocol_whitelist", "pipe,file,http,tcp"] +# rw_timeout (microseconds) ends a stalled VOD read instead of blocking forever. +VOD_PROTOCOL_ARGS = [ + "-protocol_whitelist", + "pipe,file,http,tcp", + "-rw_timeout", + "30000000", +] def build_vod_decode_command( diff --git a/frigate/test/http_api/test_http_motion_search.py b/frigate/test/http_api/test_http_motion_search.py index 380cefd8c8..aeaf19ce6b 100644 --- a/frigate/test/http_api/test_http_motion_search.py +++ b/frigate/test/http_api/test_http_motion_search.py @@ -1,5 +1,6 @@ """Tests for motion search job limits.""" +from datetime import datetime from unittest.mock import patch from frigate.jobs import motion_search @@ -23,7 +24,7 @@ class TestHttpMotionSearch(BaseTestHttp): self.addCleanup(motion_search._motion_search_jobs.clear) runner_patch = patch("frigate.jobs.motion_search.MotionSearchRunner") - runner_patch.start() + self.runner = runner_patch.start() self.addCleanup(runner_patch.stop) def start(self, client: AuthTestClient, user: str, body: dict = BODY): @@ -38,8 +39,44 @@ class TestHttpMotionSearch(BaseTestHttp): 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 + 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_holds_slot_until_grace_expires(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.status = "cancelled" + job.cancelled_at = datetime.now().timestamp() + self.assertEqual(self.start(client, "alice").status_code, 409) + + job.cancelled_at -= motion_search.CANCEL_GRACE_SECONDS + 1 + self.assertEqual(self.start(client, "alice").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_global_cap_enforced(self): with AuthTestClient(self.app) as client: diff --git a/frigate/test/test_motion_search_decode.py b/frigate/test/test_motion_search_decode.py index fda0837662..df2ff5cd13 100644 --- a/frigate/test/test_motion_search_decode.py +++ b/frigate/test/test_motion_search_decode.py @@ -101,6 +101,7 @@ 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)