Dispatch after stop() restarts _run_loop on the shut-down queue and spins a CPU core forever #38

Summary

When an EventBus gets an event after stop(), dispatch() raises QueueShutDown to the caller, but _start() has already started a new _run_loop that spins without ever waiting. The loop also swallows CancelledError, so task.cancel() cannot stop it, and the process won't exit on its own.

In browser-use this happens when anything dispatches on a session's bus after BrowserSession.kill() or stop() has stopped it. For example, on_BrowserStartEvent dispatches BrowserErrorEvent after a CDP setup failure while kill() is running. In our production deployment one such race pinned a worker process at 100% CPU for 15 hours, until the container was replaced. Requests on that worker still completed, but slowly, so nothing failed loudly.

Versions

• bubus 1.5.6 (latest on PyPI); same code on main @ de1519b
• browser-use 0.13.4 in production; 0.13.11 still pins bubus==1.5.6
• Python 3.11, Linux (production) and macOS (repro)

Minimal reproduction (bubus only)

import asyncio
import time

from bubus import BaseEvent, EventBus


class Ping(BaseEvent):
    pass


async def main() -> None:
    bus = EventBus()
    await bus.dispatch(Ping())        # starts the run loop
    await bus.stop(timeout=1)         # shuts the queue down

    try:
        bus.dispatch(Ping())          # a late dispatch, e.g. from a handler racing stop()
    except Exception as e:
        print(f"dispatch raised {type(e).__name__}")

    cpu0, wall0 = time.process_time(), time.monotonic()
    await asyncio.sleep(5)
    cpu = 100 * (time.process_time() - cpu0) / (time.monotonic() - wall0)
    print(f"_is_running={bus._is_running}  CPU over 5s: {cpu:.0f}%")

    bus._runloop_task.cancel()
    await asyncio.sleep(1)
    print(f"after cancel(): runloop done={bus._runloop_task.done()}")


asyncio.run(main())

Output (macOS; on Linux the CPU figure is ~100% of one core):

dispatch raised QueueShutDown
_is_running=True  CPU over 5s: 34%
Task exception was never retrieved
future: <Task finished ... coro=<CleanShutdownQueue.get() ...> exception=QueueShutDown()>
after cancel(): runloop done=False

The script then never exits.

Expected: after stop(), a late dispatch() either raises QueueShutDown and leaves the bus stopped, or restarts the bus cleanly with a new queue. Either way, no run loop should spin, and cancel() should stop the run-loop task.

Cause (links to main @ de1519b)

1. stop() sets _is_running = False and calls event_queue.shutdown(), but keeps the queue object.
2. On the next dispatch(), _start() sees _is_running == False. It creates a queue only if self.event_queue is None, so it keeps the shut-down queue, starts a new _run_loop task and sets _is_running = True. put_nowait then raises QueueShutDown to the caller.
3. _run_loop is written to break on QueueShutDown, but never receives one. _get_next_event catches (asyncio.CancelledError, RuntimeError, QueueShutDown) and returns None.
4. CleanShutdownQueue.get() raises right away on a shut-down queue, so the 0.1 s wait_for_timeout never applies. while self._is_running becomes a tight loop that only yields to the event loop.
5. Because _get_next_event also swallows CancelledError, cancelling _runloop_task doesn't end it.

Suggested fix

Any one of these stops the spin; the first two together seem cleanest:

• In _start(), create a fresh queue (and _on_idle) when the existing one is shut down, not only when it is None. Alternatively, refuse to restart a bus whose queue is shut down.
• In _get_next_event, let QueueShutDown (and CancelledError) propagate so _run_loop's existing except QueueShutDown: break and except asyncio.CancelledError handlers run.
• In browser-use, BrowserSession.kill()/stop() could drop or replace the old bus before awaiting stop(), so in-flight handlers can't dispatch on a stopped bus.

Related

• Bug: run agent with add_new_task，raise Exception "bubus.service.QueueShutDown" browser-use#2614: same QueueShutDown on dispatch to a stopped bus (closed; no CPU spin mentioned).

Workaround we're running

We subclass ResilientEventBus so _start() returns early when event_queue._is_shutdown is set, and make it the session bus (also after kill()/stop(), which install a plain ResilientEventBus). This depends on a private attribute, so we'd like to drop it once bubus handles this.