Worker doesn't respond when redis is unstable

[ns-hwang]

Hi, we recently find our celery worker hangs when it tries to re-establish connection to redis server.

I saw some similar discussion, but the workaround there (disable mingle) doesn't resolve mine, so I guess they are different problem.

I tried to reproduce it locally with docker-compose.yml and debug with pyrasite. It seems it might be waiting indefinitely in kombu connection collect.
Wonder if it's possible that I have missed or set any configuration incorrectly ?

We use celery 5.3.6 + redis 5.0.4 + gevent 23.9.1
I also use prefork pool to test, and the problem still exists.

Error log

Traceback (most recent call last):
File "src/gevent/_waiter.py", line 122, in gevent._gevent_c_waiter.Waiter.switch
File "/opt/venv/lib/python3.8/site-packages/celery/worker/pidbox.py", line 118, in loop
  connection.drain_events(timeout=1.0)
File "/opt/venv/lib/python3.8/site-packages/kombu/connection.py", line 341, in drain_events
  return self.transport.drain_events(self.connection, **kwargs)
File "/opt/venv/lib/python3.8/site-packages/kombu/transport/virtual/base.py", line 997, in drain_events
  get(self._deliver, timeout=timeout)
File "/opt/venv/lib/python3.8/site-packages/kombu/transport/redis.py", line 588, in get
  ret = self.handle_event(fileno, event)
File "/opt/venv/lib/python3.8/site-packages/kombu/transport/redis.py", line 570, in handle_event
  return self.on_readable(fileno), self
File "/opt/venv/lib/python3.8/site-packages/kombu/transport/redis.py", line 566, in on_readable
  chan.handlers[type]()
File "/opt/venv/lib/python3.8/site-packages/kombu/transport/redis.py", line 910, in _receive
  ret.append(self._receive_one(c))
File "/opt/venv/lib/python3.8/site-packages/kombu/transport/redis.py", line 920, in _receive_one
  response = c.parse_response()
File "/opt/venv/lib/python3.8/site-packages/redis/client.py", line 1542, in parse_response
  response = self._execute(conn, try_read)
