[BUG] Send one request per curl session

CHANGELOG.md

@@ -40,6 +40,10 @@ Increment the:
   in the value as-is rather than dropping the attribute.
   [#1536](https://github.com/open-telemetry/opentelemetry-cpp/issues/1536)
 
+* [BUG] Send one request per curl session, rather than replacing the operation
+  a running request still belongs to
+  [#4396](https://github.com/open-telemetry/opentelemetry-cpp/issues/4396)
+
 * [BUG] Report one outcome per request when a curl session is cancelled after
   the response arrives
   ([#4363](https://github.com/open-telemetry/opentelemetry-cpp/pull/4363))

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

@@ -186,10 +186,25 @@ class Session : public opentelemetry::ext::http::client::Session,
 
   std::shared_ptr<opentelemetry::ext::http::client::Request> CreateRequest() noexcept override
   {
+    if (send_started_.load(std::memory_order_acquire))
+    {
+      // The request that has been sent stays where it is. libcurl does not copy the body or the
+      // header list, so the transfer reads them for as long as it runs.
+      return http_request_;
+    }
+
     http_request_.reset(new Request());
     return http_request_;
   }
 
+  /**
+   * Send the request this session carries.
+   *
+   * A session carries one request. A second call reports CreateFailed to its own handler and
+   * sends nothing: the easy handle of the first request holds this session and the operation
+   * that owns it, and replacing that operation takes away what libcurl and the client are still
+   * reading.
+   */
   void SendRequest(
       std::shared_ptr<opentelemetry::ext::http::client::EventHandler> callback) noexcept override;
 
@@ -235,6 +250,9 @@ class Session : public opentelemetry::ext::http::client::Session,
   uint64_t session_id_ = 0UL;
   HttpClient &http_client_;
   std::atomic<bool> is_session_active_{false};
+
+  // Raised by the first SendRequest and never lowered. A session carries one request.
+  std::atomic<bool> send_started_{false};
 };
 
 class HttpClientSync : public opentelemetry::ext::http::client::HttpClientSync

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

@@ -141,6 +141,21 @@ static int deflateInPlace(z_stream *strm, unsigned char *buf, uint32_t len, uint
 void Session::SendRequest(
     std::shared_ptr<opentelemetry::ext::http::client::EventHandler> callback) noexcept
 {
+  if (send_started_.exchange(true, std::memory_order_acq_rel))
+  {
+    // The first request is not finished with this session. Its easy handle names the session in
+    // CURLOPT_PRIVATE and names its operation in every callback it was given, and the message
+    // loop resolves that name to whichever operation the session owns, so a second operation
+    // would be handed the first one's completion. Worse from a handler that sends again from
+    // OnResponse, where the operation being replaced is the one running that handler.
+    if (callback)
+    {
+      callback->OnEvent(opentelemetry::ext::http::client::SessionState::CreateFailed,
+                        "a session carries one request");
+    }
+    return;
+  }
+
   is_session_active_.store(true, std::memory_order_release);
   const auto &url       = host_ + http_request_->uri_;
   auto callback_ptr     = callback.get();

ext/test/http/curl_http_test.cc

@@ -97,6 +97,37 @@ class CustomEventHandler : public http_client::EventHandler
 // inside the Response event, which is the one moment both arms of the completion callback are
 // eligible: DispatchEvent notifies the handler before it stores the new state, and the callback
 // runs after both, so it sees an aborted operation that also has a response.
+class ReentrantSendHandler : public CustomEventHandler
+{
+public:
+  void OnResponse(http_client::Response & /* response */) noexcept override
+  {
+    got_response_.store(true, std::memory_order_release);
+
+    if (session_ != nullptr && !resent_.exchange(true, std::memory_order_acq_rel))
+    {
+      auto again = session_->CreateRequest();
+      again->SetUri("get/");
+      session_->SendRequest(self_.lock());
+    }
+  }
+
+  void OnEvent(http_client::SessionState state, nostd::string_view /* reason */) noexcept override
+  {
+    if (state == http_client::SessionState::CreateFailed)
+    {
+      create_failed_.fetch_add(1, std::memory_order_release);
+    }
+  }
+
+  http_client::Session *session_ = nullptr;
+  std::weak_ptr<ReentrantSendHandler> self_;
+  std::atomic<int> create_failed_{0};
+
+private:
+  std::atomic<bool> resent_{false};
+};
+
 class TerminalCountingHandler : public CustomEventHandler
 {
 public:
@@ -629,6 +660,63 @@ TEST_F(BasicCurlHttpTests, ResetMultiHandleWithASessionDoesNotDeadlock)
   client->FinishAllSessions();
 }
 
+// The reproduction from #4396. A handler that sends again from OnResponse replaces the operation
+// whose Cleanup is running that handler, and Cleanup reads that operation again on the way out.
+TEST_F(BasicCurlHttpTests, ASecondRequestFromInsideTheResponseIsRefused)
+{
+  received_requests_.clear();
+  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:19000");
+  auto request = session->CreateRequest();
+  request->SetUri("get/");
+
+  auto handler      = std::make_shared<ReentrantSendHandler>();
+  handler->session_ = session.get();
+  handler->self_    = handler;
+
+  session->SendRequest(handler);
+  ASSERT_TRUE(waitForRequests(30, 1));
+  session->FinishSession();
+
+  EXPECT_TRUE(handler->got_response_.load(std::memory_order_acquire));
+  EXPECT_EQ(1, handler->create_failed_.load(std::memory_order_acquire));
+
+  session_manager->FinishAllSessions();
+}
+
+// The same thing without the re-entrancy, and the request the first send is reading from is not
+// replaced under it either.
+TEST_F(BasicCurlHttpTests, ASessionSendsOneRequest)
+{
+  received_requests_.clear();
+  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:19000");
+  auto request = session->CreateRequest();
+  request->SetUri("get/");
+
+  auto first = std::make_shared<ReentrantSendHandler>();
+  session->SendRequest(first);
+  ASSERT_TRUE(waitForRequests(30, 1));
+  session->FinishSession();
+
+  EXPECT_TRUE(first->got_response_.load(std::memory_order_acquire));
+  EXPECT_EQ(0, first->create_failed_.load(std::memory_order_acquire));
+
+  auto again = session->CreateRequest();
+  EXPECT_EQ(request.get(), again.get());
+
+  auto second = std::make_shared<ReentrantSendHandler>();
+  session->SendRequest(second);
+  EXPECT_EQ(1, second->create_failed_.load(std::memory_order_acquire));
+  EXPECT_FALSE(second->got_response_.load(std::memory_order_acquire));
+
+  session_manager->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.