Telegram connect hangs and attribute errors

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #58293 fix(telegram): wall-deadline init timeout + shut down abandoned init app

  merged to main

  files: plugins/platforms/telegram/adapter.py, scripts/release.py, tests/gateway/test_telegram_init_deadline.py

• #64370 fix(telegram): diagnose blocked-loop init hangs, unbind DoH discovery from system DNS

  merged to main

  files: plugins/platforms/telegram/adapter.py, plugins/platforms/telegram/telegram_network.py, tests/gateway/test_telegram_init_deadline.py, tests/gateway/test_telegram_network.py

  Show the diff

  • plugins/platforms/telegram/adapter.py

    @@ -9,6 +9,7 @@
     
     import asyncio
     import dataclasses
    +import faulthandler
     import inspect
     import json
     import logging
    @@ -45,6 +46,35 @@ def _consume_abandoned_task(task: asyncio.Task) -> None:
             logger.debug("Abandoned Telegram init task failed after timeout", exc_info=True)
     
     
    +# Grace period after the wall-clock deadline fires: if the event loop still
    +# hasn't processed the expiry callback by then, the loop thread itself is
    +# blocked in a synchronous call — the exact state in which every asyncio-based
    +# timeout (including this helper's own expiry hand-off) goes silent, so the
    +# gateway hangs at "attempt 1/8" with no further output (#63309).
    +_LOOP_BLOCKED_DUMP_GRACE = 5.0
    +
    +
    +def _dump_loop_blocked_diagnostics(timeout: float, grace: float) -> None:
    +    """Emit diagnostics from the deadline timer thread when the loop is stuck.
    +
    +    Runs OFF the event loop, so it works precisely when the loop cannot. The
    +    faulthandler dump names the frame the loop thread is blocked in — the one
    +    piece of information #63309-class hangs otherwise never surface.
    +    """
    +    logger.warning(
    +        "[Telegram] init deadline (%.0fs) expired but the event loop has not "
    +        "processed the expiry after a further %.0fs — the loop thread appears "
    +        "BLOCKED in a synchronous call, which is why no timeout fires (#63309). "
    +        "Dumping all thread stacks to stderr to identify the blocking frame.",
    +        timeout,
    +        grace,
    +    )
    +    try:
    +        faulthandler.dump_traceback(all_threads=True)
    +    except Exception:
    +        logger.debug("faulthandler traceback dump failed", exc_info=True)
    +
    +
     async def _await_with_thread_deadline(awaitable, timeout: float, *, on_abandon=None):
         """Await with a wall-clock deadline that does not depend on loop timers.
     
    @@ -66,17 +96,34 @@ async def _await_with_thread_deadline(awaitable, timeout: float, *, on_abandon=N
         task = asyncio.ensure_future(awaitable)
         loop = asyncio.get_running_loop()
         deadline = loop.create_future()
    +    # Set the moment the loop actually runs the expiry callback (or the helper
    +    # exits normally). threading.Event so the watchdog thread can read it
    +    # without touching asyncio state from off-loop.
    +    loop_processed_expiry = threading.Event()
     
         def _mark_expired() -> None:
    +        loop_processed_expiry.set()
             if not deadline.done():
                 deadline.set_result(None)
     
         def _expire_from_thread() -> None:
             loop.call_soon_threadsafe(_mark_expired)
     
    +    def _watchdog_check() -> None:
    +        # The deadline fired _LOOP_BLOCKED_DUMP_GRACE ago but the loop never
    +        # ran _mark_expired: the loop thread is stuck in a synchronous call.
    +        # Diagnose from this thread — the loop can't.
    +        if not loop_processed_expiry.is_set():
    +            _dump_loop_blocked_diagnostics(timeout, _LOOP_BLOCKED_DUMP_GRACE)
    +
         timer = threading.Timer(max(timeout, 0.0), _expire_from_thread)
         timer.daemon = True
         timer.start()
    +    watchdog = threading.Timer(
    +        max(timeout, 0.0) + _LOOP_BLOCKED_DUMP_GRACE, _watchdog_check
    +    )
    +    watchdog.daemon = True
    +    watchdog.start()
         try:
             done, _ = await asyncio.wait(
                 {task, deadline},
    @@ -99,6 +146,11 @@ def _expire_from_thread() -> None:
             raise asyncio.TimeoutError()
         finally:
             timer.cancel()
    +        watchdog.cancel()
    +        # cancel() cannot stop a Timer whose callback is already running;
    +        # setting the event closes that race so a completed await can never
    +        # be misreported as a blocked loop.
    +        loop_processed_expiry.set()
     
     
     async def _run_abandon_cleanup(on_abandon) -> None:

  • plugins/platforms/telegram/telegram_network.py

    @@ -205,14 +205,24 @@ async def discover_fallback_ips() -> list[str]:
         """
         async with httpx.AsyncClient(timeout=httpx.Timeout(_DOH_TIMEOUT)) as client:
             doh_tasks = [_query_doh_provider(client, p) for p in _DOH_PROVIDERS]
    -        system_dns_task = asyncio.to_thread(_resolve_system_dns)
    -        results = await asyncio.gather(system_dns_task, *doh_tasks, return_exceptions=True)
    -
    -    # results[0] = system DNS IPs (set), results[1:] = DoH IP lists
    -    system_ips: set[str] = results[0] if isinstance(results[0], set) else set()
    +        system_dns_task = asyncio.ensure_future(asyncio.to_thread(_resolve_system_dns))
    +        results = await asyncio.gather(*doh_tasks, return_exceptions=True)
    +
    +    # The system-resolver leg runs socket.getaddrinfo in a worker thread with
    +    # no timeout of its own — a wedged OS resolver (broken VPN/DNS) can sit for
    +    # minutes. Its result only feeds the no-usable-answers log line below, so
    +    # it must never gate discovery: bound it and move on (#63309). The DoH legs
    +    # are already bounded by the client timeout above.
    +    system_ips: set[str] = set()
    +    try:
    +        system_result = await asyncio.wait_for(system_dns_task, timeout=_DOH_TIMEOUT)
    +        if isinstance(system_result, set):
    +            system_ips = system_result
    +    except Exception:
    +        logger.debug("System-DNS resolution for %s did not complete in time", _TELEGRAM_API_HOST)
     
         doh_ips: list[str] = []
    -    for r in results[1:]:
    +    for r in results:
             if isinstance(r, list):
                 doh_ips.extend(r)
     

  • tests/gateway/test_telegram_init_deadline.py

    @@ -199,3 +199,79 @@ async def test_shutdown_abandoned_app_handles_none_and_missing_requests():
         app.shutdown = AsyncMock(side_effect=RuntimeError("still running"))
         app.bot = None
         await tg_adapter._shutdown_abandoned_app(app)  # must not raise
    +
    +
    [redacted:email]
    +async def test_blocked_loop_after_expiry_dumps_diagnostics(monkeypatch):
    +    """#63309: when the loop thread is stuck in a synchronous call, the expiry
    +    callback never runs and every asyncio timeout goes silent. The off-loop
    +    watchdog must detect that state and emit diagnostics from its own thread."""
    +    import asyncio as _asyncio
    +    import time as _time
    +
    +    dumps = []
    +    monkeypatch.setattr(
    +        tg_adapter,
    +        "_dump_loop_blocked_diagnostics",
    +        lambda timeout, grace: dumps.append((timeout, grace)),
    +    )
    +    monkeypatch.setattr(tg_adapter, "_LOOP_BLOCKED_DUMP_GRACE", 0.15)
    +
    +    hung = _asyncio.get_running_loop().create_future()  # never completes
    +    task = _asyncio.ensure_future(
    +        tg_adapter._await_with_thread_deadline(hung, timeout=0.05)
    +    )
    +    # Let the helper start its deadline + watchdog timers…
    +    await _asyncio.sleep(0)
    +    # …then block the event loop straight through deadline (0.05s) AND the
    +    # watchdog grace (0.15s): call_soon_threadsafe stays queued, exactly like
    +    # a sync call pinning the loop during Application.initialize().
    +    _time.sleep(0.5)
    +    with pytest.raises(_asyncio.TimeoutError):
    +        await task
    +
    +    assert dumps == [(0.05, 0.15)]
    +    hung.cancel()
    +
    +
    [redacted:email]
    +async def test_responsive_loop_expiry_does_not_dump(monkeypatch):
    +    """A normal timeout on a responsive loop must not trigger the watchdog."""
    +    import asyncio as _asyncio
    +
    +    dumps = []
    +    monkeypatch.setattr(
    +        tg_adapter,
    +        "_dump_loop_blocked_diagnostics",
    +        lambda timeout, grace: dumps.append((timeout, grace)),
    +    )
    +    monkeypatch.setattr(tg_adapter, "_LOOP_BLOCKED_DUMP_GRACE", 0.1)
    +
    +    hung = _asyncio.get_running_loop().create_future()
    +    with pytest.raises(_asyncio.TimeoutError):
    +        await tg_adapter._await_with_thread_deadline(hung, timeout=0.05)
    +    # Give the (cancelled) watchdog window time to have fired if it were going to.
    +    await _asyncio.sleep(0.3)
    +    assert dumps == []
    +    hung.cancel()
    +
    +
    [redacted:email]
    +async def test_completed_await_never_reports_blocked_loop(monkeypatch):
    +    """Success before the deadline must cancel the watchdog (no false dump)."""
    +    import asyncio as _asyncio
    +
    +    dumps = []
    +    monkeypatch.setattr(
    +        tg_adapter,
    +        "_dump_loop_blocked_diagnostics",
    +        lambda timeout, grace: dumps.append((timeout, grace)),
    +    )
    +    monkeypatch.setattr(tg_adapter, "_LOOP_BLOCKED_DUMP_GRACE", 0.05)
    +
    +    async def _quick():
    +        return "ok"
    +
    +    assert await tg_adapter._await_with_thread_deadline(_quick(), timeout=0.2) == "ok"
    +    await _asyncio.sleep(0.4)
    +    assert dumps == []

  • tests/gateway/test_telegram_network.py

    @@ -705,3 +705,54 @@ async def test_invalid_ip_in_doh_response_skipped(self, monkeypatch):
     
             ips = await tnet.discover_fallback_ips()
             assert ips == ["149.154.167.220"]
    +
    +    @pytest.mark.asyncio
    +    async def test_hung_system_dns_does_not_gate_doh_results(self, monkeypatch):
    +        """#63309: socket.getaddrinfo has no timeout of its own — a wedged OS
    +        resolver must not stall discovery. DoH answers must come back promptly
    +        even while the system-DNS worker thread is still hanging."""
    +        import time as _time
    +
    +        self._patch_doh(monkeypatch, {
    +            "https://dns.google": (200, _doh_answer("149.154.167.220")),
    +            "https://cloudflare-dns.com": (200, _doh_answer()),
    +        }, system_dns_ips=["149.154.166.110"])
    +        monkeypatch.setattr(tnet, "_DOH_TIMEOUT", 0.2)
    +
    +        def _hung_getaddrinfo(*a, **kw):
    +            _time.sleep(1.5)  # far beyond the discovery bound
    +            raise OSError("resolver wedged")
    +
    +        monkeypatch.setattr(tnet.socket, "getaddrinfo", _hung_getaddrinfo)
    +
    +        start = _time.monotonic()
    +        ips = await tnet.discover_fallback_ips()
    +        elapsed = _time.monotonic() - start
    +
    +        assert ips == ["149.154.167.220"]
    +        assert elapsed < 1.0, f"discovery gated on hung system DNS ({elapsed:.2f}s)"
    +
    +    @pytest.mark.asyncio
    +    async def test_hung_system_dns_with_no_doh_answers_bounded_seed_fallback(self, monkeypatch):
    +        """Worst case — resolver wedged AND no DoH answers — must still return
    +        the seed list within the bound instead of hanging connect()."""
    +        import time as _time
    +
    +        self._patch_doh(monkeypatch, {
    +            "https://dns.google": (200, {"Status": 0}),
    +            "https://cloudflare-dns.com": (200, {"garbage": True}),
    +        }, system_dns_ips=["149.154.166.110"])
    +        monkeypatch.setattr(tnet, "_DOH_TIMEOUT", 0.2)
    +
    +        def _hung_getaddrinfo(*a, **kw):
    +            _time.sleep(1.5)
    +            raise OSError("resolver wedged")
    +
    +        monkeypatch.setattr(tnet.socket, "getaddrinfo", _hung_getaddrinfo)
    +
    +        start = _time.monotonic()
    +        ips = await tnet.discover_fallback_ips()
    +        elapsed = _time.monotonic() - start
    +
    +        assert ips == tnet._SEED_FALLBACK_IPS
    +        assert elapsed < 1.0, f"seed fallback gated on hung system DNS ({elapsed:.2f}s)"

