From 7cf2cdfd7af51575c12451823664658a79f62391 Mon Sep 17 00:00:00 2001 From: Benjamin Hindman Date: Fri, 14 Aug 2026 02:23:34 +0000 Subject: [PATCH 01/10] Ensure `TransactionShouldRetryWithoutBackoff` propagates Don't treat `TransactionShouldRetryWithoutBackoff` as an uncertain mutation within an `IdempotencyManager` since it is not uncertain and it may cause the entire transaction to later abort due to that uncertainty rather than just letting `TransactionShouldRetryWithoutBackoff` propagate. --- reboot/aio/aborted.py | 61 +++++++++--- reboot/aio/idempotency.py | 4 +- tests/reboot/BUILD.bazel | 23 +++++ tests/reboot/aborted_tests.py | 87 +++++++++++++++++ tests/reboot/idempotency_uncertainty_tests.py | 96 +++++++++++++++++++ 5 files changed, 256 insertions(+), 15 deletions(-) create mode 100644 tests/reboot/aborted_tests.py create mode 100644 tests/reboot/idempotency_uncertainty_tests.py diff --git a/reboot/aio/aborted.py b/reboot/aio/aborted.py index b38c90d9..e20af258 100644 --- a/reboot/aio/aborted.py +++ b/reboot/aio/aborted.py @@ -95,6 +95,28 @@ def is_retryable(aborted: Aborted): rbt.v1alpha1.errors_pb2.InvalidMethod, ] +# Errors that only Reboot generates, never a proxy or other component, +# and which a caller can catch with the transaction still committing. +# +# We consider it a recoverable pattern to either check if something is +# already constructed, e.g., by calling a reader on it, or try to +# construct something to ensure it is constructed, e.g., by calling a +# constructor. +FROM_BACKEND_AND_RECOVERABLE_ERROR_TYPES: tuple[type[Message], ...] = ( + rbt.v1alpha1.errors_pb2.StateNotConstructed, + rbt.v1alpha1.errors_pb2.StateAlreadyConstructed, +) + +# Errors that only Reboot generates, never a proxy or other component, +# but which require the transaction to abort and be retried rather +# than commit. +FROM_BACKEND_AND_UNRECOVERABLE_ERROR_TYPES: tuple[type[Message], ...] = ( + # Raised when a participant joins a transaction that started + # before the participant last recovered, i.e., before the + # participant ran any of the transaction's code. + rbt.v1alpha1.errors_pb2.TransactionShouldRetryWithoutBackoff, +) + # Any possible error type, i.e., possibly a `GrpcError`, a # `RebootError`, or a user declared error. ErrorT = TypeVar('ErrorT', bound=Message) @@ -415,21 +437,34 @@ def is_from_backend_and_recoverable(cls, exception: BaseException): # propagate as though they raised it themselves. return ( isinstance(exception, Aborted) and ( - cls.is_declared_error(exception.error) or - # We consider it a recoverable pattern to either check - # if something is already constructed, e.g., by calling a - # reader on it, or try to construct something to - # ensure it is constructed, e.g., by calling a - # constructor. Moreover, these errors are only - # generated by Reboot and never by a proxy or other - # component so we know that we've received this from - # the backend. - isinstance( + cls.is_declared_error(exception.error) or isinstance( exception.error, - rbt.v1alpha1.errors_pb2.StateNotConstructed, - ) or isinstance( + FROM_BACKEND_AND_RECOVERABLE_ERROR_TYPES, + ) + ) + ) + + @classmethod + def is_from_backend( + cls: type[AbortedT], + exception: BaseException, + ) -> bool: + """Helper to determine if an exception came from the backend, i.e., + the developer or Reboot raised it, as opposed to a transport + error which may have been raised before or after the server + processed the call. Because the backend raised it we know + whether or not a mutation happened. + + This is a weaker property than + `is_from_backend_and_recoverable()`: an error may tell us that + no mutation happened while still requiring the transaction to + abort and be retried. + """ + return ( + cls.is_from_backend_and_recoverable(exception) or ( + isinstance(exception, Aborted) and isinstance( exception.error, - rbt.v1alpha1.errors_pb2.StateAlreadyConstructed, + FROM_BACKEND_AND_UNRECOVERABLE_ERROR_TYPES, ) ) ) diff --git a/reboot/aio/idempotency.py b/reboot/aio/idempotency.py index d0c4854f..48f676a4 100644 --- a/reboot/aio/idempotency.py +++ b/reboot/aio/idempotency.py @@ -456,7 +456,7 @@ def idempotently( ) else: yield None - # TODO(benh): differentiate errors so that we only set + # TODO(benh): differentiate more errors so that we only set # uncertainty when we are truly uncertain. except BaseException as exception: # The `yield` threw an exception, which means the user @@ -469,7 +469,7 @@ def idempotently( if ( aborted_type is not None and - aborted_type.is_from_backend_and_recoverable(exception) + aborted_type.is_from_backend(exception) ): # We are not uncertain because we _must_ have gotten # this from the backend, so just let it propagate. diff --git a/tests/reboot/BUILD.bazel b/tests/reboot/BUILD.bazel index 7b78e930..8e66811d 100644 --- a/tests/reboot/BUILD.bazel +++ b/tests/reboot/BUILD.bazel @@ -755,6 +755,29 @@ py_test( ], ) +py_test( + name = "aborted_tests_py", + size = "small", + srcs = [":aborted_tests.py"], + main = "aborted_tests.py", + deps = [ + "//reboot/aio:aborted_py", + "@com_github_reboot_dev_reboot//rbt/v1alpha1:errors_py_proto", + ], +) + +py_test( + name = "idempotency_uncertainty_tests_py", + size = "small", + srcs = [":idempotency_uncertainty_tests.py"], + main = "idempotency_uncertainty_tests.py", + deps = [ + "//reboot/aio:aborted_py", + "//reboot/aio:idempotency_py", + "@com_github_reboot_dev_reboot//rbt/v1alpha1:errors_py_proto", + ], +) + py_test( name = "type_annotation_inference_tests_py", srcs = [":type_annotation_inference_tests.py"], diff --git a/tests/reboot/aborted_tests.py b/tests/reboot/aborted_tests.py new file mode 100644 index 00000000..dda165aa --- /dev/null +++ b/tests/reboot/aborted_tests.py @@ -0,0 +1,87 @@ +import unittest +from google.protobuf.message import Message +from rbt.v1alpha1.errors_pb2 import ( + StateAlreadyConstructed, + StateNotConstructed, + TransactionShouldRetryWithoutBackoff, + Unavailable, + Unknown, +) +from reboot.aio.aborted import SystemAborted +from reboot.api import Model + + +class DeclaresNothingAborted(SystemAborted): + """Stands in for a generated per-method `Aborted` type whose method + declares no errors of its own.""" + + @classmethod + def is_declared_error(cls, error: Message | Model) -> bool: + return False + + +class AbortedClassificationTest(unittest.TestCase): + """ + Tests how `Aborted` classifies errors, in particular the + distinction between an error that lets a transaction commit and one + that only tells us definitively whether a mutation happened. + """ + + def test_transaction_should_retry_is_from_backend(self): + # A participant raises this when it joins a transaction that + # started before the participant last recovered, i.e., before + # it ran any of the transaction's code, so we know definitively + # that no mutation happened. + aborted = SystemAborted(TransactionShouldRetryWithoutBackoff()) + + self.assertTrue( + DeclaresNothingAborted.is_from_backend(aborted), + ) + + # ... but the transaction still has to abort and be retried, so + # it must not be considered recoverable. + self.assertFalse( + DeclaresNothingAborted.is_from_backend_and_recoverable(aborted), + ) + + def test_construction_errors_are_recoverable(self): + for error in (StateNotConstructed(), StateAlreadyConstructed()): + with self.subTest(error=type(error).__name__): + aborted = SystemAborted(error) + self.assertTrue( + DeclaresNothingAborted. + is_from_backend_and_recoverable(aborted), + ) + # Anything recoverable is also from the backend. + self.assertTrue( + DeclaresNothingAborted.is_from_backend(aborted), + ) + + def test_transport_errors_are_neither(self): + # These may be raised by a proxy or the network before or after + # the server processed the call, so we can not know whether a + # mutation happened. + for error in (Unavailable(), Unknown()): + with self.subTest(error=type(error).__name__): + aborted = SystemAborted(error) + self.assertFalse( + DeclaresNothingAborted. + is_from_backend_and_recoverable(aborted), + ) + self.assertFalse( + DeclaresNothingAborted.is_from_backend(aborted), + ) + + def test_non_aborted_exception_is_neither(self): + exception = RuntimeError("not an `Aborted`") + + self.assertFalse( + DeclaresNothingAborted.is_from_backend_and_recoverable(exception), + ) + self.assertFalse( + DeclaresNothingAborted.is_from_backend(exception), + ) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/reboot/idempotency_uncertainty_tests.py b/tests/reboot/idempotency_uncertainty_tests.py new file mode 100644 index 00000000..bd13aa9f --- /dev/null +++ b/tests/reboot/idempotency_uncertainty_tests.py @@ -0,0 +1,96 @@ +import unittest +from google.protobuf.message import Message +from rbt.v1alpha1.errors_pb2 import ( + TransactionShouldRetryWithoutBackoff, + Unavailable, +) +from reboot.aio.aborted import SystemAborted +from reboot.aio.idempotency import IdempotencyManager +from reboot.aio.types import ServiceName, StateRef, StateTypeName +from reboot.api import Model +from typing import Optional + + +class DeclaresNothingAborted(SystemAborted): + """Stands in for a generated per-method `Aborted` type whose method + declares no errors of its own.""" + + @classmethod + def is_declared_error(cls, error: Message | Model) -> bool: + return False + + +class UncertainMutationTestCase(unittest.TestCase): + """ + Tests which failed mutations make an `IdempotencyManager` + uncertain. A `Node.Insert` that splits creates its siblings + concurrently, so a single failure can fail several mutations on one + manager. + """ + + STATE_TYPE_NAME = StateTypeName("test.v1.Node") + SERVICE_NAME = ServiceName("test.v1.NodeMethods") + + def _mutate_and_raise( + self, + manager: IdempotencyManager, + exception: Optional[BaseException], + *, + state_id: str, + ) -> None: + """Performs a mutation without idempotency, failing it with + `exception` if one is given.""" + with manager.idempotently( + state_type_name=self.STATE_TYPE_NAME, + state_ref=StateRef.from_id(self.STATE_TYPE_NAME, state_id), + service_name=self.SERVICE_NAME, + method="Create", + mutation=True, + request=None, + metadata=None, + idempotency=None, + aborted_type=DeclaresNothingAborted, + ): + if exception is not None: + raise exception + + def test_concurrent_retryable_failures_do_not_become_uncertain(self): + # Several sibling `Create`s failing with the same retryable + # abort must all propagate that abort so the transaction is + # retried, rather than the second one turning it into an + # `AssertionError`. + manager = IdempotencyManager() + + for state_id in ["sibling-1", "sibling-2", "sibling-3"]: + with self.assertRaises(SystemAborted): + self._mutate_and_raise( + manager, + SystemAborted(TransactionShouldRetryWithoutBackoff()), + state_id=state_id, + ) + + # A later mutation must still be allowed, i.e., nothing was + # recorded as uncertain. + self._mutate_and_raise(manager, None, state_id="sibling-4") + + def test_concurrent_transport_failures_report_uncertainty(self): + # An `Unavailable` may or may not have mutated, so the first + # failure makes the manager uncertain and the next mutation is + # refused -- with an actionable error, not an `AssertionError`. + manager = IdempotencyManager() + + with self.assertRaises(SystemAborted): + self._mutate_and_raise( + manager, + SystemAborted(Unavailable()), + state_id="sibling-1", + ) + + with self.assertRaises(Exception) as raised: + self._mutate_and_raise(manager, None, state_id="sibling-2") + + self.assertNotIsInstance(raised.exception, AssertionError) + + +if __name__ == '__main__': + unittest.main() From 52f686130574a3146ea62252ce75afe8c201a9ac Mon Sep 17 00:00:00 2001 From: Benjamin Hindman Date: Fri, 14 Aug 2026 02:52:57 +0000 Subject: [PATCH 02/10] Propagate errors from unrecoverable transaction aborts to properly retry --- reboot/aio/contexts.py | 17 ++++++++--- reboot/aio/state_managers.py | 12 ++++---- reboot/aio/stubs.py | 21 +++++++++----- tests/reboot/bank.proto | 23 +++++++++++++++ tests/reboot/bank.py | 42 +++++++++++++++++++++++++++ tests/reboot/pydantic/methods/test.py | 8 +++-- tests/reboot/transaction_tests.py | 33 ++++++++++++++++++--- 7 files changed, 134 insertions(+), 22 deletions(-) diff --git a/reboot/aio/contexts.py b/reboot/aio/contexts.py index b2b008f7..f22e8f33 100644 --- a/reboot/aio/contexts.py +++ b/reboot/aio/contexts.py @@ -856,9 +856,18 @@ def use(self) -> Iterator[None]: # opt out. outstanding_rpcs: int - # Whether or not the transaction enclosing this context should - # abort. - transaction_unrecoverable_abort: bool + # The abort that made the transaction enclosing this context + # unrecoverable, or `None` if it does not have to abort. Truthy + # exactly when the transaction must abort, so it also reads as the + # "should abort" flag it replaced. + # + # Holding the abort rather than a bool lets us re-raise it instead + # of a generic "must abort" error, so that a caller which catches + # and discards an abort still gets the same outcome as one which + # lets it propagate, e.g., an `Unavailable` or a + # `TransactionShouldRetryWithoutBackoff` still asks the + # coordinator for a retry. + transaction_unrecoverable_abort: Optional[BaseException] # Extra machinery for handling reactive contexts. Set when using # the `StateManager.reactively()` helper. @@ -920,7 +929,7 @@ def __init__( self.participants = Participants() self.outstanding_rpcs = 0 - self.transaction_unrecoverable_abort = False + self.transaction_unrecoverable_abort = None self.react = None diff --git a/reboot/aio/state_managers.py b/reboot/aio/state_managers.py index 97c6c057..e6fdff59 100644 --- a/reboot/aio/state_managers.py +++ b/reboot/aio/state_managers.py @@ -3782,9 +3782,9 @@ async def transactionally( # stub, marks the caller unrecoverable too, and # cascades up to the coordinator to abort the whole # transaction. - if context.transaction_unrecoverable_abort: + if context.transaction_unrecoverable_abort is not None: transaction.unrecoverable_abort = True - raise RuntimeError('Transaction must abort') + raise context.transaction_unrecoverable_abort except BaseException as exception: # Transaction doesn't need to abort if this is from # the backend and recoverable, i.e., declared or @@ -3794,11 +3794,11 @@ async def transactionally( aborted_type is not None and aborted_type.is_from_backend_and_recoverable(exception) ): - if context.transaction_unrecoverable_abort: + if context.transaction_unrecoverable_abort is not None: # We have a recoverable abort, but the # transaction is already doomed. transaction.unrecoverable_abort = True - raise RuntimeError('Transaction must abort') + raise context.transaction_unrecoverable_abort # We don't need to abort, but we do need to validate # the user is following the transaction requirements. @@ -4681,7 +4681,9 @@ async def complete(effects: Effects) -> None: # abort phase of two phase commit by raising an error here # which will "goto" the `except` block below to actually # run the abort. - if context.transaction_unrecoverable_abort or transaction.unrecoverable_abort: + if context.transaction_unrecoverable_abort is not None: + raise context.transaction_unrecoverable_abort + elif transaction.unrecoverable_abort: raise RuntimeError('Transaction must abort') await self._transaction_coordinator_complete( diff --git a/reboot/aio/stubs.py b/reboot/aio/stubs.py index a8c4b20d..68f41e2a 100644 --- a/reboot/aio/stubs.py +++ b/reboot/aio/stubs.py @@ -532,17 +532,23 @@ async def _call_transactionally( ) if not aborted_type.is_from_backend_and_recoverable(aborted): - # TODO(benh): considering stringifying the exception to - # include in the error we raise when doing the prepare - # stage of two phase commit. - self._context.transaction_unrecoverable_abort = True + # Keep the first abort, which is the one that would + # have propagated had the caller not caught it. + if self._context.transaction_unrecoverable_abort is None: + self._context.transaction_unrecoverable_abort = aborted raise aborted except: + # We don't know what went wrong, so we can't hand the + # caller anything better than a generic error. + # # TODO(benh): considering stringifying the exception to # include in the error we raise when doing the prepare # stage of two phase commit. - self._context.transaction_unrecoverable_abort = True + if self._context.transaction_unrecoverable_abort is None: + self._context.transaction_unrecoverable_abort = RuntimeError( + 'Transaction must abort' + ) raise finally: @@ -562,8 +568,7 @@ async def _call_transactionally( # `Unavailable` so that it can be retried, hopefully after # all servers have been upgraded. if saw_legacy_to_abort: - self._context.transaction_unrecoverable_abort = True - raise SystemAborted( + aborted = SystemAborted( errors_pb2.Unavailable(), message=( "A transaction participant was marked to abort " @@ -571,3 +576,5 @@ async def _call_transactionally( "retry required." ), ) + self._context.transaction_unrecoverable_abort = aborted + raise aborted diff --git a/tests/reboot/bank.proto b/tests/reboot/bank.proto index b60b5234..f5bbdf27 100644 --- a/tests/reboot/bank.proto +++ b/tests/reboot/bank.proto @@ -114,6 +114,15 @@ service AccountMethods { }; } + // Mimics an `Unavailable` the first time it is called and succeeds + // every time after, so that a transaction which aborts because of it + // can succeed once retried. + rpc MimicUnavailableOnce(MimicUnavailableOnceRequest) + returns (google.protobuf.Empty) { + option (rbt.v1alpha1.method).reader = { + }; + } + // Test method for testing behaviour where users raises/throws one // of their own defined errors. rpc Fail(FailRequest) returns (FailResponse) { @@ -214,10 +223,16 @@ message AssetsUnderManagementResponse { uint32 num_accounts = 2; } +message MimicUnavailableOnceRequest {} + message TryCatchUndeclaredErrorRequest { string account_id = 1; } +message TryCatchUnavailableRequest { + string account_id = 1; +} + message TryCatchDeclaredErrorRequest { string account_id = 1; bool read_before = 2; @@ -354,6 +369,14 @@ service BankMethods { }; } + // Test method for testing behaviour in case user catches a + // retryable `Unavailable` from a stub method call. + rpc TryCatchUnavailable(TryCatchUnavailableRequest) + returns (google.protobuf.Empty) { + option (rbt.v1alpha1.method).transaction = { + }; + } + // Test method for testing behaviour in case user catches a declared // error from a stub method call. rpc TryCatchDeclaredError(TryCatchDeclaredErrorRequest) diff --git a/tests/reboot/bank.py b/tests/reboot/bank.py index 89f91e30..2c125c70 100644 --- a/tests/reboot/bank.py +++ b/tests/reboot/bank.py @@ -39,6 +39,11 @@ # support state streaming. class AccountServicer(Account.singleton.Servicer): + # How many times `MimicUnavailableOnce` has been called. A + # transaction's state is rolled back when it aborts, so a retry + # would not observe a count kept in the state itself. + mimic_unavailable_once_calls = 0 + def authorizer(self): return allow() @@ -140,6 +145,22 @@ async def mimic_unavailable( # `Unavailable` which should propagate. raise Account.BalanceAborted(Unavailable()) + async def mimic_unavailable_once( + self, + context: ReaderContext, + state: Account.State, + request: bank_rbt.MimicUnavailableOnceRequest, + ) -> Empty: + AccountServicer.mimic_unavailable_once_calls += 1 + + # Only mimic the `Unavailable` the first time so that a + # transaction which aborts because of it succeeds once + # retried. + if AccountServicer.mimic_unavailable_once_calls == 1: + raise Account.MimicUnavailableOnceAborted(Unavailable()) + + return Empty() + async def fail( self, context: ReaderContext, @@ -446,6 +467,27 @@ async def try_catch_undeclared_error( return Empty() + async def try_catch_unavailable( + self, + context: TransactionContext, + request: bank_rbt.TryCatchUnavailableRequest, + ) -> Empty: + + account = Account.ref(request.account_id) + + try: + await account.mimic_unavailable_once( + context, + Options(bearer_token=context.caller_bearer_token), + ) + except Account.MimicUnavailableOnceAborted: + # Even though we caught it the transaction still aborts, + # and the `Unavailable` must still be what propagates so + # that the transaction gets retried. + assert context.transaction_unrecoverable_abort + + return Empty() + async def try_catch_declared_error( self, context: TransactionContext, diff --git a/tests/reboot/pydantic/methods/test.py b/tests/reboot/pydantic/methods/test.py index 84c0dfc2..815bb0e1 100644 --- a/tests/reboot/pydantic/methods/test.py +++ b/tests/reboot/pydantic/methods/test.py @@ -317,8 +317,10 @@ async def test_transaction(self) -> None: make_unauthorized_call=True, ) + # The abort that doomed the transaction is what propagates, so + # the caller is told what actually went wrong. self.assertIn( - "Transaction must abort", + "You are not authorized to call", str(aborted.exception), ) @@ -382,8 +384,10 @@ async def test_transaction_ongoing_updates(self) -> None: with self.assertRaises(Test.TransactionAborted) as aborted: await test.transaction(context) + # The abort that doomed the transaction is what propagates, so + # the caller is told what actually went wrong. self.assertIn( - "Transaction must abort", + "Simulated failure in transaction_writer", str(aborted.exception), ) diff --git a/tests/reboot/transaction_tests.py b/tests/reboot/transaction_tests.py index bf2f1ae6..6b476209 100644 --- a/tests/reboot/transaction_tests.py +++ b/tests/reboot/transaction_tests.py @@ -1842,8 +1842,9 @@ async def test_transaction_aborts_when_catching_undeclared_errors(self): with self.assertRaises(Bank.TryCatchUndeclaredErrorAborted) as aborted: await bank.TryCatchUndeclaredError(context, account_id='jonathan') - # TODO: better error message than just 'Transaction must abort'. - self.assertIn('Transaction must abort', str(aborted.exception)) + # The abort that doomed the transaction is what propagates, so + # the caller is told what actually went wrong. + self.assertIn('Jazz hands!', str(aborted.exception)) async def test_nested_catch_undeclared_aborts_root(self): """Test that a nested transaction which catches an undeclared @@ -1870,7 +1871,7 @@ async def test_nested_catch_undeclared_aborts_root(self): context, account_id='jonathan' ) - self.assertIn('Transaction must abort', str(aborted.exception)) + self.assertIn('Jazz hands!', str(aborted.exception)) async def test_nested_read_only_doomed_subtree_aborts_root(self): """Test that a three-level, entirely read-only tree where the @@ -1903,7 +1904,31 @@ async def test_nested_read_only_doomed_subtree_aborts_root(self): thrower_account_id='thrower', ) - self.assertIn('Transaction must abort', str(aborted.exception)) + self.assertIn('Jazz hands!', str(aborted.exception)) + + async def test_transaction_retried_when_catching_unavailable(self): + """Test that catching a retryable error still propagates it, so + that the whole transaction gets retried rather than failing + with an error the caller can not act on. + """ + + 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='jonathan') + + AccountServicer.mimic_unavailable_once_calls = 0 + + # `MimicUnavailableOnce` only aborts the first time, so once + # the transaction is retried it succeeds. + await bank.TryCatchUnavailable(context, account_id='jonathan') + + self.assertEqual(AccountServicer.mimic_unavailable_once_calls, 2) async def test_transaction_not_aborts_when_catching_declared_errors(self): """Test that catching a declared error will not abort the transaction. From 0d91681d4afa3b4ca276d302c52df1cb65088077 Mon Sep 17 00:00:00 2001 From: Benjamin Hindman Date: Fri, 14 Aug 2026 03:07:25 +0000 Subject: [PATCH 03/10] Fail early if a transaction has an unrecoverable abort If a developer caught an error that is unrecoverable don't let them make more Reboot method calls which is just wasting resources since the whole transaction needs to abort anyway. --- reboot/aio/stubs.py | 10 ++++++++ tests/reboot/bank.proto | 22 ++++++++++++++++++ tests/reboot/bank.py | 38 +++++++++++++++++++++++++++++++ tests/reboot/transaction_tests.py | 27 ++++++++++++++++++++++ 4 files changed, 97 insertions(+) diff --git a/reboot/aio/stubs.py b/reboot/aio/stubs.py index 68f41e2a..2d32cae5 100644 --- a/reboot/aio/stubs.py +++ b/reboot/aio/stubs.py @@ -333,6 +333,16 @@ async def _call( doing so correctly depending on whether or not we are reactive or in a transaction. """ + if self._context is not None and self._context.transaction_id is not None: + unrecoverable_abort = self._context.transaction_unrecoverable_abort + if unrecoverable_abort is not None: + # The transaction can only abort at this point, so + # refuse to do any more work on its behalf: every + # state we'd touch would take a lock and provisionally + # mutate only to be rolled back, and the failure would + # not surface until the method returned. + raise unrecoverable_abort + if metadata is None: metadata = () diff --git a/tests/reboot/bank.proto b/tests/reboot/bank.proto index f5bbdf27..e7cefecd 100644 --- a/tests/reboot/bank.proto +++ b/tests/reboot/bank.proto @@ -123,6 +123,13 @@ service AccountMethods { }; } + // Does nothing but count that it was called, so that a test can tell + // whether or not the call was actually dispatched. + rpc Ping(PingRequest) returns (google.protobuf.Empty) { + option (rbt.v1alpha1.method).reader = { + }; + } + // Test method for testing behaviour where users raises/throws one // of their own defined errors. rpc Fail(FailRequest) returns (FailResponse) { @@ -225,6 +232,12 @@ message AssetsUnderManagementResponse { message MimicUnavailableOnceRequest {} +message PingRequest {} + +message TryCatchThenCallRequest { + string account_id = 1; +} + message TryCatchUndeclaredErrorRequest { string account_id = 1; } @@ -377,6 +390,15 @@ service BankMethods { }; } + // Test method for testing behaviour in case user catches an + // undeclared error from a stub method call and then tries to make + // another call. + rpc TryCatchThenCall(TryCatchThenCallRequest) + returns (google.protobuf.Empty) { + option (rbt.v1alpha1.method).transaction = { + }; + } + // Test method for testing behaviour in case user catches a declared // error from a stub method call. rpc TryCatchDeclaredError(TryCatchDeclaredErrorRequest) diff --git a/tests/reboot/bank.py b/tests/reboot/bank.py index 2c125c70..be27efaa 100644 --- a/tests/reboot/bank.py +++ b/tests/reboot/bank.py @@ -44,6 +44,10 @@ class AccountServicer(Account.singleton.Servicer): # would not observe a count kept in the state itself. mimic_unavailable_once_calls = 0 + # How many times `Ping` has been called. Kept out of the state for + # the same reason as `mimic_unavailable_once_calls`. + ping_calls = 0 + def authorizer(self): return allow() @@ -161,6 +165,15 @@ async def mimic_unavailable_once( return Empty() + async def ping( + self, + context: ReaderContext, + state: Account.State, + request: bank_rbt.PingRequest, + ) -> Empty: + AccountServicer.ping_calls += 1 + return Empty() + async def fail( self, context: ReaderContext, @@ -467,6 +480,31 @@ async def try_catch_undeclared_error( return Empty() + async def try_catch_then_call( + self, + context: TransactionContext, + request: bank_rbt.TryCatchThenCallRequest, + ) -> Empty: + + account = Account.ref(request.account_id) + + try: + await account.throw_exception( + context, + Options(bearer_token=context.caller_bearer_token), + ) + except Account.ThrowExceptionAborted: + pass + + # The transaction can only abort now, so this call must be + # refused before it is ever dispatched. + await account.ping( + context, + Options(bearer_token=context.caller_bearer_token), + ) + + return Empty() + async def try_catch_unavailable( self, context: TransactionContext, diff --git a/tests/reboot/transaction_tests.py b/tests/reboot/transaction_tests.py index 6b476209..cb8d721e 100644 --- a/tests/reboot/transaction_tests.py +++ b/tests/reboot/transaction_tests.py @@ -1906,6 +1906,33 @@ async def test_nested_read_only_doomed_subtree_aborts_root(self): self.assertIn('Jazz hands!', str(aborted.exception)) + async def test_transaction_refuses_calls_after_catching_abort(self): + """Test that once a caught error has doomed the transaction any + further call is refused before being dispatched, rather than + doing work that will only be rolled back. + """ + + 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='jonathan') + + AccountServicer.ping_calls = 0 + + with self.assertRaises(Bank.TryCatchThenCallAborted) as aborted: + await bank.TryCatchThenCall(context, account_id='jonathan') + + # The abort that doomed the transaction is what propagates ... + self.assertIn('Jazz hands!', str(aborted.exception)) + + # ... and `Ping` never ran. + self.assertEqual(AccountServicer.ping_calls, 0) + async def test_transaction_retried_when_catching_unavailable(self): """Test that catching a retryable error still propagates it, so that the whole transaction gets retried rather than failing From 9dc8af9a7ddffbe3d9777bcd50b9ed8c30d1fbe7 Mon Sep 17 00:00:00 2001 From: Benjamin Hindman Date: Fri, 14 Aug 2026 06:43:48 +0000 Subject: [PATCH 04/10] Fix minor bugs setting OrderedMap degree --- .../ordered_map/v1/ordered_map.proto | 4 + .../collections/ordered_map/v1/ordered_map.py | 70 +++++---- .../ordered_map/v1/ordered_map_tests.py | 138 ++++++++++++++++++ 3 files changed, 185 insertions(+), 27 deletions(-) diff --git a/rbt/std/collections/ordered_map/v1/ordered_map.proto b/rbt/std/collections/ordered_map/v1/ordered_map.proto index dcc42bec..7f8ce375 100644 --- a/rbt/std/collections/ordered_map/v1/ordered_map.proto +++ b/rbt/std/collections/ordered_map/v1/ordered_map.proto @@ -107,6 +107,10 @@ message NodeInsertRequest { reserved "key", "value"; // Map of key to serialized `Value` bytes. map entries = 3; + // The degree to construct the node with when this `Insert` is what + // implicitly constructs it, i.e., when the `OrderedMap` itself is + // being implicitly constructed. + uint32 degree = 4; } message NodeInsertResponse { diff --git a/reboot/std/collections/ordered_map/v1/ordered_map.py b/reboot/std/collections/ordered_map/v1/ordered_map.py index 3c950ba1..593b0135 100644 --- a/reboot/std/collections/ordered_map/v1/ordered_map.py +++ b/reboot/std/collections/ordered_map/v1/ordered_map.py @@ -99,6 +99,10 @@ def _create( state: Node.State, request: NodeCreateRequest, ) -> None: + # A node of degree < 2 can not be split, which `_insert_leaf` + # and `_insert_inner` would spin forever trying to do. + assert request.degree >= 2, f"Invalid `degree` {request.degree}" + state.degree = request.degree state.is_leaf = request.is_leaf state.keys.extend(request.keys) @@ -153,13 +157,14 @@ async def Insert( request: NodeInsertRequest, ) -> NodeInsertResponse: # Implicit construction of root. This should only happen on - # implicit construction of the `OrderedMap`. + # implicit construction of the `OrderedMap`, which tells us + # what `degree` it was configured with. if context.constructor: self._create( context, state, NodeCreateRequest( - degree=DEFAULT_DEGREE, + degree=request.degree or DEFAULT_DEGREE, is_leaf=True, keys=[], ), @@ -351,6 +356,7 @@ async def insert(j: int) -> tuple[int, NodeInsertResponse]: response = await Node.ref(child_id).Insert( context, entries=entries, + degree=state.degree, ) return j, response @@ -701,6 +707,16 @@ def authorizer(self): else: return allow_if(all=[is_app_internal]) + def _check_degree(self, degree: Optional[int]) -> Optional[str]: + """ + Check that `degree` is a degree a node can actually be split + at. Returns an error message if it is not, or `None` if it is + (or was not specified). + """ + if degree is not None and degree < 2: + return f"`degree` must be >= 2, but got {degree}" + return None + def _check_construction_options( self, context, @@ -838,19 +854,22 @@ async def Create( state: OrderedMap.State, request: OrderedMapCreateRequest, ) -> OrderedMapCreateResponse: - if context.constructor: - if request.HasField("degree") and request.degree < 2: - raise OrderedMap.CreateAborted( - InvalidArgument(), - message="`degree` must be >= 2", - ) - else: + degree = request.degree if request.HasField("degree") else None + + message = self._check_degree(degree) + if message is not None: + raise OrderedMap.CreateAborted( + InvalidArgument(), + message=message, + ) + + if not context.constructor: # Already constructed. Allow if the configuration matches; # reject if it differs. message = self._check_construction_options( context, state, - degree=request.degree if request.HasField("degree") else None, + degree=degree, maintain_size=request.maintain_size, ) if message is not None: @@ -860,15 +879,13 @@ async def Create( ) return OrderedMapCreateResponse() - state.degree = ( - request.degree if request.HasField("degree") else DEFAULT_DEGREE - ) + state.degree = degree if degree is not None else DEFAULT_DEGREE state.maintain_size = request.maintain_size state.root_id = str(uuid.uuid4()) await Node.ref(state.root_id).Create( context, - degree=request.degree, + degree=state.degree, is_leaf=True, keys=[], ) @@ -911,18 +928,18 @@ async def Insert( state: OrderedMap.State, request: OrderedMapInsertRequest, ) -> OrderedMapInsertResponse: - # Construct the map if not yet constructed. - if context.constructor: - if request.HasField("degree") and request.degree < 2: - raise OrderedMap.InsertAborted( - InvalidArgument(), - message="`degree` must be >= 2", - ) + degree = request.degree if request.HasField("degree") else None - state.degree = ( - request.degree - if request.HasField("degree") else DEFAULT_DEGREE + message = self._check_degree(degree) + if message is not None: + raise OrderedMap.InsertAborted( + InvalidArgument(), + message=message, ) + + # Construct the map if not yet constructed. + if context.constructor: + state.degree = degree if degree is not None else DEFAULT_DEGREE state.maintain_size = ( request.maintain_size if request.HasField("maintain_size") else False @@ -938,9 +955,7 @@ async def Insert( message = self._check_construction_options( context, state, - degree=( - request.degree if request.HasField("degree") else None - ), + degree=degree, maintain_size=( request.maintain_size if request.HasField("maintain_size") else None @@ -992,6 +1007,7 @@ async def Insert( response = await root.Insert( context, entries=entries, + degree=state.degree, ) if state.maintain_size: diff --git a/tests/reboot/std/collections/ordered_map/v1/ordered_map_tests.py b/tests/reboot/std/collections/ordered_map/v1/ordered_map_tests.py index f0a339b2..16853664 100644 --- a/tests/reboot/std/collections/ordered_map/v1/ordered_map_tests.py +++ b/tests/reboot/std/collections/ordered_map/v1/ordered_map_tests.py @@ -1886,6 +1886,144 @@ async def test_maintain_size_false(self) -> None: response = await ordered_map.Range(context, limit=1) self.assertFalse(response.HasField("total_size")) + async def test_create_without_degree_then_insert(self) -> None: + """ + Test that a `Create` without an explicit `degree` gives the + root node the default degree, rather than no degree at all. + """ + await self.rbt.up(Application( + libraries=[ordered_map_library()], + )) + + context = self.rbt.create_external_context( + name=f"test-{self.id()}", + app_internal=True, + ) + + ordered_map = OrderedMap.ref("test-map") + + await ordered_map.Create(context) + + await ordered_map.Insert(context, key="a", value=from_str("A")) + + response = await ordered_map.Search(context, key="a") + self.assertTrue(response.found) + self.assertEqual(as_str(response.value), "A") + + # The default degree is far more than one key, so the root is + # still a single leaf. + response = await ordered_map.Stringify(context) + self.assertEqual(response.value, "Leaf: ['a']\n") + + async def test_invalid_degree_on_create(self) -> None: + """ + Test that a `Create` with a `degree` a node could never be + split at is refused with an `InvalidArgument`. + """ + await self.rbt.up(Application( + libraries=[ordered_map_library()], + )) + + context = self.rbt.create_external_context( + name=f"test-{self.id()}", + app_internal=True, + ) + + ordered_map = OrderedMap.ref("test-map") + + with self.assertRaises(OrderedMap.CreateAborted) as raised: + await ordered_map.Create(context, degree=1) + + self.assertIsInstance(raised.exception.error, InvalidArgument) + self.assertIn("`degree` must be >= 2", str(raised.exception)) + + async def test_invalid_degree_on_implicit_construction(self) -> None: + """ + Test that an `Insert` which would implicitly construct the map + with a `degree` a node could never be split at is refused with + an `InvalidArgument`. + """ + await self.rbt.up(Application( + libraries=[ordered_map_library()], + )) + + context = self.rbt.create_external_context( + name=f"test-{self.id()}", + app_internal=True, + ) + + ordered_map = OrderedMap.ref("test-map") + + with self.assertRaises(OrderedMap.InsertAborted) as raised: + await ordered_map.Insert( + context, + key="a", + value=from_str("A"), + degree=1, + ) + + self.assertIsInstance(raised.exception.error, InvalidArgument) + self.assertIn("`degree` must be >= 2", str(raised.exception)) + + async def test_invalid_degree_when_already_created(self) -> None: + """ + Test that an invalid `degree` for a map that already exists is + refused as invalid, rather than as a mismatch with the degree + the map was created with. + """ + await self.rbt.up(Application( + libraries=[ordered_map_library()], + )) + + context = self.rbt.create_external_context( + name=f"test-{self.id()}", + app_internal=True, + ) + + ordered_map = OrderedMap.ref("test-map") + + await ordered_map.Create(context, degree=4) + + with self.assertRaises(OrderedMap.CreateAborted) as raised: + await ordered_map.Create(context, degree=1) + + self.assertIsInstance(raised.exception.error, InvalidArgument) + self.assertIn("`degree` must be >= 2", str(raised.exception)) + + async def test_implicit_construction_uses_requested_degree(self) -> None: + """ + Test that an `Insert` which implicitly constructs the map gives + the root node the `degree` that was asked for, rather than the + default degree. + """ + await self.rbt.up(Application( + libraries=[ordered_map_library()], + )) + + context = self.rbt.create_external_context( + name=f"test-{self.id()}", + app_internal=True, + ) + + ordered_map = OrderedMap.ref("test-map") + + # No `Create`, so the first `Insert` constructs the map. + for key in ["b", "m", "z", "y"]: + await ordered_map.Insert( + context, + key=key, + value=from_str(key.upper()), + degree=4, + ) + + # A degree of 4 splits on the fourth key; had the root node + # been given the default degree it would still be one leaf. + response = await ordered_map.Stringify(context) + self.assertEqual( + response.value, + "Inner: ['y']\n Leaf: ['b', 'm']\n Leaf: ['y', 'z']\n", + ) + if __name__ == '__main__': unittest.main() From 59f936ad69e51062a08cccd2edfa30c69b9b1f07 Mon Sep 17 00:00:00 2001 From: Benjamin Hindman Date: Fri, 14 Aug 2026 20:07:03 +0000 Subject: [PATCH 05/10] Correctly handle concurrent `IdempotencyManager.idempotently()` We need to correctly handle multiple calls to `IdempotencyManager.idempotently()` which may create uncertain mutations that we want to properly handle. --- reboot/aio/idempotency.py | 10 +- tests/reboot/idempotency_uncertainty_tests.py | 115 ++++++++++++------ 2 files changed, 83 insertions(+), 42 deletions(-) diff --git a/reboot/aio/idempotency.py b/reboot/aio/idempotency.py index 48f676a4..f39db357 100644 --- a/reboot/aio/idempotency.py +++ b/reboot/aio/idempotency.py @@ -485,8 +485,14 @@ def idempotently( # manually _retry_ another call that did not have an # idempotency key and accidentally perform a mutation more # than once. - if self._mutations_without_idempotency: - assert not self._uncertain_mutation + # + # Concurrent mutations may fail together, in which case + # they all end up here; keep the first one, which is the + # one a later mutation will be told about. + if ( + self._mutations_without_idempotency and + not self._uncertain_mutation + ): self._uncertain_mutation = True self._uncertain_mutation_state_type_name = state_type_name self._uncertain_mutation_state_ref = state_ref diff --git a/tests/reboot/idempotency_uncertainty_tests.py b/tests/reboot/idempotency_uncertainty_tests.py index bd13aa9f..84457d95 100644 --- a/tests/reboot/idempotency_uncertainty_tests.py +++ b/tests/reboot/idempotency_uncertainty_tests.py @@ -1,3 +1,4 @@ +import asyncio import unittest from google.protobuf.message import Message from rbt.v1alpha1.errors_pb2 import ( @@ -5,10 +6,12 @@ Unavailable, ) from reboot.aio.aborted import SystemAborted -from reboot.aio.idempotency import IdempotencyManager +from reboot.aio.idempotency import ( + IdempotencyManager, + IdempotencyUncertainError, +) from reboot.aio.types import ServiceName, StateRef, StateTypeName from reboot.api import Model -from typing import Optional class DeclaresNothingAborted(SystemAborted): @@ -20,7 +23,7 @@ def is_declared_error(cls, error: Message | Model) -> bool: return False -class UncertainMutationTestCase(unittest.TestCase): +class UncertainMutationTestCase(unittest.IsolatedAsyncioTestCase): """ Tests which failed mutations make an `IdempotencyManager` uncertain. A `Node.Insert` that splits creates its siblings @@ -31,16 +34,9 @@ class UncertainMutationTestCase(unittest.TestCase): STATE_TYPE_NAME = StateTypeName("test.v1.Node") SERVICE_NAME = ServiceName("test.v1.NodeMethods") - def _mutate_and_raise( - self, - manager: IdempotencyManager, - exception: Optional[BaseException], - *, - state_id: str, - ) -> None: - """Performs a mutation without idempotency, failing it with - `exception` if one is given.""" - with manager.idempotently( + def _idempotently(self, manager: IdempotencyManager, *, state_id: str): + """A mutation without idempotency on `state_id`.""" + return manager.idempotently( state_type_name=self.STATE_TYPE_NAME, state_ref=StateRef.from_id(self.STATE_TYPE_NAME, state_id), service_name=self.SERVICE_NAME, @@ -50,46 +46,85 @@ def _mutate_and_raise( metadata=None, idempotency=None, aborted_type=DeclaresNothingAborted, - ): - if exception is not None: - raise exception + ) - def test_concurrent_retryable_failures_do_not_become_uncertain(self): - # Several sibling `Create`s failing with the same retryable - # abort must all propagate that abort so the transaction is - # retried, rather than the second one turning it into an - # `AssertionError`. + def test_retryable_failures_do_not_become_uncertain(self): + # Sibling `Create`s failing with a retryable abort must all + # propagate that abort so the transaction is retried, rather + # than being recorded as uncertain mutations. manager = IdempotencyManager() for state_id in ["sibling-1", "sibling-2", "sibling-3"]: with self.assertRaises(SystemAborted): - self._mutate_and_raise( - manager, - SystemAborted(TransactionShouldRetryWithoutBackoff()), - state_id=state_id, - ) + with self._idempotently(manager, state_id=state_id): + raise SystemAborted(TransactionShouldRetryWithoutBackoff()) # A later mutation must still be allowed, i.e., nothing was # recorded as uncertain. - self._mutate_and_raise(manager, None, state_id="sibling-4") + with self._idempotently(manager, state_id="sibling-4"): + pass - def test_concurrent_transport_failures_report_uncertainty(self): - # An `Unavailable` may or may not have mutated, so the first - # failure makes the manager uncertain and the next mutation is - # refused -- with an actionable error, not an `AssertionError`. + async def test_concurrent_failures_keep_the_first(self): + # Both mutations are in flight before either fails, so both + # get past the "are we uncertain?" check and both end up + # recording uncertainty. manager = IdempotencyManager() - with self.assertRaises(SystemAborted): - self._mutate_and_raise( - manager, - SystemAborted(Unavailable()), - state_id="sibling-1", - ) + first_exception = SystemAborted(Unavailable()) + second_exception = SystemAborted(Unavailable()) - with self.assertRaises(Exception) as raised: - self._mutate_and_raise(manager, None, state_id="sibling-2") + first_begun = asyncio.Event() + second_begun = asyncio.Event() + first_fails = asyncio.Event() + second_fails = asyncio.Event() + + async def mutate(state_id, exception, begun, fails): + with self._idempotently(manager, state_id=state_id): + begun.set() + await fails.wait() + raise exception + + first = asyncio.create_task( + mutate("first", first_exception, first_begun, first_fails) + ) + second = asyncio.create_task( + mutate("second", second_exception, second_begun, second_fails) + ) + + await first_begun.wait() + await second_begun.wait() + + # Both mutations are now inside `idempotently()`; fail them + # one at a time so we know which of them failed first. + first_fails.set() + with self.assertRaises(SystemAborted) as raised: + await first + self.assertIs(raised.exception, first_exception) + + second_fails.set() + with self.assertRaises(SystemAborted) as raised: + await second + self.assertIs(raised.exception, second_exception) + + # The first mutation to fail is the one we report as uncertain. + with self.assertRaises(IdempotencyUncertainError) as raised: + with self._idempotently(manager, state_id="later"): + pass + + self.assertIn("'first'", str(raised.exception)) + + def test_transport_failure_reports_uncertainty(self): + # An `Unavailable` may or may not have mutated, so it makes the + # manager uncertain and the next mutation is refused. + manager = IdempotencyManager() + + with self.assertRaises(SystemAborted): + with self._idempotently(manager, state_id="sibling-1"): + raise SystemAborted(Unavailable()) - self.assertNotIsInstance(raised.exception, AssertionError) + with self.assertRaises(IdempotencyUncertainError): + with self._idempotently(manager, state_id="sibling-2"): + pass if __name__ == '__main__': From 1b07797b72a34039f0d38f299613e242834753ee Mon Sep 17 00:00:00 2001 From: Benjamin Hindman Date: Fri, 14 Aug 2026 20:30:45 +0000 Subject: [PATCH 06/10] Ensure `Node` instances of an `OrderedMap` get the right degree --- .../collections/ordered_map/v1/ordered_map.py | 8 +++++ .../ordered_map/v1/ordered_map_tests.py | 36 +++++++++++++++++++ 2 files changed, 44 insertions(+) diff --git a/reboot/std/collections/ordered_map/v1/ordered_map.py b/reboot/std/collections/ordered_map/v1/ordered_map.py index 593b0135..fd48f300 100644 --- a/reboot/std/collections/ordered_map/v1/ordered_map.py +++ b/reboot/std/collections/ordered_map/v1/ordered_map.py @@ -169,6 +169,14 @@ async def Insert( keys=[], ), ) + elif state.degree == 0: + # Due to a bug in an older version of `OrderedMap.Create` + # that passed the root `Node` a request with an unset + # `degree` instead of the default a `Node` can be + # persisted without a `degree`. Such a node can not be + # split, so adopt the `degree` the `OrderedMap` recorded, + # which is the one the map was created with. + state.degree = request.degree or DEFAULT_DEGREE if state.is_leaf: return await self._insert_leaf( diff --git a/tests/reboot/std/collections/ordered_map/v1/ordered_map_tests.py b/tests/reboot/std/collections/ordered_map/v1/ordered_map_tests.py index 16853664..da96cc65 100644 --- a/tests/reboot/std/collections/ordered_map/v1/ordered_map_tests.py +++ b/tests/reboot/std/collections/ordered_map/v1/ordered_map_tests.py @@ -1,6 +1,10 @@ import unittest from google.protobuf.any_pb2 import Any from google.protobuf.struct_pb2 import Value +from rbt.std.collections.ordered_map.v1.ordered_map_rbt import ( + Node, + NodeInsertRequest, +) from rbt.std.item.v1.item_pb2 import Item from rbt.v1alpha1.errors_pb2 import ( InvalidArgument, @@ -13,9 +17,11 @@ from reboot.protobuf import as_str, from_str from reboot.std.collections.ordered_map.v1.ordered_map import ( InvalidRangeError, + NodeServicer, OrderedMap, ordered_map_library, ) +from unittest.mock import MagicMock class TestOrderedMap(unittest.IsolatedAsyncioTestCase): @@ -2024,6 +2030,36 @@ async def test_implicit_construction_uses_requested_degree(self) -> None: "Inner: ['y']\n Leaf: ['b', 'm']\n Leaf: ['y', 'z']\n", ) + async def test_insert_repairs_unusable_persisted_degree(self) -> None: + """ + Due to a bug in an older version of `OrderedMap.Create` that passed + the root `Node` a request with an unset `degree` instead of + the default a `Node` can be persisted without a `degree`. Such + a node can not be split. This test checks that we can "fix" a + persisted instance of an `OrderedMap` if an app has one. + + Calls `NodeServicer` directly because such a node can no longer + be created through `OrderedMap`, which is the point; only + already-persisted nodes are in this state. + """ + servicer = NodeServicer() + + state = Node.State() + state.is_leaf = True + state.degree = 0 + + context = MagicMock() + context.constructor = False + + await servicer.Insert( + context, + state, + NodeInsertRequest(entries={"a": b""}, degree=16), + ) + + self.assertEqual(state.degree, 16) + self.assertEqual(list(state.keys), ["a"]) + if __name__ == '__main__': unittest.main() From a9a32736f0b2e265a45fbd1c837389f8be8a8a81 Mon Sep 17 00:00:00 2001 From: Benjamin Hindman Date: Fri, 14 Aug 2026 20:32:39 +0000 Subject: [PATCH 07/10] Fix documentation drift --- documentation/docs/learn_more/auth.mdx | 6 ------ 1 file changed, 6 deletions(-) diff --git a/documentation/docs/learn_more/auth.mdx b/documentation/docs/learn_more/auth.mdx index aeb53a38..5557207c 100644 --- a/documentation/docs/learn_more/auth.mdx +++ b/documentation/docs/learn_more/auth.mdx @@ -25,18 +25,12 @@ instance, you can specify it in `ref()`: - - - ```py from_account = Account.ref( request.from_account_id, bearer_token=bearer_token, ) ``` - - From 94b238fd66ae32b6cc608433b7ae95774f1bf72a Mon Sep 17 00:00:00 2001 From: Benjamin Hindman Date: Fri, 14 Aug 2026 20:46:36 +0000 Subject: [PATCH 08/10] Address review comments --- reboot/aio/contexts.py | 4 +- .../collections/ordered_map/v1/ordered_map.py | 127 +++++++++--------- tests/reboot/zod/methods/test.ts | 4 +- 3 files changed, 68 insertions(+), 67 deletions(-) diff --git a/reboot/aio/contexts.py b/reboot/aio/contexts.py index f22e8f33..b814275c 100644 --- a/reboot/aio/contexts.py +++ b/reboot/aio/contexts.py @@ -857,9 +857,7 @@ def use(self) -> Iterator[None]: outstanding_rpcs: int # The abort that made the transaction enclosing this context - # unrecoverable, or `None` if it does not have to abort. Truthy - # exactly when the transaction must abort, so it also reads as the - # "should abort" flag it replaced. + # unrecoverable, or `None` if it does not have to abort. # # Holding the abort rather than a bool lets us re-raise it instead # of a generic "must abort" error, so that a caller which catches diff --git a/reboot/std/collections/ordered_map/v1/ordered_map.py b/reboot/std/collections/ordered_map/v1/ordered_map.py index fd48f300..97dda51e 100644 --- a/reboot/std/collections/ordered_map/v1/ordered_map.py +++ b/reboot/std/collections/ordered_map/v1/ordered_map.py @@ -76,6 +76,17 @@ DEFAULT_DEGREE = 128 +def _check_degree(degree: Optional[int]) -> Optional[str]: + """ + Check that `degree` is a degree a node can actually be split at. + Returns an error message if it is not, or `None` if it is (or was + not specified). + """ + if degree is not None and degree < 2: + return f"`degree` must be >= 2, but got {degree}" + return None + + def _item_to_value(item: Item) -> Value: """Helper that converts an `Item` to a `Value`.""" value = Value() @@ -99,9 +110,10 @@ def _create( state: Node.State, request: NodeCreateRequest, ) -> None: - # A node of degree < 2 can not be split, which `_insert_leaf` - # and `_insert_inner` would spin forever trying to do. - assert request.degree >= 2, f"Invalid `degree` {request.degree}" + # A node we can not split is one `_insert_leaf` and + # `_insert_inner` would spin forever trying to split. + message = _check_degree(request.degree) + assert message is None, message state.degree = request.degree state.is_leaf = request.is_leaf @@ -715,28 +727,34 @@ def authorizer(self): else: return allow_if(all=[is_app_internal]) - def _check_degree(self, degree: Optional[int]) -> Optional[str]: - """ - Check that `degree` is a degree a node can actually be split - at. Returns an error message if it is not, or `None` if it is - (or was not specified). - """ - if degree is not None and degree < 2: - return f"`degree` must be >= 2, but got {degree}" - return None - def _check_construction_options( self, context, state: OrderedMap.State, degree: Optional[int], maintain_size: Optional[bool], - ) -> Optional[str]: + ) -> tuple[ + Optional[InvalidArgument | StateAlreadyConstructed], + Optional[str], + ]: """ - Check that construction options match an already-constructed - map. Returns an error message if they don't match, or `None` - if they match (or were not specified). + Check the construction options: that they are valid at all, and + that they match an already-constructed map. Returns the error + to abort with and a message describing it, or `(None, None)` if + the options are good (or were not specified). + + An option that could never be valid is an `InvalidArgument`; an + option that is valid but conflicts with the map as it was + created is a `StateAlreadyConstructed`. """ + message = _check_degree(degree) + if message is not None: + return InvalidArgument(), message + + if context.constructor: + # Nothing to match against yet. + return None, None + errors: list[str] = [] if degree is not None and degree != state.degree: errors.append(f"degree={state.degree}, requested degree={degree}") @@ -746,14 +764,14 @@ def _check_construction_options( f"maintain_size={maintain_size}" ) if errors: - return ( + return StateAlreadyConstructed(), ( f"OrderedMap with ID '{context.state_id}' already " f"created with {'; '.join(errors)}. `degree` and " "`maintain_size` are only required when implicitly " "constructing the map; if it already exists, omit " "them or pass values that match the existing map." ) - return None + return None, None async def _build_root_levels( self, @@ -864,30 +882,22 @@ async def Create( ) -> OrderedMapCreateResponse: degree = request.degree if request.HasField("degree") else None - message = self._check_degree(degree) - if message is not None: - raise OrderedMap.CreateAborted( - InvalidArgument(), - message=message, - ) + # Allow if the configuration is valid and, if the map is + # already constructed, matches; reject if it differs. + error, message = self._check_construction_options( + context, + state, + degree=degree, + maintain_size=request.maintain_size, + ) + if error is not None: + assert message is not None + raise OrderedMap.CreateAborted(error, message=message) if not context.constructor: - # Already constructed. Allow if the configuration matches; - # reject if it differs. - message = self._check_construction_options( - context, - state, - degree=degree, - maintain_size=request.maintain_size, - ) - if message is not None: - raise OrderedMap.CreateAborted( - StateAlreadyConstructed(), - message=message, - ) return OrderedMapCreateResponse() - state.degree = degree if degree is not None else DEFAULT_DEGREE + state.degree = degree or DEFAULT_DEGREE state.maintain_size = request.maintain_size state.root_id = str(uuid.uuid4()) @@ -938,16 +948,24 @@ async def Insert( ) -> OrderedMapInsertResponse: degree = request.degree if request.HasField("degree") else None - message = self._check_degree(degree) - if message is not None: - raise OrderedMap.InsertAborted( - InvalidArgument(), - message=message, - ) + # Validate that the configuration is valid and, if the map is + # already constructed, that it matches. + error, message = self._check_construction_options( + context, + state, + degree=degree, + maintain_size=( + request.maintain_size + if request.HasField("maintain_size") else None + ), + ) + if error is not None: + assert message is not None + raise OrderedMap.InsertAborted(error, message=message) # Construct the map if not yet constructed. if context.constructor: - state.degree = degree if degree is not None else DEFAULT_DEGREE + state.degree = degree or DEFAULT_DEGREE state.maintain_size = ( request.maintain_size if request.HasField("maintain_size") else False @@ -957,23 +975,6 @@ async def Insert( # of the `Node` because it is a transaction, and nested transactions # cannot touch the data a parent touches. However, we do need # to give it an ID. - else: - # Already constructed. Validate that any construction - # options match. - message = self._check_construction_options( - context, - state, - degree=degree, - maintain_size=( - request.maintain_size - if request.HasField("maintain_size") else None - ), - ) - if message is not None: - raise OrderedMap.InsertAborted( - InvalidArgument(), - message=message, - ) # Build node entries from either the bulk `entries` field or # the single-key fields. diff --git a/tests/reboot/zod/methods/test.ts b/tests/reboot/zod/methods/test.ts index 79d7a0d4..3c2bc75b 100644 --- a/tests/reboot/zod/methods/test.ts +++ b/tests/reboot/zod/methods/test.ts @@ -57,7 +57,9 @@ test("Test Zod Service", async (t) => { await test.transaction(context); } catch (aborted) { assert.ok(aborted instanceof Error); - assert.ok(aborted.message.includes("Transaction must abort")); + assert.ok( + aborted.message.includes("Simulated failure in transactionWriter") + ); } }); }); From 2e860ffc9b3a6bbb1823734db2d41812aa82c717 Mon Sep 17 00:00:00 2001 From: Benjamin Hindman Date: Fri, 14 Aug 2026 20:50:42 +0000 Subject: [PATCH 09/10] Fix stale comment --- reboot/std/collections/ordered_map/v1/ordered_map.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/reboot/std/collections/ordered_map/v1/ordered_map.py b/reboot/std/collections/ordered_map/v1/ordered_map.py index 97dda51e..f602c1d7 100644 --- a/reboot/std/collections/ordered_map/v1/ordered_map.py +++ b/reboot/std/collections/ordered_map/v1/ordered_map.py @@ -970,11 +970,10 @@ async def Insert( request.maintain_size if request.HasField("maintain_size") else False ) + # Need an ID for the root `Node`; the node itself is + # constructed by the `Insert` below, which is passed the + # `degree` to construct it with. state.root_id = str(uuid.uuid4()) - # Must allow `Node.Insert` to perform the implicit construction - # of the `Node` because it is a transaction, and nested transactions - # cannot touch the data a parent touches. However, we do need - # to give it an ID. # Build node entries from either the bulk `entries` field or # the single-key fields. From bfd65eaa4043f0ce9072c8dee50be055e63fa32a Mon Sep 17 00:00:00 2001 From: Benjamin Hindman Date: Fri, 14 Aug 2026 21:13:27 +0000 Subject: [PATCH 10/10] Add more backend errors that are recoverable --- reboot/aio/aborted.py | 50 ++++++++++++++----- tests/reboot/aborted_tests.py | 41 +++++++++++++++ tests/reboot/idempotency_uncertainty_tests.py | 4 +- 3 files changed, 80 insertions(+), 15 deletions(-) diff --git a/reboot/aio/aborted.py b/reboot/aio/aborted.py index e20af258..fe106bd6 100644 --- a/reboot/aio/aborted.py +++ b/reboot/aio/aborted.py @@ -105,18 +105,44 @@ def is_retryable(aborted: Aborted): FROM_BACKEND_AND_RECOVERABLE_ERROR_TYPES: tuple[type[Message], ...] = ( rbt.v1alpha1.errors_pb2.StateNotConstructed, rbt.v1alpha1.errors_pb2.StateAlreadyConstructed, + # Status codes that the gRPC library never generates, only user + # code: https://grpc.io/docs/guides/status-codes/ + # + # A backend raising one of these persisted nothing, so nothing + # about the transaction is doomed: a developer can branch on the + # error and go on to finish the transaction some other way. + rbt.v1alpha1.errors_pb2.InvalidArgument, + rbt.v1alpha1.errors_pb2.NotFound, + rbt.v1alpha1.errors_pb2.AlreadyExists, + rbt.v1alpha1.errors_pb2.FailedPrecondition, + rbt.v1alpha1.errors_pb2.Aborted, + rbt.v1alpha1.errors_pb2.OutOfRange, + rbt.v1alpha1.errors_pb2.DataLoss, ) -# Errors that only Reboot generates, never a proxy or other component, -# but which require the transaction to abort and be retried rather -# than commit. +# Errors that tell us a backend raised them, and thus that no mutation +# happened, but which a transaction can not commit through. FROM_BACKEND_AND_UNRECOVERABLE_ERROR_TYPES: tuple[type[Message], ...] = ( - # Raised when a participant joins a transaction that started - # before the participant last recovered, i.e., before the - # participant ran any of the transaction's code. + # Raised by a participant refusing to take part in a transaction + # which started before the participant last recovered, i.e., + # before the participant ran any of the transaction's code. rbt.v1alpha1.errors_pb2.TransactionShouldRetryWithoutBackoff, ) +# Every error that tells us a backend raised it. The recoverable and +# unrecoverable types above partition this: each error belongs to +# exactly one of them, and together they are every error we know a +# backend to be the source of. +FROM_BACKEND_ERROR_TYPES: tuple[type[Message], ...] = ( + FROM_BACKEND_AND_RECOVERABLE_ERROR_TYPES + + FROM_BACKEND_AND_UNRECOVERABLE_ERROR_TYPES +) + +assert len( + set(FROM_BACKEND_ERROR_TYPES) +) == len(FROM_BACKEND_ERROR_TYPES + ), ("An error is either recoverable or it is not, never both") + # Any possible error type, i.e., possibly a `GrpcError`, a # `RebootError`, or a user declared error. ErrorT = TypeVar('ErrorT', bound=Message) @@ -457,15 +483,13 @@ def is_from_backend( This is a weaker property than `is_from_backend_and_recoverable()`: an error may tell us that - no mutation happened while still requiring the transaction to - abort and be retried. + no mutation happened while still leaving the transaction + unable to commit. """ return ( - cls.is_from_backend_and_recoverable(exception) or ( - isinstance(exception, Aborted) and isinstance( - exception.error, - FROM_BACKEND_AND_UNRECOVERABLE_ERROR_TYPES, - ) + isinstance(exception, Aborted) and ( + cls.is_declared_error(exception.error) or + isinstance(exception.error, FROM_BACKEND_ERROR_TYPES) ) ) diff --git a/tests/reboot/aborted_tests.py b/tests/reboot/aborted_tests.py index dda165aa..10a78301 100644 --- a/tests/reboot/aborted_tests.py +++ b/tests/reboot/aborted_tests.py @@ -1,8 +1,16 @@ import unittest from google.protobuf.message import Message from rbt.v1alpha1.errors_pb2 import ( + Aborted, + AlreadyExists, + DataLoss, + FailedPrecondition, + InvalidArgument, + NotFound, + OutOfRange, StateAlreadyConstructed, StateNotConstructed, + TransactionParticipantFailedToCommit, TransactionShouldRetryWithoutBackoff, Unavailable, Unknown, @@ -57,6 +65,39 @@ def test_construction_errors_are_recoverable(self): DeclaresNothingAborted.is_from_backend(aborted), ) + def test_user_code_only_status_codes_are_recoverable(self): + # The gRPC library never generates these, only user code, so + # one of them reaching us means a backend raised it and + # persisted nothing: + # https://grpc.io/docs/guides/status-codes/ + for error in ( + InvalidArgument(), + NotFound(), + AlreadyExists(), + FailedPrecondition(), + Aborted(), + OutOfRange(), + DataLoss(), + ): + with self.subTest(error=type(error).__name__): + aborted = SystemAborted(error) + self.assertTrue( + DeclaresNothingAborted.is_from_backend(aborted), + ) + # Nothing about the transaction is doomed, so a + # developer can catch one and still finish it. + self.assertTrue( + DeclaresNothingAborted. + is_from_backend_and_recoverable(aborted), + ) + + def test_failed_to_commit_is_not_from_backend(self): + # Raised once other participants may already have committed, so + # it does not tell us whether a mutation happened. + aborted = SystemAborted(TransactionParticipantFailedToCommit()) + + self.assertFalse(DeclaresNothingAborted.is_from_backend(aborted)) + def test_transport_errors_are_neither(self): # These may be raised by a proxy or the network before or after # the server processed the call, so we can not know whether a diff --git a/tests/reboot/idempotency_uncertainty_tests.py b/tests/reboot/idempotency_uncertainty_tests.py index 84457d95..017f42f1 100644 --- a/tests/reboot/idempotency_uncertainty_tests.py +++ b/tests/reboot/idempotency_uncertainty_tests.py @@ -107,11 +107,11 @@ async def mutate(state_id, exception, begun, fails): self.assertIs(raised.exception, second_exception) # The first mutation to fail is the one we report as uncertain. - with self.assertRaises(IdempotencyUncertainError) as raised: + with self.assertRaises(IdempotencyUncertainError) as uncertain: with self._idempotently(manager, state_id="later"): pass - self.assertIn("'first'", str(raised.exception)) + self.assertIn("'first'", str(uncertain.exception)) def test_transport_failure_reports_uncertainty(self): # An `Unavailable` may or may not have mutated, so it makes the