From 64b07889af9a1c5253cc2ac96e6b7f156e813d0a Mon Sep 17 00:00:00 2001 From: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com> Date: Thu, 15 Oct 2026 07:54:59 -0500 Subject: [PATCH] Tweaks and fixes (#24659) * add last time to active notices * Stop retrying the review lookup in the explore detail menu when it 404s * don't let a slow websocket client block publishers A websocket client that stopped reading its socket (for example a suspended workstation with a Frigate tab open) blocked ws4py's sendall on whatever thread was publishing, including the single inter-process responder thread, so the recording maintainer and stats hung until nginx dropped the client after 5 minutes. Each client now gets its own bounded send queue and writer thread, and a client that overflows the queue is shut down. ws4py writes close and pong frames directly on its shared manager thread, bypassing the queue, so a 10s socket send timeout keeps a stalled client from pinning that thread. The jsmpeg and birdseye outputs share the class, so their unreachable send error handling is removed. * don't let a lost object enter a zone 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. * add test * limit retry stop to 404 only * add docstrings --- frigate/comms/ws.py | 86 +++++++++++- frigate/output/birdseye.py | 7 +- frigate/output/camera.py | 7 +- frigate/test/test_tracked_object_zones.py | 120 ++++++++++++++++ frigate/test/test_ws_send_queue.py | 131 ++++++++++++++++++ frigate/track/tracked_object.py | 7 +- web/public/locales/en/views/system.json | 4 +- .../overlay/detail/DetailActionsMenu.tsx | 4 + web/src/hooks/use-health-problems.ts | 3 +- 9 files changed, 348 insertions(+), 21 deletions(-) create mode 100644 frigate/test/test_tracked_object_zones.py create mode 100644 frigate/test/test_ws_send_queue.py diff --git a/frigate/comms/ws.py b/frigate/comms/ws.py index 348640e2d6..5f9cb96f59 100644 --- a/frigate/comms/ws.py +++ b/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 + ws.send(message) def stop(self) -> None: if self.websocket_server is not None: diff --git a/frigate/output/birdseye.py b/frigate/output/birdseye.py index 4dfbb74172..44799db0f5 100644 --- a/frigate/output/birdseye.py +++ b/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}") + ws.send(buf, binary=True) elif self.converter.process.poll() is not None: break diff --git a/frigate/output/camera.py b/frigate/output/camera.py index 88d16ed4b4..96ff2abbad 100644 --- a/frigate/output/camera.py +++ b/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}") + ws.send(buf, binary=True) elif self.converter.process.poll() is not None: break diff --git a/frigate/test/test_tracked_object_zones.py b/frigate/test/test_tracked_object_zones.py new file mode 100644 index 0000000000..82fab479e8 --- /dev/null +++ b/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() diff --git a/frigate/test/test_ws_send_queue.py b/frigate/test/test_ws_send_queue.py new file mode 100644 index 0000000000..c758488c72 --- /dev/null +++ b/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() diff --git a/frigate/track/tracked_object.py b/frigate/track/tracked_object.py index 75234c3a87..24ae278ac8 100644 --- a/frigate/track/tracked_object.py +++ b/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: diff --git a/web/public/locales/en/views/system.json b/web/public/locales/en/views/system.json index a33253a9ba..fbaf1d7224 100644 --- a/web/public/locales/en/views/system.json +++ b/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": { diff --git a/web/src/components/overlay/detail/DetailActionsMenu.tsx b/web/src/components/overlay/detail/DetailActionsMenu.tsx index 3705ce9baf..1aab728647 100644 --- a/web/src/components/overlay/detail/DetailActionsMenu.tsx +++ b/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( search.data?.type === "audio" ? null : [`review/event/${search.id}`], + { + shouldRetryOnError: (err) => + !(axios.isAxiosError(err) && err.response?.status === 404), + }, ); const handleDebugReplay = useCallback(() => { diff --git a/web/src/hooks/use-health-problems.ts b/web/src/hooks/use-health-problems.ts index 9e493b3b18..70fd0a84e0 100644 --- a/web/src/hooks/use-health-problems.ts +++ b/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,