diff --git a/google/cloud/storage/internal/async/writer_connection_buffered.cc b/google/cloud/storage/internal/async/writer_connection_buffered.cc index c65ffed53b350..3d5b9bc8e80f3 100644 --- a/google/cloud/storage/internal/async/writer_connection_buffered.cc +++ b/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() && + options.get() != nullptr) { + return options.get()->IsPermanentFailure( + status); + } + return storage::internal::AsyncStatusTraits::IsPermanentFailure(status); +} + class AsyncWriterConnectionBufferedState : public std::enable_shared_from_this { public: AsyncWriterConnectionBufferedState( WriterConnectionFactory factory, std::unique_ptr 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 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( - 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 MakeWriterConnectionBuffered( std::unique_ptr impl, Options const& options) { return absl::make_unique( - std::move(factory), std::move(impl), + std::move(factory), std::move(impl), options, options.get(), options.get()); } diff --git a/google/cloud/storage/internal/async/writer_connection_buffered_test.cc b/google/cloud/storage/internal/async/writer_connection_buffered_test.cc index b86b8595fb39b..f369f1d88a807 100644 --- a/google/cloud/storage/internal/async/writer_connection_buffered_test.cc +++ b/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 sequencer; - auto flush_error = PermanentError(); + auto flush_error = TransientError(); auto resume_error = Status(StatusCode::kAborted, "resume loop failed"); auto mock = std::make_unique(); @@ -1681,6 +1682,85 @@ TEST(WriteConnectionBuffered, DuplicateFinalizeFails) { EXPECT_THAT(finalize2.get(), StatusIs(StatusCode::kFailedPrecondition)); } +TEST(WriteConnectionBuffered, PermanentErrorNoResume) { + AsyncSequencer sequencer; + auto failed_precondition = + Status(StatusCode::kFailedPrecondition, "precondition failed"); + + auto mock = std::make_unique(); + 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 clone() const override { + return std::make_unique(); + } + 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 sequencer; + auto invalid_argument = + Status(StatusCode::kInvalidArgument, "custom permanent error"); + + auto mock = std::make_unique(); + 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( + std::make_shared()); + + 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 diff --git a/google/cloud/storage/internal/async/writer_connection_resumed.cc b/google/cloud/storage/internal/async/writer_connection_resumed.cc index 23ffea9395bbc..17f6179cfa55c 100644 --- a/google/cloud/storage/internal/async/writer_connection_resumed.cc +++ b/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() && + options.get() != nullptr) { + return options.get()->IsPermanentFailure( + status); + } + return storage::internal::AsyncStatusTraits::IsPermanentFailure(status); +} + class AsyncWriterConnectionResumedState : public std::enable_shared_from_this { 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(state)) { buffer_offset_ = absl::get(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 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 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)) diff --git a/google/cloud/storage/internal/async/writer_connection_resumed_test.cc b/google/cloud/storage/internal/async/writer_connection_resumed_test.cc index 3968ba8a8fdb0..13d59167ff41b 100644 --- a/google/cloud/storage/internal/async/writer_connection_resumed_test.cc +++ b/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 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(); + 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 clone() const override { + return std::make_unique(); + } + 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 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(); + 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( + std::make_shared()); + + 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