mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-08 13:12:08 +00:00
fix(telegram): bound start_polling() at bootstrap and conflict-retry sites too; strengthen tests
Follow-up on the salvaged fix, which bounded start_polling() only in _handle_polling_network_error. The same wedge (#59614) exists at the two sibling call sites: 1. _start_polling_resilient (bootstrap): an exhausted pool hangs connect() forever. The TimeoutError from wait_for is a builtins TimeoutError (OSError subclass), so the existing except classifies it via _looks_like_network_error and schedules background recovery. 2. _handle_polling_conflict (conflict-retry ladder): identical hang wedges conflict attempt N forever; timeout now converts to RuntimeError and the existing except schedules the next attempt. Tests replaced with a stronger suite: hung-network-ladder repro (RED without the fix), bootstrap hang schedules recovery, success-path sanity, and a bug-class contract test asserting EVERY updater.start_polling( call site is wrapped in wait_for so a new unbounded site can't reintroduce the wedge. Verified RED (3 failures) with the wrappers removed, GREEN with them.
This commit is contained in:
kshitijk4poor
kshitij
parent
4aaaa206aa
commit
aaeba213d9
2 changed files with 166 additions and 83 deletions
|
|
@ -1979,10 +1979,19 @@ class TelegramAdapter(BasePlatformAdapter):
|
|||
if not (self._app and self._app.updater):
|
||||
raise RuntimeError("Telegram application/updater not initialized")
|
||||
try:
|
||||
await self._app.updater.start_polling(
|
||||
allowed_updates=Update.ALL_TYPES,
|
||||
drop_pending_updates=drop_pending_updates,
|
||||
error_callback=error_callback,
|
||||
# Same watchdog bound as the reconnect ladders: a wedged httpx
|
||||
# connection pool can hang start_polling() forever at bootstrap
|
||||
# too (#59614). A propagating TimeoutError is a builtins
|
||||
# TimeoutError (OSError subclass), so the except below classifies
|
||||
# it via _looks_like_network_error and schedules background
|
||||
# recovery instead of blocking connect() indefinitely.
|
||||
await asyncio.wait_for(
|
||||
self._app.updater.start_polling(
|
||||
allowed_updates=Update.ALL_TYPES,
|
||||
drop_pending_updates=drop_pending_updates,
|
||||
error_callback=error_callback,
|
||||
),
|
||||
timeout=_UPDATER_START_TIMEOUT,
|
||||
)
|
||||
return True
|
||||
except Exception as err:
|
||||
|
|
@ -2488,11 +2497,24 @@ class TelegramAdapter(BasePlatformAdapter):
|
|||
try:
|
||||
if not app:
|
||||
raise RuntimeError("Telegram application was torn down during conflict reconnect")
|
||||
await app.updater.start_polling(
|
||||
allowed_updates=Update.ALL_TYPES,
|
||||
drop_pending_updates=False,
|
||||
error_callback=self._polling_error_callback_ref,
|
||||
)
|
||||
# Same watchdog bound as the network-error ladder: an
|
||||
# exhausted pool hangs start_polling() on the conflict path
|
||||
# identically (#59614). Timeout converts to RuntimeError so
|
||||
# the except below logs a readable message and schedules the
|
||||
# next conflict attempt instead of wedging attempt N forever.
|
||||
try:
|
||||
await asyncio.wait_for(
|
||||
app.updater.start_polling(
|
||||
allowed_updates=Update.ALL_TYPES,
|
||||
drop_pending_updates=False,
|
||||
error_callback=self._polling_error_callback_ref,
|
||||
),
|
||||
timeout=_UPDATER_START_TIMEOUT,
|
||||
)
|
||||
except asyncio.TimeoutError:
|
||||
raise RuntimeError(
|
||||
"start_polling() timed out — connection pool may be wedged"
|
||||
)
|
||||
logger.info(
|
||||
"[%s] Telegram polling resumed after conflict retry %d/%d",
|
||||
self.name, self._polling_conflict_count, MAX_CONFLICT_RETRIES,
|
||||
|
|
|
|||
|
|
@ -1,96 +1,157 @@
|
|||
"""Test that start_polling() timeout prevents indefinite hanging.
|
||||
"""Regression tests for #59614: start_polling() must be time-bounded.
|
||||
|
||||
This is a regression test for issue #59614 where start_polling() could hang
|
||||
indefinitely when the connection pool is in a degraded state.
|
||||
When both the primary Telegram API server and all fallback IPs are unreachable,
|
||||
``await app.updater.start_polling(...)`` can block forever inside an exhausted
|
||||
httpx connection pool — it neither returns nor raises. Unbounded, that wedges:
|
||||
|
||||
1. the network-error reconnect ladder (stuck inside attempt 1, never advances),
|
||||
2. the heartbeat loop (sees the recovery task as alive-but-wedged and skips),
|
||||
3. the fatal-error escalation (never reached).
|
||||
|
||||
The fix wraps every ``start_polling()`` await in ``asyncio.wait_for`` with
|
||||
``_UPDATER_START_TIMEOUT`` so a hung call raises and feeds the existing retry
|
||||
ladder. These tests patch the timeout down to keep the suite fast.
|
||||
"""
|
||||
import asyncio
|
||||
import sys
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
# Only import TelegramAdapter if it exists
|
||||
try:
|
||||
from hermes.plugins.platforms.telegram import adapter
|
||||
_UPDATER_START_TIMEOUT = adapter._UPDATER_START_TIMEOUT
|
||||
except (ImportError, AttributeError):
|
||||
pytest.skip("Telegram adapter not available", allow_module_level=True)
|
||||
|
||||
def _ensure_telegram_mock():
|
||||
if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"):
|
||||
return
|
||||
telegram_mod = MagicMock()
|
||||
telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None)
|
||||
telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2"
|
||||
telegram_mod.constants.ChatType.GROUP = "group"
|
||||
telegram_mod.constants.ChatType.SUPERGROUP = "supergroup"
|
||||
telegram_mod.constants.ChatType.CHANNEL = "channel"
|
||||
telegram_mod.constants.ChatType.PRIVATE = "private"
|
||||
telegram_mod.error.NetworkError = type("NetworkError", (OSError,), {})
|
||||
telegram_mod.error.TimedOut = type("TimedOut", (OSError,), {})
|
||||
for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"):
|
||||
sys.modules.setdefault(name, telegram_mod)
|
||||
sys.modules.setdefault("telegram.error", telegram_mod.error)
|
||||
|
||||
|
||||
class TestStartPollingTimeout:
|
||||
"""Test that start_polling() timeout prevents indefinite hanging."""
|
||||
_ensure_telegram_mock()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_start_polling_timeout_raises_runtime_error(self):
|
||||
"""When start_polling() times out, it should raise RuntimeError."""
|
||||
from hermes.plugins.platforms.telegram.adapter import TelegramAdapter
|
||||
from plugins.platforms.telegram import adapter as tg_adapter # noqa: E402
|
||||
from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402
|
||||
|
||||
# Mock the adapter's internal state
|
||||
adapter = TelegramAdapter.__new__(TelegramAdapter)
|
||||
adapter.name = "test_bot"
|
||||
adapter.has_fatal_error = False
|
||||
adapter._polling_network_error_count = 1
|
||||
adapter._polling_error_callback_ref = None
|
||||
adapter._background_tasks = set()
|
||||
adapter._send_path_degraded = True
|
||||
|
||||
# Mock the app and updater
|
||||
mock_app = MagicMock()
|
||||
mock_updater = AsyncMock()
|
||||
mock_app.updater = mock_updater
|
||||
adapter._app = mock_app
|
||||
async def _hang_forever(**kwargs):
|
||||
await asyncio.sleep(1000)
|
||||
|
||||
# Make start_polling() hang indefinitely (simulate the bug)
|
||||
async def hanging_start_polling(**kwargs):
|
||||
await asyncio.sleep(1000) # Hang for a long time
|
||||
return None
|
||||
|
||||
mock_updater.start_polling = hanging_start_polling
|
||||
mock_updater.running = True
|
||||
def _bare_adapter():
|
||||
a = TelegramAdapter.__new__(TelegramAdapter)
|
||||
# `name` / `has_fatal_error` are read-only base-class properties; set the
|
||||
# backing fields they derive from instead.
|
||||
from gateway.config import Platform
|
||||
|
||||
# Mock _drain_polling_connections to avoid actual connection cleanup
|
||||
with patch.object(adapter, '_drain_polling_connections', new=AsyncMock()):
|
||||
# Trigger the network error handler
|
||||
task = asyncio.create_task(adapter._handle_polling_network_error(Exception("test")))
|
||||
a.platform = Platform.TELEGRAM
|
||||
a._fatal_error_code = None
|
||||
a._fatal_error_message = None
|
||||
a._fatal_error_retryable = True
|
||||
a._polling_network_error_count = 0
|
||||
a._polling_conflict_count = 0
|
||||
a._polling_error_callback_ref = None
|
||||
a._background_tasks = set()
|
||||
a._send_path_degraded = False
|
||||
return a
|
||||
|
||||
# Wait for the timeout to trigger (start_polling_timeout is 30s)
|
||||
try:
|
||||
await asyncio.wait_for(task, timeout=_UPDATER_START_TIMEOUT + 5)
|
||||
except asyncio.TimeoutError:
|
||||
task.cancel()
|
||||
pytest.fail("Network error handler did not complete within timeout")
|
||||
|
||||
# The task should have completed (either with success or error)
|
||||
# The important part is that it didn't hang forever
|
||||
assert task.done()
|
||||
@pytest.mark.asyncio
|
||||
async def test_network_ladder_start_polling_hang_does_not_wedge(monkeypatch):
|
||||
"""A hung start_polling() in _handle_polling_network_error must time out
|
||||
and advance the ladder instead of blocking forever (#59614 core repro)."""
|
||||
monkeypatch.setattr(tg_adapter, "_UPDATER_START_TIMEOUT", 0.2)
|
||||
a = _bare_adapter()
|
||||
a._polling_network_error_count = 0 # attempt 1 → 5s backoff before start_polling
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_start_polling_success_returns_normally(self):
|
||||
"""When start_polling() succeeds quickly, it should return normally."""
|
||||
from hermes.plugins.platforms.telegram.adapter import TelegramAdapter
|
||||
app = MagicMock()
|
||||
app.updater = AsyncMock()
|
||||
app.updater.start_polling = _hang_forever
|
||||
app.updater.running = False
|
||||
a._app = app
|
||||
|
||||
# Mock the adapter's internal state
|
||||
adapter = TelegramAdapter.__new__(TelegramAdapter)
|
||||
adapter.name = "test_bot"
|
||||
adapter.has_fatal_error = False
|
||||
adapter._polling_network_error_count = 1
|
||||
adapter._polling_error_callback_ref = None
|
||||
adapter._background_tasks = set()
|
||||
adapter._send_path_degraded = True
|
||||
with patch.object(a, "_drain_polling_connections", new=AsyncMock()), \
|
||||
patch.object(
|
||||
tg_adapter.asyncio, "ensure_future",
|
||||
side_effect=lambda coro: (coro.close(), asyncio.get_event_loop().create_future())[1],
|
||||
):
|
||||
# Unbounded, this await hangs past the 30s wait_for and fails the
|
||||
# test; bounded, the handler waits its 5s backoff, times out the hung
|
||||
# start_polling() in 0.2s, schedules the chained retry (captured by
|
||||
# the ensure_future patch), and returns.
|
||||
await asyncio.wait_for(
|
||||
a._handle_polling_network_error(Exception("net down")), timeout=30
|
||||
)
|
||||
|
||||
# Mock the app and updater
|
||||
mock_app = MagicMock()
|
||||
mock_updater = AsyncMock()
|
||||
mock_app.updater = mock_updater
|
||||
adapter._app = mock_app
|
||||
|
||||
# Make start_polling() succeed immediately
|
||||
mock_updater.start_polling = AsyncMock(return_value=None)
|
||||
mock_updater.running = True
|
||||
@pytest.mark.asyncio
|
||||
async def test_bootstrap_start_polling_hang_schedules_recovery(monkeypatch):
|
||||
"""_start_polling_resilient: a hung bootstrap start_polling() must raise
|
||||
TimeoutError (an OSError → classified as network error) and schedule
|
||||
background recovery instead of blocking connect() forever."""
|
||||
monkeypatch.setattr(tg_adapter, "_UPDATER_START_TIMEOUT", 0.2)
|
||||
a = _bare_adapter()
|
||||
|
||||
# Mock _drain_polling_connections and _verify_polling_after_reconnect
|
||||
with patch.object(adapter, '_drain_polling_connections', new=AsyncMock()), \
|
||||
patch.object(adapter, '_verify_polling_after_reconnect', new=AsyncMock()):
|
||||
# Trigger the network error handler
|
||||
await adapter._handle_polling_network_error(Exception("test"))
|
||||
app = MagicMock()
|
||||
app.updater = AsyncMock()
|
||||
app.updater.start_polling = _hang_forever
|
||||
a._app = app
|
||||
|
||||
# Verify that start_polling was called
|
||||
mock_updater.start_polling.assert_called_once()
|
||||
scheduled = []
|
||||
monkeypatch.setattr(
|
||||
a, "_schedule_polling_recovery",
|
||||
lambda err, reason: scheduled.append((err, reason)),
|
||||
raising=False,
|
||||
)
|
||||
|
||||
ok = await asyncio.wait_for(
|
||||
a._start_polling_resilient(drop_pending_updates=False, error_callback=None),
|
||||
timeout=10,
|
||||
)
|
||||
assert ok is False
|
||||
assert len(scheduled) == 1
|
||||
assert isinstance(scheduled[0][0], (TimeoutError, asyncio.TimeoutError))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_start_polling_success_path_unaffected(monkeypatch):
|
||||
"""Sanity: a fast start_polling() still returns True through the wrapper."""
|
||||
monkeypatch.setattr(tg_adapter, "_UPDATER_START_TIMEOUT", 5.0)
|
||||
a = _bare_adapter()
|
||||
|
||||
app = MagicMock()
|
||||
app.updater = AsyncMock()
|
||||
app.updater.start_polling = AsyncMock(return_value=None)
|
||||
a._app = app
|
||||
|
||||
ok = await a._start_polling_resilient(drop_pending_updates=False, error_callback=None)
|
||||
assert ok is True
|
||||
app.updater.start_polling.assert_awaited_once()
|
||||
|
||||
|
||||
def test_every_start_polling_call_site_is_time_bounded():
|
||||
"""Bug-class contract: every `updater.start_polling(` await in the adapter
|
||||
must be wrapped in asyncio.wait_for. A new unbounded call site reintroduces
|
||||
the #59614 wedge."""
|
||||
import inspect
|
||||
import re
|
||||
|
||||
src = inspect.getsource(tg_adapter)
|
||||
# Find each start_polling( call and check an enclosing wait_for within the
|
||||
# preceding 6 lines (the wrapper always sits directly above).
|
||||
lines = src.splitlines()
|
||||
unbounded = []
|
||||
for i, line in enumerate(lines):
|
||||
if re.search(r"updater\.start_polling\(", line) and "def " not in line:
|
||||
window = "\n".join(lines[max(0, i - 6):i + 1])
|
||||
if "wait_for" not in window:
|
||||
unbounded.append((i + 1, line.strip()))
|
||||
assert not unbounded, f"unbounded start_polling() call sites: {unbounded}"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue