Repository: redis/redis-py (python)

Issue:
IndexError on Connection Failure
**Version**: 
redis-py: `4.3.3`
Redis Version: `6.0.14`

**Platform**: 
- Kubernetes pod
- Docker image
- Python3.9

**Description**: 

We are seeing `IndexError` when we reboot a redis server and our async Redis connection is closed and reconnect happens. The IndexError is [located here](https://github.com/redis/redis-py/blob/bedf3c82a55b4b67eed93f686cb17e82f7ab19cd/redis/asyncio/connection.py#L765) in the codebase. 

Traceback is like this:

```python
  File "/usr/local/lib/python3.9/site-packages/sfdi_events_synd/clients/redis.py", line 81, in ping
    return await client.ping()
  File "/usr/local/lib/python3.9/site-packages/redis/asyncio/client.py", line 473, in execute_command
    conn = self.connection or await pool.get_connection(command_name, **options)
  File "/usr/local/lib/python3.9/site-packages/redis/asyncio/connection.py", line 1700, in get_connection
    await connection.connect()
  File "/usr/local/lib/python3.9/site-packages/redis/asyncio/connection.py", line 706, in connect
    raise ConnectionError(self._error_message(e))
  File "/usr/local/lib/python3.9/site-packages/redis/asyncio/connection.py", line 765, in _error_message
    f"Error {exception.args[0]} connecting to {self.host}:{self.port}. "
IndexError: tuple index out of range
```

It looks like our exception_args is empty?

```python
        if len(exception.args) == 1:
            return f"Error connecting to {self.host}:{self.port}. {exception.args[0]}."
        else:
            return (
                f"Error {exception.args[0]} connecting to {self.host}:{self.port}. "
                f"{exception.args[0]}."
            )
```

We have been experimenting with ConnectionPooling and reconnect strategies, and this came up this morning on an experimental attempt to drop connections and reconnect in code that looks _roughly_ like this:

```python
@asynccontextmanager
async def redis_connection(pool):
    """
    Context manager for automatically closing connections on finish
    """
    connection = redis.Redis(
        connection_pool=pool, auto_close_connection_pool=False
    )
    yield connection
    await connection.close()


async def ping(pool):
    async with redis_connection(pool) as client:
        return await client.ping()
```

As noted above, we run this while rebooting our server to see if we can simulate failing connections and reconnect. We were expecting to see a `ConnectionError` instead of an `IndexError`.

Pull request description:
fix: workaround asyncio bug on connection reset by peer
### Pull Request check-list

_Please make sure to review and check all of these items:_

- [X] Does `$ tox` pass with this change (including linting)?
- [x] Do the CI tests pass with this change (enable it first in your forked repo and wait for the github action build to finish)?
- [x] Is the new or changed code fully tested?
- [x] Is a documentation update included (if this change modifies existing APIs, or introduces new ones)?
- [x] Is there an example added to the examples folder (if applicable)?
- [x] Was the change added to CHANGES file?

### Description of change

fix: workaround asyncio bug on connection reset by peer

Fixes #2237

Patch:
diff --git a/.github/workflows/integration.yaml b/.github/workflows/integration.yaml
index 59f6c7d3..1cce9f36 100644
--- a/.github/workflows/integration.yaml
+++ b/.github/workflows/integration.yaml
@@ -13,6 +13,8 @@ on:
     branches:
       - master
       - '[0-9].[0-9]'
+  schedule:
+    - cron: '0 1 * * *' # nightly build
 
 jobs:
 
@@ -22,7 +24,7 @@ jobs:
      steps:
        - uses: actions/checkout@v2
        - name: install python
-         uses: actions/setup-python@v3
+         uses: actions/setup-python@v4
          with:
            python-version: 3.9
            cache: 'pip'
@@ -37,7 +39,7 @@ jobs:
      strategy:
        max-parallel: 15
        matrix:
-         python-version: ['3.6', '3.7', '3.8', '3.9', '3.10', 'pypy-3.7']
+         python-version: ['3.6', '3.7', '3.8', '3.9', '3.10', 'pypy-3.7', 'pypy-3.8']
          test-type: ['standalone', 'cluster']
          connection-type: ['hiredis', 'plain']
      env:
@@ -46,7 +48,7 @@ jobs:
      steps:
        - uses: actions/checkout@v2
        - name: install python
-         uses: actions/setup-python@v3
+         uses: actions/setup-python@v4
          with:
            python-version: ${{ matrix.python-version }}
            cache: 'pip'
diff --git a/CHANGES b/CHANGES
index 0af421b4..b4841ddb 100644
--- a/CHANGES
+++ b/CHANGES
@@ -1,4 +1,5 @@
 
+    * Add retry mechanism to async version of Connection
     * Compare commands case-insensitively in the asyncio command parser
     * Allow negative `retries` for `Retry` class to retry forever
     * Add `items` parameter to `hset` signature
diff --git a/README.md b/README.md
index 5a4bb717..4241da47 100644
--- a/README.md
+++ b/README.md
@@ -9,51 +9,31 @@ The Python interface to the Redis key-value store.
 [![codecov](https://codecov.io/gh/redis/redis-py/branch/master/graph/badge.svg?token=yenl5fzxxr)](https://codecov.io/gh/redis/redis-py)
 [![Total alerts](https://img.shields.io/lgtm/alerts/g/redis/redis-py.svg?logo=lgtm&logoWidth=18)](https://lgtm.com/projects/g/redis/redis-py/alerts/)
 
-[Installation](#installation) | [Contributing](#contributing) |  [Getting Started](#getting-started) | [Connecting To Redis](#connecting-to-redis)
+[Installation](#installation) | [Contributing](#contributing) |  [Getting Started](#getting-started) | [Connecting To Redis](#connecting-to-redis) | [Contributing](https://github.com/redis/redis-py/blob/master/CONTRIBUTING.md)
 
 ---------------------------------------------
 
 ## Python Notice
 
-redis-py 4.2.x will be the last generation of redis-py to support python 3.6 as it has been [End of Life'd](https://www.python.org/dev/peps/pep-0494/#schedule-last-security-only-release).  Async support was introduced in redis-py 4.2.x thanks to [aioredis](https://github.com/aio-libs/aioredis-py), which necessitates this change. We will continue to maintain 3.6 support as long as possible - but the plan is for redis-py version 5+ to officially remove 3.6.
+redis-py 4.3.x will be the last generation of redis-py to support python 3.6 as it has been [End of Life'd](https://www.python.org/dev/peps/pep-0494/#schedule-last-security-only-release).  Async support was introduced in redis-py 4.2.x thanks to [aioredis](https://github.com/aio-libs/aioredis-py), which necessitates this change. We will continue to maintain 3.6 support as long as possible - but the plan is for redis-py version 4.4+ to officially remove 3.6.
 
 ---------------------------
 
 ## Installation
 
-redis-py requires a running Redis server. See [Redis's
-quickstart](https://redis.io/topics/quickstart) for installation
-instructions.
-
-redis-py can be installed using pip similar to other
-Python packages. Do not use sudo with pip.
-It is usually good to work in a
-[virtualenv](https://virtualenv.pypa.io/en/latest/) or
-[venv](https://docs.python.org/3/library/venv.html) to avoid conflicts
-with other package managers and Python projects. For a quick
-introduction see [Python Virtual Environments in Five
-Minutes](https://bit.ly/py-env).
-
 To install redis-py, simply:
 
 ``` bash
 $ pip install redis
 ```
 
-or from source:
+Looking for a high-level library to handle object mapping? See [redis-om-python](https://github.com/redis/redis-om-python)!
 
-``` bash
-$ python setup.py install
+redis-py requires a running Redis server.  Assuming you have docker
+```bash
+docker run -p 6379:6379 -it redis/redis-stack-server:latest
 ```
 
-View the current documentation [here](https://readthedocs.org/projects/redis/).
-
-## Contributing
-
-Want to contribute a feature, bug fix, or report an issue? Check out
-our [guide to
-contributing](https://github.com/redis/redis-py/blob/master/CONTRIBUTING.md).
-
 ## Getting Started
 
 redis-py supports Python 3.7+.
@@ -81,91 +61,6 @@ encoding argument for the redis.Redis class.
 The encoding will be used to automatically encode any
 strings passed to commands, such as key names and values.
 
-
---------------------
-
-### MSET, MSETNX and ZADD
-
-These commands all accept a mapping of key/value pairs. In redis-py 2.X
-this mapping could be specified as **args* or as `**kwargs`. Both of
-these styles caused issues when Redis introduced optional flags to ZADD.
-Relying on `*args` caused issues with the optional argument order,
-especially in Python 2.7. Relying on `**kwargs` caused potential
-collision issues of user keys with the argument names in the method
-signature.
-
-To resolve this, redis-py 3.0 has changed these three commands to all
-accept a single positional argument named mapping that is expected to be
-a dict. For MSET and MSETNX, the dict is a mapping of key-names -\>
-values. For ZADD, the dict is a mapping of element-names -\> score.
-
-MSET, MSETNX and ZADD now look like:
-
-``` pycon
-def mset(self, mapping):
-def msetnx(self, mapping):
-def zadd(self, name, mapping, nx=False, xx=False, ch=False, incr=False):
-```
-
-All 2.X users that use these commands must modify their code to supply
-keys and values as a dict to these commands.
-
-### ZINCRBY
-
-redis-py 2.X accidentally modified the argument order of ZINCRBY,
-swapping the order of value and amount. ZINCRBY now looks like:
-
-``` python
-def zincrby(self, name, amount, value):
-```
-
-All 2.X users that rely on ZINCRBY must swap the order of amount and
-value for the command to continue to work as intended.
-
-### Encoding of User Input
-
-redis-py 3.0 only accepts user data as bytes, strings or numbers (ints,
-longs and floats). Attempting to specify a key or a value as any other
-type will raise a DataError exception.
-
-redis-py 2.X attempted to coerce any type of input into a string. While
-occasionally convenient, this caused all sorts of hidden errors when
-users passed boolean values (which were coerced to \'True\' or
-\'False\'), a None value (which was coerced to \'None\') or other
-values, such as user defined types.
-
-All 2.X users should make sure that the keys and values they pass into
-redis-py are either bytes, strings or numbers.
-
-### Locks
-
-redis-py 3.0 drops support for the pipeline-based Lock and now only
-supports the Lua-based lock. In doing so, LuaLock has been renamed to
-Lock. This also means that redis-py Lock objects require Redis server
-2.6 or greater.
-
-2.X users that were explicitly referring to *LuaLock* will have to now
-refer to *Lock* instead.
-
-### Locks as Context Managers
-
-redis-py 3.0 now raises a LockError when using a lock as a context
-manager and the lock cannot be acquired within the specified timeout.
-This is more of a bug fix than a backwards incompatible change. However,
-given an error is now raised where none was before, this might alarm
-some users.
-
-2.X users should make sure they're wrapping their lock code in a
-try/catch like this:
-
-``` python
-try:
-    with r.lock('my-lock-key', blocking_timeout=5) as lock:
-        # code you want executed only after the lock has been acquired
-except LockError:
-    # the lock wasn't acquired
-```
-
 ## API Reference
 
 The [official Redis command documentation](https://redis.io/commands)
@@ -1250,7 +1145,7 @@ Using scripting within pipelines in cluster mode is **not supported**.
 **RedisCluster specific options**
 
  require_full_coverage: (default=False)
- 
+
     When set to False (default value): the client will not require a
     full coverage of the slots. However, if not all slots are covered,
     and at least one node has 'cluster-require-full-coverage' set to
@@ -1260,14 +1155,14 @@ Using scripting within pipelines in cluster mode is **not supported**.
     When set to True: all slots must be covered to construct the
     cluster client. If not all slots are covered, RedisClusterException
     will be thrown.
-    
+
  read_from_replicas: (default=False)
 
      Enable read from replicas in READONLY mode. You can read possibly
      stale data.
      When set to true, read commands will be assigned between the
      primary and its replications in a Round-Robin manner.
-     
+
  dynamic_startup_nodes: (default=True)
 
      Set the RedisCluster's startup nodes to all of the discovered nodes.
@@ -1277,12 +1172,12 @@ Using scripting within pipelines in cluster mode is **not supported**.
      listed in the CLUSTER SLOTS output.
      If you use dynamic DNS endpoints for startup nodes but CLUSTER SLOTS lists
      specific IP addresses, it is best to set it to false.
-     
+
  cluster_error_retry_attempts: (default=3)
 
      Retry command execution attempts when encountering ClusterDownError
      or ConnectionError
-     
+
  reinitialize_steps: (default=10)
 
     Specifies the number of MOVED errors that need to occur before
diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
index 35536fc8..4c75d2f1 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -637,6 +637,8 @@ class Connection:
             retry_on_error = []
         if retry_on_timeout:
             retry_on_error.append(TimeoutError)
+            retry_on_error.append(socket.timeout)
+            retry_on_error.append(asyncio.TimeoutError)
         self.retry_on_error = retry_on_error
         if retry_on_error:
             if not retry:
@@ -685,7 +687,7 @@ class Connection:
 
     @property
     def is_connected(self):
-        return self._reader and self._writer
+        return self._reader is not None and self._writer is not None
 
     def register_connect_callback(self, callback):
         self._connect_callbacks.append(weakref.WeakMethod(callback))
@@ -706,7 +708,9 @@ class Connection:
         if self.is_connected:
             return
         try:
-            await self._connect()
+            await self.retry.call_with_retry(
+                lambda: self._connect(), lambda error: self.disconnect()
+            )
         except asyncio.CancelledError:
             raise
         except (socket.timeout, asyncio.TimeoutError):
@@ -767,7 +771,16 @@ class Connection:
     def _error_message(self, exception):
         # args for socket.error can either be (errno, "message")
         # or just "message"
-        if len(exception.args) == 1:
+        if not exception.args:
+            # asyncio has a bug where on Connection reset by peer, the
+            # exception is not instanciated, so args is empty. This is the
+            # workaround.
+            # See: https://github.com/redis/redis-py/issues/2237
+            # See: https://github.com/python/cpython/issues/94061
+            return (
+                f"Error connecting to {self.host}:{self.port}. Connection reset by peer"
+            )
+        elif len(exception.args) == 1:
             return f"Error connecting to {self.host}:{self.port}. {exception.args[0]}."
         else:
             return (
diff --git a/redis/asyncio/sentinel.py b/redis/asyncio/sentinel.py
index 5aefd09e..99c50749 100644
--- a/redis/asyncio/sentinel.py
+++ b/redis/asyncio/sentinel.py
@@ -44,7 +44,7 @@ class SentinelManagedConnection(Connection):
             if str_if_bytes(await self.read_response()) != "PONG":
                 raise ConnectionError("PING failed")
 
-    async def connect(self):
+    async def _connect_retry(self):
         if self._reader:
             return  # already connected
         if self.connection_pool.is_master:
@@ -57,6 +57,12 @@ class SentinelManagedConnection(Connection):
                     continue
             raise SlaveNotFoundError  # Never be here
 
+    async def connect(self):
+        return await self.retry.call_with_retry(
+            self._connect_retry,
+            lambda error: asyncio.sleep(0),
+        )
+
     async def read_response(self, disable_decoding: bool = False):
         try:
             return await super().read_response(disable_decoding=disable_decoding)


Test patch:
diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
index f6259adb..78a3efd2 100644
--- a/tests/test_asyncio/test_connection.py
+++ b/tests/test_asyncio/test_connection.py
@@ -1,10 +1,18 @@
 import asyncio
+import socket
 import types
+from unittest.mock import patch
 
 import pytest
 
-from redis.asyncio.connection import PythonParser, UnixDomainSocketConnection
-from redis.exceptions import InvalidResponse
+from redis.asyncio.connection import (
+    Connection,
+    PythonParser,
+    UnixDomainSocketConnection,
+)
+from redis.asyncio.retry import Retry
+from redis.backoff import NoBackoff
+from redis.exceptions import ConnectionError, InvalidResponse, TimeoutError
 from redis.utils import HIREDIS_AVAILABLE
 from tests.conftest import skip_if_server_version_lt
 
@@ -60,3 +68,44 @@ async def test_socket_param_regression(r):
 async def test_can_run_concurrent_commands(r):
     assert await r.ping() is True
     assert all(await asyncio.gather(*(r.ping() for _ in range(10))))
+
+
+async def test_connect_retry_on_timeout_error():
+    """Test that the _connect function is retried in case of a timeout"""
+    conn = Connection(retry_on_timeout=True, retry=Retry(NoBackoff(), 3))
+    origin_connect = conn._connect
+    conn._connect = mock.AsyncMock()
+
+    async def mock_connect():
+        # connect only on the last retry
+        if conn._connect.call_count <= 2:
+            raise socket.timeout
+        else:
+            return await origin_connect()
+
+    conn._connect.side_effect = mock_connect
+    await conn.connect()
+    assert conn._connect.call_count == 3
+
+
+async def test_connect_without_retry_on_os_error():
+    """Test that the _connect function is not being retried in case of a OSError"""
+    with patch.object(Connection, "_connect") as _connect:
+        _connect.side_effect = OSError("")
+        conn = Connection(retry_on_timeout=True, retry=Retry(NoBackoff(), 2))
+        with pytest.raises(ConnectionError):
+            await conn.connect()
+        assert _connect.call_count == 1
+
+
+async def test_connect_timeout_error_without_retry():
+    """Test that the _connect function is not being retried if retry_on_timeout is
+    set to False"""
+    conn = Connection(retry_on_timeout=False)
+    conn._connect = mock.AsyncMock()
+    conn._connect.side_effect = socket.timeout
+
+    with pytest.raises(TimeoutError) as e:
+        await conn.connect()
+    assert conn._connect.call_count == 1
+    assert str(e.value) == "Timeout connecting to server"
diff --git a/tests/test_asyncio/test_sentinel_managed_connection.py b/tests/test_asyncio/test_sentinel_managed_connection.py
new file mode 100644
index 00000000..a6e9f37a
--- /dev/null
+++ b/tests/test_asyncio/test_sentinel_managed_connection.py
@@ -0,0 +1,37 @@
+import socket
+
+import pytest
+
+from redis.asyncio.retry import Retry
+from redis.asyncio.sentinel import SentinelManagedConnection
+from redis.backoff import NoBackoff
+
+from .compat import mock
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_connect_retry_on_timeout_error():
+    """Test that the _connect function is retried in case of a timeout"""
+    connection_pool = mock.AsyncMock()
+    connection_pool.get_master_address = mock.AsyncMock(
+        return_value=("localhost", 6379)
+    )
+    conn = SentinelManagedConnection(
+        retry_on_timeout=True,
+        retry=Retry(NoBackoff(), 3),
+        connection_pool=connection_pool,
+    )
+    origin_connect = conn._connect
+    conn._connect = mock.AsyncMock()
+
+    async def mock_connect():
+        # connect only on the last retry
+        if conn._connect.call_count <= 2:
+            raise socket.timeout
+        else:
+            return await origin_connect()
+
+    conn._connect.side_effect = mock_connect
+    await conn.connect()
+    assert conn._connect.call_count == 3