fix(storage): preserve metadata in ReadAll() for 0-byte objects

google/cloud/storage/async/read_all_test.cc

@@ -30,6 +30,7 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 
 using ::google::cloud::storage::testing::canonical_errors::PermanentError;
 using ::google::cloud::storage_mocks::MockAsyncReaderConnection;
+using ::google::cloud::testing_util::IsOk;
 using ::google::cloud::testing_util::IsProtoEqual;
 using ::google::cloud::testing_util::StatusIs;
 using ::testing::ElementsAre;
@@ -149,6 +150,58 @@ TEST(ReadAll, Empty) {
   EXPECT_THAT(payload->contents(), IsEmpty());
 }
 
+namespace {
+google::storage::v2::Object MakeZeroByteTestObject() {
+  google::storage::v2::Object object;
+  object.set_name("test-only-name");
+  object.set_generation(123456789);
+  object.set_size(0);
+  return object;
+}
+}  // namespace
+
+// Reading a 0-byte object must preserve the object metadata.
+TEST(ReadAll, ZeroBytePayloadPreservesMetadata) {
+  auto mock = std::make_unique<MockAsyncReaderConnection>();
+  EXPECT_CALL(*mock, Read)
+      .WillOnce([] {
+        return make_ready_future(ReadResponse(
+            ReadPayload(std::string{}).set_metadata(MakeZeroByteTestObject())));
+      })
+      .WillOnce([] { return make_ready_future(ReadResponse(Status{})); });
+
+  AsyncToken token = storage_internal::MakeAsyncToken(mock.get());
+  StatusOr<ReadPayload> payload =
+      ReadAll(AsyncReader(std::move(mock)), std::move(token)).get();
+  ASSERT_THAT(payload, IsOk());
+  EXPECT_THAT(payload->contents(), IsEmpty());
+  EXPECT_THAT(payload->metadata(),
+              Optional(IsProtoEqual(MakeZeroByteTestObject())));
+}
+
+// An empty first payload with metadata, followed by data, must preserve both
+// the metadata and the data.
+TEST(ReadAll, EmptyPayloadWithMetadataThenData) {
+  auto mock = std::make_unique<MockAsyncReaderConnection>();
+  EXPECT_CALL(*mock, Read)
+      .WillOnce([] {
+        return make_ready_future(ReadResponse(
+            ReadPayload(std::string{}).set_metadata(MakeZeroByteTestObject())));
+      })
+      .WillOnce([] {
+        return make_ready_future(ReadResponse(ReadPayload("test-message-1")));
+      })
+      .WillOnce([] { return make_ready_future(ReadResponse(Status{})); });
+
+  AsyncToken token = storage_internal::MakeAsyncToken(mock.get());
+  StatusOr<ReadPayload> payload =
+      ReadAll(AsyncReader(std::move(mock)), std::move(token)).get();
+  ASSERT_THAT(payload, IsOk());
+  EXPECT_THAT(payload->contents(), ElementsAre("test-message-1"));
+  EXPECT_THAT(payload->metadata(),
+              Optional(IsProtoEqual(MakeZeroByteTestObject())));
+}
+
 TEST(ReadAll, Error) {
   auto mock = std::make_unique<MockAsyncReaderConnection>();
   EXPECT_CALL(*mock, Read)

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

@@ -60,6 +60,7 @@ using ::testing::AllOf;
 using ::testing::ElementsAre;
 using ::testing::HasSubstr;
 using ::testing::IsEmpty;
+using ::testing::Optional;
 using ::testing::ResultOf;
 using ::testing::Return;
 using ::testing::VariantWith;
@@ -601,8 +602,6 @@ TEST_F(AsyncConnectionImplTest, ReadObjectSilentWhenRetriesAreDisabled) {
   EXPECT_THAT(RetryRecords(log), IsEmpty());
 }
 
-// Only one test for ReadObjectRange(). The tests for `ReadAll()` and
-// `ReadObject()` cover most other cases.
 TEST_F(AsyncConnectionImplTest, ReadObjectRangePermanentError) {
   AsyncSequencer<bool> sequencer;
   auto mock = std::make_shared<storage::testing::MockStorageStub>();
@@ -626,6 +625,66 @@ TEST_F(AsyncConnectionImplTest, ReadObjectRangePermanentError) {
   EXPECT_THAT(pending.get(), StatusIs(PermanentError().code()));
 }
 
+// `ReadObjectRange()` on a 0-byte object must return the object metadata sent
+// by the service, even though there is no data.
+TEST_F(AsyncConnectionImplTest, ReadObjectRangeZeroByteObjectKeepsMetadata) {
+  auto constexpr kMetadata = R"pb(
+    bucket: "projects/_/buckets/test-bucket"
+    name: "test-object"
+    generation: 123456789
+    metageneration: 1
+    size: 0
+  )pb";
+  google::storage::v2::Object expected;
+  ASSERT_TRUE(TextFormat::ParseFromString(kMetadata, &expected));
+
+  AsyncSequencer<bool> sequencer;
+  auto mock = std::make_shared<storage::testing::MockStorageStub>();
+  EXPECT_CALL(*mock, AsyncReadObject).WillOnce([&] {
+    auto stream = std::make_unique<MockAsyncObjectMediaStream>();
+    EXPECT_CALL(*stream, Start).WillOnce([&] {
+      return sequencer.PushBack("Start");
+    });
+    EXPECT_CALL(*stream, Read)
+        .WillOnce([&] {
+          return sequencer.PushBack("Read").then([&](auto) {
+            // A 0-byte object: metadata, but no `checksummed_data`.
+            google::storage::v2::ReadObjectResponse response;
+            *response.mutable_metadata() = expected;
+            return std::make_optional(response);
+          });
+        })
+        .WillOnce([&] {
+          return sequencer.PushBack("Read").then([](auto) {
+            return std::optional<google::storage::v2::ReadObjectResponse>();
+          });
+        });
+    EXPECT_CALL(*stream, Finish).WillOnce([&] {
+      return sequencer.PushBack("Finish").then([](auto) { return Status{}; });
+    });
+    return std::unique_ptr<AsyncReadObjectStream>(std::move(stream));
+  });
+
+  internal::AutomaticallyCreatedBackgroundThreads pool(1);
+  auto connection =
+      MakeTestConnection(pool.cq(), mock,
+                         Options{}.set<storage::DownloadStallTimeoutOption>(
+                             std::chrono::seconds(0)));
+  future<StatusOr<storage::ReadPayload>> pending = connection->ReadObjectRange(
+      {google::storage::v2::ReadObjectRequest{}, connection->options()});
+
+  for (auto const* name : {"Start", "Read", "Read", "Finish"}) {
+    auto next = sequencer.PopFrontWithName();
+    EXPECT_EQ(next.second, name);
+    next.first.set_value(true);
+  }
+
+  StatusOr<storage::ReadPayload> payload = pending.get();
+  ASSERT_THAT(payload, IsOk());
+  EXPECT_THAT(payload->contents(), IsEmpty());
+  EXPECT_THAT(payload->metadata(), Optional(IsProtoEqual(expected)));
+}
+
 TEST_F(AsyncConnectionImplTest, ReadObjectDetectBadMessageChecksum) {
   AsyncSequencer<bool> sequencer;
   auto make_bad_checksum_stream = [&](AsyncSequencer<bool>& sequencer) {

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

@@ -52,7 +52,11 @@ struct ReadPayloadImpl {
   /// Append the data from @p rhs to @p lhs.
   static void Accumulate(storage::ReadPayload& lhs,
                          storage::ReadPayload&& rhs) {
-    if (lhs.impl_.empty()) {
+    // Only replace `lhs` if it is in its default-constructed state. A 0-byte
+    // object (or range) produces a payload with empty contents, but it may
+    // carry metadata, headers, an offset, or object hashes. Those must not be
+    // discarded by later (e.g. end-of-stream) payloads.
+    if (IsDefault(lhs)) {
       lhs = std::move(rhs);
       return;
     }
@@ -75,6 +79,15 @@ struct ReadPayloadImpl {
                      storage::ReadPayload new_data) {
     payload.impl_.Append(std::move(new_data.impl_));
   }
+
+ private:
+  /// Returns true if @p p is equivalent to a default-constructed payload.
+  static bool IsDefault(storage::ReadPayload const& p) {
+    // Note: read `object_hash_values_` directly, `GetObjectHashes()` moves
+    // the value out of the payload.
+    return p.impl_.empty() && p.offset_ == 0 && !p.metadata_.has_value() &&
+           p.headers_.empty() && !p.object_hash_values_.has_value();
+  }
 };
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

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

@@ -25,7 +25,9 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 namespace {
 
 using ::google::cloud::testing_util::IsProtoEqual;
+using ::testing::AllOf;
 using ::testing::ElementsAre;
+using ::testing::Field;
 using ::testing::IsEmpty;
 using ::testing::Optional;
 using ::testing::Pair;
@@ -90,6 +92,89 @@ TEST(ReadPayload, Reset) {
   EXPECT_THAT(actual.headers(), IsEmpty());
 }
 
+TEST(ReadPayload, AccumulateIntoDefault) {
+  google::storage::v2::Object const resource = MakeTestObject();
+  storage::ReadPayload actual;
+  ReadPayloadImpl::Accumulate(actual, ReadPayloadImpl::Make(absl::Cord(kQuick))
+                                          .set_metadata(resource)
+                                          .set_offset(1024));
+  EXPECT_THAT(actual.contents(), ElementsAre(absl::string_view(kQuick)));
+  EXPECT_THAT(actual.metadata(), Optional(IsProtoEqual(resource)));
+  EXPECT_EQ(actual.offset(), 1024);
+}
+
+TEST(ReadPayload, AccumulateAppendsData) {
+  google::storage::v2::Object const resource = MakeTestObject();
+  storage::ReadPayload actual = ReadPayloadImpl::Make(absl::Cord(kQuick))
+                                    .set_metadata(resource)
+                                    .set_offset(1024);
+  ReadPayloadImpl::Accumulate(
+      actual, ReadPayloadImpl::Make(absl::Cord(kQuick)).set_offset(2048));
+  EXPECT_THAT(actual.contents(), ElementsAre(absl::string_view(kQuick),
+                                             absl::string_view(kQuick)));
+  EXPECT_THAT(actual.metadata(), Optional(IsProtoEqual(resource)));
+  EXPECT_EQ(actual.offset(), 1024);
+}
+
+// A 0-byte payload with metadata must survive the EOF payload.
+TEST(ReadPayload, AccumulateEmptyWithMetadataKeepsMetadata) {
+  google::storage::v2::Object const resource = MakeTestObject();
+  storage::ReadPayload actual =
+      ReadPayloadImpl::Make(absl::Cord()).set_metadata(resource);
+  ReadPayloadImpl::Accumulate(actual, storage::ReadPayload{});
+  EXPECT_THAT(actual.contents(), IsEmpty());
+  EXPECT_EQ(actual.size(), 0);
+  EXPECT_THAT(actual.metadata(), Optional(IsProtoEqual(resource)));
+}
+
+// A 0-byte payload with metadata followed by data keeps the metadata, offset,
+// and object hashes from the first payload.
+TEST(ReadPayload, AccumulateEmptyWithMetadataThenData) {
+  google::storage::v2::Object const resource = MakeTestObject();
+  storage::ReadPayload actual = ReadPayloadImpl::Make(absl::Cord())
+                                    .set_metadata(resource)
+                                    .set_offset(1024);
+  ReadPayloadImpl::SetObjectHashes(
+      actual, storage::internal::HashValues{"test-crc32c", "test-md5"});
+  ReadPayloadImpl::Accumulate(
+      actual, ReadPayloadImpl::Make(absl::Cord(kQuick)).set_offset(1024));
+  EXPECT_THAT(actual.contents(), ElementsAre(absl::string_view(kQuick)));
+  EXPECT_THAT(actual.metadata(), Optional(IsProtoEqual(resource)));
+  EXPECT_EQ(actual.offset(), 1024);
+  EXPECT_THAT(ReadPayloadImpl::GetObjectHashes(actual),
+              Optional(AllOf(
+                  Field(&storage::internal::HashValues::crc32c, "test-crc32c"),
+                  Field(&storage::internal::HashValues::md5, "test-md5"))));
+}
+
+// An empty payload without metadata may still carry headers, an offset, or
+// object hashes. None of these should be discarded by later payloads.
+TEST(ReadPayload, AccumulateEmptyWithoutMetadataKeepsOtherFields) {
+  storage::ReadPayload actual = ReadPayloadImpl::Make(absl::Cord())
+                                    .set_headers({{"k1", "v1"}})
+                                    .set_offset(1024);
+  ReadPayloadImpl::SetObjectHashes(
+      actual, storage::internal::HashValues{"test-crc32c", "test-md5"});
+  ReadPayloadImpl::Accumulate(
+      actual, ReadPayloadImpl::Make(absl::Cord(kQuick)).set_offset(1024));
+  ReadPayloadImpl::Accumulate(actual, storage::ReadPayload{});
+  EXPECT_THAT(actual.contents(), ElementsAre(absl::string_view(kQuick)));
+  EXPECT_FALSE(actual.metadata().has_value());
+  EXPECT_THAT(actual.headers(), UnorderedElementsAre(Pair("k1", "v1")));
+  EXPECT_EQ(actual.offset(), 1024);
+  EXPECT_THAT(ReadPayloadImpl::GetObjectHashes(actual),
+              Optional(AllOf(
+                  Field(&storage::internal::HashValues::crc32c, "test-crc32c"),
+                  Field(&storage::internal::HashValues::md5, "test-md5"))));
+}
+
+TEST(ReadPayload, AccumulateEmptyIntoEmpty) {
+  storage::ReadPayload actual;
+  ReadPayloadImpl::Accumulate(actual, storage::ReadPayload{});
+  EXPECT_THAT(actual.contents(), IsEmpty());
+  EXPECT_FALSE(actual.metadata().has_value());
+}
+
 }  // namespace
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_internal