diff --git a/CHANGELOG.md b/CHANGELOG.md index 4a441e85..58ea5af6 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/UnleashClient/_async_unleash_api_client.py b/UnleashClient/_async_unleash_api_client.py index d7127fa6..6b08b4c7 100644 --- a/UnleashClient/_async_unleash_api_client.py +++ b/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 diff --git a/UnleashClient/_retry.py b/UnleashClient/_retry.py new file mode 100644 index 00000000..8cbcbcf6 --- /dev/null +++ b/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") diff --git a/UnleashClient/_transport.py b/UnleashClient/_transport.py index ac647819..4fb38ecb 100644 --- a/UnleashClient/_transport.py +++ b/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,9 +76,9 @@ 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. """ @@ -88,12 +86,10 @@ 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 ) diff --git a/UnleashClient/_unleash_api_client.py b/UnleashClient/_unleash_api_client.py index 81a5bdbc..77696373 100644 --- a/UnleashClient/_unleash_api_client.py +++ b/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 diff --git a/tests/conftest.py b/tests/conftest.py index 6a7c994e..0b6e71e1 100644 --- a/tests/conftest.py +++ b/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) diff --git a/tests/unit_tests/test_retry.py b/tests/unit_tests/test_retry.py new file mode 100644 index 00000000..990e30e6 --- /dev/null +++ b/tests/unit_tests/test_retry.py @@ -0,0 +1,180 @@ +import asyncio + +import pytest +from pytest import mark, param + +from UnleashClient import _retry +from UnleashClient._retry import ( + BACKOFF_BASE, + _send_with_retries, + _send_with_retries_async, +) +from UnleashClient._transport import Request, Response +from UnleashClient.errors import AlreadyClosedError + +# Imported before the autouse no_retry_backoff fixture replaces it. +_real_backoff_delay = _retry._backoff_delay + +REQUEST = Request(method="GET", url="http://localhost/api/client/features", headers={}) + + +def reply(status: int) -> Response: + return Response(status=status, headers={}, body=b"") + + +class SequenceTransport: + def __init__(self, *outcomes): + self.outcomes = list(outcomes) + self.calls = 0 + + def _next(self) -> Response: + outcome = self.outcomes[min(self.calls, len(self.outcomes) - 1)] + self.calls += 1 + if isinstance(outcome, BaseException): + raise outcome + return outcome + + +class SyncSequenceTransport(SequenceTransport): + def execute(self, request: Request) -> Response: + return self._next() + + +class AsyncSequenceTransport(SequenceTransport): + async def execute(self, request: Request) -> Response: + return self._next() + + +def send(flavor: str, outcomes, retries: int): + """Runs the sync or the async policy over ``outcomes``; returns (outcome, calls).""" + if flavor == "sync": + transport = SyncSequenceTransport(*outcomes) + outcome = _send_with_retries(transport, REQUEST, retries) + else: + transport = AsyncSequenceTransport(*outcomes) + outcome = asyncio.run(_send_with_retries_async(transport, REQUEST, retries)) + return outcome, transport.calls + + +both_flavors = mark.parametrize("flavor", ["sync", "async"]) + + +@both_flavors +@mark.parametrize( + "outcomes,retries,expected_status,expected_calls", + ( + param([reply(200)], 3, 200, 1, id="success-first-time"), + param([reply(500), reply(200)], 3, 200, 2, id="500-then-success"), + param([reply(502), reply(200)], 3, 200, 2, id="502-then-success"), + param([reply(504), reply(200)], 3, 200, 2, id="504-then-success"), + param([OSError("refused"), reply(200)], 3, 200, 2, id="error-then-success"), + param([reply(500)], 3, 500, 4, id="500-every-time-returns-the-last"), + param([reply(500)], 0, 500, 1, id="no-retries"), + param([reply(304)], 3, 304, 1, id="304-not-retried"), + param([reply(401)], 3, 401, 1, id="401-not-retried"), + param([reply(429)], 3, 429, 1, id="429-not-retried"), + param([reply(503)], 3, 503, 1, id="503-not-retried"), + ), +) +def test_retry_policy(flavor, outcomes, retries, expected_status, expected_calls): + response, calls = send(flavor, outcomes, retries) + + assert response.status == expected_status + assert calls == expected_calls + + +class ForeignTransportError(Exception): + """Stands for an error of any HTTP library the policy knows nothing about.""" + + +@both_flavors +@mark.parametrize( + "error", + (ForeignTransportError("boom"), OSError("refused"), TimeoutError()), + ids=("foreign", "os-error", "timeout"), +) +def test_any_exception_is_retried_and_the_last_one_raised(flavor, error): + with pytest.raises(type(error)): + send(flavor, [error], 3) + + +def test_an_exception_on_every_attempt_is_tried_retries_plus_one_times(): + transport = SyncSequenceTransport(OSError("refused")) + + with pytest.raises(OSError): + _send_with_retries(transport, REQUEST, 3) + + assert transport.calls == 4 + + +@mark.asyncio +async def test_an_exception_on_every_attempt_is_tried_retries_plus_one_times_async(): + transport = AsyncSequenceTransport(OSError("refused")) + + with pytest.raises(OSError): + await _send_with_retries_async(transport, REQUEST, 3) + + assert transport.calls == 4 + + +@both_flavors +def test_a_closed_transport_is_not_retried(flavor): + with pytest.raises(AlreadyClosedError): + send(flavor, [AlreadyClosedError()], 3) + + +def test_a_cancelled_fetch_is_not_retried(): + transport = AsyncSequenceTransport(asyncio.CancelledError()) + + with pytest.raises(asyncio.CancelledError): + asyncio.run(_send_with_retries_async(transport, REQUEST, 3)) + + assert transport.calls == 1 + + +# backoff + + +@both_flavors +def test_the_backoff_waits_between_attempts_but_not_after_the_last(flavor, monkeypatch): + slept = [] + + async def async_sleep(delay): + slept.append(delay) + + monkeypatch.setattr(_retry, "_backoff_delay", lambda retry: retry + 1) + monkeypatch.setattr(_retry, "_sleep", slept.append) + monkeypatch.setattr(_retry, "_async_sleep", async_sleep) + + send(flavor, [reply(500)], 3) + + assert slept == [1, 2, 3] + + +@both_flavors +def test_no_backoff_without_a_retry(flavor, monkeypatch): + slept = [] + + async def async_sleep(delay): + slept.append(delay) + + monkeypatch.setattr(_retry, "_sleep", slept.append) + monkeypatch.setattr(_retry, "_async_sleep", async_sleep) + + send(flavor, [reply(200)], 3) + + assert slept == [] + + +@mark.parametrize("retry", [0, 1, 2]) +def test_the_backoff_doubles_with_jitter_in_its_upper_half(retry): + ceiling = BACKOFF_BASE * 2**retry + + delays = [_real_backoff_delay(retry) for _ in range(200)] + + assert all(ceiling / 2 <= delay <= ceiling for delay in delays) + assert len(set(delays)) > 1 + + +def test_the_default_backoff_adds_at_most_3_5_seconds(): + assert sum(BACKOFF_BASE * 2**retry for retry in range(3)) == 3.5 diff --git a/tests/unit_tests/test_transport.py b/tests/unit_tests/test_transport.py index 66e37548..60748cb4 100644 --- a/tests/unit_tests/test_transport.py +++ b/tests/unit_tests/test_transport.py @@ -89,24 +89,13 @@ def test_execute_raises_when_no_response_arrives(): @responses.activate -def test_execute_retries_a_get_when_retries_are_set(): - responses.add(responses.GET, ENDPOINT, status=500) - responses.add(responses.GET, ENDPOINT, body=b"{}", status=200) - - response = RequestsTransport(_retries=1).execute( - Request(method="GET", url=ENDPOINT, headers={}) - ) - - assert len(responses.calls) == 2 - assert response.status == 200 - - -@responses.activate -def test_execute_does_not_retry_without_retries(): - responses.add(responses.GET, ENDPOINT, status=500) +@pytest.mark.parametrize("method", ["GET", "POST"]) +def test_execute_sends_once_even_on_a_retry_status(method): + responses.add(method, ENDPOINT, status=500) + responses.add(method, ENDPOINT, status=200) response = RequestsTransport().execute( - Request(method="GET", url=ENDPOINT, headers={}) + Request(method=method, url=ENDPOINT, headers={}) ) assert len(responses.calls) == 1 @@ -114,15 +103,14 @@ def test_execute_does_not_retry_without_retries(): @responses.activate -def test_execute_never_retries_a_post(): - responses.add(responses.POST, ENDPOINT, status=500) +def test_execute_sends_once_on_a_connection_error(): + responses.add(responses.GET, ENDPOINT, body=ConnectionError("refused")) + responses.add(responses.GET, ENDPOINT, status=200) - response = RequestsTransport(_retries=3).execute( - Request(method="POST", url=ENDPOINT, headers={}, body=b"{}") - ) + with pytest.raises(ConnectionError): + RequestsTransport().execute(Request(method="GET", url=ENDPOINT, headers={})) assert len(responses.calls) == 1 - assert response.status == 500 @responses.activate