R-3: Model-server crash recovery — respawn, fast-fail gate, local fallback

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
 
 

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,13 +412,22 @@ 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
         # Model-server IPC handles (model-server mode): the shared request queue and
         # 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,