fix(sdk): make retries safe and deterministic

This commit is contained in:
ZacharyZhang-NY committed 2026-07-19 15:49:13 -04:00
1 parent e29b3c9648
commit bfe068b367
8 files changed
+566 -51

No files matched your search

+3
View File
@@ -41,6 +41,9 @@ This repository owns the published `elydora` package, its CLI, synchronous and a
- Propagate unexpected errors to the CLI boundary. - Propagate unexpected errors to the CLI boundary.
- Use documented defaults only for genuinely optional configuration. - Use documented defaults only for genuinely optional configuration.
- Avoid compatibility shims without a named public or user configuration contract. - Avoid compatibility shims without a named public or user configuration contract.
- Interpret `max_retries` as retries after the initial attempt and reject negative or non-integer values.
- Automatically retry RFC-idempotent methods; retry non-idempotent methods only when the transport proves the request was never sent.
- Honor valid `Retry-After` delay-seconds and HTTP-date values, and release retryable responses before sleeping.
## Verification ## Verification
+5 -1
View File
@@ -129,11 +129,15 @@ client = ElydoraClient(
private_key="<seed>", # Base64url-encoded Ed25519 seed private_key="<seed>", # Base64url-encoded Ed25519 seed
base_url="https://...", # API base URL (default: https://api.elydora.com) base_url="https://...", # API base URL (default: https://api.elydora.com)
ttl_ms=30000, # Operation TTL in ms (default: 30000) ttl_ms=30000, # Operation TTL in ms (default: 30000)
max_retries=3, # Max retries on transient failures (default: 3) max_retries=3, # Retries after the initial attempt (default: 3)
token="<api-token>", # Optional API token token="<api-token>", # Optional API token
) )
``` ```
Automatic retries apply to RFC-idempotent methods.
Connection failures that prove a request was never sent can also retry non-idempotent methods.
Retryable responses honor both `Retry-After` seconds and HTTP-date values.
### Authentication ### Authentication
```python ```python
+89
View File
@@ -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
+95
View File
@@ -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())
+96
View File
@@ -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
+15 -21
View File
@@ -3,12 +3,13 @@
from __future__ import annotations from __future__ import annotations
import time import time
import asyncio
import warnings import warnings
from typing import Any, Dict, List, Optional, Union from typing import Any, Dict, Optional, Union
import aiohttp 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 .crypto import compute_chain_hash, compute_payload_hash, sign_eor
from .errors import ElydoraError from .errors import ElydoraError
from .integration_types import require_integration_type from .integration_types import require_integration_type
@@ -20,7 +21,6 @@ from .types import (
DeepHealthResponse, DeepHealthResponse,
DeleteAgentResponse, DeleteAgentResponse,
EOR, EOR,
FreezeAgentResponse,
GetAgentResponse, GetAgentResponse,
GetEpochResponse, GetEpochResponse,
GetExportResponse, GetExportResponse,
@@ -81,7 +81,7 @@ class AsyncElydoraClient:
self.private_key = private_key self.private_key = private_key
self.base_url = base_url.rstrip("/") self.base_url = base_url.rstrip("/")
self.ttl_ms = ttl_ms self.ttl_ms = ttl_ms
self.max_retries = max_retries self.max_retries = require_max_retries(max_retries)
self.token = token self.token = token
self._prev_chain_hash = GENESIS_CHAIN_HASH self._prev_chain_hash = GENESIS_CHAIN_HASH
@@ -129,23 +129,17 @@ class AsyncElydoraClient:
hdrs = headers or self._headers() hdrs = headers or self._headers()
session = await self._get_session() session = await self._get_session()
last_exc: Optional[Exception] = None return await request_with_retries(
for attempt in range(0, self.max_retries + 1): session,
try: method,
async with session.request( url,
method, url, json=json_body, params=params, headers=hdrs, timeout=aiohttp.ClientTimeout(total=30) path=path,
) as resp: max_retries=self.max_retries,
return await self._handle_response(resp) response_handler=self._handle_response,
except ElydoraError: json_body=json_body,
raise params=params,
except (aiohttp.ClientConnectionError, asyncio.TimeoutError) as exc: headers=hdrs,
last_exc = exc )
if attempt < self.max_retries:
await asyncio.sleep(min(2 ** attempt, 8))
continue
raise
raise last_exc # type: ignore[misc]
@staticmethod @staticmethod
async def _handle_response(resp: aiohttp.ClientResponse) -> Any: async def _handle_response(resp: aiohttp.ClientResponse) -> Any:
+14 -29
View File
@@ -4,10 +4,12 @@ from __future__ import annotations
import time import time
import warnings import warnings
from typing import Any, Dict, List, Optional, Union from typing import Any, Dict, Optional, Union
import requests 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 .crypto import compute_chain_hash, compute_payload_hash, sign_eor
from .errors import ElydoraError from .errors import ElydoraError
from .integration_types import require_integration_type from .integration_types import require_integration_type
@@ -19,7 +21,6 @@ from .types import (
DeepHealthResponse, DeepHealthResponse,
DeleteAgentResponse, DeleteAgentResponse,
EOR, EOR,
FreezeAgentResponse,
GetAgentResponse, GetAgentResponse,
GetEpochResponse, GetEpochResponse,
GetExportResponse, GetExportResponse,
@@ -80,7 +81,7 @@ class ElydoraClient:
self.private_key = private_key self.private_key = private_key
self.base_url = base_url.rstrip("/") self.base_url = base_url.rstrip("/")
self.ttl_ms = ttl_ms self.ttl_ms = ttl_ms
self.max_retries = max_retries self.max_retries = require_max_retries(max_retries)
self.token = token self.token = token
self._prev_chain_hash = GENESIS_CHAIN_HASH self._prev_chain_hash = GENESIS_CHAIN_HASH
@@ -117,33 +118,17 @@ class ElydoraClient:
url = f"{self.base_url}{path}" url = f"{self.base_url}{path}"
hdrs = headers or self._headers() hdrs = headers or self._headers()
last_exc: Optional[Exception] = None return request_with_retries(
for attempt in range(0, self.max_retries + 1): self._session,
try: method,
resp = self._session.request( url,
method, url, json=json_body, params=params, headers=hdrs, timeout=30 path=path,
max_retries=self.max_retries,
response_handler=self._handle_response,
json_body=json_body,
params=params,
headers=hdrs,
) )
# Retry on 429 or 5xx
if resp.status_code == 429 or resp.status_code >= 500:
if attempt < self.max_retries:
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:
time.sleep(min(2 ** attempt, 8))
continue
raise
raise last_exc # type: ignore[misc]
@staticmethod @staticmethod
def _handle_response(resp: requests.Response) -> Any: def _handle_response(resp: requests.Response) -> Any:
+249
View File
@@ -0,0 +1,249 @@
import json
import logging
import threading
from datetime import datetime, timedelta, timezone
from email.utils import format_datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Dict, List, Tuple, cast
import pytest
from elydora import AsyncElydoraClient, ElydoraClient, ElydoraError
from elydora._retry import (
can_retry,
retry_delay_seconds,
should_retry_response,
)
ScriptedResponse = Tuple[int, Dict[str, str], object]
class ScriptedServer(ThreadingHTTPServer):
responses: List[ScriptedResponse]
methods: List[str]
class ScriptedHandler(BaseHTTPRequestHandler):
def do_GET(self) -> None:
self._respond()
def do_POST(self) -> None:
content_length = int(self.headers.get("Content-Length", "0"))
if content_length:
self.rfile.read(content_length)
self._respond()
def _respond(self) -> None:
server = cast(ScriptedServer, self.server)
server.methods.append(self.command)
status, headers, body = server.responses.pop(0)
encoded = json.dumps(body).encode()
self.send_response(status)
for name, value in headers.items():
self.send_header(name, value)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(encoded)))
self.end_headers()
self.wfile.write(encoded)
def log_message(self, _format: str, *args: object) -> None:
pass
@pytest.fixture
def scripted_server():
server = ScriptedServer(("127.0.0.1", 0), ScriptedHandler)
server.responses = []
server.methods = []
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
yield server
finally:
server.shutdown()
server.server_close()
thread.join()
def base_url(server: ScriptedServer) -> str:
host, port = server.server_address
return f"http://{host}:{port}"
def error_response(status: int, retry_after: str = "") -> ScriptedResponse:
headers = {"Retry-After": retry_after} if retry_after else {}
return (
status,
headers,
{"error": {"code": "INTERNAL_ERROR", "message": "temporary failure"}},
)
def test_retry_policy_matches_http_semantics() -> None:
assert should_retry_response("GET", 503, 0, 1)
assert should_retry_response("delete", 429, 0, 1)
assert not should_retry_response("POST", 503, 0, 1)
assert not should_retry_response("GET", 501, 0, 1)
assert not should_retry_response("GET", 503, 1, 1)
assert can_retry("POST", 0, 1, request_known_unprocessed=True)
def test_retry_after_supports_seconds_and_http_dates() -> None:
now = datetime(2026, 7, 19, 12, 0, tzinfo=timezone.utc)
retry_at = format_datetime(now + timedelta(seconds=30), usegmt=True)
assert retry_delay_seconds(0, "15", now=now) == 15
assert retry_delay_seconds(0, retry_at, now=now) == 30
assert retry_delay_seconds(2, "invalid", now=now) == 4
assert retry_delay_seconds(4, None, now=now) == 8
@pytest.mark.parametrize("value", [-1, True, 1.5, "2"])
def test_clients_reject_invalid_retry_configuration(value: object) -> None:
with pytest.raises(ValueError, match="non-negative integer"):
ElydoraClient("org", "agent", "key", max_retries=value) # type: ignore[arg-type]
with pytest.raises(ValueError, match="non-negative integer"):
AsyncElydoraClient("org", "agent", "key", max_retries=value) # type: ignore[arg-type]
def test_zero_retries_still_performs_one_sync_request(
scripted_server: ScriptedServer,
) -> None:
scripted_server.responses = [(200, {}, {"user_id": "user-1"})]
client = ElydoraClient(
"org", "agent", "key", base_url=base_url(scripted_server), max_retries=0
)
assert client.get_me()["user_id"] == "user-1"
assert scripted_server.methods == ["GET"]
@pytest.mark.asyncio
async def test_zero_retries_still_performs_one_async_request(
scripted_server: ScriptedServer,
) -> None:
scripted_server.responses = [(200, {}, {"user_id": "user-1"})]
client = AsyncElydoraClient(
"org", "agent", "key", base_url=base_url(scripted_server), max_retries=0
)
try:
response = await client.get_me()
assert response["user_id"] == "user-1"
assert scripted_server.methods == ["GET"]
finally:
await client.close()
def test_sync_client_retries_idempotent_responses(
scripted_server: ScriptedServer,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
scripted_server.responses = [
error_response(503, "0"),
error_response(500),
(200, {}, {"user_id": "user-1"}),
]
delays: List[float] = []
monkeypatch.setattr("elydora._sync_http.time.sleep", delays.append)
caplog.set_level(logging.WARNING, logger="elydora._sync_http")
client = ElydoraClient(
"org", "agent", "key", base_url=base_url(scripted_server), max_retries=2
)
assert client.get_me()["user_id"] == "user-1"
assert scripted_server.methods == ["GET", "GET", "GET"]
assert delays == [0.0, 2.0]
retry_logs = [
record.getMessage()
for record in caplog.records
if record.name == "elydora._sync_http"
]
assert "retry=1/2 delay_seconds=0.0 reason=http_503" in retry_logs[0]
assert "retry=2/2 delay_seconds=2.0 reason=http_500" in retry_logs[1]
@pytest.mark.asyncio
async def test_async_client_retries_idempotent_responses(
scripted_server: ScriptedServer,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
scripted_server.responses = [
error_response(429, "0"),
error_response(502),
(200, {}, {"user_id": "user-1"}),
]
delays: List[float] = []
async def record_sleep(delay: float) -> None:
delays.append(delay)
monkeypatch.setattr("elydora._async_http.asyncio.sleep", record_sleep)
caplog.set_level(logging.WARNING, logger="elydora._async_http")
client = AsyncElydoraClient(
"org", "agent", "key", base_url=base_url(scripted_server), max_retries=2
)
try:
response = await client.get_me()
assert response["user_id"] == "user-1"
assert scripted_server.methods == ["GET", "GET", "GET"]
assert delays == [0.0, 2.0]
retry_logs = [
record.getMessage()
for record in caplog.records
if record.name == "elydora._async_http"
]
assert "retry=1/2 delay_seconds=0.0 reason=http_429" in retry_logs[0]
assert "retry=2/2 delay_seconds=2.0 reason=http_502" in retry_logs[1]
finally:
await client.close()
def test_sync_client_does_not_replay_non_idempotent_requests(
scripted_server: ScriptedServer,
monkeypatch: pytest.MonkeyPatch,
) -> None:
scripted_server.responses = [
error_response(503),
(200, {}, {"token": "duplicate"}),
]
delays: List[float] = []
monkeypatch.setattr("elydora._sync_http.time.sleep", delays.append)
client = ElydoraClient(
"org", "agent", "key", base_url=base_url(scripted_server), max_retries=3
)
with pytest.raises(ElydoraError) as error:
client.issue_token()
assert error.value.status_code == 503
assert scripted_server.methods == ["POST"]
assert delays == []
@pytest.mark.asyncio
async def test_async_client_does_not_replay_non_idempotent_requests(
scripted_server: ScriptedServer,
monkeypatch: pytest.MonkeyPatch,
) -> None:
scripted_server.responses = [
error_response(503),
(200, {}, {"token": "duplicate"}),
]
delays: List[float] = []
async def record_sleep(delay: float) -> None:
delays.append(delay)
monkeypatch.setattr("elydora._async_http.asyncio.sleep", record_sleep)
client = AsyncElydoraClient(
"org", "agent", "key", base_url=base_url(scripted_server), max_retries=3
)
try:
with pytest.raises(ElydoraError) as error:
await client.issue_token()
assert error.value.status_code == 503
assert scripted_server.methods == ["POST"]
assert delays == []
finally:
await client.close()