From c6b2a805fc02b60839981bc83393f41258ec24eb Mon Sep 17 00:00:00 2001 From: Zhao Junwang Date: Sun, 13 Sep 2026 22:22:59 +0800 Subject: [PATCH] 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 | 722 ++++++++++++++++-- src/iceberg/catalog/rest/http_client.h | 52 +- .../catalog/rest/http_client_internal.h | 67 ++ src/iceberg/catalog/rest/rest_catalog.cc | 11 +- src/iceberg/test/http_client_test.cc | 596 ++++++++++++++- 9 files changed, 1427 insertions(+), 112 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 a1f205a605..8059d71284 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 718a5b7371..ed6891efcf 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,41 +652,169 @@ 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) - : default_headers_{std::move(default_headers)}, - connection_timeout_{connection_timeout}, - socket_timeout_{socket_timeout}, +std::shared_ptr HttpClient::Make( + std::unordered_map default_headers, + HttpClientOptions options) { + return std::shared_ptr( + new HttpClient(WithOptions{}, std::move(default_headers), std::move(options), {})); +} + +HttpClient::HttpClient(WithOptions, + std::unordered_map default_headers, + HttpClientOptions options, + std::chrono::milliseconds request_timeout) + : 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 and will fail, // even on an empty bodied request. 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) + : HttpClient(WithOptions{}, std::move(default_headers), + HttpClientOptions{.connection_timeout_ms = + std::max(0, connection_timeout.count())}, + socket_timeout) {} + HttpClient::~HttpClient() = default; -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 url, AppendQueryString(path, 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_}); +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; @@ -225,23 +822,21 @@ Result HttpClient::Get( 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 url, AppendQueryString(path, params)); + return Execute(HttpMethod::kGet, url, MergeHeaders(default_headers_, headers), "", + error_handler, session); +} + Result HttpClient::Post( const std::string& path, const std::string& body, const std::unordered_map& headers, const ErrorHandler& error_handler, auth::AuthSession& session) { - ICEBERG_ASSIGN_OR_RAISE( - auto authenticated, - AuthenticateRequest(session, HttpMethod::kPost, path, - 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, path, MergeHeaders(default_headers_, headers), body, + error_handler, session); } Result HttpClient::PostForm( @@ -259,37 +854,15 @@ 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, path, - 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, path, 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 authenticated, - AuthenticateRequest(session, HttpMethod::kHead, path, - 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, path, MergeHeaders(default_headers_, headers), "", + error_handler, session); } Result HttpClient::Delete( @@ -297,19 +870,8 @@ Result HttpClient::Delete( const std::unordered_map& headers, const ErrorHandler& error_handler, auth::AuthSession& session) { ICEBERG_ASSIGN_OR_RAISE(auto url, AppendQueryString(path, 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 a185064041..401b081ef8 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,13 +71,41 @@ 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 = {}); + /// \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 = {}); + explicit HttpClient( + std::unordered_map default_headers = {}, + std::chrono::milliseconds connection_timeout = std::chrono::minutes(3), + std::chrono::milliseconds socket_timeout = {}); virtual ~HttpClient(); HttpClient(const HttpClient&) = delete; @@ -115,9 +145,19 @@ class ICEBERG_REST_EXPORT HttpClient { const ErrorHandler& error_handler, auth::AuthSession& session); private: - std::unordered_map default_headers_; - std::chrono::milliseconds connection_timeout_; - std::chrono::milliseconds socket_timeout_; + struct WithOptions {}; + HttpClient(WithOptions, std::unordered_map default_headers, + HttpClientOptions options, std::chrono::milliseconds request_timeout); + + /// \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); + + 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 c8d88d06fb..42d92ab2b5 100644 --- a/src/iceberg/catalog/rest/rest_catalog.cc +++ b/src/iceberg/catalog/rest/rest_catalog.cc @@ -452,14 +452,9 @@ 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))); - }; - return std::make_shared( - properties.ExtractHeaders(), - timeout_of(RestCatalogProperties::kClientConnectionTimeoutMs), - timeout_of(RestCatalogProperties::kClientSocketTimeoutMs)); + ICEBERG_ASSIGN_OR_RAISE(auto options, + HttpClientOptions::FromProperties(properties.configs())); + return HttpClient::Make(properties.ExtractHeaders(), std::move(options)); }; } 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 9e5037299b..9cfc01eb78 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. @@ -361,4 +647,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