From 7714bed6b87f4746e4ef6b5434d9aa60a5f8604b Mon Sep 17 00:00:00 2001 From: Robert-Jan Huijsman <22160949+rjhuijsman@users.noreply.github.com> Date: Wed, 9 Sep 2026 17:12:02 +0000 Subject: [PATCH 1/2] `state_managers`: shield a transaction's outcome from its waiters Keeps two phase commit's invariant that a participant transaction is resolved exactly once, so a participant always releases its state's lock and drops its participant entry instead of abandoning both. Before this change, `StateManager.Transaction.__await__` awaited the shared `_committed` future directly. `asyncio` cancels the future a task is suspended on when that task itself is cancelled, so any cancelled waiter -- `_load()` waiting on an ongoing prepared transaction, or a duplicate idempotent call waiting on the transaction it matched -- took the transaction's own outcome down with it. `finished()` then reported the transaction done while `commit()` and `abort()` could no longer resolve it, and the next `set_result()` raised from inside a section the framework marks exception-intolerant: ##### WOW! YOU'VE FOUND A BUG IN REBOOT! ##### Raised exception in critical exception-intolerant Abort section: : 'invalid state' That raise lands before `_complete_participant_transaction()`, so the participant entry stayed in `_participant_transactions` and the state's `Lock` kept a holder for the life of the process, leaving every later exclusive acquisition on that state waiting forever. Awaiting `asyncio.shield(self._committed)` keeps a cancellation with the waiter it belongs to, which is what the coordinator's participants future already does. - Add `TransactionTest`, covering both the cancelled waiter and the waiters alongside it. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_014L8UDXfhAJNKDLuEHN42AX --- reboot/aio/state_managers.py | 10 +++- tests/reboot/state_manager_tests.py | 81 +++++++++++++++++++++++++++++ 2 files changed, 90 insertions(+), 1 deletion(-) diff --git a/reboot/aio/state_managers.py b/reboot/aio/state_managers.py index 74fa62e1..990e4482 100644 --- a/reboot/aio/state_managers.py +++ b/reboot/aio/state_managers.py @@ -1283,7 +1283,15 @@ def __await__(self): """Awaits for transaction to finish, i.e., aborted or committed.""" async def closure(): - await self._committed + # Shield so that cancelling a waiter cancels only that + # waiter. `_committed` is shared with every other + # waiter and with the participant paths that resolve + # it, so a cancellation reaching it would leave + # `finished()` reporting a transaction that `commit()` + # and `abort()` can no longer resolve. The + # coordinator's participants future is shielded for + # the same reason. + await asyncio.shield(self._committed) return closure().__await__() diff --git a/tests/reboot/state_manager_tests.py b/tests/reboot/state_manager_tests.py index b8db8840..0db920eb 100644 --- a/tests/reboot/state_manager_tests.py +++ b/tests/reboot/state_manager_tests.py @@ -1646,6 +1646,87 @@ async def shared() -> None: self.assertFalse(lock.is_locked()) +class TransactionTest(unittest.IsolatedAsyncioTestCase): + """Unit tests for `StateManager.Transaction`, whose outcome every + interested party learns by awaiting the transaction itself.""" + + def _transaction(self) -> StateManager.Transaction: + state_type = StateTypeName('test.v1.Transactional') + state_ref = StateRef.from_id(state_type, 'test-1234') + root_id = uuid.uuid4() + return StateManager.Transaction( + transaction_ids=[root_id], + coordinator_state_type=state_type, + coordinator_state_ref=state_ref, + state_type=state_type, + state_ref=state_ref, + tasks_dispatcher=unittest.mock.MagicMock(spec=TasksDispatcher), + mode=Lock.Mode.SHARED, + # A first attempt is as old as its root transaction id. + age=root_id, + ) + + async def _waiter( + self, + transaction: StateManager.Transaction, + ) -> asyncio.Task: + """Returns a task that is awaiting `transaction`.""" + task = asyncio.create_task(self._await_transaction(transaction)) + # Runs the task up to its await on the transaction. + await asyncio.sleep(0) + return task + + async def _await_transaction( + self, + transaction: StateManager.Transaction, + ) -> None: + await transaction + + async def test_cancelled_waiter_leaves_transaction_unfinished( + self, + ) -> None: + """A task cancelled while awaiting a transaction takes the + cancellation with it, leaving the transaction free to be + aborted (or committed) afterwards. + + A cancellation that reached the transaction's own future + instead would make it claim to be finished while the + participant's `abort()` raises `InvalidStateError` from inside + a section that must not raise, abandoning the participant + entry and the state's lock. + """ + transaction = self._transaction() + waiter = await self._waiter(transaction) + + waiter.cancel() + with self.assertRaises(asyncio.CancelledError): + await waiter + + self.assertFalse(transaction.finished()) + + transaction.abort() + self.assertTrue(transaction.aborted()) + + async def test_cancelled_waiter_leaves_other_waiters_waiting( + self, + ) -> None: + """One waiter's cancellation leaves the other waiters to learn + the transaction's outcome as usual.""" + transaction = self._transaction() + cancelled_waiter = await self._waiter(transaction) + surviving_waiter = await self._waiter(transaction) + + cancelled_waiter.cancel() + with self.assertRaises(asyncio.CancelledError): + await cancelled_waiter + + transaction.prepare() + transaction.commit() + + await surviving_waiter + self.assertTrue(transaction.committed()) + + class EffectsRequiresExclusiveTest(unittest.TestCase): """Unit tests for `Effects.requires_exclusive()`, which decides whether a transaction running in shared mode must upgrade From 0942143e94491f14a39fe61abd27ab58aaca0325 Mon Sep 17 00:00:00 2001 From: Robert-Jan Huijsman <22160949+rjhuijsman@users.noreply.github.com> Date: Fri, 11 Sep 2026 20:24:04 +0000 Subject: [PATCH 2/2] `state_managers`: shield `acquired_lock` from its waiters too Extends the previous commit's reasoning to the transaction's other shared future, so that neither of the two futures a concurrent call may be parked on can be resolved by a party that was only reading it. Before this change, a call that found another call in the same transaction already started on this state awaited `transaction.acquired_lock` directly. That future is shared with every other concurrent call on this state, and is resolved by `_transaction_participant_start()` once the per-state lock has been acquired (or with the acquire's exception). A cancelled waiter therefore cancelled it for everybody, and the resolution that followed raised `InvalidStateError` from a path with no handling for it. This is the same hazard as `_committed`, reached through a different future: the regression tests for that one cover the mechanism, and this call site has no seam to exercise it through in isolation. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_014L8UDXfhAJNKDLuEHN42AX --- reboot/aio/state_managers.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/reboot/aio/state_managers.py b/reboot/aio/state_managers.py index 990e4482..e8b9c18c 100644 --- a/reboot/aio/state_managers.py +++ b/reboot/aio/state_managers.py @@ -4052,7 +4052,13 @@ async def transactionally( # has already started (i.e., calls were made # concurrently). We must wait for that first call to # acquire the per-state lock before we continue here. - await transaction.acquired_lock + # + # Shield for the same reason as `_committed`: + # `acquired_lock` is shared with every other concurrent + # call on this state, so a cancellation reaching it would + # leave `_transaction_participant_start()` unable to + # resolve it. + await asyncio.shield(transaction.acquired_lock) if transaction.finished(): # TODO(benh): add a test case for this!