From 3ab4c41d9b04c988db1a1ee91c08d12682c480c3 Mon Sep 17 00:00:00 2001
From: MagMueller <mamagnus00@gmail.com>
Date: Wed, 22 Jul 2026 19:57:29 -0700
Subject: [PATCH] fix: correct EventBus capacity accounting

---
 bubus/service.py                | 15 ++++++-----
 tests/test_stress_20k_events.py | 45 +++++++++++++--------------------
 2 files changed, 26 insertions(+), 34 deletions(-)

diff --git a/bubus/service.py b/bubus/service.py
index 72f652e..47e156b 100644
--- a/bubus/service.py
+++ b/bubus/service.py
@@ -542,17 +542,20 @@ def dispatch(self, event: T_ExpectedEvent) -> T_ExpectedEvent:
             f'Event.event_path must be a list of valid EventBus names, got: {event.event_path}'
         )
 
-        # Check hard limit on total pending events (queue + in-progress)
+        # Check queue and hard limits on total pending events (queue + in-progress)
         # Only enforce if we have memory limits set
         if self.max_history_size is not None:
             queue_size = self.event_queue.qsize() if self.event_queue else 0
-            pending_in_history = sum(1 for e in self.event_history.values() if e.event_status in ('pending', 'started'))
-            total_pending = queue_size + pending_in_history
+            # Queued events are also stored in history as pending, so only started events are non-overlapping.
+            processing_count = sum(1 for event in self.event_history.values() if event.event_status == 'started')
+            total_pending = queue_size + processing_count
+            queue_is_full = self.event_queue.full() if self.event_queue else False
 
-            if total_pending >= 100:
+            if total_pending >= 100 or queue_is_full:
+                capacity_reason = '100 max' if total_pending >= 100 else 'queue full'
                 raise RuntimeError(
-                    f'EventBus at capacity: {total_pending} pending events (100 max). '
-                    f'Queue: {queue_size}, Processing: {pending_in_history}. '
+                    f'EventBus at capacity: {total_pending} pending events ({capacity_reason}). '
+                    f'Queue: {queue_size}, Processing: {processing_count}. '
                     f'Cannot accept new events until some complete.'
                 )
 
diff --git a/tests/test_stress_20k_events.py b/tests/test_stress_20k_events.py
index 3a75be3..0126da6 100644
--- a/tests/test_stress_20k_events.py
+++ b/tests/test_stress_20k_events.py
@@ -156,41 +156,30 @@ async def handler(event: SimpleEvent) -> None:
 
 
 @pytest.mark.asyncio
-async def test_hard_limit_enforcement():
-    """Test that hard limit of 100 pending events is enforced"""
-    bus = EventBus(name='HardLimitTest')
+async def test_capacity_does_not_double_count_queued_events():
+    """Queued events should not also be reported as processing."""
+    bus = EventBus(name='CapacityAccountingTest')
 
     try:
-        # Create a slow handler to keep events pending
-        async def slow_handler(event: SimpleEvent) -> None:
-            await asyncio.sleep(0.5)  # Reduced from 10s to 0.5s
+        for _ in range(50):
+            bus.dispatch(SimpleEvent())
 
-        bus.on('SimpleEvent', slow_handler)
+        assert bus.event_queue is not None
+        assert bus.event_queue.qsize() == 50
+        assert len(bus.events_pending) == 50
+        assert len(bus.events_started) == 0
 
-        # Try to dispatch more than 100 events
-        events_dispatched = 0
-        errors = 0
+        with pytest.raises(RuntimeError) as exc_info:
+            bus.dispatch(SimpleEvent())
 
-        for _ in range(150):
-            try:
-                bus.dispatch(SimpleEvent())
-                events_dispatched += 1
-            except RuntimeError as e:
-                if 'EventBus at capacity' in str(e):
-                    errors += 1
-                else:
-                    raise
-
-        print(f'\nDispatched {events_dispatched} events')
-        print(f'Hit capacity error {errors} times')
-
-        # Should hit the limit
-        assert events_dispatched <= 100
-        assert errors > 0
+        error_message = str(exc_info.value)
+        assert 'EventBus at capacity: 50 pending events (queue full).' in error_message
+        assert 'Queue: 50, Processing: 0.' in error_message
+        assert '100 pending' not in error_message
+        assert 'Processing: 50' not in error_message
 
     finally:
-        # Properly stop the bus to clean up pending tasks
-        await bus.stop(timeout=0, clear=True)  # Don't wait, just force cleanup
+        await bus.stop(timeout=0, clear=True)
 
 
 @pytest.mark.asyncio