Fix uncertain idempotency on transaction should retry without backoff

documentation/docs/learn_more/auth.mdx

@@ -25,18 +25,12 @@ instance, you can specify it in `ref()`:
 
 <Tabs groupId="language">
   <TabItem value="python" label="Python">
-<!-- MARKDOWN-AUTO-DOCS:START
-(CODE:src=../../../tests/reboot/bank.py&lines=326-329) -->
-<!-- The below code snippet is automatically added from ../../../tests/reboot/bank.py -->
-
 ```py
 from_account = Account.ref(
     request.from_account_id,
     bearer_token=bearer_token,
 )
 ```
-
-<!-- MARKDOWN-AUTO-DOCS:END -->
   </TabItem>
 
   <TabItem value="typescript" label="TypeScript">

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<string, bytes> 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 {

reboot/aio/aborted.py

@@ -95,6 +95,54 @@ 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,
+    # 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 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 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)
@@ -415,25 +463,36 @@ 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(
-                    exception.error,
-                    rbt.v1alpha1.errors_pb2.StateNotConstructed,
-                ) or isinstance(
+                cls.is_declared_error(exception.error) or isinstance(
                     exception.error,
-                    rbt.v1alpha1.errors_pb2.StateAlreadyConstructed,
+                    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 leaving the transaction
+        unable to commit.
+        """
+        return (
+            isinstance(exception, Aborted) and (
+                cls.is_declared_error(exception.error) or
+                isinstance(exception.error, FROM_BACKEND_ERROR_TYPES)
+            )
+        )
+
 
 class SystemAborted(Aborted):
     """Encapsulates errors due to the system aborting."""

reboot/aio/contexts.py

@@ -856,9 +856,16 @@ 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.
+    #
+    # 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 +927,7 @@ def __init__(
 
         self.participants = Participants()
         self.outstanding_rpcs = 0
-        self.transaction_unrecoverable_abort = False
+        self.transaction_unrecoverable_abort = None
 
         self.react = None
 

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

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(

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 = ()
 
@@ -532,17 +542,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,12 +578,13 @@ 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 "
                         "by an older server during a rolling upgrade; "
                         "retry required."
                     ),
                 )
+                self._context.transaction_unrecoverable_abort = aborted
+                raise aborted