[ericdelorefice]

The Redis transport moves an unacked message back to its queue in one transaction, since #1444 made the restore atomic for #1405 so a message is never left half-moved. This keeps that transaction from committing when the queue lookup inside it fails: today a connection error there deletes the message instead of restoring it.

Problem

With ack emulation (the default), QoS.restore_by_tag and Channel._restore run restore_transaction through client.transaction(). It queues HDEL (and ZREM) on the unacked keys and then calls _do_restore_message, which queues RPUSH/LPUSH onto every queue _lookup(exchange, routing_key) returns. For a named exchange, _lookup reads the bindings with get_table, an SMEMBERS on another pooled connection, made while the transaction is open.

_do_restore_message wraps all of this in except Exception, logs "Could not restore message" at CRITICAL and returns. When that SMEMBERS times out or its connection drops, restore_transaction returns normally, redis-py sends EXEC with only the removal queued, and the message is gone from Redis. Kombu's Redis clients make no retries by default (the connection's retry has 0 retries here), so one failed SMEMBERS is enough. #1784 shows this lookup raising inside _do_restore_message in production (there, under a signal-based timeout); it does not report a lost message.

All three restore paths go through it:

• message.requeue() (QoS.reject(requeue=True)) returns normally and the message is gone.
• the visibility-timeout sweep (restore_visible) deletes it.
• the restore when a channel closes with messages unacked deletes it.

It needs a named exchange. Celery's default direct routing publishes through the default exchange (''), and _lookup answers that without a lookup, so default Celery routing does not reach it; kombu producers that publish to a named exchange, and Celery routes that set an exchange and routing key, do.

Reproduction

Real kombu from main (c35fe8d), redis-py 8.1.0 and a local Redis 8.10.2 server. The failure is injected: redis.Redis.smembers raises redis.exceptions.TimeoutError("Timeout reading from socket") once, for the binding lookup only. I have not seen the message loss in a user report. The transaction and everything else run against the server.

One message published to a named direct exchange, taken with get(no_ack=False), then restored with the lookup failing once:

On main:

