diff --git a/.github/workflows/claude-code-review.yml b/.github/workflows/claude-code-review.yml index 30cd419..7a35af1 100644 --- a/.github/workflows/claude-code-review.yml +++ b/.github/workflows/claude-code-review.yml @@ -36,6 +36,7 @@ jobs: uses: anthropics/claude-code-action@beta with: anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }} + model: claude-sonnet-4-6 # Optional: Specify model (defaults to Claude Sonnet 4, uncomment for Claude Opus 4.1) # model: "claude-opus-4-1-20250805" diff --git a/bubus/service.py b/bubus/service.py index 72f652e..2a21cdd 100644 --- a/bubus/service.py +++ b/bubus/service.py @@ -47,6 +47,10 @@ class QueueShutDown(Exception): pass +class EventBusCapacityError(asyncio.QueueFull, RuntimeError): + """A bounded bus cannot admit another event.""" + + QueueEntryType = TypeVar('QueueEntryType', bound='BaseEvent[Any]') T_ExpectedEvent = TypeVar('T_ExpectedEvent', bound='BaseEvent[Any]') @@ -330,6 +334,7 @@ def __init__( self.event_queue = None self.event_history = {} + self._outstanding_events: dict[int, BaseEvent[Any]] = {} self.handlers = defaultdict(list) self.parallel_handlers = parallel_handlers self.wal_path = Path(wal_path) if wal_path else None @@ -542,18 +547,18 @@ 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) - # Only enforce if we have memory limits set + # Outstanding admission is independent of truncated diagnostic history. + self._outstanding_events = { + key: value for key, value in self._outstanding_events.items() if value.event_status in ('pending', 'started') + } 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 - - if total_pending >= 100: - raise RuntimeError( + total_pending = len(self._outstanding_events) + if total_pending >= 100 or (self.event_queue and self.event_queue.full()): + raise EventBusCapacityError( f'EventBus at capacity: {total_pending} pending events (100 max). ' - f'Queue: {queue_size}, Processing: {pending_in_history}. ' - f'Cannot accept new events until some complete.' + f'Queue: {queue_size}, Processing: {max(0, total_pending - queue_size)}. ' + f'Queue limit: 50. Cannot accept new events until some complete.' ) # Auto-start if needed @@ -565,6 +570,7 @@ def dispatch(self, event: T_ExpectedEvent) -> T_ExpectedEvent: self.event_queue.put_nowait(event) # Only add to history after successfully queuing self.event_history[event.event_id] = event + self._outstanding_events[id(event)] = event logger.info( f'🗣️ {self}.dispatch({event.event_type}) ➡️ {event.event_type}#{event.event_id[-4:]} (#{self.event_queue.qsize()} {event.event_status})' ) @@ -789,6 +795,7 @@ async def stop(self, timeout: float | None = None, clear: bool = False) -> None: # Clear event history and handlers if requested (for memory cleanup) if clear: self.event_history.clear() + self._outstanding_events.clear() self.handlers.clear() # Remove from global instance tracking if self in EventBus.all_instances: @@ -980,6 +987,8 @@ async def process_event(self, event: 'BaseEvent[Any]', timeout: float | None = N # Mark event as complete if all handlers are done event.event_mark_complete_if_all_handlers_completed() + if event.event_status not in ('pending', 'started'): + self._outstanding_events.pop(id(event), None) # After processing this event, check if any parent events can now be marked complete # We do this by walking up the parent chain diff --git a/tests/test_unique_capacity.py b/tests/test_unique_capacity.py new file mode 100644 index 0000000..6e587b4 --- /dev/null +++ b/tests/test_unique_capacity.py @@ -0,0 +1,119 @@ +import asyncio + +import pytest + +from bubus import BaseEvent, EventBus +from bubus.service import EventBusCapacityError + + +async def test_history_eviction_cannot_hide_outstanding_events(): + bus = EventBus(max_history_size=1) + admitted = [] + try: + for _ in range(100): + event = bus.dispatch(BaseEvent()) + admitted.append(event) + assert bus.event_queue.get_nowait() is event + bus.event_queue.task_done() + assert len(bus.event_history) == 1 + assert len({e.event_id for e in admitted}) == 100 + with pytest.raises(EventBusCapacityError, match='capacity: 100 pending'): + bus.dispatch(BaseEvent()) + assert len(bus._outstanding_events) == 100 + finally: + await bus.stop(timeout=0, clear=True) + assert not bus._outstanding_events + + +async def test_queue_bound_completion_and_exception_compatibility(): + bus = EventBus(max_history_size=1) + seen = [] + + async def consume(event: BaseEvent): + seen.append(event.event_id) + + bus.on(BaseEvent, consume) + try: + events = [bus.dispatch(BaseEvent()) for _ in range(50)] + with pytest.raises(EventBusCapacityError, match='capacity: 50 pending') as error: + bus.dispatch(BaseEvent()) + assert isinstance(error.value, (RuntimeError, asyncio.QueueFull)) + assert bus.event_queue.maxsize == 50 + await bus.wait_until_idle(timeout=5) + assert sorted(seen) == sorted(e.event_id for e in events) + assert not bus._outstanding_events + await bus.dispatch(BaseEvent()) + assert len(seen) == 51 + finally: + await bus.stop(timeout=2, clear=True) + + +async def test_started_event_survives_history_eviction_in_capacity_count(): + bus = EventBus(max_history_size=1) + entered, release = asyncio.Event(), asyncio.Event() + completed = [] + + async def hold(event: BaseEvent): + entered.set() + await release.wait() + completed.append(event.event_id) + + bus.on(BaseEvent, hold) + try: + first = bus.dispatch(BaseEvent()) + await entered.wait() + queued = [bus.dispatch(BaseEvent()) for _ in range(50)] + assert first.event_id not in bus.event_history + with pytest.raises(EventBusCapacityError, match='51 pending.*Queue: 50, Processing: 1'): + bus.dispatch(BaseEvent()) + release.set() + await bus.wait_until_idle(timeout=5) + assert sorted(completed) == sorted(e.event_id for e in [first, *queued]) + finally: + release.set() + await bus.stop(timeout=2, clear=True) + + +async def test_distinct_instances_with_same_event_id_remain_counted(): + bus = EventBus(max_history_size=1) + entered, release = asyncio.Event(), asyncio.Event() + completed = [] + + async def hold(event: BaseEvent): + entered.set() + await release.wait() + completed.append(id(event)) + + bus.on(BaseEvent, hold) + try: + first = bus.dispatch(BaseEvent()) + await entered.wait() + second = bus.dispatch(BaseEvent(event_id=first.event_id)) + assert len(bus._outstanding_events) == 2 + release.set() + await bus.wait_until_idle(timeout=5) + assert sorted(completed) == sorted([id(first), id(second)]) + assert not bus._outstanding_events + finally: + release.set() + await bus.stop(timeout=2, clear=True) + + +async def test_repeated_same_instance_preserves_one_handler_execution(): + bus = EventBus() + calls = [] + + async def consume(event: BaseEvent): + calls.append(id(event)) + + bus.on(BaseEvent, consume) + try: + event = BaseEvent() + bus.dispatch(event) + bus.dispatch(event) + assert bus.event_queue.qsize() == 2 + await bus.wait_until_idle(timeout=5) + assert calls == [id(event)] + assert not bus._outstanding_events + finally: + await bus.stop(timeout=2, clear=True)