feat: add an option to the pipeline feature to be able to select either throwing an error or returning errors as is (#343)

Author: supercaracal

lib/redis_client/cluster.rb

@@ -82,9 +82,16 @@ def zscan(key, *args, **kwargs, &block)
       @router.try_delegate(node, :zscan, key, *args, **kwargs, &block)
     end
 
-    def pipelined
+    def pipelined(exception: true)
       seed = @config.use_replica? && @config.replica_affinity == :random ? nil : Random.new_seed
-      pipeline = ::RedisClient::Cluster::Pipeline.new(@router, @command_builder, @concurrent_worker, seed: seed)
+      pipeline = ::RedisClient::Cluster::Pipeline.new(
+        @router,
+        @command_builder,
+        @concurrent_worker,
+        exception: exception,
+        seed: seed
+      )
+
       yield pipeline
       return [] if pipeline.empty?
 

lib/redis_client/cluster/pipeline.rb

@@ -12,7 +12,7 @@ class Pipeline
       class Extended < ::RedisClient::Pipeline
         attr_reader :outer_indices
 
-        def initialize(command_builder)
+        def initialize(...)
           super
           @outer_indices = nil
         end
@@ -50,14 +50,14 @@ def get_block(inner_index)
       end
 
       ::RedisClient::ConnectionMixin.module_eval do
-        def call_pipelined_aware_of_redirection(commands, timeouts) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
+        def call_pipelined_aware_of_redirection(commands, timeouts, exception:) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
           size = commands.size
           results = Array.new(commands.size)
           @pending_reads += size
           write_multi(commands)
 
           redirection_indices = nil
-          exception = nil
+          first_exception = nil
           size.times do |index|
             timeout = timeouts && timeouts[index]
             result = read(timeout)
@@ -67,14 +67,14 @@ def call_pipelined_aware_of_redirection(commands, timeouts) # rubocop:disable Me
               if result.is_a?(::RedisClient::CommandError) && result.message.start_with?('MOVED', 'ASK')
                 redirection_indices ||= []
                 redirection_indices << index
-              else
-                exception ||= result
+              elsif exception
+                first_exception ||= result
               end
             end
             results[index] = result
           end
 
-          raise exception if exception
+          raise first_exception if exception && first_exception
           return results if redirection_indices.nil?
 
           err = ::RedisClient::Cluster::Pipeline::RedirectionNeeded.new
@@ -98,10 +98,11 @@ class RedirectionNeeded < ::RedisClient::Error
         attr_accessor :replies, :indices
       end
 
-      def initialize(router, command_builder, concurrent_worker, seed: Random.new_seed)
+      def initialize(router, command_builder, concurrent_worker, exception:, seed: Random.new_seed)
         @router = router
         @command_builder = command_builder
         @concurrent_worker = concurrent_worker
+        @exception = exception
         @seed = seed
         @pipelines = nil
         @size = 0
@@ -212,7 +213,7 @@ def send_pipeline(client, pipeline)
         results = client.ensure_connected_cluster_scoped(retryable: pipeline._retryable?) do |connection|
           commands = pipeline._commands
           client.middlewares.call_pipelined(commands, client.config) do
-            connection.call_pipelined_aware_of_redirection(commands, pipeline._timeouts)
+            connection.call_pipelined_aware_of_redirection(commands, pipeline._timeouts, exception: @exception)
           end
         end
 
@@ -224,15 +225,21 @@ def handle_redirection(err, pipeline, inner_index)
 
         if err.message.start_with?('MOVED')
           node = @router.assign_redirection_node(err.message)
-          redirect_command(node, pipeline, inner_index)
+          try_redirection(node, pipeline, inner_index)
         elsif err.message.start_with?('ASK')
           node = @router.assign_asking_node(err.message)
-          try_asking(node) ? redirect_command(node, pipeline, inner_index) : err
+          try_asking(node) ? try_redirection(node, pipeline, inner_index) : err
         else
           err
         end
       end
 
+      def try_redirection(node, pipeline, inner_index)
+        redirect_command(node, pipeline, inner_index)
+      rescue StandardError => e
+        @exception ? raise : e
+      end
+
       def redirect_command(node, pipeline, inner_index)
         method = pipeline.get_callee_method(inner_index)
         command = pipeline.get_command(inner_index)

test/redis_client/test_cluster.rb

@@ -170,6 +170,28 @@ def test_pipelined_with_errors
         10.times { |i| assert_equal((i + 10).to_s, @client.call('GET', "string#{i}")) }
       end
 
+      def test_pipelined_with_errors_as_is
+        got = @client.pipelined(exception: false) do |pipeline|
+          10.times do |i|
+            pipeline.call('SET', "string#{i}", i)
+            pipeline.call('SET', "string#{i}", i, 'too many args')
+            pipeline.call('SET', "string#{i}", i + 10)
+          end
+        end
+
+        assert_equal(30, got.size)
+
+        10.times do |i|
+          assert_equal('OK', got[(3 * i) + 0])
+          assert_instance_of(::RedisClient::CommandError, got[(3 * i) + 1])
+          assert_equal('OK', got[(3 * i) + 2])
+        end
+
+        wait_for_replication
+
+        10.times { |i| assert_equal((i + 10).to_s, @client.call('GET', "string#{i}")) }
+      end
+
       def test_pipelined_with_many_commands
         @client.pipelined { |pi| 1000.times { |i| pi.call('SET', i, i) } }
         wait_for_replication