# Author: Simon-Pierre Boucher # Mail: contact@spboucher.ai """RPC/REST client layer for free public endpoints — raw HTTP, no SDKs. Free endpoints differ wildly in limits and reliability, so every endpoint carries: - a client-side TOKEN BUCKET (don't hit the server's limit in the first place; per-endpoint rps/burst configurable in chains.yaml) - a HEALTH SCORE (EMA of success rate) + latency EMA - an exponential COOLDOWN after consecutive failures (2s → 4s → … 300s) Pools pick the healthiest non-cooling endpoint per request and fail over on transport errors, HTTP 429, and rate-limit-shaped JSON-RPC errors (some providers deliver rate limits as error objects, e.g. -32005 "rate limited"). Semantic errors that mean "your query is too big" (getLogs range caps) are raised to the caller, whose adaptive range halving is the right response — not failover. Endpoint spec in config: either "https://url" or {url, rps, burst}. """ import threading import time import requests DEFAULT_RPS = 5.0 DEFAULT_BURST = 10 class RpcError(Exception): """JSON-RPC semantic error the CALLER must handle (e.g. range too big).""" def __init__(self, code, message): super().__init__(f"RPC error {code}: {message}") self.code = code self.message = message or "" class RestError(Exception): """HTTP 4xx from a REST endpoint — a caller problem, not an outage.""" def __init__(self, status, body): super().__init__(f"HTTP {status}: {str(body)[:200]}") self.status = status self.body = body class AllEndpointsDown(RuntimeError): pass # "Query too big" wording (getLogs caps) — must reach the caller unhandled # so its adaptive range halving kicks in. _RANGE_HINTS = ("result", "range", "block range", "response size", "10000", "log") # Rate-limit wording delivered as a JSON-RPC error — handled by failover. _RATE_HINTS = ( "rate limit", "rate-limit", "rate limited", "too many request", "quota", "capacity", "compute unit", "upgrade", "current plan", ) def _is_rate_limit(message): m = (message or "").lower() if any(h in m for h in _RANGE_HINTS): return False return any(h in m for h in _RATE_HINTS) class TokenBucket: def __init__(self, rate, burst): self.rate = float(rate) self.burst = float(burst) self.tokens = self.burst self.t = time.monotonic() self.lock = threading.Lock() def acquire(self): """Take one token, sleeping outside the lock if we're in debt.""" with self.lock: now = time.monotonic() self.tokens = min(self.burst, self.tokens + (now - self.t) * self.rate) self.t = now self.tokens -= 1 wait = 0.0 if self.tokens >= 0 else -self.tokens / self.rate if wait > 0: time.sleep(wait) class Endpoint: def __init__(self, spec): if isinstance(spec, str): spec = {"url": spec} self.url = spec["url"] self.bucket = TokenBucket(spec.get("rps", DEFAULT_RPS), spec.get("burst", DEFAULT_BURST)) self.score = 1.0 self.latency_ms = 0.0 self.ok = 0 self.fail = 0 self.consec_fail = 0 self.cooldown_until = 0.0 def available(self): return time.monotonic() >= self.cooldown_until def record(self, success, latency_ms=None, cooldown=None): alpha = 0.15 self.score = (1 - alpha) * self.score + alpha * (1.0 if success else 0.0) if success: self.ok += 1 self.consec_fail = 0 if latency_ms is not None: self.latency_ms = ( 0.8 * self.latency_ms + 0.2 * latency_ms if self.latency_ms else latency_ms ) else: self.fail += 1 self.consec_fail += 1 self.cooldown_until = time.monotonic() + ( cooldown if cooldown is not None else min(300, 2 ** min(self.consec_fail, 8)) ) class _Pool: def __init__(self, specs, timeout=25, max_cycles=4): if not specs: raise ValueError("pool needs at least one endpoint") self.endpoints = [Endpoint(s) for s in specs] self.timeout = timeout self.max_cycles = max_cycles self.session = requests.Session() self._lock = threading.Lock() self._current = self.endpoints[0] @property def current_url(self): return self._current.url @property def urls(self): return [e.url for e in self.endpoints] def _attempts(self): return self.max_cycles * len(self.endpoints) def _pick(self): """Healthiest available endpoint; if all are cooling down, wait for the one that recovers soonest (capped so we notice config problems).""" with self._lock: avail = [e for e in self.endpoints if e.available()] if avail: ep, wait = max(avail, key=lambda e: (e.score, -e.latency_ms)), 0.0 else: ep = min(self.endpoints, key=lambda e: e.cooldown_until) wait = max(0.0, ep.cooldown_until - time.monotonic()) self._current = ep if wait: time.sleep(min(wait, 30)) return ep def rotate(self): """Nudge selection off the current endpoint (used by callers when a node keeps refusing a semantically-valid request).""" self._current.record(False, cooldown=15) def stats(self): now = time.monotonic() return [ { "url": e.url, "score": round(e.score, 3), "ok": e.ok, "fail": e.fail, "latency_ms": round(e.latency_ms, 1), "cooldown_s": round(max(0.0, e.cooldown_until - now), 1), } for e in self.endpoints ] class RpcPool(_Pool): """JSON-RPC 2.0 over HTTP (EVM chains, Solana, Starknet, Sui classic…).""" def _request(self, payload): last = None for _ in range(self._attempts()): ep = self._pick() ep.bucket.acquire() t0 = time.monotonic() try: r = self.session.post( ep.url, json=payload, timeout=self.timeout, headers={"Content-Type": "application/json"}, ) if r.status_code == 429: ep.record(False, cooldown=30) last = f"429 from {ep.url}" continue r.raise_for_status() data = r.json() except (requests.RequestException, ValueError) as e: ep.record(False) last = e continue if isinstance(data, dict) and data.get("error"): err = data["error"] or {} if _is_rate_limit(err.get("message")): ep.record(False, cooldown=20) last = f"{err.get('message')} ({ep.url})" continue ep.record(True, (time.monotonic() - t0) * 1000) return data raise AllEndpointsDown(f"all endpoints failing: {last}") def call(self, method, params=None): res = self._request( {"jsonrpc": "2.0", "id": 1, "method": method, "params": params or []} ) if not isinstance(res, dict): raise RpcError(None, f"unexpected response shape: {type(res).__name__}") if "error" in res and res["error"]: err = res["error"] raise RpcError(err.get("code"), err.get("message")) return res.get("result") def batch(self, calls): """calls: list of (method, params) → results in order (None on individual failure). Sequential fallback for batch-rejecting nodes.""" payload = [ {"jsonrpc": "2.0", "id": i, "method": m, "params": p or []} for i, (m, p) in enumerate(calls) ] try: res = self._request(payload) except AllEndpointsDown: res = None if not isinstance(res, list): out = [] for method, params in calls: try: out.append(self.call(method, params)) except (RpcError, AllEndpointsDown): out.append(None) return out by_id = {r.get("id"): r for r in res if isinstance(r, dict)} return [ by_id.get(i, {}).get("result") if not by_id.get(i, {}).get("error") else None for i in range(len(calls)) ] class RestPool(_Pool): """REST/HTTP APIs (Tron wallet API, Horizon, LCD, mirror nodes, algod…). Fails over on 429/5xx/transport errors; 4xx raises RestError to the caller (retrying a bad request on another node won't fix it).""" def _do(self, method, path, params=None, json_body=None): last = None for _ in range(self._attempts()): ep = self._pick() ep.bucket.acquire() url = ep.url.rstrip("/") + "/" + path.lstrip("/") t0 = time.monotonic() try: r = self.session.request( method, url, params=params, json=json_body, timeout=self.timeout ) except requests.RequestException as e: ep.record(False) last = e continue if r.status_code == 429 or r.status_code >= 500: ep.record(False, cooldown=30 if r.status_code == 429 else None) last = f"HTTP {r.status_code} from {url}" continue ep.record(True, (time.monotonic() - t0) * 1000) try: body = r.json() except ValueError: body = r.text if r.status_code >= 400: raise RestError(r.status_code, body) return body raise AllEndpointsDown(f"all endpoints failing: {last}") def get(self, path, params=None): return self._do("GET", path, params=params) def post(self, path, json_body=None): return self._do("POST", path, json_body=json_body)