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