fix(storage): cleanup interrupted downloads (googleapis#7064)

Downloads can be interrupted by the application. Sometimes this results
in incomplete cleanup of the handle, which when reused would start
failing some `curl_easy_setopt()` calls.

google/cloud/storage/internal/curl_download_request.cc

@@ -60,6 +60,15 @@ CurlDownloadRequest::CurlDownloadRequest()
       multi_(nullptr, &curl_multi_cleanup),
       spill_(CURL_MAX_WRITE_SIZE) {}
 
+CurlDownloadRequest::~CurlDownloadRequest() {
+  if (!multi_ && !handle_.handle_) return;  // moved-from, nothing to do
+  if (!curl_closed_) (void)Close();         // may need to shutdown handles
+  if (factory_) {
+    factory_->CleanupHandle(std::move(handle_));
+    factory_->CleanupMultiHandle(std::move(multi_));
+  }
+}
+
 StatusOr<HttpResponse> CurlDownloadRequest::Close() {
   TRACE_STATE();
   // Set the the closing_ flag to trigger a return 0 from the next read

google/cloud/storage/internal/curl_download_request.h

@@ -50,13 +50,7 @@ class CurlDownloadRequest : public ObjectReadSource {
  public:
   explicit CurlDownloadRequest();
 
-  ~CurlDownloadRequest() override {
-    if (!factory_) {
-      return;
-    }
-    factory_->CleanupHandle(std::move(handle_));
-    factory_->CleanupMultiHandle(std::move(multi_));
-  }
+  ~CurlDownloadRequest() override;
 
   CurlDownloadRequest(CurlDownloadRequest&&) = default;
   CurlDownloadRequest& operator=(CurlDownloadRequest&& rhs) = default;
@@ -76,6 +70,9 @@ class CurlDownloadRequest : public ObjectReadSource {
    */
   StatusOr<ReadSourceResult> Read(char* buf, std::size_t n) override;
 
+  /// Debug and test only, help identify download handles.
+  void* id() const { return handle_.handle_.get(); }
+
  private:
   friend class CurlRequestBuilder;
   /// Set the underlying CurlHandle options on a new CurlDownloadRequest.

google/cloud/storage/tests/curl_download_request_integration_test.cc

@@ -30,6 +30,8 @@ inline namespace STORAGE_CLIENT_NS {
 namespace internal {
 namespace {
 
+using ::google::cloud::testing_util::IsOk;
+
 std::string HttpBinEndpoint() {
   return google::cloud::internal::GetEnv("HTTPBIN_ENDPOINT")
       .value_or("https://nghttp2.org/httpbin");
@@ -72,6 +74,74 @@ TEST(CurlDownloadRequestTest, SimpleStream) {
   EXPECT_EQ(kDownloadedLines, count);
 }
 
+// Run one attempt of the Regression7051 test. This is wrapped in a retry loop,
+// as integration tests flake due to unrelated (and unavoidable) problems, e.g.,
+// trying to setup connections.
+Status AttemptRegression7051() {
+  // Download the maximum number of lines supported by httpbin.org
+  auto constexpr kDownloadedLines = 100;
+  auto constexpr kTestPoolSize = 32;
+  auto factory =
+      std::make_shared<PooledCurlHandleFactory>(kTestPoolSize, Options{});
+
+  auto make_download = [&] {
+    CurlRequestBuilder builder(
+        HttpBinEndpoint() + "/stream/" + std::to_string(kDownloadedLines),
+        factory);
+    return builder.BuildDownloadRequest(std::string{});
+  };
+
+  auto error = [](std::string msg) {
+    return Status(StatusCode::kUnknown, std::move(msg));
+  };
+
+  auto constexpr kBufferSize = kDownloadedLines;
+  char buffer[kBufferSize];
+
+  void* id;
+  {
+    auto r_no_close = make_download();
+    id = r_no_close.id();
+    if (id == nullptr) return error("r_no_close.id()==nulltptr");
+    auto read = r_no_close.Read(buffer, kBufferSize);
+    if (!read) return std::move(read).status();
+  }
+
+  {
+    auto r_partial_close = make_download();
+    if (r_partial_close.id() != id) return error("r_partial_close.id() != id");
+    auto read = r_partial_close.Read(buffer, kBufferSize);
+    if (!read) return std::move(read).status();
+    auto close = r_partial_close.Close();
+    if (!close) return std::move(close).status();
+  }
+
+  auto r_full = make_download();
+  if (r_full.id() != id) return error("r_full.id() != id");
+  do {
+    auto read = r_full.Read(buffer, kBufferSize);
+    if (!read) return std::move(read).status();
+    if (read->response.status_code != 100) break;
+  } while (true);
+  auto close = r_full.Close();
+  if (!close) return std::move(close).status();
+
+  return Status{};
+}
+
+/// @test Prevent regressions of #7051: re-using a stream after a partial read.
+TEST(CurlDownloadRequestTest, Regression7051) {
+  auto delay = std::chrono::seconds(1);
+  auto status = Status{};
+  for (int i = 0; i != 3; ++i) {
+    status = AttemptRegression7051();
+    if (status.ok()) break;
+    std::this_thread::sleep_for(delay);
+    delay *= 2;
+  }
+  EXPECT_THAT(status, IsOk());
+}
+
 }  // namespace
 }  // namespace internal
 }  // namespace STORAGE_CLIENT_NS