Improve error handling for failed socket operations.

Author: kinyoklion

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

@@ -4,7 +4,7 @@
 
 #include "http_requester.hpp"
 #include "asio_requester.hpp"
-#include <launchdarkly/network/curl_multi_manager.hpp>
+#include <launchdarkly/network/detail/curl_multi_manager.hpp>
 #include <launchdarkly/config/shared/built/http_properties.hpp>
 #include <functional>
 

libs/internal/src/network/curl_requester.cpp

@@ -234,6 +234,12 @@ void CurlRequester::PerformRequestWithMulti(std::shared_ptr<CurlMultiManager> mu
             return;
         }
 
+        // Handle manager error
+        if (result.type == CurlMultiManager::Result::Type::ManagerError) {
+            ctx->callback(HttpResult("Internal manager error"));
+            return;
+        }
+
         // Check for errors
         if (result.curl_code != CURLE_OK) {
             std::string error_message = kErrorCurlPrefix;

libs/networking/include/launchdarkly/network/curl_multi_manager.hpp renamed to libs/networking/include/launchdarkly/network/detail/curl_multi_manager.hpp

@@ -35,12 +35,13 @@ using SocketHandle = boost::asio::ip::tcp::socket;
 class CurlMultiManager : public std::enable_shared_from_this<CurlMultiManager> {
 public:
     /**
-     * Result of a CURL operation - either CURLcode or read timeout.
+     * Result of a CURL operation - CURLcode, read timeout, or manager error.
      */
     struct Result {
         enum class Type {
             CurlCode,
-            ReadTimeout
+            ReadTimeout,
+            ManagerError  // Internal error in CurlMultiManager (e.g., failed to add handle)
         };
 
         Type type;
@@ -53,6 +54,10 @@ class CurlMultiManager : public std::enable_shared_from_this<CurlMultiManager> {
         static Result FromReadTimeout() {
             return Result{Type::ReadTimeout, CURLE_OK};
         }
+
+        static Result FromManagerError() {
+            return Result{Type::ManagerError, CURLE_OK};
+        }
     };
 
     /**
@@ -123,7 +128,7 @@ class CurlMultiManager : public std::enable_shared_from_this<CurlMultiManager> {
         std::shared_ptr<std::function<void()>> write_handler;
     };
 
-    void start_socket_monitor(SocketInfo* socket_info, int action);
+    bool start_socket_monitor(SocketInfo* socket_info, int action);
 
     // Reset read timeout timer for a handle
     void reset_read_timeout(CURL* easy);

libs/networking/src/curl_multi_manager.cpp

@@ -1,11 +1,9 @@
 #ifdef LD_CURL_NETWORKING
 
-#include "launchdarkly/network/curl_multi_manager.hpp"
+#include "../include/launchdarkly/network/detail/curl_multi_manager.hpp"
 
 #include <boost/asio/post.hpp>
 
-#include <iostream>
-
 namespace launchdarkly::network {
 std::shared_ptr<CurlMultiManager> CurlMultiManager::create(
     boost::asio::any_io_executor executor) {
@@ -58,8 +56,10 @@ void CurlMultiManager::add_handle(const std::shared_ptr<CURL>& easy,
             curl_slist_free_all(headers);
         }
 
-        std::cerr << "Failed to add handle to multi: "
-            << curl_multi_strerror(rc) << std::endl;
+        // Failed to add handle - invoke callback with manager error
+        boost::asio::post(executor_, [callback = std::move(callback), easy]() {
+            callback(easy, Result::FromManagerError());
+        });
         return;
     }
 
@@ -111,7 +111,10 @@ int CurlMultiManager::socket_callback(CURL* easy,
         // Add or update socket in managed container
         auto [it, inserted] = manager->sockets_.try_emplace(
             s, SocketInfo{s, nullptr, 0});
-        manager->start_socket_monitor(&it->second, what);
+        if (!manager->start_socket_monitor(&it->second, what)) {
+            // Failed to start monitoring - return error to CURL
+            return -1;
+        }
     }
 
     return 0;
@@ -153,15 +156,11 @@ int CurlMultiManager::timer_callback(CURLM* multi,
 void CurlMultiManager::handle_socket_action(curl_socket_t s,
                                             const int event_bitmask) {
     int running_handles = 0;
-    const CURLMcode rc = curl_multi_socket_action(multi_handle_.get(), s,
-                                                  event_bitmask,
-                                                  &running_handles);
-
-    if (rc != CURLM_OK) {
-        std::cerr << "curl_multi_socket_action failed: "
-            << curl_multi_strerror(rc) << std::endl;
-    }
+    curl_multi_socket_action(multi_handle_.get(), s,
+                             event_bitmask,
+                             &running_handles);
 
+    // Check for completed transfers - any errors will be reported via CURLMsg
     check_multi_info();
 
     if (running_handles != still_running_) {
@@ -233,7 +232,7 @@ void CurlMultiManager::check_multi_info() {
     }
 }
 
-void CurlMultiManager::start_socket_monitor(SocketInfo* socket_info,
+bool CurlMultiManager::start_socket_monitor(SocketInfo* socket_info,
                                             const int action) {
     if (!socket_info->handle) {
         // Create tcp::socket and assign the native socket handle
@@ -243,14 +242,15 @@ void CurlMultiManager::start_socket_monitor(SocketInfo* socket_info,
         // Assign the CURL socket to the ASIO socket
         // tcp::socket::assign works with native socket handles on both platforms
         boost::system::error_code ec;
-        socket_info->handle->assign(boost::asio::ip::tcp::v4(),
+        // clang-tidy things that this has a return value, but instead the return
+        // is being handled by output parameter.
+        socket_info->handle->assign(boost::asio::ip::tcp::v4(), // NOLINT(*-unused-return-value)
                                     socket_info->sockfd, ec);
 
         if (ec) {
-            std::cerr << "Failed to assign socket: " << ec.message() <<
-                std::endl;
+            // Failed to assign socket - CURL will handle the error
             socket_info->handle.reset();
-            return;
+            return false;
         }
     }
 
@@ -356,6 +356,8 @@ void CurlMultiManager::start_socket_monitor(SocketInfo* socket_info,
             (*socket_info->write_handler)(); // Initial call
         }
     }
+
+    return true;
 }
 
 void CurlMultiManager::reset_read_timeout(CURL* easy) {

libs/server-sent-events/src/curl_client.cpp

@@ -81,7 +81,7 @@ void CurlClient::do_run() {
         return;
     }
 
-    auto ctx = request_context_;
+    const auto ctx = request_context_;
     auto weak_self = weak_from_this();
     std::weak_ptr<RequestContext> weak_ctx = ctx;
     ctx->set_callbacks(Callbacks([weak_self, weak_ctx](const std::string& message) {
@@ -152,7 +152,7 @@ void CurlClient::async_backoff(std::string const& reason) {
     auto weak_self = weak_from_this();
     backoff_timer_.expires_after(backoff_.delay());
     backoff_timer_.async_wait([weak_self](const boost::system::error_code& ec) {
-        if (auto self = weak_self.lock()) {
+        if (const auto self = weak_self.lock()) {
             self->on_backoff(ec);
         }
     });
@@ -234,7 +234,7 @@ bool CurlClient::SetupCurlOptions(CURL* curl,
 
     // Add Last-Event-ID if we have one from previous connection
     if (context.last_event_id && !context.last_event_id->empty()) {
-        std::string last_event_header = "Last-Event-ID: " + *context.last_event_id;
+        const std::string last_event_header = "Last-Event-ID: " + *context.last_event_id;
         headers = curl_slist_append(headers, last_event_header.c_str());
     }
 
@@ -406,8 +406,8 @@ size_t CurlClient::HeaderCallback(const char* buffer,
 }
 
 void CurlClient::PerformRequestWithMulti(
-    std::shared_ptr<CurlMultiManager> multi_manager,
-    std::shared_ptr<RequestContext> context) {
+    const std::shared_ptr<CurlMultiManager>& multi_manager,
+    const std::shared_ptr<RequestContext>& context) {
 
     if (context->is_shutting_down()) {
         return;
@@ -416,7 +416,7 @@ void CurlClient::PerformRequestWithMulti(
     // Initialize parser for new connection (last_event_id is tracked separately)
     context->init_parser();
 
-    std::shared_ptr<CURL>curl (curl_easy_init(), curl_easy_cleanup);
+    const std::shared_ptr<CURL>curl (curl_easy_init(), curl_easy_cleanup);
     if (!curl) {
         if (context->is_shutting_down()) {
             return;
@@ -456,6 +456,14 @@ void CurlClient::PerformRequestWithMulti(
             return;
         }
 
+        // Check if this was a manager error (e.g., failed to add handle)
+        if (result.type == CurlMultiManager::Result::Type::ManagerError) {
+            if (!context->is_shutting_down()) {
+                context->backoff("internal manager error");
+            }
+            return;
+        }
+
         // Handle CURLcode result
         CURLcode res = result.curl_code;
 

libs/server-sent-events/src/curl_client.hpp

@@ -3,7 +3,7 @@
 #ifdef LD_CURL_NETWORKING
 
 #include <launchdarkly/sse/client.hpp>
-#include <launchdarkly/network/curl_multi_manager.hpp>
+#include <launchdarkly/network/detail/curl_multi_manager.hpp>
 #include "backoff.hpp"
 #include "parser.hpp"
 
@@ -24,23 +24,25 @@ namespace launchdarkly::sse {
 namespace http = beast::http;
 namespace net = boost::asio;
 
-using launchdarkly::network::CurlMultiManager;
+using network::CurlMultiManager;
 
 /**
- * The lifecycle of the CurlClient is managed in an RAII manner. This introduces
- * some complexity with interaction with CURL, which requires a thread to be
- * driven. The desutruction of the CurlClient will signal that the CURL thread
- * should stop. Though depending on the operation it may linger for some
- * time.
- *
- * The CurlClient itself is not accessible from the CURL thread and instead
- * all their shared state is in a RequestContext. The lifetime of this context
- * can exceed that of the CurlClient while CURL shuts down. This approach
- * prevents the lifetime of the CurlClient being attached to that of the
- * curl thread.
+ * The CurlClient uses the CURL multi-socket interface to allow for
+ * single-threaded usage of CURL. We drive this usage using boost::asio
+ * and every thing is executed in the IO context provided during client
+ * construction. Calling into the API of the client could be done from threads
+ * other than the IO context thread, so some thread-safety is required to
+ * manage those interactions. For example the CurlClient destructor will
+ * be ran on whatever thread last retains a reference to the client.
  */
 class CurlClient final : public Client,
                          public std::enable_shared_from_this<CurlClient> {
+    /**
+     * Structure containing callbacks between the CURL interactions and the
+     * IO executor. Callbacks are set while a connection is being established,
+     * instead of at construction time, to allow the use of weak_from_self.
+     * The weak_from_self method cannot be used during the constructor.
+     */
     struct Callbacks {
         std::function<void(std::string)> do_backoff;
         std::function<void(Event)> on_receive;
@@ -63,22 +65,29 @@ class CurlClient final : public Client,
         }
     };
 
+    /**
+     * The request context represents the state required by the executing CURL
+     * request. Not directly including the shared data in the CurlClient allows
+     * for easy seperation of its lifetime from that of the CURL client. This
+     * facilitates destruction of the CurlClient being used to stop in-progress
+     * requests.
+     *
+     * Also the CURL client can be destructed and pending tasks will still
+     * have a valid RequestContext and will detect the shutdown.
+     */
     class RequestContext {
-        // Only items used by both the curl thread and the executor/main
-        // thread need to be mutex protected.
         std::mutex mutex_;
         std::atomic<bool> shutting_down_;
         std::atomic<curl_socket_t> curl_socket_;
         // End mutex protected items.
         std::optional<Callbacks> callbacks_;
 
     public:
-        // SSE parser using common parser from parser.hpp
         using ParserBody = detail::EventBody<std::function<void(Event)>>;
-        std::unique_ptr<typename ParserBody::value_type> parser_body;
-        std::unique_ptr<typename ParserBody::reader> parser_reader;
+        std::unique_ptr<ParserBody::value_type> parser_body;
+        std::unique_ptr<ParserBody::reader> parser_reader;
 
-        // Track last event ID for reconnection (separate from parser state)
+        // Track last event ID for reconnection.
         std::optional<std::string> last_event_id;
 
         // Progress tracking for read timeout
@@ -161,13 +170,6 @@ class CurlClient final : public Client,
         void shutdown() {
             std::lock_guard lock(mutex_);
             shutting_down_ = true;
-            if (curl_socket_ != CURL_SOCKET_BAD) {
-#ifdef _WIN32
-                closesocket(curl_socket_);
-#else
-                close(curl_socket_);
-#endif
-            }
         }
 
 
@@ -193,8 +195,8 @@ class CurlClient final : public Client,
         }
 
         void init_parser() {
-            parser_body = std::make_unique<typename ParserBody::value_type>();
-            parser_reader = std::make_unique<typename ParserBody::reader>(*parser_body);
+            parser_body = std::make_unique<ParserBody::value_type>();
+            parser_reader = std::make_unique<ParserBody::reader>(*parser_body);
             parser_reader->init();
         }
     };
@@ -227,8 +229,8 @@ class CurlClient final : public Client,
     void async_backoff(std::string const& reason);
     void on_backoff(boost::system::error_code const& ec);
     static void PerformRequestWithMulti(
-        std::shared_ptr<CurlMultiManager> multi_manager,
-        std::shared_ptr<RequestContext> context);
+        const std::shared_ptr<CurlMultiManager>& multi_manager,
+        const std::shared_ptr<RequestContext>& context);
 
     static size_t WriteCallback(const char* data,
                                 size_t size,