diff --git a/cloudrift/cache/__init__.py b/cloudrift/cache/__init__.py index e593711..6f33703 100644 --- a/cloudrift/cache/__init__.py +++ b/cloudrift/cache/__init__.py @@ -1,6 +1,6 @@ from urllib.parse import quote -from cloudrift.cache.base import CacheBackend +from cloudrift.cache.base import CacheBackend, resilient_client_kwargs _VALID_SSL_CERT_REQS = ("CERT_NONE", "CERT_OPTIONAL", "CERT_REQUIRED") @@ -111,4 +111,4 @@ def get_cache(provider: str, auth_method: str, **kwargs) -> CacheBackend: return factory(**kwargs) -__all__ = ["CacheBackend", "get_cache", "cache_broker_url"] +__all__ = ["CacheBackend", "get_cache", "cache_broker_url", "resilient_client_kwargs"] diff --git a/cloudrift/cache/base.py b/cloudrift/cache/base.py index 6e6800d..b659791 100644 --- a/cloudrift/cache/base.py +++ b/cloudrift/cache/base.py @@ -2,10 +2,61 @@ from contextlib import asynccontextmanager import redis.asyncio as aioredis -from redis.exceptions import RedisError +from redis.asyncio.retry import Retry +from redis.backoff import ExponentialBackoff +from redis.exceptions import ConnectionError as RedisConnectionError +from redis.exceptions import ReadOnlyError, RedisError +from redis.exceptions import TimeoutError as RedisTimeoutError from cloudrift.core.exceptions import CacheError +# Defaults for every Redis-backed cache client. +# +# redis-py ships with `health_check_interval=0` and `Retry(NoBackoff(), 0)` — +# i.e. no health checks and no retries. Managed Redis (Azure Cache for Redis, +# ElastiCache) reaps idle connections and drops every connection during +# maintenance, failover, and scale operations, so those defaults surface +# `ConnectionError: Connection closed by server.` to the caller on the first +# command after the server hangs up. These values make a pooled client ride +# through a sub-second disruption instead. Override per call site by passing the +# same keyword to any factory. +# +# `ReadOnlyError` is in the retry set for Azure Cache for Redis specifically: +# failover promotes the replica, and a pooled connection can still be pointed at +# the node that just became read-only, which answers writes with `-READONLY`. +# redis-py raises that as a `ResponseError` subclass, not a `ConnectionError`, +# so it is only retried if listed. Retrying disconnects and reconnects, which +# re-resolves DNS to the newly promoted primary. +DEFAULT_HEALTH_CHECK_INTERVAL = 30 +DEFAULT_MAX_RETRIES = 3 +DEFAULT_SOCKET_TIMEOUT = 5.0 +DEFAULT_SOCKET_CONNECT_TIMEOUT = 5.0 +DEFAULT_MAX_CONNECTIONS = 100 + + +def resilient_client_kwargs(**overrides) -> dict: + """Return `aioredis.Redis` kwargs that survive managed-Redis disruptions. + + Backoff is 0.1s, 0.2s, 0.4s — roughly 0.7s of retry budget per command, + enough for a fast failover without stalling a request behind a hard outage. + `Retry` is deep-copied per connection by redis-py, so sharing one instance + across a pool is safe. + + The cache API exposes no blocking commands, so a socket read timeout cannot + cut short a legitimately long call. + """ + kwargs = { + "retry": Retry(ExponentialBackoff(cap=1.0, base=0.1), DEFAULT_MAX_RETRIES), + "retry_on_error": [RedisConnectionError, RedisTimeoutError, ReadOnlyError], + "health_check_interval": DEFAULT_HEALTH_CHECK_INTERVAL, + "socket_keepalive": True, + "socket_timeout": DEFAULT_SOCKET_TIMEOUT, + "socket_connect_timeout": DEFAULT_SOCKET_CONNECT_TIMEOUT, + "max_connections": DEFAULT_MAX_CONNECTIONS, + } + kwargs.update(overrides) + return kwargs + class CacheBackend(ABC): """Abstract base class for cloud cache backends.""" diff --git a/cloudrift/cache/redis_azure.py b/cloudrift/cache/redis_azure.py index 899fcf6..9f083dc 100644 --- a/cloudrift/cache/redis_azure.py +++ b/cloudrift/cache/redis_azure.py @@ -1,7 +1,7 @@ import redis.asyncio as aioredis from redis.credentials import CredentialProvider -from cloudrift.cache.base import CacheBackend, _RedisMixin +from cloudrift.cache.base import CacheBackend, _RedisMixin, resilient_client_kwargs from cloudrift.core.exceptions import CacheConnectionError @@ -29,30 +29,56 @@ def from_access_key( port: int = 6380, db: int = 0, ssl: bool = True, + username: str | None = None, + ssl_cert_reqs: str = "required", decode_responses: bool = False, + **client_kwargs, ) -> "AzureRedisCacheBackend": - """Authenticate with an Azure Cache for Redis access key. + """Authenticate with an Azure Redis access key. + + Covers both Azure Redis products, which differ in host, port and auth: + + - **Azure Cache for Redis** — ``.redis.cache.windows.net`` on 6380, + password-only. The defaults here target it. + - **Azure Managed Redis** — ``..redis.azure.net`` on + **10000**, ACL-based, so it also needs ``username`` (usually + ``"default"``). + + Both are TLS-only. Connecting in plaintext gets the connection closed by + the server, surfacing as ``ConnectionError: Connection closed by server.`` Args: - host: e.g. ``.redis.cache.windows.net`` + host: Cache hostname. access_key: Primary or secondary access key from the Azure portal. - port: Redis SSL port (default 6380; non-TLS is 6379). + port: TLS port — 6380 for Azure Cache for Redis, 10000 for Azure + Managed Redis. db: Database index (default 0). - ssl: Enable TLS (default ``True``; required for Azure Cache for Redis). + ssl: Enable TLS (default ``True``; required by both products). + username: ACL username. Required by Azure Managed Redis; leave unset + for Azure Cache for Redis. + ssl_cert_reqs: Server-certificate policy: ``"required"`` (default), + ``"optional"``, or ``"none"``. Do not pass ``None`` — redis-py + resolves that to ``CERT_NONE``, silently disabling verification. decode_responses: When ``True``, reads return ``str`` instead of ``bytes``. + **client_kwargs: Overrides for the connection-resilience defaults + documented on ``resilient_client_kwargs``. """ try: client = aioredis.Redis( host=host, port=port, password=access_key, + username=username, db=db, ssl=ssl, - decode_responses=decode_responses, + ssl_cert_reqs=ssl_cert_reqs, + **resilient_client_kwargs( + decode_responses=decode_responses, **client_kwargs + ), ) return cls(client) except Exception as e: - raise CacheConnectionError(f"Failed to connect to Azure Cache for Redis: {e}") from e + raise CacheConnectionError(f"Failed to connect to Azure Redis: {e}") from e @classmethod def from_managed_identity( @@ -65,6 +91,7 @@ def from_managed_identity( client_id: str | None = None, decode_responses: bool = False, credential_options: dict | None = None, + **client_kwargs, ) -> "AzureRedisCacheBackend": """Authenticate via Azure AD (Entra ID token auth). @@ -82,6 +109,8 @@ def from_managed_identity( Omit to use the system-assigned identity. credential_options: Forwarded to ``DefaultAzureCredential`` — see :mod:`cloudrift.core.azure_credentials`. + **client_kwargs: Overrides for the connection-resilience defaults + documented on ``resilient_client_kwargs``. """ try: from cloudrift.core.azure_credentials import build_credential @@ -94,7 +123,9 @@ def from_managed_identity( db=db, ssl=ssl, credential_provider=provider, - decode_responses=decode_responses, + **resilient_client_kwargs( + decode_responses=decode_responses, **client_kwargs + ), ) return cls(client) except Exception as e: @@ -114,6 +145,7 @@ def from_service_principal( db: int = 0, ssl: bool = True, decode_responses: bool = False, + **client_kwargs, ) -> "AzureRedisCacheBackend": """Authenticate via Azure AD service principal (Entra ID token auth). @@ -129,6 +161,8 @@ def from_service_principal( port: Redis SSL port (default 6380). db: Database index (default 0). ssl: Enable TLS (default ``True``). + **client_kwargs: Overrides for the connection-resilience defaults + documented on ``resilient_client_kwargs``. """ try: from azure.identity import ClientSecretCredential @@ -142,7 +176,9 @@ def from_service_principal( db=db, ssl=ssl, credential_provider=provider, - decode_responses=decode_responses, + **resilient_client_kwargs( + decode_responses=decode_responses, **client_kwargs + ), ) return cls(client) except Exception as e: diff --git a/cloudrift/cache/redis_elasticache.py b/cloudrift/cache/redis_elasticache.py index a9d8d6f..952d7f3 100644 --- a/cloudrift/cache/redis_elasticache.py +++ b/cloudrift/cache/redis_elasticache.py @@ -1,7 +1,7 @@ import redis.asyncio as aioredis from redis.credentials import CredentialProvider -from cloudrift.cache.base import CacheBackend, _RedisMixin +from cloudrift.cache.base import CacheBackend, _RedisMixin, resilient_client_kwargs from cloudrift.core.exceptions import CacheConnectionError @@ -31,6 +31,7 @@ def from_auth_token( ssl: bool = True, ssl_ca_certs: str | None = None, decode_responses: bool = False, + **client_kwargs, ) -> "AWSElastiCacheBackend": """Connect using an ElastiCache AUTH token (shared secret). @@ -45,6 +46,8 @@ def from_auth_token( ssl: Enable TLS in-transit encryption (default ``True``). ssl_ca_certs: Optional path to the CA bundle (PEM) for server verification. decode_responses: When ``True``, reads return ``str`` instead of ``bytes``. + **client_kwargs: Overrides for the connection-resilience defaults + documented on ``resilient_client_kwargs``. """ try: client = aioredis.Redis( @@ -54,7 +57,9 @@ def from_auth_token( db=db, ssl=ssl, ssl_ca_certs=ssl_ca_certs, - decode_responses=decode_responses, + **resilient_client_kwargs( + decode_responses=decode_responses, **client_kwargs + ), ) return cls(client) except Exception as e: @@ -75,6 +80,7 @@ def from_iam_auth( aws_session_token: str | None = None, profile_name: str | None = None, decode_responses: bool = False, + **client_kwargs, ) -> "AWSElastiCacheBackend": """Connect using IAM-based authentication (ElastiCache Redis 7+ with IAM enabled). @@ -93,6 +99,8 @@ def from_iam_auth( aws_secret_access_key: Optional explicit AWS secret key. aws_session_token: Optional STS session token. profile_name: Optional named AWS credentials profile. + **client_kwargs: Overrides for the connection-resilience defaults + documented on ``resilient_client_kwargs``. """ try: provider = _ElastiCacheIAMProvider( @@ -112,7 +120,9 @@ def from_iam_auth( ssl=ssl, ssl_ca_certs=ssl_ca_certs, credential_provider=provider, - decode_responses=decode_responses, + **resilient_client_kwargs( + decode_responses=decode_responses, **client_kwargs + ), ) return cls(client) except Exception as e: @@ -129,6 +139,7 @@ def from_tls_cert( ssl_keyfile: str | None = None, ssl_ca_certs: str | None = None, decode_responses: bool = False, + **client_kwargs, ) -> "AWSElastiCacheBackend": """Connect using mutual TLS (mTLS) with a client certificate and key. @@ -141,6 +152,8 @@ def from_tls_cert( ssl_keyfile: Path to the client private key PEM file. ssl_ca_certs: Path to the CA certificate bundle (PEM) for server verification. decode_responses: When ``True``, reads return ``str`` instead of ``bytes``. + **client_kwargs: Overrides for the connection-resilience defaults + documented on ``resilient_client_kwargs``. """ try: client = aioredis.Redis( @@ -152,7 +165,9 @@ def from_tls_cert( ssl_certfile=ssl_certfile, ssl_keyfile=ssl_keyfile, ssl_ca_certs=ssl_ca_certs, - decode_responses=decode_responses, + **resilient_client_kwargs( + decode_responses=decode_responses, **client_kwargs + ), ) return cls(client) except Exception as e: diff --git a/cloudrift/cache/redis_standalone.py b/cloudrift/cache/redis_standalone.py index 142c92f..69736e4 100644 --- a/cloudrift/cache/redis_standalone.py +++ b/cloudrift/cache/redis_standalone.py @@ -1,6 +1,6 @@ import redis.asyncio as aioredis -from cloudrift.cache.base import CacheBackend, _RedisMixin +from cloudrift.cache.base import CacheBackend, _RedisMixin, resilient_client_kwargs from cloudrift.core.exceptions import CacheConnectionError @@ -26,6 +26,7 @@ def from_url( url: str, ssl_ca_certs: str | None = None, decode_responses: bool = False, + **client_kwargs, ) -> "StandaloneRedisBackend": """Connect using a Redis URL. @@ -36,9 +37,13 @@ def from_url( decode_responses: When ``True``, read operations return ``str`` instead of ``bytes`` (mirrors redis-py's ``decode_responses``). Default ``False`` keeps the cache-backend contract of returning ``bytes``. + **client_kwargs: Overrides for the connection-resilience defaults + documented on ``resilient_client_kwargs``. """ try: - kwargs: dict = {"decode_responses": decode_responses} + kwargs = resilient_client_kwargs( + decode_responses=decode_responses, **client_kwargs + ) if ssl_ca_certs: kwargs["ssl_ca_certs"] = ssl_ca_certs return cls(aioredis.from_url(url, **kwargs)) @@ -54,8 +59,10 @@ def from_credentials( username: str | None = None, db: int = 0, ssl: bool = False, + ssl_cert_reqs: str = "required", ssl_ca_certs: str | None = None, decode_responses: bool = False, + **client_kwargs, ) -> "StandaloneRedisBackend": """Connect using explicit host, port, and optional credentials. @@ -65,10 +72,20 @@ def from_credentials( password: Optional AUTH password. username: Optional ACL username (Redis 6+). db: Database index (default 0). - ssl: Enable TLS (default ``False``). + ssl: Enable TLS (default ``False``). Required by managed Redis that + only exposes a TLS port — e.g. Azure Managed Redis on 10000 or + Azure Cache for Redis on 6380. Connecting in plaintext to a + TLS-only port gets the connection closed by the server, which + surfaces as ``ConnectionError: Connection closed by server.`` + ssl_cert_reqs: Server-certificate policy: ``"required"`` (default), + ``"optional"``, or ``"none"``. Ignored unless ``ssl`` is true. + Do not pass ``None`` — redis-py resolves that to ``CERT_NONE``, + silently disabling verification. ssl_ca_certs: Optional path to the CA certificate bundle (PEM). decode_responses: When ``True``, read operations return ``str`` instead of ``bytes`` (mirrors redis-py's ``decode_responses``). + **client_kwargs: Overrides for the connection-resilience defaults + documented on ``resilient_client_kwargs``. """ try: client = aioredis.Redis( @@ -78,8 +95,11 @@ def from_credentials( username=username, db=db, ssl=ssl, + ssl_cert_reqs=ssl_cert_reqs, ssl_ca_certs=ssl_ca_certs, - decode_responses=decode_responses, + **resilient_client_kwargs( + decode_responses=decode_responses, **client_kwargs + ), ) return cls(client) except Exception as e: @@ -97,6 +117,7 @@ def from_tls_cert( ssl_keyfile: str | None = None, ssl_ca_certs: str | None = None, decode_responses: bool = False, + **client_kwargs, ) -> "StandaloneRedisBackend": """Connect using mutual TLS (mTLS) with client certificate and key files. @@ -111,6 +132,8 @@ def from_tls_cert( ssl_ca_certs: Path to the CA certificate bundle (PEM) for server verification. decode_responses: When ``True``, read operations return ``str`` instead of ``bytes`` (mirrors redis-py's ``decode_responses``). + **client_kwargs: Overrides for the connection-resilience defaults + documented on ``resilient_client_kwargs``. """ try: client = aioredis.Redis( @@ -123,7 +146,9 @@ def from_tls_cert( ssl_certfile=ssl_certfile, ssl_keyfile=ssl_keyfile, ssl_ca_certs=ssl_ca_certs, - decode_responses=decode_responses, + **resilient_client_kwargs( + decode_responses=decode_responses, **client_kwargs + ), ) return cls(client) except Exception as e: diff --git a/cloudrift/messaging/azure_bus.py b/cloudrift/messaging/azure_bus.py index 72f35cf..c00c695 100644 --- a/cloudrift/messaging/azure_bus.py +++ b/cloudrift/messaging/azure_bus.py @@ -336,6 +336,22 @@ async def receive( if not raw_messages: await receiver.__aexit__(None, None, None) return [] + def _body_bytes(m) -> bytes: + # ServiceBusReceivedMessage.body is a generator of chunks for + # the common data body type, so bytes(m) raises TypeError. + # It is single-use, hence joining eagerly here. + raw = m.body + if isinstance(raw, bytes): + return raw + if isinstance(raw, str): + return raw.encode("utf-8") + if raw is None: + return b"" + chunks = [ + c if isinstance(c, bytes) else str(c).encode("utf-8") for c in raw + ] + return b"".join(chunks) + tokens: set[str] = set() messages: list[Message] = [] for m in raw_messages: @@ -355,7 +371,7 @@ async def receive( messages.append( Message( id=str(m.message_id or ""), - body=bytes(m), + body=_body_bytes(m), receipt_handle=token, attributes=attrs, group_id=m.session_id, diff --git a/pyproject.toml b/pyproject.toml index c13ddb6..f8666c8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "lyzr-cloudrift" -version = "0.2.7" +version = "0.2.8" description = "Cloud-agnostic abstraction for storage, messaging, document databases, cache, secrets, SQL, crypto (KMS), pub/sub, and email" readme = "README.md" requires-python = ">=3.11" diff --git a/tests/test_cache.py b/tests/test_cache.py index 8297cc3..f012bb2 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -1,7 +1,15 @@ +import inspect + import pytest import fakeredis.aioredis -from cloudrift.cache import get_cache +from redis.exceptions import ConnectionError as RedisConnectionError +from redis.exceptions import ReadOnlyError +from redis.exceptions import TimeoutError as RedisTimeoutError + +from cloudrift.cache import get_cache, resilient_client_kwargs +from cloudrift.cache.redis_azure import AzureRedisCacheBackend +from cloudrift.cache.redis_elasticache import AWSElastiCacheBackend from cloudrift.cache.redis_standalone import StandaloneRedisBackend @@ -246,3 +254,104 @@ async def test_flush(cache): def test_invalid_provider(): with pytest.raises(ValueError, match="Unknown cache provider"): get_cache("gcp_memorystore", "from_url", url="redis://localhost") + + +# --------------------------------------------------------------------------- +# Connection resilience +# --------------------------------------------------------------------------- +# +# redis-py defaults to health_check_interval=0 and Retry(NoBackoff(), 0), so a +# managed Redis that reaps idle connections or fails over surfaces +# "Connection closed by server." to the caller. Every factory must opt out of +# those defaults. +# +# _FACTORIES covers the auth methods constructible without cloud credentials. +# test_every_factory_accepts_client_kwargs covers the rest by signature, so a +# newly added factory cannot silently skip the resilience settings. + +_FACTORIES = [ + ("redis", "from_url", {"url": "redis://localhost:6379/0"}), + ("redis", "from_credentials", {"host": "localhost"}), + ("redis", "from_tls_cert", {"host": "localhost"}), + ("elasticache", "from_auth_token", {"host": "ec.example.com", "auth_token": "t"}), + ("elasticache", "from_tls_cert", {"host": "ec.example.com"}), + ("azure_redis", "from_access_key", {"host": "az.example.com", "access_key": "k"}), +] + + +@pytest.mark.parametrize("provider,auth_method,kwargs", _FACTORIES) +def test_factory_sets_connection_resilience(provider, auth_method, kwargs): + backend = get_cache(provider, auth_method, **kwargs) + pool = backend._client.connection_pool + ckw = pool.connection_kwargs + + assert ckw["health_check_interval"] == 30 + assert ckw["socket_keepalive"] is True + assert ckw["socket_timeout"] == 5.0 + assert ckw["socket_connect_timeout"] == 5.0 + assert pool.max_connections == 100 + + # A retry policy with real attempts, covering connection loss and timeouts. + assert ckw["retry"].get_retries() == 3 + assert RedisConnectionError in ckw["retry_on_error"] + assert RedisTimeoutError in ckw["retry_on_error"] + # Azure failover answers writes with -READONLY from the demoted primary. + # It is a ResponseError, not a ConnectionError, so it must be listed + # explicitly or the caller sees the error instead of a reconnect. + assert ReadOnlyError in ckw["retry_on_error"] + + +@pytest.mark.parametrize("provider,auth_method,kwargs", _FACTORIES) +def test_factory_resilience_is_overridable(provider, auth_method, kwargs): + backend = get_cache( + provider, auth_method, health_check_interval=7, max_connections=5, **kwargs + ) + pool = backend._client.connection_pool + assert pool.connection_kwargs["health_check_interval"] == 7 + assert pool.max_connections == 5 + + +def test_resilient_client_kwargs_defaults_and_overrides(): + defaults = resilient_client_kwargs() + assert defaults["health_check_interval"] == 30 + assert defaults["max_connections"] == 100 + + overridden = resilient_client_kwargs(socket_timeout=0.5, decode_responses=True) + assert overridden["socket_timeout"] == 0.5 + assert overridden["decode_responses"] is True + # Untouched defaults survive an override. + assert overridden["health_check_interval"] == 30 + + +def test_connection_built_from_pool_carries_resilience(): + """The settings must reach the Connection object, not just the kwargs dict.""" + backend = get_cache("redis", "from_credentials", host="localhost") + conn = backend._client.connection_pool.make_connection() + assert conn.health_check_interval == 30 + assert conn.retry.get_retries() == 3 + + +@pytest.mark.parametrize( + "backend_cls", + [StandaloneRedisBackend, AWSElastiCacheBackend, AzureRedisCacheBackend], +) +def test_every_factory_accepts_client_kwargs(backend_cls): + """Guards the factories that need cloud credentials to actually construct. + + A factory without **client_kwargs cannot be forwarding the resilience + defaults, so this catches a new auth method that forgets them. + """ + factories = [ + name + for name in dir(backend_cls) + if name.startswith("from_") + and isinstance(inspect.getattr_static(backend_cls, name), classmethod) + ] + assert factories, f"no factories found on {backend_cls.__name__}" + + for name in factories: + params = inspect.signature(getattr(backend_cls, name)).parameters.values() + assert any(p.kind is inspect.Parameter.VAR_KEYWORD for p in params), ( + f"{backend_cls.__name__}.{name} does not accept **client_kwargs, so it " + "cannot forward the connection-resilience defaults" + ) diff --git a/tests/test_messaging_azure.py b/tests/test_messaging_azure.py index 8abb423..53409d4 100644 --- a/tests/test_messaging_azure.py +++ b/tests/test_messaging_azure.py @@ -58,8 +58,10 @@ def _sending_backend(session_enabled=False, sender=None): class _FakeReceivedMessage: """Minimal stand-in for ServiceBusReceivedMessage. - The real type exposes its raw payload via ``bytes(message)``; emulate that - deterministically (MagicMock does not reliably support ``__bytes__``). + Mirrors the real type's payload access: ``.body`` yields the raw bytes in + chunks and is single-use, while ``bytes(message)`` raises TypeError. An + earlier version of this fake implemented ``__bytes__``, which let a + ``bytes(m)`` conversion pass here and fail against a live broker. """ def __init__( @@ -73,9 +75,15 @@ def __init__( self.enqueued_time_utc = "2026-01-01" self.application_properties = application_properties self._body = body - - def __bytes__(self): - return self._body + self._body_consumed = False + + @property + def body(self): + if self._body_consumed: + raise RuntimeError("ServiceBusReceivedMessage.body is single-use") + self._body_consumed = True + # The real SDK yields the payload in chunks. + return iter([self._body[:1], self._body[1:]] if self._body else []) def _make_received_message(