CXXCBC-745: Implement lazy connections mode

Do not rush to open KV connections after the first configuration has
been received. Instead wait for the first operation that would be
mapped to the socket.

Author: avsej

core/bucket.cxx

@@ -19,6 +19,7 @@
 
 #include "collection_id_cache_entry.hxx"
 #include "core/app_telemetry_meter.hxx"
+#include "core/cluster_options.hxx"
 #include "core/config_listener.hxx"
 #include "core/document_id.hxx"
 #include "core/error_context/key_value_error_map_info.hxx"
@@ -319,11 +320,11 @@ class bucket_impl
   {
     if (req->key_.empty()) {
       if (auto server = server_by_vbucket(req->vbucket_, req->replica_index_); server) {
-        return find_session_by_index(server.value());
+        return find_or_connect_session_by_index(server.value());
       }
     } else if (auto [partition, server] = map_id(req->key_, req->replica_index_); server) {
       req->vbucket_ = partition;
-      return find_session_by_index(server.value());
+      return find_or_connect_session_by_index(server.value());
     }
     return std::nullopt;
   }
@@ -367,6 +368,53 @@ class bucket_impl
     return { 0, std::nullopt };
   }
 
+  void connect_session(std::size_t index)
+  {
+    const std::scoped_lock lock(config_mutex_, sessions_mutex_);
+    if (!config_) {
+      return;
+    }
+
+    const auto& node = config_->nodes[index];
+
+    const auto& hostname = node.hostname_for(origin_.options().network);
+    auto port = node.port_or(
+      origin_.options().network, service_type::key_value, origin_.options().enable_tls, 0);
+    if (port == 0) {
+      return;
+    }
+
+    const couchbase::core::origin origin(origin_.credentials(), hostname, port, origin_.options());
+    io::mcbp_session session =
+      origin_.options().enable_tls
+        ? io::mcbp_session(
+            client_id_, node.node_uuid, ctx_, tls_, origin, state_listener_, name_, known_features_)
+        : io::mcbp_session(
+            client_id_, node.node_uuid, ctx_, origin, state_listener_, name_, known_features_);
+    CB_LOG_DEBUG(R"({} rev={}, connect idx={}, session="{}", address="{}:{}")",
+                 log_prefix_,
+                 config_->rev_str(),
+                 node.index,
+                 session.id(),
+                 hostname,
+                 port);
+    session.bootstrap(
+      [self = shared_from_this(), session](std::error_code err,
+                                           topology::configuration cfg) mutable -> void {
+        if (err) {
+          return self->remove_session(session.id());
+        }
+        self->update_config(std::move(cfg));
+        session.on_configuration_update(self);
+        session.on_stop([id = session.id(), self]() -> void {
+          self->remove_session(id);
+        });
+        self->drain_deferred_queue({});
+      },
+      true);
+    sessions_.insert_or_assign(index, std::move(session));
+  }
+
   void restart_sessions()
   {
     const std::scoped_lock lock(config_mutex_, sessions_mutex_);
@@ -812,6 +860,10 @@ class bucket_impl
           continue;
         }
 
+        if (origin_.options().enable_lazy_connections) {
+          ++next_index;
+          continue;
+        }
         const couchbase::core::origin origin(
           origin_.credentials(), hostname, port, origin_.options());
         io::mcbp_session session =
@@ -885,6 +937,16 @@ class bucket_impl
     return {};
   }
 
+  [[nodiscard]] auto find_or_connect_session_by_index(std::size_t index)
+    -> std::optional<io::mcbp_session>
+  {
+    if (auto session = find_session_by_index(index); session) {
+      return session;
+    }
+    connect_session(index);
+    return {};
+  }
+
   [[nodiscard]] auto next_session_index() -> std::size_t
   {
     const std::scoped_lock lock(sessions_mutex_);
@@ -1206,4 +1268,10 @@ bucket::direct_re_queue(const std::shared_ptr<mcbp::queue_request>& req, bool is
 {
   return impl_->direct_re_queue(req, is_retry);
 }
+
+void
+bucket::connect_session(std::size_t index)
+{
+  return impl_->connect_session(index);
+}
 } // namespace couchbase::core

core/bucket.hxx

@@ -105,6 +105,8 @@ public:
     });
   }
 
+  void connect_session(std::size_t index);
+
   template<typename Request>
   void map_and_send(std::shared_ptr<operations::mcbp_command<bucket, Request>> cmd)
   {
@@ -142,6 +144,9 @@ public:
         session.has_value() ? session->bootstrap_address() : "",
         session.has_value() && session->has_config(),
         config_rev());
+      if (!session) {
+        connect_session(index);
+      }
       return defer_command([self = shared_from_this(), cmd](std::error_code ec) {
         if (ec == errc::common::request_canceled) {
           return cmd->cancel(retry_reason::do_not_retry);

core/cluster_options.hxx

@@ -113,6 +113,7 @@ public:
   };
   bool preserve_bootstrap_nodes_order{ false };
   bool allow_enterprise_analytics{ false };
+  bool enable_lazy_connections{ false };
 };
 
 } // namespace couchbase::core

core/utils/connection_string.cxx

@@ -590,6 +590,8 @@ extract_options(connection_string& connstr)
       parse_option(connstr.options.preserve_bootstrap_nodes_order, name, value, connstr.warnings);
     } else if (name == "allow_enterprise_analytics") {
       parse_option(connstr.options.allow_enterprise_analytics, name, value, connstr.warnings);
+    } else if (name == "enable_lazy_connections") {
+      parse_option(connstr.options.enable_lazy_connections, name, value, connstr.warnings);
     } else {
       connstr.warnings.push_back(
         fmt::format(R"(unknown parameter "{}" in connection string (value "{}"))", name, value));