From 77787cb7ca5cec532fceb250db77c78dc21d0636 Mon Sep 17 00:00:00 2001 From: 72004 Date: Fri, 21 Aug 2026 22:27:36 +0500 Subject: [PATCH 1/2] fix: don't drain other event loops' buses in BaseEvent.__await__ The drain loop in __await__ iterated EventBus.all_instances (process-global) and ran each bus's queued handlers on the awaiting task's loop. In a multi-loop app (parallel agent sessions, each on its own loop) that executed one bus's handlers on another bus's loop, where they hung and accumulated until the bus hit its 100-event capacity limit and dispatch() raised RuntimeError. Track each bus's owning loop (set in _start) and skip buses whose loop is not the current running loop; each bus's own _run_loop drains it on its own loop. Closes browser-use/browser-use#5509. --- bubus/models.py | 11 +++ bubus/service.py | 2 + tests/test_cross_loop_isolation.py | 117 +++++++++++++++++++++++++++++ 3 files changed, 130 insertions(+) create mode 100644 tests/test_cross_loop_isolation.py diff --git a/bubus/models.py b/bubus/models.py index 4079e49..d7cfbbb 100644 --- a/bubus/models.py +++ b/bubus/models.py @@ -295,6 +295,8 @@ async def wait_for_handlers_to_complete_then_return_event(): max_iterations = 1000 # Prevent infinite loops iterations = 0 + current_loop = asyncio.get_running_loop() + try: while not self.event_completed_signal.is_set() and iterations < max_iterations: iterations += 1 @@ -306,6 +308,15 @@ async def wait_for_handlers_to_complete_then_return_event(): if not bus or not bus.event_queue: continue + # Only drain buses that belong to the current event loop. Draining a + # bus owned by another loop runs its handlers on the wrong loop, where + # they hang forever and pile up until the bus hits its capacity limit + # (cross-loop contamination, browser-use/browser-use#5509). Each bus's + # own _run_loop drains it on its own loop. A bus that hasn't started yet + # (_loop is None) has no run loop to poll against, so skip it here too. + if bus._loop is not current_loop: + continue + # Process one event from this bus if available try: if bus.event_queue.qsize() > 0: diff --git a/bubus/service.py b/bubus/service.py index 72f652e..0788203 100644 --- a/bubus/service.py +++ b/bubus/service.py @@ -274,6 +274,7 @@ class EventBus: _is_running: bool = False _runloop_task: asyncio.Task[None] | None = None _on_idle: asyncio.Event | None = None + _loop: asyncio.AbstractEventLoop | None = None # the event loop this bus's run loop was started on def __init__( self, @@ -733,6 +734,7 @@ def close_with_cleanup() -> None: self._on_idle.clear() # Start in a busy state unless we confirm queue is empty by running step() at least once # Create and start the run loop task + self._loop = loop self._runloop_task = loop.create_task(self._run_loop(), name=f'{self}._run_loop') self._is_running = True except RuntimeError: diff --git a/tests/test_cross_loop_isolation.py b/tests/test_cross_loop_isolation.py new file mode 100644 index 0000000..5d0a132 --- /dev/null +++ b/tests/test_cross_loop_isolation.py @@ -0,0 +1,117 @@ +# pyright: basic +"""Regression test for cross-loop contamination in BaseEvent.__await__. + +When a handler awaits an event, __await__ enters a drain loop that used to iterate +*every* EventBus in the process and run its queued handlers on the awaiting task's +event loop. In a multi-loop application (e.g. several parallel agent sessions, each +on its own loop) that ran one bus's handlers on another bus's loop, where they hung +and piled up until the bus hit its capacity limit: + + RuntimeError: EventBus at capacity: 100 pending events (100 max) + +The drain loop must only process buses that belong to the current running loop. +See browser-use/browser-use#5509. +""" + +import asyncio +import threading + +from bubus import BaseEvent, EventBus + + +class BlockerEvent(BaseEvent): + pass + + +class ProbeEvent(BaseEvent): + pass + + +class ParentEvent(BaseEvent): + pass + + +def _spin_loop(loop: asyncio.AbstractEventLoop) -> None: + asyncio.set_event_loop(loop) + loop.run_forever() + + +async def test_await_drain_does_not_process_other_loops_buses(): + loop_a = asyncio.get_running_loop() + + # --- Bus B lives on its own event loop, running in a background thread --- + loop_b = asyncio.new_event_loop() + thread = threading.Thread(target=_spin_loop, args=(loop_b,), daemon=True) + thread.start() + + ran_on: dict[str, int] = {} + release_blocker = threading.Event() + + async def blocker_handler(event: BlockerEvent) -> None: + # Occupy bus B's serial processing so its own run loop cannot advance the + # ProbeEvent. That way the *only* thing that could move the ProbeEvent while + # the blocker is held is a (buggy) cross-loop drain from loop A. + while not release_blocker.is_set(): + await asyncio.sleep(0.01) + + async def probe_handler(event: ProbeEvent) -> None: + ran_on['probe'] = id(asyncio.get_running_loop()) + + async def build_bus_b() -> EventBus: + b = EventBus(name='BusB') + b.on(BlockerEvent, blocker_handler) + b.on(ProbeEvent, probe_handler) + b.dispatch(BlockerEvent()) # starts B's run loop and keeps it busy + return b + + bus_b = asyncio.run_coroutine_threadsafe(build_bus_b(), loop_b).result(timeout=5) + + # Create the probe on loop B and keep a reference so loop A can await the same + # object. It sits queued behind the blocker. + probe = await asyncio.wrap_future( + asyncio.run_coroutine_threadsafe(_dispatch_probe(bus_b), loop_b) + ) + await asyncio.sleep(0.1) + assert bus_b.event_queue is not None and bus_b.event_queue.qsize() >= 1 + + # --- Bus A: a handler on loop A awaits bus B's probe event -> enters the drain --- + bus_a = EventBus(name='BusA') + + async def parent_handler(event: ParentEvent) -> None: + # Awaiting from inside a handler (holding the global lock) triggers the + # cross-bus drain loop. Give it a bounded window to (wrongly) pick up the + # probe from bus B, then stop waiting so the test can assert. + try: + await asyncio.wait_for(_await_probe(probe), timeout=0.4) + except asyncio.TimeoutError: + pass + + bus_a.on(ParentEvent, parent_handler) + await bus_a.dispatch(ParentEvent()) + + # The probe belongs to bus B's loop. Loop A must NOT have run it. + assert ran_on.get('probe') != id(loop_a), ( + "ProbeEvent from bus B ran on bus A's loop — cross-loop contamination" + ) + + # Once bus B is free, its own loop processes the probe — on loop B. + release_blocker.set() + for _ in range(100): + if 'probe' in ran_on: + break + await asyncio.sleep(0.02) + assert ran_on.get('probe') == id(loop_b) + + # --- Cleanup --- + asyncio.run_coroutine_threadsafe(bus_b.stop(), loop_b).result(timeout=5) + await bus_a.stop() + loop_b.call_soon_threadsafe(loop_b.stop) + thread.join(timeout=5) + + +async def _dispatch_probe(bus: EventBus) -> ProbeEvent: + return bus.dispatch(ProbeEvent()) + + +async def _await_probe(probe: ProbeEvent) -> None: + await probe From 40d6bbb11164cf1a7bc981ee0a19e239f0deafb1 Mon Sep 17 00:00:00 2001 From: 72004 Date: Fri, 21 Aug 2026 23:03:26 +0500 Subject: [PATCH 2/2] review: skip stopped buses in drain; make test teardown unconditional - Drain guard also skips buses with _is_running False: a stopped bus can keep _loop set with events still queued, and nothing should run its handlers after stop(). - Move the regression test's assertions into try/ and all shutdown into finally/ so a failing run never leaks bus B's run loop or its thread. --- bubus/models.py | 16 ++++++----- tests/test_cross_loop_isolation.py | 46 +++++++++++++++++------------- 2 files changed, 35 insertions(+), 27 deletions(-) diff --git a/bubus/models.py b/bubus/models.py index d7cfbbb..f7767f2 100644 --- a/bubus/models.py +++ b/bubus/models.py @@ -308,13 +308,15 @@ async def wait_for_handlers_to_complete_then_return_event(): if not bus or not bus.event_queue: continue - # Only drain buses that belong to the current event loop. Draining a - # bus owned by another loop runs its handlers on the wrong loop, where - # they hang forever and pile up until the bus hits its capacity limit - # (cross-loop contamination, browser-use/browser-use#5509). Each bus's - # own _run_loop drains it on its own loop. A bus that hasn't started yet - # (_loop is None) has no run loop to poll against, so skip it here too. - if bus._loop is not current_loop: + # Only drain running buses that belong to the current event loop. + # Draining a bus owned by another loop runs its handlers on the wrong + # loop, where they hang forever and pile up until the bus hits its + # capacity limit (cross-loop contamination, browser-use/browser-use#5509). + # Each bus's own _run_loop drains it on its own loop. Skip buses that + # haven't started (_loop is None) or have been stopped (_is_running is + # False) — a stopped bus can keep _loop set with events still queued, and + # nothing should run its handlers after stop(). + if not bus._is_running or bus._loop is not current_loop: continue # Process one event from this bus if available diff --git a/tests/test_cross_loop_isolation.py b/tests/test_cross_loop_isolation.py index 5d0a132..b6e38fe 100644 --- a/tests/test_cross_loop_isolation.py +++ b/tests/test_cross_loop_isolation.py @@ -87,26 +87,32 @@ async def parent_handler(event: ParentEvent) -> None: pass bus_a.on(ParentEvent, parent_handler) - await bus_a.dispatch(ParentEvent()) - - # The probe belongs to bus B's loop. Loop A must NOT have run it. - assert ran_on.get('probe') != id(loop_a), ( - "ProbeEvent from bus B ran on bus A's loop — cross-loop contamination" - ) - - # Once bus B is free, its own loop processes the probe — on loop B. - release_blocker.set() - for _ in range(100): - if 'probe' in ran_on: - break - await asyncio.sleep(0.02) - assert ran_on.get('probe') == id(loop_b) - - # --- Cleanup --- - asyncio.run_coroutine_threadsafe(bus_b.stop(), loop_b).result(timeout=5) - await bus_a.stop() - loop_b.call_soon_threadsafe(loop_b.stop) - thread.join(timeout=5) + try: + await bus_a.dispatch(ParentEvent()) + + # The probe belongs to bus B's loop. Loop A must NOT have run it. + assert ran_on.get('probe') != id(loop_a), ( + "ProbeEvent from bus B ran on bus A's loop — cross-loop contamination" + ) + + # Once bus B is free, its own loop processes the probe — on loop B. + release_blocker.set() + for _ in range(100): + if 'probe' in ran_on: + break + await asyncio.sleep(0.02) + assert ran_on.get('probe') == id(loop_b) + finally: + # Always tear down, even if an assertion above fails, so a failing run + # never leaks bus B's still-spinning run loop or its background thread + # into later tests. + release_blocker.set() + try: + asyncio.run_coroutine_threadsafe(bus_b.stop(), loop_b).result(timeout=5) + finally: + await bus_a.stop() + loop_b.call_soon_threadsafe(loop_b.stop) + thread.join(timeout=5) async def _dispatch_probe(bus: EventBus) -> ProbeEvent: