chore: add opt-in retry for rate-limited and transport-failed requests

README.md

@@ -112,6 +112,18 @@ client = Stream(api_key="key", api_secret="secret", logger=logging.getLogger("my
 
 Each event carries structured fields via the standard `extra={}` mechanism (for example `http.response.status_code`, `duration_ms`, `stream.endpoint_name`). Query and body values for known secret keys (`api_key`, `api_secret`, `token`, `password`) are always redacted. Request/response bodies are omitted by default; pass `log_bodies=True` to include them (still redacted, and this emits one WARNING at construction since bodies can contain other sensitive data).
 
+### Retries
+
+By default the client makes exactly one attempt per request and surfaces errors unchanged. Pass a `RetryConfig` to opt in to auto-retry:
+
+```python
+from getstream import Stream, RetryConfig
+
+client = Stream(api_key=..., api_secret=..., retry=RetryConfig(enabled=True, max_attempts=3, max_backoff=30.0))
+```
+
+Only idempotent `GET`/`HEAD` requests are retried, and only on HTTP 429 (unless the backend marked it unrecoverable) or a transport-level failure (timeout, connection reset, DNS, TLS). A 429's `Retry-After` header is honored (clamped to `max_backoff`); otherwise the delay uses full jitter over an exponential backoff. A retried failure logs `http.request.failed` at DEBUG; a final, non-retried failure logs it at ERROR (or not at all for a final 429, since that's already covered by `http.response.received`).
+
 ### App configuration
 
 ```python

getstream/__init__.py

@@ -1,5 +1,6 @@
 import logging
 
+from getstream.config import RetryConfig  # noqa: F401
 from getstream.exceptions import (  # noqa: F401
     StreamApiException,
     StreamException,

getstream/base.py

@@ -2,6 +2,7 @@
 import logging
 import mimetypes
 import os
+import random
 import time
 import uuid
 import warnings
@@ -10,6 +11,8 @@
 
 from getstream.exceptions import (
     StreamApiException,
+    StreamRateLimitException,
+    StreamTransportException,
     build_api_exception,
     wrap_transport_error,
 )
@@ -44,6 +47,34 @@
 logger = logging.getLogger("getstream")
 
 
+# ── Retry policy (CHA-2959) ───────────────────────────────────────────
+def _retry_eligible(retry, exc, method: str, attempt: int) -> bool:
+    """Whether ``exc`` from the given 0-indexed ``attempt`` should be retried
+    under ``retry`` (a ``RetryConfig`` or ``None``). Only GET/HEAD, only HTTP
+    429 (unless marked unrecoverable) or a transport error, and only while
+    attempts remain."""
+    if retry is None or not retry.enabled:
+        return False
+    if method.upper() not in ("GET", "HEAD"):
+        return False
+    if attempt + 1 >= retry.max_attempts:
+        return False
+    if isinstance(exc, StreamRateLimitException):
+        return not bool(getattr(exc, "unrecoverable", False))
+    return isinstance(exc, StreamTransportException)
+
+
+def _retry_delay(retry, exc, attempt: int) -> float:
+    """Seconds to sleep before the next attempt: honors the server's
+    ``Retry-After`` when present (clamped to ``max_backoff``), else full
+    jitter over an exponential ceiling (``attempt`` is 0-indexed)."""
+    retry_after = getattr(exc, "retry_after", None)
+    if retry_after is not None and retry_after.total_seconds() > 0:
+        return min(retry_after.total_seconds(), retry.max_backoff)
+    ceil = min(retry.max_backoff, float(2**attempt))
+    return random.uniform(0.0, ceil) if ceil > 0 else 0.0
+
+
 def _resolve_pool_knobs(obj):
     """Pull the 3 pool knobs off ``obj`` if BaseStream has set them, else fall back to spec defaults. Top-level ``Stream``/``AsyncStream`` sets them on ``self`` before calling ``super().__init__()``, so a directly instantiated sub-client (or test fixture) still gets sane values.
 
@@ -294,7 +325,7 @@ def _endpoint_name(self, path: str) -> str:
         op = getattr(self, "_operation_name", None)
         return op or current_operation(self._normalize_endpoint_from_path(path)) or ""
 
-    def _request_sync(
+    def _attempt_sync(
         self,
         method: str,
         path: str,
@@ -334,19 +365,10 @@ def _request_sync(
                     url_path, params=query_params, *args, **call_kwargs
                 )
             except httpx.RequestError as err:
-                exc = wrap_transport_error(err)
-                log.error(
-                    "http.request.failed",
-                    extra={
-                        "http.request.method": method,
-                        "url.path": path,
-                        "stream.endpoint_name": endpoint,
-                        "error.type": exc.error_type,
-                        "error.message": str(err),
-                        "duration_ms": int((time.perf_counter() - start) * 1000.0),
-                    },
-                )
-                raise exc from err
+                # No failed-log here: the retry loop (_request_sync) owns
+                # http.request.failed so it can log at DEBUG when retrying
+                # and ERROR only on a final failure.
+                raise wrap_transport_error(err) from err
             duration = parse_duration_from_body(response.content)
             if duration:
                 span.set_attribute("http.server.duration", duration)
@@ -379,6 +401,72 @@ def _request_sync(
             record_metrics(duration_ms, attributes=metric_attrs)
             return self._parse_response(response, data_type or Dict[str, Any])
 
+    def _request_sync(
+        self,
+        method: str,
+        path: str,
+        *,
+        query_params=None,
+        args=(),
+        kwargs=None,
+        data_type: Optional[Type[T]] = None,
+    ):
+        """Retry loop around ``_attempt_sync``. Disabled (default) retry
+        policy means exactly one attempt, errors surface unchanged. When
+        enabled, retries GET/HEAD on HTTP 429 / transport errors per
+        ``_retry_eligible``/``_retry_delay``, owning the ``http.request.failed``
+        log level so a retried failure logs at DEBUG and only a final
+        transport failure logs at ERROR (a final 429 is already covered by
+        ``http.response.received``)."""
+        retry = getattr(self, "retry", None)
+        log = _resolve_logger(self)
+        endpoint = self._endpoint_name(path)
+        attempt = 0
+        while True:
+            t0 = time.perf_counter()
+            try:
+                return self._attempt_sync(
+                    method,
+                    path,
+                    query_params=query_params,
+                    args=args,
+                    kwargs=kwargs,
+                    data_type=data_type,
+                )
+            except (StreamRateLimitException, StreamTransportException) as exc:
+                duration_ms = int((time.perf_counter() - t0) * 1000)
+                if _retry_eligible(retry, exc, method, attempt):
+                    delay = _retry_delay(retry, exc, attempt)
+                    extra = {
+                        "http.request.method": method,
+                        "url.path": path,
+                        "stream.endpoint_name": endpoint,
+                        "retry.attempt": attempt + 1,
+                        "backoff_seconds": round(delay, 3),
+                        "error.message": str(exc.__cause__ or exc),
+                        "duration_ms": duration_ms,
+                    }
+                    if isinstance(exc, StreamTransportException):
+                        extra["error.type"] = exc.error_type
+                    log.debug("http.request.failed", extra=extra)
+                    time.sleep(delay)
+                    attempt += 1
+                    continue
+                if isinstance(exc, StreamTransportException):
+                    log.error(
+                        "http.request.failed",
+                        extra={
+                            "http.request.method": method,
+                            "url.path": path,
+                            "stream.endpoint_name": endpoint,
+                            "retry.attempt": attempt + 1,
+                            "error.type": exc.error_type,
+                            "error.message": str(exc.__cause__ or exc),
+                            "duration_ms": duration_ms,
+                        },
+                    )
+                raise
+
     def patch(
         self,
         path,
@@ -637,7 +725,7 @@ def _endpoint_name(self, path: str) -> str:
         op = getattr(self, "_operation_name", None)
         return op or current_operation(self._normalize_endpoint_from_path(path)) or ""
 
-    async def _request_async(
+    async def _attempt_async(
         self,
         method: str,
         path: str,
@@ -685,19 +773,10 @@ async def _request_async(
                     url_path, params=query_params, *args, **call_kwargs
                 )
             except httpx.RequestError as err:
-                exc = wrap_transport_error(err)
-                log.error(
-                    "http.request.failed",
-                    extra={
-                        "http.request.method": method,
-                        "url.path": path,
-                        "stream.endpoint_name": endpoint,
-                        "error.type": exc.error_type,
-                        "error.message": str(err),
-                        "duration_ms": int((time.perf_counter() - start) * 1000.0),
-                    },
-                )
-                raise exc from err
+                # No failed-log here: the retry loop (_request_async) owns
+                # http.request.failed so it can log at DEBUG when retrying
+                # and ERROR only on a final failure.
+                raise wrap_transport_error(err) from err
             duration = parse_duration_from_body(response.content)
             if duration:
                 span.set_attribute("http.server.duration", duration)
@@ -734,6 +813,66 @@ async def _request_async(
                 self._parse_response, response, data_type or Dict[str, Any]
             )
 
+    async def _request_async(
+        self,
+        method: str,
+        path: str,
+        *,
+        query_params=None,
+        args=(),
+        kwargs=None,
+        data_type: Optional[Type[T]] = None,
+    ):
+        """Async twin of ``BaseClient._request_sync``; see that docstring."""
+        retry = getattr(self, "retry", None)
+        log = _resolve_logger(self)
+        endpoint = self._endpoint_name(path)
+        attempt = 0
+        while True:
+            t0 = time.perf_counter()
+            try:
+                return await self._attempt_async(
+                    method,
+                    path,
+                    query_params=query_params,
+                    args=args,
+                    kwargs=kwargs,
+                    data_type=data_type,
+                )
+            except (StreamRateLimitException, StreamTransportException) as exc:
+                duration_ms = int((time.perf_counter() - t0) * 1000)
+                if _retry_eligible(retry, exc, method, attempt):
+                    delay = _retry_delay(retry, exc, attempt)
+                    extra = {
+                        "http.request.method": method,
+                        "url.path": path,
+                        "stream.endpoint_name": endpoint,
+                        "retry.attempt": attempt + 1,
+                        "backoff_seconds": round(delay, 3),
+                        "error.message": str(exc.__cause__ or exc),
+                        "duration_ms": duration_ms,
+                    }
+                    if isinstance(exc, StreamTransportException):
+                        extra["error.type"] = exc.error_type
+                    log.debug("http.request.failed", extra=extra)
+                    await asyncio.sleep(delay)
+                    attempt += 1
+                    continue
+                if isinstance(exc, StreamTransportException):
+                    log.error(
+                        "http.request.failed",
+                        extra={
+                            "http.request.method": method,
+                            "url.path": path,
+                            "stream.endpoint_name": endpoint,
+                            "retry.attempt": attempt + 1,
+                            "error.type": exc.error_type,
+                            "error.message": str(exc.__cause__ or exc),
+                            "duration_ms": duration_ms,
+                        },
+                    )
+                raise
+
     async def patch(
         self,
         path,

getstream/config.py

@@ -1,6 +1,26 @@
+from dataclasses import dataclass
+
 from getstream.version import VERSION
 
 
+@dataclass(frozen=True)
+class RetryConfig:
+    """Opt-in auto-retry policy. Disabled by default: the client performs
+    exactly one attempt and surfaces errors unchanged. When enabled, only
+    GET/HEAD requests failing with HTTP 429 or a transport error are retried,
+    and never when the backend marked the error unrecoverable."""
+
+    enabled: bool = False
+    max_attempts: int = 3
+    max_backoff: float = 30.0
+
+    def __post_init__(self):
+        if self.max_attempts < 1:
+            raise ValueError("max_attempts must be >= 1")
+        if self.max_backoff < 0:
+            raise ValueError("max_backoff must be >= 0")
+
+
 class BaseConfig:
     def __init__(
         self,

getstream/stream.py

@@ -13,6 +13,7 @@
 
 from getstream.base import _log_client_initialized, _resolve_logger
 from getstream.common import telemetry
+from getstream.config import RetryConfig
 from getstream.chat.client import ChatClient
 from getstream.chat.async_client import ChatClient as AsyncChatClient
 from getstream.common.async_client import CommonClient as AsyncCommonClient
@@ -95,6 +96,7 @@ def __init__(
         connect_timeout: Optional[float] = None,
         logger: Optional[logging.Logger] = None,
         log_bodies: bool = False,
+        retry: Optional[RetryConfig] = None,
     ):
         """Build a Stream client.
 
@@ -119,6 +121,7 @@ def __init__(
             connect_timeout: TCP + TLS handshake timeout in seconds. Default 10.0. Ignored when ``http_client`` is set.
             logger: Optional stdlib ``logging.Logger`` for the SDK's structured log events (``client.initialized``, ``http.request.sent``, ``http.response.received``, ``http.request.failed``). Defaults to ``logging.getLogger("getstream")``, which is a no-op until the caller attaches a handler.
             log_bodies: When ``True``, adds redacted request/response bodies to the request/response log events. Off by default. Emits one WARNING at construction when enabled.
+            retry: Optional ``RetryConfig`` enabling auto-retry of GET/HEAD requests on HTTP 429 or transport errors. Disabled by default (a single attempt; errors surface unchanged).
 
         Raises:
             ValueError: If both ``transport`` and ``http_client`` are set; if neither ``api_secret`` nor ``token`` can be resolved; if both are provided; if either is the empty string; if ``api_key`` is missing; or if ``request_timeout`` is not a positive number.
@@ -201,6 +204,11 @@ def _settings() -> _PoolSettings:
         # sub-clients in _apply_shared_client.
         self.log = logger
         self.log_bodies = log_bodies
+        # retry: same getattr(self, ...) plumbing as the pool knobs and log/
+        # log_bodies above, since the intermediate generated REST clients do
+        # not forward this kwarg either. Read by BaseClient/AsyncBaseClient's
+        # request loop and copied onto sub-clients in _apply_shared_client.
+        self.retry = retry
         # Pool knobs are read by BaseClient via getattr(self, ...) since the intermediate generated REST clients (CommonRestClient etc.) do not forward these kwargs. self.max_conns_per_host / idle_timeout / connect_timeout were set above before super().__init__().
         super().__init__(
             self.api_key, self.base_url, self.token, self.timeout, self.user_agent
@@ -259,6 +267,7 @@ def _apply_shared_client(self, sub_client):
         # through the caller's logger instead of silently falling back.
         sub_client.log = getattr(self, "log", None)
         sub_client.log_bodies = getattr(self, "log_bodies", False)
+        sub_client.retry = getattr(self, "retry", None)
         return sub_client
 
     def create_token(