[BUG] Publish the operation once, and complete it only at the end

Startup handed the operation out in pieces. The Session route went out
before the callback, and the promise was created after the first event,
so a cancel arriving in between could reach an operation whose callback
was still empty. Whoever noticed that cleanup had started could then
publish the completion, which meant FinishSession could return while the
IO thread was still inside a user handler.

The route is now the last thing stored. Nothing else can bring the IO
thread to this operation: Abort only raises a flag, and the abort queue
is reached through that same route, so by the time the operation is
reachable the callback, the promise and the future all exist. Cleanup is
then the only thing that ever fulfils the promise, at its tail, after the
terminal event and the completion callback have run, and the recheck that
used to fulfil it from the caller is gone.

A promise that exists before the first event would make a handler calling
FinishSession from that event wait on itself, which is what the late
future used to avoid. A thread local scope answers it directly: every
user callback runs inside one, and Finish returns early when the calling
thread is already inside a callback for this operation. It returns before
the finished flag, so a caller outside the callback still gets to wait,
which the flag used to swallow.

Measured with a handler that holds inside the terminal event for 200 ms.
Before, on the runs where the IO thread delivered that event, Finish
returned in 0 ms with the handler still running. After, it waits 400 ms,
the length of the two held events, and the handler is never still running
when it returns: 6 runs of 6.

The case that came out of that probe is on the branch, and its comment
says what it does not prove. Which thread delivers the terminal event is
not something it can choose, and on the caller-thread schedule there is
nothing for Finish to wait for, so reverting this change leaves it green
10 times in 10. The evidence above is the before and after, not that case.

33 tests pass, 3 of 3.

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

Author: thc1006

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

@@ -413,6 +413,39 @@ static_assert(
         alignof(opentelemetry::ext::http::client::SessionState),
     "std::atomic<SessionState> is more aligned, which changes the layout of HttpOperation");
 
