Fix multi thread error handling and resharding test cases (#41)

Author: supercaracal

lib/redis_client/cluster.rb

@@ -40,9 +40,10 @@ def empty?
         @size.zero?
       end
 
-      # TODO: https://github.com/redis-rb/redis-cluster-client/issues/37
-      def execute # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/MethodLength
+      # TODO: https://github.com/redis-rb/redis-cluster-client/issues/37 handle redirections
+      def execute # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/MethodLength, Metrics/PerceivedComplexity
         all_replies = Array.new(@size)
+        errors = {}
         threads = @grouped.map do |k, v|
           Thread.new(@client, k, v) do |client, node_key, rows|
             Thread.pass
@@ -60,11 +61,15 @@ def execute # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Met
             raise ReplySizeError, "commands: #{rows.size}, replies: #{replies.size}" if rows.size != replies.size
 
             rows.each_with_index { |row, idx| all_replies[row.first] = replies[idx] }
+          rescue StandardError => e
+            errors[node_key] = e
           end
         end
 
         threads.each(&:join)
-        all_replies
+        return all_replies if errors.empty?
+
+        raise ::RedisClient::Cluster::ErrorCollection, errors
       end
     end
 
@@ -185,10 +190,10 @@ def send_command(method, *command, **kwargs, &block) # rubocop:disable Metrics/A
       when 'acl', 'auth', 'bgrewriteaof', 'bgsave', 'quit', 'save', 'ping'
         @node.call_all(method, *command, **kwargs, &block).first
       when 'flushall', 'flushdb'
-        @node.call_primary(method, *command, **kwargs, &block).first
+        @node.call_primaries(method, *command, **kwargs, &block).first
       when 'wait'     then send_wait_command(method, *command, **kwargs, &block)
-      when 'keys'     then @node.call_replica(method, *command, **kwargs, &block).flatten.sort
-      when 'dbsize'   then @node.call_replica(method, *command, **kwargs, &block).sum
+      when 'keys'     then @node.call_replicas(method, *command, **kwargs, &block).flatten.sort
+      when 'dbsize'   then @node.call_replicas(method, *command, **kwargs, &block).sum
       when 'scan'     then _scan(*command, **kwargs)
       when 'lastsave' then @node.call_all(method, *command, **kwargs, &block).sort
       when 'role'     then @node.call_all(method, *command, **kwargs, &block)
@@ -206,14 +211,14 @@ 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::CommandErrorCollection => e
+    rescue RedisClient::Cluster::ErrorCollection => e
       update_cluster_info! if e.errors.values.map(&:class).any?(::RedisClient::ConnectionError)
       raise
     end
 
     def send_wait_command(method, *command, retry_count: 3, **kwargs, &block)
-      @node.call_primary(method, *command, **kwargs, &block).sum
-    rescue RedisClient::Cluster::CommandErrorCollection => e
+      @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?
 
       update_cluster_info!
@@ -246,13 +251,17 @@ def send_client_command(method, *command, **kwargs, &block)
       end
     end
 
-    def send_cluster_command(method, *command, **kwargs, &block)
+    def send_cluster_command(method, *command, **kwargs, &block) # rubocop:disable Metrics/MethodLength
       subcommand = command[1].to_s.downcase
       case subcommand
       when 'addslots', 'delslots', 'failover', 'forget', 'meet', 'replicate',
            'reset', 'set-config-epoch', 'setslot'
         raise ::RedisClient::Cluster::OrchestrationCommandNotSupported, ['cluster', subcommand]
       when 'saveconfig' then @node.call_all(method, *command, **kwargs, &block).first
+      when 'getkeysinslot'
+        raise ArgumentError, command.join(' ') if command.size != 4
+
+        find_node(@node.find_node_key_of_replica(command[2])).send(method, *command, **kwargs, &block)
       else assign_node(*command).send(method, *command, **kwargs, &block)
       end
     end
@@ -262,7 +271,7 @@ def send_script_command(method, *command, **kwargs, &block)
       when 'debug', 'kill'
         @node.call_all(method, *command, **kwargs, &block).first
       when 'flush', 'load'
-        @node.call_primary(method, *command, **kwargs, &block).first
+        @node.call_primaries(method, *command, **kwargs, &block).first
       else assign_node(*command).send(method, *command, **kwargs, &block)
       end
     end

lib/redis_client/cluster/errors.rb

@@ -29,7 +29,7 @@ def initialize(command)
     end
 
     # Raised when error occurs on any node of cluster.
-    class CommandErrorCollection < ::RedisClient::Error
+    class ErrorCollection < ::RedisClient::Error
       attr_reader :errors
 
       def initialize(errors)

lib/redis_client/cluster/node.rb

@@ -118,16 +118,16 @@ def call_all(method, *command, **kwargs, &block)
         try_map { |_, client| client.send(method, *command, **kwargs, &block) }.values
       end
 
-      def call_primary(method, *command, **kwargs, &block)
+      def call_primaries(method, *command, **kwargs, &block)
         try_map do |node_key, client|
           next if replica?(node_key)
 
           client.send(method, *command, **kwargs, &block)
         end.values
       end
 
-      def call_replica(method, *command, **kwargs, &block)
-        return call_primary(method, *command, **kwargs, &block) if replica_disabled?
+      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|
@@ -228,22 +228,22 @@ def build_clients(options, pool: nil, **kwargs)
       end
 
       def try_map # rubocop:disable Metrics/MethodLength
-        errors = {}
         results = {}
+        errors = {}
         threads = @clients.map do |k, v|
           Thread.new(k, v) do |node_key, client|
             Thread.pass
             reply = yield(node_key, client)
             results[node_key] = reply unless reply.nil?
-          rescue ::RedisClient::CommandError => e
+          rescue StandardError => e
             errors[node_key] = e
           end
         end
 
         threads.each(&:join)
         return results if errors.empty?
 
-        raise ::RedisClient::Cluster::CommandErrorCollection, errors
+        raise ::RedisClient::Cluster::ErrorCollection, errors
       end
     end
   end

test/redis_client/cluster/test_errors.rb

@@ -40,7 +40,7 @@ def test_orchestration_command_not_supported_error
         end
       end
 
-      def test_command_error_collection_error
+      def test_error_collection_error
         [
           {
             errors: { '127.0.0.1:6379' => DummyError.new('foo') },
@@ -54,7 +54,7 @@ def test_command_error_collection_error
           { errors: '', want: { msg: '', size: 0 } },
           { errors: nil, want: { msg: '', size: 0 } }
         ].each_with_index do |c, idx|
-          raise ::RedisClient::Cluster::CommandErrorCollection, c[:errors]
+          raise ::RedisClient::Cluster::ErrorCollection, c[:errors]
         rescue StandardError => e
           assert_equal(c[:want][:msg], e.message, "Case: #{idx}")
           assert_equal(c[:want][:size], e.errors.size, "Case: #{idx}")

test/redis_client/cluster/test_node.rb

@@ -220,19 +220,19 @@ def test_call_all
         assert_equal(want, got, 'Case: scale read')
       end
 
-      def test_call_primary
+      def test_call_primaries
         want = (1..(@test_node_info.count { |info| info[:role] == 'master' })).map { |_| 'PONG' }
-        got = @test_node.call_primary(:call, 'PING')
+        got = @test_node.call_primaries(:call, 'PING')
         assert_equal(want, got)
       end
 
-      def test_call_replica
+      def test_call_replicas
         want = (1..(@test_node_info.count { |info| info[:role] == 'master' })).map { |_| 'PONG' }
 
-        got = @test_node.call_replica(:call, 'PING')
+        got = @test_node.call_replicas(:call, 'PING')
         assert_equal(want, got, 'Case: primary only')
 
-        got = @test_node_with_scale_read.call_replica(:call, 'PING')
+        got = @test_node_with_scale_read.call_replicas(:call, 'PING')
         assert_equal(want, got, 'Case: scale read')
       end
 
@@ -408,7 +408,7 @@ 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::CommandErrorCollection }
+          { block: ->(_, client) { client.call('UNKNOWN') }, error: ::RedisClient::Cluster::ErrorCollection }
         ].each_with_index do |c, idx|
           msg = "Case: #{idx}"
           got = -> { @test_node.send(:try_map, &c[:block]) }

