Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions docs/invoke.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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)
Expand Down
34 changes: 23 additions & 11 deletions statemachine/invoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -401,11 +401,6 @@
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):
Expand All @@ -415,6 +410,9 @@
# 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(
Expand Down Expand Up @@ -484,11 +482,6 @@
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.
Expand All @@ -497,6 +490,9 @@
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(
Expand All @@ -506,6 +502,19 @@
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):
Expand Down Expand Up @@ -559,7 +568,7 @@

# --- Helpers ---

def handle_external_event(self, trigger_data) -> None:

Check failure on line 571 in statemachine/invoke.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this function to reduce its Cognitive Complexity from 16 to the 15 allowed.

See more on https://sonarcloud.io/project/issues?id=fgmacedo_python-statemachine&issues=AaE49qto3RkC5tmjpNf4&open=AaE49qto3RkC5tmjpNf4&pullRequest=698
"""Run finalize blocks and autoforward for active invocations.

Called by the engine before processing each external event.
Expand Down Expand Up @@ -589,7 +598,10 @@
# 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.
Expand Down
149 changes: 149 additions & 0 deletions tests/test_invoke.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,18 @@
"""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

from statemachine import Event
from statemachine import State
from statemachine import StateChart
from statemachine import StateMachine


class TestInvokeSimpleCallable:
Expand Down Expand Up @@ -295,6 +298,152 @@
assert "error_state" in sm.configuration_values


@pytest.fixture()

Check warning on line 301 in tests/test_invoke.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Remove empty parentheses from this decorator.

See more on https://sonarcloud.io/project/issues?id=fgmacedo_python-statemachine&issues=AaE49qor3RkC5tmjpNf3&open=AaE49qor3RkC5tmjpNf3&pullRequest=698
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."""

Expand Down
Loading