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,