feat(transport): support credential_provider on background Redis clients

[a-vorobiev]

Description

RedisTransport only passed credential_provider to the broker connection, not to its long-lived background clients (the PEL reclaimer, the group monitor, and the delete-on-ack client). Those clients are built from a whitelisted subset of connection kwargs (_BACKGROUND_CLIENT_CONNECTION_KEYS), and credential_provider was missing from that whitelist.

When the Redis server authenticates via a credential provider instead of a password in the URL — an ACL user, or a rotating token such as Azure Managed Redis / Microsoft Entra ID — the background clients connected unauthenticated and the server rejected them (AuthenticationError: HELLO must be called with the client already authenticated), even though the broker itself connected fine. This change adds credential_provider to the whitelist so every connection authenticates from the same source.

Type of Change

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to not work as expected)
• Documentation update
• Refactoring (no functional changes)
• Performance improvement
• Test coverage improvement
• CI/CD update
• Dependency update

Related Issue

N/A — no tracked issue.

Changes Made

• Added "credential_provider" to _BACKGROUND_CLIENT_CONNECTION_KEYS in sdk/eggai/transport/redis.py, so it is forwarded from RedisTransport(**kwargs) to the reclaimer, group monitor, and delete-on-ack clients (the broker already received it).
• Documented the rationale in the comment above the whitelist.
• Added test_credential_provider_propagates_to_background_clients in sdk/tests/test_redis.py asserting the provider reaches _connection_kwargs and the reclaimer manager.
• Added an [Unreleased] → Added entry to sdk/CHANGELOG.md.

Testing

• Existing tests pass locally (make test)
• Added new tests for new functionality
• Tested manually (describe below if applicable)

Manual Testing Steps

1. N/A — covered by the automated redis transport suites (test_redis.py, test_redis_delete_on_ack.py, test_redis_shared_dlq.py) against a local Redis: 90 passed.

Changelog

• I have updated sdk/CHANGELOG.md under [Unreleased] section (required for code changes)

Checklist

• My code follows the project's code style (make lint passes)
• My code is properly formatted (make format applied)
• I have added/updated tests that prove my fix/feature works
• I have added/updated documentation as needed
• All tests pass locally
• I have reviewed my own code
• My changes generate no new warnings
• My commit messages follow the Conventional Commits standard

Additional Notes

Companion to the existing fix(transport): forward connection-resilience kwargs to the PEL reclaimer change — same mechanism (keep the background clients' connections consistent with the broker's), one more kwarg the whitelist needs to carry. Enables Entra ID / managed-identity auth against Azure Managed Redis end-to-end.

[github-actions]

QualOps Code Quality Analysis

Status: ⚠️ FAILED - Critical or high severity issues found

Summary

• Total Issues: 10
• Critical: 0 🔴
• High: 1 🟠
• Medium: 6 🟡
• Low: 3 🟢
• Files Analyzed: 8

🟠 High Issues (1)

• sdk/eggai/transport/redis.py:159 - bug
  Accessing self.broker._connection_kwargs relies on a private FastStream internal attribute that may not exist or may change across versions.

🟡 Medium Issues (6)

• sdk/eggai/transport/redis.py:162 - security
  Credentials (password, username, credential_provider) are now stored in a plain dict at _connection_kwargs on both RedisTransport and PendingReclaimerManager, creating new exposure vectors compared to the old whitelist approach which intentionally excluded sensitive keys.

• sdk/eggai/transport/redis.py:916 - bug
  _open_background_client() creates a new Redis client (and connection pool) on every call in _create_groups(), but the client is only closed in the finally block of that method — if _open_background_client() itself raises (e.g. invalid connection_kwargs), the finally still runs aclose() on a potentially-None or partially-constructed object, which is fine, but more critically _monitor_stream_groups() also calls _open_background_client() and if that raises the finally block tries to aclose() a client that was never assigned, causing an UnboundLocalError.

• sdk/eggai/transport/redis.py:183 - bug
  All keys from broker._connection_kwargs are forwarded without validation to aioredis.ConnectionPool(**kwargs), so any FastStream-internal key that appears in the broker dict but is not a valid aioredis kwarg causes TypeError at connect()/start() time, breaking the transport completely.

...and 3 more medium issues

📊 Full Report

View detailed report

---

Powered by QualOps

[pontino]

Tested against a local Redis with requirepass. The change itself works: with credential_provider the broker, reclaimer, monitor and delete-on-ack client all authenticate.

But the same whitelist still drops other things, and I think the whitelist itself is the problem. Details inline. Three more outside the diff:

• redis.py:184 — with broker= the background clients get the default url and no kwargs. RedisTransport(broker=RedisBroker("rediss://x.redis.azure.net", credential_provider=p)) → reclaimer/monitor dial redis://localhost:6379 with no auth. The other way round (RedisTransport(broker=b, credential_provider=p)) authenticates the background clients but not the broker.
• redis.py:951 — the monitor loop only catches ResponseError. AuthenticationError is a ConnectionError, so when the token rotates the monitor task dies with no log, and disconnect() re-raises it and skips reclaimer.stop(), broker.stop() and the delete client. Reproduced with an ACL user: change password + CLIENT KILL USER. Pre-existing, but it is the Entra ID case this PR is for.
• pending_reclaimer.py:339 — start() opens a new client on every connect() and never closes the old one. Two agents on one transport: 4 → 8 connections, 2 stay open after stop() until GC. Pre-existing. if self._redis_client is None: like the delete client fixes it.

[valdis]

Good catch 👍

[pontino]

LGTM