[BUG] Cancel a curl session without writing to the easy handle

CHANGELOG.md

@@ -37,6 +37,9 @@ Increment the:
 * [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))
+* [BUG] Cancel a curl session without writing to the easy handle from the
+  cancelling thread
+  ([#4392](https://github.com/open-telemetry/opentelemetry-cpp/pull/4392))
 * [CODE HEALTH] Enable clang-tidy `modernize-deprecated-headers` and replace
   deprecated C headers (`stdint.h`, `stddef.h`, `stdlib.h`, `string.h`,
   `stdio.h`, `ctype.h`, `limits.h`, `assert.h`) with their C++ equivalents

ext/include/opentelemetry/ext/http/client/curl/http_operation_curl.h

@@ -362,7 +362,8 @@ class HttpOperation
 
   struct AsyncData
   {
-    Session *session{nullptr};  // Owner Session
+    // Read by Abort() on whichever thread cancels, cleared by Cleanup() on the IO thread.
+    std::atomic<Session *> session{nullptr};  // Owner Session
 
     std::thread::id callback_thread;
     std::function<void(HttpOperation &)> callback;

ext/src/http/client/curl/http_operation_curl.cc

@@ -375,12 +375,9 @@ int HttpOperation::OnProgressCallback(void *clientp,
     return -1;
   }
 
-  // CURL_PROGRESSFUNC_CONTINUE is added in 7.68.0
-#  if defined(CURL_PROGRESSFUNC_CONTINUE)
-  return CURL_PROGRESSFUNC_CONTINUE;
-#  else
+  // Not CURL_PROGRESSFUNC_CONTINUE, which asks libcurl to also run its built-in progress
+  // meter, and that meter writes to stderr.
   return 0;
-#  endif
 }
 #else
 int HttpOperation::OnProgressCallback(void *clientp,
@@ -546,11 +543,9 @@ void HttpOperation::Cleanup()
   if (async_data_)
   {
     // Just reset and move easy_handle to owner if in async mode
-    if (async_data_->session != nullptr)
+    Session *session = async_data_->session.exchange(nullptr, std::memory_order_acq_rel);
+    if (session != nullptr)
     {
-      auto session         = async_data_->session;
-      async_data_->session = nullptr;
-
       if (curl_resource_.easy_handle != nullptr)
       {
         curl_easy_setopt(curl_resource_.easy_handle, CURLOPT_PRIVATE, NULL);
@@ -1438,7 +1433,7 @@ CURLcode HttpOperation::SendAsync(Session *session, std::function<void(HttpOpera
 
   async_data_.reset(new AsyncData());
   async_data_->is_promise_running.store(false, std::memory_order_release);
-  async_data_->session = nullptr;
+  async_data_->session.store(nullptr, std::memory_order_release);
 
   ReleaseResponse();
 
@@ -1452,12 +1447,17 @@ CURLcode HttpOperation::SendAsync(Session *session, std::function<void(HttpOpera
   }
   curl_easy_setopt(curl_resource_.easy_handle, CURLOPT_PRIVATE, session);
 
+  // Only a session can be cancelled, so only this path needs the progress callback live. Setting
+  // it here, on the thread that owns the handle and before the handle is scheduled, leaves
+  // Abort() with nothing to do but raise the flag.
+  curl_easy_setopt(curl_resource_.easy_handle, CURLOPT_NOPROGRESS, 0L);
+
   DispatchEvent(opentelemetry::ext::http::client::SessionState::Connecting);
   is_finished_.store(false, std::memory_order_release);
   is_aborted_.store(false, std::memory_order_release);
   is_cleaned_.store(false, std::memory_order_release);
 
-  async_data_->session = session;
+  async_data_->session.store(session, std::memory_order_release);
   if (false == async_data_->is_promise_running.exchange(true, std::memory_order_acq_rel))
   {
     async_data_->result_promise = std::promise<CURLcode>();
@@ -1509,15 +1509,16 @@ void HttpOperation::ReleaseResponse()
 
 void HttpOperation::Abort()
 {
+  // The easy handle belongs to the thread inside curl_multi_perform, so nothing here reads or
+  // writes it. Raising the flag is enough: the progress callback polls it, and the scheduled
+  // abort removes the handle on that thread.
   is_aborted_.store(true, std::memory_order_release);
-  if (curl_resource_.easy_handle != nullptr)
+  if (async_data_)
   {
-    // Enable progress callback to abort from polling thread
-    curl_easy_setopt(curl_resource_.easy_handle, CURLOPT_NOPROGRESS, 0L);
-    if (async_data_ && nullptr != async_data_->session)
+    Session *session = async_data_->session.load(std::memory_order_acquire);
+    if (nullptr != session)
     {
-      async_data_->session->GetHttpClient().ScheduleAbortSession(
-          async_data_->session->GetSessionId());
+      session->GetHttpClient().ScheduleAbortSession(session->GetSessionId());
     }
   }
 }

ext/test/http/curl_http_test.cc

@@ -127,9 +127,8 @@ class TerminalCountingHandler : public CustomEventHandler
     }
   }
 
-  // Cancelling reaches curl_easy_setopt through Abort(), so it has to run on the thread that
-  // owns the handle. Both cases below cancel from an event the IO thread dispatches, and assert
-  // on cancelled_from_ so that a later edit cannot quietly move it back to the caller.
+  // cancel_at_ picks the event to cancel from and cancelled_from_ records the thread it ran
+  // on, so a case can pin which side of the client it covers.
   http_client::Session *cancel_target_ = nullptr;
   http_client::SessionState cancel_at_ = http_client::SessionState::Response;
   std::thread::id cancelled_from_{};
@@ -580,7 +579,7 @@ TEST_F(BasicCurlHttpTests, ACancelBeforeTheResponseReportsCancelled)
   // else reports a cancel.
   EXPECT_EQ(1, handler->cancelled_from_callback_.load(std::memory_order_acquire));
   EXPECT_NE(handler->cancelled_from_, std::this_thread::get_id())
-      << "the cancel has to run on the IO thread, see #4369";
+      << "this case cancels from an event the IO thread dispatches";
 
   session_manager->FinishAllSessions();
 }
@@ -630,6 +629,66 @@ TEST_F(BasicCurlHttpTests, ResetMultiHandleWithASessionDoesNotDeadlock)
   client->FinishAllSessions();
 }
 
+// The caller-thread side of the same cancel. The server handler takes mtx_requests before it
+// answers, so holding it keeps a response from racing the cancel and the abort lands while the
+// IO thread is still driving the easy handle. That pairing is what #4369 caught.
+TEST_F(BasicCurlHttpTests, ACancelFromTheCallerThreadReportsCancelled)
+{
+  received_requests_.clear();
+  auto session_manager = std::make_shared<http_client::curl::HttpCurlClientFactory>()->Create();
+  EXPECT_TRUE(session_manager != nullptr);
+
+  auto session = session_manager->CreateSession("http://127.0.0.1:19000");
+  auto request = session->CreateRequest();
+  request->SetUri("get/");
+
+  auto handler = std::make_shared<TerminalCountingHandler>();
+
+  {
+    std::unique_lock<std::mutex> lock_requests(mtx_requests);
+    session->SendRequest(handler);
+    session->CancelSession();
+    session->FinishSession();
+  }
+
+  EXPECT_FALSE(handler->got_response_.load(std::memory_order_acquire));
+  EXPECT_EQ(1, handler->cancelled_from_callback_.load(std::memory_order_acquire));
+
+  session_manager->FinishAllSessions();
+}
+
+// The same cancel again, repeated, because #4369 is a race and one attempt proves little.
+// Nothing listens on 19937, so every attempt fails to connect and the IO thread reaches Cleanup
+// while the caller is still inside CancelSession, which is the overlap the race needs. Under a
+// thread sanitizer this reports against the unfixed client on every run.
+TEST_F(BasicCurlHttpTests, RepeatedCallerThreadCancelsAreClean)
+{
+  int terminal_total = 0;
+
+  for (int i = 0; i < 20; ++i)
+  {
+    auto session_manager = std::make_shared<http_client::curl::HttpCurlClientFactory>()->Create();
+    ASSERT_TRUE(session_manager != nullptr);
+
+    auto session = session_manager->CreateSession("http://127.0.0.1:19937");
+    auto request = session->CreateRequest();
+    request->SetUri("get/");
+
+    auto handler = std::make_shared<TerminalCountingHandler>();
+    session->SendRequest(handler);
+    session->CancelSession();
+    session->FinishSession();
+    session_manager->FinishAllSessions();
+
+    EXPECT_FALSE(handler->got_response_.load(std::memory_order_acquire));
+    terminal_total += handler->terminal_count_.load(std::memory_order_acquire);
+  }
+
+  // A lower bound, not a count: #4360 tracks the same cancel arriving twice, and how many
+  // arrive is not what this case decides.
+  EXPECT_GE(terminal_total, 20);
+}
+
 TEST_F(BasicCurlHttpTests, SendGetRequestSync)
 {
   received_requests_.clear();