From e6aff05c3028b90994c00d25cc2dd6f8bde4c61a Mon Sep 17 00:00:00 2001 From: shixi-li Date: Wed, 19 Aug 2026 11:04:06 +0800 Subject: [PATCH] [BUG] Keep the curl retry deadline stable within an attempt --- CHANGELOG.md | 2 + .../http/client/curl/http_operation_curl.h | 8 +- .../http/client/curl/http_operation_curl.cc | 82 +++++++----- ext/test/http/curl_http_test.cc | 119 +++++++++++++++++- 4 files changed, 174 insertions(+), 37 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4d16d636e..545639177 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -61,6 +61,8 @@ Increment the: * [BUG] Draw the curl retry jitter from a per thread generator instead of one shared across HTTP client threads ([#4399](https://github.com/open-telemetry/opentelemetry-cpp/pull/4399)) +* [BUG] Keep the curl retry deadline stable within each attempt + ([#4403](https://github.com/open-telemetry/opentelemetry-cpp/issues/4403)) * [BUG] Stop the curl IO thread deadlocking on itself while recovering from a multi handle error ([#4394](https://github.com/open-telemetry/opentelemetry-cpp/pull/4394)) diff --git a/ext/include/opentelemetry/ext/http/client/curl/http_operation_curl.h b/ext/include/opentelemetry/ext/http/client/curl/http_operation_curl.h index c57309ccd..412faf7ba 100644 --- a/ext/include/opentelemetry/ext/http/client/curl/http_operation_curl.h +++ b/ext/include/opentelemetry/ext/http/client/curl/http_operation_curl.h @@ -280,8 +280,8 @@ class HttpOperation void Abort(); /** - * Perform curl message, this function only can be called in the polling thread and it can only - * be called when got a CURLMSG_DONE. + * Process a completed curl message. This is called directly by synchronous requests or from the + * polling thread after receiving CURLMSG_DONE. * * @param code CURLcode */ @@ -316,6 +316,8 @@ class HttpOperation const char *GetCurlErrorMessage(CURLcode code); + std::chrono::system_clock::time_point CalculateNextRetryTime(); + std::atomic is_aborted_{false}; // Set to 'true' when async callback is aborted std::atomic is_finished_{false}; // Set to 'true' when async callback is finished. std::atomic is_cleaned_{false}; // Set to 'true' when async callback is cleaned. @@ -348,7 +350,7 @@ class HttpOperation const RetryPolicy retry_policy_; decltype(RetryPolicy::max_attempts) retry_attempts_; std::chrono::system_clock::time_point last_attempt_time_; - std::chrono::system_clock::time_point retry_after_time_point_{}; + std::chrono::system_clock::time_point next_retry_time_point_{}; // Processed response headers and body // See CURLINFO_RESPONSE_CODE, type is long diff --git a/ext/src/http/client/curl/http_operation_curl.cc b/ext/src/http/client/curl/http_operation_curl.cc index 0f1bda403..11512b9ca 100644 --- a/ext/src/http/client/curl/http_operation_curl.cc +++ b/ext/src/http/client/curl/http_operation_curl.cc @@ -608,11 +608,16 @@ bool HttpOperation::IsRetryable() std::chrono::system_clock::time_point HttpOperation::NextRetryTime() { - if (retry_after_time_point_ != std::chrono::system_clock::time_point{}) + if (next_retry_time_point_ != std::chrono::system_clock::time_point{}) { - return retry_after_time_point_; + return next_retry_time_point_; } + return CalculateNextRetryTime(); +} + +std::chrono::system_clock::time_point HttpOperation::CalculateNextRetryTime() +{ // One engine per thread. Every HttpClient drives its own background thread, and drawing from // the engine advances its state, so a shared one is written by all of them at once. static thread_local std::mt19937 gen{std::random_device{}()}; @@ -1526,9 +1531,9 @@ void HttpOperation::Abort() void HttpOperation::PerformCurlMessage(CURLcode code) { ++retry_attempts_; - last_attempt_time_ = std::chrono::system_clock::now(); - last_curl_result_ = code; - retry_after_time_point_ = std::chrono::system_clock::time_point{}; + last_attempt_time_ = std::chrono::system_clock::now(); + last_curl_result_ = code; + next_retry_time_point_ = std::chrono::system_clock::time_point{}; if (code != CURLE_OK) { @@ -1559,34 +1564,15 @@ void HttpOperation::PerformCurlMessage(CURLcode code) curl_easy_getinfo(curl_resource_.easy_handle, CURLINFO_RESPONSE_CODE, &response_code_); } - // Transform state - if (GetSessionState() == opentelemetry::ext::http::client::SessionState::Connecting) - { - DispatchEvent(opentelemetry::ext::http::client::SessionState::Connected); - } - - if (GetSessionState() == opentelemetry::ext::http::client::SessionState::Connected) - { - DispatchEvent(opentelemetry::ext::http::client::SessionState::Sending); - } - - if (GetSessionState() == opentelemetry::ext::http::client::SessionState::Sending) - { - DispatchEvent(opentelemetry::ext::http::client::SessionState::Response); - } - - // A server-driven Retry-After may request a retry far beyond the retry - // policy's max_backoff (e.g. a malicious or misbehaving server returning - // "Retry-After: 2 years"). Such a session would linger in the pending retry - // list and block the FIFO drain in doRetrySessions(), so it must be closed - // rather than honored. Cap the requested delay at max_backoff; if the - // server-driven retry time still exceeds the maximum expected retry window, - // the session is closed below and not retried. + // Establish the deadline before dispatching Response. Event handlers run synchronously and may + // inspect it from the callback, so calculating it afterwards could expose a different jittered + // value from the one retained for scheduling this attempt. const bool is_retryable = IsRetryable(); bool retry_after_exceeds_max_delay = false; if (is_retryable) { + bool has_valid_retry_after = false; nostd::string_view retry_after; if (FindRetryAfterValue(response_headers_, retry_after)) { @@ -1597,23 +1583,55 @@ void HttpOperation::PerformCurlMessage(CURLcode code) if (parsed_delay || parsed_date) { + has_valid_retry_after = true; // Reuse the attempt timestamp instead of calling now() again, so the // retry-after delay is measured from the attempt that produced this // response and we avoid an extra system_clock syscall on the hot path. - retry_after_time_point_ = parsed_delay - ? (last_attempt_time_ + delay) - : ((date > last_attempt_time_) ? date : last_attempt_time_); + next_retry_time_point_ = parsed_delay + ? (last_attempt_time_ + delay) + : ((date > last_attempt_time_) ? date : last_attempt_time_); const auto max_retry_time = last_attempt_time_ + std::chrono::duration_cast(retry_policy_.max_backoff); - if (retry_after_time_point_ > max_retry_time) + if (next_retry_time_point_ > max_retry_time) { retry_after_exceeds_max_delay = true; } } } + if (!has_valid_retry_after) + { + next_retry_time_point_ = CalculateNextRetryTime(); + } + } + + // Transform state + if (GetSessionState() == opentelemetry::ext::http::client::SessionState::Connecting) + { + DispatchEvent(opentelemetry::ext::http::client::SessionState::Connected); + } + + if (GetSessionState() == opentelemetry::ext::http::client::SessionState::Connected) + { + DispatchEvent(opentelemetry::ext::http::client::SessionState::Sending); + } + + if (GetSessionState() == opentelemetry::ext::http::client::SessionState::Sending) + { + DispatchEvent(opentelemetry::ext::http::client::SessionState::Response); + } + + // A server-driven Retry-After may request a retry far beyond the retry + // policy's max_backoff (e.g. a malicious or misbehaving server returning + // "Retry-After: 2 years"). Such a session would linger in the pending retry + // list and block the FIFO drain in doRetrySessions(), so it must be closed + // rather than honored. Cap the requested delay at max_backoff; if the + // server-driven retry time still exceeds the maximum expected retry window, + // the session is closed below and not retried. + if (is_retryable) + { if (!retry_after_exceeds_max_delay) { // Clear any response data received in previous attempt diff --git a/ext/test/http/curl_http_test.cc b/ext/test/http/curl_http_test.cc index 90142962d..0fce98128 100644 --- a/ext/test/http/curl_http_test.cc +++ b/ext/test/http/curl_http_test.cc @@ -198,6 +198,26 @@ class RetryEventHandler : public CustomEventHandler } }; +#ifdef ENABLE_OTLP_RETRY_PREVIEW +class RetryDeadlineEventHandler : public RetryEventHandler +{ +public: + void OnEvent(http_client::SessionState state, nostd::string_view reason) noexcept override + { + CustomEventHandler::OnEvent(state, reason); + if (state == http_client::SessionState::Response && operation_ != nullptr) + { + ++response_event_count_; + retry_time_during_response_ = operation_->NextRetryTime(); + } + } + + curl::HttpOperation *operation_{nullptr}; + int response_event_count_{0}; + std::chrono::system_clock::time_point retry_time_during_response_{}; +}; +#endif // ENABLE_OTLP_RETRY_PREVIEW + class BasicCurlHttpTests : public ::testing::Test, public HTTP_SERVER_NS::HttpRequestCallback { protected: @@ -232,6 +252,8 @@ class BasicCurlHttpTests : public ::testing::Test, public HTTP_SERVER_NS::HttpRe server_.addHandler("/get/", *this); server_.addHandler("/post/", *this); server_.addHandler("/retry/", *this); + server_.addHandler("/retry-after/", *this); + server_.addHandler("/retry-after-invalid/", *this); server_.addHandler("/close/", *this); server_.start(); is_running_ = true; @@ -265,12 +287,21 @@ class BasicCurlHttpTests : public ::testing::Test, public HTTP_SERVER_NS::HttpRe response.body = "{'k1':'v1', 'k2':'v2', 'k3':'v3'}"; response_status = 200; } - else if (request.uri == "/retry/") + else if (request.uri == "/retry/" || request.uri == "/retry-after/" || + request.uri == "/retry-after-invalid/") { std::unique_lock lk1(mtx_requests); received_requests_.push_back(request); response.headers["Content-Type"] = "text/plain"; - response_status = 429; + if (request.uri == "/retry-after/") + { + response.headers["Retry-After"] = "2"; + } + else if (request.uri == "/retry-after-invalid/") + { + response.headers["Retry-After"] = "invalid"; + } + response_status = 429; } else if (request.uri == "/close/") { @@ -525,6 +556,90 @@ TEST_F(BasicCurlHttpTests, ExponentialBackoffRetry) ASSERT_EQ(CURLE_OK, operation.Send()); ASSERT_FALSE(operation.IsRetryable()); } + +TEST_F(BasicCurlHttpTests, RetryDeadlineIsStableWithinAttempt) +{ + RetryEventHandler handler; + http_client::HttpSslOptions no_ssl; + http_client::Body body; + http_client::Headers headers; + http_client::Compression compression = http_client::Compression::kNone; + http_client::RetryPolicy retry_policy = {4, std::chrono::duration{1.0f}, + std::chrono::duration{5.0f}, 2.0f}; + + curl::HttpOperation operation(http_client::Method::Post, "http://127.0.0.1:19000/retry/", no_ssl, + &handler, headers, body, compression, false, + curl::kDefaultHttpConnTimeout, false, false, retry_policy); + + ASSERT_EQ(CURLE_OK, operation.Send()); + ASSERT_TRUE(operation.IsRetryable()); + + const auto retry_time = operation.NextRetryTime(); + for (int i = 0; i < 8; ++i) + { + EXPECT_EQ(retry_time, operation.NextRetryTime()); + } +} + +TEST_F(BasicCurlHttpTests, RetryDeadlineIsReadyDuringResponseEvent) +{ + RetryDeadlineEventHandler handler; + http_client::HttpSslOptions no_ssl; + http_client::Body body; + http_client::Headers headers; + http_client::Compression compression = http_client::Compression::kNone; + http_client::RetryPolicy retry_policy = {4, std::chrono::duration{1.0f}, + std::chrono::duration{5.0f}, 2.0f}; + + curl::HttpOperation operation(http_client::Method::Post, "http://127.0.0.1:19000/retry-after/", + no_ssl, &handler, headers, body, compression, false, + curl::kDefaultHttpConnTimeout, false, false, retry_policy); + handler.operation_ = &operation; + + const auto before_send = std::chrono::system_clock::now(); + ASSERT_EQ(CURLE_OK, operation.Send()); + const auto after_send = std::chrono::system_clock::now(); + + ASSERT_TRUE(operation.IsRetryable()); + ASSERT_EQ(1, handler.response_event_count_); + EXPECT_EQ(handler.retry_time_during_response_, operation.NextRetryTime()); + EXPECT_GE(handler.retry_time_during_response_.time_since_epoch().count(), + (before_send + std::chrono::seconds{2}).time_since_epoch().count()); + EXPECT_LE(handler.retry_time_during_response_.time_since_epoch().count(), + (after_send + std::chrono::seconds{2}).time_since_epoch().count()); +} + +TEST_F(BasicCurlHttpTests, InvalidRetryAfterUsesStableBackoff) +{ + RetryDeadlineEventHandler handler; + http_client::HttpSslOptions no_ssl; + http_client::Body body; + http_client::Headers headers; + http_client::Compression compression = http_client::Compression::kNone; + http_client::RetryPolicy retry_policy = {4, std::chrono::duration{1.0f}, + std::chrono::duration{5.0f}, 2.0f}; + + curl::HttpOperation operation( + http_client::Method::Post, "http://127.0.0.1:19000/retry-after-invalid/", no_ssl, &handler, + headers, body, compression, false, curl::kDefaultHttpConnTimeout, false, false, retry_policy); + handler.operation_ = &operation; + + const auto before_send = std::chrono::system_clock::now(); + ASSERT_EQ(CURLE_OK, operation.Send()); + const auto after_send = std::chrono::system_clock::now(); + + ASSERT_TRUE(operation.IsRetryable()); + ASSERT_EQ(1, handler.response_event_count_); + for (int i = 0; i < 8; ++i) + { + EXPECT_EQ(handler.retry_time_during_response_, operation.NextRetryTime()); + } + EXPECT_GE(handler.retry_time_during_response_.time_since_epoch().count(), + (before_send + std::chrono::milliseconds{750}).time_since_epoch().count()); + EXPECT_LE(handler.retry_time_during_response_.time_since_epoch().count(), + (after_send + std::chrono::milliseconds{1250}).time_since_epoch().count()); +} + #endif // ENABLE_OTLP_RETRY_PREVIEW // A cancel that arrives once the server has answered used to deliver Cancelled and the response,