Tests for subsequent stream

Author: bajajneha27

google/cloud/storage/async/object_descriptor_test.cc

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 #include "google/cloud/storage/async/object_descriptor.h"
+#include "google/cloud/storage/async/options.h"
 #include "google/cloud/storage/mocks/mock_async_object_descriptor_connection.h"
 #include "google/cloud/storage/mocks/mock_async_reader_connection.h"
 #include "google/cloud/testing_util/status_matchers.h"
@@ -147,6 +148,48 @@ TEST(ObjectDescriptor, ReadLast) {
   EXPECT_FALSE(token.valid());
 }
 
+TEST(ObjectDescriptor, ReadExceedsMaxRange) {
+  auto mock = std::make_shared<MockAsyncObjectDescriptorConnection>();
+  auto constexpr kMaxRange = 1024;
+  EXPECT_CALL(*mock, options)
+      .WillRepeatedly(
+          Return(Options{}.set<storage_experimental::MaximumRangeSizeOption>(
+              kMaxRange)));
+
+  EXPECT_CALL(*mock, MakeSubsequentStream).Times(1);
+
+  EXPECT_CALL(*mock, Read)
+      .WillOnce([&](ReadParams p) -> std::unique_ptr<AsyncReaderConnection> {
+        EXPECT_EQ(p.start, 100);
+        EXPECT_EQ(p.length, kMaxRange + 1);
+        auto reader = std::make_unique<MockAsyncReaderConnection>();
+        EXPECT_CALL(*reader, Read)
+            .WillOnce([] {
+              return make_ready_future(
+                  ReadResponse(ReadPayload(std::string("some data"))));
+            })
+            .WillOnce([] { return make_ready_future(ReadResponse(Status{})); });
+        return reader;
+      });
+
+  auto tested = ObjectDescriptor(mock);
+  AsyncReader reader;
+  AsyncToken token;
+  std::tie(reader, token) = tested.Read(100, kMaxRange + 1);
+  ASSERT_TRUE(token.valid());
+
+  auto r1 = reader.Read(std::move(token)).get();
+  ASSERT_STATUS_OK(r1);
+  ReadPayload payload;
+  std::tie(payload, token) = *std::move(r1);
+  EXPECT_THAT(payload.contents(), ElementsAre("some data"));
+
+  auto r2 = reader.Read(std::move(token)).get();
+  ASSERT_STATUS_OK(r2);
+  std::tie(payload, token) = *std::move(r2);
+  EXPECT_FALSE(token.valid());
+}
+
 }  // namespace
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_experimental

google/cloud/storage/async/resume_policy.cc

@@ -35,7 +35,7 @@ class LimitedErrorCountResumePolicyImpl : public ResumePolicy {
   }
   Action OnFinish(Status const& s) override {
     if (!s.ok()) ++error_count_;
-    return error_count_ > maximum_resumes_ ? kStop : kContinue;
+    return error_count_ >= maximum_resumes_ ? kStop : kContinue;
   }
 
  private:

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

