fix(rest): too many debug headers (#10054)

We were injecting a `:curl-peer` pseudo-header (usually used for
troubleshooting) for each `Read()` call.  The existing tests did not
catch this because this headers were never exposed to the application
layer either.

google/cloud/internal/curl_http_payload.cc

@@ -31,6 +31,10 @@ StatusOr<std::size_t> CurlHttpPayload::Read(absl::Span<char> buffer) {
   return impl_->Read(buffer);
 }
 
+std::multimap<std::string, std::string> CurlHttpPayload::headers() const {
+  return impl_->headers();
+}
+
 StatusOr<std::string> ReadAll(std::unique_ptr<HttpPayload> payload,
                               std::size_t read_size) {
   std::string output_buffer;

google/cloud/internal/curl_http_payload.h

@@ -43,6 +43,8 @@ class CurlHttpPayload : public HttpPayload {
 
   StatusOr<std::size_t> Read(absl::Span<char> buffer) override;
 
+  std::multimap<std::string, std::string> headers() const;
+
  private:
   friend class CurlRestResponse;
   CurlHttpPayload(std::unique_ptr<CurlImpl> impl, Options options);

google/cloud/internal/curl_impl.cc

@@ -342,6 +342,8 @@ std::size_t CurlImpl::WriteCallback(void* ptr, std::size_t size,
   if (!all_headers_received_ && buffer_.empty()) {
     all_headers_received_ = true;
     http_code_ = handle_.GetResponseCode();
+    // Capture the peer (the HTTP server), used for troubleshooting.
+    received_headers_.emplace(":curl-peer", handle_.GetPeer());
     return WriteAllBytesToSpillBuffer(ptr, size, nmemb);
   }
   return WriteToUserBuffer(ptr, size, nmemb);
@@ -419,8 +421,6 @@ void CurlImpl::SetUrl(
 
 void CurlImpl::OnTransferDone() {
   http_code_ = handle_.GetResponseCode();
-  // Capture the peer (the HTTP server), used for troubleshooting.
-  received_headers_.emplace(":curl-peer", handle_.GetPeer());
   TRACE_STATE() << "\n";
 
   // handle_ was removed from multi_ as part of the transfer completing in
@@ -656,7 +656,6 @@ StatusOr<std::size_t> CurlImpl::ReadImpl(absl::Span<char> output) {
     return bytes_read;
   }
   TRACE_STATE() << ", http code=" << http_code_ << "\n";
-  received_headers_.emplace(":curl-peer", handle_.GetPeer());
   return bytes_read;
 }
 

google/cloud/internal/curl_rest_client_integration_test.cc

@@ -14,10 +14,12 @@
 
 #include "google/cloud/common_options.h"
 #include "google/cloud/credentials.h"
+#include "google/cloud/internal/curl_http_payload.h"
 #include "google/cloud/internal/curl_options.h"
 #include "google/cloud/internal/getenv.h"
 #include "google/cloud/internal/rest_client.h"
 #include "google/cloud/log.h"
+#include "google/cloud/testing_util/contains_once.h"
 #include "google/cloud/testing_util/status_matchers.h"
 #include <gmock/gmock.h>
 #include <nlohmann/json.hpp>
@@ -28,10 +30,14 @@ namespace rest_internal {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 namespace {
 
+using ::google::cloud::testing_util::ContainsOnce;
+using ::testing::_;
 using ::testing::Contains;
 using ::testing::Eq;
 using ::testing::HasSubstr;
 using ::testing::Not;
+using ::testing::NotNull;
+using ::testing::Pair;
 
 class RestClientIntegrationTest : public ::testing::Test {
  protected:
@@ -420,6 +426,31 @@ TEST_F(RestClientIntegrationTest, PostFormData) {
   EXPECT_THAT((*form)[form_pair_3.first], Eq(form_pair_3.second));
 }
 
+TEST_F(RestClientIntegrationTest, PeerPseudoHeader) {
+  auto client = MakeDefaultRestClient(url_, {});
+  RestRequest request;
+  request.SetPath("stream/100");
+  auto response_status = RetryRestRequest([&] { return client->Get(request); });
+  ASSERT_STATUS_OK(response_status);
+  auto response = *std::move(response_status);
+  EXPECT_THAT(response->StatusCode(), Eq(HttpStatusCode::kOk));
+  EXPECT_THAT(response->Headers(), ContainsOnce(Pair(":curl-peer", _)));
+
+  // Reading in small buffers used to cause errors.
+  auto payload = std::move(*response).ExtractPayload();
+  auto constexpr kBufferSize = 16;
+  char buffer[kBufferSize];
+  while (true) {
+    auto bytes = payload->Read(absl::MakeSpan(buffer, kBufferSize));
+    ASSERT_STATUS_OK(bytes);
+    if (*bytes == 0) break;
+  }
+  auto* payload_impl =
+      dynamic_cast<rest_internal::CurlHttpPayload*>(payload.get());
+  ASSERT_THAT(payload_impl, NotNull());
+  EXPECT_THAT(payload_impl->headers(), ContainsOnce(Pair(":curl-peer", _)));
+}
+
 }  // namespace
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace rest_internal

google/cloud/storage/tests/CMakeLists.txt

@@ -54,6 +54,7 @@ set(storage_client_integration_tests
     object_plenty_clients_serially_integration_test.cc
     object_plenty_clients_simultaneously_integration_test.cc
     object_read_headers_integration_test.cc
+    object_read_large_integration_test.cc
     object_read_preconditions_integration_test.cc
     object_read_range_integration_test.cc
     object_read_stream_integration_test.cc

google/cloud/storage/tests/object_read_large_integration_test.cc

@@ -0,0 +1,98 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "google/cloud/storage/testing/storage_integration_test.h"
+#include "google/cloud/internal/getenv.h"
+#include "google/cloud/testing_util/status_matchers.h"
+#include "absl/strings/str_split.h"
+#include <gmock/gmock.h>
+#include <fstream>
+#include <iterator>
+#include <string>
+#include <vector>
+
+namespace google {
+namespace cloud {
+namespace storage {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+namespace {
+
+// This test depends on Linux-specific features.
+#if GTEST_OS_LINUX
+using ::google::cloud::internal::GetEnv;
+
+class ObjectReadLargeIntegrationTest
+    : public google::cloud::storage::testing::StorageIntegrationTest {};
+
+std::size_t CurrentRss() {
+  auto is = std::ifstream("/proc/self/statm");
+  std::vector<std::string> fields = absl::StrSplit(
+      std::string{std::istreambuf_iterator<char>{is.rdbuf()}, {}}, ' ');
+  // The fields are documented in proc(5).
+  return std::stoull(fields[1]) * 4096;
+}
+
+std::string DebugRss() {
+  std::ostringstream os;
+  for (auto const* path : {"/proc/self/status", "/proc/self/maps"}) {
+    os << "\n" << path << "\n";
+    auto is = std::ifstream(path);
+    os << std::string{std::istreambuf_iterator<char>{is.rdbuf()}, {}};
+  }
+  return std::move(os).str();
+}
+
+TEST_F(ObjectReadLargeIntegrationTest, LimitedMemoryGrowth) {
+  StatusOr<Client> client = MakeIntegrationTestClient();
+  ASSERT_STATUS_OK(client);
+
+  auto bucket_name = GetEnv("GOOGLE_CLOUD_CPP_STORAGE_TEST_BUCKET_NAME");
+  ASSERT_TRUE(bucket_name.has_value());
+
+  // This environment variable is not defined in the CI builds. It can be used
+  // to override the object in manual tests.
+  auto object_name = GetEnv("GOOGLE_CLOUD_CPP_STORAGE_TEST_OBJECT_NAME_HUGE");
+  if (!object_name) {
+    object_name = MakeRandomObjectName();
+    auto data = MakeRandomData(10 * 1024 * 1024);
+    auto meta =
+        client->InsertObject(*bucket_name, *object_name, std::move(data));
+    ASSERT_STATUS_OK(meta);
+    ScheduleForDelete(*meta);
+  }
+
+  auto constexpr kBufferSize = 128 * 1024;
+  auto constexpr kRssTolerance = 32 * 1024 * 1024;
+  std::vector<char> buffer(kBufferSize);
+  auto reader = client->ReadObject(*bucket_name, *object_name);
+  auto const initial_rss = CurrentRss();
+  std::cout << "Initial RSS = " << initial_rss << DebugRss() << std::endl;
+  auto tolerance = initial_rss + kRssTolerance;
+  std::int64_t offset = 0;
+  while (reader) {
+    reader.read(buffer.data(), buffer.size());
+    offset += reader.gcount();
+    auto const current_rss = CurrentRss();
+    EXPECT_LE(current_rss, tolerance) << "offset=" << offset << DebugRss();
+    if (current_rss >= tolerance) tolerance = current_rss + kRssTolerance;
+  }
+  EXPECT_STATUS_OK(reader.status());
+}
+#endif  // GTEST_OS_LINUX
+
+}  // namespace
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace storage
+}  // namespace cloud
+}  // namespace google

google/cloud/storage/tests/storage_client_integration_tests.bzl

@@ -55,6 +55,7 @@ storage_client_integration_tests = [
     "object_plenty_clients_serially_integration_test.cc",
     "object_plenty_clients_simultaneously_integration_test.cc",
     "object_read_headers_integration_test.cc",
+    "object_read_large_integration_test.cc",
     "object_read_preconditions_integration_test.cc",
     "object_read_range_integration_test.cc",
     "object_read_stream_integration_test.cc",