Fix node down test cases (#46)

Author: supercaracal

README.md

@@ -39,7 +39,7 @@ RedisClient.cluster(timeout: 3.0).new_client
 ```
 
 ```ruby
-# To connect with subset nodes for startup
+# To connect with a subset of nodes for startup
 RedisClient.cluster(nodes: %w[redis://node1:6379 redis://node2:6379]).new_client
 ```
 
@@ -70,7 +70,7 @@ The other methods are not implemented because the client cannot operate with clu
 `#pipelined` method splits and sends commands to each node and aggregates replies.
 
 ## Multiple keys and CROSSSLOT error
-A part of commands can be passed multiple keys. But it has a constraint the keys are in the same hash slot.
+A subset of commands can be passed multiple keys. But it has a constraint the keys are in the same hash slot.
 The following error occurs because keys must be in the same hash slot and not just the same node.
 
 ```ruby

lib/redis_client/cluster.rb

@@ -170,8 +170,10 @@ def pubsub
     end
 
     def close
-      @node.each(&:close)
+      @node.call_all(:close)
       nil
+    rescue StandardError
+      # ignore
     end
 
     private
@@ -187,10 +189,11 @@ def fetch_cluster_info!(config, pool: nil, **kwargs)
     def send_command(method, *command, **kwargs, &block) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/MethodLength
       cmd = command.first.to_s.downcase
       case cmd
-      when 'acl', 'auth', 'bgrewriteaof', 'bgsave', 'quit', 'save', 'ping'
+      when 'acl', 'auth', 'bgrewriteaof', 'bgsave', 'quit', 'save'
         @node.call_all(method, *command, **kwargs, &block).first
       when 'flushall', 'flushdb'
         @node.call_primaries(method, *command, **kwargs, &block).first
+      when 'ping'     then @node.send_ping(method, *command, **kwargs, &block).first
       when 'wait'     then send_wait_command(method, *command, **kwargs, &block)
       when 'keys'     then @node.call_replicas(method, *command, **kwargs, &block).flatten.sort
       when 'dbsize'   then @node.call_replicas(method, *command, **kwargs, &block).sum
@@ -211,15 +214,18 @@ def send_command(method, *command, **kwargs, &block) # rubocop:disable Metrics/A
         node = assign_node(*command)
         try_send(node, method, *command, **kwargs, &block)
       end
-    rescue RedisClient::Cluster::ErrorCollection => e
-      update_cluster_info! if e.errors.values.map(&:class).any?(::RedisClient::ConnectionError)
-      raise
+    rescue RedisClient::Cluster::Node::ReloadNeeded
+      update_cluster_info!
+      raise ::RedisClient::Cluster::NodeMightBeDown
     end
 
     def send_wait_command(method, *command, retry_count: 3, **kwargs, &block)
       @node.call_primaries(method, *command, **kwargs, &block).sum
     rescue RedisClient::Cluster::ErrorCollection => e
-      raise if retry_count <= 0 || e.errors.values.map(&:message).grep(/ERR WAIT cannot be used with replica instances/).empty?
+      raise if retry_count <= 0
+      raise if e.errors.values.none? do |err|
+        err.message.include?('WAIT cannot be used with replica instances')
+      end
 
       update_cluster_info!
       retry_count -= 1
@@ -353,44 +359,30 @@ def assign_node(*command)
       find_node(node_key)
     end
 
-    def find_node_key(*command, primary_only: false) # rubocop:disable Metrics/MethodLength
+    def find_node_key(*command, primary_only: false)
       key = @command.extract_first_key(command)
-      if key.empty?
-        return @node.primary_node_keys.sample if @command.should_send_to_primary?(command) || primary_only
-
-        return @node.replica_node_keys.sample
-      end
-
-      slot = ::RedisClient::Cluster::KeySlotConverter.convert(key)
-      return unless @node.slot_exists?(slot)
+      slot = key.empty? ? nil : ::RedisClient::Cluster::KeySlotConverter.convert(key)
 
       if @command.should_send_to_primary?(command) || primary_only
-        @node.find_node_key_of_primary(slot)
+        @node.find_node_key_of_primary(slot) || @node.primary_node_keys.sample
       else
-        @node.find_node_key_of_replica(slot)
+        @node.find_node_key_of_replica(slot) || @node.replica_node_keys.sample
       end
     end
 
     def find_node(node_key, retry_count: 3)
-      return @node.sample if node_key.nil?
-
       @node.find_by(node_key)
     rescue ::RedisClient::Cluster::Node::ReloadNeeded
-      raise(::RedisClient::ConnectionError, 'unstable cluster state') if retry_count <= 0
+      raise ::RedieClient::Cluster::NodeMightBeDown if retry_count <= 0
 
-      update_cluster_info!(node_key)
+      update_cluster_info!
       retry_count -= 1
       retry
     end
 
-    def update_cluster_info!(node_key = nil)
+    def update_cluster_info!
       @mutex.synchronize do
-        unless node_key.nil?
-          host, port = ::RedisClient::Cluster::NodeKey.split(node_key)
-          @config.add_node(host, port)
-        end
-
-        @node.each(&:close)
+        close
         @node = fetch_cluster_info!(@config, pool: @pool, **@client_kwargs)
       end
     end

lib/redis_client/cluster/command.rb

@@ -68,12 +68,14 @@ def dig_details(command, key)
         @details.fetch(name).fetch(key)
       end
 
-      def determine_first_key_position(command) # rubocop:disable Metrics/CyclomaticComplexity
+      def determine_first_key_position(command) # rubocop:disable Metrics/CyclomaticComplexity, Metrics/MethodLength
         case command&.flatten&.first.to_s.downcase
-        when 'eval', 'evalsha', 'migrate', 'zinterstore', 'zunionstore' then 3
+        when 'eval', 'evalsha', 'zinterstore', 'zunionstore' then 3
         when 'object' then 2
         when 'memory'
           command[1].to_s.casecmp('usage').zero? ? 2 : 0
+        when 'migrate'
+          command[3] == '""' ? determine_optional_key_position(command, 'keys') : 3
         when 'xread', 'xreadgroup'
           determine_optional_key_position(command, 'streams')
         else

lib/redis_client/cluster/errors.rb

@@ -51,5 +51,15 @@ def initialize(command)
         super("Cluster client doesn't know which node the #{command} command should be sent to.")
       end
     end
+
+    class NodeMightBeDown < ::RedisClient::Error
+      def initialize(_ = '')
+        super(
+          'The client is trying to fetch the latest cluster state '\
+          'because a subset of nodes might be down. '\
+          'It might continue to raise errors for a while.'
+        )
+      end
+    end
   end
 end

lib/redis_client/cluster/key_slot_converter.rb

@@ -43,6 +43,8 @@ module KeySlotConverter
       module_function
 
       def convert(key)
+        return nil if key.nil?
+
         crc = 0
         key.each_byte do |b|
           crc = ((crc << 8) & 0xffff) ^ XMODEM_CRC16_LOOKUP[((crc >> 8) ^ b) & 0xff]

lib/redis_client/cluster/node.rb

@@ -12,6 +12,7 @@ class Node
       SLOT_SIZE = 16_384
       MIN_SLOT = 0
       MAX_SLOT = SLOT_SIZE - 1
+      MAX_STARTUP_SAMPLE = 37
       IGNORE_GENERIC_CONFIG_KEYS = %i[url host port path].freeze
 
       ReloadNeeded = Class.new(::RedisClient::Error)
@@ -32,19 +33,33 @@ def build_connection_prelude
       end
 
       class << self
-        def load_info(options, **kwargs)
-          startup_nodes = ::RedisClient::Cluster::Node.new(options, **kwargs)
-
-          errors = startup_nodes.map do |n|
-            reply = n.call('CLUSTER', 'NODES')
-            return parse_node_info(reply)
-          rescue ::RedisClient::ConnectionError, ::RedisClient::CommandError => e
-            e
+        def load_info(options, **kwargs) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/MethodLength, Metrics/PerceivedComplexity
+          startup_size = options.size > MAX_STARTUP_SAMPLE ? MAX_STARTUP_SAMPLE : options.size
+          node_info_list = Array.new(startup_size)
+          errors = Array.new(startup_size)
+          startup_options = options.to_a.sample(MAX_STARTUP_SAMPLE).to_h
+          startup_nodes = ::RedisClient::Cluster::Node.new(startup_options, **kwargs)
+          threads = startup_nodes.each_with_index.map do |raw_client, idx|
+            Thread.new(raw_client, idx) do |cli, i|
+              Thread.pass
+              reply = cli.call('CLUSTER', 'NODES')
+              node_info_list[i] = parse_node_info(reply)
+            rescue StandardError => e
+              errors[i] = e
+            ensure
+              cli&.close
+            end
+          end
+          threads.each(&:join)
+          raise ::RedisClient::Cluster::InitialSetupError, errors if node_info_list.all?(&:nil?)
+
+          grouped = node_info_list.compact.group_by do |rows|
+            rows.sort_by { |row| row[:id] }
+                .map { |r| "#{r[:id]}#{r[:node_key]}#{r[:role]}#{r[:primary_id]}#{r[:config_epoch]}" }
+                .join
           end
 
-          raise ::RedisClient::Cluster::InitialSetupError, errors
-        ensure
-          startup_nodes&.each(&:close)
+          grouped.max_by { |_, v| v.size }[1].first
         end
 
         private
@@ -109,32 +124,58 @@ def replica_node_keys
       end
 
       def find_by(node_key)
+        raise ReloadNeeded if node_key.nil? || !@clients.key?(node_key)
+
         @clients.fetch(node_key)
-      rescue KeyError
-        raise ReloadNeeded
       end
 
       def call_all(method, *command, **kwargs, &block)
-        try_map { |_, client| client.send(method, *command, **kwargs, &block) }.values
+        results, errors = try_map do |_, client|
+          client.send(method, *command, **kwargs, &block)
+        end
+
+        return results.values if errors.empty?
+
+        raise ::RedisClient::Cluster::ErrorCollection, errors
       end
 
       def call_primaries(method, *command, **kwargs, &block)
-        try_map do |node_key, client|
+        results, errors = try_map do |node_key, client|
           next if replica?(node_key)
 
           client.send(method, *command, **kwargs, &block)
-        end.values
+        end
+
+        return results.values if errors.empty?
+
+        raise ::RedisClient::Cluster::ErrorCollection, errors
       end
 
       def call_replicas(method, *command, **kwargs, &block)
         return call_primaries(method, *command, **kwargs, &block) if replica_disabled?
 
         replica_node_keys = @replications.values.map(&:sample)
-        try_map do |node_key, client|
+        results, errors = try_map do |node_key, client|
           next if primary?(node_key) || !replica_node_keys.include?(node_key)
 
           client.send(method, *command, **kwargs, &block)
-        end.values
+        end
+
+        return results.values if errors.empty?
+
+        raise ::RedisClient::Cluster::ErrorCollection, errors
+      end
+
+      def send_ping(method, *command, **kwargs, &block)
+        results, errors = try_map do |_, client|
+          client.send(method, *command, **kwargs, &block)
+        end
+
+        return results.values if errors.empty?
+
+        raise ReloadNeeded if errors.values.any?(::RedisClient::ConnectionError)
+
+        raise ::RedisClient::Cluster::ErrorCollection, errors
       end
 
       def scale_reading_clients
@@ -144,21 +185,18 @@ def scale_reading_clients
         end
       end
 
-      def slot_exists?(slot)
-        slot = Integer(slot)
-        return false if slot < MIN_SLOT || slot > MAX_SLOT
-
-        !@slots[slot].nil?
-      end
-
       def find_node_key_of_primary(slot)
+        return if slot.nil?
+
         slot = Integer(slot)
         return if slot < MIN_SLOT || slot > MAX_SLOT
 
         @slots[slot]
       end
 
       def find_node_key_of_replica(slot)
+        return if slot.nil?
+
         slot = Integer(slot)
         return if slot < MIN_SLOT || slot > MAX_SLOT
 
@@ -241,9 +279,7 @@ def try_map # rubocop:disable Metrics/MethodLength
         end
 
         threads.each(&:join)
-        return results if errors.empty?
-
-        raise ::RedisClient::Cluster::ErrorCollection, errors
+        [results, errors]
       end
     end
   end

test/redis_client/cluster/test_command.rb

@@ -176,6 +176,7 @@ def test_determine_first_key_position
           { command: [['EVAL'], '"return ARGV[1]"', 0, 'hello'], want: 3 },
           { command: %w[EVALSHA sha1 2 foo bar baz zap], want: 3 },
           { command: %w[MIGRATE host port key 0 5 COPY], want: 3 },
+          { command: %w[MIGRATE host port "" 0 5 COPY KEYS key], want: 8 },
           { command: %w[ZINTERSTORE out 2 zset1 zset2 WEIGHTS 2 3], want: 3 },
           { command: %w[ZUNIONSTORE out 2 zset1 zset2 WEIGHTS 2 3], want: 3 },
           { command: %w[OBJECT HELP], want: 2 },

test/redis_client/cluster/test_key_slot_converter.rb

@@ -19,6 +19,13 @@ def test_convert
           got = ::RedisClient::Cluster::KeySlotConverter.convert(key)
           assert_equal(want, got, "Case: #{idx}")
         end
+
+        assert_nil(::RedisClient::Cluster::KeySlotConverter.convert(nil), 'Case: nil')
+
+        multi_byte_key = 'あいうえお'
+        want = @raw_clients.first.call('CLUSTER', 'KEYSLOT', multi_byte_key)
+        got = ::RedisClient::Cluster::KeySlotConverter.convert(multi_byte_key)
+        assert_equal(want, got, "Case: #{multi_byte_key}")
       end
     end
   end

test/redis_client/cluster/test_node.rb

@@ -246,19 +246,12 @@ def test_scale_reading_clients # rubocop:disable Metrics/CyclomaticComplexity
         got.each { |e| assert_includes(want, e, 'Case: scale read') }
       end
 
-      def test_slot_exists?
-        refute(@test_node.slot_exists?(-1))
-        assert(@test_node.slot_exists?(0))
-        assert(@test_node.slot_exists?(16_383))
-        refute(@test_node.slot_exists?(16_384))
-        assert_raises(TypeError) { @test_node.slot_exists?(:foo) }
-      end
-
       def test_find_node_key_of_primary
         sample_node = @test_node_info.find { |info| info[:role] == 'master' }
         sample_slot = sample_node[:slots].first.first
         got = @test_node.find_node_key_of_primary(sample_slot)
-        assert_equal(sample_node[:node_key], got)
+        assert_equal(sample_node[:node_key], got, 'Case: sample slot')
+        assert_nil(@test_node.find_node_key_of_primary(nil), 'Case: nil')
       end
 
       def test_find_node_key_of_replica
@@ -273,6 +266,8 @@ def test_find_node_key_of_replica
         got = @test_node_with_scale_read.find_node_key_of_replica(sample_slot)
         want = sample_replicas.map { |info| info[:node_key] }
         assert_includes(want, got, 'Case: scale read')
+
+        assert_nil(@test_node.find_node_key_of_replica(nil), 'Case: nil')
       end
 
       def test_update_slot
@@ -407,15 +402,15 @@ def test_build_clients # rubocop:disable Metrics/CyclomaticComplexity, Metrics/P
       def test_try_map
         primary_node_keys = @test_node_info.select { |info| info[:role] == 'master' }.map { |info| info[:node_key] }
         [
-          { block: ->(_, client) { client.call('PING') }, want: primary_node_keys.to_h { |k| [k, 'PONG'] } },
-          { block: ->(_, client) { client.call('UNKNOWN') }, error: ::RedisClient::Cluster::ErrorCollection }
+          { block: ->(_, client) { client.call('PING') }, results: primary_node_keys.to_h { |k| [k, 'PONG'] } },
+          { block: ->(_, client) { client.call('UNKNOWN') }, errors: ::RedisClient::CommandError }
         ].each_with_index do |c, idx|
           msg = "Case: #{idx}"
-          got = -> { @test_node.send(:try_map, &c[:block]) }
-          if c.key?(:error)
-            assert_raises(c[:error], msg, &got)
+          results, errors = @test_node.send(:try_map, &c[:block])
+          if c.key?(:errors)
+            errors.each_value { |e| assert_instance_of(c[:errors], e, msg) }
           else
-            assert_equal(c[:want], got.call, msg)
+            assert_equal(c[:results], results, msg)
           end
         end
       end