RCBC-304: Add Cluster#wait_until_ready and Bucket#wait_until_ready

ext/rcb_backend.cxx

@@ -648,7 +648,7 @@ cb_backend_to_public_api_cluster(VALUE self) -> couchbase::cluster
   TypedData_Get_Struct(self, cb_backend_data, &cb_backend_type, backend);
 
   if (backend->instance == nullptr) {
-    rb_raise(exc_cluster_closed(), "Cluster has been closed already");
+    throw ruby_exception(exc_cluster_closed(), "Cluster has been closed already");
   }
 
   return *backend->instance;

ext/rcb_backend.hxx

@@ -33,6 +33,8 @@ class cluster;
 
 namespace couchbase::ruby
 {
+// Both throw ruby_exception with Error::ClusterClosed once the backend is closed, so they may be
+// called only from a method body registered through cb_method.
 auto
 cb_backend_to_public_api_cluster(VALUE self) -> couchbase::cluster;
 

ext/rcb_diagnostics.cxx

@@ -17,19 +17,53 @@
 
 #include <core/cluster.hxx>
 
+#include <couchbase/bucket.hxx>
+#include <couchbase/cluster.hxx>
+
 #include <spdlog/fmt/bundled/core.h>
 
 #include <future>
+#include <optional>
 
 #include <ruby.h>
 
 #include "rcb_backend.hxx"
+#include "rcb_exceptions.hxx"
 #include "rcb_utils.hxx"
 
 namespace couchbase::ruby
 {
 namespace
 {
+// Enumerator names are shared by core::service_type and couchbase::service_type.
+template<typename ServiceType>
+auto
+cb_service_type_from_symbol(VALUE entry) -> std::optional<ServiceType>
+{
+  if (entry == rb_id2sym(rb_intern("kv"))) {
+    return ServiceType::key_value;
+  }
+  if (entry == rb_id2sym(rb_intern("query"))) {
+    return ServiceType::query;
+  }
+  if (entry == rb_id2sym(rb_intern("analytics"))) {
+    return ServiceType::analytics;
+  }
+  if (entry == rb_id2sym(rb_intern("search"))) {
+    return ServiceType::search;
+  }
+  if (entry == rb_id2sym(rb_intern("views"))) {
+    return ServiceType::view;
+  }
+  if (entry == rb_id2sym(rb_intern("management"))) {
+    return ServiceType::management;
+  }
+  if (entry == rb_id2sym(rb_intern("eventing"))) {
+    return ServiceType::eventing;
+  }
+  return std::nullopt;
+}
+
 VALUE
 cb_Backend_diagnostics(VALUE self, VALUE report_id)
 {
@@ -151,21 +185,9 @@ cb_Backend_ping(VALUE self, VALUE bucket, VALUE options)
     if (!NIL_P(services)) {
       auto entries_num = static_cast<std::size_t>(RARRAY_LEN(services));
       for (std::size_t i = 0; i < entries_num; ++i) {
-        VALUE entry = rb_ary_entry(services, static_cast<long>(i));
-        if (entry == rb_id2sym(rb_intern("kv"))) {
-          selected_services.insert(core::service_type::key_value);
-        } else if (entry == rb_id2sym(rb_intern("query"))) {
-          selected_services.insert(core::service_type::query);
-        } else if (entry == rb_id2sym(rb_intern("analytics"))) {
-          selected_services.insert(core::service_type::analytics);
-        } else if (entry == rb_id2sym(rb_intern("search"))) {
-          selected_services.insert(core::service_type::search);
-        } else if (entry == rb_id2sym(rb_intern("views"))) {
-          selected_services.insert(core::service_type::view);
-        } else if (entry == rb_id2sym(rb_intern("management"))) {
-          selected_services.insert(core::service_type::management);
-        } else if (entry == rb_id2sym(rb_intern("eventing"))) {
-          selected_services.insert(core::service_type::eventing);
+        if (auto type = cb_service_type_from_symbol<core::service_type>(
+              rb_ary_entry(services, static_cast<long>(i)))) {
+          selected_services.insert(*type);
         }
       }
     }
@@ -248,12 +270,85 @@ cb_Backend_ping(VALUE self, VALUE bucket, VALUE options)
   }
   return Qnil;
 }
+
+VALUE
+cb_Backend_wait_until_ready(VALUE self, VALUE bucket, VALUE timeout, VALUE options)
+{
+  if (!NIL_P(bucket)) {
+    Check_Type(bucket, T_STRING);
+  }
+  if (!NIL_P(options)) {
+    Check_Type(options, T_HASH);
+  }
+
+  try {
+    const auto wait_timeout = cb_to_timeout(timeout, "timeout");
+
+    couchbase::wait_until_ready_options wait_options{};
+
+    VALUE desired_state = Qnil;
+    cb_extract_option_symbol(desired_state, options, "desired_state");
+    if (!NIL_P(desired_state)) {
+      if (desired_state == rb_id2sym(rb_intern("online"))) {
+        wait_options.desired_state(couchbase::cluster_state::online);
+      } else if (desired_state == rb_id2sym(rb_intern("degraded"))) {
+        wait_options.desired_state(couchbase::cluster_state::degraded);
+      } else {
+        VALUE message = Qnil;
+        cb_protect([&message, desired_state] {
+          message = rb_sprintf("unsupported desired_state, given %+" PRIsVALUE, desired_state);
+        });
+        throw ruby_exception(exc_invalid_argument(), message);
+      }
+    }
+
+    VALUE services = Qnil;
+    cb_extract_option_array(services, options, "service_types");
+    if (!NIL_P(services)) {
+      std::set<couchbase::service_type> selected_services{};
+      auto entries_num = static_cast<std::size_t>(RARRAY_LEN(services));
+      for (std::size_t i = 0; i < entries_num; ++i) {
+        VALUE entry = rb_ary_entry(services, static_cast<long>(i));
+        auto type = cb_service_type_from_symbol<couchbase::service_type>(entry);
+        if (!type) {
+          // inspect may raise; cb_protect lets selected_services unwind.
+          VALUE message = Qnil;
+          cb_protect([&message, entry] {
+            message = rb_sprintf("unsupported service type, given %+" PRIsVALUE, entry);
+          });
+          throw ruby_exception(exc_invalid_argument(), message);
+        }
+        selected_services.insert(*type);
+      }
+      wait_options.service_types(std::move(selected_services));
+    }
+
+    // Acquired after the options are parsed: a Ruby raise above must not skip its destructor.
+    auto cluster = cb_backend_to_public_api_cluster(self);
+    couchbase::error err{};
+    if (NIL_P(bucket)) {
+      auto f = cluster.wait_until_ready(wait_timeout, wait_options);
+      err = cb_wait_for_future(f);
+    } else {
+      auto f = cluster.bucket(cb_string_new(bucket)).wait_until_ready(wait_timeout, wait_options);
+      err = cb_wait_for_future(f);
+    }
+    if (err) {
+      cb_throw_error(err, "unable to wait until ready");
+    }
+  } catch (const std::system_error& se) {
+    throw ruby_exception(cb_map_error_code(
+      se.code(), fmt::format("failed to perform {}: {}", __func__, se.what()), false));
+  }
+  return Qnil;
+}
 } // namespace
 
 void
 init_diagnostics(VALUE cBackend)
 {
   rb_define_method(cBackend, "diagnostics", cb_method<cb_Backend_diagnostics>::invoke, 1);
   rb_define_method(cBackend, "ping", cb_method<cb_Backend_ping>::invoke, 2);
+  rb_define_method(cBackend, "wait_until_ready", cb_method<cb_Backend_wait_until_ready>::invoke, 3);
 }
 } // namespace couchbase::ruby

ext/rcb_observability.cxx

@@ -129,6 +129,6 @@ cb_Backend_cluster_labels(VALUE self)
 void
 init_observability(VALUE cBackend)
 {
-  rb_define_method(cBackend, "cluster_labels", cb_Backend_cluster_labels, 0);
+  rb_define_method(cBackend, "cluster_labels", cb_method<cb_Backend_cluster_labels>::invoke, 0);
 }
 } // namespace couchbase::ruby