@@ -248,7 +248,6 @@ bool ObjectDescriptorImpl::IsResumable(
   for (auto const& any : proto_status.details()) {
     auto error = google::storage::v2::BidiReadObjectError{};
     if (!any.UnpackTo(&error)) continue;
-    auto ranges = CopyActiveRanges();
     for (auto const& range : CopyActiveRanges()) {
       for (auto const& range_error : error.read_range_errors()) {
         if (range.first != range_error.read_id()) continue;

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

@@ -1190,6 +1190,183 @@ TEST(ObjectDescriptorImpl, RecoverFromPartialFailure) {
   EXPECT_THAT(s3r1.get(), VariantWith<Status>(PermanentError()));
 }
 
+/// @test Verify that we can create a subsequent stream and read from it.
+TEST(ObjectDescriptorImpl, ReadWithSubsequentStream) {
+  // Setup
+  auto constexpr kResponse0 = R"pb(
+    metadata {
+      bucket: "projects/_/buckets/test-bucket"
+      name: "test-object"
+      generation: 42
+    }
+    read_handle { handle: "handle-12345" }
+  )pb";
+  auto constexpr kRequest1 = R"pb(
+    read_ranges { read_id: 1 read_offset: 100 read_length: 100 }
+  )pb";
+  auto constexpr kResponse1 = R"pb(
+    object_data_ranges {
+      range_end: true
+      read_range { read_id: 1 read_offset: 100 }
+      checksummed_data { content: "payload-for-stream-1" }
+    }
+  )pb";
+  auto constexpr kRequest2 = R"pb(
+    read_ranges { read_id: 2 read_offset: 200 read_length: 200 }
+  )pb";
+  auto constexpr kResponse2 = R"pb(
+    object_data_ranges {
+      range_end: true
+      read_range { read_id: 2 read_offset: 200 }
+      checksummed_data { content: "payload-for-stream-2" }
+    }
+  )pb";
+
+  AsyncSequencer<bool> sequencer;
+
+  // First stream setup
+  auto stream1 = std::make_unique<MockStream>();
+  EXPECT_CALL(*stream1, Write)
+      .WillOnce([&](Request const& request, grpc::WriteOptions) {
+        auto expected = Request{};
+        EXPECT_TRUE(TextFormat::ParseFromString(kRequest1, &expected));
+        EXPECT_THAT(request, IsProtoEqual(expected));
+        return sequencer.PushBack("Write[1]").then([](auto f) {
+          return f.get();
+        });
+      });
+  EXPECT_CALL(*stream1, Read)
+      .WillOnce([&]() {
+        return sequencer.PushBack("Read[1]").then([&](auto) {
+          auto response = Response{};
+          EXPECT_TRUE(TextFormat::ParseFromString(kResponse1, &response));
+          return absl::make_optional(response);
+        });
+      })
+      .WillOnce([&]() {
+        return sequencer.PushBack("Read[1.eos]").then([&](auto) {
+          return absl::optional<Response>{};
+        });
+      });
+  EXPECT_CALL(*stream1, Finish).WillOnce([&]() {
+    return sequencer.PushBack("Finish[1]").then([](auto) { return Status{}; });
+  });
+  EXPECT_CALL(*stream1, Cancel).Times(1);
+
+  // Second stream setup
+  auto stream2 = std::make_unique<MockStream>();
+  EXPECT_CALL(*stream2, Write)
+      .WillOnce([&](Request const& request, grpc::WriteOptions) {
+        auto expected = Request{};
+        EXPECT_TRUE(TextFormat::ParseFromString(kRequest2, &expected));
+        EXPECT_THAT(request, IsProtoEqual(expected));
+        return sequencer.PushBack("Write[2]").then([](auto f) {
+          return f.get();
+        });
+      });
+  EXPECT_CALL(*stream2, Read)
+      .WillOnce([&]() {
+        return sequencer.PushBack("Read[2]").then([&](auto) {
+          auto response = Response{};
+          EXPECT_TRUE(TextFormat::ParseFromString(kResponse2, &response));
+          return absl::make_optional(response);
+        });
+      })
+      .WillOnce([&]() {
+        return sequencer.PushBack("Read[2.eos]").then([](auto) {
+          return absl::optional<Response>{};
+        });
+      });
+  EXPECT_CALL(*stream2, Finish).WillOnce([&]() {
+    return sequencer.PushBack("Finish[2]").then([](auto) { return Status{}; });
+  });
+  EXPECT_CALL(*stream2, Cancel).Times(1);
+
+  // Mock factory for subsequent streams
+  MockFactory factory;
+  EXPECT_CALL(factory, Call).WillOnce([&](Request const&) {
+    auto stream_result = OpenStreamResult{
+        std::make_shared<OpenStream>(std::move(stream2)), Response{}};
+    return make_ready_future(make_status_or(std::move(stream_result)));
+  });
+
+  // Create the ObjectDescriptorImpl
+  auto tested = std::make_shared<ObjectDescriptorImpl>(
+      NoResume(), factory.AsStdFunction(),
+      google::storage::v2::BidiReadObjectSpec{},
+      std::make_shared<OpenStream>(std::move(stream1)));
+
+  auto response0 = Response{};
+  EXPECT_TRUE(TextFormat::ParseFromString(kResponse0, &response0));
+  tested->Start(std::move(response0));
+
+  auto read1 = sequencer.PopFrontWithName();
+  EXPECT_EQ(read1.second, "Read[1]");
+  // Start a read on the first stream
+  auto reader1 = tested->Read({100, 100});
+  auto future1 = reader1->Read();
+  // The implementation starts a read loop eagerly after Start(), and then
+  // the call to tested->Read() schedules a write.
+  auto write1 = sequencer.PopFrontWithName();
+  EXPECT_EQ(write1.second, "Write[1]");
+  write1.first.set_value(true);
+
+  // Now we can satisfy the read. This will deliver the data to the reader.
+  read1.first.set_value(true);
+
+  EXPECT_THAT(future1.get(),
+              VariantWith<storage_experimental::ReadPayload>(ResultOf(
+                  "contents are",
+                  [](storage_experimental::ReadPayload const& p) {
+                    return p.contents();
+                  },
+                  ElementsAre(absl::string_view{"payload-for-stream-1"}))));
+
+  EXPECT_THAT(reader1->Read().get(), VariantWith<Status>(IsOk()));
+
+  auto next = sequencer.PopFrontWithName();
+  EXPECT_EQ(next.second, "Read[1.eos]");
+  next.first.set_value(true);
+
+  // The first stream should be finishing now.
+  auto finish1 = sequencer.PopFrontWithName();
+  EXPECT_EQ(finish1.second, "Finish[1]");
+  finish1.first.set_value(true);
+
+  // Create and switch to a new stream
+  tested->MakeSubsequentStream();
+
+  auto read2 = sequencer.PopFrontWithName();
+  EXPECT_EQ(read2.second, "Read[2]");
+  // Start a read on the second stream
+  auto reader2 = tested->Read({200, 200});
+  auto future2 = reader2->Read();
+
+  auto write2 = sequencer.PopFrontWithName();
+  EXPECT_EQ(write2.second, "Write[2]");
+  write2.first.set_value(true);
+
+  read2.first.set_value(true);
+
+  EXPECT_THAT(future2.get(),
+              VariantWith<storage_experimental::ReadPayload>(ResultOf(
+                  "contents are",
+                  [](storage_experimental::ReadPayload const& p) {
+                    return p.contents();
+                  },
+                  ElementsAre(absl::string_view{"payload-for-stream-2"}))));
+
+  EXPECT_THAT(reader2->Read().get(), VariantWith<Status>(IsOk()));
+
+  auto read2_eos = sequencer.PopFrontWithName();
+  EXPECT_EQ(read2_eos.second, "Read[2.eos]");
+  read2_eos.first.set_value(true);
+
+  auto finish2 = sequencer.PopFrontWithName();
+  EXPECT_EQ(finish2.second, "Finish[2]");
+  finish2.first.set_value(true);
+}
+
 }  // namespace
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_internal

