RCBC-387: implement replica reads (#48)

Author: avsej

Gemfile

@@ -27,7 +27,7 @@ group :development do
   gem "minitest"
   gem "minitest-reporters"
   gem "rack"
-  gem "rails"
+  gem "rails", "~> 7.0.3"
   gem "reek"
   gem "rubocop"
   gem "rubocop-minitest"

ext/couchbase.cxx

@@ -924,7 +924,7 @@ cb_map_error_code(const couchbase::key_value_error_context& ctx, const std::stri
 {
     VALUE exc = cb_map_error_code(ctx.ec(), message);
     VALUE error_context = rb_hash_new();
-    std::string error(fmt::format("{}, {}", ctx.ec().value(), ctx.ec().message()));
+    std::string error(ctx.ec().message());
     rb_hash_aset(error_context, rb_id2sym(rb_intern("error")), cb_str_new(error));
     rb_hash_aset(error_context, rb_id2sym(rb_intern("id")), cb_str_new(ctx.id()));
     rb_hash_aset(error_context, rb_id2sym(rb_intern("scope")), cb_str_new(ctx.scope()));
@@ -2135,6 +2135,95 @@ cb_Backend_document_get(VALUE self, VALUE bucket, VALUE scope, VALUE collection,
     return Qnil;
 }
 
+struct passthrough_transcoder {
+    using document_type = couchbase::codec::encoded_value;
+
+    static auto decode(const couchbase::codec::encoded_value& data) -> document_type
+    {
+        return data;
+    }
+};
+
+template<>
+struct couchbase::codec::is_transcoder<passthrough_transcoder> : public std::true_type {
+};
+
+static VALUE
+cb_Backend_document_get_any_replica(VALUE self, VALUE bucket, VALUE scope, VALUE collection, VALUE id, VALUE options)
+{
+    const auto& core = cb_backend_to_cluster(self);
+
+    Check_Type(bucket, T_STRING);
+    Check_Type(scope, T_STRING);
+    Check_Type(collection, T_STRING);
+    Check_Type(id, T_STRING);
+
+    try {
+        couchbase::get_any_replica_options opts;
+        cb_options_set_timeout(opts, options);
+
+        auto f = couchbase::cluster(core)
+                   .bucket(cb_string_new(bucket))
+                   .scope(cb_string_new(scope))
+                   .collection(cb_string_new(collection))
+                   .get_any_replica(cb_string_new(id), opts);
+        auto [ctx, resp] = cb_wait_for_future(f);
+        if (ctx.ec()) {
+            cb_throw_error_code(ctx, "unable to get replica of the document");
+        }
+
+        auto value = resp.content_as<passthrough_transcoder>();
+        VALUE res = rb_hash_new();
+        rb_hash_aset(res, rb_id2sym(rb_intern("content")), cb_str_new(value.data));
+        rb_hash_aset(res, rb_id2sym(rb_intern("cas")), cb_cas_to_num(resp.cas()));
+        rb_hash_aset(res, rb_id2sym(rb_intern("flags")), UINT2NUM(value.flags));
+        return res;
+    } catch (const ruby_exception& e) {
+        rb_exc_raise(e.exception_object());
+    }
+    return Qnil;
+}
+
+static VALUE
+cb_Backend_document_get_all_replicas(VALUE self, VALUE bucket, VALUE scope, VALUE collection, VALUE id, VALUE options)
+{
+    const auto& core = cb_backend_to_cluster(self);
+
+    Check_Type(bucket, T_STRING);
+    Check_Type(scope, T_STRING);
+    Check_Type(collection, T_STRING);
+    Check_Type(id, T_STRING);
+
+    try {
+        couchbase::get_all_replicas_options opts;
+        cb_options_set_timeout(opts, options);
+
+        auto f = couchbase::cluster(core)
+                   .bucket(cb_string_new(bucket))
+                   .scope(cb_string_new(scope))
+                   .collection(cb_string_new(collection))
+                   .get_all_replicas(cb_string_new(id), opts);
+        auto [ctx, resp] = cb_wait_for_future(f);
+        if (ctx.ec()) {
+            cb_throw_error_code(ctx, "unable to get all replicas for the document");
+        }
+
+        VALUE res = rb_ary_new_capa(static_cast<long>(resp.size()));
+        for (const auto& entry : resp) {
+            VALUE response = rb_hash_new();
+            auto value = entry.content_as<passthrough_transcoder>();
+            rb_hash_aset(response, rb_id2sym(rb_intern("content")), cb_str_new(value.data));
+            rb_hash_aset(response, rb_id2sym(rb_intern("cas")), cb_cas_to_num(entry.cas()));
+            rb_hash_aset(response, rb_id2sym(rb_intern("flags")), UINT2NUM(value.flags));
+            rb_ary_push(res, response);
+        }
+        return res;
+    } catch (const ruby_exception& e) {
+        rb_exc_raise(e.exception_object());
+    }
+    return Qnil;
+}
+
 static VALUE
 cb_Backend_document_get_multi(VALUE self, VALUE keys, VALUE options)
 {
@@ -7671,6 +7760,8 @@ init_backend(VALUE mCouchbase)
     rb_define_method(cBackend, "ping", VALUE_FUNC(cb_Backend_ping), 2);
 
     rb_define_method(cBackend, "document_get", VALUE_FUNC(cb_Backend_document_get), 5);
+    rb_define_method(cBackend, "document_get_any_replica", VALUE_FUNC(cb_Backend_document_get_any_replica), 5);
+    rb_define_method(cBackend, "document_get_all_replicas", VALUE_FUNC(cb_Backend_document_get_all_replicas), 5);
     rb_define_method(cBackend, "document_get_multi", VALUE_FUNC(cb_Backend_document_get_multi), 2);
     rb_define_method(cBackend, "document_get_projected", VALUE_FUNC(cb_Backend_document_get_projected), 5);
     rb_define_method(cBackend, "document_get_and_lock", VALUE_FUNC(cb_Backend_document_get_and_lock), 6);

lib/couchbase/collection.rb

@@ -168,15 +168,46 @@ def get_and_touch(id, expiry, options = Options::GetAndTouch.new)
     # @param [Options::GetAllReplicas] options request customization
     #
     # @return [Array<GetReplicaResult>]
-    def get_all_replicas(id, options = Options::GetAllReplicas.new) end
+    def get_all_replicas(id, options = Options::GetAllReplicas.new)
+      resp = @backend.document_get_all_replicas(@bucket_name, @scope_name, @name, id, options.to_backend)
+      resp.map do |entry|
+        GetReplicaResult.new do |res|
+          res.transcoder = options.transcoder
+          res.cas = entry[:cas]
+          res.flags = entry[:flags]
+          res.encoded = entry[:content]
+          res.is_replica = entry[:is_replica]
+        end
+      end
+    end
 
-    # Reads all available replicas, and returns the first found
+    # Reads all available replicas and active, and returns the first found.
     #
     # @param [String] id the document id which is used to uniquely identify it.
     # @param [Options::GetAnyReplica] options request customization
     #
+    # @example Get document contents
+    #   res = collection.get_any_replica("customer123")
+    #   res.is_active #=> false
+    #   res.content["addresses"]
+    #
+    #   # {"billing"=>
+    #   #   {"line1"=>"123 Any Street", "line2"=>"Anytown", "country"=>"United Kingdom"},
+    #   #  "delivery"=>
+    #   #   {"line1"=>"123 Any Street", "line2"=>"Anytown", "country"=>"United Kingdom"}}
+    #
+    #
     # @return [GetReplicaResult]
-    def get_any_replica(id, options = Options::GetAnyReplica.new) end
+    def get_any_replica(id, options = Options::GetAnyReplica.new)
+      resp = @backend.document_get_any_replica(@bucket_name, @scope_name, @name, id, options.to_backend)
+      GetReplicaResult.new do |res|
+        res.transcoder = options.transcoder
+        res.cas = resp[:cas]
+        res.flags = resp[:flags]
+        res.encoded = resp[:content]
+        res.is_replica = resp[:is_replica]
+      end
+    end
 
     # Checks if the given document ID exists on the active partition.
     #

test/active_support/behaviors/local_cache_behavior.rb

@@ -291,4 +291,13 @@ def test_local_cache_should_read_and_write_false
       assert_equal false, @cache.read(key)
     end
   end
+
+  def test_local_cache_should_deserialize_entries_on_multi_get
+    keys = Array.new(5) { SecureRandom.uuid }
+    values = keys.index_with(true)
+    @cache.with_local_cache do
+      assert @cache.write_multi(values)
+      assert_equal values, @cache.read_multi(*keys)
+    end
+  end
 end

test/crud_test.rb

@@ -48,6 +48,23 @@ def test_removes_documents
       end
     end
 
+    def test_reads_from_replica
+      doc_id = uniq_id(:foo)
+      document = {"value" => 42}
+      @collection.upsert(doc_id, document)
+
+      res = @collection.get_any_replica(doc_id)
+      assert_equal document, res.content
+      assert_respond_to res, :replica?
+
+      res = @collection.get_all_replicas(doc_id)
+      refute_empty res
+      res.each do |entry|
+        assert_equal document, entry.content
+        assert_respond_to entry, :replica?
+      end
+    end
+
     def test_touch_sets_expiration
       document = {"value" => 42}
       doc_id = uniq_id(:foo)