diff --git a/bubus/service.py b/bubus/service.py index 72f652e..047cc12 100644 --- a/bubus/service.py +++ b/bubus/service.py @@ -1211,13 +1211,27 @@ async def deadlock_monitor(): inside_handler_context.reset(handler_token) _current_handler_id_context.reset(handler_id_token) - # Ensure handler task is cancelled if it's still running + # Ensure handler task is cancelled if it's still running. + # If the handler is blocked in synchronous code (e.g. via + # run_in_executor or a C-level call), CancelledError cannot be + # delivered and the task won't respond to cancel(). We try once + # with a short grace period, then abandon the task so we don't + # hold the global lock indefinitely. The zombie thread is logged + # so it can be diagnosed; it will be cleaned up when the process + # exits (it's a daemon thread under the default executor). if handler_task and not handler_task.done(): handler_task.cancel() try: - await asyncio.wait_for(handler_task, timeout=0.1) + await asyncio.wait_for(handler_task, timeout=1.0) except (asyncio.CancelledError, TimeoutError): - pass # Expected when we cancel the task + if not handler_task.done(): + logger.warning( + f'⚠️ {self} handler {get_handler_name(handler)}()' + f' could not be cancelled (still running after 1s).' + f' It is likely blocked in synchronous code and will' + f' be abandoned as a zombie thread.' + ) + pass # Abandon the task — can't cancel blocking sync code # Ensure monitor task is cancelled try: diff --git a/tests/test_handler_timeout_zombie_thread.py b/tests/test_handler_timeout_zombie_thread.py new file mode 100644 index 0000000..403fb77 --- /dev/null +++ b/tests/test_handler_timeout_zombie_thread.py @@ -0,0 +1,58 @@ +"""Test that handler timeout doesn't hang the event bus. + +Option C: when a handler can't be cancelled (blocked in sync code), abandon +it with a warning instead of holding the global lock indefinitely. +""" + +import asyncio +import logging +import threading +import time + +import pytest + +from bubus import BaseEvent, EventBus + + +class BlockEvent(BaseEvent[str]): + pass + + +@pytest.mark.asyncio +async def test_timeout_does_not_hang_on_blocking_handler(caplog): + """A handler blocked in sync code must not hang the event bus. + + The timeout fires after 1s. The fix: the finally block abandons the + task after a short grace period and logs a warning, releasing the global + lock so other events can proceed. + """ + thread_started = threading.Event() + thread_should_stop = threading.Event() + + def _block_sync(): + thread_started.set() + thread_should_stop.wait(timeout=30) + return "done" + + async def blocking_handler(event: BlockEvent) -> str: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, _block_sync) + + bus = EventBus(name="test_zombie") + bus.on(BlockEvent, blocking_handler) + bus._start() + + bus.dispatch(BlockEvent(event_timeout=1.0)) + + # step() must return within 5s — without the fix it hangs for 30s + # because the global lock is held waiting for the uncancellable task. + try: + await asyncio.wait_for(bus.step(timeout=1.0), timeout=5.0) + except (asyncio.TimeoutError, TimeoutError, Exception): + pass # Expected — the handler timed out + + # The warning about the abandoned thread should have been logged + thread_should_stop.set() + await bus.stop(timeout=1, clear=True) + + assert thread_started.is_set(), "Handler thread should have started"