Add multiple replicas test (#35)

Author: supercaracal

.github/workflows/test.yaml

@@ -35,8 +35,7 @@ jobs:
       matrix:
         redis: ['6.2.7', '7.0.1']
         ruby: ['3.1', '3.0', '2.7']
-        #driver: ['ruby', 'hiredis'] # FIXME: freaky
-        driver: ['ruby']
+        driver: ['ruby'] # FIXME: freaky hiredis
         docker: ['docker-compose.yaml', 'docker-compose.ssl.yaml']
         os: ['ubuntu-latest']
     runs-on: ${{ matrix.os }}
@@ -64,3 +63,33 @@ jobs:
         run: bundle exec rake test
       - name: Stop containers
         run: docker compose -f $DOCKER_COMPOSE_FILE down
+  multiple-replicas:
+    name: Multiple Replicas
+    timeout-minutes: 15
+    strategy:
+      fail-fast: false
+    runs-on: ubuntu-latest
+    env:
+      REDIS_VERSION: '7.0.1'
+      REDIS_REPLICA_SIZE: '2'
+      DOCKER_COMPOSE_FILE: 'docker-compose.replica.yaml'
+    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
+      - name: Stop containers
+        run: docker compose -f $DOCKER_COMPOSE_FILE down

Rakefile

@@ -16,6 +16,9 @@ task :wait do
   $LOAD_PATH.unshift(File.expand_path('test', __dir__))
   require 'constants'
   require 'cluster_controller'
-  ::ClusterController.new(TEST_NODE_URIS, **TEST_GENERIC_OPTIONS)
-                     .wait_for_cluster_to_be_ready
+  ::ClusterController.new(
+    TEST_NODE_URIS,
+    replica_size: TEST_REPLICA_SIZE,
+    **TEST_GENERIC_OPTIONS
+  ).wait_for_cluster_to_be_ready
 end

docker-compose.replica.yaml

@@ -0,0 +1,88 @@
+---
+# 3 shards with double replicas
+services:
+  node1: &node
+    image: "redis:${REDIS_VERSION:-7}"
+    command: >
+      redis-server
+      --maxmemory            64mb
+      --maxmemory-policy     allkeys-lru
+      --appendonly           yes
+      --cluster-enabled      yes
+      --cluster-config-file  nodes.conf
+      --cluster-node-timeout 5000
+    healthcheck:
+      test: ["CMD", "redis-cli", "ping"]
+      interval: "7s"
+      timeout: "5s"
+      retries: 10
+    ports:
+      - "6379:6379"
+  node2:
+    <<: *node
+    ports:
+      - "6380:6379"
+  node3:
+    <<: *node
+    ports:
+      - "6381:6379"
+  node4:
+    <<: *node
+    ports:
+      - "6382:6379"
+  node5:
+    <<: *node
+    ports:
+      - "6383:6379"
+  node6:
+    <<: *node
+    ports:
+      - "6384:6379"
+  node7:
+    <<: *node
+    ports:
+      - "6385:6379"
+  node8:
+    <<: *node
+    ports:
+      - "6386:6379"
+  node9:
+    <<: *node
+    ports:
+      - "6387:6379"
+  clustering:
+    image: "redis:${REDIS_VERSION:-7}"
+    command: >
+      bash -c "apt-get update > /dev/null
+      && apt-get install --no-install-recommends --no-install-suggests -y dnsutils > /dev/null
+      && rm -rf /var/lib/apt/lists/*
+      && yes yes | redis-cli --cluster create
+      $$(dig node1 +short):6379
+      $$(dig node2 +short):6379
+      $$(dig node3 +short):6379
+      $$(dig node4 +short):6379
+      $$(dig node5 +short):6379
+      $$(dig node6 +short):6379
+      $$(dig node7 +short):6379
+      $$(dig node8 +short):6379
+      $$(dig node9 +short):6379
+      --cluster-replicas 2"
+    depends_on:
+      node1:
+        condition: service_healthy
+      node2:
+        condition: service_healthy
+      node3:
+        condition: service_healthy
+      node4:
+        condition: service_healthy
+      node5:
+        condition: service_healthy
+      node6:
+        condition: service_healthy
+      node7:
+        condition: service_healthy
+      node8:
+        condition: service_healthy
+      node9:
+        condition: service_healthy

lib/redis_client/cluster/node.rb

@@ -129,19 +129,17 @@ def call_primary(method, *command, **kwargs, &block)
       def call_replica(method, *command, **kwargs, &block)
         return call_primary(method, *command, **kwargs, &block) if replica_disabled?
 
+        replica_node_keys = @replications.values.map(&:sample)
         try_map do |node_key, client|
-          next if primary?(node_key)
+          next if primary?(node_key) || !replica_node_keys.include?(node_key)
 
           client.send(method, *command, **kwargs, &block)
         end.values
       end
 
       def scale_reading_clients
-        clients = @clients.select do |node_key, _|
-          replica_disabled? ? primary?(node_key) : replica?(node_key)
-        end
-
-        clients.values.sort_by do |client|
+        keys = replica_disabled? ? @replications.keys : @replications.values.map(&:first)
+        @clients.select { |k, _| keys.include?(k) }.values.sort_by do |client|
           ::RedisClient::Cluster::NodeKey.build_from_host_port(client.config.host, client.config.port)
         end
       end
@@ -184,7 +182,7 @@ def primary?(node_key)
       end
 
       def replica?(node_key)
-        !(@replications.nil? || @replications.size.zero?) && @replications[node_key].size.zero?
+        !(@replications.nil? || @replications.size.zero?) && !@replications.key?(node_key)
       end
 
       def build_slot_node_mappings(node_info)
@@ -203,7 +201,6 @@ def build_replication_mappings(node_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]]
         end
       end
 

test/cluster_controller.rb

@@ -2,10 +2,12 @@
 
 class ClusterController
   SLOT_SIZE = 16_384
+  SHARD_SIZE = 3
 
-  def initialize(node_addrs, timeout: 30.0, reconnect_attempts: 3, **kwargs)
-    raise 'Redis Cluster requires at least 3 master nodes.' if node_addrs.size < 3
+  def initialize(node_addrs, replica_size: 1, timeout: 30.0, reconnect_attempts: 3, **kwargs)
+    raise "Redis Cluster requires at least #{SHARD_SIZE} master nodes." if node_addrs.size < SHARD_SIZE
 
+    @replica_size = replica_size
     @timeout = kwargs[:timeout]
     @clients = node_addrs.map do |addr|
       ::RedisClient.new(
@@ -199,7 +201,7 @@ def wait_cluster_building(clients, max_attempts: 600)
   def wait_replication(clients, max_attempts: 600)
     wait_for_state(clients, max_attempts) do |client|
       flags = hashify_cluster_node_flags(client)
-      flags.values.count { |f| f == 'slave' } == 3
+      flags.values.count { |f| f == 'slave' } == @replica_size * SHARD_SIZE
     end
   end
 
@@ -266,17 +268,11 @@ def hashify_node_map(client)
   end
 
   def take_masters(clients)
-    size = clients.size / 2
-    return clients if size < 3
-
-    clients.take(size)
+    clients.take(SHARD_SIZE)
   end
 
   def take_slaves(clients)
-    size = clients.size / 2
-    return [] if size < 3
-
-    clients[size..size * 2]
+    clients[SHARD_SIZE..]
   end
 
   def take_replication_pairs(clients)

test/constants.rb

@@ -1,18 +1,35 @@
 # frozen_string_literal: true
 
+# rubocop:disable Lint/UnderscorePrefixedVariableName
+
 require 'redis_client'
 
 TEST_REDIS_HOST = '127.0.0.1'
-TEST_REDIS_PORTS = (6379..6384).freeze
+TEST_REDIS_PORT = 6379
 TEST_TIMEOUT_SEC = 5.0
 TEST_RECONNECT_ATTEMPTS = 3
 
+_new_raw_cli = ->(**opts) { ::RedisClient.config(host: TEST_REDIS_HOST, port: TEST_REDIS_PORT, **opts).new_client }
+_test_cert_path = ->(f) { File.expand_path(File.join('ssl_certs', f), __dir__) }
+
+TEST_SSL_PARAMS = {
+  ca_file: _test_cert_path.call('redis-rb-ca.crt'),
+  cert: _test_cert_path.call('redis-rb-cert.crt'),
+  key: _test_cert_path.call('redis-rb-cert.key')
+}.freeze
+
+_base_opts = {
+  timeout: TEST_TIMEOUT_SEC,
+  reconnect_attempts: TEST_RECONNECT_ATTEMPTS
+}.freeze
+
+_ssl_opts = {
+  ssl: true,
+  ssl_params: TEST_SSL_PARAMS
+}.freeze
+
 begin
-  ::RedisClient.config(
-    host: TEST_REDIS_HOST,
-    port: TEST_REDIS_PORTS.first,
-    timeout: TEST_TIMEOUT_SEC
-  ).new_client.call('PING')
+  _new_raw_cli.call(**_base_opts).call('PING')
   TEST_REDIS_SCHEME = 'redis'
 rescue ::RedisClient::ConnectionError => e
   raise e if e.message != 'Connection reset by peer'
@@ -21,28 +38,17 @@
 end
 
 TEST_REDIS_SSL = TEST_REDIS_SCHEME == 'rediss'
-TEST_REPLICA_SIZE = 1
-TEST_NUMBER_OF_REPLICAS = 3
 TEST_FIXED_HOSTNAME = TEST_REDIS_SSL ? TEST_REDIS_HOST : nil
 
+TEST_SHARD_SIZE = 3
+TEST_REPLICA_SIZE = ENV.fetch('REDIS_REPLICA_SIZE', '1').to_i
+TEST_NUMBER_OF_REPLICAS = TEST_REPLICA_SIZE * TEST_SHARD_SIZE
+TEST_NUMBER_OF_NODES = TEST_SHARD_SIZE + TEST_NUMBER_OF_REPLICAS
+
+TEST_REDIS_PORTS = TEST_REDIS_PORT.upto(TEST_REDIS_PORT + TEST_NUMBER_OF_NODES - 1).to_a.freeze
 TEST_NODE_URIS = TEST_REDIS_PORTS.map { |v| "#{TEST_REDIS_SCHEME}://#{TEST_REDIS_HOST}:#{v}" }.freeze
 TEST_NODE_OPTIONS = TEST_REDIS_PORTS.to_h { |v| ["#{TEST_REDIS_HOST}:#{v}", { host: TEST_REDIS_HOST, port: v }] }.freeze
 
-GET_TEST_CERT_PATH = ->(f) { File.expand_path(File.join('ssl_certs', f), __dir__) }
-TEST_GENERIC_OPTIONS = if TEST_REDIS_SSL
-                         {
-                           timeout: TEST_TIMEOUT_SEC,
-                           reconnect_attempts: TEST_RECONNECT_ATTEMPTS,
-                           ssl: true,
-                           ssl_params: {
-                             ca_file: GET_TEST_CERT_PATH.call('redis-rb-ca.crt'),
-                             cert: GET_TEST_CERT_PATH.call('redis-rb-cert.crt'),
-                             key: GET_TEST_CERT_PATH.call('redis-rb-cert.key')
-                           }
-                         }.freeze
-                       else
-                         {
-                           timeout: TEST_TIMEOUT_SEC,
-                           reconnect_attempts: TEST_RECONNECT_ATTEMPTS
-                         }.freeze
-                       end
+TEST_GENERIC_OPTIONS = TEST_REDIS_SSL ? _base_opts.merge(_ssl_opts).freeze : _base_opts
+
+# rubocop:enable Lint/UnderscorePrefixedVariableName

test/redis_client/cluster/test_command.rb

@@ -8,7 +8,7 @@
 
 class RedisClient
   class Cluster
-    class TestCommand < Minitest::Test
+    class TestCommand < TestingWrapper
       def setup
         @raw_clients = TEST_NODE_URIS.map { |addr| ::RedisClient.config(url: addr, **TEST_GENERIC_OPTIONS).new_client }
       end

test/redis_client/cluster/test_errors.rb

@@ -5,7 +5,7 @@
 
 class RedisClient
   class Cluster
-    class TestErrors < Minitest::Test
+    class TestErrors < TestingWrapper
       DummyError = Struct.new('DummyError', :message)
 
       def test_initial_setup_error

test/redis_client/cluster/test_key_slot_converter.rb

@@ -5,7 +5,7 @@
 
 class RedisClient
   class Cluster
-    class TestKeySlotConverter < Minitest::Test
+    class TestKeySlotConverter < TestingWrapper
       def setup
         @raw_clients = TEST_NODE_URIS.map { |addr| ::RedisClient.config(url: addr, **TEST_GENERIC_OPTIONS).new_client }
       end

test/redis_client/cluster/test_node.rb

@@ -10,7 +10,7 @@
 class RedisClient
   class Cluster
     class Node
-      class TestConfig < Minitest::Test
+      class TestConfig < TestingWrapper
         def test_connection_prelude
           [
             { params: { scale_read: true }, want: [%w[HELLO 3], %w[READONLY]] },
@@ -24,7 +24,7 @@ def test_connection_prelude
       end
     end
 
-    class TestNode < Minitest::Test
+    class TestNode < TestingWrapper
       def setup
         @test_config = ::RedisClient::ClusterConfig.new(
           nodes: TEST_NODE_URIS,
@@ -232,22 +232,22 @@ def test_call_primary
 
       def test_call_replica
         want = (1..(@test_node_info.count { |info| info[:role] == 'master' })).map { |_| 'PONG' }
+
         got = @test_node.call_replica(:call, 'PING')
         assert_equal(want, got, 'Case: primary only')
 
-        want = (1..(@test_node_info.count { |info| info[:role] == 'slave' })).map { |_| 'PONG' }
         got = @test_node_with_scale_read.call_replica(:call, 'PING')
         assert_equal(want, got, 'Case: scale read')
       end
 
-      def test_scale_reading_clients
+      def test_scale_reading_clients # rubocop:disable Metrics/CyclomaticComplexity
         want = @test_node_info.select { |info| info[:role] == 'master' }.map { |info| info[:node_key] }.sort
         got = @test_node.scale_reading_clients.map { |client| "#{client.config.host}:#{client.config.port}" }
         assert_equal(want, got, 'Case: primary only')
 
-        want = @test_node_info.select { |info| info[:role] == 'slave' }.map { |info| info[:node_key] }.sort
+        want = @test_node_info.select { |info| info[:role] == 'slave' }.map { |info| info[:node_key] }
         got = @test_node_with_scale_read.scale_reading_clients.map { |client| "#{client.config.host}:#{client.config.port}" }
-        assert_equal(want, got, 'Case: scale read')
+        got.each { |e| assert_includes(want, e, 'Case: scale read') }
       end
 
       def test_slot_exists?
@@ -271,11 +271,12 @@ def test_find_node_key_of_replica
         got = @test_node.find_node_key_of_replica(sample_slot)
         assert_equal(sample_node[:node_key], got, 'Case: primary only')
 
-        sample_replica = @test_node_info.find { |info| info[:role] == 'slave' }
-        sample_primary = @test_node_info.find { |info| info[:id] == sample_replica[:primary_id] }
+        sample_replicas = @test_node_info.select { |info| info[:role] == 'slave' }
+        sample_primary = @test_node_info.find { |info| info[:id] == sample_replicas.first[:primary_id] }
         sample_slot = sample_primary[:slots].first.first
         got = @test_node_with_scale_read.find_node_key_of_replica(sample_slot)
-        assert_equal(sample_replica[:node_key], got, 'Case: scale read')
+        want = sample_replicas.map { |info| info[:node_key] }
+        assert_includes(want, got, 'Case: scale read')
       end
 
       def test_update_slot