diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 7f451f8c4c0fc..f37c41c0c6dbf 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -543,6 +543,14 @@ _UPDATER_STOP_TIMEOUT = 15.0 # reconnect ladder from stalling indefinitely and allows the heartbeat loop to # trigger its own recovery path. Refs: NousResearch/hermes-agent#59614 _UPDATER_START_TIMEOUT = 30.0 +# shutdown()/initialize() on the getUpdates httpx request close and rebuild the +# connection pool. When a connection is wedged on a stale CLOSE-WAIT socket that +# close can block forever, hanging _drain_polling_connections() and freezing the +# whole reconnect ladder (the tracked _polling_error_task never completes, so +# every escalation path stays gated behind its in-flight guard). Bound the drain +# so the ladder always advances toward the fatal-restart escalation. Matches +# _UPDATER_STOP_TIMEOUT. Refs: NousResearch/hermes-agent#66377 +_DRAIN_TIMEOUT = 15.0 # A generation is not healthy until the dedicated getUpdates request returns # successfully. This exceeds a normal long-poll cycle for healthy idle bots. _POLLING_PROGRESS_TIMEOUT = 60.0 @@ -1987,20 +1995,22 @@ class TelegramAdapter(BasePlatformAdapter): except Exception: return try: - await polling_req.shutdown() + # Bounded: a wedged CLOSE-WAIT socket can make this close hang + # forever and freeze the reconnect ladder (#66377). + await asyncio.wait_for(polling_req.shutdown(), timeout=_DRAIN_TIMEOUT) except Exception: logger.debug( - "[%s] Polling request shutdown failed (non-fatal)", + "[%s] Polling request shutdown failed/timed out (non-fatal)", self.name, exc_info=True, ) try: - await polling_req.initialize() + await asyncio.wait_for(polling_req.initialize(), timeout=_DRAIN_TIMEOUT) logger.debug( "[%s] Polling request pool drained before reconnect", self.name ) except Exception: logger.debug( - "[%s] Polling request re-initialize failed (non-fatal)", + "[%s] Polling request re-initialize failed/timed out (non-fatal)", self.name, exc_info=True, ) diff --git a/tests/gateway/test_telegram_network_reconnect.py b/tests/gateway/test_telegram_network_reconnect.py index c1c10726755dd..7b8ed87a0ba43 100644 --- a/tests/gateway/test_telegram_network_reconnect.py +++ b/tests/gateway/test_telegram_network_reconnect.py @@ -325,6 +325,54 @@ async def test_initialize_still_runs_when_shutdown_fails(): mock_app.updater.start_polling.assert_called_once() +@pytest.mark.asyncio +async def test_reconnect_continues_if_drain_hangs(monkeypatch): + """If the polling request drain HANGS (wedged httpx pool close on a + CLOSE-WAIT socket), the reconnect ladder must still advance rather than + freezing the tracked _polling_error_task forever. + + Regression test for #66377: an unbounded ``shutdown()`` / + ``initialize()`` in ``_drain_polling_connections`` leaves the handler + task pending, which gates every escalation path and silently kills the + gateway. The drain awaits are bounded by ``_DRAIN_TIMEOUT``, so the + handler must complete and reach ``start_polling`` within a hard bound. + """ + adapter = _make_adapter() + adapter._polling_network_error_count = 1 + + mock_app, mock_polling_req = _make_mock_app() + + async def _hang(*args, **kwargs): + await asyncio.Event().wait() # never returns + + # Both drain awaits wedge indefinitely. + mock_polling_req.shutdown = AsyncMock(side_effect=_hang) + mock_polling_req.initialize = AsyncMock(side_effect=_hang) + adapter._app = mock_app + + # Keep the drain timeout tiny so the test stays fast; the real default + # is generous enough not to truncate healthy closes. + monkeypatch.setattr(tg_adapter, "_DRAIN_TIMEOUT", 0.01, raising=False) + + with patch("asyncio.sleep", new_callable=AsyncMock): + # Hard outer bound: on unfixed code the drain hangs forever and this + # trips; with the fix the inner wait_for releases well before it. + await asyncio.wait_for( + adapter._handle_polling_network_error(Exception("Timed out")), + timeout=5, + ) + + # Ladder advanced past the wedged drain despite it never returning. + mock_app.updater.start_polling.assert_called_once() + assert adapter._polling_network_error_count == 2 + # The tracked task must not be stuck pending — otherwise every + # escalation path stays gated behind an in-flight guard. + assert ( + adapter._polling_error_task is None + or adapter._polling_error_task.done() + ) + + @pytest.mark.asyncio async def test_conflict_retry_also_drains_polling_connections(): """_handle_polling_conflict must also drain the polling pool on retry."""