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.
This commit is contained in:
kshitijk4poor
@@ -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") \
|
||||
|
||||
141
plugins/platforms/telegram/telegram_held_inbound.py
Normal file
141
plugins/platforms/telegram/telegram_held_inbound.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user