Redis: deliver rotating StreamingCredentialProvider tokens to long-lived BRPOP and pub/sub connections

[ThomasAitken]

Fixes #2509. Builds on #2507.

Summary

When a redis.credentials.StreamingCredentialProvider (e.g. the Microsoft Entra ID
provider from redis-entraid, or an AWS ElastiCache IAM provider) is used with the
Redis transport, the rotated re-auth tokens it emits never reach the two connections
the transport holds for the entire lifetime of the worker: the BRPOP connection and
the pub/sub LISTEN connection. As a result the broker eventually severs those
connections when the original credentials expire (AWS ElastiCache with IAM auth enforces
a hard 12-hour connection limit), causing unacked messages to be redelivered, in-flight
tasks to be interrupted, and brief worker unavailability during the forced reconnect.

This PR makes the transport deliver those tokens itself, at safe points on the event loop.

Root cause

redis-py's streaming re-auth (RegisterReAuthForPooledConnections) handles in-use
pooled connections by calling Connection.set_re_auth_token(token), which only stores
the token; the actual AUTH command is deferred until the connection is released back to
the pool (AfterConnectionReleasedEvent → ReAuthConnectionListener → Connection.re_auth()).

The transport, however, pins two connections that are never released:

1. the BRPOP connection, assigned to client.connection in
   MultiChannelPoller._client_registered, and
2. the pub/sub LISTEN connection, opened when subclient.subscribe() runs.

Both live in ConnectionPool._in_use_connections for the worker's lifetime, so
re_auth() is never triggered and the stored token is never flushed. #2507 correctly
made sure the re-auth listener gets registered (by passing credential_provider through
to redis.Redis(...)); this PR makes sure the stored tokens actually get applied.

What this PR does

A periodic, event-loop-driven flush of any pending token onto the two long-lived
connections. Because it runs on the single event-loop thread, it can never race the
socket reads:

• BRPOP connection — re-authenticated in place with Connection.re_auth(), but only
  when the connection is live and no blocking pop is in flight (not _in_poll), so the
  AUTH reply can't interleave with the pop's reply. It is flushed both from the periodic
  timer and from _brpop_read's finally block (the guaranteed-idle moment right
  after a reply is consumed), so rotated tokens are applied promptly even under heavy
  consumption. Crucially, the flush is skipped when the socket is already down: sending
  AUTH on a disconnected connection would transparently reconnect a fresh socket
  behind the poller's back (which _register_BRPOP would then never re-register, silently
  stalling the channel). Instead the socket is left down and _register_BRPOP reconnects
  and re-registers it, with on_connect re-authenticating via the current
  credentials. On an AUTH failure the socket is dropped and the stale token cleared, so
  the reconnect's fresh on_connect auth stands and the same (possibly expired) token
  isn't re-sent in place on every tick.
• Pub/sub LISTEN connection — a subscribed RESP2 connection cannot process AUTH, so
  it is transparently reconnected (and re-subscribed) via the existing poller machinery,
  which re-authenticates on on_connect with the current credentials. Fanout delivery is
  best-effort, so the sub-second reconnect is far less disruptive than a broker-forced
  disconnect. Under RESP3, redis-py re-authenticates pub/sub connections itself
  (RegisterReAuthForPubSub), so the transport leaves those untouched to avoid
  double-AUTH and background-thread races.

Wiring:

