Contract tests passing with multi support.

Author: kinyoklion

libs/internal/include/launchdarkly/network/curl_multi_manager.hpp

@@ -54,9 +54,10 @@ class CurlMultiManager : public std::enable_shared_from_this<CurlMultiManager> {
     /**
      * Add an easy handle to be managed.
      * @param easy The CURL easy handle (must be configured)
+     * @param headers The curl_slist headers (will be freed automatically)
      * @param callback Called when the transfer completes
      */
-    void add_handle(CURL* easy, CompletionCallback callback);
+    void add_handle(CURL* easy, curl_slist* headers, CompletionCallback callback);
 
     /**
      * Remove an easy handle from management.
@@ -86,7 +87,7 @@ class CurlMultiManager : public std::enable_shared_from_this<CurlMultiManager> {
     // Per-socket data
     struct SocketInfo {
         curl_socket_t sockfd;
-        std::unique_ptr<boost::asio::posix::stream_descriptor> descriptor;
+        std::shared_ptr<boost::asio::posix::stream_descriptor> descriptor;
         int action{0}; // CURL_POLL_IN, CURL_POLL_OUT, etc.
     };
 
@@ -99,6 +100,7 @@ class CurlMultiManager : public std::enable_shared_from_this<CurlMultiManager> {
 
     std::mutex mutex_;
     std::map<CURL*, CompletionCallback> callbacks_;
+    std::map<CURL*, curl_slist*> headers_;
     int still_running_{0};
 };
 

libs/internal/src/network/curl_multi_manager.cpp

@@ -33,20 +33,54 @@ CurlMultiManager::CurlMultiManager(boost::asio::any_io_executor executor)
 
 CurlMultiManager::~CurlMultiManager() {
     if (multi_handle_) {
+        // Extract and clear pending handles, callbacks, and headers
+        std::map<CURL*, CompletionCallback> pending_callbacks;
+        std::map<CURL*, curl_slist*> pending_headers;
+        {
+            std::lock_guard<std::mutex> lock(mutex_);
+            pending_callbacks = std::move(callbacks_);
+            pending_headers = std::move(headers_);
+            callbacks_.clear();
+            headers_.clear();
+        }
+
+        // Remove handles from multi and cleanup resources
+        // Do NOT invoke callbacks as they may access destroyed objects
+        for (auto& [easy, callback] : pending_callbacks) {
+            curl_multi_remove_handle(multi_handle_, easy);
+
+            // Free headers if they exist for this handle
+            auto header_it = pending_headers.find(easy);
+            if (header_it != pending_headers.end() && header_it->second) {
+                curl_slist_free_all(header_it->second);
+            }
+
+            curl_easy_cleanup(easy);
+        }
+
         curl_multi_cleanup(multi_handle_);
     }
 }
 
-void CurlMultiManager::add_handle(CURL* easy, CompletionCallback callback) {
+void CurlMultiManager::add_handle(CURL* easy, curl_slist* headers, CompletionCallback callback) {
     {
         std::lock_guard<std::mutex> lock(mutex_);
         callbacks_[easy] = std::move(callback);
+        headers_[easy] = headers;
     }
 
     CURLMcode rc = curl_multi_add_handle(multi_handle_, easy);
     if (rc != CURLM_OK) {
         std::lock_guard<std::mutex> lock(mutex_);
         callbacks_.erase(easy);
+
+        // Free headers on error
+        auto header_it = headers_.find(easy);
+        if (header_it != headers_.end() && header_it->second) {
+            curl_slist_free_all(header_it->second);
+        }
+        headers_.erase(easy);
+
         std::cerr << "Failed to add handle to multi: "
                   << curl_multi_strerror(rc) << std::endl;
     }
@@ -57,6 +91,13 @@ void CurlMultiManager::remove_handle(CURL* easy) {
 
     std::lock_guard<std::mutex> lock(mutex_);
     callbacks_.erase(easy);
+
+    // Free headers if they exist
+    auto header_it = headers_.find(easy);
+    if (header_it != headers_.end() && header_it->second) {
+        curl_slist_free_all(header_it->second);
+    }
+    headers_.erase(easy);
 }
 
 int CurlMultiManager::socket_callback(CURL* easy, curl_socket_t s, int what,
@@ -144,18 +185,31 @@ void CurlMultiManager::check_multi_info() {
             CURLcode result = msg->data.result;
 
             CompletionCallback callback;
+            curl_slist* headers = nullptr;
             {
                 std::lock_guard<std::mutex> lock(mutex_);
                 auto it = callbacks_.find(easy);
                 if (it != callbacks_.end()) {
                     callback = std::move(it->second);
                     callbacks_.erase(it);
                 }
+
+                // Get and remove headers
+                auto header_it = headers_.find(easy);
+                if (header_it != headers_.end()) {
+                    headers = header_it->second;
+                    headers_.erase(header_it);
+                }
             }
 
             // Remove from multi handle
             curl_multi_remove_handle(multi_handle_, easy);
 
+            // Free headers
+            if (headers) {
+                curl_slist_free_all(headers);
+            }
+
             // Invoke completion callback
             if (callback) {
                 boost::asio::post(executor_, [callback = std::move(callback),
@@ -170,7 +224,7 @@ void CurlMultiManager::check_multi_info() {
 void CurlMultiManager::start_socket_monitor(SocketInfo* socket_info, int action) {
     if (!socket_info->descriptor) {
         // Create descriptor for this socket
-        socket_info->descriptor = std::make_unique<
+        socket_info->descriptor = std::make_shared<
             boost::asio::posix::stream_descriptor>(executor_);
         socket_info->descriptor->assign(socket_info->sockfd);
     }
@@ -182,28 +236,72 @@ void CurlMultiManager::start_socket_monitor(SocketInfo* socket_info, int action)
 
     // Monitor for read events
     if (action & CURL_POLL_IN) {
-        socket_info->descriptor->async_wait(
-            boost::asio::posix::stream_descriptor::wait_read,
-            [weak_self, sockfd](const boost::system::error_code& ec) {
-                if (!ec) {
+        // Use weak_ptr to safely detect when descriptor is deleted
+        std::weak_ptr<boost::asio::posix::stream_descriptor> weak_descriptor = socket_info->descriptor;
+
+        // Use shared_ptr for recursive lambda
+        auto read_handler = std::make_shared<std::function<void()>>();
+        *read_handler = [weak_self, sockfd, weak_descriptor, read_handler]() {
+            // Check if manager and descriptor are still valid
+            auto self = weak_self.lock();
+            auto descriptor = weak_descriptor.lock();
+            if (!self || !descriptor) {
+                return;
+            }
+
+            descriptor->async_wait(
+                boost::asio::posix::stream_descriptor::wait_read,
+                [weak_self, sockfd, weak_descriptor, read_handler](const boost::system::error_code& ec) {
+                    // If operation was canceled or had an error, don't re-register
+                    if (ec) {
+                        return;
+                    }
+
                     if (auto self = weak_self.lock()) {
                         self->handle_socket_action(sockfd, CURL_CSELECT_IN);
+
+                        // Always try to re-register for continuous monitoring
+                        // The validity check at the top of read_handler will stop it if needed
+                        (*read_handler)();  // Recursive call
                     }
-                }
-            });
+                });
+        };
+        (*read_handler)();  // Initial call
     }
 
     // Monitor for write events
     if (action & CURL_POLL_OUT) {
-        socket_info->descriptor->async_wait(
-            boost::asio::posix::stream_descriptor::wait_write,
-            [weak_self, sockfd](const boost::system::error_code& ec) {
-                if (!ec) {
+        // Use weak_ptr to safely detect when descriptor is deleted
+        std::weak_ptr<boost::asio::posix::stream_descriptor> weak_descriptor = socket_info->descriptor;
+
+        // Use shared_ptr for recursive lambda
+        auto write_handler = std::make_shared<std::function<void()>>();
+        *write_handler = [weak_self, sockfd, weak_descriptor, write_handler]() {
+            // Check if manager and descriptor are still valid
+            auto self = weak_self.lock();
+            auto descriptor = weak_descriptor.lock();
+            if (!self || !descriptor) {
+                return;
+            }
+
+            descriptor->async_wait(
+                boost::asio::posix::stream_descriptor::wait_write,
+                [weak_self, sockfd, weak_descriptor, write_handler](const boost::system::error_code& ec) {
+                    // If operation was canceled or had an error, don't re-register
+                    if (ec) {
+                        return;
+                    }
+
                     if (auto self = weak_self.lock()) {
                         self->handle_socket_action(sockfd, CURL_CSELECT_OUT);
+
+                        // Always try to re-register for continuous monitoring
+                        // The validity check at the top of write_handler will stop it if needed
+                        (*write_handler)();  // Recursive call
                     }
-                }
-            });
+                });
+        };
+        (*write_handler)();  // Initial call
     }
 }
 

libs/internal/src/network/curl_requester.cpp

@@ -103,17 +103,14 @@ void CurlRequester::PerformRequestWithMulti(std::shared_ptr<CurlMultiManager> mu
     // This will be cleaned up in the completion callback
     struct RequestContext {
         CURL* curl;
-        curl_slist* headers;
         std::string url;
         std::string body; // Keep body alive
         std::string response_body;
         HttpResult::HeadersType response_headers;
         std::function<void(const HttpResult&)> callback;
 
         ~RequestContext() {
-            if (headers) {
-                curl_slist_free_all(headers);
-            }
+            // Headers are managed by CurlMultiManager
             if (curl) {
                 curl_easy_cleanup(curl);
             }
@@ -122,9 +119,11 @@ void CurlRequester::PerformRequestWithMulti(std::shared_ptr<CurlMultiManager> mu
 
     auto ctx = std::make_shared<RequestContext>();
     ctx->curl = curl;
-    ctx->headers = nullptr;
     ctx->callback = std::move(cb);
 
+    // Headers will be managed by CurlMultiManager
+    curl_slist* headers = nullptr;
+
     // Helper macro to check curl_easy_setopt return values
     #define CURL_SETOPT_CHECK(handle, option, parameter) \
         do { \
@@ -133,6 +132,9 @@ void CurlRequester::PerformRequestWithMulti(std::shared_ptr<CurlMultiManager> mu
                 std::string error_message = kErrorCurlPrefix; \
                 error_message += "curl_easy_setopt failed for " #option ": "; \
                 error_message += curl_easy_strerror(code); \
+                if (headers) { \
+                    curl_slist_free_all(headers); \
+                } \
                 ctx->callback(HttpResult(error_message)); \
                 return; \
             } \
@@ -170,19 +172,18 @@ void CurlRequester::PerformRequestWithMulti(std::shared_ptr<CurlMultiManager> mu
     auto const& base_headers = request.Properties().BaseHeaders();
     for (auto const& [key, value] : base_headers) {
         std::string header = key + kHeaderSeparator + value;
-        const auto appendResult = curl_slist_append(ctx->headers, header.c_str());
+        const auto appendResult = curl_slist_append(headers, header.c_str());
         if (!appendResult) {
-            if (ctx->headers) {
-                curl_slist_free_all(ctx->headers);
-                ctx->headers = nullptr;
+            if (headers) {
+                curl_slist_free_all(headers);
             }
             ctx->callback(HttpResult(kErrorHeaderAppend));
             return;
         }
-        ctx->headers = appendResult;
+        headers = appendResult;
     }
-    if (ctx->headers) {
-        CURL_SETOPT_CHECK(curl, CURLOPT_HTTPHEADER, ctx->headers);
+    if (headers) {
+        CURL_SETOPT_CHECK(curl, CURLOPT_HTTPHEADER, headers);
     }
 
     // Set timeouts with millisecond precision
@@ -232,7 +233,8 @@ void CurlRequester::PerformRequestWithMulti(std::shared_ptr<CurlMultiManager> mu
     #undef CURL_SETOPT_CHECK
 
     // Add handle to multi manager for async processing
-    multi_manager->add_handle(curl, [ctx](CURL* easy, CURLcode result) {
+    // Headers will be freed automatically by CurlMultiManager
+    multi_manager->add_handle(curl, headers, [ctx](CURL* easy, CURLcode result) {
         // This callback runs on the executor when the request completes
 
         // Check for errors