diff --git a/gateway/platforms/telegram.py b/gateway/platforms/telegram.py index d2b425b52b9f3..24d1737a04114 100644 --- a/gateway/platforms/telegram.py +++ b/gateway/platforms/telegram.py @@ -1120,7 +1120,7 @@ async def _handle_polling_conflict(self, error: Exception) -> None: try: await self._app.updater.start_polling( allowed_updates=Update.ALL_TYPES, - drop_pending_updates=False, + drop_pending_updates=True, error_callback=self._polling_error_callback_ref, ) logger.info( @@ -1128,6 +1128,13 @@ async def _handle_polling_conflict(self, error: Exception) -> None: self.name, self._polling_conflict_count, MAX_CONFLICT_RETRIES, ) self._polling_conflict_count = 0 # reset counter on success + # Schedule a verification probe (same pattern as network error + # recovery) to detect a wedged updater that reports running=True + # but has a dead consumer task — regression guard for #40691. + if not self.has_fatal_error: + probe = asyncio.ensure_future(self._verify_polling_after_reconnect()) + self._background_tasks.add(probe) + probe.add_done_callback(self._background_tasks.discard) return except Exception as retry_err: logger.warning( diff --git a/tests/gateway/test_telegram_network_reconnect.py b/tests/gateway/test_telegram_network_reconnect.py index 81b7bed12e43e..5c7402a4a4859 100644 --- a/tests/gateway/test_telegram_network_reconnect.py +++ b/tests/gateway/test_telegram_network_reconnect.py @@ -141,13 +141,7 @@ async def test_reconnect_success_resets_error_count(): assert adapter._polling_network_error_count == 0 # Clean up the heartbeat-probe task scheduled after a successful reconnect. - pending = [t for t in adapter._background_tasks if not t.done()] - for t in pending: - t.cancel() - try: - await t - except (asyncio.CancelledError, Exception): - pass + await _cancel_pending_background_tasks(adapter) @pytest.mark.asyncio @@ -196,6 +190,15 @@ def _make_mock_app(): return mock_app, mock_polling_req +async def _cancel_pending_background_tasks(adapter: TelegramAdapter) -> None: + for task in [task for task in adapter._background_tasks if not task.done()]: + task.cancel() + try: + await task + except (asyncio.CancelledError, Exception): + pass + + @pytest.mark.asyncio async def test_reconnect_drains_polling_request_only(): """During reconnect, only the polling request (_request[0]) must be cycled. @@ -286,6 +289,60 @@ async def test_conflict_retry_also_drains_polling_connections(): mock_polling_req.initialize.assert_called_once() mock_app.updater.start_polling.assert_called_once() + # Clean up the heartbeat-probe task scheduled after a successful retry. + await _cancel_pending_background_tasks(adapter) + + +@pytest.mark.asyncio +async def test_conflict_retry_drops_pending_updates_and_preserves_callback(): + """Conflict retries must clear stale updates without losing the error callback.""" + adapter = _make_adapter() + callback = lambda error: None + adapter._polling_error_callback_ref = callback + + mock_app, _ = _make_mock_app() + adapter._app = mock_app + + with patch("asyncio.sleep", new_callable=AsyncMock): + await adapter._handle_polling_conflict( + Exception("Conflict: terminated by other getUpdates request") + ) + + mock_app.updater.start_polling.assert_awaited_once() + kwargs = mock_app.updater.start_polling.await_args.kwargs + assert kwargs["drop_pending_updates"] is True + assert kwargs["error_callback"] is callback + + # Clean up the heartbeat-probe task scheduled after a successful retry. + await _cancel_pending_background_tasks(adapter) + + +@pytest.mark.asyncio +async def test_conflict_retry_schedules_heartbeat_probe_on_success(): + """Successful conflict recovery must verify that polling is truly live.""" + adapter = _make_adapter() + adapter._polling_error_callback_ref = lambda error: None + + mock_app, _ = _make_mock_app() + adapter._app = mock_app + + async def never_finishes_probe(): + await asyncio.Event().wait() + + adapter._verify_polling_after_reconnect = never_finishes_probe + initial_count = len(adapter._background_tasks) + + with patch("asyncio.sleep", new_callable=AsyncMock): + await adapter._handle_polling_conflict( + Exception("Conflict: terminated by other getUpdates request") + ) + + assert len(adapter._background_tasks) > initial_count, ( + "Expected a heartbeat probe task after successful conflict retry" + ) + + await _cancel_pending_background_tasks(adapter) + @pytest.mark.asyncio async def test_drain_helper_noop_without_app(): @@ -466,10 +523,4 @@ async def test_reconnect_schedules_heartbeat_probe_on_success(): ) # Clean up. - pending = [t for t in adapter._background_tasks if not t.done()] - for t in pending: - t.cancel() - try: - await t - except (asyncio.CancelledError, Exception): - pass + await _cancel_pending_background_tasks(adapter)