fix(telegram): serialize application teardown

This commit is contained in:
chengyongru
2026-08-18 00:41:57 +08:00
committed by chengyongru
parent 8a928592c5
commit 2b4a04fb71
2 changed files with 58 additions and 14 deletions
+18 -14
View File
@@ -516,6 +516,7 @@ class TelegramChannel(BaseChannel):
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
self._app_ready = asyncio.Event() # cleared while the app is being rebuilt
self._teardown_lock = asyncio.Lock()
def _require_app(self) -> TelegramApplication:
if self._app is None:
@@ -768,21 +769,22 @@ class TelegramChannel(BaseChannel):
async def _teardown_app(self) -> None:
"""Shut down the application, tolerating partially started state."""
app, self._app = self._app, None
self._app_ready.clear()
if not app:
return
for step in (cast(Any, app.updater).stop, app.stop, app.shutdown):
async with self._teardown_lock:
app, self._app = self._app, None
self._app_ready.clear()
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)
# Application.shutdown() skips the HTTPX pools unless initialize()
# finished, so a failed startup leaks one per retry. This is idempotent.
try:
await step()
await app.bot.shutdown()
except Exception as e:
self.logger.debug("teardown step failed: {}", e)
# Application.shutdown() skips the HTTPX pools unless initialize()
# finished, so a failed startup leaks one per retry. This is idempotent.
try:
await app.bot.shutdown()
except Exception as e:
self.logger.debug("bot shutdown failed: {}", e)
self.logger.debug("bot shutdown failed: {}", e)
async def stop(self) -> None:
"""Stop the Telegram bot."""
@@ -804,7 +806,9 @@ class TelegramChannel(BaseChannel):
if self._app:
self.logger.info("Stopping bot...")
await self._teardown_app()
# Join an in-flight supervisor teardown before ChannelManager cancels
# start(), otherwise cancellation can strand the old HTTPX pools.
await self._teardown_app()
@staticmethod
def _get_media_type(path: str) -> str:
@@ -563,6 +563,46 @@ async def test_stop_during_startup_does_not_leak_app(monkeypatch) -> None:
assert channel._app is None # torn down, not leaked
@pytest.mark.asyncio
async def test_stop_waits_for_inflight_watchdog_teardown(monkeypatch) -> None:
"""Manager cancellation after stop() must not interrupt an active teardown."""
_FakeHTTPXRequest.clear()
channel = TelegramChannel(
TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
MessageBus(),
)
teardown_started = asyncio.Event()
finish_teardown = asyncio.Event()
app = _FakeApp(lambda: None)
async def slow_updater_stop() -> None:
teardown_started.set()
await finish_teardown.wait()
app.updater.stop = slow_updater_stop
monkeypatch.setattr("nanobot.channels.telegram.runtime.HTTPXRequest", _FakeHTTPXRequest)
monkeypatch.setattr(
"nanobot.channels.telegram.runtime.Application",
SimpleNamespace(builder=lambda: _FakeBuilder(app)),
)
monkeypatch.setattr("nanobot.channels.telegram.runtime.POLL_STALE_SECONDS", -1.0)
monkeypatch.setattr("nanobot.channels.telegram.runtime.POLL_WATCH_INTERVAL", 0.0)
start_task = asyncio.create_task(channel.start())
await teardown_started.wait()
assert channel._app is None
stop_task = asyncio.create_task(channel.stop())
await asyncio.sleep(0)
assert not stop_task.done()
finish_teardown.set()
await stop_task
await start_task
assert app.bot.shutdown_calls == 1
@pytest.mark.asyncio
async def test_send_during_rebuild_fails_instead_of_dropping(monkeypatch) -> None:
"""A send that cannot reach Telegram must raise so the manager can retry."""