fix(storage): allow providing expected CRC32C to Finalize()

Author: v-pratap

google/cloud/storage/examples/storage_async_samples.cc

@@ -728,6 +728,40 @@ void CreateAndWriteAppendableObject(google::cloud::storage::AsyncClient& client,
   std::cout << "File successfully uploaded " << object.DebugString() << "\n";
 }
 
+void CreateAndWriteAppendableObjectWithChecksum(
+    google::cloud::storage::AsyncClient& client,
+    std::vector<std::string> const& argv) {
+  //! [create-and-write-appendable-object-with-checksum]
+  // [START storage_create_and_write_appendable_object_with_checksum]
+  namespace gcs = google::cloud::storage;
+  auto coro = [](gcs::AsyncClient& client, std::string bucket_name,
+                 std::string object_name)
+      -> google::cloud::future<google::storage::v2::Object> {
+    auto [writer, token] =
+        (co_await client.StartAppendableObjectUpload(
+             gcs::BucketName(std::move(bucket_name)), std::move(object_name)))
+            .value();
+    std::cout << "Appendable upload started for object " << object_name << "\n";
+
+    token = (co_await writer.Write(std::move(token),
+                                   gcs::WritePayload("Some data\n")))
+                .value();
+    std::cout << "Wrote some data.\n";
+
+    // Set the expected CRC32C checksum in the current options scope
+    // just before calling Finalize().
+    google::cloud::internal::OptionsSpan span(
+        google::cloud::Options{}.set<gcs::UseCrc32cValueOption>(1848151177U));
+    co_return (co_await writer.Finalize(std::move(token))).value();
+  };
+  // [END storage_create_and_write_appendable_object_with_checksum]
+  //! [create-and-write-appendable-object-with-checksum]
+  // The example is easier to test and run if we call the coroutine and block
+  // until it completes..
+  auto const object = coro(client, argv.at(0), argv.at(1)).get();
+  std::cout << "File successfully uploaded " << object.DebugString() << "\n";
+}
+
 void PauseAndResumeAppendableUpload(google::cloud::storage::AsyncClient& client,
                                     std::vector<std::string> const& argv) {
   //! [pause-and-resume-appendable-upload]
@@ -1035,6 +1069,12 @@ void CreateAndWriteAppendableObject(google::cloud::storage::AsyncClient&,
                "coroutines\n";
 }
 
+void CreateAndWriteAppendableObjectWithChecksum(
+    google::cloud::storage::AsyncClient&, std::vector<std::string> const&) {
+  std::cerr << "AsyncClient::CreateAndWriteAppendableObjectWithChecksum() "
+               "example requires coroutines\n";
+}
+
 void PauseAndResumeAppendableUpload(google::cloud::storage::AsyncClient&,
                                     std::vector<std::string> const&) {
   std::cerr << "AsyncClient::PauseAndResumeAppendableUpload() example requires "
@@ -1505,6 +1545,8 @@ int main(int argc, char* argv[]) try {
 
       make_entry("create-and-write-appendable-object", {},
                  CreateAndWriteAppendableObject),
+      make_entry("create-and-write-appendable-object-with-checksum", {},
+                 CreateAndWriteAppendableObjectWithChecksum),
       make_entry("pause-and-resume-appendable-upload", {},
                  PauseAndResumeAppendableUpload),
       make_entry("finalize-appendable-object-upload", {},

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

@@ -14,9 +14,11 @@
 
 #include "google/cloud/storage/internal/async/writer_connection_impl.h"
 #include "google/cloud/mocks/mock_async_streaming_read_write_rpc.h"
+#include "google/cloud/storage/async/options.h"
 #include "google/cloud/storage/internal/async/write_object.h"
 #include "google/cloud/storage/internal/crc32c.h"
 #include "google/cloud/storage/internal/grpc/ctype_cord_workaround.h"
+#include "google/cloud/storage/internal/grpc/object_metadata_parser.h"
 #include "google/cloud/storage/internal/hash_function_impl.h"
 #include "google/cloud/storage/options.h"
 #include "google/cloud/storage/testing/canonical_errors.h"
@@ -744,6 +746,123 @@ TEST(AsyncWriterConnectionTest, FinalizeAppendableNoChecksum) {
   next.first.set_value(true);
 }
 
+TEST(AsyncWriterConnectionTest, FinalizeAppendableWithExpectedChecksum) {
+  AsyncSequencer<bool> sequencer;
+  auto mock = std::make_unique<MockStream>();
+  EXPECT_CALL(*mock, Cancel).Times(1);
+  EXPECT_CALL(*mock, Write)
+      .WillOnce([&](Request const& request, grpc::WriteOptions wopt) {
+        EXPECT_TRUE(request.finish_write());
+        EXPECT_TRUE(wopt.is_last_message());
+        EXPECT_EQ(request.common_object_request_params().encryption_algorithm(),
+                  "test-only-algo");
+        EXPECT_TRUE(request.has_object_checksums());
+        EXPECT_EQ(request.object_checksums().crc32c(),
+                  123456);  // wait, it might be an int in proto
+        return sequencer.PushBack("Write");
+      });
+  EXPECT_CALL(*mock, Read).WillOnce([&]() {
+    return sequencer.PushBack("Read").then([](auto f) {
+      if (!f.get()) return absl::optional<Response>();
+      return absl::make_optional(MakeTestResponse());
+    });
+  });
+  EXPECT_CALL(*mock, Finish).WillOnce([&] {
+    return sequencer.PushBack("Finish").then([](auto f) {
+      if (f.get()) return Status{};
+      return PermanentError();
+    });
+  });
+  auto hash = std::make_shared<MockHashFunction>();
+  EXPECT_CALL(*hash, Update(_, An<absl::Cord const&>(), _)).Times(1);
+  EXPECT_CALL(*hash, Finish)
+      .WillOnce(Return(storage::internal::HashValues{
+          storage_internal::Crc32cFromProto(123456), {}}));
+
+  auto request = MakeRequest();
+  request.mutable_write_object_spec()->set_appendable(true);
+
+  auto options = internal::MakeImmutableOptions(
+      Options{}.set<storage::UseCrc32cValueOption>(123456));
+
+  auto tested = std::make_unique<AsyncWriterConnectionImpl>(
+      options, std::move(request), std::move(mock), hash, 1024);
+  auto response = tested->Finalize(WritePayload{});
+
+  auto next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Write");
+  next.first.set_value(true);
+  next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Read");
+  next.first.set_value(true);
+  auto object = response.get();
+  EXPECT_THAT(object, IsOkAndHolds(IsProtoEqual(MakeTestObject())))
+      << "=" << object->DebugString();
+
+  tested = {};
+  next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Finish");
+  next.first.set_value(true);
+}
+
+TEST(AsyncWriterConnectionTest,
+     FinalizeAppendableWithExpectedChecksumFromCurrentOptions) {
+  AsyncSequencer<bool> sequencer;
+  auto mock = std::make_unique<MockStream>();
+  EXPECT_CALL(*mock, Cancel).Times(1);
+  EXPECT_CALL(*mock, Write)
+      .WillOnce([&](Request const& request, grpc::WriteOptions wopt) {
+        EXPECT_TRUE(request.finish_write());
+        EXPECT_TRUE(wopt.is_last_message());
+        EXPECT_EQ(request.common_object_request_params().encryption_algorithm(),
+                  "test-only-algo");
+        EXPECT_TRUE(request.has_object_checksums());
+        EXPECT_EQ(request.object_checksums().crc32c(), 654321);
+        return sequencer.PushBack("Write");
+      });
+  EXPECT_CALL(*mock, Read).WillOnce([&]() {
+    return sequencer.PushBack("Read").then([](auto f) {
+      if (!f.get()) return absl::optional<Response>();
+      return absl::make_optional(MakeTestResponse());
+    });
+  });
+  EXPECT_CALL(*mock, Finish).WillOnce([&] {
+    return sequencer.PushBack("Finish").then([](auto f) {
+      if (f.get()) return Status{};
+      return PermanentError();
+    });
+  });
+  auto hash = std::make_shared<MockHashFunction>();
+  EXPECT_CALL(*hash, Update(_, An<absl::Cord const&>(), _)).Times(1);
+  // It shouldn't call Finish() because we use kFinalize!
+  EXPECT_CALL(*hash, Finish).Times(0);
+
+  auto request = MakeRequest();
+  request.mutable_write_object_spec()->set_appendable(true);
+
+  auto tested = std::make_unique<AsyncWriterConnectionImpl>(
+      TestOptions(), std::move(request), std::move(mock), hash, 1024);
+
+  internal::OptionsSpan span(
+      Options{}.set<storage::UseCrc32cValueOption>(654321));
+  auto response = tested->Finalize(WritePayload{});
+
+  auto next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Write");
+  next.first.set_value(true);
+  next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Read");
+  next.first.set_value(true);
+  auto object = response.get();
+  EXPECT_THAT(object, IsOkAndHolds(IsProtoEqual(MakeTestObject())))
+      << "=" << object->DebugString();
+
+  tested = {};
+  next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Finish");
+  next.first.set_value(true);
+}
+
 TEST(AsyncWriterConnectionTest, ResumeWithHandle) {
   AsyncSequencer<bool> sequencer;
   auto mock = std::make_unique<MockStream>();