fix(camera-api): stop serving frozen frames when a patrol hangs

docker-compose.yml

@@ -17,12 +17,18 @@ services:
       - ./data:/usr/src/app/data
     restart: always
     network_mode: host
+    # tini as PID 1, so the healthcheck's kill -9 -1 also reaches uvicorn
+    init: true
     healthcheck:
-      test: ["CMD-SHELL", "curl -sf http://localhost:8081/health || exit 1"]
+      # Docker never restarts an unhealthy container by itself: on a hung patrol (503) or an API
+      # that stopped answering, kill everything but PID 1 so `restart: always` brings it back.
+      # curl retries twice so a single slow answer does not restart it. Connection refused (rc 7)
+      # only means uvicorn is still starting, as it binds after camera setup.
+      test: ["CMD-SHELL", "curl -sf --max-time 5 --retry 2 --retry-delay 5 http://localhost:8081/health/patrols; rc=$$?; [ $$rc -eq 0 ] && exit 0; [ $$rc -eq 7 ] && exit 1; kill -9 -1"]
       interval: 30s
       retries: 3
       start_period: 20s
-      timeout: 5s
+      timeout: 30s
     logging:
       driver: json-file
       options:

pyro_camera_api/pyro_camera_api/api/routes_cameras.py

@@ -238,21 +238,28 @@ def get_latest_image(
         le=100,
         description="JPEG quality that controls compression level. Higher means better quality and larger file size.",
     ),
+    max_age_s: int = Query(
+        default=300,
+        ge=1,
+        description="Images stored longer ago than this are treated as missing.",
+    ),
 ):
     """
     Return the last stored image for a given camera and pose.
 
     The camera adapter may cache captured frames per preset pose in its
     last_images mapping. This endpoint exposes that cache. If there is
-    no image for the requested pose, the endpoint returns HTTP 204 with
+    no image for the requested pose, or it is older than max_age_s (for
+    example a hung patrol thread), the endpoint returns HTTP 204 with
     an empty body.
     """
     cam = CAMERA_REGISTRY.get(camera_ip)
     if cam is None:
         raise HTTPException(status_code=404, detail="Unknown camera")
 
     image = cam.last_images.get(pose)
-    if image is None:
+    stored_at = cam.last_image_at.get(pose)
+    if image is None or stored_at is None or time.monotonic() - stored_at > max_age_s:
         return Response(status_code=status.HTTP_204_NO_CONTENT)
 
     buffer = BytesIO()

pyro_camera_api/pyro_camera_api/api/routes_health.py

@@ -7,6 +7,9 @@
 from __future__ import annotations
 
 from fastapi import APIRouter
+from fastapi.responses import JSONResponse
+
+from pyro_camera_api.camera.patrol import stalled_patrols
 
 router = APIRouter()
 
@@ -16,7 +19,20 @@ def health():
     """
     Return a basic heartbeat response to confirm that the API is running.
 
-    This endpoint does not perform any camera or service check.
-    It is intended for uptime monitoring and infrastructure-level health probes.
+    Kept free of camera checks: the Shelly watchdog reads it as Pi liveness and
+    power-cycles the Pi and the router when it fails.
+    """
+    return {"status": "ok"}
+
+
+@router.get("/health/patrols")
+def health_patrols():
+    """
+    Return 200 while every running patrol makes progress, 503 with the hung cameras otherwise.
+
+    Used by the compose healthcheck, which restarts the container on a 503.
     """
+    stalled = stalled_patrols()
+    if stalled:
+        return JSONResponse(status_code=503, content={"status": "patrol_stalled", "cameras": stalled})
     return {"status": "ok"}

pyro_camera_api/pyro_camera_api/camera/adapters/reolink.py

@@ -30,7 +30,19 @@
 logger = logging.getLogger(__name__)
 
 
-class _LegacyTLSAdapter(HTTPAdapter):
+# A PTZ call without timeout once hung a patrol thread forever (pyro-engine#422):
+# every request gets this unless the call site passes its own.
+_DEFAULT_TIMEOUT = 10
+
+
+class _TimeoutAdapter(HTTPAdapter):
+    def send(self, request: Any, *args: Any, **kwargs: Any) -> Any:
+        if kwargs.get("timeout") is None:
+            kwargs["timeout"] = _DEFAULT_TIMEOUT
+        return super().send(request, *args, **kwargs)
+
+
+class _LegacyTLSAdapter(_TimeoutAdapter):
     """Old Reolink firmwares only offer RSA-key-exchange TLS ciphers, which
     Python 3.10+ dropped from its default cipher list: restore OpenSSL's own."""
 
@@ -51,6 +63,7 @@ def init_poolmanager(self, *args: Any, **kwargs: Any) -> None:
 # hands back a stale socket (matches the per-call behaviour of requests.get).
 _session.headers["Connection"] = "close"
 _session.mount("https://", _LegacyTLSAdapter())
