fix: prevent stopped event buses from restarting a busy loop

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.
+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

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()
 
@@ -815,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 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
 
         self._start()
         assert self._on_idle and self.event_queue, 'EventBus._start() must be called before wait_until_idle() is reached'
@@ -824,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)
 
@@ -843,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)
@@ -899,9 +899,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 +922,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

tests/test_shutdown.py

@@ -0,0 +1,140 @@
+# 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()
+
+
+@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
+
+
+@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())
+    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()