spb/coinexplorer Public MIT

1# Author: Simon-Pierre Boucher2# Mail: contact@spboucher.ai3"""RPC/REST client layer for free public endpoints — raw HTTP, no SDKs.45Free endpoints differ wildly in limits and reliability, so every endpoint6carries:7  - a client-side TOKEN BUCKET (don't hit the server's limit in the first8    place; per-endpoint rps/burst configurable in chains.yaml)9  - a HEALTH SCORE (EMA of success rate) + latency EMA10  - an exponential COOLDOWN after consecutive failures (2s → 4s → … 300s)1112Pools pick the healthiest non-cooling endpoint per request and fail over13on transport errors, HTTP 429, and rate-limit-shaped JSON-RPC errors14(some providers deliver rate limits as error objects, e.g. -3200515"rate limited"). Semantic errors that mean "your query is too big"16(getLogs range caps) are raised to the caller, whose adaptive range17halving is the right response — not failover.1819Endpoint spec in config: either "https://url" or {url, rps, burst}.20"""2122import threading23import time2425import requests2627DEFAULT_RPS = 5.028DEFAULT_BURST = 10293031class RpcError(Exception):32    """JSON-RPC semantic error the CALLER must handle (e.g. range too big)."""3334    def __init__(self, code, message):35        super().__init__(f"RPC error {code}: {message}")36        self.code = code37        self.message = message or ""383940class RestError(Exception):41    """HTTP 4xx from a REST endpoint — a caller problem, not an outage."""4243    def __init__(self, status, body):44        super().__init__(f"HTTP {status}: {str(body)[:200]}")45        self.status = status46        self.body = body474849class AllEndpointsDown(RuntimeError):50    pass515253# "Query too big" wording (getLogs caps) — must reach the caller unhandled54# so its adaptive range halving kicks in.55_RANGE_HINTS = ("result", "range", "block range", "response size", "10000", "log")56# Rate-limit wording delivered as a JSON-RPC error — handled by failover.57_RATE_HINTS = (58    "rate limit", "rate-limit", "rate limited", "too many request",59    "quota", "capacity", "compute unit", "upgrade", "current plan",60)616263def _is_rate_limit(message):64    m = (message or "").lower()65    if any(h in m for h in _RANGE_HINTS):66        return False67    return any(h in m for h in _RATE_HINTS)686970class TokenBucket:71    def __init__(self, rate, burst):72        self.rate = float(rate)73        self.burst = float(burst)74        self.tokens = self.burst75        self.t = time.monotonic()76        self.lock = threading.Lock()7778    def acquire(self):79        """Take one token, sleeping outside the lock if we're in debt."""80        with self.lock:81            now = time.monotonic()82            self.tokens = min(self.burst, self.tokens + (now - self.t) * self.rate)83            self.t = now84            self.tokens -= 185            wait = 0.0 if self.tokens >= 0 else -self.tokens / self.rate86        if wait > 0:87            time.sleep(wait)888990class Endpoint:91    def __init__(self, spec):92        if isinstance(spec, str):93            spec = {"url": spec}94        self.url = spec["url"]95        self.bucket = TokenBucket(spec.get("rps", DEFAULT_RPS), spec.get("burst", DEFAULT_BURST))96        self.score = 1.097        self.latency_ms = 0.098        self.ok = 099        self.fail = 0100        self.consec_fail = 0101        self.cooldown_until = 0.0102103    def available(self):104        return time.monotonic() >= self.cooldown_until105106    def record(self, success, latency_ms=None, cooldown=None):107        alpha = 0.15108        self.score = (1 - alpha) * self.score + alpha * (1.0 if success else 0.0)109        if success:110            self.ok += 1111            self.consec_fail = 0112            if latency_ms is not None:113                self.latency_ms = (114                    0.8 * self.latency_ms + 0.2 * latency_ms if self.latency_ms else latency_ms115                )116        else:117            self.fail += 1118            self.consec_fail += 1119            self.cooldown_until = time.monotonic() + (120                cooldown if cooldown is not None else min(300, 2 ** min(self.consec_fail, 8))121            )122123124class _Pool:125    def __init__(self, specs, timeout=25, max_cycles=4):126        if not specs:127            raise ValueError("pool needs at least one endpoint")128        self.endpoints = [Endpoint(s) for s in specs]129        self.timeout = timeout130        self.max_cycles = max_cycles131        self.session = requests.Session()132        self._lock = threading.Lock()133        self._current = self.endpoints[0]134135    @property136    def current_url(self):137        return self._current.url138139    @property140    def urls(self):141        return [e.url for e in self.endpoints]142143    def _attempts(self):144        return self.max_cycles * len(self.endpoints)145146    def _pick(self):147        """Healthiest available endpoint; if all are cooling down, wait for148        the one that recovers soonest (capped so we notice config problems)."""149        with self._lock:150            avail = [e for e in self.endpoints if e.available()]151            if avail:152                ep, wait = max(avail, key=lambda e: (e.score, -e.latency_ms)), 0.0153            else:154                ep = min(self.endpoints, key=lambda e: e.cooldown_until)155                wait = max(0.0, ep.cooldown_until - time.monotonic())156            self._current = ep157        if wait:158            time.sleep(min(wait, 30))159        return ep160161    def rotate(self):162        """Nudge selection off the current endpoint (used by callers when a163        node keeps refusing a semantically-valid request)."""164        self._current.record(False, cooldown=15)165166    def stats(self):167        now = time.monotonic()168        return [169            {170                "url": e.url,171                "score": round(e.score, 3),172                "ok": e.ok,173                "fail": e.fail,174                "latency_ms": round(e.latency_ms, 1),175                "cooldown_s": round(max(0.0, e.cooldown_until - now), 1),176            }177            for e in self.endpoints178        ]179180181class RpcPool(_Pool):182    """JSON-RPC 2.0 over HTTP (EVM chains, Solana, Starknet, Sui classic…)."""183184    def _request(self, payload):185        last = None186        for _ in range(self._attempts()):187            ep = self._pick()188            ep.bucket.acquire()189            t0 = time.monotonic()190            try:191                r = self.session.post(192                    ep.url, json=payload, timeout=self.timeout,193                    headers={"Content-Type": "application/json"},194                )195                if r.status_code == 429:196                    ep.record(False, cooldown=30)197                    last = f"429 from {ep.url}"198                    continue199                r.raise_for_status()200                data = r.json()201            except (requests.RequestException, ValueError) as e:202                ep.record(False)203                last = e204                continue205            if isinstance(data, dict) and data.get("error"):206                err = data["error"] or {}207                if _is_rate_limit(err.get("message")):208                    ep.record(False, cooldown=20)209                    last = f"{err.get('message')} ({ep.url})"210                    continue211            ep.record(True, (time.monotonic() - t0) * 1000)212            return data213        raise AllEndpointsDown(f"all endpoints failing: {last}")214215    def call(self, method, params=None):216        res = self._request(217            {"jsonrpc": "2.0", "id": 1, "method": method, "params": params or []}218        )219        if not isinstance(res, dict):220            raise RpcError(None, f"unexpected response shape: {type(res).__name__}")221        if "error" in res and res["error"]:222            err = res["error"]223            raise RpcError(err.get("code"), err.get("message"))224        return res.get("result")225226    def batch(self, calls):227        """calls: list of (method, params) → results in order (None on228        individual failure). Sequential fallback for batch-rejecting nodes."""229        payload = [230            {"jsonrpc": "2.0", "id": i, "method": m, "params": p or []}231            for i, (m, p) in enumerate(calls)232        ]233        try:234            res = self._request(payload)235        except AllEndpointsDown:236            res = None237        if not isinstance(res, list):238            out = []239            for method, params in calls:240                try:241                    out.append(self.call(method, params))242                except (RpcError, AllEndpointsDown):243                    out.append(None)244            return out245        by_id = {r.get("id"): r for r in res if isinstance(r, dict)}246        return [247            by_id.get(i, {}).get("result")248            if not by_id.get(i, {}).get("error") else None249            for i in range(len(calls))250        ]251252253class RestPool(_Pool):254    """REST/HTTP APIs (Tron wallet API, Horizon, LCD, mirror nodes, algod…).255256    Fails over on 429/5xx/transport errors; 4xx raises RestError to the257    caller (retrying a bad request on another node won't fix it)."""258259    def _do(self, method, path, params=None, json_body=None):260        last = None261        for _ in range(self._attempts()):262            ep = self._pick()263            ep.bucket.acquire()264            url = ep.url.rstrip("/") + "/" + path.lstrip("/")265            t0 = time.monotonic()266            try:267                r = self.session.request(268                    method, url, params=params, json=json_body, timeout=self.timeout269                )270            except requests.RequestException as e:271                ep.record(False)272                last = e273                continue274            if r.status_code == 429 or r.status_code >= 500:275                ep.record(False, cooldown=30 if r.status_code == 429 else None)276                last = f"HTTP {r.status_code} from {url}"277                continue278            ep.record(True, (time.monotonic() - t0) * 1000)279            try:280                body = r.json()281            except ValueError:282                body = r.text283            if r.status_code >= 400:284                raise RestError(r.status_code, body)285            return body286        raise AllEndpointsDown(f"all endpoints failing: {last}")287288    def get(self, path, params=None):289        return self._do("GET", path, params=params)290291    def post(self, path, json_body=None):292        return self._do("POST", path, json_body=json_body)293