ext/rcb_utils.cxx

@@ -24,6 +24,8 @@
 #include "rcb_exceptions.hxx"
 #include "rcb_utils.hxx"
 
+#include <chrono>
+#include <cinttypes>
 #include <cstdint>
 #include <ruby.h>
 
@@ -705,4 +707,38 @@ get_string(VALUE options, VALUE name)
   return {};
 }
 } // namespace options
+
+auto
+cb_to_timeout(VALUE timeout, const char* name, std::chrono::milliseconds max_timeout)
+  -> std::chrono::milliseconds
+{
+  if (!RB_INTEGER_TYPE_P(timeout)) {
+    VALUE message = Qnil;
+    cb_protect([&message, timeout, name] {
+      message = rb_sprintf("%s must be an Integer, but given %+" PRIsVALUE, name, timeout);
+    });
+    throw ruby_exception(rb_eArgError, message);
+  }
+  std::uint64_t timeout_ms = 0;
+  // The sign is -2 or -1 for a negative value, and 2 for one that does not fit timeout_ms.
+  if (int sign = rb_integer_pack(timeout,
+                                 &timeout_ms,
+                                 1,
+                                 sizeof(timeout_ms),
+                                 0,
+                                 INTEGER_PACK_LSWORD_FIRST | INTEGER_PACK_NATIVE_BYTE_ORDER);
+      sign < 0 || sign > 1 || timeout_ms > static_cast<std::uint64_t>(max_timeout.count())) {
+    VALUE message = Qnil;
+    cb_protect([&message, timeout, name, max = static_cast<std::int64_t>(max_timeout.count())] {
+      message =
+        rb_sprintf("%s must be between 0 and %" PRId64 " milliseconds, but given %+" PRIsVALUE,
+                   name,
+                   max,
+                   timeout);
+    });
+    throw ruby_exception(exc_invalid_argument(), message);
+  }
+  return std::chrono::milliseconds{ static_cast<std::int64_t>(timeout_ms) };
+}
+
 } // namespace couchbase::ruby

