From 041b733913fa967867f13b8848ba2e7b1124f5a5 Mon Sep 17 00:00:00 2001 From: Fernando Macedo Date: Wed, 14 Oct 2026 02:49:27 -0300 Subject: [PATCH] fix: report errors raised while an invocation processes its done.invoke When an invoke handler finishes while the machine is idle, its own thread or task sends done.invoke and runs the macrostep. An error raised there was swallowed when the transition exited the invoke's state, and otherwise re-sent as error.execution even on StateMachine. done.invoke is now sent outside the handler's error path: the sync engine lets the error propagate out of the handler thread, and the async engine passes it to the event loop's exception handler. Errors raised by on_finalize go through the engine's error handling, so they become error.execution on StateChart and propagate on StateMachine. Refs #692 Signed-off-by: Fernando Macedo --- docs/invoke.md | 11 ++- statemachine/invoke.py | 34 +++++++--- tests/test_invoke.py | 149 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 181 insertions(+), 13 deletions(-) diff --git a/docs/invoke.md b/docs/invoke.md index ba4c22f7..a21be177 100644 --- a/docs/invoke.md +++ b/docs/invoke.md @@ -425,8 +425,8 @@ True ## Error handling -If an invoke handler raises an exception, `error.execution` is sent to the machine's -internal queue (when `catch_errors_as_events=True`, the default for `StateChart`). You can +If an invoke handler raises an exception while its state is active, `error.execution` is +sent to the machine's external queue, whatever `catch_errors_as_events` is set to. You can handle it with a transition for `error.execution`: ```py @@ -451,6 +451,13 @@ True ``` +An error raised while the machine processes `done.invoke`, by a callback or by the +handler's `on_finalize`, is handled like an error in any other step: it becomes `error.execution` when `catch_errors_as_events=True`, +and propagates otherwise. When the machine is idle as the handler finishes, the handler's +own thread or task processes `done.invoke`, and nobody awaits it: the sync engine lets the +error propagate out of the thread, to `threading.excepthook`, and the async engine passes +it to the event loop's exception handler. + ## Multiple invokes ### Independent invokes (one event each) diff --git a/statemachine/invoke.py b/statemachine/invoke.py index 798066b3..1959c7d1 100644 --- a/statemachine/invoke.py +++ b/statemachine/invoke.py @@ -401,11 +401,6 @@ def _run_sync_handler( result = handler.run(ctx) else: result = callback.call(ctx=ctx, machine=ctx.machine, **ctx.kwargs) - if not ctx.cancelled.is_set(): - self.sm.send( - f"done.invoke.{ctx.invokeid}", - data=result, - ) except Exception as e: if not ctx.cancelled.is_set(): # Intentionally using the external queue (no internal=True): @@ -415,6 +410,9 @@ def _run_sync_handler( # active (the internal queue is only drained within a macrostep). # This matches done.invoke, which also uses the external queue. self.sm.send("error.execution", error=e) + else: + if not ctx.cancelled.is_set(): + self.sm.send(f"done.invoke.{ctx.invokeid}", data=result) finally: invocation.terminated = True self._debug( @@ -484,11 +482,6 @@ async def _run_async_handler( result = await loop.run_in_executor( None, lambda: callback.call(ctx=ctx, machine=ctx.machine, **ctx.kwargs) ) - if not ctx.cancelled.is_set(): - await self.sm.send( - f"done.invoke.{ctx.invokeid}", - data=result, - ) except asyncio.CancelledError: # Intentionally swallowed: the owning state was exited, so this # invocation was cancelled — there is nothing to propagate. @@ -497,6 +490,9 @@ async def _run_async_handler( if not ctx.cancelled.is_set(): # External queue — see comment in _run_sync_handler. await self.sm.send("error.execution", error=e) + else: + if not ctx.cancelled.is_set(): + await self._send_done_async(ctx.invokeid, result) finally: invocation.terminated = True self._debug( @@ -506,6 +502,19 @@ async def _run_async_handler( ctx.cancelled.is_set(), ) + async def _send_done_async(self, invokeid: str, result: Any): + """Send ``done.invoke`` and report an error raised while processing it. + + The invocation's task drives the macrostep when the machine is idle, and no caller + awaits that task, so the error goes to the event loop's exception handler. + """ + try: + await self.sm.send(f"done.invoke.{invokeid}", data=result) + except Exception as e: + asyncio.get_running_loop().call_exception_handler( + {"message": f"Error while processing done.invoke.{invokeid}", "exception": e} + ) + # --- Cancel --- def _cancel(self, invokeid: str): @@ -589,7 +598,10 @@ def handle_external_event(self, trigger_data) -> None: # Note: finalize must run even after the invocation terminates, because # child events may still be queued when the handler thread completes. if is_from_child and hasattr(handler, "on_finalize"): - handler.on_finalize(trigger_data) + try: + handler.on_finalize(trigger_data) + except Exception as e: + self._engine._handle_error(e, trigger_data) # Autoforward: forward parent events to child (not events from child itself). # Only forward if the invocation is still running. diff --git a/tests/test_invoke.py b/tests/test_invoke.py index c9dc271c..305e34c2 100644 --- a/tests/test_invoke.py +++ b/tests/test_invoke.py @@ -1,8 +1,10 @@ """Tests for the invoke callback group.""" +import asyncio import threading import time +import pytest from statemachine.invoke import IInvoke from statemachine.invoke import InvokeContext from statemachine.invoke import invoke_group @@ -10,6 +12,7 @@ from statemachine import Event from statemachine import State from statemachine import StateChart +from statemachine import StateMachine class TestInvokeSimpleCallable: @@ -295,6 +298,152 @@ def on_enter_error_state(self, **kwargs): assert "error_state" in sm.configuration_values +@pytest.fixture() +def thread_errors(monkeypatch) -> "list[BaseException]": + errors: "list[BaseException]" = [] + monkeypatch.setattr(threading, "excepthook", lambda args: errors.append(args.exc_value)) + return errors + + +async def finish_invocation( + sm, release: threading.Event, thread_errors: "list[BaseException]" +) -> "list[BaseException]": + """Let the only invocation finish and return the errors its thread or task reported.""" + (invocation,) = sm._engine._invoke_manager._active.values() + if invocation.task is None: + release.set() + invocation.thread.join(timeout=5) + return thread_errors + loop_errors: "list[BaseException]" = [] + loop = asyncio.get_running_loop() + previous_handler = loop.get_exception_handler() + loop.set_exception_handler(lambda _loop, context: loop_errors.append(context["exception"])) + try: + release.set() + await asyncio.wait([invocation.task], timeout=5) + finally: + loop.set_exception_handler(previous_handler) + assert invocation.task.exception() is None + return loop_errors + + +class TestInvokeCompletionProcessingErrors: + """An error raised while the invocation itself processes its ``done.invoke`` is not lost. + + The invocation finishes after the machine settled, so its own thread or task processes + ``done.invoke`` and the error has no other caller to reach. + """ + + async def test_error_leaving_the_state_propagates(self, sm_runner, thread_errors): + release = threading.Event() + + class SM(StateMachine): + s1 = State(initial=True) + s2 = State(final=True) + done_invoke_s1 = s1.to(s2, on="boom") + + def on_invoke_s1(self): + release.wait(timeout=5) + + def boom(self): + raise ValueError("boom") + + sm = await sm_runner.start(SM) + errors = await finish_invocation(sm, release, thread_errors) + + assert [repr(e) for e in errors] == [repr(ValueError("boom"))] + assert sm.configuration_values == {"s1"} + + async def test_error_staying_in_the_state_propagates(self, sm_runner, thread_errors): + release = threading.Event() + + class SM(StateMachine): + s1 = State(initial=True) + s2 = State(final=True) + done_invoke_s1 = s1.to.itself(internal=True, on="boom") + error_execution = s1.to(s2) + + def on_invoke_s1(self): + release.wait(timeout=5) + + def boom(self): + raise ValueError("boom") + + sm = await sm_runner.start(SM) + errors = await finish_invocation(sm, release, thread_errors) + + assert [repr(e) for e in errors] == [repr(ValueError("boom"))] + assert sm.configuration_values == {"s1"} + + async def test_error_leaving_the_state_becomes_error_execution(self, sm_runner, thread_errors): + release = threading.Event() + + class SM(StateChart): + s1 = State(initial=True) + s2 = State() + handled = State(final=True) + done_invoke_s1 = s1.to(s2, on="boom") + error_execution = s2.to(handled) + + def on_invoke_s1(self): + release.wait(timeout=5) + + def boom(self): + raise ValueError("boom") + + sm = await sm_runner.start(SM) + errors = await finish_invocation(sm, release, thread_errors) + + assert errors == [] + assert sm.configuration_values == {"handled"} + + +class FailingFinalize: + def __init__(self, release: threading.Event): + self.release = release + + def run(self, ctx): + self.release.wait(timeout=5) + + def on_finalize(self, trigger_data): + raise ValueError("finalize") + + +class TestInvokeFinalizeErrors: + """An error raised by ``on_finalize`` is handled like an error in any other step.""" + + async def test_finalize_error_becomes_error_execution(self, sm_runner, thread_errors): + release = threading.Event() + + class SM(StateChart): + s1 = State(initial=True, invoke=FailingFinalize(release)) + s2 = State() + handled = State(final=True) + done_invoke_s1 = s1.to(s2) + error_execution = s2.to(handled) + + sm = await sm_runner.start(SM) + errors = await finish_invocation(sm, release, thread_errors) + + assert errors == [] + assert sm.configuration_values == {"handled"} + + async def test_finalize_error_propagates(self, sm_runner, thread_errors): + release = threading.Event() + + class SM(StateMachine): + s1 = State(initial=True, invoke=FailingFinalize(release)) + s2 = State(final=True) + done_invoke_s1 = s1.to(s2) + error_execution = s1.to(s2) + + sm = await sm_runner.start(SM) + errors = await finish_invocation(sm, release, thread_errors) + + assert [repr(e) for e in errors] == [repr(ValueError("finalize"))] + assert sm.configuration_values == {"s1"} + + class TestInvokeMultiple: """Multiple invokes per state — all run concurrently."""