File "/opt/venv/lib/python3.8/site-packages/redis/client.py", line 1518, in _execute
  return conn.retry.call_with_retry(
File "/opt/venv/lib/python3.8/site-packages/redis/retry.py", line 49, in call_with_retry
  fail(error)
File "/opt/venv/lib/python3.8/site-packages/redis/client.py", line 1520, in <lambda>
  lambda error: self._disconnect_raise_connect(conn, error),
File "/opt/venv/lib/python3.8/site-packages/redis/client.py", line 1507, in _disconnect_raise_connect
  raise error
File "/opt/venv/lib/python3.8/site-packages/redis/retry.py", line 46, in call_with_retry
  return do()
File "/opt/venv/lib/python3.8/site-packages/redis/client.py", line 1519, in <lambda>
  lambda: command(*args, **kwargs),
File "/opt/venv/lib/python3.8/site-packages/redis/client.py", line 1540, in try_read
  return conn.read_response(disconnect_on_error=False)
File "/opt/venv/lib/python3.8/site-packages/redis/sentinel.py", line 61, in read_response
  return super().read_response(
File "/opt/venv/lib/python3.8/site-packages/redis/connection.py", line 882, in read_response
  response = self._parser.read_response(disable_decoding=disable_decoding)
File "/opt/venv/lib/python3.8/site-packages/redis/connection.py", line 349, in read_response
  result = self._read_response(disable_decoding=disable_decoding)
File "/opt/venv/lib/python3.8/site-packages/redis/connection.py", line 359, in _read_response
  raw = self._buffer.readline()
File "/opt/venv/lib/python3.8/site-packages/redis/connection.py", line 262, in readline
  self._read_from_socket()
File "/opt/venv/lib/python3.8/site-packages/redis/connection.py", line 215, in _read_from_socket
  raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
redis.exceptions.ConnectionError: Connection closed by server.
2023-09-17T09:05:51Z <built-in method switch of gevent._gevent_c_greenlet_primitives.TrackedRawGreenlet object at 0x7fbde2cdba40> failed with ConnectionError

The worker doesn't show any log or receive any task after this error message.

part of gevent dump info by pyrasite shell

********************************************************************************
* Greenlets
********************************************************************************
---- Thread boundary
<greenlet.greenlet object at 0x7f365c6e2c80 (otid=0x7f365c6d54e0) suspended active started main>
 :    Parent: None
 :    Running:
 :      File "/opt/venv/bin/celery", line 8, in <module>
 :        sys.exit(main())
 :      File "/opt/venv/lib/python3.8/site-packages/celery/__main__.py", line 15, in main
 :        sys.exit(_main())
 :      File "/opt/venv/lib/python3.8/site-packages/celery/bin/celery.py", line 236, in main
 :        return celery(auto_envvar_prefix="CELERY")
 :      File "/opt/venv/lib/python3.8/site-packages/click/core.py", line 1157, in __call__
 :        return self.main(*args, **kwargs)
 :      File "/opt/venv/lib/python3.8/site-packages/click/core.py", line 1078, in main
 :        rv = self.invoke(ctx)
 :      File "/opt/venv/lib/python3.8/site-packages/click/core.py", line 1688, in invoke
 :        return _process_result(sub_ctx.command.invoke(sub_ctx))
 :      File "/opt/venv/lib/python3.8/site-packages/click/core.py", line 1434, in invoke
 :        return ctx.invoke(self.callback, **ctx.params)
 :      File "/opt/venv/lib/python3.8/site-packages/click/core.py", line 783, in invoke
 :        return __callback(*args, **kwargs)
 :      File "/opt/venv/lib/python3.8/site-packages/click/decorators.py", line 33, in new_func
 :        return f(get_current_context(), *args, **kwargs)
 :      File "/opt/venv/lib/python3.8/site-packages/celery/bin/base.py", line 134, in caller
 :        return f(ctx, *args, **kwargs)
 :      File "/opt/venv/lib/python3.8/site-packages/celery/bin/worker.py", line 356, in worker
 :        worker.start()
 :      File "/opt/venv/lib/python3.8/site-packages/celery/worker/worker.py", line 202, in start
 :        self.blueprint.start(self)
 :      File "/opt/venv/lib/python3.8/site-packages/celery/bootsteps.py", line 116, in start
 :        step.start(parent)
 :      File "/opt/venv/lib/python3.8/site-packages/celery/bootsteps.py", line 365, in start
 :        return self.obj.start()
 :      File "/opt/venv/lib/python3.8/site-packages/celery/worker/consumer/consumer.py", line 361, in start
 :        self.on_connection_error_after_connected(exc)
 :      File "/opt/venv/lib/python3.8/site-packages/celery/worker/consumer/consumer.py", line 380, in on_connection_error_after_connected
 :        self.connection.collect()
 :      File "/opt/venv/lib/python3.8/site-packages/kombu/connection.py", line 384, in collect
 :        self._do_close_self()
 :      File "/opt/venv/lib/python3.8/site-packages/kombu/connection.py", line 354, in _do_close_self
 :        self.maybe_close_channel(self._default_channel)
 :      File "/opt/venv/lib/python3.8/site-packages/kombu/connection.py", line 346, in maybe_close_channel
 :        channel.close()
 :      File "/opt/venv/lib/python3.8/site-packages/kombu/transport/redis.py", line 1086, in close
 :        self._brpop_read()
 :      File "/opt/venv/lib/python3.8/site-packages/kombu/transport/redis.py", line 961, in _brpop_read
 :        dest__item = self.client.parse_response(self.client.connection,
 :      File "/opt/venv/lib/python3.8/site-packages/redis/client.py", line 1286, in parse_response
 :        response = connection.read_response()
 :      File "/opt/venv/lib/python3.8/site-packages/redis/connection.py", line 882, in read_response
 :        response = self._parser.read_response(disable_decoding=disable_decoding)
 :      File "/opt/venv/lib/python3.8/site-packages/redis/connection.py", line 349, in read_response
 :        result = self._read_response(disable_decoding=disable_decoding)
 :      File "/opt/venv/lib/python3.8/site-packages/redis/connection.py", line 359, in _read_response
 :        raw = self._buffer.readline()
 :      File "/opt/venv/lib/python3.8/site-packages/redis/connection.py", line 262, in readline
 :        self._read_from_socket()
 :      File "/opt/venv/lib/python3.8/site-packages/redis/connection.py", line 212, in _read_from_socket
 :        data = self._sock.recv(socket_read_size)
 :      File "/opt/venv/lib/python3.8/site-packages/gevent/_socketcommon.py", line 666, in recv
 :        self._wait(self._read_event)