From 944fe6ed196b7641bf70d0a779aacfc0da7f44f9 Mon Sep 17 00:00:00 2001 From: Gauri Kalra Date: Tue, 21 Jul 2026 09:40:38 +0000 Subject: [PATCH 1/3] fix(storage): consume server responses before finishing stream in AsyncWriter Close/Finalize --- .../internal/async/writer_connection_impl.cc | 89 +++-- .../async/writer_connection_impl_test.cc | 317 +++++++++++++----- 2 files changed, 304 insertions(+), 102 deletions(-) diff --git a/google/cloud/storage/internal/async/writer_connection_impl.cc b/google/cloud/storage/internal/async/writer_connection_impl.cc index 59c4dce471365..53c42c2b82323 100644 --- a/google/cloud/storage/internal/async/writer_connection_impl.cc +++ b/google/cloud/storage/internal/async/writer_connection_impl.cc @@ -261,17 +261,24 @@ future AsyncWriterConnectionImpl::OnClose(std::size_t upload_size, HandleFinishAfterError("Expected Finish() error after non-ok Write()")); } offset_ += upload_size; - std::unique_lock lk(mu_); - auto impl = impl_; - lk.unlock(); - return impl->Read() - .then([this](auto f) { return OnQuery(f.get()); }) - .then([this](auto g) { - auto status = g.get(); - if (!status) return make_ready_future(std::move(status).status()); - return Finish(); - }) - .then([](auto f) { return f.get(); }); + struct ConsumeLoop { + AsyncWriterConnectionImpl* self; + future> operator()( + future> f) { + auto response = f.get(); + if (!response.has_value()) return make_ready_future(StatusOr(true)); + return self->impl_->Read().then(*this); + } + }; + + return impl_->Read().then(ConsumeLoop{this}).then([this](auto f) { + auto res = f.get(); + if (!res) { + return self->Finish().then( + HandleFinishAfterError(std::move(res).status())); + } + return self->Finish().then([](auto f2) { return f2.get(); }); + }); } future> @@ -293,21 +300,53 @@ AsyncWriterConnectionImpl::OnFinalUpload(std::size_t upload_size, } offset_ += upload_size; - std::unique_lock lk(mu_); - auto impl = impl_; - lk.unlock(); - return impl->Read() - .then([this](auto f) { return OnQuery(f.get()); }) - .then([this](auto g) -> StatusOr { - auto status = g.get(); - if (!status) return std::move(status).status(); - if (!absl::holds_alternative( - persisted_state_)) { - return internal::InternalError( - "no object metadata returned after finalizing upload", - GCP_ERROR_INFO()); + struct ConsumeLoop { + AsyncWriterConnectionImpl* self; + future> operator()( + future> f) { + auto response = f.get(); + if (!response.has_value()) { + return make_ready_future(StatusOr(true)); + } + // Process intermediate messages. + if (response->has_write_handle()) { + self->latest_write_handle_ = response->write_handle(); + } + if (response->has_persisted_size()) { + self->persisted_state_ = response->persisted_size(); + } + if (response->has_resource()) { + self->persisted_state_ = response->resource(); + } + return self->impl_->Read().then(*this); + } + }; + + return impl_->Read() + .then(ConsumeLoop{this}) + .then([this]( + auto f) -> future> { + auto res = f.get(); + if (!res) { + return self->Finish() + .then(HandleFinishAfterError(std::move(res).status())) + .then([](auto f) { + return StatusOr(f.get()); + }); } - return absl::get(persisted_state_); + return self->Finish().then( + [self](auto f2) -> StatusOr { + auto status = f2.get(); + if (!status.ok()) return status; + if (!absl::holds_alternative( + self->persisted_state_)) { + return internal::InternalError( + "no object metadata returned after finalizing upload", + GCP_ERROR_INFO()); + } + return absl::get( + self->persisted_state_); + }); }); } diff --git a/google/cloud/storage/internal/async/writer_connection_impl_test.cc b/google/cloud/storage/internal/async/writer_connection_impl_test.cc index 48acede5d54a8..65c2ec8828a4f 100644 --- a/google/cloud/storage/internal/async/writer_connection_impl_test.cc +++ b/google/cloud/storage/internal/async/writer_connection_impl_test.cc @@ -275,12 +275,19 @@ TEST(AsyncWriterConnectionTest, FinalizeEmpty) { "test-only-algo"); return sequencer.PushBack("Write"); }); - EXPECT_CALL(*mock, Read).WillOnce([&]() { - return sequencer.PushBack("Read").then([](auto f) { - if (!f.get()) return std::optional(); - return std::make_optional(MakeTestResponse()); - }); - }); + EXPECT_CALL(*mock, Read) + .WillOnce([&]() { + return sequencer.PushBack("Read1").then([](auto f) { + if (!f.get()) return std::optional(); + return std::make_optional(MakeTestResponse()); + }); + }) + .WillOnce([&]() { + return sequencer.PushBack("Read2").then([](auto f) { + f.get(); + return std::optional(); + }); + }); EXPECT_CALL(*mock, Finish).WillOnce([&] { return sequencer.PushBack("Finish").then([](auto f) { if (f.get()) return Status{}; @@ -298,16 +305,20 @@ TEST(AsyncWriterConnectionTest, FinalizeEmpty) { ASSERT_THAT(next.second, "Write"); next.first.set_value(true); next = sequencer.PopFrontWithName(); - ASSERT_THAT(next.second, "Read"); + ASSERT_THAT(next.second, "Read1"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + ASSERT_THAT(next.second, "Read2"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + ASSERT_THAT(next.second, "Finish"); 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, FinalizeFails) { @@ -451,12 +462,19 @@ TEST(AsyncWriterConnectionTest, UnexpectedQueryFinalMissingResource) { EXPECT_CALL(*mock, Write).WillOnce([&](Request const&, grpc::WriteOptions) { return sequencer.PushBack("Write"); }); - EXPECT_CALL(*mock, Read).WillOnce([&]() { - return sequencer.PushBack("Read").then([](auto f) { - if (!f.get()) return std::optional(); - return std::make_optional(Response{}); - }); - }); + EXPECT_CALL(*mock, Read) + .WillOnce([&]() { + return sequencer.PushBack("Read1").then([](auto f) { + if (!f.get()) return std::optional(); + return std::make_optional(Response{}); + }); + }) + .WillOnce([&]() { + return sequencer.PushBack("Read2").then([](auto f) { + f.get(); + return std::optional(); + }); + }); EXPECT_CALL(*mock, Finish).WillOnce([&] { return sequencer.PushBack("Finish").then([](auto f) { if (f.get()) return Status{}; @@ -474,14 +492,18 @@ TEST(AsyncWriterConnectionTest, UnexpectedQueryFinalMissingResource) { ASSERT_THAT(next.second, "Write"); next.first.set_value(true); next = sequencer.PopFrontWithName(); - ASSERT_THAT(next.second, "Read"); + ASSERT_THAT(next.second, "Read1"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + ASSERT_THAT(next.second, "Read2"); next.first.set_value(true); - EXPECT_THAT(response.get(), StatusIs(StatusCode::kInternal)); - - tested.reset(); next = sequencer.PopFrontWithName(); ASSERT_THAT(next.second, "Finish"); next.first.set_value(true); + + EXPECT_THAT(response.get(), StatusIs(StatusCode::kInternal)); + + tested = {}; } TEST(AsyncWriterConnectionTest, FlushEmpty) { @@ -709,12 +731,19 @@ TEST(AsyncWriterConnectionTest, FinalizeAppendableNoChecksum) { EXPECT_FALSE(request.has_object_checksums()); return sequencer.PushBack("Write"); }); - EXPECT_CALL(*mock, Read).WillOnce([&]() { - return sequencer.PushBack("Read").then([](auto f) { - if (!f.get()) return std::optional(); - return std::make_optional(MakeTestResponse()); - }); - }); + EXPECT_CALL(*mock, Read) + .WillOnce([&]() { + return sequencer.PushBack("Read1").then([](auto f) { + if (!f.get()) return std::optional(); + return std::make_optional(MakeTestResponse()); + }); + }) + .WillOnce([&]() { + return sequencer.PushBack("Read2").then([](auto f) { + f.get(); + return std::optional(); + }); + }); EXPECT_CALL(*mock, Finish).WillOnce([&] { return sequencer.PushBack("Finish").then([](auto f) { if (f.get()) return Status{}; @@ -734,16 +763,20 @@ TEST(AsyncWriterConnectionTest, FinalizeAppendableNoChecksum) { ASSERT_THAT(next.second, "Write"); next.first.set_value(true); next = sequencer.PopFrontWithName(); - ASSERT_THAT(next.second, "Read"); + ASSERT_THAT(next.second, "Read1"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + ASSERT_THAT(next.second, "Read2"); next.first.set_value(true); + next = sequencer.PopFrontWithName(); + ASSERT_THAT(next.second, "Finish"); + 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, FinalizeAppendableWithExpectedChecksum) { @@ -760,12 +793,19 @@ TEST(AsyncWriterConnectionTest, FinalizeAppendableWithExpectedChecksum) { EXPECT_EQ(request.object_checksums().crc32c(), 123456); return sequencer.PushBack("Write"); }); - EXPECT_CALL(*mock, Read).WillOnce([&]() { - return sequencer.PushBack("Read").then([](auto f) { - if (!f.get()) return std::optional(); - return std::make_optional(MakeTestResponse()); - }); - }); + EXPECT_CALL(*mock, Read) + .WillOnce([&]() { + return sequencer.PushBack("Read1").then([](auto f) { + if (!f.get()) return std::optional(); + return std::make_optional(MakeTestResponse()); + }); + }) + .WillOnce([&]() { + return sequencer.PushBack("Read2").then([](auto f) { + f.get(); + return std::optional(); + }); + }); EXPECT_CALL(*mock, Finish).WillOnce([&] { return sequencer.PushBack("Finish").then([](auto f) { if (f.get()) return Status{}; @@ -790,16 +830,20 @@ TEST(AsyncWriterConnectionTest, FinalizeAppendableWithExpectedChecksum) { ASSERT_THAT(next.second, "Write"); next.first.set_value(true); next = sequencer.PopFrontWithName(); - ASSERT_THAT(next.second, "Read"); + ASSERT_THAT(next.second, "Read1"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + ASSERT_THAT(next.second, "Read2"); next.first.set_value(true); + next = sequencer.PopFrontWithName(); + ASSERT_THAT(next.second, "Finish"); + 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, @@ -817,12 +861,19 @@ TEST(AsyncWriterConnectionTest, 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 std::optional(); - return std::make_optional(MakeTestResponse()); - }); - }); + EXPECT_CALL(*mock, Read) + .WillOnce([&]() { + return sequencer.PushBack("Read1").then([](auto f) { + if (!f.get()) return std::optional(); + return std::make_optional(MakeTestResponse()); + }); + }) + .WillOnce([&]() { + return sequencer.PushBack("Read2").then([](auto f) { + f.get(); + return std::optional(); + }); + }); EXPECT_CALL(*mock, Finish).WillOnce([&] { return sequencer.PushBack("Finish").then([](auto f) { if (f.get()) return Status{}; @@ -848,16 +899,20 @@ TEST(AsyncWriterConnectionTest, ASSERT_THAT(next.second, "Write"); next.first.set_value(true); next = sequencer.PopFrontWithName(); - ASSERT_THAT(next.second, "Read"); + ASSERT_THAT(next.second, "Read1"); next.first.set_value(true); + next = sequencer.PopFrontWithName(); + ASSERT_THAT(next.second, "Read2"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + ASSERT_THAT(next.second, "Finish"); + 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, @@ -875,12 +930,19 @@ TEST(AsyncWriterConnectionTest, 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 std::optional(); - return std::make_optional(MakeTestResponse()); - }); - }); + EXPECT_CALL(*mock, Read) + .WillOnce([&]() { + return sequencer.PushBack("Read1").then([](auto f) { + if (!f.get()) return std::optional(); + return std::make_optional(MakeTestResponse()); + }); + }) + .WillOnce([&]() { + return sequencer.PushBack("Read2").then([](auto f) { + f.get(); + return std::optional(); + }); + }); EXPECT_CALL(*mock, Finish).WillOnce([&] { return sequencer.PushBack("Finish").then([](auto f) { if (f.get()) return Status{}; @@ -907,16 +969,20 @@ TEST(AsyncWriterConnectionTest, ASSERT_THAT(next.second, "Write"); next.first.set_value(true); next = sequencer.PopFrontWithName(); - ASSERT_THAT(next.second, "Read"); + ASSERT_THAT(next.second, "Read1"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + ASSERT_THAT(next.second, "Read2"); next.first.set_value(true); + next = sequencer.PopFrontWithName(); + ASSERT_THAT(next.second, "Finish"); + 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, @@ -934,12 +1000,19 @@ TEST(AsyncWriterConnectionTest, EXPECT_EQ(request.object_checksums().md5_hash(), "test-md5"); return sequencer.PushBack("Write"); }); - EXPECT_CALL(*mock, Read).WillOnce([&]() { - return sequencer.PushBack("Read").then([](auto f) { - if (!f.get()) return std::optional(); - return std::make_optional(MakeTestResponse()); - }); - }); + EXPECT_CALL(*mock, Read) + .WillOnce([&]() { + return sequencer.PushBack("Read1").then([](auto f) { + if (!f.get()) return std::optional(); + return std::make_optional(MakeTestResponse()); + }); + }) + .WillOnce([&]() { + return sequencer.PushBack("Read2").then([](auto f) { + f.get(); + return std::optional(); + }); + }); EXPECT_CALL(*mock, Finish).WillOnce([&] { return sequencer.PushBack("Finish").then([](auto f) { if (f.get()) return Status{}; @@ -966,16 +1039,20 @@ TEST(AsyncWriterConnectionTest, ASSERT_THAT(next.second, "Write"); next.first.set_value(true); next = sequencer.PopFrontWithName(); - ASSERT_THAT(next.second, "Read"); + ASSERT_THAT(next.second, "Read1"); next.first.set_value(true); + next = sequencer.PopFrontWithName(); + ASSERT_THAT(next.second, "Read2"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + ASSERT_THAT(next.second, "Finish"); + 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) { @@ -1084,12 +1161,19 @@ TEST(AsyncWriterConnectionTest, CloseEmpty) { "test-only-algo"); return sequencer.PushBack("Write"); }); - EXPECT_CALL(*mock, Read).WillOnce([&] { - return sequencer.PushBack("Read").then([](auto f) { - if (!f.get()) return std::optional(); - return std::make_optional(Response{}); - }); - }); + EXPECT_CALL(*mock, Read) + .WillOnce([&] { + return sequencer.PushBack("Read1").then([](auto f) { + if (!f.get()) return std::optional(); + return std::make_optional(Response{}); + }); + }) + .WillOnce([&] { + return sequencer.PushBack("Read2").then([](auto f) { + f.get(); + return std::optional(); + }); + }); EXPECT_CALL(*mock, Finish).WillOnce([&] { return sequencer.PushBack("Finish").then([](auto f) { if (f.get()) return Status{}; @@ -1108,7 +1192,11 @@ TEST(AsyncWriterConnectionTest, CloseEmpty) { next.first.set_value(true); next = sequencer.PopFrontWithName(); - ASSERT_THAT(next.second, "Read"); + ASSERT_THAT(next.second, "Read1"); + next.first.set_value(true); + + next = sequencer.PopFrontWithName(); + ASSERT_THAT(next.second, "Read2"); next.first.set_value(true); next = sequencer.PopFrontWithName(); @@ -1148,6 +1236,81 @@ TEST(AsyncWriterConnectionTest, CloseError) { EXPECT_THAT(response.get(), StatusIs(PermanentError().code())); } +TEST(AsyncWriterConnectionTest, CloseMultipleResponsesBeforeEOF) { + AsyncSequencer sequencer; + auto mock = std::make_unique(); + EXPECT_CALL(*mock, Cancel).Times(1); + EXPECT_CALL(*mock, Write) + .WillOnce([&](Request const& request, grpc::WriteOptions wopt) { + EXPECT_TRUE(request.flush()); + EXPECT_TRUE(request.state_lookup()); + EXPECT_TRUE(wopt.is_last_message()); + return sequencer.PushBack("Write"); + }); + + // Simulate the scenario where Close() prompts the server to return multiple + // intermediate response messages before sending EOF. The client should + // successfully consume all responses and complete Finish() cleanly without + // hanging. + EXPECT_CALL(*mock, Read) + .WillOnce([&] { + return sequencer.PushBack("Read1").then([](auto f) { + f.get(); + return std::make_optional(Response{}); + }); + }) + .WillOnce([&] { + return sequencer.PushBack("Read2").then([](auto f) { + f.get(); + auto r = Response{}; + r.mutable_write_handle()->set_handle("intermediate-handle"); + return std::make_optional(r); + }); + }) + .WillOnce([&] { + return sequencer.PushBack("Read3").then([](auto f) { + f.get(); + return std::optional(); + }); + }); + EXPECT_CALL(*mock, Finish).WillOnce([&] { + return sequencer.PushBack("Finish").then([](auto f) { + if (f.get()) return Status{}; + return PermanentError(); + }); + }); + auto hash = std::make_shared(); + EXPECT_CALL(*hash, Update(_, An(), _)).Times(1); + EXPECT_CALL(*hash, Finish).Times(0); + + auto tested = std::make_unique( + TestOptions(), MakeRequest(), std::move(mock), hash, 1024); + auto close = tested->Close(WritePayload{}); + + auto next = sequencer.PopFrontWithName(); + ASSERT_THAT(next.second, "Write"); + next.first.set_value(true); + + next = sequencer.PopFrontWithName(); + ASSERT_THAT(next.second, "Read1"); + next.first.set_value(true); + + next = sequencer.PopFrontWithName(); + ASSERT_THAT(next.second, "Read2"); + next.first.set_value(true); + + next = sequencer.PopFrontWithName(); + ASSERT_THAT(next.second, "Read3"); + next.first.set_value(true); + + next = sequencer.PopFrontWithName(); + ASSERT_THAT(next.second, "Finish"); + next.first.set_value(true); + + EXPECT_THAT(close.get(), IsOk()); + tested = {}; +} + GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END } // namespace storage_internal } // namespace cloud From 77af81c14e821ad00e5cfe33091c9da4755308a9 Mon Sep 17 00:00:00 2001 From: Gauri Kalra Date: Tue, 21 Jul 2026 09:52:13 +0000 Subject: [PATCH 2/3] Address feedback from Gemini code assistant --- .../connection_impl_appendable_upload_test.cc | 42 +++- .../async/connection_impl_upload_hash_test.cc | 206 +++++++++++------- .../async/connection_impl_upload_test.cc | 187 ++++++++++------ .../internal/async/writer_connection_impl.cc | 78 +++---- 4 files changed, 319 insertions(+), 194 deletions(-) diff --git a/google/cloud/storage/internal/async/connection_impl_appendable_upload_test.cc b/google/cloud/storage/internal/async/connection_impl_appendable_upload_test.cc index e216b73dc1c2c..189af38f74830 100644 --- a/google/cloud/storage/internal/async/connection_impl_appendable_upload_test.cc +++ b/google/cloud/storage/internal/async/connection_impl_appendable_upload_test.cc @@ -104,6 +104,12 @@ std::unique_ptr MakeCommonAppendStream( response.mutable_resource()->set_size(persisted_size + 1024); return std::make_optional(std::move(response)); }); + }) + // The third `Read()` call returns EOF. + .WillOnce([&] { + return sequencer.PushBack("Read(EOF)").then([](auto) { + return std::optional(); + }); }); EXPECT_CALL(*stream, Cancel).Times(1); @@ -304,6 +310,12 @@ TEST_F(AsyncConnectionImplAppendableTest, StartAppendableObjectUploadSuccess) { next = sequencer.PopFrontWithName(); EXPECT_EQ(next.second, "Read(FinalObject)"); next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read(EOF)"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Finish"); + next.first.set_value(true); auto response = w2.get(); ASSERT_STATUS_OK(response); @@ -312,9 +324,6 @@ TEST_F(AsyncConnectionImplAppendableTest, StartAppendableObjectUploadSuccess) { EXPECT_EQ(response->size(), 1024); writer.reset(); - next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Finish"); - next.first.set_value(true); } TEST_F(AsyncConnectionImplAppendableTest, ResumeAppendableObjectUploadSuccess) { @@ -370,15 +379,18 @@ TEST_F(AsyncConnectionImplAppendableTest, ResumeAppendableObjectUploadSuccess) { next = sequencer.PopFrontWithName(); EXPECT_EQ(next.second, "Read(FinalObject)"); next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read(EOF)"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Finish"); + next.first.set_value(true); auto response = w2.get(); ASSERT_STATUS_OK(response); EXPECT_EQ(response->size(), kPersistedSize + 1024); writer.reset(); - next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Finish"); - next.first.set_value(true); } TEST_F(AsyncConnectionImplAppendableTest, AppendableUploadTooManyTransients) { @@ -517,6 +529,12 @@ TEST_F(AsyncConnectionImplAppendableTest, AppendableUploadRedirect) { next = sequencer.PopFrontWithName(); EXPECT_EQ(next.second, "Read(FinalObject)"); next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read(EOF)"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Finish"); + next.first.set_value(true); auto response = w2.get(); ASSERT_STATUS_OK(response); @@ -525,9 +543,6 @@ TEST_F(AsyncConnectionImplAppendableTest, AppendableUploadRedirect) { EXPECT_EQ(response->size(), 1024 + 1024); writer.reset(); - next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Finish"); - next.first.set_value(true); } TEST_F(AsyncConnectionImplAppendableTest, AppendableUploadRedirectNoHandle) { @@ -614,6 +629,12 @@ TEST_F(AsyncConnectionImplAppendableTest, AppendableUploadRedirectNoHandle) { next = sequencer.PopFrontWithName(); EXPECT_EQ(next.second, "Read(FinalObject)"); next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read(EOF)"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Finish"); + next.first.set_value(true); auto response = w2.get(); ASSERT_STATUS_OK(response); @@ -622,9 +643,6 @@ TEST_F(AsyncConnectionImplAppendableTest, AppendableUploadRedirectNoHandle) { EXPECT_EQ(response->size(), 1024 + 1024); writer.reset(); - next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Finish"); - next.first.set_value(true); } } // namespace GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END diff --git a/google/cloud/storage/internal/async/connection_impl_upload_hash_test.cc b/google/cloud/storage/internal/async/connection_impl_upload_hash_test.cc index 37d2b3a71ffaa..7faabd3a7f5f8 100644 --- a/google/cloud/storage/internal/async/connection_impl_upload_hash_test.cc +++ b/google/cloud/storage/internal/async/connection_impl_upload_hash_test.cc @@ -216,16 +216,23 @@ TEST_P(AsyncConnectionImplUploadHashTest, StartUnbuffered) { EXPECT_TRUE(wopt.is_last_message()); return sequencer.PushBack("Write"); }); - EXPECT_CALL(*stream, Read).WillOnce([&] { - return sequencer.PushBack("Read").then([](auto) { - auto response = google::storage::v2::BidiWriteObjectResponse{}; - response.mutable_resource()->set_bucket( - "projects/_/buckets/test-bucket"); - response.mutable_resource()->set_name("test-object"); - response.mutable_resource()->set_generation(123456); - return std::make_optional(std::move(response)); - }); - }); + EXPECT_CALL(*stream, Read) + .WillOnce([&] { + return sequencer.PushBack("Read1").then([](auto) { + auto response = google::storage::v2::BidiWriteObjectResponse{}; + response.mutable_resource()->set_bucket( + "projects/_/buckets/test-bucket"); + response.mutable_resource()->set_name("test-object"); + response.mutable_resource()->set_generation(123456); + return std::make_optional(std::move(response)); + }); + }) + .WillOnce([&] { + return sequencer.PushBack("Read2").then([](auto) { + return std::optional< + google::storage::v2::BidiWriteObjectResponse>(); + }); + }); EXPECT_CALL(*stream, Cancel).Times(1); EXPECT_CALL(*stream, Finish).WillOnce([&] { return sequencer.PushBack("Finish").then([](auto) { return Status{}; }); @@ -267,7 +274,13 @@ TEST_P(AsyncConnectionImplUploadHashTest, StartUnbuffered) { EXPECT_EQ(next.second, "Write"); next.first.set_value(true); next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Read"); + EXPECT_EQ(next.second, "Read1"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read2"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Finish"); next.first.set_value(true); auto response = w2.get(); @@ -277,9 +290,6 @@ TEST_P(AsyncConnectionImplUploadHashTest, StartUnbuffered) { EXPECT_EQ(response->generation(), 123456); writer.reset(); - next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Finish"); - next.first.set_value(true); } TEST_P(AsyncConnectionImplUploadHashTest, @@ -316,16 +326,23 @@ TEST_P(AsyncConnectionImplUploadHashTest, EXPECT_TRUE(wopt.is_last_message()); return sequencer.PushBack("Write"); }); - EXPECT_CALL(*stream, Read).WillOnce([&] { - return sequencer.PushBack("Read").then([](auto) { - auto response = google::storage::v2::BidiWriteObjectResponse{}; - response.mutable_resource()->set_bucket( - "projects/_/buckets/test-bucket"); - response.mutable_resource()->set_name("test-object"); - response.mutable_resource()->set_generation(123456); - return std::make_optional(std::move(response)); - }); - }); + EXPECT_CALL(*stream, Read) + .WillOnce([&] { + return sequencer.PushBack("Read1").then([](auto) { + auto response = google::storage::v2::BidiWriteObjectResponse{}; + response.mutable_resource()->set_bucket( + "projects/_/buckets/test-bucket"); + response.mutable_resource()->set_name("test-object"); + response.mutable_resource()->set_generation(123456); + return std::make_optional(std::move(response)); + }); + }) + .WillOnce([&] { + return sequencer.PushBack("Read2").then([](auto) { + return std::optional< + google::storage::v2::BidiWriteObjectResponse>(); + }); + }); EXPECT_CALL(*stream, Cancel).Times(1); EXPECT_CALL(*stream, Finish).WillOnce([&] { return sequencer.PushBack("Finish").then([](auto) { return Status{}; }); @@ -367,7 +384,13 @@ TEST_P(AsyncConnectionImplUploadHashTest, EXPECT_EQ(next.second, "Write"); next.first.set_value(true); next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Read"); + EXPECT_EQ(next.second, "Read1"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read2"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Finish"); next.first.set_value(true); auto response = w2.get(); @@ -377,9 +400,6 @@ TEST_P(AsyncConnectionImplUploadHashTest, EXPECT_EQ(response->generation(), 123456); writer.reset(); - next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Finish"); - next.first.set_value(true); } TEST_P(AsyncConnectionImplUploadHashTest, ResumeUnbufferedWithPersistedData) { @@ -414,16 +434,23 @@ TEST_P(AsyncConnectionImplUploadHashTest, ResumeUnbufferedWithPersistedData) { EXPECT_TRUE(wopt.is_last_message()); return sequencer.PushBack("Write"); }); - EXPECT_CALL(*stream, Read).WillOnce([&] { - return sequencer.PushBack("Read").then([](auto) { - auto response = google::storage::v2::BidiWriteObjectResponse{}; - response.mutable_resource()->set_bucket( - "projects/_/buckets/test-bucket"); - response.mutable_resource()->set_name("test-object"); - response.mutable_resource()->set_generation(123456); - return std::make_optional(std::move(response)); - }); - }); + EXPECT_CALL(*stream, Read) + .WillOnce([&] { + return sequencer.PushBack("Read1").then([](auto) { + auto response = google::storage::v2::BidiWriteObjectResponse{}; + response.mutable_resource()->set_bucket( + "projects/_/buckets/test-bucket"); + response.mutable_resource()->set_name("test-object"); + response.mutable_resource()->set_generation(123456); + return std::make_optional(std::move(response)); + }); + }) + .WillOnce([&] { + return sequencer.PushBack("Read2").then([](auto) { + return std::optional< + google::storage::v2::BidiWriteObjectResponse>(); + }); + }); EXPECT_CALL(*stream, Cancel).Times(1); EXPECT_CALL(*stream, Finish).WillOnce([&] { return sequencer.PushBack("Finish").then([](auto) { return Status{}; }); @@ -465,7 +492,13 @@ TEST_P(AsyncConnectionImplUploadHashTest, ResumeUnbufferedWithPersistedData) { EXPECT_EQ(next.second, "Write"); next.first.set_value(true); next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Read"); + EXPECT_EQ(next.second, "Read1"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read2"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Finish"); next.first.set_value(true); auto response = w2.get(); @@ -475,9 +508,6 @@ TEST_P(AsyncConnectionImplUploadHashTest, ResumeUnbufferedWithPersistedData) { EXPECT_EQ(response->generation(), 123456); writer.reset(); - next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Finish"); - next.first.set_value(true); } TEST_P(AsyncConnectionImplUploadHashTest, StartBuffered) { @@ -538,6 +568,13 @@ TEST_P(AsyncConnectionImplUploadHashTest, StartBuffered) { response.mutable_resource()->set_generation(123456); return std::make_optional(std::move(response)); }); + }) + .WillOnce([&]() { + return sequencer.PushBack("Read(3)").then( + [](auto f) -> std::optional< + google::storage::v2::BidiWriteObjectResponse> { + return std::nullopt; + }); }); EXPECT_CALL(*stream, Cancel).Times(1); EXPECT_CALL(*stream, Finish).WillOnce([&] { @@ -590,6 +627,12 @@ TEST_P(AsyncConnectionImplUploadHashTest, StartBuffered) { next = sequencer.PopFrontWithName(); EXPECT_EQ(next.second, "Read(2)"); next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read(3)"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Finish"); + next.first.set_value(true); auto response = w2.get(); ASSERT_STATUS_OK(response); @@ -598,9 +641,6 @@ TEST_P(AsyncConnectionImplUploadHashTest, StartBuffered) { EXPECT_EQ(response->generation(), 123456); writer.reset(); - next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Finish"); - next.first.set_value(true); } TEST_P(AsyncConnectionImplUploadHashTest, ResumeBufferedWithoutPersistedData) { @@ -636,16 +676,23 @@ TEST_P(AsyncConnectionImplUploadHashTest, ResumeBufferedWithoutPersistedData) { EXPECT_TRUE(wopt.is_last_message()); return sequencer.PushBack("Write"); }); - EXPECT_CALL(*stream, Read).WillOnce([&] { - return sequencer.PushBack("Read").then([](auto) { - auto response = google::storage::v2::BidiWriteObjectResponse{}; - response.mutable_resource()->set_bucket( - "projects/_/buckets/test-bucket"); - response.mutable_resource()->set_name("test-object"); - response.mutable_resource()->set_generation(123456); - return std::make_optional(std::move(response)); - }); - }); + EXPECT_CALL(*stream, Read) + .WillOnce([&] { + return sequencer.PushBack("Read1").then([](auto) { + auto response = google::storage::v2::BidiWriteObjectResponse{}; + response.mutable_resource()->set_bucket( + "projects/_/buckets/test-bucket"); + response.mutable_resource()->set_name("test-object"); + response.mutable_resource()->set_generation(123456); + return std::make_optional(std::move(response)); + }); + }) + .WillOnce([&] { + return sequencer.PushBack("Read2").then([](auto) { + return std::optional< + google::storage::v2::BidiWriteObjectResponse>(); + }); + }); EXPECT_CALL(*stream, Cancel).Times(1); EXPECT_CALL(*stream, Finish).WillOnce([&] { return sequencer.PushBack("Finish").then([](auto) { return Status{}; }); @@ -687,7 +734,13 @@ TEST_P(AsyncConnectionImplUploadHashTest, ResumeBufferedWithoutPersistedData) { EXPECT_EQ(next.second, "Write"); next.first.set_value(true); next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Read"); + EXPECT_EQ(next.second, "Read1"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read2"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Finish"); next.first.set_value(true); auto response = w2.get(); @@ -697,9 +750,6 @@ TEST_P(AsyncConnectionImplUploadHashTest, ResumeBufferedWithoutPersistedData) { EXPECT_EQ(response->generation(), 123456); writer.reset(); - next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Finish"); - next.first.set_value(true); } TEST_P(AsyncConnectionImplUploadHashTest, ResumeBufferedWithPersistedData) { @@ -734,16 +784,23 @@ TEST_P(AsyncConnectionImplUploadHashTest, ResumeBufferedWithPersistedData) { EXPECT_TRUE(wopt.is_last_message()); return sequencer.PushBack("Write"); }); - EXPECT_CALL(*stream, Read).WillOnce([&] { - return sequencer.PushBack("Read").then([](auto) { - auto response = google::storage::v2::BidiWriteObjectResponse{}; - response.mutable_resource()->set_bucket( - "projects/_/buckets/test-bucket"); - response.mutable_resource()->set_name("test-object"); - response.mutable_resource()->set_generation(123456); - return std::make_optional(std::move(response)); - }); - }); + EXPECT_CALL(*stream, Read) + .WillOnce([&] { + return sequencer.PushBack("Read1").then([](auto) { + auto response = google::storage::v2::BidiWriteObjectResponse{}; + response.mutable_resource()->set_bucket( + "projects/_/buckets/test-bucket"); + response.mutable_resource()->set_name("test-object"); + response.mutable_resource()->set_generation(123456); + return std::make_optional(std::move(response)); + }); + }) + .WillOnce([&] { + return sequencer.PushBack("Read2").then([](auto) { + return std::optional< + google::storage::v2::BidiWriteObjectResponse>(); + }); + }); EXPECT_CALL(*stream, Cancel).Times(1); EXPECT_CALL(*stream, Finish).WillOnce([&] { return sequencer.PushBack("Finish").then([](auto) { return Status{}; }); @@ -785,7 +842,13 @@ TEST_P(AsyncConnectionImplUploadHashTest, ResumeBufferedWithPersistedData) { EXPECT_EQ(next.second, "Write"); next.first.set_value(true); next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Read"); + EXPECT_EQ(next.second, "Read1"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read2"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Finish"); next.first.set_value(true); auto response = w2.get(); @@ -795,9 +858,6 @@ TEST_P(AsyncConnectionImplUploadHashTest, ResumeBufferedWithPersistedData) { EXPECT_EQ(response->generation(), 123456); writer.reset(); - next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Finish"); - next.first.set_value(true); } } // namespace diff --git a/google/cloud/storage/internal/async/connection_impl_upload_test.cc b/google/cloud/storage/internal/async/connection_impl_upload_test.cc index 8fce65544a5d2..d0b617fef28d1 100644 --- a/google/cloud/storage/internal/async/connection_impl_upload_test.cc +++ b/google/cloud/storage/internal/async/connection_impl_upload_test.cc @@ -168,16 +168,23 @@ TEST_F(AsyncConnectionImplTest, StartUnbufferedUpload) { EXPECT_TRUE(wopt.is_last_message()); return sequencer.PushBack("Write"); }); - EXPECT_CALL(*stream, Read).WillOnce([&] { - return sequencer.PushBack("Read").then([](auto) { - auto response = google::storage::v2::BidiWriteObjectResponse{}; - response.mutable_resource()->set_bucket( - "projects/_/buckets/test-bucket"); - response.mutable_resource()->set_name("test-object"); - response.mutable_resource()->set_generation(123456); - return std::make_optional(std::move(response)); - }); - }); + EXPECT_CALL(*stream, Read) + .WillOnce([&] { + return sequencer.PushBack("Read1").then([](auto) { + auto response = google::storage::v2::BidiWriteObjectResponse{}; + response.mutable_resource()->set_bucket( + "projects/_/buckets/test-bucket"); + response.mutable_resource()->set_name("test-object"); + response.mutable_resource()->set_generation(123456); + return std::make_optional(std::move(response)); + }); + }) + .WillOnce([&] { + return sequencer.PushBack("Read2").then([](auto) { + return std::optional< + google::storage::v2::BidiWriteObjectResponse>(); + }); + }); EXPECT_CALL(*stream, Cancel).Times(1); EXPECT_CALL(*stream, Finish).WillOnce([&] { return sequencer.PushBack("Finish").then( @@ -255,7 +262,13 @@ TEST_F(AsyncConnectionImplTest, StartUnbufferedUpload) { EXPECT_EQ(next.second, "Write"); next.first.set_value(true); next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Read"); + EXPECT_EQ(next.second, "Read1"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read2"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Finish"); next.first.set_value(true); auto response = w2.get(); @@ -265,9 +278,6 @@ TEST_F(AsyncConnectionImplTest, StartUnbufferedUpload) { EXPECT_EQ(response->generation(), 123456); writer.reset(); - next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Finish"); - next.first.set_value(true); } TEST_F(AsyncConnectionImplTest, UnbufferedUploadNewUploadWithTimeout) { @@ -288,16 +298,23 @@ TEST_F(AsyncConnectionImplTest, UnbufferedUploadNewUploadWithTimeout) { EXPECT_CALL(*stream, Write).WillOnce([&sequencer] { return sequencer.PushBack("Write"); }); - EXPECT_CALL(*stream, Read).WillOnce([&sequencer] { - return sequencer.PushBack("Read").then([](auto) { - auto response = google::storage::v2::BidiWriteObjectResponse{}; - response.mutable_resource()->set_bucket( - "projects/_/buckets/test-bucket"); - response.mutable_resource()->set_name("test-object"); - response.mutable_resource()->set_generation(123456); - return std::make_optional(std::move(response)); - }); - }); + EXPECT_CALL(*stream, Read) + .WillOnce([&sequencer] { + return sequencer.PushBack("Read1").then([](auto) { + auto response = google::storage::v2::BidiWriteObjectResponse{}; + response.mutable_resource()->set_bucket( + "projects/_/buckets/test-bucket"); + response.mutable_resource()->set_name("test-object"); + response.mutable_resource()->set_generation(123456); + return std::make_optional(std::move(response)); + }); + }) + .WillOnce([&sequencer] { + return sequencer.PushBack("Read2").then([](auto) { + return std::optional< + google::storage::v2::BidiWriteObjectResponse>(); + }); + }); EXPECT_CALL(*stream, Cancel).Times(1); EXPECT_CALL(*stream, Finish).WillOnce([&sequencer] { return sequencer.PushBack("Finish").then([](auto) { return Status{}; }); @@ -356,17 +373,23 @@ TEST_F(AsyncConnectionImplTest, UnbufferedUploadNewUploadWithTimeout) { timer = sequencer.PopFrontWithName(); EXPECT_EQ(timer.second, "MakeRelativeTimer"); next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Read"); + EXPECT_EQ(next.second, "Read1"); + timer.first.set_value(false); // simulate a cancelled timer. + next.first.set_value(true); + timer = sequencer.PopFrontWithName(); + EXPECT_EQ(timer.second, "MakeRelativeTimer"); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read2"); timer.first.set_value(false); // simulate a cancelled timer. next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Finish"); + next.first.set_value(true); auto response = w2.get(); ASSERT_STATUS_OK(response); writer.reset(); - next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Finish"); - next.first.set_value(true); } class MockAsyncBidiWriteObjectStreamLifetime @@ -512,16 +535,23 @@ TEST_F(AsyncConnectionImplTest, ResumeUnbufferedUpload) { EXPECT_TRUE(wopt.is_last_message()); return sequencer.PushBack("Write"); }); - EXPECT_CALL(*stream, Read).WillOnce([&] { - return sequencer.PushBack("Read").then([](auto) { - auto response = google::storage::v2::BidiWriteObjectResponse{}; - response.mutable_resource()->set_bucket( - "projects/_/buckets/test-bucket"); - response.mutable_resource()->set_name("test-object"); - response.mutable_resource()->set_generation(123456); - return std::make_optional(std::move(response)); - }); - }); + EXPECT_CALL(*stream, Read) + .WillOnce([&] { + return sequencer.PushBack("Read1").then([](auto) { + auto response = google::storage::v2::BidiWriteObjectResponse{}; + response.mutable_resource()->set_bucket( + "projects/_/buckets/test-bucket"); + response.mutable_resource()->set_name("test-object"); + response.mutable_resource()->set_generation(123456); + return std::make_optional(std::move(response)); + }); + }) + .WillOnce([&] { + return sequencer.PushBack("Read2").then([](auto) { + return std::optional< + google::storage::v2::BidiWriteObjectResponse>(); + }); + }); EXPECT_CALL(*stream, Cancel).Times(1); EXPECT_CALL(*stream, Finish).WillOnce([&] { return sequencer.PushBack("Finish").then( @@ -574,7 +604,13 @@ TEST_F(AsyncConnectionImplTest, ResumeUnbufferedUpload) { EXPECT_EQ(next.second, "Write"); next.first.set_value(true); next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Read"); + EXPECT_EQ(next.second, "Read1"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read2"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Finish"); next.first.set_value(true); auto response = w2.get(); @@ -584,9 +620,6 @@ TEST_F(AsyncConnectionImplTest, ResumeUnbufferedUpload) { EXPECT_EQ(response->generation(), 123456); writer.reset(); - next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Finish"); - next.first.set_value(true); } TEST_F(AsyncConnectionImplTest, ResumeUnbufferedUploadFinalized) { @@ -868,16 +901,23 @@ TEST_F(AsyncConnectionImplTest, BufferedUploadNewUpload) { EXPECT_TRUE(wopt.is_last_message()); return sequencer.PushBack("Write"); }); - EXPECT_CALL(*stream, Read).WillOnce([&] { - return sequencer.PushBack("Read").then([](auto) { - auto response = google::storage::v2::BidiWriteObjectResponse{}; - response.mutable_resource()->set_bucket( - "projects/_/buckets/test-bucket"); - response.mutable_resource()->set_name("test-object"); - response.mutable_resource()->set_generation(123456); - return std::make_optional(std::move(response)); - }); - }); + EXPECT_CALL(*stream, Read) + .WillOnce([&] { + return sequencer.PushBack("Read1").then([](auto) { + auto response = google::storage::v2::BidiWriteObjectResponse{}; + response.mutable_resource()->set_bucket( + "projects/_/buckets/test-bucket"); + response.mutable_resource()->set_name("test-object"); + response.mutable_resource()->set_generation(123456); + return std::make_optional(std::move(response)); + }); + }) + .WillOnce([&] { + return sequencer.PushBack("Read2").then([](auto) { + return std::optional< + google::storage::v2::BidiWriteObjectResponse>(); + }); + }); EXPECT_CALL(*stream, Cancel).Times(1); EXPECT_CALL(*stream, Finish).WillOnce([&] { return sequencer.PushBack("Finish").then( @@ -924,7 +964,13 @@ TEST_F(AsyncConnectionImplTest, BufferedUploadNewUpload) { EXPECT_EQ(next.second, "Write"); next.first.set_value(true); next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Read"); + EXPECT_EQ(next.second, "Read1"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read2"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Finish"); next.first.set_value(true); auto response = w1.get(); @@ -934,9 +980,6 @@ TEST_F(AsyncConnectionImplTest, BufferedUploadNewUpload) { EXPECT_EQ(response->generation(), 123456); writer.reset(); - next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Finish"); - next.first.set_value(true); } TEST_F(AsyncConnectionImplTest, ResumeBufferedUploadNewUploadResume) { @@ -1114,13 +1157,20 @@ TEST_F(AsyncConnectionImplTest, ResumeBufferedUpload) { EXPECT_TRUE(wopt.is_last_message()); return sequencer.PushBack("Write"); }); - EXPECT_CALL(*stream, Read).WillOnce([&] { - return sequencer.PushBack("Read").then([](auto) { - auto response = google::storage::v2::BidiWriteObjectResponse{}; - *response.mutable_resource() = TestProtoObject(); - return std::make_optional(std::move(response)); - }); - }); + EXPECT_CALL(*stream, Read) + .WillOnce([&] { + return sequencer.PushBack("Read1").then([](auto) { + auto response = google::storage::v2::BidiWriteObjectResponse{}; + *response.mutable_resource() = TestProtoObject(); + return std::make_optional(std::move(response)); + }); + }) + .WillOnce([&] { + return sequencer.PushBack("Read2").then([](auto) { + return std::optional< + google::storage::v2::BidiWriteObjectResponse>(); + }); + }); EXPECT_CALL(*stream, Cancel).Times(1); EXPECT_CALL(*stream, Finish).WillOnce([&] { return sequencer.PushBack("Finish").then( @@ -1173,16 +1223,19 @@ TEST_F(AsyncConnectionImplTest, ResumeBufferedUpload) { EXPECT_EQ(next.second, "Write"); next.first.set_value(true); next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Read"); + EXPECT_EQ(next.second, "Read1"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Read2"); + next.first.set_value(true); + next = sequencer.PopFrontWithName(); + EXPECT_EQ(next.second, "Finish"); next.first.set_value(true); auto response = w2.get(); EXPECT_THAT(response, IsOkAndHolds(IsProtoEqual(TestProtoObject()))); writer.reset(); - next = sequencer.PopFrontWithName(); - EXPECT_EQ(next.second, "Finish"); - next.first.set_value(true); } TEST_F(AsyncConnectionImplTest, ResumeBufferedUploadFinalized) { diff --git a/google/cloud/storage/internal/async/writer_connection_impl.cc b/google/cloud/storage/internal/async/writer_connection_impl.cc index 53c42c2b82323..4a49f045b4f0f 100644 --- a/google/cloud/storage/internal/async/writer_connection_impl.cc +++ b/google/cloud/storage/internal/async/writer_connection_impl.cc @@ -261,23 +261,24 @@ future AsyncWriterConnectionImpl::OnClose(std::size_t upload_size, HandleFinishAfterError("Expected Finish() error after non-ok Write()")); } offset_ += upload_size; + std::unique_lock lk(mu_); + auto impl = impl_; + lk.unlock(); + struct ConsumeLoop { - AsyncWriterConnectionImpl* self; - future> operator()( + std::shared_ptr impl; + + future operator()( future> f) { auto response = f.get(); - if (!response.has_value()) return make_ready_future(StatusOr(true)); - return self->impl_->Read().then(*this); + if (!response.has_value()) return make_ready_future(); + return impl->Read().then(*this); } }; - return impl_->Read().then(ConsumeLoop{this}).then([this](auto f) { - auto res = f.get(); - if (!res) { - return self->Finish().then( - HandleFinishAfterError(std::move(res).status())); - } - return self->Finish().then([](auto f2) { return f2.get(); }); + return impl->Read().then(ConsumeLoop{impl}).then([this](auto f) { + f.get(); + return Finish(); }); } @@ -300,14 +301,18 @@ AsyncWriterConnectionImpl::OnFinalUpload(std::size_t upload_size, } offset_ += upload_size; + std::unique_lock lk(mu_); + auto impl = impl_; + lk.unlock(); + struct ConsumeLoop { AsyncWriterConnectionImpl* self; - future> operator()( + std::shared_ptr impl; + + future operator()( future> f) { auto response = f.get(); - if (!response.has_value()) { - return make_ready_future(StatusOr(true)); - } + if (!response.has_value()) return make_ready_future(); // Process intermediate messages. if (response->has_write_handle()) { self->latest_write_handle_ = response->write_handle(); @@ -318,36 +323,25 @@ AsyncWriterConnectionImpl::OnFinalUpload(std::size_t upload_size, if (response->has_resource()) { self->persisted_state_ = response->resource(); } - return self->impl_->Read().then(*this); + return impl->Read().then(*this); } }; - return impl_->Read() - .then(ConsumeLoop{this}) - .then([this]( - auto f) -> future> { - auto res = f.get(); - if (!res) { - return self->Finish() - .then(HandleFinishAfterError(std::move(res).status())) - .then([](auto f) { - return StatusOr(f.get()); - }); - } - return self->Finish().then( - [self](auto f2) -> StatusOr { - auto status = f2.get(); - if (!status.ok()) return status; - if (!absl::holds_alternative( - self->persisted_state_)) { - return internal::InternalError( - "no object metadata returned after finalizing upload", - GCP_ERROR_INFO()); - } - return absl::get( - self->persisted_state_); - }); - }); + return impl->Read().then(ConsumeLoop{this, impl}).then([this](auto f) { + f.get(); + return Finish().then( + [this](auto f2) -> StatusOr { + auto status = f2.get(); + if (!status.ok()) return status; + if (!absl::holds_alternative( + persisted_state_)) { + return internal::InternalError( + "no object metadata returned after finalizing upload", + GCP_ERROR_INFO()); + } + return absl::get(persisted_state_); + }); + }); } future> AsyncWriterConnectionImpl::OnQuery( From 85a4af424eb2240d15bd9667b4c2752c9f11a018 Mon Sep 17 00:00:00 2001 From: Gauri Kalra Date: Thu, 23 Jul 2026 04:30:40 +0000 Subject: [PATCH 3/3] Fix failing tests --- .../internal/async/connection_impl_upload_hash_test.cc | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/google/cloud/storage/internal/async/connection_impl_upload_hash_test.cc b/google/cloud/storage/internal/async/connection_impl_upload_hash_test.cc index 7faabd3a7f5f8..8d208f9cc3b9a 100644 --- a/google/cloud/storage/internal/async/connection_impl_upload_hash_test.cc +++ b/google/cloud/storage/internal/async/connection_impl_upload_hash_test.cc @@ -570,11 +570,10 @@ TEST_P(AsyncConnectionImplUploadHashTest, StartBuffered) { }); }) .WillOnce([&]() { - return sequencer.PushBack("Read(3)").then( - [](auto f) -> std::optional< - google::storage::v2::BidiWriteObjectResponse> { - return std::nullopt; - }); + return sequencer.PushBack("Read(3)").then([](auto) { + return std::optional< + google::storage::v2::BidiWriteObjectResponse>{}; + }); }); EXPECT_CALL(*stream, Cancel).Times(1); EXPECT_CALL(*stream, Finish).WillOnce([&] {