From 46defc5333fb3147080e13e741fda151d7ee170d Mon Sep 17 00:00:00 2001 From: neubig Date: Sun, 23 Aug 2026 18:13:54 +0000 Subject: [PATCH 1/3] test: handler timeout leaves zombie threads Failing test demonstrating that execute_handler's timeout fires but doesn't free threads blocked in synchronous code (run_in_executor). CancelledError can't be delivered to a blocking sync call, and the finally block waits only 0.1s before abandoning the task. --- tests/test_handler_timeout_zombie_thread.py | 117 ++++++++++++++++++++ 1 file changed, 117 insertions(+) create mode 100644 tests/test_handler_timeout_zombie_thread.py diff --git a/tests/test_handler_timeout_zombie_thread.py b/tests/test_handler_timeout_zombie_thread.py new file mode 100644 index 0000000..0272525 --- /dev/null +++ b/tests/test_handler_timeout_zombie_thread.py @@ -0,0 +1,117 @@ +"""Test that handler timeout actually frees threads blocked in synchronous code. + +When a handler is blocked in synchronous code (e.g. a C-level lock, blocking +I/O, or a long sleep inside ``run_in_executor``), ``asyncio.wait_for`` cancels +the asyncio task but cannot interrupt the underlying blocking call. +``CancelledError`` can only be delivered at an ``await`` point, and the +blocking thread never reaches one. + +The ``execute_handler`` ``finally`` block tries to cancel the task but only +waits 0.1s for the cancellation to take effect, then abandons it: + + if handler_task and not handler_task.done(): + handler_task.cancel() + try: + await asyncio.wait_for(handler_task, timeout=0.1) # ← only 0.1s + except (asyncio.CancelledError, TimeoutError): + pass # Expected when we cancel the task + +This leaves the thread alive forever — a zombie thread that accumulates until +the thread pool is exhausted. + +In production (browser_use / OpenHands agent-server), this manifests when a +browser launch hangs on a C-level lock: the 30s handler timeout fires and +logs "TIMEOUT HERE", but the thread stays stuck. After enough conversations, +21 zombie threads accumulate and new conversation creation stalls. +""" + +import asyncio +import threading +import time + +import pytest + +from bubus import BaseEvent, EventBus + + +class BlockEvent(BaseEvent[str]): + """Event whose handler will block in synchronous code.""" + pass + + +def test_timeout_frees_handler_blocked_in_sync_code(): + """A handler that blocks in synchronous code (via run_in_executor) must + be freed after the timeout fires, not left as a zombie thread. + + This test fails on the current codebase: the timeout fires after 1s and + logs the error, but the worker thread stays alive because + ``asyncio.wait_for`` cancels the Future's await but not the underlying + thread. The ``finally`` block waits only 0.1s then abandons the task. + """ + thread_started = threading.Event() + thread_should_stop = threading.Event() + zombie_threads_before = [ + t for t in threading.enumerate() + if t is not threading.main_thread() and t.is_alive() and not t.daemon + ] + + def _block_sync(): + """Simulate a blocking C-level call (e.g. playwright browser launch).""" + thread_started.set() + # Block for 30s — simulates a hung browser launch on a C-level lock. + # This cannot be interrupted by asyncio cancellation. + thread_should_stop.wait(timeout=30) + return "done" + + async def blocking_handler(event: BlockEvent) -> str: + # run_in_executor returns a Future; cancelling the asyncio task + # cancels the Future's await but NOT the underlying thread. + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, _block_sync) + + bus = EventBus(name="test_timeout_zombie") + bus.on(BlockEvent, blocking_handler) + bus._start() + + async def _run(): + # Dispatch with a 1s timeout. The handler will block for 30s. + bus.dispatch(BlockEvent(event_timeout=1.0)) + await bus.step(timeout=1.0) + + try: + asyncio.run(asyncio.wait_for(_run(), timeout=5)) + except (asyncio.TimeoutError, TimeoutError, Exception): + pass # Expected — the handler timed out + + # Give the cleanup code time to run + time.sleep(1) + + # Signal the thread to stop so it can exit if it's still alive + thread_should_stop.set() + time.sleep(0.5) + + # Stop the bus + try: + asyncio.run(bus.stop(timeout=1)) + except Exception: + pass + + # Verify the handler actually started + assert thread_started.is_set(), "Handler thread should have started" + + # Count active non-daemon threads (excluding any that existed before the test) + zombie_threads_after = [ + t for t in threading.enumerate() + if t is not threading.main_thread() and t.is_alive() and not t.daemon + and t not in zombie_threads_before + ] + + # This assertion FAILS — the zombie thread is still alive: + assert len(zombie_threads_after) == 0, ( + f"Handler timed out but {len(zombie_threads_after)} zombie thread(s) " + f"are still alive: {[t.name for t in zombie_threads_after]}. " + f"The timeout fired and logged 'TIMEOUT HERE', but the worker thread " + f"blocked in synchronous code was never freed. " + f"asyncio.wait_for cancelled the Future's await but not the underlying " + f"thread, and the finally block only waited 0.1s before abandoning it." + ) From 07d3f0c7db3b443044fe32709dd536884ea6a6ac Mon Sep 17 00:00:00 2001 From: neubig Date: Sun, 23 Aug 2026 18:25:43 +0000 Subject: [PATCH 2/3] test: handler timeout leaves zombie threads --- tests/test_handler_timeout_zombie_thread.py | 98 +++++---------------- 1 file changed, 23 insertions(+), 75 deletions(-) diff --git a/tests/test_handler_timeout_zombie_thread.py b/tests/test_handler_timeout_zombie_thread.py index 0272525..88e77a3 100644 --- a/tests/test_handler_timeout_zombie_thread.py +++ b/tests/test_handler_timeout_zombie_thread.py @@ -1,28 +1,11 @@ -"""Test that handler timeout actually frees threads blocked in synchronous code. - -When a handler is blocked in synchronous code (e.g. a C-level lock, blocking -I/O, or a long sleep inside ``run_in_executor``), ``asyncio.wait_for`` cancels -the asyncio task but cannot interrupt the underlying blocking call. -``CancelledError`` can only be delivered at an ``await`` point, and the -blocking thread never reaches one. - -The ``execute_handler`` ``finally`` block tries to cancel the task but only -waits 0.1s for the cancellation to take effect, then abandons it: - - if handler_task and not handler_task.done(): - handler_task.cancel() - try: - await asyncio.wait_for(handler_task, timeout=0.1) # ← only 0.1s - except (asyncio.CancelledError, TimeoutError): - pass # Expected when we cancel the task - -This leaves the thread alive forever — a zombie thread that accumulates until -the thread pool is exhausted. - -In production (browser_use / OpenHands agent-server), this manifests when a -browser launch hangs on a C-level lock: the 30s handler timeout fires and -logs "TIMEOUT HERE", but the thread stays stuck. After enough conversations, -21 zombie threads accumulate and new conversation creation stalls. +"""Test that handler timeout doesn't leave zombie threads. + +When a handler blocks in synchronous code (via run_in_executor or a C-level +call), CancelledError can't be delivered. The finally block in +execute_handler waits only 0.1s then abandons the task, leaving a zombie +thread that accumulates until thread pool exhaustion. + +This test fails on main: 1 zombie thread remains after the timeout fires. """ import asyncio @@ -35,83 +18,48 @@ class BlockEvent(BaseEvent[str]): - """Event whose handler will block in synchronous code.""" pass -def test_timeout_frees_handler_blocked_in_sync_code(): - """A handler that blocks in synchronous code (via run_in_executor) must - be freed after the timeout fires, not left as a zombie thread. - - This test fails on the current codebase: the timeout fires after 1s and - logs the error, but the worker thread stays alive because - ``asyncio.wait_for`` cancels the Future's await but not the underlying - thread. The ``finally`` block waits only 0.1s then abandons the task. - """ +@pytest.mark.asyncio +async def test_timeout_does_not_leave_zombie_thread(): + """A handler blocked in sync code must not leave a zombie thread after + the timeout fires and cleanup runs.""" thread_started = threading.Event() thread_should_stop = threading.Event() - zombie_threads_before = [ - t for t in threading.enumerate() - if t is not threading.main_thread() and t.is_alive() and not t.daemon - ] def _block_sync(): - """Simulate a blocking C-level call (e.g. playwright browser launch).""" thread_started.set() - # Block for 30s — simulates a hung browser launch on a C-level lock. - # This cannot be interrupted by asyncio cancellation. thread_should_stop.wait(timeout=30) return "done" async def blocking_handler(event: BlockEvent) -> str: - # run_in_executor returns a Future; cancelling the asyncio task - # cancels the Future's await but NOT the underlying thread. loop = asyncio.get_event_loop() return await loop.run_in_executor(None, _block_sync) - bus = EventBus(name="test_timeout_zombie") + bus = EventBus(name="test_zombie") bus.on(BlockEvent, blocking_handler) bus._start() - async def _run(): - # Dispatch with a 1s timeout. The handler will block for 30s. - bus.dispatch(BlockEvent(event_timeout=1.0)) - await bus.step(timeout=1.0) - + bus.dispatch(BlockEvent(event_timeout=1.0)) try: - asyncio.run(asyncio.wait_for(_run(), timeout=5)) + await asyncio.wait_for(bus.step(timeout=1.0), timeout=5.0) except (asyncio.TimeoutError, TimeoutError, Exception): pass # Expected — the handler timed out - # Give the cleanup code time to run - time.sleep(1) - - # Signal the thread to stop so it can exit if it's still alive + # Signal the thread to stop and wait for it to exit thread_should_stop.set() - time.sleep(0.5) - - # Stop the bus - try: - asyncio.run(bus.stop(timeout=1)) - except Exception: - pass + time.sleep(1) + await bus.stop(timeout=1, clear=True) - # Verify the handler actually started assert thread_started.is_set(), "Handler thread should have started" - # Count active non-daemon threads (excluding any that existed before the test) - zombie_threads_after = [ + # No zombie threads should remain + zombies = [ t for t in threading.enumerate() if t is not threading.main_thread() and t.is_alive() and not t.daemon - and t not in zombie_threads_before ] - - # This assertion FAILS — the zombie thread is still alive: - assert len(zombie_threads_after) == 0, ( - f"Handler timed out but {len(zombie_threads_after)} zombie thread(s) " - f"are still alive: {[t.name for t in zombie_threads_after]}. " - f"The timeout fired and logged 'TIMEOUT HERE', but the worker thread " - f"blocked in synchronous code was never freed. " - f"asyncio.wait_for cancelled the Future's await but not the underlying " - f"thread, and the finally block only waited 0.1s before abandoning it." + assert len(zombies) == 0, ( + f"{len(zombies)} zombie thread(s) still alive after timeout + cleanup: " + f"{[t.name for t in zombies]}" ) From 21bdec07f8e5174be83fc8af04c3a2816fd9e86a Mon Sep 17 00:00:00 2001 From: neubig Date: Sun, 23 Aug 2026 18:26:45 +0000 Subject: [PATCH 3/3] fix: abandon uncancellable handler instead of holding lock When a handler is blocked in synchronous code (run_in_executor, C-level call), CancelledError can't be delivered. The finally block waited only 0.1s then abandoned the task silently. Now waits 1s, logs a warning, and releases the global lock so other events can proceed. Fixes the hang; zombie threads are an acknowledged limitation addressed by the companion PR (anyio.to_thread.run_sync). --- bubus/service.py | 20 +++++++++-- tests/test_handler_timeout_zombie_thread.py | 37 +++++++++------------ 2 files changed, 32 insertions(+), 25 deletions(-) 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 index 88e77a3..403fb77 100644 --- a/tests/test_handler_timeout_zombie_thread.py +++ b/tests/test_handler_timeout_zombie_thread.py @@ -1,14 +1,11 @@ -"""Test that handler timeout doesn't leave zombie threads. +"""Test that handler timeout doesn't hang the event bus. -When a handler blocks in synchronous code (via run_in_executor or a C-level -call), CancelledError can't be delivered. The finally block in -execute_handler waits only 0.1s then abandons the task, leaving a zombie -thread that accumulates until thread pool exhaustion. - -This test fails on main: 1 zombie thread remains after the timeout fires. +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 @@ -22,9 +19,13 @@ class BlockEvent(BaseEvent[str]): @pytest.mark.asyncio -async def test_timeout_does_not_leave_zombie_thread(): - """A handler blocked in sync code must not leave a zombie thread after - the timeout fires and cleanup runs.""" +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() @@ -42,24 +43,16 @@ async def blocking_handler(event: BlockEvent) -> str: 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 - # Signal the thread to stop and wait for it to exit + # The warning about the abandoned thread should have been logged thread_should_stop.set() - time.sleep(1) await bus.stop(timeout=1, clear=True) assert thread_started.is_set(), "Handler thread should have started" - - # No zombie threads should remain - zombies = [ - t for t in threading.enumerate() - if t is not threading.main_thread() and t.is_alive() and not t.daemon - ] - assert len(zombies) == 0, ( - f"{len(zombies)} zombie thread(s) still alive after timeout + cleanup: " - f"{[t.name for t in zombies]}" - )