From fa2bd5f1c48bb8e99ac2b781e856ee647afeb750 Mon Sep 17 00:00:00 2001 From: Sneh Date: Mon, 27 Jul 2026 05:17:49 -0400 Subject: [PATCH] Fix network_retry_loop blocking shutdown and orphaned tasks during network outages Bug 1: Replace bare asyncio.sleep(cur_interval) with an event-aware asyncio.wait_for(stop_event.wait(), timeout=cur_interval) so that the backoff sleep is interrupted immediately when the stop_event is set. Previously, a bot experiencing a network outage could hang for up to 30 seconds on graceful shutdown while the backoff sleep ran uninterrupted. Bug 2: After calling .cancel() on pending asyncio tasks inside do_action(), await asyncio.gather(*pending, return_exceptions=True) to ensure cancelled tasks are fully cleaned up before returning. Without this, merely requesting cancellation left tasks in a pending state, causing 'Task was destroyed but it is pending!' warnings from the garbage collector when the tasks were eventually collected. --- .../5310.networkloop-stop-event-sleep.toml | 5 + src/telegram/ext/_utils/networkloop.py | 28 ++- tests/ext/_utils/test_networkloop.py | 165 ++++++++++++++++++ 3 files changed, 194 insertions(+), 4 deletions(-) create mode 100644 changes/unreleased/5310.networkloop-stop-event-sleep.toml diff --git a/changes/unreleased/5310.networkloop-stop-event-sleep.toml b/changes/unreleased/5310.networkloop-stop-event-sleep.toml new file mode 100644 index 00000000000..6995f61e89d --- /dev/null +++ b/changes/unreleased/5310.networkloop-stop-event-sleep.toml @@ -0,0 +1,5 @@ +bugfixes = "Fixed ``network_retry_loop`` hanging up to 30 seconds on shutdown during network outages by interrupting the backoff sleep when ``stop_event`` is set. Also fixed orphaned asyncio tasks (causing ``Task was destroyed but it is pending!`` warnings) by properly awaiting cancelled tasks in ``do_action``." +[[pull_requests]] +uid = "5310" +author_uids = ["xsneh"] +closes_threads = [] diff --git a/src/telegram/ext/_utils/networkloop.py b/src/telegram/ext/_utils/networkloop.py index f3696b589aa..925e4847f46 100644 --- a/src/telegram/ext/_utils/networkloop.py +++ b/src/telegram/ext/_utils/networkloop.py @@ -141,9 +141,14 @@ async def do_action() -> None: done, pending = await asyncio.wait( (action_cb_task, stop_task), return_when=asyncio.FIRST_COMPLETED ) - with contextlib.suppress(asyncio.CancelledError): - for task in pending: - task.cancel() + # Cancel pending tasks and await their completion so they are properly cleaned up. + # Merely calling .cancel() schedules cancellation but does not await it, which can + # leave tasks in a pending state and cause "Task was destroyed but it is pending!" + # warnings from the garbage collector. + for task in pending: + task.cancel() + if pending: + await asyncio.gather(*pending, return_exceptions=True) if stop_task in done: _LOGGER.debug("%s Cancelled", log_prefix) @@ -159,6 +164,9 @@ async def do_action() -> None: while effective_is_running(): try: await do_action() + if stop_event and stop_event.is_set(): + _LOGGER.debug("%s Stop event set. Stopping loop.", log_prefix) + break if not repeat_on_success: _LOGGER.debug("%s Action succeeded. Stopping loop.", log_prefix) break @@ -197,4 +205,16 @@ async def do_action() -> None: retries += 1 if cur_interval: - await asyncio.sleep(cur_interval) + # If a stop_event is provided, use it to interrupt the backoff sleep early. + # A bare asyncio.sleep() would block shutdown for the full backoff duration + # (up to 30 s) even after stop_event is set. + if stop_event: + with contextlib.suppress(asyncio.TimeoutError): + await asyncio.wait_for(stop_event.wait(), timeout=cur_interval) + if stop_event.is_set(): + _LOGGER.debug( + "%s Stop event set during backoff sleep. Stopping loop.", log_prefix + ) + break + else: + await asyncio.sleep(cur_interval) diff --git a/tests/ext/_utils/test_networkloop.py b/tests/ext/_utils/test_networkloop.py index ba3f64a3e10..16cfd1ed043 100644 --- a/tests/ext/_utils/test_networkloop.py +++ b/tests/ext/_utils/test_networkloop.py @@ -24,6 +24,10 @@ and the error callback handling, which were added as part of the bug fix in #5030. """ +import asyncio +import logging +import time + import pytest from telegram.error import InvalidToken, RetryAfter, TelegramError, TimedOut @@ -243,3 +247,164 @@ async def action_succeeds_after_few_tries(): ) assert call_count == success_after + + async def test_stop_event_interrupts_backoff_sleep(self, caplog): + """Regression test for the bug where asyncio.sleep(cur_interval) during backoff was not + interrupted by stop_event, causing shutdowns to hang for up to 30 seconds. + + Verifies that setting stop_event while the loop is sleeping between retries causes the + loop to exit promptly instead of waiting for the full backoff duration. + """ + caplog.set_level(logging.DEBUG) + stop_event = asyncio.Event() + + # Use an interval that is long enough to be detectable but short enough to keep the + # test suite fast. If the fix regresses, the loop would sleep this full duration on + # every retry; the wall-clock assertion below would then fail reliably. + INTERVAL = 5.0 + error_handled = asyncio.Event() + + async def failing_action(): + raise TelegramError("Simulated network outage") + + def on_err(exc): + error_handled.set() + + # Deterministically wait until the network_retry_loop has processed the error + # and is about to enter its backoff sleep. + async def trigger_stop(): + await error_handled.wait() + # Yield to the event loop once more to ensure network_retry_loop has + # entered `await asyncio.wait_for(...)` inside the backoff block. + await asyncio.sleep(0) + stop_event.set() + + trigger_task = asyncio.create_task(trigger_stop()) + t_start = time.perf_counter() + await network_retry_loop( + action_cb=failing_action, + on_err_cb=on_err, + description="test-backoff-interrupt", + interval=INTERVAL, + stop_event=stop_event, + max_retries=-1, + repeat_on_success=True, + ) + elapsed = time.perf_counter() - t_start + trigger_task.cancel() + await asyncio.gather(trigger_task, return_exceptions=True) + + # The loop must exit well before the full INTERVAL; 1 s is a generous threshold. + assert elapsed < 1.0, ( + f"Loop took {elapsed:.2f}s to exit after stop_event was set — " + "stop_event is not interrupting the backoff sleep." + ) + # Assert the specific log message to prove the loop broke from inside the backoff wait, + # preventing a race condition where the stop_event might trigger an earlier exit check. + assert "Stop event set during backoff sleep. Stopping loop." in caplog.text + + async def test_stop_event_breaks_repeat_on_success_loop(self): + """Regression test for the bug where network_retry_loop would loop infinitely if + repeat_on_success=True and stop_event was set during do_action(). + + Verifies that when stop_event is set mid-action, the outer loop breaks cleanly + and does not re-enter do_action() despite repeat_on_success being True. + """ + stop_event = asyncio.Event() + action_started = asyncio.Event() + call_count = 0 + + async def action(): + nonlocal call_count + call_count += 1 + action_started.set() + # Wait for cancellation by the stop_event + await asyncio.sleep(60) + + async def trigger_stop(): + await action_started.wait() + stop_event.set() + + trigger_task = asyncio.create_task(trigger_stop()) + await network_retry_loop( + action_cb=action, + description="test-break-repeat-loop", + interval=0, + stop_event=stop_event, + max_retries=-1, + repeat_on_success=True, + ) + trigger_task.cancel() + await asyncio.gather(trigger_task, return_exceptions=True) + + # If the bug were present, the loop would restart and call action() again. + assert call_count == 1, f"Action was called {call_count} times, expected exactly 1." + + async def test_stop_event_breaks_repeat_on_success_after_successful_action(self): + """Verifies that if an action completes successfully but stop_event is set, + the loop breaks cleanly and does not repeat despite repeat_on_success=True. + """ + stop_event = asyncio.Event() + call_count = 0 + + async def action(): + nonlocal call_count + call_count += 1 + # Action completes successfully. + # We set stop_event from within the action to simulate it being set + # concurrently just before the action finishes. + stop_event.set() + + await network_retry_loop( + action_cb=action, + description="test-break-repeat-after-success", + interval=0, + stop_event=stop_event, + max_retries=-1, + repeat_on_success=True, + ) + + assert call_count == 1, f"Action was called {call_count} times, expected exactly 1." + + async def test_no_pending_tasks_after_stop_event(self): + """Regression test for the bug where pending tasks were only .cancel()ed but never + awaited inside do_action(), leaving them in a pending state and producing + 'Task was destroyed but it is pending!' warnings from the garbage collector. + + Verifies that after stop_event fires mid-action, all asyncio tasks created by the loop + are fully completed (not merely cancelled) before network_retry_loop returns. + """ + stop_event = asyncio.Event() + action_started = asyncio.Event() + + async def slow_action(): + action_started.set() + # Simulate a slow in-flight HTTP call; will be cancelled via stop_event. + await asyncio.sleep(5) + + tasks_before = set(asyncio.all_tasks()) + + # Concurrently set stop_event the moment the slow action begins. + async def trigger_stop(): + await action_started.wait() + stop_event.set() + + trigger_task = asyncio.create_task(trigger_stop()) + await network_retry_loop( + action_cb=slow_action, + description="test-no-pending-tasks", + interval=0, + stop_event=stop_event, + max_retries=-1, + repeat_on_success=True, + ) + trigger_task.cancel() + await asyncio.gather(trigger_task, return_exceptions=True) + + tasks_after = asyncio.all_tasks() - tasks_before + pending = {t for t in tasks_after if not t.done()} + + assert not pending, ( + f"{len(pending)} task(s) remain pending after network_retry_loop returned with " + "stop_event set — cancelled tasks are not being properly awaited." + )