diff --git a/gateway/platforms/qqbot/adapter.py b/gateway/platforms/qqbot/adapter.py index 5b4a396ed2fd..05697dd81f8f 100644 --- a/gateway/platforms/qqbot/adapter.py +++ b/gateway/platforms/qqbot/adapter.py @@ -675,11 +675,39 @@ async def _reconnect(self, backoff_idx: int) -> bool: return True except Exception as exc: logger.warning("[%s] Reconnect failed: %s", self._log_tag, exc) + # Clear stale WebSocket reference so _read_events() raises on + # the next call instead of returning silently on a closed socket. + # Without this, a closed-but-assigned _ws causes _read_events() + # to skip its while-loop and return normally, which resets + # backoff_idx in _listen_loop and creates a CPU-spinning + # tight loop with no await points. See #17703. + self._ws = None + # Also close the session if _open_ws() created one before + # failing — otherwise it leaks until the next successful + # reconnect or final disconnect cleanup. + if self._session and not self._session.closed: + await self._session.close() + self._session = None return False async def _read_events(self) -> None: - """Read WebSocket frames until connection closes.""" - if not self._ws: + """Read WebSocket frames until the connection closes. + + This coroutine must terminate by raising whenever the underlying + transport is unusable. Returning normally lets the outer + ``_listen_loop`` reset ``backoff_idx`` and re-enter immediately; + because no ``await`` in that path suspends, the event loop is + starved and the gateway spins at 100 % CPU (issue #17703). + + Three independent guards prevent that: + 1. Entry check — reject a stale closed socket up front. + 2. The while-loop itself — ``await receive()`` is the + suspension point that keeps the event loop alive. + 3. Post-loop fallback — if the loop exits without raising + while the adapter is still running, force an error so + ``_listen_loop`` takes the reconnect path. + """ + if not self._ws or self._ws.closed: raise RuntimeError("WebSocket not connected") while self._running and self._ws and not self._ws.closed: @@ -696,6 +724,15 @@ async def _read_events(self) -> None: elif msg.type in {aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR}: raise RuntimeError("WebSocket closed") + # Fallback: the while-loop exited without raising (e.g. ws.closed + # became True between the entry guard and the loop, or _ws was + # cleared externally). If the adapter is still supposed to be + # running, raise so _listen_loop schedules a reconnect instead of + # resetting backoff and re-entering this coroutine in a tight + # CPU-spinning loop with no await points. + if self._running: + raise RuntimeError("WebSocket closed") + async def _heartbeat_loop(self) -> None: """Send periodic heartbeats (QQ Gateway expects op 1 heartbeat with latest seq). diff --git a/tests/gateway/test_qqbot.py b/tests/gateway/test_qqbot.py index e1f41aeccdc6..e077f79dbbdc 100644 --- a/tests/gateway/test_qqbot.py +++ b/tests/gateway/test_qqbot.py @@ -2196,3 +2196,126 @@ def test_fatal_codes_include_intent_errors(self): assert 4014 in fatal_codes assert 4001 in fatal_codes assert 4915 in fatal_codes + + +# --------------------------------------------------------------------------- +# Issue #17703 — CPU-spinning tight loop regression tests +# --------------------------------------------------------------------------- + +class TestCPUSpinningRegression: + """Regression tests for #17703: WebSocket reconnect failure causes + CPU-spinning tight loop with no await points.""" + + def _make_adapter(self): + from gateway.platforms.qqbot.adapter import QQAdapter + return QQAdapter(_make_config(app_id="test", client_secret="test")) + + @pytest.mark.asyncio + async def test_read_events_raises_when_ws_closed(self): + """Guard 2: _read_events() must raise when _ws exists but is closed. + + Without this guard, _read_events() returns normally (skips the + while-loop), which resets backoff_idx in _listen_loop and creates + a CPU-spinning tight loop. See #17703. + """ + adapter = self._make_adapter() + adapter._running = True + adapter._ws = SimpleNamespace(closed=True, receive=mock.AsyncMock()) + + with pytest.raises(RuntimeError, match="WebSocket not connected"): + await adapter._read_events() + + @pytest.mark.asyncio + async def test_read_events_raises_when_ws_none(self): + """Guard 2: _read_events() must raise when _ws is None.""" + adapter = self._make_adapter() + adapter._running = True + adapter._ws = None + + with pytest.raises(RuntimeError, match="WebSocket not connected"): + await adapter._read_events() + + @pytest.mark.asyncio + async def test_read_events_closed_msg_raises(self): + """When ws.receive() returns CLOSED mid-loop, _read_events must raise + immediately instead of silently exiting. (Regression for #17703)""" + import aiohttp + adapter = self._make_adapter() + adapter._running = True + + # Simulate a ws that is open at entry but immediately reports + # CLOSED, causing the while-loop to exit via the CLOSED branch. + async def recv_closed(): + return SimpleNamespace( + type=aiohttp.WSMsgType.CLOSED, data=None, extra="" + ) + + adapter._ws = SimpleNamespace(closed=False, receive=recv_closed) + + with pytest.raises(RuntimeError, match="WebSocket closed"): + await adapter._read_events() + + @pytest.mark.asyncio + async def test_read_events_post_loop_guard(self): + """Guard 3: if ws.closed flips from False to True between entry + check and while condition, the loop never executes and the + post-loop fallback must raise. (Regression for #17703)""" + adapter = self._make_adapter() + adapter._running = True + + class ClosingWS: + """Returns closed=False on first check (entry guard), + then closed=True on second check (while condition).""" + def __init__(self): + self._closed_checks = 0 + + @property + def closed(self): + self._closed_checks += 1 + return self._closed_checks >= 2 + + adapter._ws = ClosingWS() + + with pytest.raises(RuntimeError, match="WebSocket closed"): + await adapter._read_events() + + @pytest.mark.asyncio + async def test_reconnect_failure_clears_ws_and_session(self): + """_reconnect() must clear _ws AND _session on failure so that + stale references don't leak resources or confuse _read_events().""" + adapter = self._make_adapter() + adapter._running = True + + # Pre-populate stale references + old_ws = SimpleNamespace(closed=True) + old_session = SimpleNamespace(closed=False, close=mock.AsyncMock()) + adapter._ws = old_ws + adapter._session = old_session + + # Make _ensure_token fail to trigger the except branch + adapter._ensure_token = mock.AsyncMock(side_effect=RuntimeError("DNS fail")) + adapter._http_client = mock.MagicMock() + + result = await adapter._reconnect(0) + + assert result is False + assert adapter._ws is None, "_ws should be cleared after failed reconnect" + assert adapter._session is None, "_session should be cleared after failed reconnect" + old_session.close.assert_called_once() + + @pytest.mark.asyncio + async def test_reconnect_failure_no_session_leak(self): + """_reconnect() should not fail if _session is already None.""" + adapter = self._make_adapter() + adapter._running = True + adapter._ws = SimpleNamespace(closed=True) + adapter._session = None # no session to clean + + adapter._ensure_token = mock.AsyncMock(side_effect=RuntimeError("DNS fail")) + adapter._http_client = mock.MagicMock() + + result = await adapter._reconnect(0) + + assert result is False + assert adapter._ws is None + assert adapter._session is None