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 <noreply@anthropic.com>

Author: tmunzer-AIDE

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. |
 

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()

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,