ext/rcb_utils.hxx

@@ -340,6 +340,20 @@ cb_extract_duration(Field& field, VALUE options, const char* name)
   }
 }
 
+// The default upper bound of cb_to_timeout. It lies far inside the range of steady_clock, so a
+// deadline steady_clock::now() + timeout cannot overflow.
+constexpr std::chrono::milliseconds cb_max_timeout{ std::chrono::hours{ 24 * 365 } };
+static_assert(cb_max_timeout < std::chrono::steady_clock::duration::max() / 2);
+
+// Converts timeout, an Integer number of milliseconds from 0 to max_timeout, which must not be
+// negative. Throws ArgumentError for a non-Integer, and Error::InvalidArgument for a value out of
+// that range; name labels the value in the messages. Call it only from a method body registered
+// through cb_method.
+[[nodiscard]] auto
+cb_to_timeout(VALUE timeout,
+              const char* name,
+              std::chrono::milliseconds max_timeout = cb_max_timeout) -> std::chrono::milliseconds;
+
 void
 cb_extract_content(std::vector<std::byte>& field, VALUE content);
 

lib/couchbase/bucket.rb

@@ -149,5 +149,25 @@ def ping(options = Options::Ping::DEFAULT)
         end
       end
     end
+
+    # Waits until the bucket reaches the desired state, actively pinging its services
+    #
+    # With +:online+, the wait also requires the bucket's vbucket map to be placed: every vbucket has its active copy
+    # and each replica the data nodes can hold assigned to a node. +:degraded+ only pings.
+    #
+    # +Thread#raise+ does not interrupt the wait. It ends on success, on +timeout+, or when the cluster is disconnected.
+    #
+    # @param [Integer, #in_milliseconds] timeout the maximum time to wait
+    # @param [Options::WaitUntilReady] options
+    #
+    # @return [void]
+    #
+    # @raise [Error::UnambiguousTimeout] if the desired state is not reached within +timeout+
+    # @raise [Error::ClusterClosed] if the cluster has been disconnected
+    # @raise [Error::InvalidArgument] if +timeout+ is negative or longer than 365 days, or +options+ name an
+    #   unsupported desired state or service type
+    def wait_until_ready(timeout, options = Options::WaitUntilReady::DEFAULT)
+      @backend.wait_until_ready(@name, Utils::Time.extract_duration(timeout), options.to_backend)
+    end
   end
 end

lib/couchbase/cluster.rb

@@ -348,6 +348,23 @@ def ping(options = Options::Ping::DEFAULT)
       end
     end
 
