add docstrings

When the detector lost an object, the tracker kept it at its last box and every held frame still counted toward the zone's inertia. A car detected in a zone once or twice as it drove out of view could reach an inertia of 8 and be marked as entering, which sent it to GenAI and any other required zones check even though Review never showed the zone. Inertia and loitering still count on frames without a detection, but an object can only newly enter a zone on a frame where it was actually detected.

frigate/comms/ws.py

@@ -3,6 +3,9 @@
 import errno
 import json
 import logging
+import queue
+import socket
+import struct
 import threading
 from collections.abc import Callable
 from typing import Any
@@ -76,6 +79,12 @@ _WS_VIEWER_TOPICS = frozenset(
 # Camera-scoped command topics a camera-authorized (non-admin) user may send.
 _WS_CAMERA_COMMAND_TOPICS = frozenset({"ptz"})
 
+# Max outbound messages waiting on a client's writer thread.
+WS_MAX_PENDING_MESSAGES = 256
+
+# Seconds a single send() may wait for a client to free socket buffer space.
+WS_SEND_TIMEOUT = 10
+
 
 def _check_ws_authorization(
     topic: str,
@@ -464,6 +473,78 @@ def _materialize_for_ws(
 
 
 class WebSocket(WebSocket_):  # type: ignore[misc]
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self._send_queue: queue.Queue[tuple[Any, bool] | None] = queue.Queue(
+            maxsize=WS_MAX_PENDING_MESSAGES
+        )
+        self._writer: threading.Thread | None = None
+        self._aborted = False
+
+    def opened(self) -> None:
+        """Set a send timeout on the socket and start this client's writer thread."""
+
+        # ws4py writes close and pong frames on the shared manager thread, so a
+        # stalled client must not block sendall forever. SO_SNDTIMEO leaves recv
+        # (and idle clients) untouched, unlike settimeout().
+        self.sock.setsockopt(
+            socket.SOL_SOCKET, socket.SO_SNDTIMEO, struct.pack("ll", WS_SEND_TIMEOUT, 0)
+        )
+
+        # every client gets its own writer so a client that stops reading only
+        # blocks itself, never the thread that called publish()
+        self._writer = threading.Thread(
+            target=self._drain_send_queue, name="ws_writer", daemon=True
+        )
+        self._writer.start()
+
+    def send(self, payload: Any, binary: bool = False) -> None:
+        """Queue a message for this client's writer thread.
+
+        Returns without waiting for delivery. Disconnects the client if its queue
+        is full.
+        """
+        try:
+            self._send_queue.put_nowait((payload, binary))
+        except queue.Full:
+            self._abort("Websocket client is not keeping up, disconnecting it")
+
+    def _drain_send_queue(self) -> None:
+        while True:
+            item = self._send_queue.get()
+            if item is None or self.terminated or self.sock is None:
+                return
+            try:
+                super().send(*item)
+            except Exception:
+                self._abort()
+                return
+
+    def _abort(self, reason: str | None = None) -> None:
+        # publish() keeps hitting a full queue until the manager thread removes
+        # the connection, so only act (and log) the first time
+        if self._aborted:
+            return
+        self._aborted = True
+        if reason:
+            logger.warning(reason)
+
+        # shutdown rather than close so the ws4py manager thread sees EOF and
+        # runs its normal unregister/terminate; this also unblocks a stuck sendall
+        sock = self.sock
+        if sock is not None:
+            try:
+                sock.shutdown(socket.SHUT_RDWR)
+            except OSError:
+                pass
+
+    def closed(self, code: int, reason: str | None = None) -> None:
+        """Signal the writer thread to exit."""
+        try:
+            self._send_queue.put_nowait(None)
+        except queue.Full:
+            pass
+
     def unhandled_error(self, error: Any) -> None:
         """
         Handles the unfriendly socket closures on the server side
@@ -597,10 +678,7 @@ class WebSocketClient(Communicator):
             )
             if message is None:
                 continue
-            try:
             ws.send(message)
-            except (ConnectionResetError, BrokenPipeError, ValueError):
-                pass
 
     def stop(self) -> None:
         if self.websocket_server is not None:

frigate/output/birdseye.py

@@ -277,12 +277,7 @@ class BroadcastThread(threading.Thread):
                         and ws.environ["PATH_INFO"] == f"/{self.camera}"
                         and ws_has_camera_access(ws, self.camera, self.config)
                     ):
-                        try:
                         ws.send(buf, binary=True)
-                        except ValueError:
-                            pass
-                        except (BrokenPipeError, ConnectionResetError, OSError) as e:
-                            logger.debug(f"Websocket unexpectedly closed {e}")
             elif self.converter.process.poll() is not None:
                 break
 

frigate/output/camera.py

@@ -127,12 +127,7 @@ class BroadcastThread(threading.Thread):
                         and ws.environ["PATH_INFO"] == f"/{self.camera}"
                         and ws_has_camera_access(ws, self.camera, self.config)
                     ):
-                        try:
                         ws.send(buf, binary=True)
-                        except ValueError:
-                            pass
-                        except (BrokenPipeError, ConnectionResetError) as e:
-                            logger.debug(f"Websocket unexpectedly closed {e}")
             elif self.converter.process.poll() is not None:
                 break
 

frigate/test/test_tracked_object_zones.py

@@ -0,0 +1,120 @@
+"""Zone entry for tracked objects the detector has lost."""
+
+import unittest
+
+from frigate.config import FrigateConfig
+from frigate.track.tracked_object import TrackedObject
+
+CAMERA = "cam"
+FPS = 5
+INERTIA = 3
+OUTSIDE_X = 200
+INSIDE_X = 480
+
+
+def _config() -> FrigateConfig:
+    return FrigateConfig(
+        **{
+            "mqtt": {"enabled": False},
+            "cameras": {
+                CAMERA: {
+                    "ffmpeg": {
+                        "inputs": [
+                            {"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]}
+                        ]
+                    },
+                    "detect": {"width": 640, "height": 480, "fps": FPS},
+                    "objects": {"track": ["car"]},
+                    "zones": {
+                        "zone": {
+                            "coordinates": "0.5,0,1,0,1,1,0.5,1",
+                            "inertia": INERTIA,
+                        }
+                    },
+                }
+            },
+        }
+    )
+
+
+def _obj_data(x: int, frame_time: float) -> dict:
+    return {
+        "id": "1000.0-abcdef",
+        "label": "car",
+        "frame_time": frame_time,
+        "start_time": 1000.0,
+        "score": 0.9,
+        "box": (x - 40, 200, x + 40, 260),
+        "area": 4800,
+        "ratio": 1.33,
+        "region": (0, 0, 640, 480),
+        "centroid": (x, 230),
+        "position_changes": 1,
+        "motionless_count": 0,
+        "attributes": [],
+    }
+
+
+class TestTrackedObjectZones(unittest.TestCase):
+    def setUp(self):
+        config = _config()
+        camera_config = config.cameras[CAMERA]
+        self.obj = TrackedObject(
+            config.model_for_camera(CAMERA),
+            camera_config,
+            config.ui,
+            {},
+            {**_obj_data(OUTSIDE_X, 1000.0), "score_history": [0.9]},
+        )
+        self.frame = 0
+
+    def _step(self, x: int, detected: bool = True) -> None:
+        """Advance one frame. A lost object keeps the box from its last detection."""
+        self.frame += 1
+        current_frame_time = 1000.0 + self.frame / FPS
+
+        if detected:
+            self.last_detection = _obj_data(x, current_frame_time)
+
+        self.obj.update(current_frame_time, dict(self.last_detection), False)
+
+    def test_lost_object_does_not_enter_zone(self):
+        self._step(INSIDE_X)
+
+        for _ in range(INERTIA * 3):
+            self._step(INSIDE_X, detected=False)
+
+        self.assertEqual(self.obj.current_zones, [])
+        self.assertEqual(self.obj.entered_zones, [])
+
+    def test_lost_object_enters_zone_when_detected_again(self):
+        self._step(INSIDE_X)
+
+        for _ in range(INERTIA):
+            self._step(INSIDE_X, detected=False)
+
+        self._step(INSIDE_X)
+        self.assertEqual(self.obj.entered_zones, ["zone"])
+
+    def test_detected_object_enters_zone(self):
+        for _ in range(INERTIA - 1):
+            self._step(INSIDE_X)
+
+        self.assertEqual(self.obj.entered_zones, [])
+
+        self._step(INSIDE_X)
+        self.assertEqual(self.obj.current_zones, ["zone"])
+        self.assertEqual(self.obj.entered_zones, ["zone"])
+
+    def test_object_in_zone_stays_while_lost(self):
+        for _ in range(INERTIA):
+            self._step(INSIDE_X)
+
+        for _ in range(INERTIA):
+            self._step(INSIDE_X, detected=False)
+
+        self.assertEqual(self.obj.current_zones, ["zone"])
+
+
+if __name__ == "__main__":
+    unittest.main()

frigate/test/test_ws_send_queue.py

@@ -0,0 +1,131 @@
+"""Outbound websocket sends must never block the thread calling publish()."""
+
+import socket
+import threading
+import unittest
+
+from frigate.comms.ws import WS_MAX_PENDING_MESSAGES, WebSocket
+
+
+class _FakeSock:
+    """Socket stand-in; ``block`` makes sendall hang like a client that stopped reading."""
+
+    def __init__(self, block: bool = False) -> None:
+        self.block = block
+        self.released = threading.Event()
+        self.shutdown_called = threading.Event()
+        self.frames: list[bytes] = []
+
+    def sendall(self, data: bytes) -> None:
+        if self.shutdown_called.is_set():
+            raise BrokenPipeError()
+        if self.block and not self.released.is_set():
+            self.released.wait(timeout=10)
+            raise BrokenPipeError()
+        self.frames.append(data)
+
+    def shutdown(self, how: int) -> None:
+        assert how == socket.SHUT_RDWR
+        self.shutdown_called.set()
+        self.released.set()
+
+    def close(self) -> None:
+        pass
+
+    def setsockopt(self, *args) -> None:
+        pass
+
+    def fileno(self) -> int:
+        return 99
+
+
+def _wait_for(predicate, timeout: float = 2.0) -> bool:
+    deadline = threading.Event()
+    for _ in range(int(timeout / 0.01)):
+        if predicate():
+            return True
+        deadline.wait(0.01)
+    return predicate()
+
+
+class TestWebSocketSendQueue(unittest.TestCase):
+    def _open(self, sock: _FakeSock) -> WebSocket:
+        ws = WebSocket(sock)
+        ws.opened()
+        return ws
+
+    def test_stalled_client_does_not_block_publisher(self):
+        sock = _FakeSock(block=True)
+        ws = self._open(sock)
+
+        def publish_many():
+            for i in range(WS_MAX_PENDING_MESSAGES + 5):
+                ws.send(f"message {i}")
+
+        publisher = threading.Thread(target=publish_many, daemon=True)
+        publisher.start()
+        publisher.join(timeout=2.0)
+
+        self.assertFalse(publisher.is_alive(), "publish() blocked on a stalled client")
+        self.assertTrue(
+            sock.shutdown_called.wait(timeout=2.0),
+            "a client that cannot keep up should be disconnected",
+        )
+        ws._writer.join(2)
+        self.assertFalse(ws._writer.is_alive())
+
+    def test_overflow_warns_and_shuts_down_once(self):
+        sock = _FakeSock(block=True)
+        ws = self._open(sock)
+        shutdown_calls = []
+        original_shutdown = sock.shutdown
+        sock.shutdown = lambda how: (shutdown_calls.append(how), original_shutdown(how))
+
+        with self.assertLogs("frigate.comms.ws", level="WARNING") as logs:
+            # keep publishing after overflow, as the dispatcher does until the
+            # manager thread removes the connection
+            for i in range(WS_MAX_PENDING_MESSAGES * 3):
+                ws.send(f"message {i}")
+
+        self.assertEqual(len(logs.output), 1)
+        self.assertEqual(len(shutdown_calls), 1)
+
+    def test_messages_delivered_in_order(self):
+        sock = _FakeSock()
+        ws = self._open(sock)
+        for i in range(3):
+            ws.send(f"message {i}")
+
+        self.assertTrue(_wait_for(lambda: len(sock.frames) == 3))
+        for i, frame in enumerate(sock.frames):
+            self.assertIn(f"message {i}".encode(), frame)
+        self.assertFalse(sock.shutdown_called.is_set())
+
+    def test_closed_stops_writer_thread(self):
+        sock = _FakeSock()
+        ws = self._open(sock)
+        writer = ws._writer
+        ws.closed(1000, "bye")
+        writer.join(timeout=2.0)
+        self.assertFalse(writer.is_alive())
+
+    def test_closed_with_full_queue_and_blocked_writer_exits_after_shutdown(self):
+        sock = _FakeSock(block=True)
+        ws = self._open(sock)
+        ws.send("in flight")
+        self.assertTrue(_wait_for(ws._send_queue.empty))
+        for i in range(WS_MAX_PENDING_MESSAGES):
+            ws.send(f"message {i}")
+        self.assertTrue(ws._send_queue.full())
+
+        ws.closed(1006)
+        self.assertTrue(ws._writer.is_alive())
+        self.assertFalse(sock.shutdown_called.is_set())
+
+        sock.shutdown(socket.SHUT_RDWR)
+        ws._writer.join(2)
+        self.assertFalse(ws._writer.is_alive())
+
+
+if __name__ == "__main__":
+    unittest.main()

frigate/track/tracked_object.py

@@ -186,6 +186,7 @@ class TrackedObject:
         # check zones
         current_zones = []
         bottom_center = (obj_data["centroid"][0], obj_data["box"][3])
+        detected = obj_data["frame_time"] == current_frame_time
         in_loitering_zone = False
         in_speed_zone = False
 
@@ -275,11 +276,13 @@ class TrackedObject:
 
                         loitering_score = self.zone_loitering.get(name, 0) + 1
 
-                        # loitering time is configured as seconds, convert to count of frames
+                        # loitering time is configured as seconds, convert to count of frames.
+                        # a lost object keeps counting but can't newly enter a zone
+                        # until it is detected again
                         if loitering_score >= (
                             self.camera_config.zones[name].loitering_time
                             * self.camera_config.detect.fps
-                        ):
+                        ) and (detected or name in self.current_zones):
                             current_zones.append(name)
 
                             if name not in self.entered_zones:

web/public/locales/en/views/system.json

@@ -35,8 +35,8 @@
       "showAll": "Show all again",
       "showAllTitle": "Show all hidden notices?",
       "showAllDesc": "Every acknowledged and muted notice returns to the Notices list, including config and stream checks.",
-      "firstSeen_one": "First seen {{time}}",
-      "firstSeen_other": "First seen {{time}} · {{count}} times",
+      "occurred_one": "Occurred {{time}}",
+      "occurred_other": "Last occurred {{lastTime}} · {{count}} times since {{time}}",
       "acknowledgedAt": "Acknowledged {{time}}",
       "mutedAt": "Muted {{time}}",
       "filter": {

web/src/components/overlay/detail/DetailActionsMenu.tsx

@@ -55,6 +55,10 @@ export default function DetailActionsMenu({
   // currently, audio event ids are not saved in review items
   const { data: reviewItem } = useSWR<ReviewSegment>(
     search.data?.type === "audio" ? null : [`review/event/${search.id}`],
+    {
+      shouldRetryOnError: (err) =>
+        !(axios.isAxiosError(err) && err.response?.status === 404),
+    },
   );
 
   const handleDebugReplay = useCallback(() => {

web/src/hooks/use-health-problems.ts

@@ -118,9 +118,10 @@ export function useHealthProblems(
                   ns: "views/system",
                   time: formatTime(notice.acknowledged_at),
                 })
-              : t("health.notices.firstSeen", {
+              : t("health.notices.occurred", {
                   ns: "views/system",
                   time: formatTime(notice.first_seen),
+                  lastTime: formatTime(notice.last_seen),
                   count: notice.count,
                 }),
         link: external ? undefined : link,