test/test_against_cluster_state.rb

@@ -37,30 +37,29 @@ def test_the_state_of_cluster_failover
     end
 
     def test_the_state_of_cluster_resharding
-      @client.pipelined { |pipeline| 10_000.times { |i| pipeline.call('SET', "{key}#{i}", i) } }
-      wait_for_replication
-
-      slot = SLOT_SIZE.times.max_by { |i| @client.call('CLUSTER', 'COUNTKEYSINSLOT', i) }
-      src = @client.instance_variable_get(:@node).find_node_key_of_primary(slot)
-      dest = @client.instance_variable_get(:@node).primary_node_keys.reject { |k| k == src }.sample
-
-      @controller.start_resharding(slot: slot, src_node_key: src, dest_node_key: dest)
-      10_000.times { |i| assert_equal(i.to_s, @client.call('GET', "{key}#{i}"), "Case: GET: #{i}") }
-      @controller.finish_resharding(slot: slot, dest_node_key: dest)
+      do_resharding_test do |keys|
+        keys.each do |key|
+          want = key
+          got = @client.call('GET', key)
+          assert_equal(want, got, "Case: GET: #{key}")
+        end
+      end
     end
 
     def test_the_state_of_cluster_resharding_with_pipelining
       skip('TODO: https://github.com/redis-rb/redis-cluster-client/issues/37')
