feat(storage): add checksumming of bidiread messages (#78)

* feat(storage): add checksumming of bidiread messages

* remove move

Author: ddelgrosso1

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

@@ -13,11 +13,15 @@
 // limitations under the License.
 
 #include "google/cloud/storage/internal/async/object_descriptor_impl.h"
+#include "google/cloud/storage/async/options.h"
 #include "google/cloud/storage/internal/async/handle_redirect_error.h"
 #include "google/cloud/storage/internal/async/object_descriptor_reader_tracing.h"
+#include "google/cloud/storage/internal/hash_function.h"
+#include "google/cloud/storage/internal/hash_function_impl.h"
 #include "google/cloud/grpc_error_delegate.h"
 #include "google/cloud/internal/opentelemetry.h"
 #include <google/rpc/status.pb.h>
+#include <memory>
 #include <utility>
 
 namespace google {
@@ -53,7 +57,15 @@ absl::optional<google::storage::v2::Object> ObjectDescriptorImpl::metadata()
 
 std::unique_ptr<storage_experimental::AsyncReaderConnection>
 ObjectDescriptorImpl::Read(ReadParams p) {
-  auto range = std::make_shared<ReadRange>(p.start, p.limit);
+  std::shared_ptr<storage::internal::HashFunction> hash_function =
+      std::shared_ptr<storage::internal::HashFunction>(
+          storage::internal::CreateNullHashFunction());
+  if (options_.has<storage_experimental::EnableCrc32cValidationOption>()) {
+    hash_function =
+        std::make_shared<storage::internal::Crc32cMessageHashFunction>(
+            storage::internal::CreateNullHashFunction());
+  }
+  auto range = std::make_shared<ReadRange>(p.start, p.limit, hash_function);
 
   std::unique_lock<std::mutex> lk(mu_);
   auto const id = ++read_id_generator_;

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

@@ -69,8 +69,16 @@ void ReadRange::OnRead(google::storage::v2::ObjectRangeData data) {
   std::unique_lock<std::mutex> lk(mu_);
   if (status_) return;
   if (data.range_end()) status_ = Status{};
-  // TODO(#28) - verify the checksum
+
+  auto check_summed_data = data.mutable_checksummed_data();
   auto content = StealMutableContent(*data.mutable_checksummed_data());
+  auto status =
+      hash_function_->Update(offset_, content, check_summed_data->crc32c());
+  if (!status.ok()) {
+    status_ = std::move(status);
+    return Notify(std::move(lk), ReadPayloadImpl::Make(std::move(content)));
+  }
+
   offset_ += content.size();
   if (limit_ != 0) limit_ -= std::min<std::int64_t>(content.size(), limit_);
   auto p = ReadPayloadImpl::Make(std::move(content));

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

@@ -16,12 +16,14 @@
 #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_ASYNC_READ_RANGE_H
 
 #include "google/cloud/storage/async/reader_connection.h"
+#include "google/cloud/storage/internal/hash_function.h"
 #include "google/cloud/future.h"
 #include "google/cloud/status.h"
 #include "google/cloud/version.h"
 #include "absl/types/optional.h"
 #include <google/storage/v2/storage.pb.h>
 #include <cstdint>
+#include <memory>
 #include <mutex>
 
 namespace google {
@@ -43,8 +45,12 @@ class ReadRange {
   using ReadResponse =
       storage_experimental::AsyncReaderConnection::ReadResponse;
 
-  ReadRange(std::int64_t offset, std::int64_t limit)
-      : offset_(offset), limit_(limit) {}
+  ReadRange(std::int64_t offset, std::int64_t limit,
+            std::shared_ptr<storage::internal::HashFunction> hash_function =
+                storage::internal::CreateNullHashFunction())
+      : offset_(offset),
+        limit_(limit),
+        hash_function_(std::move(hash_function)) {}
 
   bool IsDone() const;
 
@@ -66,6 +72,7 @@ class ReadRange {
   absl::optional<storage_experimental::ReadPayload> payload_;
   absl::optional<Status> status_;
   absl::optional<promise<ReadResponse>> wait_;
+  std::shared_ptr<storage::internal::HashFunction> hash_function_;
 };
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

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

@@ -13,11 +13,18 @@
 // limitations under the License.
 
 #include "google/cloud/storage/internal/async/read_range.h"
+#include "google/cloud/storage/internal/hash_function.h"
+#include "google/cloud/storage/internal/hash_values.h"
 #include "google/cloud/storage/testing/canonical_errors.h"
 #include "google/cloud/testing_util/is_proto_equal.h"
 #include "google/cloud/testing_util/status_matchers.h"
+#include "absl/strings/cord.h"
+#include "absl/strings/string_view.h"
 #include <google/protobuf/text_format.h>
 #include <gmock/gmock.h>
+#include <gtest/gtest.h>
+#include <memory>
+#include <string>
 
 namespace google {
 namespace cloud {
@@ -31,11 +38,30 @@ using ::google::cloud::testing_util::IsOk;
 using ::google::cloud::testing_util::IsProtoEqual;
 using ::google::cloud::testing_util::StatusIs;
 using ::google::protobuf::TextFormat;
+using ::testing::_;
+using ::testing::AtLeast;
 using ::testing::ElementsAre;
 using ::testing::Optional;
 using ::testing::ResultOf;
 using ::testing::VariantWith;
 
+class MockHashFunction : public storage::internal::HashFunction {
+ public:
+  MOCK_METHOD(void, Update, (absl::string_view buffer), (override));
+  MOCK_METHOD(Status, Update, (std::int64_t offset, absl::string_view buffer),
+              (override));
+  MOCK_METHOD(Status, Update,
+              (std::int64_t offset, absl::string_view buffer,
+               std::uint32_t buffer_crc),
+              (override));
+  MOCK_METHOD(Status, Update,
+              (std::int64_t offset, absl::Cord const& buffer,
+               std::uint32_t buffer_crc),
+              (override));
+  MOCK_METHOD(std::string, Name, (), (const, override));
+  MOCK_METHOD(storage::internal::HashValues, Finish, (), (override));
+};
+
 TEST(ReadRange, BasicLifecycle) {
   ReadRange actual(10000, 40);
   EXPECT_FALSE(actual.IsDone());
@@ -156,6 +182,23 @@ TEST(ReadRange, Queue) {
   EXPECT_THAT(actual.Read().get(), VariantWith<ReadPayload>(matcher));
 }
 
+TEST(ReadRange, HashFunctionCalled) {
+  auto hash_function = std::make_shared<MockHashFunction>();
+  absl::Cord contents("1234567890");
+  EXPECT_CALL(*hash_function, Update(0, contents, _)).Times(AtLeast(1));
+
+  ReadRange actual(0, 0, hash_function);
+  auto data = google::storage::v2::ObjectRangeData{};
+  auto constexpr kData0 = R"pb(
+    checksummed_data { content: "1234567890" }
+    read_range { read_offset: 0 read_limit: 10 read_id: 7 }
+    range_end: false
+  )pb";
+
+  EXPECT_TRUE(TextFormat::ParseFromString(kData0, &data));
+  actual.OnRead(std::move(data));
+}
+
 }  // namespace
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_internal