Impl(ACv2): add appendable takeover (#102)

Author: bajajneha27

google/cloud/storage/async/client.cc

@@ -144,6 +144,33 @@ AsyncClient::StartAppendableObjectUpload(
       });
 }
 
+future<StatusOr<std::pair<AsyncWriter, AsyncToken>>>
+AsyncClient::ResumeAppendableObjectUpload(BucketName const& bucket_name,
+                                          std::string object_name,
+                                          std::int64_t generation,
+                                          Options opts) {
+  auto request = google::storage::v2::BidiWriteObjectRequest{};
+  auto& append_object_spec = *request.mutable_append_object_spec();
+
+  append_object_spec.set_bucket(BucketName(bucket_name).FullName());
+  append_object_spec.set_object(std::move(object_name));
+  append_object_spec.set_generation(std::move(generation));
+
+  return connection_
+      ->ResumeAppendableObjectUpload(
+          {std::move(request),
+           internal::MergeOptions(std::move(opts), connection_->options())})
+      .then([](auto f) -> StatusOr<std::pair<AsyncWriter, AsyncToken>> {
+        auto w = f.get();
+        if (!w) return std::move(w).status();
+        auto t = absl::holds_alternative<google::storage::v2::Object>(
+                     (*w)->PersistedState())
+                     ? AsyncToken()
+                     : storage_internal::MakeAsyncToken(w->get());
+        return std::make_pair(AsyncWriter(*std::move(w)), std::move(t));
+      });
+}
+
 future<StatusOr<std::pair<AsyncWriter, AsyncToken>>>
 AsyncClient::StartBufferedUpload(BucketName const& bucket_name,
                                  std::string object_name, Options opts) {

google/cloud/storage/async/client.h

@@ -426,6 +426,23 @@ class AsyncClient {
   StartAppendableObjectUpload(
       google::storage::v2::BidiWriteObjectRequest request, Options opts = {});
 
+  /**
+   * Resume a resumable upload session for appendable objects and automatic
+   * recovery from transient failures.
+   *
+   * @snippet{doc} async/client.h resume-appendable-object-upload
+   *
+   * @param bucket_name the name of the bucket that contains the object.
+   * @param object_name the name of the object to be uploaded.
+   * @param generation the object generation to be uploaded.
+   * @param opts options controlling the behaviour of this RPC, for example the
+   * application may change the retry policy.
+   */
+  future<StatusOr<std::pair<AsyncWriter, AsyncToken>>>
+  ResumeAppendableObjectUpload(BucketName const& bucket_name,
+                               std::string object_name, std::int64_t generation,
+                               Options opts = {});
+
   /*
   [start-buffered-upload-common]
   This function always uses [resumable uploads][resumable-link]. The objects

google/cloud/storage/async/connection.h

@@ -131,6 +131,11 @@ class AsyncConnection {
       StatusOr<std::unique_ptr<storage_experimental::AsyncWriterConnection>>>
   StartAppendableObjectUpload(AppendableUploadParams p) = 0;
 
+  /// Resume an appendable upload configured for persistent sources.
+  virtual future<
+      StatusOr<std::unique_ptr<storage_experimental::AsyncWriterConnection>>>
+  ResumeAppendableObjectUpload(AppendableUploadParams p) = 0;
+
   /**
    * A thin wrapper around the `WriteObject()` parameters.
    *

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

@@ -293,6 +293,17 @@ AsyncConnectionImpl::ReadObjectRange(ReadObjectParams p) {
 
 future<StatusOr<std::unique_ptr<storage_experimental::AsyncWriterConnection>>>
 AsyncConnectionImpl::StartAppendableObjectUpload(AppendableUploadParams p) {
+  return AppendableObjectUploadImpl(std::move(p));
+}
+
+future<StatusOr<std::unique_ptr<storage_experimental::AsyncWriterConnection>>>
+AsyncConnectionImpl::ResumeAppendableObjectUpload(AppendableUploadParams p) {
+  return AppendableObjectUploadImpl(std::move(p), true);
+}
+
+future<StatusOr<std::unique_ptr<storage_experimental::AsyncWriterConnection>>>
+AsyncConnectionImpl::AppendableObjectUploadImpl(AppendableUploadParams p,
+                                                bool takeover) {
   auto current = internal::MakeImmutableOptions(std::move(p.options));
   auto request = p.request;
   std::int64_t persisted_size = 0;
@@ -313,10 +324,10 @@ AsyncConnectionImpl::StartAppendableObjectUpload(AppendableUploadParams p) {
 
   auto factory = WriteResultFactory(
       [stub = stub_, cq = cq_, retry = std::move(retry),
-       backoff = std::move(backoff), current, function_name = __func__](
-          google::storage::v2::BidiWriteObjectRequest req) {
-        auto call = [stub, request = std::move(req)](
-                        CompletionQueue& cq,
+       backoff = std::move(backoff), current, function_name = __func__,
+       takeover](google::storage::v2::BidiWriteObjectRequest req) {
+        auto call = [stub, request = std::move(req), takeover](
+                        google::cloud::CompletionQueue& cq,
                         std::shared_ptr<grpc::ClientContext> context,
                         google::cloud::internal::ImmutableOptions options,
                         RequestPlaceholder const&) mutable
@@ -325,7 +336,13 @@ AsyncConnectionImpl::StartAppendableObjectUpload(AppendableUploadParams p) {
               options->get<storage::TransferStallTimeoutOption>(),
               options->get<storage::TransferStallMinimumRateOption>(),
               google::storage::v2::ServiceConstants::MAX_WRITE_CHUNK_BYTES);
-          ApplyRoutingHeaders(*context, request.write_object_spec());
+
+          // Apply the routing header
+          if (takeover)
+            ApplyRoutingHeaders(*context, request.append_object_spec());
+          else
+            ApplyRoutingHeaders(*context, request.write_object_spec());
+
           auto rpc = stub->AsyncBidiWriteObject(cq, std::move(context),
                                                 std::move(options));
           rpc = std::make_unique<StreamingRpcTimeout>(cq, timeout, timeout,

google/cloud/storage/internal/async/connection_impl.h

@@ -73,6 +73,9 @@ class AsyncConnectionImpl
   future<StatusOr<std::unique_ptr<storage_experimental::AsyncWriterConnection>>>
   StartAppendableObjectUpload(AppendableUploadParams p) override;
 
+  future<StatusOr<std::unique_ptr<storage_experimental::AsyncWriterConnection>>>
+  ResumeAppendableObjectUpload(AppendableUploadParams p) override;
+
   future<StatusOr<std::unique_ptr<storage_experimental::AsyncWriterConnection>>>
   StartUnbufferedUpload(UploadParams p) override;
 
@@ -137,6 +140,9 @@ class AsyncConnectionImpl
       std::shared_ptr<storage::internal::HashFunction> hash_function,
       std::int64_t persisted_size);
 
+  future<StatusOr<std::unique_ptr<storage_experimental::AsyncWriterConnection>>>
+  AppendableObjectUploadImpl(AppendableUploadParams p, bool takeover = false);
+
   CompletionQueue cq_;
   std::shared_ptr<GrpcChannelRefresh> refresh_;
   std::shared_ptr<StorageStub> stub_;

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

@@ -113,6 +113,23 @@ class AsyncConnectionTracing : public storage_experimental::AsyncConnection {
         });
   }
 
+  future<StatusOr<std::unique_ptr<storage_experimental::AsyncWriterConnection>>>
+  ResumeAppendableObjectUpload(AppendableUploadParams p) override {
+    auto span = internal::MakeSpan(
+        "storage::AsyncConnection::ResumeAppendableObjectUpload");
+    internal::OTelScope scope(span);
+    return impl_->ResumeAppendableObjectUpload(std::move(p))
+        .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
+               span = std::move(span)](auto f)
+                  -> StatusOr<std::unique_ptr<
+                      storage_experimental::AsyncWriterConnection>> {
+          auto w = f.get();
+          internal::DetachOTelContext(oc);
+          if (!w) return internal::EndSpan(*span, std::move(w).status());
+          return MakeTracingWriterConnection(span, *std::move(w));
+        });
+  }
+
   future<StatusOr<std::unique_ptr<storage_experimental::AsyncWriterConnection>>>
   StartUnbufferedUpload(UploadParams p) override {
     auto span =

google/cloud/storage/internal/grpc/configure_client_context.cc

@@ -50,6 +50,13 @@ void ApplyRoutingHeaders(grpc::ClientContext& context,
       "bucket=" + google::cloud::internal::UrlEncode(spec.resource().bucket()));
 }
 
+void ApplyRoutingHeaders(grpc::ClientContext& context,
+                         google::storage::v2::AppendObjectSpec const& spec) {
+  context.AddMetadata(
+      "x-goog-request-params",
+      "bucket=" + google::cloud::internal::UrlEncode(spec.bucket()));
+}
+
 void ApplyRoutingHeaders(grpc::ClientContext& context,
                          storage::internal::UploadChunkRequest const& request) {
   ApplyResumableUploadRoutingHeader(context, request.upload_session_url());

google/cloud/storage/internal/grpc/configure_client_context.h

@@ -79,6 +79,10 @@ void ApplyRoutingHeaders(
 void ApplyRoutingHeaders(grpc::ClientContext& context,
                          google::storage::v2::WriteObjectSpec const& spec);
 
+/// @copydoc ApplyRoutingHeaders(grpc::ClientContext&,)
+void ApplyRoutingHeaders(grpc::ClientContext& context,
+                         google::storage::v2::AppendObjectSpec const& spec);
+
 /**
  * The generated `StorageMetadata` stub can not handle dynamic routing headers
  * for client side streaming. So we manually match and extract the headers in

google/cloud/storage/mocks/mock_async_connection.h

@@ -48,6 +48,10 @@ class MockAsyncConnection : public storage_experimental::AsyncConnection {
       future<StatusOr<
           std::unique_ptr<storage_experimental::AsyncWriterConnection>>>,
       StartAppendableObjectUpload, (AppendableUploadParams), (override));
+  MOCK_METHOD(
+      future<StatusOr<
+          std::unique_ptr<storage_experimental::AsyncWriterConnection>>>,
+      ResumeAppendableObjectUpload, (AppendableUploadParams), (override));
   MOCK_METHOD(
       future<StatusOr<
           std::unique_ptr<storage_experimental::AsyncWriterConnection>>>,