Merge remote-tracking branch 'origin/main' into CXXCBC-496-error

* origin/main:
  CXXCBC-407: allow to use 0 max expiry for new collections (#569)
  CXXCBC-445: return request_canceled on IO error in HTTP session (#568)
  CXXCBC-509: Zone-Aware replica reads (#566)
  CXXCBC-511: Prevent use of HTTP session if idle timer has expired (#565)
  Improve test stability (#563)
  Always attempt to extract common query code if error has not been set (#561)
  Add feature check for scoped analyze_document in tests (#555)
  CXXCBC-30 Inconsistent behaviour when using subdoc opcodes incorrectly (#559)
  CXXCBC-503 - Additions for more protection against picking up a config with an empty vbucket map. (#558)
  CXXCBC-503: Ignore configuration if it contains an empty vBucketMap (#556)
  CXXCBC-489: Add version_7_2_0 eventing function language compatibility (#554)

.github/workflows/tests.yml

@@ -12,7 +12,8 @@ jobs:
       fail-fast: false
       matrix:
         server:
-          - 7.2.3
+          - 7.6.1
+          - 7.2.4
           - 7.1.6
           - 7.0.5
         suite:

CMakeLists.txt

@@ -168,6 +168,7 @@ set(couchbase_cxx_client_FILES
     core/impl/query_index_manager.cxx
     core/impl/query_string_query.cxx
     core/impl/regexp_query.cxx
+    core/impl/replica_utils.cxx
     core/impl/retry_action.cxx
     core/impl/retry_reason.cxx
     core/impl/scope.cxx

core/bucket.cxx

@@ -591,7 +591,25 @@ class bucket_impl
         bool sequence_changed = false;
         {
             std::scoped_lock lock(config_mutex_);
-            if (!config_) {
+            // MB-60405 fixes this for 7.6.2, but for earlier versions we need to protect against using a
+            // config that has an empty vbucket map.  Ideally we only run into this condition on initial
+            // bootstrap and that is handled in the session's update_config(), but just in case, only accept
+            // a config w/ a non-empty vbucket map.
+            if (config.vbmap && config.vbmap->size() == 0) {
+                if (!config_) {
+                    CB_LOG_WARNING("{} will not initialize configuration rev={} because config has an empty partition map",
+                                   log_prefix_,
+                                   config.rev_str());
+                } else {
+                    CB_LOG_WARNING("{} will not update the configuration old={} -> new={}, because new config has an empty partition map",
+                                   log_prefix_,
+                                   config_->rev_str(),
+                                   config.rev_str());
+                }
+                // this is to make sure we can get a correct config soon
+                poll_config(errc::network::configuration_not_available);
+                return;
+            } else if (!config_) {
                 CB_LOG_DEBUG("{} initialize configuration rev={}", log_prefix_, config.rev_str());
             } else if (config.force) {
                 CB_LOG_DEBUG("{} forced to accept configuration rev={}", log_prefix_, config.rev_str());

core/cluster.cxx

@@ -232,7 +232,9 @@ class cluster_impl : public std::enable_shared_from_this<cluster_impl>
                   return self->dns_srv_tracker_->get_srv_nodes([self, hostname = std::move(hostname), handler = std::move(handler)](
                                                                  origin::node_list nodes, std::error_code ec) mutable {
                       if (ec) {
-                          return self->close([ec, handler = std::move(handler)]() mutable { handler(ec); });
+                          return self->close([ec, handler = std::move(handler)]() mutable {
+                              handler(ec);
+                          });
                       }
                       if (!nodes.empty()) {
                           self->origin_.set_nodes(std::move(nodes));
@@ -510,7 +512,9 @@ class cluster_impl : public std::enable_shared_from_this<cluster_impl>
                     tls_.load_verify_file(origin_.options().trust_certificate, ec);
                     if (ec) {
                         CB_LOG_ERROR("[{}]: unable to load verify file \"{}\": {}", id_, origin_.options().trust_certificate, ec.message());
-                        return close([ec, handler = std::move(handler)]() mutable { return handler(ec); });
+                        return close([ec, handler = std::move(handler)]() mutable {
+                            return handler(ec);
+                        });
                     }
                 }
             }
@@ -520,13 +524,17 @@ class cluster_impl : public std::enable_shared_from_this<cluster_impl>
                 tls_.use_certificate_chain_file(origin_.certificate_path(), ec);
                 if (ec) {
                     CB_LOG_ERROR("[{}]: unable to load certificate chain \"{}\": {}", id_, origin_.certificate_path(), ec.message());
-                    return close([ec, handler = std::move(handler)]() mutable { return handler(ec); });
+                    return close([ec, handler = std::move(handler)]() mutable {
+                        return handler(ec);
+                    });
                 }
                 CB_LOG_DEBUG(R"([{}]: use TLS private key: "{}")", id_, origin_.key_path());
                 tls_.use_private_key_file(origin_.key_path(), asio::ssl::context::file_format::pem, ec);
                 if (ec) {
                     CB_LOG_ERROR("[{}]: unable to load private key \"{}\": {}", id_, origin_.key_path(), ec.message());
-                    return close([ec, handler = std::move(handler)]() mutable { return handler(ec); });
+                    return close([ec, handler = std::move(handler)]() mutable {
+                        return handler(ec);
+                    });
                 }
             }
             session_ = io::mcbp_session(id_, ctx_, tls_, origin_, dns_srv_tracker_);
@@ -559,7 +567,9 @@ class cluster_impl : public std::enable_shared_from_this<cluster_impl>
                   });
               }
               if (ec) {
-                  return self->close([ec, handler = std::move(handler)]() mutable { handler(ec); });
+                  return self->close([ec, handler = std::move(handler)]() mutable {
+                      handler(ec);
+                  });
               }
               handler(ec);
           });
@@ -616,7 +626,9 @@ class cluster_impl : public std::enable_shared_from_this<cluster_impl>
                       if (cluster->session_) {
                           cluster->session_->ping(collector->build_reporter(), timeout);
                       }
-                      cluster->for_each_bucket([&collector, &timeout](auto bucket) { bucket->ping(collector, timeout); });
+                      cluster->for_each_bucket([&collector, &timeout](auto bucket) {
+                          bucket->ping(collector, timeout);
+                      });
                   }
                   cluster->session_manager_->ping(services, timeout, collector, cluster->origin_.credentials());
               }
@@ -636,7 +648,9 @@ class cluster_impl : public std::enable_shared_from_this<cluster_impl>
             if (self->session_) {
                 res.services[service_type::key_value].emplace_back(self->session_->diag_info());
             }
-            self->for_each_bucket([&res](const auto& bucket) { bucket->export_diag_info(res); });
+            self->for_each_bucket([&res](const auto& bucket) {
+                bucket->export_diag_info(res);
+            });
             self->session_manager_->export_diag_info(res);
             handler(std::move(res));
         }));
@@ -653,7 +667,9 @@ class cluster_impl : public std::enable_shared_from_this<cluster_impl>
                 self->session_->stop(retry_reason::do_not_retry);
                 self->session_.reset();
             }
-            self->for_each_bucket([](auto bucket) { bucket->close(); });
+            self->for_each_bucket([](auto bucket) {
+                bucket->close();
+            });
             self->session_manager_->close();
             handler();
             self->work_.reset();
@@ -853,7 +869,13 @@ void
 cluster::execute(operations::get_all_replicas_request request,
                  utils::movable_function<void(operations::get_all_replicas_response)>&& handler) const
 {
-    return request.execute(impl_, std::move(handler));
+    auto bucket_name = request.id.bucket();
+    return open_bucket(bucket_name, [impl = impl_, request = std::move(request), handler = std::move(handler)](auto ec) mutable {
+        if (ec) {
+            return handler(operations::get_all_replicas_response{ make_key_value_error_context(ec, request.id) });
+        }
+        return request.execute(impl, std::move(handler));
+    });
 }
 
 void
@@ -873,7 +895,13 @@ void
 cluster::execute(operations::get_any_replica_request request,
                  utils::movable_function<void(operations::get_any_replica_response)>&& handler) const
 {
-    return request.execute(impl_, std::move(handler));
+    auto bucket_name = request.id.bucket();
+    return open_bucket(bucket_name, [impl = impl_, request = std::move(request), handler = std::move(handler)](auto ec) mutable {
+        if (ec) {
+            return handler(operations::get_any_replica_response{ make_key_value_error_context(ec, request.id) });
+        }
+        return request.execute(impl, std::move(handler));
+    });
 }
 
 void

core/cluster_options.hxx

@@ -85,6 +85,7 @@ struct cluster_options {
     std::size_t max_http_connections{ 0 };
     std::chrono::milliseconds idle_http_connection_timeout = timeout_defaults::idle_http_connection_timeout;
     std::string user_agent_extra{};
+    std::string server_group{};
     couchbase::transactions::transactions_config::built transactions{};
 
     bool dump_configuration{ false };

core/impl/analytics_index_manager.cxx

@@ -512,8 +512,8 @@ analytics_index_manager::create_dataverse(std::string dataverse_name,
 }
 
 auto
-analytics_index_manager::create_dataverse(std::string dataverse_name, const create_dataverse_analytics_options& options) const
-  -> std::future<error>
+analytics_index_manager::create_dataverse(std::string dataverse_name,
+                                          const create_dataverse_analytics_options& options) const -> std::future<error>
 {
     auto barrier = std::make_shared<std::promise<error>>();
     auto future = barrier->get_future();
@@ -532,8 +532,8 @@ analytics_index_manager::drop_dataverse(std::string dataverse_name,
 }
 
 auto
-analytics_index_manager::drop_dataverse(std::string dataverse_name, const drop_dataverse_analytics_options& options) const
-  -> std::future<error>
+analytics_index_manager::drop_dataverse(std::string dataverse_name,
+                                        const drop_dataverse_analytics_options& options) const -> std::future<error>
 {
     auto barrier = std::make_shared<std::promise<error>>();
     auto future = barrier->get_future();
@@ -637,8 +637,9 @@ analytics_index_manager::drop_index(std::string index_name,
 }
 
 auto
-analytics_index_manager::drop_index(std::string index_name, std::string dataset_name, const drop_index_analytics_options& options) const
-  -> std::future<error>
+analytics_index_manager::drop_index(std::string index_name,
+                                    std::string dataset_name,
+                                    const drop_index_analytics_options& options) const -> std::future<error>
 {
     auto barrier = std::make_shared<std::promise<error>>();
     auto future = barrier->get_future();
@@ -713,8 +714,7 @@ auto
 analytics_index_manager::get_pending_mutations(const get_pending_mutations_analytics_options& options) const
   -> std::future<std::pair<error, std::map<std::string, std::map<std::string, std::int64_t>>>>
 {
-    auto barrier =
-      std::make_shared<std::promise<std::pair<error, std::map<std::string, std::map<std::string, std::int64_t>>>>>();
+    auto barrier = std::make_shared<std::promise<std::pair<error, std::map<std::string, std::map<std::string, std::int64_t>>>>>();
     auto future = barrier->get_future();
     get_pending_mutations(options, [barrier](auto err, auto resp) mutable {
         barrier->set_value({ std::move(err), std::move(resp) });
@@ -731,8 +731,8 @@ analytics_index_manager::create_link(const management::analytics_link& link,
 }
 
 auto
-analytics_index_manager::create_link(const management::analytics_link& link, const create_link_analytics_options& options) const
-  -> std::future<error>
+analytics_index_manager::create_link(const management::analytics_link& link,
+                                     const create_link_analytics_options& options) const -> std::future<error>
 {
     auto barrier = std::make_shared<std::promise<error>>();
     auto future = barrier->get_future();
@@ -751,8 +751,8 @@ analytics_index_manager::replace_link(const management::analytics_link& link,
 }
 
 auto
-analytics_index_manager::replace_link(const management::analytics_link& link, const replace_link_analytics_options& options) const
-  -> std::future<error>
+analytics_index_manager::replace_link(const management::analytics_link& link,
+                                      const replace_link_analytics_options& options) const -> std::future<error>
 {
     auto barrier = std::make_shared<std::promise<error>>();
     auto future = barrier->get_future();
@@ -772,8 +772,9 @@ analytics_index_manager::drop_link(std::string link_name,
 }
 
 auto
-analytics_index_manager::drop_link(std::string link_name, std::string dataverse_name, const drop_link_analytics_options& options) const
-  -> std::future<error>
+analytics_index_manager::drop_link(std::string link_name,
+                                   std::string dataverse_name,
+                                   const drop_link_analytics_options& options) const -> std::future<error>
 {
     auto barrier = std::make_shared<std::promise<error>>();
     auto future = barrier->get_future();

core/impl/cluster.cxx

@@ -100,6 +100,7 @@ options_to_origin(const std::string& connection_string, const couchbase::cluster
     user_options.user_agent_extra = opts.behavior.user_agent_extra;
     user_options.network = opts.behavior.network;
 
+    user_options.server_group = opts.network.server_group;
     user_options.enable_tcp_keep_alive = opts.network.enable_tcp_keep_alive;
     user_options.tcp_keep_alive_interval = opts.network.tcp_keep_alive_interval;
     user_options.config_poll_interval = opts.network.config_poll_interval;
@@ -121,6 +122,7 @@ options_to_origin(const std::string& connection_string, const couchbase::cluster
             user_options.use_ip_protocol = core::io::ip_protocol::force_ipv6;
             break;
     }
+    user_options.server_group = opts.network.server_group;
 
     user_options.enable_compression = opts.compression.enabled;