Refactored retries, fixed comments, fixed linters

Author: vladvildanov

docs/advanced_features.rst

@@ -180,7 +180,7 @@ the server.
 Please note:
 
 -  RedisCluster pipelines currently only support key-based commands.
--  The pipeline gets its ‘read_from_replicas’ value from the
+-  The pipeline gets its ‘load_balancing_strategy’ value from the
    cluster’s parameter. Thus, if read from replications is enabled in
    the cluster instance, the pipeline will also direct read commands to
    replicas.
@@ -204,7 +204,7 @@ is exactly that: a compromise. While this behavior is different from
 non-transactional cluster pipelines, it simplifies migration of clients
 from standalone to cluster under some circumstances. Note that application
 code that issues multi/exec commands on a standalone client without
-embedding them within a pipeline would eventually get ‘AttributeError’s.
+embedding them within a pipeline would eventually get ‘AttributeError’.
 With this approach, if the application uses ‘client.pipeline(transaction=True)’,
 then switching the client with a cluster-aware instance would simplify
 code changes (to some extent). This may be true for application code that

redis/cluster.py

@@ -1237,8 +1237,6 @@ def _execute_command(self, target_node, *args, **kwargs):
             except AuthenticationError:
                 raise
             except (ConnectionError, TimeoutError) as e:
-                # Connection retries are being handled in the node's
-                # Retry object.
                 # ConnectionError can also be raised if we couldn't get a
                 # connection from the pool before timing out, so check that
                 # this is an actual connection before attempting to disconnect.
@@ -1854,7 +1852,7 @@ def remap_host_port(self, host: str, port: int) -> Tuple[str, int]:
 
     def find_connection_owner(self, connection: Connection) -> Optional[Redis]:
         node_name = get_node_name(connection.host, connection.port)
-        for node in self.nodes_cache.values():
+        for node in tuple(self.nodes_cache.values()):
             if node.redis_connection:
                 conn_args = node.redis_connection.connection_pool.connection_kwargs
                 if node_name == get_node_name(
@@ -2550,7 +2548,7 @@ def multi(self):
         """
         Starts transactional context.
 
-        See: ClusterPipeline.reset()
+        See: ClusterPipeline.multi()
         """
         pass
 
@@ -2706,9 +2704,9 @@ def send_cluster_commands(
         self, stack, raise_on_error=True, allow_redirections=True
     ):
         """
-        Wrapper for CLUSTERDOWN error handling.
+        Wrapper for RedisCluster.ERRORS_ALLOW_RETRY errors handling.
 
-        If the cluster reports it is down it is assumed that:
+        If one of the retryable exceptions has been thrown we assume that:
          - connection_pool was disconnected
          - connection_pool was reseted
          - refereh_table_asap set to True
@@ -2816,11 +2814,10 @@ def _send_cluster_commands(
         # we  write to all the open sockets for each node first,
         # before reading anything
         # this allows us to flush all the requests out across the
-        # network essentially in parallel
-        # so that we can read them all in parallel as they come back.
+        # network
+        # so that we can read them from different sockets as they come back.
         # we dont' multiplex on the sockets as they come available,
         # but that shouldn't make too much difference.
-        node_commands = nodes.values()
         try:
             node_commands = nodes.values()
             for n in node_commands:
@@ -3024,7 +3021,7 @@ class TransactionStrategy(AbstractStrategy):
     IMMEDIATE_EXECUTE_COMMANDS = {"WATCH", "UNWATCH"}
     UNWATCH_COMMANDS = {"DISCARD", "EXEC", "UNWATCH"}
     SLOT_REDIRECT_ERRORS = (AskError, MovedError)
-    CONNECTION_ERRORS = (ConnectionError,OSError,ClusterDownError)
+    CONNECTION_ERRORS = (ConnectionError, OSError, ClusterDownError)
 
     def __init__(self, pipe: ClusterPipeline):
         super().__init__(pipe)
@@ -3033,6 +3030,10 @@ def __init__(self, pipe: ClusterPipeline):
         self._pipeline_slots: Set[int] = set()
         self._transaction_connection: Optional[Connection] = None
         self._executing = False
+        self._retry = copy(self._pipe.retry)
+        self._retry.update_supported_errors(
+            RedisCluster.ERRORS_ALLOW_RETRY + self.SLOT_REDIRECT_ERRORS
+        )
 
     def _get_client_and_connection_for_transaction(self) -> Tuple[Redis, Connection]:
         """
@@ -3105,9 +3106,7 @@ def _validate_watch(self):
         self._watching = True
 
     def _immediate_execute_command(self, *args, **options):
-        retry = copy(self._pipe.retry)
-        retry.update_supported_errors([AskError, MovedError])
-        return retry.call_with_retry(
+        return self._retry.call_with_retry(
             lambda: self._get_connection_and_send_command(*args, **options),
             self._reinitialize_on_error,
         )
@@ -3137,7 +3136,10 @@ def _reinitialize_on_error(self, error):
             if type(error) in self.SLOT_REDIRECT_ERRORS and self._executing:
                 raise WatchError("Slot rebalancing occurred while watching keys")
 
-        if type(error) in self.SLOT_REDIRECT_ERRORS or type(error) in self.CONNECTION_ERRORS:
+        if (
+            type(error) in self.SLOT_REDIRECT_ERRORS
+            or type(error) in self.CONNECTION_ERRORS
+        ):
             if self._transaction_connection:
                 self._transaction_connection = None
 
@@ -3169,9 +3171,7 @@ def execute(self, raise_on_error: bool = True) -> List[Any]:
     def _execute_transaction_with_retries(
         self, stack: List["PipelineCommand"], raise_on_error: bool
     ):
-        retry = copy(self._pipe.retry)
-        retry.update_supported_errors([AskError, MovedError])
-        return retry.call_with_retry(
+        return self._retry.call_with_retry(
             lambda: self._execute_transaction(stack, raise_on_error),
             self._reinitialize_on_error,
         )

tests/conftest.py

@@ -30,7 +30,7 @@
 from tests.ssl_utils import get_tls_certificates
 
 REDIS_INFO = {}
-default_redis_url = "redis://localhost:6379/0"
+default_redis_url = "redis://localhost:16379/0"
 default_protocol = "2"
 default_redismod_url = "redis://localhost:6479"
 

tests/test_cluster_transaction.py

@@ -11,7 +11,7 @@
 from redis.cluster import PRIMARY, ClusterNode, NodesManager, RedisCluster
 from redis.retry import Retry
 
-from .conftest import skip_if_server_version_lt, wait_for_command
+from .conftest import skip_if_server_version_lt
 
 
 def _find_source_and_target_node_for_slot(