feat(transport): support credential_provider on background Redis clients

sdk/CHANGELOG.md

@@ -20,6 +20,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   the retry stream. An explicit `max_records` is kept. This means one
   `XREADGROUP` per entry: setups that relied on the unbounded read for
   throughput with small, fast messages can set `max_records` explicitly.
+- `RedisTransport` now dials its background clients (the PEL reclaimer, the group
+  monitor, and the delete-on-ack client) with the broker's own connection settings,
+  instead of a hand-maintained whitelist of forwarded kwargs. The whitelist silently
+  dropped everything outside it, so against a server that needs authentication the
+  background clients connected unauthenticated and were rejected
+  (`AuthenticationError: HELLO must be called with the client already authenticated`)
+  even while the broker connected fine — and `db=` / `ssl=` / `client_name` diverged
+  too (e.g. the reclaimer scanned db 0 while the broker ran on db 3). Now `username`,
+  `password`, `db`, `ssl`, `credential_provider` (ACL users and rotating tokens such
+  as Azure Managed Redis / Microsoft Entra ID) and the resilience kwargs all match the
+  broker, including when a pre-built `broker=` is supplied.
+  Requires **faststream >= 0.7** — on 0.6.0 `RedisBroker(url, credential_provider=…)`
+  (and `password=`) raises `TypeError`.
+- The group monitor no longer dies on a transient connection or auth failure
+  (`AuthenticationError` is a `ConnectionError`, not a `ResponseError`, so it escaped
+  the loop's handler): a failed cycle is logged and retried next interval, by when
+  redis-py has reconnected and the credential provider has re-authenticated. A dying
+  monitor previously left `disconnect()` re-raising and skipping reclaimer/broker/
+  delete-client shutdown.
+- The PEL reclaimer no longer leaks a Redis client on each `connect()`: a second
+  `connect()` on a shared transport reuses the existing client instead of replacing
+  it without closing the old one.
 
 ## [0.6.0] - 2026-09-25
 

sdk/eggai/transport/pending_reclaimer.py

@@ -289,23 +289,21 @@
 
     def __init__(
         self,
-        redis_url: str,
         connection_kwargs: dict[str, Any] | None = None,
         delete_on_ack_disabled: set[str] | None = None,
     ):
-        self._redis_url = redis_url
         # Stream keys where delete_on_ack was turned off at runtime (another
         # consumer group appeared). Shared with, and mutated by, the transport's
         # group monitor; _ack() falls back to a plain XACK for these.
         self._delete_on_ack_disabled: set[str] = (
             delete_on_ack_disabled if delete_on_ack_disabled is not None else set()
         )
-        # Connection-resilience settings (socket_timeout, socket_keepalive,
-        # health_check_interval, retry_on_timeout, …) forwarded from the
-        # transport so this independent client recovers from a silently dropped
-        # connection the same way the broker does. Without them a blocking read
-        # against a half-dead socket (e.g. cloud Redis failover) hangs forever
-        # with no socket timeout to break it.
+        # The broker's own connection dict (host/port/db/username/password/ssl/
+        # credential_provider plus resilience kwargs), forwarded from the transport so
+        # this independent client reaches the same server and database and authenticates
+        # the same way the broker does — and recovers from a silently dropped connection
+        # (without a socket timeout a blocking read against a half-dead socket, e.g. a
+        # cloud Redis failover, would hang forever).
         self._connection_kwargs: dict[str, Any] = connection_kwargs or {}
         self._redis_client: aioredis.Redis | None = None
         self._configs: dict[tuple[str, str, str], ReclaimerConfig] = {}
@@ -336,10 +334,15 @@
         # decode_responses=False: field values are kept as raw bytes so that
         # FastStream's binary-encoded __data__ field is passed through unchanged.
         # decode_responses is pinned here and must not be overridden by callers.
-        self._redis_client = aioredis.from_url(
-            self._redis_url,
-            **{**self._connection_kwargs, "decode_responses": False},
-        )
+        # Guarded so a second connect() on a shared transport reuses the client rather
+        # than leaking the previous one (the client is closed only in stop()). A
+        # ConnectionPool is used (not Redis(**kwargs)) because the broker's dict can
+        # carry connection_class/path for rediss:// and unix://.
+        if self._redis_client is None:
+            pool = aioredis.ConnectionPool(
+                **{**self._connection_kwargs, "decode_responses": False}
+            )
+            self._redis_client = aioredis.Redis.from_pool(pool)
         self._running = True
         for key, config in self._configs.items():
             if key in self._tasks and not self._tasks[key].done():
@@ -527,7 +530,7 @@
             stale_ids: list[Any] = [msg_id for msg_id, _ in candidates]
         else:
             stale_ids = []
             for msg_id, idle in candidates:
                 retry_count = await self._read_retry_count(config.stream, msg_id)
                 if idle >= self._effective_idle_ms(config, retry_count):
                     stale_ids.append(msg_id)

sdk/eggai/transport/redis.py

@@ -31,30 +31,6 @@
 # the stream and own their own PEL entries — the competing-consumers pattern.
 _CONSUMER_INSTANCE = f"{socket.gethostname()}-{os.getpid()}"
 
-# Connection settings that, when passed to the broker, must also be forwarded to
-# the transport's other long-lived background clients — the PEL reclaimer and the
-# consumer-group monitor — so every connection recovers from a silently dropped
-# socket the same way. Kept to redis-py connection/resilience kwargs that are valid
-# for ``aioredis.from_url`` — broker/FastStream-only kwargs (decoder, middlewares,
-# asyncapi_*, …) are intentionally excluded. ``decode_responses`` is omitted on
-# purpose: each background client pins it itself (False for the reclaimer's binary
-# passthrough, True for the monitor's string commands).
-_BACKGROUND_CLIENT_CONNECTION_KEYS = (
-    "socket_timeout",
-    "socket_connect_timeout",
-    "socket_keepalive",
-    "socket_keepalive_options",
-    "health_check_interval",
-    "retry_on_timeout",
-    "retry_on_error",
-    "max_connections",
-    "ssl_keyfile",
-    "ssl_certfile",
-    "ssl_cert_reqs",
-    "ssl_ca_certs",
-    "ssl_check_hostname",
-)
-
 
 @dataclass(frozen=True)
 class _StreamGroupInfo:
@@ -173,16 +149,17 @@
             self.broker = broker
         else:
             self.broker = RedisBroker(url, log_level=logging.INFO, **kwargs)
-        self._redis_url = url
-        # Forward connection-resilience kwargs to the transport's background clients
-        # (PEL reclaimer, group monitor) so they don't hang on a silently dropped
-        # connection while the broker recovers.
-        self._connection_kwargs: dict[str, Any] = {
-            k: kwargs[k] for k in _BACKGROUND_CLIENT_CONNECTION_KEYS if k in kwargs
-        }
+        # The exact connection settings the broker dials with (host/port/db/username/
+        # password/ssl/credential_provider/resilience kwargs), read from the broker so
+        # the transport's own long-lived clients — the PEL reclaimer, the group monitor,
+        # and the delete-on-ack client — reach the same server and database and
+        # authenticate the same way. Taking the broker's dict (instead of re-deriving a
+        # whitelist from kwargs) also covers the ``broker=`` case and stops auth-only
+        # params like ``password`` and ``credential_provider`` from being silently dropped.
+        self._connection_kwargs: dict[str, Any] = dict(self.broker._connection_kwargs)
         self._max_len = max_len
         self._retry_max_len = retry_max_len
         self._dlq_max_len = dlq_max_len
         self._running = False
         self._reclaimer_manager: PendingReclaimerManager | None = None
         # A set so repeated subscribe() calls with an identical (stream_key, group,
@@ -203,6 +180,19 @@
         # group appeared; shared with the reclaimer. Plain XACK from then on.
         self._delete_on_ack_disabled: set[str] = set()
 
+    def _open_background_client(self, *, decode_responses: bool) -> aioredis.Redis:
+        """Open a long-lived background client (reclaimer / monitor / delete-on-ack)
+        that dials the same server as the broker. ``self._connection_kwargs`` is the
+        broker's own connection dict, so host/port/db/username/password/ssl/
+        credential_provider all match. A ConnectionPool (rather than ``Redis(**kwargs)``)
+        is used because the broker's dict can carry ``connection_class``/``path`` for
+        ``rediss://`` and ``unix://``, which the ``Redis`` constructor rejects.
+        ``decode_responses`` is pinned per client and must not be overridden by it."""
+        pool = aioredis.ConnectionPool(
+            **{**self._connection_kwargs, "decode_responses": decode_responses}
+        )
+        return aioredis.Redis.from_pool(pool)
+
     async def connect(self):
         """
         Establishes a connection to the Redis server by starting the RedisBroker instance.
@@ -229,9 +219,7 @@
         # an earlier connect().
         opened_delete_client = False
         if self._delete_on_ack_groups and self._delete_client is None:
-            self._delete_client = aioredis.from_url(
-                self._redis_url, **self._connection_kwargs
-            )
+            self._delete_client = self._open_background_client(decode_responses=False)
             opened_delete_client = True
         try:
             await self._create_groups()
@@ -867,10 +855,7 @@
         use ">" from a group that starts where the caller asked."""
         if not self._stream_subscriptions:
             return
-        client = aioredis.from_url(
-            self._redis_url,
-            **{**self._connection_kwargs, "decode_responses": True},
-        )
+        client = self._open_background_client(decode_responses=True)
         try:
             for info in list(self._stream_subscriptions):
                 try:
@@ -925,46 +910,56 @@
         group_create_id with MKSTREAM when the stream itself is gone.
         """
         # decode_responses=True (string commands here) is pinned and must not be
-        # overridden; the forwarded resilience kwargs (socket_timeout, keepalive, …)
-        # keep this client from hanging on a half-dead socket during the very
-        # failover this monitor exists to recover from.
-        client = aioredis.from_url(
-            self._redis_url,
-            **{**self._connection_kwargs, "decode_responses": True},
-        )
+        # overridden; the broker's connection kwargs (socket_timeout, keepalive,
+        # credentials, …) keep this client authenticated and from hanging on a
+        # half-dead socket during the very failover this monitor exists to recover from.
+        client = self._open_background_client(decode_responses=True)
         try:
             while self._running:
                 await asyncio.sleep(self._group_monitor_interval_s)
-                # Iterate a snapshot: a concurrent subscribe() on a shared
-                # transport may mutate the set between awaits below.
-                for info in list(self._stream_subscriptions):
-                    try:
-                        stream_exists = await client.exists(info.stream_key)
-                        create_id = "0" if stream_exists else info.group_create_id
-                        await client.xgroup_create(
-                            name=info.stream_key,
-                            groupname=info.group,
-                            id=create_id,
-                            mkstream=True,
-                        )
-                        logger.info(
-                            "Recreated missing consumer group %s on stream %s (id=%s)",
-                            info.group,
-                            info.stream_key,
-                            create_id,
-                        )
-                    except ResponseError as e:
-                        if "BUSYGROUP" not in str(e):
-                            logger.warning(
-                                "Failed to ensure consumer group %s on %s: %s",
+                try:
+                    # Iterate a snapshot: a concurrent subscribe() on a shared
+                    # transport may mutate the set between awaits below.
+                    for info in list(self._stream_subscriptions):
+                        try:
+                            stream_exists = await client.exists(info.stream_key)
+                            create_id = "0" if stream_exists else info.group_create_id
+                            await client.xgroup_create(
+                                name=info.stream_key,
+                                groupname=info.group,
+                                id=create_id,
+                                mkstream=True,
+                            )
+                            logger.info(
+                                "Recreated missing consumer group %s on stream %s (id=%s)",
                                 info.group,
                                 info.stream_key,
-                                e,
+                                create_id,
                             )
-                await self._check_delete_on_ack_groups(client)
+                        except ResponseError as e:
+                            if "BUSYGROUP" not in str(e):
+                                logger.warning(
+                                    "Failed to ensure consumer group %s on %s: %s",
+                                    info.group,
+                                    info.stream_key,
+                                    e,
+                                )
+                    await self._check_delete_on_ack_groups(client)
+                except Exception as e:
+                    # A dropped/half-dead socket or a rotated auth token must not kill
+                    # the monitor task (AuthenticationError is a ConnectionError, not a
+                    # ResponseError, so it escapes the inner handler above). Log and
+                    # retry next interval, by when redis-py has reconnected and the
+                    # credential provider has re-authenticated. CancelledError is a
+                    # BaseException and still propagates to the outer handler.
+                    logger.warning(
+                        "Consumer-group monitor cycle failed; retrying in %.1fs: %s",
+                        self._group_monitor_interval_s,
+                        e,
+                    )
         except asyncio.CancelledError:
             pass
         finally:
             await client.aclose()
 
     async def _check_delete_on_ack_groups(self, client: Any) -> None:
@@ -1097,7 +1092,6 @@
     ) -> tuple[str, str, str]:
         if self._reclaimer_manager is None:
             self._reclaimer_manager = PendingReclaimerManager(
-                self._redis_url,
                 connection_kwargs=self._connection_kwargs,
                 delete_on_ack_disabled=self._delete_on_ack_disabled,
             )

sdk/pyproject.toml

@@ -34,7 +34,7 @@ classifiers = [
 [tool.poetry.dependencies]
 python = "^3.10"
 pydantic = "^2.11.4"
-faststream = {extras = ["kafka", "redis"], version = ">=0.6,<0.8"}
+faststream = {extras = ["kafka", "redis"], version = ">=0.7,<0.8"}
 click = {version = "^8.0.0", optional = true}
 jinja2 = {version = "^3.1.6", optional = true}
 a2a-sdk = {version = "^0.3.0", optional = true}