From 11875d92cd69a9cd866f12f357afbdbf4a4647ce Mon Sep 17 00:00:00 2001 From: only1dreamgene Date: Mon, 19 Oct 2026 15:57:03 -0700 Subject: [PATCH] Add hardened provider HTTP client with retries, rate limiting and Redis caching --- .env.example | 5 + docs/provider-http-client.md | 48 +++++ src/oracle/httpClient.ts | 337 +++++++++++++++++++++++++++++++++++ 3 files changed, 390 insertions(+) create mode 100644 docs/provider-http-client.md create mode 100644 src/oracle/httpClient.ts diff --git a/.env.example b/.env.example index 18db953..360dcc9 100644 --- a/.env.example +++ b/.env.example @@ -159,3 +159,8 @@ ORACLE_RELAYER_SECRET_ID= # ── Ops API ────────────────────────────────────────────────────────────────── # Bearer token for GET/POST /api/v1/claims/ops/* endpoints. # OPS_API_TOKEN= + +# ── Oracle provider HTTP client (issue #188) ───────────────────────────────── +# Timeout/retry/rate-limit budgets are defined in src/oracle/httpClient.ts and +# documented in docs/provider-http-client.md. Responses are cached in Redis +# (REDIS_URL / REDIS_HOST); if Redis is down, calls go directly upstream. diff --git a/docs/provider-http-client.md b/docs/provider-http-client.md new file mode 100644 index 0000000..3997bcf --- /dev/null +++ b/docs/provider-http-client.md @@ -0,0 +1,48 @@ +# Oracle provider HTTP client + +`src/oracle/httpClient.ts` (`ProviderHttpClient`) is the only way adapters call +CoinGecko, Horizon and DeFiLlama. Never call `fetch` directly from an adapter. + +## Usage + +```ts +const client = new ProviderHttpClient(PROVIDER_DEFAULTS.coingecko, cacheService); +const data = await client.getJson(url, { + cacheKey: "simple-price:xlm,usdc", // feeds hitting the same endpoint share this key + ttlMs: feedFreshnessMs, // cache never serves older than this + budgetMs: cycleRemainingMs, // retries + backoff must fit in this +}); +``` + +## Retry policy + +- Total attempts = `maxRetries + 1` (default 3), each bounded by `timeoutMs` (5s). +- Retried: timeout, connection errors, HTTP 5xx. **Never** retried: 4xx, invalid JSON, local rate-limit rejection. +- Backoff: full jitter, `rand() * min(maxBackoffMs, baseBackoffMs * 2^n)`. +- A retry is skipped if its backoff would push past `budgetMs`; pass the remaining time of the + poll cycle so no work outlives the cycle. + +## Rate-limit budgets (process-wide token buckets, per provider) + +| Provider | Burst | Sustained | Concurrency | Public limit (approx.) | +|-----------|-------|-----------|-------------|------------------------| +| CoinGecko | 3 | 5/min | 2 | 5–15/min free tier | +| Horizon | 10 | 30/min | 4 | 3600/hr per IP | +| DeFiLlama | 10 | 120/min | 4 | ~500/min | + +Every attempt (including retries) consumes a token. An empty bucket fails fast with a +`rate_limited` error — no queuing, so no retry storms. + +## Caching + +Responses are cached in Redis under `provider::` with TTL = feed freshness. +Entries carry their write time and are re-checked on read, so a value past the freshness +window is never served. Redis errors are counted in `stats.cacheErrors` and the call goes +upstream. `client.stats` / `client.cacheHitRatio` expose hits, misses, retries, rate-limited +counts and upstream requests for metrics. + +## Errors + +All failures are `ProviderHttpError` with `provider`, `kind` +(`timeout | network | http_4xx | http_5xx | rate_limited | budget_exceeded | invalid_body`), +`status`, `attempts` and `retryable`, for consumption by health tracking. diff --git a/src/oracle/httpClient.ts b/src/oracle/httpClient.ts new file mode 100644 index 0000000..df568ca --- /dev/null +++ b/src/oracle/httpClient.ts @@ -0,0 +1,337 @@ +/** + * Shared, hardened HTTP client for outbound oracle data-provider calls + * (CoinGecko, Horizon, DeFiLlama, ...). Adapters must go through this client + * rather than calling `fetch` directly so that timeout, retry, rate-limit and + * cache behaviour is uniform. See docs/provider-http-client.md. + */ + +export type ProviderErrorKind = + | "timeout" + | "network" + | "http_4xx" + | "http_5xx" + | "rate_limited" + | "budget_exceeded" + | "invalid_body"; + +function isRetryableKind(kind: ProviderErrorKind): boolean { + return kind === "timeout" || kind === "network" || kind === "http_5xx"; +} + +/** Structured failure the health-tracking layer can consume. */ +export class ProviderHttpError extends Error { + constructor( + readonly provider: string, + readonly kind: ProviderErrorKind, + message: string, + readonly attempts: number, + readonly status?: number + ) { + super(`[${provider}] ${kind}: ${message}`); + this.name = "ProviderHttpError"; + } + + get retryable(): boolean { + return isRetryableKind(this.kind); + } +} + +export interface ProviderConfig { + name: string; + /** Per-attempt timeout. */ + timeoutMs: number; + /** Retries after the first attempt (total attempts = maxRetries + 1). */ + maxRetries: number; + baseBackoffMs: number; + maxBackoffMs: number; + /** Max simultaneous in-flight requests to this provider. */ + maxConcurrency: number; + /** Token bucket: burst size and sustained refill rate. */ + rateLimit: { capacity: number; refillPerSec: number }; +} + +/** + * Default budgets, comfortably under each provider's public limit + * (CoinGecko free ≈ 5–15 req/min, Horizon public ≈ 3600 req/hr per IP, + * DeFiLlama ≈ 500 req/min). + */ +export const PROVIDER_DEFAULTS: Record = { + coingecko: { + name: "coingecko", + timeoutMs: 5000, + maxRetries: 2, + baseBackoffMs: 500, + maxBackoffMs: 4000, + maxConcurrency: 2, + rateLimit: { capacity: 3, refillPerSec: 5 / 60 }, + }, + horizon: { + name: "horizon", + timeoutMs: 5000, + maxRetries: 2, + baseBackoffMs: 250, + maxBackoffMs: 2000, + maxConcurrency: 4, + rateLimit: { capacity: 10, refillPerSec: 0.5 }, + }, + defillama: { + name: "defillama", + timeoutMs: 5000, + maxRetries: 2, + baseBackoffMs: 250, + maxBackoffMs: 2000, + maxConcurrency: 4, + rateLimit: { capacity: 10, refillPerSec: 2 }, + }, +}; + +/** Minimal cache surface; CacheService satisfies it. */ +export interface ProviderCache { + get(key: string): Promise; + set(key: string, value: unknown, ttlSeconds?: number): Promise; +} + +export type Transport = ( + url: string, + init: { signal: AbortSignal; headers?: Record } +) => Promise<{ ok: boolean; status: number; json(): Promise }>; + +export interface RequestOptions { + /** Cache key; requests sharing a key (and TTL window) share one upstream call. */ + cacheKey?: string; + /** Feed freshness window. Cached values older than this are never served. */ + ttlMs?: number; + /** + * Overall time budget for this call including retries and backoff + * (typically the remainder of the poll cycle). + */ + budgetMs: number; + headers?: Record; +} + +export interface ClientStats { + cacheHits: number; + cacheMisses: number; + cacheErrors: number; + upstreamRequests: number; + retries: number; + rateLimited: number; +} + +class TokenBucket { + private tokens: number; + private last = Date.now(); + constructor( + private readonly capacity: number, + private readonly refillPerSec: number + ) { + this.tokens = capacity; + } + tryTake(): boolean { + const now = Date.now(); + this.tokens = Math.min( + this.capacity, + this.tokens + ((now - this.last) / 1000) * this.refillPerSec + ); + this.last = now; + if (this.tokens >= 1) { + this.tokens -= 1; + return true; + } + return false; + } +} + +class Semaphore { + private active = 0; + private waiters: Array<() => void> = []; + constructor(private readonly max: number) {} + async acquire(deadline: number): Promise { + if (this.active < this.max) { + this.active++; + return true; + } + return new Promise((resolve) => { + const grant = () => { + clearTimeout(timer); + this.active++; + resolve(true); + }; + const timer = setTimeout( + () => { + this.waiters = this.waiters.filter((w) => w !== grant); + resolve(false); + }, + Math.max(0, deadline - Date.now()) + ); + this.waiters.push(grant); + }); + } + release(): void { + this.active--; + this.waiters.shift()?.(); + } +} + +// Process-wide limiter state, shared by every client instance per provider. +const buckets = new Map(); +const semaphores = new Map(); + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +/** Full-jitter exponential backoff: uniform in [0, min(max, base * 2^attempt)). */ +export function backoffDelayMs( + cfg: ProviderConfig, + attempt: number, + rand: () => number = Math.random +): number { + return Math.floor(rand() * Math.min(cfg.maxBackoffMs, cfg.baseBackoffMs * 2 ** attempt)); +} + +export class ProviderHttpClient { + readonly stats: ClientStats = { + cacheHits: 0, + cacheMisses: 0, + cacheErrors: 0, + upstreamRequests: 0, + retries: 0, + rateLimited: 0, + }; + + constructor( + private readonly config: ProviderConfig, + private readonly cache?: ProviderCache, + private readonly transport: Transport = (url, init) => + fetch(url, init) as ReturnType, + private readonly rand: () => number = Math.random + ) { + if (!buckets.has(config.name)) { + buckets.set( + config.name, + new TokenBucket(config.rateLimit.capacity, config.rateLimit.refillPerSec) + ); + semaphores.set(config.name, new Semaphore(config.maxConcurrency)); + } + } + + get cacheHitRatio(): number { + const total = this.stats.cacheHits + this.stats.cacheMisses; + return total === 0 ? 0 : this.stats.cacheHits / total; + } + + async getJson(url: string, opts: RequestOptions): Promise { + const key = opts.cacheKey ? `provider:${this.config.name}:${opts.cacheKey}` : undefined; + const ttlMs = opts.ttlMs ?? 0; + const useCache = !!key && ttlMs > 0 && !!this.cache; + + if (useCache) { + try { + const hit = await this.cache!.get<{ at: number; v: T }>(key!); + // Re-check age ourselves: never serve past the feed's freshness window. + if (hit && Date.now() - hit.at < ttlMs) { + this.stats.cacheHits++; + return hit.v; + } + } catch { + this.stats.cacheErrors++; // Redis down: degrade to a direct call. + } + this.stats.cacheMisses++; + } + + const value = await this.fetchWithRetry(url, opts); + + if (useCache) { + try { + await this.cache!.set( + key!, + { at: Date.now(), v: value }, + Math.max(1, Math.ceil(ttlMs / 1000)) + ); + } catch { + this.stats.cacheErrors++; + } + } + return value; + } + + private async fetchWithRetry(url: string, opts: RequestOptions): Promise { + const cfg = this.config; + const deadline = Date.now() + opts.budgetMs; + let attempts = 0; + let last: ProviderHttpError | undefined; + + for (let attempt = 0; attempt <= cfg.maxRetries; attempt++) { + if (attempt > 0) { + const delay = backoffDelayMs(cfg, attempt - 1, this.rand); + // A retry that cannot finish inside the budget is wasted work. + if (Date.now() + delay + 1 >= deadline) break; + this.stats.retries++; + await sleep(delay); + } + if (deadline - Date.now() <= 0) break; + + if (!buckets.get(cfg.name)!.tryTake()) { + this.stats.rateLimited++; + throw new ProviderHttpError(cfg.name, "rate_limited", "local token bucket exhausted", attempts); + } + const sem = semaphores.get(cfg.name)!; + if (!(await sem.acquire(deadline))) break; + try { + attempts++; + const timeoutMs = Math.min(cfg.timeoutMs, deadline - Date.now()); + return await this.attemptOnce(url, opts, timeoutMs, attempts); + } catch (err) { + last = err as ProviderHttpError; + if (!last.retryable) throw last; + } finally { + sem.release(); + } + } + throw ( + last ?? + new ProviderHttpError( + cfg.name, + "budget_exceeded", + `no time left in ${opts.budgetMs}ms budget`, + attempts + ) + ); + } + + private async attemptOnce( + url: string, + opts: RequestOptions, + timeoutMs: number, + attempts: number + ): Promise { + const name = this.config.name; + const ctrl = new AbortController(); + let timedOut = false; + const timer = setTimeout(() => { + timedOut = true; + ctrl.abort(); + }, Math.max(1, timeoutMs)); + this.stats.upstreamRequests++; + try { + const res = await this.transport(url, { signal: ctrl.signal, headers: opts.headers }); + if (!res.ok) { + const kind: ProviderErrorKind = res.status >= 500 ? "http_5xx" : "http_4xx"; + throw new ProviderHttpError(name, kind, `HTTP ${res.status}`, attempts, res.status); + } + try { + return (await res.json()) as T; + } catch (e) { + if (timedOut) throw e; + throw new ProviderHttpError(name, "invalid_body", "response is not valid JSON", attempts, res.status); + } + } catch (err) { + if (err instanceof ProviderHttpError) throw err; + if (timedOut) { + throw new ProviderHttpError(name, "timeout", `no response in ${timeoutMs}ms`, attempts); + } + throw new ProviderHttpError(name, "network", (err as Error).message, attempts); + } finally { + clearTimeout(timer); + } + } +}