feat: Copy multi_curl_http_fetcher_async and file_util from B&A

This is part of a series of CLs to copy
the KV BYOS clients from B&A to common repo.

Bug: 411430242
Change-Id: Icf105e520ce1a068548e94db092a3464e2448853
GitOrigin-RevId: d587666a2fb3f2768089f93515530dc45c72c6b2

Author: Privacy Sandbox Team

.bazelrc

@@ -14,6 +14,7 @@ build --config=otel
 build --@com_google_googleurl//build_config:system_icu=0
 
 test --test_output=errors
+test --test_tag_filters=-flaky
 
 build:run_all_tests --cache_test_results=no
 test:run_all_tests --test_verbose_timeout_warnings

src/clients/http_fetcher_async/BUILD.bazel

@@ -34,6 +34,7 @@ cc_library(
     hdrs = [
         "curl_request_data.h",
     ],
+    visibility = ["//visibility:public"],
     deps = [
         "@com_google_absl//absl/container:flat_hash_map",
         "@com_google_absl//absl/functional:any_invocable",
@@ -115,6 +116,58 @@ cc_library(
     ],
 )
 
+cc_library(
+    name = "multi_curl_http_fetcher_async",
+    srcs = [
+        "multi_curl_http_fetcher_async.cc",
+        "multi_curl_request_manager.cc",
+        "multi_curl_request_manager.h",
+    ],
+    hdrs = [
+        "multi_curl_http_fetcher_async.h",
+    ],
+    deps = [
+        ":curl_request_data",
+        ":curl_request_queue",
+        ":curl_request_worker",
+        ":http_fetcher_async",
+        "//src/concurrent:executor",
+        "//src/util/file:file_util",
+        "@com_google_absl//absl/container:flat_hash_map",
+        "@com_google_absl//absl/log:check",
+        "@com_google_absl//absl/status",
+        "@com_google_absl//absl/strings",
+        "@com_google_absl//absl/time",
+        "@curl",
+        "@libevent//:event",
+    ],
+)
+
+cc_test(
+    name = "multi_curl_http_fetcher_async_test",
+    size = "medium",
+    srcs = ["multi_curl_http_fetcher_async_test.cc"],
+    flaky = 1,
+    tags = [
+        "noasan",  # TODO(b/411430242) - Fix ASAN errors in this test.
+        "notsan",  # TSAN cannot detect event library locking mechanism
+        "requires-network",
+    ],
+    deps = [
+        ":multi_curl_http_fetcher_async",
+        "//src/concurrent:executor",
+        "//src/logger:request_context_logger",
+        "@com_github_grpc_grpc//:grpc++",
+        "@com_google_absl//absl/status",
+        "@com_google_absl//absl/strings",
+        "@com_google_absl//absl/synchronization",
+        "@com_google_absl//absl/time",
+        "@com_google_googletest//:gtest",
+        "@com_google_googletest//:gtest_main",
+        "@rapidjson",
+    ],
+)
+
 cc_test(
     name = "curl_request_queue_test",
     size = "small",

src/clients/http_fetcher_async/curl_request_data.h

@@ -112,6 +112,7 @@ struct MultiCurlHttpFetcherAsyncOptions {
   // TODO(b/412330778): Rename to `curl_max_wait_time_ms`
   absl::Duration curl_max_wait_time = kDefaultMaxRequestWaitTime;
   int curl_queue_length = kDefaultMaxCurlPendingRequests;
+  bool skip_tls_verification = false;
 };
 
 // This struct maintains the data related to a Curl request, some of which

src/clients/http_fetcher_async/http_fetcher_async.h

@@ -1,4 +1,4 @@
-//  Copyright 2022 Google LLC
+//  Copyright 2025 Google LLC
 //
 //  Licensed under the Apache License, Version 2.0 (the "License");
 //  you may not use this file except in compliance with the License.

src/clients/http_fetcher_async/multi_curl_http_fetcher_async.cc

@@ -0,0 +1,274 @@
+//  Copyright 2025 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
+//
+//       http://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 "src/clients/http_fetcher_async/multi_curl_http_fetcher_async.h"
+
+#include <stdlib.h>
+
+#include <algorithm>
+#include <random>
+#include <sstream>
+#include <string>
+#include <utility>
+
+#include <curl/curl.h>
+
+#include "absl/log/check.h"
+#include "absl/strings/str_cat.h"
+#include "absl/time/time.h"
+#include "event2/thread.h"
+#include "src/util/file/file_util.h"
+
+namespace privacy_sandbox::server_common::clients {
+using ::grpc_event_engine::experimental::EventEngine;
+
+MultiCurlHttpFetcherAsync::MultiCurlHttpFetcherAsync(
+    server_common::Executor* executor,
+    const MultiCurlHttpFetcherAsyncOptions& options)
+    : executor_(executor),
+      keepalive_idle_sec_(options.keepalive_idle_sec),
+      keepalive_interval_sec_(options.keepalive_interval_sec),
+      skip_tls_verification_(options.skip_tls_verification),
+      ca_cert_(options.ca_cert),
+      num_curl_workers_(options.num_curl_workers) {
+  DCHECK_GT(num_curl_workers_, 0);
+  auto ca_cert_blob = GetFileContent(ca_cert_, /*log_on_error=*/true);
+  CHECK_OK(ca_cert_blob);
+  ca_cert_blob_ = *std::move(ca_cert_blob);
+
+  // Setup curl workers and their work queues.
+  curl_request_workers_.reserve(num_curl_workers_);
+  for (int i = 0; i < options.num_curl_workers; ++i) {
+    auto request_queue = std::make_unique<CurlRequestQueue>(
+        executor_, options.curl_queue_length, options.curl_max_wait_time);
+    curl_request_workers_.push_back(std::make_unique<CurlRequestWorker>(
+        executor_, *request_queue, options.curlmopt_maxconnects,
+        options.curlmopt_max_total_connections,
+        options.curlmopt_max_host_connections));
+    request_queues_.emplace_back(std::move(request_queue));
+  }
+}
+
+// The function declaration of WriteCallback is specified by libcurl.
+// Please do not modify the parameter types or ordering, although you
+// may modify the function name and body.
+// libcurl documentation: https://curl.se/libcurl/c/CURLOPT_WRITEFUNCTION.html
+//
+// data: A pointer to the data that was delivered over the wire.
+// size: (legacy) size is always 1. Represents 1 byte.
+// number_elements: the number of elements (each of size 1 byte) to write
+// output: a libcurl-client-provided pointer of where to save the data
+// return: number of bytes actually written to output
+static size_t WriteCallback(char* data, size_t size, size_t number_elements,
+                            std::string* output) {
+  output->append(reinterpret_cast<char*>(data), size * number_elements);
+  return size * number_elements;
+}
+
+// The function declaration of ReadCallback is specified by libcurl.
+// Please do not modify the parameter types or ordering, although you
+// may modify the function name and body.
+// libcurl documentation: https://curl.se/libcurl/c/CURLOPT_READFUNCTION.html
+static size_t ReadCallback(char* data, size_t size, size_t num_items,
+                           void* userdata) {
+  auto* to_upload = static_cast<DataToUpload*>(userdata);
+  if (to_upload->offset >= to_upload->data.size()) {
+    // No more data to upload.
+    return 0;
+  }
+  size_t num_bytes_to_upload =
+      std::min(to_upload->data.size() - to_upload->offset, num_items * size);
+  memcpy(data, to_upload->data.c_str() + to_upload->offset,
+         num_bytes_to_upload);
+  PS_VLOG(8) << "PUT-ing data (offset: " << to_upload->offset
+             << ", chunk size: " << num_bytes_to_upload
+             << "): " << to_upload->data;
+  to_upload->offset += num_bytes_to_upload;
+  return num_bytes_to_upload;
+}
+
+void MultiCurlHttpFetcherAsync::FetchUrls(
+    const std::vector<HTTPRequest>& requests, absl::Duration timeout,
+    OnDoneFetchUrls done_callback) {
+  FetchUrlsWithMetadata(
+      requests, timeout,
+      [done_callback = std::move(done_callback)](
+          std::vector<absl::StatusOr<HTTPResponse>> response_vector) mutable {
+        std::vector<absl::StatusOr<std::string>> results;
+        results.reserve(response_vector.size());
+        for (auto& response : response_vector) {
+          if (response.ok()) {
+            results.emplace_back(std::move(response)->body);
+          } else {
+            results.emplace_back(std::move(response).status());
+          }
+        }
+        std::move(done_callback)(std::move(results));
+      });
+}
+
+void MultiCurlHttpFetcherAsync::FetchUrlsWithMetadata(
+    const std::vector<HTTPRequest>& requests, absl::Duration timeout,
+    OnDoneFetchUrlsWithMetadata done_callback) {
+  // The FetchUrl lambdas are the owners of the underlying FetchUrlsLifetime and
+  // this shared_ptr will be destructed once the last FetchUrl lambda finishes.
+  // Using a shared_ptr here allows us to avoid making MultiCurlHttpFetcherAsync
+  // the owner of the FetchUrlsLifetime, which would complicate cleanup on
+  // MultiCurlHttpFetcherAsync destruction during a pending FetchUrls call.
+  auto shared_lifetime = std::make_shared<
+      MultiCurlHttpFetcherAsync::FetchUrlsWithMetadataLifetime>();
+  shared_lifetime->pending_results = requests.size();
+  shared_lifetime->all_done_callback = std::move(done_callback);
+  shared_lifetime->results =
+      std::vector<absl::StatusOr<HTTPResponse>>(requests.size());
+  if (requests.empty()) {
+    // Execute callback immediately if there are no requests.
+    std::move(shared_lifetime->all_done_callback)(
+        std::move(shared_lifetime->results));
+    return;
+  }
+  for (int i = 0; i < requests.size(); i++) {
+    FetchUrlWithMetadata(
+        requests.at(i), absl::ToInt64Milliseconds(timeout),
+        [i, shared_lifetime](absl::StatusOr<HTTPResponse> result) {
+          absl::MutexLock lock_results(&shared_lifetime->results_mu);
+          shared_lifetime->results[i] = std::move(result);
+          if (--shared_lifetime->pending_results == 0) {
+            std::move(shared_lifetime->all_done_callback)(
+                std::move(shared_lifetime->results));
+          }
+        });
+  }
+}
+
+std::unique_ptr<CurlRequestData> MultiCurlHttpFetcherAsync::CreateCurlRequest(
+    const HTTPRequest& request, int timeout_ms, int64_t keepalive_idle_sec,
+    int64_t keepalive_interval_sec, OnDoneFetchUrlWithMetadata done_callback) {
+  auto curl_request_data = std::make_unique<CurlRequestData>(
+      request.headers, std::move(done_callback), request.include_headers,
+      request.redirect_config.get_redirect_url);
+  CURL* req_handle = curl_request_data->req_handle;
+  curl_easy_setopt(req_handle, CURLOPT_WRITEFUNCTION, WriteCallback);
+  curl_easy_setopt(req_handle, CURLOPT_URL, request.url.begin());
+  curl_easy_setopt(req_handle, CURLOPT_WRITEDATA,
+                   &curl_request_data->response_with_metadata.body);
+  curl_easy_setopt(req_handle, CURLOPT_FOLLOWLOCATION, 1);
+  if (request.redirect_config.strict_http) {
+    curl_easy_setopt(req_handle, CURLOPT_REDIR_PROTOCOLS_STR, "http,https");
+  }
+
+  curl_easy_setopt(req_handle, CURLOPT_TIMEOUT_MS, timeout_ms);
+  // Enable TCP keep-alive to keep connection warm.
+  curl_easy_setopt(req_handle, CURLOPT_TCP_KEEPALIVE, 1L);
+  curl_easy_setopt(req_handle, CURLOPT_TCP_KEEPIDLE,
+                   static_cast<int64_t>(keepalive_idle_sec));
+  curl_easy_setopt(req_handle, CURLOPT_TCP_KEEPINTVL,
+                   static_cast<int64_t>(keepalive_interval_sec));
+  // Allow upto 1200 seconds idle time.
+  curl_easy_setopt(req_handle, CURLOPT_MAXAGE_CONN, 1200L);
+  // Set CURLOPT_ACCEPT_ENCODING to an empty string to pass all supported
+  // encodings. See https://curl.se/libcurl/c/CURLOPT_ACCEPT_ENCODING.html.
+  curl_easy_setopt(req_handle, CURLOPT_ACCEPT_ENCODING, "");
+
+  if (skip_tls_verification_) {
+    curl_easy_setopt(req_handle, CURLOPT_SSL_VERIFYPEER, 0L);
+    curl_easy_setopt(req_handle, CURLOPT_SSL_VERIFYHOST, 0L);
+    curl_easy_setopt(req_handle, CURLOPT_PROXY_SSL_VERIFYPEER, 0L);
+    curl_easy_setopt(req_handle, CURLOPT_PROXY_SSL_VERIFYHOST, 0L);
+  } else {
+    curl_easy_setopt(req_handle, CURLOPT_CAINFO_BLOB, ca_cert_blob_.c_str());
+  }
+  // Set HTTP headers.
+  if (!request.headers.empty()) {
+    curl_easy_setopt(req_handle, CURLOPT_HTTPHEADER,
+                     curl_request_data->headers_list_ptr);
+  }
+  return curl_request_data;
+}
+
+void MultiCurlHttpFetcherAsync::ScheduleAsyncCurlRequest(
+    std::unique_ptr<CurlRequestData> request) {
+  // Spreads the requests in a uniformly distributed (random) manner across
+  // the available queues/workers.
+  static std::random_device random_device;
+  static std::mt19937 generator(random_device());
+  std::uniform_int_distribution<int> distribute(0, num_curl_workers_ - 1);
+  int num_worker = distribute(generator);
+
+  auto& request_queue = request_queues_[num_worker];
+  absl::MutexLock lock(&request_queue->Mu());
+  if (request_queue->Full()) {
+    std::move(request->done_callback)(
+        absl::ResourceExhaustedError("Request Queue Limit Exceeded."));
+    return;
+  }
+
+  request_queue->Enqueue(std::move(request));
+}
+
+void MultiCurlHttpFetcherAsync::FetchUrl(const HTTPRequest& request,
+                                         int timeout_ms,
+                                         OnDoneFetchUrl done_callback) {
+  FetchUrlWithMetadata(request, timeout_ms,
+                       [callback = std::move(done_callback)](
+                           absl::StatusOr<HTTPResponse> response) mutable {
+                         if (response.ok()) {
+                           absl::StatusOr<std::string> string_response =
+                               std::move(response->body);
+                           std::move(callback)(std::move(string_response));
+                         } else {
+                           std::move(callback)(response.status());
+                         }
+                       });
+}
+
+void MultiCurlHttpFetcherAsync::FetchUrlWithMetadata(
+    const HTTPRequest& request, int timeout_ms,
+    OnDoneFetchUrlWithMetadata done_callback) {
+  ScheduleAsyncCurlRequest(
+      CreateCurlRequest(request, timeout_ms, keepalive_idle_sec_,
+                        keepalive_interval_sec_, std::move(done_callback)));
+}
+
+void MultiCurlHttpFetcherAsync::PutUrl(const HTTPRequest& http_request,
+                                       int timeout_ms,
+                                       OnDoneFetchUrl done_callback) {
+  absl::AnyInvocable<void(absl::StatusOr<HTTPResponse>)> on_done_with_metadata =
+      [callback = std::move(done_callback)](
+          absl::StatusOr<HTTPResponse> response) mutable {
+        if (response.ok()) {
+          absl::StatusOr<std::string> string_response =
+              std::move(response->body);
+          std::move(callback)(std::move(string_response));
+        } else {
+          std::move(callback)(response.status());
+        }
+      };
+  auto request = CreateCurlRequest(http_request, timeout_ms,
+                                   keepalive_idle_sec_, keepalive_interval_sec_,
+                                   std::move(on_done_with_metadata));
+
+  request->body =
+      std::make_unique<DataToUpload>(DataToUpload{http_request.body});
+  curl_easy_setopt(request->req_handle, CURLOPT_UPLOAD, 1L);
+  curl_easy_setopt(request->req_handle, CURLOPT_PUT, 1L);
+  curl_easy_setopt(request->req_handle, CURLOPT_POSTFIELDSIZE_LARGE,
+                   static_cast<curl_off_t>(http_request.body.size()));
+  curl_easy_setopt(request->req_handle, CURLOPT_READDATA, request->body.get());
+  curl_easy_setopt(request->req_handle, CURLOPT_READFUNCTION, ReadCallback);
+
+  ScheduleAsyncCurlRequest(std::move(request));
+}
+
+}  // namespace privacy_sandbox::server_common::clients