Count unique unfinished events for EventBus capacity

[sarath-menon]

bubus36.mp4

EventBus capacity double-counts queued events that also appear in history and can lose track of unfinished work when history evicts it. This change tracks unique unfinished event objects independently of history, preserving the 50-entry queue limit and 100-event capacity while allowing existing RuntimeError and QueueFull handlers to catch capacity failures. All 69 focused tests pass, including history eviction, distinct objects sharing an event ID, and repeated dispatch of the same object. This dependency-only draft leaves Cloud pinned to bubus 1.5.6, so package publication and Cloud adoption remain separate steps; the animation illustrates local tests. The pre-existing mutation of parent/path metadata before failed admission remains a limitation and should be addressed separately before broad overload rollout.

The automated-review credential was repaired and the validated Sonnet 4.6 workflow setting adopted through #37; all checks on this head, including automated review, now pass.

[claude]

Claude encountered an error —— View job

---

I'll analyze this and get back to you.

[claude]

Claude encountered an error —— View job

---

I'll analyze this and get back to you.

[claude]

Claude finished @sarath-menon's task —— View job

---

PR Review: Count unique unfinished events for EventBus capacity

Tasks:

• Gather context and read changed files
• Review code quality and best practices
• Check for potential bugs or issues
• Evaluate performance considerations
• Assess test coverage
• Post final review

---

Summary

This PR correctly fixes two related bugs in EventBus capacity tracking:

1. Double-counting: the old code counted queued events from the queue size plus matching events from event_history, but queued events are also in history, so they were counted twice.
2. History eviction hiding outstanding work: when event_history was trimmed below max_history_size, in-progress events could disappear from history, making the capacity check unaware of them.

The fix introduces _outstanding_events: dict[int, BaseEvent[Any]] keyed by id(event) (Python object identity) rather than event_id (UUID string), allowing distinct event objects that share the same UUID to each be counted independently.

Overall the approach is sound. A few observations below.

---

Issues and Observations

1. Race / stale-key accumulation in the admission filter (medium)

bubus/service.py:551-553

self._outstanding_events = {
    key: value for key, value in self._outstanding_events.items() if value.event_status in ('pending', 'started')
}

This replaces the whole dict on every dispatch() call. Because id(event) can be reused by Python for new objects after an old one is garbage-collected, a freshly-allocated event could theoretically inherit the id of a previously-completed event that was already purged from the dict — though in practice this is unlikely. More practically, the dict comprehension is O(n) on every dispatch(), which is fine at current scale but worth noting.

A minor alternative would be to remove completed entries lazily in process_event (already done at line 991) and skip the comprehension in dispatch(), instead relying only on the lazy removal. The current dual-cleanup approach is fine and defensive; just something to consider for the future.

2. _outstanding_events is not pruned in stop() without clear=True (low)

bubus/service.py:795-799 – the clear=True branch clears _outstanding_events, but the normal stop() path does not. After a stop without clear, _outstanding_events may retain completed-but-not-yet-pruned events. This is benign for current usage but worth being aware of.

3. __str__ still reads from event_history (low / cosmetic)

bubus/service.py:374

def __str__(self) -> str:
    icon = '🟢' if self._is_running else '🔴'
    return f'{self.name}{icon}(⏳ {len(self.events_pending or [])} | ▶️ {len(self.events_started or [])} | ✅ {len(self.events_completed or [])} ➡️ {len(self.handlers)} 👂)'

events_pending and events_started still derive from event_history, so they can undercount after history eviction. This was pre-existing, but now that _outstanding_events exists it would be easy to make __str__ more accurate. Not a blocker.

4. Capacity error message: "Processing" count can be misleading (low)

bubus/service.py:559-561

f'Queue: {queue_size}, Processing: {max(0, total_pending - queue_size)}. '

total_pending counts all outstanding events by object identity, while queue_size counts items currently in the queue. For the repeated-same-instance case (test_repeated_same_instance_preserves_one_handler_execution), dispatching the same object twice puts it in the queue twice (qsize()==2) but only adds one entry to _outstanding_events, so total_pending - queue_size could be negative — hence the max(0, …) guard. The guard prevents a crash but the message may say "Processing: 0" when there actually is an active handler. Minor cosmetic issue.

5. Admitted event mutation before capacity check (acknowledged limitation)

The PR description already calls this out: event.event_path is mutated at line 539 before the capacity check at line 554. If admission is refused, the caller's event object has already been modified. This is a pre-existing issue not introduced by this PR, but worth tracking.

---

Test Coverage

The new test file tests/test_unique_capacity.py covers the four key scenarios well:

Test	What it checks
test_history_eviction_cannot_hide_outstanding_events	History at max_history_size=1 can't hide 100 admitted events
test_queue_bound_completion_and_exception_compatibility	EventBusCapacityError is both RuntimeError and asyncio.QueueFull; slots reopen after completion
test_started_event_survives_history_eviction_in_capacity_count	In-flight event evicted from history still counts toward capacity
test_distinct_instances_with_same_event_id_remain_counted	Two objects sharing a UUID are counted as two outstanding items

One scenario that could be worth adding (or documenting as a known gap): dispatching the same object instance more than once currently adds two queue entries but only one _outstanding_events entry (by id). test_repeated_same_instance_preserves_one_handler_execution shows the handler only runs once (loop prevention), but the capacity accounting for that path is not explicitly asserted.

---

Code Quality

• EventBusCapacityError(asyncio.QueueFull, RuntimeError) is a clean addition — callers can catch either parent.
• The id(event) key choice is correctly motivated and well-explained in the PR description.
• The two-commit separation (dict key type → UUIDStr, then corrected to int) leaves the history clean.
• No security concerns.

---

Overall verdict: The fix is correct, well-motivated, and the tests cover the core edge cases. The observations above are mostly low-severity or cosmetic. The acknowledged limitation around path mutation before failed admission is the most impactful remaining gap, and the PR description is appropriately transparent about it.