From 2e15601ecaf1f5119ede1db9bfef11f357b140ae Mon Sep 17 00:00:00 2001 From: Zhao Junwang Date: Sun, 13 Sep 2026 22:22:59 +0800 Subject: [PATCH 1/5] feat(rest): align HTTP retries and timeouts with Java Configure REST transport through rest.client.* properties for connection and socket timeouts, retry limits, user-agent and idempotency-key lifetime. Apply initial and merged catalog properties through the client factory, preserving custom factories and the existing timeout constructor. Use Java's three-minute default connection timeout and zero-value timeout semantics. Enforce response inactivity deadlines in milliseconds with a synchronous libcurl transfer loop, without an additional executor. Authenticate once and replay the same request across retries. Decide HTTP status retries after headers and before reading the entity. Honor Idempotency-Key for retry-safe requests and bound keyed retries by the first-send time and maximum next interval, including Retry-After or jitter. Parse key lifetimes as ISO-8601 durations with Java-compatible handling of invalid and non-positive values. Retry eligible request-execution I/O failures without a delay. Propagate final response-body failures without replaying the request. Preserve redirect protection, transfer-error reporting and commit-state-unknown handling. Saturate oversized Retry-After conversions without overflow and avoid echoing malformed OAuth token responses. Document the transport policy and add regression coverage for retry rules, key lifetime budgets, stable authentication, timeout precision and duration boundaries. --- mkdocs/docs/rest-catalog.md | 79 ++ mkdocs/mkdocs.yml | 1 + src/iceberg/catalog/rest/auth/oauth2_util.cc | 9 +- src/iceberg/catalog/rest/catalog_properties.h | 2 +- src/iceberg/catalog/rest/http_client.cc | 728 ++++++++++++++++-- src/iceberg/catalog/rest/http_client.h | 55 +- .../catalog/rest/http_client_internal.h | 67 ++ src/iceberg/catalog/rest/rest_catalog.cc | 12 +- src/iceberg/test/http_client_test.cc | 633 ++++++++++++++- 9 files changed, 1460 insertions(+), 126 deletions(-) create mode 100644 mkdocs/docs/rest-catalog.md create mode 100644 src/iceberg/catalog/rest/http_client_internal.h diff --git a/mkdocs/docs/rest-catalog.md b/mkdocs/docs/rest-catalog.md new file mode 100644 index 0000000000..eadf1f29c6 --- /dev/null +++ b/mkdocs/docs/rest-catalog.md @@ -0,0 +1,79 @@ + + +# REST catalog transport + +The transport and retry properties follow the Java `HTTPClient` and +`ExponentialHttpRequestRetryStrategy`. They apply to configuration discovery, +OAuth token requests, catalog operations, and storage credential refresh requests. +Initial requests use client properties; subsequent requests use merged catalog +properties. + +| Property | Default | Notes | +| --- | --- | --- | +| `rest.client.connection-timeout-ms` | 180000 | Non-negative integer; zero disables the configured deadline | +| `rest.client.socket-timeout-ms` | Unbounded | Non-negative integer; zero disables the response inactivity timeout | +| `rest.client.max-retries` | 5 | Positive integer; excludes the first attempt | +| `rest.client.user-agent` | `iceberg-cpp/` | Used when the request has no explicit `User-Agent` header | +| `rest.client.idempotency-key-lifetime` | Unbounded | ISO-8601 duration, for example `PT30M`; zero or negative durations disable keyed retries; malformed or out-of-range values are ignored | + +The socket timeout measures response inactivity rather than average transfer +speed or total request duration. Connection setup and writing a request body +use their own transport behavior. Transfers run synchronously in the calling +thread, polling the remaining read deadline in milliseconds; no executor is +required for HTTP retries. + +HTTP response retries are decided after receiving the headers, before reading +the response body: + +- HTTP 429, and HTTP 503 carrying `Retry-After`, permit retries for any method. +- GET, HEAD and DELETE, and requests carrying an `Idempotency-Key` header, also + permit retries on HTTP 408, 500, 502, 503 and 504. +- Positive numeric and HTTP-date `Retry-After` values are honored. Otherwise the + wait doubles from 1 second up to 64 seconds, with up to 10% jitter. Invalid or + non-positive values use this backoff; numeric durations beyond the millisecond + range saturate at the maximum representable duration. + +Recoverable I/O failures before response headers are received permit retries +for idempotent methods or requests carrying `Idempotency-Key`. As in Java's +I/O-exception retry path, these retries have no delay. Name resolution, +connection, TLS, timeout, cancellation and premature-response-closure failures +are not retried. Failures while reading a final response body are propagated, +including when its HTTP status was successful. + +For keyed requests, the lifetime starts at the first send. A retry is abandoned +when elapsed time plus the maximum next retry interval reaches the configured +lifetime. The bound includes `Retry-After` or the largest possible HTTP backoff +jitter. As in Java, the I/O-error lifetime check uses the exponential interval +even though the exception retry itself has no delay. Unkeyed requests are not +constrained by this property. + +Authentication runs once before the first send. Retries replay the same +headers, body and idempotency key. A subsequent logical request authenticates +again and can use an updated token. + +Errors are classified from the `code` reported in the error body, falling back +to the HTTP status when the body cannot be parsed. Service diagnostics remain +available. Malformed OAuth token responses never echo response content. + +Other Java transport options, including `rest.client.max-connections`, +`rest.client.connections-per-route`, `rest.client.proxy.*` and +`rest.client.tls.configurer-impl`, are not yet supported by this transport. + +For temporary storage credentials and their lifetime, see [FileIO](file-io.md#refreshing-temporary-s3-credentials). diff --git a/mkdocs/mkdocs.yml b/mkdocs/mkdocs.yml index 7cf0ec152e..9ccbcc92b4 100644 --- a/mkdocs/mkdocs.yml +++ b/mkdocs/mkdocs.yml @@ -61,6 +61,7 @@ nav: - Verify a Release Candidate: verify-rc.md - API Documentation: api/index.html - SQL Catalog: sql-catalog.md + - REST Catalog: rest-catalog.md extra: social: diff --git a/src/iceberg/catalog/rest/auth/oauth2_util.cc b/src/iceberg/catalog/rest/auth/oauth2_util.cc index d62ac7e1a0..9de28eaa06 100644 --- a/src/iceberg/catalog/rest/auth/oauth2_util.cc +++ b/src/iceberg/catalog/rest/auth/oauth2_util.cc @@ -49,9 +49,12 @@ constexpr std::string_view kAuthorizationHeader = "Authorization"; constexpr std::string_view kBearerPrefix = "Bearer "; Result ParseTokenResponse(const std::string& response_body) { - ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response_body)); - ICEBERG_ASSIGN_OR_RAISE(auto token_response, FromJson(json)); - ICEBERG_RETURN_UNEXPECTED(token_response.Validate()); + // Never propagate the underlying JSON errors: they echo the response body, + // which may contain token material. FromJson already validates the response. + auto json = FromJsonString(response_body); + if (!json) return InvalidArgument("Invalid OAuth2 token response"); + auto token_response = FromJson(*json); + if (!token_response) return InvalidArgument("Invalid OAuth2 token response"); return token_response; } diff --git a/src/iceberg/catalog/rest/catalog_properties.h b/src/iceberg/catalog/rest/catalog_properties.h index 0b24af58ca..f5e7d629ed 100644 --- a/src/iceberg/catalog/rest/catalog_properties.h +++ b/src/iceberg/catalog/rest/catalog_properties.h @@ -74,7 +74,7 @@ class ICEBERG_REST_EXPORT RestCatalogProperties inline static constexpr std::string_view kHeaderPrefix = "header."; /// \brief HTTP connection timeout in milliseconds, matching Java REST clients. inline static Entry kClientConnectionTimeoutMs{ - "rest.client.connection-timeout-ms", 0}; + "rest.client.connection-timeout-ms", 180000}; /// \brief HTTP socket/read timeout in milliseconds, matching Java REST clients. inline static Entry kClientSocketTimeoutMs{"rest.client.socket-timeout-ms", 0}; diff --git a/src/iceberg/catalog/rest/http_client.cc b/src/iceberg/catalog/rest/http_client.cc index 019f78209d..b6de8949d9 100644 --- a/src/iceberg/catalog/rest/http_client.cc +++ b/src/iceberg/catalog/rest/http_client.cc @@ -19,19 +19,177 @@ #include "iceberg/catalog/rest/http_client.h" +#include +#include +#include +#include +#include #include +#include +#include +#include +#include +#include #include +#include #include "iceberg/catalog/rest/auth/auth_session.h" #include "iceberg/catalog/rest/constant.h" #include "iceberg/catalog/rest/error_handlers.h" +#include "iceberg/catalog/rest/http_client_internal.h" #include "iceberg/catalog/rest/rest_util.h" +#include "iceberg/logging/log_macros.h" #include "iceberg/result.h" #include "iceberg/util/macros.h" namespace iceberg::rest { +namespace { + +// Property names follow the Java HTTPClient. +constexpr std::string_view kRestMaxRetries = "rest.client.max-retries"; +constexpr std::string_view kRestConnectionTimeoutMs = "rest.client.connection-timeout-ms"; +constexpr std::string_view kRestSocketTimeoutMs = "rest.client.socket-timeout-ms"; +constexpr std::string_view kRestUserAgent = "rest.client.user-agent"; +constexpr std::string_view kRestKeyLifetime = "rest.client.idempotency-key-lifetime"; + +// Parse an integer property. Error messages name the property but never +// echo its value, since server-provided configuration may reach these options. +template +Status ParseIntegerProperty( + const std::unordered_map& properties, std::string_view name, + std::optional& target, Int minimum = 0) { + auto it = properties.find(std::string(name)); + if (it == properties.end()) return {}; + std::string_view value = it->second; + // Integer.parseInt/parseLong accept an explicit leading plus sign. + if (value.starts_with('+')) { + value.remove_prefix(1); + if (value.starts_with('-')) { + return InvalidArgument("Invalid integer HTTP client property '{}'", name); + } + } + Int parsed{}; + auto [end, ec] = std::from_chars(value.data(), value.data() + value.size(), parsed); + if (ec != std::errc{} || end != value.data() + value.size() || parsed < minimum) { + return InvalidArgument("Invalid HTTP client property '{}': must be {} integer", name, + minimum == 0 ? "a non-negative" : "a positive"); + } + target = parsed; + return {}; +} + +bool CheckedAdd(int64_t left, int64_t right, int64_t& result) { + if ((right > 0 && left > std::numeric_limits::max() - right) || + (right < 0 && left < std::numeric_limits::min() - right)) { + return false; + } + result = left + right; + return true; +} + +bool CheckedSubtract(int64_t left, int64_t right, int64_t& result) { + if ((right > 0 && left < std::numeric_limits::min() + right) || + (right < 0 && left > std::numeric_limits::max() + right)) { + return false; + } + result = left - right; + return true; +} + +// Match Java Duration.parse's day/time syntax, including signed components and +// fractional seconds. Keep seconds and nanoseconds separate to avoid narrowing +// an otherwise valid duration before checking the millisecond bound. +std::optional ParseKeyLifetime(const std::string& value) { + static const std::regex pattern( + R"(([+-]?)P(?:([+-]?[0-9]+)D)?(?:T(?:([+-]?[0-9]+)H)?(?:([+-]?[0-9]+)M)?(?:([+-]?[0-9]+)(?:[.,]([0-9]{0,9}))?S)?)?)", + std::regex::icase); + std::smatch match; + if (!std::regex_match(value, match, pattern)) return std::nullopt; + const bool has_time = match[3].matched || match[4].matched || match[5].matched; + if ((!match[2].matched && !has_time) || + ((value.find('T') != std::string::npos || value.find('t') != std::string::npos) && + !has_time)) { + return std::nullopt; + } + constexpr int64_t multipliers[] = {86400, 3600, 60, 1}; + int64_t components[4]{}; + for (size_t i = 0; i < 4; ++i) { + if (!match[i + 2].matched) continue; + const auto text = match[i + 2].str(); + auto view = std::string_view(text); + if (view.starts_with('+')) view.remove_prefix(1); + int64_t number{}; + auto [end, ec] = std::from_chars(view.data(), view.data() + view.size(), number); + if (ec != std::errc{} || end != view.data() + view.size() || + number > std::numeric_limits::max() / multipliers[i] || + number < std::numeric_limits::min() / multipliers[i]) { + return std::nullopt; + } + components[i] = number * multipliers[i]; + } + int64_t seconds = components[3]; + for (int i = 2; i >= 0; --i) { + if (!CheckedAdd(seconds, components[i], seconds)) return std::nullopt; + } + int64_t nanos = 0; + const auto fraction = match[6].str(); + for (char digit : fraction) nanos = nanos * 10 + (digit - '0'); + if (!fraction.empty()) { + for (size_t i = fraction.size(); i < 9; ++i) nanos *= 10; + if (match[5].str().starts_with('-')) nanos = -nanos; + } + if (nanos < 0) { + if (!CheckedAdd(seconds, -1, seconds)) return std::nullopt; + nanos += 1000000000; + } + if (match[1].str() == "-") { + if (nanos == 0) { + if (seconds == std::numeric_limits::min()) return std::nullopt; + seconds = -seconds; + } else { + seconds = seconds >= 0 ? -seconds - 1 : -(seconds + 1); + nanos = 1000000000 - nanos; + } + } + if (seconds < 0 || (seconds == 0 && nanos == 0)) { + return std::chrono::milliseconds::zero(); + } + constexpr auto kMaxMillis = std::chrono::milliseconds::max().count(); + if (seconds > kMaxMillis / 1000 || + (seconds == kMaxMillis / 1000 && nanos > (kMaxMillis % 1000) * 1000000)) { + return std::nullopt; + } + return std::chrono::milliseconds(seconds * 1000 + nanos / 1000000); +} + +} // namespace + +Result HttpClientOptions::FromProperties( + const std::unordered_map& properties) { + HttpClientOptions options; + ICEBERG_RETURN_UNEXPECTED(ParseIntegerProperty(properties, kRestConnectionTimeoutMs, + options.connection_timeout_ms)); + ICEBERG_RETURN_UNEXPECTED( + ParseIntegerProperty(properties, kRestSocketTimeoutMs, options.socket_timeout_ms)); + std::optional max_retries; + ICEBERG_RETURN_UNEXPECTED( + ParseIntegerProperty(properties, kRestMaxRetries, max_retries, 1)); + if (max_retries.has_value()) options.max_retries = *max_retries; + if (auto it = properties.find(std::string(kRestUserAgent)); it != properties.end()) { + options.user_agent = it->second; + } + if (auto it = properties.find(std::string(kRestKeyLifetime)); it != properties.end()) { + options.idempotency_key_lifetime = ParseKeyLifetime(it->second); + if (!options.idempotency_key_lifetime.has_value()) { + ICEBERG_LOG_WARN("Ignoring malformed or out-of-range HTTP client property '{}'", + kRestKeyLifetime); + } + } + return options; +} + class HttpResponse::Impl { public: explicit Impl(cpr::Response&& response) : response_(std::move(response)) {} @@ -50,7 +208,6 @@ class HttpResponse::Impl { }; HttpResponse::HttpResponse() = default; - HttpResponse::HttpResponse(int32_t status_code, std::string body) { cpr::Response cpr_response; cpr_response.status_code = status_code; @@ -72,6 +229,312 @@ std::unordered_map HttpResponse::headers() const { namespace { +// The retry rules below mirror +// org.apache.iceberg.rest.ExponentialHttpRequestRetryStrategy. + +constexpr std::string_view kHeaderRetryAfter = "Retry-After"; + +/// \brief HTTP methods whose repetition is safe (RFC 9110 idempotent methods). +bool IsIdempotent(HttpMethod method) { + switch (method) { + case HttpMethod::kGet: + case HttpMethod::kHead: + case HttpMethod::kDelete: + case HttpMethod::kPut: + return true; + default: + return false; + } +} + +/// \brief Request-execution I/O failures eligible for Java-style retries. +/// An empty reply corresponds to NoHttpResponseException. Premature entity +/// closure, malformed responses, DNS, connect, TLS and timeout errors are excluded. +bool IsRetriableTransportError(cpr::ErrorCode code) { + switch (code) { + case cpr::ErrorCode::GOT_NOTHING: + case cpr::ErrorCode::SEND_ERROR: + case cpr::ErrorCode::RECV_ERROR: + case cpr::ErrorCode::HTTP2: + case cpr::ErrorCode::HTTP2_STREAM: + return true; + default: + return false; + } +} + +} // namespace + +namespace { + +std::chrono::milliseconds Backoff(int64_t exec_count) { + if (exec_count <= 0) return std::chrono::milliseconds::zero(); + return std::chrono::milliseconds( + 1000 * (exec_count > 7 ? int64_t{64} : int64_t{1} << (exec_count - 1))); +} + +std::optional RetryAfter( + const cpr::Response& response, std::chrono::system_clock::time_point now) { + auto it = response.header.find(std::string(kHeaderRetryAfter)); + if (it == response.header.end()) return std::nullopt; + auto value = std::string_view(it->second); + if (value.starts_with('+')) value.remove_prefix(1); + int64_t seconds = 0; + auto [end, ec] = std::from_chars(value.data(), value.data() + value.size(), seconds); + if (ec == std::errc{} && end == value.data() + value.size()) { + if (seconds <= 0) return std::nullopt; + constexpr auto kMaxSeconds = std::chrono::milliseconds::max().count() / 1000; + if (seconds > kMaxSeconds) return std::chrono::milliseconds::max(); + return std::chrono::milliseconds(seconds * 1000); + } + const auto date = curl_getdate(it->second.c_str(), nullptr); + if (date < 0) return std::nullopt; + const auto now_ms = + std::chrono::duration_cast(now.time_since_epoch()) + .count(); + const auto date_seconds = static_cast(date); + if (date_seconds > std::chrono::milliseconds::max().count() / 1000) { + return std::chrono::milliseconds::max(); + } + const auto date_ms = date_seconds * 1000; + if (date_ms <= now_ms) return std::nullopt; + int64_t delay; + if (!CheckedSubtract(date_ms, now_ms, delay)) return std::chrono::milliseconds::max(); + return std::chrono::milliseconds(delay); +} + +} // namespace + +namespace detail { + +bool ShouldRetry(const cpr::Response& response, const HttpRequest& request, + int64_t exec_count, const HttpClientOptions& options, + std::optional elapsed) { + if (exec_count > options.max_retries) return false; + const bool has_key = request.headers.contains("Idempotency-Key"); + const bool safe = IsIdempotent(request.method) || has_key; + if (response.error) { + // Java reads the final response entity outside HttpRequestRetryExec. A body + // read failure is propagated, even for a successful status or keyed request. + if (response.status_code != 0 || !safe || + !IsRetriableTransportError(response.error.code)) + return false; + } else { + const auto code = response.status_code; + if (code != 429 && + !(code == 503 && response.header.contains(std::string(kHeaderRetryAfter))) && + !(safe && + (code == 408 || code == 500 || code == 502 || code == 503 || code == 504))) + return false; + } + if (has_key && options.idempotency_key_lifetime.has_value() && elapsed.has_value()) { + const auto lifetime = options.idempotency_key_lifetime->count(); + if (elapsed->count() >= lifetime) return false; + int64_t remaining; + if (!CheckedSubtract(lifetime, elapsed->count(), remaining)) { + remaining = std::numeric_limits::max(); + } + auto upper_delay = Backoff(exec_count); + if (!response.error) { + if (auto retry_after = RetryAfter(response, std::chrono::system_clock::now())) { + upper_delay = *retry_after; + } else { + upper_delay += upper_delay / 10; + } + } + if (upper_delay.count() >= remaining) return false; + } + return true; +} + +std::chrono::milliseconds RetryInterval(const cpr::Response& response, int64_t exec_count, + std::chrono::system_clock::time_point now) { + // HttpRequestRetryStrategy's I/O-exception overload defaults to no delay. + if (response.error) return std::chrono::milliseconds::zero(); + if (auto retry_after = RetryAfter(response, now)) return *retry_after; + const auto delay = Backoff(exec_count); + thread_local std::mt19937 generator{std::random_device{}()}; + std::uniform_int_distribution jitter( + 0, std::max(1, delay.count() / 10) - 1); + return delay + std::chrono::milliseconds(jitter(generator)); +} + +bool SocketIdleTimeout::Check(int64_t downloaded, int64_t uploaded, int64_t header_bytes, + Clock::time_point now) { + if (!last_progress_.has_value() || downloaded != downloaded_ || uploaded != uploaded_ || + header_bytes != header_bytes_) { + last_progress_ = now; + downloaded_ = downloaded; + uploaded_ = uploaded; + header_bytes_ = header_bytes; + } + return now - *last_progress_ < timeout_; +} + +std::optional SocketIdleTimeout::Remaining( + Clock::time_point now) const { + if (!last_progress_.has_value()) return std::nullopt; + return std::max( + std::chrono::milliseconds::zero(), + std::chrono::ceil(timeout_ - (now - *last_progress_))); +} + +} // namespace detail + +namespace { + +struct Attempt { + cpr::Response response; + bool retry_approved = false; +}; + +/// Retry the same authenticated request, matching Java's retry execution chain. +template +Result SendWithRetry(const HttpClientOptions& options, + const HttpRequest& request, Send send) { + const auto first_attempt = std::chrono::time_point_cast( + std::chrono::system_clock::now()); + for (int64_t exec_count = 1;; ++exec_count) { + ICEBERG_ASSIGN_OR_RAISE(auto attempt, send(exec_count, first_attempt)); + const auto elapsed = std::chrono::duration_cast( + std::chrono::time_point_cast( + std::chrono::system_clock::now()) - + first_attempt); + if (!attempt.retry_approved && + !detail::ShouldRetry(attempt.response, request, exec_count, options, elapsed)) { + return std::move(attempt.response); + } + std::this_thread::sleep_for(detail::RetryInterval(attempt.response, exec_count)); + } +} + +void ApplyOptions(cpr::Session& session, const HttpClientOptions& options) { + const auto timeout = options.connection_timeout_ms.value_or(180000); + // libcurl interprets zero as its own five-minute default. Use its largest + // native deadline when Java's disabled connection timeout is requested. + session.SetConnectTimeout(cpr::ConnectTimeout{std::chrono::milliseconds( + timeout == 0 ? std::numeric_limits::max() + : std::min(timeout, std::numeric_limits::max()))}); +} + +// Drive transfers synchronously through libcurl's socket API. Observe socket +// readability before libcurl consumes a fragment: header callbacks only report +// complete lines, while the read deadline must include partial header progress. +template +Result PerformSession(cpr::Session& session, + detail::SocketIdleTimeout* idle_timeout, + CheckIdle check_idle) { + auto* curl = session.GetCurlHolder()->handle; + using SocketInterests = std::map; + // Keep callback data alive until remove_handle/cleanup finish invoking it. + SocketInterests sockets; + const auto cleanup = [curl](CURLM* multi) { + curl_multi_remove_handle(multi, curl); + curl_multi_cleanup(multi); + }; + std::unique_ptr multi(curl_multi_init(), cleanup); + if (!multi) return IOError("Failed to initialize HTTP transfer"); + const auto socket_callback = + +[](CURL*, curl_socket_t socket, int action, void* data, void*) -> int { + auto& interests = *static_cast(data); + try { + if (action == CURL_POLL_REMOVE) { + interests.erase(socket); + } else { + interests.insert_or_assign(socket, action); + } + } catch (...) { + // Do not let an allocation failure escape through libcurl's C callback. + return -1; + } + return 0; + }; + if (curl_multi_setopt(multi.get(), CURLMOPT_SOCKETFUNCTION, socket_callback) != + CURLM_OK || + curl_multi_setopt(multi.get(), CURLMOPT_SOCKETDATA, &sockets) != CURLM_OK) { + return IOError("Failed to initialize HTTP socket monitoring"); + } + if (const auto result = curl_multi_add_handle(multi.get(), curl); result != CURLM_OK) { + return IOError("Failed to prepare HTTP transfer: {}", curl_multi_strerror(result)); + } + int running = 0; + auto result = curl_multi_socket_action(multi.get(), CURL_SOCKET_TIMEOUT, 0, &running); + CURLcode transfer_result = CURLE_OK; + while (true) { + if (result != CURLM_OK) { + return IOError("HTTP transfer failed: {}", curl_multi_strerror(result)); + } + if (running == 0) { + int remaining = 0; + const auto* message = curl_multi_info_read(multi.get(), &remaining); + if (message == nullptr || message->msg != CURLMSG_DONE) { + return IOError("HTTP transfer ended without a completion result"); + } + transfer_result = message->data.result; + break; + } + if (idle_timeout != nullptr && !check_idle()) { + transfer_result = CURLE_OPERATION_TIMEDOUT; + break; + } + auto wait_ms = std::chrono::milliseconds(1000); + long curl_timeout = -1; + if (const auto status = curl_multi_timeout(multi.get(), &curl_timeout); + status != CURLM_OK) { + return IOError("HTTP timer query failed: {}", curl_multi_strerror(status)); + } + if (curl_timeout >= 0) { + wait_ms = std::min(wait_ms, std::chrono::milliseconds(curl_timeout)); + } + if (idle_timeout != nullptr) { + if (auto remaining = idle_timeout->Remaining()) { + wait_ms = std::min(wait_ms, *remaining); + } + } + std::vector watched; + watched.reserve(sockets.size()); + for (const auto& [socket, action] : sockets) { + short events = 0; + if (action & CURL_POLL_IN) events |= CURL_WAIT_POLLIN; + if (action & CURL_POLL_OUT) events |= CURL_WAIT_POLLOUT; + if (events != 0) watched.push_back({socket, events, 0}); + } + int activity = 0; +#if LIBCURL_VERSION_NUM >= 0x074200 + const auto poll_result = curl_multi_poll( + multi.get(), watched.data(), static_cast(watched.size()), + static_cast(wait_ms.count()), &activity); +#else + const auto poll_started = std::chrono::steady_clock::now(); + const auto poll_result = curl_multi_wait( + multi.get(), watched.data(), static_cast(watched.size()), + static_cast(wait_ms.count()), &activity); + if (activity == 0) std::this_thread::sleep_until(poll_started + wait_ms); +#endif + if (poll_result != CURLM_OK) { + return IOError("HTTP transfer poll failed: {}", curl_multi_strerror(poll_result)); + } + bool handled_socket = false; + for (const auto& socket : watched) { + if (socket.revents == 0 || !sockets.contains(socket.fd)) continue; + int events = 0; + if (socket.revents & CURL_WAIT_POLLIN) { + events |= CURL_CSELECT_IN; + if (idle_timeout != nullptr) idle_timeout->RecordReadActivity(); + } + if (socket.revents & CURL_WAIT_POLLOUT) events |= CURL_CSELECT_OUT; + if (socket.revents & CURL_WAIT_POLLPRI) events |= CURL_CSELECT_ERR; + result = curl_multi_socket_action(multi.get(), socket.fd, events, &running); + handled_socket = true; + if (result != CURLM_OK || running == 0) break; + } + if (!handled_socket) { + result = curl_multi_socket_action(multi.get(), CURL_SOCKET_TIMEOUT, 0, &running); + } + } + return session.Complete(transfer_result); +} + /// \brief Default error type for unparseable REST responses. constexpr std::string_view kRestExceptionType = "RESTException"; @@ -88,7 +551,7 @@ const cpr::Redirect kNoFollowRedirects{/*maximum=*/0L, /*follow=*/false, /// \brief Merge default headers with per-request headers (per-request wins). HttpHeaders MergeHeaders( - const std::unordered_map& default_headers, + const HttpHeaders& default_headers, const std::unordered_map& request_headers) { HttpHeaders merged; for (const auto& [key, val] : default_headers) { @@ -149,11 +612,19 @@ bool IsSuccessful(int32_t status_code) { /// \brief Builds a default ErrorResponse when the response body cannot be parsed. ErrorResponse BuildDefaultErrorResponse(const cpr::Response& response) { + std::string message; + if (response.status_code == 0 && !response.error.message.empty()) { + // Transport failure: libcurl's description never contains response content. + message = response.error.message; + } else if (!response.reason.empty()) { + message = response.reason; + } else { + message = GetStandardReasonPhrase(response.status_code); + } return { .code = static_cast(response.status_code), .type = std::string(kRestExceptionType), - .message = !response.reason.empty() ? response.reason - : GetStandardReasonPhrase(response.status_code), + .message = std::move(message), }; } @@ -173,8 +644,6 @@ Status CheckTransfer(const cpr::Response& response) { } /// \brief Handles failure responses by invoking the provided error handler. -/// -/// A failed transfer is checked first, as its status code may be 0 or a success. Status HandleFailureResponse(const cpr::Response& response, const ErrorHandler& error_handler) { ICEBERG_RETURN_UNEXPECTED(CheckTransfer(response)); @@ -183,28 +652,52 @@ Status HandleFailureResponse(const cpr::Response& response, } auto parse_result = error_handler.ParseResponse( static_cast(response.status_code), response.text); - const ErrorResponse final_error = - parse_result.value_or(BuildDefaultErrorResponse(response)); - return error_handler.Accept(final_error); + const auto final_error = parse_result.value_or(BuildDefaultErrorResponse(response)); + auto status = error_handler.Accept(final_error); + if (status.has_value()) { + return RestError("Unhandled error (code: {}, type: {}): {}", final_error.code, + final_error.type, final_error.message); + } + return status; } } // namespace -HttpClient::HttpClient(std::unordered_map default_headers, - std::chrono::milliseconds connection_timeout, - std::chrono::milliseconds socket_timeout, +std::shared_ptr HttpClient::Make( + std::unordered_map default_headers, + HttpClientOptions options, std::optional base_uri) { + return std::shared_ptr( + new HttpClient(WithOptions{}, std::move(default_headers), std::move(options), {}, + std::move(base_uri))); +} + +HttpClient::HttpClient(WithOptions, + std::unordered_map default_headers, + HttpClientOptions options, + std::chrono::milliseconds request_timeout, std::optional base_uri) - : base_uri_{std::move(base_uri)}, - default_headers_{std::move(default_headers)}, - connection_timeout_{connection_timeout}, - socket_timeout_{socket_timeout}, + : base_uri_(std::move(base_uri)), + options_(std::move(options)), + default_headers_{default_headers.begin(), default_headers.end()}, + socket_timeout_(request_timeout), connection_pool_{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; - default_headers_[kHeaderUserAgent] = kUserAgent; + default_headers_.try_emplace(kHeaderAccept, kMimeTypeApplicationJson); + default_headers_.try_emplace(kHeaderUserAgent, + options_.user_agent.value_or(kUserAgent)); } +HttpClient::HttpClient(std::unordered_map default_headers, + std::chrono::milliseconds connection_timeout, + std::chrono::milliseconds socket_timeout, + std::optional base_uri) + : HttpClient(WithOptions{}, std::move(default_headers), + HttpClientOptions{.connection_timeout_ms = + std::max(0, connection_timeout.count())}, + socket_timeout, std::move(base_uri)) {} + HttpClient::~HttpClient() = default; Result HttpClient::RequestUri(const std::string& path) const { @@ -225,24 +718,139 @@ Result HttpClient::RequestUri(const std::string& path) const { return full_path; } +Result HttpClient::Execute(HttpMethod method, const std::string& url, + HttpHeaders headers, std::string body, + const ErrorHandler& error_handler, + auth::AuthSession& session) { + ICEBERG_ASSIGN_OR_RAISE( + auto request, + AuthenticateRequest(session, method, url, std::move(headers), std::move(body))); + ICEBERG_ASSIGN_OR_RAISE( + auto response, + SendWithRetry( + options_, request, + [&](int64_t exec_count, + std::chrono::system_clock::time_point first_attempt) -> Result { + detail::SocketIdleTimeout idle_timeout( + std::chrono::milliseconds(options_.socket_timeout_ms.value_or(0))); + bool socket_timed_out = false; + bool retry_approved = false; + bool final_response_started = false; + std::string raw_headers; + cpr::Session cpr_session; + cpr_session.SetConnectionPool(*connection_pool_); + ApplyOptions(cpr_session, options_); + cpr_session.SetRedirect(kNoFollowRedirects); + cpr_session.SetTimeout(cpr::Timeout{socket_timeout_}); + cpr_session.SetUrl(cpr::Url{request.url}); + cpr_session.SetHeader(ToCprHeader(request)); + auto* curl = cpr_session.GetCurlHolder()->handle; + cpr_session.SetHeaderCallback(cpr::HeaderCallback{[&](std::string_view header, + intptr_t /*userdata*/) { + raw_headers.append(header); + cpr::Response candidate; + curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &candidate.status_code); + if (candidate.status_code >= 200) final_response_started = true; + if (header != "\r\n" && header != "\n") return true; + if (candidate.status_code < 200) return true; + candidate.header = cpr::util::parseHeader(raw_headers); + const auto elapsed = std::chrono::duration_cast( + std::chrono::time_point_cast( + std::chrono::system_clock::now()) - + first_attempt); + if (detail::ShouldRetry(candidate, request, exec_count, options_, + elapsed)) { + // Apache decides HTTP-status retries before reading the entity. + // Do not wait for an error response body that may never arrive. + retry_approved = true; + return false; + } + return true; + }}); + const bool has_socket_timeout = options_.socket_timeout_ms.value_or(0) > 0; + cpr::cpr_pf_arg_t download_progress = 0; + cpr::cpr_pf_arg_t upload_progress = 0; + cpr::cpr_pf_arg_t upload_size = 0; + const auto check_idle = [&] { + double pretransfer_seconds = 0; + long request_bytes = 0; + curl_easy_getinfo(curl, CURLINFO_PRETRANSFER_TIME, &pretransfer_seconds); + curl_easy_getinfo(curl, CURLINFO_REQUEST_SIZE, &request_bytes); + if ((pretransfer_seconds == 0 && request_bytes == 0 && + download_progress == 0 && upload_progress == 0 && + raw_headers.empty()) || + (!final_response_started && upload_progress < upload_size)) { + idle_timeout.Reset(); + return true; + } + if (!idle_timeout.Check(download_progress, + final_response_started ? 0 : upload_progress, + raw_headers.size())) { + socket_timed_out = true; + return false; + } + return true; + }; + if (has_socket_timeout) { + cpr_session.SetProgressCallback(cpr::ProgressCallback{ + [&](cpr::cpr_pf_arg_t /*download_total*/, cpr::cpr_pf_arg_t downloaded, + cpr::cpr_pf_arg_t upload_total, cpr::cpr_pf_arg_t uploaded, + intptr_t /*userdata*/) { + download_progress = downloaded; + upload_size = upload_total; + upload_progress = uploaded; + return check_idle(); + }}); + } + switch (request.method) { + case HttpMethod::kGet: + cpr_session.PrepareGet(); + break; + case HttpMethod::kHead: + cpr_session.PrepareHead(); + break; + case HttpMethod::kPost: + cpr_session.SetBody(cpr::Body{request.body}); + cpr_session.PreparePost(); + break; + case HttpMethod::kDelete: + cpr_session.PrepareDelete(); + break; + default: + return NotSupported("Unsupported HTTP method"); + } + ICEBERG_ASSIGN_OR_RAISE( + auto result, + PerformSession(cpr_session, has_socket_timeout ? &idle_timeout : nullptr, + check_idle)); + result.raw_header = std::move(raw_headers); + result.header = cpr::util::parseHeader(result.raw_header, &result.status_line, + &result.reason); + if (retry_approved) { + result.error = {}; + } else if (socket_timed_out) { + result.error.code = cpr::ErrorCode::OPERATION_TIMEDOUT; + result.error.message = + "No response progress within the configured socket timeout"; + } + return Attempt{.response = std::move(result), + .retry_approved = retry_approved}; + })); + + ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler)); + HttpResponse http_response; + http_response.impl_ = std::make_unique(std::move(response)); + return http_response; +} + Result HttpClient::Get( const std::string& path, const std::unordered_map& params, 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 url, AppendQueryString(request_uri, params)); - 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 Execute(HttpMethod::kGet, url, MergeHeaders(default_headers_, headers), "", + error_handler, session); } Result HttpClient::Post( @@ -250,19 +858,8 @@ Result HttpClient::Post( 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::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 Execute(HttpMethod::kPost, request_uri, MergeHeaders(default_headers_, headers), + body, error_handler, session); } Result HttpClient::PostForm( @@ -281,39 +878,17 @@ Result HttpClient::PostForm( // Sign the exact bytes cpr will put on the wire. std::string encoded_body = cpr::Payload(pair_list.begin(), pair_list.end()).GetContent(); - ICEBERG_ASSIGN_OR_RAISE( - auto authenticated, - 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 Execute(HttpMethod::kPost, request_uri, + MergeHeaders(default_headers_, form_headers), std::move(encoded_body), + error_handler, session); } 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; + return Execute(HttpMethod::kHead, request_uri, MergeHeaders(default_headers_, headers), + "", error_handler, session); } Result HttpClient::Delete( @@ -322,19 +897,8 @@ Result HttpClient::Delete( const ErrorHandler& error_handler, auth::AuthSession& session) { ICEBERG_ASSIGN_OR_RAISE(auto request_uri, RequestUri(path)); ICEBERG_ASSIGN_OR_RAISE(auto url, AppendQueryString(request_uri, params)); - ICEBERG_ASSIGN_OR_RAISE( - 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 Execute(HttpMethod::kDelete, url, MergeHeaders(default_headers_, headers), "", + error_handler, session); } } // namespace iceberg::rest diff --git a/src/iceberg/catalog/rest/http_client.h b/src/iceberg/catalog/rest/http_client.h index 9a24593878..15435ad04e 100644 --- a/src/iceberg/catalog/rest/http_client.h +++ b/src/iceberg/catalog/rest/http_client.h @@ -22,9 +22,11 @@ #include #include #include +#include #include #include +#include "iceberg/catalog/rest/http_request.h" #include "iceberg/catalog/rest/iceberg_rest_export.h" #include "iceberg/catalog/rest/type_fwd.h" #include "iceberg/result.h" @@ -69,14 +71,42 @@ class ICEBERG_REST_EXPORT HttpResponse { std::unique_ptr impl_; }; +/// \brief Transport settings shared by the config, init and catalog clients. +/// +/// Property names and defaults mirror the Java `HTTPClient` (`rest.client.*`). +struct ICEBERG_REST_EXPORT HttpClientOptions { + /// `rest.client.connection-timeout-ms`; Java defaults to three minutes. + /// Zero disables the configured connection deadline. + std::optional connection_timeout_ms = 180000; + /// `rest.client.socket-timeout-ms`, the maximum transfer inactivity. + /// Unset or zero leaves it unbounded. + std::optional socket_timeout_ms; + /// `rest.client.max-retries`, the maximum number of retries per request. + int max_retries = 5; + /// `rest.client.user-agent`; unset sends `iceberg-cpp/`. + std::optional user_agent; + /// `rest.client.idempotency-key-lifetime`, parsed from an ISO-8601 duration. + /// Unset disables the bound; zero disables retries of keyed requests. + std::optional idempotency_key_lifetime; + + static Result FromProperties( + const std::unordered_map& properties); +}; + /// \brief HTTP client for making requests to Iceberg REST Catalog API. class ICEBERG_REST_EXPORT HttpClient { public: + /// \brief Create a client using Java-style transport options. + static std::shared_ptr Make( + std::unordered_map default_headers = {}, + HttpClientOptions options = {}, std::optional base_uri = std::nullopt); + /// \brief Construct with a connection timeout and a legacy total request timeout. /// \param connection_timeout,socket_timeout Zero leaves requests unbounded. - explicit HttpClient(std::unordered_map default_headers = {}, - std::chrono::milliseconds connection_timeout = {}, - std::chrono::milliseconds socket_timeout = {}, - std::optional base_uri = std::nullopt); + explicit HttpClient( + std::unordered_map default_headers = {}, + std::chrono::milliseconds connection_timeout = std::chrono::minutes(3), + std::chrono::milliseconds socket_timeout = {}, + std::optional base_uri = std::nullopt); virtual ~HttpClient(); HttpClient(const HttpClient&) = delete; @@ -116,13 +146,24 @@ class ICEBERG_REST_EXPORT HttpClient { const ErrorHandler& error_handler, auth::AuthSession& session); private: + struct WithOptions {}; + HttpClient(WithOptions, std::unordered_map default_headers, + HttpClientOptions options, std::chrono::milliseconds request_timeout, + std::optional base_uri); + /// \brief Build a request URI from this client's base URI and a path. Result RequestUri(const std::string& path) const; + /// \brief Authenticate once, then send and retry the authenticated request. + Result Execute(HttpMethod method, const std::string& url, + HttpHeaders headers, std::string body, + const ErrorHandler& error_handler, + auth::AuthSession& session); + std::optional base_uri_; - std::unordered_map default_headers_; - std::chrono::milliseconds connection_timeout_; - std::chrono::milliseconds socket_timeout_; + HttpClientOptions options_; + HttpHeaders default_headers_; + std::chrono::milliseconds socket_timeout_{}; std::unique_ptr connection_pool_; }; diff --git a/src/iceberg/catalog/rest/http_client_internal.h b/src/iceberg/catalog/rest/http_client_internal.h new file mode 100644 index 0000000000..7e3c09e427 --- /dev/null +++ b/src/iceberg/catalog/rest/http_client_internal.h @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + */ + +#pragma once + +#include +#include +#include +#include + +#include "iceberg/catalog/rest/http_client.h" +#include "iceberg/catalog/rest/iceberg_rest_export.h" + +namespace cpr { +class Response; +} // namespace cpr + +namespace iceberg::rest::detail { + +ICEBERG_REST_EXPORT bool ShouldRetry( + const cpr::Response& response, const HttpRequest& request, int64_t exec_count, + const HttpClientOptions& options, + std::optional elapsed = std::nullopt); + +ICEBERG_REST_EXPORT std::chrono::milliseconds RetryInterval( + const cpr::Response& response, int64_t exec_count, + std::chrono::system_clock::time_point now = std::chrono::system_clock::now()); + +/// Tracks transfer inactivity independently of the average byte rate. +class ICEBERG_REST_EXPORT SocketIdleTimeout { + public: + using Clock = std::chrono::steady_clock; + + explicit SocketIdleTimeout(std::chrono::milliseconds timeout) : timeout_(timeout) {} + + bool Check(int64_t downloaded, int64_t uploaded, int64_t header_bytes, + Clock::time_point now = Clock::now()); + void Reset() { last_progress_.reset(); } + void RecordReadActivity(Clock::time_point now = Clock::now()) { last_progress_ = now; } + std::optional Remaining( + Clock::time_point now = Clock::now()) const; + + private: + std::chrono::milliseconds timeout_; + std::optional last_progress_; + int64_t downloaded_ = 0; + int64_t uploaded_ = 0; + int64_t header_bytes_ = 0; +}; + +} // namespace iceberg::rest::detail diff --git a/src/iceberg/catalog/rest/rest_catalog.cc b/src/iceberg/catalog/rest/rest_catalog.cc index a25d28eb47..5319b4cad8 100644 --- a/src/iceberg/catalog/rest/rest_catalog.cc +++ b/src/iceberg/catalog/rest/rest_catalog.cc @@ -453,15 +453,11 @@ Result> RestCatalog::Make( if (!client_factory) { client_factory = [](const RestCatalogProperties& properties) -> Result> { - // A value that is not positive leaves requests unbounded. - const auto timeout_of = [&properties](const auto& entry) { - return std::chrono::milliseconds(std::max(0, properties.Get(entry))); - }; + ICEBERG_ASSIGN_OR_RAISE(auto options, + HttpClientOptions::FromProperties(properties.configs())); ICEBERG_ASSIGN_OR_RAISE(auto uri, properties.Uri()); - return std::make_shared( - properties.ExtractHeaders(), - timeout_of(RestCatalogProperties::kClientConnectionTimeoutMs), - timeout_of(RestCatalogProperties::kClientSocketTimeoutMs), std::string(uri)); + return HttpClient::Make(properties.ExtractHeaders(), std::move(options), + std::string(uri)); }; } ICEBERG_ASSIGN_OR_RAISE(auto uri, config.Uri()); diff --git a/src/iceberg/test/http_client_test.cc b/src/iceberg/test/http_client_test.cc index 6cc653131e..6c18ae1855 100644 --- a/src/iceberg/test/http_client_test.cc +++ b/src/iceberg/test/http_client_test.cc @@ -26,11 +26,15 @@ #include #include #include +#include +#include #include #include #include #include +#include +#include #include #include @@ -46,6 +50,7 @@ #include "iceberg/catalog/rest/auth/auth_session.h" #include "iceberg/catalog/rest/error_handlers.h" +#include "iceberg/catalog/rest/http_client_internal.h" #include "iceberg/test/matchers.h" namespace iceberg::rest { @@ -58,7 +63,16 @@ namespace { /// a fixed response and closes the connection. class LoopbackHttpServer { public: - explicit LoopbackHttpServer(std::string response) : response_(std::move(response)) {} + explicit LoopbackHttpServer(std::string response, + std::chrono::milliseconds body_byte_interval = {}, + std::chrono::milliseconds header_byte_interval = {}) + : response_(std::move(response)), + body_byte_interval_(body_byte_interval), + header_byte_interval_(header_byte_interval) {} + + explicit LoopbackHttpServer(std::vector responses, + bool wait_for_client_close = false) + : responses_(std::move(responses)), wait_for_client_close_(wait_for_client_close) {} ~LoopbackHttpServer() { stop_ = true; @@ -101,6 +115,10 @@ class LoopbackHttpServer { int port() const { return port_; } int request_count() const { return request_count_.load(); } + std::vector requests() const { + std::lock_guard lock(requests_mutex_); + return requests_; + } std::string url(std::string_view path) const { return std::format("http://127.0.0.1:{}{}", port_, path); } @@ -116,16 +134,46 @@ class LoopbackHttpServer { if (conn < 0) { continue; } - ReadRequest(conn); - ++request_count_; - SendAll(conn, response_); + auto request = ReadRequest(conn); + { + std::lock_guard lock(requests_mutex_); + requests_.push_back(std::move(request)); + } + const auto index = static_cast(request_count_.fetch_add(1)); + const auto& response = responses_.empty() + ? response_ + : responses_[std::min(index, responses_.size() - 1)]; + if (header_byte_interval_ > std::chrono::milliseconds::zero()) { + const auto line_end = response.find("\r\n") + 2; + const auto body_start = response.find("\r\n\r\n") + 4; + SendAll(conn, response.substr(0, line_end)); + for (size_t i = line_end; i < body_start && !stop_; ++i) { + std::this_thread::sleep_for(header_byte_interval_); + SendAll(conn, response.substr(i, 1)); + } + if (!stop_) SendAll(conn, response.substr(body_start)); + } else if (body_byte_interval_ == std::chrono::milliseconds::zero()) { + SendAll(conn, response); + } else { + const auto body_start = response.find("\r\n\r\n") + 4; + SendAll(conn, response.substr(0, body_start)); + for (size_t i = body_start; i < response.size() && !stop_; ++i) { + std::this_thread::sleep_for(body_byte_interval_); + SendAll(conn, response.substr(i, 1)); + } + } + if (wait_for_client_close_) { + std::array buffer{}; + while (recv(conn, buffer.data(), buffer.size(), 0) > 0) { + } + } close(conn); } } /// \brief Consume the request headers and any Content-Length body so the /// client does not see a connection reset before it reads the response. - static void ReadRequest(int conn) { + static std::string ReadRequest(int conn) { timeval timeout{.tv_sec = 10, .tv_usec = 0}; setsockopt(conn, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)); setsockopt(conn, SOL_SOCKET, SO_SNDTIMEO, &timeout, sizeof(timeout)); @@ -137,7 +185,7 @@ class LoopbackHttpServer { while (true) { ssize_t n = recv(conn, buf.data(), buf.size(), 0); if (n <= 0) { - return; + return request; } request.append(buf.data(), static_cast(n)); if (header_end == std::string::npos) { @@ -148,7 +196,7 @@ class LoopbackHttpServer { content_length = ParseContentLength(request.substr(0, header_end)); } if (request.size() >= header_end + 4 + content_length) { - return; + return request; } } } @@ -184,13 +232,43 @@ class LoopbackHttpServer { } std::string response_; + std::chrono::milliseconds body_byte_interval_{}; + std::chrono::milliseconds header_byte_interval_{}; + std::vector responses_; + bool wait_for_client_close_ = false; int listen_fd_ = -1; int port_ = 0; std::atomic stop_{false}; std::atomic request_count_{0}; + mutable std::mutex requests_mutex_; + std::vector requests_; std::thread thread_; }; +std::string ResponseBytes(int code, std::string_view body = "{}", + std::string_view extra_headers = "") { + return std::format( + "HTTP/1.1 {} Response\r\nContent-Length: {}\r\n{}" + "Connection: close\r\n\r\n{}", + code, body.size(), extra_headers, body); +} + +class CountingAuthSession final : public auth::AuthSession { + public: + explicit CountingAuthSession(bool add_key = false) : add_key_(add_key) {} + + Result Authenticate(HttpRequest request) override { + request.headers["Authorization"] = std::format("Bearer attempt-{}", ++calls); + if (add_key_) request.headers["idempotency-key"] = "test-key"; + return request; + } + + int calls = 0; + + private: + bool add_key_; +}; + // A loopback socket listening on an ephemeral port. struct Listener { int fd = -1; @@ -220,10 +298,11 @@ void Listen(Listener& listener) { // Accepts one connection, reads the request head and sends `response`. Then // ends the connection if `end`, or else waits for the client to close it. -void ServeOnce(int listener, std::string_view response, bool end) { +void ServeOnce(int listener, std::string_view response, bool end, + int timeout_seconds = 10) { // Bound every wait, so the test cannot hang. timeval wait{}; - wait.tv_sec = 10; + wait.tv_sec = timeout_seconds; ::setsockopt(listener, SOL_SOCKET, SO_RCVTIMEO, &wait, sizeof(wait)); const int connection = ::accept(listener, nullptr, nullptr); if (connection < 0) { @@ -267,11 +346,7 @@ TEST(HttpClientTest, GetFailsWhenItsTransferFails) { Case{.name = "body stalls", .response = kTruncated, .end = false, - .error = "timed out"}, - Case{.name = "body ends early", - .response = kTruncated, - .end = true, - .error = "failed after status 200"}}) { + .error = "timed out"}}) { SCOPED_TRACE(test_case.name); Listener listener; ASSERT_NO_FATAL_FAILURE(Listen(listener)); @@ -289,6 +364,217 @@ TEST(HttpClientTest, GetFailsWhenItsTransferFails) { } } +TEST(HttpClientTest, SocketTimeoutAbortsAStalledTransfer) { + for (std::string_view bytes : + {std::string_view(""), std::string_view("HTTP/1.1 200 OK\r\n" + "Content-Length: 100\r\n\r\n{")}) { + SCOPED_TRACE(bytes); + Listener listener; + ASSERT_NO_FATAL_FAILURE(Listen(listener)); + const auto url = std::format("http://127.0.0.1:{}/v1/config", listener.port); + auto client = HttpClient::Make( + {}, HttpClientOptions{.socket_timeout_ms = 250, .max_retries = 0}); + auto session = auth::AuthSession::MakeDefault({}); + // Bound the server independently so a broken idle timer cannot hang the test. + std::thread server( + [&] { ServeOnce(listener.fd, bytes, /*end=*/false, /*timeout_seconds=*/3); }); + auto response = client->Get(url, {}, {}, *DefaultErrorHandler::Instance(), *session); + server.join(); + EXPECT_THAT(response, IsError(ErrorKind::kIOError)); + EXPECT_THAT(response, HasErrorMessage("timed out")); + } +} + +TEST(HttpClientTest, SocketTimeoutUsesTheMillisecondReadDeadline) { + LoopbackHttpServer server( + "HTTP/1.1 200 OK\r\nContent-Length: 4\r\n" + "Connection: close\r\n\r\nnull", + std::chrono::milliseconds(400)); + ASSERT_TRUE(server.Start()); + auto client = + HttpClient::Make({}, HttpClientOptions{.socket_timeout_ms = 250, .max_retries = 0}); + CountingAuthSession session; + auto response = client->Get(server.url("/v1/config"), {}, {}, + *DefaultErrorHandler::Instance(), session); + EXPECT_THAT(response, IsError(ErrorKind::kIOError)); + EXPECT_THAT(response, HasErrorMessage("timed out")); + EXPECT_EQ(server.request_count(), 1); +} + +TEST(HttpClientTest, SocketTimeoutAppliesAfterEarlyPostRejection) { + Listener listener; + ASSERT_NO_FATAL_FAILURE(Listen(listener)); + const auto url = std::format("http://127.0.0.1:{}/v1/commit", listener.port); + auto client = + HttpClient::Make({}, HttpClientOptions{.socket_timeout_ms = 250, .max_retries = 1}); + CountingAuthSession session; + // The server rejects Expect: 100-continue before the large upload begins, + // then leaves the final response body incomplete. + std::thread server([&] { + ServeOnce(listener.fd, "HTTP/1.1 403 Forbidden\r\nContent-Length: 100\r\n\r\n{", + /*end=*/false, /*timeout_seconds=*/2); + }); + auto response = client->Post(url, std::string(2 * 1024 * 1024, 'x'), {}, + *DefaultErrorHandler::Instance(), session); + server.join(); + EXPECT_THAT(response, IsError(ErrorKind::kIOError)); + EXPECT_THAT(response, HasErrorMessage("timed out")); +} + +TEST(HttpClientTest, SocketTimeoutAllowsAHeaderArrivingInFragments) { + LoopbackHttpServer server(ResponseBytes(200, "{}", "X-Slow: fragment\r\n"), + /*body_byte_interval=*/{}, + /*header_byte_interval=*/std::chrono::milliseconds(40)); + ASSERT_TRUE(server.Start()); + auto client = + HttpClient::Make({}, HttpClientOptions{.socket_timeout_ms = 250, .max_retries = 1}); + CountingAuthSession session; + ICEBERG_UNWRAP_OR_FAIL(auto response, + client->Get(server.url("/v1/config"), {}, {}, + *DefaultErrorHandler::Instance(), session)); + EXPECT_EQ(response.body(), "{}"); + EXPECT_EQ(response.headers().at("X-Slow"), "fragment"); + EXPECT_EQ(server.request_count(), 1); +} + +TEST(HttpClientTest, SocketTimeoutAllowsSlowContinuousProgress) { + // Below one byte per second, but every gap is shorter than the idle timeout. + LoopbackHttpServer server( + "HTTP/1.1 200 OK\r\nContent-Length: 4\r\n" + "Connection: close\r\n\r\nnull", + std::chrono::milliseconds(1200)); + ASSERT_TRUE(server.Start()); + auto client = HttpClient::Make( + {}, HttpClientOptions{.socket_timeout_ms = 2000, .max_retries = 0}); + auto session = auth::AuthSession::MakeDefault({}); + ICEBERG_UNWRAP_OR_FAIL(auto response, + client->Get(server.url("/v1/config"), {}, {}, + *DefaultErrorHandler::Instance(), *session)); + EXPECT_EQ(response.status_code(), 200); + EXPECT_EQ(response.body(), "null"); + EXPECT_EQ(server.request_count(), 1); +} + +TEST(HttpClientTest, DoesNotRetryTruncatedGetResponse) { + LoopbackHttpServer server( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n" + "Content-Length: 100\r\nConnection: close\r\n\r\n{"); + ASSERT_TRUE(server.Start()); + auto client = HttpClient::Make({}, HttpClientOptions{.max_retries = 1}); + auto session = auth::AuthSession::MakeDefault({}); + auto response = client->Get(server.url("/v1/config"), {}, {}, + *DefaultErrorHandler::Instance(), *session); + EXPECT_THAT(response, IsError(ErrorKind::kIOError)); + EXPECT_THAT(response, HasErrorMessage("failed after status 200")); + EXPECT_EQ(server.request_count(), 1); +} + +TEST(HttpClientTest, Retries503AtHeadersBeforeReadingTheBody) { + LoopbackHttpServer server( + std::vector{"HTTP/1.1 503 Service Unavailable\r\nRetry-After: 1\r\n" + "Content-Length: 100\r\n\r\n{", + ResponseBytes(200)}, + /*wait_for_client_close=*/true); + ASSERT_TRUE(server.Start()); + auto client = + HttpClient::Make({}, HttpClientOptions{.socket_timeout_ms = 500, .max_retries = 1}); + CountingAuthSession session; + ICEBERG_UNWRAP_OR_FAIL(auto response, + client->Post(server.url("/v1/commit"), "{}", {}, + *DefaultErrorHandler::Instance(), session)); + EXPECT_EQ(response.body(), "{}"); + EXPECT_EQ(server.request_count(), 2); + EXPECT_EQ(session.calls, 1); +} + +TEST(HttpClientTest, RetriesKeyedPostWithTheSameAuthentication) { + LoopbackHttpServer server(std::vector{ + ResponseBytes(500, "{}", "Retry-After: 1\r\n"), ResponseBytes(200)}); + ASSERT_TRUE(server.Start()); + auto client = HttpClient::Make( + {}, HttpClientOptions{.max_retries = 1, + .idempotency_key_lifetime = std::chrono::seconds(5)}); + CountingAuthSession session(/*add_key=*/true); + ICEBERG_UNWRAP_OR_FAIL(auto response, + client->Post(server.url("/v1/commit"), "{}", {}, + *DefaultErrorHandler::Instance(), session)); + EXPECT_EQ(response.status_code(), 200); + EXPECT_EQ(server.request_count(), 2); + EXPECT_EQ(session.calls, 1); + const auto requests = server.requests(); + ASSERT_EQ(requests.size(), 2); + EXPECT_EQ(requests[0], requests[1]); + EXPECT_THAT(requests[0], testing::HasSubstr("Bearer attempt-1")); +} + +TEST(HttpClientTest, KeyLifetimeStopsBeforeWaitingForTheNextAttempt) { + LoopbackHttpServer server(ResponseBytes(500, "{}", "Retry-After: 3600\r\n")); + ASSERT_TRUE(server.Start()); + auto client = HttpClient::Make( + {}, HttpClientOptions{.max_retries = 5, + .idempotency_key_lifetime = std::chrono::seconds(1)}); + CountingAuthSession session(/*add_key=*/true); + auto response = client->Post(server.url("/v1/commit"), "{}", {}, + *DefaultErrorHandler::Instance(), session); + EXPECT_THAT(response, IsError(ErrorKind::kInternalServerError)); + EXPECT_EQ(server.request_count(), 1); + EXPECT_EQ(session.calls, 1); +} + +TEST(HttpClientTest, UnkeyedPostIsNotRetriedOn500) { + LoopbackHttpServer server(ResponseBytes(500)); + ASSERT_TRUE(server.Start()); + auto client = HttpClient::Make({}, HttpClientOptions{.max_retries = 1}); + CountingAuthSession session; + EXPECT_THAT(client->Post(server.url("/v1/commit"), "{}", {}, + *DefaultErrorHandler::Instance(), session), + IsError(ErrorKind::kInternalServerError)); + EXPECT_EQ(server.request_count(), 1); + EXPECT_EQ(session.calls, 1); +} + +TEST(HttpClientTest, RetriesAnEmptyReplyWithoutReauthenticating) { + LoopbackHttpServer server(std::vector{"", ResponseBytes(200)}); + ASSERT_TRUE(server.Start()); + auto client = HttpClient::Make({}, HttpClientOptions{.max_retries = 1}); + CountingAuthSession session; + ICEBERG_UNWRAP_OR_FAIL(auto response, + client->Get(server.url("/v1/config"), {}, {}, + *DefaultErrorHandler::Instance(), session)); + EXPECT_EQ(response.status_code(), 200); + EXPECT_EQ(server.request_count(), 2); + EXPECT_EQ(session.calls, 1); +} + +TEST(HttpClientTest, FailsWhenAnErrorHandlerDoesNotReturnAnError) { + class NonThrowingHandler final : public DefaultErrorHandler { + public: + Status Accept(const ErrorResponse&) const override { return {}; } + } handler; + LoopbackHttpServer server(ResponseBytes(400)); + ASSERT_TRUE(server.Start()); + HttpClient client; + CountingAuthSession session; + auto response = client.Get(server.url("/v1/config"), {}, {}, handler, session); + EXPECT_THAT(response, IsError(ErrorKind::kRestError)); + EXPECT_THAT(response, HasErrorMessage("Unhandled error")); +} + +TEST(HttpClientTest, ExplicitUserAgentHeaderWinsOverTheProperty) { + LoopbackHttpServer server(ResponseBytes(200)); + ASSERT_TRUE(server.Start()); + auto client = HttpClient::Make({{"user-agent", "explicit/2.0"}}, + HttpClientOptions{.user_agent = "property/1.0"}); + CountingAuthSession session; + ICEBERG_UNWRAP_OR_FAIL(auto response, + client->Get(server.url("/v1/config"), {}, {}, + *DefaultErrorHandler::Instance(), session)); + const auto requests = server.requests(); + ASSERT_EQ(requests.size(), 1); + EXPECT_THAT(requests[0], testing::HasSubstr("explicit/2.0")); + EXPECT_THAT(requests[0], testing::Not(testing::HasSubstr("property/1.0"))); +} + TEST(HttpClientTest, PostReportsTransferFailureAfterSuccessfulStatus) { // A write that got its status may have taken effect, so the caller must treat // the result as unknown rather than retrying or cleaning up automatically. @@ -328,18 +614,33 @@ class CapturingAuthSession final : public auth::AuthSession { }; TEST(HttpClientTest, BuildsRequestUriLikeJavaHttpRequest) { - HttpClient client({}, {}, {}, std::string("https://catalog.example/api/")); - CapturingAuthSession session; - - auto relative = client.PostForm("v1/oauth/tokens///", {}, {}, - *DefaultErrorHandler::Instance(), session); - EXPECT_THAT(relative, IsError(ErrorKind::kInvalidArgument)); - EXPECT_EQ(session.request().url, "https://catalog.example/api/v1/oauth/tokens"); + const std::string base_uri = "https://catalog.example/api/"; + for (const auto& client : + {std::make_shared(std::unordered_map{}, + std::chrono::milliseconds{}, + std::chrono::milliseconds{}, base_uri), + HttpClient::Make({}, {}, base_uri)}) { + CapturingAuthSession session; + const auto& handler = *DefaultErrorHandler::Instance(); + const auto expect_url = [&](const Result& response, + std::string_view expected_url) { + EXPECT_THAT(response, IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(response, HasErrorMessage("captured")); + EXPECT_EQ(session.request().url, expected_url); + }; - auto absolute = client.PostForm("https://auth.example/token/", {}, {}, - *DefaultErrorHandler::Instance(), session); - EXPECT_THAT(absolute, IsError(ErrorKind::kInvalidArgument)); - EXPECT_EQ(session.request().url, "https://auth.example/token/"); + const std::string path = "v1/oauth/tokens///"; + const std::string url = "https://catalog.example/api/v1/oauth/tokens"; + expect_url(client->Get(path, {{"page", "2"}}, {}, handler, session), url + "?page=2"); + expect_url(client->Post(path, "{}", {}, handler, session), url); + expect_url(client->PostForm(path, {}, {}, handler, session), url); + expect_url(client->Head(path, {}, handler, session), url); + expect_url(client->Delete(path, {{"page", "2"}}, {}, handler, session), + url + "?page=2"); + + expect_url(client->PostForm("https://auth.example/token/", {}, {}, handler, session), + "https://auth.example/token/"); + } } // Redirect responses must be reported without replaying the request at the target. @@ -389,4 +690,286 @@ TEST(HttpClientTest, DoesNotFollowRedirects) { #endif } +TEST(HttpClientRetryTest, TransportRetriesMatchTheJavaExceptionBoundary) { + struct Case { + HttpMethod method; + int status; + cpr::ErrorCode error; + bool retry; + }; + const Case cases[] = { + {HttpMethod::kGet, 0, cpr::ErrorCode::RECV_ERROR, true}, + {HttpMethod::kHead, 0, cpr::ErrorCode::GOT_NOTHING, true}, + {HttpMethod::kDelete, 0, cpr::ErrorCode::SEND_ERROR, true}, + {HttpMethod::kGet, 0, cpr::ErrorCode::OPERATION_TIMEDOUT, false}, + {HttpMethod::kGet, 0, cpr::ErrorCode::SSL_CONNECT_ERROR, false}, + {HttpMethod::kGet, 0, cpr::ErrorCode::COULDNT_CONNECT, false}, + {HttpMethod::kGet, 0, cpr::ErrorCode::PARTIAL_FILE, false}, + {HttpMethod::kGet, 0, cpr::ErrorCode::WEIRD_SERVER_REPLY, false}, + {HttpMethod::kGet, 200, cpr::ErrorCode::PARTIAL_FILE, false}, + {HttpMethod::kGet, 200, cpr::ErrorCode::RECV_ERROR, false}, + {HttpMethod::kPost, 0, cpr::ErrorCode::RECV_ERROR, false}, + {HttpMethod::kPost, 200, cpr::ErrorCode::PARTIAL_FILE, false}, + }; + for (const auto& test_case : cases) { + SCOPED_TRACE(std::format("method={}, status={}, error={}", + static_cast(test_case.method), test_case.status, + static_cast(test_case.error))); + cpr::Response response; + response.status_code = test_case.status; + response.error.code = test_case.error; + const HttpRequest request{.method = test_case.method}; + EXPECT_EQ(detail::ShouldRetry(response, request, 1, {}), test_case.retry); + } +} + +TEST(HttpClientRetryTest, StatusRetriesRespectMethodAndIdempotencyKey) { + for (const int code : {200, 400, 408, 429, 500, 502, 503, 504}) { + SCOPED_TRACE(code); + cpr::Response response; + response.status_code = code; + HttpRequest request{.method = HttpMethod::kGet}; + EXPECT_EQ(detail::ShouldRetry(response, request, 1, {}), code != 200 && code != 400); + request.method = HttpMethod::kPost; + EXPECT_EQ(detail::ShouldRetry(response, request, 1, {}), code == 429); + request.headers["iDeMpOtEnCy-KeY"] = "key"; + EXPECT_EQ(detail::ShouldRetry(response, request, 1, {}), code != 200 && code != 400); + request.headers.clear(); + response.header["retry-after"] = "1"; + EXPECT_EQ(detail::ShouldRetry(response, request, 1, {}), code == 429 || code == 503); + EXPECT_FALSE(detail::ShouldRetry(response, request, 6, {})); + } +} + +TEST(HttpClientRetryTest, KeyedNetworkFailuresRetryWithoutAnExceptionDelay) { + cpr::Response response; + response.error.code = cpr::ErrorCode::RECV_ERROR; + const HttpRequest request{.method = HttpMethod::kPost, + .headers = {{"Idempotency-Key", "key"}}}; + EXPECT_TRUE(detail::ShouldRetry(response, request, 1, {})); + EXPECT_EQ(detail::RetryInterval(response, 1), std::chrono::milliseconds::zero()); +} + +TEST(HttpClientRetryTest, KeyLifetimeMatchesJavaBudgetChecks) { + struct Case { + std::chrono::milliseconds lifetime; + std::chrono::milliseconds elapsed; + bool has_key; + std::string retry_after; + int attempt; + bool transport_error; + bool retry; + }; + const Case cases[] = { + {std::chrono::seconds(1), {}, true, "3600", 1, false, false}, + {std::chrono::hours(1), {}, true, "1", 1, false, true}, + {std::chrono::milliseconds(1), {}, true, "", 1, true, false}, + {std::chrono::milliseconds(1), {}, false, "3600", 1, false, true}, + {std::chrono::minutes(30), std::chrono::minutes(31), true, "1", 1, false, false}, + {std::chrono::seconds(2), std::chrono::milliseconds(1100), true, "1", 2, false, + false}, + {std::chrono::milliseconds(1), {}, true, "", 1, false, false}, + {{}, {}, true, "1", 1, false, false}, + {{}, {}, false, "1", 1, false, true}, + {std::chrono::milliseconds(1050), {}, true, "", 1, false, false}, + {std::chrono::milliseconds(1101), {}, true, "", 1, false, true}, + }; + for (const auto& test_case : cases) { + cpr::Response response; + response.status_code = test_case.transport_error ? 0 : 500; + if (test_case.transport_error) response.error.code = cpr::ErrorCode::RECV_ERROR; + if (!test_case.retry_after.empty()) + response.header["Retry-After"] = test_case.retry_after; + HttpRequest request{.method = + test_case.has_key ? HttpMethod::kPost : HttpMethod::kGet}; + if (test_case.has_key) request.headers["Idempotency-Key"] = "key"; + HttpClientOptions options{.idempotency_key_lifetime = test_case.lifetime}; + EXPECT_EQ(detail::ShouldRetry(response, request, test_case.attempt, options, + test_case.elapsed), + test_case.retry); + } + // Like Java, a missing first-attempt timestamp does not impose a bound. + cpr::Response response; + response.status_code = 500; + response.header["Retry-After"] = "3600"; + const HttpRequest request{.method = HttpMethod::kPost, + .headers = {{"Idempotency-Key", "key"}}}; + EXPECT_TRUE(detail::ShouldRetry( + response, request, 1, {.idempotency_key_lifetime = std::chrono::milliseconds(1)})); +} + +TEST(HttpClientRetryTest, RetryAfterSupportsSecondsAndMillisecondDateDifferences) { + cpr::Response response; + response.header["Retry-After"] = "+2"; + EXPECT_EQ(detail::RetryInterval(response, 1), std::chrono::seconds(2)); + response.header["Retry-After"] = "Thu, 01 Jan 1970 00:00:02 GMT"; + EXPECT_EQ(detail::RetryInterval(response, 1, + std::chrono::system_clock::time_point{} + + std::chrono::milliseconds(1750)), + std::chrono::milliseconds(250)); +} + +TEST(HttpClientRetryTest, LargeRetryAfterValuesSaturateWithoutOverflow) { + for (const auto& value : + {"9223372036854776", "18446744073709552", "9223372036854775807"}) { + cpr::Response response; + response.header["Retry-After"] = value; + EXPECT_EQ(detail::RetryInterval(response, 1), std::chrono::milliseconds::max()); + } +} + +TEST(HttpClientRetryTest, InvalidRetryAfterFallsBackWithoutOverflow) { + for (const auto& value : {"9223372036854775808", "-9223372036854775808", "0", "-1", + "invalid", "Thu, 01 Jan 1970 00:00:00 GMT"}) { + SCOPED_TRACE(value); + cpr::Response response; + response.header["Retry-After"] = value; + const auto delay = detail::RetryInterval( + response, 1, + std::chrono::system_clock::time_point{} + std::chrono::milliseconds(1)); + EXPECT_GE(delay, std::chrono::seconds(1)); + EXPECT_LT(delay, std::chrono::milliseconds(1100)); + } +} + +TEST(HttpClientRetryTest, ExponentialBackoffIsCapped) { + cpr::Response response; + EXPECT_EQ(detail::RetryInterval(response, 0), std::chrono::milliseconds::zero()); + for (int64_t attempt : {int64_t{7}, int64_t{8}, std::numeric_limits::max()}) { + const auto delay = detail::RetryInterval(response, attempt); + EXPECT_GE(delay, std::chrono::seconds(64)); + EXPECT_LT(delay, std::chrono::milliseconds(70400)); + } +} + +TEST(HttpClientOptionsTest, SocketTimeoutAcceptsMaximumMilliseconds) { + ICEBERG_UNWRAP_OR_FAIL(auto options, + HttpClientOptions::FromProperties( + {{"rest.client.socket-timeout-ms", "2147483647"}})); + ASSERT_TRUE(options.socket_timeout_ms.has_value()); + const auto timeout = std::chrono::milliseconds(*options.socket_timeout_ms); + detail::SocketIdleTimeout idle(timeout); + const auto start = detail::SocketIdleTimeout::Clock::time_point{}; + EXPECT_TRUE(idle.Check(0, 0, 0, start)); + EXPECT_TRUE(idle.Check(0, 0, 0, start + timeout - std::chrono::milliseconds(1))); + EXPECT_FALSE(idle.Check(0, 0, 0, start + timeout)); +} + +TEST(HttpClientOptionsTest, SocketTimeoutMeasuresGapsRatherThanAverageSpeed) { + const auto timeout = std::chrono::seconds(10); + detail::SocketIdleTimeout idle(timeout); + const auto start = detail::SocketIdleTimeout::Clock::time_point{}; + for (int bytes = 0; bytes < 10; ++bytes) { + EXPECT_TRUE(idle.Check(bytes, 0, 0, start + std::chrono::seconds(2 * bytes))); + } + // Download, upload and response-header progress each restart the idle timer. + EXPECT_TRUE(idle.Check(9, 1, 0, start + std::chrono::seconds(27))); + EXPECT_TRUE(idle.Check(9, 1, 1, start + std::chrono::seconds(36))); + EXPECT_EQ(idle.Remaining(start + std::chrono::seconds(40)), std::chrono::seconds(6)); + EXPECT_FALSE(idle.Check(9, 1, 1, start + std::chrono::seconds(46))); + EXPECT_EQ(idle.Remaining(start + std::chrono::seconds(46)), + std::chrono::milliseconds::zero()); + idle.Reset(); + EXPECT_FALSE(idle.Remaining().has_value()); + EXPECT_TRUE(idle.Check(0, 0, 0, start + std::chrono::seconds(60))); +} + +TEST(HttpClientOptionsTest, PreservesLegacyBraceInitializedConstructors) { + HttpClient default_client; + HttpClient headers_client({}); + HttpClient connection_timeout_client({}, {}); + HttpClient total_timeout_client({}, {}, {}); +} + +TEST(HttpClientOptionsTest, DefaultsMatchJavaClient) { + ICEBERG_UNWRAP_OR_FAIL(auto options, HttpClientOptions::FromProperties({})); + EXPECT_EQ(options.connection_timeout_ms, 180000); + EXPECT_FALSE(options.socket_timeout_ms.has_value()); + EXPECT_EQ(options.max_retries, 5); + EXPECT_FALSE(options.user_agent.has_value()); + EXPECT_FALSE(options.idempotency_key_lifetime.has_value()); +} + +TEST(HttpClientOptionsTest, AppliesTransportProperties) { + ICEBERG_UNWRAP_OR_FAIL(auto options, + HttpClientOptions::FromProperties( + {{"rest.client.connection-timeout-ms", "17"}, + {"rest.client.socket-timeout-ms", "29"}, + {"rest.client.max-retries", "2"}, + {"rest.client.user-agent", "custom/1.0"}, + {"rest.client.idempotency-key-lifetime", "PT30M"}})); + EXPECT_EQ(options.connection_timeout_ms, 17); + EXPECT_EQ(options.socket_timeout_ms, 29); + EXPECT_EQ(options.max_retries, 2); + EXPECT_EQ(options.user_agent, "custom/1.0"); + EXPECT_EQ(options.idempotency_key_lifetime, std::chrono::minutes(30)); +} + +TEST(HttpClientOptionsTest, TimeoutsAcceptZeroAndExplicitPlusSigns) { + ICEBERG_UNWRAP_OR_FAIL(auto zero, HttpClientOptions::FromProperties( + {{"rest.client.connection-timeout-ms", "0"}, + {"rest.client.socket-timeout-ms", "0"}})); + EXPECT_EQ(zero.connection_timeout_ms, 0); + EXPECT_EQ(zero.socket_timeout_ms, 0); + ICEBERG_UNWRAP_OR_FAIL(auto positive, HttpClientOptions::FromProperties( + {{"rest.client.connection-timeout-ms", "+17"}, + {"rest.client.socket-timeout-ms", "+29"}, + {"rest.client.max-retries", "+2"}})); + EXPECT_EQ(positive.connection_timeout_ms, 17); + EXPECT_EQ(positive.socket_timeout_ms, 29); + EXPECT_EQ(positive.max_retries, 2); +} + +TEST(HttpClientOptionsTest, RejectsInvalidSettingsWithoutEchoingValues) { + for (auto key : {"rest.client.connection-timeout-ms", "rest.client.socket-timeout-ms", + "rest.client.max-retries"}) { + for (auto value : + {"-1", "9999999999999999999999", "secret-response", "1.5", "", "+", "+-0"}) { + auto result = HttpClientOptions::FromProperties({{key, value}}); + ASSERT_FALSE(result) << key << "=" << value; + EXPECT_NE(result.error().message.find(key), std::string::npos); + EXPECT_EQ(result.error().message.find("secret-response"), std::string::npos); + } + } + EXPECT_FALSE(HttpClientOptions::FromProperties({{"rest.client.max-retries", "0"}})); +} + +TEST(HttpClientOptionsTest, ParsesJavaDurationSyntaxForKeyLifetime) { + const std::pair cases[] = { + {"PT30M", 1800000}, + {"PT0S", 0}, + {"PT-5M", 0}, + {"-PT5M", 0}, + {"P2D", 172800000}, + {"P2DT3H4M", 183840000}, + {"PT20.345S", 20345}, + {"pt20,345s", 20345}, + {"PT0.000000001S", 0}, + {"PT-0.5S", 0}, + {"-PT-6H+3M", 21420000}, + {"+PT+1S", 1000}, + {"PT1.S", 1000}, + {"PT9223372036854775.807S", std::numeric_limits::max()}, + }; + for (const auto& [value, millis] : cases) { + SCOPED_TRACE(value); + ICEBERG_UNWRAP_OR_FAIL(auto options, + HttpClientOptions::FromProperties( + {{"rest.client.idempotency-key-lifetime", value}})); + EXPECT_EQ(options.idempotency_key_lifetime, std::chrono::milliseconds(millis)); + } +} + +TEST(HttpClientOptionsTest, IgnoresMalformedOrOutOfRangeKeyLifetimeLikeJava) { + for (const auto* value : + {"PT9999999999999H", "PT99999999999999999999H", "garbage", "30m", "", "P", "PT", + "P1DT", "PT.5S", "PT1.1234567890S", "PT9223372036854775.807000001S"}) { + SCOPED_TRACE(value); + ICEBERG_UNWRAP_OR_FAIL(auto options, + HttpClientOptions::FromProperties( + {{"rest.client.idempotency-key-lifetime", value}})); + EXPECT_FALSE(options.idempotency_key_lifetime.has_value()); + } +} + } // namespace iceberg::rest From 46e20f99c085250ff77997e49e5b1b9d43f6ddee Mon Sep 17 00:00:00 2001 From: Zhao Junwang Date: Fri, 30 Oct 2026 08:28:11 +0800 Subject: [PATCH 2/5] refactor(rest): improve transport testability and coverage Inject retry timing for deterministic tests while preserving Java retry behavior. Cover connection reuse, shorten slow socket tests, share timeout defaults and clarify legacy timeout naming and documentation. --- mkdocs/docs/rest-catalog.md | 12 +- src/iceberg/catalog/rest/catalog_properties.h | 3 +- src/iceberg/catalog/rest/constant.h | 3 + src/iceberg/catalog/rest/http_client.cc | 87 ++--- src/iceberg/catalog/rest/http_client.h | 15 +- .../catalog/rest/http_client_internal.h | 30 +- src/iceberg/test/http_client_test.cc | 318 ++++++++++++++++-- 7 files changed, 373 insertions(+), 95 deletions(-) diff --git a/mkdocs/docs/rest-catalog.md b/mkdocs/docs/rest-catalog.md index eadf1f29c6..ba838615b2 100644 --- a/mkdocs/docs/rest-catalog.md +++ b/mkdocs/docs/rest-catalog.md @@ -29,14 +29,14 @@ properties. | --- | --- | --- | | `rest.client.connection-timeout-ms` | 180000 | Non-negative integer; zero disables the configured deadline | | `rest.client.socket-timeout-ms` | Unbounded | Non-negative integer; zero disables the response inactivity timeout | -| `rest.client.max-retries` | 5 | Positive integer; excludes the first attempt | +| `rest.client.max-retries` | 5 | Positive integer; excludes the first attempt; does not bound total wait time | | `rest.client.user-agent` | `iceberg-cpp/` | Used when the request has no explicit `User-Agent` header | | `rest.client.idempotency-key-lifetime` | Unbounded | ISO-8601 duration, for example `PT30M`; zero or negative durations disable keyed retries; malformed or out-of-range values are ignored | The socket timeout measures response inactivity rather than average transfer -speed or total request duration. Connection setup and writing a request body -use their own transport behavior. Transfers run synchronously in the calling -thread, polling the remaining read deadline in milliseconds; no executor is +speed or total request duration. It does not bound request-body uploads. The +legacy `HttpClient` constructor's third argument remains a total timeout per +attempt. Transfers run synchronously in the calling thread; no executor is required for HTTP retries. HTTP response retries are decided after receiving the headers, before reading @@ -68,6 +68,10 @@ Authentication runs once before the first send. Retries replay the same headers, body and idempotency key. A subsequent logical request authenticates again and can use an updated token. +The client does not generate idempotency keys. Supply a distinct key for each +logical mutation through per-request headers or an `AuthSession`. A static +`header.Idempotency-Key` is unsuitable for unrelated mutations. + Errors are classified from the `code` reported in the error body, falling back to the HTTP status when the body cannot be parsed. Service diagnostics remain available. Malformed OAuth token responses never echo response content. diff --git a/src/iceberg/catalog/rest/catalog_properties.h b/src/iceberg/catalog/rest/catalog_properties.h index f5e7d629ed..fcb30058e9 100644 --- a/src/iceberg/catalog/rest/catalog_properties.h +++ b/src/iceberg/catalog/rest/catalog_properties.h @@ -23,6 +23,7 @@ #include #include +#include "iceberg/catalog/rest/constant.h" #include "iceberg/catalog/rest/iceberg_rest_export.h" #include "iceberg/result.h" #include "iceberg/util/config.h" @@ -74,7 +75,7 @@ class ICEBERG_REST_EXPORT RestCatalogProperties inline static constexpr std::string_view kHeaderPrefix = "header."; /// \brief HTTP connection timeout in milliseconds, matching Java REST clients. inline static Entry kClientConnectionTimeoutMs{ - "rest.client.connection-timeout-ms", 180000}; + "rest.client.connection-timeout-ms", kDefaultConnectionTimeoutMs}; /// \brief HTTP socket/read timeout in milliseconds, matching Java REST clients. inline static Entry kClientSocketTimeoutMs{"rest.client.socket-timeout-ms", 0}; diff --git a/src/iceberg/catalog/rest/constant.h b/src/iceberg/catalog/rest/constant.h index 0a6e8d0c01..fc7ac12241 100644 --- a/src/iceberg/catalog/rest/constant.h +++ b/src/iceberg/catalog/rest/constant.h @@ -19,6 +19,7 @@ #pragma once +#include #include #include "iceberg/version.h" @@ -28,6 +29,8 @@ namespace iceberg::rest { +inline constexpr int64_t kDefaultConnectionTimeoutMs = 180000; + inline const std::string kHeaderContentType = "Content-Type"; inline const std::string kHeaderAccept = "Accept"; inline const std::string kHeaderXClientVersion = "X-Client-Version"; diff --git a/src/iceberg/catalog/rest/http_client.cc b/src/iceberg/catalog/rest/http_client.cc index b6de8949d9..e9263827f6 100644 --- a/src/iceberg/catalog/rest/http_client.cc +++ b/src/iceberg/catalog/rest/http_client.cc @@ -35,6 +35,7 @@ #include #include "iceberg/catalog/rest/auth/auth_session.h" +#include "iceberg/catalog/rest/catalog_properties.h" #include "iceberg/catalog/rest/constant.h" #include "iceberg/catalog/rest/error_handlers.h" #include "iceberg/catalog/rest/http_client_internal.h" @@ -49,8 +50,6 @@ namespace { // Property names follow the Java HTTPClient. constexpr std::string_view kRestMaxRetries = "rest.client.max-retries"; -constexpr std::string_view kRestConnectionTimeoutMs = "rest.client.connection-timeout-ms"; -constexpr std::string_view kRestSocketTimeoutMs = "rest.client.socket-timeout-ms"; constexpr std::string_view kRestUserAgent = "rest.client.user-agent"; constexpr std::string_view kRestKeyLifetime = "rest.client.idempotency-key-lifetime"; @@ -169,10 +168,12 @@ std::optional ParseKeyLifetime(const std::string& val Result HttpClientOptions::FromProperties( const std::unordered_map& properties) { HttpClientOptions options; - ICEBERG_RETURN_UNEXPECTED(ParseIntegerProperty(properties, kRestConnectionTimeoutMs, - options.connection_timeout_ms)); - ICEBERG_RETURN_UNEXPECTED( - ParseIntegerProperty(properties, kRestSocketTimeoutMs, options.socket_timeout_ms)); + ICEBERG_RETURN_UNEXPECTED(ParseIntegerProperty( + properties, RestCatalogProperties::kClientConnectionTimeoutMs.key(), + options.connection_timeout_ms)); + ICEBERG_RETURN_UNEXPECTED(ParseIntegerProperty( + properties, RestCatalogProperties::kClientSocketTimeoutMs.key(), + options.socket_timeout_ms)); std::optional max_retries; ICEBERG_RETURN_UNEXPECTED( ParseIntegerProperty(properties, kRestMaxRetries, max_retries, 1)); @@ -309,7 +310,8 @@ namespace detail { bool ShouldRetry(const cpr::Response& response, const HttpRequest& request, int64_t exec_count, const HttpClientOptions& options, - std::optional elapsed) { + std::optional elapsed, + std::chrono::system_clock::time_point now) { if (exec_count > options.max_retries) return false; const bool has_key = request.headers.contains("Idempotency-Key"); const bool safe = IsIdempotent(request.method) || has_key; @@ -336,7 +338,7 @@ bool ShouldRetry(const cpr::Response& response, const HttpRequest& request, } auto upper_delay = Backoff(exec_count); if (!response.error) { - if (auto retry_after = RetryAfter(response, std::chrono::system_clock::now())) { + if (auto retry_after = RetryAfter(response, now)) { upper_delay = *retry_after; } else { upper_delay += upper_delay / 10; @@ -359,6 +361,31 @@ std::chrono::milliseconds RetryInterval(const cpr::Response& response, int64_t e return delay + std::chrono::milliseconds(jitter(generator)); } +RetryTiming::RetryTiming() + : now(std::chrono::system_clock::now), + sleep([](std::chrono::milliseconds delay) { std::this_thread::sleep_for(delay); }) { +} + +Result SendWithRetry(const HttpClientOptions& options, + const HttpRequest& request, const SendAttempt& send, + const RetryTiming& timing) { + // Java records FIRST_ATTEMPT_EPOCH with System.currentTimeMillis(). + const auto first_attempt = + std::chrono::time_point_cast(timing.now()); + for (int64_t exec_count = 1;; ++exec_count) { + ICEBERG_ASSIGN_OR_RAISE(auto attempt, send(exec_count, first_attempt)); + const auto now = timing.now(); + const auto elapsed = std::chrono::duration_cast( + std::chrono::time_point_cast(now) - first_attempt); + if (!attempt.retry_approved && + !ShouldRetry(attempt.response, request, exec_count, options, elapsed, now)) { + return std::move(attempt.response); + } + const auto delay = RetryInterval(attempt.response, exec_count, timing.now()); + if (delay > std::chrono::milliseconds::zero()) timing.sleep(delay); + } +} + bool SocketIdleTimeout::Check(int64_t downloaded, int64_t uploaded, int64_t header_bytes, Clock::time_point now) { if (!last_progress_.has_value() || downloaded != downloaded_ || uploaded != uploaded_ || @@ -383,33 +410,9 @@ std::optional SocketIdleTimeout::Remaining( namespace { -struct Attempt { - cpr::Response response; - bool retry_approved = false; -}; - -/// Retry the same authenticated request, matching Java's retry execution chain. -template -Result SendWithRetry(const HttpClientOptions& options, - const HttpRequest& request, Send send) { - const auto first_attempt = std::chrono::time_point_cast( - std::chrono::system_clock::now()); - for (int64_t exec_count = 1;; ++exec_count) { - ICEBERG_ASSIGN_OR_RAISE(auto attempt, send(exec_count, first_attempt)); - const auto elapsed = std::chrono::duration_cast( - std::chrono::time_point_cast( - std::chrono::system_clock::now()) - - first_attempt); - if (!attempt.retry_approved && - !detail::ShouldRetry(attempt.response, request, exec_count, options, elapsed)) { - return std::move(attempt.response); - } - std::this_thread::sleep_for(detail::RetryInterval(attempt.response, exec_count)); - } -} - void ApplyOptions(cpr::Session& session, const HttpClientOptions& options) { - const auto timeout = options.connection_timeout_ms.value_or(180000); + const auto timeout = + options.connection_timeout_ms.value_or(kDefaultConnectionTimeoutMs); // libcurl interprets zero as its own five-minute default. Use its largest // native deadline when Java's disabled connection timeout is requested. session.SetConnectTimeout(cpr::ConnectTimeout{std::chrono::milliseconds( @@ -679,7 +682,7 @@ HttpClient::HttpClient(WithOptions, : base_uri_(std::move(base_uri)), options_(std::move(options)), default_headers_{default_headers.begin(), default_headers.end()}, - socket_timeout_(request_timeout), + request_timeout_(request_timeout), connection_pool_{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. @@ -727,10 +730,10 @@ Result HttpClient::Execute(HttpMethod method, const std::string& u AuthenticateRequest(session, method, url, std::move(headers), std::move(body))); ICEBERG_ASSIGN_OR_RAISE( auto response, - SendWithRetry( + detail::SendWithRetry( options_, request, - [&](int64_t exec_count, - std::chrono::system_clock::time_point first_attempt) -> Result { + [&](int64_t exec_count, std::chrono::system_clock::time_point first_attempt) + -> Result { detail::SocketIdleTimeout idle_timeout( std::chrono::milliseconds(options_.socket_timeout_ms.value_or(0))); bool socket_timed_out = false; @@ -741,7 +744,7 @@ Result HttpClient::Execute(HttpMethod method, const std::string& u cpr_session.SetConnectionPool(*connection_pool_); ApplyOptions(cpr_session, options_); cpr_session.SetRedirect(kNoFollowRedirects); - cpr_session.SetTimeout(cpr::Timeout{socket_timeout_}); + cpr_session.SetTimeout(cpr::Timeout{request_timeout_}); cpr_session.SetUrl(cpr::Url{request.url}); cpr_session.SetHeader(ToCprHeader(request)); auto* curl = cpr_session.GetCurlHolder()->handle; @@ -762,6 +765,8 @@ Result HttpClient::Execute(HttpMethod method, const std::string& u elapsed)) { // Apache decides HTTP-status retries before reading the entity. // Do not wait for an error response body that may never arrive. + // Aborting before draining the body sacrifices connection reuse + // for this response; subsequent successful requests still reuse the pool. retry_approved = true; return false; } @@ -833,8 +838,8 @@ Result HttpClient::Execute(HttpMethod method, const std::string& u result.error.message = "No response progress within the configured socket timeout"; } - return Attempt{.response = std::move(result), - .retry_approved = retry_approved}; + return detail::Attempt{.response = std::move(result), + .retry_approved = retry_approved}; })); ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler)); diff --git a/src/iceberg/catalog/rest/http_client.h b/src/iceberg/catalog/rest/http_client.h index 15435ad04e..83c49425f5 100644 --- a/src/iceberg/catalog/rest/http_client.h +++ b/src/iceberg/catalog/rest/http_client.h @@ -26,6 +26,7 @@ #include #include +#include "iceberg/catalog/rest/constant.h" #include "iceberg/catalog/rest/http_request.h" #include "iceberg/catalog/rest/iceberg_rest_export.h" #include "iceberg/catalog/rest/type_fwd.h" @@ -77,7 +78,7 @@ class ICEBERG_REST_EXPORT HttpResponse { struct ICEBERG_REST_EXPORT HttpClientOptions { /// `rest.client.connection-timeout-ms`; Java defaults to three minutes. /// Zero disables the configured connection deadline. - std::optional connection_timeout_ms = 180000; + std::optional connection_timeout_ms = kDefaultConnectionTimeoutMs; /// `rest.client.socket-timeout-ms`, the maximum transfer inactivity. /// Unset or zero leaves it unbounded. std::optional socket_timeout_ms; @@ -102,11 +103,11 @@ class ICEBERG_REST_EXPORT HttpClient { HttpClientOptions options = {}, std::optional base_uri = std::nullopt); /// \brief Construct with a connection timeout and a legacy total request timeout. /// \param connection_timeout,socket_timeout Zero leaves requests unbounded. - explicit HttpClient( - std::unordered_map default_headers = {}, - std::chrono::milliseconds connection_timeout = std::chrono::minutes(3), - std::chrono::milliseconds socket_timeout = {}, - std::optional base_uri = std::nullopt); + explicit HttpClient(std::unordered_map default_headers = {}, + std::chrono::milliseconds connection_timeout = + std::chrono::milliseconds(kDefaultConnectionTimeoutMs), + std::chrono::milliseconds socket_timeout = {}, + std::optional base_uri = std::nullopt); virtual ~HttpClient(); HttpClient(const HttpClient&) = delete; @@ -163,7 +164,7 @@ class ICEBERG_REST_EXPORT HttpClient { std::optional base_uri_; HttpClientOptions options_; HttpHeaders default_headers_; - std::chrono::milliseconds socket_timeout_{}; + std::chrono::milliseconds request_timeout_{}; std::unique_ptr connection_pool_; }; diff --git a/src/iceberg/catalog/rest/http_client_internal.h b/src/iceberg/catalog/rest/http_client_internal.h index 7e3c09e427..00e8951377 100644 --- a/src/iceberg/catalog/rest/http_client_internal.h +++ b/src/iceberg/catalog/rest/http_client_internal.h @@ -22,26 +22,46 @@ #include #include #include +#include #include +#include + #include "iceberg/catalog/rest/http_client.h" #include "iceberg/catalog/rest/iceberg_rest_export.h" -namespace cpr { -class Response; -} // namespace cpr - namespace iceberg::rest::detail { ICEBERG_REST_EXPORT bool ShouldRetry( const cpr::Response& response, const HttpRequest& request, int64_t exec_count, const HttpClientOptions& options, - std::optional elapsed = std::nullopt); + std::optional elapsed = std::nullopt, + std::chrono::system_clock::time_point now = std::chrono::system_clock::now()); ICEBERG_REST_EXPORT std::chrono::milliseconds RetryInterval( const cpr::Response& response, int64_t exec_count, std::chrono::system_clock::time_point now = std::chrono::system_clock::now()); +/// Internal timing seam for testing the retry execution loop without real waits. +struct ICEBERG_REST_EXPORT RetryTiming { + RetryTiming(); + std::function now; + std::function sleep; +}; + +struct Attempt { + cpr::Response response; + bool retry_approved = false; +}; + +using SendAttempt = + std::function(int64_t, std::chrono::system_clock::time_point)>; + +ICEBERG_REST_EXPORT Result SendWithRetry(const HttpClientOptions& options, + const HttpRequest& request, + const SendAttempt& send, + const RetryTiming& timing = {}); + /// Tracks transfer inactivity independently of the average byte rate. class ICEBERG_REST_EXPORT SocketIdleTimeout { public: diff --git a/src/iceberg/test/http_client_test.cc b/src/iceberg/test/http_client_test.cc index 6c18ae1855..b0f20a9916 100644 --- a/src/iceberg/test/http_client_test.cc +++ b/src/iceberg/test/http_client_test.cc @@ -49,6 +49,7 @@ #endif #include "iceberg/catalog/rest/auth/auth_session.h" +#include "iceberg/catalog/rest/catalog_properties.h" #include "iceberg/catalog/rest/error_handlers.h" #include "iceberg/catalog/rest/http_client_internal.h" #include "iceberg/test/matchers.h" @@ -60,7 +61,7 @@ namespace iceberg::rest { namespace { /// \brief Minimal HTTP/1.1 server on 127.0.0.1 that answers every request with -/// a fixed response and closes the connection. +/// fixed responses, optionally keeping the connection open for the next request. class LoopbackHttpServer { public: explicit LoopbackHttpServer(std::string response, @@ -71,8 +72,10 @@ class LoopbackHttpServer { header_byte_interval_(header_byte_interval) {} explicit LoopbackHttpServer(std::vector responses, - bool wait_for_client_close = false) - : responses_(std::move(responses)), wait_for_client_close_(wait_for_client_close) {} + bool wait_for_client_close = false, bool keep_alive = false) + : responses_(std::move(responses)), + wait_for_client_close_(wait_for_client_close), + keep_alive_(keep_alive) {} ~LoopbackHttpServer() { stop_ = true; @@ -115,6 +118,7 @@ class LoopbackHttpServer { int port() const { return port_; } int request_count() const { return request_count_.load(); } + int connection_count() const { return connection_count_.load(); } std::vector requests() const { std::lock_guard lock(requests_mutex_); return requests_; @@ -134,39 +138,45 @@ class LoopbackHttpServer { if (conn < 0) { continue; } - auto request = ReadRequest(conn); - { - std::lock_guard lock(requests_mutex_); - requests_.push_back(std::move(request)); - } - const auto index = static_cast(request_count_.fetch_add(1)); - const auto& response = responses_.empty() - ? response_ - : responses_[std::min(index, responses_.size() - 1)]; - if (header_byte_interval_ > std::chrono::milliseconds::zero()) { - const auto line_end = response.find("\r\n") + 2; - const auto body_start = response.find("\r\n\r\n") + 4; - SendAll(conn, response.substr(0, line_end)); - for (size_t i = line_end; i < body_start && !stop_; ++i) { - std::this_thread::sleep_for(header_byte_interval_); - SendAll(conn, response.substr(i, 1)); + ++connection_count_; + do { + auto request = ReadRequest(conn); + if (request.empty()) break; + { + std::lock_guard lock(requests_mutex_); + requests_.push_back(std::move(request)); } - if (!stop_) SendAll(conn, response.substr(body_start)); - } else if (body_byte_interval_ == std::chrono::milliseconds::zero()) { - SendAll(conn, response); - } else { - const auto body_start = response.find("\r\n\r\n") + 4; - SendAll(conn, response.substr(0, body_start)); - for (size_t i = body_start; i < response.size() && !stop_; ++i) { - std::this_thread::sleep_for(body_byte_interval_); - SendAll(conn, response.substr(i, 1)); + const auto index = static_cast(request_count_.fetch_add(1)); + const auto& response = responses_.empty() + ? response_ + : responses_[std::min(index, responses_.size() - 1)]; + if (header_byte_interval_ > std::chrono::milliseconds::zero()) { + const auto line_end = response.find("\r\n") + 2; + // Fragment one complete header line. Its total transfer time can exceed + // the idle timeout without delaying every header in the response. + const auto header_end = response.find("\r\n", line_end) + 2; + SendAll(conn, response.substr(0, line_end)); + for (size_t i = line_end; i < header_end && !stop_; ++i) { + std::this_thread::sleep_for(header_byte_interval_); + SendAll(conn, response.substr(i, 1)); + } + if (!stop_) SendAll(conn, response.substr(header_end)); + } else if (body_byte_interval_ == std::chrono::milliseconds::zero()) { + SendAll(conn, response); + } else { + const auto body_start = response.find("\r\n\r\n") + 4; + SendAll(conn, response.substr(0, body_start)); + for (size_t i = body_start; i < response.size() && !stop_; ++i) { + std::this_thread::sleep_for(body_byte_interval_); + SendAll(conn, response.substr(i, 1)); + } } - } - if (wait_for_client_close_) { - std::array buffer{}; - while (recv(conn, buffer.data(), buffer.size(), 0) > 0) { + if (wait_for_client_close_) { + std::array buffer{}; + while (recv(conn, buffer.data(), buffer.size(), 0) > 0) { + } } - } + } while (keep_alive_ && !stop_); close(conn); } } @@ -236,10 +246,12 @@ class LoopbackHttpServer { std::chrono::milliseconds header_byte_interval_{}; std::vector responses_; bool wait_for_client_close_ = false; + bool keep_alive_ = false; int listen_fd_ = -1; int port_ = 0; std::atomic stop_{false}; std::atomic request_count_{0}; + std::atomic connection_count_{0}; mutable std::mutex requests_mutex_; std::vector requests_; std::thread thread_; @@ -427,7 +439,7 @@ TEST(HttpClientTest, SocketTimeoutAllowsAHeaderArrivingInFragments) { /*header_byte_interval=*/std::chrono::milliseconds(40)); ASSERT_TRUE(server.Start()); auto client = - HttpClient::Make({}, HttpClientOptions{.socket_timeout_ms = 250, .max_retries = 1}); + HttpClient::Make({}, HttpClientOptions{.socket_timeout_ms = 500, .max_retries = 1}); CountingAuthSession session; ICEBERG_UNWRAP_OR_FAIL(auto response, client->Get(server.url("/v1/config"), {}, {}, @@ -438,14 +450,14 @@ TEST(HttpClientTest, SocketTimeoutAllowsAHeaderArrivingInFragments) { } TEST(HttpClientTest, SocketTimeoutAllowsSlowContinuousProgress) { - // Below one byte per second, but every gap is shorter than the idle timeout. + // The response takes longer than the idle timeout, but each gap is shorter. LoopbackHttpServer server( "HTTP/1.1 200 OK\r\nContent-Length: 4\r\n" "Connection: close\r\n\r\nnull", - std::chrono::milliseconds(1200)); + std::chrono::milliseconds(300)); ASSERT_TRUE(server.Start()); auto client = HttpClient::Make( - {}, HttpClientOptions{.socket_timeout_ms = 2000, .max_retries = 0}); + {}, HttpClientOptions{.socket_timeout_ms = 1000, .max_retries = 0}); auto session = auth::AuthSession::MakeDefault({}); ICEBERG_UNWRAP_OR_FAIL(auto response, client->Get(server.url("/v1/config"), {}, {}, @@ -469,6 +481,28 @@ TEST(HttpClientTest, DoesNotRetryTruncatedGetResponse) { EXPECT_EQ(server.request_count(), 1); } +TEST(HttpClientTest, ReusesConnectionsAcrossRequestsWithFreshMultiHandles) { + for (const int32_t socket_timeout : {0, 2000}) { + SCOPED_TRACE(socket_timeout); + const std::string response = "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n{}"; + LoopbackHttpServer server(std::vector{response, response}, + /*wait_for_client_close=*/false, /*keep_alive=*/true); + ASSERT_TRUE(server.Start()); + auto client = HttpClient::Make({}, {.connection_timeout_ms = 5000, + .socket_timeout_ms = socket_timeout, + .max_retries = 0}); + auto session = auth::AuthSession::MakeDefault({}); + for (int request = 0; request < 2; ++request) { + ICEBERG_UNWRAP_OR_FAIL(auto result, + client->Get(server.url("/v1/config"), {}, {}, + *DefaultErrorHandler::Instance(), *session)); + EXPECT_EQ(result.body(), "{}"); + } + EXPECT_EQ(server.request_count(), 2); + EXPECT_EQ(server.connection_count(), 1); + } +} + TEST(HttpClientTest, Retries503AtHeadersBeforeReadingTheBody) { LoopbackHttpServer server( std::vector{"HTTP/1.1 503 Service Unavailable\r\nRetry-After: 1\r\n" @@ -842,6 +876,213 @@ TEST(HttpClientRetryTest, ExponentialBackoffIsCapped) { } } +class HttpClientRetryExecutionTest : public ::testing::Test { + protected: + HttpClientRetryExecutionTest() { + timing_.now = [&] { return std::chrono::system_clock::time_point{} + elapsed_; }; + timing_.sleep = [&](std::chrono::milliseconds delay) { + sleeps_.push_back(delay); + elapsed_ += delay; + }; + } + + static detail::Attempt Reply(int status, const std::string& retry_after = {}) { + cpr::Response response; + response.status_code = status; + if (!retry_after.empty()) response.header["Retry-After"] = retry_after; + return {.response = std::move(response)}; + } + + std::chrono::milliseconds elapsed_{}; + std::vector sleeps_; + detail::RetryTiming timing_; +}; + +TEST_F(HttpClientRetryExecutionTest, UnkeyedRequestsHonorRetryAfterWithoutCap) { + const std::pair cases[] = { + {"3600", std::chrono::hours(1)}, + {"9999999999", std::chrono::milliseconds(9999999999000LL)}, + {"9223372036854775807", std::chrono::milliseconds::max()}, + {"Thu, 01 Jan 1970 01:00:00 GMT", std::chrono::hours(1)}}; + // Record even the largest durations without advancing a nanosecond clock + // beyond its representable range or actually sleeping. + timing_.sleep = [&](std::chrono::milliseconds delay) { sleeps_.push_back(delay); }; + for (const auto& [value, delay] : cases) { + SCOPED_TRACE(value); + sleeps_.clear(); + int attempts = 0; + ICEBERG_UNWRAP_OR_FAIL(auto response, + detail::SendWithRetry( + {}, {}, + [&](int64_t, auto) -> Result { + return Reply(++attempts == 1 ? 429 : 200, value); + }, + timing_)); + EXPECT_EQ(response.status_code, 200); + EXPECT_EQ(attempts, 2); + EXPECT_THAT(sleeps_, ::testing::ElementsAre(delay)); + } +} + +TEST_F(HttpClientRetryExecutionTest, KeyLifetimeRejectsLongRetryAfterBeforeWaiting) { + const HttpRequest request{.method = HttpMethod::kPost, + .headers = {{"Idempotency-Key", "key"}}}; + int attempts = 0; + ICEBERG_UNWRAP_OR_FAIL( + auto response, detail::SendWithRetry( + {.idempotency_key_lifetime = std::chrono::seconds(1)}, request, + [&](int64_t, auto) -> Result { + ++attempts; + auto attempt = Reply(429, "3600"); + attempt.response.text = "server error"; + return attempt; + }, + timing_)); + EXPECT_EQ(response.status_code, 429); + EXPECT_EQ(response.text, "server error"); + EXPECT_EQ(attempts, 1); + EXPECT_TRUE(sleeps_.empty()); +} + +TEST_F(HttpClientRetryExecutionTest, KeyLifetimeIncludesAttemptsAndAllPriorWaits) { + const HttpRequest request{.headers = {{"Idempotency-Key", "key"}}}; + int attempts = 0; + ICEBERG_UNWRAP_OR_FAIL( + auto response, + detail::SendWithRetry( + {.max_retries = 10, + .idempotency_key_lifetime = std::chrono::milliseconds(2500)}, + request, + [&](int64_t count, auto first_attempt) -> Result { + EXPECT_EQ(count, ++attempts); + EXPECT_EQ(first_attempt, std::chrono::system_clock::time_point{}); + elapsed_ += std::chrono::milliseconds(100); + return Reply(503, "1"); + }, + timing_)); + EXPECT_EQ(response.status_code, 503); + EXPECT_EQ(attempts, 3); + EXPECT_THAT(sleeps_, + ::testing::ElementsAre(std::chrono::seconds(1), std::chrono::seconds(1))); +} + +TEST_F(HttpClientRetryExecutionTest, RespectsRetryCountWithBackoffAndJitter) { + int attempts = 0; + ICEBERG_UNWRAP_OR_FAIL(auto response, + detail::SendWithRetry( + {.max_retries = 2}, {}, + [&](int64_t, auto) -> Result { + ++attempts; + return Reply(500); + }, + timing_)); + EXPECT_EQ(response.status_code, 500); + EXPECT_EQ(attempts, 3); + ASSERT_EQ(sleeps_.size(), 2); + EXPECT_GE(sleeps_[0], std::chrono::milliseconds(1000)); + EXPECT_LT(sleeps_[0], std::chrono::milliseconds(1100)); + EXPECT_GE(sleeps_[1], std::chrono::milliseconds(2000)); + EXPECT_LT(sleeps_[1], std::chrono::milliseconds(2200)); +} + +TEST_F(HttpClientRetryExecutionTest, IoRetriesHaveNoWait) { + int attempts = 0; + ICEBERG_UNWRAP_OR_FAIL(auto response, + detail::SendWithRetry( + {.max_retries = 2}, {}, + [&](int64_t, auto) -> Result { + if (++attempts == 3) return Reply(200); + auto attempt = Reply(0); + attempt.response.error.code = cpr::ErrorCode::RECV_ERROR; + return attempt; + }, + timing_)); + EXPECT_EQ(response.status_code, 200); + EXPECT_EQ(attempts, 3); + EXPECT_TRUE(sleeps_.empty()); +} + +TEST_F(HttpClientRetryExecutionTest, ApprovedRetryStillRunsAfterSleepOvershootsLifetime) { + const HttpRequest request{.headers = {{"Idempotency-Key", "key"}}}; + timing_.sleep = [&](std::chrono::milliseconds delay) { + sleeps_.push_back(delay); + elapsed_ += std::chrono::seconds(2); + }; + int attempts = 0; + ICEBERG_UNWRAP_OR_FAIL( + auto response, detail::SendWithRetry( + {.idempotency_key_lifetime = std::chrono::seconds(2)}, request, + [&](int64_t, auto) -> Result { + return Reply(++attempts == 1 ? 429 : 200, "1"); + }, + timing_)); + // Java checks the lifetime when approving a retry, not again after sleeping. + EXPECT_EQ(response.status_code, 200); + EXPECT_EQ(attempts, 2); + EXPECT_THAT(sleeps_, ::testing::ElementsAre(std::chrono::seconds(1))); +} + +TEST_F(HttpClientRetryExecutionTest, PreservesRetryApprovedAtResponseHeaders) { + const HttpRequest request{.headers = {{"Idempotency-Key", "key"}}}; + int attempts = 0; + ICEBERG_UNWRAP_OR_FAIL( + auto response, detail::SendWithRetry( + {.idempotency_key_lifetime = std::chrono::seconds(2)}, request, + [&](int64_t, auto) -> Result { + if (++attempts == 2) return Reply(200); + auto attempt = Reply(429, "1"); + // The retry was approved before closing the current response + // took time. + attempt.retry_approved = true; + elapsed_ += std::chrono::milliseconds(1500); + return attempt; + }, + timing_)); + EXPECT_EQ(response.status_code, 200); + EXPECT_EQ(attempts, 2); + EXPECT_THAT(sleeps_, ::testing::ElementsAre(std::chrono::seconds(1))); +} + +TEST_F(HttpClientRetryExecutionTest, KeyLifetimeAccountsForLargestPossibleJitter) { + const HttpRequest request{.headers = {{"Idempotency-Key", "key"}}}; + for (const int64_t lifetime : {1050, 1101}) { + SCOPED_TRACE(lifetime); + elapsed_ = std::chrono::milliseconds::zero(); + sleeps_.clear(); + int attempts = 0; + ICEBERG_UNWRAP_OR_FAIL( + auto response, + detail::SendWithRetry( + {.idempotency_key_lifetime = std::chrono::milliseconds(lifetime)}, request, + [&](int64_t, auto) -> Result { + return Reply(++attempts == 1 ? 500 : 200); + }, + timing_)); + EXPECT_EQ(response.status_code, lifetime == 1050 ? 500 : 200); + EXPECT_EQ(attempts, lifetime == 1050 ? 1 : 2); + EXPECT_EQ(sleeps_.size(), lifetime == 1050 ? 0 : 1); + } +} + +TEST_F(HttpClientRetryExecutionTest, IoFailureCannotRetryAfterKeyLifetimeExpires) { + const HttpRequest request{.headers = {{"Idempotency-Key", "key"}}}; + int attempts = 0; + ICEBERG_UNWRAP_OR_FAIL( + auto response, detail::SendWithRetry( + {.idempotency_key_lifetime = std::chrono::seconds(1)}, request, + [&](int64_t, auto) -> Result { + ++attempts; + elapsed_ += std::chrono::seconds(1); + auto attempt = Reply(0); + attempt.response.error.code = cpr::ErrorCode::RECV_ERROR; + return attempt; + }, + timing_)); + EXPECT_EQ(response.error.code, cpr::ErrorCode::RECV_ERROR); + EXPECT_EQ(attempts, 1); + EXPECT_TRUE(sleeps_.empty()); +} + TEST(HttpClientOptionsTest, SocketTimeoutAcceptsMaximumMilliseconds) { ICEBERG_UNWRAP_OR_FAIL(auto options, HttpClientOptions::FromProperties( @@ -888,6 +1129,9 @@ TEST(HttpClientOptionsTest, DefaultsMatchJavaClient) { EXPECT_EQ(options.max_retries, 5); EXPECT_FALSE(options.user_agent.has_value()); EXPECT_FALSE(options.idempotency_key_lifetime.has_value()); + const RestCatalogProperties properties; + EXPECT_EQ(options.connection_timeout_ms, + properties.Get(RestCatalogProperties::kClientConnectionTimeoutMs)); } TEST(HttpClientOptionsTest, AppliesTransportProperties) { From 7eb8420b38b8647145db1799e390ed021cafc3fe Mon Sep 17 00:00:00 2001 From: Zhao Junwang Date: Sun, 1 Nov 2026 23:40:44 +0800 Subject: [PATCH 3/5] fix(rest): resolve Windows macro conflicts and CI lint errors Protect min/max calls from Windows SDK macros. Fix clang-tidy findings in transport code and retry tests while retaining libcurl native integer types and Java retry behavior. --- src/iceberg/catalog/rest/http_client.cc | 66 ++++++------ src/iceberg/test/http_client_test.cc | 137 ++++++++++++++++-------- 2 files changed, 127 insertions(+), 76 deletions(-) diff --git a/src/iceberg/catalog/rest/http_client.cc b/src/iceberg/catalog/rest/http_client.cc index e9263827f6..dd11d10ced 100644 --- a/src/iceberg/catalog/rest/http_client.cc +++ b/src/iceberg/catalog/rest/http_client.cc @@ -20,6 +20,7 @@ #include "iceberg/catalog/rest/http_client.h" #include +#include #include #include #include @@ -80,8 +81,8 @@ Status ParseIntegerProperty( } bool CheckedAdd(int64_t left, int64_t right, int64_t& result) { - if ((right > 0 && left > std::numeric_limits::max() - right) || - (right < 0 && left < std::numeric_limits::min() - right)) { + if ((right > 0 && left > (std::numeric_limits::max)() - right) || + (right < 0 && left < (std::numeric_limits::min)() - right)) { return false; } result = left + right; @@ -89,8 +90,8 @@ bool CheckedAdd(int64_t left, int64_t right, int64_t& result) { } bool CheckedSubtract(int64_t left, int64_t right, int64_t& result) { - if ((right > 0 && left < std::numeric_limits::min() + right) || - (right < 0 && left > std::numeric_limits::max() + right)) { + if ((right > 0 && left < (std::numeric_limits::min)() + right) || + (right < 0 && left > (std::numeric_limits::max)() + right)) { return false; } result = left - right; @@ -112,8 +113,8 @@ std::optional ParseKeyLifetime(const std::string& val !has_time)) { return std::nullopt; } - constexpr int64_t multipliers[] = {86400, 3600, 60, 1}; - int64_t components[4]{}; + constexpr std::array multipliers = {86400, 3600, 60, 1}; + std::array components{}; for (size_t i = 0; i < 4; ++i) { if (!match[i + 2].matched) continue; const auto text = match[i + 2].str(); @@ -122,8 +123,8 @@ std::optional ParseKeyLifetime(const std::string& val int64_t number{}; auto [end, ec] = std::from_chars(view.data(), view.data() + view.size(), number); if (ec != std::errc{} || end != view.data() + view.size() || - number > std::numeric_limits::max() / multipliers[i] || - number < std::numeric_limits::min() / multipliers[i]) { + number > (std::numeric_limits::max)() / multipliers[i] || + number < (std::numeric_limits::min)() / multipliers[i]) { return std::nullopt; } components[i] = number * multipliers[i]; @@ -145,7 +146,7 @@ std::optional ParseKeyLifetime(const std::string& val } if (match[1].str() == "-") { if (nanos == 0) { - if (seconds == std::numeric_limits::min()) return std::nullopt; + if (seconds == (std::numeric_limits::min)()) return std::nullopt; seconds = -seconds; } else { seconds = seconds >= 0 ? -seconds - 1 : -(seconds + 1); @@ -155,7 +156,7 @@ std::optional ParseKeyLifetime(const std::string& val if (seconds < 0 || (seconds == 0 && nanos == 0)) { return std::chrono::milliseconds::zero(); } - constexpr auto kMaxMillis = std::chrono::milliseconds::max().count(); + constexpr auto kMaxMillis = (std::chrono::milliseconds::max)().count(); if (seconds > kMaxMillis / 1000 || (seconds == kMaxMillis / 1000 && nanos > (kMaxMillis % 1000) * 1000000)) { return std::nullopt; @@ -284,8 +285,8 @@ std::optional RetryAfter( auto [end, ec] = std::from_chars(value.data(), value.data() + value.size(), seconds); if (ec == std::errc{} && end == value.data() + value.size()) { if (seconds <= 0) return std::nullopt; - constexpr auto kMaxSeconds = std::chrono::milliseconds::max().count() / 1000; - if (seconds > kMaxSeconds) return std::chrono::milliseconds::max(); + constexpr auto kMaxSeconds = (std::chrono::milliseconds::max)().count() / 1000; + if (seconds > kMaxSeconds) return (std::chrono::milliseconds::max)(); return std::chrono::milliseconds(seconds * 1000); } const auto date = curl_getdate(it->second.c_str(), nullptr); @@ -294,13 +295,13 @@ std::optional RetryAfter( std::chrono::duration_cast(now.time_since_epoch()) .count(); const auto date_seconds = static_cast(date); - if (date_seconds > std::chrono::milliseconds::max().count() / 1000) { - return std::chrono::milliseconds::max(); + if (date_seconds > (std::chrono::milliseconds::max)().count() / 1000) { + return (std::chrono::milliseconds::max)(); } const auto date_ms = date_seconds * 1000; if (date_ms <= now_ms) return std::nullopt; int64_t delay; - if (!CheckedSubtract(date_ms, now_ms, delay)) return std::chrono::milliseconds::max(); + if (!CheckedSubtract(date_ms, now_ms, delay)) return (std::chrono::milliseconds::max)(); return std::chrono::milliseconds(delay); } @@ -319,22 +320,24 @@ bool ShouldRetry(const cpr::Response& response, const HttpRequest& request, // Java reads the final response entity outside HttpRequestRetryExec. A body // read failure is propagated, even for a successful status or keyed request. if (response.status_code != 0 || !safe || - !IsRetriableTransportError(response.error.code)) + !IsRetriableTransportError(response.error.code)) { return false; + } } else { const auto code = response.status_code; if (code != 429 && !(code == 503 && response.header.contains(std::string(kHeaderRetryAfter))) && !(safe && - (code == 408 || code == 500 || code == 502 || code == 503 || code == 504))) + (code == 408 || code == 500 || code == 502 || code == 503 || code == 504))) { return false; + } } if (has_key && options.idempotency_key_lifetime.has_value() && elapsed.has_value()) { const auto lifetime = options.idempotency_key_lifetime->count(); if (elapsed->count() >= lifetime) return false; int64_t remaining; if (!CheckedSubtract(lifetime, elapsed->count(), remaining)) { - remaining = std::numeric_limits::max(); + remaining = (std::numeric_limits::max)(); } auto upper_delay = Backoff(exec_count); if (!response.error) { @@ -401,9 +404,9 @@ bool SocketIdleTimeout::Check(int64_t downloaded, int64_t uploaded, int64_t head std::optional SocketIdleTimeout::Remaining( Clock::time_point now) const { if (!last_progress_.has_value()) return std::nullopt; - return std::max( - std::chrono::milliseconds::zero(), - std::chrono::ceil(timeout_ - (now - *last_progress_))); + return (std::max)(std::chrono::milliseconds::zero(), + std::chrono::ceil( + timeout_ - (now - *last_progress_))); } } // namespace detail @@ -415,9 +418,11 @@ void ApplyOptions(cpr::Session& session, const HttpClientOptions& options) { options.connection_timeout_ms.value_or(kDefaultConnectionTimeoutMs); // libcurl interprets zero as its own five-minute default. Use its largest // native deadline when Java's disabled connection timeout is requested. + // Its native long timeout is 32-bit on Windows, so int64_t is not equivalent. + constexpr auto kMaxCurlTimeoutMs = + (std::numeric_limits::max)(); // NOLINT(google-runtime-int) session.SetConnectTimeout(cpr::ConnectTimeout{std::chrono::milliseconds( - timeout == 0 ? std::numeric_limits::max() - : std::min(timeout, std::numeric_limits::max()))}); + timeout == 0 ? kMaxCurlTimeoutMs : std::min(timeout, kMaxCurlTimeoutMs))}); } // Drive transfers synchronously through libcurl's socket API. Observe socket @@ -481,26 +486,27 @@ Result PerformSession(cpr::Session& session, break; } auto wait_ms = std::chrono::milliseconds(1000); - long curl_timeout = -1; + // curl_multi_timeout requires a native long pointer, also on Windows. + long curl_timeout = -1; // NOLINT(google-runtime-int) if (const auto status = curl_multi_timeout(multi.get(), &curl_timeout); status != CURLM_OK) { return IOError("HTTP timer query failed: {}", curl_multi_strerror(status)); } if (curl_timeout >= 0) { - wait_ms = std::min(wait_ms, std::chrono::milliseconds(curl_timeout)); + wait_ms = (std::min)(wait_ms, std::chrono::milliseconds(curl_timeout)); } if (idle_timeout != nullptr) { if (auto remaining = idle_timeout->Remaining()) { - wait_ms = std::min(wait_ms, *remaining); + wait_ms = (std::min)(wait_ms, *remaining); } } std::vector watched; watched.reserve(sockets.size()); for (const auto& [socket, action] : sockets) { - short events = 0; - if (action & CURL_POLL_IN) events |= CURL_WAIT_POLLIN; - if (action & CURL_POLL_OUT) events |= CURL_WAIT_POLLOUT; - if (events != 0) watched.push_back({socket, events, 0}); + curl_waitfd watched_socket{.fd = socket, .events = 0, .revents = 0}; + if (action & CURL_POLL_IN) watched_socket.events |= CURL_WAIT_POLLIN; + if (action & CURL_POLL_OUT) watched_socket.events |= CURL_WAIT_POLLOUT; + if (watched_socket.events != 0) watched.push_back(watched_socket); } int activity = 0; #if LIBCURL_VERSION_NUM >= 0x074200 diff --git a/src/iceberg/test/http_client_test.cc b/src/iceberg/test/http_client_test.cc index b0f20a9916..b387810f10 100644 --- a/src/iceberg/test/http_client_test.cc +++ b/src/iceberg/test/http_client_test.cc @@ -149,7 +149,7 @@ class LoopbackHttpServer { const auto index = static_cast(request_count_.fetch_add(1)); const auto& response = responses_.empty() ? response_ - : responses_[std::min(index, responses_.size() - 1)]; + : responses_[(std::min)(index, responses_.size() - 1)]; if (header_byte_interval_ > std::chrono::milliseconds::zero()) { const auto line_end = response.find("\r\n") + 2; // Fragment one complete header line. Its total transfer time can exceed @@ -726,25 +726,61 @@ TEST(HttpClientTest, DoesNotFollowRedirects) { TEST(HttpClientRetryTest, TransportRetriesMatchTheJavaExceptionBoundary) { struct Case { - HttpMethod method; int status; cpr::ErrorCode error; + HttpMethod method; bool retry; }; - const Case cases[] = { - {HttpMethod::kGet, 0, cpr::ErrorCode::RECV_ERROR, true}, - {HttpMethod::kHead, 0, cpr::ErrorCode::GOT_NOTHING, true}, - {HttpMethod::kDelete, 0, cpr::ErrorCode::SEND_ERROR, true}, - {HttpMethod::kGet, 0, cpr::ErrorCode::OPERATION_TIMEDOUT, false}, - {HttpMethod::kGet, 0, cpr::ErrorCode::SSL_CONNECT_ERROR, false}, - {HttpMethod::kGet, 0, cpr::ErrorCode::COULDNT_CONNECT, false}, - {HttpMethod::kGet, 0, cpr::ErrorCode::PARTIAL_FILE, false}, - {HttpMethod::kGet, 0, cpr::ErrorCode::WEIRD_SERVER_REPLY, false}, - {HttpMethod::kGet, 200, cpr::ErrorCode::PARTIAL_FILE, false}, - {HttpMethod::kGet, 200, cpr::ErrorCode::RECV_ERROR, false}, - {HttpMethod::kPost, 0, cpr::ErrorCode::RECV_ERROR, false}, - {HttpMethod::kPost, 200, cpr::ErrorCode::PARTIAL_FILE, false}, - }; + const std::array cases{{ + {.status = 0, + .error = cpr::ErrorCode::RECV_ERROR, + .method = HttpMethod::kGet, + .retry = true}, + {.status = 0, + .error = cpr::ErrorCode::GOT_NOTHING, + .method = HttpMethod::kHead, + .retry = true}, + {.status = 0, + .error = cpr::ErrorCode::SEND_ERROR, + .method = HttpMethod::kDelete, + .retry = true}, + {.status = 0, + .error = cpr::ErrorCode::OPERATION_TIMEDOUT, + .method = HttpMethod::kGet, + .retry = false}, + {.status = 0, + .error = cpr::ErrorCode::SSL_CONNECT_ERROR, + .method = HttpMethod::kGet, + .retry = false}, + {.status = 0, + .error = cpr::ErrorCode::COULDNT_CONNECT, + .method = HttpMethod::kGet, + .retry = false}, + {.status = 0, + .error = cpr::ErrorCode::PARTIAL_FILE, + .method = HttpMethod::kGet, + .retry = false}, + {.status = 0, + .error = cpr::ErrorCode::WEIRD_SERVER_REPLY, + .method = HttpMethod::kGet, + .retry = false}, + {.status = 200, + .error = cpr::ErrorCode::PARTIAL_FILE, + .method = HttpMethod::kGet, + .retry = false}, + {.status = 200, + .error = cpr::ErrorCode::RECV_ERROR, + .method = HttpMethod::kGet, + .retry = false}, + {.status = 0, + .error = cpr::ErrorCode::RECV_ERROR, + .method = HttpMethod::kPost, + .retry = false}, + {.status = 200, + .error = cpr::ErrorCode::PARTIAL_FILE, + .method = HttpMethod::kPost, + .retry = false}, + }}; for (const auto& test_case : cases) { SCOPED_TRACE(std::format("method={}, status={}, error={}", static_cast(test_case.method), test_case.status, @@ -787,33 +823,41 @@ TEST(HttpClientRetryTest, KeyedNetworkFailuresRetryWithoutAnExceptionDelay) { TEST(HttpClientRetryTest, KeyLifetimeMatchesJavaBudgetChecks) { struct Case { std::chrono::milliseconds lifetime; - std::chrono::milliseconds elapsed; - bool has_key; + std::chrono::milliseconds elapsed{}; std::string retry_after; - int attempt; - bool transport_error; - bool retry; - }; - const Case cases[] = { - {std::chrono::seconds(1), {}, true, "3600", 1, false, false}, - {std::chrono::hours(1), {}, true, "1", 1, false, true}, - {std::chrono::milliseconds(1), {}, true, "", 1, true, false}, - {std::chrono::milliseconds(1), {}, false, "3600", 1, false, true}, - {std::chrono::minutes(30), std::chrono::minutes(31), true, "1", 1, false, false}, - {std::chrono::seconds(2), std::chrono::milliseconds(1100), true, "1", 2, false, - false}, - {std::chrono::milliseconds(1), {}, true, "", 1, false, false}, - {{}, {}, true, "1", 1, false, false}, - {{}, {}, false, "1", 1, false, true}, - {std::chrono::milliseconds(1050), {}, true, "", 1, false, false}, - {std::chrono::milliseconds(1101), {}, true, "", 1, false, true}, + int attempt = 1; + bool has_key = true; + bool transport_error = false; + bool retry = false; }; + const std::array cases{{ + {.lifetime = std::chrono::seconds(1), .retry_after = "3600"}, + {.lifetime = std::chrono::hours(1), .retry_after = "1", .retry = true}, + {.lifetime = std::chrono::milliseconds(1), .transport_error = true}, + {.lifetime = std::chrono::milliseconds(1), + .retry_after = "3600", + .has_key = false, + .retry = true}, + {.lifetime = std::chrono::minutes(30), + .elapsed = std::chrono::minutes(31), + .retry_after = "1"}, + {.lifetime = std::chrono::seconds(2), + .elapsed = std::chrono::milliseconds(1100), + .retry_after = "1", + .attempt = 2}, + {.lifetime = std::chrono::milliseconds(1)}, + {.lifetime = {}, .retry_after = "1"}, + {.lifetime = {}, .retry_after = "1", .has_key = false, .retry = true}, + {.lifetime = std::chrono::milliseconds(1050)}, + {.lifetime = std::chrono::milliseconds(1101), .retry = true}, + }}; for (const auto& test_case : cases) { cpr::Response response; response.status_code = test_case.transport_error ? 0 : 500; if (test_case.transport_error) response.error.code = cpr::ErrorCode::RECV_ERROR; - if (!test_case.retry_after.empty()) + if (!test_case.retry_after.empty()) { response.header["Retry-After"] = test_case.retry_after; + } HttpRequest request{.method = test_case.has_key ? HttpMethod::kPost : HttpMethod::kGet}; if (test_case.has_key) request.headers["Idempotency-Key"] = "key"; @@ -848,7 +892,7 @@ TEST(HttpClientRetryTest, LargeRetryAfterValuesSaturateWithoutOverflow) { {"9223372036854776", "18446744073709552", "9223372036854775807"}) { cpr::Response response; response.header["Retry-After"] = value; - EXPECT_EQ(detail::RetryInterval(response, 1), std::chrono::milliseconds::max()); + EXPECT_EQ(detail::RetryInterval(response, 1), (std::chrono::milliseconds::max)()); } } @@ -869,7 +913,8 @@ TEST(HttpClientRetryTest, InvalidRetryAfterFallsBackWithoutOverflow) { TEST(HttpClientRetryTest, ExponentialBackoffIsCapped) { cpr::Response response; EXPECT_EQ(detail::RetryInterval(response, 0), std::chrono::milliseconds::zero()); - for (int64_t attempt : {int64_t{7}, int64_t{8}, std::numeric_limits::max()}) { + for (int64_t attempt : + {int64_t{7}, int64_t{8}, (std::numeric_limits::max)()}) { const auto delay = detail::RetryInterval(response, attempt); EXPECT_GE(delay, std::chrono::seconds(64)); EXPECT_LT(delay, std::chrono::milliseconds(70400)); @@ -899,11 +944,11 @@ class HttpClientRetryExecutionTest : public ::testing::Test { }; TEST_F(HttpClientRetryExecutionTest, UnkeyedRequestsHonorRetryAfterWithoutCap) { - const std::pair cases[] = { - {"3600", std::chrono::hours(1)}, - {"9999999999", std::chrono::milliseconds(9999999999000LL)}, - {"9223372036854775807", std::chrono::milliseconds::max()}, - {"Thu, 01 Jan 1970 01:00:00 GMT", std::chrono::hours(1)}}; + const std::array, 4> cases{ + {{"3600", std::chrono::hours(1)}, + {"9999999999", std::chrono::milliseconds(9999999999000LL)}, + {"9223372036854775807", (std::chrono::milliseconds::max)()}, + {"Thu, 01 Jan 1970 01:00:00 GMT", std::chrono::hours(1)}}}; // Record even the largest durations without advancing a nanosecond clock // beyond its representable range or actually sleeping. timing_.sleep = [&](std::chrono::milliseconds delay) { sleeps_.push_back(delay); }; @@ -1179,7 +1224,7 @@ TEST(HttpClientOptionsTest, RejectsInvalidSettingsWithoutEchoingValues) { } TEST(HttpClientOptionsTest, ParsesJavaDurationSyntaxForKeyLifetime) { - const std::pair cases[] = { + const std::array, 14> cases{{ {"PT30M", 1800000}, {"PT0S", 0}, {"PT-5M", 0}, @@ -1193,8 +1238,8 @@ TEST(HttpClientOptionsTest, ParsesJavaDurationSyntaxForKeyLifetime) { {"-PT-6H+3M", 21420000}, {"+PT+1S", 1000}, {"PT1.S", 1000}, - {"PT9223372036854775.807S", std::numeric_limits::max()}, - }; + {"PT9223372036854775.807S", (std::numeric_limits::max)()}, + }}; for (const auto& [value, millis] : cases) { SCOPED_TRACE(value); ICEBERG_UNWRAP_OR_FAIL(auto options, From 4c67304f9eb5108683937b4a1311136f5d405917 Mon Sep 17 00:00:00 2001 From: Zhao Junwang Date: Sun, 1 Nov 2026 23:57:33 +0800 Subject: [PATCH 4/5] refactor(rest): separate OAuth response redaction from transport Restore the upstream token-response parser and keep the targeted validation-error fix and regression coverage in PR #1079. --- mkdocs/docs/rest-catalog.md | 2 +- src/iceberg/catalog/rest/auth/oauth2_util.cc | 9 +++------ 2 files changed, 4 insertions(+), 7 deletions(-) diff --git a/mkdocs/docs/rest-catalog.md b/mkdocs/docs/rest-catalog.md index ba838615b2..2fc2722645 100644 --- a/mkdocs/docs/rest-catalog.md +++ b/mkdocs/docs/rest-catalog.md @@ -74,7 +74,7 @@ logical mutation through per-request headers or an `AuthSession`. A static Errors are classified from the `code` reported in the error body, falling back to the HTTP status when the body cannot be parsed. Service diagnostics remain -available. Malformed OAuth token responses never echo response content. +available. Other Java transport options, including `rest.client.max-connections`, `rest.client.connections-per-route`, `rest.client.proxy.*` and diff --git a/src/iceberg/catalog/rest/auth/oauth2_util.cc b/src/iceberg/catalog/rest/auth/oauth2_util.cc index 9de28eaa06..d62ac7e1a0 100644 --- a/src/iceberg/catalog/rest/auth/oauth2_util.cc +++ b/src/iceberg/catalog/rest/auth/oauth2_util.cc @@ -49,12 +49,9 @@ constexpr std::string_view kAuthorizationHeader = "Authorization"; constexpr std::string_view kBearerPrefix = "Bearer "; Result ParseTokenResponse(const std::string& response_body) { - // Never propagate the underlying JSON errors: they echo the response body, - // which may contain token material. FromJson already validates the response. - auto json = FromJsonString(response_body); - if (!json) return InvalidArgument("Invalid OAuth2 token response"); - auto token_response = FromJson(*json); - if (!token_response) return InvalidArgument("Invalid OAuth2 token response"); + ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response_body)); + ICEBERG_ASSIGN_OR_RAISE(auto token_response, FromJson(json)); + ICEBERG_RETURN_UNEXPECTED(token_response.Validate()); return token_response; } From cfabc7d08cecb8a6a36154e91f11783324d5ef16 Mon Sep 17 00:00:00 2001 From: Zhao Junwang Date: Mon, 2 Nov 2026 00:02:43 +0800 Subject: [PATCH 5/5] fix(util): guard string comparison against Windows min macro --- src/iceberg/util/string_util.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/iceberg/util/string_util.h b/src/iceberg/util/string_util.h index f1431e49d4..16cde04c53 100644 --- a/src/iceberg/util/string_util.h +++ b/src/iceberg/util/string_util.h @@ -196,7 +196,7 @@ class ICEBERG_EXPORT StringUtils { // knows. static std::optional AsciiEqualsIgnoreCase(std::string_view a, std::string_view b) { - const size_t n = std::min(a.size(), b.size()); + const size_t n = (std::min)(a.size(), b.size()); for (size_t i = 0; i < n; ++i) { if (!IsAsciiByte(a[i]) || !IsAsciiByte(b[i])) { return std::nullopt;