From 4c05ee567cbf2fc595d4a353ae0a678761c81f2a Mon Sep 17 00:00:00 2001 From: Thomas Munzer Date: Fri, 24 Apr 2026 12:09:14 +0200 Subject: [PATCH 1/8] Improve websocket reliability per Mist best practices --- README.md | 11 +- src/mistapi/websockets/__ws_client.py | 351 ++++++++++++++++++++++++-- src/mistapi/websockets/location.py | 40 +-- src/mistapi/websockets/orgs.py | 24 +- src/mistapi/websockets/session.py | 8 +- src/mistapi/websockets/sites.py | 56 ++-- tests/unit/test_websocket_client.py | 70 ++++- 7 files changed, 465 insertions(+), 95 deletions(-) diff --git a/README.md b/README.md index e488ea8..0d5ec6e 100644 --- a/README.md +++ b/README.md @@ -583,19 +583,22 @@ All channel classes accept the following optional keyword arguments: | Parameter | Type | Default | Description | |-----------|------|---------|-------------| -| `ping_interval` | `int` | `30` | Seconds between automatic ping frames. Set to `0` to disable pings. | -| `ping_timeout` | `int` | `10` | Seconds to wait for a pong response before treating the connection as dead. | +| `ping_interval` | `int` | `60` | Seconds between automatic ping frames. Set to `0` to disable pings. | +| `ping_timeout` | `int` | `45` | Seconds to wait for a pong response before treating the connection as dead. Must be lower than `ping_interval`. | | `auto_reconnect` | `bool` | `False` | Automatically reconnect on transient failures using exponential backoff. | | `max_reconnect_attempts` | `int` | `5` | Maximum number of reconnect attempts before giving up. | | `reconnect_backoff` | `float` | `2.0` | Base backoff delay in seconds. Doubles after each failed attempt (2s, 4s, 8s, ...). Resets on successful reconnection. | | `queue_maxsize` | `int` | `0` | Maximum messages buffered in the internal queue for `receive()`. `0` means unbounded. When set, incoming messages are dropped with a warning when the queue is full, preventing memory growth on high-frequency streams. | +| `subscription_watchdog_timeout` | `float` | `10.0` | Maximum time to wait for all `channel_subscribed` acknowledgements after connect. On timeout, the connection is closed to trigger a clean reconnect. | +| `rate_limit_backoff` | `float` | `30.0` | Minimum reconnect delay after a 429 rate-limit response. | +| `throughput_log_interval` | `int` | `100` | Logs queue depth and processed counts every N messages. Set to `0` to disable periodic throughput logs. | ```python ws = mistapi.websockets.sites.DeviceStatsEvents( apisession, site_ids=[""], ping_interval=60, # ping every 60 s - ping_timeout=20, # wait up to 20 s for pong + ping_timeout=45, # wait up to 45 s for pong auto_reconnect=True, # reconnect on transient failures ) ws.connect() @@ -609,6 +612,8 @@ ws.connect() | `ws.on_message(cb)` | `cb(data: dict)` | Register callback for incoming messages. Mutually exclusive with `receive()`. | | `ws.on_error(cb)` | `cb(error: Exception)` | Register callback for WebSocket errors | | `ws.on_close(cb)` | `cb(code: int \| None, msg: str \| None)` | Register callback for connection close. Safe to call `connect()` from within. | +| `ws.on_ping(cb)` | `cb(message: str \| bytes \| None)` | Register callback for received ping frames. | +| `ws.on_pong(cb)` | `cb(message: str \| bytes \| None)` | Register callback for received pong frames. | | `ws.connect(run_in_background)` | | Open the connection. `True` (default) runs in a daemon thread; `False` blocks. | | `ws.disconnect(wait, timeout)` | | Close the connection. `wait=True` blocks until the background thread finishes. | | `ws.receive()` | `-> Generator[dict]` | Blocking generator yielding messages. Mutually exclusive with `on_message`. | diff --git a/src/mistapi/websockets/__ws_client.py b/src/mistapi/websockets/__ws_client.py index e351c5d..58fabb5 100644 --- a/src/mistapi/websockets/__ws_client.py +++ b/src/mistapi/websockets/__ws_client.py @@ -19,7 +19,7 @@ import ssl import threading from collections.abc import Callable, Generator -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any import websocket @@ -48,6 +48,10 @@ def filter(self, record: logging.LogRecord) -> bool: from mistapi import APISession +MAX_CHANNELS_PER_CONNECTION = 2000 +HIGH_CHANNEL_COUNT_WARNING = 1500 + + class _MistWebsocket: """ Base class for Mist API WebSocket channels. @@ -64,14 +68,25 @@ def __init__( self, mist_session: "APISession", channels: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int = 45, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: + if ping_interval < 0: + raise ValueError("ping_interval must be >= 0") + if ping_timeout <= 0: + raise ValueError("ping_timeout must be > 0") + if ping_interval and ping_interval <= ping_timeout: + raise ValueError( + "ping_interval must be greater than ping_timeout when enabled" + ) if max_reconnect_attempts < 0: raise ValueError("max_reconnect_attempts must be >= 0 (0 = unlimited)") if reconnect_backoff <= 0: @@ -80,32 +95,75 @@ def __init__( raise ValueError("max_reconnect_backoff must be > 0") if queue_maxsize < 0: raise ValueError("queue_maxsize must be >= 0") + if subscription_watchdog_timeout <= 0: + raise ValueError("subscription_watchdog_timeout must be > 0") + if rate_limit_backoff <= 0: + raise ValueError("rate_limit_backoff must be > 0") + if throughput_log_interval < 0: + raise ValueError("throughput_log_interval must be >= 0") + + deduped_channels = list(dict.fromkeys(channels)) + if len(deduped_channels) != len(channels): + logger.warning( + "Duplicate channels detected; using %d unique channels instead of %d", + len(deduped_channels), + len(channels), + ) + if len(deduped_channels) > MAX_CHANNELS_PER_CONNECTION: + raise ValueError( + f"Too many channels ({len(deduped_channels)}). " + f"Mist supports up to {MAX_CHANNELS_PER_CONNECTION} channels per connection" + ) + if len(deduped_channels) >= HIGH_CHANNEL_COUNT_WARNING: + logger.warning( + "High channel count (%d). Consider spreading subscriptions over multiple " + "WebSocket connections to reduce message backlog risk.", + len(deduped_channels), + ) self._mist_session = mist_session - self._channels = channels + self._channels = deduped_channels + self._expected_channels = set(deduped_channels) self._ping_interval = ping_interval self._ping_timeout = ping_timeout self._auto_reconnect = auto_reconnect self._max_reconnect_attempts = max_reconnect_attempts self._reconnect_backoff = reconnect_backoff self._max_reconnect_backoff = max_reconnect_backoff + self._subscription_watchdog_timeout = subscription_watchdog_timeout + self._rate_limit_backoff = rate_limit_backoff + self._throughput_log_interval = throughput_log_interval self._lock = threading.Lock() + self._subscription_lock = threading.Lock() self._ws: websocket.WebSocketApp | None = None self._thread: threading.Thread | None = None + self._callback_thread: threading.Thread | None = None + self._subscription_watchdog: threading.Timer | None = None self._queue: queue.Queue[dict | None] = queue.Queue(maxsize=queue_maxsize) + self._callback_queue: queue.Queue[dict | None] = queue.Queue( + maxsize=queue_maxsize + ) self._connected = ( threading.Event() ) # tracks whether the WebSocket connection is currently open self._user_disconnect = threading.Event() + self._callback_stop = threading.Event() self._finished = threading.Event() self._finished.set() # not running initially self._reconnect_attempts = 0 self._last_close_code: int | None = None self._last_close_msg: str | None = None + self._last_http_status: int | None = None + self._subscribed_channels: set[str] = set() + self._messages_received = 0 + self._messages_dropped = 0 + self._messages_processed = 0 self._on_message_cb: Callable[[dict], None] | None = None self._on_error_cb: Callable[[Exception], None] | None = None self._on_open_cb: Callable[[], None] | None = None self._on_close_cb: Callable[[int | None, str | None], None] | None = None + self._on_ping_cb: Callable[[str | bytes | None], None] | None = None + self._on_pong_cb: Callable[[str | bytes | None], None] | None = None # ------------------------------------------------------------------ # Auth / URL helpers @@ -184,10 +242,198 @@ def on_close(self, callback: Callable[[int | None, str | None], None]) -> None: """Register a callback invoked when the connection closes.""" self._on_close_cb = callback + def on_ping(self, callback: Callable[[str | bytes | None], None]) -> None: + """Register a callback invoked when a ping frame is received.""" + self._on_ping_cb = callback + + def on_pong(self, callback: Callable[[str | bytes | None], None]) -> None: + """Register a callback invoked when a pong frame is received.""" + self._on_pong_cb = callback + + # ------------------------------------------------------------------ + # Internal helpers + + @staticmethod + def _extract_status_code(error: Exception) -> int | None: + status_code = getattr(error, "status_code", None) + if isinstance(status_code, int): + return status_code + response = getattr(error, "response", None) + if response is not None: + response_code = getattr(response, "status_code", None) + if isinstance(response_code, int): + return response_code + return None + + def _drain_queue(self, target_queue: queue.Queue[Any]) -> None: + while not target_queue.empty(): + try: + target_queue.get_nowait() + except queue.Empty: + break + + def _start_callback_worker(self) -> None: + if self._callback_thread is not None and self._callback_thread.is_alive(): + return + self._callback_stop.clear() + self._callback_thread = threading.Thread( + target=self._run_callback_worker, daemon=True + ) + self._callback_thread.start() + + def _run_callback_worker(self) -> None: + while True: + if self._callback_stop.is_set(): + break + try: + item = self._callback_queue.get(timeout=1) + except queue.Empty: + if self._finished.is_set() and self._callback_queue.empty(): + break + continue + if item is None: + if self._callback_stop.is_set() or self._finished.is_set(): + break + continue + callback = self._on_message_cb + if callback is None: + continue + try: + callback(item) + except Exception: + logger.exception("on_message callback raised") + self._messages_processed += 1 + if ( + self._throughput_log_interval + and self._messages_processed % self._throughput_log_interval == 0 + ): + logger.info( + "WebSocket callback worker processed %d messages. " + "Callback queue size=%d dropped=%d", + self._messages_processed, + self._callback_queue.qsize(), + self._messages_dropped, + ) + + def _cancel_subscription_watchdog(self) -> None: + with self._lock: + timer = self._subscription_watchdog + self._subscription_watchdog = None + if timer is not None: + timer.cancel() + + def _arm_subscription_watchdog(self, ws: websocket.WebSocketApp) -> None: + if not self._expected_channels: + return + + def _watchdog_expired() -> None: + if self._user_disconnect.is_set(): + return + with self._lock: + current_ws = self._ws + if ws is not current_ws: + return + with self._subscription_lock: + missing = sorted(self._expected_channels - self._subscribed_channels) + if not missing: + return + preview = ", ".join(missing[:5]) + if len(missing) > 5: + preview = f"{preview}, ..." + self._last_close_code = 1008 + self._last_close_msg = ( + f"subscription watchdog timeout: missing {len(missing)} channels" + ) + logger.error( + "Subscription watchdog timeout after %.1fs: received %d/%d subscriptions. " + "Missing: %s", + self._subscription_watchdog_timeout, + len(self._expected_channels) - len(missing), + len(self._expected_channels), + preview, + ) + ws.close() + + timer = threading.Timer(self._subscription_watchdog_timeout, _watchdog_expired) + timer.daemon = True + self._cancel_subscription_watchdog() + with self._lock: + self._subscription_watchdog = timer + timer.start() + + def _process_subscription_event( + self, ws: websocket.WebSocketApp, data: dict + ) -> None: + event = data.get("event") + channel = data.get("channel") + if not isinstance(channel, str): + channel = None + + if event == "channel_subscribed" and channel: + with self._subscription_lock: + self._subscribed_channels.add(channel) + subscribed_count = len(self._subscribed_channels) + expected_count = len(self._expected_channels) + logger.info( + "Channel subscribed (%d/%d): %s", + subscribed_count, + expected_count, + channel, + ) + if channel not in self._expected_channels: + logger.warning( + "Received channel_subscribed for unexpected channel: %s", channel + ) + if subscribed_count >= expected_count: + self._cancel_subscription_watchdog() + logger.info("All requested channels subscribed (%d)", expected_count) + return + + if event == "subscribe_failed": + detail = data.get("detail") + logger.error( + "Subscription failed for channel %s: %s. Closing to trigger reconnect.", + channel, + detail, + ) + self._last_close_code = 1008 + self._last_close_msg = f"subscribe_failed channel={channel} detail={detail}" + self._cancel_subscription_watchdog() + ws.close() + + def _enqueue_message(self, message: dict, to_callback_queue: bool) -> None: + target_queue = self._callback_queue if to_callback_queue else self._queue + queue_name = "callback" if to_callback_queue else "receive" + self._messages_received += 1 + try: + target_queue.put_nowait(message) + except queue.Full: + self._messages_dropped += 1 + logger.warning("%s queue full; dropping message", queue_name.capitalize()) + return + if ( + self._throughput_log_interval + and self._messages_received % self._throughput_log_interval == 0 + ): + logger.info( + "WebSocket received %d messages. %s queue size=%d dropped=%d", + self._messages_received, + queue_name.capitalize(), + target_queue.qsize(), + self._messages_dropped, + ) + # ------------------------------------------------------------------ # Internal WebSocketApp handlers def _handle_open(self, ws: websocket.WebSocketApp) -> None: + logger.info( + "WebSocket opened. Requesting %d channel subscription(s)", + len(self._channels), + ) + self._last_http_status = None + with self._subscription_lock: + self._subscribed_channels.clear() try: for channel in self._channels: ws.send(json.dumps({"subscribe": channel})) @@ -195,6 +441,8 @@ def _handle_open(self, ws: websocket.WebSocketApp) -> None: logger.error("Subscription send failed: %s", exc) ws.close() return + if self._expected_channels: + self._arm_subscription_watchdog(ws) self._reconnect_attempts = 0 self._last_close_code = None self._last_close_msg = None @@ -212,33 +460,70 @@ def _handle_message(self, ws: websocket.WebSocketApp, message: str | bytes) -> N data = json.loads(message) except (json.JSONDecodeError, TypeError): data = {"raw": message} + + if isinstance(data, dict): + self._process_subscription_event(ws, data) + if self._on_message_cb: - try: - self._on_message_cb(data) - except Exception: - logger.exception("on_message callback raised") - else: - try: - self._queue.put_nowait(data) - except queue.Full: - logger.warning("Receive queue full; dropping message") + self._start_callback_worker() + self._enqueue_message(data, to_callback_queue=True) + return - def _handle_error(self, ws: websocket.WebSocketApp, error: Exception) -> None: + self._enqueue_message(data, to_callback_queue=False) + + def _handle_error(self, _ws: websocket.WebSocketApp, error: Exception) -> None: + status_code = self._extract_status_code(error) + if status_code is not None: + self._last_http_status = status_code + if status_code == 429: + logger.warning( + "WebSocket received HTTP 429 (rate limit). " + "Reconnect backoff will be raised to at least %.1fs", + self._rate_limit_backoff, + ) + else: + logger.error("WebSocket error: %s", error) if self._on_error_cb: try: self._on_error_cb(error) except Exception: logger.exception("on_error callback raised") + def _handle_ping( + self, _ws: websocket.WebSocketApp, message: str | bytes | None + ) -> None: + logger.info("WebSocket ping received") + if self._on_ping_cb: + try: + self._on_ping_cb(message) + except Exception: + logger.exception("on_ping callback raised") + + def _handle_pong( + self, _ws: websocket.WebSocketApp, message: str | bytes | None + ) -> None: + logger.info("WebSocket pong received") + if self._on_pong_cb: + try: + self._on_pong_cb(message) + except Exception: + logger.exception("on_pong callback raised") + def _handle_close( self, - ws: websocket.WebSocketApp, + _ws: websocket.WebSocketApp, close_status_code: int | None, close_msg: str | None, ) -> None: self._connected.clear() + self._cancel_subscription_watchdog() self._last_close_code = close_status_code self._last_close_msg = close_msg + logger.info( + "WebSocket closed. code=%s message=%s", + close_status_code, + close_msg, + ) # ------------------------------------------------------------------ # Lifecycle @@ -253,6 +538,8 @@ def _create_ws_app(self) -> websocket.WebSocketApp: on_message=self._handle_message, on_error=self._handle_error, on_close=self._handle_close, + on_ping=self._handle_ping, + on_pong=self._handle_pong, ) def connect(self, run_in_background: bool = True) -> None: @@ -270,15 +557,18 @@ def connect(self, run_in_background: bool = True) -> None: raise RuntimeError("Already connected; call disconnect() first") self._finished.clear() self._user_disconnect.clear() + self._callback_stop.clear() self._reconnect_attempts = 0 + self._messages_received = 0 + self._messages_dropped = 0 + self._messages_processed = 0 # Drain stale sentinel from previous connection - while not self._queue.empty(): - try: - self._queue.get_nowait() - except queue.Empty: - break + self._drain_queue(self._queue) + self._drain_queue(self._callback_queue) self._ws = self._create_ws_app() + if self._on_message_cb: + self._start_callback_worker() if run_in_background: self._thread = threading.Thread( target=self._run_forever_safe, daemon=True @@ -294,6 +584,8 @@ def _run_forever_safe(self) -> None: while True: with self._lock: ws = self._ws + if ws is None: + break try: sslopt = self._build_sslopt() ws.run_forever( @@ -322,6 +614,8 @@ def _run_forever_safe(self) -> None: delay = self._reconnect_backoff * (2 ** (self._reconnect_attempts - 1)) if self._max_reconnect_backoff is not None: delay = min(delay, self._max_reconnect_backoff) + if self._last_http_status == 429: + delay = max(delay, self._rate_limit_backoff) if self._max_reconnect_attempts > 0: logger.info( "Reconnecting in %.1fs (attempt %d/%d)", @@ -353,10 +647,16 @@ def _run_forever_safe(self) -> None: pass finally: + self._cancel_subscription_watchdog() + self._callback_stop.set() try: self._queue.put_nowait(None) # sentinel — unblocks receive() except queue.Full: pass # _finished.set() below will unblock receive() independently + try: + self._callback_queue.put_nowait(None) # sentinel — unblocks worker + except queue.Full: + pass self._finished.set() # mark as not running — unblocks connect() if self._on_close_cb: try: @@ -376,13 +676,22 @@ def disconnect(self, wait: bool = False, timeout: float | None = None) -> None: when *wait* is True). ``None`` means wait indefinitely. """ self._user_disconnect.set() + self._callback_stop.set() + self._cancel_subscription_watchdog() with self._lock: ws = self._ws if ws: ws.close() + try: + self._callback_queue.put_nowait(None) + except queue.Full: + pass if wait and self._thread is not None: if self._thread is not threading.current_thread(): self._thread.join(timeout=timeout) + if wait and self._callback_thread is not None: + if self._callback_thread is not threading.current_thread(): + self._callback_thread.join(timeout=timeout) def receive(self) -> Generator[dict, None, None]: """ @@ -442,4 +751,4 @@ def __exit__(self, *args) -> None: def ready(self) -> bool: """Returns True if the WebSocket connection is open and ready.""" - return self._ws is not None and self._ws.ready() + return bool(self._ws is not None and self._ws.ready()) diff --git a/src/mistapi/websockets/location.py b/src/mistapi/websockets/location.py index ad2773c..16e9622 100644 --- a/src/mistapi/websockets/location.py +++ b/src/mistapi/websockets/location.py @@ -29,9 +29,9 @@ class BleAssetsEvents(_MistWebsocket): UUID of the site to stream events from. map_ids : list[str] UUIDs of the maps to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 + ping_timeout : int, default 45 Time in seconds to wait for a ping response before considering the connection dead. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. @@ -77,8 +77,8 @@ def __init__( mist_session: APISession, site_id: str, map_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int = 45, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -113,9 +113,9 @@ class ConnectedClientsEvents(_MistWebsocket): UUID of the site to stream events from. map_ids : list[str] UUIDs of the maps to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 + ping_timeout : int, default 45 Time in seconds to wait for a ping response before considering the connection dead. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. @@ -161,8 +161,8 @@ def __init__( mist_session: APISession, site_id: str, map_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int = 45, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -197,9 +197,9 @@ class SdkClientsEvents(_MistWebsocket): UUID of the site to stream events from. map_ids : list[str] UUIDs of the maps to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 + ping_timeout : int, default 45 Time in seconds to wait for a ping response before considering the connection dead. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. @@ -245,8 +245,8 @@ def __init__( mist_session: APISession, site_id: str, map_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int = 45, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -281,9 +281,9 @@ class UnconnectedClientsEvents(_MistWebsocket): UUID of the site to stream events from. map_ids : list[str] UUIDs of the maps to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 + ping_timeout : int, default 45 Time in seconds to wait for a ping response before considering the connection dead. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. @@ -329,8 +329,8 @@ def __init__( mist_session: APISession, site_id: str, map_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int = 45, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -367,9 +367,9 @@ class DiscoveredBleAssetsEvents(_MistWebsocket): UUID of the site to stream events from. map_ids : list[str] UUIDs of the maps to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 + ping_timeout : int, default 45 Time in seconds to wait for a ping response before considering the connection dead. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. @@ -415,8 +415,8 @@ def __init__( mist_session: APISession, site_id: str, map_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int = 45, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, diff --git a/src/mistapi/websockets/orgs.py b/src/mistapi/websockets/orgs.py index 9a04e7f..546d6b2 100644 --- a/src/mistapi/websockets/orgs.py +++ b/src/mistapi/websockets/orgs.py @@ -27,9 +27,9 @@ class InsightsEvents(_MistWebsocket): Authenticated API session. org_id : str UUID of the organization to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 + ping_timeout : int, default 45 Time in seconds to wait for a ping response before considering the connection dead. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. @@ -74,8 +74,8 @@ def __init__( self, mist_session: APISession, org_id: str, - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int = 45, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -107,9 +107,9 @@ class MxEdgesStatsEvents(_MistWebsocket): Authenticated API session. org_id : str UUID of the organization to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 + ping_timeout : int, default 45 Time in seconds to wait for a ping response before considering the connection dead. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. @@ -154,8 +154,8 @@ def __init__( self, mist_session: APISession, org_id: str, - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int = 45, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -187,9 +187,9 @@ class MxEdgesEvents(_MistWebsocket): Authenticated API session. org_id : str UUID of the org to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 + ping_timeout : int, default 45 Time in seconds to wait for a ping response before considering the connection dead. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. @@ -234,8 +234,8 @@ def __init__( self, mist_session: APISession, org_id: str, - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int = 45, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, diff --git a/src/mistapi/websockets/session.py b/src/mistapi/websockets/session.py index e7365dc..f15ae93 100644 --- a/src/mistapi/websockets/session.py +++ b/src/mistapi/websockets/session.py @@ -36,9 +36,9 @@ class SessionWithUrl(_MistWebsocket): The session's authentication credentials (API token or cookies) are sent to whatever host is specified in this URL. Only use trusted URLs — never pass user-supplied or untrusted input. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 + ping_timeout : int, default 45 Time in seconds to wait for a ping response before considering the connection dead. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. @@ -83,8 +83,8 @@ def __init__( self, mist_session: APISession, url: str, - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int = 45, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, diff --git a/src/mistapi/websockets/sites.py b/src/mistapi/websockets/sites.py index 64b2b88..1ba11fb 100644 --- a/src/mistapi/websockets/sites.py +++ b/src/mistapi/websockets/sites.py @@ -27,9 +27,9 @@ class ClientsStatsEvents(_MistWebsocket): Authenticated API session. site_ids : list[str] UUIDs of the sites to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 + ping_timeout : int, default 45 Time in seconds to wait for a ping response before considering the connection dead. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. @@ -74,8 +74,8 @@ def __init__( self, mist_session: APISession, site_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int = 45, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -116,9 +116,9 @@ class DeviceCmdEvents(_MistWebsocket): UUID of the site to stream events from. device_ids : list[str] UUIDs of the devices to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 + ping_timeout : int, default 45 Time in seconds to wait for a ping response before considering the connection dead. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. @@ -164,8 +164,8 @@ def __init__( mist_session: APISession, site_id: str, device_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int = 45, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -200,9 +200,9 @@ class DeviceStatsEvents(_MistWebsocket): Authenticated API session. site_ids : list[str] UUIDs of the sites to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 + ping_timeout : int, default 45 Time in seconds to wait for a ping response before considering the connection dead. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. @@ -247,8 +247,8 @@ def __init__( self, mist_session: APISession, site_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int = 45, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -281,9 +281,9 @@ class DeviceEvents(_MistWebsocket): Authenticated API session. site_ids : list[str] UUIDs of the sites to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 + ping_timeout : int, default 45 Time in seconds to wait for a ping response before considering the connection dead. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. @@ -328,8 +328,8 @@ def __init__( self, mist_session: APISession, site_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int = 45, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -362,9 +362,9 @@ class MxEdgesStatsEvents(_MistWebsocket): Authenticated API session. site_ids : list[str] UUIDs of the sites to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 + ping_timeout : int, default 45 Time in seconds to wait for a ping response before considering the connection dead. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. @@ -409,8 +409,8 @@ def __init__( self, mist_session: APISession, site_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int = 45, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -443,9 +443,9 @@ class MxEdgesEvents(_MistWebsocket): Authenticated API session. site_ids : list[str] UUIDs of the sites to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 + ping_timeout : int, default 45 Time in seconds to wait for a ping response before considering the connection dead. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. @@ -490,8 +490,8 @@ def __init__( self, mist_session: APISession, site_ids: list[str], - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int = 45, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -524,9 +524,9 @@ class PcapEvents(_MistWebsocket): Authenticated API session. site_id : str UUID of the site to stream events from. - ping_interval : int, default 30 + ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 10 + ping_timeout : int, default 45 Time in seconds to wait for a ping response before considering the connection dead. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. @@ -571,8 +571,8 @@ def __init__( self, mist_session: APISession, site_id: str, - ping_interval: int = 30, - ping_timeout: int = 10, + ping_interval: int = 60, + ping_timeout: int = 45, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, diff --git a/tests/unit/test_websocket_client.py b/tests/unit/test_websocket_client.py index 7177476..5bc793e 100644 --- a/tests/unit/test_websocket_client.py +++ b/tests/unit/test_websocket_client.py @@ -363,17 +363,43 @@ def test_wraps_invalid_json_in_raw_key(self, ws_client) -> None: def test_calls_on_message_callback_with_parsed_data(self, ws_client) -> None: cb = Mock() - ws_client.on_message(cb) + called = threading.Event() + + def cb_wrapper(data): + cb(data) + called.set() + + ws_client.on_message(cb_wrapper) + ws_client._finished.clear() # keep worker alive for this assertion + ws_client._start_callback_worker() payload = {"type": "event"} ws_client._handle_message(Mock(), json.dumps(payload)) + + assert called.wait(timeout=1), "callback was not invoked by worker" cb.assert_called_once_with(payload) + ws_client._callback_stop.set() + ws_client._callback_queue.put_nowait(None) + def test_calls_on_message_callback_with_raw_fallback(self, ws_client) -> None: cb = Mock() - ws_client.on_message(cb) + called = threading.Event() + + def cb_wrapper(data): + cb(data) + called.set() + + ws_client.on_message(cb_wrapper) + ws_client._finished.clear() # keep worker alive for this assertion + ws_client._start_callback_worker() ws_client._handle_message(Mock(), "plain text") + + assert called.wait(timeout=1), "callback was not invoked by worker" cb.assert_called_once_with({"raw": "plain text"}) + ws_client._callback_stop.set() + ws_client._callback_queue.put_nowait(None) + def test_no_error_without_on_message_callback(self, ws_client) -> None: ws_client._handle_message(Mock(), '{"ok": true}') # Should not raise @@ -461,6 +487,8 @@ def test_connect_creates_websocket_app(self, mock_ws_cls, ws_client) -> None: on_message=ws_client._handle_message, on_error=ws_client._handle_error, on_close=ws_client._handle_close, + on_ping=ws_client._handle_ping, + on_pong=ws_client._handle_pong, ) mock_ws_instance.run_forever.assert_called_once() @@ -538,8 +566,8 @@ def test_passes_sslopt_when_verify_false(self, mock_session) -> None: client._ws = mock_ws client._run_forever_safe() mock_ws.run_forever.assert_called_once_with( - ping_interval=30, - ping_timeout=10, + ping_interval=60, + ping_timeout=45, sslopt={"cert_reqs": ssl.CERT_NONE, "check_hostname": False}, ) @@ -687,8 +715,8 @@ class TestInit: """Tests for __init__ defaults.""" def test_default_ping_interval_and_timeout(self, ws_client) -> None: - assert ws_client._ping_interval == 30 - assert ws_client._ping_timeout == 10 + assert ws_client._ping_interval == 60 + assert ws_client._ping_timeout == 45 def test_custom_ping_interval_and_timeout(self, single_channel_client) -> None: assert single_channel_client._ping_interval == 15 @@ -729,6 +757,22 @@ def test_negative_queue_maxsize_raises(self, mock_session) -> None: with pytest.raises(ValueError, match="queue_maxsize must be >= 0"): _MistWebsocket(mock_session, channels=["/ch"], queue_maxsize=-1) + def test_ping_interval_must_be_greater_than_ping_timeout( + self, mock_session + ) -> None: + with pytest.raises(ValueError, match="ping_interval must be greater"): + _MistWebsocket( + mock_session, + channels=["/ch"], + ping_interval=10, + ping_timeout=10, + ) + + def test_channel_limit_enforced(self, mock_session) -> None: + channels = [f"/sites/{i}/stats/devices" for i in range(2001)] + with pytest.raises(ValueError, match="Too many channels"): + _MistWebsocket(mock_session, channels=channels) + def test_negative_max_reconnect_backoff_raises(self, mock_session) -> None: with pytest.raises(ValueError, match="max_reconnect_backoff must be > 0"): _MistWebsocket(mock_session, channels=["/ch"], max_reconnect_backoff=-1.0) @@ -1209,12 +1253,24 @@ class TestQueueCallbackBehavior: def test_message_callback_skips_queue(self, ws_client) -> None: cb = Mock() - ws_client.on_message(cb) + called = threading.Event() + + def cb_wrapper(data): + cb(data) + called.set() + + ws_client.on_message(cb_wrapper) + ws_client._finished.clear() # keep worker alive for this assertion + ws_client._start_callback_worker() ws_client._handle_message(Mock(), '{"event": "data"}') + assert called.wait(timeout=1), "callback was not invoked by worker" cb.assert_called_once_with({"event": "data"}) assert ws_client._queue.empty() + ws_client._callback_stop.set() + ws_client._callback_queue.put_nowait(None) + def test_no_callback_uses_queue(self, ws_client) -> None: ws_client._handle_message(Mock(), '{"event": "data"}') assert not ws_client._queue.empty() From 9fbef406aeaaed5f96871648b882414a694efe6f Mon Sep 17 00:00:00 2001 From: Thomas Munzer Date: Fri, 24 Apr 2026 12:36:20 +0200 Subject: [PATCH 2/8] Address PR feedback for websocket reliability changes --- README.md | 4 +- src/mistapi/websockets/__ws_client.py | 15 ++++--- src/mistapi/websockets/location.py | 30 ++++++++++++++ src/mistapi/websockets/orgs.py | 18 +++++++++ src/mistapi/websockets/session.py | 6 +++ src/mistapi/websockets/sites.py | 42 +++++++++++++++++++ tests/unit/test_websocket_client.py | 58 ++++++++++++++++++++++++--- 7 files changed, 159 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index 0d5ec6e..e8dbeab 100644 --- a/README.md +++ b/README.md @@ -584,11 +584,11 @@ All channel classes accept the following optional keyword arguments: | Parameter | Type | Default | Description | |-----------|------|---------|-------------| | `ping_interval` | `int` | `60` | Seconds between automatic ping frames. Set to `0` to disable pings. | -| `ping_timeout` | `int` | `45` | Seconds to wait for a pong response before treating the connection as dead. Must be lower than `ping_interval`. | +| `ping_timeout` | `int` | `45` | Seconds to wait for a pong response before treating the connection as dead. When `ping_interval > 0`, this must be lower than `ping_interval`. | | `auto_reconnect` | `bool` | `False` | Automatically reconnect on transient failures using exponential backoff. | | `max_reconnect_attempts` | `int` | `5` | Maximum number of reconnect attempts before giving up. | | `reconnect_backoff` | `float` | `2.0` | Base backoff delay in seconds. Doubles after each failed attempt (2s, 4s, 8s, ...). Resets on successful reconnection. | -| `queue_maxsize` | `int` | `0` | Maximum messages buffered in the internal queue for `receive()`. `0` means unbounded. When set, incoming messages are dropped with a warning when the queue is full, preventing memory growth on high-frequency streams. | +| `queue_maxsize` | `int` | `0` | Maximum messages buffered in the internal queues used for both `receive()` and callback delivery. `0` means unbounded. When set, incoming messages are dropped with a warning when either queue is full, preventing memory growth on high-frequency streams. | | `subscription_watchdog_timeout` | `float` | `10.0` | Maximum time to wait for all `channel_subscribed` acknowledgements after connect. On timeout, the connection is closed to trigger a clean reconnect. | | `rate_limit_backoff` | `float` | `30.0` | Minimum reconnect delay after a 429 rate-limit response. | | `throughput_log_interval` | `int` | `100` | Logs queue depth and processed counts every N messages. Set to `0` to disable periodic throughput logs. | diff --git a/src/mistapi/websockets/__ws_client.py b/src/mistapi/websockets/__ws_client.py index 58fabb5..1556a76 100644 --- a/src/mistapi/websockets/__ws_client.py +++ b/src/mistapi/websockets/__ws_client.py @@ -460,6 +460,8 @@ def _handle_message(self, ws: websocket.WebSocketApp, message: str | bytes) -> N data = json.loads(message) except (json.JSONDecodeError, TypeError): data = {"raw": message} + if not isinstance(data, dict): + data = {"data": data} if isinstance(data, dict): self._process_subscription_event(ws, data) @@ -473,8 +475,7 @@ def _handle_message(self, ws: websocket.WebSocketApp, message: str | bytes) -> N def _handle_error(self, _ws: websocket.WebSocketApp, error: Exception) -> None: status_code = self._extract_status_code(error) - if status_code is not None: - self._last_http_status = status_code + self._last_http_status = status_code if status_code == 429: logger.warning( "WebSocket received HTTP 429 (rate limit). " @@ -517,12 +518,14 @@ def _handle_close( ) -> None: self._connected.clear() self._cancel_subscription_watchdog() - self._last_close_code = close_status_code - self._last_close_msg = close_msg + if close_status_code is not None: + self._last_close_code = close_status_code + if close_msg not in (None, ""): + self._last_close_msg = close_msg logger.info( "WebSocket closed. code=%s message=%s", - close_status_code, - close_msg, + self._last_close_code, + self._last_close_msg, ) # ------------------------------------------------------------------ diff --git a/src/mistapi/websockets/location.py b/src/mistapi/websockets/location.py index 16e9622..f9bc0aa 100644 --- a/src/mistapi/websockets/location.py +++ b/src/mistapi/websockets/location.py @@ -84,6 +84,9 @@ def __init__( reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [f"/sites/{site_id}/stats/maps/{mid}/assets" for mid in map_ids] super().__init__( @@ -96,6 +99,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -168,6 +174,9 @@ def __init__( reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [f"/sites/{site_id}/stats/maps/{mid}/clients" for mid in map_ids] super().__init__( @@ -180,6 +189,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -252,6 +264,9 @@ def __init__( reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [f"/sites/{site_id}/stats/maps/{mid}/sdkclients" for mid in map_ids] super().__init__( @@ -264,6 +279,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -336,6 +354,9 @@ def __init__( reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [ f"/sites/{site_id}/stats/maps/{mid}/unconnected_clients" for mid in map_ids @@ -350,6 +371,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -422,6 +446,9 @@ def __init__( reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [ f"/sites/{site_id}/stats/maps/{mid}/discovered_assets" for mid in map_ids @@ -436,4 +463,7 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) diff --git a/src/mistapi/websockets/orgs.py b/src/mistapi/websockets/orgs.py index 546d6b2..e23a6d6 100644 --- a/src/mistapi/websockets/orgs.py +++ b/src/mistapi/websockets/orgs.py @@ -81,6 +81,9 @@ def __init__( reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: super().__init__( mist_session, @@ -92,6 +95,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -161,6 +167,9 @@ def __init__( reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: super().__init__( mist_session, @@ -172,6 +181,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -241,6 +253,9 @@ def __init__( reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: super().__init__( mist_session, @@ -252,4 +267,7 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) diff --git a/src/mistapi/websockets/session.py b/src/mistapi/websockets/session.py index f15ae93..29f7b16 100644 --- a/src/mistapi/websockets/session.py +++ b/src/mistapi/websockets/session.py @@ -90,6 +90,9 @@ def __init__( reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: parsed = urlparse(url) if parsed.scheme.lower() != "wss" or not parsed.netloc: @@ -105,6 +108,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) def _build_ws_url(self) -> str: diff --git a/src/mistapi/websockets/sites.py b/src/mistapi/websockets/sites.py index 1ba11fb..04d902c 100644 --- a/src/mistapi/websockets/sites.py +++ b/src/mistapi/websockets/sites.py @@ -81,6 +81,9 @@ def __init__( reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [f"/sites/{site_id}/stats/clients" for site_id in site_ids] super().__init__( @@ -93,6 +96,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -171,6 +177,9 @@ def __init__( reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [ f"/sites/{site_id}/devices/{device_id}/cmd" for device_id in device_ids @@ -185,6 +194,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -254,6 +266,9 @@ def __init__( reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [f"/sites/{site_id}/stats/devices" for site_id in site_ids] super().__init__( @@ -266,6 +281,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -335,6 +353,9 @@ def __init__( reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [f"/sites/{site_id}/devices" for site_id in site_ids] super().__init__( @@ -347,6 +368,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -416,6 +440,9 @@ def __init__( reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [f"/sites/{site_id}/stats/mxedges" for site_id in site_ids] super().__init__( @@ -428,6 +455,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -497,6 +527,9 @@ def __init__( reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [f"/sites/{site_id}/mxedges" for site_id in site_ids] super().__init__( @@ -509,6 +542,9 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) @@ -578,6 +614,9 @@ def __init__( reconnect_backoff: float = 2.0, max_reconnect_backoff: float | None = None, queue_maxsize: int = 0, + subscription_watchdog_timeout: float = 10.0, + rate_limit_backoff: float = 30.0, + throughput_log_interval: int = 100, ) -> None: channels = [f"/sites/{site_id}/pcaps"] super().__init__( @@ -590,4 +629,7 @@ def __init__( reconnect_backoff=reconnect_backoff, max_reconnect_backoff=max_reconnect_backoff, queue_maxsize=queue_maxsize, + subscription_watchdog_timeout=subscription_watchdog_timeout, + rate_limit_backoff=rate_limit_backoff, + throughput_log_interval=throughput_log_interval, ) diff --git a/tests/unit/test_websocket_client.py b/tests/unit/test_websocket_client.py index 5bc793e..58722cc 100644 --- a/tests/unit/test_websocket_client.py +++ b/tests/unit/test_websocket_client.py @@ -378,8 +378,7 @@ def cb_wrapper(data): assert called.wait(timeout=1), "callback was not invoked by worker" cb.assert_called_once_with(payload) - ws_client._callback_stop.set() - ws_client._callback_queue.put_nowait(None) + ws_client.disconnect(wait=True, timeout=1) def test_calls_on_message_callback_with_raw_fallback(self, ws_client) -> None: cb = Mock() @@ -397,8 +396,7 @@ def cb_wrapper(data): assert called.wait(timeout=1), "callback was not invoked by worker" cb.assert_called_once_with({"raw": "plain text"}) - ws_client._callback_stop.set() - ws_client._callback_queue.put_nowait(None) + ws_client.disconnect(wait=True, timeout=1) def test_no_error_without_on_message_callback(self, ws_client) -> None: ws_client._handle_message(Mock(), '{"ok": true}') # Should not raise @@ -831,6 +829,18 @@ def test_inherits_from_mist_websocket(self, mock_session) -> None: ws = DeviceCmdEvents(mock_session, site_id="s1", device_ids=["d1"]) assert isinstance(ws, _MistWebsocket) + def test_supports_reliability_kwargs(self, mock_session) -> None: + ws = DeviceStatsEvents( + mock_session, + site_ids=["s1"], + subscription_watchdog_timeout=3.0, + rate_limit_backoff=12.0, + throughput_log_interval=250, + ) + assert ws._subscription_watchdog_timeout == 3.0 + assert ws._rate_limit_backoff == 12.0 + assert ws._throughput_log_interval == 250 + class TestOrgChannels: """Tests for public org-level WebSocket channel classes.""" @@ -851,6 +861,18 @@ def test_inherits_from_mist_websocket(self, mock_session) -> None: ws = InsightsEvents(mock_session, org_id="o1") assert isinstance(ws, _MistWebsocket) + def test_supports_reliability_kwargs(self, mock_session) -> None: + ws = InsightsEvents( + mock_session, + org_id="o1", + subscription_watchdog_timeout=5.0, + rate_limit_backoff=20.0, + throughput_log_interval=400, + ) + assert ws._subscription_watchdog_timeout == 5.0 + assert ws._rate_limit_backoff == 20.0 + assert ws._throughput_log_interval == 400 + class TestLocationChannels: """Tests for public location-level WebSocket channel classes.""" @@ -882,6 +904,19 @@ def test_inherits_from_mist_websocket(self, mock_session) -> None: ws = BleAssetsEvents(mock_session, site_id="s1", map_ids=["m1"]) assert isinstance(ws, _MistWebsocket) + def test_supports_reliability_kwargs(self, mock_session) -> None: + ws = ConnectedClientsEvents( + mock_session, + site_id="s1", + map_ids=["m1"], + subscription_watchdog_timeout=4.0, + rate_limit_backoff=18.0, + throughput_log_interval=150, + ) + assert ws._subscription_watchdog_timeout == 4.0 + assert ws._rate_limit_backoff == 18.0 + assert ws._throughput_log_interval == 150 + class TestSessionChannel: """Tests for the SessionWithUrl WebSocket channel class.""" @@ -895,6 +930,18 @@ def test_inherits_from_mist_websocket(self, mock_session) -> None: ws = SessionWithUrl(mock_session, url="wss://example.com/custom") assert isinstance(ws, _MistWebsocket) + def test_supports_reliability_kwargs(self, mock_session) -> None: + ws = SessionWithUrl( + mock_session, + url="wss://example.com/custom", + subscription_watchdog_timeout=6.0, + rate_limit_backoff=25.0, + throughput_log_interval=300, + ) + assert ws._subscription_watchdog_timeout == 6.0 + assert ws._rate_limit_backoff == 25.0 + assert ws._throughput_log_interval == 300 + # --------------------------------------------------------------------------- # Auto-reconnect @@ -1268,8 +1315,7 @@ def cb_wrapper(data): cb.assert_called_once_with({"event": "data"}) assert ws_client._queue.empty() - ws_client._callback_stop.set() - ws_client._callback_queue.put_nowait(None) + ws_client.disconnect(wait=True, timeout=1) def test_no_callback_uses_queue(self, ws_client) -> None: ws_client._handle_message(Mock(), '{"event": "data"}') From 527841c2b4052faf01570ed6cbb209a468d3dc73 Mon Sep 17 00:00:00 2001 From: Thomas Munzer Date: Fri, 24 Apr 2026 12:37:46 +0200 Subject: [PATCH 3/8] Tune websocket logging and metrics thread safety --- src/mistapi/websockets/__ws_client.py | 43 +++++++++++++++++++-------- 1 file changed, 31 insertions(+), 12 deletions(-) diff --git a/src/mistapi/websockets/__ws_client.py b/src/mistapi/websockets/__ws_client.py index 1556a76..9b37c00 100644 --- a/src/mistapi/websockets/__ws_client.py +++ b/src/mistapi/websockets/__ws_client.py @@ -135,6 +135,7 @@ def __init__( self._throughput_log_interval = throughput_log_interval self._lock = threading.Lock() self._subscription_lock = threading.Lock() + self._metrics_lock = threading.Lock() self._ws: websocket.WebSocketApp | None = None self._thread: threading.Thread | None = None self._callback_thread: threading.Thread | None = None @@ -302,17 +303,20 @@ def _run_callback_worker(self) -> None: callback(item) except Exception: logger.exception("on_message callback raised") - self._messages_processed += 1 + with self._metrics_lock: + self._messages_processed += 1 + messages_processed = self._messages_processed + messages_dropped = self._messages_dropped if ( self._throughput_log_interval - and self._messages_processed % self._throughput_log_interval == 0 + and messages_processed % self._throughput_log_interval == 0 ): logger.info( "WebSocket callback worker processed %d messages. " "Callback queue size=%d dropped=%d", - self._messages_processed, + messages_processed, self._callback_queue.qsize(), - self._messages_dropped, + messages_dropped, ) def _cancel_subscription_watchdog(self) -> None: @@ -374,12 +378,22 @@ def _process_subscription_event( self._subscribed_channels.add(channel) subscribed_count = len(self._subscribed_channels) expected_count = len(self._expected_channels) - logger.info( + logger.debug( "Channel subscribed (%d/%d): %s", subscribed_count, expected_count, channel, ) + if expected_count and ( + subscribed_count == 1 + or subscribed_count % 100 == 0 + or subscribed_count >= expected_count + ): + logger.info( + "Subscription progress: received %d/%d channel acknowledgements", + subscribed_count, + expected_count, + ) if channel not in self._expected_channels: logger.warning( "Received channel_subscribed for unexpected channel: %s", channel @@ -404,23 +418,28 @@ def _process_subscription_event( def _enqueue_message(self, message: dict, to_callback_queue: bool) -> None: target_queue = self._callback_queue if to_callback_queue else self._queue queue_name = "callback" if to_callback_queue else "receive" - self._messages_received += 1 + with self._metrics_lock: + self._messages_received += 1 + messages_received = self._messages_received try: target_queue.put_nowait(message) except queue.Full: - self._messages_dropped += 1 + with self._metrics_lock: + self._messages_dropped += 1 logger.warning("%s queue full; dropping message", queue_name.capitalize()) return if ( self._throughput_log_interval - and self._messages_received % self._throughput_log_interval == 0 + and messages_received % self._throughput_log_interval == 0 ): + with self._metrics_lock: + messages_dropped = self._messages_dropped logger.info( "WebSocket received %d messages. %s queue size=%d dropped=%d", - self._messages_received, + messages_received, queue_name.capitalize(), target_queue.qsize(), - self._messages_dropped, + messages_dropped, ) # ------------------------------------------------------------------ @@ -493,7 +512,7 @@ def _handle_error(self, _ws: websocket.WebSocketApp, error: Exception) -> None: def _handle_ping( self, _ws: websocket.WebSocketApp, message: str | bytes | None ) -> None: - logger.info("WebSocket ping received") + logger.debug("WebSocket ping received") if self._on_ping_cb: try: self._on_ping_cb(message) @@ -503,7 +522,7 @@ def _handle_ping( def _handle_pong( self, _ws: websocket.WebSocketApp, message: str | bytes | None ) -> None: - logger.info("WebSocket pong received") + logger.debug("WebSocket pong received") if self._on_pong_cb: try: self._on_pong_cb(message) From 844d588a12dbf77eabff43b3e1b5eaadc7efbe02 Mon Sep 17 00:00:00 2001 From: Thomas Munzer Date: Tue, 14 Jul 2026 19:25:35 +0200 Subject: [PATCH 4/8] fix(websockets): address review findings on reliability rework - derive ping_timeout as min(45, ping_interval - 1) when not supplied, so existing callers passing only ping_interval keep working - drain the callback queue on shutdown instead of dropping pending messages; the None sentinel now marks end-of-stream - report subscription watchdog timeouts and subscribe_failed events through on_error so callers without auto_reconnect get a signal - don't resurrect the callback worker after disconnect (stop flag is now only cleared by connect()) - cite the Mist rate-limits doc for the channel-count constants - document the new reliability kwargs in all channel class docstrings - add tests: ping derivation, worker drain, worker stop guard, watchdog expiry/cancel, subscribe_failed error reporting Co-Authored-By: Claude Fable 5 --- README.md | 4 +- src/mistapi/websockets/__ws_client.py | 70 ++++++++++---- src/mistapi/websockets/location.py | 95 +++++++++++++++--- src/mistapi/websockets/orgs.py | 57 +++++++++-- src/mistapi/websockets/session.py | 19 +++- src/mistapi/websockets/sites.py | 133 ++++++++++++++++++++++---- tests/unit/test_websocket_client.py | 108 +++++++++++++++++++++ 7 files changed, 415 insertions(+), 71 deletions(-) diff --git a/README.md b/README.md index e8dbeab..f2f596d 100644 --- a/README.md +++ b/README.md @@ -584,12 +584,12 @@ All channel classes accept the following optional keyword arguments: | Parameter | Type | Default | Description | |-----------|------|---------|-------------| | `ping_interval` | `int` | `60` | Seconds between automatic ping frames. Set to `0` to disable pings. | -| `ping_timeout` | `int` | `45` | Seconds to wait for a pong response before treating the connection as dead. When `ping_interval > 0`, this must be lower than `ping_interval`. | +| `ping_timeout` | `int \| None` | `None` | Seconds to wait for a pong response before treating the connection as dead. Defaults to `min(45, ping_interval - 1)`. When `ping_interval > 0`, this must be lower than `ping_interval`. | | `auto_reconnect` | `bool` | `False` | Automatically reconnect on transient failures using exponential backoff. | | `max_reconnect_attempts` | `int` | `5` | Maximum number of reconnect attempts before giving up. | | `reconnect_backoff` | `float` | `2.0` | Base backoff delay in seconds. Doubles after each failed attempt (2s, 4s, 8s, ...). Resets on successful reconnection. | | `queue_maxsize` | `int` | `0` | Maximum messages buffered in the internal queues used for both `receive()` and callback delivery. `0` means unbounded. When set, incoming messages are dropped with a warning when either queue is full, preventing memory growth on high-frequency streams. | -| `subscription_watchdog_timeout` | `float` | `10.0` | Maximum time to wait for all `channel_subscribed` acknowledgements after connect. On timeout, the connection is closed to trigger a clean reconnect. | +| `subscription_watchdog_timeout` | `float` | `10.0` | Maximum time to wait for all `channel_subscribed` acknowledgements after connect. On timeout, the error is reported to `on_error` and the connection is closed; with `auto_reconnect=True` this triggers a clean reconnect. | | `rate_limit_backoff` | `float` | `30.0` | Minimum reconnect delay after a 429 rate-limit response. | | `throughput_log_interval` | `int` | `100` | Logs queue depth and processed counts every N messages. Set to `0` to disable periodic throughput logs. | diff --git a/src/mistapi/websockets/__ws_client.py b/src/mistapi/websockets/__ws_client.py index 9b37c00..f763cf0 100644 --- a/src/mistapi/websockets/__ws_client.py +++ b/src/mistapi/websockets/__ws_client.py @@ -48,9 +48,15 @@ def filter(self, record: logging.LogRecord) -> bool: from mistapi import APISession +# Channel limits from the Mist WebSocket documentation: +# https://www.juniper.net/documentation/us/en/software/mist/api/http/guides/websockets/rate-limits MAX_CHANNELS_PER_CONNECTION = 2000 HIGH_CHANNEL_COUNT_WARNING = 1500 +# Default pong timeout, and the value ping_timeout is derived from when only +# ping_interval is supplied (min(DEFAULT_PING_TIMEOUT, ping_interval - 1)). +DEFAULT_PING_TIMEOUT = 45 + class _MistWebsocket: """ @@ -69,7 +75,7 @@ def __init__( mist_session: "APISession", channels: list[str], ping_interval: int = 60, - ping_timeout: int = 45, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -81,6 +87,20 @@ def __init__( ) -> None: if ping_interval < 0: raise ValueError("ping_interval must be >= 0") + if ping_timeout is None: + # Derive a valid pong timeout so callers who only set + # ping_interval keep working (websocket-client requires + # ping_timeout < ping_interval). + ping_timeout = ( + min(DEFAULT_PING_TIMEOUT, ping_interval - 1) + if ping_interval > 0 + else DEFAULT_PING_TIMEOUT + ) + if ping_timeout < 1: + raise ValueError( + "ping_interval must be >= 2 to leave room for the pong " + "timeout, or set ping_interval=0 to disable pings" + ) if ping_timeout <= 0: raise ValueError("ping_timeout must be > 0") if ping_interval and ping_interval <= ping_timeout: @@ -274,28 +294,31 @@ def _drain_queue(self, target_queue: queue.Queue[Any]) -> None: break def _start_callback_worker(self) -> None: + if self._callback_stop.is_set(): + # Disconnecting/stopped: don't resurrect the worker for a late + # message. connect() clears the flag before starting a new run. + return if self._callback_thread is not None and self._callback_thread.is_alive(): return - self._callback_stop.clear() self._callback_thread = threading.Thread( target=self._run_callback_worker, daemon=True ) self._callback_thread.start() def _run_callback_worker(self) -> None: + # On stop, keep draining so every message received before the stop + # (i.e. before the None sentinel) is still delivered to the callback. while True: - if self._callback_stop.is_set(): - break try: item = self._callback_queue.get(timeout=1) except queue.Empty: + if self._callback_stop.is_set(): + break # stop requested and queue fully drained if self._finished.is_set() and self._callback_queue.empty(): break continue if item is None: - if self._callback_stop.is_set() or self._finished.is_set(): - break - continue + break # end-of-stream sentinel; everything before it was delivered callback = self._on_message_cb if callback is None: continue @@ -348,13 +371,18 @@ def _watchdog_expired() -> None: self._last_close_msg = ( f"subscription watchdog timeout: missing {len(missing)} channels" ) - logger.error( - "Subscription watchdog timeout after %.1fs: received %d/%d subscriptions. " - "Missing: %s", - self._subscription_watchdog_timeout, - len(self._expected_channels) - len(missing), - len(self._expected_channels), - preview, + # Surface through on_error so callers without auto_reconnect + # still get an actionable signal (the close alone reconnects + # only when auto_reconnect is enabled). + self._handle_error( + ws, + TimeoutError( + f"subscription watchdog timeout after " + f"{self._subscription_watchdog_timeout:.1f}s: received " + f"{len(self._expected_channels) - len(missing)}/" + f"{len(self._expected_channels)} subscriptions. " + f"Missing: {preview}" + ), ) ws.close() @@ -405,14 +433,15 @@ def _process_subscription_event( if event == "subscribe_failed": detail = data.get("detail") - logger.error( - "Subscription failed for channel %s: %s. Closing to trigger reconnect.", - channel, - detail, - ) self._last_close_code = 1008 self._last_close_msg = f"subscribe_failed channel={channel} detail={detail}" self._cancel_subscription_watchdog() + # Surface through on_error so callers without auto_reconnect + # still get an actionable signal before the connection closes. + self._handle_error( + ws, + ConnectionError(f"subscription failed for channel {channel}: {detail}"), + ) ws.close() def _enqueue_message(self, message: dict, to_callback_queue: bool) -> None: @@ -482,8 +511,7 @@ def _handle_message(self, ws: websocket.WebSocketApp, message: str | bytes) -> N if not isinstance(data, dict): data = {"data": data} - if isinstance(data, dict): - self._process_subscription_event(ws, data) + self._process_subscription_event(ws, data) if self._on_message_cb: self._start_callback_worker() diff --git a/src/mistapi/websockets/location.py b/src/mistapi/websockets/location.py index f9bc0aa..cc2ecf5 100644 --- a/src/mistapi/websockets/location.py +++ b/src/mistapi/websockets/location.py @@ -31,8 +31,10 @@ class BleAssetsEvents(_MistWebsocket): UUIDs of the maps to stream events from. ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 45 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be + lower than ping_interval. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -46,6 +48,17 @@ class BleAssetsEvents(_MistWebsocket): ``receive()`` generator. ``0`` means unbounded. When set, incoming messages are dropped with a warning when the queue is full, preventing memory growth on high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -78,7 +91,7 @@ def __init__( site_id: str, map_ids: list[str], ping_interval: int = 60, - ping_timeout: int = 45, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -121,8 +134,10 @@ class ConnectedClientsEvents(_MistWebsocket): UUIDs of the maps to stream events from. ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 45 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be + lower than ping_interval. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -136,6 +151,17 @@ class ConnectedClientsEvents(_MistWebsocket): ``receive()`` generator. ``0`` means unbounded. When set, incoming messages are dropped with a warning when the queue is full, preventing memory growth on high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -168,7 +194,7 @@ def __init__( site_id: str, map_ids: list[str], ping_interval: int = 60, - ping_timeout: int = 45, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -211,8 +237,10 @@ class SdkClientsEvents(_MistWebsocket): UUIDs of the maps to stream events from. ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 45 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be + lower than ping_interval. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -226,6 +254,17 @@ class SdkClientsEvents(_MistWebsocket): ``receive()`` generator. ``0`` means unbounded. When set, incoming messages are dropped with a warning when the queue is full, preventing memory growth on high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -258,7 +297,7 @@ def __init__( site_id: str, map_ids: list[str], ping_interval: int = 60, - ping_timeout: int = 45, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -301,8 +340,10 @@ class UnconnectedClientsEvents(_MistWebsocket): UUIDs of the maps to stream events from. ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 45 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be + lower than ping_interval. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -316,6 +357,17 @@ class UnconnectedClientsEvents(_MistWebsocket): ``receive()`` generator. ``0`` means unbounded. When set, incoming messages are dropped with a warning when the queue is full, preventing memory growth on high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -348,7 +400,7 @@ def __init__( site_id: str, map_ids: list[str], ping_interval: int = 60, - ping_timeout: int = 45, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -393,8 +445,10 @@ class DiscoveredBleAssetsEvents(_MistWebsocket): UUIDs of the maps to stream events from. ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 45 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be + lower than ping_interval. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -408,6 +462,17 @@ class DiscoveredBleAssetsEvents(_MistWebsocket): ``receive()`` generator. ``0`` means unbounded. When set, incoming messages are dropped with a warning when the queue is full, preventing memory growth on high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -440,7 +505,7 @@ def __init__( site_id: str, map_ids: list[str], ping_interval: int = 60, - ping_timeout: int = 45, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, diff --git a/src/mistapi/websockets/orgs.py b/src/mistapi/websockets/orgs.py index e23a6d6..aee9d5a 100644 --- a/src/mistapi/websockets/orgs.py +++ b/src/mistapi/websockets/orgs.py @@ -29,8 +29,10 @@ class InsightsEvents(_MistWebsocket): UUID of the organization to stream events from. ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 45 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be + lower than ping_interval. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -44,6 +46,17 @@ class InsightsEvents(_MistWebsocket): ``receive()`` generator. ``0`` means unbounded. When set, incoming messages are dropped with a warning when the queue is full, preventing memory growth on high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -75,7 +88,7 @@ def __init__( mist_session: APISession, org_id: str, ping_interval: int = 60, - ping_timeout: int = 45, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -115,8 +128,10 @@ class MxEdgesStatsEvents(_MistWebsocket): UUID of the organization to stream events from. ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 45 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be + lower than ping_interval. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -130,6 +145,17 @@ class MxEdgesStatsEvents(_MistWebsocket): ``receive()`` generator. ``0`` means unbounded. When set, incoming messages are dropped with a warning when the queue is full, preventing memory growth on high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -161,7 +187,7 @@ def __init__( mist_session: APISession, org_id: str, ping_interval: int = 60, - ping_timeout: int = 45, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -201,8 +227,10 @@ class MxEdgesEvents(_MistWebsocket): UUID of the org to stream events from. ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 45 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be + lower than ping_interval. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -216,6 +244,17 @@ class MxEdgesEvents(_MistWebsocket): ``receive()`` generator. ``0`` means unbounded. When set, incoming messages are dropped with a warning when the queue is full, preventing memory growth on high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -247,7 +286,7 @@ def __init__( mist_session: APISession, org_id: str, ping_interval: int = 60, - ping_timeout: int = 45, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, diff --git a/src/mistapi/websockets/session.py b/src/mistapi/websockets/session.py index 29f7b16..d396a89 100644 --- a/src/mistapi/websockets/session.py +++ b/src/mistapi/websockets/session.py @@ -38,8 +38,10 @@ class SessionWithUrl(_MistWebsocket): trusted URLs — never pass user-supplied or untrusted input. ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 45 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be + lower than ping_interval. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -53,6 +55,17 @@ class SessionWithUrl(_MistWebsocket): ``receive()`` generator. ``0`` means unbounded. When set, incoming messages are dropped with a warning when the queue is full, preventing memory growth on high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -84,7 +97,7 @@ def __init__( mist_session: APISession, url: str, ping_interval: int = 60, - ping_timeout: int = 45, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, diff --git a/src/mistapi/websockets/sites.py b/src/mistapi/websockets/sites.py index 04d902c..c0b142e 100644 --- a/src/mistapi/websockets/sites.py +++ b/src/mistapi/websockets/sites.py @@ -29,8 +29,10 @@ class ClientsStatsEvents(_MistWebsocket): UUIDs of the sites to stream events from. ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 45 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be + lower than ping_interval. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -44,6 +46,17 @@ class ClientsStatsEvents(_MistWebsocket): ``receive()`` generator. ``0`` means unbounded. When set, incoming messages are dropped with a warning when the queue is full, preventing memory growth on high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -75,7 +88,7 @@ def __init__( mist_session: APISession, site_ids: list[str], ping_interval: int = 60, - ping_timeout: int = 45, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -124,8 +137,10 @@ class DeviceCmdEvents(_MistWebsocket): UUIDs of the devices to stream events from. ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 45 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be + lower than ping_interval. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -139,6 +154,17 @@ class DeviceCmdEvents(_MistWebsocket): ``receive()`` generator. ``0`` means unbounded. When set, incoming messages are dropped with a warning when the queue is full, preventing memory growth on high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -171,7 +197,7 @@ def __init__( site_id: str, device_ids: list[str], ping_interval: int = 60, - ping_timeout: int = 45, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -214,8 +240,10 @@ class DeviceStatsEvents(_MistWebsocket): UUIDs of the sites to stream events from. ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 45 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be + lower than ping_interval. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -229,6 +257,17 @@ class DeviceStatsEvents(_MistWebsocket): ``receive()`` generator. ``0`` means unbounded. When set, incoming messages are dropped with a warning when the queue is full, preventing memory growth on high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -260,7 +299,7 @@ def __init__( mist_session: APISession, site_ids: list[str], ping_interval: int = 60, - ping_timeout: int = 45, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -301,8 +340,10 @@ class DeviceEvents(_MistWebsocket): UUIDs of the sites to stream events from. ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 45 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be + lower than ping_interval. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -316,6 +357,17 @@ class DeviceEvents(_MistWebsocket): ``receive()`` generator. ``0`` means unbounded. When set, incoming messages are dropped with a warning when the queue is full, preventing memory growth on high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -347,7 +399,7 @@ def __init__( mist_session: APISession, site_ids: list[str], ping_interval: int = 60, - ping_timeout: int = 45, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -388,8 +440,10 @@ class MxEdgesStatsEvents(_MistWebsocket): UUIDs of the sites to stream events from. ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 45 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be + lower than ping_interval. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -403,6 +457,17 @@ class MxEdgesStatsEvents(_MistWebsocket): ``receive()`` generator. ``0`` means unbounded. When set, incoming messages are dropped with a warning when the queue is full, preventing memory growth on high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -434,7 +499,7 @@ def __init__( mist_session: APISession, site_ids: list[str], ping_interval: int = 60, - ping_timeout: int = 45, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -475,8 +540,10 @@ class MxEdgesEvents(_MistWebsocket): UUIDs of the sites to stream events from. ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 45 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be + lower than ping_interval. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -490,6 +557,17 @@ class MxEdgesEvents(_MistWebsocket): ``receive()`` generator. ``0`` means unbounded. When set, incoming messages are dropped with a warning when the queue is full, preventing memory growth on high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -521,7 +599,7 @@ def __init__( mist_session: APISession, site_ids: list[str], ping_interval: int = 60, - ping_timeout: int = 45, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, @@ -562,8 +640,10 @@ class PcapEvents(_MistWebsocket): UUID of the site to stream events from. ping_interval : int, default 60 Interval in seconds to send WebSocket ping frames (keep-alive). - ping_timeout : int, default 45 - Time in seconds to wait for a ping response before considering the connection dead. + ping_timeout : int | None, default None + Time in seconds to wait for a ping response before considering the + connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be + lower than ping_interval. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -577,6 +657,17 @@ class PcapEvents(_MistWebsocket): ``receive()`` generator. ``0`` means unbounded. When set, incoming messages are dropped with a warning when the queue is full, preventing memory growth on high-frequency streams. + subscription_watchdog_timeout : float, default 10.0 + Maximum time in seconds to wait for all channel subscription + acknowledgements after connect. On timeout, the error is reported to + ``on_error`` and the connection is closed (auto_reconnect, when + enabled, then reconnects). + rate_limit_backoff : float, default 30.0 + Minimum reconnect delay in seconds after an HTTP 429 rate-limit + response. + throughput_log_interval : int, default 100 + Log queue depth and processed counts every N messages. ``0`` disables + periodic throughput logs. EXAMPLE ----------- @@ -608,7 +699,7 @@ def __init__( mist_session: APISession, site_id: str, ping_interval: int = 60, - ping_timeout: int = 45, + ping_timeout: int | None = None, auto_reconnect: bool = False, max_reconnect_attempts: int = 5, reconnect_backoff: float = 2.0, diff --git a/tests/unit/test_websocket_client.py b/tests/unit/test_websocket_client.py index 58722cc..2efe0a7 100644 --- a/tests/unit/test_websocket_client.py +++ b/tests/unit/test_websocket_client.py @@ -766,6 +766,21 @@ def test_ping_interval_must_be_greater_than_ping_timeout( ping_timeout=10, ) + def test_ping_timeout_derived_from_ping_interval(self, mock_session) -> None: + client = _MistWebsocket(mock_session, channels=["/ch"], ping_interval=30) + assert client._ping_timeout == 29 + + client = _MistWebsocket(mock_session, channels=["/ch"], ping_interval=120) + assert client._ping_timeout == 45 + + def test_ping_timeout_derived_with_pings_disabled(self, mock_session) -> None: + client = _MistWebsocket(mock_session, channels=["/ch"], ping_interval=0) + assert client._ping_timeout == 45 + + def test_tiny_ping_interval_without_timeout_raises(self, mock_session) -> None: + with pytest.raises(ValueError, match="ping_interval must be >= 2"): + _MistWebsocket(mock_session, channels=["/ch"], ping_interval=1) + def test_channel_limit_enforced(self, mock_session) -> None: channels = [f"/sites/{i}/stats/devices" for i in range(2001)] with pytest.raises(ValueError, match="Too many channels"): @@ -1559,3 +1574,96 @@ def blocking_run_forever(**kwargs): barrier.set() # release blocking thread t.join(timeout=5) assert not t.is_alive() + + +# --------------------------------------------------------------------------- +# Reliability behaviors (callback worker, watchdog, subscribe_failed) +# --------------------------------------------------------------------------- + + +class TestReliabilityBehavior: + """Tests for the callback worker lifecycle and subscription safeguards.""" + + def test_callback_worker_drains_queue_on_stop(self, ws_client) -> None: + received = [] + ws_client.on_message(received.append) + for i in range(3): + ws_client._callback_queue.put_nowait({"n": i}) + ws_client._callback_queue.put_nowait(None) # end-of-stream sentinel + ws_client._callback_stop.set() + + ws_client._run_callback_worker() # run synchronously + + assert received == [{"n": 0}, {"n": 1}, {"n": 2}] + + def test_start_callback_worker_refuses_after_stop(self, ws_client) -> None: + ws_client._callback_stop.set() + ws_client._start_callback_worker() + assert ws_client._callback_thread is None + + def test_watchdog_timeout_reports_on_error_and_closes(self, mock_session) -> None: + client = _MistWebsocket( + mock_session, + channels=["/ch1", "/ch2"], + subscription_watchdog_timeout=0.05, + ) + errors = [] + error_seen = threading.Event() + + def on_error(exc): + errors.append(exc) + error_seen.set() + + client.on_error(on_error) + mock_ws = Mock() + client._ws = mock_ws + + client._arm_subscription_watchdog(mock_ws) + + assert error_seen.wait(timeout=2), "watchdog did not fire" + assert isinstance(errors[0], TimeoutError) + assert "subscription watchdog timeout" in str(errors[0]) + mock_ws.close.assert_called_once() + assert client._last_close_code == 1008 + + def test_watchdog_cancelled_when_all_channels_subscribed( + self, mock_session + ) -> None: + client = _MistWebsocket( + mock_session, + channels=["/ch1"], + subscription_watchdog_timeout=0.05, + ) + errors = [] + client.on_error(errors.append) + mock_ws = Mock() + client._ws = mock_ws + + client._arm_subscription_watchdog(mock_ws) + client._process_subscription_event( + mock_ws, {"event": "channel_subscribed", "channel": "/ch1"} + ) + + threading.Event().wait(timeout=0.2) # let a stray timer fire if any + assert errors == [] + mock_ws.close.assert_not_called() + + def test_subscribe_failed_reports_on_error_and_closes(self, ws_client) -> None: + errors = [] + ws_client.on_error(errors.append) + mock_ws = Mock() + + ws_client._process_subscription_event( + mock_ws, + { + "event": "subscribe_failed", + "channel": "/test/channel1", + "detail": "denied", + }, + ) + + assert len(errors) == 1 + assert isinstance(errors[0], ConnectionError) + assert "/test/channel1" in str(errors[0]) + mock_ws.close.assert_called_once() + assert ws_client._last_close_code == 1008 From e205134d287c4fd6454d4904d5b6734ce43ac404 Mon Sep 17 00:00:00 2001 From: Thomas Munzer Date: Tue, 14 Jul 2026 19:44:24 +0200 Subject: [PATCH 5/8] docs(websockets): clarify ping_timeout derivation and queue_maxsize scope - note that ping_timeout falls back to 45 when ping_interval=0 (pings disabled, value unused) in the constants comment, README, and all channel class docstrings - document that queue_maxsize bounds both the receive() and callback queues in the channel class docstrings Co-Authored-By: Claude Fable 5 --- README.md | 2 +- src/mistapi/websockets/__ws_client.py | 5 +- src/mistapi/websockets/location.py | 75 ++++++++++-------- src/mistapi/websockets/orgs.py | 45 ++++++----- src/mistapi/websockets/session.py | 15 ++-- src/mistapi/websockets/sites.py | 105 +++++++++++++++----------- 6 files changed, 148 insertions(+), 99 deletions(-) diff --git a/README.md b/README.md index f2f596d..2b51880 100644 --- a/README.md +++ b/README.md @@ -584,7 +584,7 @@ All channel classes accept the following optional keyword arguments: | Parameter | Type | Default | Description | |-----------|------|---------|-------------| | `ping_interval` | `int` | `60` | Seconds between automatic ping frames. Set to `0` to disable pings. | -| `ping_timeout` | `int \| None` | `None` | Seconds to wait for a pong response before treating the connection as dead. Defaults to `min(45, ping_interval - 1)`. When `ping_interval > 0`, this must be lower than `ping_interval`. | +| `ping_timeout` | `int \| None` | `None` | Seconds to wait for a pong response before treating the connection as dead. Defaults to `min(45, ping_interval - 1)` when pings are enabled, or `45` when `ping_interval=0` (unused since pings are disabled). When `ping_interval > 0`, this must be lower than `ping_interval`. | | `auto_reconnect` | `bool` | `False` | Automatically reconnect on transient failures using exponential backoff. | | `max_reconnect_attempts` | `int` | `5` | Maximum number of reconnect attempts before giving up. | | `reconnect_backoff` | `float` | `2.0` | Base backoff delay in seconds. Doubles after each failed attempt (2s, 4s, 8s, ...). Resets on successful reconnection. | diff --git a/src/mistapi/websockets/__ws_client.py b/src/mistapi/websockets/__ws_client.py index f763cf0..e874b41 100644 --- a/src/mistapi/websockets/__ws_client.py +++ b/src/mistapi/websockets/__ws_client.py @@ -53,8 +53,9 @@ def filter(self, record: logging.LogRecord) -> bool: MAX_CHANNELS_PER_CONNECTION = 2000 HIGH_CHANNEL_COUNT_WARNING = 1500 -# Default pong timeout, and the value ping_timeout is derived from when only -# ping_interval is supplied (min(DEFAULT_PING_TIMEOUT, ping_interval - 1)). +# Default pong timeout. When ping_timeout is not supplied it is derived as +# min(DEFAULT_PING_TIMEOUT, ping_interval - 1), or DEFAULT_PING_TIMEOUT when +# ping_interval is 0 (pings disabled, so the value is unused). DEFAULT_PING_TIMEOUT = 45 diff --git a/src/mistapi/websockets/location.py b/src/mistapi/websockets/location.py index cc2ecf5..f75fe30 100644 --- a/src/mistapi/websockets/location.py +++ b/src/mistapi/websockets/location.py @@ -33,8 +33,10 @@ class BleAssetsEvents(_MistWebsocket): Interval in seconds to send WebSocket ping frames (keep-alive). ping_timeout : int | None, default None Time in seconds to wait for a ping response before considering the - connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be - lower than ping_interval. + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -44,10 +46,11 @@ class BleAssetsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. subscription_watchdog_timeout : float, default 10.0 Maximum time in seconds to wait for all channel subscription acknowledgements after connect. On timeout, the error is reported to @@ -136,8 +139,10 @@ class ConnectedClientsEvents(_MistWebsocket): Interval in seconds to send WebSocket ping frames (keep-alive). ping_timeout : int | None, default None Time in seconds to wait for a ping response before considering the - connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be - lower than ping_interval. + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -147,10 +152,11 @@ class ConnectedClientsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. subscription_watchdog_timeout : float, default 10.0 Maximum time in seconds to wait for all channel subscription acknowledgements after connect. On timeout, the error is reported to @@ -239,8 +245,10 @@ class SdkClientsEvents(_MistWebsocket): Interval in seconds to send WebSocket ping frames (keep-alive). ping_timeout : int | None, default None Time in seconds to wait for a ping response before considering the - connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be - lower than ping_interval. + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -250,10 +258,11 @@ class SdkClientsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. subscription_watchdog_timeout : float, default 10.0 Maximum time in seconds to wait for all channel subscription acknowledgements after connect. On timeout, the error is reported to @@ -342,8 +351,10 @@ class UnconnectedClientsEvents(_MistWebsocket): Interval in seconds to send WebSocket ping frames (keep-alive). ping_timeout : int | None, default None Time in seconds to wait for a ping response before considering the - connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be - lower than ping_interval. + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -353,10 +364,11 @@ class UnconnectedClientsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. subscription_watchdog_timeout : float, default 10.0 Maximum time in seconds to wait for all channel subscription acknowledgements after connect. On timeout, the error is reported to @@ -447,8 +459,10 @@ class DiscoveredBleAssetsEvents(_MistWebsocket): Interval in seconds to send WebSocket ping frames (keep-alive). ping_timeout : int | None, default None Time in seconds to wait for a ping response before considering the - connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be - lower than ping_interval. + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -458,10 +472,11 @@ class DiscoveredBleAssetsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. subscription_watchdog_timeout : float, default 10.0 Maximum time in seconds to wait for all channel subscription acknowledgements after connect. On timeout, the error is reported to diff --git a/src/mistapi/websockets/orgs.py b/src/mistapi/websockets/orgs.py index aee9d5a..65463f5 100644 --- a/src/mistapi/websockets/orgs.py +++ b/src/mistapi/websockets/orgs.py @@ -31,8 +31,10 @@ class InsightsEvents(_MistWebsocket): Interval in seconds to send WebSocket ping frames (keep-alive). ping_timeout : int | None, default None Time in seconds to wait for a ping response before considering the - connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be - lower than ping_interval. + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -42,10 +44,11 @@ class InsightsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. subscription_watchdog_timeout : float, default 10.0 Maximum time in seconds to wait for all channel subscription acknowledgements after connect. On timeout, the error is reported to @@ -130,8 +133,10 @@ class MxEdgesStatsEvents(_MistWebsocket): Interval in seconds to send WebSocket ping frames (keep-alive). ping_timeout : int | None, default None Time in seconds to wait for a ping response before considering the - connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be - lower than ping_interval. + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -141,10 +146,11 @@ class MxEdgesStatsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. subscription_watchdog_timeout : float, default 10.0 Maximum time in seconds to wait for all channel subscription acknowledgements after connect. On timeout, the error is reported to @@ -229,8 +235,10 @@ class MxEdgesEvents(_MistWebsocket): Interval in seconds to send WebSocket ping frames (keep-alive). ping_timeout : int | None, default None Time in seconds to wait for a ping response before considering the - connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be - lower than ping_interval. + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -240,10 +248,11 @@ class MxEdgesEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. subscription_watchdog_timeout : float, default 10.0 Maximum time in seconds to wait for all channel subscription acknowledgements after connect. On timeout, the error is reported to diff --git a/src/mistapi/websockets/session.py b/src/mistapi/websockets/session.py index d396a89..2a5f56c 100644 --- a/src/mistapi/websockets/session.py +++ b/src/mistapi/websockets/session.py @@ -40,8 +40,10 @@ class SessionWithUrl(_MistWebsocket): Interval in seconds to send WebSocket ping frames (keep-alive). ping_timeout : int | None, default None Time in seconds to wait for a ping response before considering the - connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be - lower than ping_interval. + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -51,10 +53,11 @@ class SessionWithUrl(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. subscription_watchdog_timeout : float, default 10.0 Maximum time in seconds to wait for all channel subscription acknowledgements after connect. On timeout, the error is reported to diff --git a/src/mistapi/websockets/sites.py b/src/mistapi/websockets/sites.py index c0b142e..df194c2 100644 --- a/src/mistapi/websockets/sites.py +++ b/src/mistapi/websockets/sites.py @@ -31,8 +31,10 @@ class ClientsStatsEvents(_MistWebsocket): Interval in seconds to send WebSocket ping frames (keep-alive). ping_timeout : int | None, default None Time in seconds to wait for a ping response before considering the - connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be - lower than ping_interval. + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -42,10 +44,11 @@ class ClientsStatsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. subscription_watchdog_timeout : float, default 10.0 Maximum time in seconds to wait for all channel subscription acknowledgements after connect. On timeout, the error is reported to @@ -139,8 +142,10 @@ class DeviceCmdEvents(_MistWebsocket): Interval in seconds to send WebSocket ping frames (keep-alive). ping_timeout : int | None, default None Time in seconds to wait for a ping response before considering the - connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be - lower than ping_interval. + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -150,10 +155,11 @@ class DeviceCmdEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. subscription_watchdog_timeout : float, default 10.0 Maximum time in seconds to wait for all channel subscription acknowledgements after connect. On timeout, the error is reported to @@ -242,8 +248,10 @@ class DeviceStatsEvents(_MistWebsocket): Interval in seconds to send WebSocket ping frames (keep-alive). ping_timeout : int | None, default None Time in seconds to wait for a ping response before considering the - connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be - lower than ping_interval. + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -253,10 +261,11 @@ class DeviceStatsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. subscription_watchdog_timeout : float, default 10.0 Maximum time in seconds to wait for all channel subscription acknowledgements after connect. On timeout, the error is reported to @@ -342,8 +351,10 @@ class DeviceEvents(_MistWebsocket): Interval in seconds to send WebSocket ping frames (keep-alive). ping_timeout : int | None, default None Time in seconds to wait for a ping response before considering the - connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be - lower than ping_interval. + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -353,10 +364,11 @@ class DeviceEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. subscription_watchdog_timeout : float, default 10.0 Maximum time in seconds to wait for all channel subscription acknowledgements after connect. On timeout, the error is reported to @@ -442,8 +454,10 @@ class MxEdgesStatsEvents(_MistWebsocket): Interval in seconds to send WebSocket ping frames (keep-alive). ping_timeout : int | None, default None Time in seconds to wait for a ping response before considering the - connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be - lower than ping_interval. + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -453,10 +467,11 @@ class MxEdgesStatsEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. subscription_watchdog_timeout : float, default 10.0 Maximum time in seconds to wait for all channel subscription acknowledgements after connect. On timeout, the error is reported to @@ -542,8 +557,10 @@ class MxEdgesEvents(_MistWebsocket): Interval in seconds to send WebSocket ping frames (keep-alive). ping_timeout : int | None, default None Time in seconds to wait for a ping response before considering the - connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be - lower than ping_interval. + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -553,10 +570,11 @@ class MxEdgesEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. subscription_watchdog_timeout : float, default 10.0 Maximum time in seconds to wait for all channel subscription acknowledgements after connect. On timeout, the error is reported to @@ -642,8 +660,10 @@ class PcapEvents(_MistWebsocket): Interval in seconds to send WebSocket ping frames (keep-alive). ping_timeout : int | None, default None Time in seconds to wait for a ping response before considering the - connection dead. Defaults to ``min(45, ping_interval - 1)``. Must be - lower than ping_interval. + connection dead. Defaults to ``min(45, ping_interval - 1)`` when + pings are enabled, or ``45`` when ``ping_interval`` is 0 (unused + since pings are disabled). Must be lower than ping_interval when + pings are enabled. auto_reconnect : bool, default False Automatically reconnect on unexpected disconnections using exponential backoff. max_reconnect_attempts : int, default 5 @@ -653,10 +673,11 @@ class PcapEvents(_MistWebsocket): max_reconnect_backoff : float | None, default None Maximum backoff delay in seconds. If None, backoff grows indefinitely. queue_maxsize : int, default 0 - Maximum number of messages buffered in the internal queue for the - ``receive()`` generator. ``0`` means unbounded. When set, - incoming messages are dropped with a warning when the queue is - full, preventing memory growth on high-frequency streams. + Maximum number of messages buffered in each of the internal queues + used for the ``receive()`` generator and callback delivery. ``0`` + means unbounded. When set, incoming messages are dropped with a + warning when the target queue is full, preventing memory growth on + high-frequency streams. subscription_watchdog_timeout : float, default 10.0 Maximum time in seconds to wait for all channel subscription acknowledgements after connect. On timeout, the error is reported to From 977d20d2b543b6464ddf5158783aa81771a6181e Mon Sep 17 00:00:00 2001 From: Thomas Munzer Date: Tue, 14 Jul 2026 20:02:28 +0200 Subject: [PATCH 6/8] fix(websockets): reset reconnect state after subscriptions --- README.md | 2 +- src/mistapi/websockets/__ws_client.py | 24 ++++-- tests/unit/test_websocket_client.py | 106 +++++++++++++++++++++++++- 3 files changed, 122 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 2b51880..e3d80ae 100644 --- a/README.md +++ b/README.md @@ -587,7 +587,7 @@ All channel classes accept the following optional keyword arguments: | `ping_timeout` | `int \| None` | `None` | Seconds to wait for a pong response before treating the connection as dead. Defaults to `min(45, ping_interval - 1)` when pings are enabled, or `45` when `ping_interval=0` (unused since pings are disabled). When `ping_interval > 0`, this must be lower than `ping_interval`. | | `auto_reconnect` | `bool` | `False` | Automatically reconnect on transient failures using exponential backoff. | | `max_reconnect_attempts` | `int` | `5` | Maximum number of reconnect attempts before giving up. | -| `reconnect_backoff` | `float` | `2.0` | Base backoff delay in seconds. Doubles after each failed attempt (2s, 4s, 8s, ...). Resets on successful reconnection. | +| `reconnect_backoff` | `float` | `2.0` | Base backoff delay in seconds. Doubles after each failed attempt (2s, 4s, 8s, ...). Resets once the connection is fully established and all requested subscriptions are acknowledged. | | `queue_maxsize` | `int` | `0` | Maximum messages buffered in the internal queues used for both `receive()` and callback delivery. `0` means unbounded. When set, incoming messages are dropped with a warning when either queue is full, preventing memory growth on high-frequency streams. | | `subscription_watchdog_timeout` | `float` | `10.0` | Maximum time to wait for all `channel_subscribed` acknowledgements after connect. On timeout, the error is reported to `on_error` and the connection is closed; with `auto_reconnect=True` this triggers a clean reconnect. | | `rate_limit_backoff` | `float` | `30.0` | Minimum reconnect delay after a 429 rate-limit response. | diff --git a/src/mistapi/websockets/__ws_client.py b/src/mistapi/websockets/__ws_client.py index e874b41..2422894 100644 --- a/src/mistapi/websockets/__ws_client.py +++ b/src/mistapi/websockets/__ws_client.py @@ -403,6 +403,11 @@ def _process_subscription_event( channel = None if event == "channel_subscribed" and channel: + if channel not in self._expected_channels: + logger.warning( + "Received channel_subscribed for unexpected channel: %s", channel + ) + return with self._subscription_lock: self._subscribed_channels.add(channel) subscribed_count = len(self._subscribed_channels) @@ -423,12 +428,11 @@ def _process_subscription_event( subscribed_count, expected_count, ) - if channel not in self._expected_channels: - logger.warning( - "Received channel_subscribed for unexpected channel: %s", channel - ) - if subscribed_count >= expected_count: + if subscribed_count == expected_count: self._cancel_subscription_watchdog() + self._reconnect_attempts = 0 + self._last_close_code = None + self._last_close_msg = None logger.info("All requested channels subscribed (%d)", expected_count) return @@ -492,9 +496,13 @@ def _handle_open(self, ws: websocket.WebSocketApp) -> None: return if self._expected_channels: self._arm_subscription_watchdog(ws) - self._reconnect_attempts = 0 - self._last_close_code = None - self._last_close_msg = None + else: + # Clients without managed subscriptions (for example + # SessionWithUrl) are fully established as soon as the transport + # opens because no channel acknowledgements will arrive. + self._reconnect_attempts = 0 + self._last_close_code = None + self._last_close_msg = None self._connected.set() if self._on_open_cb: try: diff --git a/tests/unit/test_websocket_client.py b/tests/unit/test_websocket_client.py index 2efe0a7..dccc7fe 100644 --- a/tests/unit/test_websocket_client.py +++ b/tests/unit/test_websocket_client.py @@ -1057,11 +1057,83 @@ def disconnect_when_ready(): assert entered_backoff.is_set(), "Backoff was never entered" assert client._queue.get_nowait() is None - def test_handle_open_resets_reconnect_attempts(self, mock_session) -> None: + def test_reconnect_counter_resets_after_all_channels_subscribed( + self, mock_session + ) -> None: client = self._make_client(mock_session, max_reconnect_attempts=3) client._reconnect_attempts = 5 + client._last_close_code = 1006 + client._last_close_msg = "previous failure" + mock_ws = Mock() + + client._handle_open(mock_ws) + + assert client._reconnect_attempts == 5 + assert client._last_close_code == 1006 + assert client._last_close_msg == "previous failure" + + client._process_subscription_event( + mock_ws, {"event": "channel_subscribed", "channel": "/ch"} + ) + + assert client._reconnect_attempts == 0 + assert client._last_close_code is None + assert client._last_close_msg is None + + def test_handle_open_resets_reconnect_attempts_without_channels( + self, mock_session + ) -> None: + client = self._make_client(mock_session, channels=[], max_reconnect_attempts=3) + client._reconnect_attempts = 5 + client._last_close_code = 1006 + client._last_close_msg = "previous failure" + client._handle_open(Mock()) + assert client._reconnect_attempts == 0 + assert client._last_close_code is None + assert client._last_close_msg is None + + def test_subscription_failures_honor_retry_limit_and_backoff( + self, mock_session + ) -> None: + client = self._make_client( + mock_session, max_reconnect_attempts=2, reconnect_backoff=0.01 + ) + call_count = 0 + observed_delays: list[float] = [] + + def fake_run_forever(**kwargs): + nonlocal call_count + call_count += 1 + client._handle_open(client._ws) + client._process_subscription_event( + client._ws, + { + "event": "subscribe_failed", + "channel": "/ch", + "detail": "denied", + }, + ) + client._handle_close(client._ws, None, None) + + def capture_delay(timeout=None): + if timeout is not None: + observed_delays.append(timeout) + return False + + mock_ws = Mock() + mock_ws.run_forever.side_effect = fake_run_forever + with ( + patch.object(client, "_create_ws_app", return_value=mock_ws), + patch.object(client, "_arm_subscription_watchdog"), + patch.object(client._user_disconnect, "wait", side_effect=capture_delay), + ): + client._ws = mock_ws + client._run_forever_safe() + + assert call_count == 3 # initial connection plus two retries + assert observed_delays == pytest.approx([0.01, 0.02]) def test_successful_reconnect_resets_counter(self, mock_session) -> None: client = self._make_client(mock_session, max_reconnect_attempts=2) @@ -1076,10 +1148,18 @@ def fake_run_forever(**kwargs): elif call_count == 2: # Reconnect succeeds, then simulate open + later drop client._handle_open(client._ws) + client._process_subscription_event( + client._ws, + {"event": "channel_subscribed", "channel": "/ch"}, + ) client._handle_close(client._ws, 1006, "drop again") elif call_count == 3: # Another reconnect succeeds, then clean exit client._handle_open(client._ws) + client._process_subscription_event( + client._ws, + {"event": "channel_subscribed", "channel": "/ch"}, + ) client._handle_close(client._ws, 1006, "drop again") elif call_count == 4: # Final reconnect succeeds then user disconnects @@ -1648,6 +1728,30 @@ def test_watchdog_cancelled_when_all_channels_subscribed( assert errors == [] mock_ws.close.assert_not_called() + def test_unexpected_ack_does_not_count_toward_subscription_completion( + self, mock_session + ) -> None: + client = _MistWebsocket( + mock_session, + channels=["/ch1", "/ch2"], + ) + mock_ws = Mock() + watchdog = Mock() + client._subscription_watchdog = watchdog + + client._process_subscription_event( + mock_ws, + {"event": "channel_subscribed", "channel": "/unexpected"}, + ) + client._process_subscription_event( + mock_ws, + {"event": "channel_subscribed", "channel": "/ch1"}, + ) + + assert client._subscribed_channels == {"/ch1"} + assert client._subscription_watchdog is watchdog + watchdog.cancel.assert_not_called() + def test_subscribe_failed_reports_on_error_and_closes(self, ws_client) -> None: errors = [] ws_client.on_error(errors.append) From 055cd3603758137f27dd5b6f067490a36ad8b941 Mon Sep 17 00:00:00 2001 From: Thomas Munzer Date: Tue, 14 Jul 2026 20:09:19 +0200 Subject: [PATCH 7/8] docs(changelog): add 0.63.3 websocket reliability notes --- CHANGELOG.md | 43 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9842de4..36a0bc1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,47 @@ # CHANGELOG + +## Version 0.63.3 (July 2026) + +**Released**: July 14, 2026 + +This patch release improves WebSocket subscription reliability, isolates user callbacks from the receive loop, and adds stronger reconnect, rate-limit, and connection-health safeguards. (#26) + +--- + +### 1. NEW FEATURES + +#### **Subscription Acknowledgement Watchdog** +WebSocket clients now track `channel_subscribed` acknowledgements for every requested channel. The configurable `subscription_watchdog_timeout` closes an incomplete connection and reports a `TimeoutError` through `on_error`; when `auto_reconnect=True`, the client reconnects and retries the subscriptions. Explicit `subscribe_failed` events are also surfaced through `on_error` before reconnecting. + +#### **Ping and Pong Hooks** +Added `on_ping()` and `on_pong()` callbacks for applications that need visibility into WebSocket keepalive frames. + +--- + +### 2. IMPROVEMENTS + +#### **Non-Blocking Callback Delivery** +Incoming frames are now placed on a dedicated callback-worker queue so slow user callbacks no longer block the WebSocket receive loop. Pending callback messages are drained during shutdown, and `queue_maxsize` bounds both generator and callback delivery queues. + +#### **Reconnect and Rate-Limit Handling** +- HTTP 429 handshake failures use the configurable `rate_limit_backoff` as a minimum reconnect delay. +- Exponential reconnect state is reset only after all requested subscriptions are acknowledged, so persistent subscription failures respect `max_reconnect_attempts` and continue increasing the backoff. +- `SessionWithUrl` connections with no managed channels reset reconnect state when the transport opens. + +#### **Connection Health and Observability** +- Updated keepalive defaults to `ping_interval=60` and a derived `ping_timeout` of up to 45 seconds. +- Added periodic, thread-safe throughput and queue-depth logging through `throughput_log_interval`. +- Duplicate channels are removed while preserving order, high channel counts produce a warning, and connections above the 2,000-channel limit are rejected. + +--- + +### 3. BUG FIXES + +#### **Subscription Completion Accounting** +Unexpected `channel_subscribed` acknowledgements are now ignored before updating subscription progress, preventing them from cancelling the watchdog while requested channels are still missing. + +--- + ## Version 0.63.2 (July 2026) **Released**: July 14, 2026 From 9de1ffcc8e56acc12d4939dc7610bff846cec202 Mon Sep 17 00:00:00 2001 From: Thomas Munzer Date: Tue, 14 Jul 2026 20:11:17 +0200 Subject: [PATCH 8/8] fix: update version to 0.63.3 in pyproject.toml --- pyproject.toml | 2 +- src/mistapi/__version.py | 2 +- src/mistapi/api/v1/sites/sle.py | 4 ++-- uv.lock | 2 +- 4 files changed, 5 insertions(+), 5 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index f854d13..d36784b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "mistapi" -version = "0.63.2" +version = "0.63.3" authors = [{ name = "Thomas Munzer", email = "tmunzer@juniper.net" }] description = "Python package to simplify the Mist System APIs usage" keywords = ["Mist", "Juniper", "API"] diff --git a/src/mistapi/__version.py b/src/mistapi/__version.py index 71462fc..88d01e1 100644 --- a/src/mistapi/__version.py +++ b/src/mistapi/__version.py @@ -1,2 +1,2 @@ -__version__ = "0.63.2" +__version__ = "0.63.3" __author__ = "Thomas Munzer " diff --git a/src/mistapi/api/v1/sites/sle.py b/src/mistapi/api/v1/sites/sle.py index 7e478c8..2d044aa 100644 --- a/src/mistapi/api/v1/sites/sle.py +++ b/src/mistapi/api/v1/sites/sle.py @@ -18,7 +18,7 @@ @deprecation.deprecated( deprecated_in="0.59.2", removed_in="0.65.0", - current_version="0.63.2", + current_version="0.63.3", details="function replaced with getSiteSleClassifierSummaryTrend", ) def getSiteSleClassifierDetails( @@ -764,7 +764,7 @@ def listSiteSleImpactedWirelessClients( @deprecation.deprecated( deprecated_in="0.59.2", removed_in="0.65.0", - current_version="0.63.2", + current_version="0.63.3", details="function replaced with getSiteSleSummaryTrend", ) def getSiteSleSummary( diff --git a/uv.lock b/uv.lock index 2233da5..8866d8e 100644 --- a/uv.lock +++ b/uv.lock @@ -552,7 +552,7 @@ wheels = [ [[package]] name = "mistapi" -version = "0.63.2" +version = "0.63.3" source = { editable = "." } dependencies = [ { name = "deprecation" },