From 8c77e8fc0c2fcaa46501575b6b576e39daa052b3 Mon Sep 17 00:00:00 2001 From: Anton Vorobiev Date: Mon, 5 Oct 2026 15:49:07 +0200 Subject: [PATCH 1/3] feat(transport): support credential_provider on background Redis clients --- sdk/CHANGELOG.md | 11 +++++++++++ sdk/eggai/transport/redis.py | 8 ++++++++ sdk/tests/test_redis.py | 32 ++++++++++++++++++++++++++++++++ 3 files changed, 51 insertions(+) diff --git a/sdk/CHANGELOG.md b/sdk/CHANGELOG.md index afe3aa72..a5ff7f76 100644 --- a/sdk/CHANGELOG.md +++ b/sdk/CHANGELOG.md @@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- `RedisTransport` now forwards `credential_provider` to its background clients + (the PEL reclaimer, the group monitor, and the delete-on-ack client), not just + to the broker. This lets every connection authenticate from the same source + when the server requires a credential provider rather than a password in the + URL — an ACL user, or a rotating token such as Azure Managed Redis / Microsoft + Entra ID. Previously the background clients connected unauthenticated and the + server rejected them (`AuthenticationError: HELLO must be called with the + client already authenticated`) even while the broker was connected fine. + ## [0.6.0] - 2026-09-25 ### Added diff --git a/sdk/eggai/transport/redis.py b/sdk/eggai/transport/redis.py index 53d64216..d8d19efa 100644 --- a/sdk/eggai/transport/redis.py +++ b/sdk/eggai/transport/redis.py @@ -39,6 +39,13 @@ # 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). +# +# ``credential_provider`` is forwarded for the same reason: when the server requires +# authentication (ACL, or a rotating token such as Azure Managed Redis / Entra ID), +# the background clients authenticate against the same server as the broker, so they +# need the same credential source. Without it they connect unauthenticated and the +# server rejects them (``AuthenticationError: HELLO must be called with the client +# already authenticated``) even while the broker is connected fine. _BACKGROUND_CLIENT_CONNECTION_KEYS = ( "socket_timeout", "socket_connect_timeout", @@ -53,6 +60,7 @@ "ssl_cert_reqs", "ssl_ca_certs", "ssl_check_hostname", + "credential_provider", ) diff --git a/sdk/tests/test_redis.py b/sdk/tests/test_redis.py index ad939ed0..0cf3025d 100644 --- a/sdk/tests/test_redis.py +++ b/sdk/tests/test_redis.py @@ -1588,6 +1588,38 @@ async def handler(message): } +@pytest.mark.asyncio +async def test_credential_provider_propagates_to_background_clients(): + """A credential_provider passed to the transport reaches the background clients + (reclaimer, group monitor, delete-on-ack). When the server authenticates with a + rotating token (e.g. Azure Managed Redis / Entra ID) and no password is in the + URL, these clients would otherwise connect unauthenticated and be rejected.""" + from redis.credentials import CredentialProvider + + class _StubCredentialProvider(CredentialProvider): + def get_credentials(self): + return ("user", "token") + + provider = _StubCredentialProvider() + + transport = RedisTransport(credential_provider=provider) + + assert transport._connection_kwargs == {"credential_provider": provider} + + async def handler(message): + return message + + await transport.subscribe( + "orders", handler, handler_id="orders-handler-cred", retry_on_idle_ms=500 + ) + + assert transport._reclaimer_manager is not None + assert ( + transport._reclaimer_manager._connection_kwargs["credential_provider"] + is provider + ) + + @pytest.mark.asyncio async def test_reclaimer_start_applies_connection_kwargs(monkeypatch): """start() forwards connection_kwargs to from_url while pinning From ae75f350e567aa0caf5620d2c4e5b18d46754ec4 Mon Sep 17 00:00:00 2001 From: Anton Vorobiev Date: Tue, 6 Oct 2026 07:58:18 +0200 Subject: [PATCH 2/3] fix(transport): dial background Redis clients with the broker's connection settings --- sdk/CHANGELOG.md | 32 +++-- sdk/eggai/transport/pending_reclaimer.py | 27 ++-- sdk/eggai/transport/redis.py | 142 +++++++++----------- sdk/pyproject.toml | 2 +- sdk/tests/test_redis.py | 163 ++++++++++++----------- sdk/tests/test_redis_delete_on_ack.py | 4 +- sdk/tests/test_redis_shared_dlq.py | 2 +- 7 files changed, 188 insertions(+), 184 deletions(-) diff --git a/sdk/CHANGELOG.md b/sdk/CHANGELOG.md index a5ff7f76..a5d326f9 100644 --- a/sdk/CHANGELOG.md +++ b/sdk/CHANGELOG.md @@ -7,16 +7,30 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] -### Added +### Fixed -- `RedisTransport` now forwards `credential_provider` to its background clients - (the PEL reclaimer, the group monitor, and the delete-on-ack client), not just - to the broker. This lets every connection authenticate from the same source - when the server requires a credential provider rather than a password in the - URL — an ACL user, or a rotating token such as Azure Managed Redis / Microsoft - Entra ID. Previously the background clients connected unauthenticated and the - server rejected them (`AuthenticationError: HELLO must be called with the - client already authenticated`) even while the broker was connected fine. +- `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 diff --git a/sdk/eggai/transport/pending_reclaimer.py b/sdk/eggai/transport/pending_reclaimer.py index db696752..f52f8e23 100644 --- a/sdk/eggai/transport/pending_reclaimer.py +++ b/sdk/eggai/transport/pending_reclaimer.py @@ -289,23 +289,21 @@ class PendingReclaimerManager: 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 @@ async def start(self) -> None: # 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(): diff --git a/sdk/eggai/transport/redis.py b/sdk/eggai/transport/redis.py index d8d19efa..358adaaa 100644 --- a/sdk/eggai/transport/redis.py +++ b/sdk/eggai/transport/redis.py @@ -31,38 +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). -# -# ``credential_provider`` is forwarded for the same reason: when the server requires -# authentication (ACL, or a rotating token such as Azure Managed Redis / Entra ID), -# the background clients authenticate against the same server as the broker, so they -# need the same credential source. Without it they connect unauthenticated and the -# server rejects them (``AuthenticationError: HELLO must be called with the client -# already authenticated``) even while the broker is connected fine. -_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", - "credential_provider", -) - @dataclass(frozen=True) class _StreamGroupInfo: @@ -181,13 +149,14 @@ def __init__( 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 @@ -211,6 +180,19 @@ def __init__( # 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. @@ -237,9 +219,7 @@ async def connect(self): # 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() @@ -862,10 +842,7 @@ async def _create_groups(self) -> None: 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: @@ -920,43 +897,53 @@ async def _monitor_stream_groups(self) -> None: 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: @@ -1092,7 +1079,6 @@ def _setup_reclaimer( ) -> 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, ) diff --git a/sdk/pyproject.toml b/sdk/pyproject.toml index b4595900..c219e078 100644 --- a/sdk/pyproject.toml +++ b/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} diff --git a/sdk/tests/test_redis.py b/sdk/tests/test_redis.py index 0cf3025d..1b22f522 100644 --- a/sdk/tests/test_redis.py +++ b/sdk/tests/test_redis.py @@ -898,7 +898,7 @@ async def test_reclaimer_manager_start_stop_cycles(): ReclaimerConfig, ) - manager = PendingReclaimerManager("redis://localhost:6379") + manager = PendingReclaimerManager() assert manager._redis_client is None manager.add( @@ -930,7 +930,7 @@ async def test_reclaimer_manager_stop_without_start(): """Calling stop() before start() must not raise.""" from eggai.transport.pending_reclaimer import PendingReclaimerManager - manager = PendingReclaimerManager("redis://localhost:6379") + manager = PendingReclaimerManager() await manager.stop() # should not raise @@ -1276,7 +1276,7 @@ async def test_reclaimer_nogroup_recovery(): await redis_client.xgroup_create(stream_name, group_name, id="$", mkstream=True) await redis_client.delete(stream_name) - manager = PendingReclaimerManager("redis://localhost:6379") + manager = PendingReclaimerManager() manager.add( ReclaimerConfig( stream=stream_name, @@ -1560,17 +1560,19 @@ async def test_connection_kwargs_propagate_to_reclaimer(): socket_keepalive=True, health_check_interval=30, retry_on_timeout=True, - # A broker/FastStream-only kwarg that must NOT leak into the reclaimer's - # aioredis.from_url call. + # A broker/FastStream-only kwarg that is not a connection kwarg and must NOT + # reach the background clients. graceful_timeout=20.0, ) - assert transport._connection_kwargs == { - "socket_timeout": 10.0, - "socket_keepalive": True, - "health_check_interval": 30, - "retry_on_timeout": True, - } + # The broker's own connection dict is forwarded, so the resilience kwargs are + # present (alongside host/port/db); the FastStream-only one is not. + ck = transport._connection_kwargs + assert ck["socket_timeout"] == 10.0 + assert ck["socket_keepalive"] is True + assert ck["health_check_interval"] == 30 + assert ck["retry_on_timeout"] is True + assert "graceful_timeout" not in ck async def handler(message): return message @@ -1580,50 +1582,15 @@ async def handler(message): ) assert transport._reclaimer_manager is not None - assert transport._reclaimer_manager._connection_kwargs == { - "socket_timeout": 10.0, - "socket_keepalive": True, - "health_check_interval": 30, - "retry_on_timeout": True, - } - - -@pytest.mark.asyncio -async def test_credential_provider_propagates_to_background_clients(): - """A credential_provider passed to the transport reaches the background clients - (reclaimer, group monitor, delete-on-ack). When the server authenticates with a - rotating token (e.g. Azure Managed Redis / Entra ID) and no password is in the - URL, these clients would otherwise connect unauthenticated and be rejected.""" - from redis.credentials import CredentialProvider - - class _StubCredentialProvider(CredentialProvider): - def get_credentials(self): - return ("user", "token") - - provider = _StubCredentialProvider() - - transport = RedisTransport(credential_provider=provider) - - assert transport._connection_kwargs == {"credential_provider": provider} - - async def handler(message): - return message - - await transport.subscribe( - "orders", handler, handler_id="orders-handler-cred", retry_on_idle_ms=500 - ) - - assert transport._reclaimer_manager is not None - assert ( - transport._reclaimer_manager._connection_kwargs["credential_provider"] - is provider - ) + # The reclaimer gets the same connection dict the transport forwards. + assert transport._reclaimer_manager._connection_kwargs is ck @pytest.mark.asyncio async def test_reclaimer_start_applies_connection_kwargs(monkeypatch): - """start() forwards connection_kwargs to from_url while pinning - decode_responses=False (callers cannot override the binary-passthrough flag).""" + """start() builds its client from the forwarded connection dict — credentials, db + and resilience kwargs all reach it — while pinning decode_responses=False (callers + cannot override the binary-passthrough flag).""" from eggai.transport import pending_reclaimer from eggai.transport.pending_reclaimer import PendingReclaimerManager @@ -1633,27 +1600,36 @@ class _FakeClient: async def aclose(self): pass - def fake_from_url(url, **kwargs): - captured["url"] = url + def fake_pool(**kwargs): captured["kwargs"] = kwargs - return _FakeClient() + return object() - monkeypatch.setattr(pending_reclaimer.aioredis, "from_url", fake_from_url) + monkeypatch.setattr(pending_reclaimer.aioredis, "ConnectionPool", fake_pool) + monkeypatch.setattr( + pending_reclaimer.aioredis.Redis, + "from_pool", + staticmethod(lambda pool: _FakeClient()), + ) + provider = object() manager = PendingReclaimerManager( - "redis://localhost:6379", connection_kwargs={ + "host": "db.example", + "port": 6380, + "db": 3, + "password": "s3cret", + "credential_provider": provider, "socket_timeout": 10.0, - "socket_keepalive": True, # Even if a caller sneaks decode_responses in, it must stay False. "decode_responses": True, }, ) await manager.start() - assert captured["url"] == "redis://localhost:6379" + assert captured["kwargs"]["db"] == 3 + assert captured["kwargs"]["password"] == "s3cret" + assert captured["kwargs"]["credential_provider"] is provider assert captured["kwargs"]["socket_timeout"] == 10.0 - assert captured["kwargs"]["socket_keepalive"] is True assert captured["kwargs"]["decode_responses"] is False await manager.stop() @@ -1661,15 +1637,17 @@ def fake_from_url(url, **kwargs): @pytest.mark.asyncio async def test_monitor_applies_connection_kwargs(monkeypatch): - """The group monitor's background client gets the transport's resilience - kwargs while pinning decode_responses=True (its string commands need decoding, - and that flag must not be overridable).""" + """The group monitor's background client dials the same server as the broker — db, + password and credential_provider all reach it — while pinning decode_responses=True + (its string commands need decoding, and that flag must not be overridable).""" from eggai.transport import redis as redis_module + provider = object() transport = RedisTransport( + url="redis://localhost:6379/3", + password="s3cret", + credential_provider=provider, socket_timeout=10.0, - socket_keepalive=True, - health_check_interval=30, ) captured: dict = {} @@ -1678,24 +1656,49 @@ class _FakeClient: async def aclose(self): pass - def fake_from_url(url, **kwargs): - captured["url"] = url + def fake_pool(**kwargs): captured["kwargs"] = kwargs - return _FakeClient() + return object() - monkeypatch.setattr(redis_module.aioredis, "from_url", fake_from_url) + monkeypatch.setattr(redis_module.aioredis, "ConnectionPool", fake_pool) + monkeypatch.setattr( + redis_module.aioredis.Redis, + "from_pool", + staticmethod(lambda pool: _FakeClient()), + ) # _running stays False, so the monitor creates its client, skips the loop body, - # and closes the client via the finally block — enough to capture from_url. + # and closes the client via the finally block — enough to capture the pool kwargs. await transport._monitor_stream_groups() - assert captured["url"] == transport._redis_url + assert captured["kwargs"]["db"] == 3 + assert captured["kwargs"]["password"] == "s3cret" + assert captured["kwargs"]["credential_provider"] is provider assert captured["kwargs"]["socket_timeout"] == 10.0 - assert captured["kwargs"]["socket_keepalive"] is True - assert captured["kwargs"]["health_check_interval"] == 30 assert captured["kwargs"]["decode_responses"] is True +def test_prebuilt_broker_forwards_its_connection_settings(): + """With an explicit broker=, the background clients take their settings from the + broker's own connection dict, so they dial its server/db/credentials rather than a + default localhost. This is the case the old kwargs whitelist could not cover.""" + from faststream.redis import RedisBroker + + provider = object() + broker = RedisBroker( + "rediss://redis.example:10000/4", + password="s3cret", + credential_provider=provider, + ) + transport = RedisTransport(broker=broker) + + ck = transport._connection_kwargs + assert ck["host"] == "redis.example" + assert ck["db"] == 4 + assert ck["password"] == "s3cret" + assert ck["credential_provider"] is provider + + @pytest.mark.asyncio async def test_publish_maxlen_bounds_stream_length(): """Integration: publishing well past max_len keeps the stream approximately @@ -1919,7 +1922,7 @@ async def test_effective_idle_ms_escalates_and_caps(): """_effective_idle_ms grows geometrically with retry count and respects the cap.""" from eggai.transport.pending_reclaimer import PendingReclaimerManager - manager = PendingReclaimerManager("redis://localhost:6379") + manager = PendingReclaimerManager() config = _make_backoff_config( "s", "s.retry", backoff_multiplier=2.0, backoff_max_ms=5000 ) @@ -1943,7 +1946,7 @@ async def test_effective_idle_ms_huge_retry_count_does_not_overflow(): """ from eggai.transport.pending_reclaimer import PendingReclaimerManager - manager = PendingReclaimerManager("redis://localhost:6379") + manager = PendingReclaimerManager() # 2.0 ** 5000 overflows the max float (~1.8e308) several times over. capped = _make_backoff_config( @@ -1966,7 +1969,7 @@ async def test_effective_idle_ms_constant_when_multiplier_one(): """multiplier=1.0 reproduces the original constant cadence at every count.""" from eggai.transport.pending_reclaimer import PendingReclaimerManager - manager = PendingReclaimerManager("redis://localhost:6379") + manager = PendingReclaimerManager() config = _make_backoff_config("s", "s.retry") # multiplier defaults to 1.0 for count in range(5): assert manager._effective_idle_ms(config, count) == 1000 @@ -1984,7 +1987,7 @@ async def test_effective_idle_ms_jitter_spreads_upward_above_base(): """ from eggai.transport.pending_reclaimer import PendingReclaimerManager - manager = PendingReclaimerManager("redis://localhost:6379") + manager = PendingReclaimerManager() config = _make_backoff_config("s", "s.retry", backoff_jitter=0.5) # multiplier 1.0 samples = [manager._effective_idle_ms(config, 0) for _ in range(200)] # Upward only: never below the base floor (1000), never above base*(1+jitter). @@ -2002,7 +2005,7 @@ async def test_effective_idle_ms_jitter_stays_under_cap(): """Jitter is applied before the cap, so backoff_max_ms remains a hard ceiling.""" from eggai.transport.pending_reclaimer import PendingReclaimerManager - manager = PendingReclaimerManager("redis://localhost:6379") + manager = PendingReclaimerManager() # base*mult^count = 1000*2**3 = 8000; jitter would push to 12000, but the cap # pins it at 5000 regardless. config = _make_backoff_config( @@ -2106,7 +2109,7 @@ async def test_backoff_skips_high_retry_count_message_not_yet_due(): envelope = await _make_envelope({"type": "t", "_retry_count": "3", "id": test_id}) await _deliver_to_pel(client, stream, group, f"{group}-live", envelope) - manager = PendingReclaimerManager("redis://localhost:6379") + manager = PendingReclaimerManager() manager._redis_client = redis.Redis( host="localhost", port=6379, decode_responses=False ) @@ -2147,7 +2150,7 @@ async def test_backoff_reclaims_message_once_threshold_elapsed(): envelope = await _make_envelope({"type": "t", "_retry_count": "0", "id": test_id}) await _deliver_to_pel(client, stream, group, f"{group}-live", envelope) - manager = PendingReclaimerManager("redis://localhost:6379") + manager = PendingReclaimerManager() manager._redis_client = redis.Redis( host="localhost", port=6379, decode_responses=False ) @@ -2189,7 +2192,7 @@ async def test_reclaimer_stop_completes_when_cancellation_is_swallowed(monkeypat ReclaimerConfig, ) - manager = PendingReclaimerManager("redis://localhost:6379") + manager = PendingReclaimerManager() manager.add( ReclaimerConfig( stream="eggai.swallow", diff --git a/sdk/tests/test_redis_delete_on_ack.py b/sdk/tests/test_redis_delete_on_ack.py index 9f411806..7729a544 100644 --- a/sdk/tests/test_redis_delete_on_ack.py +++ b/sdk/tests/test_redis_delete_on_ack.py @@ -433,9 +433,7 @@ async def test_reclaimer_ack_respects_runtime_disabled_streams(): ) disabled: set[str] = set() - manager = PendingReclaimerManager( - "redis://localhost:6379", delete_on_ack_disabled=disabled - ) + manager = PendingReclaimerManager(delete_on_ack_disabled=disabled) client = MagicMock() client.xack = AsyncMock() manager._redis_client = client diff --git a/sdk/tests/test_redis_shared_dlq.py b/sdk/tests/test_redis_shared_dlq.py index d8724f0f..17d50242 100644 --- a/sdk/tests/test_redis_shared_dlq.py +++ b/sdk/tests/test_redis_shared_dlq.py @@ -563,7 +563,7 @@ async def test_reclaimer_wraps_poison_entry_into_dlq(): await redis_client.xreadgroup(group, "worker", {stream: ">"}, count=2) await asyncio.sleep(0.05) - manager = PendingReclaimerManager("redis://localhost:6379") + manager = PendingReclaimerManager() config = ReclaimerConfig( stream=stream, group=group, From bdaeee7e7c769cce77a1db829260b115c31a4514 Mon Sep 17 00:00:00 2001 From: Anton Vorobiev Date: Tue, 6 Oct 2026 08:06:19 +0200 Subject: [PATCH 3/3] update poetry lock --- sdk/poetry.lock | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/sdk/poetry.lock b/sdk/poetry.lock index a0ecb282..6dc14959 100644 --- a/sdk/poetry.lock +++ b/sdk/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 2.4.3 and should not be changed by hand. +# This file is automatically @generated by Poetry 2.2.1 and should not be changed by hand. [[package]] name = "a2a-sdk" @@ -1662,7 +1662,6 @@ files = [ {file = "jinja2-3.1.6-py3-none-any.whl", hash = "sha256:85ece4451f492d0c13c5dd7c13a64681a86afae63a5f347908daf103ce6d2f67"}, {file = "jinja2-3.1.6.tar.gz", hash = "sha256:0137fb05990d35f1275a587e9aee6d56da821fc83491a0fb838183be43f66d6d"}, ] -markers = {main = "extra == \"cli\""} [package.dependencies] MarkupSafe = ">=2.0" @@ -1717,7 +1716,7 @@ files = [ [package.dependencies] attrs = ">=22.2.0" -jsonschema-specifications = ">=2023.3.6" +jsonschema-specifications = ">=2023.03.6" referencing = ">=0.28.4" rpds-py = ">=0.25.0" @@ -2065,7 +2064,6 @@ files = [ {file = "markupsafe-3.0.3-cp39-cp39-win_arm64.whl", hash = "sha256:38664109c14ffc9e7437e86b4dceb442b0096dfe3541d7864d9cbe1da4cf36c8"}, {file = "markupsafe-3.0.3.tar.gz", hash = "sha256:722695808f4b6457b320fdc131280796bdceb04ab50fe1795cd540799ebe1698"}, ] -markers = {main = "extra == \"cli\""} [[package]] name = "mcp" @@ -4190,4 +4188,4 @@ otel = ["opentelemetry-api", "opentelemetry-exporter-otlp-proto-grpc", "opentele [metadata] lock-version = "2.1" python-versions = "^3.10" -content-hash = "59e37b5cd75c9dc95bc18b0a99e481f5d52edbc58d6cbfac6bec4c2d9f696c9a" +content-hash = "906bdc375aa13665155773f26c15b340e5bd45258e02bb17ce4c57d1ba883fb1"