+_session.mount("http://", _TimeoutAdapter())
 
 # Zoom/focus reads and the zoom restore run under the stream lock in
 # routes_stream, so a camera that stops answering must not hang them forever.

pyro_camera_api/pyro_camera_api/camera/base.py

@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+import time
 from abc import ABC, abstractmethod
 from typing import Dict, List, Optional
 
@@ -36,6 +37,12 @@ def __init__(self, camera_id: str, cam_type: str = "static") -> None:
         # PTZ cameras can use pose -> image
         # Static cameras can use -1 -> image
         self.last_images: Dict[int, Optional[Image.Image]] = {}
+        # time.monotonic() of each stored image, so stale frames can be refused
+        self.last_image_at: Dict[int, float] = {}
+
+    def store_image(self, pose: int, image: Image.Image) -> None:
+        self.last_images[pose] = image
+        self.last_image_at[pose] = time.monotonic()
 
     @abstractmethod
     def capture(self, **kwargs) -> Optional[Image.Image]:

pyro_camera_api/pyro_camera_api/camera/patrol.py

@@ -9,9 +9,9 @@
 import logging
 import threading
 import time
-from typing import Any, Dict
+from typing import Any, Dict, List
 
-from pyro_camera_api.camera.registry import CAMERA_REGISTRY
+from pyro_camera_api.camera.registry import CAMERA_REGISTRY, PATROL_PROGRESS, PATROL_THREADS
 from pyro_camera_api.services.stream import is_camera_streaming
 
 logger = logging.getLogger(__name__)
@@ -23,6 +23,10 @@
 # seconds between stream re-checks while the patrol is paused by a live stream
 STREAM_CHECK_INTERVAL = 5.0
 
+# A loop iteration takes well under a minute even when every camera call times
+# out, so this much silence from a live patrol thread means it is hung.
+STALL_LIMIT = 10 * 60.0
+
 # per camera state for the static loop
 FAILURE_COUNT: Dict[str, int] = {}
 SKIP_UNTIL: Dict[str, float] = {}
@@ -68,7 +72,17 @@ def is_stream_running_for(app: Any, camera_ip: str) -> bool:
     return False
 
 
+def stalled_patrols() -> List[str]:
+    now = time.monotonic()
+    return [
+        ip
+        for ip, thread in PATROL_THREADS.items()
+        if thread.is_alive() and now - PATROL_PROGRESS.get(ip, now) > STALL_LIMIT
+    ]
+
+
 def patrol_loop(camera_ip: str, stop_flag: threading.Event) -> None:
+    PATROL_PROGRESS[camera_ip] = time.monotonic()
     cam = CAMERA_REGISTRY[camera_ip]
     poses = getattr(cam, "cam_poses", []) or []
 
@@ -79,6 +93,7 @@ def patrol_loop(camera_ip: str, stop_flag: threading.Event) -> None:
     logger.info("[%s] Starting patrol cycle with %d poses", camera_ip, len(poses))
 
     while not stop_flag.is_set():
+        PATROL_PROGRESS[camera_ip] = time.monotonic()
         # Live stream has priority: hold the patrol while a stream is active
         # so the camera never moves under the viewer, and resume automatically
         # once the stream stops (the thread keeps running, patrol stays "on").
@@ -99,7 +114,7 @@ def patrol_loop(camera_ip: str, stop_flag: threading.Event) -> None:
 
                 image = cam.capture()
                 if image:
-                    cam.last_images[pose] = image
+                    cam.store_image(pose, image)
                     logger.info("[%s] Stored image for pose %s", camera_ip, pose)
 
             except Exception as exc:
@@ -133,6 +148,7 @@ def patrol_loop(camera_ip: str, stop_flag: threading.Event) -> None:
 
 
 def static_loop(camera_ip: str, stop_flag: threading.Event) -> None:
+    PATROL_PROGRESS[camera_ip] = time.monotonic()
     cam = CAMERA_REGISTRY[camera_ip]
     logger.info("[%s] Starting static camera loop", camera_ip)
 
@@ -143,6 +159,7 @@ def static_loop(camera_ip: str, stop_flag: threading.Event) -> None:
     settle_until = 0.0
 
     while not stop_flag.is_set():
+        PATROL_PROGRESS[camera_ip] = time.monotonic()
         now = time.time()
 
         # skip window
@@ -161,7 +178,7 @@ def static_loop(camera_ip: str, stop_flag: threading.Event) -> None:
                     settle_until = opened_at + 1.0
 
                 if image and now >= settle_until:
-                    cam.last_images[-1] = image
+                    cam.store_image(-1, image)
                     logger.info("[%s] Updated static image (pose -1)", camera_ip)
                     # success reset failure counter and clear skip
                     FAILURE_COUNT[camera_ip] = 0

pyro_camera_api/pyro_camera_api/camera/registry.py

