fix(rest): make RestCatalog safe to share across threads

src/iceberg/catalog/rest/http_client.cc

@@ -20,6 +20,9 @@
 #include "iceberg/catalog/rest/http_client.h"
 
 #include <map>
+#include <mutex>
+#include <utility>
+#include <vector>
 
 #include <cpr/cpr.h>
 
@@ -34,19 +37,25 @@ namespace iceberg::rest {
 
 class HttpResponse::Impl {
  public:
-  explicit Impl(cpr::Response&& response) : response_(std::move(response)) {}
+  explicit Impl(cpr::Response response)
+      : status_code_(static_cast<int32_t>(response.status_code)),
+        body_(std::move(response.text)),
+        headers_(response.header.begin(), response.header.end()) {}
   ~Impl() = default;
 
-  int32_t status_code() const { return static_cast<int32_t>(response_.status_code); }
+  static Result<HttpResponse> Make(cpr::Response response,
+                                   const ErrorHandler& error_handler);
 
-  std::string body() const { return response_.text; }
+  int32_t status_code() const { return status_code_; }
 
-  std::unordered_map<std::string, std::string> headers() const {
-    return {response_.header.begin(), response_.header.end()};
-  }
+  std::string body() const { return body_; }
+
+  std::unordered_map<std::string, std::string> headers() const { return headers_; }
 
  private:
-  cpr::Response response_;
+  int32_t status_code_;
+  std::string body_;
+  std::unordered_map<std::string, std::string> headers_;
 };
 
 HttpResponse::HttpResponse() = default;
@@ -190,6 +199,50 @@ Status HandleFailureResponse(const cpr::Response& response,
 
 }  // namespace
 
+Result<HttpResponse> HttpResponse::Impl::Make(cpr::Response response,
+                                              const ErrorHandler& error_handler) {
+  ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
+  HttpResponse http_response;
+  http_response.impl_ = std::make_unique<Impl>(std::move(response));
+  return http_response;
+}
+
+// libcurl does not support using a connection cache from several threads at once
+// (CURLSHOPT_SHARE(3)).
+class HttpClient::ConnectionPools {
+ public:
+  template <typename Send>
+  Result<HttpResponse> Run(const Send& send, const ErrorHandler& error_handler) {
+    std::unique_ptr<cpr::ConnectionPool> pool = Acquire();
+    // Make() consumes the CPR response, freeing its curl handle before pool reuse.
+    Result<HttpResponse> response = HttpResponse::Impl::Make(send(*pool), error_handler);
+    Release(std::move(pool));
+    return response;
+  }
+
+ private:
+  std::unique_ptr<cpr::ConnectionPool> Acquire() {
+    {
+      std::lock_guard lock(mutex_);
+      if (!idle_.empty()) {
+        auto pool = std::move(idle_.back());
+        idle_.pop_back();
+        return pool;
+      }
+    }
+    return std::make_unique<cpr::ConnectionPool>();
+  }
+
+  void Release(std::unique_ptr<cpr::ConnectionPool> pool) {
+    std::lock_guard lock(mutex_);
+    idle_.push_back(std::move(pool));
+  }
+
+  std::mutex mutex_;
+  // LIFO: the most recently released pool is the likeliest to have a live connection.
+  std::vector<std::unique_ptr<cpr::ConnectionPool>> idle_;
+};
+
 HttpClient::HttpClient(std::unordered_map<std::string, std::string> default_headers,
                        std::chrono::milliseconds connection_timeout,
                        std::chrono::milliseconds socket_timeout,
@@ -198,7 +251,7 @@ HttpClient::HttpClient(std::unordered_map<std::string, std::string> default_head
       default_headers_{std::move(default_headers)},
       connection_timeout_{connection_timeout},
       socket_timeout_{socket_timeout},
-      connection_pool_{std::make_unique<cpr::ConnectionPool>()} {
+      connection_pools_{std::make_unique<ConnectionPools>()} {
   // Set default Content-Type for all requests (including GET/HEAD/DELETE).
   // Many systems require that content type is set regardless, even on an empty body.
   default_headers_[kHeaderContentType] = kMimeTypeApplicationJson;
@@ -234,15 +287,14 @@ Result<HttpResponse> HttpClient::Get(
   ICEBERG_ASSIGN_OR_RAISE(auto authenticated,
                           AuthenticateRequest(session, HttpMethod::kGet, std::move(url),
                                               MergeHeaders(default_headers_, headers)));
-  cpr::Response response =
-      cpr::Get(cpr::Url{authenticated.url}, ToCprHeader(authenticated),
-               kNoFollowRedirects, *connection_pool_,
-               cpr::ConnectTimeout{connection_timeout_}, cpr::Timeout{socket_timeout_});
-
-  ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
-  HttpResponse http_response;
-  http_response.impl_ = std::make_unique<HttpResponse::Impl>(std::move(response));
-  return http_response;
+  return connection_pools_->Run(
+      [&](const cpr::ConnectionPool& pool) {
+        return cpr::Get(cpr::Url{authenticated.url}, ToCprHeader(authenticated),
+                        kNoFollowRedirects, pool,
+                        cpr::ConnectTimeout{connection_timeout_},
+                        cpr::Timeout{socket_timeout_});
+      },
+      error_handler);
 }
 
 Result<HttpResponse> HttpClient::Post(
@@ -254,15 +306,14 @@ Result<HttpResponse> HttpClient::Post(
       auto authenticated,
       AuthenticateRequest(session, HttpMethod::kPost, std::move(request_uri),
                           MergeHeaders(default_headers_, headers), body));
-  cpr::Response response =
-      cpr::Post(cpr::Url{authenticated.url}, cpr::Body{authenticated.body},
-                ToCprHeader(authenticated), kNoFollowRedirects, *connection_pool_,
-                cpr::ConnectTimeout{connection_timeout_}, cpr::Timeout{socket_timeout_});
-
-  ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
-  HttpResponse http_response;
-  http_response.impl_ = std::make_unique<HttpResponse::Impl>(std::move(response));
-  return http_response;
+  return connection_pools_->Run(
+      [&](const cpr::ConnectionPool& pool) {
+        return cpr::Post(cpr::Url{authenticated.url}, cpr::Body{authenticated.body},
+                         ToCprHeader(authenticated), kNoFollowRedirects, pool,
+                         cpr::ConnectTimeout{connection_timeout_},
+                         cpr::Timeout{socket_timeout_});
+      },
+      error_handler);
 }
 
 Result<HttpResponse> HttpClient::PostForm(
@@ -286,15 +337,14 @@ Result<HttpResponse> HttpClient::PostForm(
       AuthenticateRequest(session, HttpMethod::kPost, std::move(request_uri),
                           MergeHeaders(default_headers_, form_headers),
                           std::move(encoded_body)));
-  cpr::Response response =
-      cpr::Post(cpr::Url{authenticated.url}, cpr::Body{authenticated.body},
-                ToCprHeader(authenticated), kNoFollowRedirects, *connection_pool_,
-                cpr::ConnectTimeout{connection_timeout_}, cpr::Timeout{socket_timeout_});
-
-  ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
-  HttpResponse http_response;
-  http_response.impl_ = std::make_unique<HttpResponse::Impl>(std::move(response));
-  return http_response;
+  return connection_pools_->Run(
+      [&](const cpr::ConnectionPool& pool) {
+        return cpr::Post(cpr::Url{authenticated.url}, cpr::Body{authenticated.body},
+                         ToCprHeader(authenticated), kNoFollowRedirects, pool,
+                         cpr::ConnectTimeout{connection_timeout_},
+                         cpr::Timeout{socket_timeout_});
+      },
+      error_handler);
 }
 
 Result<HttpResponse> HttpClient::Head(
@@ -305,15 +355,14 @@ Result<HttpResponse> HttpClient::Head(
       auto authenticated,
       AuthenticateRequest(session, HttpMethod::kHead, std::move(request_uri),
                           MergeHeaders(default_headers_, headers)));
-  cpr::Response response =
-      cpr::Head(cpr::Url{authenticated.url}, ToCprHeader(authenticated),
-                kNoFollowRedirects, *connection_pool_,
-                cpr::ConnectTimeout{connection_timeout_}, cpr::Timeout{socket_timeout_});
-
-  ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
-  HttpResponse http_response;
-  http_response.impl_ = std::make_unique<HttpResponse::Impl>(std::move(response));
-  return http_response;
+  return connection_pools_->Run(
+      [&](const cpr::ConnectionPool& pool) {
+        return cpr::Head(cpr::Url{authenticated.url}, ToCprHeader(authenticated),
+                         kNoFollowRedirects, pool,
+                         cpr::ConnectTimeout{connection_timeout_},
+                         cpr::Timeout{socket_timeout_});
+      },
+      error_handler);
 }
 
 Result<HttpResponse> HttpClient::Delete(
@@ -326,15 +375,14 @@ Result<HttpResponse> HttpClient::Delete(
       auto authenticated,
       AuthenticateRequest(session, HttpMethod::kDelete, std::move(url),
                           MergeHeaders(default_headers_, headers)));
-  cpr::Response response = cpr::Delete(
-      cpr::Url{authenticated.url}, ToCprHeader(authenticated), kNoFollowRedirects,
-      *connection_pool_, cpr::ConnectTimeout{connection_timeout_},
-      cpr::Timeout{socket_timeout_});
-
-  ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
-  HttpResponse http_response;
-  http_response.impl_ = std::make_unique<HttpResponse::Impl>(std::move(response));
-  return http_response;
+  return connection_pools_->Run(
+      [&](const cpr::ConnectionPool& pool) {
+        return cpr::Delete(cpr::Url{authenticated.url}, ToCprHeader(authenticated),
+                           kNoFollowRedirects, pool,
+                           cpr::ConnectTimeout{connection_timeout_},
+                           cpr::Timeout{socket_timeout_});
+      },
+      error_handler);
 }
 
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/http_client.h

@@ -32,17 +32,9 @@
 /// \file iceberg/catalog/rest/http_client.h
 /// \brief Http client for Iceberg REST API.
 
-namespace cpr {
-class ConnectionPool;
-}  // namespace cpr
-
 namespace iceberg::rest {
 
-/// \brief A simple wrapper for cpr::Response.
-///
-/// This class encapsulates the details of the underlying cpr library's response,
-/// providing a consistent interface that is independent of the specific network
-/// library used.
+/// \brief HTTP response status, body, and headers.
 class ICEBERG_REST_EXPORT HttpResponse {
  public:
   HttpResponse();
@@ -70,6 +62,9 @@ class ICEBERG_REST_EXPORT HttpResponse {
 };
 
 /// \brief HTTP client for making requests to Iceberg REST Catalog API.
+///
+/// Thread-safe, provided the AuthSession passed to each request supports
+/// concurrent calls to Authenticate().
 class ICEBERG_REST_EXPORT HttpClient {
  public:
   /// \param connection_timeout,socket_timeout Zero leaves requests unbounded.
@@ -116,14 +111,16 @@ class ICEBERG_REST_EXPORT HttpClient {
       const ErrorHandler& error_handler, auth::AuthSession& session);
 
  private:
+  class ConnectionPools;
+
   /// \brief Build a request URI from this client's base URI and a path.
   Result<std::string> RequestUri(const std::string& path) const;
 
   std::optional<std::string> base_uri_;
   std::unordered_map<std::string, std::string> default_headers_;
   std::chrono::milliseconds connection_timeout_;
   std::chrono::milliseconds socket_timeout_;
-  std::unique_ptr<cpr::ConnectionPool> connection_pool_;
+  std::unique_ptr<ConnectionPools> connection_pools_;
 };
 
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/rest_catalog.cc

@@ -584,6 +584,7 @@ RestCatalog::RestCatalog(RestCatalogProperties config, std::shared_ptr<FileIO> f
 std::string_view RestCatalog::name() const { return name_; }
 
 Result<std::shared_ptr<Catalog>> RestCatalog::AsCatalog() {
+  std::lock_guard lock(default_catalog_mutex_);
   if (auto catalog = default_catalog_.lock()) {
     return catalog;
   }

src/iceberg/catalog/rest/rest_catalog.h

@@ -21,6 +21,7 @@
 
 #include <functional>
 #include <memory>
+#include <mutex>
 #include <optional>
 #include <string>
 #include <unordered_set>
@@ -43,6 +44,10 @@
 namespace iceberg::rest {
 
 /// \brief Session-aware REST catalog root.
+///
+/// Thread-safe, including the catalogs from AsCatalog() and WithContext(), as long as
+/// the configured HTTP client, FileIO, metrics reporters, executors, auth manager and
+/// its sessions are.
 class ICEBERG_REST_EXPORT RestCatalog final
     : public SessionCatalog,
       public std::enable_shared_from_this<RestCatalog> {
@@ -235,7 +240,8 @@ class ICEBERG_REST_EXPORT RestCatalog final
   SnapshotMode snapshot_mode_;
   std::optional<ScanPlanningMode> client_scan_planning_mode_;
   SessionContext default_context_;
-  std::weak_ptr<Catalog> default_catalog_;
+  std::mutex default_catalog_mutex_;
+  std::weak_ptr<Catalog> default_catalog_;  // Guarded by default_catalog_mutex_.
   std::shared_ptr<MetricsReporter> reporter_;
   Executor* metrics_executor_ = nullptr;
 };

src/iceberg/test/http_client_test.cc

@@ -26,6 +26,7 @@
 #include <cstddef>
 #include <cstdint>
 #include <format>
+#include <memory>
 #include <string>
 #include <string_view>
 #include <thread>
@@ -301,7 +302,6 @@ TEST(HttpClientTest, PostReportsTransferFailureAfterSuccessfulStatus) {
   ASSERT_NO_FATAL_FAILURE(Listen(listener));
   const auto url = std::format("http://127.0.0.1:{}/v1/commit", listener.port);
 
-  // Outlives the response, which keeps a curl handle in its connection pool.
   HttpClient client;
   auto session = auth::AuthSession::MakeDefault({});
   std::thread server([&] { ServeOnce(listener.fd, response_bytes, /*end=*/true); });
@@ -312,6 +312,40 @@ TEST(HttpClientTest, PostReportsTransferFailureAfterSuccessfulStatus) {
   EXPECT_THAT(response, HasErrorMessage("failed after status 200"));
 }
 
+TEST(HttpClientTest, ResponseOutlivesLaterRequestsAndClient) {
+  constexpr std::string_view kBody = R"({"defaults":{}})";
+  const auto response_bytes = std::format(
+      "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nX-Test: kept\r\n"
+      "Content-Length: {}\r\nConnection: close\r\n\r\n{}",
+      kBody.size(), kBody);
+  Listener listener;
+  ASSERT_NO_FATAL_FAILURE(Listen(listener));
+  const auto url = std::format("http://127.0.0.1:{}/v1/config", listener.port);
+
+  auto client = std::make_unique<HttpClient>();
+  auto session = auth::AuthSession::MakeDefault({});
+  std::thread server([&] {
+    ServeOnce(listener.fd, response_bytes, /*end=*/true);
+    ServeOnce(listener.fd, response_bytes, /*end=*/true);
+  });
+  auto first = client->Get(url, /*params=*/{}, /*headers=*/{},
+                           *DefaultErrorHandler::Instance(), *session);
+  auto second = client->Get(url, /*params=*/{}, /*headers=*/{},
+                            *DefaultErrorHandler::Instance(), *session);
+  server.join();
+  client.reset();
+
+  auto expect_intact = [&](const Result<HttpResponse>& response) {
+    ASSERT_THAT(response, IsOk());
+    EXPECT_EQ(response->status_code(), 200);
+    EXPECT_EQ(response->body(), kBody);
+    EXPECT_THAT(response->headers(),
+                ::testing::Contains(::testing::Pair("X-Test", "kept")));
+  };
+  expect_intact(first);
+  expect_intact(second);
+}
+
 #endif
 
 class CapturingAuthSession final : public auth::AuthSession {
@@ -340,6 +374,22 @@ TEST(HttpClientTest, BuildsRequestUriLikeJavaHttpRequest) {
                                   *DefaultErrorHandler::Instance(), session);
   EXPECT_THAT(absolute, IsError(ErrorKind::kInvalidArgument));
   EXPECT_EQ(session.request().url, "https://auth.example/token/");
+
+  // A different path per method, so a call that fails before authenticating can't
+  // pass by matching the URL left over from the previous call.
+  const auto& handler = *DefaultErrorHandler::Instance();
+  EXPECT_THAT(client.Get("v1/get", {}, {}, handler, session),
+              IsError(ErrorKind::kInvalidArgument));
+  EXPECT_EQ(session.request().url, "https://catalog.example/api/v1/get");
+  EXPECT_THAT(client.Post("v1/post", "{}", {}, handler, session),
+              IsError(ErrorKind::kInvalidArgument));
+  EXPECT_EQ(session.request().url, "https://catalog.example/api/v1/post");
+  EXPECT_THAT(client.Head("v1/head", {}, handler, session),
+              IsError(ErrorKind::kInvalidArgument));
+  EXPECT_EQ(session.request().url, "https://catalog.example/api/v1/head");
+  EXPECT_THAT(client.Delete("v1/delete", {}, {}, handler, session),
+              IsError(ErrorKind::kInvalidArgument));
+  EXPECT_EQ(session.request().url, "https://catalog.example/api/v1/delete");
 }
 
 // Redirect responses must be reported without replaying the request at the target.