diff --git a/mkdocs/docs/rest-catalog.md b/mkdocs/docs/rest-catalog.md new file mode 100644 index 000000000..ba838615b --- /dev/null +++ b/mkdocs/docs/rest-catalog.md @@ -0,0 +1,83 @@ + + +# 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; 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. 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 +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. + +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. + +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 7cf0ec152..9ccbcc92b 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 d62ac7e1a..9de28eaa0 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 0b24af58c..fcb30058e 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", 0}; + "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 0a6e8d0c0..fc7ac1224 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 019f78209..e9263827f 100644 --- a/src/iceberg/catalog/rest/http_client.cc +++ b/src/iceberg/catalog/rest/http_client.cc @@ -19,19 +19,178 @@ #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/catalog_properties.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 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, 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)); + 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 +209,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 +230,314 @@ 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, + 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; + 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, 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)); +} + +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_ || + 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 { + +void ApplyOptions(cpr::Session& session, const HttpClientOptions& options) { + 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( + 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 +554,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 +615,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 +647,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 +655,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()}, + 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. 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 +721,141 @@ 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, + detail::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{request_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. + // Aborting before draining the body sacrifices connection reuse + // for this response; subsequent successful requests still reuse the pool. + 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 detail::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 +863,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 +883,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 +902,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 9a2459387..83c49425f 100644 --- a/src/iceberg/catalog/rest/http_client.h +++ b/src/iceberg/catalog/rest/http_client.h @@ -22,9 +22,12 @@ #include #include #include +#include #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" #include "iceberg/result.h" @@ -69,12 +72,40 @@ 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 = kDefaultConnectionTimeoutMs; + /// `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 connection_timeout = + std::chrono::milliseconds(kDefaultConnectionTimeoutMs), std::chrono::milliseconds socket_timeout = {}, std::optional base_uri = std::nullopt); virtual ~HttpClient(); @@ -116,13 +147,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 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 new file mode 100644 index 000000000..00e895137 --- /dev/null +++ b/src/iceberg/catalog/rest/http_client_internal.h @@ -0,0 +1,87 @@ +/* + * 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 + +#include + +#include "iceberg/catalog/rest/http_client.h" +#include "iceberg/catalog/rest/iceberg_rest_export.h" + +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::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: + 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 a25d28eb4..5319b4cad 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 6cc653131..b0f20a991 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 @@ -45,7 +49,9 @@ #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" namespace iceberg::rest { @@ -55,10 +61,21 @@ 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) : 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, bool keep_alive = false) + : responses_(std::move(responses)), + wait_for_client_close_(wait_for_client_close), + keep_alive_(keep_alive) {} ~LoopbackHttpServer() { stop_ = true; @@ -101,6 +118,11 @@ 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_; + } std::string url(std::string_view path) const { return std::format("http://127.0.0.1:{}{}", port_, path); } @@ -116,16 +138,52 @@ class LoopbackHttpServer { if (conn < 0) { continue; } - ReadRequest(conn); - ++request_count_; - SendAll(conn, response_); + ++connection_count_; + do { + auto request = ReadRequest(conn); + if (request.empty()) break; + { + 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; + // 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) { + } + } + } while (keep_alive_ && !stop_); 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 +195,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 +206,7 @@ class LoopbackHttpServer { content_length = ParseContentLength(request.substr(0, header_end)); } if (request.size() >= header_end + 4 + content_length) { - return; + return request; } } } @@ -184,13 +242,45 @@ 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; + 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_; }; +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 +310,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 +358,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 +376,239 @@ 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 = 500, .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) { + // 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(300)); + ASSERT_TRUE(server.Start()); + auto client = HttpClient::Make( + {}, 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"), {}, {}, + *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, 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" + "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 +648,33 @@ class CapturingAuthSession final : public auth::AuthSession { }; TEST(HttpClientTest, BuildsRequestUriLikeJavaHttpRequest) { - HttpClient client({}, {}, {}, std::string("https://catalog.example/api/")); - CapturingAuthSession session; + 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 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 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"); - 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/"); + 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 +724,496 @@ 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)); + } +} + +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( + {{"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()); + const RestCatalogProperties properties; + EXPECT_EQ(options.connection_timeout_ms, + properties.Get(RestCatalogProperties::kClientConnectionTimeoutMs)); +} + +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