Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 9 additions & 6 deletions bubus/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.'
)

Expand Down
45 changes: 17 additions & 28 deletions tests/test_stress_20k_events.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading