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."""