broadcast: propagate subscriber exceptions instead of swallowing

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):

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):