How to safely handle exceptions in async context?

[ale18V]

• Python State Machine version: 2.5.0
• Python version: 3.12
• Operating System: Linux

Description

I ran into this issue: there may be multiple coroutines issuing events to the machine, yet if a coroutine triggers an invalid transition, a different coroutine may receive the exception instead.

Here is a simple example:

from statemachine import Event, State, StateMachine
import asyncio


class Test(StateMachine):
    INITIAL = State(initial=True)
    FINAL = State(final=True)

    noop = Event(INITIAL.to(FINAL))

    @noop.on
    async def do_nothing(self, name):
        await asyncio.sleep(10)
        print(f"Did nothing via {name}")


test = Test()


if __name__ == "__main__":

    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()

    async def fn1():
        await test.noop("fn1")

# Try/Except block is not working here
    async def fn2():
        try:
            await test.noop("fn2")
        except Exception as e:
            print(e)

    loop.create_task(fn1())
    loop.create_task(fn2())
    loop.run_forever()

Output is:

   ➜ python3 test.py 
Did nothing via fn1
Task exception was never retrieved
future: <Task finished name='Task-1' coro=<fn1() done> 
exception=TransitionNotAllowed("Can't Noop when in Final.")>
Traceback (most recent call last):
  File "test.py", line 29, in fn1
    await test.noop("fn1")
  File "/python3.12/site-packages/statemachine/engines/async_.py", line 66, in processing_loop
    result = await self._trigger(trigger_data)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/python3.12/site-packages/statemachine/engines/async_.py", line 96, in _trigger
    raise TransitionNotAllowed(trigger_data.event, state)
statemachine.exceptions.TransitionNotAllowed: Can't Noop when in Final.

It is clear that the invalid transition is triggered in fn2 yet the exception is thrown in fn1

[ale18V]

While this is just a proof of concept, in my original use-case there are some places where I know that triggering a transition will not result in an exception and some other places where I know a try/catch is needed. However, if the exception is thrown in an unpredictable place I would be forced to implement exception handling everywhere and this makes stacking events complex.

[fgmacedo]

Is just ignoring events that didn't result in transitions a valid option for your use case?

If so, you can just create the SM instance with Test(allow_event_without_transition=True)

Ref: https://python-statemachine.readthedocs.io/en/latest/api.html#statemachine

[ale18V]

Hi, thanks for the quick answer and suggestion. This seems to work well for invalid transitions. Unluckily this is not a viable option for me however, as I was using transition's validators to throw meaningful exceptions that need to be catched by the same coroutine issuing the event.

Concretely:

class Test(StateMachine):
    INITIAL = State(initial=True)
    FINAL = State()

    noop = Event(INITIAL.to(FINAL))
    noop2 = Event(INITIAL.to(FINAL) | FINAL.to.itself())

    @noop2.on
    @noop.on
    async def do_nothing(self, name):
        await asyncio.sleep(5)
        print(f"Did nothing via {name}")

    @noop2.validators
    def raise_exception(self):
        # Do some checks ...
        # If checks fail
        print("noop2 is not allowed", flush=True)
        raise ValueError("noop2 is not allowed")

test = Test(allow_event_without_transition=True)

try:
    loop = asyncio.get_running_loop()
except RuntimeError:
    loop = asyncio.new_event_loop()

async def fn1():
    await test.noop("fn1")

async def fn2():
    try:
        await test.noop2("fn2")
    except ValueError as e:
        # Some logic
        print(e)
loop.create_task(fn1())
loop.create_task(fn2())
loop.run_forever()

However validator exceptions too are not guaranteed to be thrown in the same coroutine so I guess i'll have to revaluate my application's flow

[fgmacedo]

I'm afraid that currently this is not supported out of the box. The problem is that the event's queue is consumed by a "critical section" that only the first caller can enter, and then the queue is consumed to exhaustion. While the critical section is closed by the first caller, other callers don't block and just put the event into the queue returning immediately.

So that's why eventually the first caller gets the second event.

But our thread gives me an idea that I can work on.... I don't know if will work and probably will break the public API of the library, but looks like promising: For each caller that don't get the chance to enter the critical section, instead of returning None, I could return a Promisse-like object, like a asyncio.Future or concurrent.futures.Future. And when the event is consumed, set the future.set_result(result) or future.set_exception(exception).

[ale18V]

Seems like a good idea. Let me know if help is wanted