Broken connections are reused in connection pools

[evgenymarkov]

Describe the bug

If an application has a connection break with the network, it accumulates broken connections with the Redis server and tries to reuse them to execute commands. This is due to the fact that the connect() methods do not check whether an exception is set in the StreamReaders:

class Connection:
    async def connect(self):
        """Connects to the Redis server if not already connected"""
        if self.is_connected:   # <-- self._reader may contain an exception (e.g. OSError No route to host).
            return

class SentinelManagedConnection:
    async def connect(self):
        if self._reader:  # <-- self._reader may contain an exception (e.g. OSError No route to host).
            return  # already connected

This cases should be handled.

To Reproduce

1. Initialize base Redis client or Sentinel client, connect to Redis server
2. Send pings periodically
3. Turn off wi-fi
4. Look at ping errors
5. Turn on wi-fi
6. Look at ping errors

Expected behavior

There should be no errors in step 6

Logs/tracebacks

https://gist.github.com/evgenymarkov/1af3a5cb12887068cb9c6a72c09168ea

Python Version

3.9.10

aioredis Version

2.0.1

Additional context

No response

Code of Conduct

• I agree to follow the aio-libs Code of Conduct

[evgenymarkov]

#1313

[evgenymarkov]

My issue can be related with #1174