fix: async engine answers callers after their macrostep and hands the loop over on cancellation

docs/async.md

@@ -152,7 +152,8 @@ processing loop at a time. The sync engine does not support this pattern.
 Under the hood, the async engine attaches an `asyncio.Future` to each
 externally enqueued event. The coroutine that acquires the processing lock
 resolves each event's future as it processes the queue. Callers that didn't
-acquire the lock simply `await` their future.
+acquire the lock `await` their future, or the release of the lock, to take
+the loop over if the coroutine holding it was cancelled.
 
 ```{note}
 Futures are created for every event sent from outside the processing loop,
@@ -165,9 +166,18 @@ model: they are enqueued and processed by the running loop.
 If an exception occurs during processing (with `catch_errors_as_events=False`),
 the exception is routed to the caller whose event raised it. Other callers
 whose events were still pending will also receive the exception, since the
-processing loop clears the queue on failure. An exception raised by the
-internal events or eventless transitions that follow an event goes only to
-the callers still pending: the caller of that event already has its result.
+processing loop clears the queue on failure. A caller is answered once the
+macrostep of its event ends, so an exception raised by the internal events or
+eventless transitions that follow its event reaches it too. When no coroutine
+awaits that event, because a callback sent it or its caller was cancelled while
+awaiting it, the exception goes to the coroutine running the processing loop.
+
+If the coroutine running the processing loop is cancelled, only that coroutine
+stops: a coroutine awaiting a queued event takes the loop over, and returns
+once the queue is empty. With no coroutine waiting, the events left in the
+queue run at the next `send()`. The caller of the event being processed when
+the cancellation hit gets a `StateMachineError`, since its callbacks were
+interrupted.
 
 
 (syncengine)=

docs/releases/3.2.2.md

@@ -74,6 +74,9 @@ fail after upgrading:
   transition. Internal events raised before a callback fails, and events sent before a failed
   lazy initial activation on the async engine, are dropped too, instead of running at the next
   event. See {ref}`release-3.2.2-macrostep`.
+- On the async engine, with `catch_errors_as_events=False`, `await send()` raises an exception
+  that escapes the internal events or eventless transitions following its event. It used to
+  return the event's result, and the exception was lost. See {ref}`release-3.2.2-async-loop`.
 - An attribute named with the `error_`, `done_state_` or `done_invoke_` prefix now registers
   two events, so `events`, `allowed_events` and `enabled_events()` list both ids. See
   {ref}`release-3.2.2-prefixed-events`.
@@ -377,18 +380,82 @@ dropped event gets that exception. Any exception escapes with `catch_errors_as_e
 as on a `StateMachine`; on a `StateChart`, one from a validator or `prepare_event` does. On
 3.2.1 the queued external events ran before those internal events and eventless transitions,
 and their callers got their own results; the exception was then raised by the sync `send()`
