diff --git a/eventforge/observers.py b/eventforge/observers.py index 9746079..8515bb9 100644 --- a/eventforge/observers.py +++ b/eventforge/observers.py @@ -149,7 +149,25 @@ def dispatch( class BroadcastDispatcher(Dispatcher): - """Default: call every subscriber in order; swallow each one's exceptions.""" + """Default: call every subscriber in order; a subscriber's exception + propagates out of ``dispatch`` (and thus out of ``fire``). + + Earlier subscribers in the list still run and any side effects they had + already committed stand -- broadcast delivery isn't transactional -- but + the first exception aborts the remaining subscribers for this fire and + surfaces to the caller instead of being logged and swallowed. A buggy or + assertion-raising subscriber must be able to fail the operation it's + observing, not just get an exception logged while the caller sails on. + + This is unconditional: there is no opt-in to swallow again. Fire-and-log + is still available where it's genuinely wanted -- ``ConcurrentDispatcher`` + logs submit failures since subscribers run out-of-band, and the + class-level (``@observe``/``Reporter``) subscriber path in :meth:`Eventful.fire` + isolates its own subscribers because reporting/metrics code should not be + able to break the thing it's reporting on. Competing-consumer delivery + (:class:`~eventforge.work_queue.WorkQueue`) has its own ack/nack/DLQ + failure isolation and does not go through this dispatcher. + """ def dispatch( self, @@ -158,10 +176,7 @@ def dispatch( kwargs: Dict[str, Any], ) -> None: for fn in subscribers: - try: - fn(*args, **kwargs) - except Exception: - logger.exception("broadcast subscriber failed: %r", fn) + fn(*args, **kwargs) class RoundRobinDispatcher(Dispatcher): diff --git a/tests/test_observers.py b/tests/test_observers.py index 10d6098..a3e318b 100644 --- a/tests/test_observers.py +++ b/tests/test_observers.py @@ -55,7 +55,7 @@ def test_unsubscribe(self): e.fire(2) assert results == [1] - def test_failing_subscriber_does_not_stop_chain(self): + def test_failing_subscriber_propagates(self): e = Eventful() ok = [] @@ -64,7 +64,22 @@ def bad(x): e.on(bad) e.on(lambda x: ok.append(x)) - e.fire("ping") + with pytest.raises(RuntimeError, match="boom"): + e.fire("ping") + # bad() ran before raising; the subscriber after it never got a turn. + assert ok == [] + + def test_earlier_subscribers_run_before_later_one_raises(self): + e = Eventful() + ok = [] + + def bad(x): + raise RuntimeError("boom") + + e.on(lambda x: ok.append(x)) + e.on(bad) + with pytest.raises(RuntimeError, match="boom"): + e.fire("ping") assert ok == ["ping"] def test_round_robin_dispatch(self):