[BUG] Settle a refused session before reporting it

doAddSessions logged the refusal, then read GetOperation(), then finished it. The
log goes through the global handler, which is replaceable application code and is
not inside a callback scope for that operation, so a handler answering with
Session::FinishSession() reached HttpOperation::Finish(). That is not
InsideCallbackFor(this), so it waits on a promise only the FinishUnscheduled
below it fulfils, on the same thread.

Reading the operation after the log was the other half. A handler calling
SendRequest() from there swaps the operation, and the loop then finishes the new
one and leaves the refused one unsettled.

FinishUnscheduled dispatches its event before it cleans up, and a handler calling
FinishSession() from that event is inside a callback for the operation, so
Finish() returns instead of waiting. Reporting after it is what puts the log
handler inside the same protection.

ARejectedSessionIsSettledBeforeItIsReported installs a global log handler that
records whether the terminal event has arrived by the time the refusal is logged,
and restores the previous handler and level through a scope guard. It passes with
this change and fails at the ordering assertion with the report moved back above
the settlement.

The first version of that assertion asked only whether the handler had seen any
event. That is true in both orders, because SendAsync dispatches Created during
setup, and the case passed with the defect put back. It asks for CreateFailed now.

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

Author: thc1006

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

@@ -784,7 +784,6 @@ bool HttpClient::doAddSessions()
   for (auto &rejected : rejected_by_multi)
   {
     const char *reason = curl_multi_strerror(rejected.second);
-    OTEL_INTERNAL_LOG_ERROR("[HTTP Client Curl] curl_multi_add_handle failed: " << reason);
 
     // 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.
@@ -793,6 +792,11 @@ bool HttpClient::doAddSessions()
     {
       operation->FinishUnscheduled(reason);
     }
+
+    // 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);
   }
 
   // Finishing a rejected session queues its removal, and the loop's idle check has already run

ext/test/http/curl_http_test.cc

@@ -13,6 +13,7 @@
 #  include <numeric>
 #endif  // ENABLE_OTLP_COMPRESSION_PREVIEW
 
+#include <algorithm>
 #include <atomic>
 #include <chrono>
 #include <condition_variable>
@@ -34,12 +35,14 @@
 #include "opentelemetry/ext/http/server/http_server.h"
 #include "opentelemetry/nostd/function_ref.h"
 #include "opentelemetry/nostd/string_view.h"
+#include "opentelemetry/sdk/common/global_log_handler.h"
 #include "opentelemetry/version.h"
 
 constexpr int HTTP_PORT{19000};
 
 namespace curl        = opentelemetry::ext::http::client::curl;
 namespace http_client = opentelemetry::ext::http::client;
+namespace sdk_common  = opentelemetry::sdk::common;
 namespace nostd       = opentelemetry::nostd;
 
 OPENTELEMETRY_BEGIN_NAMESPACE
@@ -783,6 +786,110 @@ TEST_F(BasicCurlHttpTests, AnUnscheduledSessionTellsAHandlerNothingElseHolds)
   http_client::curl::HttpClientTestPeer::RemoveSessions(client);
 }
 
