[aurangzaib048]

Summary

Fixes celery/celery#11054.

A celery 5.7 worker on the Redis broker exits with AttributeError: 'NoneType' object has no attribute 'hostname' when Redis restarts while a prefetched task is waiting in the worker. The Redis Channel.close() could stop part-way when the server was gone, and the channel then stayed open until a later close, which ran after Connection.collect() had detached the transport it needs.

Root cause

Before it closes the channel, Channel.close() drains a BRPOP that is in flight and deletes the auto-delete fanout bindings. Both need the server. A connection error from either propagated out of close() before super().close(), so the channel stayed open (closed is False) with its unacked deliveries.

Connection.collect() swallows that error, in maybe_close_channel() and around close_connection(), and then _do_close_transport() sets transport.client = None. The channel is still the connection's _default_channel, so the next Connection.close() closes it again. If the server is back by then, that close gets as far as restore_unacked_once(), which builds a connection pool from transport.client and raises AttributeError. That is neither a connection nor a channel error, so nothing ignores it.

Since celery 5.7 (celery/celery#10242) the worker calls collect() and then close() after a connection error, so a Redis restart with a prefetched task waiting killed the worker. In a run with a logging wrapper around maybe_close_channel(), the first close inside collect() failed in the BRPOP drain, and the second one, from close_connection(), in the fanout queue_delete() the reporter found.

Changes

• kombu/transport/redis.py: Channel.close() skips the drain and the fanout binding cleanup on a connection error, then disconnects its pools and clients and closes the channel as before. Unacked deliveries are restored at that point when the server is reachable; when it is not, they stay in the unacked hash until visibility_timeout, as they did before.
• t/unit/transport/test_redis.py: a close whose drain fails still closes the channel, and the 5.7 regression: worker dies with AttributeError ('NoneType' object has no attribute 'hostname') when the Redis broker restarts while a prefetched task waits celery#11054 sequence (fanout cleanup fails while the server is down during collect(), then close() once it is back) restores unacked messages only while the transport is attached.

Test plan

• 2 new unit tests fail on main (the drain error escapes close(); unacked messages are restored through a transport whose client is None) and pass with the fix
• Reproduced on Linux with the scenario from the issue: celery main (1a48583d4) prefork -c 2, three 8 s tasks so one waits prefetched, docker restart -t 0 of a redis:7.2 --appendonly yes container in the worker's network namespace, then a ping task; redis-py 6.4.0
  • kombu main: the worker exits with the AttributeError on the first restart
  • this branch: 10 of 10 restarts, every ping task consumed, the worker stays up
  • idle worker, 15 restarts: both pass, but main logs Error on restarting Gossip: AttributeError(...) on all 15 (the Gossip channel's fanout cleanup fails the same way) and this branch on none
• The ack path from 5.7 regression: worker dies with AttributeError ('NoneType' object has no attribute 'hostname') when the Redis broker restarts while a prefetched task waits celery#11054 (acks_late, -c 2, three 6 s tasks, Redis restarted 2 s in), 8 runs each: on main the worker exits in all 8 when a task that was running acks through the closed channel (_connparams() on a None transport.client); on this branch none exit and both finished tasks are removed from unacked. The close inside collect() now completes while the transport is attached, and its restore attempt leaves a pool the late ack uses. With CLIENT KILL instead of a restart both survive, with the same 2 redeliveries per run from the restore at close
• Unit suite: Linux, tox 3.14-unit and 3.12-unit as an unprivileged user, 2342 passed each; macOS (3.12) 2228 passed
• Integration in a Linux container with each broker on localhost like the CI services (tox 3.14-linux-integration-*, CI's pytest command): py-amqp 51, redis 129, mongodb 26, kafka 38, sqs 42, pgmq 29 passed
• flake8, pydocstyle, mypy, apicheck