@@ -55,6 +55,8 @@ def repl(match: re.Match[str]) -> str:
 # Patrol threading state, later managed in camera.patrol
 PATROL_THREADS: Dict[str, threading.Thread] = {}
 PATROL_FLAGS: Dict[str, threading.Event] = {}
+# time.monotonic() of the last loop iteration of each patrol thread
+PATROL_PROGRESS: Dict[str, float] = {}
 
 # Stuck-detector threading state, later managed in camera.stuck_detector
 STUCK_CHECK_THREADS: Dict[str, threading.Thread] = {}

pyro_camera_api/tests/test_reolink_tls.py

@@ -3,6 +3,9 @@
 # This program is licensed under the Apache License 2.0.
 # See LICENSE or go to <https://opensource.org/licenses/Apache-2.0> for full license details.
 
+import pytest
+import requests
+
 from pyro_camera_api.camera.adapters.reolink import _session
 
 
@@ -15,3 +18,19 @@ def test_https_session_offers_rsa_key_exchange_ciphers():
     assert "AES128-GCM-SHA256" in names
     # Proxy env vars must not reroute camera calls around the TLS adapter
     assert not _session.trust_env
+
+
+def test_session_applies_default_timeout(monkeypatch):
+    # A call without timeout must not wait forever on a silent camera (pyro-engine#422)
+    seen = []
+
+    def fake_send(_self, _request, **kwargs):
+        seen.append(kwargs["timeout"])
+        raise requests.ConnectionError
+
+    monkeypatch.setattr("requests.adapters.HTTPAdapter.send", fake_send)
+    for url in ("https://192.168.1.10", "http://192.168.1.10"):
+        for kwargs in ({}, {"timeout": 2}):
+            with pytest.raises(requests.ConnectionError):
+                _session.post(url, **kwargs)
+    assert seen == [10, 2, 10, 2]

pyro_camera_api/tests/test_stale_patrol.py

@@ -0,0 +1,63 @@
+# Copyright (C) 2022-2026, Pyronear.
+
+# This program is licensed under the Apache License 2.0.
+# See LICENSE or go to <https://opensource.org/licenses/Apache-2.0> for full license details.
+
+import threading
+import time
+
+from PIL import Image
+
+from pyro_camera_api.api import routes_health
+from pyro_camera_api.api.routes_cameras import get_latest_image
+from pyro_camera_api.api.routes_health import health, health_patrols
+from pyro_camera_api.camera import patrol
+from pyro_camera_api.camera.adapters.mock import MockCamera
+from pyro_camera_api.camera.registry import CAMERA_REGISTRY
+
+
+def test_latest_image_refuses_stale_frame(monkeypatch):
+    # A hung patrol thread leaves an old frame in the cache: it must not be served
+    cam_id = "stale-cam"
+    cam = MockCamera(camera_id=cam_id, cam_type="ptz", cam_poses=[0])
+    CAMERA_REGISTRY[cam_id] = cam
+    try:
+        assert get_latest_image(cam_id, 0, quality=95, max_age_s=300).status_code == 204
+
+        cam.store_image(0, Image.new("RGB", (8, 8)))
+        assert get_latest_image(cam_id, 0, quality=95, max_age_s=300).media_type == "image/jpeg"
+
+        monkeypatch.setattr("time.monotonic", lambda: cam.last_image_at[0] + 301)
+        assert get_latest_image(cam_id, 0, quality=95, max_age_s=300).status_code == 204
+    finally:
+        CAMERA_REGISTRY.pop(cam_id, None)
+
+
+def test_watchdog_flags_only_hung_patrol_threads(monkeypatch):
+    # A live thread silent for too long is hung, a finished one is not
+    hung, done = threading.Event(), threading.Event()
+    threads = {
+        "hung": threading.Thread(target=hung.wait, daemon=True),
+        "busy": threading.Thread(target=hung.wait, daemon=True),
+        "done": threading.Thread(target=done.set),
+    }
+    for thread in threads.values():
+        thread.start()
+    threads["done"].join()
+    monkeypatch.setattr(patrol, "PATROL_THREADS", threads)
+    now = time.monotonic()
+    monkeypatch.setattr(
+        patrol, "PATROL_PROGRESS", {"hung": now - patrol.STALL_LIMIT - 1, "busy": now, "done": now - 3600}
+    )
+    try:
+        assert patrol.stalled_patrols() == ["hung"]
+    finally:
+        hung.set()
+
+
+def test_health_patrols_reports_stalled_patrols(monkeypatch):
+    assert health_patrols() == {"status": "ok"}
+    monkeypatch.setattr(routes_health, "stalled_patrols", lambda: ["192.168.1.12"])
+    assert health_patrols().status_code == 503
+    # The Shelly reads /health as Pi liveness: a hung patrol must not power-cycle the Pi
+    assert health() == {"status": "ok"}