fix(storage): do not retry permanent errors in async writer resume (#16340)

AsyncWriterConnectionResumed and AsyncWriterConnectionBuffered previously
unconditionally invoked Resume() when stream operations returned an error,
even if the error was a permanent failure such as FAILED_PRECONDITION.

For appendable uploads (BidiWriteObject), issuing a new BidiWriteObject
RPC on Resume() caused the client to request writer exclusivity again,
resulting in concurrent writers endlessly stealing exclusivity back and
forth from each other on FAILED_PRECONDITION errors.

This change checks if the error status is a permanent failure (via
AsyncRetryPolicy / AsyncStatusTraits) before attempting Resume(). If
permanent, the upload immediately terminates with the error status.

Author: v-pratap

google/cloud/storage/internal/async/writer_connection_buffered.cc

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 #include "google/cloud/storage/internal/async/writer_connection_buffered.h"
+#include "google/cloud/storage/async/retry_policy.h"
 #include "google/cloud/storage/internal/async/write_payload_impl.h"
 #include "google/cloud/future.h"
 #include "google/cloud/internal/make_status.h"
@@ -60,14 +61,25 @@ Status MakeFastForwardError(absl::string_view upload_id,
           .WithMetadata("gcloud-cpp.storage.persisted_size", returned));
 }
 
+bool IsPermanentFailure(Options const& options, Status const& status) {
+  if (options.has<storage::AsyncRetryPolicyOption>() &&
+      options.get<storage::AsyncRetryPolicyOption>() != nullptr) {
+    return options.get<storage::AsyncRetryPolicyOption>()->IsPermanentFailure(
+        status);
+  }
+  return storage::internal::AsyncStatusTraits::IsPermanentFailure(status);
+}
+
 class AsyncWriterConnectionBufferedState
     : public std::enable_shared_from_this<AsyncWriterConnectionBufferedState> {
  public:
   AsyncWriterConnectionBufferedState(
       WriterConnectionFactory factory,
       std::unique_ptr<storage::AsyncWriterConnection> impl,
-      std::size_t buffer_size_lwm, std::size_t buffer_size_hwm)
+      Options const& options, std::size_t buffer_size_lwm,
+      std::size_t buffer_size_hwm)
       : factory_(std::move(factory)),