• #64574 fix(telegram): slots-safe getUpdates instrumentation for Python 3.13 (#64482)

  merged to main

  files: plugins/platforms/telegram/adapter.py, tests/test_telegram_polling_progress_ptb.py

  Show the diff

  • plugins/platforms/telegram/adapter.py

    @@ -1969,34 +1969,62 @@ def _record_polling_progress(self, generation: int) -> None:
             self._polling_conflict_count = 0
             self._send_path_degraded = False
     
    +    def _observe_polling_request_result(self, request, generation, result):
    +        """Record getUpdates progress from an observed do_request result.
    +
    +        Purely observational: PTB still parses the untouched payload and owns
    +        any resulting exception. Kept as its own method so the observation
    +        logic is shared and independently testable.
    +        """
    +        status_code, payload = result
    +        if generation is None or not (200 <= status_code < 300):
    +            return
    +        try:
    +            # Use the request's own parser so health observation agrees
    +            # exactly with PTB's authoritative response handling (e.g.
    +            # UTF-8 replacement decoding and BOM rejection).
    +            envelope = request.parse_json_payload(payload)
    +        except Exception:
    +            return
    +        if (
    +            isinstance(envelope, dict)
    +            and envelope.get("ok") is True
    +            and "result" in envelope
    +        ):
    +            self._record_polling_progress(generation)
    +
         def _instrument_polling_request(self, request):
    -        """Wrap one dedicated PTB getUpdates request with progress tracking."""
    -        do_request = request.do_request
    -
    -        async def _do_request(*args, **kwargs):
    -            generation = _POLLING_GENERATION_CONTEXT.get()
    -            result = await do_request(*args, **kwargs)
    -            status_code, payload = result
    -            if generation is not None and 200 <= status_code < 300:
    -                try:
    -                    # Use the request's own parser so health observation agrees
    -                    # exactly with PTB's authoritative response handling (e.g.
    -                    # UTF-8 replacement decoding and BOM rejection).
    -                    envelope = request.parse_json_payload(payload)
    -                except Exception:
    -                    # Instrumentation is observational: PTB still parses the
    -                    # untouched payload and owns the resulting exception.
    -                    pass
    -                else:
    -                    if (
    -                        isinstance(envelope, dict)
    -                        and envelope.get("ok") is True
    -                        and "result" in envelope
    -                    ):
    -                        self._record_polling_progress(generation)
    -            return result
    -
    -        request.do_request = _do_request
    +        """Instrument one dedicated PTB getUpdates request with progress tracking.
    +
    +        PTB's request classes (``BaseRequest`` / ``HTTPXRequest``) use
    +        ``__slots__``. On Python 3.13 their instances no longer carry a
    +        ``__dict__`` (the ``AbstractAsyncContextManager`` MRO stopped yielding
    +        one), so ``request.do_request = wrapper`` raises
    +        ``AttributeError: 'HTTPXRequest' object attribute 'do_request' is
    +        read-only`` and the whole Telegram connect fails (#64482). It only
    +        appeared to work on Python 3.12, where those instances still had a
    +        ``__dict__``.
    +
    +        Instead of monkey-patching the instance, re-tag it to a thin subclass
    +        that overrides ``do_request``. This is portable across Python versions
    +        and works for both the real request and the test doubles. The subclass
    +        declares ``__slots__ = ()`` so its instance layout stays identical to
    +        the base, which is what makes the ``__class__`` swap legal on a slotted
    +        instance.
    +        """
    +        adapter = self
    +        base_cls = type(request)
    +
    +        class _InstrumentedPollingRequest(base_cls):
    +            __slots__ = ()
    +
    +            async def do_request(self, *args, **kwargs):
    +                generation = _POLLING_GENERATION_CONTEXT.get()
    +                result = await super().do_request(*args, **kwargs)
    +                adapter._observe_polling_request_result(self, generation, result)
    +                return result
    +
    +        request.__class__ = _InstrumentedPollingRequest
             return request
     
         async def _start_polling_once(

  • tests/test_telegram_polling_progress_ptb.py

    @@ -92,6 +92,34 @@ async def do_request(self, url, method, request_data=None, **_kwargs):
             return 200, self.payload
     
     
    +class _SlottedEnvelopeRequest(BaseRequest):
    +    """A getUpdates request with no instance ``__dict__``.
    +
    +    Reproduces PTB's real HTTPXRequest shape on Python 3.13, where every
    +    class in the MRO defines ``__slots__`` and instances therefore reject an
    +    instance-attribute ``do_request`` monkey-patch as "read-only" (#64482).
    +    ``__slots__`` names the payload so the double needs no ``__dict__``.
    +    """
    +
    +    __slots__ = ("_payload",)
    +
    +    def __init__(self, payload):
    +        self._payload = payload
    +
    +    @property
    +    def read_timeout(self):
    +        return 10
    +
    +    async def initialize(self):
    +        return None
    +
    +    async def shutdown(self):
    +        return None
    +
    +    async def do_request(self, url, method, request_data=None, **_kwargs):
    +        return 200, self._payload
    +
    +
     async def _cancel_task(task):
         if task is None or task.done():
             return
    @@ -226,6 +254,47 @@ async def test_real_base_request_valid_success_envelope_records_progress():
         assert adapter._send_path_degraded is False
     
     
    [redacted:email]
    +async def test_slotted_request_instrumented_without_read_only_error():
    +    """Instrumenting a __slots__ request must not raise, and must still record.
    +
    +    Regression for #64482: PTB's HTTPXRequest carries no instance ``__dict__``
    +    on Python 3.13, so the old ``request.do_request = wrapper`` monkey-patch
    +    raised ``AttributeError: ... 'do_request' is read-only`` and broke every
    +    Telegram connect. The subclass re-tag must instrument such an instance and
    +    still observe getUpdates progress.
    +    """
    +    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="[redacted:secret-assignment]"))
    +    generation, progress = adapter._begin_polling_generation()
    +    adapter._polling_network_error_count = 4
    +    adapter._polling_conflict_count = 3
    +
    +    request = _SlottedEnvelopeRequest(b'{"ok":true,"result":[]}')
    +    # Guard the premise where the runtime actually enforces it: PTB's request
    +    # MRO is only fully slotted on Python 3.13+ (contextlib's
    +    # AbstractAsyncContextManager gained ``__slots__ = ()`` there). On older
    +    # runtimes the instance still carries a ``__dict__`` and the legacy
    +    # monkey-patch is accepted, so the read-only premise cannot be asserted.
    +    if not hasattr(request, "__dict__"):
    +        with pytest.raises(AttributeError, match="read-only"):
    +            request.do_request = lambda *a, **k: None
    +
    +    instrumented = adapter._instrument_polling_request(request)
    +    context_token = [redacted:secret-assignment]
    +    try:
    +        result = await instrumented.post(
    +            "https://api.telegram.org/bot-token/getUpdates"
    +        )
    +    finally:
    +        tg_adapter._POLLING_GENERATION_CONTEXT.reset(context_token)
    +
    +    assert result == []
    +    assert progress.is_set()
    +    assert adapter._polling_network_error_count == 0
    +    assert adapter._polling_conflict_count == 0
    +    assert adapter._send_path_degraded is False
    +
    +
     @pytest.mark.asyncio
     async def test_real_ptb_stop_cleanup_cannot_heal_recovery_generation():
         assert tg_adapter.TELEGRAM_AVAILABLE is True

• #70884 fix(telegram+tools): prevent connect hang with retry watchdog and fresh app per attempt (#67498, #68915)

  merged to main

  files: contributors/emails/[redacted:email], plugins/platforms/telegram/adapter.py, tests/tools/test_process_registry.py, tools/process_registry.py

• #71118 fix(telegram): require proven polling readiness on cold start

  merged to main

  files: contributors/emails/[redacted:email], gateway/run.py, plugins/platforms/telegram/adapter.py, tests/gateway/test_platform_reconnect.py, tests/gateway/test_telegram_conflict.py, tests/gateway/test_telegram_polling_progress.py, tests/gateway/test_telegram_start_polling_timeout.py