diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 0f425cc33b..c6fa1a4b1b 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -509,15 +509,6 @@ class _PollingStallError(RuntimeError): """ -class _IngressDispatchStallError(_PollingStallError): - """A confirmed PTB dispatcher stall: getUpdates fetches but no handler dispatch (#102260, #130407). - - Subclasses ``_PollingStallError`` so the recovery ladder hands the adapter to the supervisor for a - rebuild instead of restarting the same Updater in place: an in-place polling restart keeps the same - ``Application`` (and its wedged dispatcher task), while a rebuild constructs a fresh dispatcher. - """ - - class TelegramAdapter(BasePlatformAdapter): """Telegram bot adapter: users/groups, MarkdownV2 replies, forum topics, media.""" @@ -2365,8 +2356,8 @@ class TelegramAdapter(BasePlatformAdapter): ``received`` and ``dispatched`` count the same population (every fetched update reaches the group-99 catch-all: no handler raises ApplicationHandlerStop, no error handler is registered), so a backlog with no dispatch progress across ``_INGRESS_DISPATCH_STALL_HEARTBEATS`` heartbeats - is a wedged dispatcher at any traffic rate. Warns once per stall, marks the adapter degraded, - and hands it to the supervisor for a rebuild; re-arms on dispatch progress. + is a wedged dispatcher at any traffic rate. Hands the adapter to the supervisor for a rebuild once + per stall (``_PollingStallError``: an in-place restart keeps the wedged dispatcher); re-arms on progress. """ if self._webhook_mode or self._teardown_started or self.has_fatal_error or self._recovery_in_flight(): return @@ -2382,19 +2373,13 @@ class TelegramAdapter(BasePlatformAdapter): self._ingress_stalled_heartbeats = stalled + 1 if stalled + 1 < _INGRESS_DISPATCH_STALL_HEARTBEATS: return - backlog = received - dispatched - logger.warning( - "[%s] Telegram ingress is healthy but deaf: %d update(s) fetched by getUpdates have not been " - "dispatched to any handler across %d heartbeats (%d received, %d dispatched, generation %d). " - "Polling is fine; PTB's dispatcher is not draining its queue.", - self.name, backlog, _INGRESS_DISPATCH_STALL_HEARTBEATS, received, dispatched, - getattr(self, "_polling_generation", 0)) + # No pre-log: the handoff warning and ``_go_fatal_network`` carry this text (see _check_polling_stall). + generation = getattr(self, "_polling_generation", 0) self._schedule_polling_recovery( - _IngressDispatchStallError( - "PTB dispatcher made no progress for %d heartbeats " - "(%d received, %d dispatched, generation %d; ingress dispatch stall watchdog)" - % (_INGRESS_DISPATCH_STALL_HEARTBEATS, received, dispatched, - getattr(self, "_polling_generation", 0))), + _PollingStallError( + "ingress healthy but deaf: PTB dispatcher made no progress for %d heartbeats with %d update(s) " + "fetched but not dispatched (%d received, %d dispatched, generation %d)" + % (_INGRESS_DISPATCH_STALL_HEARTBEATS, received - dispatched, received, dispatched, generation)), reason="ingress dispatch stall watchdog") async def _check_polling_stall(self) -> None: diff --git a/tests/gateway/test_telegram_ingress_delivery_gap.py b/tests/gateway/test_telegram_ingress_delivery_gap.py index 1f76d10d1d..c3687d18d5 100644 --- a/tests/gateway/test_telegram_ingress_delivery_gap.py +++ b/tests/gateway/test_telegram_ingress_delivery_gap.py @@ -49,6 +49,13 @@ def _deaf_reports(caplog) -> list[str]: return [r.getMessage() for r in caplog.records if _DEAF in r.message] +def _stall_reports(sched) -> list[str]: + """Errors handed to the (stubbed) recovery handoff: the stall's only announcement.""" + errors = [call.args[0] for call in sched.call_args_list] + assert all(type(e) is tg_adapter._PollingStallError for e in errors) + return [str(e) for e in errors if _DEAF in str(e)] + + @pytest.mark.asyncio async def test_stall_reported_once_on_backlog_regardless_of_update_age(caplog): """Healthy dispatch never reports; a wedged dispatcher is reported after four heartbeats even @@ -60,11 +67,11 @@ async def test_stall_reported_once_on_backlog_regardless_of_update_age(caplog): _heartbeats(adapter, 3) assert _deaf_reports(caplog) == [] - with patch.object(adapter, "_schedule_polling_recovery"): + with patch.object(adapter, "_schedule_polling_recovery") as sched: for _ in range(5): # a fresh update lands before every heartbeat, none dispatched _receive(adapter, 1) adapter._check_ingress_dispatch_stall() - (report,) = _deaf_reports(caplog) + (report,) = _stall_reports(sched) assert "4 update(s) fetched" in report and "6 received, 2 dispatched" in report @@ -77,15 +84,14 @@ async def test_dispatch_progress_rearms_the_report(caplog): _heartbeats(adapter, 3) # 270s: a slow-but-bounded (<=300s) sequential handler is not a wedge assert sched.call_count == 0 _heartbeats(adapter, 2) - assert len(_deaf_reports(caplog)) == 1 and sched.call_count == 1 + assert len(_stall_reports(sched)) == 1 await _dispatch(adapter, 1) # partial drain: progress, backlog remains adapter._check_ingress_dispatch_stall() - assert len(_deaf_reports(caplog)) == 1 and sched.call_count == 1 + assert len(_stall_reports(sched)) == 1 _heartbeats(adapter, 4) - assert len(_deaf_reports(caplog)) == 2 and sched.call_count == 2 - for call in sched.call_args_list: - assert isinstance(call.args[0], tg_adapter._IngressDispatchStallError) + assert len(_stall_reports(sched)) == 2 + assert _deaf_reports(caplog) == [] # stubbed handoff: no separate pre-log announces the stall @pytest.mark.asyncio