ws client queue_maxsize hardening

Author: tmunzer-AIDE

README.md

@@ -588,7 +588,7 @@ All channel classes accept the following optional keyword arguments:
 | `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, the receive thread blocks if the queue is full, providing backpressure for high-frequency streams. |
+| `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. |
 
 ```python
 ws = mistapi.websockets.sites.DeviceStatsEvents(

src/mistapi/websockets/__ws_client.py

@@ -196,7 +196,10 @@ def _handle_message(self, ws: websocket.WebSocketApp, message: str | bytes) -> N
             except Exception:
                 logger.exception("on_message callback raised")
         else:
-            self._queue.put(data)
+            try:
+                self._queue.put_nowait(data)
+            except queue.Full:
+                logger.warning("Receive queue full; dropping message")
 
     def _handle_error(self, ws: websocket.WebSocketApp, error: Exception) -> None:
         if self._on_error_cb:

src/mistapi/websockets/location.py

@@ -42,8 +42,8 @@ class BleAssetsEvents(_MistWebsocket):
     queue_maxsize : int, default 0
         Maximum number of messages buffered in the internal queue for the
         ``receive()`` generator. ``0`` means unbounded. When set, the
-        websocket-client receive thread blocks if the queue is full,
-        providing backpressure for high-frequency streams.
+        incoming messages are dropped with a warning when the queue is
+        full, preventing memory growth on high-frequency streams.
 
     EXAMPLE
     -----------
@@ -122,8 +122,8 @@ class ConnectedClientsEvents(_MistWebsocket):
     queue_maxsize : int, default 0
         Maximum number of messages buffered in the internal queue for the
         ``receive()`` generator. ``0`` means unbounded. When set, the
-        websocket-client receive thread blocks if the queue is full,
-        providing backpressure for high-frequency streams.
+        incoming messages are dropped with a warning when the queue is
+        full, preventing memory growth on high-frequency streams.
 
     EXAMPLE
     -----------
@@ -202,8 +202,8 @@ class SdkClientsEvents(_MistWebsocket):
     queue_maxsize : int, default 0
         Maximum number of messages buffered in the internal queue for the
         ``receive()`` generator. ``0`` means unbounded. When set, the
-        websocket-client receive thread blocks if the queue is full,
-        providing backpressure for high-frequency streams.
+        incoming messages are dropped with a warning when the queue is
+        full, preventing memory growth on high-frequency streams.
 
     EXAMPLE
     -----------
@@ -282,8 +282,8 @@ class UnconnectedClientsEvents(_MistWebsocket):
     queue_maxsize : int, default 0
         Maximum number of messages buffered in the internal queue for the
         ``receive()`` generator. ``0`` means unbounded. When set, the
-        websocket-client receive thread blocks if the queue is full,
-        providing backpressure for high-frequency streams.
+        incoming messages are dropped with a warning when the queue is
+        full, preventing memory growth on high-frequency streams.
 
     EXAMPLE
     -----------
@@ -364,8 +364,8 @@ class DiscoveredBleAssetsEvents(_MistWebsocket):
     queue_maxsize : int, default 0
         Maximum number of messages buffered in the internal queue for the
         ``receive()`` generator. ``0`` means unbounded. When set, the
-        websocket-client receive thread blocks if the queue is full,
-        providing backpressure for high-frequency streams.
+        incoming messages are dropped with a warning when the queue is
+        full, preventing memory growth on high-frequency streams.
 
     EXAMPLE
     -----------

src/mistapi/websockets/orgs.py

@@ -40,8 +40,8 @@ class InsightsEvents(_MistWebsocket):
     queue_maxsize : int, default 0
         Maximum number of messages buffered in the internal queue for the
         ``receive()`` generator. ``0`` means unbounded. When set, the
-        websocket-client receive thread blocks if the queue is full,
-        providing backpressure for high-frequency streams.
+        incoming messages are dropped with a warning when the queue is
+        full, preventing memory growth on high-frequency streams.
 
     EXAMPLE
     -----------
@@ -116,8 +116,8 @@ class MxEdgesStatsEvents(_MistWebsocket):
     queue_maxsize : int, default 0
         Maximum number of messages buffered in the internal queue for the
         ``receive()`` generator. ``0`` means unbounded. When set, the
-        websocket-client receive thread blocks if the queue is full,
-        providing backpressure for high-frequency streams.
+        incoming messages are dropped with a warning when the queue is
+        full, preventing memory growth on high-frequency streams.
 
     EXAMPLE
     -----------
@@ -192,8 +192,8 @@ class MxEdgesEvents(_MistWebsocket):
     queue_maxsize : int, default 0
         Maximum number of messages buffered in the internal queue for the
         ``receive()`` generator. ``0`` means unbounded. When set, the
-        websocket-client receive thread blocks if the queue is full,
-        providing backpressure for high-frequency streams.
+        incoming messages are dropped with a warning when the queue is
+        full, preventing memory growth on high-frequency streams.
 
     EXAMPLE
     -----------

src/mistapi/websockets/session.py

@@ -47,8 +47,8 @@ class SessionWithUrl(_MistWebsocket):
     queue_maxsize : int, default 0
         Maximum number of messages buffered in the internal queue for the
         ``receive()`` generator. ``0`` means unbounded. When set, the
-        websocket-client receive thread blocks if the queue is full,
-        providing backpressure for high-frequency streams.
+        incoming messages are dropped with a warning when the queue is
+        full, preventing memory growth on high-frequency streams.
 
     EXAMPLE
     -----------

src/mistapi/websockets/sites.py

@@ -40,8 +40,8 @@ class ClientsStatsEvents(_MistWebsocket):
     queue_maxsize : int, default 0
         Maximum number of messages buffered in the internal queue for the
         ``receive()`` generator. ``0`` means unbounded. When set, the
-        websocket-client receive thread blocks if the queue is full,
-        providing backpressure for high-frequency streams.
+        incoming messages are dropped with a warning when the queue is
+        full, preventing memory growth on high-frequency streams.
 
     EXAMPLE
     -----------
@@ -125,8 +125,8 @@ class DeviceCmdEvents(_MistWebsocket):
     queue_maxsize : int, default 0
         Maximum number of messages buffered in the internal queue for the
         ``receive()`` generator. ``0`` means unbounded. When set, the
-        websocket-client receive thread blocks if the queue is full,
-        providing backpressure for high-frequency streams.
+        incoming messages are dropped with a warning when the queue is
+        full, preventing memory growth on high-frequency streams.
 
     EXAMPLE
     -----------
@@ -205,8 +205,8 @@ class DeviceStatsEvents(_MistWebsocket):
     queue_maxsize : int, default 0
         Maximum number of messages buffered in the internal queue for the
         ``receive()`` generator. ``0`` means unbounded. When set, the
-        websocket-client receive thread blocks if the queue is full,
-        providing backpressure for high-frequency streams.
+        incoming messages are dropped with a warning when the queue is
+        full, preventing memory growth on high-frequency streams.
 
     EXAMPLE
     -----------
@@ -282,8 +282,8 @@ class DeviceEvents(_MistWebsocket):
     queue_maxsize : int, default 0
         Maximum number of messages buffered in the internal queue for the
         ``receive()`` generator. ``0`` means unbounded. When set, the
-        websocket-client receive thread blocks if the queue is full,
-        providing backpressure for high-frequency streams.
+        incoming messages are dropped with a warning when the queue is
+        full, preventing memory growth on high-frequency streams.
 
     EXAMPLE
     -----------
@@ -359,8 +359,8 @@ class MxEdgesStatsEvents(_MistWebsocket):
     queue_maxsize : int, default 0
         Maximum number of messages buffered in the internal queue for the
         ``receive()`` generator. ``0`` means unbounded. When set, the
-        websocket-client receive thread blocks if the queue is full,
-        providing backpressure for high-frequency streams.
+        incoming messages are dropped with a warning when the queue is
+        full, preventing memory growth on high-frequency streams.
 
     EXAMPLE
     -----------
@@ -436,8 +436,8 @@ class MxEdgesEvents(_MistWebsocket):
     queue_maxsize : int, default 0
         Maximum number of messages buffered in the internal queue for the
         ``receive()`` generator. ``0`` means unbounded. When set, the
-        websocket-client receive thread blocks if the queue is full,
-        providing backpressure for high-frequency streams.
+        incoming messages are dropped with a warning when the queue is
+        full, preventing memory growth on high-frequency streams.
 
     EXAMPLE
     -----------
@@ -513,8 +513,8 @@ class PcapEvents(_MistWebsocket):
     queue_maxsize : int, default 0
         Maximum number of messages buffered in the internal queue for the
         ``receive()`` generator. ``0`` means unbounded. When set, the
-        websocket-client receive thread blocks if the queue is full,
-        providing backpressure for high-frequency streams.
+        incoming messages are dropped with a warning when the queue is
+        full, preventing memory growth on high-frequency streams.
 
     EXAMPLE
     -----------

tests/unit/test_websocket_client.py

@@ -1096,6 +1096,17 @@ def test_receive_raises_when_message_callback_registered(self, ws_client) -> Non
         with pytest.raises(RuntimeError, match="on_message callback"):
             list(ws_client.receive())
 
+    def test_queue_full_drops_message(self, mock_session) -> None:
+        client = _MistWebsocket(mock_session, channels=["/ch"], queue_maxsize=1)
+        # Fill the queue
+        client._handle_message(Mock(), '{"event": "first"}')
+        assert not client._queue.empty()
+        # Second message should be dropped, not block
+        client._handle_message(Mock(), '{"event": "dropped"}')
+        # Only the first message should be in the queue
+        assert client._queue.get_nowait() == {"event": "first"}
+        assert client._queue.empty()
+
 
 # ---------------------------------------------------------------------------
 # disconnect(wait=...)