fix(storage): prevent indefinite hang when calling Close() on async appendable uploads

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

@@ -225,7 +225,17 @@ future<Status> AsyncWriterConnectionImpl::OnClose(std::size_t upload_size,
         HandleFinishAfterError("Expected Finish() error after non-ok Write()"));
   }
   offset_ += upload_size;
-  return Finish().then([](auto f) { return f.get(); });
+  std::unique_lock<std::mutex> 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(); });
 }
 
 future<StatusOr<google::storage::v2::Object>>

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

@@ -850,6 +850,12 @@ 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 absl::optional<Response>();
+      return absl::make_optional(Response{});
+    });
+  });
   EXPECT_CALL(*mock, Finish).WillOnce([&] {
     return sequencer.PushBack("Finish").then([](auto f) {
       if (f.get()) return Status{};
@@ -867,6 +873,10 @@ TEST(AsyncWriterConnectionTest, CloseEmpty) {
   ASSERT_THAT(next.second, "Write");
   next.first.set_value(true);
 
+  next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Read");
+  next.first.set_value(true);
+
   next = sequencer.PopFrontWithName();
   ASSERT_THAT(next.second, "Finish");
   next.first.set_value(true);

google/cloud/storage/tests/async_client_integration_test.cc

@@ -810,7 +810,7 @@ TEST_F(AsyncClientIntegrationTest, ResumeAppendableObjectUpload) {
     token = *std::move(p);
   }
 
-  writer.Close();
+  ASSERT_STATUS_OK(writer.Close().get());
 
   // Reset the existing writer and resume the upload.
   writer = AsyncWriter();