From 9c450d3d81ffef8fa33ed348616930da09380860 Mon Sep 17 00:00:00 2001 From: Gauri Kalra Date: Wed, 15 Jul 2026 07:45:23 +0000 Subject: [PATCH 1/2] feat(storage): implement internal pre-warmed stream dispatch and caching for BiDi reads --- google/cloud/storage/async/client.cc | 6 +- google/cloud/storage/async/connection.h | 6 + .../storage/internal/async/connection_impl.cc | 14 +- .../async/connection_impl_open_test.cc | 12 +- .../internal/async/object_descriptor_impl.cc | 30 ++- .../internal/async/object_descriptor_impl.h | 17 +- .../async/object_descriptor_impl_test.cc | 188 +++++++++++++++++- 7 files changed, 250 insertions(+), 23 deletions(-) diff --git a/google/cloud/storage/async/client.cc b/google/cloud/storage/async/client.cc index 9f9baf89da6ca..4999e3527546b 100644 --- a/google/cloud/storage/async/client.cc +++ b/google/cloud/storage/async/client.cc @@ -61,8 +61,10 @@ future> AsyncClient::Open( future> AsyncClient::Open( google::storage::v2::BidiReadObjectSpec spec, Options opts) { return connection_ - ->Open({std::move(spec), google::cloud::internal::MergeOptions( - std::move(opts), connection_->options())}) + ->Open({std::move(spec), + google::cloud::internal::MergeOptions(std::move(opts), + connection_->options()), + /*initial_read_range=*/absl::nullopt}) .then([](auto f) -> StatusOr { auto connection = f.get(); if (!connection) return std::move(connection).status(); diff --git a/google/cloud/storage/async/connection.h b/google/cloud/storage/async/connection.h index 8e9df06d9f9e9..012e967a0c9f1 100644 --- a/google/cloud/storage/async/connection.h +++ b/google/cloud/storage/async/connection.h @@ -82,9 +82,15 @@ class AsyncConnection { * We use a single struct as the input parameter for this function to * prevent breaking any mocks when additional parameters are needed. */ + struct InitialReadRange { + std::int64_t offset; + std::int64_t length; + }; + struct OpenParams { google::storage::v2::BidiReadObjectSpec read_spec; Options options; + absl::optional initial_read_range; }; /// Open an object to perform multiple reads. diff --git a/google/cloud/storage/internal/async/connection_impl.cc b/google/cloud/storage/internal/async/connection_impl.cc index 976fea52f2337..bd0dfb00d0a20 100644 --- a/google/cloud/storage/internal/async/connection_impl.cc +++ b/google/cloud/storage/internal/async/connection_impl.cc @@ -191,7 +191,13 @@ future> AsyncConnectionImpl::InsertObject( future>> AsyncConnectionImpl::Open(OpenParams p) { auto initial_request = google::storage::v2::BidiReadObjectRequest{}; - *initial_request.mutable_read_object_spec() = p.read_spec; + *initial_request.mutable_read_object_spec() = std::move(p.read_spec); + if (p.initial_read_range) { + auto* range = initial_request.add_read_ranges(); + range->set_read_offset(p.initial_read_range->offset); + range->set_read_length(p.initial_read_range->length); + range->set_read_id(1); + } auto current = internal::MakeImmutableOptions(p.options); // Get the policy factory and immediately create a policy. auto resume_policy = current->get()(); @@ -234,8 +240,8 @@ AsyncConnectionImpl::Open(OpenParams p) { using ReturnType = std::shared_ptr; return pending.then([rp = std::move(resume_policy), fa = std::move(factory), rs = std::move(p.read_spec), - options = std::move(p.options), refresh = refresh_]( - auto f) mutable -> StatusOr { + options = std::move(p.options), eager = p.initial_read_range, + refresh = refresh_](auto f) mutable -> StatusOr { auto result = f.get(); if (!result) return std::move(result).status(); @@ -255,7 +261,7 @@ AsyncConnectionImpl::Open(OpenParams p) { }; auto impl = std::make_shared( std::move(rp), std::move(fa), std::move(rs), std::move(result->stream), - std::move(options), std::move(transport_ok)); + std::move(options), std::move(transport_ok), eager); impl->Start(std::move(result->first_response)); return ReturnType(impl); }); diff --git a/google/cloud/storage/internal/async/connection_impl_open_test.cc b/google/cloud/storage/internal/async/connection_impl_open_test.cc index 941df81f692de..837e7af8e96cb 100644 --- a/google/cloud/storage/internal/async/connection_impl_open_test.cc +++ b/google/cloud/storage/internal/async/connection_impl_open_test.cc @@ -214,7 +214,8 @@ TEST(AsyncConnectionImplTest, OpenSimple) { auto request = google::storage::v2::BidiReadObjectSpec{}; ASSERT_TRUE(TextFormat::ParseFromString(kExpectedRequest, &request)); - auto pending = connection->Open({std::move(request), connection->options()}); + auto pending = connection->Open( + {std::move(request), connection->options(), /*initial_read_range=*/absl::nullopt}); auto next = sequencer.PopFrontWithName(); EXPECT_EQ(next.second, "Start"); @@ -310,7 +311,8 @@ TEST(AsyncConnectionImplTest, HandleRedirectErrors) { auto request = google::storage::v2::BidiReadObjectSpec{}; ASSERT_TRUE(TextFormat::ParseFromString(kExpectedRequest0, &request)); - auto pending = connection->Open({std::move(request), connection->options()}); + auto pending = connection->Open( + {std::move(request), connection->options(), /*initial_read_range=*/absl::nullopt}); for (int i = 0; i != kRetryAttempts + 1; ++i) { auto next = sequencer.PopFrontWithName(); @@ -354,7 +356,8 @@ TEST(AsyncConnectionImplTest, StopOnPermanentError) { auto request = google::storage::v2::BidiReadObjectSpec{}; ASSERT_TRUE(TextFormat::ParseFromString(kExpectedRequest, &request)); - auto pending = connection->Open({std::move(request), connection->options()}); + auto pending = connection->Open( + {std::move(request), connection->options(), /*initial_read_range=*/absl::nullopt}); auto next = sequencer.PopFrontWithName(); EXPECT_EQ(next.second, "Start"); @@ -390,7 +393,8 @@ TEST(AsyncConnectionImplTest, TooManyTransienErrors) { auto request = google::storage::v2::BidiReadObjectSpec{}; ASSERT_TRUE(TextFormat::ParseFromString(kExpectedRequest, &request)); - auto pending = connection->Open({std::move(request), connection->options()}); + auto pending = connection->Open( + {std::move(request), connection->options(), /*initial_read_range=*/absl::nullopt}); for (int i = 0; i != kRetryAttempts + 1; ++i) { auto next = sequencer.PopFrontWithName(); diff --git a/google/cloud/storage/internal/async/object_descriptor_impl.cc b/google/cloud/storage/internal/async/object_descriptor_impl.cc index 4f145986521ad..f8a359d048c27 100644 --- a/google/cloud/storage/internal/async/object_descriptor_impl.cc +++ b/google/cloud/storage/internal/async/object_descriptor_impl.cc @@ -40,10 +40,14 @@ ObjectDescriptorImpl::ObjectDescriptorImpl( OpenStreamFactory make_stream, google::storage::v2::BidiReadObjectSpec read_object_spec, std::shared_ptr stream, Options options, - std::function transport_ok) + std::function transport_ok, + absl::optional initial_read_range) : resume_policy_prototype_(std::move(resume_policy)), make_stream_(std::move(make_stream)), read_object_spec_(std::move(read_object_spec)), + initial_read_range_(std::move(initial_read_range)), + initial_cache_consumed_(!initial_read_range_.has_value()), + read_id_generator_(initial_read_range_.has_value() ? 1 : 0), options_(std::move(options)), transport_ok_(std::move(transport_ok)) { stream_manager_ = std::make_unique( @@ -203,12 +207,26 @@ std::unique_ptr ObjectDescriptorImpl::Read( } auto it = stream_manager_->GetLeastBusyStream(); - auto const id = ++read_id_generator_; + bool is_cache_hit = false; + if (!initial_cache_consumed_ && initial_read_range_.has_value()) { + auto const& initial = *initial_read_range_; + // Ensure both offset and length match. + if (p.start == initial.offset && limit.value_or(0) == initial.length) { + is_cache_hit = true; + } + initial_cache_consumed_ = true; + } + + auto const id = is_cache_hit ? 1 : ++read_id_generator_; it->active_ranges.emplace(id, range); - auto& read_range = *it->stream->next_request.add_read_ranges(); - read_range.set_read_id(id); - read_range.set_read_offset(p.start); - read_range.set_read_length(p.length); + + if (!is_cache_hit) { + auto& read_range = *it->stream->next_request.add_read_ranges(); + read_range.set_read_id(id); + read_range.set_read_offset(p.start); + read_range.set_read_length(p.length); + } + Flush(std::move(lk), it); if (!internal::TracingEnabled(options_)) { diff --git a/google/cloud/storage/internal/async/object_descriptor_impl.h b/google/cloud/storage/internal/async/object_descriptor_impl.h index 7f5f654951f3b..68ea46ec1c4a3 100644 --- a/google/cloud/storage/internal/async/object_descriptor_impl.h +++ b/google/cloud/storage/internal/async/object_descriptor_impl.h @@ -15,6 +15,7 @@ #ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_ASYNC_OBJECT_DESCRIPTOR_IMPL_H #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_ASYNC_OBJECT_DESCRIPTOR_IMPL_H +#include "google/cloud/storage/async/connection.h" #include "google/cloud/storage/async/object_descriptor_connection.h" #include "google/cloud/storage/async/resume_policy.h" #include "google/cloud/storage/internal/async/multi_stream_manager.h" @@ -57,11 +58,14 @@ class ObjectDescriptorImpl : public storage::ObjectDescriptorConnection, public std::enable_shared_from_this { public: - ObjectDescriptorImpl(std::unique_ptr resume_policy, - OpenStreamFactory make_stream, - google::storage::v2::BidiReadObjectSpec read_object_spec, - std::shared_ptr stream, Options options = {}, - std::function transport_ok = {}); + ObjectDescriptorImpl( + std::unique_ptr resume_policy, + OpenStreamFactory make_stream, + google::storage::v2::BidiReadObjectSpec read_object_spec, + std::shared_ptr stream, Options options = {}, + std::function transport_ok = {}, + absl::optional + initial_read_range = absl::nullopt); ~ObjectDescriptorImpl() override; // Start the read loop. @@ -122,6 +126,9 @@ class ObjectDescriptorImpl mutable std::mutex mu_; google::storage::v2::BidiReadObjectSpec read_object_spec_; absl::optional metadata_; + absl::optional + initial_read_range_; + bool initial_cache_consumed_ = false; std::int64_t read_id_generator_ = 0; Options options_; diff --git a/google/cloud/storage/internal/async/object_descriptor_impl_test.cc b/google/cloud/storage/internal/async/object_descriptor_impl_test.cc index 5423423b4a08d..61dd43474dc3a 100644 --- a/google/cloud/storage/internal/async/object_descriptor_impl_test.cc +++ b/google/cloud/storage/internal/async/object_descriptor_impl_test.cc @@ -72,13 +72,14 @@ auto MakeTested( OpenStreamFactory make_stream, google::storage::v2::BidiReadObjectSpec read_object_spec, std::shared_ptr stream, - std::function transport_ok = [] { return true; }) { + std::function transport_ok = [] { return true; }, + absl::optional initial_read_range = absl::nullopt) { Options options; options.set(true); return std::make_shared( std::move(resume_policy), std::move(make_stream), std::move(read_object_spec), std::move(stream), std::move(options), - std::move(transport_ok)); + std::move(transport_ok), std::move(initial_read_range)); } MATCHER_P(IsProtoEqualModuloRepeatedFieldOrdering, value, @@ -286,6 +287,189 @@ TEST(ObjectDescriptorImpl, ReadSingleRange) { next.first.set_value(true); } +/// @test Verify that reading a pre-warmed cache range skips making a request. +TEST(ObjectDescriptorImpl, ReadCacheHit) { + auto constexpr kResponse0 = R"pb( + metadata { + bucket: "projects/_/buckets/test-bucket" + name: "test-object" + generation: 42 + } + read_handle { handle: "handle-12345" } + )pb"; + + auto constexpr kResponse1 = R"pb( + read_handle { handle: "handle-23456" } + object_data_ranges { + range_end: true + read_range { read_id: 1 read_offset: 20000 } + checksummed_data { + content: "The quick brown fox jumps over the lazy dog" + } + } + )pb"; + + AsyncSequencer sequencer; + auto stream = std::make_unique(); + EXPECT_CALL(*stream, Write).Times(0); // Cache hit means NO new request! + + EXPECT_CALL(*stream, Read) + .WillOnce([=, &sequencer]() { + return sequencer.PushBack("Read[1]").then([&](auto) { + auto response = Response{}; + EXPECT_TRUE(TextFormat::ParseFromString(kResponse1, &response)); + return absl::make_optional(response); + }); + }) + .WillOnce([&sequencer]() { + return sequencer.PushBack("Read[2]").then( + [](auto) { return absl::optional{}; }); + }); + EXPECT_CALL(*stream, Finish).WillOnce([&sequencer]() { + return sequencer.PushBack("Finish").then( + [](auto) { return PermanentError(); }); + }); + EXPECT_CALL(*stream, Cancel).Times(AtMost(1)); + + MockFactory factory; + EXPECT_CALL(factory, Call).WillOnce([](Request const&) { + return make_ready_future(StatusOr(PermanentError())); + }); + + auto tested = MakeTested(NoResume(), factory.AsStdFunction(), + google::storage::v2::BidiReadObjectSpec{}, + std::make_shared(std::move(stream)), + [] { return true; }, + storage::AsyncConnection::InitialReadRange{20000, 100}); + auto response = Response{}; + EXPECT_TRUE(TextFormat::ParseFromString(kResponse0, &response)); + tested->Start(std::move(response)); + + auto read1 = sequencer.PopFrontWithName(); + EXPECT_EQ(read1.second, "Read[1]"); + + // Call ReadParams exactly matching the cache + auto s1 = tested->Read({20000, 100}); + auto s1r1 = s1->Read(); + + read1.first.set_value(true); + + EXPECT_THAT(s1r1.get(), + VariantWith(ResultOf( + "contents are", + [](storage::ReadPayload const& p) { return p.contents(); }, + ElementsAre(absl::string_view{ + "The quick brown fox jumps over the lazy dog"})))); + + EXPECT_THAT(s1->Read().get(), VariantWith(IsOk())); + + auto next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read[2]"); + next.first.set_value(true); + + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Finish"); + next.first.set_value(true); +} + +/// @test Verify that reading a cache miss spawns a request with ID 2. +TEST(ObjectDescriptorImpl, ReadCacheMiss) { + 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: 2 read_offset: 35000 read_length: 100 } + )pb"; + auto constexpr kResponse1 = R"pb( + read_handle { handle: "handle-23456" } + object_data_ranges { + range_end: true + read_range { read_id: 2 read_offset: 35000 } + checksummed_data { + content: "Data for ID 2" + } + } + )pb"; + + AsyncSequencer sequencer; + auto stream = std::make_unique(); + EXPECT_CALL(*stream, 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(*stream, Read) + .WillOnce([=, &sequencer]() { + return sequencer.PushBack("Read[1]").then([&](auto) { + auto response = Response{}; + EXPECT_TRUE(TextFormat::ParseFromString(kResponse1, &response)); + return absl::make_optional(response); + }); + }) + .WillOnce([&sequencer]() { + return sequencer.PushBack("Read[2]").then( + [](auto) { return absl::optional{}; }); + }); + EXPECT_CALL(*stream, Finish).WillOnce([&sequencer]() { + return sequencer.PushBack("Finish").then( + [](auto) { return PermanentError(); }); + }); + EXPECT_CALL(*stream, Cancel).Times(AtMost(1)); + + MockFactory factory; + EXPECT_CALL(factory, Call).WillOnce([](Request const&) { + return make_ready_future(StatusOr(PermanentError())); + }); + + auto tested = MakeTested(NoResume(), factory.AsStdFunction(), + google::storage::v2::BidiReadObjectSpec{}, + std::make_shared(std::move(stream)), + [] { return true; }, + storage::AsyncConnection::InitialReadRange{20000, 100}); + auto response = Response{}; + EXPECT_TRUE(TextFormat::ParseFromString(kResponse0, &response)); + tested->Start(std::move(response)); + + auto read1 = sequencer.PopFrontWithName(); + EXPECT_EQ(read1.second, "Read[1]"); + + // Call ReadParams that MISSES the cache. (Requests offset 35000 instead of 20000). + auto s1 = tested->Read({35000, 100}); + auto s1r1 = s1->Read(); + + auto next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Write[1]"); + next.first.set_value(true); + read1.first.set_value(true); + + EXPECT_THAT(s1r1.get(), + VariantWith(ResultOf( + "contents are", + [](storage::ReadPayload const& p) { return p.contents(); }, + ElementsAre(absl::string_view{"Data for ID 2"})))); + + EXPECT_THAT(s1->Read().get(), VariantWith(IsOk())); + + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read[2]"); + next.first.set_value(true); + + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Finish"); + next.first.set_value(true); +} + /// @test Reading multiple ranges creates a single request. TEST(ObjectDescriptorImpl, ReadMultipleRanges) { auto constexpr kLength = 100; From ab0554961bc3f94cfc15edebda00047843e456b4 Mon Sep 17 00:00:00 2001 From: Gauri Kalra Date: Wed, 15 Jul 2026 09:23:17 +0000 Subject: [PATCH 2/2] Address feedback from Gemini code assistant --- google/cloud/storage/async/connection.h | 14 +++++--- .../storage/internal/async/connection_impl.cc | 6 ++-- .../internal/async/object_descriptor_impl.cc | 11 ++++--- .../internal/async/object_descriptor_impl.h | 6 ++-- .../async/object_descriptor_impl_test.cc | 33 ++++++++++--------- 5 files changed, 38 insertions(+), 32 deletions(-) diff --git a/google/cloud/storage/async/connection.h b/google/cloud/storage/async/connection.h index 012e967a0c9f1..6017de35d09cd 100644 --- a/google/cloud/storage/async/connection.h +++ b/google/cloud/storage/async/connection.h @@ -76,17 +76,21 @@ class AsyncConnection { virtual future> InsertObject( InsertObjectParams p) = 0; + /// A parameter to specify the initial offset and length of an eagerly-started + /// read. + struct InitialReadRange { + /// The starting offset of the read. + std::int64_t offset; + /// The number of bytes to read. + std::int64_t length; + }; + /** * A thin wrapper around the `Open()` parameters. * * We use a single struct as the input parameter for this function to * prevent breaking any mocks when additional parameters are needed. */ - struct InitialReadRange { - std::int64_t offset; - std::int64_t length; - }; - struct OpenParams { google::storage::v2::BidiReadObjectSpec read_spec; Options options; diff --git a/google/cloud/storage/internal/async/connection_impl.cc b/google/cloud/storage/internal/async/connection_impl.cc index bd0dfb00d0a20..6f4ffcb7c3fa3 100644 --- a/google/cloud/storage/internal/async/connection_impl.cc +++ b/google/cloud/storage/internal/async/connection_impl.cc @@ -191,7 +191,7 @@ future> AsyncConnectionImpl::InsertObject( future>> AsyncConnectionImpl::Open(OpenParams p) { auto initial_request = google::storage::v2::BidiReadObjectRequest{}; - *initial_request.mutable_read_object_spec() = std::move(p.read_spec); + *initial_request.mutable_read_object_spec() = p.read_spec; if (p.initial_read_range) { auto* range = initial_request.add_read_ranges(); range->set_read_offset(p.initial_read_range->offset); @@ -240,7 +240,7 @@ AsyncConnectionImpl::Open(OpenParams p) { using ReturnType = std::shared_ptr; return pending.then([rp = std::move(resume_policy), fa = std::move(factory), rs = std::move(p.read_spec), - options = std::move(p.options), eager = p.initial_read_range, + options = std::move(p.options), initial_read_range = p.initial_read_range, refresh = refresh_](auto f) mutable -> StatusOr { auto result = f.get(); if (!result) return std::move(result).status(); @@ -261,7 +261,7 @@ AsyncConnectionImpl::Open(OpenParams p) { }; auto impl = std::make_shared( std::move(rp), std::move(fa), std::move(rs), std::move(result->stream), - std::move(options), std::move(transport_ok), eager); + initial_read_range, std::move(options), std::move(transport_ok)); impl->Start(std::move(result->first_response)); return ReturnType(impl); }); diff --git a/google/cloud/storage/internal/async/object_descriptor_impl.cc b/google/cloud/storage/internal/async/object_descriptor_impl.cc index f8a359d048c27..9545f98acffd5 100644 --- a/google/cloud/storage/internal/async/object_descriptor_impl.cc +++ b/google/cloud/storage/internal/async/object_descriptor_impl.cc @@ -39,15 +39,16 @@ ObjectDescriptorImpl::ObjectDescriptorImpl( std::unique_ptr resume_policy, OpenStreamFactory make_stream, google::storage::v2::BidiReadObjectSpec read_object_spec, - std::shared_ptr stream, Options options, - std::function transport_ok, - absl::optional initial_read_range) + std::shared_ptr stream, + absl::optional initial_read_range, + Options options, + std::function transport_ok) : resume_policy_prototype_(std::move(resume_policy)), make_stream_(std::move(make_stream)), read_object_spec_(std::move(read_object_spec)), initial_read_range_(std::move(initial_read_range)), - initial_cache_consumed_(!initial_read_range_.has_value()), - read_id_generator_(initial_read_range_.has_value() ? 1 : 0), + initial_cache_consumed_(!initial_read_range.has_value()), + read_id_generator_(initial_read_range.has_value() ? 1 : 0), options_(std::move(options)), transport_ok_(std::move(transport_ok)) { stream_manager_ = std::make_unique( diff --git a/google/cloud/storage/internal/async/object_descriptor_impl.h b/google/cloud/storage/internal/async/object_descriptor_impl.h index 68ea46ec1c4a3..7ac8228927b4b 100644 --- a/google/cloud/storage/internal/async/object_descriptor_impl.h +++ b/google/cloud/storage/internal/async/object_descriptor_impl.h @@ -62,10 +62,10 @@ class ObjectDescriptorImpl std::unique_ptr resume_policy, OpenStreamFactory make_stream, google::storage::v2::BidiReadObjectSpec read_object_spec, - std::shared_ptr stream, Options options = {}, - std::function transport_ok = {}, + std::shared_ptr stream, absl::optional - initial_read_range = absl::nullopt); + initial_read_range, + Options options = {}, std::function transport_ok = {}); ~ObjectDescriptorImpl() override; // Start the read loop. diff --git a/google/cloud/storage/internal/async/object_descriptor_impl_test.cc b/google/cloud/storage/internal/async/object_descriptor_impl_test.cc index 61dd43474dc3a..c4d479fea1848 100644 --- a/google/cloud/storage/internal/async/object_descriptor_impl_test.cc +++ b/google/cloud/storage/internal/async/object_descriptor_impl_test.cc @@ -72,14 +72,15 @@ auto MakeTested( OpenStreamFactory make_stream, google::storage::v2::BidiReadObjectSpec read_object_spec, std::shared_ptr stream, - std::function transport_ok = [] { return true; }, - absl::optional initial_read_range = absl::nullopt) { + absl::optional initial_read_range = absl::nullopt, + std::function transport_ok = [] { return true; }) { Options options; options.set(true); return std::make_shared( std::move(resume_policy), std::move(make_stream), - std::move(read_object_spec), std::move(stream), std::move(options), - std::move(transport_ok), std::move(initial_read_range)); + std::move(read_object_spec), std::move(stream), + std::move(initial_read_range), std::move(options), + std::move(transport_ok)); } MATCHER_P(IsProtoEqualModuloRepeatedFieldOrdering, value, @@ -311,7 +312,8 @@ TEST(ObjectDescriptorImpl, ReadCacheHit) { AsyncSequencer sequencer; auto stream = std::make_unique(); - EXPECT_CALL(*stream, Write).Times(0); // Cache hit means NO new request! + // Cache hit means no new request. + EXPECT_CALL(*stream, Write).Times(0); EXPECT_CALL(*stream, Read) .WillOnce([=, &sequencer]() { @@ -339,7 +341,6 @@ TEST(ObjectDescriptorImpl, ReadCacheHit) { auto tested = MakeTested(NoResume(), factory.AsStdFunction(), google::storage::v2::BidiReadObjectSpec{}, std::make_shared(std::move(stream)), - [] { return true; }, storage::AsyncConnection::InitialReadRange{20000, 100}); auto response = Response{}; EXPECT_TRUE(TextFormat::ParseFromString(kResponse0, &response)); @@ -348,7 +349,7 @@ TEST(ObjectDescriptorImpl, ReadCacheHit) { auto read1 = sequencer.PopFrontWithName(); EXPECT_EQ(read1.second, "Read[1]"); - // Call ReadParams exactly matching the cache + // Call ReadParams exactly matching the cache. auto s1 = tested->Read({20000, 100}); auto s1r1 = s1->Read(); @@ -435,7 +436,6 @@ TEST(ObjectDescriptorImpl, ReadCacheMiss) { auto tested = MakeTested(NoResume(), factory.AsStdFunction(), google::storage::v2::BidiReadObjectSpec{}, std::make_shared(std::move(stream)), - [] { return true; }, storage::AsyncConnection::InitialReadRange{20000, 100}); auto response = Response{}; EXPECT_TRUE(TextFormat::ParseFromString(kResponse0, &response)); @@ -444,7 +444,7 @@ TEST(ObjectDescriptorImpl, ReadCacheMiss) { auto read1 = sequencer.PopFrontWithName(); EXPECT_EQ(read1.second, "Read[1]"); - // Call ReadParams that MISSES the cache. (Requests offset 35000 instead of 20000). + // Call ReadParams that misses the cache. (Requests offset 35000 instead of 20000). auto s1 = tested->Read({35000, 100}); auto s1r1 = s1->Read(); @@ -1964,7 +1964,7 @@ TEST(ObjectDescriptorImpl, MultiStreamOptimizationDisabled) { auto tested = std::make_shared( NoResume(), factory.AsStdFunction(), google::storage::v2::BidiReadObjectSpec{}, - std::make_shared(std::move(stream)), options); + std::make_shared(std::move(stream)), /*initial_read_range=*/absl::nullopt, options); tested->MakeSubsequentStream(); @@ -2009,6 +2009,7 @@ TEST(ObjectDescriptorImpl, IsOpenFalseOnPermanentError) { auto tested = MakeTested(NoResume(), factory.AsStdFunction(), google::storage::v2::BidiReadObjectSpec{}, std::make_shared(std::move(stream)), + /*initial_read_range=*/absl::nullopt, transport_ok_callback); EXPECT_TRUE(tested->IsOpen()); @@ -2027,7 +2028,7 @@ TEST(ObjectDescriptorImpl, IsOpenFalseOnTransportFailure) { auto tested = std::make_shared( NoResume(), factory.AsStdFunction(), google::storage::v2::BidiReadObjectSpec{}, - std::make_shared(std::move(stream)), Options{}, + std::make_shared(std::move(stream)), /*initial_read_range=*/absl::nullopt, Options{}, std::move(transport_ok)); EXPECT_FALSE(tested->IsOpen()); } @@ -2043,7 +2044,7 @@ TEST(ObjectDescriptorImpl, IsOpenTrueOnTransportSuccess) { auto tested = std::make_shared( NoResume(), factory.AsStdFunction(), google::storage::v2::BidiReadObjectSpec{}, - std::make_shared(std::move(stream)), Options{}, + std::make_shared(std::move(stream)), /*initial_read_range=*/absl::nullopt, Options{}, std::move(transport_ok)); EXPECT_TRUE(tested->IsOpen()); } @@ -2120,7 +2121,7 @@ TEST(ObjectDescriptorImpl, FullReadChecksumValidationSuccess) { auto tested = std::make_shared( NoResume(), factory.AsStdFunction(), google::storage::v2::BidiReadObjectSpec{}, - std::make_shared(std::move(stream)), options); + std::make_shared(std::move(stream)), /*initial_read_range=*/absl::nullopt, options); auto response = Response{}; EXPECT_TRUE(TextFormat::ParseFromString(kResponseMetadata, &response)); @@ -2227,7 +2228,7 @@ TEST(ObjectDescriptorImpl, FullReadChecksumValidationMismatchCrc32c) { auto tested = std::make_shared( NoResume(), factory.AsStdFunction(), google::storage::v2::BidiReadObjectSpec{}, - std::make_shared(std::move(stream)), options); + std::make_shared(std::move(stream)), /*initial_read_range=*/absl::nullopt, options); auto response = Response{}; EXPECT_TRUE(TextFormat::ParseFromString(kResponseMetadata, &response)); @@ -2335,7 +2336,7 @@ TEST(ObjectDescriptorImpl, FullReadChecksumValidationMismatchMd5) { auto tested = std::make_shared( NoResume(), factory.AsStdFunction(), google::storage::v2::BidiReadObjectSpec{}, - std::make_shared(std::move(stream)), options); + std::make_shared(std::move(stream)), /*initial_read_range=*/absl::nullopt, options); auto response = Response{}; EXPECT_TRUE(TextFormat::ParseFromString(kResponseMetadata, &response)); @@ -2443,7 +2444,7 @@ TEST(ObjectDescriptorImpl, PartialReadChecksumValidationBypassed) { auto tested = std::make_shared( NoResume(), factory.AsStdFunction(), google::storage::v2::BidiReadObjectSpec{}, - std::make_shared(std::move(stream)), options); + std::make_shared(std::move(stream)), /*initial_read_range=*/absl::nullopt, options); auto response = Response{}; EXPECT_TRUE(TextFormat::ParseFromString(kResponseMetadata, &response));