Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 17 additions & 3 deletions bubus/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Comment thread
neubig marked this conversation as resolved.
handler, event, abandon_on_cancel=True
Comment thread
neubig marked this conversation as resolved.
)
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
Expand Down
72 changes: 72 additions & 0 deletions tests/test_handler_timeout_zombie_thread.py
Original file line number Diff line number Diff line change
@@ -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)
Comment thread
neubig marked this conversation as resolved.
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):
Comment thread
neubig marked this conversation as resolved.
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()
Comment thread
neubig marked this conversation as resolved.
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"