fix(messaging): Azure receive() could never decode a body; Redis resilience for managed cache

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"]

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."""

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** — ``<name>.redis.cache.windows.net`` on 6380,
+          password-only. The defaults here target it.
+        - **Azure Managed Redis** — ``<name>.<region>.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. ``<name>.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:

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: