diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index ede20eb967..1cb69b2095 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -544,6 +544,13 @@ class TelegramAdapter(BasePlatformAdapter): self._seen_update_ids: dict = {} self._inflight_update_ids: dict = {} self._update_admission = None + # Completed update IDs survive adapter replacement and restarts (update_admission.py). + # Resolved now: secondary profiles construct adapters inside their own home scope. + from hermes_constants import get_hermes_home + self._update_receipt_dir = get_hermes_home() + self._update_receipts_loaded: set = set() + self._update_receipts_dirty: set = set() + self._update_receipt_flush: Optional[asyncio.Task] = None self._bot: Optional[Bot] = None self._webhook_mode: bool = False self._mention_patterns = self._compile_mention_patterns() @@ -3453,6 +3460,10 @@ class TelegramAdapter(BasePlatformAdapter): logger.warning("[%s] Error during Telegram disconnect: %s", self.name, _redact_telegram_error_text(e)) self._app = None self._bot = None + # Land the last completed receipts before a replacement adapter reads them. + flush = getattr(self, "_update_receipt_flush", None) + if flush is not None and not flush.done(): + await self._await_disconnect_step(asyncio.shield(flush), _DISCONNECT_STEP_TIMEOUT, "update-receipt flush") logger.info("[%s] Disconnected from Telegram", self.name) def _should_thread_reply(self, reply_to: Optional[str], chunk_index: int) -> bool: diff --git a/plugins/platforms/telegram/update_admission.py b/plugins/platforms/telegram/update_admission.py index 0ae321a444..982f0068a4 100644 --- a/plugins/platforms/telegram/update_admission.py +++ b/plugins/platforms/telegram/update_admission.py @@ -1,11 +1,17 @@ """Telegram admission before every PTB handler group, including native plugins. Claims belong to the receiving adapter across Application rebuilds. Completed -history is bounded and has no TTL; dispatch and its PTB tasks pin active claims. -No disk receipt, cross-process coordination or exactly-once effects are promised. +history is bounded and has no TTL in memory; dispatch and its PTB tasks pin active +claims. Completed IDs are also written to a per-bot receipt file under the adapter's +Hermes home, so a replacement adapter or a restarted gateway still drops updates that +Telegram redelivers because their getUpdates/webhook acknowledgement never landed. +No cross-process coordination or exactly-once effects are promised. """ import asyncio +import json +import logging +import time from contextvars import ContextVar from dataclasses import dataclass from functools import wraps @@ -14,9 +20,74 @@ from telegram import Update from telegram.ext import Application, ApplicationHandlerStop, ConversationHandler from gateway.platforms.helpers import bounded_put +from utils import atomic_json_write +logger = logging.getLogger(__name__) _DEFAULT_BLOCK = object() +_SEEN_CAP = 4096 +# The Bot API keeps an unconfirmed update for at most 24 hours (getUpdates), so an older +# receipt can never match a redelivery. That is also well inside the week of silence after +# which Telegram may restart update IDs at a random value, so a set lookup stays safe +# where a persisted numeric high-watermark would not. +RECEIPT_TTL_SECONDS = 24 * 60 * 60 + + +def _receipt_path(adapter, bot_id): + return adapter._update_receipt_dir / f"telegram_update_receipts_{bot_id}.json" + + +def _load_receipts(adapter, bot_id) -> None: + """Seed completed history from disk once per bot, before its first admission check.""" + if bot_id in adapter._update_receipts_loaded: + return + adapter._update_receipts_loaded.add(bot_id) + path = _receipt_path(adapter, bot_id) + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError: + return + except (OSError, ValueError): + logger.warning("[Telegram] Ignoring unreadable update receipts at %s", path, exc_info=True) + return + ids = payload.get("update_ids") if isinstance(payload, dict) else None + if not isinstance(ids, dict): + return + cutoff = time.time() - RECEIPT_TTL_SECONDS + fresh = sorted((float(ts), uid) for uid, ts in ids.items() + if isinstance(ts, (int, float)) and ts > cutoff and str(uid).lstrip("-").isdigit()) + seen = adapter._seen_update_ids + for ts, uid in fresh: + key = f"{bot_id}:{uid}" + if key not in seen: + bounded_put(seen, key, ts, _SEEN_CAP) + + +def _record_receipt(adapter, key: str) -> None: + bounded_put(adapter._seen_update_ids, key, time.time(), _SEEN_CAP) + adapter._update_receipts_dirty.add(key.split(":", 1)[0]) + task = adapter._update_receipt_flush + if task is None or task.done(): + # One writer per adapter coalesces bursts; the fsync stays off the event loop. + adapter._update_receipt_flush = asyncio.get_running_loop().create_task( + _flush_receipts(adapter), name="telegram-update-receipts") + + +async def _flush_receipts(adapter) -> None: + from hermes_constants import mkdir_under_hermes_home + + while adapter._update_receipts_dirty: + bot_id = adapter._update_receipts_dirty.pop() + prefix, cutoff = f"{bot_id}:", time.time() - RECEIPT_TTL_SECONDS + # Snapshot on the loop: admission mutates this dict only from loop callbacks. + ids = {key[len(prefix):]: ts for key, ts in adapter._seen_update_ids.items() + if key.startswith(prefix) and isinstance(ts, float) and ts > cutoff} + path = _receipt_path(adapter, bot_id) + try: + mkdir_under_hermes_home(path.parent) + await asyncio.to_thread(atomic_json_write, path, {"update_ids": ids}, indent=None) + except Exception: + logger.warning("[Telegram] Failed to persist update receipts to %s", path, exc_info=True) @dataclass @@ -140,7 +211,7 @@ class TelegramApplication(Application): return del self.adapter._inflight_update_ids[claim.key] if claim.accepted or (claim.completed and not claim.failed): - bounded_put(self.adapter._seen_update_ids, claim.key, None, 4096) + _record_receipt(self.adapter, claim.key) async def process_error(self, update, error, job=None, coroutine=None): claim = self._current_claim.get() @@ -153,7 +224,9 @@ class TelegramApplication(Application): async def process_update(self, update): if not isinstance(update, Update): return await super().process_update(update) - key = f"{self.bot.id}:{update.update_id}" + bot_id = self.bot.id + key = f"{bot_id}:{update.update_id}" + _load_receipts(self.adapter, bot_id) # Dispatch happened even when preparation fails before the group-99 observer. self.adapter._updates_dispatched_total += 1 seen = self.adapter._seen_update_ids diff --git a/tests/plugins/test_telegram_update_admission.py b/tests/plugins/test_telegram_update_admission.py index b345515bfb..af2ebb1459 100644 --- a/tests/plugins/test_telegram_update_admission.py +++ b/tests/plugins/test_telegram_update_admission.py @@ -84,7 +84,7 @@ def update(bot, uid=10, kind="text", *, edited=False, chat=42, text="hello", gro @asynccontextmanager -async def connected(monkeypatch, *, extra=None, bot_id=111): +async def connected(monkeypatch, *, extra=None, bot_id=111, is_reconnect=False): adapter = TelegramAdapter(PlatformConfig(enabled=True, token=f"{bot_id}:offline-test", extra=extra or {})) # Only transport/lifecycle services and the final model-work boundary are replaced. monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", "88") @@ -102,7 +102,7 @@ async def connected(monkeypatch, *, extra=None, bot_id=111): return True monkeypatch.setattr(adapter, "_start_session_processing", start) - assert await adapter.connect() + assert await adapter.connect(is_reconnect=is_reconnect) try: yield adapter, adapter._app, delivered finally: @@ -192,13 +192,14 @@ async def test_replay_is_admitted_once_before_dispatch(monkeypatch, tmp_path, ki await app.process_update(update(app.bot, **args)) assert len(delivered) == 2 if mode == "owners": - for profile, bot_id in (("alpha", 222), ("beta", 333), ("alpha", 222)): + for profile, bot_id, fresh in (("alpha", 222, 1), ("beta", 333, 1), ("alpha", 222, 0)): token = set_hermes_home_override(tmp_path / profile) try: async with connected(monkeypatch, bot_id=bot_id) as (other, other_app, other_delivered): await other_app.process_update(update(other_app.bot)) await asyncio.gather(*other._pending_text_batch_tasks.values()) - assert len(other_delivered) == 1 + # A rebuilt adapter for the same bot and home reads that home's receipt. + assert len(other_delivered) == fresh finally: reset_hermes_home_override(token) await app.process_update(update(app.bot)) @@ -739,3 +740,53 @@ async def test_only_pre_handoff_failure_reopens_admission(monkeypatch, tmp_path, assert len(delivered) == 1 assert delivered[0].text == ("/status" if stage == "dispatch" else "hello") assert adapter._platform_event_handler.await_count == (1 if stage in ("prepare", "pressure", "batch_prepare") else 0) + + +@pytest.mark.asyncio +async def test_redelivery_to_rebuilt_adapter_is_dropped(monkeypatch, tmp_path): + """The reconnect watcher and a gateway restart both build a new adapter, and a new PTB + Updater polls from offset 0: Telegram resends every update whose acknowledgement never + landed. The receipt must outlive the adapter that completed the update.""" + from hermes_constants import get_hermes_home + from plugins.platforms.telegram.update_admission import RECEIPT_TTL_SECONDS + + receipts = get_hermes_home() / "telegram_update_receipts_111.json" + async with connected(monkeypatch) as (adapter, app, delivered): + await app.process_update(update(app.bot, 10)) + with monkeypatch.context() as broken: + broken.setattr(adapter, "_cache_replied_media", AsyncMock(side_effect=OSError("before enqueue"))) + await app.process_update(update(app.bot, 20, text="unaccepted")) + await asyncio.gather(*adapter._pending_text_batch_tasks.values()) + assert [event.text for event in delivered] == ["hello"] + # disconnect() waits for the receipt write; the failed preparation is not a receipt. + assert set(json.loads(receipts.read_text())["update_ids"]) == {"10"} + + # A fresh adapter, connected the way the gateway reconnect watcher does it. + async with connected(monkeypatch, is_reconnect=True) as (adapter, app, delivered): + await app.process_update(update(app.bot, 10)) + await app.process_update(update(app.bot, 20, text="unaccepted")) + await asyncio.gather(*adapter._pending_text_batch_tasks.values()) + await app.process_update(update(app.bot, 21, edited=True, text="changed")) + await asyncio.gather(*adapter._pending_text_batch_tasks.values()) + # Old update dropped; the retry of an unaccepted one and an edit of message 472 still run. + assert [event.text for event in delivered] == ["unaccepted", "changed"] + assert adapter._updates_dispatched_total == 3 + + # Receipts are bot-scoped, and older than Telegram's 24h retention they cannot match. + async with connected(monkeypatch, bot_id=222) as (adapter, app, delivered): + await app.process_update(update(app.bot, 10)) + await asyncio.gather(*adapter._pending_text_batch_tasks.values()) + assert len(delivered) == 1 + stale = time.time() - RECEIPT_TTL_SECONDS - 1 + receipts.write_text(json.dumps({"update_ids": {"10": stale, "20": "bad", "x": time.time()}})) + async with connected(monkeypatch) as (adapter, app, delivered): + await app.process_update(update(app.bot, 10)) + await asyncio.gather(*adapter._pending_text_batch_tasks.values()) + assert len(delivered) == 1 + assert set(json.loads(receipts.read_text())["update_ids"]) == {"10"} + + receipts.write_text("{not json") + async with connected(monkeypatch) as (adapter, app, delivered): + await app.process_update(update(app.bot, 30)) + await asyncio.gather(*adapter._pending_text_batch_tasks.values()) + assert len(delivered) == 1