-      @client.pipelined { |pipeline| 100_000.times { |i| pipeline.call('SET', "{key}#{i}", i) } }
-
-      slot = SLOT_SIZE.times.max_by { |i| @client.call('CLUSTER', 'COUNTKEYSINSLOT', i) }
-      src = @client.instance_variable_get(:@node).find_node_key_of_primary(slot)
-      dest = @client.instance_variable_get(:@node).primary_node_keys.reject { |k| k == src }.sample
 
-      @controller.start_resharding(slot: slot, src_node_key: src, dest_node_key: dest)
-      values = @client.pipelined { |pipeline| 100_000.times { |i| pipeline.call('GET', "{key}#{i}") } }
-      100_000.times { |i| assert_equal(i.to_s, values[i], "Case: GET: #{i}") }
-      @controller.finish_resharding(slot: slot, dest_node_key: dest)
+      do_resharding_test do |keys|
+        values = @client.pipelined do |pipeline|
+          keys.each { |key| pipeline.call('GET', key) }
+        end
+
+        keys.each_with_index do |key, i|
+          want = key
+          got = values[i]
+          assert_equal(want, got, "Case: GET: #{key}")
+        end
+      end
     end
 
     private
@@ -72,6 +71,22 @@ def wait_for_replication
     def fetch_cluster_info(key)
       @client.call('CLUSTER', 'INFO').split("\r\n").to_h { |v| v.split(':') }.fetch(key)
     end
+
+    def do_resharding_test(number_of_keys: 1000)
+      @client.pipelined { |pipeline| number_of_keys.times { |i| pipeline.call('SET', "key#{i}", "key#{i}") } }
+      wait_for_replication
+      count, slot = @client.pipelined { |pi| SLOT_SIZE.times { |i| pi.call('CLUSTER', 'COUNTKEYSINSLOT', i) } }
+                           .each_with_index.max_by { |c, _| c }
+      refute_equal(0, count)
+      keys = @client.call('CLUSTER', 'GETKEYSINSLOT', slot, count)
+      refute_empty(keys)
+      src = @client.instance_variable_get(:@node).find_node_key_of_primary(slot)
+      dest = @client.instance_variable_get(:@node).primary_node_keys.reject { |k| k == src }.sample
+      @controller.start_resharding(slot: slot, src_node_key: src, dest_node_key: dest)
+      wait_for_replication
+      yield(keys)
+      @controller.finish_resharding(slot: slot, dest_node_key: dest)
+    end
   end
 
   class PrimaryOnly < TestingWrapper