From 4efe73323ca35de6ea6f9261fe7e66b79d07bd0e Mon Sep 17 00:00:00 2001 From: Sreesh Maheshwar Date: Fri, 25 Sep 2026 10:47:56 +0100 Subject: [PATCH 1/3] fix(rest): make HttpClient safe for concurrent requests HttpClient sent every request through one cpr::ConnectionPool, and libcurl does not support sharing a connection cache between threads. Overlapping requests, for example from token refresh or the metrics executor, could crash. Give each in-flight request its own pool and reuse idle pools. HttpResponse now copies the status, body and headers out of cpr::Response, whose curl handle must be freed before its pool is reused. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/iceberg/catalog/rest/http_client.cc | 164 +++++++++++------- src/iceberg/catalog/rest/http_client.h | 11 +- src/iceberg/test/http_client_test.cc | 38 +++- .../test/rest_catalog_integration_test.cc | 28 +++ 4 files changed, 177 insertions(+), 64 deletions(-) diff --git a/src/iceberg/catalog/rest/http_client.cc b/src/iceberg/catalog/rest/http_client.cc index 019f78209..f72668e6f 100644 --- a/src/iceberg/catalog/rest/http_client.cc +++ b/src/iceberg/catalog/rest/http_client.cc @@ -20,6 +20,9 @@ #include "iceberg/catalog/rest/http_client.h" #include +#include +#include +#include #include @@ -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(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(response_.status_code); } + static Result 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 headers() const { - return {response_.header.begin(), response_.header.end()}; - } + std::string body() const { return body_; } + + std::unordered_map headers() const { return headers_; } private: - cpr::Response response_; + int32_t status_code_; + std::string body_; + std::unordered_map headers_; }; HttpResponse::HttpResponse() = default; @@ -190,6 +199,52 @@ Status HandleFailureResponse(const cpr::Response& response, } // namespace +Result 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(std::move(response)); + return http_response; +} + +// libcurl does not support using a connection cache from several threads at once +// (CURLSHOPT_SHARE(3)), so each request takes a cpr::ConnectionPool that no other +// request is using, and puts it back afterwards for reuse. +class HttpClient::ConnectionPools { + public: + template + Result Run(const Send& send, const ErrorHandler& error_handler) { + std::unique_ptr pool = Acquire(); + // The cpr::Response owns a curl handle that stays attached to the pool until freed. + // Make() takes it by value, so it is freed before the pool is released. + Result response = HttpResponse::Impl::Make(send(*pool), error_handler); + Release(std::move(pool)); + return response; + } + + private: + std::unique_ptr Acquire() { + { + std::lock_guard lock(mutex_); + if (!idle_.empty()) { + auto pool = std::move(idle_.back()); + idle_.pop_back(); + return pool; + } + } + return std::make_unique(); + } + + void Release(std::unique_ptr 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> idle_; +}; + HttpClient::HttpClient(std::unordered_map default_headers, std::chrono::milliseconds connection_timeout, std::chrono::milliseconds socket_timeout, @@ -198,7 +253,7 @@ HttpClient::HttpClient(std::unordered_map default_head default_headers_{std::move(default_headers)}, connection_timeout_{connection_timeout}, socket_timeout_{socket_timeout}, - connection_pool_{std::make_unique()} { + connection_pools_{std::make_unique()} { // 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 +289,14 @@ Result 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(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 HttpClient::Post( @@ -254,15 +308,14 @@ Result 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(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 HttpClient::PostForm( @@ -286,34 +339,30 @@ Result 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(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 HttpClient::Head( const std::string& path, const std::unordered_map& headers, const ErrorHandler& error_handler, auth::AuthSession& session) { - ICEBERG_ASSIGN_OR_RAISE(auto request_uri, RequestUri(path)); - ICEBERG_ASSIGN_OR_RAISE( - 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(std::move(response)); - return http_response; + ICEBERG_ASSIGN_OR_RAISE(auto authenticated, + AuthenticateRequest(session, HttpMethod::kHead, path, + MergeHeaders(default_headers_, headers))); + 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 HttpClient::Delete( @@ -326,15 +375,14 @@ Result 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(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 diff --git a/src/iceberg/catalog/rest/http_client.h b/src/iceberg/catalog/rest/http_client.h index 9a2459387..c20e3bbde 100644 --- a/src/iceberg/catalog/rest/http_client.h +++ b/src/iceberg/catalog/rest/http_client.h @@ -32,10 +32,6 @@ /// \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. @@ -70,6 +66,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,6 +115,8 @@ 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 RequestUri(const std::string& path) const; @@ -123,7 +124,7 @@ class ICEBERG_REST_EXPORT HttpClient { std::unordered_map default_headers_; std::chrono::milliseconds connection_timeout_; std::chrono::milliseconds socket_timeout_; - std::unique_ptr connection_pool_; + std::unique_ptr connection_pools_; }; } // namespace iceberg::rest diff --git a/src/iceberg/test/http_client_test.cc b/src/iceberg/test/http_client_test.cc index 6cc653131..ca0dbba2f 100644 --- a/src/iceberg/test/http_client_test.cc +++ b/src/iceberg/test/http_client_test.cc @@ -26,6 +26,7 @@ #include #include #include +#include #include #include #include @@ -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,42 @@ TEST(HttpClientTest, PostReportsTransferFailureAfterSuccessfulStatus) { EXPECT_THAT(response, HasErrorMessage("failed after status 200")); } +// A response owns its data, so it stays readable after a later request reuses +// its connection pool and after the client is destroyed. +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(); + 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& 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 { diff --git a/src/iceberg/test/rest_catalog_integration_test.cc b/src/iceberg/test/rest_catalog_integration_test.cc index 0e3ba6a3d..187aa738e 100644 --- a/src/iceberg/test/rest_catalog_integration_test.cc +++ b/src/iceberg/test/rest_catalog_integration_test.cc @@ -26,6 +26,7 @@ #include #include #include +#include #include #include @@ -425,6 +426,33 @@ TEST_F(RestCatalogIntegrationTest, FetchServerConfigDirect) { } } +// RestCatalog uses one HttpClient from several threads (e.g. token refresh, metrics). +// Without the fix this crashes on some libcurl versions, such as 8.22.0. +TEST_F(RestCatalogIntegrationTest, ConcurrentHttpClientRequests) { + constexpr int kThreads = 32; + constexpr int kRequestsPerThread = 200; + + // Finite timeouts, so a stalled fixture fails the test instead of hanging it. + HttpClient client({}, std::chrono::seconds(30), std::chrono::seconds(30)); + auto noop_session = auth::AuthSession::MakeDefault({}); + const std::string config_url = std::format("{}/v1/config", CatalogUri()); + + std::vector threads; + threads.reserve(kThreads); + for (int i = 0; i < kThreads; ++i) { + threads.emplace_back([&] { + for (int j = 0; j < kRequestsPerThread; ++j) { + ASSERT_THAT(client.Get(config_url, {}, /*headers=*/{}, + *DefaultErrorHandler::Instance(), *noop_session), + IsOk()); + } + }); + } + for (auto& thread : threads) { + thread.join(); + } +} + // -- Namespace operations -- TEST_F(RestCatalogIntegrationTest, ListNamespaces) { From 64d859c045cb77484ac69c4fa8942b747289599c Mon Sep 17 00:00:00 2001 From: Sreesh Maheshwar Date: Sun, 18 Oct 2026 21:29:43 +0100 Subject: [PATCH 2/3] docs(rest): trim HTTP client comments --- src/iceberg/catalog/rest/http_client.cc | 6 ++---- src/iceberg/catalog/rest/http_client.h | 6 +----- src/iceberg/test/http_client_test.cc | 2 -- src/iceberg/test/rest_catalog_integration_test.cc | 4 +--- 4 files changed, 4 insertions(+), 14 deletions(-) diff --git a/src/iceberg/catalog/rest/http_client.cc b/src/iceberg/catalog/rest/http_client.cc index f72668e6f..4183d253e 100644 --- a/src/iceberg/catalog/rest/http_client.cc +++ b/src/iceberg/catalog/rest/http_client.cc @@ -208,15 +208,13 @@ Result HttpResponse::Impl::Make(cpr::Response response, } // libcurl does not support using a connection cache from several threads at once -// (CURLSHOPT_SHARE(3)), so each request takes a cpr::ConnectionPool that no other -// request is using, and puts it back afterwards for reuse. +// (CURLSHOPT_SHARE(3)). class HttpClient::ConnectionPools { public: template Result Run(const Send& send, const ErrorHandler& error_handler) { std::unique_ptr pool = Acquire(); - // The cpr::Response owns a curl handle that stays attached to the pool until freed. - // Make() takes it by value, so it is freed before the pool is released. + // Make() consumes the CPR response, freeing its curl handle before pool reuse. Result response = HttpResponse::Impl::Make(send(*pool), error_handler); Release(std::move(pool)); return response; diff --git a/src/iceberg/catalog/rest/http_client.h b/src/iceberg/catalog/rest/http_client.h index c20e3bbde..bc7e60a2a 100644 --- a/src/iceberg/catalog/rest/http_client.h +++ b/src/iceberg/catalog/rest/http_client.h @@ -34,11 +34,7 @@ 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(); diff --git a/src/iceberg/test/http_client_test.cc b/src/iceberg/test/http_client_test.cc index ca0dbba2f..3a170adae 100644 --- a/src/iceberg/test/http_client_test.cc +++ b/src/iceberg/test/http_client_test.cc @@ -312,8 +312,6 @@ TEST(HttpClientTest, PostReportsTransferFailureAfterSuccessfulStatus) { EXPECT_THAT(response, HasErrorMessage("failed after status 200")); } -// A response owns its data, so it stays readable after a later request reuses -// its connection pool and after the client is destroyed. TEST(HttpClientTest, ResponseOutlivesLaterRequestsAndClient) { constexpr std::string_view kBody = R"({"defaults":{}})"; const auto response_bytes = std::format( diff --git a/src/iceberg/test/rest_catalog_integration_test.cc b/src/iceberg/test/rest_catalog_integration_test.cc index 187aa738e..a13d2dbd3 100644 --- a/src/iceberg/test/rest_catalog_integration_test.cc +++ b/src/iceberg/test/rest_catalog_integration_test.cc @@ -426,13 +426,11 @@ TEST_F(RestCatalogIntegrationTest, FetchServerConfigDirect) { } } -// RestCatalog uses one HttpClient from several threads (e.g. token refresh, metrics). -// Without the fix this crashes on some libcurl versions, such as 8.22.0. +// RestCatalog shares its HttpClient with token refresh and metrics reporting. TEST_F(RestCatalogIntegrationTest, ConcurrentHttpClientRequests) { constexpr int kThreads = 32; constexpr int kRequestsPerThread = 200; - // Finite timeouts, so a stalled fixture fails the test instead of hanging it. HttpClient client({}, std::chrono::seconds(30), std::chrono::seconds(30)); auto noop_session = auth::AuthSession::MakeDefault({}); const std::string config_url = std::format("{}/v1/config", CatalogUri()); From 1f820cc5618fcc8ea1e05dc5df3ab68e45d8092e Mon Sep 17 00:00:00 2001 From: Sreesh Maheshwar Date: Thu, 5 Nov 2026 17:28:41 +0000 Subject: [PATCH 3/3] fix(rest): resolve Head() paths against the base URI Head() lost its RequestUri() call in a rebase, so relative paths were sent without the base URI. Restore it, and check that every request method resolves relative paths. Co-Authored-By: Claude Opus 5.5 --- src/iceberg/catalog/rest/http_client.cc | 8 +++++--- src/iceberg/test/http_client_test.cc | 16 ++++++++++++++++ src/iceberg/test/rest_table_scan_test.cc | 3 +-- 3 files changed, 22 insertions(+), 5 deletions(-) diff --git a/src/iceberg/catalog/rest/http_client.cc b/src/iceberg/catalog/rest/http_client.cc index 4183d253e..48bef3c75 100644 --- a/src/iceberg/catalog/rest/http_client.cc +++ b/src/iceberg/catalog/rest/http_client.cc @@ -350,9 +350,11 @@ Result HttpClient::PostForm( Result HttpClient::Head( const std::string& path, const std::unordered_map& headers, const ErrorHandler& error_handler, auth::AuthSession& session) { - ICEBERG_ASSIGN_OR_RAISE(auto authenticated, - AuthenticateRequest(session, HttpMethod::kHead, path, - MergeHeaders(default_headers_, headers))); + ICEBERG_ASSIGN_OR_RAISE(auto request_uri, RequestUri(path)); + ICEBERG_ASSIGN_OR_RAISE( + auto authenticated, + AuthenticateRequest(session, HttpMethod::kHead, std::move(request_uri), + MergeHeaders(default_headers_, headers))); return connection_pools_->Run( [&](const cpr::ConnectionPool& pool) { return cpr::Head(cpr::Url{authenticated.url}, ToCprHeader(authenticated), diff --git a/src/iceberg/test/http_client_test.cc b/src/iceberg/test/http_client_test.cc index 3a170adae..37b2ed85e 100644 --- a/src/iceberg/test/http_client_test.cc +++ b/src/iceberg/test/http_client_test.cc @@ -374,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. diff --git a/src/iceberg/test/rest_table_scan_test.cc b/src/iceberg/test/rest_table_scan_test.cc index 7610f8249..4877ac255 100644 --- a/src/iceberg/test/rest_table_scan_test.cc +++ b/src/iceberg/test/rest_table_scan_test.cc @@ -105,8 +105,7 @@ MATCHER_P(JsonBodyLacks, key, "") { // -------------------------------------------------------------------------- // Mock HTTP client that overrides the virtual methods of HttpClient. -// The base class constructor creates a cpr::ConnectionPool, which is a -// lightweight allocation (no network connections are opened at construction). +// The base class constructor opens no network connections. // -------------------------------------------------------------------------- class MockHttpClient : public HttpClient { public: