refactor: common retry handling

CHANGELOG.md

@@ -8,6 +8,7 @@
 * (Major): The constructors raise `ValueError` if `api_key`, `app_name` or `IdentityOptions.instance_id` contains a line break, such as an API token read from a file with a trailing newline, and `TypeError` if one isn't a str. These values are sent as headers.
 * (Major): `initialize_client()` no longer raises when registration fails. A failed registration is logged and the client keeps running, the same way for both clients.
 * (Major): Upgraded to yggdrasil-engine 3, new engine built with PyO3 for better performance.
+* (Minor): Feature fetches are retried the same way by both clients: on a 500, 502 or 504, or when no response arrives, with a short randomised backoff between attempts (up to about 0.5s, 1s and 2s). Before, retries were immediate.
 * (Minor): New `FETCH_COMPLETED` event, emitted after every feature fetch with how it ended: `CHANGED`, `NOT_CHANGED`, `UNAUTHORIZED`, `RATE_LIMITED`, `UNAVAILABLE` or `INVALID_RESPONSE`.
 * (Minor): `feature_definitions()` reports whether each feature is enabled in the environment.
 * (Minor): New `evaluate_all_toggles()` evaluates every feature for one context in a single call, without recording usage metrics.

UnleashClient/_async_unleash_api_client.py

@@ -8,6 +8,7 @@
 from UnleashClient._aiohttp_transport import AiohttpTransport
 from UnleashClient._event_dispatcher import _EventDispatcher
 from UnleashClient._headers import _HeaderFactory
