The Third Listener Was Never Called

Coverage said 100%. Statements and branches, both 100%. In the same tree an event bus could lose two of three listeners' failures and never call the third listener at all — and there was no number anywhere that could have told me.

The number was not lying

$ pytest -q
1227 passed, 2 warnings in 25.75s
TOTAL   7369   0   2818   0   100%

Line coverage asks one question: did this line run. Every line inside that dispatch loop ran, in that suite and in every dispatch before it. Coverage counts executions. It does not ask what the returned value still knows about what happened.

A 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 now
        tasks.append(res if inspect.isawaitable(res) else _sync_wrapper(res))
    if not tasks:
        return []
    return await asyncio.gather(*tasks)             # awaited later

Invocation and scheduling are interleaved; gather only sees the pile once the loop is done. Three quiet losses fall out of that ordering:

1. A synchronous raise aborts the loop, so no listener after it is ever called. In my case that was the third listener — the healthy one, the one that was supposed to write the record.
2. Any coroutine already created for an earlier async listener is dropped unawaited (RuntimeWarning: coroutine ... was never awaited).
3. When the raise comes from an awaited listener instead, gather propagates the first exception and returns while its siblings are still running. The docs are explicit that they are not cancelled. Whatever they return or raise afterwards goes nowhere.

The method's own docstring said the listeners "run concurrently and are awaited together". Two of those three clauses were true.

Same caller, three different realities

Python 3.12.16, checkout of 2432c28. One synchronous raiser, one async raiser, one healthy listener that sleeps 50ms and records that it ran.

[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

Line 2 is the expensive one: the healthy listener never ran, then or ever, because a synchronous raise ended the loop two lines before gather. Line 4 is the near miss: it did run, and finished after await parallel() had already handed the exception to my except block.

1 of 2 in both lines is the other half of it. The ValueError from the second listener is simply gone. From the caller's chair, "one listener failed" and "two failed, and one never got scheduled" are the same object: a single exception with no count attached. That means no retry of the failed subset, no 2 of 3 listeners failed for event 'save' line in the log, and no reason to go look at the other two.

The test that was almost right

Exactly one test out of 1227 dispatched a raising listener through parallel(): tests/integration/test_phase1_acceptance.py::test_acceptance_dispatch_modes.

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"]

Its own comment records the divergence — "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 exactly what both implementations agree on. That is careful work.

Now look at where the assertion sits relative to the sleep. Pre-fix, gather returned at the raise with slow_ok still an orphan; the sleep(0.1) handed that orphan 100ms to finish; by the time anyone looked, the list was non-empty. The property was false at the instant await parallel() returned and true 100ms later, and the test only ever sampled the 100ms.

Two other properties were never asserted anywhere: how many failures came back, and whether the siblings had finished by the time the caller got control.

The fix: invoke inside the coroutine, collect instead of propagate

-        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, failures included, and hands the exceptions back as values. From there the failures get attributed: each one carries 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; two or more raise an ExceptionGroup that names every 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

[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

$ 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 count of failures, and the state of the siblings at return. The single-failure case still raises ValueError — that line is a compatibility guarantee, not residue.

Three things that line up

1. A dispatch loop that invokes callbacks while it is still assembling the list of things to await.
2. A return channel where "one failed" and "two failed, one skipped" are the same value.
3. A test whose clock outlives the work it observes.

The third is the one I would carry forward. It was the most careful test in that file, written by somebody who already knew about the divergence — and its sleep is what turned a red property green.

What I'd carry forward

• Assert the count, not the presence. len(group.exceptions) == 2 fails loudly the day a failure is dropped. pytest.raises(RuntimeError) passes whether one listener broke or all three 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(): raise as listener zero is the ordering that aborts dispatch before anything is scheduled, and the ordering nobody writes by hand, because it looks odd sitting next to a pile of async def.
• Keep one healthy listener in the same dispatch, recording its own invocation. "Did the good path survive a bad sibling" is a different question from "did the bad sibling get reported", and the suite only ever asked the second one.

One thing I can't measure from my side: if your dispatcher fans out N listeners and two of them raise, does the value you get back carry a count — and if there is a 100ms sleep sitting between an await and an assertion in your suite today, did you ever find it by chasing a flake, or is it still passing on the wrong instant?