• MultiChannelPoller.maybe_reauth() iterates channels and swallows connection_errors
  (like the existing maybe_restore_messages / maybe_check_subclient_health timers, so a
  transient error can't tear down the event loop).
• A new _reauth_tref timer is registered in register_with_event_loop, with the same
  stale-timer cancellation as the existing timers.
• New transport option reauth_check_interval (seconds, default 10) controls the cadence.

Design notes

• Decoupled from the credential provider. Detection keys off the connection's stored
  _re_auth_token (the attribute redis-py itself populates for in-use pooled connections),
  so there is no coupling to any specific provider and no new import surface.
• No eager reconnects. Both long-lived connections are only ever re-authenticated in
  place on a live socket; whenever a socket must be dropped, reconnection and
  re-registration are left to the existing _register_BRPOP / _register_LISTEN paths, so
  the poller's bookkeeping (_chan_to_sock / _fd_to_chan) stays consistent.
• No-op unless streaming re-auth is active. When no StreamingCredentialProvider is
  configured the token is never set, so every check is a cheap early return. Behaviour for
  existing users is unchanged.
• Backwards compatible. No signature changes; the new transport option is optional with
  a sensible default.

Tests

28 new unit tests in t/unit/transport/test_redis.py covering:

• the token helper (including a test against a real redis.Connection to guard against
  redis-py renaming the stored-token attribute),
• RESP2 vs RESP3 protocol gating (including real connections),
• the BRPOP flush across live-idle / in-flight / socket-down / no-token / no-client /
  ConnectionError / ResponseError / disconnect-error paths, and that a rejected/failed
  AUTH both drops the socket and clears the stale token,
• an end-to-end regression test proving a BRPOP connection error does not eager-reconnect
  an unregistered socket (the socket is left down for _register_BRPOP),
• the pub/sub reconnect (RESP2 reconnect, RESP3 left to redis-py, no-token, no-subclient,
  missing-method tolerance),
• the _brpop_read finally hook, and
• the poller-level delegation and error swallowing.

The 10 existing register_with_event_loop tests were updated for the new timer.

Docs

• New "Streaming credentials / automatic re-authentication" section in
  docs/reference/kombu.transport.redis.rst.
• reauth_check_interval documented in the transport module docstring.

I tagged the docs versionadded:: 5.7.0 — happy to adjust to whatever the next release will be.

🤖 Heavily assisted by Claude Code

[Copilot]

Pull request overview

This PR enhances Kombu’s Redis transport to properly apply rotated authentication tokens from redis.credentials.StreamingCredentialProvider to the transport’s two long-lived connections (the BRPOP consumer connection and the pub/sub LISTEN connection), preventing broker-forced disconnects when credentials expire.

Changes:

• Adds periodic, event-loop-driven re-authentication checks (maybe_reauth) and a new reauth_check_interval transport option (default: 10s).
• Flushes pending re-auth tokens onto the BRPOP connection at safe idle points (including _brpop_read’s finally block), and refreshes the RESP2 pub/sub connection by reconnecting when needed.
• Adds extensive unit tests and new documentation describing streaming credential support and configuration.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.

File	Description
kombu/transport/redis.py	Implements token flushing/reconnect logic and registers a new event-loop timer controlled by reauth_check_interval.
t/unit/transport/test_redis.py	Adds unit tests covering token detection, protocol gating (RESP2/RESP3), BRPOP/LISTEN behavior, and timer registration updates.
docs/reference/kombu.transport.redis.rst	Documents streaming credential behavior and the new reauth_check_interval option.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[codecov]

Codecov Report

❌ Patch coverage is 96.77419% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.89%. Comparing base (9537a96) to head (024dc56).
⚠️ Report is 3 commits behind head on main.
✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
kombu/transport/redis.py	96.77%	2 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2563      +/-   ##
==========================================
+ Coverage   82.81%   82.89%   +0.08%     
==========================================
  Files          79       79              
  Lines       10333    10394      +61     
  Branches     1187     1195       +8     
==========================================
+ Hits         8557     8616      +59     
- Misses       1575     1577       +2     
  Partials      201      201              

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

[Copilot]

Pull request overview

Copilot reviewed 3 out of 3 changed files in this pull request and generated no new comments.

Suppressed comments (3)

docs/reference/kombu.transport.redis.rst:84

• This second snippet repeats the unsupported placement of credential_provider in transport options, so copying the complete example still does not activate streaming credentials. Keep only reauth_check_interval in broker_transport_options and reference the corrected credential-provider setup above.

        app.conf.broker_transport_options = {
            "credential_provider": my_streaming_credential_provider,
            "reauth_check_interval": 10,
        }

docs/reference/kombu.transport.redis.rst:51

• This example does not configure the credential provider. broker_transport_options becomes Connection.transport_options, while the Redis channel obtains credentials from Connection.client.credential_provider (see kombu/connection.py:189-198 and kombu/transport/redis.py:1422-1440); credential_provider is also absent from Channel.from_transport_options. Consequently, users following this example get passwordless/default Redis connections and the documented re-authentication never activates. Show a supported route such as passing credential_provider= to kombu.Connection, or a broker URL credential_provider query parameter for an importable provider.

This issue also appears on line 81 of the same file.

        app.conf.broker_transport_options = {
            "credential_provider": my_streaming_credential_provider,
        }

kombu/transport/redis.py:1727

• The newly documented configurable interval is not exercised: all updated registration tests use empty transport options and assert only the default 10. Add a test supplying a non-default reauth_check_interval and asserting that value is passed to call_repeatedly, analogous to test_configurable_health_check, so a typo or ignored option cannot regress unnoticed.

        reauth_check_interval = connection.client.transport_options.get(
            'reauth_check_interval',
            DEFAULT_REAUTH_CHECK_INTERVAL
        )
        cycle._reauth_tref = loop.call_repeatedly(
            reauth_check_interval,
            cycle.maybe_reauth