Bug in retry-handling in the PubSub-client

[sondrfos]

Version:
redis-py: 5.0.3
redis: 7.2.4

Platform:
Python 3.12.2 running on Debian bookworm

Description:
First off I want to say thanks for maintaining this library :)

So to the issue at hand:
It seems like the retry-functionality in the PubSub-client (both async and sync) is not working correctly.
To simulate a crash of redis I am closing and restarting redis while our system is running. Even though our connection is set up with an infinite amount of retry, the PubSub-client always crashes and never recovers.

Having dug though the source code I have found the pathways causing the crash.

redis-py/redis/asyncio/client.py

Lines 885 to 896 in 1784b37

	async def _execute(self, conn, command, *args, **kwargs):
	"""
	Connect manually upon disconnection. If the Redis server is down,
	this will fail and raise a ConnectionError as desired.
	After reconnection, the ``on_connect`` callback should have been
	called by the # connection to resubscribe us to any channels and
	patterns we were previously listening to
	"""
	return await conn.retry.call_with_retry(
	lambda: command(*args, **kwargs),
	lambda error: self._disconnect_raise_connect(conn, error),
	)

The _execute-function runs periodicially, and whenever redis is closed the command fails.
This causes the _disconnect_raise_connect-function to be called:

redis-py/redis/asyncio/client.py

Lines 870 to 883 in 1784b37

	async def _disconnect_raise_connect(self, conn, error):
	"""
	Close the connection and raise an exception
	if retry_on_error is not set or the error is not one
	of the specified error types. Otherwise, try to
	reconnect
	"""
	await conn.disconnect()
	if (
	conn.retry_on_error is None
	or isinstance(error, tuple(conn.retry_on_error)) is False
	):
	raise error
	await conn.connect()

_disconnect_raise_connect closes the connection, conn.retry_on_error is True making the function skip raising error.
This is followed by an attempt at reconnecting.

redis-py/redis/asyncio/connection.py

Lines 269 to 284 in 1784b37

	async def connect(self):
	"""Connects to the Redis server if not already connected"""
	if self.is_connected:
	return
	try:
	await self.retry.call_with_retry(
	lambda: self._connect(), lambda error: self.disconnect()
	)
	except asyncio.CancelledError:
	raise # in 3.7 and earlier, this is an Exception, not BaseException
	except (socket.timeout, asyncio.TimeoutError):
	raise TimeoutError("Timeout connecting to server")
	except OSError as e:
	raise ConnectionError(self._error_message(e))
	except Exception as exc:
	raise ConnectionError(exc) from exc

Since we are not currently connected, the first if-statement is False. A call with retry is made to the _connect-function.

redis-py/redis/asyncio/connection.py

Lines 678 to 700 in 1784b37

	async def _connect(self):
	"""Create a TCP socket connection"""
	async with async_timeout(self.socket_connect_timeout):
	reader, writer = await asyncio.open_connection(
	**self._connection_arguments()
	)
	self._reader = reader
	self._writer = writer
	sock = writer.transport.get_extra_info("socket")
	if sock:
	sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
	try:
	# TCP_KEEPALIVE
	if self.socket_keepalive:
	sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
	for k, v in self.socket_keepalive_options.items():
	sock.setsockopt(socket.SOL_TCP, k, v)
	except (OSError, TypeError):
	# `socket_keepalive_options` might contain invalid options
	# causing an error. Do not leave the connection open.
	writer.close()
	raise

Inside the _connect-function an attempt is made to connect to redis. Since redis is unavailable an OSError is created. OSError is not in the list of supported_errors causing it to bypass the error handling in the call_with_retry-function and is propagated to the connect-function. Here it is converted to a ConnectionError and is propagated all the way to the top unhandled.

Is this intended behaviour?
Is this something that could be handled in a different way?

We have temporarily fixed this by adding OSError to the list of supported_errors, but this feels like quite a hacky solution.

[ManelCoutinhoSensei]

Hey,

Based on your description, I believe I've encountered the same issue with the synchronous version (as you said).

Minimal Reproducible Example

The following pseudo-code snippet helps to reproduce the issue:

def event_handler(msg):
    print(msg)

redis_pub_sub.psubscribe(**{"test": event_handler})

def exception_handler(ex, pubsub, thread):
    print(traceback.format_exc())

redis_pub_sub.run_in_thread(sleep_time=0.1, exception_handler=exception_handler)

# Close Redis connection

Error Trace

