From 113ccb61f27cd30c3345ade4cdaf24ad0d70d79c Mon Sep 17 00:00:00 2001 From: zhang wenkai <81221238+zwk110@users.noreply.github.com> Date: Wed, 21 Oct 2026 14:47:57 +0800 Subject: [PATCH 1/3] fix: keep shut-down event buses stopped --- bubus/service.py | 14 ++++++--- tests/test_shutdown.py | 65 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 75 insertions(+), 4 deletions(-) create mode 100644 tests/test_shutdown.py diff --git a/bubus/service.py b/bubus/service.py index 72f652e..3b58331 100644 --- a/bubus/service.py +++ b/bubus/service.py @@ -684,6 +684,9 @@ def notify_expect_handler(event: 'BaseEvent[Any]') -> None: def _start(self) -> None: """Start the event bus if not already running""" if not self._is_running: + if self.event_queue is not None and self.event_queue._is_shutdown: # pyright: ignore[reportPrivateUsage] + raise QueueShutDown + try: loop = asyncio.get_running_loop() @@ -899,9 +902,9 @@ async def _get_next_event(self, wait_for_timeout: float = 0.1) -> 'BaseEvent[Any if not self._is_running: return None + # Create a task for queue.get() so we can cancel it cleanly + get_next_queued_event = asyncio.create_task(self.event_queue.get()) try: - # Create a task for queue.get() so we can cancel it cleanly - get_next_queued_event = asyncio.create_task(self.event_queue.get()) if hasattr(get_next_queued_event, '_log_destroy_pending'): get_next_queued_event._log_destroy_pending = False # type: ignore # Suppress warnings on this task in case of cleanup @@ -922,9 +925,12 @@ async def _get_next_event(self, wait_for_timeout: float = 0.1) -> 'BaseEvent[Any self._on_idle.set() return None - except (asyncio.CancelledError, RuntimeError, QueueShutDown): - # Clean cancellation during shutdown or queue was shut down + except RuntimeError: + # Event loop is closing. return None + finally: + get_next_queued_event.cancel() + await asyncio.gather(get_next_queued_event, return_exceptions=True) async def step( self, event: 'BaseEvent[Any] | None' = None, timeout: float | None = None, wait_for_timeout: float = 0.1 diff --git a/tests/test_shutdown.py b/tests/test_shutdown.py new file mode 100644 index 0000000..5fd413c --- /dev/null +++ b/tests/test_shutdown.py @@ -0,0 +1,65 @@ +# pyright: basic +import asyncio + +import pytest + +from bubus import BaseEvent, EventBus +from bubus.service import QueueShutDown + + +class PingEvent(BaseEvent): + pass + + +@pytest.mark.parametrize('clear', [False, True]) +async def test_late_dispatch_does_not_restart_stopped_bus(clear: bool): + bus = EventBus() + await bus.dispatch(PingEvent()) + await bus.stop(timeout=1, clear=clear) + + try: + with pytest.raises(QueueShutDown): + bus.dispatch(PingEvent()) + assert not bus._is_running + assert bus._runloop_task is None + finally: + await bus.stop() + + +async def test_cancel_idle_runloop_closes_queue_waiter(): + bus = EventBus() + await bus.dispatch(PingEvent()) + await bus.wait_until_idle() + task = bus._runloop_task + assert task is not None + assert bus.event_queue is not None + + # Cancel while the loop is waiting for its next event. + async with asyncio.timeout(1): + while not bus.event_queue._getters: # noqa: ASYNC110 - wait for the real queue waiter before cancellation + await asyncio.sleep(0) + + try: + task.cancel() + done, _ = await asyncio.wait({task}, timeout=1) + assert task in done + assert not bus._is_running + assert not bus.event_queue._getters + finally: + await bus.stop() + + +async def test_queue_shutdown_exits_runloop(): + bus = EventBus() + await bus.dispatch(PingEvent()) + task = bus._runloop_task + assert task is not None + assert bus.event_queue is not None + + try: + bus.event_queue.shutdown() + done, _ = await asyncio.wait({task}, timeout=1) + assert task in done + assert not bus._is_running + finally: + await bus.stop() From e0b8cc6d130dd5b65cc0728245b4e6b650a4b71c Mon Sep 17 00:00:00 2001 From: zhang wenkai <81221238+zwk110@users.noreply.github.com> Date: Wed, 21 Oct 2026 18:23:46 +0800 Subject: [PATCH 2/3] fix: keep idle waits safe after event bus shutdown --- README.md | 1 + bubus/service.py | 5 ++++- tests/test_shutdown.py | 18 ++++++++++++++++++ 3 files changed, 23 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index afd7ed8..52676ec 100644 --- a/README.md +++ b/README.md @@ -572,6 +572,7 @@ event = await bus.expect( ##### `wait_until_idle(timeout: float | None=None)` Wait until all events are processed and the bus is idle. +After `stop()`, this returns immediately without restarting the bus. ```python await bus.wait_until_idle() # wait indefinitely until EventBus has finished processing all events diff --git a/bubus/service.py b/bubus/service.py index 3b58331..0a4f820 100644 --- a/bubus/service.py +++ b/bubus/service.py @@ -818,7 +818,10 @@ async def stop(self, timeout: float | None = None, clear: bool = False) -> None: pass async def wait_until_idle(self, timeout: float | None = None) -> None: - """Wait until the event bus is idle (no events being processed and all handlers completed)""" + """Wait until idle, or return immediately if the bus has already been stopped.""" + + if not self._is_running and self.event_queue is not None and self.event_queue._is_shutdown: # pyright: ignore[reportPrivateUsage] + return self._start() assert self._on_idle and self.event_queue, 'EventBus._start() must be called before wait_until_idle() is reached' diff --git a/tests/test_shutdown.py b/tests/test_shutdown.py index 5fd413c..8be0460 100644 --- a/tests/test_shutdown.py +++ b/tests/test_shutdown.py @@ -49,17 +49,35 @@ async def test_cancel_idle_runloop_closes_queue_waiter(): await bus.stop() +@pytest.mark.parametrize('clear', [False, True]) +async def test_wait_until_idle_after_stop_does_not_restart_bus(clear: bool): + bus = EventBus() + await bus.dispatch(PingEvent()) + await bus.stop(timeout=1, clear=clear) + + await bus.wait_until_idle(timeout=1) + + assert not bus._is_running + assert bus._runloop_task is None + + async def test_queue_shutdown_exits_runloop(): bus = EventBus() await bus.dispatch(PingEvent()) + await bus.wait_until_idle() task = bus._runloop_task assert task is not None assert bus.event_queue is not None + async with asyncio.timeout(1): + while not bus.event_queue._getters: # noqa: ASYNC110 - wait for the real queue waiter before shutdown + await asyncio.sleep(0) + try: bus.event_queue.shutdown() done, _ = await asyncio.wait({task}, timeout=1) assert task in done assert not bus._is_running + assert not bus.event_queue._getters finally: await bus.stop() From 6fe20273c890f78fe5c209f5d3be1542d9f6062f Mon Sep 17 00:00:00 2001 From: zhang wenkai <81221238+zwk110@users.noreply.github.com> Date: Wed, 21 Oct 2026 23:16:58 +0800 Subject: [PATCH 3/3] fix: release pending idle waits when the event bus stops --- README.md | 2 +- bubus/service.py | 18 +++++-------- tests/test_shutdown.py | 57 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 64 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 52676ec..2b665db 100644 --- a/README.md +++ b/README.md @@ -572,7 +572,7 @@ event = await bus.expect( ##### `wait_until_idle(timeout: float | None=None)` Wait until all events are processed and the bus is idle. -After `stop()`, this returns immediately without restarting the bus. +If `stop()` is called, any pending idle wait returns. Waiting after `stop()` also returns immediately without restarting the bus. ```python await bus.wait_until_idle() # wait indefinitely until EventBus has finished processing all events diff --git a/bubus/service.py b/bubus/service.py index 0a4f820..173f6b5 100644 --- a/bubus/service.py +++ b/bubus/service.py @@ -818,7 +818,7 @@ async def stop(self, timeout: float | None = None, clear: bool = False) -> None: pass async def wait_until_idle(self, timeout: float | None = None) -> None: - """Wait until idle, or return immediately if the bus has already been stopped.""" + """Wait until the bus is idle or stopped.""" if not self._is_running and self.event_queue is not None and self.event_queue._is_shutdown: # pyright: ignore[reportPrivateUsage] return @@ -830,16 +830,8 @@ async def wait_until_idle(self, timeout: float | None = None) -> None: remaining_timeout = timeout try: - # First wait for the queue to be empty - join_task = asyncio.create_task(self.event_queue.join()) - await asyncio.wait_for(join_task, timeout=remaining_timeout) - - # Update remaining timeout - if timeout is not None: - elapsed = asyncio.get_event_loop().time() - start_time - remaining_timeout = max(0, timeout - elapsed) - - # Wait for idle state + # The run loop signals idle after draining the queue; stop() also + # sets this signal when aborting events that cannot finish queue.join(). idle_task = asyncio.create_task(self._on_idle.wait()) await asyncio.wait_for(idle_task, timeout=remaining_timeout) @@ -849,7 +841,9 @@ async def wait_until_idle(self, timeout: float | None = None) -> None: await asyncio.sleep(0) # Yield to event loop # Double-check we're truly idle - if new events came in, wait again - while not self._on_idle.is_set() or self.events_started or self.events_pending: + while self._is_running and ( + not self._on_idle.is_set() or self.events_started or self.events_pending or self.event_queue.qsize() + ): if timeout is not None: elapsed = asyncio.get_event_loop().time() - start_time remaining_timeout = max(0, timeout - elapsed) diff --git a/tests/test_shutdown.py b/tests/test_shutdown.py index 8be0460..91d11df 100644 --- a/tests/test_shutdown.py +++ b/tests/test_shutdown.py @@ -61,6 +61,63 @@ async def test_wait_until_idle_after_stop_does_not_restart_bus(clear: bool): assert bus._runloop_task is None +@pytest.mark.parametrize('clear', [False, True]) +async def test_stop_releases_pending_idle_wait(clear: bool): + bus = EventBus() + started, release = asyncio.Event(), asyncio.Event() + + async def handle(event: PingEvent): + started.set() + await release.wait() + + bus.on(PingEvent, handle) + bus.dispatch(PingEvent()) + bus.dispatch(PingEvent()) + await asyncio.wait_for(started.wait(), timeout=1) + waiter = asyncio.create_task(bus.wait_until_idle()) + await asyncio.sleep(0) + assert not waiter.done() + + try: + await bus.stop(timeout=0, clear=clear) + done, _ = await asyncio.wait({waiter}, timeout=1) + assert waiter in done + assert waiter.result() is None + assert not bus._is_running + finally: + release.set() + waiter.cancel() + await asyncio.gather(waiter, return_exceptions=True) + await bus.stop() + + +async def test_idle_wait_after_new_dispatch_waits_for_handler(): + bus = EventBus() + await bus.wait_until_idle() + started, release = asyncio.Event(), asyncio.Event() + + async def handle(event: PingEvent): + started.set() + await release.wait() + + bus.on(PingEvent, handle) + event = bus.dispatch(PingEvent()) + waiter = asyncio.create_task(bus.wait_until_idle()) + try: + await asyncio.wait_for(started.wait(), timeout=1) + assert not waiter.done() + release.set() + await asyncio.wait_for(waiter, timeout=1) + assert event.event_status == 'completed' + assert bus.event_queue is not None + assert bus.event_queue.empty() + finally: + release.set() + waiter.cancel() + await asyncio.gather(waiter, return_exceptions=True) + await bus.stop() + + async def test_queue_shutdown_exits_runloop(): bus = EventBus() await bus.dispatch(PingEvent())