remove motion search range check from the UI

With rw_timeout, the HLS demuxer skipped a timed-out segment and exited cleanly, so a run reported success with a hole in it, and when the keyframe probe and the decode lost different segments the equal frame counts paired frames with the wrong timestamps. The keyframe path now requires the probe to cover the whole run without a gap over 5 s, a decode whose stderr shows a skipped segment fails the run, and a cadence decode more than 5 s short of the run's duration fails it too, so these runs land in segments_with_errors. The keyframe probe polls should_stop and is killed on cancel, probe_video_dimensions gets the read timeout, and a user's own cancelled job no longer blocks their next search while it still counts toward the global cap.

docker/main/rootfs/etc/s6-overlay/s6-rc.d/certsync/run

@@ -36,8 +36,8 @@ else
     lefile="/config/tls/fullchain.pem"
 fi
 
-tls_enabled=`python3 -I /usr/local/nginx/get_nginx_settings.py | jq -r .tls.enabled`
-listen_external_port=`python3 -I /usr/local/nginx/get_nginx_settings.py | jq -r .listen.external_port`
+tls_enabled=`python3 /usr/local/nginx/get_nginx_settings.py | jq -r .tls.enabled`
+listen_external_port=`python3 /usr/local/nginx/get_nginx_settings.py | jq -r .listen.external_port`
 
 while true
 do

docker/main/rootfs/etc/s6-overlay/s6-rc.d/frigate/run

@@ -11,13 +11,30 @@ if [[ "$(id -u)" -eq 0 ]]; then
     fi
 fi
 
+# /root survives s6-setuidgid and breaks cache writes after the drop; set
+# before opt_in_out so the opt-out marker lands where the service will look
+if [[ "$runs_as_root" -eq 0 ]]; then
+    export HOME=/config
+fi
+
+# detector runtimes installed at first start (pip install --user) live under
+# $HOME/.local; the dynamic loader only reads LD_LIBRARY_PATH at exec time
+export LD_LIBRARY_PATH="${LD_LIBRARY_PATH:+${LD_LIBRARY_PATH}:}${HOME}/.local/lib"
+export PATH="${PATH}:${HOME}/.local/bin"
+
+# opt out of openvino telemetry
+if [ -e /usr/local/bin/opt_in_out ]; then
+  /usr/local/bin/opt_in_out --opt_out > /dev/null 2>&1
+fi
+
+# Logs should be sent to stdout so that s6 can collect them
+
 # Tell S6-Overlay not to restart this service
 s6-svc -O .
 
-# Runs as root with root's HOME and LD_LIBRARY_PATH; -I skips user site-packages
 function set_libva_version() {
     local ffmpeg_path
-    ffmpeg_path=$(python3 -I /usr/local/ffmpeg/get_ffmpeg_path.py)
+    ffmpeg_path=$(python3 /usr/local/ffmpeg/get_ffmpeg_path.py)
     LIBAVFORMAT_VERSION_MAJOR=$("$ffmpeg_path" -version | grep -Po "libavformat\W+\K\d+")
     export LIBAVFORMAT_VERSION_MAJOR
 }
