From d0be70b85af68a31573751f16aae70e4e66f40d1 Mon Sep 17 00:00:00 2001 From: Fernando Macedo Date: Tue, 13 Oct 2026 17:21:15 -0300 Subject: [PATCH] fix: answer callers of events dropped when the machine terminates Once the machine reaches a top-level final state, external events still queued are dropped. Their futures were never resolved, so a coroutine awaiting send() of one hung forever on the async engine. The processing loop now drains the external queue when the machine stops and answers each caller as the stopped machine answers a new event: None, or TransitionNotAllowed when allow_event_without_transition is False. A machine stopped during a macrostep, by an eventless or internal transition to a top-level final state or by a top-level parallel state completing, still took one more queued external event in the same iteration; the loop now stops there. Fixes #678 Signed-off-by: Fernando Macedo --- docs/releases/3.2.2.md | 10 +- statemachine/engines/async_.py | 5 + statemachine/engines/base.py | 21 ++++ statemachine/engines/sync.py | 5 + tests/test_async_futures.py | 215 +++++++++++++++++++++++++++++++++ 5 files changed, 255 insertions(+), 1 deletion(-) diff --git a/docs/releases/3.2.2.md b/docs/releases/3.2.2.md index 02c26e82..07db4213 100644 --- a/docs/releases/3.2.2.md +++ b/docs/releases/3.2.2.md @@ -413,7 +413,15 @@ engines. Eventless transitions are now checked after every external event instea batch, so a burst of queued external events costs one guard evaluation each, and an event queued behind one that reaches a final state is no longer processed. -[#668](https://github.com/fgmacedo/python-statemachine/pull/668). +Such an event, delayed or not, is dropped when the machine terminates, and a coroutine awaiting +its `send()` gets the answer the terminated machine gives to a new event: `None`, or +`TransitionNotAllowed` for that event when `allow_event_without_transition` is `False`, as on a +`StateMachine`. This also applies when the machine terminates because every region of a +top-level parallel state reached a final state. On v3.2.1 a coroutine awaiting a delayed event +still pending when the machine terminated was never answered. + +[#668](https://github.com/fgmacedo/python-statemachine/pull/668), +[#678](https://github.com/fgmacedo/python-statemachine/issues/678). ### Graphviz errors reported only the exit code diff --git a/statemachine/engines/async_.py b/statemachine/engines/async_.py index 4c17a0e1..d4908764 100644 --- a/statemachine/engines/async_.py +++ b/statemachine/engines/async_.py @@ -361,12 +361,17 @@ async def processing_loop(self, caller_future: "asyncio.Future[object] | None" = # Spawn invoke handlers for states entered during this macrostep. await self._invoke_manager.spawn_pending_async() self._check_root_final_state() + if not self.running: + break took_external_events, first_result = await self._take_due_external_event( first_result ) took_events = took_events or took_external_events + if not self.running: + self._drop_external_events() + except Exception as exc: if caller_future is not None: # Route the exception to the caller's future if still pending. diff --git a/statemachine/engines/base.py b/statemachine/engines/base.py index 81cafdfa..b9ad56cc 100644 --- a/statemachine/engines/base.py +++ b/statemachine/engines/base.py @@ -105,6 +105,13 @@ def clear(self): with self.queue.mutex: self.queue.queue.clear() + def drain(self) -> list[TriggerData]: + """Remove every trigger from the queue and return those not cancelled.""" + with self.queue.mutex: + triggers = [trigger for trigger in self.queue.queue if not trigger.cancelled] + self.queue.queue.clear() + return triggers + def reject_futures(self, exc: Exception): """Reject all unresolved futures in the queue. @@ -180,6 +187,20 @@ def put(self, trigger_data: TriggerData, internal: bool = False): def clear(self): self.external_queue.clear() + def _drop_external_events(self): + """Drop the external events left once the machine has stopped, answering a caller + awaiting one as the stopped machine answers a new event.""" + for trigger_data in self.external_queue.drain(): + future = trigger_data.future + if future is None or future.done(): + continue + if self.sm.allow_event_without_transition: + future.set_result(None) + else: + future.set_exception( + TransitionNotAllowed(trigger_data.event, self.sm.configuration) + ) + def cancel_event(self, send_id: str): """Cancel the event with the given send_id.""" self.external_queue.cancel(send_id) diff --git a/statemachine/engines/sync.py b/statemachine/engines/sync.py index d4bcc334..fea3606e 100644 --- a/statemachine/engines/sync.py +++ b/statemachine/engines/sync.py @@ -75,11 +75,16 @@ def processing_loop(self): # Spawn invoke handlers for states entered during this macrostep. self._invoke_manager.spawn_pending_sync() self._check_root_final_state() + if not self.running: + break took_external_events, first_result = self._take_due_external_event( first_result ) took_events = took_events or took_external_events + + if not self.running: + self._drop_external_events() finally: self._processing.release() self._debug("%s Processing loop ended", self._log_id) diff --git a/tests/test_async_futures.py b/tests/test_async_futures.py index 661ab105..66612230 100644 --- a/tests/test_async_futures.py +++ b/tests/test_async_futures.py @@ -12,9 +12,12 @@ import pytest from statemachine.engines.base import EventQueue from statemachine.event_data import TriggerData +from statemachine.exceptions import TransitionNotAllowed +from statemachine import Event from statemachine import State from statemachine import StateChart +from statemachine import StateMachine # --------------------------------------------------------------------------- # Fixtures / helpers @@ -324,6 +327,218 @@ async def fn2(): assert str(errors["fn2"]) == "noop2 is not allowed" +def mount_doom_quest(base): + class MountDoom(base): + road = State(initial=True) + orodruin = State() + destroyed = State(final=True) + + climb = road.to(orodruin) | orodruin.to.itself() + cast_ring = road.to(destroyed) | orodruin.to(destroyed) + + def __init__(self, *args, **kwargs): + self.ring_falls = asyncio.Event() + super().__init__(*args, **kwargs) + + async def on_cast_ring(self): + await self.ring_falls.wait() + return "destroyed" + + return MountDoom + + +def shire_journey(base): + class Shire(base): + bag_end = State(initial=True) + rivendell = State() + grey_havens = State(final=True) + + journey = bag_end.to(rivendell) + long_journey = Event(bag_end.to(rivendell), delay=60000) + sail = bag_end.to(grey_havens, on="say_goodbye") | rivendell.to(grey_havens) + + def __init__(self, *args, **kwargs): + self.journeys = 0 + super().__init__(*args, **kwargs) + + def say_goodbye(self): + self.journey.put() + self.long_journey.put() + + def on_enter_rivendell(self): + self.journeys += 1 + + return Shire + + +async def start_casting_the_ring(sm): + """Start ``cast_ring`` in a task that holds the processing lock until ``ring_falls``.""" + await sm.activate_initial_state() + holder = asyncio.ensure_future(sm.send("cast_ring")) + await asyncio.sleep(0) + return holder + + +async def queue_behind(sm, *sends): + waiters = [asyncio.ensure_future(sm.send(event, **kwargs)) for event, kwargs in sends] + await asyncio.sleep(0) + return waiters + + +class TestEventsDroppedAtFinalState: + """Events still queued when the machine reaches a top-level final state are not run, + and a coroutine awaiting one gets what a terminated machine answers to a new event. + + See: https://github.com/fgmacedo/python-statemachine/issues/678 + """ + + @pytest.mark.timeout(5) + @pytest.mark.asyncio() + async def test_statechart_callers_get_none(self): + sm = mount_doom_quest(StateChart)() + holder = await start_casting_the_ring(sm) + waiters = await queue_behind(sm, ("climb", {}), ("climb", {}), ("climb", {"delay": 60000})) + + sm.ring_falls.set() + + assert await holder == "destroyed" + assert await asyncio.gather(*waiters) == [None, None, None] + assert sm.configuration_values == {"destroyed"} + + @pytest.mark.timeout(5) + @pytest.mark.asyncio() + async def test_statemachine_callers_get_transition_not_allowed(self): + sm = mount_doom_quest(StateMachine)() + holder = await start_casting_the_ring(sm) + waiters = await queue_behind(sm, ("climb", {}), ("climb", {}), ("climb", {"delay": 60000})) + + sm.ring_falls.set() + + assert await holder == "destroyed" + results = await asyncio.gather(*waiters, return_exceptions=True) + assert all(isinstance(result, TransitionNotAllowed) for result in results) + assert [result.event.id for result in results] == ["climb", "climb", "climb"] + assert sm.configuration_values == {"destroyed"} + + @pytest.mark.timeout(5) + @pytest.mark.asyncio() + async def test_each_caller_gets_its_own_event(self): + sm = mount_doom_quest(StateMachine)() + holder = await start_casting_the_ring(sm) + waiters = await queue_behind(sm, ("climb", {}), ("cast_ring", {})) + + sm.ring_falls.set() + + assert await holder == "destroyed" + results = await asyncio.gather(*waiters, return_exceptions=True) + assert [result.event.id for result in results] == ["climb", "cast_ring"] + + @pytest.mark.timeout(5) + @pytest.mark.asyncio() + async def test_cancelled_caller_does_not_stop_the_others(self): + sm = mount_doom_quest(StateChart)() + holder = await start_casting_the_ring(sm) + gives_up, *waiters = await queue_behind( + sm, ("climb", {}), ("climb", {}), ("climb", {"delay": 60000}) + ) + gives_up.cancel() + await asyncio.sleep(0) + + sm.ring_falls.set() + + assert await holder == "destroyed" + assert await asyncio.gather(*waiters) == [None, None] + assert gives_up.cancelled() + + @pytest.mark.timeout(5) + @pytest.mark.parametrize("base", [StateChart, StateMachine]) + async def test_events_queued_behind_final_state_leave_no_trigger(self, sm_runner, base): + Shire = shire_journey(base) + sm = await sm_runner.start(Shire) + await sm_runner.send(sm, "sail") + + assert sm.configuration_values == {"grey_havens"} + assert sm.journeys == 0 + assert sm._engine.external_queue.is_empty() + + @pytest.mark.timeout(5) + async def test_statechart_send_after_final_state_leaves_no_trigger(self, sm_runner): + sm = await sm_runner.start(shire_journey(StateChart)) + await sm_runner.send(sm, "sail") + + assert await sm_runner.send(sm, "journey") is None + assert await sm_runner.send(sm, "journey", delay=60000) is None + assert sm.journeys == 0 + assert sm._engine.external_queue.is_empty() + + @pytest.mark.timeout(5) + @pytest.mark.parametrize("base", [StateChart, StateMachine]) + async def test_events_queued_behind_root_parallel_completion_are_dropped( + self, sm_runner, base + ): + class LastAlliance(base): + class war(State.Parallel): + class mordor(State.Compound): + besieged = State(initial=True) + fallen = State(final=True) + victory = besieged.to(fallen) + + class barad_dur(State.Compound): + standing = State(initial=True) + ruined = State(final=True) + victory = standing.to(ruined) + + rally = war.to.itself() + + def __init__(self, *args, **kwargs): + self.rallies = 0 + super().__init__(*args, **kwargs) + + def on_victory(self): + self.rally.put() + + def on_rally(self): + self.rallies += 1 + + sm = await sm_runner.start(LastAlliance) + await sm_runner.send(sm, "victory") + + assert sm.is_terminated + assert sm.rallies == 0 + assert sm._engine.external_queue.is_empty() + + @pytest.mark.timeout(5) + @pytest.mark.parametrize("base", [StateChart, StateMachine]) + async def test_events_queued_behind_an_eventless_final_transition_are_dropped( + self, sm_runner, base + ): + class Fellowship(base): + rivendell = State(initial=True) + moria = State() + lothlorien = State(final=True) + + enter_moria = rivendell.to(moria, on="hear_drums") + moria.to(lothlorien) + rest = rivendell.to(lothlorien) + + def __init__(self, *args, **kwargs): + self.rests = 0 + super().__init__(*args, **kwargs) + + def hear_drums(self): + self.rest.put() + + def on_rest(self): + self.rests += 1 + + sm = await sm_runner.start(Fellowship) + await sm_runner.send(sm, "enter_moria") + + assert sm.configuration_values == {"lothlorien"} + assert sm.rests == 0 + assert sm._engine.external_queue.is_empty() + + class TestEventQueueRejectFutures: """Unit tests for EventQueue.reject_futures."""