"""retry — run a use case again when DataK3 said "not now", and only when that is safe.
A rollout, a writer restart or the reader's concurrency cap answers with an error that means "try again":
on 2026-10-16 a tables hotfix gave crm-demo a 90-second window where every write would have been a 500.
The unit retried is the WHOLE use case, never one statement: DataK3 has no read-your-writes inside an open
transaction, so a statement re-run inside a half-done transaction is not the same work.
WHICH failures may be retried is the whole design:
SAFE nothing can have committed — the database refused or aborted before COMMIT
57P03 cannot connect now · 08001 could not connect · 53300 too many connections
53400 configuration limit (the reader's 16-concurrent-read cap) · 40001 serialization
failure (the transaction was rolled back) · "ConfigurationLimitExceeded"
AMBIGUOUS the connection dropped mid-transaction (08006, 08003): the COMMIT may have landed.
Retried only for a query or a command declared idempotent — re-running `open_quote`
after its commit reached the database would make a second quote.
Everything else — a refused rule, a missing row, a bug — is not retried: running it again gives the same
answer, slower.
"""
import random
import re
import time
from typing import Any, Callable, Optional
SAFE = frozenset({"57P03", "08001", "53300", "53400", "40001"})
AMBIGUOUS = frozenset({"08006", "08003", "08000"})
SAFE_MARKERS = ("configurationlimitexceeded",)
AMBIGUOUS_MARKERS = ("could not confirm the wal", "server closed the connection unexpectedly",
"connection reset by peer")
_SQLSTATE = re.compile(r"SQLSTATE[\s:=]*\(?([0-9A-Z]{5})\)?")
ATTEMPTS = 4
BACKOFF = (0.2, 0.6, 1.5) # seconds before attempts 2, 3, 4 — 2.3s worst case, inside any timeout
def sqlstate(exc: BaseException) -> Optional[str]:
"""The SQLSTATE behind an exception, however it was wrapped (SQLAlchemy → psycopg, or text)."""
seen = set()
while exc is not None and id(exc) not in seen:
seen.add(id(exc))
for attr in ("sqlstate", "pgcode"):
code = getattr(exc, attr, None)
if isinstance(code, str) and len(code) == 5:
return code
m = _SQLSTATE.search(str(exc))
if m:
return m.group(1)
exc = getattr(exc, "orig", None) or exc.__cause__ or exc.__context__
return None
def classify(exc: BaseException) -> Optional[str]:
""""safe", "ambiguous", or None (not a transient database failure — do not retry)."""
code = sqlstate(exc)
text = str(exc).lower()
if code in SAFE or any(m in text for m in SAFE_MARKERS):
return "safe"
if code in AMBIGUOUS or any(m in text for m in AMBIGUOUS_MARKERS):
return "ambiguous"
return None
def run(fn: Callable[[], Any], *, idempotent: bool, attempts: int = ATTEMPTS,
sleep: Callable[[float], None] = time.sleep, on_retry: Optional[Callable[[int, str, BaseException], None]] = None) -> Any:
"""Call `fn`; on a transient failure that is safe to repeat, wait and call it again."""
for attempt in range(1, attempts + 1):
try:
return fn()
except Exception as exc: # noqa: BLE001 — re-raised unless retryable
kind = classify(exc)
retryable = kind == "safe" or (kind == "ambiguous" and idempotent)
if not retryable or attempt == attempts:
raise
if on_retry:
on_retry(attempt, kind, exc)
base = BACKOFF[min(attempt - 1, len(BACKOFF) - 1)]
sleep(base * (0.75 + random.random() / 2)) # jitter: a fleet does not retry in step
raise AssertionError("unreachable")