[BUG] Keep the curl retry deadline stable within an attempt

CHANGELOG.md

@@ -63,6 +63,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))

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<bool> is_aborted_{false};   // Set to 'true' when async callback is aborted
   std::atomic<bool> is_finished_{false};  // Set to 'true' when async callback is finished.
   std::atomic<bool> 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

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<std::chrono::milliseconds>(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

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<std::mutex> 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<float>{1.0f},
+                                           std::chrono::duration<float>{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<float>{1.0f},
+                                           std::chrono::duration<float>{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<float>{1.0f},
+                                           std::chrono::duration<float>{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,