From 5d7913fb4ffc563d467f0a97723959e2cc1c54f7 Mon Sep 17 00:00:00 2001 From: Caleigh Runge-Hottman Date: Thu, 5 Nov 2026 16:27:10 -0500 Subject: [PATCH] Re-sign each retry [RHELDST-47596, RHELDST-47601] Retries were performed by urllib3's Retry on the HTTPAdapter, which resends the request below the requests auth layer -- so EdgeGridAuth signed once and every retry reused the same Authorization header. Once a request's signature aged past Akamai's ~30s timestamp tolerance, every retry was rejected with 400 "Invalid timestamp", regardless of whether the original rate limiting had cleared, until the retry budget was exhausted and the commit failed. Replace the transport-level retry with an application-level retry loop in __start_purge: each attempt re-issues session.post(), so EdgeGridAuth re-signs with a current timestamp and a retry can succeed once the rate limit clears. - compute_backoff() mirrors urllib3's exponential backoff (first retry 0, then factor*2**(n-1) + jitter, capped), so the FAST_PURGE_RETRY_* tunables keep their meaning. - Retryable statuses are an explicit allowlist (anything else fails fast), configurable via FAST_PURGE_RETRY_STATUSES. The default covers 429, all 5xx, and the EdgeGrid-specific 400 (returned for a stale timestamp and for WAF rate-control, both of which recover on a fresh attempt). - Connection/timeout errors are retried. - Failed attempts now log the response body and Retry-After, so the actual Akamai error (e.g. "Invalid timestamp") is visible. - A non-JSON 201 is tolerated rather than raising. Co-Authored-By: Claude Opus 4.8 --- fastpurge/_client.py | 178 +++++++++++++++++++++++++++++-------------- tests/test_purge.py | 175 +++++++++++++++++++++++++++++++++++++++++- 2 files changed, 295 insertions(+), 58 deletions(-) diff --git a/fastpurge/_client.py b/fastpurge/_client.py index 5ffa8b5..4d4fd9b 100644 --- a/fastpurge/_client.py +++ b/fastpurge/_client.py @@ -1,15 +1,13 @@ import json import logging import os +import random +import time from collections import namedtuple from threading import local, Lock import requests -from requests.adapters import HTTPAdapter -from requests.exceptions import RetryError -from urllib3.util import Retry -from http import HTTPStatus try: from time import monotonic @@ -36,33 +34,47 @@ ]) -class LoggingRetry(Retry): - def __init__(self, *args, **kwargs, ): - self._logger = kwargs.pop('logger', None) - super(LoggingRetry, self).__init__(*args, **kwargs) +class FastPurgeError(RuntimeError): + """Raised when the Fast Purge API reports an error. - def new(self, **kw): - kw['logger'] = self._logger - return super(LoggingRetry, self).new(**kw) + The exception message contains a summary of the error encountered.""" - def increment(self, method, url, *args, **kwargs): - response = kwargs.get("response") - if response: - self._logger.error("An invalid status code %s was received " - "when trying to %s to %s: %s", - response.status, method, url, response.reason) - else: # pragma: no cover - self._logger.error( - "An unknown error occurred when trying to %s to %s", method, - url) - return super(LoggingRetry, self).increment(method, url, *args, - **kwargs) +def compute_backoff(failures, backoff_factor, backoff_max, jitter, + _rand=random.random): + """Delay (seconds) before retry number ``failures``. -class FastPurgeError(RuntimeError): - """Raised when the Fast Purge API reports an error. + Mirrors urllib3's exponential backoff so the ``FAST_PURGE_RETRY_*`` + tunables keep their existing meaning: the first retry (``failures == 1``) + has no delay, then ``backoff_factor * 2 ** (failures - 1)`` plus up to + ``jitter`` seconds of random jitter, capped at ``backoff_max``. + """ + if failures <= 1: + return 0.0 + delay = backoff_factor * (2 ** (failures - 1)) + if jitter: + delay += _rand() * jitter + return min(backoff_max, delay) - The exception message contains a summary of the error encountered.""" + +def parse_retry_statuses(spec): + """Parse a FAST_PURGE_RETRY_STATUSES spec into a frozenset of ints. + + ``spec`` is a comma-separated list where each item is either a single + status code (``"429"``) or an inclusive range (``"500-599"``). Whitespace + and empty items are ignored. + """ + statuses = set() + for item in spec.split(","): + item = item.strip() + if not item: + continue + if "-" in item: + low, high = item.split("-", 1) + statuses.update(range(int(low), int(high) + 1)) + else: + statuses.add(int(item)) + return frozenset(statuses) class FastPurgeClient(object): @@ -109,6 +121,16 @@ class FastPurgeClient(object): RETRY_BACKOFF_MAX = float(os.environ.get("FAST_PURGE_RETRY_BACKOFF_MAX", "150")) RETRY_BACKOFF_JITTER = float(os.environ.get("FAST_PURGE_RETRY_JITTER", "2")) + # HTTP statuses worth retrying, as an allowlist -- anything not listed + # fails fast (e.g. 401/403/404/413 are not retried pointlessly). The + # default covers rate limiting (429), all server errors (500-599), and + # 400: Akamai returns 400 "Invalid timestamp" for a stale signature (fixed + # by re-signing on retry) and for WAF rate-control, both of which recover + # on a fresh attempt. Override with FAST_PURGE_RETRY_STATUSES, a + # comma-separated list of codes and/or inclusive ranges, e.g. "429,500-599". + RETRY_STATUSES = parse_retry_statuses( + os.environ.get("FAST_PURGE_RETRY_STATUSES", "400,429,500-599")) + # Default purge type. # Akamai recommend "invalidate", so why is "delete" our default? # Here's what Akamai docs have to say: @@ -231,37 +253,25 @@ def __baseurl(self): return '{out}:{port}'.format(out=out, port=self.__port) - @property - def __retry_policy(self): - retries = getattr(self.__local, 'retries', None) - if not retries: - retries = LoggingRetry( - total=self.MAX_RETRIES, - backoff_factor=self.RETRY_BACKOFF, - backoff_max=self.RETRY_BACKOFF_MAX, - backoff_jitter=self.RETRY_BACKOFF_JITTER, - # We strictly require 201 here since that's how the server - # tells us we queued something async, as expected - status_forcelist=[status.value for status in HTTPStatus - if status.value != 201], - allowed_methods={'POST'}, - logger=LOG, - ) - self.__local.retries = retries - return retries - @property def __session(self): session = getattr(self.__local, 'session', None) if not session: session = requests.Session() session.auth = EdgeGridAuth(**self.__auth) - session.mount(self.__baseurl, - HTTPAdapter(max_retries=self.__retry_policy)) - + # Deliberately no transport-level (urllib3) retries: those resend + # the request below the requests auth layer, reusing the original + # EdgeGrid signature, so any retry past Akamai's ~30s timestamp + # window fails with "Invalid timestamp". Retries are handled in + # __start_purge, which re-issues post() so each attempt is re-signed. self.__local.session = session return session + def __backoff_time(self, failures): + return compute_backoff(failures, self.RETRY_BACKOFF, + self.RETRY_BACKOFF_MAX, + self.RETRY_BACKOFF_JITTER) + def __get_request_bodies(self, objects): out_object = {"objects": objects} out_str = json.dumps(out_object) @@ -279,16 +289,72 @@ def __get_request_bodies(self, objects): def __start_purge(self, endpoint, request_body): headers = {'Content-Type': 'application/json'} LOG.debug("POST JSON of size %s to %s", len(request_body), endpoint) + + attempts = self.MAX_RETRIES + 1 + last_reason = "no attempts made" + # One initial attempt plus up to MAX_RETRIES retries. + for failures in range(0, attempts): + delay = self.__backoff_time(failures) + if delay: + time.sleep(delay) + + try: + # A fresh post() re-prepares the request, so EdgeGridAuth signs + # it again with a current timestamp. That is what lets a retry + # succeed after Akamai's ~30s signature window -- urllib3-level + # retries resend the original request with its now-stale + # signature and can never recover. + response = self.__session.post( + endpoint, data=request_body, headers=headers) + except requests.exceptions.RequestException as error: + # Connection reset / timeout / etc: transient, always retryable. + last_reason = str(error) + LOG.error("An error occurred when trying to POST to %s: %s", + endpoint, error) + continue + + if response.status_code == 201: + # 201 is how Fast Purge acknowledges a queued async purge. + return self.__purge_from_response(response) + + # Same message as before, expanded with Retry-After and the response + # body so the real Akamai error (e.g. "Invalid timestamp") is + # visible. .get()/.text are safe even if the API omits them. + LOG.error( + "An invalid status code %s was received when trying to POST " + "to %s: %s (Retry-After=%s): %s", + response.status_code, endpoint, response.reason, + response.headers.get("Retry-After"), (response.text or "")[:500]) + + if response.status_code not in self.RETRY_STATUSES: + # Permanent error (bad credentials/permissions/request) that + # won't self-correct -- fail fast instead of burning the window. + message = ("Request to %s failed with non-retryable status %s" + % (endpoint, response.status_code)) + LOG.debug("%s", message) + raise FastPurgeError(message) + + last_reason = "too many %s error responses" % response.status_code + + message = ("Request to {endpoint} was unsuccessful after {retries} " + "retries: {reason}").format( + endpoint=endpoint, retries=self.MAX_RETRIES, + reason=last_reason) + LOG.debug("%s", message) + raise FastPurgeError(message) + + @staticmethod + def __purge_from_response(response): + # Tolerate an unexpected non-JSON 201 (e.g. a future API change): still + # treat the purge as accepted, falling back to the default estimate. try: - response = self.__session.post(endpoint, data=request_body, headers=headers) response_body = response.json() - estimated_seconds = response_body.get('estimatedSeconds', 5) - return Purge(response_body, monotonic() + estimated_seconds) - except RetryError as e: - message = "Request to {endpoint} was unsuccessful after {retries} retries: {reason}". \ - format(endpoint=endpoint, retries=self.MAX_RETRIES, reason=e.args[0].reason) - LOG.debug("%s", message) - raise FastPurgeError(message) from e + except ValueError: + LOG.warning("Purge accepted (201) with non-JSON body: %s", + (response.text or "")[:200]) + response_body = {} + estimated_seconds = response_body.get('estimatedSeconds', 5) + return Purge(response_body, monotonic() + estimated_seconds) def purge_objects(self, object_type, objects, **kwargs): """Purge a collection of objects. diff --git a/tests/test_purge.py b/tests/test_purge.py index b11ef19..b2d85de 100644 --- a/tests/test_purge.py +++ b/tests/test_purge.py @@ -1,4 +1,5 @@ import pytest +import requests import requests_mock import mock import responses @@ -9,9 +10,16 @@ from monotonic import monotonic from fastpurge import FastPurgeClient, FastPurgeError +from fastpurge._client import compute_backoff, parse_retry_statuses # pylint: disable=unused-argument +# NOTE: FastPurgeClient class attributes (MAX_RETRIES, RETRY_BACKOFF, etc.) are +# populated from environment variables once at import time, so monkeypatch.setenv +# has no effect on tests -- the class is already defined. Use +# monkeypatch.setattr(FastPurgeClient, "ATTR", value) instead, which is what the +# corresponding env var would have set had it been present before import. + @pytest.fixture def client_auth(): @@ -136,8 +144,9 @@ def test_scheme_port(client_auth, requests_mocker): def test_response_fails(client, no_thread_retries, monkeypatch): """Requests fail with a FastPurgeError if API gives unsuccessful response.""" url = 'https://fastpurge.example.com/ccu/v3/delete/cpcode/production' - # Decrease backoff, otherwise the test will run for 5 minutes - monkeypatch.setenv("FAST_PURGE_RETRY_BACKOFF", "0.001") + # Decrease backoff to avoid a multi-minute test run. + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF", 0.001) + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF_JITTER", 0) responses.add(responses.POST, url, status=503, content_type="application/json", body="Error") @@ -224,3 +233,165 @@ def test_retries_on_error(client_auth): assert len(err_1.calls) == 1 assert len(err_2.calls) == 1 assert len(res.calls) == 1 + + +@responses.activate(registry=responses.registries.OrderedRegistry) +def test_resigns_each_retry(client_auth, monkeypatch): + """Each retry is re-signed, so the EdgeGrid Authorization header differs + between attempts. (urllib3-level retries reused the original signature, + which failed with "Invalid timestamp" past Akamai's ~30s window.)""" + url = 'http://fastpurge.example.com:42/ccu/v3/delete/tag/staging' + responses.add(responses.POST, url, status=429, body="Rate limited") + responses.add(responses.POST, url, status=429, body="Rate limited") + responses.add(responses.POST, url, status=201, + content_type="application/json", + json={'estimatedSeconds': 0.1}) + + # No real delay between attempts. + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF", 0) + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF_JITTER", 0) + + client = FastPurgeClient(auth=client_auth, scheme='http', port=42) + + assert client.purge_by_tag(['red'], network='staging').result() + + auth_headers = [call.request.headers['Authorization'] + for call in responses.calls] + assert len(auth_headers) == 3 + # A fresh signature (new timestamp + nonce) per attempt => all distinct. + assert len(set(auth_headers)) == 3 + + +@responses.activate +def test_does_not_retry_auth_errors(client_auth): + """401/403 are not retried -- bad credentials/permissions won't self-correct.""" + url = 'http://fastpurge.example.com:42/ccu/v3/delete/tag/staging' + responses.add(responses.POST, url, status=401, body="Unauthorized") + + client = FastPurgeClient(auth=client_auth, scheme='http', port=42) + exception = client.purge_by_tag(['red'], network='staging').exception() + + assert isinstance(exception, FastPurgeError) + assert 'non-retryable status 401' in str(exception) + # Failed immediately, no retries. + assert len(responses.calls) == 1 + + +@responses.activate(registry=responses.registries.OrderedRegistry) +def test_retries_connection_errors(client_auth, monkeypatch): + """Connection/network errors are retried, not just HTTP statuses.""" + url = 'http://fastpurge.example.com:42/ccu/v3/delete/tag/staging' + responses.add(responses.POST, url, + body=requests.exceptions.ConnectionError("boom")) + responses.add(responses.POST, url, status=201, + content_type="application/json", + json={'estimatedSeconds': 0.1}) + + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF", 0) + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF_JITTER", 0) + + client = FastPurgeClient(auth=client_auth, scheme='http', port=42) + + assert client.purge_by_tag(['red'], network='staging').result() + assert len(responses.calls) == 2 + + +@responses.activate +def test_accepts_non_json_201(client_auth): + """A 201 with a non-JSON body is still treated as an accepted purge + (robust to an unexpected response-shape change), using the default estimate.""" + url = 'http://fastpurge.example.com:42/ccu/v3/delete/tag/staging' + responses.add(responses.POST, url, status=201, body="OK") # not JSON + + client = FastPurgeClient(auth=client_auth, scheme='http', port=42) + result = client.purge_by_tag(['red'], network='staging').result() + + assert result == [{}] # succeeded, with empty metadata + + +def test_default_retry_statuses(): + """Default allowlist: 429, 400 (EdgeGrid timestamp/WAF) and any 5xx are + retryable; other client errors are not.""" + for status in (400, 429, 500, 501, 502, 503, 504, 599): + assert status in FastPurgeClient.RETRY_STATUSES, status + for status in (201, 401, 403, 404, 405, 413, 422): + assert status not in FastPurgeClient.RETRY_STATUSES, status + + +def test_parse_retry_statuses(): + """FAST_PURGE_RETRY_STATUSES parsing: single codes, inclusive ranges, and + tolerant of whitespace / empty items.""" + assert parse_retry_statuses("429") == frozenset((429,)) + assert parse_retry_statuses("400,429") == frozenset((400, 429)) + assert parse_retry_statuses("500-504") == frozenset(range(500, 505)) + assert parse_retry_statuses(" 429 , 500-599 ") == frozenset((429,)) | frozenset(range(500, 600)) + assert parse_retry_statuses("") == frozenset() + + +@responses.activate +def test_retry_statuses_is_configurable(client_auth, monkeypatch): + """A status removed from RETRY_STATUSES is no longer retried (fails fast), + confirming the allowlist drives behavior.""" + url = 'http://fastpurge.example.com:42/ccu/v3/delete/tag/staging' + responses.add(responses.POST, url, status=503, body="Error") + + # Narrow the allowlist so 503 is NOT retryable. + monkeypatch.setattr(FastPurgeClient, "RETRY_STATUSES", frozenset((429,))) + + client = FastPurgeClient(auth=client_auth, scheme='http', port=42) + exception = client.purge_by_tag(['red'], network='staging').exception() + + assert isinstance(exception, FastPurgeError) + assert 'non-retryable status 503' in str(exception) + assert len(responses.calls) == 1 # failed immediately, no retries + + +def test_compute_backoff_matches_urllib3_schedule(): + """No jitter: first retry 0, then factor*2**(n-1), capped at backoff_max.""" + delays = [compute_backoff(n, 1.5, 9, 0) for n in range(1, 6)] + assert delays == [0.0, 3.0, 6.0, 9.0, 9.0] # 3, 6, 12->9, 24->9 + + # Jitter adds up to `jitter` seconds (here forced to its max) before the cap. + assert compute_backoff(2, 1.5, 100, 2.0, _rand=lambda: 1.0) == 3.0 + 2.0 + # Neither the initial attempt (0) nor the first retry (1) ever sleeps, + # regardless of jitter. + assert compute_backoff(0, 1.5, 100, 2.0, _rand=lambda: 1.0) == 0.0 + assert compute_backoff(1, 1.5, 100, 2.0, _rand=lambda: 1.0) == 0.0 + + +@responses.activate +def test_honors_max_retries(client_auth, monkeypatch): + """A persistently failing status is attempted exactly MAX_RETRIES + 1 times.""" + url = 'http://fastpurge.example.com:42/ccu/v3/delete/tag/staging' + responses.add(responses.POST, url, status=503, body="Error") + + monkeypatch.setattr(FastPurgeClient, "MAX_RETRIES", 2) + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF", 0) + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF_JITTER", 0) + + client = FastPurgeClient(auth=client_auth, scheme='http', port=42) + exception = client.purge_by_tag(['red'], network='staging').exception() + + assert isinstance(exception, FastPurgeError) + assert len(responses.calls) == 3 # 1 initial attempt + 2 retries + assert 'too many 503 error responses' in str(exception) + + +@responses.activate +def test_connection_error_exhaustion(client_auth, monkeypatch): + """When every attempt hits a connection error, a FastPurgeError is raised + reporting the underlying error (not a 'too many N' status message).""" + url = 'http://fastpurge.example.com:42/ccu/v3/delete/tag/staging' + responses.add(responses.POST, url, + body=requests.exceptions.ConnectionError("boom")) + + monkeypatch.setattr(FastPurgeClient, "MAX_RETRIES", 2) + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF", 0) + monkeypatch.setattr(FastPurgeClient, "RETRY_BACKOFF_JITTER", 0) + + client = FastPurgeClient(auth=client_auth, scheme='http', port=42) + exception = client.purge_by_tag(['red'], network='staging').exception() + + assert isinstance(exception, FastPurgeError) + assert len(responses.calls) == 3 # 1 initial attempt + 2 retries + assert 'boom' in str(exception)