diff --git a/bubus/service.py b/bubus/service.py index 72f652e..5834e67 100644 --- a/bubus/service.py +++ b/bubus/service.py @@ -1127,9 +1127,23 @@ async def deadlock_monitor(): # This allows us to process child events when the handler awaits them result_value: Any = await asyncio.wait_for(handler_task, timeout=event_result.timeout) elif inspect.isfunction(handler) or inspect.ismethod(handler): - # If handler function is sync function, run it directly in the main thread - # This blocks but ensures we have access to the event loop, dont run it in a subthread! - result_value: Any = handler(event) + # If handler function is sync function, run it in a worker thread + # via anyio.to_thread.run_sync so that the handler timeout can + # actually cancel it. anyio's thread cancellation works by + # cancelling the portal task, unlike run_in_executor which + # cannot cancel the underlying thread. + # + # Previously this ran directly in the event loop thread, which + # blocked the loop entirely and made the timeout useless. + if event_result.timeout is not None: + with anyio.fail_after(event_result.timeout): + result_value: Any = await anyio.to_thread.run_sync( + handler, event, abandon_on_cancel=True + ) + else: + result_value: Any = await anyio.to_thread.run_sync( + handler, event + ) # If the sync handler returned a BaseEvent (from dispatch), DON'T await it # For forwarding handlers like bus.on('*', other_bus.dispatch), the handler diff --git a/tests/test_handler_timeout_zombie_thread.py b/tests/test_handler_timeout_zombie_thread.py new file mode 100644 index 0000000..28c891c --- /dev/null +++ b/tests/test_handler_timeout_zombie_thread.py @@ -0,0 +1,72 @@ +"""Test that sync handler timeout doesn't hang the event bus. + +Option A: sync handlers now run via anyio.to_thread.run_sync with +abandon_on_cancel=True, so the timeout can cancel the await and the bus +stays responsive. The worker thread may still be alive (Python can't kill +threads), but the bus is no longer blocked on it. +""" + +import asyncio +import threading + +import pytest + +from bubus import BaseEvent, EventBus + + +class BlockEvent(BaseEvent[str]): + pass + + +class FastEvent(BaseEvent[str]): + pass + + +@pytest.mark.asyncio +async def test_sync_handler_timeout_does_not_block_subsequent_events(): + """A sync handler that blocks must not prevent subsequent events from + being processed after the timeout fires. + + Before the fix: sync handlers ran directly in the event loop thread, + so a blocking sync handler blocked the entire loop — the timeout never + fired and no other events could be processed. + + After the fix: sync handlers run via anyio.to_thread.run_sync, so the + event loop stays free to enforce the timeout and process other events. + """ + thread_started = threading.Event() + thread_should_stop = threading.Event() + fast_event_handled = threading.Event() + + def blocking_sync_handler(event: BlockEvent) -> str: + thread_started.set() + thread_should_stop.wait(timeout=30) + return "done" + + def fast_sync_handler(event: FastEvent) -> str: + fast_event_handled.set() + return "fast" + + bus = EventBus(name="test_sync_timeout") + bus.on(BlockEvent, blocking_sync_handler) + bus.on(FastEvent, fast_sync_handler) + bus._start() + + # Dispatch the blocking event + bus.dispatch(BlockEvent(event_timeout=1.0)) + try: + await asyncio.wait_for(bus.step(timeout=1.0), timeout=5.0) + except (asyncio.TimeoutError, TimeoutError, Exception): + pass # Expected — the handler timed out + + # Now dispatch a fast event — it must be processed quickly. + # Before the fix, the event loop was blocked by the sync handler + # and this would hang. + bus.dispatch(FastEvent()) + await asyncio.wait_for(bus.step(timeout=1.0), timeout=5.0) + + thread_should_stop.set() + await bus.stop(timeout=1, clear=True) + + assert thread_started.is_set(), "Blocking handler should have started" + assert fast_event_handled.is_set(), "Fast event should have been processed"