Repository Public
1// Provider runtime: per-provider deadline, circuit breaker, token-bucket rate limit and latency statistics.2import { ProviderError, type HttpClient, type ProviderContext, type SwapProvider, type SwapQuote, type SwapQuoteRequest } from './types.ts';34export interface ProviderPolicy { timeoutMs: number; ratePerSec: number; burst: number; breakerFailures: number; breakerCooldownMs: number }5export const DEFAULT_POLICY: ProviderPolicy = { timeoutMs: 2500, ratePerSec: 2, burst: 4, breakerFailures: 4, breakerCooldownMs: 30_000 };67export interface ProviderStats {8 id: string; enabled: boolean; state: 'closed' | 'open' | 'half_open';9 calls: number; failures: number; timeouts: number; lastError: string | null; lastOkAt: number | null;10 p50: number | null; p95: number | null; p99: number | null;11}1213const pct = (sorted: number[], p: number) => (sorted.length ? sorted[Math.min(sorted.length - 1, Math.ceil((p / 100) * sorted.length) - 1)]! : null);1415export class ProviderRuntime {16 private lat: number[] = []; private calls = 0; private failures = 0; private timeouts = 0; private consecutive = 0;17 private openedAt = 0; private lastError: string | null = null; private lastOkAt: number | null = null;18 private tokens: number; private refilledAt: number;19 enabled = true;20 constructor(readonly provider: SwapProvider, readonly policy: ProviderPolicy, private http: HttpClient, private now: () => number = Date.now) {21 this.tokens = policy.burst; this.refilledAt = now();22 }23 private state(): 'closed' | 'open' | 'half_open' {24 if (this.consecutive < this.policy.breakerFailures) return 'closed';25 return this.now() - this.openedAt >= this.policy.breakerCooldownMs ? 'half_open' : 'open';26 }27 private take(): boolean {28 const t = this.now();29 this.tokens = Math.min(this.policy.burst, this.tokens + ((t - this.refilledAt) / 1000) * this.policy.ratePerSec); this.refilledAt = t;30 if (this.tokens < 1) return false;31 this.tokens -= 1; return true;32 }33 /** Quote with the provider's own deadline. Never throws anything but ProviderError. */34 async quote(req: SwapQuoteRequest, outer?: AbortSignal): Promise<SwapQuote[]> {35 const id = this.provider.id;36 if (!this.enabled) throw new ProviderError(id, 'unavailable', 'provider disabled');37 if (this.state() === 'open') throw new ProviderError(id, 'unavailable', 'circuit open after repeated failures');38 if (!this.take()) throw new ProviderError(id, 'rate_limited', 'local rate limit');39 const started = this.now(); this.calls += 1;40 const ac = new AbortController();41 const timer = setTimeout(() => ac.abort(), this.policy.timeoutMs);42 const onOuter = () => ac.abort(); outer?.addEventListener('abort', onOuter, { once: true });43 const ctx: ProviderContext = { signal: ac.signal, http: this.http, now: this.now };44 try {45 const quotes = await this.provider.quote(req, ctx);46 this.lat.push(this.now() - started); if (this.lat.length > 500) this.lat.shift();47 this.consecutive = 0; this.lastOkAt = this.now();48 return quotes;49 } catch (e) {50 // a caller that gave up (outer abort) says nothing about the provider's health51 if (outer?.aborted) throw new ProviderError(id, 'timeout', 'request cancelled by caller');52 this.failures += 1; this.consecutive += 1; if (this.consecutive === this.policy.breakerFailures) this.openedAt = this.now();53 const timedOut = ac.signal.aborted; if (timedOut) this.timeouts += 1;54 const err = e instanceof ProviderError ? e : new ProviderError(id, timedOut ? 'timeout' : 'bad_response', (e as Error).message?.slice(0, 200) ?? 'error');55 if (timedOut && err.code !== 'timeout') err.code = 'timeout';56 this.lastError = `${err.code}: ${err.message}`;57 if (this.state() === 'half_open') this.openedAt = this.now(); // failed probe: stay open for another cooldown58 throw err;59 } finally { clearTimeout(timer); outer?.removeEventListener('abort', onOuter); }60 }61 stats(): ProviderStats {62 const s = [...this.lat].sort((a, b) => a - b);63 return { id: this.provider.id, enabled: this.enabled, state: this.state(), calls: this.calls, failures: this.failures, timeouts: this.timeouts,64 lastError: this.lastError, lastOkAt: this.lastOkAt, p50: pct(s, 50), p95: pct(s, 95), p99: pct(s, 99) };65 }66}67