Handle domain event handlers concurrently with exception isolation

.gitignore

@@ -169,3 +169,5 @@ tmp/
 uv.lock
 
 .codspeed/
+
+.vscode

pyproject.toml

@@ -4,8 +4,8 @@ requires = ["setuptools>=42", "wheel"]
 
 [project]
 authors = [
-  {name = "Vadim Kozyrevskiy", email = "vadikko2@mail.ru"},
-  {name = "Dmitry Kutlubaev", email = "kutlubaev00@mail.ru"}
+  { name = "Vadim Kozyrevskiy", email = "vadikko2@mail.ru" },
+  { name = "Dmitry Kutlubaev", email = "kutlubaev00@mail.ru" },
 ]
 classifiers = [
   "Development Status :: 4 - Beta",
@@ -14,7 +14,7 @@ classifiers = [
   "Programming Language :: Python :: 3.10",
   "Programming Language :: Python :: 3.11",
   "Programming Language :: Python :: 3.12",
-  "Programming Language :: Python :: 3.13"
+  "Programming Language :: Python :: 3.13",
 ]
 dependencies = [
   "dataclass-wizard==0.*",
@@ -24,14 +24,14 @@ dependencies = [
   "pydantic==2.*",
   "sqlalchemy[asyncio]==2.0.*",
   "python-dotenv==1.*",
-  "typing-extensions>=4.0"
+  "typing-extensions>=4.0",
 ]
 description = "Event-Driven Architecture Framework for Distributed Systems"
-maintainers = [{name = "Vadim Kozyrevskiy", email = "vadikko2@mail.ru"}]
+maintainers = [{ name = "Vadim Kozyrevskiy", email = "vadikko2@mail.ru" }]
 name = "python-cqrs"
 readme = "README.md"
 requires-python = ">=3.10"
-version = "4.11.0"
+version = "4.12.0"
 
 [project.optional-dependencies]
 aiobreaker = ["aiobreaker>=0.3.0"]
@@ -45,10 +45,10 @@ dev = [
   "pytest-cov>=4.0.0",
   "pytest-codspeed==4.2.0",
   # Tests
-  "aio-pika==9.3.0",  # from rabbit
-  "aiokafka==0.10.0",  # from kafka
-  "retry-async==0.1.*", # from kafka
-  "requests==2.*",  # from aiokafka
+  "aio-pika==9.3.0",        # from rabbit
+  "aiokafka==0.10.0",       # from kafka
+  "retry-async==0.1.*",     # from kafka
+  "requests==2.*",          # from aiokafka
   "pytest~=7.4.2",
   "pytest-asyncio~=0.21.1",
   "pytest-env==0.6.2",
@@ -57,7 +57,7 @@ dev = [
   "asyncpg>=0.29.0",
   "redis>=5.0.0",
   # Circuit breaker for tests
-  "aiobreaker>=0.3.0"  # from aiobreaker
+  "aiobreaker>=0.3.0", # from aiobreaker
 ]
 examples = [
   "fastapi==0.109.*",

src/cqrs/events/event_emitter.py

@@ -1,3 +1,4 @@
+import asyncio
 import functools
 import logging
 import typing
@@ -169,26 +170,55 @@ async def _(self, event: IDomainEvent) -> typing.Sequence[IEvent]:
                 type(event).__name__,
             )
             return ()
+
+        results = await asyncio.gather(
+            *(self._process_single_handler(event, item) for item in handlers_types),
+            return_exceptions=True,
+        )
+
         follow_ups: list[IEvent] = []
-        for handler_item in handlers_types:
-            if isinstance(handler_item, EventHandlerFallback):
-                follow_ups.extend(
-                    await self._handle_with_fallback(event, handler_item),
+        for handler_item, res in zip(handlers_types, results):
+            if isinstance(res, Exception):
+                handler_name = (
+                    handler_item.primary.__name__
+                    if isinstance(handler_item, EventHandlerFallback)
+                    else handler_item.__name__
                 )
-                continue
-            handler_type = handler_item
-            handler: _H = await self._container.resolve(handler_type)
-            logger.debug(
-                "Handling Event(%s) via event handler(%s)",
-                type(event).__name__,
-                handler_type.__name__,
-            )
-            await handler.handle(event)
-            follow_ups.extend(list(handler.events))
+                logger.exception(
+                    "Error occurred while processing domain event %s in handler %s: %s",
+                    type(event).__name__,
+                    handler_name,
+                    res,
+                    exc_info=res,
+                )
+            elif isinstance(res, BaseException):
+                raise res
+            else:
+                follow_ups.extend(res)
+
         return follow_ups
 
     @emit.register(INotificationEvent)
     async def _(self, event: INotificationEvent) -> typing.Sequence[IEvent]:
         """Emit notification event: send to message broker; no follow-ups."""
         await self._send_to_broker(event)
         return ()
+
+    async def _process_single_handler(
+        self,
+        event: IDomainEvent,
+        handler_item: typing.Union[typing.Type[_H], EventHandlerFallback],
+    ) -> typing.Sequence[IEvent]:
+        """Process a single event handler and return its follow-up events."""
+        if isinstance(handler_item, EventHandlerFallback):
+            return await self._handle_with_fallback(event, handler_item)
+
+        handler_type = handler_item
+        handler: _H = await self._container.resolve(handler_type)
+        logger.debug(
+            "Handling Event(%s) via event handler(%s)",
+            type(event).__name__,
+            handler_type.__name__,
+        )
+        await handler.handle(event)
+        return list(handler.events)

tests/benchmarks/dataclasses/test_benchmark_event_handler_chain.py

@@ -112,11 +112,15 @@ def test_benchmark_event_chain_three_levels_parallel(
 ) -> None:
     """Benchmark: 1 root event -> 10 L2 -> 50 L3 (61 total), semaphore 4 (dataclass)."""
     processor = event_processor_chain_parallel
+    loop = asyncio.new_event_loop()
 
     async def run() -> None:
         await processor.emit_events([_EventL1(id_="root")])
 
-    benchmark(lambda: asyncio.run(run()))
+    try:
+        benchmark(lambda: loop.run_until_complete(run()))
+    finally:
+        loop.close()
 
 
 @pytest.fixture
@@ -132,8 +136,12 @@ def test_benchmark_event_chain_three_levels_sequential(
 ) -> None:
     """Benchmark: same 3-level chain, sequential (BFS), dataclass events."""
     processor = event_processor_chain_sequential
+    loop = asyncio.new_event_loop()
 
     async def run() -> None:
         await processor.emit_events([_EventL1(id_="root")])
 
-    benchmark(lambda: asyncio.run(run()))
+    try:
+        benchmark(lambda: loop.run_until_complete(run()))
+    finally:
+        loop.close()

tests/benchmarks/default/test_benchmark_event_handler_chain.py

@@ -102,7 +102,7 @@ def _make_processor(parallel: bool) -> EventProcessor:
 
 @pytest.fixture
 def event_processor_chain_parallel() -> EventProcessor:
-    """EventProcessor with 3-level chain, parallel, semaphore=4."""
+    """EventProcessor with 3-level chain, parallel, semaphore=4 (dataclass events)."""
     return _make_processor(parallel=True)
 
 
@@ -111,18 +111,22 @@ def test_benchmark_event_chain_three_levels_parallel(
     benchmark,
     event_processor_chain_parallel: EventProcessor,
 ) -> None:
-    """Benchmark: 1 root event -> 10 L2 -> 50 L3 (61 total), semaphore 4."""
+    """Benchmark: 1 root event -> 10 L2 -> 50 L3 (61 total), semaphore 4 (dataclass)."""
     processor = event_processor_chain_parallel
+    loop = asyncio.new_event_loop()
 
     async def run() -> None:
         await processor.emit_events([_EventL1(id="root")])
 
-    benchmark(lambda: asyncio.run(run()))
+    try:
+        benchmark(lambda: loop.run_until_complete(run()))
+    finally:
+        loop.close()
 
 
 @pytest.fixture
 def event_processor_chain_sequential() -> EventProcessor:
-    """EventProcessor with 3-level chain, sequential."""
+    """EventProcessor with 3-level chain, sequential (dataclass events)."""
     return _make_processor(parallel=False)
 
 
@@ -131,10 +135,14 @@ def test_benchmark_event_chain_three_levels_sequential(
     benchmark,
     event_processor_chain_sequential: EventProcessor,
 ) -> None:
-    """Benchmark: same 3-level chain, sequential (BFS)."""
+    """Benchmark: same 3-level chain, sequential (BFS), dataclass events."""
     processor = event_processor_chain_sequential
+    loop = asyncio.new_event_loop()
 
     async def run() -> None:
         await processor.emit_events([_EventL1(id="root")])
 
-    benchmark(lambda: asyncio.run(run()))
+    try:
+        benchmark(lambda: loop.run_until_complete(run()))
+    finally:
+        loop.close()

tests/unit/test_event_emitter_handler_events.py

@@ -1,5 +1,6 @@
 """Unit tests for EventEmitter returning follow-up events from handler.events."""
 
+import asyncio
 import typing
 
 import pydantic
@@ -122,3 +123,107 @@ async def resolve(self, type_: type) -> EventHandler[IEvent]:
     assert len(follow_ups) == 2
     xs = [e.x for e in follow_ups]  # type: ignore[attr-defined]
     assert "a_1" in xs and "a_2" in xs
+
+
+async def test_event_emitter_executes_handlers_in_parallel() -> None:
+    """Arrange: one event with two handlers that wait on each other. Act: emit in parallel. Assert: completes without deadlock."""
+
+    class _ParallelEvent(DomainEvent, frozen=True):
+        id: str
+
+    handler_1_started = asyncio.Event()
+    handler_2_started = asyncio.Event()
+    handler_1_completed = asyncio.Event()
+    handler_2_completed = asyncio.Event()
+
+    class _ParallelHandler1(EventHandler[_ParallelEvent]):
+        async def handle(self, event: _ParallelEvent) -> None:
+            handler_1_started.set()
+            await asyncio.wait_for(handler_2_started.wait(), timeout=1.0)
+            handler_1_completed.set()
+
+    class _ParallelHandler2(EventHandler[_ParallelEvent]):
+        async def handle(self, event: _ParallelEvent) -> None:
+            handler_2_started.set()
+            await asyncio.wait_for(handler_1_started.wait(), timeout=1.0)
+            handler_2_completed.set()
+
+    h1 = _ParallelHandler1()
+    h2 = _ParallelHandler2()
+
+    event_map = EventMap()
+    event_map.bind(_ParallelEvent, _ParallelHandler1)
+    event_map.bind(_ParallelEvent, _ParallelHandler2)
+
+    class C:
+        async def resolve(self, type_: type) -> EventHandler[IEvent]:
+            if type_ is _ParallelHandler1:
+                return h1  # type: ignore[return-value]
+            return h2  # type: ignore[return-value]
+
+    emitter = EventEmitter(
+        event_map=event_map,
+        container=C(),  # type: ignore[arg-type]
+    )
+
+    # If executed sequentially, this call will deadlock and raise asyncio.TimeoutError
+    await emitter.emit(_ParallelEvent(id="p1"))
+
+    assert handler_1_started.is_set()
+    assert handler_2_started.is_set()
+    assert handler_1_completed.is_set()
+    assert handler_2_completed.is_set()
+
+
+async def test_event_emitter_continues_execution_when_handler_raises_exception() -> None:
+    """Arrange: two handlers where handler 1 fails. Act: emit. Assert: handler 2 completes without flow crashing."""
+
+    class _SampleEvent(DomainEvent, frozen=True):
+        id: str
+
+    handler_1_executed = asyncio.Event()
+    handler_2_executed = asyncio.Event()
+
+    class _FailingHandler(EventHandler[_SampleEvent]):
+        async def handle(self, event: _SampleEvent) -> None:
+            handler_1_executed.set()
+            raise RuntimeError("Database connection lost")
+
+    class _HealthyHandler(EventHandler[_SampleEvent]):
+        def __init__(self) -> None:
+            self._out: list[IEvent] = []
+
+        @property
+        def events(self) -> typing.Sequence[IEvent]:
+            return tuple(self._out)
+
+        async def handle(self, event: _SampleEvent) -> None:
+            # Brief pause to ensure concurrency while handler 1 fails
+            await asyncio.sleep(0.01)
+            handler_2_executed.set()
+            self._out.append(_EventB(id="success_" + event.id))
+
+    failing_h = _FailingHandler()
+    healthy_h = _HealthyHandler()
+
+    event_map = EventMap()
+    event_map.bind(_SampleEvent, _FailingHandler)
+    event_map.bind(_SampleEvent, _HealthyHandler)
+
+    class Container:
+        async def resolve(self, type_: type) -> EventHandler[IEvent]:
+            if type_ is _FailingHandler:
+                return failing_h  # type: ignore[return-value]
+            return healthy_h  # type: ignore[return-value]
+
+    emitter = EventEmitter(
+        event_map=event_map,
+        container=Container(),  # type: ignore[arg-type]
+    )
+    follow_ups = await emitter.emit(_SampleEvent(id="123"))
+
+    assert handler_1_executed.is_set()
+    assert handler_2_executed.is_set()
+    assert len(follow_ups) == 1
+    assert isinstance(follow_ups[0], _EventB)
+    assert follow_ups[0].id_ == "success_123"  # type: ignore[attr-defined]

tests/unit/test_event_fallback.py

@@ -102,8 +102,7 @@ async def test_event_fallback_failure_exceptions_only_matching_triggers_fallback
     container: Container[Any] = _TestEventContainer()
     emitter = EventEmitter(event_map=event_map, container=container)
 
-    with pytest.raises(RuntimeError, match="Primary failed"):
-        await emitter.emit(SampleEvent(id="e1"))
+    await emitter.emit(SampleEvent(id="e1"))
 
     assert container._primary.called
     assert not container._fallback.called