+from UnleashClient._retry import _send_with_retries_async
 from UnleashClient._transport import AsyncTransport, Request
 from UnleashClient._unleash_api_client import (
     FetchResult,
@@ -24,18 +25,6 @@
 from UnleashClient.events import FetchStatus
 from UnleashClient.utils import LOGGER
 
-try:
-    import aiohttp
-except ImportError as exc:  # pragma: no cover
-    raise ImportError(
-        "AsyncUnleashClient requires aiohttp, which is not installed. "
-        "Install it with: pip install UnleashClient[async]"
-    ) from exc
-
-RETRY_STATUSES = frozenset({500, 502, 504})
-
-RETRYABLE_ERRORS = (aiohttp.ClientConnectionError, asyncio.TimeoutError)
-
 
 class _AsyncUnleashApiClient:
     """The asyncio twin of :class:`UnleashClient._unleash_api_client._UnleashApiClient`."""
@@ -133,47 +122,38 @@ async def _fetch_features(
             if options.project_name:
                 base_params = {"project": options.project_name}
 
-            request = Request(
-                method="GET", url=base_url, headers=headers, params=base_params
+            resp = await _send_with_retries_async(
+                self._transport,
+                Request(
+                    method="GET", url=base_url, headers=headers, params=base_params
+                ),
+                config.http.retries,
             )
-            attempts = 1 + max(config.http.retries, 0)
-
-            for attempt in range(attempts):
-                last_attempt = attempt == attempts - 1
-                try:
-                    resp = await self._transport.execute(request)
-                except RETRYABLE_ERRORS:
-                    if last_attempt:
-                        raise
-                    continue
-
-                if resp.status in RETRY_STATUSES and not last_attempt:
-                    continue
-
-                http_status_code = resp.status
-                if resp.status not in [200, 304]:
-                    _log_resp_info(resp)
-                    LOGGER.warning(
-                        "Unleash Client feature fetch failed due to unexpected HTTP status code: %s",
-                        resp.status,
-                    )
-                    raise Exception(  # pylint: disable=broad-exception-raised
-                        "Unleash Client feature fetch failed!"
-                    )
-
-                fetched_etag = resp.headers.get("etag", "")
-
-                if resp.status == 304:
-                    return (
-                        FetchResult(None, fetched_etag, not_modified=True),
-                        FetchStatus.NOT_CHANGED,
-                    )
-
-                raw_state = _text(resp)
+
+            http_status_code = resp.status
+            if resp.status not in [200, 304]:
+                _log_resp_info(resp)
+                LOGGER.warning(
+                    "Unleash Client feature fetch failed due to unexpected HTTP status code: %s",
+                    resp.status,
+                )
+                raise Exception(  # pylint: disable=broad-exception-raised
+                    "Unleash Client feature fetch failed!"
+                )
+
+            fetched_etag = resp.headers.get("etag", "")
+
+            if resp.status == 304:
                 return (
-                    FetchResult(raw_state, fetched_etag),
-                    _fetched_status(raw_state),
+                    FetchResult(None, fetched_etag, not_modified=True),
+                    FetchStatus.NOT_CHANGED,
                 )
+
+            raw_state = _text(resp)
+            return (
+                FetchResult(raw_state, fetched_etag),
+                _fetched_status(raw_state),
+            )
         except Exception as exc:
             LOGGER.exception(
                 "Unleash Client feature fetch failed due to exception: %s", exc

UnleashClient/_retry.py

@@ -0,0 +1,106 @@
+"""
+The retry policy for feature fetches, shared by the sync and async clients.
+
+Transports send each request once; retrying is decided here. Register and
+metrics are never retried.
+"""
+
+import asyncio
+import random
+import time
+from typing import Union
+
+from UnleashClient._transport import AsyncTransport, Request, Response, Transport
+from UnleashClient.errors import AlreadyClosedError
+from UnleashClient.utils import LOGGER
+
+RETRY_STATUSES = frozenset({500, 502, 504})
+BACKOFF_BASE = 0.5
+
+_sleep = time.sleep
+_async_sleep = asyncio.sleep
+
+
+def _backoff_delay(retry: int) -> float:
+    """
+    Seconds to wait before retry number ``retry`` (0 for the first).
+
+    Exponential with jitter: a random point in the upper half of
+    ``BACKOFF_BASE * 2 ** retry``, so 0.25-0.5s, 0.5-1s, 1-2s, and clients that
+    failed together don't retry together.
+    """
+    ceiling = BACKOFF_BASE * 2**retry
+    return random.uniform(ceiling / 2, ceiling)
+
+
+_Outcome = Union[Response, Exception]
+
+
+def _attempts(retries: int) -> range:
+    """The attempt numbers: the first attempt and ``retries`` retries."""
+    return range(1 + max(retries, 0))
+
+
+def _should_retry(attempt: int, retries: int, outcome: _Outcome) -> bool:
+    if attempt >= max(retries, 0):
+        return False
+    if isinstance(outcome, Response):
+        return outcome.status in RETRY_STATUSES
+    return not isinstance(outcome, AlreadyClosedError)
+
+
+def _next_step(attempt: int, retries: int, outcome: _Outcome) -> Union[Response, float]:
+    if not _should_retry(attempt, retries, outcome):
+        if isinstance(outcome, Exception):
+            raise outcome
+        return outcome
+
+    delay = _backoff_delay(attempt)
+    reason = (
+        f"HTTP status {outcome.status}"
+        if isinstance(outcome, Response)
+        else repr(outcome)
+    )
+    LOGGER.debug("Retrying feature fetch in %.2fs after %s", delay, reason)
+    return delay
+
+
+def _send_with_retries(
+    transport: Transport, request: Request, retries: int
+) -> Response:
+    """
+    Send ``request``, retrying on a status in :data:`RETRY_STATUSES` and on any
+    exception, with a backoff between attempts.ca
+
+    :param retries: Retries after the first attempt; negative counts as 0.
+    :return: The last response, which can still have a retry status.
+    :raises Exception: The last attempt's exception, if it raised.
+    """
+    for attempt in _attempts(retries):
+        try:
+            outcome: _Outcome = transport.execute(request)
+        except Exception as exc:  # pylint: disable=broad-except
+            outcome = exc
+        step = _next_step(attempt, retries, outcome)
+        if isinstance(step, Response):
+            return step
+        _sleep(step)
+
+    raise AssertionError("unreachable: the last attempt returns or raises")
+
+
+async def _send_with_retries_async(
+    transport: AsyncTransport, request: Request, retries: int
+) -> Response:
+    """The asyncio twin of :func:`_send_with_retries`: only the awaits differ."""
+    for attempt in _attempts(retries):
+        try:
+            outcome: _Outcome = await transport.execute(request)
+        except Exception as exc:  # pylint: disable=broad-except
+            outcome = exc
+        step = _next_step(attempt, retries, outcome)
+        if isinstance(step, Response):
+            return step
+        await _async_sleep(step)
+
+    raise AssertionError("unreachable: the last attempt returns or raises")

UnleashClient/_transport.py

@@ -7,8 +7,6 @@
 from typing import Any, Mapping, Optional, Protocol, runtime_checkable
 
 import requests
-from requests.adapters import HTTPAdapter
-from urllib3 import Retry
 
 from UnleashClient.constants import REQUEST_TIMEOUT
 
@@ -78,22 +76,20 @@ class RequestsTransport:
     """
     Sends requests with ``requests``, on a fresh session for every request.
 
+    Sends each request once: retrying is the API client's job.
+
     :param timeout: Seconds to wait for the connection and for each read.
-    :param _retries: GET requests are retried this many times on a connection
-                     error or a 500, 502 or 504, by urllib3.
     :param _request_kwargs: extra keyword arguments for every ``requests`` call.
     """
 
     def __init__(
         self,
         *,
         timeout: float = REQUEST_TIMEOUT,
-        # these two are are going away too
-        _retries: Optional[int] = None,
+        # going away with HttpOptions
         _request_kwargs: Optional[Mapping[str, Any]] = None,
     ) -> None:
         self._timeout = timeout
-        self._retries = _retries
         self._request_kwargs = dict(_request_kwargs or {})
 
     def execute(self, request: Request) -> Response:
@@ -104,15 +100,6 @@ def execute(self, request: Request) -> Response:
             kwargs["data"] = request.body
 
         with requests.Session() as session:
-            if request.method == "GET" and self._retries is not None:
-                adapter = HTTPAdapter(
-                    max_retries=Retry(
-                        total=self._retries, status_forcelist=[500, 502, 504]
-                    )
-                )
-                session.mount("https://", adapter)
-                session.mount("http://", adapter)
-
             resp = session.request(
                 request.method, request.url, **kwargs, **self._request_kwargs
             )

UnleashClient/_unleash_api_client.py

@@ -4,6 +4,7 @@
 
 from UnleashClient._event_dispatcher import _EventDispatcher
 from UnleashClient._headers import _HeaderFactory
+from UnleashClient._retry import _send_with_retries
 from UnleashClient._transport import Request, RequestsTransport, Response, Transport
 from UnleashClient.config import PollingOptions, UnleashConfig
 from UnleashClient.constants import FEATURES_URL, METRICS_URL, REGISTER_URL
@@ -109,7 +110,6 @@ def __init__(
 
         self._transport: Transport = transport or RequestsTransport(
             timeout=config.http.timeout,
-            _retries=config.http.retries,
             # this is going to go away soon, just like the whole httpOptions in the client constructor
             _request_kwargs=config.http.options,
         )
@@ -157,7 +157,8 @@ def _fetch_features(
             if options.project_name:
                 base_params = {"project": options.project_name}
 
-            resp = self._transport.execute(
+            resp = _send_with_retries(
+                self._transport,
                 Request(
                     method="GET",
                     url=base_url,
@@ -166,7 +167,8 @@ def _fetch_features(
                         **request_specific_headers,
                     },
                     params=base_params,
-                )
+                ),
+                config.http.retries,
             )
 
             http_status_code = resp.status

tests/conftest.py

@@ -79,3 +79,8 @@ def cache_segments(tmp_path):
     )
     yield temporary_cache
     temporary_cache.destroy()
+
+
+@pytest.fixture(autouse=True)
+def no_retry_backoff(monkeypatch):
+    monkeypatch.setattr("UnleashClient._retry._backoff_delay", lambda retry: 0)