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

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

@@ -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:

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)
 

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 == []

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)"