+namespace
+{
+// Which operations the calling thread is currently inside a callback for. A handler is allowed to
+// call FinishSession() on the request it is being told about, and that call must not wait for a
+// completion only the thread it is running on can publish.
+thread_local std::vector<const HttpOperation *> callbacks_in_progress;
+
+class CallbackScope
+{
+public:
+  explicit CallbackScope(const HttpOperation *operation) noexcept
+  {
+    callbacks_in_progress.push_back(operation);
+  }
+  ~CallbackScope() { callbacks_in_progress.pop_back(); }
+
+  CallbackScope(const CallbackScope &)            = delete;
+  CallbackScope &operator=(const CallbackScope &) = delete;
+
+  static bool InsideCallbackFor(const HttpOperation *operation) noexcept
+  {
+    for (const auto *entry : callbacks_in_progress)
+    {
+      if (entry == operation)
+      {
+        return true;
+      }
+    }
+    return false;
+  }
+};
+}  // namespace
+
 void HttpOperation::DispatchEvent(opentelemetry::ext::http::client::SessionState type,
                                   const std::string &reason)
 {
@@ -422,6 +455,7 @@ void HttpOperation::DispatchEvent(opentelemetry::ext::http::client::SessionState
 
   if (event_handle_ != nullptr)
   {
+    const CallbackScope scope{this};
     event_handle_->OnEvent(type, reason);
   }
 }
@@ -514,6 +548,14 @@ HttpOperation::~HttpOperation()
 
 void HttpOperation::Finish()
 {
+  // Called from inside a callback this operation dispatched, so the completion being waited for is
+  // the one this thread has not published yet. Returning before the flag below leaves the wait
+  // available to a caller that is not inside a callback.
+  if (CallbackScope::InsideCallbackFor(this))
+  {
+    return;
+  }
+
   if (is_finished_.exchange(true, std::memory_order_acq_rel))
   {
     return;
@@ -572,6 +614,7 @@ void HttpOperation::Cleanup()
     callback.swap(async_data_->callback);
     if (callback)
     {
+      const CallbackScope scope{this};
       HttpOperationAccessor::SetThreadId(*async_data_, std::this_thread::get_id());
       callback(*this);
       HttpOperationAccessor::SetThreadId(*async_data_, std::thread::id());
@@ -1472,26 +1515,18 @@ CURLcode HttpOperation::SendAsync(Session *session, std::function<void(HttpOpera
   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.store(session, std::memory_order_release);
-  async_data_->callback = std::move(callback);
-
-  DispatchEvent(opentelemetry::ext::http::client::SessionState::Connecting);
-
-  // The future alone stays unpublished until after the event, so a handler calling
-  // FinishSession() from it returns instead of waiting on an unscheduled transfer.
+  async_data_->callback       = std::move(callback);
   async_data_->result_promise = std::promise<CURLcode>();
   async_data_->result_future  = async_data_->result_promise.get_future();
   async_data_->is_promise_running.store(true, std::memory_order_release);
 
-  // A cancel from the event may have torn the operation down before the future existed.
-  if (is_cleaned_.load(std::memory_order_acquire))
-  {
-    if (async_data_->is_promise_running.exchange(false, std::memory_order_acq_rel))
-    {
-      async_data_->result_promise.set_value(last_curl_result_);
-    }
-    return CURLE_OK;
-  }
+  // Last, and the only thing that makes this operation reachable from the IO thread: an abort is
+  // queued through this route, so nothing can bring the operation there before the callback and
+  // the completion above exist. Cleanup is therefore the only thing that ever fulfils the promise,
+  // at its tail, once the terminal event and the completion callback have run.
+  async_data_->session.store(session, std::memory_order_release);
+
+  DispatchEvent(opentelemetry::ext::http::client::SessionState::Connecting);
 
   if (WasAborted())
   {

ext/test/http/curl_http_test.cc

@@ -912,6 +912,83 @@ class OverlappingCancelHandler : public TerminalCountingHandler
 // 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.
+namespace
+{
+// Cancels from Connecting, then holds inside the terminal event for a bounded 200 ms. The bound is
+// its own, so this can never block for good and a case built on it can only fail, not hang.
+class LatchedCancelHandler : public TerminalCountingHandler
+{
+public:
+  void OnEvent(http_client::SessionState state, nostd::string_view reason) noexcept override
+  {
+    TerminalCountingHandler::OnEvent(state, reason);
+    if (state == http_client::SessionState::Cancelled)
+    {
+      entries_.fetch_add(1, std::memory_order_relaxed);
+      if (std::this_thread::get_id() == owner_)
+      {
+        on_owner_thread_.fetch_add(1, std::memory_order_relaxed);
+      }
+      inside_.fetch_add(1, std::memory_order_release);
+      std::this_thread::sleep_for(std::chrono::milliseconds(200));
+      inside_.fetch_sub(1, std::memory_order_release);
+    }
+  }
+
+  std::thread::id owner_{};
+  std::atomic<int> entries_{0};
+  std::atomic<int> on_owner_thread_{0};
+  std::atomic<int> inside_{0};
+};
+}  // namespace
+
+// Holds that Finish does not return while a handler is still running. Which thread delivers the
+// terminal event is not something this case can choose: when the caller thread delivers it there
+// is nothing for Finish to wait for and the case only confirms that. Measured across separate
+// runs, the IO thread takes it roughly one time in three, and that is the run that matters. So
+// this asserts a true invariant on either schedule but does not on its own discriminate the
+// ordering change it came from; the evidence for that is in the commit that made it.
+TEST_F(BasicCurlHttpTests, FinishDoesNotReturnWhileAHandlerIsRunning)
+{
+  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();
+  }
+
+  auto session = session_manager->CreateSession("http://127.0.0.1:19937");
+  auto request = session->CreateRequest();
+  request->SetUri("get/");
+
+  auto handler            = std::make_shared<LatchedCancelHandler>();
+  handler->owner_         = std::this_thread::get_id();
+  handler->cancel_target_ = session.get();
+  handler->cancel_at_     = http_client::SessionState::Connecting;
+
+  session->SendRequest(handler);
+
+  const auto start = std::chrono::steady_clock::now();
+  session->FinishSession();
+  const auto finished = std::chrono::duration_cast<std::chrono::milliseconds>(
+                            std::chrono::steady_clock::now() - start)
+                            .count();
+  const int still_inside = handler->inside_.load(std::memory_order_acquire);
+  session_manager->FinishAllSessions();
+
+  EXPECT_EQ(0, still_inside) << "Finish returned after " << finished
+                             << " ms with a handler still running";
+  EXPECT_GE(handler->entries_.load(std::memory_order_relaxed), 1)
+      << "the terminal event never arrived, so nothing was tested";
+}
+
 TEST_F(BasicCurlHttpTests, CancelFromConnectingWhilePollingCompletes)
 {
   received_requests_.clear();