diff --git a/tests/gateway/test_cron_shutdown_drain.py b/tests/gateway/test_cron_shutdown_drain.py new file mode 100644 index 00000000000..8029c6e70c2 --- /dev/null +++ b/tests/gateway/test_cron_shutdown_drain.py @@ -0,0 +1,73 @@ +"""Regression tests for #58818. + +On restart the gateway must drain an in-flight cron delivery instead of +dropping it. A cron delivery is a coroutine scheduled onto the gateway event +loop (``safe_schedule_threadsafe``) while the ticker thread blocks on its +future. The shutdown wait therefore must NOT block the loop with a synchronous +``thread.join()`` — doing so deadlocks the delivery (the loop can never run it) +and the message is silently lost. ``_await_thread_exit`` waits cooperatively so +the pending delivery completes first. +""" +import asyncio +import threading + +import pytest + +import gateway.run as gateway_run + + +@pytest.mark.asyncio +async def test_await_thread_exit_lets_loop_scheduled_delivery_complete(): + # Reproduces the drop: the worker schedules a coroutine onto THIS loop and + # blocks on its result, exactly like cron/_deliver_result. A blocking join + # would deadlock it; the cooperative wait lets it finish. + loop = asyncio.get_running_loop() + delivered = threading.Event() + worker_done = threading.Event() + + async def _delivery(): + await asyncio.sleep(0.05) + delivered.set() + return "ok" + + def _cron_worker(): + fut = asyncio.run_coroutine_threadsafe(_delivery(), loop) + fut.result(timeout=10) + worker_done.set() + + thread = threading.Thread(target=_cron_worker, daemon=True) + thread.start() + + exited = await gateway_run._await_thread_exit(thread, timeout=10) + + assert exited is True + assert delivered.is_set(), "in-flight delivery coroutine never ran (loop was blocked)" + assert worker_done.is_set() + + +@pytest.mark.asyncio +async def test_await_thread_exit_returns_false_on_timeout(): + keep_alive = threading.Event() + + def _spin(): + keep_alive.wait(5) + + thread = threading.Thread(target=_spin, daemon=True) + thread.start() + try: + exited = await gateway_run._await_thread_exit(thread, timeout=0.2, poll=0.02) + assert exited is False + assert thread.is_alive() + finally: + keep_alive.set() + thread.join(timeout=2) + + +@pytest.mark.asyncio +async def test_await_thread_exit_handles_none_and_dead_thread(): + assert await gateway_run._await_thread_exit(None, timeout=1) is True + + thread = threading.Thread(target=lambda: None, daemon=True) + thread.start() + thread.join(timeout=2) + assert await gateway_run._await_thread_exit(thread, timeout=1) is True