Asycnio ClusterPipeline execute won't execute any commands when cluster_error_retry_attempts is 0

[ryan-smylski]

Version: 5.0.3 (code hasn't changed in any subsequent version)

Platform: Python 3.8.6 / Almalinux 3.9 / AWS

Description: Calls to delete in asyncio RedisCluster do not execute if cluster_error_retry_attempts is 0
Last part of stack trace

    return sum(await self._execute_pipeline_by_slot(command, slots_to_keys))
  File "/usr/local/lib/python3.8/site-packages/redis/commands/cluster.py", line 341, in _execute_pipeline_by_slot
    return await pipe.execute()
  File "/usr/local/lib/python3.8/site-packages/redis/asyncio/cluster.py", line 1502, in execute
    raise exception
UnboundLocalError: local variable 'exception' referenced before assignment

The retry logic in ClusterPipeline is incorrect, in that it will execute cluster_error_retry_attempts - 1 times as range(n) creates a list that is n-1 in length. So, when the variable is 0, it will not even execute the command once.

[ryan-smylski]

Updating ClusterPipeline's execute function to the follow will fix this. (Uses the same approach as RedisCluster.execute_command)

async def execute(
        self, raise_on_error: bool = True, allow_redirections: bool = True
    ) -> List[Any]:
        """
        Execute the pipeline.

        It will retry the commands as specified by :attr:`cluster_error_retry_attempts`
        & then raise an exception.

        :param raise_on_error:
            | Raise the first error if there are any errors
        :param allow_redirections:
            | Whether to retry each failed command individually in case of redirection
              errors

        :raises RedisClusterException: if target_nodes is not provided & the command
            can't be mapped to a slot
        """
        if not self._command_stack:
            return []

        try:
            # Add one for the first execution
            for _ in range(self._client.cluster_error_retry_attempts + 1):
                if self._client._initialize:
                    await self._client.initialize()

                try:
                    return await self._execute(
                        self._client,
                        self._command_stack,
                        raise_on_error=raise_on_error,
                        allow_redirections=allow_redirections,
                    )
                except BaseException as e:
                    if type(e) in self.__class__.ERRORS_ALLOW_RETRY:
                        # Try again with the new cluster setup.
                        exception = e
                        await self._client.aclose()
                        await asyncio.sleep(0.25)
                    else:
                        # All other errors should be raised.
                        raise

            # If it fails the configured number of times then raise an exception
            raise exception
        finally:
            self._command_stack = []