CXXCBC-618: fix memory leak and data race

using std::function along with std::shared_ptr for promise might affect
the lifetime of the cluster object and result in memory leak and data
races during destruction

Author: avsej

core/impl/cluster.cxx

@@ -240,7 +240,7 @@ class cluster_impl : public std::enable_shared_from_this<cluster_impl>
 
   void open(const std::string& connection_string,
             const cluster_options& options,
-            cluster_connect_handler&& handler)
+            core::utils::movable_function<void(error, cluster)>&& handler)
   {
     core_.open(
       options_to_origin(connection_string, options),
@@ -512,39 +512,52 @@ cluster::search(std::string index_name,
   return barrier->get_future();
 }
 
-auto
-cluster::connect(const std::string& connection_string,
-                 const cluster_options& options) -> std::future<std::pair<error, cluster>>
+namespace
 {
-  auto barrier = std::make_shared<std::promise<std::pair<error, cluster>>>();
-  auto future = barrier->get_future();
-  connect(connection_string, options, [barrier](auto err, auto c) {
-    barrier->set_value({ std::move(err), std::move(c) });
-  });
-  return future;
-}
-
 void
-cluster::connect(const std::string& connection_string,
-                 const cluster_options& options,
-                 cluster_connect_handler&& handler)
+connect_in_background(const std::string& connection_string,
+                      const cluster_options& options,
+                      core::utils::movable_function<void(error, cluster)>&& handler)
 {
   // Spawn new thread for connection to ensure that cluster_impl pointer will
   // not be deallocated in IO thread in case of error.
   std::thread([connection_string, options, handler = std::move(handler)]() {
-    auto barrier = std::make_shared<std::promise<std::pair<error, cluster>>>();
-    auto future = barrier->get_future();
+    std::promise<std::pair<error, cluster>> barrier;
+    auto future = barrier.get_future();
     {
       auto impl = std::make_shared<cluster_impl>();
-      impl->open(connection_string, options, [barrier](auto err, auto c) {
-        barrier->set_value({ std::move(err), std::move(c) });
-      });
+      impl->open(
+        connection_string, options, [barrier = std::move(barrier)](auto err, auto c) mutable {
+          barrier.set_value({ std::move(err), std::move(c) });
+        });
     }
 
     auto [err, c] = future.get();
     handler(std::move(err), std::move(c));
   }).detach();
 }
+} // namespace
+
+auto
+cluster::connect(const std::string& connection_string,
+                 const cluster_options& options) -> std::future<std::pair<error, cluster>>
+{
+  std::promise<std::pair<error, cluster>> barrier;
+  auto future = barrier.get_future();
+  connect_in_background(
+    connection_string, options, [barrier = std::move(barrier)](auto err, auto c) mutable {
+      barrier.set_value({ std::move(err), std::move(c) });
+    });
+  return future;
+}
+
+void
+cluster::connect(const std::string& connection_string,
+                 const cluster_options& options,
+                 cluster_connect_handler&& handler)
+{
+  connect_in_background(connection_string, options, std::move(handler));
+}
 
 auto
 cluster::notify_fork(fork_event event) -> void

core/io/http_command.hxx

@@ -49,13 +49,13 @@ struct http_command : public std::enable_shared_from_this<http_command<Request>>
   asio::steady_timer deadline;
   Request request;
   encoded_request_type encoded;
-  std::shared_ptr<tracing::tracer_wrapper> tracer_;
+  std::shared_ptr<tracing::tracer_wrapper> tracer_{ nullptr };
   std::shared_ptr<couchbase::tracing::request_span> span_{ nullptr };
   std::shared_ptr<metrics::meter_wrapper> meter_{};
   std::shared_ptr<io::http_session> session_{};
   http_command_handler handler_{};
   std::chrono::milliseconds timeout_{};
-  std::string client_context_id_;
+  std::string client_context_id_{};
   std::shared_ptr<couchbase::tracing::request_span> parent_span{ nullptr };
 #ifdef COUCHBASE_CXX_CLIENT_COLUMNAR
   std::chrono::milliseconds dispatch_timeout_{};

core/io/http_message.hxx

@@ -36,9 +36,9 @@ struct streaming_settings {
 };
 
 struct http_request {
-  service_type type;
+  service_type type{};
   std::string method{};
-  std::string path;
+  std::string path{};
   std::map<std::string, std::string> headers{};
   std::string body{};
   std::optional<streaming_settings> streaming{};

core/io/http_session_manager.hxx

@@ -208,8 +208,7 @@ public:
             std::scoped_lock lock(sessions_mutex_);
             busy_sessions_[type].push_back(session);
           }
-          operations::http_noop_request request{};
-          request.type = type;
+          operations::http_noop_request request{ type };
           request.timeout = timeout;
 #ifdef COUCHBASE_CXX_CLIENT_COLUMNAR
           auto cmd = std::make_shared<operations::http_command<operations::http_noop_request>>(

core/metrics/meter_wrapper.hxx

@@ -33,9 +33,9 @@
 namespace couchbase::core::metrics
 {
 struct metric_attributes {
-  couchbase::core::service_type service;
-  std::string operation;
-  std::error_code ec;
+  couchbase::core::service_type service{};
+  std::string operation{};
+  std::error_code ec{};
   std::optional<std::string> bucket_name{};
   std::optional<std::string> scope_name{};
   std::optional<std::string> collection_name{};