diff --git a/gateway/platforms/telegram.py b/gateway/platforms/telegram.py index e17d104a68ae..e05b35e2f5cb 100644 --- a/gateway/platforms/telegram.py +++ b/gateway/platforms/telegram.py @@ -8,6 +8,7 @@ """ import asyncio +import httpx import json import logging import os @@ -178,6 +179,27 @@ def _looks_like_network_error(error: Exception) -> bool: pass return isinstance(error, OSError) + @staticmethod + def _is_retryable_send_network_error(error: Exception) -> bool: + """Retry only when the request clearly never reached Telegram. + + PTB wraps underlying httpx failures in higher-level telegram errors. + Connect/pool failures are safe to retry because no request was + delivered. Read/write timeouts are ambiguous and can duplicate sends. + """ + if "Request was *not* sent to Telegram" in str(error): + return True + + current: BaseException | None = error + seen: set[int] = set() + while current is not None and id(current) not in seen: + seen.add(id(current)) + if isinstance(current, (httpx.ConnectError, httpx.ConnectTimeout, httpx.PoolTimeout)): + return True + current = current.__cause__ or current.__context__ + + return False + async def _handle_polling_network_error(self, error: Exception) -> None: """Reconnect polling after a transient network interruption. @@ -520,8 +542,28 @@ async def connect(self) -> bool: ", ".join(fallback_ips), ) transport = TelegramFallbackTransport(fallback_ips) - request = HTTPXRequest(httpx_kwargs={"transport": transport}) - get_updates_request = HTTPXRequest(httpx_kwargs={"transport": transport}) + # Explicit timeouts are required here because passing a + # pre-built HTTPXRequest to ApplicationBuilder bypasses + # the builder's default timeout configuration. The + # fallback transport adds latency (DoH discovery + IP + # rotation), so these values must be generous enough to + # absorb that overhead. Without them the library's 5s + # defaults cause frequent TimedOut errors which the + # retry logic then compounds into duplicate deliveries. + request = HTTPXRequest( + read_timeout=20.0, + write_timeout=20.0, + connect_timeout=10.0, + pool_timeout=5.0, + httpx_kwargs={"transport": transport}, + ) + get_updates_request = HTTPXRequest( + read_timeout=30.0, + write_timeout=5.0, + connect_timeout=10.0, + pool_timeout=5.0, + httpx_kwargs={"transport": transport}, + ) builder = builder.request(request).get_updates_request(get_updates_request) self._app = builder.build() self._bot = self._app.bot @@ -820,6 +862,8 @@ async def send( continue # Other BadRequest errors are permanent — don't retry raise + if not self._is_retryable_send_network_error(send_err): + raise if _send_attempt < 2: wait = 2 ** _send_attempt logger.warning("[%s] Network error on send (attempt %d/3), retrying in %ds: %s", diff --git a/tests/gateway/test_telegram_thread_fallback.py b/tests/gateway/test_telegram_thread_fallback.py index e2817d834000..a3deb4455c08 100644 --- a/tests/gateway/test_telegram_thread_fallback.py +++ b/tests/gateway/test_telegram_thread_fallback.py @@ -14,6 +14,7 @@ import types from types import SimpleNamespace +import httpx import pytest from gateway.config import PlatformConfig, Platform @@ -33,11 +34,16 @@ class FakeBadRequest(FakeNetworkError): pass +class FakeTimedOut(FakeNetworkError): + pass + + # Build a fake telegram module tree so the adapter's internal imports work _fake_telegram = types.ModuleType("telegram") _fake_telegram_error = types.ModuleType("telegram.error") _fake_telegram_error.NetworkError = FakeNetworkError _fake_telegram_error.BadRequest = FakeBadRequest +_fake_telegram_error.TimedOut = FakeTimedOut _fake_telegram.error = _fake_telegram_error _fake_telegram_constants = types.ModuleType("telegram.constants") _fake_telegram_constants.ParseMode = SimpleNamespace(MARKDOWN_V2="MarkdownV2") @@ -146,7 +152,7 @@ async def mock_send_message(**kwargs): @pytest.mark.asyncio async def test_send_retries_network_errors_normally(): - """Real transient network errors (not BadRequest) should still be retried.""" + """Connect-stage network errors should still be retried.""" adapter = _make_adapter() attempt = [0] @@ -154,7 +160,9 @@ async def test_send_retries_network_errors_normally(): async def mock_send_message(**kwargs): attempt[0] += 1 if attempt[0] < 3: - raise FakeNetworkError("Connection reset") + raise FakeNetworkError("httpx.ConnectError: Connection reset") from httpx.ConnectError( + "Connection reset" + ) return SimpleNamespace(message_id=200) adapter._bot = SimpleNamespace(send_message=mock_send_message) @@ -168,6 +176,31 @@ async def mock_send_message(**kwargs): assert attempt[0] == 3 # Two retries then success +@pytest.mark.asyncio +async def test_send_does_not_retry_ambiguous_timeouts(): + """Read/write timeouts may happen after delivery starts, so don't resend.""" + adapter = _make_adapter() + + attempt = [0] + + async def mock_send_message(**kwargs): + attempt[0] += 1 + raise FakeTimedOut("Timed out waiting for Telegram response") from httpx.ReadTimeout( + "read timeout" + ) + + adapter._bot = SimpleNamespace(send_message=mock_send_message) + + result = await adapter.send( + chat_id="123", + content="test message", + ) + + assert result.success is False + assert "Timed out waiting for Telegram response" in result.error + assert attempt[0] == 1 + + @pytest.mark.asyncio async def test_thread_fallback_only_fires_once(): """After clearing thread_id, subsequent chunks should also use None."""