+        options_(internal::MakeImmutableOptions(options)),
         buffer_size_lwm_(buffer_size_lwm),
         buffer_size_hwm_(buffer_size_hwm),
         impl_(std::move(impl)) {
@@ -422,6 +434,9 @@ class AsyncWriterConnectionBufferedState
       if (!s.ok() && cancelled_) {
         return SetError(std::move(lk), std::move(s));
       }
+      if (!s.ok() && IsPermanentFailure(*options_, s)) {
+        return SetError(std::move(lk), std::move(s));
+      }
       // Guard against concurrent resume attempts.
       if (resuming_) return;
       resuming_ = true;
@@ -630,6 +645,8 @@ class AsyncWriterConnectionBufferedState
   // Creates new `impl_` instances when needed.
   WriterConnectionFactory const factory_;
 
+  google::cloud::internal::ImmutableOptions options_;
+
   // Request a server-side flush if the buffer goes over this threshold.
   std::size_t const buffer_size_lwm_;
 
@@ -780,9 +797,10 @@ class AsyncWriterConnectionBuffered : public storage::AsyncWriterConnection {
   explicit AsyncWriterConnectionBuffered(
       WriterConnectionFactory factory,
       std::unique_ptr<storage::AsyncWriterConnection> impl,
-      std::size_t buffer_size_lwm, std::size_t buffer_size_hwm)
+      Options const& options, std::size_t buffer_size_lwm,
+      std::size_t buffer_size_hwm)
       : state_(std::make_shared<AsyncWriterConnectionBufferedState>(
-            std::move(factory), std::move(impl), buffer_size_lwm,
+            std::move(factory), std::move(impl), options, buffer_size_lwm,
             buffer_size_hwm)) {}
 
   void Cancel() override { return state_->Cancel(); }
@@ -833,7 +851,7 @@ std::unique_ptr<storage::AsyncWriterConnection> MakeWriterConnectionBuffered(
     std::unique_ptr<storage::AsyncWriterConnection> impl,
     Options const& options) {
   return absl::make_unique<AsyncWriterConnectionBuffered>(
-      std::move(factory), std::move(impl),
+      std::move(factory), std::move(impl), options,
       options.get<storage::BufferedUploadLwmOption>(),
       options.get<storage::BufferedUploadHwmOption>());
 }

google/cloud/storage/internal/async/writer_connection_buffered_test.cc

@@ -14,6 +14,7 @@
 
 #include "google/cloud/storage/internal/async/writer_connection_buffered.h"
 #include "google/cloud/storage/async/connection.h"
+#include "google/cloud/storage/async/retry_policy.h"
 #include "google/cloud/storage/mocks/mock_async_writer_connection.h"
 #include "google/cloud/storage/testing/canonical_errors.h"
 #include "google/cloud/testing_util/async_sequencer.h"
@@ -707,7 +708,7 @@ TEST(WriteConnectionBuffered, FlushWithEmptyPayload) {
 
 TEST(WriteConnectionBuffered, ErrorFailsPendingFlushes) {
   AsyncSequencer<bool> sequencer;
-  auto flush_error = PermanentError();
+  auto flush_error = TransientError();
   auto resume_error = Status(StatusCode::kAborted, "resume loop failed");
 
   auto mock = std::make_unique<MockAsyncWriterConnection>();
@@ -1681,6 +1682,85 @@ TEST(WriteConnectionBuffered, DuplicateFinalizeFails) {
   EXPECT_THAT(finalize2.get(), StatusIs(StatusCode::kFailedPrecondition));
 }
 
+TEST(WriteConnectionBuffered, PermanentErrorNoResume) {
+  AsyncSequencer<bool> sequencer;
+  auto failed_precondition =
+      Status(StatusCode::kFailedPrecondition, "precondition failed");
+
+  auto mock = std::make_unique<MockAsyncWriterConnection>();
+  EXPECT_CALL(*mock, UploadId).WillRepeatedly(Return("test-upload-id"));
+  EXPECT_CALL(*mock, PersistedState)
+      .WillRepeatedly(Return(MakePersistedState(0)));
+  EXPECT_CALL(*mock, Write).WillOnce([&](auto) {
+    return sequencer.PushBack("Write").then(
+        [failed_precondition](auto) { return failed_precondition; });
+  });
+
+  MockFactory mock_factory;
+  // factory_ should NOT be called because error is permanent
+  // (FAILED_PRECONDITION).
+  EXPECT_CALL(mock_factory, Call).Times(0);
+
+  auto connection = MakeWriterConnectionBuffered(
+      mock_factory.AsStdFunction(), std::move(mock), TestOptions());
+
+  auto write1 = connection->Write(TestPayload(1));
+  EXPECT_STATUS_OK(write1.get());
+
+  auto next = sequencer.PopFrontWithName();
+  EXPECT_THAT(next.second, Eq("Write"));
+  next.first.set_value(true);
+
+  auto write2 = connection->Write(TestPayload(1));
+  EXPECT_THAT(write2.get(), StatusIs(StatusCode::kFailedPrecondition));
+}
+
+TEST(WriteConnectionBuffered, CustomRetryPolicyOption) {
+  struct CustomAsyncRetryPolicy : public storage::AsyncRetryPolicy {
+    std::unique_ptr<storage::AsyncRetryPolicy> clone() const override {
+      return std::make_unique<CustomAsyncRetryPolicy>();
+    }
+    bool OnFailure(Status const&) override { return false; }
+    bool IsExhausted() const override { return false; }
+    bool IsPermanentFailure(Status const& s) const override {
+      return s.code() == StatusCode::kInvalidArgument;
+    }
+  };
+
+  AsyncSequencer<bool> sequencer;
+  auto invalid_argument =
+      Status(StatusCode::kInvalidArgument, "custom permanent error");
+
+  auto mock = std::make_unique<MockAsyncWriterConnection>();
+  EXPECT_CALL(*mock, UploadId).WillRepeatedly(Return("test-upload-id"));
+  EXPECT_CALL(*mock, PersistedState)
+      .WillRepeatedly(Return(MakePersistedState(0)));
+  EXPECT_CALL(*mock, Write).WillOnce([&](auto) {
+    return sequencer.PushBack("Write").then(
+        [invalid_argument](auto) { return invalid_argument; });
+  });
+
+  MockFactory mock_factory;
+  // factory_ should NOT be called because error is permanent per custom policy.
+  EXPECT_CALL(mock_factory, Call).Times(0);
+
+  auto options = TestOptions().set<storage::AsyncRetryPolicyOption>(
+      std::make_shared<CustomAsyncRetryPolicy>());
+
+  auto connection = MakeWriterConnectionBuffered(mock_factory.AsStdFunction(),
+                                                 std::move(mock), options);
+
+  auto write1 = connection->Write(TestPayload(1));
+  EXPECT_STATUS_OK(write1.get());
+
+  auto next = sequencer.PopFrontWithName();
+  EXPECT_THAT(next.second, Eq("Write"));
+  next.first.set_value(true);
+
+  auto write2 = connection->Write(TestPayload(1));
+  EXPECT_THAT(write2.get(), StatusIs(StatusCode::kInvalidArgument));
+}
+
 }  // namespace
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_internal

google/cloud/storage/internal/async/writer_connection_resumed.cc

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 #include "google/cloud/storage/internal/async/writer_connection_resumed.h"
+#include "google/cloud/storage/async/retry_policy.h"
 #include "google/cloud/storage/internal/async/write_payload_impl.h"
 #include "google/cloud/storage/internal/async/writer_connection_impl.h"
 #include "google/cloud/future.h"
@@ -57,6 +58,15 @@ Status MakeFastForwardError(std::int64_t resend_offset,
           .WithMetadata("gcloud-cpp.storage.persisted_size", returned));
 }
 
+bool IsPermanentFailure(Options const& options, Status const& status) {
+  if (options.has<storage::AsyncRetryPolicyOption>() &&
+      options.get<storage::AsyncRetryPolicyOption>() != nullptr) {
+    return options.get<storage::AsyncRetryPolicyOption>()->IsPermanentFailure(
+        status);
+  }
+  return storage::internal::AsyncStatusTraits::IsPermanentFailure(status);
+}
+
 class AsyncWriterConnectionResumedState
     : public std::enable_shared_from_this<AsyncWriterConnectionResumedState> {
  public:
@@ -72,12 +82,12 @@ class AsyncWriterConnectionResumedState
         impl_(std::move(impl)),
         initial_request_(std::move(initial_request)),
         hash_function_(std::move(hash_function)),
+        options_(internal::MakeImmutableOptions(options)),
         first_response_(std::move(first_response)),
         buffer_size_lwm_(buffer_size_lwm),
         buffer_size_hwm_(buffer_size_hwm) {
     finalized_future_ = finalized_.get_future();
     closed_future_ = closed_.get_future();
-    options_ = internal::MakeImmutableOptions(options);
     auto state = impl_->PersistedState();
     if (absl::holds_alternative<google::storage::v2::Object>(state)) {
       buffer_offset_ = absl::get<google::storage::v2::Object>(state).size();
@@ -419,6 +429,24 @@ class AsyncWriterConnectionResumedState
   }
 
   void Resume(Status const& s) {
+    // Capture the finalization and close state *before* starting the async
+    // resume.
+    bool was_finalizing;
+    bool was_closing;
+    {
+      std::unique_lock<std::mutex> lk(mu_);
+      if (state_ == State::kResuming) return;
+      was_finalizing = finalizing_;
+      was_closing = closing_;
+      if (!s.ok() && cancelled_) {
+        return SetError(std::move(lk), std::move(s));
+      }
+      if (!s.ok() && IsPermanentFailure(*options_, s)) {
+        return SetError(std::move(lk), std::move(s));
+      }
+      state_ = State::kResuming;
+    }
+
     auto proto_status = ExtractGrpcStatus(s);
     auto request = google::storage::v2::BidiWriteObjectRequest{};
     auto& append_object_spec = *request.mutable_append_object_spec();
@@ -439,20 +467,6 @@ class AsyncWriterConnectionResumedState
     append_object_spec.set_generation(first_response_.resource().generation());
     ApplyWriteRedirectErrors(append_object_spec, std::move(proto_status));
 
-    // Capture the finalization and close state *before* starting the async
-    // resume.
-    bool was_finalizing;
-    bool was_closing;
-    {
-      std::unique_lock<std::mutex> lk(mu_);
-      if (state_ == State::kResuming) return;
-      was_finalizing = finalizing_;
-      was_closing = closing_;
-      if (!s.ok() && cancelled_) {
-        return SetError(std::move(lk), std::move(s));
-      }
-      state_ = State::kResuming;
-    }
     // Pass the original status `s`, `was_finalizing`, and `was_closing` to the
     // callback.
     factory_(std::move(request))

google/cloud/storage/internal/async/writer_connection_resumed_test.cc

@@ -14,6 +14,7 @@
 #include "google/cloud/storage/internal/async/writer_connection_resumed.h"
 #include "google/cloud/mocks/mock_async_streaming_read_write_rpc.h"
 #include "google/cloud/storage/async/connection.h"
+#include "google/cloud/storage/async/retry_policy.h"
 #include "google/cloud/storage/internal/grpc/ctype_cord_workaround.h"
 #include "google/cloud/storage/mocks/mock_async_writer_connection.h"
 #include "google/cloud/storage/testing/canonical_errors.h"
@@ -1234,6 +1235,89 @@ TEST(WriterConnectionResumed, DuplicateFinalizeFails) {
   EXPECT_THAT(finalize2.get(), StatusIs(StatusCode::kFailedPrecondition));
 }
 
+TEST(WriteConnectionResumed, PermanentErrorNoResume) {
+  AsyncSequencer<bool> sequencer;
+  auto initial_request = google::storage::v2::BidiWriteObjectRequest{};
+  auto first_response = google::storage::v2::BidiWriteObjectResponse{};
+  auto failed_precondition = Status(StatusCode::kFailedPrecondition,
+                                    "another writer became exclusive");
+
+  auto mock = std::make_unique<MockAsyncWriterConnection>();
+  EXPECT_CALL(*mock, UploadId).WillRepeatedly(Return("test-upload-id"));
+  EXPECT_CALL(*mock, PersistedState)
+      .WillRepeatedly(Return(MakePersistedState(0)));
+  EXPECT_CALL(*mock, Flush).WillOnce([&](auto) {
+    return sequencer.PushBack("Flush").then(
+        [failed_precondition](auto) { return failed_precondition; });
+  });
+
+  MockFactory mock_factory;
+  // factory_ should NOT be called because error is permanent
+  // (FAILED_PRECONDITION).
+  EXPECT_CALL(mock_factory, Call).Times(0);
+
+  auto connection = MakeWriterConnectionResumed(
+      mock_factory.AsStdFunction(), std::move(mock), initial_request, nullptr,
+      first_response, Options{});
+
+  auto write = connection->Write(TestPayload(10));
+  ASSERT_FALSE(write.is_ready());
+
+  auto next = sequencer.PopFrontWithName();
+  EXPECT_THAT(next.second, Eq("Flush"));
+  next.first.set_value(true);
+
+  EXPECT_THAT(write.get(), StatusIs(StatusCode::kFailedPrecondition));
+}
+
+TEST(WriteConnectionResumed, CustomRetryPolicyOption) {
+  struct CustomAsyncRetryPolicy : public storage::AsyncRetryPolicy {
+    std::unique_ptr<storage::AsyncRetryPolicy> clone() const override {
+      return std::make_unique<CustomAsyncRetryPolicy>();
+    }
+    bool OnFailure(Status const&) override { return false; }
+    bool IsExhausted() const override { return false; }
+    bool IsPermanentFailure(Status const& s) const override {
+      return s.code() == StatusCode::kInvalidArgument;
+    }
+  };
+
+  AsyncSequencer<bool> sequencer;
+  auto initial_request = google::storage::v2::BidiWriteObjectRequest{};
+  auto first_response = google::storage::v2::BidiWriteObjectResponse{};
+  auto invalid_argument =
+      Status(StatusCode::kInvalidArgument, "custom permanent error");
+
+  auto mock = std::make_unique<MockAsyncWriterConnection>();
+  EXPECT_CALL(*mock, UploadId).WillRepeatedly(Return("test-upload-id"));
+  EXPECT_CALL(*mock, PersistedState)
+      .WillRepeatedly(Return(MakePersistedState(0)));
+  EXPECT_CALL(*mock, Flush).WillOnce([&](auto) {
+    return sequencer.PushBack("Flush").then(
+        [invalid_argument](auto) { return invalid_argument; });
+  });
+
+  MockFactory mock_factory;
+  // factory_ should NOT be called because error is permanent per custom policy.
+  EXPECT_CALL(mock_factory, Call).Times(0);
+
+  auto options = Options{}.set<storage::AsyncRetryPolicyOption>(
+      std::make_shared<CustomAsyncRetryPolicy>());
+
+  auto connection = MakeWriterConnectionResumed(
+      mock_factory.AsStdFunction(), std::move(mock), initial_request, nullptr,
+      first_response, options);
+
+  auto write = connection->Write(TestPayload(10));
+  ASSERT_FALSE(write.is_ready());
+
+  auto next = sequencer.PopFrontWithName();
+  EXPECT_THAT(next.second, Eq("Flush"));
+  next.first.set_value(true);
+
+  EXPECT_THAT(write.get(), StatusIs(StatusCode::kInvalidArgument));
+}
+
 }  // namespace
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_internal