fix(telegram): prevent duplicate message delivery on send timeout (#5153)

TimedOut is a subclass of NetworkError in python-telegram-bot. The
inner retry loop in send() and the outer _send_with_retry() in base.py
both treated it as a transient connection error and retried — but
send_message is not idempotent. When the request reaches Telegram but
the HTTP response times out, the message is already delivered. Retrying
sends duplicates. Worst case: up to 9 copies (inner 3x × outer 3x).

Inner loop (telegram.py):
- Import TimedOut separately, isinstance-check before generic
  NetworkError retry (same pattern as BadRequest carve-out from #3390)
- Re-raise immediately — no retry
- Mark as retryable=False in outer exception handler

Outer loop (base.py):
- Remove 'timeout', 'timed out', 'readtimeout', 'writetimeout' from
  _RETRYABLE_ERROR_PATTERNS (read/write timeouts are delivery-ambiguous)
- Add 'connecttimeout' (safe — connection never established)
- Keep 'network' (other platforms still need it)
- Add _is_timeout_error() + early return to prevent plain-text fallback
  on timeout errors (would also cause duplicate delivery)

Connection errors (ConnectionReset, ConnectError, etc.) are still
retried — these fail before the request reaches the server.

Credit: tmdgusya (PR #3899), barun1997 (PR #3904) for identifying the
bug and proposing fixes.

Closes #3899, closes #3904.

tests/gateway/test_send_retry.py

@@ -72,6 +72,43 @@ class TestIsRetryableError:
     def test_case_insensitive(self):
         assert _StubAdapter._is_retryable_error("CONNECTERROR: host unreachable")
 
+    def test_timeout_not_retryable(self):
+        assert not _StubAdapter._is_retryable_error("ReadTimeout: request timed out")
+
+    def test_timed_out_not_retryable(self):
+        assert not _StubAdapter._is_retryable_error("Timed out waiting for response")
+
+    def test_connect_timeout_is_retryable(self):
+        assert _StubAdapter._is_retryable_error("ConnectTimeout: connection timed out")
+
+
+# ---------------------------------------------------------------------------
+# _is_timeout_error
+# ---------------------------------------------------------------------------
+
+class TestIsTimeoutError:
+    def test_none_is_not_timeout(self):
+        assert not _StubAdapter._is_timeout_error(None)
+
+    def test_empty_is_not_timeout(self):
+        assert not _StubAdapter._is_timeout_error("")
+
+    def test_timed_out(self):
+        assert _StubAdapter._is_timeout_error("Timed out waiting for response")
+
+    def test_read_timeout(self):
+        assert _StubAdapter._is_timeout_error("ReadTimeout: request timed out")
+
+    def test_write_timeout(self):
+        assert _StubAdapter._is_timeout_error("WriteTimeout: send stalled")
+
+    def test_connect_timeout_not_flagged(self):
+        """ConnectTimeout is a connection error, not a delivery-ambiguous timeout."""
+        assert not _StubAdapter._is_timeout_error("ConnectTimeout: host unreachable")
+
+    def test_connection_error_not_timeout(self):
+        assert not _StubAdapter._is_timeout_error("ConnectionError: host unreachable")
+
 
 # ---------------------------------------------------------------------------
 # _send_with_retry — success on first attempt
@@ -112,17 +149,33 @@ class TestSendWithRetryNetworkRetry:
         assert len(adapter._send_calls) == 2  # initial + 1 retry
 
     @pytest.mark.asyncio
-    async def test_retries_on_timeout_and_succeeds(self):
+    async def test_timeout_not_retried_to_prevent_duplicates(self):
+        """ReadTimeout is NOT retried because the request may have reached
+        the server — retrying a non-idempotent send risks duplicate delivery.
+        It also skips plain-text fallback (timeout is not a formatting issue)."""
         adapter = _StubAdapter()
         adapter._send_results = [
             SendResult(success=False, error="ReadTimeout: request timed out"),
-            SendResult(success=False, error="ReadTimeout: request timed out"),
+        ]
+        with patch("asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
+            result = await adapter._send_with_retry("chat1", "hello", max_retries=3, base_delay=0)
+        # No retry, no fallback — timeout returns failure immediately
+        mock_sleep.assert_not_called()
+        assert not result.success
+        assert len(adapter._send_calls) == 1
+
+    @pytest.mark.asyncio
+    async def test_connect_timeout_still_retried(self):
+        """ConnectTimeout is safe to retry — the connection was never established."""
+        adapter = _StubAdapter()
+        adapter._send_results = [
+            SendResult(success=False, error="ConnectTimeout: connection timed out"),
             SendResult(success=True, message_id="ok"),
         ]
         with patch("asyncio.sleep", new_callable=AsyncMock):
-            result = await adapter._send_with_retry("chat1", "hello", max_retries=3, base_delay=0)
+            result = await adapter._send_with_retry("chat1", "hello", max_retries=2, base_delay=0)
         assert result.success
-        assert len(adapter._send_calls) == 3
+        assert len(adapter._send_calls) == 2
 
     @pytest.mark.asyncio
     async def test_retryable_flag_respected(self):

tests/gateway/test_telegram_thread_fallback.py

@@ -33,11 +33,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")
@@ -168,6 +173,34 @@ async def test_send_retries_network_errors_normally():
     assert attempt[0] == 3  # Two retries then success
 
 
+@pytest.mark.asyncio
+async def test_send_does_not_retry_timeout():
+    """TimedOut (subclass of NetworkError) should NOT be retried in send().
+
+    The request may have already been delivered to the user — retrying
+    would send duplicate messages.
+    """
+    adapter = _make_adapter()
+
+    attempt = [0]
+
+    async def mock_send_message(**kwargs):
+        attempt[0] += 1
+        raise FakeTimedOut("Timed out waiting for Telegram response")
+
+    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" in result.error
+    # CRITICAL: only 1 attempt — no retry for TimedOut
+    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."""