== message.requeue()
  log CRITICAL kombu.transport.redis: Could not restore message: {'body': 'eyJqb2IiOiAxfQ==', ...
  call returned normally
  after the call: queue 'work' holds 0, unacked hash holds 0, unacked index holds 0
== visibility sweep (maybe_restore_messages -> restore_visible)
  after the call: queue 'work' holds 0, unacked hash holds 0, unacked index holds 0
== channel.close() with the message unacked
  after the call: queue 'work' holds 0, unacked hash holds 0, unacked index holds 0

With this change:

== message.requeue()
  call raised redis.exceptions.TimeoutError: Timeout reading from socket
  after the call: queue 'work' holds 0, unacked hash holds 1, unacked index holds 1
  after a later healthy restore_visible sweep: queue 'work' holds 1, unacked hash holds 0, unacked index holds 0
== visibility sweep (maybe_restore_messages -> restore_visible)
  call returned normally
  after the call: queue 'work' holds 0, unacked hash holds 1, unacked index holds 1
  after a later healthy restore_visible sweep: queue 'work' holds 1, unacked hash holds 0, unacked index holds 0
== channel.close() with the message unacked
  call raised redis.exceptions.TimeoutError: Timeout reading from socket
  after the call: queue 'work' holds 0, unacked hash holds 1, unacked index holds 1
  after a later healthy restore_visible sweep: queue 'work' holds 1, unacked hash holds 0, unacked index holds 0

Without the injected failure, all three paths requeue the message on both main and this branch.

Change

_do_restore_message re-raises the channel's connection_errors (4 added lines), so restore_transaction raises, redis-py discards the queued commands, and the message stays in the unacked keys for a later restore. The sweep path already expects this: MultiChannelPoller.maybe_restore_messages catches connection_errors from restore_visible and retries on the next cycle (#2498). Any other exception is still logged with the payload and the message dropped, as before, so a payload that can never be restored does not come back on every sweep.

Intended behaviour changes, on a connection error during the restore only:

• message.requeue() raises the connection error instead of returning. The message is still unacked, so the requeue can be retried, or the visibility-timeout sweep restores it.
• Channel.close() raises it from the restore at close, as it already does when the restore's own transaction cannot reach Redis. The message, and any other message that close had not yet restored, stays unacked until the visibility-timeout sweep requeues it.

In the unit tests, the test Transport uses KeyError as its connection error, so test_do_restore_message now makes rpush raise ValueError to keep testing the log-and-drop path it was written for.

Testing

macOS, Python 3.12.14, redis-py 8.1.0, plain pytest.

• t/unit/transport/test_redis.py: 3 new tests fail on main and pass with this change: _do_restore_message re-raises a connection error from the lookup without logging it or queueing a push, and Channel._restore and QoS.restore_by_tag raise it and leave the stored unacked entry (and index) unchanged, through a transaction stand-in that only applies the queued commands when the callable returns, as redis-py does. 2 new controls pass both ways: a non-connection error is still logged with the payload, and a successful lookup requeues the message with redelivered set.
• t/integration/test_redis.py, new test_RedisRestoreBindingLookupFailure, against Redis 8.10.2 with and without global_keyprefix: requeue() and restore_visible with the lookup failing raise the timeout, leave the unacked entry unchanged and the queue empty, and a later healthy requeue or sweep delivers the message exactly once (4 runs, fail on main); a plain requeue control (2 runs) passes both ways.
• With the source reverted to main and the new tests kept, the 7 fail, and they pass with the change, ten runs each way; the controls pass both ways.
• t/unit: 1512 passed on main, 1517 on this branch (190 skipped). Six modules were not collected here because their optional dependencies (botocore, azure, google) are not installed.
• t/integration/test_redis.py against a local Redis 8.10.2 started without Docker (not the tox Docker environment): 111 passed on main, 117 on this branch. The same 6 health-check tests (test_RedisSubclientHealthCheck::test_missed_pongs_drop_connection_and_resubscribe and two test_RedisBrpopHealthCheck half-open tests, each with both connection params) fail on this machine on main and on this branch alike.
• pre-commit run --files on the three changed files: all hooks that apply passed (pyupgrade, autoflake, flake8, yesqa, codespell, check-merge-conflict, mixed-line-ending, isort, mypy). pydocstyle kombu/transport/redis.py (the tox pydocstyle env) reports nothing.

This PR was written with AI assistance (Claude Code); the tests above were run before opening it.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 85.50%. Comparing base (c35fe8d) to head (b95d479).
⚠️ Report is 1 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2767      +/-   ##
==========================================
+ Coverage   85.47%   85.50%   +0.02%     
==========================================
  Files          80       80              
  Lines       11460    11462       +2     
  Branches     1374     1374              
==========================================
+ Hits         9796     9801       +5     
+ Misses       1442     1440       -2     
+ Partials      222      221       -1     

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

[MehrazRumman]

Checked at 4c08e75. t/unit/transport/test_redis.py passes (288 passed). The three new failure-path tests fail against main's redis.py, and the controls pass both ways. CI is green, including the Redis integration jobs that run the new test_RedisRestoreBindingLookupFailure.

The change looks right

• Re-raising connection_errors from _do_restore_message makes restore_transaction raise, so redis-py's transaction() never sends EXEC. The HDEL/ZREM queued with the push are discarded, and the message stays in the unacked hash and index. Without it, a failed SMEMBERS lets EXEC commit the removal with no push, which is the loss described.
• Other exceptions still go to the existing log-and-drop path, so a payload that can never be routed doesn't come back on every sweep. That is the right split.
• How each caller handles the new exception:
  • restore_visible: MultiChannelPoller.maybe_restore_messages already catches connection_errors and retries next cycle (#2498). The Mutex is released in its finally, and the remaining due tags of that batch wait for the next sweep, which is fine.
  • QoS.reject(requeue=True): when restore_by_tag raises, super().ack(tag) is skipped, so the tag stays in _delivered and a later close or sweep can still restore it. Celery calls this through Message.reject_log_error, which catches the transport's connection errors and logs them, so a worker doesn't crash on this path.

Suggestions

1. message.requeue() / reject(requeue=True) and Channel.close() can now raise a Redis connection error where they used to return. That is the correct trade (an exception rather than a silently lost message), but it is visible to direct kombu users who call requeue() without a try. A changelog line saying so would help.
2. For Channel.close(), the description notes that messages after the failing one stay unacked until the visibility timeout. That is worth one sentence in the same changelog line, since recovery then waits up to visibility_timeout (1 h by default) instead of happening at close.

[ericdelorefice]

Thanks @MehrazRumman for checking it against main and for walking through each caller, especially the reject_log_error path in Celery.

Agreed on the changelog note. Since the changelog here is written at release time, I left it out of the PR, but I'm happy to add a line if @auvipy prefers it in the PR. Suggested wording:

• Redis: a connection error while restoring an unacked message now raises from Message.requeue() and Channel.close() instead of deleting the message. The message stays unacked, and messages a failed close did not restore are requeued by the visibility-timeout sweep.

[auvipy]

no changelog entry is needed in the PR. only doc update is OK only if it is needed.