@@ -25,32 +42,14 @@ function set_libva_version() {
 echo "[INFO] Preparing Frigate..."
 set_libva_version
 
-# HOME=/root survives s6-setuidgid, so any run that is not root gets
-# HOME=/config explicitly. The runtime user owns /config, so these paths go
-# only to the service process, never to root. Detector runtimes installed at
-# first start (pip install --user) live under $HOME/.local.
-run_as=()
-runtime_home="${HOME}"
-if [[ "$runs_as_root" -eq 0 ]]; then
-    runtime_home=/config
-    if [[ "$(id -u)" -eq 0 ]]; then
-        run_as=(s6-setuidgid frigate)
-    fi
-fi
-run_as+=(env
-    "HOME=${runtime_home}"
-    "LD_LIBRARY_PATH=${LD_LIBRARY_PATH:+${LD_LIBRARY_PATH}:}${runtime_home}/.local/lib"
-    "PATH=${PATH}:${runtime_home}/.local/bin")
-
-# opt out of openvino telemetry; the marker lands under the service's HOME
-if [ -e /usr/local/bin/opt_in_out ]; then
-  "${run_as[@]}" /usr/local/bin/opt_in_out --opt_out > /dev/null 2>&1 || true
-fi
-
 echo "[INFO] Starting Frigate..."
 
 cd /opt/frigate || echo "[ERROR] Failed to change working directory to /opt/frigate"
 
 # Replace the bash process with the Frigate process, redirecting stderr to stdout
 exec 2>&1
-exec "${run_as[@]}" python3 -u -m frigate
+if [[ "$(id -u)" -ne 0 || "$runs_as_root" -eq 1 ]]; then
+    exec python3 -u -m frigate
+else
+    exec s6-setuidgid frigate python3 -u -m frigate
+fi

docker/main/rootfs/etc/s6-overlay/s6-rc.d/go2rtc/run

@@ -59,33 +59,38 @@ function get_ip_and_port_from_supervisor() {
 
 function set_libva_version() {
     local ffmpeg_path
-    ffmpeg_path=$(python3 -I /usr/local/ffmpeg/get_ffmpeg_path.py)
+    ffmpeg_path=$(python3 /usr/local/ffmpeg/get_ffmpeg_path.py)
     LIBAVFORMAT_VERSION_MAJOR=$("$ffmpeg_path" -version | grep -Po "libavformat\W+\K\d+")
     export LIBAVFORMAT_VERSION_MAJOR
 }
 
 set_libva_version
 
-# create_config.py creates the file exclusively and fails if anything
-# reappears at this path
-rm -f /dev/shm/go2rtc.yaml
-
-echo "[INFO] Preparing new go2rtc config..."
-
-if [[ -n "${SUPERVISOR_TOKEN:-}" ]]; then
-    # Running as a Home Assistant Add-on, infer the IP address and port
-    get_ip_and_port_from_supervisor
+if [[ -f "/dev/shm/go2rtc.yaml" ]]; then
+    echo "[INFO] Removing stale config from last run..."
+    rm /dev/shm/go2rtc.yaml
 fi
 
-python3 -I /usr/local/go2rtc/create_config.py
+if [[ ! -f "/dev/shm/go2rtc.yaml" ]]; then
+    echo "[INFO] Preparing new go2rtc config..."
+
+    if [[ -n "${SUPERVISOR_TOKEN:-}" ]]; then
+        # Running as a Home Assistant Add-on, infer the IP address and port
+        get_ip_and_port_from_supervisor
+    fi
+
+    python3 /usr/local/go2rtc/create_config.py
+else
+    echo "[WARNING] Unable to remove existing go2rtc config. Changes made to your frigate config file may not be recognized. Please remove the /dev/shm/go2rtc.yaml from your docker host manually."
+fi
 
 # HomeKit persistence. The helper is symlink-safe; hand off to go2rtc only when dropping.
 readonly homekit_config_path="/config/go2rtc_homekit.yml"
 if [[ "$(id -u)" -eq 0 && "$runs_as_root" -eq 0 ]]; then
-    python3 -I /usr/local/go2rtc/prepare_homekit.py "${homekit_config_path}" --chown
-    chown -h go2rtc:go2rtc /dev/shm/go2rtc.yaml 2>/dev/null || true
+    python3 /usr/local/go2rtc/prepare_homekit.py "${homekit_config_path}" --chown
+    chown go2rtc:go2rtc /dev/shm/go2rtc.yaml 2>/dev/null || true
 else
-    python3 -I /usr/local/go2rtc/prepare_homekit.py "${homekit_config_path}"
+    python3 /usr/local/go2rtc/prepare_homekit.py "${homekit_config_path}"
 fi
 
 readonly config_path="/config"

docker/main/rootfs/etc/s6-overlay/s6-rc.d/nginx/run

@@ -124,7 +124,7 @@ fi
 mkdir -p /etc/letsencrypt/www 2>/dev/null || true
 
 # nginx settings are read once; both templates consume them
-nginx_settings=$(python3 -I /usr/local/nginx/get_nginx_settings.py)
+nginx_settings=$(python3 /usr/local/nginx/get_nginx_settings.py)
 
 # build templates for optional FRIGATE_BASE_PATH environment variable
 echo "$nginx_settings" | \

docker/main/rootfs/usr/local/go2rtc/create_config.py

@@ -198,13 +198,9 @@ if config.get("birdseye", {}).get("restream", False):
     else:
         go2rtc_config["streams"] = {"birdseye": ffmpeg_cmd}
 
-# The run script removes the old file first. Exclusive create never follows a
-# link the go2rtc user planted in sticky /dev/shm. The config contains camera
-# credentials, so it is not world-readable.
-fd = os.open(
-    "/dev/shm/go2rtc.yaml",
-    os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
-    0o640,
-)
-with os.fdopen(fd, "w") as f:
+# Write go2rtc_config to /dev/shm/go2rtc.yaml
+with open("/dev/shm/go2rtc.yaml", "w") as f:
     yaml.dump(go2rtc_config, f)
+
+# config contains camera credentials; do not leave it world-readable
+os.chmod("/dev/shm/go2rtc.yaml", 0o640)

docker/main/rootfs/usr/local/nginx/get_nginx_settings.py

@@ -8,7 +8,7 @@ from typing import Any
 from ruamel.yaml import YAML
 
 sys.path.insert(0, "/opt/frigate")
-from frigate.util.config import find_config_file, parse_listen_address
+from frigate.util.config import find_config_file
 
 sys.path.remove("/opt/frigate")
 
@@ -35,29 +35,20 @@ ipv6_config: dict[str, Any] = networking_config.get("ipv6", {})
 ipv6_config.setdefault("enabled", False)
 
 listen_config: dict[str, Any] = networking_config.get("listen", {})
+listen_config.setdefault("internal", 5000)
+listen_config.setdefault("external", 8971)
 
-# Frigate's own validation runs elsewhere, and a bad value here must not stop nginx
-# from serving the safe-mode UI. Valid values are kept so a fallback never widens
-# the unauthenticated internal listener beyond the configured address.
-for key, default in (("internal", 5000), ("external", 8971)):
-    try:
-        value, port = parse_listen_address(listen_config.get(key, default))
-    except ValueError:
-        print(
-            f"[WARN] Invalid networking.listen.{key}, using {default}",
-            file=sys.stderr,
-        )
-        value, port = default, default
+# handle case where internal port is a string with ip:port
+internal_port = listen_config["internal"]
+if type(internal_port) is str:
+    internal_port = int(internal_port.split(":")[-1])
+listen_config["internal_port"] = internal_port
 
-    listen_config[key] = value
-    listen_config[f"{key}_port"] = port
-
-if listen_config["internal_port"] == listen_config["external_port"]:
-    print(
-        "[ERROR] networking.listen.internal and external share a port",
-        file=sys.stderr,
-    )
-    sys.exit(1)
+# handle case where external port is a string with ip:port
+external_port = listen_config["external"]
+if type(external_port) is str:
+    external_port = int(external_port.split(":")[-1])
+listen_config["external_port"] = external_port
 
 base_path = os.environ.get("FRIGATE_BASE_PATH", "")
 

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

frigate/config/network.py

@@ -1,12 +1,18 @@
-from pydantic import Field, field_validator, model_validator
-
-from frigate.util.config import parse_listen_address
+from pydantic import Field, model_validator
 
 from .base import FrigateBaseModel
 
 __all__ = ["IPv6Config", "ListenConfig", "NetworkingConfig"]
 
 
+def parse_listen_port(value: int | str) -> int:
+    """Return the port number from a bare port or an "address:port" value."""
+    if isinstance(value, str):
+        return int(value.split(":")[-1])
+
+    return value
+
+
 class IPv6Config(FrigateBaseModel):
     enabled: bool = Field(
         default=False,
@@ -27,20 +33,13 @@ class ListenConfig(FrigateBaseModel):
         description="External listening port for Frigate (default 8971).",
     )
 
-    # Runs before pydantic coerces a bool to an int port, which
-    # get_nginx_settings.py rejects when it reads the raw YAML
-    @field_validator("internal", "external", mode="before")
-    @classmethod
-    def validate_listen(cls, value: int | str) -> int | str:
-        return parse_listen_address(value)[0]
-
     @property
     def internal_port(self) -> int:
-        return parse_listen_address(self.internal)[1]
+        return parse_listen_port(self.internal)
 
     @property
     def external_port(self) -> int:
-        return parse_listen_address(self.external)[1]
+        return parse_listen_port(self.external)
 
     @model_validator(mode="after")
     def validate_distinct_ports(self) -> "ListenConfig":

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)
 

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 []
 

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)

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