+// Restores the global log handler and level whatever the case does, since both are process wide
+// and gtest runs one process per case only under CTest.
+class ScopedLogHandler
+{
+public:
+  explicit ScopedLogHandler(nostd::shared_ptr<sdk_common::internal_log::LogHandler> handler)
+      : previous_handler_(sdk_common::internal_log::GlobalLogHandler::GetLogHandler()),
+        previous_level_(sdk_common::internal_log::GlobalLogHandler::GetLogLevel())
+  {
+    sdk_common::internal_log::GlobalLogHandler::SetLogHandler(handler);
+    sdk_common::internal_log::GlobalLogHandler::SetLogLevel(
+        sdk_common::internal_log::LogLevel::Debug);
+  }
+
+  ~ScopedLogHandler()
+  {
+    sdk_common::internal_log::GlobalLogHandler::SetLogHandler(previous_handler_);
+    sdk_common::internal_log::GlobalLogHandler::SetLogLevel(previous_level_);
+  }
+
+  ScopedLogHandler(const ScopedLogHandler &)            = delete;
+  ScopedLogHandler &operator=(const ScopedLogHandler &) = delete;
+
+private:
+  nostd::shared_ptr<sdk_common::internal_log::LogHandler> previous_handler_;
+  sdk_common::internal_log::LogLevel previous_level_;
+};
+
+// Runs whatever the case gives it, on the thread that wrote the log line.
+class ObservingLogHandler : public sdk_common::internal_log::LogHandler
+{
+public:
+  explicit ObservingLogHandler(std::function<void()> observe) : observe_(std::move(observe)) {}
+
+  void Handle(sdk_common::internal_log::LogLevel /* level */,
+              const char * /* file */,
+              int /* line */,
+              const char * /* msg */,
+              const sdk_common::AttributeMap & /* attributes */) noexcept override
+  {
+    observe_();
+  }
+
+private:
+  std::function<void()> observe_;
+};
+
+// The refusal is reported through the global log handler, which is application code that can call
+// back into this client. Reporting before the operation is settled leaves a handler that answers
+// with FinishSession() waiting on a promise only this thread can fulfil.
+TEST_F(BasicCurlHttpTests, ARejectedSessionIsSettledBeforeItIsReported)
+{
+  curl::HttpClient client;
+
+  auto session = client.CreateSession("http://127.0.0.1:19000");
+  ASSERT_TRUE(session != nullptr);
+  auto curl_session = std::static_pointer_cast<curl::Session>(session);
+
+  auto handler = std::make_shared<RecordingHandler>();
+  http_client::HttpSslOptions no_ssl;
+  http_client::Body body;
+  http_client::Headers headers;
+  http_client::RetryPolicy no_retry{};
+  http_client::Compression compression = http_client::Compression::kNone;
+
+  curl_session->GetOperation().reset(new curl::HttpOperation(
+      http_client::Method::Get, "http://127.0.0.1:19000/get/", no_ssl, handler.get(), headers, body,
+      compression, false, curl::kDefaultHttpConnTimeout, false, false, no_retry));
+
+  ASSERT_EQ(CURLE_OK, curl_session->GetOperation()->SendAsync(
+                          curl_session.get(), [](curl::HttpOperation & /* operation */) {}));
+
+  std::atomic<int> log_calls{0};
+  std::atomic<int> settled_when_logged{0};
+  auto observer = nostd::shared_ptr<sdk_common::internal_log::LogHandler>(
+      new ObservingLogHandler([&handler, &log_calls, &settled_when_logged]() {
+        log_calls.fetch_add(1, std::memory_order_release);
+
+        // The terminal event specifically. Created arrives during setup, so "the handler has
+        // seen something" is true in either order.
+        const std::vector<http_client::SessionState> states = handler->States();
+        if (std::find(states.begin(), states.end(), http_client::SessionState::CreateFailed) !=
+            states.end())
+        {
+          settled_when_logged.fetch_add(1, std::memory_order_release);
+        }
+      }));
+
+  {
+    ScopedLogHandler scoped{observer};
+    http_client::curl::HttpClientTestPeer::RefuseAdds(client);
+    http_client::curl::HttpClientTestPeer::AddSessions(client);
+    http_client::curl::HttpClientTestPeer::AllowAdds(client);
+  }
+
+  ASSERT_EQ(1, log_calls.load(std::memory_order_acquire))
+      << "the refusal was not reported once, so this case is not observing what it says it is";
+  EXPECT_EQ(1, settled_when_logged.load(std::memory_order_acquire))
+      << "the refusal was reported while the operation was still unsettled, so a log handler "
+      << "calling FinishSession() from here would wait on a promise only this thread fulfils";
+
+  http_client::curl::HttpClientTestPeer::RemoveSessions(client);
+}
+
 // The third way an operation ends up with nothing to run it. SendAsync does the whole async
 // setup and Session::SendRequest is what spawns the worker, so the add runs here on one thread
 // against a multi handle that refuses it, which is what a multi handle that failed to initialize