From 4fdf31fc01d0f34e40309e4e57b7358238a40668 Mon Sep 17 00:00:00 2001 From: Stevenson Chittumuri Date: Fri, 3 Jul 2026 13:54:40 -0400 Subject: [PATCH 1/2] Add model-server crash recovery: respawn, fast-fail gate, local fallback The shared model-server process (AUTOPTZ_MODEL_SERVER=1) was spawn-once with no recovery: a dead server meant every camera's detect() blocked its full 5s timeout forever. Add supervisor-side respawn on the same health-scan cadence, reusing the existing per-camera backoff pattern (exponential 1->30s, max 5 attempts) for the single shared process. Reuses the same request/response queues across a respawn (shm re-attach was already lazy per-request in serve(), pinned with a test rather than changed). Client-side: InferenceClient's default timeout drops 5.0s -> 2.0s and gains a server_down gate so detect() fails fast during an outage instead of stalling each camera at the full timeout per frame. On restart-budget exhaustion, Supervisor.model_server_failed() latches and every model-server-mode worker's refresh_detector_from_pool() is invoked; RemotePool.detector() now checks a failed flag and swaps in a lazily-built local detector instead of the dead IPC client, so detection continues in degraded per-worker mode. Co-Authored-By: Claude Fable 5 --- autoptz/engine/pipeline/inference_server.py | 51 ++++- autoptz/engine/process_worker.py | 35 ++- autoptz/engine/supervisor.py | 178 +++++++++++++-- tests/test_inference_server_ipc.py | 184 ++++++++++++++++ tests/test_supervisor_health.py | 227 ++++++++++++++++++++ 5 files changed, 654 insertions(+), 21 deletions(-) diff --git a/autoptz/engine/pipeline/inference_server.py b/autoptz/engine/pipeline/inference_server.py index 19e0b76f..355ee365 100644 --- a/autoptz/engine/pipeline/inference_server.py +++ b/autoptz/engine/pipeline/inference_server.py @@ -67,13 +67,24 @@ class InferenceClient: """ def __init__( - self, camera_id: str, req_q: Any, resp_q: Any, shm_writer: Any, timeout_s: float = 5.0 + self, + camera_id: str, + req_q: Any, + resp_q: Any, + shm_writer: Any, + timeout_s: float = 2.0, + server_down: Any = None, ) -> None: self._cam = camera_id self._req_q = req_q self._resp_q = resp_q self._shm = shm_writer self._timeout_s = timeout_s + # Supervisor-owned gate (multiprocessing.Event or equivalent — anything with + # ``is_set()``): set while the shared server is down/being restarted, so + # detect() fast-fails instead of blocking a whole timeout per frame during + # an outage. None (tests/no-recovery callers) → gate is always "up". + self._server_down = server_down self._seq = 0 # per-request id so a timed-out reply can't desync the next call # The response queue may be reused across a worker restart; drop any replies # left by a previous (crashed) run so they can't be matched to a fresh request. @@ -88,6 +99,10 @@ def ep(self) -> str: return "model-server" def detect(self, frame: Any) -> Any: + if self._server_down is not None and self._server_down.is_set(): + # Supervisor has flagged the server as down/being restarted — fail fast + # instead of waiting on a queue nothing is going to answer. + return [] try: h, w = int(self._shm.height), int(self._shm.width) oh, ow = int(frame.shape[0]), int(frame.shape[1]) @@ -130,14 +145,40 @@ def detect(self, frame: Any) -> Any: class RemotePool: """Pool-shaped wrapper so the worker's pooled detect-stack build uses the IPC - client as its detector (the tracker stays local per-camera).""" + client as its detector (the tracker stays local per-camera). + + Degraded fallback (R-3): once the supervisor gives up trying to respawn a dead + model-server (``failed`` set), ``detector()`` stops handing back the (now + permanently useless) IPC client and instead lazily builds + caches a LOCAL + detector via ``build_local_fn``. This is the seam + ``CameraWorker.refresh_detector_from_pool()`` already re-invokes on command, so + the supervisor's only job on budget exhaustion is: set ``failed`` and ask every + model-server worker to refresh — no new IPC, no protocol change. + """ detector_ep = "model-server" - def __init__(self, client: InferenceClient) -> None: + def __init__( + self, + client: InferenceClient, + *, + failed: Any = None, + build_local_fn: Any = None, + ) -> None: self._client = client - - def detector(self) -> InferenceClient: + self._failed = failed + self._build_local_fn = build_local_fn + self._local: Any | None = None + + def detector(self) -> Any: + if self._failed is not None and self._failed.is_set(): + if self._local is None and self._build_local_fn is not None: + try: + self._local = self._build_local_fn() + except Exception: # noqa: BLE001 — local fallback must never crash the worker + log.warning("model-server fallback: local detector build failed", exc_info=True) + if self._local is not None: + return self._local return self._client diff --git a/autoptz/engine/process_worker.py b/autoptz/engine/process_worker.py index 484afeb6..9c455534 100644 --- a/autoptz/engine/process_worker.py +++ b/autoptz/engine/process_worker.py @@ -213,6 +213,8 @@ def run_camera_process( identity_q: Any, infer_req_q: Any = None, infer_resp_q: Any = None, + infer_down_ev: Any = None, + infer_failed_ev: Any = None, ) -> None: """Child-process entrypoint: build + run a CameraWorker, driven by ``cmd_q``. @@ -272,8 +274,26 @@ def _emit_identity(rec: Any) -> None: from autoptz.engine.runtime.shm import ShmWriter _infer_writer = ShmWriter(shm_name_for(spec.camera_id), SERVER_FRAME_H, SERVER_FRAME_W) - client = InferenceClient(spec.camera_id, infer_req_q, infer_resp_q, _infer_writer) - worker.set_inference_pool(RemotePool(client)) + client = InferenceClient( + spec.camera_id, infer_req_q, infer_resp_q, _infer_writer, server_down=infer_down_ev + ) + + # Local-fallback factory (R-3): only used if the supervisor ever sets + # infer_failed_ev (model-server restart budget exhausted). Built lazily + # and cached by RemotePool so a healthy model-server never pays for it. + def _build_local_detector(_spec: WorkerSpec = spec) -> Any: + from autoptz.engine.pipeline.pool import build_inference_pool + + pool = build_inference_pool( + detector_tier=_spec.detector_tier, + unified_pose=_spec.unified_pose, + allow_model_download=False, + ) + return pool.detector() if pool is not None else None + + worker.set_inference_pool( + RemotePool(client, failed=infer_failed_ev, build_local_fn=_build_local_detector) + ) worker._infer_shm_writer = _infer_writer # keep the shm alive for the worker's life elif not spec.synthetic: _wire_models_and_identity(worker, spec) @@ -392,6 +412,8 @@ def __init__( unified_pose: bool = False, infer_req_q: Any = None, infer_resp_q: Any = None, + infer_down_ev: Any = None, + infer_failed_ev: Any = None, ) -> None: self.camera_id = camera_id self.config = config @@ -399,6 +421,13 @@ def __init__( # this camera's response queue. None → the child builds its own detector. self._infer_req_q = infer_req_q self._infer_resp_q = infer_resp_q + # R-3 crash recovery: shared gates set by the supervisor. ``infer_down_ev`` + # makes the child's InferenceClient fast-fail while the server is being + # respawned; ``infer_failed_ev`` tells RemotePool to fall back to a local + # detector once the restart budget is exhausted. Both None → no recovery + # wiring (matches today's behavior). + self._infer_down_ev = infer_down_ev + self._infer_failed_ev = infer_failed_ev self.shm_name = f"cam_{camera_id[:8]}_preview" # must match CameraWorker's self._on_telemetry = on_telemetry self._on_identity: Any | None = None @@ -472,6 +501,8 @@ def start(self) -> None: self._identity_q, self._infer_req_q, self._infer_resp_q, + self._infer_down_ev, + self._infer_failed_ev, ), name=f"camproc-{self.camera_id[:8]}", daemon=True, diff --git a/autoptz/engine/supervisor.py b/autoptz/engine/supervisor.py index e6302f8d..c24c566e 100644 --- a/autoptz/engine/supervisor.py +++ b/autoptz/engine/supervisor.py @@ -165,6 +165,18 @@ def __init__( self._model_server_stop: Any | None = None self._infer_req_q: Any | None = None self._infer_resp_qs: dict[str, Any] = {} + # R-3 crash recovery. ``_model_server_down`` is set while the server is + # dead/being respawned so every camera's InferenceClient fast-fails instead + # of blocking a full timeout per frame; cleared once the respawned server + # signals ready. ``_model_server_failed_ev`` latches once the restart budget + # is exhausted — handed to RemotePool so it swaps in a local detector, and + # queryable via :meth:`model_server_failed`. ``_ms_restart_state`` mirrors + # the per-camera ``_restart_state`` shape (attempts, next_allowed_t, failed) + # for this single shared process, reusing the SAME backoff constants. + self._model_server_down: Any | None = None + self._model_server_failed_ev: Any | None = None + self._ms_restart_state: tuple[int, float, bool] = (0, 0.0, False) + self._model_server_camera_ids: list[str] = [] # Global ML-subsystem switches (detection / tracking / face_recognition / # pose), broadcast via SetFeaturesCmd and applied to every worker. @@ -309,6 +321,10 @@ def stop(self) -> None: self._model_server_stop = None self._infer_req_q = None self._infer_resp_qs = {} + self._model_server_down = None + self._model_server_failed_ev = None + self._ms_restart_state = (0, 0.0, False) + self._model_server_camera_ids = [] if proc is not None: try: if stop_ev is not None: @@ -499,6 +515,7 @@ def tick(self) -> None: if now - self._last_health_scan_t >= _HEALTH_SCAN_INTERVAL_S: self._last_health_scan_t = now self._scan_worker_health(now) + self._scan_model_server_health(now) # Throttled (~1 Hz) system-CPU sample fanned out to every worker so the # per-camera governor can back off collectively when the machine is hot. if now - getattr(self, "_last_cpu_sample_t", 0.0) >= _CPU_SAMPLE_INTERVAL_S: @@ -689,6 +706,121 @@ def _scan_worker_health(self, now: float) -> None: except Exception: # noqa: BLE001 log.warning("camera_id=%s respawn failed", cid, exc_info=True) + def _scan_model_server_health(self, now: float) -> None: + """Respawn the shared model-server process if it died (model-server mode). + + Mirrors :meth:`_scan_worker_health`'s backoff/budget shape exactly (same + constants, same exponential curve, same cap) rather than inventing a new + mechanism — there is just one "camera" here (the server process) instead of + one per real camera, so the state is a single tuple, not a dict. + + On respawn: reuse the SAME ``_infer_req_q`` / ``_infer_resp_qs`` (shm re- + attach on the server side is already lazy per-request — see + :func:`autoptz.engine.pipeline.inference_server.serve`), holding + ``_model_server_down`` set for the duration so every camera's + ``InferenceClient`` fast-fails instead of blocking a timeout per frame. + + On budget exhaustion: latch ``_model_server_failed_ev`` (queryable via + :meth:`model_server_failed`) and ask every model-server-mode worker to + ``refresh_detector_from_pool()`` — the seam ``RemotePool.detector()`` uses + to swap in a local detector — so detection continues in degraded mode. + """ + from autoptz.engine.runtime.flags import env_model_server + + if not env_model_server() or self._model_server_proc is None: + return + + try: + alive = bool(self._model_server_proc.is_alive()) + except Exception: # noqa: BLE001 + alive = False + if alive: + if self._ms_restart_state != (0, 0.0, False): + self._ms_restart_state = (0, 0.0, False) + return + + attempts, next_allowed_t, _failed = self._ms_restart_state + if attempts >= _MAX_RESTART_ATTEMPTS: + return # already gave up — the permanent-failure ERROR was logged once + if now < next_allowed_t: + return # still in the back-off window + + log.info( + "model-server process died — restarting (attempt %d/%d)", + attempts + 1, + _MAX_RESTART_ATTEMPTS, + ) + new_attempts = attempts + 1 + backoff = min(_MAX_BACKOFF_S, _BASE_BACKOFF_S * (2 ** (new_attempts - 1))) + + if new_attempts >= _MAX_RESTART_ATTEMPTS: + self._ms_restart_state = (new_attempts, now + backoff, True) + log.error( + "model-server permanently failed: %d auto-restart attempts exhausted " + "— falling back to local per-camera detectors.", + new_attempts, + ) + if self._model_server_failed_ev is not None: + self._model_server_failed_ev.set() + self._refresh_model_server_workers() + return + + self._ms_restart_state = (new_attempts, now + backoff, False) + if self._model_server_down is not None: + self._model_server_down.set() + self._respawn_model_server() + + def _respawn_model_server(self) -> None: + """Start a fresh server process reusing the existing queues, then clear the + down-gate once it signals ready (or on any failure, so cameras don't stay + latched fast-fail forever over a respawn that itself failed to start).""" + try: + import multiprocessing as mp + + from autoptz.engine.pipeline.inference_server import run_inference_server + + ctx = mp.get_context("spawn") + self._model_server_stop = ctx.Event() + ready = ctx.Event() + self._model_server_proc = ctx.Process( + target=run_inference_server, + args=( + self._infer_req_q, + self._infer_resp_qs, + list(self._model_server_camera_ids), + self._resolve_detector_tier(), + self._any_unified_pose(), + ready, + self._model_server_stop, + ), + name="model-server", + daemon=True, + ) + self._model_server_proc.start() + if not ready.wait(timeout=30.0): + log.warning("respawned model-server slow to accept requests.") + except Exception: # noqa: BLE001 — respawn is best-effort; next scan retries + log.warning("model-server respawn failed", exc_info=True) + finally: + if self._model_server_down is not None: + self._model_server_down.clear() + + def _refresh_model_server_workers(self) -> None: + """Ask every model-server-mode camera worker to re-pull its detector from + the pool — the seam RemotePool uses to swap in a local detector once the + supervisor has given up on the shared server.""" + with self._lock: + workers = list(self._workers.values()) + for worker in workers: + if not getattr(worker, "_is_process_worker", False): + continue + refresh = getattr(worker, "refresh_detector_from_pool", None) + if callable(refresh): + try: + refresh() + except Exception: # noqa: BLE001 + log.debug("model-server fallback refresh failed", exc_info=True) + # ── command routing ───────────────────────────────────────────────────────── def _route(self, cmd: Any) -> None: @@ -986,6 +1118,16 @@ def _ensure_inference_pool(self) -> Any | None: self._inference_pool = None return self._inference_pool + def _resolve_detector_tier(self) -> str: + tier = "auto" + try: + getter = getattr(self._client, "getDetectorModelTier", None) + if callable(getter): + tier = str(getter() or "auto") + except Exception: # noqa: BLE001 + tier = "auto" + return tier + def _ensure_model_server(self, camera_ids: list[str]) -> None: """Spawn the ONE shared model-server process (model-server mode only). @@ -1008,14 +1150,12 @@ def _ensure_model_server(self, camera_ids: list[str]) -> None: self._infer_req_q = ctx.Queue() self._infer_resp_qs = {cid: ctx.Queue() for cid in camera_ids} self._model_server_stop = ctx.Event() + self._model_server_down = ctx.Event() + self._model_server_failed_ev = ctx.Event() + self._model_server_camera_ids = list(camera_ids) + self._ms_restart_state = (0, 0.0, False) ready = ctx.Event() - tier = "auto" - try: - getter = getattr(self._client, "getDetectorModelTier", None) - if callable(getter): - tier = str(getter() or "auto") - except Exception: # noqa: BLE001 - tier = "auto" + tier = self._resolve_detector_tier() self._model_server_proc = ctx.Process( target=run_inference_server, args=( @@ -1050,6 +1190,8 @@ def _ensure_model_server(self, camera_ids: list[str]) -> None: self._model_server_stop = None self._infer_req_q = None self._infer_resp_qs = {} + self._model_server_down = None + self._model_server_failed_ev = None def _apply_hardware_env(self, camera_count: int) -> None: """Publish global hardware prefs into the environment before workers start. @@ -1236,13 +1378,7 @@ def _make_worker(self, camera_id: str, config: CameraConfig) -> Any: ) return self._worker_factory(camera_id, config, on_telemetry) - tier = "auto" - try: - getter = getattr(self._client, "getDetectorModelTier", None) - if callable(getter): - tier = str(getter() or "auto") - except Exception: # noqa: BLE001 - tier = "auto" + tier = self._resolve_detector_tier() db_path = "" try: db_path = str(getattr(self._store, "_path", "") or "") @@ -1270,6 +1406,11 @@ def _make_worker(self, camera_id: str, config: CameraConfig) -> Any: # response queue so it delegates detection instead of loading a detector. infer_req_q=self._infer_req_q, infer_resp_q=infer_resp_q, + # R-3 crash recovery gates (both None until the flag first spawns the + # server): fast-fail while down, fall back to a local detector once the + # restart budget is exhausted. + infer_down_ev=self._model_server_down, + infer_failed_ev=self._model_server_failed_ev, ) def _spawn_worker(self, camera_id: str, *, defer_inference: bool = False) -> None: @@ -1377,6 +1518,15 @@ def failed_cameras(self) -> list[str]: """Camera ids that auto-restart has permanently given up on.""" return [cid for cid, st in self._restart_state.items() if st[2]] + def model_server_failed(self) -> bool: + """True iff the shared model-server's restart budget is exhausted. + + The UI can use this to surface "model server failed — using local + detectors"; workers have already been told to fall back (see + :meth:`_refresh_model_server_workers`) by the time this latches. + """ + return bool(self._ms_restart_state[2]) + def _default_worker_factory( self, camera_id: str, diff --git a/tests/test_inference_server_ipc.py b/tests/test_inference_server_ipc.py index c61003dc..2038214d 100644 --- a/tests/test_inference_server_ipc.py +++ b/tests/test_inference_server_ipc.py @@ -251,3 +251,187 @@ def detect_fn(frame): # noqa: ANN001 t.join(timeout=1.0) writer.close() reader.close() + + +# ── R-3: model-server crash recovery ───────────────────────────────────────── + + +def test_default_timeout_is_two_seconds() -> None: + """The client must fast-fail at 2.0s (not the old 5.0s) so a dead server does + not stall a camera's inference loop for 5 seconds on every single frame.""" + import inspect + + sig = inspect.signature(InferenceClient.__init__) + assert sig.parameters["timeout_s"].default == 2.0 + + +def test_server_down_gate_short_circuits_detect_without_waiting_on_queue() -> None: + """(b) While the supervisor has the server marked down, detect() must return [] + IMMEDIATELY (well under timeout_s) instead of blocking on the (dead) response + queue — otherwise every camera crawls at timeout_s per frame during an outage. + """ + import queue + + name = f"itest_{uuid.uuid4().hex[:8]}" + writer = ShmWriter(name, 32, 32) + server_down = threading.Event() + server_down.set() # supervisor has flagged the server as down/being restarted + # No server thread running at all — if the gate didn't short-circuit, this would + # block for the full timeout waiting on an empty resp_q. + client = InferenceClient( + "camA", queue.Queue(), queue.Queue(), writer, timeout_s=5.0, server_down=server_down + ) + t0 = time.monotonic() + assert client.detect(_frame(1, 32, 32)) == [] + elapsed = time.monotonic() - t0 + assert elapsed < 0.5, f"detect() blocked {elapsed:.2f}s despite the down-gate being set" + writer.close() + + +def test_server_down_gate_clears_and_detect_resumes() -> None: + """Once the supervisor clears the down-gate (respawned server signalled ready), + detect() must go back to actually talking to the server instead of staying + latched in the fast-fail path.""" + import queue + + cam = "camA" + name = f"itest_{uuid.uuid4().hex[:8]}" + writer = ShmWriter(name, 32, 32) + reader = ShmReader(name, 32, 32) + req_q: queue.Queue = queue.Queue() + resp_q: queue.Queue = queue.Queue() + stop = threading.Event() + server_down = threading.Event() + server_down.set() + + def detect_fn(frame): # noqa: ANN001 + return [("ok", int(frame[0, 0, 0]))] + + t = threading.Thread( + target=serve, args=(req_q, {cam: resp_q}, {cam: reader}, detect_fn, stop), daemon=True + ) + t.start() + try: + client = InferenceClient(cam, req_q, resp_q, writer, timeout_s=2.0, server_down=server_down) + assert client.detect(_frame(3, 32, 32)) == [] # gate set → fast-fail, no real call + server_down.clear() # supervisor: respawned server is ready again + assert client.detect(_frame(3, 32, 32)) == [("ok", 3)] # resumes real detection + finally: + stop.set() + t.join(timeout=1.0) + writer.close() + reader.close() + + +def test_remote_pool_uses_client_while_server_is_healthy() -> None: + """RemotePool must not fall back to a local detector while the model-server + has not been declared permanently failed — the IPC client stays authoritative.""" + import queue + + name = f"itest_{uuid.uuid4().hex[:8]}" + writer = ShmWriter(name, 16, 16) + client = InferenceClient("camA", queue.Queue(), queue.Queue(), writer) + built_local = [] + + def _build_local(): + built_local.append(True) + return object() + + pool = RemotePool(client, failed=threading.Event(), build_local_fn=_build_local) + assert pool.detector() is client + assert built_local == [] + writer.close() + + +def test_remote_pool_falls_back_to_local_detector_once_failed() -> None: + """(c) After the supervisor exhausts the model-server restart budget, it sets + the pool's failed flag. RemotePool.detector() must then hand back a LOCAL + detector (built via the injected factory) instead of the dead IPC client, so + ``worker.refresh_detector_from_pool()`` — which just re-calls ``pool.detector()`` + — is enough to make detection resume in degraded (per-worker) mode. + """ + import queue + + name = f"itest_{uuid.uuid4().hex[:8]}" + writer = ShmWriter(name, 16, 16) + client = InferenceClient("camA", queue.Queue(), queue.Queue(), writer) + local_detector = object() + calls = [] + + def _build_local(): + calls.append(True) + return local_detector + + failed = threading.Event() + pool = RemotePool(client, failed=failed, build_local_fn=_build_local) + assert pool.detector() is client # healthy → still the IPC client + + failed.set() # supervisor: restart budget exhausted + assert pool.detector() is local_detector + assert len(calls) == 1 + + # Subsequent calls reuse the cached local detector, not rebuild it every time. + assert pool.detector() is local_detector + assert len(calls) == 1 + writer.close() + + +def test_respawned_server_reuses_same_queues_no_client_reconstruction() -> None: + """(a)+(d) Kill the server-side thread mid-run, "respawn" it (a fresh serve() + loop reusing the SAME req/resp queues and shm reader dict — exactly what the + supervisor does across a real process respawn), and prove the ORIGINAL client + resumes getting real detections with NO reconstruction. This pins that shm + re-attach is lazy per-request (serve()'s ``attach`` callback), so reusing the + same queues + a fresh reader dict is sufficient for recovery. + """ + import queue + + cam = "camA" + name = f"itest_{uuid.uuid4().hex[:8]}" + writer = ShmWriter(name, 32, 32) + req_q: queue.Queue = queue.Queue() + resp_q: queue.Queue = queue.Queue() + + def detect_fn(frame): # noqa: ANN001 + return [("ok", int(frame[0, 0, 0]))] + + def attach(_c: str): # noqa: ANN202 + try: + return ShmReader(name, 32, 32) + except FileNotFoundError: + return None + + stop1 = threading.Event() + t1 = threading.Thread( + target=serve, + args=(req_q, {cam: resp_q}, {}, detect_fn, stop1), + kwargs={"attach": attach}, + daemon=True, + ) + t1.start() + client = InferenceClient(cam, req_q, resp_q, writer, timeout_s=2.0) + try: + assert client.detect(_frame(4, 32, 32)) == [("ok", 4)] # server #1 alive + + # Kill server #1 ("the child process died"). + stop1.set() + t1.join(timeout=1.0) + + # Respawn: a brand-new serve() loop, SAME req_q/resp_q, fresh reader dict — + # mirrors the supervisor building a new Process with the queues it already + # held. No new InferenceClient is constructed. + stop2 = threading.Event() + t2 = threading.Thread( + target=serve, + args=(req_q, {cam: resp_q}, {}, detect_fn, stop2), + kwargs={"attach": attach}, + daemon=True, + ) + t2.start() + try: + assert client.detect(_frame(9, 32, 32)) == [("ok", 9)] # resumed, same client + finally: + stop2.set() + t2.join(timeout=1.0) + finally: + writer.close() diff --git a/tests/test_supervisor_health.py b/tests/test_supervisor_health.py index fd97d502..e63a8730 100644 --- a/tests/test_supervisor_health.py +++ b/tests/test_supervisor_health.py @@ -686,6 +686,233 @@ def factory(camera_id, config, on_tel): sup.stop() +class _FakeModelServerProc: + """Minimal fake for supervisor._model_server_proc with a settable is_alive().""" + + def __init__(self, *_a, **_k) -> None: # noqa: ANN002, ANN003 + self._alive = True + self.start_calls = 0 + self.terminate_calls = 0 + + def start(self) -> None: + self.start_calls += 1 + + def is_alive(self) -> bool: + return self._alive + + def terminate(self) -> None: + self.terminate_calls += 1 + self._alive = False + + def join(self, timeout: float = 0.0) -> None: # noqa: ARG002 + pass + + +class _FakeEvent: + """Stand-in for ctx.Event() used by both the ready-handshake and the + stop/down/failed gates. The fake process never runs run_inference_server to + set "ready" for real, so wait() always reports ready instantly instead of + blocking the full 30s production timeout; set()/clear()/is_set() behave + normally so down-gate assertions still work.""" + + def __init__(self) -> None: + self._flag = False + + def set(self) -> None: + self._flag = True + + def clear(self) -> None: + self._flag = False + + def is_set(self) -> bool: + return self._flag + + def wait(self, timeout: float = 0.0) -> bool: # noqa: ARG002 + return True + + +class _MsFakeProcessWorker(_HealthFakeWorker): + """Fake model-server-mode camera worker: flags itself as a process worker (like + ProcessWorkerHandle) and records refresh_detector_from_pool() calls so tests can + assert the supervisor invoked the local-fallback seam.""" + + _is_process_worker = True + + def __init__(self, camera_id, config, on_telemetry) -> None: # noqa: ANN001 + super().__init__(camera_id, config, on_telemetry) + self.refresh_calls = 0 + + def refresh_detector_from_pool(self) -> None: + self.refresh_calls += 1 + + +class TestModelServerHealthScan: + """R-3: the supervisor's model-server respawn path mirrors the worker + auto-restart machinery (same backoff constants, same budget/cap shape) instead + of inventing a new mechanism.""" + + def _build_ms(self, qapp, monkeypatch): # noqa: ANN001 + """Supervisor in model-server mode with a fake process + one fake + process-worker camera, bypassing the real mp.Process spawn.""" + import multiprocessing as mp + + from autoptz.ui.engine_client import EngineClient + + monkeypatch.setenv("AUTOPTZ_MODEL_SERVER", "1") + client = EngineClient() + cid = client.addCamera("usb://0", "X") + client.drain_commands() + + factory_log: list[_MsFakeProcessWorker] = [] + + def factory(camera_id, config, on_tel): # noqa: ANN001 + w = _MsFakeProcessWorker(camera_id, config, on_tel) + factory_log.append(w) + return w + + sup = _make_sup_with_factory(client, factory) + sup._ensure_identity_service = lambda: None # type: ignore[method-assign] + sup._ensure_inference_pool = lambda: None # type: ignore[method-assign] + + real_ctx = mp.get_context("spawn") + proc_log: list[_FakeModelServerProc] = [] + + class _FakeCtx: + Queue = staticmethod(real_ctx.Queue) + Event = staticmethod(_FakeEvent) + + def Process(self, *_a, **_k): # noqa: ANN002, ANN003, N802 + p = _FakeModelServerProc() + proc_log.append(p) + return p + + monkeypatch.setattr(mp, "get_context", lambda *_a, **_k: _FakeCtx()) + # The fake process never actually serves, so don't block start() waiting + # for the real ready-event handshake. + monkeypatch.setattr( + "autoptz.engine.supervisor.Supervisor._ensure_model_server", + lambda self, camera_ids: _fake_ensure_model_server(self, camera_ids, proc_log), + ) + sup.start() + return sup, client, factory_log, proc_log, cid + + def test_dead_model_server_is_respawned(self, qapp, monkeypatch) -> None: + sup, client, factory_log, proc_log, cid = self._build_ms(qapp, monkeypatch) + try: + assert len(proc_log) == 1 + proc_log[0]._alive = False + sup._scan_model_server_health(1000.0) + assert len(proc_log) == 2 # respawned + assert sup._model_server_proc is proc_log[1] + finally: + sup.stop() + + def test_respawn_reuses_same_queues_no_reconstruction(self, qapp, monkeypatch) -> None: + """(d) Across a respawn the SAME request queue / response queues stay in + place — clients need no reconstruction, only the server process changes.""" + sup, client, factory_log, proc_log, cid = self._build_ms(qapp, monkeypatch) + try: + req_q_before = sup._infer_req_q + resp_qs_before = sup._infer_resp_qs + proc_log[0]._alive = False + sup._scan_model_server_health(1000.0) + assert sup._infer_req_q is req_q_before + assert sup._infer_resp_qs is resp_qs_before + finally: + sup.stop() + + def test_backoff_prevents_immediate_second_respawn(self, qapp, monkeypatch) -> None: + sup, client, factory_log, proc_log, cid = self._build_ms(qapp, monkeypatch) + try: + proc_log[0]._alive = False + now = 1000.0 + sup._scan_model_server_health(now) + assert len(proc_log) == 2 + proc_log[1]._alive = False + sup._scan_model_server_health(now + 0.1) # still inside backoff window + assert len(proc_log) == 2 + finally: + sup.stop() + + def test_backoff_expires_and_allows_respawn(self, qapp, monkeypatch) -> None: + sup, client, factory_log, proc_log, cid = self._build_ms(qapp, monkeypatch) + try: + proc_log[0]._alive = False + now = 1000.0 + sup._scan_model_server_health(now) + assert len(proc_log) == 2 + proc_log[1]._alive = False + sup._scan_model_server_health(now + _BASE_BACKOFF_S + 0.1) + assert len(proc_log) == 3 + finally: + sup.stop() + + def test_budget_exhaustion_sets_failed_flag_and_triggers_worker_fallback( + self, qapp, monkeypatch + ) -> None: + """(c) After restart attempts are exhausted, model_server_failed() is True + and every model-server-mode worker's refresh_detector_from_pool() fires + (the seam that lets RemotePool swap in a local detector).""" + sup, client, factory_log, proc_log, cid = self._build_ms(qapp, monkeypatch) + try: + now = 1000.0 + for _attempt in range(_MAX_RESTART_ATTEMPTS): + sup._model_server_proc._alive = False + now += _MAX_BACKOFF_S + 1.0 + sup._scan_model_server_health(now) + + assert sup.model_server_failed() is True + assert factory_log[0].refresh_calls == 1 # fired exactly once, at exhaustion + finally: + sup.stop() + + def test_healthy_server_is_not_touched(self, qapp, monkeypatch) -> None: + sup, client, factory_log, proc_log, cid = self._build_ms(qapp, monkeypatch) + try: + sup._scan_model_server_health(1000.0) + assert len(proc_log) == 1 # no respawn + assert sup.model_server_failed() is False + finally: + sup.stop() + + def test_not_enabled_is_a_noop(self, qapp, monkeypatch) -> None: + """With AUTOPTZ_MODEL_SERVER off, scanning must never try to touch a + model-server process that was never started.""" + from autoptz.ui.engine_client import EngineClient + + monkeypatch.delenv("AUTOPTZ_MODEL_SERVER", raising=False) + client = EngineClient() + sup = _make_sup_with_factory(client, lambda cid, cfg, tel: _HealthFakeWorker(cid, cfg, tel)) + sup._ensure_identity_service = lambda: None # type: ignore[method-assign] + sup._ensure_inference_pool = lambda: None # type: ignore[method-assign] + sup._running = True + sup._scan_model_server_health(1000.0) # must not raise + assert sup._model_server_proc is None + assert sup.model_server_failed() is False + + +def _fake_ensure_model_server(sup, camera_ids, proc_log) -> None: # noqa: ANN001 + """Deterministic stand-in for Supervisor._ensure_model_server: builds the same + queue/event handles but skips the real ready-event wait (the fake process never + serves), so tests can drive _scan_model_server_health directly.""" + import multiprocessing as mp + + from autoptz.engine.runtime.flags import env_model_server + + if not env_model_server() or sup._model_server_proc is not None: + return + ctx = mp.get_context("spawn") + sup._infer_req_q = ctx.Queue() + sup._infer_resp_qs = {cid: ctx.Queue() for cid in camera_ids} + sup._model_server_stop = ctx.Event() + sup._model_server_down = ctx.Event() + sup._model_server_failed_ev = ctx.Event() + sup._model_server_camera_ids = list(camera_ids) + sup._ms_restart_state = (0, 0.0, False) + sup._model_server_proc = ctx.Process() + sup._model_server_proc.start() + + class TestPermanentFailed: def test_failed_flag_and_accessor_set_at_cap(self, qapp, caplog) -> None: import logging From 70fccba01bd8c8ea872bce07bdb613844530e674 Mon Sep 17 00:00:00 2001 From: Stevenson Chittumuri Date: Fri, 3 Jul 2026 14:19:12 -0400 Subject: [PATCH 2/2] Make model-server respawn non-blocking on the tick/GUI thread MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _respawn_model_server() previously did proc.start() then a synchronous ready.wait(timeout=30.0) inside _scan_model_server_health(), which tick() drives from a QTimer on the GUI thread — a slow respawn could freeze the UI for up to 30s per attempt, up to 5 attempts. Split the respawn into two phases across scan ticks: starting the child now only records a spawn deadline and the ready Event, returning immediately. Later ticks poll ready.is_set() (never .wait() with a timeout) until either it signals ready (success — gate cleared, attempts reset) or the deadline passes (treated as a failed attempt — stuck child killed, existing backoff/budget accounting applies). The server_down gate stays set for the whole spawning+backoff window so clients keep fast-failing throughout. Also add a real-behavior companion to test_default_timeout_is_two_seconds and simplify an existing conditional-then-reassign in the health scan. Co-Authored-By: Claude Fable 5 --- autoptz/engine/supervisor.py | 118 ++++++++++-- tests/test_inference_server_ipc.py | 19 ++ tests/test_supervisor_health.py | 278 ++++++++++++++++++++++++++++- 3 files changed, 395 insertions(+), 20 deletions(-) diff --git a/autoptz/engine/supervisor.py b/autoptz/engine/supervisor.py index c24c566e..2ac7ccb4 100644 --- a/autoptz/engine/supervisor.py +++ b/autoptz/engine/supervisor.py @@ -71,6 +71,10 @@ _BASE_BACKOFF_S = 1.0 # initial restart back-off _MAX_BACKOFF_S = 30.0 # maximum restart back-off (exponential cap) _MAX_RESTART_ATTEMPTS = 5 # give up after this many consecutive failures +# How long a respawned model-server gets to signal "ready" before the attempt is +# treated as failed. Polled across ticks (never awaited) so a slow spawn cannot +# block the GUI thread — see Supervisor._scan_model_server_health. +_MS_SPAWN_TIMEOUT_S = 30.0 # A worker can be ALIVE yet HUNG (capture/inference threads stuck, no telemetry). # Treat it as unhealthy and respawn it through the same backoff path once its # telemetry is older than this. Deliberately above the 2.0 s inference-stall @@ -177,6 +181,13 @@ def __init__( self._model_server_failed_ev: Any | None = None self._ms_restart_state: tuple[int, float, bool] = (0, 0.0, False) self._model_server_camera_ids: list[str] = [] + # Non-blocking respawn bookkeeping (tick() runs on the GUI thread — the + # health scan must never block it). ``_ms_ready_ev`` is the ready-handshake + # Event for the IN-FLIGHT respawn attempt; ``_ms_spawn_deadline`` is the + # monotonic time by which it must have signalled ready. Both None when no + # respawn is currently in flight. + self._ms_ready_ev: Any | None = None + self._ms_spawn_deadline: float | None = None # Global ML-subsystem switches (detection / tracking / face_recognition / # pose), broadcast via SetFeaturesCmd and applied to every worker. @@ -325,6 +336,8 @@ def stop(self) -> None: self._model_server_failed_ev = None self._ms_restart_state = (0, 0.0, False) self._model_server_camera_ids = [] + self._ms_ready_ev = None + self._ms_spawn_deadline = None if proc is not None: try: if stop_ev is not None: @@ -714,11 +727,18 @@ def _scan_model_server_health(self, now: float) -> None: mechanism — there is just one "camera" here (the server process) instead of one per real camera, so the state is a single tuple, not a dict. + ``tick()`` (and therefore this scan) runs on the GUI thread in the shipped + app, so a respawn attempt must never block it: starting the child and + waiting for its ready-handshake are split across ticks. ``_ms_spawn_deadline`` + is None when no respawn is in flight; while it is set, this method only + POLLS the ready Event (never ``.wait()`` with a timeout) and otherwise + no-ops until the deadline passes. ``_model_server_down`` stays set for the + whole spawning+backoff window so every camera's ``InferenceClient`` + fast-fails instead of blocking a timeout per frame. + On respawn: reuse the SAME ``_infer_req_q`` / ``_infer_resp_qs`` (shm re- attach on the server side is already lazy per-request — see - :func:`autoptz.engine.pipeline.inference_server.serve`), holding - ``_model_server_down`` set for the duration so every camera's - ``InferenceClient`` fast-fails instead of blocking a timeout per frame. + :func:`autoptz.engine.pipeline.inference_server.serve`). On budget exhaustion: latch ``_model_server_failed_ev`` (queryable via :meth:`model_server_failed`) and ask every model-server-mode worker to @@ -730,13 +750,16 @@ def _scan_model_server_health(self, now: float) -> None: if not env_model_server() or self._model_server_proc is None: return + if self._ms_spawn_deadline is not None: + self._poll_model_server_spawn(now) + return + try: alive = bool(self._model_server_proc.is_alive()) except Exception: # noqa: BLE001 alive = False if alive: - if self._ms_restart_state != (0, 0.0, False): - self._ms_restart_state = (0, 0.0, False) + self._ms_restart_state = (0, 0.0, False) return attempts, next_allowed_t, _failed = self._ms_restart_state @@ -768,12 +791,74 @@ def _scan_model_server_health(self, now: float) -> None: self._ms_restart_state = (new_attempts, now + backoff, False) if self._model_server_down is not None: self._model_server_down.set() - self._respawn_model_server() + self._respawn_model_server(now) + + def _poll_model_server_spawn(self, now: float) -> None: + """Non-blocking continuation of an in-flight respawn attempt. + + Called on every scan tick while ``_ms_spawn_deadline`` is set. Only ever + polls ``is_set()`` — never ``.wait()`` with a nonzero timeout — so a slow + or stuck child cannot stall the GUI thread. The down-gate stays set for the + whole spawning window regardless of outcome; it is only cleared on success. + """ + ready = self._ms_ready_ev + deadline = self._ms_spawn_deadline + assert deadline is not None # guarded by caller + + if ready is not None and ready.is_set(): + # Respawn succeeded — same reset the old alive-branch did. + self._ms_spawn_deadline = None + self._ms_ready_ev = None + self._ms_restart_state = (0, 0.0, False) + if self._model_server_down is not None: + self._model_server_down.clear() + log.info("model-server respawned successfully — resuming shared detection.") + return + + if now < deadline: + return # still spawning; check again next tick + + # Deadline exceeded without a ready signal — treat exactly like a failed + # attempt: kill the stuck child (best-effort, non-blocking join) and apply + # the same backoff/budget accounting _scan_model_server_health uses. The + # down-gate is deliberately left set — the caller (next tick) may either + # retry or, at budget exhaustion, trigger the local-fallback path. + log.warning("respawned model-server slow to accept requests — treating as failed.") + self._ms_spawn_deadline = None + self._ms_ready_ev = None + proc = self._model_server_proc + if proc is not None: + try: + proc.terminate() + except Exception: # noqa: BLE001 + log.debug("failed to terminate stuck model-server child", exc_info=True) + try: + proc.join(timeout=0.5) + except Exception: # noqa: BLE001 + log.debug("failed to join stuck model-server child", exc_info=True) - def _respawn_model_server(self) -> None: - """Start a fresh server process reusing the existing queues, then clear the - down-gate once it signals ready (or on any failure, so cameras don't stay - latched fast-fail forever over a respawn that itself failed to start).""" + attempts, _next_allowed_t, _failed = self._ms_restart_state + backoff = min(_MAX_BACKOFF_S, _BASE_BACKOFF_S * (2 ** (attempts - 1))) + if attempts >= _MAX_RESTART_ATTEMPTS: + self._ms_restart_state = (attempts, now + backoff, True) + log.error( + "model-server permanently failed: %d auto-restart attempts exhausted " + "— falling back to local per-camera detectors.", + attempts, + ) + if self._model_server_failed_ev is not None: + self._model_server_failed_ev.set() + self._refresh_model_server_workers() + else: + self._ms_restart_state = (attempts, now + backoff, False) + # Gate stays set (never cleared here) — next scan either retries after the + # backoff window or the caller above has already routed to fallback. + + def _respawn_model_server(self, now: float) -> None: + """Start a fresh server process reusing the existing queues and record a + spawn deadline; the ready-handshake is polled on later ticks by + :meth:`_poll_model_server_spawn`, never awaited here — starting a child + process is fast, so this itself does not block the caller.""" try: import multiprocessing as mp @@ -797,13 +882,16 @@ def _respawn_model_server(self) -> None: daemon=True, ) self._model_server_proc.start() - if not ready.wait(timeout=30.0): - log.warning("respawned model-server slow to accept requests.") + self._ms_ready_ev = ready + self._ms_spawn_deadline = now + _MS_SPAWN_TIMEOUT_S except Exception: # noqa: BLE001 — respawn is best-effort; next scan retries log.warning("model-server respawn failed", exc_info=True) - finally: - if self._model_server_down is not None: - self._model_server_down.clear() + self._ms_ready_ev = None + self._ms_spawn_deadline = None + # Gate stays set: the outer backoff/budget accounting already ran for + # this attempt, so the next scan either retries (after backoff) or + # routes to fallback (budget exhausted) — see + # _scan_model_server_health / _poll_model_server_spawn. def _refresh_model_server_workers(self) -> None: """Ask every model-server-mode camera worker to re-pull its detector from diff --git a/tests/test_inference_server_ipc.py b/tests/test_inference_server_ipc.py index 2038214d..ffeda2dc 100644 --- a/tests/test_inference_server_ipc.py +++ b/tests/test_inference_server_ipc.py @@ -265,6 +265,25 @@ def test_default_timeout_is_two_seconds() -> None: assert sig.parameters["timeout_s"].default == 2.0 +def test_default_timeout_actually_bounds_detect_when_server_never_replies() -> None: + """Behavioral companion to the signature check above: construct a client with + the DEFAULT timeout_s (no override) against a request queue nobody is + servicing, and prove detect() actually gives up around 2s — not the old 5s — + instead of only trusting the __init__ default value in isolation.""" + import queue + + name = f"itest_{uuid.uuid4().hex[:8]}" + writer = ShmWriter(name, 32, 32) + # No server thread at all — the response queue is never populated, so detect() + # must fall through its own timeout_s deadline rather than hang. + client = InferenceClient("camA", queue.Queue(), queue.Queue(), writer) + t0 = time.monotonic() + assert client.detect(_frame(1, 32, 32)) == [] + elapsed = time.monotonic() - t0 + assert 1.5 <= elapsed < 4.0, f"detect() took {elapsed:.2f}s — expected ~2.0s default timeout" + writer.close() + + def test_server_down_gate_short_circuits_detect_without_waiting_on_queue() -> None: """(b) While the supervisor has the server marked down, detect() must return [] IMMEDIATELY (well under timeout_s) instead of blocking on the (dead) response diff --git a/tests/test_supervisor_health.py b/tests/test_supervisor_health.py index e63a8730..a6510314 100644 --- a/tests/test_supervisor_health.py +++ b/tests/test_supervisor_health.py @@ -13,6 +13,7 @@ _INFER_RESTART_S, _MAX_BACKOFF_S, _MAX_RESTART_ATTEMPTS, + _MS_SPAWN_TIMEOUT_S, _WORKER_HANG_S, _WORKER_WARMUP_GRACE_S, ) @@ -835,15 +836,35 @@ def test_backoff_prevents_immediate_second_respawn(self, qapp, monkeypatch) -> N sup.stop() def test_backoff_expires_and_allows_respawn(self, qapp, monkeypatch) -> None: + """Updated for non-blocking respawn (R-3 review fix): a single + _scan_model_server_health call now only STARTS a respawn attempt (spawn + deadline pending) instead of synchronously spawning-and-waiting-for-ready + in one call. ``_FakeEvent`` here never actually signals ready (nothing + drives a real server loop that would call ``.set()``), so — like a + genuinely stuck child — the in-flight attempt only resolves once its + spawn deadline passes; see TestModelServerRespawnNonBlocking for + dedicated polling-across-ticks coverage including the ready-succeeds + case. This test now walks a full spawn-timeout -> backoff -> respawn + cycle instead of asserting a same-tick third spawn.""" sup, client, factory_log, proc_log, cid = self._build_ms(qapp, monkeypatch) try: proc_log[0]._alive = False now = 1000.0 - sup._scan_model_server_health(now) + sup._scan_model_server_health(now) # attempt 1: starts spawning assert len(proc_log) == 2 - proc_log[1]._alive = False + + # Spawn deadline passes without a ready signal -> attempt 1 fails and + # enters its backoff window. + now += _MS_SPAWN_TIMEOUT_S + 0.1 + sup._scan_model_server_health(now) + assert len(proc_log) == 2 # no new process yet — still in backoff + assert sup._ms_spawn_deadline is None + + sup._scan_model_server_health(now + 0.01) # still inside backoff window + assert len(proc_log) == 2 # untouched + sup._scan_model_server_health(now + _BASE_BACKOFF_S + 0.1) - assert len(proc_log) == 3 + assert len(proc_log) == 3 # backoff expired — attempt 2 starts finally: sup.stop() @@ -852,14 +873,23 @@ def test_budget_exhaustion_sets_failed_flag_and_triggers_worker_fallback( ) -> None: """(c) After restart attempts are exhausted, model_server_failed() is True and every model-server-mode worker's refresh_detector_from_pool() fires - (the seam that lets RemotePool swap in a local detector).""" + (the seam that lets RemotePool swap in a local detector). + + Updated for non-blocking respawn (R-3 review fix): starting a respawn and + observing its outcome are now two separate ticks, and ``_FakeEvent`` + never actually signals ready, so each cycle below ticks twice: once to + start the respawn, once more past the spawn deadline to resolve it as a + failed attempt (mirroring a server that never comes up) before advancing + past the backoff window for the next cycle.""" sup, client, factory_log, proc_log, cid = self._build_ms(qapp, monkeypatch) try: now = 1000.0 for _attempt in range(_MAX_RESTART_ATTEMPTS): sup._model_server_proc._alive = False now += _MAX_BACKOFF_S + 1.0 - sup._scan_model_server_health(now) + sup._scan_model_server_health(now) # starts the respawn attempt + now += _MS_SPAWN_TIMEOUT_S + 0.1 + sup._scan_model_server_health(now) # deadline exceeded -> failed assert sup.model_server_failed() is True assert factory_log[0].refresh_calls == 1 # fired exactly once, at exhaustion @@ -891,6 +921,244 @@ def test_not_enabled_is_a_noop(self, qapp, monkeypatch) -> None: assert sup.model_server_failed() is False +class _AssertNoBlockingWaitEvent: + """Stand-in for ctx.Event() that records any ``.wait(timeout=...)`` call made + with a nonzero timeout, so a test can assert the tick path never blocked — + the health-scan/tick thread must only ever poll ``is_set()``. + + Deliberately does NOT raise from inside ``wait()``: production code wraps the + old blocking respawn in a broad ``except Exception``, which would silently + swallow an AssertionError raised here and give a false-negative (test passes + even though the code blocked in real life, where wait() actually sleeps + instead of raising). Recording to a class-level list and asserting on it + from the test body survives that broad except clause. ``wait(timeout=0)`` (a + non-blocking check) is tolerated and not recorded.""" + + blocking_wait_calls: list[float] = [] + + def __init__(self) -> None: + self._flag = False + + def set(self) -> None: + self._flag = True + + def clear(self) -> None: + self._flag = False + + def is_set(self) -> bool: + return self._flag + + def wait(self, timeout: float = 0.0) -> bool: + if timeout: + _AssertNoBlockingWaitEvent.blocking_wait_calls.append(timeout) + return self._flag + + +class _ControllableModelServerProc(_FakeModelServerProc): + """Fake model-server process whose is_alive() can be flipped after start().""" + + +class TestModelServerRespawnNonBlocking: + """IMPORTANT-1: _respawn_model_server (called from _scan_model_server_health, + which tick() drives on the GUI thread in the shipped app) must never block — + `proc.start()` + record a spawn deadline, then poll `ready.is_set()` on later + ticks instead of `ready.wait(timeout=...)`. All tests here use a synthetic + monotonic clock (`now`) — no real sleeps.""" + + def _build_ms(self, qapp, monkeypatch, event_cls=_AssertNoBlockingWaitEvent): # noqa: ANN001 + """Same shape as TestModelServerHealthScan._build_ms but with a pluggable + Event class so this suite can prove the tick path never awaits it.""" + import multiprocessing as mp + + from autoptz.ui.engine_client import EngineClient + + _AssertNoBlockingWaitEvent.blocking_wait_calls.clear() + monkeypatch.setenv("AUTOPTZ_MODEL_SERVER", "1") + client = EngineClient() + cid = client.addCamera("usb://0", "X") + client.drain_commands() + + factory_log: list[_MsFakeProcessWorker] = [] + + def factory(camera_id, config, on_tel): # noqa: ANN001 + w = _MsFakeProcessWorker(camera_id, config, on_tel) + factory_log.append(w) + return w + + sup = _make_sup_with_factory(client, factory) + sup._ensure_identity_service = lambda: None # type: ignore[method-assign] + sup._ensure_inference_pool = lambda: None # type: ignore[method-assign] + + real_ctx = mp.get_context("spawn") + proc_log: list[_ControllableModelServerProc] = [] + + class _FakeCtx: + Queue = staticmethod(real_ctx.Queue) + Event = staticmethod(event_cls) + + def Process(self, *_a, **_k): # noqa: ANN002, ANN003, N802 + p = _ControllableModelServerProc() + proc_log.append(p) + return p + + monkeypatch.setattr(mp, "get_context", lambda *_a, **_k: _FakeCtx()) + monkeypatch.setattr( + "autoptz.engine.supervisor.Supervisor._ensure_model_server", + lambda self, camera_ids: _fake_ensure_model_server(self, camera_ids, proc_log), + ) + sup.start() + return sup, client, factory_log, proc_log, cid + + def test_tick_path_never_calls_blocking_wait(self, qapp, monkeypatch) -> None: + """RED against fc6eee1: the old _respawn_model_server called + ready.wait(timeout=30.0) synchronously from the scan. With the fix, the + scan only starts the child and polls is_set() on later ticks, so no + nonzero-timeout wait() call is ever recorded. + + Asserts on the event's call recorder rather than letting wait() raise, + because the old code wraps the whole respawn in a broad + ``except Exception`` that would otherwise swallow an in-wait assertion + and produce a false-negative RED.""" + sup, client, factory_log, proc_log, cid = self._build_ms(qapp, monkeypatch) + try: + proc_log[0]._alive = False + sup._scan_model_server_health(1000.0) + assert len(proc_log) == 2 # respawn attempt was started + assert _AssertNoBlockingWaitEvent.blocking_wait_calls == [] + finally: + sup.stop() + + def test_spawn_pending_tick_is_a_noop(self, qapp, monkeypatch) -> None: + """While a respawn is in flight and not yet at its deadline, subsequent + scan ticks must not start another child or touch restart state.""" + sup, client, factory_log, proc_log, cid = self._build_ms(qapp, monkeypatch) + try: + proc_log[0]._alive = False + now = 1000.0 + sup._scan_model_server_health(now) + assert len(proc_log) == 2 + assert sup._ms_spawn_deadline == now + _MS_SPAWN_TIMEOUT_S + + # Well before the deadline — should be a pure no-op. + sup._scan_model_server_health(now + 1.0) + assert len(proc_log) == 2 # no new spawn + assert sup._ms_spawn_deadline == now + _MS_SPAWN_TIMEOUT_S # unchanged + assert sup._ms_restart_state == (1, now + _BASE_BACKOFF_S, False) + finally: + sup.stop() + + def test_ready_on_later_tick_succeeds(self, qapp, monkeypatch) -> None: + """Ready-signal arriving on a LATER tick (not the same one that started + the spawn) must clear the gate and reset attempts, mirroring the old + synchronous-success semantics — just detected asynchronously.""" + sup, client, factory_log, proc_log, cid = self._build_ms(qapp, monkeypatch) + try: + proc_log[0]._alive = False + now = 1000.0 + sup._scan_model_server_health(now) + assert sup._ms_spawn_deadline is not None + assert sup._model_server_down.is_set() is True # fast-fail gate held + + # Simulate the child signalling ready sometime later, well before the + # spawn deadline. + sup._model_server_proc._alive = True + ready_ev = sup._ms_ready_ev + ready_ev.set() + + sup._scan_model_server_health(now + 2.0) + + assert sup._ms_spawn_deadline is None + assert sup._ms_restart_state == (0, 0.0, False) + assert sup._model_server_down.is_set() is False # gate cleared + finally: + sup.stop() + + def test_deadline_exceeded_counts_as_failed_attempt_with_backoff( + self, qapp, monkeypatch + ) -> None: + """If the child never signals ready before the spawn deadline, that tick + must count as a failed attempt (existing backoff/budget accounting) and + kill the stuck child — not hang waiting for it.""" + sup, client, factory_log, proc_log, cid = self._build_ms(qapp, monkeypatch) + try: + proc_log[0]._alive = False + now = 1000.0 + sup._scan_model_server_health(now) + assert sup._ms_spawn_deadline == now + _MS_SPAWN_TIMEOUT_S + stuck_proc = sup._model_server_proc + + # Never signals ready; advance past the spawn deadline. + sup._scan_model_server_health(now + _MS_SPAWN_TIMEOUT_S + 0.1) + + assert sup._ms_spawn_deadline is None + assert stuck_proc.terminate_calls == 1 # stuck child was killed + attempts, next_allowed_t, failed = sup._ms_restart_state + # The attempt was already counted (attempts=1) when the respawn was + # initiated — mirrors the old synchronous code, which incremented + # before the (blocking) wait. A timed-out spawn does not double-count. + assert attempts == 1 + assert next_allowed_t > now + _MS_SPAWN_TIMEOUT_S # backoff window set + assert failed is False + assert sup._model_server_down.is_set() is True # gate still held + finally: + sup.stop() + + def test_gate_stays_set_across_spawning_and_backoff_window(self, qapp, monkeypatch) -> None: + """The down-gate must remain set continuously from the moment the server + is first found dead through spawning AND the subsequent backoff wait — + clients must keep fast-failing the whole time, never just during one + phase.""" + sup, client, factory_log, proc_log, cid = self._build_ms(qapp, monkeypatch) + try: + proc_log[0]._alive = False + now = 1000.0 + sup._scan_model_server_health(now) + assert sup._model_server_down.is_set() is True # spawning phase + + # Deadline exceeded -> failed attempt -> now in backoff phase. + now += _MS_SPAWN_TIMEOUT_S + 0.1 + sup._scan_model_server_health(now) + assert sup._ms_spawn_deadline is None + assert sup._model_server_down.is_set() is True # backoff phase + + # Still inside the backoff window — must stay a no-op, gate still set. + now += 0.1 + sup._scan_model_server_health(now) + assert sup._model_server_down.is_set() is True + finally: + sup.stop() + + def test_deadline_exceeded_at_budget_exhaustion_triggers_fallback( + self, qapp, monkeypatch + ) -> None: + """A spawn that times out on the FINAL allowed attempt must exhaust the + budget exactly like the old synchronous failure path did: latch + model_server_failed() and fire the worker fallback.""" + sup, client, factory_log, proc_log, cid = self._build_ms(qapp, monkeypatch) + try: + now = 1000.0 + proc_log[0]._alive = False + for _attempt in range(_MAX_RESTART_ATTEMPTS - 1): + sup._scan_model_server_health(now) + now += _MS_SPAWN_TIMEOUT_S + 0.1 # spawn never signals ready + sup._scan_model_server_health(now) # deadline exceeded -> failed + now += _MAX_BACKOFF_S + 1.0 # clear backoff before next attempt + if sup._model_server_proc is not None: + sup._model_server_proc._alive = False + + assert sup.model_server_failed() is False # not yet — one more to go + + sup._scan_model_server_health(now) # final attempt starts spawning + now += _MS_SPAWN_TIMEOUT_S + 0.1 + sup._scan_model_server_health(now) # final deadline exceeded -> exhausted + + assert sup.model_server_failed() is True + assert factory_log[0].refresh_calls == 1 + assert sup._model_server_down.is_set() is True + finally: + sup.stop() + + def _fake_ensure_model_server(sup, camera_ids, proc_log) -> None: # noqa: ANN001 """Deterministic stand-in for Supervisor._ensure_model_server: builds the same queue/event handles but skips the real ready-event wait (the fake process never