+    # Waits until the cluster reaches the desired state, actively pinging its services
+    #
+    # +Thread#raise+ does not interrupt the wait. It ends on success, on +timeout+, or when the cluster is disconnected.
+    #
+    # @param [Integer, #in_milliseconds] timeout the maximum time to wait
+    # @param [Options::WaitUntilReady] options
+    #
+    # @return [void]
+    #
+    # @raise [Error::UnambiguousTimeout] if the desired state is not reached within +timeout+
+    # @raise [Error::ClusterClosed] if the cluster has been disconnected
+    # @raise [Error::InvalidArgument] if +timeout+ is negative or longer than 365 days, or +options+ name an
+    #   unsupported desired state or service type
+    def wait_until_ready(timeout, options = Options::WaitUntilReady::DEFAULT)
+      @backend.wait_until_ready(nil, Utils::Time.extract_duration(timeout), options.to_backend)
+    end
+
     private
 
     # Initialize {Cluster} object

lib/couchbase/options.rb

@@ -1975,6 +1975,41 @@ def to_backend
       DEFAULT = Ping.new.freeze
     end
 
+    # Options for {Couchbase::Cluster#wait_until_ready} and {Couchbase::Bucket#wait_until_ready}
+    class WaitUntilReady
+      attr_accessor :desired_state # @return [Symbol]
+      attr_accessor :service_types # @return [Array<Symbol>, nil]
+
+      # Creates an instance of options for {Couchbase::Cluster#wait_until_ready} and {Couchbase::Bucket#wait_until_ready}
+      #
+      # @param [Symbol] desired_state the state to wait for
+      #   +:online+:: every endpoint the ping reports for each requested service responds
+      #   +:degraded+:: at least one endpoint the ping reports for each requested service responds
+      # @param [Array<Symbol>, nil] service_types the service types to wait for (same names as {Ping#service_types}).
+      #   When +nil+ or empty, every service in the ping report is considered. A requested service that no node runs
+      #   is never ready, so the wait times out.
+      #
+      # @yieldparam [WaitUntilReady] self
+      def initialize(desired_state: :online,
+                     service_types: nil)
+        @desired_state = desired_state
+        @service_types = service_types
+
+        yield self if block_given?
+      end
+
+      # @api private
+      def to_backend
+        {
+          desired_state: @desired_state,
+          service_types: @service_types,
+        }
+      end
+
+      # @api private
+      DEFAULT = WaitUntilReady.new.freeze
+    end
+
     # Options for {Couchbase::Cluster#analytics_query}
     class Analytics < Base
       attr_accessor :client_context_id # @return [String]

lib/couchbase/protostellar/bucket.rb

@@ -50,6 +50,10 @@ def collections
       def ping(_options = Options::Ping::DEFAULT)
         raise Couchbase::Error::FeatureNotAvailable, "The #{Protostellar::NAME} protocol does not support ping"
       end
+
+      def wait_until_ready(_timeout, _options = Options::WaitUntilReady::DEFAULT)
+        raise Couchbase::Error::FeatureNotAvailable, "The #{Protostellar::NAME} protocol does not support wait_until_ready"
+      end
     end
   end
 end

lib/couchbase/protostellar/cluster.rb

@@ -155,6 +155,10 @@ def ping(_options = Options::Ping::DEFAULT)
         raise Couchbase::Error::FeatureNotAvailable, "The #{Protostellar::NAME} protocol does not support ping"
       end
 
+      def wait_until_ready(_timeout, _options = Options::WaitUntilReady::DEFAULT)
+        raise Couchbase::Error::FeatureNotAvailable, "The #{Protostellar::NAME} protocol does not support wait_until_ready"
+      end
+
       def search_indexes
         raise Couchbase::Error::FeatureNotAvailable, "The #{Protostellar::NAME} protocol does not support search index management"
       end

test/test_helper.rb

@@ -189,6 +189,12 @@ def bucket
       @bucket ||= ENV.fetch("TEST_BUCKET", nil) || "default"
     end
 
+    # CAVES starts a cluster per test. Its KV endpoints must answer before the first operation;
+    # the bootstrap itself is already awaited by Cluster#bucket.
+    def wait_until_kv_ready(cluster)
+      cluster.bucket(bucket).wait_until_ready(10_000, Options::WaitUntilReady.new(service_types: [:kv]))
+    end
+
     def management_endpoint
       @management_endpoint ||= ENV.fetch("TEST_MANAGEMENT_ENDPOINT") do
         if connection_string
@@ -320,7 +326,7 @@ def use_server?
 
     def connect(options = Options::Cluster.new)
       options.authenticate(env.username, env.password)
-      @cluster = Cluster.connect(env.connection_string, options)
+      (@cluster = Cluster.connect(env.connection_string, options)).tap { |c| env.wait_until_kv_ready(c) if use_caves? }
     end
 
     def ensure_primary_index!