Three listeners on one event. Two of them raise. The await hands back a single RuntimeError, and the third listener — the healthy one, the one that was supposed to write the record — is never invoked at all. Not late, not slow: never. That happened in a tree where 1227 tests passed and coverage reported 100% of statements and branches.
The method's own docstring said: "All listeners run concurrently and are awaited together."
A dispatch loop that called before it scheduled
# src/anima/events.py at 2432c28 async def parallel(self, event, *args, this_arg=None, **kwargs): listeners = self._filter_listeners(event, this_arg) tasks = [] for listener in listeners: res = listener.callback(*args, **kwargs) # invoked here if inspect.isawaitable(res): tasks.append(res) else: async def _sync_wrapper(val): return val tasks.append(_sync_wrapper(res)) if not tasks: return [] return await asyncio.gather(*tasks) # awaited here Every callback is invoked while the list of things to await is still being assembled. asyncio.gather gets a say only after the loop finishes. That ordering produces two separate defects:
- A listener that raises synchronously raises inside the
forloop. The loop is abandoned, so no listener after it is ever called — and any coroutine already created for an earlier async listener is dropped without being awaited (RuntimeWarning: coroutine ... was never awaited). - When the failure comes from an awaited listener instead,
gatherpropagates that first exception immediately and leaves the sibling tasks running. The docs are explicit that they are not cancelled. Soawait parallel(...)returns while healthy listeners are still mid-flight, and whatever they return or raise afterwards goes nowhere.
Two defects, one report. The caller sees a single exception either way.
What the caller can tell apart
Measured on a checkout of 2432c28, Python 3.12.16, two listener shapes:
bus = EventBus() def bad(e): raise RuntimeError("boom-sync") async def bad2(e): raise ValueError("boom-async") async def good(e): await asyncio.sleep(0.05) healthy.append("ran") return "ok" for cb in (bad, bad2, good): bus.on("save", cb) [sync raiser first] raised RuntimeError, 1 of 2 failures: boom-sync [sync raiser first] healthy at return=False healthy after 200ms=False [async raiser first] raised ValueError, 1 of 2 failures: boom-async [async raiser first] healthy at return=False healthy after 200ms=True Read the two healthy values. In the first ordering the healthy listener never ran, no matter how long I waited: it was never scheduled, because the synchronous raise ended the loop two lines before gather. In the second ordering it did run, finishing after await parallel() had already handed the exception to my except block.
And 1 of 2 in both: the ValueError from the second listener is simply gone. From the caller's side, one bad listener and two bad listeners plus one skipped listener produce the same thing — a single exception with no count attached. The caller cannot retry the failed subset, cannot log "2 of 3 listeners failed", cannot even know it should look at the other two.
Why 1227 green tests did not see it
$ pytest -q 1227 passed, 2 warnings in 25.75s TOTAL 7369 0 2818 0 100% Required test coverage of 100% reached. Total coverage: 100.00% Exactly one of those tests dispatched a raising listener through parallel(): tests/integration/test_phase1_acceptance.py::test_acceptance_dispatch_modes.
failing_ctx = Context() settled = [] async def boom(_x): raise RuntimeError("kaboom") async def slow_ok(_x): await asyncio.sleep(0.05) settled.append("slow-ok") return "slow-ok" failing_ctx.on("work", boom, mode=EventMode.PARALLEL) failing_ctx.on("work", slow_ok, mode=EventMode.PARALLEL) with pytest.raises(RuntimeError, match="kaboom"): await failing_ctx.parallel("work", "w") await asyncio.sleep(0.1) assert settled == ["slow-ok"] That test is not careless. Its comment names the divergence outright — "Anima raises the first error instead of an aggregate; pin the shared half here (all listeners settle) and the first error" — so somebody read the specification, saw the gap, and pinned what both implementations do share.
Now look at where the assertion sits relative to the sleep. Pre-fix, gather returned at the raise with slow_ok still running as an orphan; the sleep(0.1) gave that orphan 100ms to finish; then settled == ["slow-ok"] passed. The property was false at the instant await parallel() returned and true 100ms later, and the test only ever looked at 100ms. It measured the right property on a clock that outlives the work.
Two properties were never asserted anywhere: how many failures surfaced, and whether the siblings had finished by the time the caller got control back. Coverage reported 100% because every line ran — including the line that discarded the second exception.
The fix
Invoke each listener inside its own coroutine, and collect exceptions instead of propagating the first one:
- tasks = [] - for listener in listeners: - res = listener.callback(*args, **kwargs) - if inspect.isawaitable(res): - tasks.append(res) - else: - tasks.append(_sync_wrapper(res)) - if not tasks: - return [] - return await asyncio.gather(*tasks) + results = await asyncio.gather( + *(_dispatch_one(listener, args, kwargs) for listener in listeners), + return_exceptions=True, + ) async def _dispatch_one(listener, args, kwargs): result = listener.callback(*args, **kwargs) if inspect.isawaitable(result): return await result return result return_exceptions=True makes gather wait for every listener before returning, failures included, and hands the exceptions back as values. The failures are then attributed: each gets a note naming its listener (index plus qualname, in registration order), a single failure re-raises the original exception object so existing except ValueError: callers keep working, and two or more raise an ExceptionGroup that names each failing listener. Non-Exception base exceptions (CancelledError and friends) cannot live inside an ExceptionGroup and must not be swallowed, so the first one is re-raised once every listener has settled.
The same probe after the fix
[sync raiser first] raised ExceptionGroup, 2 of 2 failures: 2 of 3 listeners failed for event 'save': [0] sync_raiser.<locals>.bad, [1] sync_raiser.<locals>.bad2 (2 sub-exceptions) [sync raiser first] healthy at return=True healthy after 200ms=True [async raiser first] raised ValueError, 1 of 2 failures: boom-async [async raiser first] healthy at return=True healthy after 200ms=True The healthy listener now runs in the ordering that used to skip it entirely, and in both orderings it has finished before parallel() returns. The single-failure case still raises ValueError unchanged; that line is a compatibility guarantee, not a leftover.
$ pytest -q 1232 passed, 2 warnings in 26.00s TOTAL 7388 0 2830 0 100% $ mypy src/anima/events.py Success: no issues found in 1 source file Five new tests. The two properties the old suite never checked are now assertions: the number of failures, and the state of the siblings at return.
The general shape
Three things line up, and I have seen this shape in more than one codebase:
- A dispatch loop that invokes callbacks while it is still assembling the list of things to await.
- A return channel where "one listener failed" and "two failed, one skipped" are the same value — a bare exception carrying no count.
- A test whose clock outlives the work it observes: a
sleepplaced after the raise, which lets a settle assertion pass for the wrong reason.
Number 3 is the one I would carry forward. It was the most careful test in that file.
What I'd test first
- Assert the count, not the presence.
len(group.exceptions) == 2fails loudly when a listener's failure is dropped;pytest.raises(RuntimeError)passes whether one listener broke or all of them did. - Assert at the moment of return. No sleep between the dispatch and the assertion. If a property needs a sleep to hold, the code is telling you who is still running.
- Register the synchronous raiser first.
def bad(): raiseas listener zero is the ordering that aborts dispatch before anything is scheduled, and it is the ordering nobody writes by hand, because it looks odd sitting next to a pile ofasync def. - Keep one healthy listener in the same dispatch and have it record its own invocation. "Did the successful path survive a failing sibling" is a different question from "did the failing sibling get reported", and the suite only ever asked the second one.
One question I would like an answer to, since I have only measured this from my side: if you fan out with asyncio.gather on Python 3.11+, have you moved to TaskGroup — and when gather returned at the first exception and left the siblings running, was that fail-fast behaviour something you chose, or something nobody picked and nobody measured?
