[BUG] Complete a curl operation that is never scheduled

CHANGELOG.md

@@ -110,6 +110,23 @@ Increment the:
 * [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))
+* [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] Report a curl request the multi handle refused as a failed create,
+  rather than as a cancel nobody asked for
+  ([#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))
+* [BUG] Let a handler finish the request it is being told about from any event,
+  including the ones the IO thread delivers, instead of waiting on a completion
+  only that thread goes on to publish
+  ([#4395](https://github.com/open-telemetry/opentelemetry-cpp/pull/4395))
+* [BUG] Decide whether a curl thread is inside a callback without an
+  unsynchronized member, which was read and written from two threads behind a
+  sanitizer suppression
+  ([#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_client_curl.h

@@ -348,7 +348,7 @@ class HttpClient : public opentelemetry::ext::http::client::HttpClient
   // return true if create background thread, false is already exist background thread
   bool MaybeSpawnBackgroundThread();
 
-  void ScheduleAddSession(uint64_t session_id);
+  bool ScheduleAddSession(uint64_t session_id);
   void ScheduleAbortSession(uint64_t session_id);
   void ScheduleRemoveSession(uint64_t session_id, HttpCurlEasyResource &&resource);
 
@@ -368,6 +368,12 @@ class HttpClient : public opentelemetry::ext::http::client::HttpClient
 
   std::mutex multi_handle_m_;
   CURLM *multi_handle_;
+
+  // How a session is handed to the multi handle. Only a test replaces it, so that a case about
+  // what happens to a refused session can name the code libcurl returns instead of arranging a
+  // multi handle libcurl documents as unusable and relying on what it does with one.
+  CURLMcode (*add_handle_)(CURLM *, CURL *) = &curl_multi_add_handle;
+
   std::atomic<uint64_t> next_session_id_{0};
   uint64_t max_sessions_per_connection_;
 

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

@@ -13,13 +13,13 @@
 #  include <future>
 #endif
 
+#include <atomic>
 #include <cstdint>
 #include <functional>
 #include <map>
 #include <regex>
 #include <sstream>
 #include <string>
-#include <thread>
 #include <vector>
 #ifdef _WIN32
 #  include <io.h>
@@ -246,7 +246,7 @@ class HttpOperation
    */
   opentelemetry::ext::http::client::SessionState GetSessionState() const noexcept
   {
-    return session_state_;
+    return session_state_.load(std::memory_order_acquire);
   }
 
   /**
@@ -290,6 +290,22 @@ class HttpOperation
   inline CURL *GetCurlEasyHandle() noexcept { return curl_resource_.easy_handle; }
 
 private:
+  // The client is what discovers that nothing will run a request, so it is what gets to say so.
+  friend class HttpClient;
+
+  /**
+   * Finish an operation that nothing is going to run, and say why. Not for callers: it forces a
+   * terminal state, cleans up, dispatches the terminal event, fulfils the promise and hands the
+   * easy resource back, none of which is safe to ask for from outside the client.
+   *
+   * The event goes in ahead of the cleanup, because the only strong reference to the caller's
+   * handler is the one the completion callback holds and the cleanup is what lets that go. The
+   * event carries the terminal state with it, so the cleanup still does not report a cancel
+   * nobody asked for, and cleaning up last is what makes a Finish() on another thread wait for
+   * the event rather than for the promise alone.
+   */
+  void FinishUnscheduled(const char *reason);
+
   CURLcode SetCurlPtrOption(CURLoption option, void *value);
 
   CURLcode SetCurlStrOption(CURLoption option, const char *str)
@@ -338,7 +354,8 @@ 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_{
+  // Atomic because a handler that cancels from one event can overlap the next dispatch.
+  std::atomic<opentelemetry::ext::http::client::SessionState> session_state_{
       opentelemetry::ext::http::client::SessionState::Created};
 
   const opentelemetry::ext::http::client::Compression &compression_;
@@ -365,13 +382,11 @@ class HttpOperation
     // 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;
     std::atomic<bool> is_promise_running{false};
     std::promise<CURLcode> result_promise;
     std::future<CURLcode> result_future;
   };
-  friend class HttpOperationAccessor;
   std::unique_ptr<AsyncData> async_data_;
 };
 }  // namespace curl

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

@@ -11,6 +11,7 @@
 #include <functional>
 #include <list>
 #include <mutex>
+#include <ostream>
 #include <string>
 #include <thread>
 #include <unordered_map>
@@ -23,6 +24,7 @@
 #include "opentelemetry/ext/http/common/url_parser.h"
 #include "opentelemetry/nostd/shared_ptr.h"
 #include "opentelemetry/nostd/string_view.h"
+#include "opentelemetry/sdk/common/global_log_handler.h"
 #include "opentelemetry/sdk/common/thread_instrumentation.h"
 #include "opentelemetry/version.h"
 
@@ -33,8 +35,6 @@
 #  include <array>
 
 #  include "opentelemetry/nostd/type_traits.h"
-#else
-#  include "opentelemetry/sdk/common/global_log_handler.h"
 #endif
 
 OPENTELEMETRY_BEGIN_NAMESPACE
@@ -327,7 +327,12 @@ std::shared_ptr<opentelemetry::ext::http::client::Session> HttpClient::CreateSes
   const auto parsedUrl = common::UrlParser(std::string(url));
   if (!parsedUrl.success_)
   {
-    return std::make_shared<Session>(*this);
+    // Unregistered, but given an id all the same. Pending removals are keyed by session id, so
+    // two sessions sharing one displace each other's easy handle and header list, which nothing
+    // then frees.
+    auto unregistered = std::make_shared<Session>(*this);
+    unregistered->SetId(++next_session_id_);
+    return unregistered;
   }
   auto session =
       std::make_shared<Session>(*this, parsedUrl.scheme_, parsedUrl.host_, parsedUrl.port_);
@@ -638,16 +643,25 @@ bool HttpClient::MaybeSpawnBackgroundThread()
   return true;
 }
 
-void HttpClient::ScheduleAddSession(uint64_t session_id)
+bool HttpClient::ScheduleAddSession(uint64_t session_id)
 {
   {
+    std::lock_guard<std::mutex> sessions_lock{sessions_m_};
+    if (sessions_.end() == sessions_.find(session_id))
+    {
+      // Whatever removed the session owns its teardown. Adding it back would run an operation
+      // nobody waits on, and leave the caller waiting on one nobody runs.
+      return false;
+    }
+
     std::lock_guard<std::recursive_mutex> lock_guard{session_ids_m_};
     pending_to_add_session_ids_.insert(session_id);
     pending_to_remove_session_handles_.erase(session_id);
     pending_to_abort_sessions_.erase(session_id);
   }
 
   wakeupBackgroundThread();
+  return true;
 }
 
 void HttpClient::ScheduleAbortSession(uint64_t session_id)
@@ -728,32 +742,67 @@ bool HttpClient::doAddSessions()
   }
 
   bool has_data = false;
+  std::list<std::pair<std::shared_ptr<Session>, CURLMcode>> rejected_by_multi;
 
-  std::lock_guard<std::mutex> lock_guard{sessions_m_};
-  for (auto &session_id : pending_to_add_session_ids)
   {
-    auto session = sessions_.find(session_id);
-    if (session == sessions_.end())
+    std::lock_guard<std::mutex> lock_guard{sessions_m_};
+    for (auto &session_id : pending_to_add_session_ids)
     {
-      continue;
-    }
+      auto session = sessions_.find(session_id);
+      if (session == sessions_.end())
+      {
+        continue;
+      }
 
-    if (!session->second->GetOperation())
-    {
-      continue;
+      if (!session->second->GetOperation())
+      {
+        continue;
+      }
+
+      CURL *easy_handle = session->second->GetOperation()->GetCurlEasyHandle();
+      if (nullptr == easy_handle)
+      {
+        continue;
+      }
+
+      const CURLMcode rc = add_handle_(multi_handle_, easy_handle);
+      if (CURLM_OK != rc)
+      {
+        // Nothing will drive this transfer. Leaving it here is what makes a caller wait on a
+        // future nobody can complete, so hand it to the loop below to be finished. Reported
+        // there too: the log handler is application code that can re-enter this client.
+        rejected_by_multi.emplace_back(session->second, rc);
+        continue;
+      }
+
+      has_data = true;
     }
+  }
 
-    CURL *easy_handle = session->second->GetOperation()->GetCurlEasyHandle();
-    if (nullptr == easy_handle)
+  // Outside sessions_m_ on purpose. FinishOperation runs the caller's handler, and a handler
+  // that cancels from it takes that lock again. See #4389.
+  for (auto &rejected : rejected_by_multi)
+  {
+    const char *reason = curl_multi_strerror(rejected.second);
+
+    // Told the same way as a session this client never registered, because it is the same thing
+    // from the handler's side: nothing is going to run this request.
+    auto &operation = rejected.first->GetOperation();
+    if (operation)
     {
-      continue;
+      operation->FinishUnscheduled(reason);
     }
 
-    curl_multi_add_handle(multi_handle_, easy_handle);
-    has_data = true;
+    // Settled first. A log handler is replaceable application code, and one that calls
+    // FinishSession() from here would otherwise wait on a promise only the line above fulfils,
+    // on this thread.
+    OTEL_INTERNAL_LOG_ERROR("[HTTP Client Curl] curl_multi_add_handle failed: " << reason);
   }
 
-  return has_data;
+  // Finishing a rejected session queues its removal, and the loop's idle check has already run
+  // doRemoveSessions by the time it calls this. Answering false here lets the worker exit with
+  // that removal still queued.
+  return has_data || !rejected_by_multi.empty();
 }
 
 bool HttpClient::doAbortSessions()