Add node down test (#39)

Author: supercaracal

.github/workflows/test.yaml

@@ -124,3 +124,35 @@ jobs:
         run: bundle exec rake test_cluster_state
       - name: Stop containers
         run: docker compose -f $DOCKER_COMPOSE_FILE down
+  cluster-broken:
+    name: Cluster Broken
+    timeout-minutes: 10
+    strategy:
+      fail-fast: false
+      matrix:
+        redis: ['7.0.1', '6.2.7']
+    runs-on: ubuntu-latest
+    env:
+      REDIS_VERSION: ${{ matrix.redis }}
+      DOCKER_COMPOSE_FILE: 'docker-compose.yaml'
+      RESTART_POLICY: 'no'
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v3
+      - name: Set up Ruby
+        uses: ruby/setup-ruby@v1
+        with:
+          ruby-version: '3.1'
+          bundler-cache: true
+      - name: Pull Docker images
+        run: docker pull redis:$REDIS_VERSION
+      - name: Run containers
+        run: docker compose -f $DOCKER_COMPOSE_FILE up -d
+      - name: Wait for Redis cluster to be ready
+        run: bundle exec rake wait
+      - name: Print containers
+        run: docker compose -f $DOCKER_COMPOSE_FILE ps
+      - name: Run minitest
+        run: bundle exec rake test_cluster_broken
+      - name: Stop containers
+        run: docker compose -f $DOCKER_COMPOSE_FILE down

Rakefile

@@ -7,7 +7,7 @@ task default: :test
 Rake::TestTask.new :test do |t|
   t.libs << :test
   t.libs << :lib
-  files = Dir['test/**/test_*.rb'].grep_v(/test_against_cluster_state/)
+  files = Dir['test/**/test_*.rb'].grep_v(/test_against_cluster_(state|broken)/)
   files = ARGV[1..] if ARGV.size > 1
   t.test_files = files
   t.options = '-v'
@@ -20,6 +20,13 @@ Rake::TestTask.new :test_cluster_state do |t|
   t.options = '-v'
 end
 
+Rake::TestTask.new :test_cluster_broken do |t|
+  t.libs << :test
+  t.libs << :lib
+  t.test_files = %w[test/test_against_cluster_broken.rb]
+  t.options = '-v'
+end
+
 desc 'Wait for cluster to be ready'
 task :wait do
   $LOAD_PATH.unshift(File.expand_path('test', __dir__))

docker-compose.replica.yaml

@@ -11,6 +11,7 @@ services:
       --cluster-enabled      yes
       --cluster-config-file  nodes.conf
       --cluster-node-timeout 5000
+    restart: "${RESTART_POLICY:-always}"
     healthcheck:
       test: ["CMD", "redis-cli", "ping"]
       interval: "7s"

docker-compose.ssl.yaml

@@ -18,6 +18,7 @@ services:
       --tls-ca-cert-file     /etc/ssl/private/redis-rb-ca.crt
       --tls-cluster          yes
       --tls-replication      yes
+    restart: "${RESTART_POLICY:-always}"
     healthcheck:
       test:
         - "CMD-SHELL"

docker-compose.yaml

@@ -10,6 +10,7 @@ services:
       --cluster-enabled      yes
       --cluster-config-file  nodes.conf
       --cluster-node-timeout 5000
+    restart: "${RESTART_POLICY:-always}"
     healthcheck:
       test: ["CMD", "redis-cli", "ping"]
       interval: "7s"

lib/redis_client/cluster.rb

@@ -182,7 +182,7 @@ 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'
+      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
@@ -206,12 +206,15 @@ 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
+      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
-      raise if e.errors.values.map(&:message).grep(/ERR WAIT cannot be used with replica instances/).size.zero?
+      raise if retry_count <= 0 || e.errors.values.map(&:message).grep(/ERR WAIT cannot be used with replica instances/).empty?
 
       update_cluster_info!
       retry_count -= 1
@@ -277,18 +280,16 @@ def send_pubsub_command(method, *command, **kwargs, &block) # rubocop:disable Me
 
     # @see https://redis.io/topics/cluster-spec#redirection-and-resharding
     #   Redirection and resharding
-    def try_send(node, method, *args, retry_count: 3, **kwargs, &block) # rubocop:disable Metrics/AbcSize, Metrics/MethodLength
+    def try_send(node, method, *args, retry_count: 3, **kwargs, &block) # rubocop:disable Metrics/MethodLength, Metrics/AbcSize
       node.send(method, *args, **kwargs, &block)
     rescue ::RedisClient::CommandError => e
-      if e.message.start_with?(REPLY_MOVED)
-        raise if retry_count <= 0
+      raise if retry_count <= 0
 
+      if e.message.start_with?(REPLY_MOVED)
         node = assign_redirection_node(e.message)
         retry_count -= 1
         retry
       elsif e.message.start_with?(REPLY_ASK)
-        raise if retry_count <= 0
-
         node = assign_asking_node(e.message)
         node.call(CMD_ASKING)
         retry_count -= 1
@@ -297,8 +298,11 @@ def try_send(node, method, *args, retry_count: 3, **kwargs, &block) # rubocop:di
         raise
       end
     rescue ::RedisClient::ConnectionError
+      raise if retry_count <= 0
+
       update_cluster_info!
-      raise
+      retry_count -= 1
+      retry
     end
 
     def _scan(*command, **kwargs) # rubocop:disable Metrics/MethodLength, Metrics/AbcSize
@@ -358,13 +362,16 @@ def find_node_key(*command, primary_only: false) # rubocop:disable Metrics/Metho
       end
     end
 
-    def find_node(node_key)
+    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
+
       update_cluster_info!(node_key)
-      @node.find_by(node_key)
+      retry_count -= 1
+      retry
     end
 
     def update_cluster_info!(node_key = nil)

lib/redis_client/cluster/node.rb

@@ -171,6 +171,12 @@ def update_slot(slot, node_key)
         @mutex.synchronize { @slots[slot] = node_key }
       end
 
+      def replicated?(primary_node_key, replica_node_key)
+        return false if @replications.nil? || @replications.size.zero?
+
+        @replications.fetch(primary_node_key).include?(replica_node_key)
+      end
+
       private
 
       def replica_disabled?
@@ -182,7 +188,9 @@ def primary?(node_key)
       end
 
       def replica?(node_key)
-        !(@replications.nil? || @replications.size.zero?) && !@replications.key?(node_key)
+        return false if @replications.nil? || @replications.size.zero?
+
+        !@replications.key?(node_key)
       end
 
       def build_slot_node_mappings(node_info)
@@ -196,11 +204,12 @@ def build_slot_node_mappings(node_info)
         slots
       end
 
-      def build_replication_mappings(node_info)
+      def build_replication_mappings(node_info) # rubocop:disable Metrics/AbcSize
         dict = node_info.to_h { |info| [info[:id], info] }
         node_info.each_with_object(Hash.new { |h, k| h[k] = [] }) do |info, acc|
           primary_info = dict[info[:primary_id]]
           acc[primary_info[:node_key]] << info[:node_key] unless primary_info.nil?
+          acc[info[:node_key]] if info[:role] == 'master' # for the primary which have no replicas
         end
       end
 

test/cluster_controller.rb

@@ -140,6 +140,8 @@ def wait_meeting(clients, max_attempts:)
     wait_for_state(clients, max_attempts: max_attempts) do |client|
       info = hashify_cluster_info(client)
       info['cluster_known_nodes'].to_s == clients.size.to_s
+    rescue ::RedisClient::ConnectionError
+      true
     end
   end
 
@@ -175,20 +177,26 @@ def wait_cluster_building(clients, max_attempts:)
     wait_for_state(clients, max_attempts: max_attempts) do |client|
       info = hashify_cluster_info(client)
       info['cluster_state'] == 'ok'
+    rescue ::RedisClient::ConnectionError
+      true
     end
   end
 
   def wait_replication(clients, number_of_replicas:, max_attempts:)
     wait_for_state(clients, max_attempts: max_attempts) do |client|
       flags = hashify_cluster_node_flags(clients, client: client)
       flags.values.count { |f| f == 'slave' } == number_of_replicas
+    rescue ::RedisClient::ConnectionError
+      true
     end
   end
 
   def wait_failover(clients, master_key:, slave_key:, max_attempts:)
     wait_for_state(clients, max_attempts: max_attempts) do |client|
       flags = hashify_cluster_node_flags(clients, client: client)
       flags[master_key] == 'slave' && flags[slave_key] == 'master'
+    rescue ::RedisClient::ConnectionError
+      true
     end
   end
 
@@ -197,6 +205,8 @@ def wait_replication_delay(clients, timeout:)
     wait_for_state(clients, max_attempts: clients.size + 1) do |client|
       client.blocking_call(timeout, 'WAIT', @replica_size, timeout_msec - 100) if client.call('ROLE').first == 'master'
       true
+    rescue ::RedisClient::ConnectionError
+      true
     end
   end
 
@@ -214,6 +224,8 @@ def wait_cluster_recovering(clients, max_attempts:)
       else
         true
       end
+    rescue ::RedisClient::ConnectionError
+      true
     end
   end
 
@@ -241,9 +253,13 @@ def hashify_cluster_node_flags(clients, client: nil)
       .to_h { |arr| [id2key[arr[0]], (arr[2].split(',') & %w[master slave]).first] }
   end
 
-  def hashify_node_map(clients, client: nil)
+  def hashify_node_map(clients)
     id2key = fetch_internal_id_to_node_key_mappings(clients)
-    fetch_cluster_nodes(client || clients.first).to_h { |arr| [id2key[arr[0]], arr[0]] }
+    clients.each do |client|
+      return fetch_cluster_nodes(client).to_h { |arr| [id2key[arr[0]], arr[0]] }
+    rescue ::RedisClient::ConnectionError
+      next
+    end
   end
 
   def fetch_internal_id_by_natted_node_key(client, node_key)

test/redis_client/cluster/test_node.rb

@@ -283,6 +283,13 @@ def test_update_slot
         assert_equal(another_node_key, @test_node.find_node_key_of_primary(sample_slot))
       end
 
+      def test_replicated?
+        @test_node_info.select { |info| info[:role] == 'slave' }.each do |replica_info|
+          primary_info = @test_node_info.find { |info| info[:id] == replica_info[:primary_id] }
+          assert(@test_node.replicated?(primary_info[:node_key], replica_info[:node_key]))
+        end
+      end
+
       def test_replica_disabled?
         assert(@test_node.send(:replica_disabled?))
         refute(@test_node_with_scale_read.send(:replica_disabled?))
@@ -331,6 +338,7 @@ def test_build_replication_mappings
         node_key7 = '127.0.0.1:7007'
         node_key8 = '127.0.0.1:7008'
         node_key9 = '127.0.0.1:7009'
+
         node_info = [
           { id: '1', node_key: node_key1, primary_id: '-' },
           { id: '2', node_key: node_key2, primary_id: '-' },
@@ -344,12 +352,26 @@ def test_build_replication_mappings
         ]
         got = @test_node.send(:build_replication_mappings, node_info)
         got.transform_values!(&:sort!)
-        assert_same(node_key4, got[node_key1][0])
-        assert_same(node_key7, got[node_key1][1])
-        assert_same(node_key5, got[node_key2][0])
-        assert_same(node_key8, got[node_key2][1])
-        assert_same(node_key6, got[node_key3][0])
-        assert_same(node_key9, got[node_key3][1])
+        assert_same(node_key4, got[node_key1][0], 'Case: regular')
+        assert_same(node_key7, got[node_key1][1], 'Case: regular')
+        assert_same(node_key5, got[node_key2][0], 'Case: regular')
+        assert_same(node_key8, got[node_key2][1], 'Case: regular')
+        assert_same(node_key6, got[node_key3][0], 'Case: regular')
+        assert_same(node_key9, got[node_key3][1], 'Case: regular')
+
+        node_info = [
+          { id: '1', role: 'master', node_key: node_key1, primary_id: '-' },
+          { id: '3', role: 'master', node_key: node_key3, primary_id: '-' },
+          { id: '4', role: 'slave', node_key: node_key4, primary_id: '1' },
+          { id: '5', role: 'master', node_key: node_key5, primary_id: '-' },
+          { id: '6', role: 'slave', node_key: node_key6, primary_id: '3' }
+        ]
+        got = @test_node.send(:build_replication_mappings, node_info)
+        got.transform_values!(&:sort!)
+        assert_equal(3, got.size, 'Case: lack of replica')
+        assert_same(node_key4, got[node_key1][0], 'Case: lack of replica')
+        assert_same(node_key6, got[node_key3][0], 'Case: lack of replica')
+        assert_empty(got[node_key5], 'Case: lack of replica')
       end
 
       def test_build_clients # rubocop:disable Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity

test/test_against_cluster_broken.rb

@@ -0,0 +1,72 @@
+# frozen_string_literal: true
+
+require 'testing_helper'
+
+class TestAgainstClusterBroken < TestingWrapper
+  def setup
+    @client = new_test_client
+    @controller = new_cluster_controller
+  end
+
+  def teardown
+    @client.close
+    @controller.close
+  end
+
+  def test_several_nodes_are_down # rubocop:disable Metrics/CyclomaticComplexity
+    skip('TODO: https://github.com/redis-rb/redis-cluster-client/issues/9')
+
+    node = @client.instance_variable_get(:@node)
+    primary_node_key = node.primary_node_keys.sample
+    replica_node_key = node.replica_node_keys.reject { |k| node.replicated?(primary_node_key, k) }.sample
+
+    msgs = []
+    msgs += @client.call('CLUSTER', 'NODES').split("\n")
+    msgs += ["Primary: #{primary_node_key}, Replica: #{replica_node_key}"]
+
+    node.instance_variable_get(:@clients).each do |node_key, raw_cli|
+      next if node_key != primary_node_key && node_key != replica_node_key
+
+      # @see https://github.com/redis/redis/blob/475563e2e941ebbdb83f50474bf2daa5ae276fcf/src/debug.c#L387-L493
+      raw_cli.call('SHUTDOWN')
+    end
+
+    @controller.wait_for_cluster_to_be_ready
+
+    errors = []
+    begin
+      @client.call_once('PING')
+    rescue ::RedisClient::Error => e
+      errors << e
+    end
+
+    msgs += @client.call('CLUSTER', 'NODES').split("\n")
+    msgs += errors.group_by(&:message).map do |k, v|
+      "#{v.first.class.name}: #{k}\n#{v.first.backtrace.take(10).join("\n")}"
+    end
+    msg = msgs.join("\n")
+
+    assert_equal([1], @client.instance_variable_get(:@node).node_keys, msg)
+    assert_equal(0, errors.size, msg)
+  end
+
+  private
+
+  def new_test_client
+    config = ::RedisClient::ClusterConfig.new(
+      nodes: TEST_NODE_URIS,
+      replica: true,
+      fixed_hostname: TEST_FIXED_HOSTNAME,
+      **TEST_GENERIC_OPTIONS
+    )
+    ::RedisClient::Cluster.new(config)
+  end
+
+  def new_cluster_controller
+    @controller = ClusterController.new(
+      TEST_NODE_URIS,
+      replica_size: TEST_REPLICA_SIZE,
+      **TEST_GENERIC_OPTIONS.merge(timeout: 30.0)
+    )
+  end
+end