frigate/test/test_network_config.py

@@ -36,73 +36,6 @@ class TestListenConfig(unittest.TestCase):
         self.assertEqual(listen.internal_port, 5001)
         self.assertEqual(listen.external_port, 8971)
 
-    def test_valid_listen_values_are_rebuilt(self):
-        cases = {
-            5000: 5000,
-            "5000": 5000,
-            "127.0.0.1:5000": "127.0.0.1:5000",
-            "0.0.0.0:5000": "0.0.0.0:5000",
-            "[::1]:5000": "[::1]:5000",
-            "[::]:5000": "[::]:5000",
-            "[2001:DB8:0:0::1]:5000": "[2001:db8::1]:5000",
-            " 5000 ": 5000,
-            "5000\n": 5000,
-            " 127.0.0.1:5000 ": "127.0.0.1:5000",
-            "localhost:5000": "localhost:5000",
-            "frigate-host.lan:5000": "frigate-host.lan:5000",
-            "*:5000": "*:5000",
-        }
-
-        for value, expected in cases.items():
-            with self.subTest(value=value):
-                listen = ListenConfig(internal=value)
-
-                self.assertEqual(listen.internal, expected)
-                self.assertEqual(listen.internal_port, 5000)
-
-    def test_invalid_listen_values_rejected(self):
-        cases = [
-            "5000; location /x/ { alias /; autoindex on; } #:5000",
-            "127.0.0.1:5000; include /etc/passwd",
-            "5000 ssl",
-            "5000;",
-            "::1:5000",
-            "[127.0.0.1]:5000",
-            "[fe80::1%eth0]:5000",
-            "local host:5000",
-            "localhost;:5000",
-            "local{host}:5000",
-            "host#:5000",
-            "'localhost':5000",
-            '"localhost":5000',
-            "-host:5000",
-            "host-:5000",
-            "host..lan:5000",
-            "**:5000",
-            "*.lan:5000",
-            "unix:/tmp/sock:5000",
-            "5000 5001",
-            ":5000",
-            "127.0.0.1:",
-            "127.0.0.1:+5000",
-            0,
-            65536,
-            -1,
-        ]
-
-        for value in cases:
-            with self.subTest(value=value):
-                with self.assertRaises(ValidationError):
-                    ListenConfig(internal=value)
-
-    def test_bool_listen_value_rejected(self):
-        # get_nginx_settings.py sees the raw bool and falls back to the default,
-        # so accepting it as port 1 would split nginx and the auth port check
-        for value in (True, False):
-            with self.subTest(value=value):
-                with self.assertRaises(ValidationError):
-                    ListenConfig(internal=value)
-
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)

