diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index fd5fc0177eba..5d4c095e3fbe 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -1175,12 +1175,23 @@ def _looks_like_polling_conflict(error: Exception) -> bool: @staticmethod def _looks_like_network_error(error: Exception) -> bool: - """Return True for transient network errors that warrant a reconnect attempt.""" + """Return True for transient transport failures that warrant reconnect.""" name = error.__class__.__name__.lower() + if name in {"badrequest", "invalidtoken", "forbidden", "retryafter"}: + return False if name in {"networkerror", "timedout", "connectionerror"}: return True try: - from telegram.error import NetworkError, TimedOut + from telegram.error import ( + BadRequest, + Forbidden, + InvalidToken, + NetworkError, + RetryAfter, + TimedOut, + ) + if isinstance(error, (BadRequest, InvalidToken, Forbidden, RetryAfter)): + return False if isinstance(error, (NetworkError, TimedOut)): return True except ImportError: @@ -2204,17 +2215,11 @@ async def _polling_heartbeat_loop(self) -> None: except asyncio.CancelledError: return except (asyncio.TimeoutError, OSError) as probe_err: - logger.warning( - "[%s] Polling heartbeat probe failed (%s); triggering reconnect", - self.name, probe_err, - ) - if self._polling_error_task and not self._polling_error_task.done(): - continue # reconnect already in progress - loop = asyncio.get_running_loop() - self._polling_error_task = loop.create_task( - self._handle_polling_network_error(probe_err) - ) - except Exception: + self._schedule_polling_recovery(probe_err, reason="heartbeat probe") + except Exception as probe_err: + if self._looks_like_network_error(probe_err): + self._schedule_polling_recovery(probe_err, reason="heartbeat probe") + continue # Non-connectivity errors (e.g. TelegramError 401) are not # CLOSE-WAIT symptoms — let PTB's own handlers surface them. pass @@ -3192,10 +3197,6 @@ def _with_limits(httpx_kwargs: Optional[dict] = None) -> dict: # Start polling — retry initialize() for transient TLS resets. # Each attempt is capped by _init_timeout so a single unreachable # fallback-IP chain can't block startup indefinitely. - try: - from telegram.error import NetworkError, TimedOut - except ImportError: - NetworkError = TimedOut = OSError # type: ignore[misc,assignment] _max_connect = 8 _init_timeout = _env_float("HERMES_TELEGRAM_INIT_TIMEOUT", 30.0) for _attempt in range(_max_connect): @@ -3230,7 +3231,19 @@ def _with_limits(httpx_kwargs: Optional[dict] = None) -> dict: f"({_init_timeout:.0f}s each). Check network connectivity to api.telegram.org " f"or set HERMES_TELEGRAM_HTTP_CONNECT_TIMEOUT to a lower value." ) - except (NetworkError, TimedOut, OSError) as init_err: + except OSError as init_err: + if _attempt < _max_connect - 1: + wait = min(2 ** _attempt, 15) + logger.warning( + "[%s] Connect attempt %d/%d failed: %s — retrying in %ds", + self.name, _attempt + 1, _max_connect, init_err, wait, + ) + await asyncio.sleep(wait) + else: + raise + except Exception as init_err: + if not self._looks_like_network_error(init_err): + raise if _attempt < _max_connect - 1: wait = min(2 ** _attempt, 15) logger.warning( diff --git a/tests/gateway/conftest.py b/tests/gateway/conftest.py index 3546b09cc754..978d634331a6 100644 --- a/tests/gateway/conftest.py +++ b/tests/gateway/conftest.py @@ -69,15 +69,21 @@ def _ensure_telegram_mock() -> None: mod.constants.ChatType.SUPERGROUP = "supergroup" mod.constants.ChatType.CHANNEL = "channel" - # Real exception classes so ``except (NetworkError, ...)`` clauses - # in production code don't blow up with TypeError. - mod.error.NetworkError = type("NetworkError", (OSError,), {}) - mod.error.TimedOut = type("TimedOut", (OSError,), {}) - mod.error.BadRequest = type("BadRequest", (Exception,), {}) - mod.error.Forbidden = type("Forbidden", (Exception,), {}) - mod.error.InvalidToken = type("InvalidToken", (Exception,), {}) - mod.error.RetryAfter = type("RetryAfter", (Exception,), {"retry_after": 1}) - mod.error.Conflict = type("Conflict", (Exception,), {}) + # Mirror PTB's exception hierarchy: BadRequest is a semantic API error, + # but inherits from NetworkError in python-telegram-bot 22.x. + mod.error.TelegramError = type("TelegramError", (Exception,), {}) + mod.error.NetworkError = type("NetworkError", (mod.error.TelegramError,), {}) + mod.error.TimedOut = type("TimedOut", (mod.error.NetworkError,), {}) + mod.error.BadRequest = type("BadRequest", (mod.error.NetworkError,), {}) + mod.error.Forbidden = type("Forbidden", (mod.error.TelegramError,), {}) + mod.error.InvalidToken = type("InvalidToken", (mod.error.TelegramError,), {}) + + class RetryAfter(mod.error.TelegramError): + def __init__(self, retry_after=1): + self.retry_after = retry_after + + mod.error.RetryAfter = RetryAfter + mod.error.Conflict = type("Conflict", (mod.error.TelegramError,), {}) # Update.ALL_TYPES used in start_polling() mod.Update.ALL_TYPES = [] diff --git a/tests/gateway/test_telegram_network_reconnect.py b/tests/gateway/test_telegram_network_reconnect.py index 0016b0e32e75..463877bf0728 100644 --- a/tests/gateway/test_telegram_network_reconnect.py +++ b/tests/gateway/test_telegram_network_reconnect.py @@ -6,7 +6,9 @@ rather than silently leaving polling dead. """ +import ast import asyncio +from pathlib import Path import sys from unittest.mock import AsyncMock, MagicMock, patch @@ -678,6 +680,123 @@ async def telegram_error_wait_for(coro, timeout): adapter._handle_polling_network_error.assert_not_awaited() +async def _heartbeat_exception_case(exc, *, pending_probe=False): + adapter = _make_adapter() + reconnect_handler = AsyncMock() + adapter._handle_polling_network_error = reconnect_handler # type: ignore[method-assign] + mock_app = MagicMock() + mock_app.updater.running = True + if pending_probe: + mock_app.bot.get_me = AsyncMock(return_value=MagicMock()) + mock_app.bot.get_webhook_info = AsyncMock(side_effect=exc) + else: + mock_app.bot.get_me = AsyncMock(side_effect=exc) + adapter._app = mock_app + + sleep_calls = 0 + + async def fast_sleep(_seconds): + nonlocal sleep_calls + sleep_calls += 1 + if sleep_calls >= 2: + raise asyncio.CancelledError() + + with patch("asyncio.sleep", side_effect=fast_sleep): + await adapter._polling_heartbeat_loop() + await asyncio.sleep(0) + return adapter + + +@pytest.mark.asyncio +@pytest.mark.parametrize("pending_probe", [False, True]) +async def test_heartbeat_routes_ptb_transport_errors_to_reconnect(pending_probe): + from telegram.error import NetworkError, TimedOut + + for exc in (NetworkError("network"), TimedOut("timeout")): + adapter = await _heartbeat_exception_case(exc, pending_probe=pending_probe) + reconnect_handler = adapter._handle_polling_network_error + assert isinstance(reconnect_handler, AsyncMock) + reconnect_handler.assert_awaited_once_with(exc) + assert adapter._polling_error_task is not None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("pending_probe", [False, True]) +async def test_heartbeat_ignores_ptb_semantic_errors(pending_probe): + from telegram.error import BadRequest, Forbidden, InvalidToken, RetryAfter + + for exc in ( + BadRequest("bad request"), + Forbidden("forbidden"), + InvalidToken("invalid token"), + RetryAfter(1), + ): + adapter = await _heartbeat_exception_case(exc, pending_probe=pending_probe) + reconnect_handler = adapter._handle_polling_network_error + assert isinstance(reconnect_handler, AsyncMock) + reconnect_handler.assert_not_awaited() + assert adapter._polling_error_task is None + + +@pytest.mark.parametrize( + ("error_name", "expected"), + [ + ("NetworkError", True), + ("TimedOut", True), + ("BadRequest", False), + ("Forbidden", False), + ("InvalidToken", False), + ("RetryAfter", False), + ], +) +def test_network_error_classifier_matches_ptb_semantics(error_name, expected): + import telegram.error as telegram_error + + error_type = getattr(telegram_error, error_name) + error = error_type(1) if error_name == "RetryAfter" else error_type(error_name) + assert TelegramAdapter._looks_like_network_error(error) is expected + + +def _calls_shared_network_classifier(node): + return any( + isinstance(child, ast.Call) + and isinstance(child.func, ast.Attribute) + and child.func.attr == "_looks_like_network_error" + for child in ast.walk(node) + ) + + +def test_polling_error_callback_uses_shared_network_classifier(): + source = Path(TelegramAdapter.connect.__code__.co_filename).read_text(encoding="utf-8") + tree = ast.parse(source) + callbacks = [ + node + for node in ast.walk(tree) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and node.name == "_polling_error_callback" + ] + assert len(callbacks) == 1 + assert _calls_shared_network_classifier(callbacks[0]) + + +def test_connect_initialize_retry_uses_shared_network_classifier(): + source = Path(TelegramAdapter.connect.__code__.co_filename).read_text(encoding="utf-8") + tree = ast.parse(source) + connect = next( + node + for node in ast.walk(tree) + if isinstance(node, ast.AsyncFunctionDef) and node.name == "connect" + ) + exception_handlers = [ + node + for node in ast.walk(connect) + if isinstance(node, ast.ExceptHandler) + and isinstance(node.type, ast.Name) + and node.type.id == "Exception" + ] + assert any(_calls_shared_network_classifier(handler) for handler in exception_handlers) + + @pytest.mark.asyncio async def test_heartbeat_loop_exits_on_fatal_error(): """A fatal error short-circuits the loop before probing get_me()."""