fix(buzz): recover from uninterruptible websocket reads

plugins/platforms/buzz/adapter.py

@@ -161,6 +161,13 @@ _BARE_MEDIA_RE = re.compile(_MEDIA_URL_PATTERN, re.IGNORECASE)
 _MEDIA_PATH_RE = re.compile(r"^/media/(?P<sha>[0-9a-f]{64})(?P<ext>\.[a-z0-9]{1,10})?/?$", re.IGNORECASE)
 
 
+def _consume_ws_read_task(task: asyncio.Task) -> None:
+    """Retrieve a detached stalled receive task's result without blocking recovery."""
+    if not task.cancelled():
+        with contextlib.suppress(Exception):
+            task.exception()
+
+
 def _effective_port(parsed) -> Optional[int]:
     try:
         if parsed.port is not None:
@@ -1117,12 +1124,30 @@ class BuzzAdapter(BasePlatformAdapter):
         """Read frames until the relay closes; an idle read raises ConnectionError to reconnect."""
         frame_iter = websocket.__aiter__()
         while True:
+            read_task = asyncio.ensure_future(frame_iter.__anext__())
             try:
-                raw = await asyncio.wait_for(frame_iter.__anext__(), timeout=_WS_READ_IDLE_TIMEOUT)
+                done, _ = await asyncio.wait(
+                    {read_task}, timeout=_WS_READ_IDLE_TIMEOUT, return_when=asyncio.FIRST_COMPLETED
+                )
+                if not done:
+                    # wait_for() cancels and then waits for its awaitable to acknowledge the cancellation.
+                    # A transport receive stuck below asyncio can ignore that cancellation forever, leaving the
+                    # adapter healthy-looking. Detach the read instead so the outer loop can close and reconnect.
+                    self._mark_degraded()
+                    logger.warning(
+                        "Buzz: WebSocket read watchdog expired after %.0fs; receive task remained parked; reconnecting",
+                        _WS_READ_IDLE_TIMEOUT,
+                    )
+                    raise ConnectionError(
+                        f"no WebSocket frame for {_WS_READ_IDLE_TIMEOUT:.0f}s; assuming the connection went silent"
+                    )
+                raw = read_task.result()
             except StopAsyncIteration:
                 return
-            except asyncio.TimeoutError:
-                raise ConnectionError(f"no WebSocket frame for {_WS_READ_IDLE_TIMEOUT:.0f}s; assuming the connection went silent") from None
+            finally:
+                if not read_task.done():
+                    read_task.cancel()
+                    read_task.add_done_callback(_consume_ws_read_task)
             try:
                 message = json.loads(raw)
             except (ValueError, TypeError):

tests/gateway/test_buzz_websocket.py

@@ -147,11 +147,21 @@ async def test_websocket_loop_reconnects_when_read_goes_silent(monkeypatch, capl
     adapter = _make_adapter()
     monkeypatch.setattr(_buzz_mod, "_WS_READ_IDLE_TIMEOUT", 0.05)
     caplog.set_level(logging.WARNING)
+    degraded = []
+    monkeypatch.setattr(adapter, "_mark_degraded", lambda: degraded.append(True))
 
     sockets = []
+    release_parked_receive = asyncio.Event()
 
     async def dead_anext():
-        await asyncio.Event().wait()  # never yields, never raises
+        # Some transports do not acknowledge cancellation while their receive
+        # call is parked.  ``asyncio.wait_for`` then waits for this coroutine
+        # to finish, so its own timeout cannot return.
+        try:
+            await asyncio.Event().wait()
+        except asyncio.CancelledError:
+            await release_parked_receive.wait()
+            raise
 
     def fake_connect(*args, **kwargs):
         ws = _ScriptedWebSocket(dead_anext)
@@ -168,6 +178,7 @@ async def test_websocket_loop_reconnects_when_read_goes_silent(monkeypatch, capl
         while len(sockets) < 2 and time.monotonic() < deadline:
             await asyncio.sleep(0.02)
     finally:
+        release_parked_receive.set()
         task.cancel()
         try:
             await asyncio.wait_for(task, 5.0)
@@ -177,6 +188,8 @@ async def test_websocket_loop_reconnects_when_read_goes_silent(monkeypatch, capl
     assert len(sockets) >= 2, "idle read watchdog did not force a reconnect"
     assert sockets[0].exited, "the silent connection was not closed before reconnecting"
     assert any("went silent" in record.message for record in caplog.records)
+    assert any("receive task remained parked" in record.message for record in caplog.records)
+    assert degraded, "a parked receive must publish degraded health"
 
 
 @pytest.mark.asyncio