Add middleware to ensure Router only handles errors for its clients

If you nest usages of RedisClient (perhaps becuase you're scanning from
one client whilst inserting into a different cluster, perhaps), it's
possible for errors which were raised from one cluster's RedisClients to
instead be handled by the wrong cluster.

This is bad, because it can mean processing bogus topology "updates"
when e.g. MOVED or ASKING responses get caught by the wrong client.

To fix this, we can add a middleware to our RedisClient instances
(through the documented middleware interface), and tag errors at the
source based on what config object they came from. This can then be used
inside the router to identify errors that it should and should not
handle.

Author: KJ Tsanaktsidis

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