fix: prevent EDEADLK self-join in ~CurlHttpOperation on async-thread destruction

lib/http/HttpClient_Curl.cpp

@@ -83,11 +83,9 @@ namespace MAT_NS_BEGIN {
 
         auto curlOperation = std::make_shared<CurlHttpOperation>(curlRequest->m_method, curlRequest->m_url, callback, requestHeaders, curlRequest->m_body, false, HTTP_CONN_TIMEOUT, m_sslVerify, sslCaInfo);
         curlRequest->SetOperation(curlOperation);
-
-        // The lifetime of curlOperation is guarnteed by the call to result.wait() in the d'tor.  
-        curlOperation->SendAsync([this, callback, requestId](CurlHttpOperation& operation) {
-            this->EraseRequest(requestId);
 
+        curlOperation->SendAsync([this, callback, requestId](CurlHttpOperation& operation) {
+            EraseRequest(requestId);
             auto response = std::unique_ptr<SimpleHttpResponse>(new SimpleHttpResponse(requestId));
             response->m_result = HttpResult_OK;
 
@@ -161,4 +159,3 @@ namespace MAT_NS_BEGIN {
 } MAT_NS_END
 
 #endif
-

lib/http/HttpClient_Curl.hpp

@@ -17,11 +17,15 @@
 #include <sstream>
 #include <vector>
 #include <iterator>
+#include <map>
 
 #include <algorithm>
 #include <numeric>
-#include <future>
+#include <limits>
 #include <atomic>
+#include <thread>
+#include <mutex>
+#include <stdexcept>
 
 #include <poll.h>
 #include <curl/curl.h>
@@ -94,9 +98,10 @@ class CurlHttpOperation {
             std::string method,
             std::string url,
             IHttpResponseCallback* callback,
-            // requestHeaders is copied into the curl_slist during construction
-            // and need not outlive this operation. requestBody is stored by
-            // reference and read by Send(), so it must outlive this operation.
+            // requestHeaders is copied into the curl_slist during construction and
+            // need not outlive this operation. requestBody is stored by reference;
+            // CurlHttpRequest destroys this operation (which joins the worker) before
+            // destroying its inherited request-body storage.
             const std::map<std::string, std::string>& requestHeaders,
             const std::vector<uint8_t>& requestBody,
             // Default connectivity and response size options
@@ -173,13 +178,22 @@ class CurlHttpOperation {
      */
     virtual ~CurlHttpOperation()
     {
-        // Given the request has not been aborted we should wait for completion here
-        // This guarantees the lifetime of this request.
-        if (result.valid())
+        if (m_worker.joinable())
         {
-            result.wait();
+            if (m_worker.get_id() == std::this_thread::get_id())
+            {
+                // The completion callback can release the owning request on this
+                // worker. Detach rather than joining the current thread; Send() has
+                // finished and the worker does not touch this operation afterward.
+                m_worker.detach();
+            }
+            else
+            {
+                m_worker.join();
+            }
         }
-        DispatchEvent(OnDestroy);
+
+        DispatchDestroyEvent();
         res = CURLE_OK;
         curl_easy_cleanup(curl);
         curl_slist_free_all(m_headersChunk);
@@ -317,14 +331,46 @@ class CurlHttpOperation {
         return res;
     }
 
-    std::future<long> & SendAsync(std::function<void(CurlHttpOperation &)> callback = nullptr) {
-        result = std::async(std::launch::async, [this, callback] {
-            long result = Send();
-            if (callback!=nullptr)
-                callback(*this);
-            return result;
-        });
-        return result;
+    void SendAsync(std::function<void(CurlHttpOperation &)> callback = nullptr) {
+        // A newly created std::thread may run before it is assigned to m_worker.
+        // Hold this gate until the assignment completes so a fast failure cannot
+        // destroy the operation from its callback while SendAsync still uses it.
+        {
+            std::lock_guard<std::mutex> startGuard(m_workerStartMtx);
+            if (m_sendAttempted)
+            {
+                throw std::logic_error("CurlHttpOperation is single-use");
+            }
+            m_sendAttempted = true;
+
+            try
+            {
+                m_worker = std::thread([this, callback]() {
+                    {
+                        std::lock_guard<std::mutex> startGuard(m_workerStartMtx);
+                    }
+                    try
+                    {
+                        Send();
+                    }
+                    catch (...)
+                    {
+                        // std::async stored worker exceptions in its unobserved
+                        // future. A raw thread must contain them.
+                        res = CURLE_FAILED_INIT;
+                    }
+                    Complete(callback);
+                });
+                return;
+            }
+            catch (...)
+            {
+                // Callable allocation/copy or std::thread creation failed.
+            }
+        }
+
+        res = CURLE_FAILED_INIT;
+        Complete(callback);
     }
 
     /**
@@ -437,17 +483,15 @@ class CurlHttpOperation {
 
     CURL *curl;                     // Local curl instance
     CURLcode res = CURLE_OK;        // Curl result OR HTTP status code if successful
-    
+
     IHttpResponseCallback* m_callback = nullptr;
 
     // Request values
     std::string m_method;
     std::string m_url;
     std::string m_sslCaInfo;
-    // The SDK upload path keeps the owning IHttpRequest alive through the
-    // callback context until Send() completes; copying this body would duplicate
-    // every upload payload. Unlike CURLOPT_CAINFO, the body pointer is set and
-    // consumed during Send(), not retained from construction.
+    // The owning CurlHttpRequest destroys this operation before its inherited
+    // request-body storage, and cross-thread destruction joins the worker.
     const std::vector<uint8_t>& requestBody;
     struct curl_slist *m_headersChunk = nullptr;
 
@@ -464,7 +508,46 @@ class CurlHttpOperation {
     size_t sendlen   = 0;        // # bytes sent by client
     size_t acklen    = 0;        // # bytes ack by server
 
-    std::future<long>       result;
+    std::mutex m_workerStartMtx;
+    bool m_sendAttempted = false;
+    std::thread m_worker;
+    std::atomic<bool> m_destroyEventDispatched { false };
+
+    void DispatchDestroyEvent() noexcept
+    {
+        bool expected = false;
+        if (m_destroyEventDispatched.compare_exchange_strong(
+                expected, true, std::memory_order_acq_rel))
+        {
+            try
+            {
+                DispatchEvent(OnDestroy);
+            }
+            catch (...)
+            {
+                // State observers must not terminate the worker or destructor.
+            }
+        }
+    }
+
+    void Complete(const std::function<void(CurlHttpOperation &)>& callback) noexcept
+    {
+        // Preserve the documented state event while m_callback is still valid.
+        // The completion callback can release the last owner, so this must remain
+        // the worker's final access to the operation.
+        DispatchDestroyEvent();
+        try
+        {
+            if (callback != nullptr)
+            {
+                callback(*this);
+            }
+        }
+        catch (...)
+        {
+            // Match the old unobserved-future behavior at the thread boundary.
+        }
+    }
 
     /**
      * Helper routine to wait for data on socket

tests/unittests/HttpClientCurlTests.cpp

@@ -13,6 +13,15 @@
 #include "http/HttpClient_Curl.hpp"
 #include "config/RuntimeConfig_Default.hpp"
 
+#include <future>
+#include <chrono>
+#include <memory>
+#include <atomic>
+#include <cstdlib>
+#include <functional>
+#include <stdexcept>
+#include <utility>
+
 using namespace testing;
 using namespace MAT;
 
@@ -127,6 +136,68 @@ TEST_F(HttpClientCurlTests, SetSslVerification_ConcurrentCallsNoRace)
     SUCCEED();
 }
 
+// --- Regression: EDEADLK self-join in ~CurlHttpOperation ---
+
+TEST_F(HttpClientCurlTests, SendAsync_DestroyOnWorkerThread_NoSelfJoin)
+{
+    struct TrackingCallback : public IHttpResponseCallback
+    {
+        std::atomic<int> destroyEvents { 0 };
+        void OnHttpResponse(IHttpResponse* response) override { delete response; }
+        void OnHttpStateEvent(HttpStateEvent state, void*, size_t) override
+        {
+            if (state == OnDestroy)
+            {
+                ++destroyEvents;
+            }
+        }
+    };
+
+    auto callback = std::make_shared<TrackingCallback>();
+    auto callbackDone = std::make_shared<std::promise<void>>();
+    auto done = callbackDone->get_future();
+
+    auto op = std::make_shared<CurlHttpOperation>(
+        "GET", "://malformed", callback.get(), m_headers, m_body,
+        false, 1 /*connTimeout*/, false /*sslVerify*/, "");
+
+    auto box = std::make_shared<std::shared_ptr<CurlHttpOperation>>(std::move(op));
+    (*box)->SendAsync([box, callback, callbackDone](CurlHttpOperation&) {
+        box->reset();
+        callbackDone->set_value();
+    });
+
+    if (done.wait_for(std::chrono::seconds(5)) != std::future_status::ready)
+    {
+        ADD_FAILURE() << "curl worker did not finish before fixture teardown";
+        std::abort();
+    }
+    EXPECT_EQ(callback->destroyEvents.load(), 1);
+}
+
+TEST_F(HttpClientCurlTests, SendAsync_CallbackCopyFailureStillCompletes)
+{
+    struct ThrowOnCopy
+    {
+        explicit ThrowOnCopy(bool& invoked) : invoked(&invoked) {}
+        ThrowOnCopy(ThrowOnCopy&&) = default;
+        ThrowOnCopy(const ThrowOnCopy&) { throw std::logic_error("copy failed"); }
+        void operator()(CurlHttpOperation&) const { *invoked = true; }
+        bool* invoked;
+    };
+
+    CurlHttpOperation op(
+        "GET", "://malformed", nullptr, m_headers, m_body,
+        false, 1 /*connTimeout*/, false /*sslVerify*/, "");
+    bool callbackInvoked = false;
+    std::function<void(CurlHttpOperation&)> callback { ThrowOnCopy(callbackInvoked) };
+
+    EXPECT_NO_THROW(op.SendAsync(std::move(callback)));
+    EXPECT_TRUE(callbackInvoked);
+    EXPECT_EQ(op.GetResponseCode(), CURLE_FAILED_INIT);
+    EXPECT_THROW(op.SendAsync(), std::logic_error);
+}
+
 // --- Response-size cap (memory-amplification DoS hardening) ---
 
 class HttpClientCurlResponseCapTests : public ::testing::Test,
@@ -138,9 +209,7 @@ class HttpClientCurlResponseCapTests : public ::testing::Test,
     HttpClient_Curl m_client;
     // The client never takes ownership of the request (it only stores a raw pointer
     // and erases it); the fixture owns it and frees it in TearDown -- on the main
-    // thread, after the transfer has completed. Freeing it inside OnHttpResponse
-    // would destroy the CurlHttpOperation from within its own async task, whose
-    // destructor waits on that task (a self-join deadlock).
+    // thread, after the transfer has completed.
     std::unique_ptr<IHttpRequest> m_request;
     std::string     m_hostname;
     size_t          m_responseBodySize {0};