frigate/util/config.py

@@ -1,10 +1,8 @@
 """configuration utils."""
 
 import asyncio
-import ipaddress
 import logging
 import os
-import re
 import shutil
 from functools import cache
 from typing import Any
@@ -128,49 +126,6 @@ def redact_credential(obj: dict[str, Any], key: str) -> None:
         obj.pop(key, None)
 
 
-LISTEN_HOSTNAME = re.compile(
-    r"[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?"
-    r"(?:\.[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?)*"
-)
-
-
-def parse_listen_address(value: int | str) -> tuple[int | str, int]:
-    """Validate a networking.listen value and rebuild it from its parsed parts.
-
-    Accepts a port, "host:port", "*:port", or "[IPv6]:port", where host is an
-    IPv4 address or an RFC 1123 hostname. nginx receives the rebuilt value
-    verbatim, so anything else raises ValueError.
-
-    Returns:
-        The normalized value (an int for a bare port) and the port
-    """
-    if isinstance(value, bool) or not isinstance(value, int | str):
-        raise ValueError(f"invalid listen value: {value!r}")
-
-    host, sep, port_text = str(value).strip().rpartition(":")
-
-    if not re.fullmatch(r"[0-9]{1,5}", port_text) or not (1 <= int(port_text) <= 65535):
-        raise ValueError(f"invalid listen port: {value!r}")
-
-    port = int(port_text)
-
-    if not sep:
-        return port, port
-
-    if host.startswith("[") and host.endswith("]"):
-        address = ipaddress.IPv6Address(host[1:-1])
-
-        if address.scope_id:
-            raise ValueError(f"listen address must not have a scope ID: {value!r}")
-
-        return f"[{address}]:{port}", port
-
-    if host != "*" and not LISTEN_HOSTNAME.fullmatch(host):
-        raise ValueError(f"invalid listen address: {value!r}")
-
-    return f"{host}:{port}", port
-
-
 def find_config_file() -> str:
     """Return the path of the config file to use.
 

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"

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