Traceback (most recent call last):
  File "/path/to/env/lib/python3.10/site-packages/redis/retry.py", line 62, in call_with_retry
    return do()
  File "/path/to/env/lib/python3.10/site-packages/redis/client.py", line 842, in <lambda>
    lambda: command(*args, **kwargs),
  File "/path/to/env/lib/python3.10/site-packages/redis/client.py", line 859, in try_read
    if not conn.can_read(timeout=timeout):
  File "/path/to/env/lib/python3.10/site-packages/redis/connection.py", line 600, in can_read
    return self._parser.can_read(timeout)
  File "/path/to/env/lib/python3.10/site-packages/redis/_parsers/hiredis.py", line 79, in can_read
    return self.read_from_socket(timeout=timeout, raise_on_timeout=False)
  File "/path/to/env/lib/python3.10/site-packages/redis/_parsers/hiredis.py", line 90, in read_from_socket
    raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
redis.exceptions.ConnectionError: Connection closed by server.

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/path/to/env/lib/python3.10/site-packages/redis/connection.py", line 622, in read_response
    response = self._parser.read_response(disable_decoding=disable_decoding)
  File "/path/to/env/lib/python3.10/site-packages/redis/_parsers/hiredis.py", line 128, in read_response
    self.read_from_socket()
  File "/path/to/env/lib/python3.10/site-packages/redis/_parsers/hiredis.py", line 88, in read_from_socket
    bufflen = self._sock.recv_into(self._buffer)
ConnectionResetError: [Errno 104] Connection reset by peer

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/path/to/env/lib/python3.10/site-packages/redis/client.py", line 1192, in run
    pubsub.get_message(ignore_subscribe_messages=True, timeout=sleep_time)
  File "/path/to/env/lib/python3.10/site-packages/redis/client.py", line 1054, in get_message
    response = self.parse_response(block=(timeout is None), timeout=timeout)
  File "/path/to/env/lib/python3.10/site-packages/redis/client.py", line 865, in parse_response
    response = self._execute(conn, try_read)
  File "/path/to/env/lib/python3.10/site-packages/redis/client.py", line 841, in _execute
    return conn.retry.call_with_retry(
  File "/path/to/env/lib/python3.10/site-packages/redis/retry.py", line 65, in call_with_retry
    fail(error)
  File "/path/to/env/lib/python3.10/site-packages/redis/client.py", line 843, in <lambda>
    lambda error: self._disconnect_raise_connect(conn, error),
  File "/path/to/env/lib/python3.10/site-packages/redis/client.py", line 831, in _disconnect_raise_connect
    conn.connect()
  File "/path/to/env/lib/python3.10/site-packages/redis/connection.py", line 370, in connect
    self.on_connect()
  File "/path/to/env/lib/python3.10/site-packages/redis/connection.py", line 501, in on_connect
    self.read_response()
  File "/path/to/env/lib/python3.10/site-packages/redis/connection.py", line 630, in read_response
    raise ConnectionError(
redis.exceptions.ConnectionError: Error while reading from localhost:6379 : (104, 'Connection reset by peer')

Possible Cause

It seems that the issue in this version is more related to the connect method since the second part of the error trace occurs outside the retry scope.

Quick Patch

Temporarily adding OSError to the list of supported_errors doesn't solve the problem, but a quick workaround is to wrap everything inside a retry call, as shown below. However, this might introduce unexpected side effects.

def connect(self):
    """Connects to the Redis server if not already connected."""
    if self._sock:
        return

    def _full_connect():
        self._sock = self._connect()
        if self.redis_connect_func is None:
            # Use the default on_connect function
            self.on_connect()
        else:
            # Use the passed function redis_connect_func
            self.redis_connect_func(self)

        # Run any user callbacks. Currently, the only internal callback
        # is for pubsub channel/pattern resubscription.
        # First, remove any dead weak references
        self._connect_callbacks = [ref for ref in self._connect_callbacks if ref()]
        for ref in self._connect_callbacks:
            callback = ref()
            if callback:
                callback(self)

    try:
        sock = self.retry.call_with_retry(
            lambda: _full_connect(),
            lambda error: self.disconnect(error)
        )
    except socket.timeout:
        raise TimeoutError("Timeout connecting to server")
    except OSError as e:
        raise ConnectionError(self._error_message(e))

I see that this issue is already assigned to you, @gerzse. Do you have any updates on its progress?

[d33m00n]

Having the same issue with pipelines. Solved using this hack:

    async def redis_connect_func_patch(self: AbstractConnection):
        async def _noop(_):
            pass

        await self.retry.call_with_retry(
            do=lambda: self.on_connect_check_health(check_health=True),
            fail=_noop,
        )

    redis = Redis(
        protocol=3,
        retry_on_error=[
            asyncio.TimeoutError,
            ConnectionError,
            exceptions.ConnectionError,
            exceptions.TimeoutError,
            TimeoutError,
        ],
        retry=RedisRetry(backoff=ExponentialBackoff(cap=10, base=0.1), retries=100),
        redis_connect_func=redis_connect_func_patch, # the hack
    )

[ManelCoutinhoSensei]

Btw @petyaslavova , I have a better explanation of the problem here