diff --git a/gateway/platforms/helpers.py b/gateway/platforms/helpers.py index 0635334e98..6ddb232ff5 100644 --- a/gateway/platforms/helpers.py +++ b/gateway/platforms/helpers.py @@ -72,18 +72,28 @@ def inbound_dedup_caches(adapter: Any) -> dict[str, MessageDeduplicator]: return {name: v for name, v in vars(adapter).items() if isinstance(v, MessageDeduplicator)} -def carry_inbound_dedup(caches: Optional[dict], adapter: Any) -> None: - """Seed a rebuilt adapter's dedup caches from the instance it replaces. +def carry_inbound_dedup(predecessor: Any, adapter: Any) -> None: + """Seed a rebuilt adapter's dedup caches from the instance it replaces (call before connect). The runner's reconnect path builds a NEW adapter; without this a platform replaying a recent inbound ID after the reconnect (websocket resume, webhook retry, unacked poll batch) is admitted and answered a second time.""" - for name, previous in (caches or {}).items(): + for name, previous in (inbound_dedup_caches(predecessor) if predecessor is not None else {}).items(): current = getattr(adapter, name, None) if isinstance(current, MessageDeduplicator) and current is not previous: current.absorb(previous) +def hand_over_held_inbound(predecessor: Any, adapter: Any) -> None: + """Move inbound the retired instance is holding to its now-published replacement (#132829). + + Runs at publish time, not before connect: a candidate that fails to connect never owns the + queue, and the predecessor keeps holding until a replacement is registered.""" + adopt = getattr(adapter, "adopt_held_inbound", None) + if predecessor is not None and callable(adopt): + adopt(predecessor) + + # Worker-thread handoff used by the off-loop persist paths. A module attribute # so tests can replace THIS seam instead of patching ``asyncio.to_thread`` # globally. diff --git a/gateway/run_adapters.py b/gateway/run_adapters.py index d705c429c2..f280172e20 100644 --- a/gateway/run_adapters.py +++ b/gateway/run_adapters.py @@ -26,7 +26,7 @@ from gateway.config import ( platform_binds_port as _platform_binds_port, ) from gateway.platforms.base import BasePlatformAdapter -from gateway.platforms.helpers import carry_inbound_dedup, inbound_dedup_caches +from gateway.platforms.helpers import carry_inbound_dedup, hand_over_held_inbound from gateway.restart import is_global_startup_conflict from gateway.run_shutdown import _log_suppressed from gateway.session import SessionSource @@ -244,7 +244,8 @@ class GatewayAdapterLifecycleMixin: **({"queued_at": now} if queued else {}), "credential_claim": self._adapter_credential_claim(platform, adapter), "listener_claim": self._adapter_listener_claim(platform, adapter), - "inbound_dedup": inbound_dedup_caches(adapter) if adapter is not None else None, + # The replaced instance: its dedup caches seed each candidate, its held inbound moves on publish. + "predecessor": adapter, } def _queue_retryable_fatal_platform(self, adapter: BasePlatformAdapter) -> bool: @@ -827,7 +828,7 @@ class GatewayAdapterLifecycleMixin: ) logger.info("Reconnect %s: no adapter yet, next retry in %ds", platform.value, backoff) return - carry_inbound_dedup(info.get("inbound_dedup"), adapter) + carry_inbound_dedup(info.get("predecessor"), adapter) self._wire_adapter_handlers(adapter) # is_reconnect keeps the server-side update queue so offline-period messages are delivered. success = await self._connect_adapter_with_timeout(adapter, platform, is_reconnect=True) @@ -891,7 +892,7 @@ class GatewayAdapterLifecycleMixin: """Publish a freshly reconnected primary adapter and replay what it missed while down.""" self._publish_primary_adapter(platform, adapter) self.delivery_router.adapters = self.adapters - del self._failed_platforms[platform] + hand_over_held_inbound(self._failed_platforms.pop(platform).get("predecessor"), adapter) # connect() returning True does not mean the receive path is confirmed -- Telegram's degraded # reconnect returns True so the gateway stays up while its own ladder retries. Stamping "connected" # here would undo the adapter's accurate status. @@ -1398,7 +1399,7 @@ class GatewayAdapterLifecycleMixin: and _platform_binds_port(platform.value, getattr(getattr(adapter, "config", None), "extra", None)): adapter._shared_listener_profile = profile_name - async def _secondary_reconnect_attempt(self, profile_name: str, platform: Platform, inbound_dedup=None): + async def _secondary_reconnect_attempt(self, profile_name: str, platform: Platform, predecessor=None): """One scoped attempt to rebuild+connect a secondary adapter → ``(adapter, success)``; ``(None, None)`` = give up for good (disabled, credential removed, adapter unavailable). Caller tears down a RETURNED adapter; one whose configure/connect raised is torn down here.""" @@ -1430,7 +1431,7 @@ class GatewayAdapterLifecycleMixin: platform.value, profile_name, ) return None, None - carry_inbound_dedup(inbound_dedup, adapter) + carry_inbound_dedup(predecessor, adapter) try: self._configure_profile_adapter(adapter, profile_name, platform) success = await self._connect_adapter_with_timeout(adapter, platform, is_reconnect=True) @@ -1441,7 +1442,7 @@ class GatewayAdapterLifecycleMixin: return adapter, success async def _run_secondary_profile_reconnect( - self, profile_name: str, platform: Platform, inbound_dedup=None + self, profile_name: str, platform: Platform, predecessor=None ) -> None: """Reconnect a retryable secondary adapter under its own profile scope.""" from gateway.run import _profile_runtime_scope, _reconnect_backoff @@ -1454,7 +1455,7 @@ class GatewayAdapterLifecycleMixin: adapter = None try: adapter, success = await self._secondary_reconnect_attempt( - profile_name, platform, inbound_dedup + profile_name, platform, predecessor ) if adapter is None: return @@ -1462,6 +1463,7 @@ class GatewayAdapterLifecycleMixin: profile_map = self._profile_adapters.setdefault(profile_name, {}) if platform not in profile_map: profile_map[platform] = adapter + hand_over_held_inbound(predecessor, adapter) self._sync_voice_mode_state_to_adapter(adapter) logger.info("✓ %s reconnected (profile: %s)", platform.value, profile_name) await self._redeliver_failed_obligations_for_platform( @@ -1576,7 +1578,7 @@ class GatewayAdapterLifecycleMixin: if platform in profile_pending: return profile_pending[platform] = self._retain_background_task(asyncio.create_task( - self._run_secondary_profile_reconnect(profile_name, platform, inbound_dedup_caches(adapter)), + self._run_secondary_profile_reconnect(profile_name, platform, adapter), name=f"secondary-reconnect:{profile_name}:{platform.value}", )) diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 2e3dcf9528..0758bf9e51 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -10,6 +10,7 @@ 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 @@ -818,6 +819,10 @@ class TelegramAdapter(BasePlatformAdapter): 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) + if successor is not None and successor() is not None: # retired by a runner rebuild (#132829) + successor()._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 = [] @@ -837,6 +842,24 @@ class TelegramAdapter(BasePlatformAdapter): 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.""" + if predecessor is self or not isinstance(predecessor, TelegramAdapter): + return + predecessor._held_inbound_successor = weakref.ref(self) + held = getattr(predecessor, "_held_inbound_events", None) or [] + events, held[:] = list(held), [] + 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:]: diff --git a/tests/gateway/test_platform_reconnect.py b/tests/gateway/test_platform_reconnect.py index b194dd55b3..608c02ea3b 100644 --- a/tests/gateway/test_platform_reconnect.py +++ b/tests/gateway/test_platform_reconnect.py @@ -412,6 +412,49 @@ class TestReconnectKeepsInboundDedup: assert new._dedup.is_duplicate("m1") is True assert new._dedup.is_duplicate("m2") is False + @pytest.mark.asyncio + async def test_held_telegram_inbound_reaches_published_replacement_once(self): + """PTB already acked a held update, so the queue on the retired Telegram adapter must be + delivered exactly once by the replacement the watcher publishes, not by a failed candidate (#132829).""" + from gateway.platforms.event import MessageEvent + from gateway.session import SessionSource + from plugins.platforms.telegram.adapter import TelegramAdapter + + class _Telegram(TelegramAdapter): + def __init__(self, succeed): + super().__init__(PlatformConfig(enabled=True, token="123:abc")) + self.succeed, self.handle_message = succeed, AsyncMock() + + async def connect(self, *, is_reconnect=False): + if self.succeed: + self._mark_connected() + return self.succeed + + async def disconnect(self): + return None + + runner = _make_runner() + runner.stop = AsyncMock() + runner._sync_voice_mode_state_to_adapter = MagicMock() + old, failed, new = _Telegram(True), _Telegram(False), _Telegram(True) + runner.adapters[Platform.TELEGRAM] = old + old._set_fatal_error("telegram_network_error", "stall", retryable=True) + event = MessageEvent(text="held", source=SessionSource(platform=Platform.TELEGRAM, chat_id="1", chat_type="dm")) + old._hold_inbound_event(event, where="text-enqueue") + await runner._handle_adapter_fatal_error(old) + for candidate in (failed, new): + runner._failed_platforms[Platform.TELEGRAM]["next_retry"] = 0 + with patch.object(runner, "_create_adapter", return_value=candidate): + await runner._reconnect_failed_platform(Platform.TELEGRAM, time.monotonic() + 1) + await asyncio.sleep(0) + await new._held_inbound_redispatch_task + + assert runner.adapters[Platform.TELEGRAM] is new + failed.handle_message.assert_not_called() + new.handle_message.assert_awaited_once_with(event) + assert event.source._transport_adapter_ref() is new + assert old._held_inbound_events == [] and new._held_inbound_events == [] + # --- Pause / resume circuit breaker --- diff --git a/tests/gateway/test_telegram_text_batching.py b/tests/gateway/test_telegram_text_batching.py index 9e17c52130..2e2a5162c4 100644 --- a/tests/gateway/test_telegram_text_batching.py +++ b/tests/gateway/test_telegram_text_batching.py @@ -195,6 +195,21 @@ class TestHoldInboundAcrossReconnect: assert adapter.handle_message.call_args[0][0].text == "should survive disconnect" assert adapter._held_inbound_events == [] + @pytest.mark.asyncio + async def test_late_teardown_salvage_on_retired_adapter_reaches_replacement(self): + """Teardown of a rebuilt-away adapter can salvage a batch after the replacement drained (#132829).""" + old, new = _make_adapter(), _make_adapter() + old._mark_disconnected() + new.adopt_held_inbound(old) + old._pending_text_batches["k"] = _make_event("late-salvage") + + await old._cancel_pending_delivery_tasks() + await new._held_inbound_redispatch_task + + old.handle_message.assert_not_called() + assert [c.args[0].text for c in new.handle_message.call_args_list] == ["late-salvage"] + assert old._held_inbound_events == [] + @pytest.mark.asyncio async def test_flush_during_disconnect_holds_popped_event(self): """After pop, drop-guard must hold — not destroy — the event.