From d169a2f3a9ef2b69dc8e49ec526e46aa7fcdb7af Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Mon, 5 Oct 2026 15:34:23 +0530 Subject: [PATCH] refactor(telegram): move the held-inbound queue into telegram_held_inbound to keep adapter.py under its size cap Code-health ratchet: plugins/platforms/telegram/adapter.py was 7446 lines > its 7422 cap after the held-inbound hand-over (#132829) added adopt_held_inbound/_adopt_held_event. The whole hold queue (_schedule_held_inbound_redispatch, _hold_inbound_event, adopt_held_inbound, _adopt_held_event, _rehold_from, _redispatch_held_inbound) moves verbatim into TelegramHeldInboundMixin; TelegramAdapter inherits it, so call sites and instance-level patches are unchanged. The sibling logs under the adapter's logger name so existing log filters still match. --- plugins/platforms/telegram/adapter.py | 130 +--------------- .../telegram/telegram_held_inbound.py | 141 ++++++++++++++++++ 2 files changed, 143 insertions(+), 128 deletions(-) create mode 100644 plugins/platforms/telegram/telegram_held_inbound.py diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index f832684828..b41b82963c 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -10,7 +10,6 @@ import os import html as _html import re import time -import weakref from contextvars import ContextVar from datetime import datetime, timezone from typing import Any, Awaitable, Callable, Dict, Iterator, List, Optional, Set @@ -179,6 +178,7 @@ _MEDIA_KIND_KEYS = { from gateway.platforms.event import MessageEvent, MessageType, ProcessingOutcome from plugins.platforms.telegram.telegram_entities import expand_link_entities +from plugins.platforms.telegram.telegram_held_inbound import TelegramHeldInboundMixin from plugins.platforms.telegram.telegram_ids import normalize_telegram_chat_id from plugins.platforms.telegram.telegram_network import ( SEED_FALLBACK_IPS, TelegramFallbackTransport, discover_fallback_ips, parse_fallback_ip_env, tcp_keepalive_socket_options) @@ -401,7 +401,6 @@ def _separate_chunk_indicator_from_fence(text: str) -> str: # MarkdownV2 has no table syntax, so pipe tables become bullet groups via convert_table_to_bullets(). from gateway.platforms.helpers import ( TABLE_SEPARATOR_RE as _TABLE_SEPARATOR_RE, compile_mention_patterns, convert_table_to_bullets as _wrap_markdown_tables) -from gateway.platforms.helpers import cancel_task # Rich-message regions whose internal newlines must stay bare (Telegram renders them natively): # fenced code blocks OR GFM pipe-table blocks (header row, delimiter row, data rows). @@ -511,7 +510,7 @@ class _PollingStallError(RuntimeError): """ -class TelegramAdapter(BasePlatformAdapter): +class TelegramAdapter(TelegramHeldInboundMixin, BasePlatformAdapter): """Telegram bot adapter: users/groups, MarkdownV2 replies, forum topics, media.""" # Bound for the per-(chat_id, status_key) status-message cache; FIFO half-trim on overflow. @@ -787,131 +786,6 @@ class TelegramAdapter(BasePlatformAdapter): Callers must NOT destroy the event (PTB already advanced the offset) — hold and redispatch.""" return bool(getattr(self, "_drop_delayed_deliveries", False)) - def _schedule_held_inbound_redispatch(self) -> None: - """Ensure a tracked drain runs when held events exist and delivery is live (no-op while - down or after permanent fatal; an in-flight drain schedules its own follow-up).""" - if self._is_permanent_fatal() or self._should_drop_delayed_delivery(): - return - if not getattr(self, "_held_inbound_events", None): - return - try: - loop = asyncio.get_running_loop() - except RuntimeError: - return - prior = getattr(self, "_held_inbound_redispatch_task", None) - try: - current = asyncio.current_task() - except RuntimeError: - current = None - if prior is not None and not prior.done() and prior is not current: - return - self._held_inbound_redispatch_task = loop.create_task(self._redispatch_held_inbound(prior=None if prior is current else prior)) - - def _hold_inbound_event(self, event: "MessageEvent", *, where: str, schedule: bool = True) -> None: - """Preserve an inbound event that cannot be dispatched now (PTB already acked the update, so dropping is silent loss). - Capped, identity-deduped; permanent fatal discards. ``schedule=False`` inside a drain avoids poison-event loops. - - The disconnect drop-guard (#55971) correctly prevents dispatch into a torn-down session. Destroying - the event is wrong: by the time we reach enqueue/flush, python-telegram-bot has already acked the - update and advanced the offset — silent permanent loss, no log, no error. - """ - if self._is_permanent_fatal(): - logger.warning( - "[Telegram] Discarding inbound under non-retryable fatal (%s, %d chars)", where, len(getattr(event, "text", None) or "")) - return - successor = getattr(self, "_held_inbound_successor", None) - target = successor() if successor is not None else None - if target is not None: # retired by a runner rebuild (#132829) - self._accept_update() # the claim being dispatched is ours, not the successor's - target._adopt_held_event(event, where=f"{where}-forwarded", schedule=schedule) - return - held = getattr(self, "_held_inbound_events", None) - if held is None: - self._held_inbound_events = held = [] - if any(existing is event for existing in held): - return - max_n = int(getattr(self, "HELD_INBOUND_MAX", 64) or 64) - while len(held) >= max_n: - dropped = held.pop(0) - logger.warning( - "[Telegram] Held-inbound queue full (%d); dropping oldest (%d chars)", max_n, len(getattr(dropped, "text", None) or "")) - held.append(event) - self._accept_update() - logger.warning( - "[Telegram] Holding inbound (%s, %d chars, queue=%d)%s", where, len(getattr(event, "text", None) or ""), len(held), - " - will redispatch on reconnect" if self._should_drop_delayed_delivery() else (" - scheduling redispatch" if schedule else "")) - # A live-path hold must not orphan the event waiting for a reconnect that never comes. - if schedule and not self._should_drop_delayed_delivery(): - self._schedule_held_inbound_redispatch() - - def adopt_held_inbound(self, predecessor: "TelegramAdapter") -> None: - """Take over the hold queue of the instance the runner just replaced with us (#132829): it only - drains on its own ``_mark_connected``, which never comes; later holds there forward here.""" - predecessor._held_inbound_successor = weakref.ref(self) - held = getattr(predecessor, "_held_inbound_events", None) or [] - events = list(held) - held.clear() - for event in events: - self._adopt_held_event(event, where="adopted", schedule=False) - self._schedule_held_inbound_redispatch() - - def _adopt_held_event(self, event: "MessageEvent", *, where: str, schedule: bool) -> None: - """Hold a predecessor's event here; only the transport ref moves (profile/authz stay as resolved).""" - if getattr(event, "source", None) is not None: - event.source._transport_adapter_ref = weakref.ref(self) - self._hold_inbound_event(event, where=where, schedule=schedule) - - def _rehold_from(self, events: list, idx: int, where: str) -> None: - """Re-hold ``events[idx:]`` without rescheduling (drain interrupted / failed / cancelled).""" - for rest in events[idx:]: - self._hold_inbound_event(rest, where=where, schedule=False) - - async def _redispatch_held_inbound(self, prior: Optional[asyncio.Task] = None) -> None: - """Drain the hold queue after reconnect or a connected-path hold; ``prior`` (previous - redispatch task) is cancelled+awaited here so ``_mark_connected`` stays synchronous.""" - if prior is not asyncio.current_task(): # a self-redispatch must not cancel itself - await cancel_task(prior) - held = getattr(self, "_held_inbound_events", None) - if self._is_permanent_fatal(): - if held: - n = len(held) - held.clear() - logger.warning("[Telegram] Redispatch aborted; discarded %d held inbound under non-retryable fatal", n) - return - if not held: - return - # Take ownership atomically; concurrent holds append to the fresh list for a follow-up. - events = list(held) - held.clear() - logger.warning("[Telegram] Redispatching %d held inbound message(s)", len(events)) - allow_followup_schedule = True - try: - for idx, event in enumerate(events): - if self._is_permanent_fatal() or self._should_drop_delayed_delivery(): - self._rehold_from(events, idx, "redispatch-interrupted") - return - try: - await self.handle_message(event) - except asyncio.CancelledError: - self._rehold_from(events, idx, "redispatch-cancelled") - raise - except Exception: - # Retryable failure: re-hold but do NOT reschedule now (a poison event would - # tight-loop); the next mark_connected/live hold drains. - logger.exception( - "[Telegram] Failed to redispatch held inbound (%d chars); re-holding", len(getattr(event, "text", None) or "")) - self._rehold_from(events, idx, "redispatch-failed") - allow_followup_schedule = False - return - finally: - # Events that arrived mid-drain while still connected need another pass. - if ( - allow_followup_schedule - and getattr(self, "_held_inbound_events", None) - and not self._should_drop_delayed_delivery() - and not self._is_permanent_fatal()): - self._schedule_held_inbound_redispatch() - def _notification_kwargs(self, metadata: Optional[Dict[str, Any]]) -> Dict[str, Any]: """In "important" mode return disable_notification=True unless ``notify`` or ``is_approval_prompt`` (#132516).""" if getattr(self, "_notifications_mode", "important") != "important" or (metadata or {}).get("notify") \ diff --git a/plugins/platforms/telegram/telegram_held_inbound.py b/plugins/platforms/telegram/telegram_held_inbound.py new file mode 100644 index 0000000000..79d7d3878c --- /dev/null +++ b/plugins/platforms/telegram/telegram_held_inbound.py @@ -0,0 +1,141 @@ +"""Telegram held-inbound queue: events PTB already acked that cannot dispatch yet are held, +redispatched on reconnect, and handed to the adapter that replaces this one (#132829).""" + +import asyncio +import logging +import weakref +from typing import Optional + +from gateway.platforms.event import MessageEvent +from gateway.platforms.helpers import cancel_task + +logger = logging.getLogger("plugins.platforms.telegram.adapter") + + +class TelegramHeldInboundMixin: + """Hold / redispatch / hand-over of inbound events for ``TelegramAdapter``.""" + + def _schedule_held_inbound_redispatch(self) -> None: + """Ensure a tracked drain runs when held events exist and delivery is live (no-op while + down or after permanent fatal; an in-flight drain schedules its own follow-up).""" + if self._is_permanent_fatal() or self._should_drop_delayed_delivery(): + return + if not getattr(self, "_held_inbound_events", None): + return + try: + loop = asyncio.get_running_loop() + except RuntimeError: + return + prior = getattr(self, "_held_inbound_redispatch_task", None) + try: + current = asyncio.current_task() + except RuntimeError: + current = None + if prior is not None and not prior.done() and prior is not current: + return + self._held_inbound_redispatch_task = loop.create_task(self._redispatch_held_inbound(prior=None if prior is current else prior)) + + def _hold_inbound_event(self, event: "MessageEvent", *, where: str, schedule: bool = True) -> None: + """Preserve an inbound event that cannot be dispatched now (PTB already acked the update, so dropping is silent loss). + Capped, identity-deduped; permanent fatal discards. ``schedule=False`` inside a drain avoids poison-event loops. + + The disconnect drop-guard (#55971) correctly prevents dispatch into a torn-down session. Destroying + the event is wrong: by the time we reach enqueue/flush, python-telegram-bot has already acked the + update and advanced the offset — silent permanent loss, no log, no error. + """ + if self._is_permanent_fatal(): + logger.warning( + "[Telegram] Discarding inbound under non-retryable fatal (%s, %d chars)", where, len(getattr(event, "text", None) or "")) + return + successor = getattr(self, "_held_inbound_successor", None) + target = successor() if successor is not None else None + if target is not None: # retired by a runner rebuild (#132829) + self._accept_update() # the claim being dispatched is ours, not the successor's + target._adopt_held_event(event, where=f"{where}-forwarded", schedule=schedule) + return + held = getattr(self, "_held_inbound_events", None) + if held is None: + self._held_inbound_events = held = [] + if any(existing is event for existing in held): + return + max_n = int(getattr(self, "HELD_INBOUND_MAX", 64) or 64) + while len(held) >= max_n: + dropped = held.pop(0) + logger.warning( + "[Telegram] Held-inbound queue full (%d); dropping oldest (%d chars)", max_n, len(getattr(dropped, "text", None) or "")) + held.append(event) + self._accept_update() + logger.warning( + "[Telegram] Holding inbound (%s, %d chars, queue=%d)%s", where, len(getattr(event, "text", None) or ""), len(held), + " - will redispatch on reconnect" if self._should_drop_delayed_delivery() else (" - scheduling redispatch" if schedule else "")) + # A live-path hold must not orphan the event waiting for a reconnect that never comes. + if schedule and not self._should_drop_delayed_delivery(): + self._schedule_held_inbound_redispatch() + + def adopt_held_inbound(self, predecessor: "TelegramHeldInboundMixin") -> None: + """Take over the hold queue of the instance the runner just replaced with us (#132829): it only + drains on its own ``_mark_connected``, which never comes; later holds there forward here.""" + predecessor._held_inbound_successor = weakref.ref(self) + held = getattr(predecessor, "_held_inbound_events", None) or [] + events = list(held) + held.clear() + for event in events: + self._adopt_held_event(event, where="adopted", schedule=False) + self._schedule_held_inbound_redispatch() + + def _adopt_held_event(self, event: "MessageEvent", *, where: str, schedule: bool) -> None: + """Hold a predecessor's event here; only the transport ref moves (profile/authz stay as resolved).""" + if getattr(event, "source", None) is not None: + event.source._transport_adapter_ref = weakref.ref(self) + self._hold_inbound_event(event, where=where, schedule=schedule) + + def _rehold_from(self, events: list, idx: int, where: str) -> None: + """Re-hold ``events[idx:]`` without rescheduling (drain interrupted / failed / cancelled).""" + for rest in events[idx:]: + self._hold_inbound_event(rest, where=where, schedule=False) + + async def _redispatch_held_inbound(self, prior: Optional[asyncio.Task] = None) -> None: + """Drain the hold queue after reconnect or a connected-path hold; ``prior`` (previous + redispatch task) is cancelled+awaited here so ``_mark_connected`` stays synchronous.""" + if prior is not asyncio.current_task(): # a self-redispatch must not cancel itself + await cancel_task(prior) + held = getattr(self, "_held_inbound_events", None) + if self._is_permanent_fatal(): + if held: + n = len(held) + held.clear() + logger.warning("[Telegram] Redispatch aborted; discarded %d held inbound under non-retryable fatal", n) + return + if not held: + return + # Take ownership atomically; concurrent holds append to the fresh list for a follow-up. + events = list(held) + held.clear() + logger.warning("[Telegram] Redispatching %d held inbound message(s)", len(events)) + allow_followup_schedule = True + try: + for idx, event in enumerate(events): + if self._is_permanent_fatal() or self._should_drop_delayed_delivery(): + self._rehold_from(events, idx, "redispatch-interrupted") + return + try: + await self.handle_message(event) + except asyncio.CancelledError: + self._rehold_from(events, idx, "redispatch-cancelled") + raise + except Exception: + # Retryable failure: re-hold but do NOT reschedule now (a poison event would + # tight-loop); the next mark_connected/live hold drains. + logger.exception( + "[Telegram] Failed to redispatch held inbound (%d chars); re-holding", len(getattr(event, "text", None) or "")) + self._rehold_from(events, idx, "redispatch-failed") + allow_followup_schedule = False + return + finally: + # Events that arrived mid-drain while still connected need another pass. + if ( + allow_followup_schedule + and getattr(self, "_held_inbound_events", None) + and not self._should_drop_delayed_delivery() + and not self._is_permanent_fatal()): + self._schedule_held_inbound_redispatch()