[BUG] Publish the curl session state before the event that reports it

DispatchEvent stored session_state_ after the handler had returned. A handler
that cancels from the first event of a client that is already polling lets the
background thread finish the operation and dispatch a state of its own, so both
threads wrote that member, and the store left until after the handler returned
overwrote the cancel with a state the operation had left. Found by @lalitb in
review.

The store moves ahead of the handler and the member becomes atomic. SessionState
is a uint8_t enum and std::atomic of it is one byte with alignment one, so the
layout of an installed type does not move.

The new case warms the client with a completed request first, because the
background thread is only spawned after SendAsync returns and without one there
is nothing for the event to overlap. It is the only case in this file that
reaches the is_cleaned_ recheck at the end of SendAsync. Its counters are
relaxed deliberately, since an acquire or a release on them gives the two
threads an ordering the code under test does not have and the racing store stops
being reported.

Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>

Author: thc1006

CHANGELOG.md

@@ -40,6 +40,9 @@ Increment the:
 * [BUG] Finish a curl operation that never gets scheduled, instead of leaving
   FinishSession blocked forever
   ([#4395](https://github.com/open-telemetry/opentelemetry-cpp/pull/4395))
+* [BUG] Store the curl session state before the event that reports it, so a
+  cancel dispatched by the IO thread is not overwritten or raced
+  ([#4395](https://github.com/open-telemetry/opentelemetry-cpp/pull/4395))
 * [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

@@ -13,6 +13,7 @@
 #  include <future>
 #endif
 
+#include <atomic>
 #include <cstdint>
 #include <functional>
 #include <map>
@@ -246,7 +247,7 @@ class HttpOperation
    */
   opentelemetry::ext::http::client::SessionState GetSessionState() const noexcept
   {
-    return session_state_;
+    return session_state_.load(std::memory_order_acquire);
   }
 
   /**
@@ -338,7 +339,9 @@ class HttpOperation
   const Headers &request_headers_;
   const opentelemetry::ext::http::client::Body &request_body_;
   size_t request_nwrite_{0};
-  opentelemetry::ext::http::client::SessionState session_state_{
+  // Written by whichever thread dispatches an event. A handler that cancels from one event
+  // lets the background thread dispatch another, so the two dispatches can overlap.
+  std::atomic<opentelemetry::ext::http::client::SessionState> session_state_{
       opentelemetry::ext::http::client::SessionState::Created};
 
   const opentelemetry::ext::http::client::Compression &compression_;

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

@@ -407,12 +407,15 @@ int HttpOperation::OnProgressCallback(void *clientp,
 void HttpOperation::DispatchEvent(opentelemetry::ext::http::client::SessionState type,
                                   const std::string &reason)
 {
+  // Published before the handler runs. A handler can cancel from here, which lets the background
+  // thread finish the operation and publish a state of its own, and a store left until after the
+  // handler returned would overwrite that one.
+  session_state_.store(type, std::memory_order_release);
+
   if (event_handle_ != nullptr)
   {
     event_handle_->OnEvent(type, reason);
   }
-
-  session_state_ = type;
 }
 
 HttpOperation::HttpOperation(opentelemetry::ext::http::client::Method method,

ext/test/http/curl_http_test.cc

@@ -698,6 +698,85 @@ TEST_F(BasicCurlHttpTests, InvalidUrlCompletes)
   EXPECT_GE(handler->terminal_count_.load(std::memory_order_acquire), 1);
 }
 
+// The counters say whether the IO thread reached this handler while the caller was still
+// inside an event of its own. They are relaxed on purpose. An acquire or a release on them
+// would give the two threads an ordering the code under test does not have, and a state store
+// racing another would stop being reported.
+class OverlappingCancelHandler : public TerminalCountingHandler
+{
+public:
+  void OnEvent(http_client::SessionState state, nostd::string_view reason) noexcept override
+  {
+    const int depth = inside_events_.fetch_add(1, std::memory_order_relaxed) + 1;
+    int highest     = max_concurrent_events_.load(std::memory_order_relaxed);
+    while (depth > highest &&
+           !max_concurrent_events_.compare_exchange_weak(highest, depth, std::memory_order_relaxed,
+                                                         std::memory_order_relaxed))
+    {
+    }
+
+    TerminalCountingHandler::OnEvent(state, reason);
+
+    if (state == cancel_at_)
+    {
+      // Cancelling wakes the IO thread, which finishes the operation and dispatches a Cancelled
+      // of its own. Stay in this event until that one arrives so the two really do overlap, and
+      // give up rather than hang if it never does.
+      const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(30);
+      while (inside_events_.load(std::memory_order_relaxed) < 2 &&
+             std::chrono::steady_clock::now() < deadline)
+      {
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+      }
+    }
+
+    inside_events_.fetch_sub(1, std::memory_order_relaxed);
+  }
+
+  std::atomic<int> inside_events_{0};
+  std::atomic<int> max_concurrent_events_{0};
+};
+
+// A client spawns its IO thread only after a request has been scheduled, so cancelling from
+// the first event of the first request has nothing to overlap. On a client that is already
+// polling, the IO thread finishes the operation while the handler is still inside that event,
+// and the caller reaches the end of SendAsync with the operation already cleaned up.
+TEST_F(BasicCurlHttpTests, CancelFromConnectingWhilePollingCompletes)
+{
+  received_requests_.clear();
+  auto session_manager = std::make_shared<http_client::curl::HttpCurlClientFactory>()->Create();
+  ASSERT_TRUE(session_manager != nullptr);
+
+  {
+    auto warm         = session_manager->CreateSession("http://127.0.0.1:19000");
+    auto warm_request = warm->CreateRequest();
+    warm_request->SetUri("get/");
+    auto warm_handler = std::make_shared<TerminalCountingHandler>();
+    warm->SendRequest(warm_handler);
+    ASSERT_TRUE(waitForRequests(30, 1));
+    warm->FinishSession();
+    ASSERT_TRUE(warm_handler->got_response_.load(std::memory_order_acquire));
+  }
+
+  // Nothing listens on 19937, so this operation cannot answer on its own.
+  auto session = session_manager->CreateSession("http://127.0.0.1:19937");
+  auto request = session->CreateRequest();
+  request->SetUri("get/");
+
+  auto handler            = std::make_shared<OverlappingCancelHandler>();
+  handler->cancel_target_ = session.get();
+  handler->cancel_at_     = http_client::SessionState::Connecting;
+
+  session->SendRequest(handler);
+  session->FinishSession();
+  session_manager->FinishAllSessions();
+
+  EXPECT_EQ(2, handler->max_concurrent_events_.load(std::memory_order_acquire));
+  EXPECT_FALSE(handler->got_response_.load(std::memory_order_acquire));
+  EXPECT_EQ(1, handler->cancelled_from_callback_.load(std::memory_order_acquire));
+  EXPECT_FALSE(session->IsSessionActive());
+}
+
 TEST_F(BasicCurlHttpTests, SendGetRequestSync)
 {
   received_requests_.clear();