diff --git a/plugins/platforms/buzz/adapter.py b/plugins/platforms/buzz/adapter.py index b99be70d95..6e2b16ff34 100644 --- a/plugins/platforms/buzz/adapter.py +++ b/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[0-9a-f]{64})(?P\.[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): diff --git a/tests/gateway/test_buzz_websocket.py b/tests/gateway/test_buzz_websocket.py index 543a55896b..320d4d050f 100644 --- a/tests/gateway/test_buzz_websocket.py +++ b/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