From baa7a608f81250ee09e9d69489985c28671c299d Mon Sep 17 00:00:00 2001 From: Robert-Jan Huijsman <22160949+rjhuijsman@users.noreply.github.com> Date: Thu, 10 Sep 2026 14:11:23 +0000 Subject: [PATCH 1/3] `state_managers`: let a read-only participant answer a re-sent `Prepare` Stops a transaction that prepared successfully from being aborted because the coordinator asked a second time, which is what left `concurrent_transactions_same_state` hanging for its full Bazel timeout on MacOS arm64. Before this change, two behaviours that are each correct did not compose. A read-only participant elides its prepare/commit on the first `Prepare`: it marks the transaction prepared and committed in memory, drops its participant entry and releases its shared lock straight away, since it has nothing to persist and the coordinator skips it at commit. Separately, the coordinator retries `Prepare` indefinitely on any RPC-level error, because such an error says nothing about whether the participant prepared and so must never be read as an abort. Put together, a `Prepare` whose response was lost got re-sent to a participant that had forgotten the transaction precisely because it had succeeded. It answered `abort=True`, which the coordinator does treat as definitive, and a prepared transaction became an aborted one. The recovery path already avoids this by re-preparing with `skip_read_only=True`, "because by then read-only participants may have already forgotten the transaction"; the live retry loop had no equivalent. The coordinator now tells each participant, in `PrepareRequest`, whether it recorded that participant as read-only. A participant asked about a transaction it no longer holds can then answer prepared -- but only once its own restart detection confirms it has not restarted since the transaction began, which is the single way it could have lost the transaction rather than completed it. `abort=True` keeps meaning exactly one thing, and the coordinator needs no new interpretation of it. Old participants ignore the new field and answer `abort=True` as before; old coordinators never set it, so new participants keep today's behaviour. Both directions of a rolling upgrade degrade to what happens today. - Add five tests covering the elision, the re-sent `Prepare`, and the three cases that must still abort: an old coordinator, a coordinator that is not `read_only_aware`, and a participant that restarted or cannot detect a restart. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_014L8UDXfhAJNKDLuEHN42AX --- rbt/v1alpha1/transactions.proto | 5 + reboot/aio/state_managers.py | 58 ++++++- tests/reboot/state_manager_tests.py | 230 +++++++++++++++++++++++++++- 3 files changed, 288 insertions(+), 5 deletions(-) diff --git a/rbt/v1alpha1/transactions.proto b/rbt/v1alpha1/transactions.proto index 2f40a4ee9..f97ea70d7 100644 --- a/rbt/v1alpha1/transactions.proto +++ b/rbt/v1alpha1/transactions.proto @@ -25,6 +25,11 @@ message PrepareRequest { // false (the default) and new participants fall back to writing to // disk for prepare/commit. bool read_only_aware = 3; + // When true, the coordinator has recorded this participant as a + // read-only participant of this transaction. New coordinators set + // this alongside `read_only_aware`; old coordinators leave it false + // (the default). + bool read_only = 4; } message PrepareResponse { diff --git a/reboot/aio/state_managers.py b/reboot/aio/state_managers.py index e8b9c18c4..3084bf380 100644 --- a/reboot/aio/state_managers.py +++ b/reboot/aio/state_managers.py @@ -5681,7 +5681,14 @@ async def _transaction_coordinator_prepare( any `SystemAborted` as "safe to abort". """ - async def prepare(state_type: StateTypeName, state_ref: StateRef): + read_only_participants = set(participants.read_only()) + + async def prepare( + state_type: StateTypeName, + state_ref: StateRef, + *, + read_only: bool, + ): # We retry indefinitely on any non-definitive outcome: we # cannot report the transaction as aborted to the caller # unless we have a definitive "never prepared" answer from @@ -5721,6 +5728,7 @@ async def prepare(state_type: StateTypeName, state_ref: StateRef): # `Prepare`. Old participants ignore this # field and do the disk-writes. read_only_aware=True, + read_only=read_only, ), metadata=Headers( application_id=application_id, @@ -5772,7 +5780,11 @@ async def prepare(state_type: StateTypeName, state_ref: StateRef): ) from None await concurrently( - prepare(state_type, state_ref) for (state_type, state_ref) in + prepare( + state_type, + state_ref, + read_only=(state_type, state_ref) in read_only_participants, + ) for (state_type, state_ref) in # On re-prepare `skip_read_only=True` because by then # read-only participants may have already forgotten the # transaction. @@ -6102,8 +6114,11 @@ async def Prepare( state_type, state_ref, transaction_id ) if transaction is None: + can_use_restart_detection = self._can_use_restart_detection( + transaction_id, state_type + ) if ( - self._can_use_restart_detection(transaction_id, state_type) and + can_use_restart_detection and # `_can_use_restart_detection` ensures that # `self._recovery_timestamp_ms` is not `None`. self._recovery_timestamp_ms # type: ignore[operator] @@ -6137,6 +6152,43 @@ async def Prepare( f"recovered at {recovery_time})." ) if request.abort_via_response: + if ( + request.read_only and request.read_only_aware and + can_use_restart_detection + ): + # A `read_only_aware` coordinator recorded this + # transaction as read-only here, and the check + # above establishes that this server has not + # restarted since this transaction began. Were we + # still holding this transaction, the lookup above + # would have found it, so the one remaining + # explanation is that an earlier `Prepare` + # prepared and committed it in memory and released + # the shared lock (see + # `transaction_participant_prepare`). Answer + # prepared: the coordinator retries `Prepare` on + # any RPC-level error, and "abort" here would turn + # a transaction that did prepare into an abort. + # + # The `read_only` flag alone establishes none of + # this: it reflects our own classification, made + # when we joined and sent up to the coordinator, + # and `Participants.retain_as_read_only()` can + # move a participant into that set later. The + # local restart check is what rules out having + # lost this transaction along with the rest of + # memory. + # + # That check is per process: shards are fixed when + # a state manager is constructed, so a shard + # changes owner only via a new process, which the + # recovery timestamp catches. If we implement + # reassignment of shards into already running + # servers, this check needs a per-shard recovery + # timestamp, or this answer would claim a + # transaction the new owner never saw was + # prepared. + return transactions_pb2.PrepareResponse() logger.warning( f"Failed to prepare transaction '{transaction_id}': " f"No pending transaction for state type '{state_type}' " diff --git a/tests/reboot/state_manager_tests.py b/tests/reboot/state_manager_tests.py index 0db920eb0..1850e65df 100644 --- a/tests/reboot/state_manager_tests.py +++ b/tests/reboot/state_manager_tests.py @@ -8,7 +8,7 @@ from google.protobuf.empty_pb2 import Empty from google.protobuf.timestamp_pb2 import Timestamp from google.protobuf.wrappers_pb2 import StringValue -from rbt.v1alpha1 import database_pb2, tasks_pb2 +from rbt.v1alpha1 import database_pb2, tasks_pb2, transactions_pb2 from rbt.v1alpha1.errors_pb2 import ( StateAlreadyConstructed, StateNotConstructed, @@ -26,7 +26,11 @@ ) from reboot.aio.headers import Headers from reboot.aio.internals.channel_manager import _ChannelManager -from reboot.aio.internals.contextvars import Servicing, _servicing +from reboot.aio.internals.contextvars import ( + Servicing, + _servicing, + use_application_id, +) from reboot.aio.internals.tasks_dispatcher import TasksDispatcher from reboot.aio.placement import StaticPlacementClient from reboot.aio.resolvers import NoResolver @@ -840,6 +844,228 @@ async def test_transaction_with_idempotency_key_is_exclusive(self): assert transaction is not None self.assertEqual(transaction.mode, Lock.Mode.EXCLUSIVE) + def create_grpc_context_mock( + self, + state_ref: StateRef, + ) -> grpc.aio.ServicerContext: + """Create the gRPC context a `Participant` servicer method + reads its headers from, so these tests can call `Prepare` + the way a coordinator does.""" + grpc_context = unittest.mock.MagicMock(spec=grpc.aio.ServicerContext) + grpc_context.invocation_metadata.return_value = Headers( + application_id=ApplicationId('test-app'), + state_ref=state_ref, + ).to_grpc_metadata() + return grpc_context + + async def join_read_only_participant( + self, + state_id: StateId, + *, + database_timestamp_ms: int, + ) -> TransactionContext: + """Join a read-only transaction (shared lock, no idempotency + key, restart detection available) as a participant on + `state_id` and leave it joined, i.e. in the state a + coordinator's first `Prepare` finds it in.""" + context = self.create_transaction_context( + state_id, + database_timestamp_ms=database_timestamp_ms, + ) + async with self.state_manager.transactionally( + context, + self.create_task_dispatcher_mock(), + aborted_type=None, + ) as transaction: + assert transaction is not None + self.assertEqual(transaction.mode, Lock.Mode.SHARED) + self.assertTrue(transaction.using_restart_detection) + return context + + async def send_prepare( + self, + context: TransactionContext, + *, + read_only: bool, + read_only_aware: bool = True, + ) -> transactions_pb2.PrepareResponse: + """Send the `Prepare` a modern coordinator sends, telling the + participant whether the coordinator recorded it as + read-only.""" + assert context.transaction_root_id is not None + # A `Participant` servicer method reads the application id + # from the asyncio context variable that every server's + # `UseApplicationIdInterceptor` sets. + with use_application_id(ApplicationId('test-app')): + return await self.state_manager.Prepare( + transactions_pb2.PrepareRequest( + transaction_id=context.transaction_root_id.bytes, + abort_via_response=True, + read_only_aware=read_only_aware, + read_only=read_only, + ), + self.create_grpc_context_mock(context._state_ref), + ) + + async def test_read_only_prepare_elides_and_releases_the_lock( + self, + ) -> None: + """A read-only participant's first `Prepare` prepares and + commits in memory, drops the participant entry and releases + the shared lock, all without a disk write.""" + self.state_manager._recovery_timestamp_ms = 1000 + + context = await self.join_read_only_participant( + "test-1234", + database_timestamp_ms=2000, + ) + state_type = MyGreeterServicer.__state_type_name__ + state_ref = context._state_ref + self.assertTrue( + self.state_manager._locks[state_type][state_ref].is_shared_locked() + ) + + response = await self.send_prepare(context, read_only=True) + + self.assertFalse(response.abort) + self.assertIsNone( + self.state_manager._lookup_participant_transaction( + state_type, + state_ref, + context.transaction_root_id, + ) + ) + self.assertFalse( + self.state_manager._locks[state_type][state_ref].is_locked() + ) + + async def test_reprepared_read_only_participant_answers_prepared( + self, + ) -> None: + """A `Prepare` re-sent to a read-only participant that already + elided is answered "prepared". + + The coordinator retries `Prepare` on any RPC-level error, + because such an error says nothing about whether the + participant prepared. If the first `Prepare` did arrive, the + participant elided it and forgot the transaction, so the retry + asks about a transaction that is gone precisely because it + succeeded. The participant has not restarted since the + transaction began, so that is the only way it can have + forgotten, and answering "abort" would turn a transaction that + did prepare into an abort. + """ + self.state_manager._recovery_timestamp_ms = 1000 + + context = await self.join_read_only_participant( + "test-1234", + database_timestamp_ms=2000, + ) + elided = await self.send_prepare(context, read_only=True) + self.assertFalse(elided.abort) + + response = await self.send_prepare(context, read_only=True) + + self.assertFalse(response.abort) + self.assertFalse(response.restart_detected) + + async def test_reprepared_participant_aborts_for_old_coordinator( + self, + ) -> None: + """An old coordinator does not set `read_only`, so it still + gets the definitive abort it expects. + + Its participant record may not distinguish read-only + participants at all, so "I have no pending transaction" has to + keep meaning abort for it. + """ + self.state_manager._recovery_timestamp_ms = 1000 + + context = await self.join_read_only_participant( + "test-1234", + database_timestamp_ms=2000, + ) + elided = await self.send_prepare(context, read_only=True) + self.assertFalse(elided.abort) + + response = await self.send_prepare(context, read_only=False) + + self.assertTrue(response.abort) + self.assertFalse(response.restart_detected) + + async def test_reprepared_participant_aborts_without_read_only_aware( + self, + ) -> None: + """A coordinator that has not promised to skip read-only + participants on recovery still gets a definitive abort. + + That promise is what permits the elision, so without it a + missing participant entry has no benign explanation. + """ + self.state_manager._recovery_timestamp_ms = 1000 + + context = await self.join_read_only_participant( + "test-1234", + database_timestamp_ms=2000, + ) + elided = await self.send_prepare(context, read_only=True) + self.assertFalse(elided.abort) + + response = await self.send_prepare( + context, + read_only=True, + read_only_aware=False, + ) + + self.assertTrue(response.abort) + + async def test_reprepared_read_only_participant_aborts_after_restart( + self, + ) -> None: + """A participant that restarted since the transaction began + reports the restart rather than claiming it prepared: it may + have lost the transaction with its memory, so the coordinator + must retry rather than count it as prepared.""" + self.state_manager._recovery_timestamp_ms = 1000 + + context = await self.join_read_only_participant( + "test-1234", + database_timestamp_ms=2000, + ) + elided = await self.send_prepare(context, read_only=True) + self.assertFalse(elided.abort) + + # The server recovered after this transaction began, which is + # what it looks like to have restarted mid-transaction. + self.state_manager._recovery_timestamp_ms = 3000 + + response = await self.send_prepare(context, read_only=True) + + self.assertTrue(response.abort) + self.assertTrue(response.restart_detected) + + async def test_reprepared_read_only_participant_aborts_without_uuid7( + self, + ) -> None: + """Without restart detection a participant cannot tell "I + finished and forgot" from "I lost my state", so it keeps + answering abort. + + A UUIDv4 transaction id carries no timestamp to compare the + recovery timestamp against. Such a transaction never elides + either, so a missing participant entry really is unexplained. + """ + self.state_manager._recovery_timestamp_ms = 1000 + + context = self.create_transaction_context("test-1234") + assert context.transaction_root_id is not None + self.assertEqual(context.transaction_root_id.version, 4) + + response = await self.send_prepare(context, read_only=True) + + self.assertTrue(response.abort) + self.assertFalse(response.restart_detected) + class PresumedDeadlockedNestedTransactionTest(unittest.TestCase): """Which nested transaction, if any, a call waiting for ownership From 0474948011111cda23d17f8736b496914039ba5a Mon Sep 17 00:00:00 2001 From: Robert-Jan Huijsman <22160949+rjhuijsman@users.noreply.github.com> Date: Fri, 11 Sep 2026 20:42:07 +0000 Subject: [PATCH 2/3] `state_managers`: abort a participant told to commit without preparing Stops a read-only participant from holding its state's shared lock for the life of the process after its coordinator crashes, which blocks every later exclusive claim on that state. Before this change, a participant whose watch of the coordinator reported the transaction committed went on to commit, whether or not it had ever prepared. A coordinator writes its participant list to disk and fans `Prepare` out concurrently, so it can crash with the list durably recorded and a read-only participant's `Prepare` never sent. That participant stays joined, unprepared, holding its shared lock. The recovered coordinator re-prepares with `skip_read_only=True` -- read-only participants may already have elided and forgotten the transaction -- and then answers this one's `Watch` with "committed". The database persists a participant transaction only once it is prepared and refuses to commit one that is not, so that commit failed: with restart detection "Missing transaction for state type ...", and on the legacy path "Txn not prepared". The watch loop treats the failure as transient, backs off and asks again, gets the same answer, and repeats forever, so `_complete_participant_transaction()` is never reached and the lock is never released. Aborting is the terminal outcome such a participant can still reach, and it is safe: a read-only participant has nothing to apply. A participant that did elide is already `finished()`, which makes the abort a no-op for it. - Add tests for both: the unprepared participant told to commit, and the elided participant tolerating a later abort. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_014L8UDXfhAJNKDLuEHN42AX --- reboot/aio/state_managers.py | 27 ++++--- tests/reboot/state_manager_tests.py | 119 ++++++++++++++++++++++++++++ 2 files changed, 137 insertions(+), 9 deletions(-) diff --git a/reboot/aio/state_managers.py b/reboot/aio/state_managers.py index 3084bf380..e890eddb1 100644 --- a/reboot/aio/state_managers.py +++ b/reboot/aio/state_managers.py @@ -5994,15 +5994,24 @@ async def _transaction_participant_watch( ).to_grpc_metadata(), ) - if not watch_response.aborted: - # It is worth noting here that if this participant - # was read-only then - # `transaction_participant_commit` will be a no-op - # because the only way a transaction commits is if - # it was prepared and thus this read-only - # participant must have prepared so - # `transaction_participant_commit` will find a - # `finished` transaction. + # Committing requires a prepared transaction: the + # database persists a participant transaction only + # once it is prepared, and rejects a commit for one + # that is not, so an unprepared participant reaches a + # terminal outcome by aborting, which is also what + # releases this state's lock. + # + # A recovering coordinator re-prepares with + # `skip_read_only=True`, so a read-only participant + # whose original `Prepare` was lost when the + # coordinator crashed is still unprepared by the time + # that coordinator's `Watch` reports the transaction + # as committed. + # + # A participant that elided its prepare and commit is + # already `finished()`, which makes either call below + # a no-op for it. + if not watch_response.aborted and transaction.prepared(): await self.transaction_participant_commit(transaction) else: await self.transaction_participant_abort(transaction) diff --git a/tests/reboot/state_manager_tests.py b/tests/reboot/state_manager_tests.py index 1850e65df..3cc6f750d 100644 --- a/tests/reboot/state_manager_tests.py +++ b/tests/reboot/state_manager_tests.py @@ -907,6 +907,125 @@ async def send_prepare( self.create_grpc_context_mock(context._state_ref), ) + async def watch_coordinator_answering( + self, + transaction: StateManager.Transaction, + *, + aborted: bool, + ) -> None: + """Run the participant's watch control loop once against a + coordinator whose `Watch` answers `aborted`, which is how a + participant that the coordinator does not contact directly + learns its transaction's outcome.""" + # The participant started a watch task of its own when it + # joined; this loop replaces it, so stop that one. + assert transaction.watch_task is not None + transaction.watch_task.cancel() + + stub = unittest.mock.MagicMock() + + async def watch(request, metadata): + return transactions_pb2.WatchResponse(aborted=aborted) + + stub.Watch = watch + + with unittest.mock.patch( + 'reboot.aio.state_managers.transactions_pb2_grpc.CoordinatorStub', + return_value=stub, + ): + await self.state_manager._transaction_participant_watch( + ApplicationId('test-app'), + unittest.mock.MagicMock(spec=_ChannelManager), + transaction, + ) + + def lookup_joined_transaction( + self, + context: TransactionContext, + ) -> StateManager.Transaction: + """Return the participant transaction `context` joined.""" + assert context.transaction_root_id is not None + transaction = self.state_manager._lookup_participant_transaction( + MyGreeterServicer.__state_type_name__, + context._state_ref, + context.transaction_root_id, + ) + assert transaction is not None + return transaction + + async def test_unprepared_read_only_participant_told_to_commit_aborts( + self, + ) -> None: + """A read-only participant told that a transaction it never + prepared committed aborts, releasing its shared lock. + + A coordinator writes its participants to disk and sends + `Prepare` concurrently, so it can crash with the participants + durably recorded and a read-only participant's `Prepare` never + sent; that participant stays joined, unprepared and holding + its shared lock. The recovered coordinator re-prepares with + `skip_read_only=True` and then answers this participant's + `Watch` with "committed", which is a transaction the database + never prepared and therefore refuses to commit. Aborting is + the outcome this participant can still reach, and it is safe + because a read-only participant has nothing to apply. + """ + self.state_manager._recovery_timestamp_ms = 1000 + + context = await self.join_read_only_participant( + "test-1234", + database_timestamp_ms=2000, + ) + state_type = MyGreeterServicer.__state_type_name__ + state_ref = context._state_ref + transaction = self.lookup_joined_transaction(context) + self.assertFalse(transaction.prepared()) + self.assertTrue( + self.state_manager._locks[state_type][state_ref].is_shared_locked() + ) + + await self.watch_coordinator_answering(transaction, aborted=False) + + self.assertTrue(transaction.aborted()) + self.assertIsNone( + self.state_manager._lookup_participant_transaction( + state_type, + state_ref, + context.transaction_root_id, + ) + ) + self.assertFalse( + self.state_manager._locks[state_type][state_ref].is_locked() + ) + + async def test_abort_of_an_elided_read_only_participant_is_a_no_op( + self, + ) -> None: + """A read-only participant that already elided its prepare and + commit tolerates a later abort: it is `finished()`, so the + abort leaves it committed and leaves its released lock + alone.""" + self.state_manager._recovery_timestamp_ms = 1000 + + context = await self.join_read_only_participant( + "test-1234", + database_timestamp_ms=2000, + ) + state_type = MyGreeterServicer.__state_type_name__ + state_ref = context._state_ref + transaction = self.lookup_joined_transaction(context) + + elided = await self.send_prepare(context, read_only=True) + self.assertFalse(elided.abort) + self.assertTrue(transaction.committed()) + + await self.state_manager.transaction_participant_abort(transaction) + + self.assertTrue(transaction.committed()) + self.assertFalse( + self.state_manager._locks[state_type][state_ref].is_locked() + ) + async def test_read_only_prepare_elides_and_releases_the_lock( self, ) -> None: From b005cc088120667fea018cbe582188e83c976282 Mon Sep 17 00:00:00 2001 From: Robert-Jan Huijsman <22160949+rjhuijsman@users.noreply.github.com> Date: Tue, 15 Sep 2026 15:33:52 +0000 Subject: [PATCH 3/3] Address review comments Replaces the state manager unit tests with tests that drive the real coordinator and participant end to end, per benh's review: the old tests called the `Prepare` servicer method with a mocked gRPC context, hand-drove the participant's watch loop against a patched `CoordinatorStub` and poked `_recovery_timestamp_ms`, `_locks` and `_lookup_participant_transaction` directly, which mimicked the implementation rather than testing it. The new tests in `transaction_tests.py` run a real `Reboot()` application with `Bank.Transferrable`, whose accounts are read-only participants, and inject only the fault: a `Prepare` response that goes missing (the participant handles the RPC in full and the RPC then fails), a `Prepare` that is never handled before the coordinator crashes, an old coordinator (`read_only` cleared from its requests), and a participant server restart between the elision and the re-sent `Prepare`. Outcomes are observed through the client: whether `Transferrable` commits or aborts and with what error, and whether an exclusive write on the account goes through, which is what shows its shared lock was released. Kept: the re-sent `Prepare` (commits), the old coordinator (still a definitive abort, and the coordinator's `Abort` is a no-op for the elided participant), the restart (reported over claiming prepared), and the unprepared participant told to commit after a coordinator crash (aborts, releasing its lock; the wait is bounded so the unfixed behaviour fails rather than hangs). Dropped: the `read_only_aware=False` case, since a coordinator that sets `read_only` always sets `read_only_aware` too and a participant that is not told `read_only_aware` never elides, and the UUIDv4 case, since a participant without restart detection never elides either, so neither can reach the new branch with a real coordinator. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01DvoLW5RiQ6EWUKzZQoC4oX --- tests/reboot/state_manager_tests.py | 349 +----------------- tests/reboot/transaction_tests.py | 534 ++++++++++++++++++++++++++++ 2 files changed, 536 insertions(+), 347 deletions(-) diff --git a/tests/reboot/state_manager_tests.py b/tests/reboot/state_manager_tests.py index 3cc6f750d..0db920eb0 100644 --- a/tests/reboot/state_manager_tests.py +++ b/tests/reboot/state_manager_tests.py @@ -8,7 +8,7 @@ from google.protobuf.empty_pb2 import Empty from google.protobuf.timestamp_pb2 import Timestamp from google.protobuf.wrappers_pb2 import StringValue -from rbt.v1alpha1 import database_pb2, tasks_pb2, transactions_pb2 +from rbt.v1alpha1 import database_pb2, tasks_pb2 from rbt.v1alpha1.errors_pb2 import ( StateAlreadyConstructed, StateNotConstructed, @@ -26,11 +26,7 @@ ) from reboot.aio.headers import Headers from reboot.aio.internals.channel_manager import _ChannelManager -from reboot.aio.internals.contextvars import ( - Servicing, - _servicing, - use_application_id, -) +from reboot.aio.internals.contextvars import Servicing, _servicing from reboot.aio.internals.tasks_dispatcher import TasksDispatcher from reboot.aio.placement import StaticPlacementClient from reboot.aio.resolvers import NoResolver @@ -844,347 +840,6 @@ async def test_transaction_with_idempotency_key_is_exclusive(self): assert transaction is not None self.assertEqual(transaction.mode, Lock.Mode.EXCLUSIVE) - def create_grpc_context_mock( - self, - state_ref: StateRef, - ) -> grpc.aio.ServicerContext: - """Create the gRPC context a `Participant` servicer method - reads its headers from, so these tests can call `Prepare` - the way a coordinator does.""" - grpc_context = unittest.mock.MagicMock(spec=grpc.aio.ServicerContext) - grpc_context.invocation_metadata.return_value = Headers( - application_id=ApplicationId('test-app'), - state_ref=state_ref, - ).to_grpc_metadata() - return grpc_context - - async def join_read_only_participant( - self, - state_id: StateId, - *, - database_timestamp_ms: int, - ) -> TransactionContext: - """Join a read-only transaction (shared lock, no idempotency - key, restart detection available) as a participant on - `state_id` and leave it joined, i.e. in the state a - coordinator's first `Prepare` finds it in.""" - context = self.create_transaction_context( - state_id, - database_timestamp_ms=database_timestamp_ms, - ) - async with self.state_manager.transactionally( - context, - self.create_task_dispatcher_mock(), - aborted_type=None, - ) as transaction: - assert transaction is not None - self.assertEqual(transaction.mode, Lock.Mode.SHARED) - self.assertTrue(transaction.using_restart_detection) - return context - - async def send_prepare( - self, - context: TransactionContext, - *, - read_only: bool, - read_only_aware: bool = True, - ) -> transactions_pb2.PrepareResponse: - """Send the `Prepare` a modern coordinator sends, telling the - participant whether the coordinator recorded it as - read-only.""" - assert context.transaction_root_id is not None - # A `Participant` servicer method reads the application id - # from the asyncio context variable that every server's - # `UseApplicationIdInterceptor` sets. - with use_application_id(ApplicationId('test-app')): - return await self.state_manager.Prepare( - transactions_pb2.PrepareRequest( - transaction_id=context.transaction_root_id.bytes, - abort_via_response=True, - read_only_aware=read_only_aware, - read_only=read_only, - ), - self.create_grpc_context_mock(context._state_ref), - ) - - async def watch_coordinator_answering( - self, - transaction: StateManager.Transaction, - *, - aborted: bool, - ) -> None: - """Run the participant's watch control loop once against a - coordinator whose `Watch` answers `aborted`, which is how a - participant that the coordinator does not contact directly - learns its transaction's outcome.""" - # The participant started a watch task of its own when it - # joined; this loop replaces it, so stop that one. - assert transaction.watch_task is not None - transaction.watch_task.cancel() - - stub = unittest.mock.MagicMock() - - async def watch(request, metadata): - return transactions_pb2.WatchResponse(aborted=aborted) - - stub.Watch = watch - - with unittest.mock.patch( - 'reboot.aio.state_managers.transactions_pb2_grpc.CoordinatorStub', - return_value=stub, - ): - await self.state_manager._transaction_participant_watch( - ApplicationId('test-app'), - unittest.mock.MagicMock(spec=_ChannelManager), - transaction, - ) - - def lookup_joined_transaction( - self, - context: TransactionContext, - ) -> StateManager.Transaction: - """Return the participant transaction `context` joined.""" - assert context.transaction_root_id is not None - transaction = self.state_manager._lookup_participant_transaction( - MyGreeterServicer.__state_type_name__, - context._state_ref, - context.transaction_root_id, - ) - assert transaction is not None - return transaction - - async def test_unprepared_read_only_participant_told_to_commit_aborts( - self, - ) -> None: - """A read-only participant told that a transaction it never - prepared committed aborts, releasing its shared lock. - - A coordinator writes its participants to disk and sends - `Prepare` concurrently, so it can crash with the participants - durably recorded and a read-only participant's `Prepare` never - sent; that participant stays joined, unprepared and holding - its shared lock. The recovered coordinator re-prepares with - `skip_read_only=True` and then answers this participant's - `Watch` with "committed", which is a transaction the database - never prepared and therefore refuses to commit. Aborting is - the outcome this participant can still reach, and it is safe - because a read-only participant has nothing to apply. - """ - self.state_manager._recovery_timestamp_ms = 1000 - - context = await self.join_read_only_participant( - "test-1234", - database_timestamp_ms=2000, - ) - state_type = MyGreeterServicer.__state_type_name__ - state_ref = context._state_ref - transaction = self.lookup_joined_transaction(context) - self.assertFalse(transaction.prepared()) - self.assertTrue( - self.state_manager._locks[state_type][state_ref].is_shared_locked() - ) - - await self.watch_coordinator_answering(transaction, aborted=False) - - self.assertTrue(transaction.aborted()) - self.assertIsNone( - self.state_manager._lookup_participant_transaction( - state_type, - state_ref, - context.transaction_root_id, - ) - ) - self.assertFalse( - self.state_manager._locks[state_type][state_ref].is_locked() - ) - - async def test_abort_of_an_elided_read_only_participant_is_a_no_op( - self, - ) -> None: - """A read-only participant that already elided its prepare and - commit tolerates a later abort: it is `finished()`, so the - abort leaves it committed and leaves its released lock - alone.""" - self.state_manager._recovery_timestamp_ms = 1000 - - context = await self.join_read_only_participant( - "test-1234", - database_timestamp_ms=2000, - ) - state_type = MyGreeterServicer.__state_type_name__ - state_ref = context._state_ref - transaction = self.lookup_joined_transaction(context) - - elided = await self.send_prepare(context, read_only=True) - self.assertFalse(elided.abort) - self.assertTrue(transaction.committed()) - - await self.state_manager.transaction_participant_abort(transaction) - - self.assertTrue(transaction.committed()) - self.assertFalse( - self.state_manager._locks[state_type][state_ref].is_locked() - ) - - async def test_read_only_prepare_elides_and_releases_the_lock( - self, - ) -> None: - """A read-only participant's first `Prepare` prepares and - commits in memory, drops the participant entry and releases - the shared lock, all without a disk write.""" - self.state_manager._recovery_timestamp_ms = 1000 - - context = await self.join_read_only_participant( - "test-1234", - database_timestamp_ms=2000, - ) - state_type = MyGreeterServicer.__state_type_name__ - state_ref = context._state_ref - self.assertTrue( - self.state_manager._locks[state_type][state_ref].is_shared_locked() - ) - - response = await self.send_prepare(context, read_only=True) - - self.assertFalse(response.abort) - self.assertIsNone( - self.state_manager._lookup_participant_transaction( - state_type, - state_ref, - context.transaction_root_id, - ) - ) - self.assertFalse( - self.state_manager._locks[state_type][state_ref].is_locked() - ) - - async def test_reprepared_read_only_participant_answers_prepared( - self, - ) -> None: - """A `Prepare` re-sent to a read-only participant that already - elided is answered "prepared". - - The coordinator retries `Prepare` on any RPC-level error, - because such an error says nothing about whether the - participant prepared. If the first `Prepare` did arrive, the - participant elided it and forgot the transaction, so the retry - asks about a transaction that is gone precisely because it - succeeded. The participant has not restarted since the - transaction began, so that is the only way it can have - forgotten, and answering "abort" would turn a transaction that - did prepare into an abort. - """ - self.state_manager._recovery_timestamp_ms = 1000 - - context = await self.join_read_only_participant( - "test-1234", - database_timestamp_ms=2000, - ) - elided = await self.send_prepare(context, read_only=True) - self.assertFalse(elided.abort) - - response = await self.send_prepare(context, read_only=True) - - self.assertFalse(response.abort) - self.assertFalse(response.restart_detected) - - async def test_reprepared_participant_aborts_for_old_coordinator( - self, - ) -> None: - """An old coordinator does not set `read_only`, so it still - gets the definitive abort it expects. - - Its participant record may not distinguish read-only - participants at all, so "I have no pending transaction" has to - keep meaning abort for it. - """ - self.state_manager._recovery_timestamp_ms = 1000 - - context = await self.join_read_only_participant( - "test-1234", - database_timestamp_ms=2000, - ) - elided = await self.send_prepare(context, read_only=True) - self.assertFalse(elided.abort) - - response = await self.send_prepare(context, read_only=False) - - self.assertTrue(response.abort) - self.assertFalse(response.restart_detected) - - async def test_reprepared_participant_aborts_without_read_only_aware( - self, - ) -> None: - """A coordinator that has not promised to skip read-only - participants on recovery still gets a definitive abort. - - That promise is what permits the elision, so without it a - missing participant entry has no benign explanation. - """ - self.state_manager._recovery_timestamp_ms = 1000 - - context = await self.join_read_only_participant( - "test-1234", - database_timestamp_ms=2000, - ) - elided = await self.send_prepare(context, read_only=True) - self.assertFalse(elided.abort) - - response = await self.send_prepare( - context, - read_only=True, - read_only_aware=False, - ) - - self.assertTrue(response.abort) - - async def test_reprepared_read_only_participant_aborts_after_restart( - self, - ) -> None: - """A participant that restarted since the transaction began - reports the restart rather than claiming it prepared: it may - have lost the transaction with its memory, so the coordinator - must retry rather than count it as prepared.""" - self.state_manager._recovery_timestamp_ms = 1000 - - context = await self.join_read_only_participant( - "test-1234", - database_timestamp_ms=2000, - ) - elided = await self.send_prepare(context, read_only=True) - self.assertFalse(elided.abort) - - # The server recovered after this transaction began, which is - # what it looks like to have restarted mid-transaction. - self.state_manager._recovery_timestamp_ms = 3000 - - response = await self.send_prepare(context, read_only=True) - - self.assertTrue(response.abort) - self.assertTrue(response.restart_detected) - - async def test_reprepared_read_only_participant_aborts_without_uuid7( - self, - ) -> None: - """Without restart detection a participant cannot tell "I - finished and forgot" from "I lost my state", so it keeps - answering abort. - - A UUIDv4 transaction id carries no timestamp to compare the - recovery timestamp against. Such a transaction never elides - either, so a missing participant entry really is unexplained. - """ - self.state_manager._recovery_timestamp_ms = 1000 - - context = self.create_transaction_context("test-1234") - assert context.transaction_root_id is not None - self.assertEqual(context.transaction_root_id.version, 4) - - response = await self.send_prepare(context, read_only=True) - - self.assertTrue(response.abort) - self.assertFalse(response.restart_detected) - class PresumedDeadlockedNestedTransactionTest(unittest.TestCase): """Which nested transaction, if any, a call waiting for ownership diff --git a/tests/reboot/transaction_tests.py b/tests/reboot/transaction_tests.py index 106875551..a3df0e722 100644 --- a/tests/reboot/transaction_tests.py +++ b/tests/reboot/transaction_tests.py @@ -2161,6 +2161,540 @@ async def mock_commit(state_manager, request, grpc_context): # observed). self.assertEqual([], unexpected) + async def test_read_only_participant_answers_re_sent_prepare( + self, + ) -> None: + """A read-only participant whose `Prepare` response was lost + answers the coordinator's re-sent `Prepare` "prepared", so the + transaction commits. + + A read-only participant elides its prepare and commit on its + first `Prepare`: it marks the transaction prepared and + committed in memory, drops its participant entry and releases + its shared lock. The coordinator retries `Prepare` on any + RPC-level error, because such an error says nothing about + whether the participant prepared. So a `Prepare` whose + response was lost is re-sent to a participant that has + forgotten the transaction precisely because it succeeded, and + an "abort" from it would turn a transaction that did prepare + into an abort. + + Simulates the lost response by letting `alice` handle her + first `Prepare` of the `Transferrable` transaction in full and + then failing the RPC; the re-sent `Prepare` reaches her + unmodified. + """ + prepare = SidecarStateManager.Prepare + + alice_ref = StateRef.from_id(Account.__state_type_name__, 'alice') + bob_ref = StateRef.from_id(Account.__state_type_name__, 'bob') + + # The responses `alice` gave to each `Prepare` of the + # `Transferrable` transaction, in order. Only tracked once the + # setup transactions have drained (see below), so the first + # one is her first `Prepare` of that transaction. + alice_responses: list[transactions_pb2.PrepareResponse] = [] + track = False + + async def mock_prepare(state_manager, request, grpc_context): + state_ref = Headers.from_grpc_context(grpc_context).state_ref + response = await prepare(state_manager, request, grpc_context) + if track and state_ref == alice_ref: + alice_responses.append(response) + if len(alice_responses) == 1: + # The participant handled the `Prepare` in full; + # only its response goes missing. + raise RuntimeError('Simulating a lost Prepare response') + return response + + with mock.patch( + 'reboot.aio.state_managers.SidecarStateManager.Prepare', + mock_prepare, + ): + await self.rbt.up( + Application(servicers=[AccountServicer, BankServicer]), + ) + context = self.rbt.create_external_context(name=self.id()) + + bank, _ = await Bank.Create(context, SINGLETON_BANK_ID) + await bank.SignUp( + context, account_id=alice_ref.id, initial_deposit=100 + ) + await bank.SignUp( + context, account_id=bob_ref.id, initial_deposit=200 + ) + + # Reading a state waits for any prepared-but-not-yet- + # committed transaction on it to complete, and + # `AssetsUnderManagement` reads the Bank and every + # account, so once it returns the setup transactions' + # commit phases have fully drained and no further + # `Prepare` RPCs are coming from them. + await bank.AssetsUnderManagement( + context, + wait_for_amount_at_least=0, + ) + + track = True + + response = await bank.Transferrable( + context, + from_account_id=alice_ref.id, + to_account_id=bob_ref.id, + amount=50, + ) + self.assertTrue(response.transferrable) + + # `alice` elided on her first `Prepare` and answered the + # re-sent one "prepared" too. + self.assertEqual( + [False, False], + [response.abort for response in alice_responses], + ) + + # Her elision released her shared lock, so an exclusive + # write on her goes through. + alice = Account.ref(alice_ref.id) + await alice.Deposit(context, amount=1) + balance = await alice.Balance(context) + self.assertEqual(balance.amount, 101) + + async def test_re_sent_prepare_from_old_coordinator_still_aborts( + self, + ) -> None: + """A coordinator that does not say it recorded a participant + as read-only gets the definitive abort it expects when it + re-sends a `Prepare` to a participant that has forgotten the + transaction. + + An old coordinator has no `read_only` field in its + `PrepareRequest`, and its participant record may not even + distinguish read-only participants, so "no pending + transaction" has to keep meaning abort for it, as it did + before the field existed. The participant still elided on the + first `Prepare`, so the coordinator's `Abort` finds nothing to + abort and the account is left usable. + + Simulates the old coordinator by clearing `read_only` from + every `Prepare` of the `Transferrable` transaction, with the + same lost first response as + `test_read_only_participant_answers_re_sent_prepare`. + """ + prepare = SidecarStateManager.Prepare + + alice_ref = StateRef.from_id(Account.__state_type_name__, 'alice') + bob_ref = StateRef.from_id(Account.__state_type_name__, 'bob') + + alice_responses: list[transactions_pb2.PrepareResponse] = [] + track = False + + async def mock_prepare(state_manager, request, grpc_context): + state_ref = Headers.from_grpc_context(grpc_context).state_ref + if track: + request.read_only = False + response = await prepare(state_manager, request, grpc_context) + if track and state_ref == alice_ref: + alice_responses.append(response) + if len(alice_responses) == 1: + raise RuntimeError('Simulating a lost Prepare response') + return response + + with mock.patch( + 'reboot.aio.state_managers.SidecarStateManager.Prepare', + mock_prepare, + ): + await self.rbt.up( + Application(servicers=[AccountServicer, BankServicer]), + ) + context = self.rbt.create_external_context(name=self.id()) + + bank, _ = await Bank.Create(context, SINGLETON_BANK_ID) + await bank.SignUp( + context, account_id=alice_ref.id, initial_deposit=100 + ) + await bank.SignUp( + context, account_id=bob_ref.id, initial_deposit=200 + ) + + # See `test_read_only_participant_answers_re_sent_prepare` + # for why this drains the setup transactions. + await bank.AssetsUnderManagement( + context, + wait_for_amount_at_least=0, + ) + + track = True + + with self.assertRaises(Bank.TransferrableAborted) as aborted: + await bank.Transferrable( + context, + from_account_id=alice_ref.id, + to_account_id=bob_ref.id, + amount=50, + ) + self.assertEqual( + type(aborted.exception.error), + errors_pb2.TransactionParticipantFailedToPrepare, + ) + + # Need to acknowledge idempotency uncertainty so that we + # can continue running the test! + context.acknowledge_idempotency_uncertainty() + + # `alice` elided on her first `Prepare`; the re-sent one + # was answered with a definitive abort, so the coordinator + # asked no further. + self.assertEqual( + [False, True], + [response.abort for response in alice_responses], + ) + + # Her elision released her shared lock and the abort left + # that alone, so an exclusive write on her goes through. + alice = Account.ref(alice_ref.id) + await alice.Deposit(context, amount=1) + balance = await alice.Balance(context) + self.assertEqual(balance.amount, 101) + + async def test_restarted_read_only_participant_reports_restart( + self, + ) -> None: + """A read-only participant that restarted between eliding and + the re-sent `Prepare` reports the restart rather than claiming + it prepared, and the transaction is retried from scratch. + + Restarting is the one way a participant can have lost a + transaction rather than completed it, so restart detection + takes precedence over answering a re-sent `Prepare` + "prepared". The coordinator turns the reported restart into + `Unavailable`, the client retries with the same idempotency + key, and the retried transaction commits. + + The account's first `Prepare` of the first `Transferrable` + transaction is handled in full and then its response is lost; + the account's server is restarted before the re-sent + `Prepare` is allowed to reach the real handler. + """ + prepare = SidecarStateManager.Prepare + + account_ref = StateRef.from_id( + Account.__state_type_name__, 'jonathan-2345' + ) + + # The responses the account gave to each `Prepare` of the + # first `Transferrable` transaction, in order; tracked once + # the setup transactions have drained. + lost_transaction_id: Optional[bytes] = None + account_responses: list[transactions_pb2.PrepareResponse] = [] + account_elided = asyncio.Event() + account_restarted = asyncio.Event() + track = False + + async def mock_prepare(state_manager, request, grpc_context): + nonlocal lost_transaction_id + state_ref = Headers.from_grpc_context(grpc_context).state_ref + if track and state_ref == account_ref: + if lost_transaction_id is None: + lost_transaction_id = request.transaction_id + if request.transaction_id == lost_transaction_id: + if len(account_responses) > 0: + # A re-sent `Prepare`. Hold it until the + # account's server has restarted; one held on + # the old server is cancelled along with that + # server, and the coordinator re-sends it. + await account_restarted.wait() + response = await prepare( + state_manager, request, grpc_context + ) + account_responses.append(response) + if len(account_responses) == 1: + account_elided.set() + raise RuntimeError( + 'Simulating a lost Prepare response' + ) + return response + return await prepare(state_manager, request, grpc_context) + + # Records that the client retried on `Unavailable`, proving + # the participant's reported restart made it all the way to + # the client. + should_retry = UnaryRetriedCall._should_retry + retried_unavailable = asyncio.Event() + + def mock_should_retry(unary_retried_call, error): + if error.code() == grpc.StatusCode.UNAVAILABLE: + retried_unavailable.set() + return should_retry(unary_retried_call, error) + + with mock.patch( + 'reboot.aio.state_managers.SidecarStateManager.Prepare', + mock_prepare, + ), mock.patch( + 'reboot.aio.stubs.UnaryRetriedCall._should_retry', + mock_should_retry, + ): + await self.rbt.up( + Application(servicers=[AccountServicer, BankServicer]), + local_envoy=True, + servers=2, + ) + context = self.rbt.create_external_context(name=self.id()) + + bank, _ = await Bank.Create(context, SINGLETON_BANK_ID) + + # Bank and account on different servers, so that only the + # participant restarts. + _, account_server_id = await self.rbt.unique_servers( + bank._state_ref, + account_ref, + ) + + await bank.SignUp(context, account_id=account_ref.id) + + # See `test_read_only_participant_answers_re_sent_prepare` + # for why this drains the setup transactions. + await bank.AssetsUnderManagement( + context, + wait_for_amount_at_least=0, + ) + + track = True + + async def Transferrable(): + return await bank.Transferrable( + context, + from_account_id=account_ref.id, + to_account_id=account_ref.id, + amount=0, + ) + + transferrable_task = asyncio.create_task(Transferrable()) + + await account_elided.wait() + + account_server = await self.rbt.server_stop(account_server_id) + await self.rbt.server_start(account_server) + account_restarted.set() + + response = await transferrable_task + self.assertTrue(response.transferrable) + + self.assertTrue(retried_unavailable.is_set()) + + # The account elided on its first `Prepare` and reported + # its restart on the re-sent one, after which the + # coordinator asked no further about that transaction. + self.assertEqual( + [(False, False), (True, True)], + [ + (response.abort, response.restart_detected) + for response in account_responses + ], + ) + + async def test_unprepared_read_only_participant_aborts_after_crash( + self, + ) -> None: + """A read-only participant whose `Prepare` was never handled + before its coordinator crashed aborts, releasing its shared + lock, when the recovered coordinator reports the transaction + committed. + + A coordinator writes its participant list to disk and fans + `Prepare` out concurrently, so it can crash with the list + durably recorded and a read-only participant's `Prepare` + unhandled. That participant stays joined, unprepared, holding + its shared lock. The recovered coordinator re-prepares with + `skip_read_only=True`, because read-only participants may have + elided and forgotten the transaction, and then answers this + participant's `Watch` with "committed". The database refuses + to commit a participant transaction it never prepared, so + committing can never succeed; aborting is the terminal outcome + the participant can still reach, and it is safe because a + read-only participant has nothing to apply. + + The account's `Prepare` of the `Transferrable` transaction is + held, never reaching the real handler, until the Bank's server + is stopped; the dying coordinator's cleanup fails so its + record of its participants survives until recovery. Bounds + the wait for the account's lock so that a participant that + never reaches a terminal outcome fails this test rather than + hanging it. + """ + prepare = SidecarStateManager.Prepare + transaction_coordinator_prepare = ( + DatabaseClient.transaction_coordinator_prepare + ) + transaction_coordinator_cleanup = ( + DatabaseClient.transaction_coordinator_cleanup + ) + + bank_ref = StateRef.from_id(Bank.__state_type_name__, SINGLETON_BANK_ID) + account_ref = StateRef.from_id( + Account.__state_type_name__, 'jonathan-2345' + ) + + # All tracked only once the setup transactions have drained. + stranded_transaction_id: Optional[bytes] = None + account_prepare_held = asyncio.Event() + bank_prepared = asyncio.Event() + transaction_coordinator_prepare_written = asyncio.Event() + track = False + + # Whether `transaction_coordinator_cleanup` fails, simulating + # a coordinator that crashes while stopping rather than + # cleanly aborting: its record of its participants stays in + # the database and no participant is told to abort. Set from + # just before the coordinator is stopped until the account's + # outcome is observed, because the dying coordinator's cleanup + # can still run after `server_stop` returns. + fail_transaction_coordinator_cleanup = False + + async def mock_prepare(state_manager, request, grpc_context): + nonlocal stranded_transaction_id + state_ref = Headers.from_grpc_context(grpc_context).state_ref + if track and state_ref == account_ref: + if stranded_transaction_id is None: + stranded_transaction_id = request.transaction_id + if request.transaction_id == stranded_transaction_id: + # Never handled: held until the coordinator's + # server stops, which cancels the RPC. + account_prepare_held.set() + await asyncio.Event().wait() + response = await prepare(state_manager, request, grpc_context) + if track and state_ref == bank_ref: + bank_prepared.set() + return response + + async def mock_transaction_coordinator_prepare( + database_client, + **kwargs, + ): + result = await transaction_coordinator_prepare( + database_client, **kwargs + ) + if track: + transaction_coordinator_prepare_written.set() + return result + + async def mock_transaction_coordinator_cleanup( + database_client, + **kwargs, + ): + if fail_transaction_coordinator_cleanup: + raise RuntimeError('Simulating a coordinator crash') + return await transaction_coordinator_cleanup( + database_client, **kwargs + ) + + # Whether the client may retry. Off while the bank server is + # stopped, so that its going down surfaces as `Unavailable` + # rather than an endless retry; on again once it is back, + # since the exclusive write below may wait on the account's + # lock past the lock deadline and then be asked to retry. + retries_enabled = True + should_retry = UnaryRetriedCall._should_retry + + def mock_should_retry(unary_retried_call, error): + return retries_enabled and should_retry(unary_retried_call, error) + + with mock.patch( + 'reboot.aio.state_managers.SidecarStateManager.Prepare', + mock_prepare, + ), mock.patch( + 'reboot.server.database.DatabaseClient.' + 'transaction_coordinator_prepare', + mock_transaction_coordinator_prepare, + ), mock.patch( + 'reboot.server.database.DatabaseClient.' + 'transaction_coordinator_cleanup', + mock_transaction_coordinator_cleanup, + ), mock.patch( + 'reboot.aio.stubs.UnaryRetriedCall._should_retry', + mock_should_retry, + ): + await self.rbt.up( + Application(servicers=[AccountServicer, BankServicer]), + local_envoy=True, + servers=2, + ) + context = self.rbt.create_external_context(name=self.id()) + + bank, _ = await Bank.Create(context, SINGLETON_BANK_ID) + + # Bank and account on different servers, so that stopping + # the coordinator's server leaves the participant running. + bank_server_id, _ = await self.rbt.unique_servers( + bank._state_ref, + account_ref, + ) + + await bank.SignUp(context, account_id=account_ref.id) + + # See `test_read_only_participant_answers_re_sent_prepare` + # for why this drains the setup transactions. + await bank.AssetsUnderManagement( + context, + wait_for_amount_at_least=0, + ) + + track = True + + async def Transferrable(): + return await bank.Transferrable( + context, + from_account_id=account_ref.id, + to_account_id=account_ref.id, + amount=0, + ) + + transferrable_task = asyncio.create_task(Transferrable()) + + # The coordinator's record of its participants persisted + # and the Bank durably prepared, while the account's + # `Prepare` is being held. + await account_prepare_held.wait() + await bank_prepared.wait() + await transaction_coordinator_prepare_written.wait() + + fail_transaction_coordinator_cleanup = True + + retries_enabled = False + bank_server = await self.rbt.server_stop(bank_server_id) + + with self.assertRaises(Bank.TransferrableAborted) as aborted: + await transferrable_task + + self.assertEqual( + type(aborted.exception.error), errors_pb2.Unavailable + ) + + # Need to acknowledge idempotency uncertainty so that we + # can continue running the test! + context.acknowledge_idempotency_uncertainty() + + await self.rbt.server_start(bank_server) + retries_enabled = True + + # The recovered coordinator re-prepares only the Bank, + # which is durably prepared, and then reports the + # transaction committed to the account's `Watch`. The + # account, unprepared, aborts its part of the transaction + # and releases its shared lock, which is what lets an + # exclusive write on it go through. + account = Account.ref(account_ref.id) + await asyncio.wait_for( + account.Deposit(context, amount=1), + timeout=90, + ) + balance = await account.Balance(context) + self.assertEqual(balance.amount, 1) + + # The recovered coordinator's commit control loop retries + # its cleanup until it succeeds, so it may now delete the + # record. + fail_transaction_coordinator_cleanup = False + async def test_transaction_recovery_after_coordinator_preparing( self, ) -> None: