feat: do not rethrow RuntimeStatusError from a Commit() mutator (#1320)

Treat the exception thrown by the mutator as if the enclosed `Status`
had been returned instead (OK is converted to an "unknown" status).
This allows the transaction to be rerun (when the status is transient).

Author: devbww

google/cloud/spanner/client.cc

@@ -178,6 +178,13 @@ StatusOr<CommitResult> Client::Commit(
 #endif
       mutations = mutator(txn);
 #if GOOGLE_CLOUD_CPP_HAVE_EXCEPTIONS
+    } catch (RuntimeStatusError const& error) {
+      // Treat this like mutator() returned a bad Status.
+      Status status = error.status();
+      if (status.ok()) {
+        status = Status(StatusCode::kUnknown, "OK Status thrown from mutator");
+      }
+      mutations = status;
     } catch (...) {
       auto rb_status = Rollback(txn);
       if (!RerunnablePolicy::IsOk(rb_status)) {

google/cloud/spanner/client.h

@@ -461,7 +461,9 @@ class Client {
    * @param backoff_policy controls how long `Commit` waits between reruns.
    *
    * @throw Rethrows any exception thrown by @p `mutator` (after rolling back
-   *     the transaction).
+   *     the transaction). However, a `RuntimeStatusError` exception is
+   *     instead consumed and converted into a `mutator` return value of the
+   *     enclosed `Status`.
    */
   StatusOr<CommitResult> Commit(
       std::function<StatusOr<Mutations>(Transaction)> const& mutator,

google/cloud/spanner/client_test.cc

@@ -584,6 +584,30 @@ TEST(ClientTest, CommitMutatorException) {
     FAIL();
   }
 }
+
+TEST(ClientTest, CommitMutatorRuntimeStatusException) {
+  auto conn = std::make_shared<MockConnection>();
+  EXPECT_CALL(*conn, Rollback(_)).WillRepeatedly(Return(Status()));
+  Client client(conn);
+  try {
+    auto result = client.Commit([](Transaction const&) -> StatusOr<Mutations> {
+      throw RuntimeStatusError(Status());  // OK
+    });
+    EXPECT_EQ(StatusCode::kUnknown, result.status().code());
+    EXPECT_THAT(result.status().message(), HasSubstr("OK Status thrown"));
+  } catch (...) {
+    FAIL();  // exception consumed
+  }
+  try {
+    auto result = client.Commit([](Transaction const&) -> StatusOr<Mutations> {
+      throw RuntimeStatusError(Status(StatusCode::kCancelled, "uh oh"));
+    });
+    EXPECT_EQ(StatusCode::kCancelled, result.status().code());
+    EXPECT_EQ(result.status().message(), "uh oh");
+  } catch (...) {
+    FAIL();  // exception consumed
+  }
+}
 #endif
 
 TEST(ClientTest, CommitMutatorRerunTransientFailures) {