[ishan-1010]

Fixes #2841.

With retry=True, Producer._publish declares the queue through a nested ensure (added in #2174). When that reconnects, the producer's channel gets closed but _publish keeps publishing on the one it read before the declare, which raises an AttributeError on Redis and never retries. This revives the producer onto the connection's live channel after the declare and prepares the message again.

Tested with the injection from the issue on a real Redis (one ConnectionError in _queue_bind). On main the publish raises and the queue stays empty, with this change the queue length is 1. The new unit test uses the memory transport and fails without the change.

@auvipy

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 85.78%. Comparing base (445027e) to head (67bdbaf).
⚠️ Report is 1 commits behind head on main.

Additional details and impacted files

@@           Coverage Diff           @@
##             main    #2844   +/-   ##
=======================================
  Coverage   85.77%   85.78%           
=======================================
  Files          80       80           
  Lines       11595    11599    +4     
  Branches     1403     1404    +1     
=======================================
+ Hits         9946     9950    +4     
  Misses       1428     1428           
  Partials      221      221           

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

[MehrazRumman]

Read 67bdbaf against #2841. The cause is as described. _imaybe_declare() runs the declare through entity.channel.connection.client.ensure(...). On a recoverable error, Connection.ensure() calls collect(), which closes every channel on the connection, including the producer's. Then it revives only the entity, onto the new default_channel. _publish() then publishes on the channel it read before the declare. Checking channel.connection after the declare and reviving the producer fixes this for producers created from a Connection.

Preparing the message again is safe. Nothing between the two prepare_message() calls changes headers or properties, and on the virtual transports prepare_message() only uses setdefault() on properties, so calling it twice gives the same message.

All 37 checks pass on 67bdbaf.

One case is still left, inline.

[MehrazRumman]

This still fails for a producer created from a channel, such as Producer(conn.channel()). Its __connection__ is None, so Producer.connection falls back to self.channel.connection.client. Here the channel is already closed, so that raises AttributeError, the property catches it and returns None, and this line raises AttributeError: 'NoneType' object has no attribute 'default_channel'. That's still not a recoverable error, so the outer ensure gives up as before.

conn.Producer() and kombu's ProducerPool, which Celery uses, pass the Connection, so they're covered. Code that passes a channel isn't.

Reading the connection before the declare, while the channel is still open, would cover both:

        channel = self.channel
        connection = self.connection
        ...
            if channel.connection is None:
                # A reconnect during the declare closed our channel.
                self.revive(connection.default_channel)

A second case in the test, with Producer(conn.channel()), would show it. As written now it should fail with the NoneType error above.