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 @@ -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)
Comment thread
neubig marked this conversation as resolved.
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:
Expand Down
58 changes: 58 additions & 0 deletions tests/test_handler_timeout_zombie_thread.py
Original file line number Diff line number Diff line change
@@ -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
Comment thread
neubig marked this conversation as resolved.

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
Comment thread
neubig marked this conversation as resolved.

# The warning about the abandoned thread should have been logged
Comment thread
neubig marked this conversation as resolved.
thread_should_stop.set()
await bus.stop(timeout=1, clear=True)

assert thread_started.is_set(), "Handler thread should have started"