-that started processing, and on the async engine it reached no caller. On the async engine, as
-on 3.2.1, the coroutine whose event enabled the failing transitions gets its own event's
-result. Internal events raised by a callback before it fails are dropped too: they used to run
-at the next event. The same applies when the lazy initial activation fails on the async engine,
-where a second coroutine awaiting an event sent before it used to wait forever, and the events
-sent before it, which used to run at the next `send()`, are dropped.
+that started processing, and on the async engine it reached no caller. The coroutine whose
+event enabled the failing transitions also gets the exception, see
+{ref}`release-3.2.2-async-loop`. Internal events raised by a callback before it fails are
+dropped too: they used to run at the next event. The same applies when the lazy initial
+activation fails on the async engine, where a second coroutine awaiting an event sent before it
+used to wait forever, and the events sent before it, which used to run at the next `send()`,
+are dropped.
 
 [#668](https://github.com/fgmacedo/python-statemachine/pull/668),
 [#678](https://github.com/fgmacedo/python-statemachine/issues/678),
 [#686](https://github.com/fgmacedo/python-statemachine/issues/686),
 [#685](https://github.com/fgmacedo/python-statemachine/pull/685).
 
+(release-3.2.2-async-loop)=
+
+#### Async engine: an exception after the event, and a cancelled processing loop
+
+**An exception raised after a coroutine's event was lost.** With
+`catch_errors_as_events=False`, as on a `StateMachine`, an exception raised by the internal
+events or eventless transitions that follow an external event reached no caller on the async
+engine: `await send()` returned the event's result, and the exception was neither raised nor
+logged. The sync engine raises it from `send()`. A coroutine is now answered once the
+macrostep of its event ends, so the exception reaches it:
+
+```py
+>>> import asyncio
+
+>>> class Alarm(StateMachine):
+...     idle = State(initial=True)
+...     armed = State()
+...     ringing = State()
+...
+...     arm = idle.to(armed)
+...     trip = armed.to(ringing)
+...     reset = ringing.to(idle)
+...
+...     async def on_enter_armed(self):
+...         await self.raise_("trip")
+...
+...     async def on_trip(self):
+...         raise ValueError("sensor failure")
+
+>>> async def arm():
+...     sm = Alarm()
+...     await sm.activate_initial_state()
+...     await sm.send("arm")
+
+>>> asyncio.run(arm())
+Traceback (most recent call last):
+...
+ValueError: sensor failure
+
+```
+
+When no coroutine awaits the event whose macrostep failed, because a callback sent it or its
+caller was cancelled while awaiting it, the coroutine running the processing loop gets the
+exception, as the
+sync engine raises it from the `send()` running the loop. The callers of the events still
+queued get the exception as before. A `StateChart` is unchanged: the error becomes
+`error.execution`.
+
+**Cancelling the coroutine running the processing loop left the others waiting.** When the
+coroutine holding the processing loop was cancelled, as `asyncio.wait_for()` or a request
+timeout does, the coroutines awaiting the `send()` of events queued behind it were not
+answered until a later `send()` ran the loop again: `CancelledError` is not an `Exception`, so
+the loop's error handling did not run. The cancellation now stops only that coroutine. The
+internal events left by its macrostep are dropped, and a coroutine awaiting a queued event
+takes the processing loop over, so each gets its own result, delayed events included. Like any
+coroutine running the loop, the one taking over returns once the queue is empty. A coroutine
+awaiting the event being processed when the cancellation hit, whose callbacks were interrupted
+halfway, gets a `StateMachineError` saying so. With no coroutine waiting, the events left in
+the queue, the cancelled coroutine's own included, run at the next `send()`.
+
+[#689](https://github.com/fgmacedo/python-statemachine/issues/689),
+[#690](https://github.com/fgmacedo/python-statemachine/issues/690).
+
 (release-3.2.2-prefixed-events)=
 
 #### Prefixed events were registered under a glued id

statemachine/engines/async_.py

@@ -7,7 +7,9 @@
 from ..event_data import EventData
 from ..event_data import TriggerData
 from ..exceptions import InvalidDefinition
+from ..exceptions import StateMachineError
 from ..exceptions import TransitionNotAllowed
+from ..i18n import _
 from ..orderedset import OrderedSet
 from ..state import State
 from .base import _ERROR_EXECUTION
@@ -16,16 +18,53 @@
 from .base import _running_engine
 
 if TYPE_CHECKING:
+    from ..statemachine import StateChart
     from ..transition import Transition
 
 
+class _MacrostepCaller:
+    """The caller of the external event whose macrostep is running, answered once the
+    macrostep ends, so that an exception raised by its internal events or eventless
+    transitions still reaches it."""
+
+    def __init__(self):
+        self.future: "asyncio.Future[object] | None" = None
+        self.result: object = None
+
+    def answer(self):
+        AsyncEngine._resolve_future(self.future, self.result)
+        self.future = None
+        self.result = None
+
+    def reject(self, exc: Exception) -> bool:
+        """Reject the caller with ``exc``; returns whether a caller was still waiting."""
+        if self.future is None or self.future.done():
+            return False
+        self.future.set_exception(exc)
+        return True
+
+    def interrupt(self):
+        self.reject(
+            StateMachineError(
+                _(
+                    "The task running the processing loop was cancelled while it processed "
+                    "this event."
+                )
+            )
+        )
+
+
 class AsyncEngine(BaseEngine):
     """Async engine with full StateChart support.
 
     Mirrors :class:`SyncEngine` algorithm but uses ``async``/``await`` for callback dispatch.
     All pure-computation helpers are inherited from :class:`BaseEngine`.
     """
 
+    def __init__(self, sm: "StateChart"):
+        super().__init__(sm)
+        self._lock_waiters: "set[asyncio.Future[object]]" = set()
+
     def put(self, trigger_data: TriggerData, internal: bool = False):
         """Override to attach an asyncio.Future to a trigger put from outside the loop.
 
@@ -50,12 +89,6 @@
         if future is not None and not future.done():
             future.set_result(result)
 
-    @staticmethod
-    def _reject_future(future: "asyncio.Future[object] | None", exc: Exception):
-        """Reject a future with the given exception, if present and not yet done."""
-        if future is not None and not future.done():
-            future.set_exception(exc)
-
     def _reject_pending_futures(self, exc: Exception):
         """Reject all unresolved futures in the external queue."""
         self.external_queue.reject_futures(exc)
@@ -340,17 +373,45 @@
 
         When ``caller_future`` is provided, the caller can ``await`` it to
         receive its own event's result — even if another coroutine holds the
-        processing lock.
+        processing lock. If that coroutine is cancelled, the caller takes the loop over.
         """
-        if not self._processing.acquire(blocking=False):
-            # Another coroutine holds the lock and will process our event.
-            # Await the caller's future so we get our own result back.
-            if caller_future is not None:
-                return await caller_future
-            return None
+        while not self._processing.acquire(blocking=False):
+            if caller_future is None:
+                return None
+            await self._wait_for_result_or_release(caller_future)
+            if caller_future.done():
+                return caller_future.result()
+
+        try:
+            return await self._run_processing_loop(caller_future)
+        finally:
+            self._processing.release()
+            self._wake_lock_waiters()
+
+    async def _wait_for_result_or_release(self, caller_future: "asyncio.Future[object]"):
+        """Wait until the caller's event is answered or the processing lock is released.
 
+        A caller cancelled here cancels its future, so its event, still processed, answers
+        nobody.
+        """
+        released = caller_future.get_loop().create_future()
+        self._lock_waiters.add(released)
+        try:
+            await asyncio.wait((caller_future, released), return_when=asyncio.FIRST_COMPLETED)
+        except BaseException:
+            caller_future.cancel()
+            raise
+        finally:
+            self._lock_waiters.discard(released)
+
+    def _wake_lock_waiters(self):
+        for released in self._lock_waiters:
+            self._resolve_future(released, None)
+
+    async def _run_processing_loop(self, caller_future: "asyncio.Future[object] | None"):
         self._debug("%s Processing loop started: %s", self._log_id, self.sm.current_state_value)
         first_result = self._sentinel
+        macrostep = _MacrostepCaller()
         try:
             with _processing_step(self):
                 took_events = True
@@ -360,37 +421,35 @@
 
                     # Spawn invoke handlers for states entered during this macrostep.
                     await self._invoke_manager.spawn_pending_async()
+                    macrostep.answer()
                     self._check_root_final_state()
                     if not self.running:
                         break
 
                     took_external_events, first_result = await self._take_due_external_event(
-                        first_result
+                        first_result, macrostep
                     )
                     took_events = took_events or took_external_events
 
+                macrostep.answer()
                 if not self.running:
                     self._drop_external_events()
 
         except Exception as exc:
+            answered = caller_future is None or caller_future.done()
             self._reject_pending_futures(exc)
             self.clear()
-            if caller_future is not None:
-                # Route the exception to the caller's future if still pending.
-                # If already resolved (caller's own event succeeded before a
-                # later event failed), suppress the exception — the caller will
-                # get their successful result via ``await future`` below, and
-                # the failing event's exception was already routed to *its*
-                # caller's future by ``_reject_future(event_future, ...)``.
-                self._reject_future(caller_future, exc)
-            else:
+            if not macrostep.reject(exc) and answered:
                 raise
-        finally:
-            self._processing.release()
+        except BaseException:
+            self.internal_queue.clear()
+            macrostep.interrupt()
+            if caller_future is not None:
+                caller_future.cancel()
+            raise
 
         self._debug("%s Processing loop ended", self._log_id)
         result = first_result if first_result is not self._sentinel else None
-        # If the caller has a future, await it (already resolved by now).
         if caller_future is not None:
             # Resolve the future if it wasn't processed (e.g. machine terminated).
             self._resolve_future(caller_future, result)
@@ -424,11 +483,12 @@
                 took_events = True
                 await self.microstep(list(enabled_transitions), internal_event)
 
-    async def _take_due_external_event(self, first_result):
+    async def _take_due_external_event(self, first_result, macrostep: "_MacrostepCaller"):
         """Take the external event at the head of the queue, if it is due.
 
         One at a time, so the macrostep that follows runs the internal events and eventless
-        transitions it enabled before the next external event is taken.
+        transitions it enabled before the next external event is taken. The event's caller is
+        left in ``macrostep``, to be answered once that macrostep ends.
 
         Returns whether the queue held an event, and ``first_result`` updated with the result
         of the first event that enabled a transition.
@@ -458,35 +518,29 @@
             await self._enter_states(transitions, external_event, OrderedSet(), OrderedSet())
             return True, first_result
 
+        macrostep.future = external_event.future
         result = await self._take_external_event(external_event)
+        if result is not self._sentinel:
+            macrostep.result = result
         if first_result is self._sentinel:
             first_result = result
         return True, first_result
 
     async def _take_external_event(self, external_event: TriggerData):
-        """Run one external event and settle its future.
+        """Run one external event.
 
         Returns the event's microstep result, or the sentinel when it enabled no transition.
         """
         # Finalize + autoforward for active invocations
         self._invoke_manager.handle_external_event(external_event)
 
-        event_future = external_event.future
-        try:
-            enabled_transitions = await self.select_transitions(external_event)
-            self._debug("%s Enabled transitions: %s", self._log_id, enabled_transitions)
-            if enabled_transitions:
-                result = await self.microstep(list(enabled_transitions), external_event)
-                self._resolve_future(event_future, result)
-                return result
-            if not self.sm.allow_event_without_transition:
-                raise TransitionNotAllowed(external_event.event, self.sm.configuration)
-            # Event allowed but no transition — resolve with None
-            self._resolve_future(event_future, None)
-            return self._sentinel
-        except Exception as exc:
-            self._reject_future(event_future, exc)
-            raise
+        enabled_transitions = await self.select_transitions(external_event)
+        self._debug("%s Enabled transitions: %s", self._log_id, enabled_transitions)
+        if enabled_transitions:
+            return await self.microstep(list(enabled_transitions), external_event)
+        if not self.sm.allow_event_without_transition:
+            raise TransitionNotAllowed(external_event.event, self.sm.configuration)
+        return self._sentinel
 
     async def enabled_events(self, *args, **kwargs):
         sm = self.sm