From e16743b0d5d1899785a5dfa87d4f5f8d8e503226 Mon Sep 17 00:00:00 2001 From: Jaret Bottoms <25535214+jbbottoms@users.noreply.github.com> Date: Sun, 12 Jul 2026 13:15:21 -0500 Subject: [PATCH] fix(telegram): diagnose blocked-loop init hangs, unbind DoH from system DNS MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The #63309 hang class — gateway stuck at 'Connecting to Telegram (attempt 1/8)' with no retry, no timeout, for minutes — can only occur when the event loop thread itself is blocked in a synchronous call: _await_with_thread_deadline's timer fires off-loop, but its expiry hand-off (call_soon_threadsafe) still needs the loop to run, and the gateway's outer wait_for is a pure loop timer. When the loop is pinned, every layer goes silent simultaneously and the process wedges with no evidence of where. Two changes: 1. Loop-blocked watchdog in _await_with_thread_deadline: a second daemon timer fires one grace period (5s) after the deadline; if the loop still hasn't processed the expiry, it logs a WARNING from the timer thread and faulthandler-dumps all thread stacks to stderr — converting the silent hang into a trace that names the exact blocking frame. A threading.Event set by the expiry callback (and on normal exit) keeps completed awaits from ever being misreported. 2. discover_fallback_ips: the system-resolver leg runs socket.getaddrinfo in a worker thread with no timeout, and asyncio.gather waited on it unboundedly — a wedged OS resolver stalled discovery for minutes between the two startup log lines. Its result only feeds a log message, so it no longer gates discovery: DoH legs (already client-bounded) are gathered alone and the system leg is awaited with a _DOH_TIMEOUT cap, best-effort. Refs #63309 Tests: 3 watchdog regressions (blocked-loop dump fires; responsive-loop timeout does not; completed await does not) + 2 hung-resolver regressions (DoH results returned promptly; worst-case seed fallback stays bounded). --- plugins/platforms/telegram/adapter.py | 52 +++++++++++++ .../platforms/telegram/telegram_network.py | 20 +++-- tests/gateway/test_telegram_init_deadline.py | 76 +++++++++++++++++++ tests/gateway/test_telegram_network.py | 51 +++++++++++++ 4 files changed, 194 insertions(+), 5 deletions(-) diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 7b959b1edf61..05506c79da48 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -9,6 +9,7 @@ Uses python-telegram-bot library for: 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 @@ async def _await_with_thread_deadline(awaitable, timeout: float, *, on_abandon=N 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: diff --git a/plugins/platforms/telegram/telegram_network.py b/plugins/platforms/telegram/telegram_network.py index 319f4d2accf8..a0fd14ebb5d3 100644 --- a/plugins/platforms/telegram/telegram_network.py +++ b/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) + system_dns_task = asyncio.ensure_future(asyncio.to_thread(_resolve_system_dns)) + results = await asyncio.gather(*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() + # 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) diff --git a/tests/gateway/test_telegram_init_deadline.py b/tests/gateway/test_telegram_init_deadline.py index ca697276e8b5..5e3d045ba48f 100644 --- a/tests/gateway/test_telegram_init_deadline.py +++ b/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 + + +@pytest.mark.asyncio +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() + + +@pytest.mark.asyncio +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() + + +@pytest.mark.asyncio +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 == [] diff --git a/tests/gateway/test_telegram_network.py b/tests/gateway/test_telegram_network.py index cede492fe5bd..8406973b9483 100644 --- a/tests/gateway/test_telegram_network.py +++ b/tests/gateway/test_telegram_network.py @@ -705,3 +705,54 @@ class TestDiscoverFallbackIps: 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)"