google/cloud/storage/tests/async_client_integration_test.cc

@@ -17,6 +17,7 @@
 #include "google/cloud/storage/async/bucket_name.h"
 #include "google/cloud/storage/async/client.h"
 #include "google/cloud/storage/async/idempotency_policy.h"
+#include "google/cloud/storage/async/options.h"
 #include "google/cloud/storage/async/read_all.h"
 #include "google/cloud/storage/grpc_plugin.h"
 #include "google/cloud/storage/testing/storage_integration_test.h"
@@ -962,6 +963,61 @@ TEST_F(AsyncClientIntegrationTest, Open) {
                       storage::Generation(metadata->generation()));
 }
 
+TEST_F(AsyncClientIntegrationTest, OpenExceedMaximumRange) {
+  if (!UsingEmulator()) GTEST_SKIP();
+  auto async = AsyncClient(
+      TestOptions().set<storage_experimental::MaximumRangeSizeOption>(1024));
+  auto client = MakeIntegrationTestClient(true, TestOptions());
+  auto object_name = MakeRandomObjectName();
+
+  auto create = client.CreateBucket(
+      bucket_name(), storage::BucketMetadata{}.set_location("us-west4"));
+  if (!create && create.status().code() != StatusCode::kAlreadyExists) {
+    GTEST_FAIL() << "cannot create bucket: " << create.status();
+  }
+
+  auto constexpr kSize = 2048;
+  auto const block = MakeRandomData(kSize);
+
+  auto w =
+      async.StartAppendableObjectUpload(BucketName(bucket_name()), object_name)
+          .get();
+  ASSERT_STATUS_OK(w);
+  AsyncWriter writer;
+  AsyncToken token;
+  std::tie(writer, token) = *std::move(w);
+  auto p = writer.Write(std::move(token), WritePayload(block)).get();
+  ASSERT_STATUS_OK(p);
+  token = *std::move(p);
+
+  auto metadata = writer.Finalize(std::move(token)).get();
+  ASSERT_STATUS_OK(metadata);
+
+  auto spec = google::storage::v2::BidiReadObjectSpec{};
+  spec.set_bucket(BucketName(bucket_name()).FullName());
+  spec.set_object(object_name);
+  auto descriptor = async.Open(spec).get();
+  ASSERT_STATUS_OK(descriptor);
+
+  AsyncReader r0;
+  AsyncToken t0;
+  auto actual0 = std::string{};
+  std::tie(r0, t0) = descriptor->Read(0, kSize);
+  while (t0.valid()) {
+    auto read = r0.Read(std::move(t0)).get();
+    ASSERT_STATUS_OK(read);
+    ReadPayload p;
+    AsyncToken t;
+    std::tie(p, t) = *std::move(read);
+    for (auto sv : p.contents()) actual0 += std::string(sv);
+    t0 = std::move(t);
+  }
+
+  EXPECT_EQ(actual0.size(), kSize);
+  client.DeleteObject(bucket_name(), object_name,
+                      storage::Generation(metadata->generation()));
+}
+
 }  // namespace
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_experimental