From e2501d72efe857bef6eeadafe4cd497d71b7364a Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Sat, 3 Oct 2026 15:41:41 +0530 Subject: [PATCH] refactor(telegram): raise plain _PollingStallError for ingress dispatch stall Gate finding (quality, Low): _IngressDispatchStallError had no consumer - every branch checks isinstance(error, _PollingStallError) - so the subclass and its docstring were dead weight. One stall was also announced three times (the "healthy but deaf" pre-log, the handoff warning, the fatal error line); the sibling _check_polling_stall deliberately skips a pre-log. Drop the subclass, drop the pre-log and carry the "healthy but deaf" counts in the error text instead, and read the polling generation once. Tests assert on the error handed to the stubbed handoff rather than caplog. --- plugins/platforms/telegram/adapter.py | 31 +++++-------------- .../test_telegram_ingress_delivery_gap.py | 20 +++++++----- 2 files changed, 21 insertions(+), 30 deletions(-) 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