limit concurrent motion search jobs and search range

Any viewer could start unlimited motion search jobs over an unbounded time range, and finished jobs were never dropped from memory. Each user now gets one active job (409), at most 3 run at once (429), the range is capped at 48 hours, and finished jobs are evicted 10 minutes after they end. Jobs are no longer filed in the shared job manager, which kept every finished job forever.

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={

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:

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())