Add middleware to ensure Router only handles errors for its clients (#310)

Author: supercaracal

lib/redis_client/cluster/error_identification.rb

@@ -0,0 +1,43 @@
+# frozen_string_literal: true
+
+class RedisClient
+  class Cluster
+    module ErrorIdentification
+      def self.client_owns_error?(err, client)
+        err.is_a?(TaggedError) && err.from?(client)
+      end
+
+      module TaggedError
+        attr_accessor :config_instance
+
+        def from?(client)
+          client.config.equal?(config_instance)
+        end
+      end
+
+      module Middleware
+        def connect(config)
+          super
+        rescue RedisClient::Error => e
+          identify_error(e, config)
+          raise
+        end
+
+        def call(_command, config)
+          super
+        rescue RedisClient::Error => e
+          identify_error(e, config)
+          raise
+        end
+        alias call_pipelined call
+
+        private
+
+        def identify_error(err, config)
+          err.singleton_class.include(TaggedError)
+          err.config_instance = config
+        end
+      end
+    end
+  end
+end

lib/redis_client/cluster/node.rb

@@ -2,6 +2,7 @@
 
 require 'redis_client'
 require 'redis_client/config'
+require 'redis_client/cluster/error_identification'
 require 'redis_client/cluster/errors'
 require 'redis_client/cluster/node/primary_only'
 require 'redis_client/cluster/node/random_replica'
@@ -78,9 +79,11 @@ def []=(index, element)
       end
 
       class Config < ::RedisClient::Config
-        def initialize(scale_read: false, **kwargs)
+        def initialize(scale_read: false, middlewares: nil, **kwargs)
           @scale_read = scale_read
-          super(**kwargs)
+          middlewares ||= []
+          middlewares.unshift ErrorIdentification::Middleware
+          super(middlewares: middlewares, **kwargs)
         end
 
         private
@@ -214,6 +217,10 @@ def reload!
         end
       end
 
+      def owns_error?(err)
+        any? { |c| ErrorIdentification.client_owns_error?(err, c) }
+      end
+
       private
 
       def make_topology_class(with_replica, replica_affinity)

lib/redis_client/cluster/router.rb

@@ -65,7 +65,7 @@ def send_command(method, command, *args, &block) # rubocop:disable Metrics/AbcSi
         raise if e.errors.any?(::RedisClient::CircuitBreaker::OpenCircuitError)
 
         update_cluster_info! if e.errors.values.any? do |err|
-          err.message.start_with?('CLUSTERDOWN Hash slot not served')
+          @node.owns_error?(err) && err.message.start_with?('CLUSTERDOWN Hash slot not served')
         end
 
         raise
@@ -94,6 +94,8 @@ def handle_redirection(node, retry_count:) # rubocop:disable Metrics/AbcSize, Me
       rescue ::RedisClient::CircuitBreaker::OpenCircuitError
         raise
       rescue ::RedisClient::CommandError => e
+        raise unless ErrorIdentification.client_owns_error?(e, node)
+
         if e.message.start_with?('MOVED')
           node = assign_redirection_node(e.message)
           retry_count -= 1
@@ -111,7 +113,9 @@ def handle_redirection(node, retry_count:) # rubocop:disable Metrics/AbcSize, Me
           retry if retry_count >= 0
         end
         raise
-      rescue ::RedisClient::ConnectionError
+      rescue ::RedisClient::ConnectionError => e
+        raise unless ErrorIdentification.client_owns_error?(e, node)
+
         update_cluster_info!
 
         raise if retry_count <= 0

test/redis_client/test_cluster.rb

@@ -485,6 +485,24 @@ def test_circuit_breakers
         cli&.close
       end
 
+      def test_only_reshards_own_errors
+        @client.call_v(%w[SADD testkey testvalue1])
+        @client.call_v(%w[SADD testkey testvalue2])
+        slot = ::RedisClient::Cluster::KeySlotConverter.convert('testkey')
+        router = @client.instance_variable_get(:@router)
+        correct_primary_key = router.find_node_key_by_key('testkey', primary: true)
+        broken_primary_key = (router.node_keys - [correct_primary_key]).first
+        assert_raises(RedisClient::CommandError) do
+          @client.sscan('testkey', retry_count: 0) do
+            raise RedisClient::CommandError, "MOVED #{slot} #{broken_primary_key}"
+          end
+        end
+
+        # The exception should not have causes @client to update its shard mappings, because it didn't
+        # come from a RedisClient instance that @client knows about.
+        assert_equal correct_primary_key, router.find_node_key_by_key('testkey', primary: true)
+      end
+
       private
 
       def wait_for_replication