Don't attempt redirection on the actual MULTI for WATCH...MULTI

If we have a WATCH, then the MULTI _must_ exec on exactly that node; it
should not be allowed for that to be redirected to a different node,
because then the MULTI isn't on the same connection as the WATCH
anymore!

The first part of this patch fixes this by disabling redirection
handling in transaction.rb if we are in a watch. I also added a test
test_the_state_of_cluster_resharding_with_reexecuted_watch for this.

However, the second part of this patch is a lot trickier...

This change causes a different test,
test_the_state_of_cluster_resharding_with_transaction_and_watch, to
break. That test is asserting that
- if we watch something where the slot is in the middle of migrating
  between nodes,
- and all the keys we wanted to watch are on the new node,
- that the client correctly retries the transaction with ASKING against
  the new node

Disabling redirection handling obviously makes this stop working, and it
_is_ possible to handle this case correctly. We need to record whether
or not we had to issue an ASKING on the WATCH for the transaction, and
if so, pre-emptively issue an ASKING on the MULTI too. That's because
this slot is not yet actually assigned to the node we're connected to
(it's IMPORTING).

It may well not be worth it, and I'm also OK with just failing
WATCH/MULTI on slots which are currently migrating. That would imply:

- Keeping test_the_state_of_cluster_resharding_with_reexecuted_watch
- Deleting test_the_state_of_cluster_resharding_with_transaction_and_watch

Author: KJ Tsanaktsidis

lib/redis_client/cluster.rb

@@ -98,9 +98,9 @@ def multi(watch: nil)
         return transaction.execute
       end
 
-      ::RedisClient::Cluster::OptimisticLocking.new(@router).watch(watch) do |c, slot|
+      ::RedisClient::Cluster::OptimisticLocking.new(@router).watch(watch) do |c, slot, asking|
         transaction = ::RedisClient::Cluster::Transaction.new(
-          @router, @command_builder, node: c, slot: slot
+          @router, @command_builder, node: c, slot: slot, asking: asking
         )
         yield transaction
         transaction.execute

lib/redis_client/cluster/optimistic_locking.rb

@@ -8,6 +8,7 @@ class Cluster
     class OptimisticLocking
       def initialize(router)
         @router = router
+        @asking = false
       end
 
       def watch(keys)
@@ -17,12 +18,13 @@ def watch(keys)
         # We have not yet selected a node for this transaction, initially, which means we can handle
         # redirections freely initially (i.e. for the first WATCH call)
         node = @router.find_primary_node_by_slot(slot)
-        @router.handle_redirection(node, retry_count: 1) do |nd|
+        handle_redirection(node, retry_count: 1) do |nd|
           nd.with do |c|
             c.ensure_connected_cluster_scoped(retryable: false) do
+              c.call('ASKING') if @asking
               c.call('WATCH', *keys)
               begin
-                yield(c, slot)
+                yield(c, slot, @asking)
               rescue ::RedisClient::ConnectionError
                 # No need to unwatch on a connection error.
                 raise
@@ -37,6 +39,25 @@ def watch(keys)
 
       private
 
+      def handle_redirection(node, retry_count: 1, &blk)
+        @router.handle_redirection(node, retry_count: retry_count) do |nd|
+          handle_asking_once(nd, &blk)
+        end
+      end
+
+      def handle_asking_once(node)
+        yield node
+      rescue ::RedisClient::CommandError => e
+        raise unless ErrorIdentification.client_owns_error?(e, node)
+        raise unless e.message.start_with?('ASK')
+
+        node = @router.assign_asking_node(e.message)
+        @asking = true
+        yield node
+      ensure
+        @asking = false
+      end
+
       def find_slot(keys)
         return if keys.empty?
         return if keys.any? { |k| k.nil? || k.empty? }

lib/redis_client/cluster/transaction.rb

@@ -9,7 +9,7 @@ class Transaction
       ConsistencyError = Class.new(::RedisClient::Error)
       MAX_REDIRECTION = 2
 
-      def initialize(router, command_builder, node: nil, slot: nil)
+      def initialize(router, command_builder, node: nil, slot: nil, asking: false)
         @router = router
         @command_builder = command_builder
         @retryable = true
@@ -18,6 +18,7 @@ def initialize(router, command_builder, node: nil, slot: nil)
         @node = node
         prepare_tx unless @node.nil?
         @watching_slot = slot
+        @asking = asking
       end
 
       def call(*command, **kwargs, &block)
@@ -93,7 +94,11 @@ def prepare_tx
 
       def settle
         @pipeline.call('EXEC')
-        send_transaction(@node, redirect: MAX_REDIRECTION)
+        # If we needed ASKING on the watch, we need ASKING on the multi as well.
+        @node.call('ASKING') if @asking
+        # Don't handle redirections at this level if we're in a watch (the watcher handles redirections
+        # at the whole-transaction level.)
+        send_transaction(@node, redirect: !!@watching_slot ? 0 : MAX_REDIRECTION)
       end
 
       def send_transaction(client, redirect:)
@@ -110,7 +115,8 @@ def send_pipeline(client, redirect:)
           client.middlewares.call_pipelined(commands, client.config) do
             connection.call_pipelined(commands, nil)
           rescue ::RedisClient::CommandError => e
-            return handle_command_error!(commands, e, redirect: redirect) unless redirect.zero?
+            ensure_the_same_slot!(commands)
+            return handle_command_error!(e, redirect: redirect) unless redirect.zero?
 
             raise
           end
@@ -139,15 +145,13 @@ def coerce_results!(results, offset: 1)
         results
       end
 
-      def handle_command_error!(commands, err, redirect:) # rubocop:disable Metrics/AbcSize
+      def handle_command_error!(err, redirect:) # rubocop:disable Metrics/AbcSize
         if err.message.start_with?('CROSSSLOT')
           raise ConsistencyError, "#{err.message}: #{err.command}"
         elsif err.message.start_with?('MOVED')
-          ensure_the_same_slot!(commands)
           node = @router.assign_redirection_node(err.message)
           send_transaction(node, redirect: redirect - 1)
         elsif err.message.start_with?('ASK')
-          ensure_the_same_slot!(commands)
           node = @router.assign_asking_node(err.message)
           try_asking(node) ? send_transaction(node, redirect: redirect - 1) : err
         else

test/test_against_cluster_state.rb

@@ -103,6 +103,35 @@ def test_the_state_of_cluster_resharding_with_transaction_and_watch
       assert_equal(1, call_cnt)
     end
 
+    def test_the_state_of_cluster_resharding_with_reexecuted_watch
+      client2 = new_test_client
+      call_cnt = 0
+
+      @client.call('SET', 'watch_key', 'original_value')
+      @client.multi(watch: %w[watch_key]) do |tx|
+        # Use client2 to change the value of watch_key, which would cause this transaction to fail
+        if call_cnt == 0
+          client2.call('SET', 'watch_key', 'client2_value')
+
+          # Now perform (and _finish_) a reshard, which should make this transaction receive a MOVED
+          # redirection when it goes to commit. That should result in the entire block being retried
+          slot = ::RedisClient::Cluster::KeySlotConverter.convert('watch_key')
+          src, dest = @controller.select_resharding_target(slot)
+          @controller.start_resharding(slot: slot, src_node_key: src, dest_node_key: dest)
+          @controller.finish_resharding(slot: slot, src_node_key: src, dest_node_key: dest)
+        end
+        call_cnt += 1
+
+        tx.call('SET', 'watch_key', "@client_value_#{call_cnt}")
+      end
+      # It should have retried the entire transaction block.
+      assert_equal(2, call_cnt)
+      # The second call succeeded
+      assert_equal('@client_value_2', @client.call('GET', 'watch_key'))
+    ensure
+      client2&.close
+    end
+
     def test_the_state_of_cluster_resharding_with_pipelining_on_new_connection
       # This test is excercising a very delicate race condition; i think the use of @client to set
       # the keys in do_resharding_test is actually causing the race condition not to happen, so this