diff --git a/mkdocs/docs/rest-catalog.md b/mkdocs/docs/rest-catalog.md new file mode 100644 index 000000000..eadf1f29c --- /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 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 a1f205a60..8059d7128 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 718a5b737..ed6891efc 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 a18506404..401b081ef 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 000000000..7e3c09e42 --- /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 c8d88d06f..42d92ab2b 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 9e5037299..9cfc01eb7 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