fix(python): make retries safe and deterministic
This commit is contained in:
1 parent
f64d00673c
commit
82e5807752
8 files changed
+567
-50
No files matched your search
@@ -0,0 +1,89 @@
|
||||
"""Asynchronous HTTP execution with the Elydora retry contract."""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import Any, Awaitable, Callable, Dict, Optional
|
||||
|
||||
import aiohttp
|
||||
|
||||
from ._retry import can_retry, log_retry, retry_delay_seconds, should_retry_response
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def request_with_retries(
|
||||
session: aiohttp.ClientSession,
|
||||
method: str,
|
||||
url: str,
|
||||
*,
|
||||
path: str,
|
||||
max_retries: int,
|
||||
response_handler: Callable[[aiohttp.ClientResponse], Awaitable[Any]],
|
||||
json_body: Any = None,
|
||||
params: Optional[Dict[str, str]] = None,
|
||||
headers: Optional[Dict[str, str]] = None,
|
||||
) -> Any:
|
||||
retry_count = 0
|
||||
while True:
|
||||
try:
|
||||
delay: Optional[float] = None
|
||||
reason = ""
|
||||
async with session.request(
|
||||
method,
|
||||
url,
|
||||
json=json_body,
|
||||
params=params,
|
||||
headers=headers,
|
||||
timeout=aiohttp.ClientTimeout(total=30),
|
||||
) as response:
|
||||
if should_retry_response(
|
||||
method, response.status, retry_count, max_retries
|
||||
):
|
||||
delay = retry_delay_seconds(
|
||||
retry_count, response.headers.get("Retry-After")
|
||||
)
|
||||
reason = f"http_{response.status}"
|
||||
else:
|
||||
return await response_handler(response)
|
||||
if delay is not None:
|
||||
log_retry(logger, method, path, retry_count, max_retries, delay, reason)
|
||||
await asyncio.sleep(delay)
|
||||
retry_count += 1
|
||||
continue
|
||||
except aiohttp.ClientConnectorError as error:
|
||||
if can_retry(
|
||||
method,
|
||||
retry_count,
|
||||
max_retries,
|
||||
request_known_unprocessed=True,
|
||||
):
|
||||
delay = retry_delay_seconds(retry_count)
|
||||
log_retry(
|
||||
logger,
|
||||
method,
|
||||
path,
|
||||
retry_count,
|
||||
max_retries,
|
||||
delay,
|
||||
type(error).__name__,
|
||||
)
|
||||
await asyncio.sleep(delay)
|
||||
retry_count += 1
|
||||
continue
|
||||
raise
|
||||
except (aiohttp.ClientConnectionError, asyncio.TimeoutError) as error:
|
||||
if can_retry(method, retry_count, max_retries):
|
||||
delay = retry_delay_seconds(retry_count)
|
||||
log_retry(
|
||||
logger,
|
||||
method,
|
||||
path,
|
||||
retry_count,
|
||||
max_retries,
|
||||
delay,
|
||||
type(error).__name__,
|
||||
)
|
||||
await asyncio.sleep(delay)
|
||||
retry_count += 1
|
||||
continue
|
||||
raise
|
||||
@@ -0,0 +1,95 @@
|
||||
"""Shared retry policy for the synchronous and asynchronous clients."""
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from email.utils import parsedate_to_datetime
|
||||
from logging import Logger
|
||||
from typing import Optional
|
||||
|
||||
IDEMPOTENT_METHODS = frozenset({"DELETE", "GET", "HEAD", "OPTIONS", "PUT"})
|
||||
RETRYABLE_STATUS_CODES = frozenset({408, 429, 500, 502, 503, 504})
|
||||
MAX_BACKOFF_SECONDS = 8
|
||||
|
||||
|
||||
def require_max_retries(value: object) -> int:
|
||||
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
|
||||
raise ValueError("max_retries must be a non-negative integer")
|
||||
return value
|
||||
|
||||
|
||||
def can_retry(
|
||||
method: str,
|
||||
retry_count: int,
|
||||
max_retries: int,
|
||||
*,
|
||||
request_known_unprocessed: bool = False,
|
||||
) -> bool:
|
||||
if retry_count >= max_retries:
|
||||
return False
|
||||
return request_known_unprocessed or method.upper() in IDEMPOTENT_METHODS
|
||||
|
||||
|
||||
def should_retry_response(
|
||||
method: str,
|
||||
status_code: int,
|
||||
retry_count: int,
|
||||
max_retries: int,
|
||||
) -> bool:
|
||||
return status_code in RETRYABLE_STATUS_CODES and can_retry(
|
||||
method, retry_count, max_retries
|
||||
)
|
||||
|
||||
|
||||
def retry_delay_seconds(
|
||||
retry_count: int,
|
||||
retry_after: Optional[str] = None,
|
||||
*,
|
||||
now: Optional[datetime] = None,
|
||||
) -> float:
|
||||
parsed_delay = _parse_retry_after(retry_after, now=now)
|
||||
if parsed_delay is not None:
|
||||
return parsed_delay
|
||||
return float(min(2**retry_count, MAX_BACKOFF_SECONDS))
|
||||
|
||||
|
||||
def log_retry(
|
||||
logger: Logger,
|
||||
method: str,
|
||||
path: str,
|
||||
retry_count: int,
|
||||
max_retries: int,
|
||||
delay: float,
|
||||
reason: str,
|
||||
) -> None:
|
||||
logger.warning(
|
||||
"Retrying Elydora request method=%s path=%s retry=%d/%d "
|
||||
"delay_seconds=%s reason=%s",
|
||||
method.upper(),
|
||||
path,
|
||||
retry_count + 1,
|
||||
max_retries,
|
||||
delay,
|
||||
reason,
|
||||
)
|
||||
|
||||
|
||||
def _parse_retry_after(
|
||||
value: Optional[str],
|
||||
*,
|
||||
now: Optional[datetime],
|
||||
) -> Optional[float]:
|
||||
if value is None:
|
||||
return None
|
||||
|
||||
normalized = value.strip()
|
||||
if normalized.isascii() and normalized.isdecimal():
|
||||
return float(int(normalized))
|
||||
|
||||
try:
|
||||
retry_at = parsedate_to_datetime(normalized)
|
||||
except (TypeError, ValueError, OverflowError):
|
||||
return None
|
||||
if retry_at.tzinfo is None:
|
||||
retry_at = retry_at.replace(tzinfo=timezone.utc)
|
||||
|
||||
current_time = now or datetime.now(timezone.utc)
|
||||
return max(0.0, (retry_at - current_time).total_seconds())
|
||||
@@ -0,0 +1,96 @@
|
||||
"""Synchronous HTTP execution with the Elydora retry contract."""
|
||||
|
||||
import logging
|
||||
import time
|
||||
from typing import Any, Callable, Dict, Optional
|
||||
|
||||
import requests
|
||||
|
||||
from ._retry import can_retry, log_retry, retry_delay_seconds, should_retry_response
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def request_with_retries(
|
||||
session: requests.Session,
|
||||
method: str,
|
||||
url: str,
|
||||
*,
|
||||
path: str,
|
||||
max_retries: int,
|
||||
response_handler: Callable[[requests.Response], Any],
|
||||
json_body: Any = None,
|
||||
params: Optional[Dict[str, str]] = None,
|
||||
headers: Optional[Dict[str, str]] = None,
|
||||
) -> Any:
|
||||
retry_count = 0
|
||||
while True:
|
||||
try:
|
||||
response = session.request(
|
||||
method,
|
||||
url,
|
||||
json=json_body,
|
||||
params=params,
|
||||
headers=headers,
|
||||
timeout=30,
|
||||
)
|
||||
if should_retry_response(
|
||||
method, response.status_code, retry_count, max_retries
|
||||
):
|
||||
delay = retry_delay_seconds(
|
||||
retry_count, response.headers.get("Retry-After")
|
||||
)
|
||||
response.close()
|
||||
log_retry(
|
||||
logger,
|
||||
method,
|
||||
path,
|
||||
retry_count,
|
||||
max_retries,
|
||||
delay,
|
||||
f"http_{response.status_code}",
|
||||
)
|
||||
time.sleep(delay)
|
||||
retry_count += 1
|
||||
continue
|
||||
return response_handler(response)
|
||||
except requests.exceptions.ConnectTimeout as error:
|
||||
if can_retry(
|
||||
method,
|
||||
retry_count,
|
||||
max_retries,
|
||||
request_known_unprocessed=True,
|
||||
):
|
||||
delay = retry_delay_seconds(retry_count)
|
||||
log_retry(
|
||||
logger,
|
||||
method,
|
||||
path,
|
||||
retry_count,
|
||||
max_retries,
|
||||
delay,
|
||||
type(error).__name__,
|
||||
)
|
||||
time.sleep(delay)
|
||||
retry_count += 1
|
||||
continue
|
||||
raise
|
||||
except (
|
||||
requests.exceptions.ConnectionError,
|
||||
requests.exceptions.Timeout,
|
||||
) as error:
|
||||
if can_retry(method, retry_count, max_retries):
|
||||
delay = retry_delay_seconds(retry_count)
|
||||
log_retry(
|
||||
logger,
|
||||
method,
|
||||
path,
|
||||
retry_count,
|
||||
max_retries,
|
||||
delay,
|
||||
type(error).__name__,
|
||||
)
|
||||
time.sleep(delay)
|
||||
retry_count += 1
|
||||
continue
|
||||
raise
|
||||
@@ -3,12 +3,13 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
import asyncio
|
||||
import warnings
|
||||
from typing import Any, Dict, List, Optional, Union
|
||||
from typing import Any, Dict, Optional, Union
|
||||
|
||||
import aiohttp
|
||||
|
||||
from ._async_http import request_with_retries
|
||||
from ._retry import require_max_retries
|
||||
from .crypto import compute_chain_hash, compute_payload_hash, sign_eor
|
||||
from .errors import ElydoraError
|
||||
from .integration_types import require_integration_type
|
||||
@@ -71,7 +72,7 @@ class AsyncElydoraClient:
|
||||
self.private_key = private_key
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.ttl_ms = ttl_ms
|
||||
self.max_retries = max_retries
|
||||
self.max_retries = require_max_retries(max_retries)
|
||||
self.token = token
|
||||
|
||||
self._prev_chain_hash = GENESIS_CHAIN_HASH
|
||||
@@ -119,23 +120,17 @@ class AsyncElydoraClient:
|
||||
hdrs = headers or self._headers()
|
||||
session = await self._get_session()
|
||||
|
||||
last_exc: Optional[Exception] = None
|
||||
for attempt in range(self.max_retries):
|
||||
try:
|
||||
async with session.request(
|
||||
method, url, json=json_body, params=params, headers=hdrs, timeout=aiohttp.ClientTimeout(total=30)
|
||||
) as resp:
|
||||
return await self._handle_response(resp)
|
||||
except ElydoraError:
|
||||
raise
|
||||
except (aiohttp.ClientConnectionError, asyncio.TimeoutError) as exc:
|
||||
last_exc = exc
|
||||
if attempt < self.max_retries - 1:
|
||||
await asyncio.sleep(min(2 ** attempt, 8))
|
||||
continue
|
||||
raise
|
||||
|
||||
raise last_exc # type: ignore[misc]
|
||||
return await request_with_retries(
|
||||
session,
|
||||
method,
|
||||
url,
|
||||
path=path,
|
||||
max_retries=self.max_retries,
|
||||
response_handler=self._handle_response,
|
||||
json_body=json_body,
|
||||
params=params,
|
||||
headers=hdrs,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
async def _handle_response(resp: aiohttp.ClientResponse) -> Any:
|
||||
|
||||
@@ -4,10 +4,12 @@ from __future__ import annotations
|
||||
|
||||
import time
|
||||
import warnings
|
||||
from typing import Any, Dict, List, Optional, Union
|
||||
from typing import Any, Dict, Optional, Union
|
||||
|
||||
import requests
|
||||
|
||||
from ._retry import require_max_retries
|
||||
from ._sync_http import request_with_retries
|
||||
from .crypto import compute_chain_hash, compute_payload_hash, sign_eor
|
||||
from .errors import ElydoraError
|
||||
from .integration_types import require_integration_type
|
||||
@@ -70,7 +72,7 @@ class ElydoraClient:
|
||||
self.private_key = private_key
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.ttl_ms = ttl_ms
|
||||
self.max_retries = max_retries
|
||||
self.max_retries = require_max_retries(max_retries)
|
||||
self.token = token
|
||||
|
||||
self._prev_chain_hash = GENESIS_CHAIN_HASH
|
||||
@@ -107,33 +109,17 @@ class ElydoraClient:
|
||||
url = f"{self.base_url}{path}"
|
||||
hdrs = headers or self._headers()
|
||||
|
||||
last_exc: Optional[Exception] = None
|
||||
for attempt in range(self.max_retries):
|
||||
try:
|
||||
resp = self._session.request(
|
||||
method, url, json=json_body, params=params, headers=hdrs, timeout=30
|
||||
)
|
||||
# Retry on 429 or 5xx
|
||||
if resp.status_code == 429 or resp.status_code >= 500:
|
||||
if attempt < self.max_retries - 1:
|
||||
retry_after = resp.headers.get("Retry-After")
|
||||
if retry_after and retry_after.isdigit():
|
||||
delay = int(retry_after)
|
||||
else:
|
||||
delay = min(2 ** attempt, 8)
|
||||
time.sleep(delay)
|
||||
continue
|
||||
return self._handle_response(resp)
|
||||
except ElydoraError:
|
||||
raise
|
||||
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as exc:
|
||||
last_exc = exc
|
||||
if attempt < self.max_retries - 1:
|
||||
time.sleep(min(2 ** attempt, 8))
|
||||
continue
|
||||
raise
|
||||
|
||||
raise last_exc # type: ignore[misc]
|
||||
return request_with_retries(
|
||||
self._session,
|
||||
method,
|
||||
url,
|
||||
path=path,
|
||||
max_retries=self.max_retries,
|
||||
response_handler=self._handle_response,
|
||||
json_body=json_body,
|
||||
params=params,
|
||||
headers=hdrs,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _handle_response(resp: requests.Response) -> Any:
|
||||
|
||||
Reference in new issue
Block a user