diff --git a/nanobot/channels/telegram/runtime.py b/nanobot/channels/telegram/runtime.py index 4f5ae0ddd..5a8947b0f 100644 --- a/nanobot/channels/telegram/runtime.py +++ b/nanobot/channels/telegram/runtime.py @@ -25,9 +25,9 @@ from telegram import ( Update, User, ) -from telegram.error import BadRequest, NetworkError, TimedOut +from telegram.error import BadRequest, InvalidToken, NetworkError, TimedOut from telegram.ext import Application, CallbackQueryHandler, ContextTypes, MessageHandler, filters -from telegram.request import HTTPXRequest +from telegram.request import BaseRequest, HTTPXRequest from nanobot.bus.events import OutboundMessage from nanobot.bus.outbound_events import ProgressEvent @@ -38,6 +38,7 @@ from nanobot.config.paths import get_media_dir from nanobot.config.schema import Base from nanobot.security.network import validate_url_target from nanobot.utils.helpers import split_message +from nanobot.utils.logging_bridge import redirect_lib_logging TELEGRAM_MAX_MESSAGE_LEN = 4000 # Telegram message character limit # Telegram's actual API limit is 4096; we split raw markdown at 4000 as a @@ -53,6 +54,39 @@ TELEGRAM_REPLY_CONTEXT_MAX_LEN = TELEGRAM_MAX_MESSAGE_LEN # Max length for repl TelegramApplication: TypeAlias = Application[Any, Any, Any, Any, Any, Any] _T = TypeVar("_T") +# A healthy getUpdates long poll completes every ~10s even with no traffic; +# PTB retries timeouts silently, so stalls must be detected here. +POLL_STALE_SECONDS = 120.0 +POLL_WATCH_INTERVAL = 1.0 +RESTART_BACKOFF_INITIAL_SECONDS = 5.0 +RESTART_BACKOFF_MAX_SECONDS = 300.0 + + +class _LivenessTrackedRequest(BaseRequest): + """Wrap the getUpdates request pool, reporting each completed round trip.""" + + __slots__ = ("inner", "_on_round_trip") + + def __init__(self, inner: BaseRequest, on_round_trip: Callable[[], None]) -> None: + super().__init__() + self.inner = inner + self._on_round_trip = on_round_trip + + @property + def read_timeout(self) -> float | None: + return self.inner.read_timeout + + async def initialize(self) -> None: + await self.inner.initialize() + + async def shutdown(self) -> None: + await self.inner.shutdown() + + async def do_request(self, *args: Any, **kwargs: Any) -> tuple[int, bytes]: + result = await self.inner.do_request(*args, **kwargs) + self._on_round_trip() + return result + def _split_telegram_markdown(content: str, max_len: int) -> list[str]: """Split raw Telegram Markdown without leaving fenced code blocks unbalanced.""" @@ -477,6 +511,7 @@ class TelegramChannel(BaseChannel): self._inbound_buffers: dict[str, list[_QueuedTelegramUpdate]] = {} self._inbound_workers: dict[str, asyncio.Task[None]] = {} self._rich_send_disabled: bool = False # Latch off if Bot API < 10.1 + self._last_poll_ok: float = 0.0 # monotonic time of last getUpdates round trip def _require_app(self) -> TelegramApplication: if self._app is None: @@ -516,13 +551,56 @@ class TelegramChannel(BaseChannel): return content async def start(self) -> None: - """Start the Telegram bot.""" + """Start the Telegram bot, rebuilding the app whenever polling stalls.""" if not self.config.token: self.logger.error("bot token not configured") return - self._running = True + redirect_lib_logging("telegram") + redirect_lib_logging("httpx", level="WARNING") + self._running = True + backoff = RESTART_BACKOFF_INITIAL_SECONDS + while self._running: + try: + await self._start_app() + except InvalidToken as e: + # A rejected token is a config error, not a transient network + # failure — retrying forever would only spam the log. + await self._teardown_app() + self.logger.error("bot token rejected: {}", self._format_telegram_error(e)) + return + except Exception as e: + await self._teardown_app() + if not self._running: + break + self.logger.error( + "startup failed: {}; retrying in {:.0f}s", + self._format_telegram_error(e), + backoff, + ) + await self._idle(backoff) + backoff = min(backoff * 2, RESTART_BACKOFF_MAX_SECONDS) + continue + + backoff = RESTART_BACKOFF_INITIAL_SECONDS + if not self._running: + # stop() ran while _start_app() was mid-flight and tore down the + # previous (possibly None) app; this one would leak otherwise. + await self._teardown_app() + break + stalled = await self._watch_polling() + if not stalled or not self._running: + break + self.logger.warning( + "polling stalled: no getUpdates round trip for {:.0f}s; " + "rebuilding connection pools and restarting", + time.monotonic() - self._last_poll_ok, + ) + await self._teardown_app() + + async def _start_app(self) -> None: + """Build, initialize and start the Telegram application.""" proxy = self.config.proxy or None # Separate pools so long-polling (getUpdates) never starves outbound sends. @@ -544,7 +622,7 @@ class TelegramChannel(BaseChannel): Application.builder() .token(self.config.token) .request(api_request) - .get_updates_request(poll_request) + .get_updates_request(_LivenessTrackedRequest(poll_request, self._note_poll_ok)) ) self._app = builder.build() self._app.add_error_handler(self._on_error) @@ -621,16 +699,43 @@ class TelegramChannel(BaseChannel): max_connections=self.config.webhook_max_connections, ) else: - # Start polling (this runs until stopped) + self._last_poll_ok = time.monotonic() await cast(Any, self._app.updater).start_polling( allowed_updates=allowed_updates, drop_pending_updates=False, # Process pending messages on startup error_callback=self._on_polling_error, ) - # Keep running until stopped + def _note_poll_ok(self) -> None: + # HTTP error statuses count too: the watchdog detects transport stalls, + # not logical failures. + self._last_poll_ok = time.monotonic() + + async def _watch_polling(self) -> bool: + """Idle until stop(); in polling mode, return True when getUpdates goes stale.""" + watch = self.config.mode != "webhook" while self._running: - await asyncio.sleep(1) + await asyncio.sleep(POLL_WATCH_INTERVAL) + if watch and time.monotonic() - self._last_poll_ok > POLL_STALE_SECONDS: + return True + return False + + async def _idle(self, seconds: float) -> None: + """Sleep in short steps so stop() stays responsive.""" + deadline = time.monotonic() + seconds + while self._running and time.monotonic() < deadline: + await asyncio.sleep(POLL_WATCH_INTERVAL) + + async def _teardown_app(self) -> None: + """Shut down the application, tolerating partially started state.""" + app, self._app = self._app, None + if not app: + return + for step in (cast(Any, app.updater).stop, app.stop, app.shutdown): + try: + await step() + except Exception as e: + self.logger.debug("teardown step failed: {}", e) async def stop(self) -> None: """Stop the Telegram bot.""" @@ -652,10 +757,7 @@ class TelegramChannel(BaseChannel): if self._app: self.logger.info("Stopping bot...") - await cast(Any, self._app.updater).stop() - await self._app.stop() - await self._app.shutdown() - self._app = None + await self._teardown_app() @staticmethod def _get_media_type(path: str) -> str: diff --git a/nanobot/channels/telegram/tests/test_telegram_channel.py b/nanobot/channels/telegram/tests/test_telegram_channel.py index 160b3acd9..5c4e7682a 100644 --- a/nanobot/channels/telegram/tests/test_telegram_channel.py +++ b/nanobot/channels/telegram/tests/test_telegram_channel.py @@ -337,7 +337,7 @@ async def test_start_creates_separate_pools_with_proxy(monkeypatch) -> None: assert api_req.kwargs["connection_pool_size"] == 32 assert poll_req.kwargs["connection_pool_size"] == 4 assert builder.request_value is api_req - assert builder.get_updates_request_value is poll_req + assert builder.get_updates_request_value.inner is poll_req assert callable(app.updater.start_polling_kwargs["error_callback"]) assert any(cmd.command == "status" for cmd in app.bot.commands) assert any(cmd.command == "history" for cmd in app.bot.commands) @@ -378,6 +378,161 @@ async def test_start_respects_custom_pool_config(monkeypatch) -> None: assert poll_req.kwargs["pool_timeout"] == 10.0 +@pytest.mark.asyncio +async def test_stalled_polling_triggers_pool_rebuild(monkeypatch) -> None: + """When no getUpdates round trip completes for too long, the app is rebuilt.""" + _FakeHTTPXRequest.clear() + config = TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]) + bus = MessageBus() + channel = TelegramChannel(config, bus) + + apps: list[_FakeApp] = [] + + def on_start_polling() -> None: + if len(apps) >= 2: + channel._running = False + + def make_builder(): + app = _FakeApp(on_start_polling) + apps.append(app) + return _FakeBuilder(app) + + monkeypatch.setattr("nanobot.channels.telegram.runtime.HTTPXRequest", _FakeHTTPXRequest) + monkeypatch.setattr( + "nanobot.channels.telegram.runtime.Application", + SimpleNamespace(builder=make_builder), + ) + monkeypatch.setattr("nanobot.channels.telegram.runtime.POLL_STALE_SECONDS", -1.0) + monkeypatch.setattr("nanobot.channels.telegram.runtime.POLL_WATCH_INTERVAL", 0.0) + + await channel.start() + + assert len(apps) == 2 + assert apps[0].updater.start_polling_kwargs is not None + assert apps[1].updater.start_polling_kwargs is not None + # 2 fresh pools per app + assert len(_FakeHTTPXRequest.instances) == 4 + + +@pytest.mark.asyncio +async def test_startup_failure_retries_with_backoff(monkeypatch) -> None: + """Transient startup failures back off and retry until the app comes up.""" + from telegram.error import NetworkError + + _FakeHTTPXRequest.clear() + config = TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]) + bus = MessageBus() + channel = TelegramChannel(config, bus) + + apps: list[_FakeApp] = [] + + def make_builder(): + app = _FakeApp(lambda: setattr(channel, "_running", False)) + if len(apps) < 2: + async def _fail() -> None: + raise NetworkError("connect failed") + + app.initialize = _fail + apps.append(app) + return _FakeBuilder(app) + + monkeypatch.setattr("nanobot.channels.telegram.runtime.HTTPXRequest", _FakeHTTPXRequest) + monkeypatch.setattr( + "nanobot.channels.telegram.runtime.Application", + SimpleNamespace(builder=make_builder), + ) + monkeypatch.setattr("nanobot.channels.telegram.runtime.POLL_WATCH_INTERVAL", 0.0) + monkeypatch.setattr("nanobot.channels.telegram.runtime.RESTART_BACKOFF_INITIAL_SECONDS", 0.0) + + await channel.start() + + assert len(apps) == 3 + assert apps[0].updater.start_polling_kwargs is None + assert apps[1].updater.start_polling_kwargs is None + assert apps[2].updater.start_polling_kwargs is not None + + +@pytest.mark.asyncio +async def test_invalid_token_stops_without_retry(monkeypatch) -> None: + """A rejected token is a config error: log it and give up instead of retrying.""" + from telegram.error import InvalidToken + + _FakeHTTPXRequest.clear() + config = TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]) + bus = MessageBus() + channel = TelegramChannel(config, bus) + + apps: list[_FakeApp] = [] + + def make_builder(): + app = _FakeApp(lambda: None) + + async def _reject() -> None: + raise InvalidToken("token rejected by Telegram") + + app.initialize = _reject + apps.append(app) + return _FakeBuilder(app) + + monkeypatch.setattr("nanobot.channels.telegram.runtime.HTTPXRequest", _FakeHTTPXRequest) + monkeypatch.setattr( + "nanobot.channels.telegram.runtime.Application", + SimpleNamespace(builder=make_builder), + ) + monkeypatch.setattr("nanobot.channels.telegram.runtime.RESTART_BACKOFF_INITIAL_SECONDS", 0.0) + + await channel.start() + + assert len(apps) == 1 + assert channel._app is None + + +@pytest.mark.asyncio +async def test_stop_during_startup_does_not_leak_app(monkeypatch) -> None: + """stop() landing while _start_app() is mid-flight must not leave the app running.""" + _FakeHTTPXRequest.clear() + config = TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]) + bus = MessageBus() + channel = TelegramChannel(config, bus) + + # Simulate stop() winning the race just before start_polling returns. + app = _FakeApp(lambda: setattr(channel, "_running", False)) + builder = _FakeBuilder(app) + + monkeypatch.setattr("nanobot.channels.telegram.runtime.HTTPXRequest", _FakeHTTPXRequest) + monkeypatch.setattr( + "nanobot.channels.telegram.runtime.Application", + SimpleNamespace(builder=lambda: builder), + ) + + await channel.start() + + assert channel._app is None # torn down, not leaked + + +@pytest.mark.asyncio +async def test_liveness_tracked_request_stamps_on_round_trip() -> None: + from nanobot.channels.telegram.runtime import _LivenessTrackedRequest + + stamps: list[int] = [] + + class _Inner: + read_timeout = 5.0 + + async def initialize(self) -> None: + pass + + async def shutdown(self) -> None: + pass + + async def do_request(self, *args, **kwargs): + return 200, b"{}" + + wrapped = _LivenessTrackedRequest(_Inner(), lambda: stamps.append(1)) + assert await wrapped.do_request(url="https://example.org", method="POST") == (200, b"{}") + assert stamps == [1] + + def test_webhook_config_requires_https_url_and_secret() -> None: with pytest.raises(ValueError, match="webhook_url is required"): TelegramConfig(enabled=True, token="123:abc", mode="webhook")