From fb56823cd98811a728ab0fd05e500628e519604e Mon Sep 17 00:00:00 2001 From: sherifatolanike Date: Tue, 28 Jul 2026 05:42:45 +0000 Subject: [PATCH] feat: implement RPC timeout and circuit breaker for external blockchain calls - Configurable timeout per RPC endpoint (rpcConfig.ts with per-chain defaults) - Circuit breaker state machine (CLOSED/OPEN/HALF_OPEN) with failure threshold & recovery timeout - RPC provider fallback across multiple endpoints per chain - Circuit state monitoring dashboard (GET /rpc/dashboard, GET /rpc/health/:chainId) - Graceful degradation on circuit open (cache_stale / fail_fast modes) - Manual circuit reset (POST /rpc/reset with reset/open/reset_all/reset_metrics) - Prometheus metrics at GET /metrics/rpc - New error codes: RPC_TIMEOUT, RPC_CIRCUIT_OPEN, RPC_ALL_PROVIDERS_FAILED --- backend/server.ts | 155 ++++++- backend/services/index.ts | 32 ++ backend/services/rpc/circuitBreaker.ts | 364 +++++++++++++++ backend/services/rpc/index.ts | 64 +++ backend/services/rpc/rpcConfig.ts | 311 +++++++++++++ backend/services/rpc/rpcMonitorService.ts | 487 ++++++++++++++++++++ backend/services/rpc/rpcProviderFallback.ts | 467 +++++++++++++++++++ backend/services/shared/apiResponse.ts | 12 +- 8 files changed, 1889 insertions(+), 3 deletions(-) create mode 100644 backend/services/rpc/circuitBreaker.ts create mode 100644 backend/services/rpc/index.ts create mode 100644 backend/services/rpc/rpcConfig.ts create mode 100644 backend/services/rpc/rpcMonitorService.ts create mode 100644 backend/services/rpc/rpcProviderFallback.ts diff --git a/backend/server.ts b/backend/server.ts index 522687f0..b2ce7900 100644 --- a/backend/server.ts +++ b/backend/server.ts @@ -36,6 +36,11 @@ import { rateLimitingService } from './services/shared/rateLimitingService'; import { createRateLimitMiddleware, RATE_LIMIT_HEADERS } from './services/shared/rateLimitMiddleware'; import { applyETagToRawHandler } from './shared/middleware/etagMiddleware'; import { SubscriptionTier } from '../src/types/subscription'; +import { + rpcMonitorService, + DEFAULT_CHAIN_ENDPOINTS, + RpcProviderFallback, +} from './services/rpc'; export interface StartServerOptions { port?: number; @@ -76,8 +81,8 @@ async function ensurePlanCache(pool: Pool): Promise { function buildRateLimitMiddleware() { return createRateLimitMiddleware({ service: rateLimitingService, - // Bypass paths (health/metrics never throttled) - bypassPaths: ['/health', '/metrics', '/metrics/plan-cache'], + // Bypass paths (health/metrics/rpc never throttled) + bypassPaths: ['/health', '/metrics', '/metrics/plan-cache', '/metrics/rpc', '/rpc/dashboard', '/rpc/health', '/rpc/events'], // Tier resolver: reads x-subscription-tier header; defaults to FREE tierFn: (apiKey, _userId) => { // In production this would look up the tier from a DB / cache. @@ -149,6 +154,40 @@ function matchPlanId(pathname: string): string | null { const match = pathname.match(/^\/plans\/([^/]+)$/); return match?.[1] ?? null; } +// ── RPC Provider Fallback instances (hot-reloadable) ───────────────────────── +const rpcProviders = new Map(); + +function getOrCreateRpcProvider(chainId: number): RpcProviderFallback { + let provider = rpcProviders.get(chainId); + if (!provider) { + const chainConfig = DEFAULT_CHAIN_ENDPOINTS[chainId]; + if (!chainConfig) { + throw new Error(`No RPC configuration for chain ${chainId}`); + } + provider = new RpcProviderFallback(chainConfig, rpcMonitorService); + rpcProviders.set(chainId, provider); + + // Register circuits with the monitor + for (const snapshot of provider.getCircuitStates()) { + rpcMonitorService.registerCircuit(snapshot); + } + } + return provider; +} + +function registerAllChainProviders(): void { + for (const chainIdStr of Object.keys(DEFAULT_CHAIN_ENDPOINTS)) { + const chainId = Number(chainIdStr); + try { + const provider = getOrCreateRpcProvider(chainId); + for (const snapshot of provider.getCircuitStates()) { + rpcMonitorService.registerCircuit(snapshot); + } + } catch (err) { + console.warn(`[Server] Could not register RPC provider for chain ${chainId}:`, err); + } + } +} export async function startServer(options: StartServerOptions = {}): Promise { const pool = options.pool ?? (await getPool()); @@ -166,6 +205,9 @@ export async function startServer(options: StartServerOptions = {}): Promise { const url = new URL(req.url ?? '/', `http://${req.headers.host ?? 'localhost'}`); const { pathname } = url; @@ -187,6 +229,110 @@ export async function startServer(options: StartServerOptions = {}): Promise | null = null; + private readonly config: CircuitBreakerConfig; + public readonly chainId: number; + public readonly endpointUrl: string; + public readonly endpointLabel: string; + + constructor(endpoint: Pick) { + super(); + this.chainId = endpoint.chainId; + this.endpointUrl = endpoint.url; + this.endpointLabel = endpoint.label; + this.config = { ...DEFAULT_CONFIG, ...endpoint.circuitBreaker }; + } + + // ── Public API ──────────────────────────────────────────────────────────── + + /** Returns the current state of the circuit. */ + getState(): CircuitState { + // Lazy transition: if OPEN and recovery timeout has elapsed, move to HALF_OPEN + if (this.state === 'OPEN' && this.recoveryScheduledAt !== null) { + const elapsed = Date.now() - this.recoveryScheduledAt; + if (elapsed >= this.config.recoveryTimeoutMs) { + this.transitionTo('HALF_OPEN', 'Recovery timeout elapsed'); + } + } + return this.state; + } + + /** + * Should be called BEFORE making an RPC call. + * Returns true if the request is allowed to proceed. + * Throws CircuitOpenError if the circuit is OPEN. + */ + allowRequest(): boolean { + const currentState = this.getState(); + + if (currentState === 'CLOSED') { + return true; + } + + if (currentState === 'OPEN') { + throw new CircuitOpenError(this.endpointUrl, this.chainId, currentState); + } + + // HALF_OPEN: limit concurrent probe requests + if (this.halfOpenProbeCount >= this.config.halfOpenMaxRequests) { + return false; + } + + this.halfOpenProbeCount++; + this.pendingProbeRequests++; + return true; + } + + /** + * Record a successful RPC call. + * @param wasProbe Whether this was a half-open probe request. + */ + recordSuccess(wasProbe = false): void { + this.totalSuccesses++; + this.lastSuccessAt = Date.now(); + this.consecutiveFailures = 0; + + if (wasProbe) { + this.pendingProbeRequests--; + } + + if (this.state === 'HALF_OPEN') { + // Probe succeeded — close the circuit + this.transitionTo('CLOSED', 'Half-open probe succeeded'); + } else if (this.state === 'CLOSED' && this.consecutiveFailures > 0) { + // Reset consecutive failures on success + this.consecutiveFailures = 0; + } + } + + /** + * Record a failed RPC call. + * @param wasProbe Whether this was a half-open probe request. + */ + recordFailure(wasProbe = false): void { + this.totalFailures++; + this.lastFailureAt = Date.now(); + this.consecutiveFailures++; + + if (wasProbe) { + this.pendingProbeRequests--; + } + + if (this.state === 'HALF_OPEN') { + // Probe failed — open the circuit again + this.transitionTo('OPEN', 'Half-open probe failed'); + } else if ( + this.state === 'CLOSED' && + this.consecutiveFailures >= this.config.failureThreshold + ) { + // Threshold reached — open the circuit + this.transitionTo('OPEN', `Failure threshold reached (${this.consecutiveFailures} consecutive)`); + } + } + + /** + * Manually reset the circuit to CLOSED state. + * Clears all failure counts and cancels any pending recovery timer. + */ + manualReset(): void { + this.cancelRecoveryTimer(); + this.consecutiveFailures = 0; + this.totalFailures = 0; + this.totalSuccesses = 0; + this.halfOpenProbeCount = 0; + this.pendingProbeRequests = 0; + this._manuallyReset = true; + this.transitionTo('CLOSED', 'Manual reset by operator'); + } + + /** + * Manually open the circuit (force trip). + */ + manualOpen(): void { + this.cancelRecoveryTimer(); + this._manuallyReset = true; + this.transitionTo('OPEN', 'Manually opened by operator'); + } + + /** Returns a snapshot of the current circuit state. */ + snapshot(): CircuitStateSnapshot { + const now = Date.now(); + // Compute current downtime duration without mutating the stored value. + const currentDowntime = this.state === 'OPEN' && this.openedAt !== null + ? this.cumulativeDowntimeMs + (now - this.openedAt) + : this.cumulativeDowntimeMs; + + return { + endpointUrl: this.endpointUrl, + endpointLabel: this.endpointLabel, + chainId: this.chainId, + state: this.state, + consecutiveFailures: this.consecutiveFailures, + totalFailures: this.totalFailures, + totalSuccesses: this.totalSuccesses, + lastFailureAt: this.lastFailureAt, + lastSuccessAt: this.lastSuccessAt, + lastStateChangeAt: this.lastStateChangeAt, + openedAt: this.openedAt, + recoveryScheduledAt: this.recoveryScheduledAt, + halfOpenProbeCount: this.halfOpenProbeCount, + pendingProbeRequests: this.pendingProbeRequests, + cumulativeDowntimeMs: currentDowntime, + manuallyReset: this._manuallyReset, + }; + } + + /** Resets metrics counters (keeps state). */ + resetMetrics(): void { + this.consecutiveFailures = 0; + this.totalFailures = 0; + this.totalSuccesses = 0; + this.halfOpenProbeCount = 0; + this.pendingProbeRequests = 0; + this.cumulativeDowntimeMs = 0; + this._manuallyReset = false; + } + + // ── Private ─────────────────────────────────────────────────────────────── + + private transitionTo(newState: CircuitState, reason: string): void { + const previousState = this.state; + this.state = newState; + this.lastStateChangeAt = Date.now(); + + if (newState === 'OPEN') { + this.openedAt = Date.now(); + this.recoveryScheduledAt = Date.now(); + this.halfOpenProbeCount = 0; + this.pendingProbeRequests = 0; + + if (this.config.autoReset) { + this.scheduleRecovery(); + } + } else if (newState === 'HALF_OPEN') { + this.openedAt = null; + this.recoveryScheduledAt = null; + this.halfOpenProbeCount = 0; + this.pendingProbeRequests = 0; + } else if (newState === 'CLOSED') { + // Accumulate downtime if transitioning from OPEN to CLOSED + if (previousState === 'OPEN' && this.openedAt !== null) { + this.cumulativeDowntimeMs += Date.now() - this.openedAt; + } + this.openedAt = null; + this.recoveryScheduledAt = null; + this.halfOpenProbeCount = 0; + this.pendingProbeRequests = 0; + this.consecutiveFailures = 0; + this.cancelRecoveryTimer(); + } + + const event: CircuitBreakerEvent = { + endpoint: this.endpointUrl, + chainId: this.chainId, + previousState, + newState, + timestamp: Date.now(), + reason, + failuresAtTransition: this.consecutiveFailures, + }; + + this.emit('stateChange', event); + } + + private scheduleRecovery(): void { + this.cancelRecoveryTimer(); + this.recoveryTimer = setTimeout(() => { + if (this.state === 'OPEN') { + this.transitionTo('HALF_OPEN', 'Recovery timer fired'); + } + }, this.config.recoveryTimeoutMs); + } + + private cancelRecoveryTimer(): void { + if (this.recoveryTimer !== null) { + clearTimeout(this.recoveryTimer); + this.recoveryTimer = null; + } + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Errors +// ───────────────────────────────────────────────────────────────────────────── + +export class CircuitOpenError extends Error { + readonly endpointUrl: string; + readonly chainId: number; + readonly circuitState: CircuitState; + readonly code = 'RPC_CIRCUIT_OPEN'; + + constructor(endpointUrl: string, chainId: number, circuitState: CircuitState) { + super( + `Circuit breaker is ${circuitState} for RPC endpoint ${endpointUrl} (chain ${chainId}). ` + + 'Request blocked to prevent cascading failure.', + ); + this.name = 'CircuitOpenError'; + this.endpointUrl = endpointUrl; + this.chainId = chainId; + this.circuitState = circuitState; + Object.setPrototypeOf(this, new.target.prototype); + } +} + +export class RpcTimeoutError extends Error { + readonly endpointUrl: string; + readonly chainId: number; + readonly timeoutMs: number; + readonly code = 'RPC_TIMEOUT'; + + constructor(endpointUrl: string, chainId: number, timeoutMs: number) { + super( + `RPC call to ${endpointUrl} (chain ${chainId}) timed out after ${timeoutMs}ms.`, + ); + this.name = 'RpcTimeoutError'; + this.endpointUrl = endpointUrl; + this.chainId = chainId; + this.timeoutMs = timeoutMs; + Object.setPrototypeOf(this, new.target.prototype); + } +} + +export class AllProvidersFailedError extends Error { + readonly chainId: number; + readonly attemptedEndpoints: string[]; + readonly code = 'RPC_ALL_PROVIDERS_FAILED'; + + constructor(chainId: number, attemptedEndpoints: string[]) { + super( + `All RPC providers failed for chain ${chainId}. Attempted endpoints: ${attemptedEndpoints.join(', ')}`, + ); + this.name = 'AllProvidersFailedError'; + this.chainId = chainId; + this.attemptedEndpoints = attemptedEndpoints; + Object.setPrototypeOf(this, new.target.prototype); + } +} diff --git a/backend/services/rpc/index.ts b/backend/services/rpc/index.ts new file mode 100644 index 00000000..05a10f3c --- /dev/null +++ b/backend/services/rpc/index.ts @@ -0,0 +1,64 @@ +/** + * RPC Circuit Breaker & Timeout Module — Issue #RPC-CB + * + * Provides configurable timeout, circuit breaker, and provider fallback + * for external blockchain RPC calls. + */ + +// ── Configuration ──────────────────────────────────────────────────────────── + +export { + DEFAULT_CIRCUIT_BREAKER_CONFIG, + DEFAULT_RPC_GLOBAL_CONFIG, + DEFAULT_CHAIN_ENDPOINTS, + DEFAULT_STELLAR_CHAIN_CONFIG, + DEFAULT_ENDPOINT_TIMEOUT_MS, + resolveEndpointUrl, +} from './rpcConfig'; + +export type { + CircuitBreakerConfig, + RpcEndpointConfig, + RpcChainConfig, + RpcGlobalConfig, +} from './rpcConfig'; + +// ── Circuit Breaker ────────────────────────────────────────────────────────── + +export { + CircuitBreaker, + CircuitOpenError, + RpcTimeoutError, + AllProvidersFailedError, +} from './circuitBreaker'; + +export type { + CircuitState, + CircuitStateSnapshot, + CircuitBreakerEvent, +} from './circuitBreaker'; + +// ── RPC Provider Fallback ──────────────────────────────────────────────────── + +export { + RpcProviderFallback, +} from './rpcProviderFallback'; + +export type { + RpcCallOptions, + RpcCallResult, +} from './rpcProviderFallback'; + +// ── Monitor Service ────────────────────────────────────────────────────────── + +export { + RpcMonitorService, + rpcMonitorService, +} from './rpcMonitorService'; + +export type { + RpcMonitorMetrics, + ChainHealthSummary, + RpcMonitorDashboard, + RpcDashboardQuery, +} from './rpcMonitorService'; diff --git a/backend/services/rpc/rpcConfig.ts b/backend/services/rpc/rpcConfig.ts new file mode 100644 index 00000000..a94da309 --- /dev/null +++ b/backend/services/rpc/rpcConfig.ts @@ -0,0 +1,311 @@ +/** + * RPC Endpoint & Circuit Breaker Configuration — Issue #RPC-CB + * + * Provides typed configuration for: + * - Timeout per RPC endpoint (ms) + * - Circuit breaker parameters (failure threshold, recovery timeout, half-open max requests) + * - Provider fallback ordering per chain + * - Graceful degradation behaviour + */ + +// ───────────────────────────────────────────────────────────────────────────── +// Circuit breaker configuration +// ───────────────────────────────────────────────────────────────────────────── + +export interface CircuitBreakerConfig { + /** Number of consecutive failures before the circuit opens. Default: 5 */ + failureThreshold: number; + /** Milliseconds to wait before transitioning from OPEN to HALF_OPEN. Default: 30_000 */ + recoveryTimeoutMs: number; + /** Max requests allowed in HALF_OPEN state to probe the endpoint. Default: 3 */ + halfOpenMaxRequests: number; + /** If true, the circuit automatically resets after recoveryTimeoutMs. Default: true */ + autoReset: boolean; +} + +// ───────────────────────────────────────────────────────────────────────────── +// Per-endpoint configuration +// ───────────────────────────────────────────────────────────────────────────── + +export interface RpcEndpointConfig { + /** Chain ID this endpoint serves (1 = Ethereum Mainnet, 137 = Polygon, etc.). */ + chainId: number; + /** Human-readable label (e.g. 'Cloudflare Ethereum Public RPC'). */ + label: string; + /** The RPC URL (https or wss). */ + url: string; + /** Optional WebSocket URL for subscriptions (falls back to url if omitted). */ + wsUrl?: string; + /** Request timeout in milliseconds. Default: 10_000 */ + timeoutMs: number; + /** Optional custom headers to attach to every RPC request. */ + headers?: Record; + /** Circuit breaker settings for this endpoint. */ + circuitBreaker: CircuitBreakerConfig; + /** Whether this endpoint supports read-only calls. Default: true */ + supportsRead: boolean; + /** Whether this endpoint supports write (transaction submission). Default: true */ + supportsWrite: boolean; + /** Max requests per second (soft rate-limit). 0 = unlimited. Default: 0 */ + maxRps: number; +} + +// ───────────────────────────────────────────────────────────────────────────── +// Chain-level configuration (groups endpoints for a chain) +// ───────────────────────────────────────────────────────────────────────────── + +export interface RpcChainConfig { + /** Chain ID. */ + chainId: number; + /** Human-readable chain name (e.g. 'Ethereum', 'Polygon'). */ + chainName: string; + /** Ordered list of RPC endpoint configurations. First is primary, rest are fallbacks. */ + endpoints: RpcEndpointConfig[]; + /** + * Behaviour when circuit is open and no fallback responds: + * - 'fail_fast' — throw instantly (default) + * - 'cache_stale' — return last successful cached response (if available) + * - 'degraded' — return a minimal degraded response + */ + onCircuitOpen: 'fail_fast' | 'cache_stale' | 'degraded'; + /** + * Behaviour when all providers have been exhausted: + * - 'fail_fast' — throw instantly (default) + * - 'cache_stale' — return last successful cached response + */ + onAllFailed: 'fail_fast' | 'cache_stale'; +} + +// ───────────────────────────────────────────────────────────────────────────── +// Global RPC configuration +// ───────────────────────────────────────────────────────────────────────────── + +export interface RpcGlobalConfig { + /** Global default timeout applied to any endpoint without an explicit timeoutMs. */ + defaultTimeoutMs: number; + /** Global default circuit breaker config applied when not specified per-endpoint. */ + defaultCircuitBreaker: CircuitBreakerConfig; + /** Whether to enable circuit breaker logging. Default: true */ + enableLogging: boolean; + /** Whether to emit metrics for Prometheus collection. Default: true */ + enableMetrics: boolean; +} + +// ───────────────────────────────────────────────────────────────────────────── +// Defaults +// ───────────────────────────────────────────────────────────────────────────── + +export const DEFAULT_CIRCUIT_BREAKER_CONFIG: CircuitBreakerConfig = { + failureThreshold: 5, + recoveryTimeoutMs: 30_000, + halfOpenMaxRequests: 3, + autoReset: true, +}; + +export const DEFAULT_RPC_GLOBAL_CONFIG: RpcGlobalConfig = { + defaultTimeoutMs: 10_000, + defaultCircuitBreaker: { ...DEFAULT_CIRCUIT_BREAKER_CONFIG }, + enableLogging: true, + enableMetrics: true, +}; + +export const DEFAULT_ENDPOINT_TIMEOUT_MS = 10_000; + +// ───────────────────────────────────────────────────────────────────────────── +// EVM chain defaults (mirrors src/config/evm.ts with fallback providers) +// ───────────────────────────────────────────────────────────────────────────── + +/** + * Built-in RPC endpoint configs with primary + fallback providers per chain. + * These mirror the existing EVM_RPC_URLS in src/config/evm.ts but add + * fallback providers and timeout/circuit-breaker settings. + */ +export const DEFAULT_CHAIN_ENDPOINTS: Record = { + 1: { + chainId: 1, + chainName: 'Ethereum Mainnet', + endpoints: [ + { + chainId: 1, + label: 'Cloudflare Ethereum Gateway', + url: 'https://cloudflare-eth.com', + timeoutMs: 10_000, + circuitBreaker: { ...DEFAULT_CIRCUIT_BREAKER_CONFIG }, + supportsRead: true, + supportsWrite: false, + maxRps: 100, + }, + { + chainId: 1, + label: 'Ethereum Foundation (eth-mainnet)', + url: 'https://mainnet.infura.io/v3/${INFURA_PROJECT_ID}', + timeoutMs: 10_000, + circuitBreaker: { failureThreshold: 3, recoveryTimeoutMs: 15_000, halfOpenMaxRequests: 2, autoReset: true }, + supportsRead: true, + supportsWrite: true, + maxRps: 200, + }, + { + chainId: 1, + label: 'Alchemy Ethereum (eth-mainnet)', + url: 'https://eth-mainnet.g.alchemy.com/v2/${ALCHEMY_API_KEY}', + timeoutMs: 10_000, + circuitBreaker: { failureThreshold: 3, recoveryTimeoutMs: 15_000, halfOpenMaxRequests: 2, autoReset: true }, + supportsRead: true, + supportsWrite: true, + maxRps: 200, + }, + ], + onCircuitOpen: 'cache_stale', + onAllFailed: 'cache_stale', + }, + 137: { + chainId: 137, + chainName: 'Polygon Mainnet', + endpoints: [ + { + chainId: 137, + label: 'Polygon Public RPC', + url: 'https://polygon-rpc.com', + timeoutMs: 15_000, + circuitBreaker: { ...DEFAULT_CIRCUIT_BREAKER_CONFIG }, + supportsRead: true, + supportsWrite: false, + maxRps: 50, + }, + { + chainId: 137, + label: 'Polygon Infura', + url: 'https://polygon-mainnet.infura.io/v3/${INFURA_PROJECT_ID}', + timeoutMs: 10_000, + circuitBreaker: { failureThreshold: 3, recoveryTimeoutMs: 15_000, halfOpenMaxRequests: 2, autoReset: true }, + supportsRead: true, + supportsWrite: true, + maxRps: 200, + }, + ], + onCircuitOpen: 'fail_fast', + onAllFailed: 'cache_stale', + }, + 42161: { + chainId: 42161, + chainName: 'Arbitrum One', + endpoints: [ + { + chainId: 42161, + label: 'Arbitrum Public RPC', + url: 'https://arb1.arbitrum.io/rpc', + timeoutMs: 15_000, + circuitBreaker: { ...DEFAULT_CIRCUIT_BREAKER_CONFIG }, + supportsRead: true, + supportsWrite: false, + maxRps: 50, + }, + { + chainId: 42161, + label: 'Arbitrum Infura', + url: 'https://arbitrum-mainnet.infura.io/v3/${INFURA_PROJECT_ID}', + timeoutMs: 10_000, + circuitBreaker: { failureThreshold: 3, recoveryTimeoutMs: 15_000, halfOpenMaxRequests: 2, autoReset: true }, + supportsRead: true, + supportsWrite: true, + maxRps: 200, + }, + ], + onCircuitOpen: 'fail_fast', + onAllFailed: 'cache_stale', + }, + 10: { + chainId: 10, + chainName: 'Optimism', + endpoints: [ + { + chainId: 10, + label: 'Optimism Public RPC', + url: 'https://mainnet.optimism.io', + timeoutMs: 15_000, + circuitBreaker: { ...DEFAULT_CIRCUIT_BREAKER_CONFIG }, + supportsRead: true, + supportsWrite: false, + maxRps: 50, + }, + { + chainId: 10, + label: 'Optimism Infura', + url: 'https://optimism-mainnet.infura.io/v3/${INFURA_PROJECT_ID}', + timeoutMs: 10_000, + circuitBreaker: { failureThreshold: 3, recoveryTimeoutMs: 15_000, halfOpenMaxRequests: 2, autoReset: true }, + supportsRead: true, + supportsWrite: true, + maxRps: 200, + }, + ], + onCircuitOpen: 'fail_fast', + onAllFailed: 'cache_stale', + }, + 8453: { + chainId: 8453, + chainName: 'Base', + endpoints: [ + { + chainId: 8453, + label: 'Base Public RPC', + url: 'https://mainnet.base.org', + timeoutMs: 15_000, + circuitBreaker: { ...DEFAULT_CIRCUIT_BREAKER_CONFIG }, + supportsRead: true, + supportsWrite: false, + maxRps: 50, + }, + ], + onCircuitOpen: 'fail_fast', + onAllFailed: 'cache_stale', + }, +}; + +// ───────────────────────────────────────────────────────────────────────────── +// Stellar (Soroban) defaults +// ───────────────────────────────────────────────────────────────────────────── + +export const DEFAULT_STELLAR_CHAIN_CONFIG: RpcChainConfig = { + chainId: 0x8000, // Stellar chain ID convention + chainName: 'Stellar (Soroban)', + endpoints: [ + { + chainId: 0x8000, + label: 'Stellar Soroban RPC', + url: 'https://soroban-rpc.stellar.org', + timeoutMs: 15_000, + circuitBreaker: { ...DEFAULT_CIRCUIT_BREAKER_CONFIG }, + supportsRead: true, + supportsWrite: true, + maxRps: 100, + }, + { + chainId: 0x8000, + label: 'Stellar Soroban Testnet', + url: 'https://soroban-testnet.stellar.org', + timeoutMs: 15_000, + circuitBreaker: { failureThreshold: 3, recoveryTimeoutMs: 15_000, halfOpenMaxRequests: 2, autoReset: true }, + supportsRead: true, + supportsWrite: true, + maxRps: 100, + }, + ], + onCircuitOpen: 'cache_stale', + onAllFailed: 'cache_stale', +}; + +// ───────────────────────────────────────────────────────────────────────────── +// Helper: resolve an endpoint URL with env var placeholders +// ───────────────────────────────────────────────────────────────────────────── + +/** + * Replaces ${ENV_VAR_NAME} placeholders in a URL string with actual environment + * variable values. Unknown variables are replaced with an empty string. + */ +export function resolveEndpointUrl(url: string): string { + return url.replace(/\$\{(\w+)\}/g, (_match, varName) => { + return process.env[varName] ?? ''; + }); +} diff --git a/backend/services/rpc/rpcMonitorService.ts b/backend/services/rpc/rpcMonitorService.ts new file mode 100644 index 00000000..6d099207 --- /dev/null +++ b/backend/services/rpc/rpcMonitorService.ts @@ -0,0 +1,487 @@ +/** + * RPC Monitor Service — Issue #RPC-CB + * + * Central monitoring service for all RPC circuit breakers. + * Provides: + * - Per-endpoint circuit state tracking + * - Aggregated chain health summaries + * - Prometheus-style metrics + * - Alerting on circuit state changes + * - Manual circuit reset + * - Historical event log + */ + +import type { CircuitBreakerEvent, CircuitStateSnapshot } from './circuitBreaker'; + +// ───────────────────────────────────────────────────────────────────────────── +// Types +// ───────────────────────────────────────────────────────────────────────────── + +export interface RpcMonitorMetrics { + /** Total RPC calls made. */ + totalCalls: number; + /** Successful RPC calls. */ + successfulCalls: number; + /** Failed RPC calls. */ + failedCalls: number; + /** Calls skipped due to open circuit. */ + skippedCalls: number; + /** Degraded responses served from cache/fallback. */ + degradedResponses: number; + /** Average response time in ms. */ + avgResponseTimeMs: number; + /** P50 response time in ms. */ + p50ResponseTimeMs: number; + /** P95 response time in ms. */ + p95ResponseTimeMs: number; + /** P99 response time in ms. */ + p99ResponseTimeMs: number; +} + +export interface ChainHealthSummary { + chainId: number; + chainName: string; + totalEndpoints: number; + healthyEndpoints: number; + degradedEndpoints: number; + openEndpoints: number; + halfOpenEndpoints: number; + closedEndpoints: number; + overallHealth: 'healthy' | 'degraded' | 'unhealthy'; + metrics: RpcMonitorMetrics; + circuits: CircuitStateSnapshot[]; +} + +export interface RpcMonitorDashboard { + timestamp: string; + totalChains: number; + healthyChains: number; + degradedChains: number; + unhealthyChains: number; + globalMetrics: RpcMonitorMetrics; + chains: ChainHealthSummary[]; + recentEvents: CircuitBreakerEvent[]; +} + +export interface RpcDashboardQuery { + chainId?: number; + endpointUrl?: string; + eventLimit?: number; +} + +// ───────────────────────────────────────────────────────────────────────────── +// RPC Monitor Service +// ───────────────────────────────────────────────────────────────────────────── + +export class RpcMonitorService { + private circuits = new Map(); + private events: CircuitBreakerEvent[] = []; + private readonly maxEvents = 10_000; + + // Per-endpoint call metrics + private callMetrics = new Map(); + + // Chain-level degraded response tracking + private degradedResponses: { + chainId: number; + method: string; + mode: string; + timestamp: number; + }[] = []; + + // ── Public API ──────────────────────────────────────────────────────────── + + /** + * Record a circuit breaker state change event. + */ + recordCircuitEvent(event: CircuitBreakerEvent): void { + this.events.push(event); + if (this.events.length > this.maxEvents) { + this.events = this.events.slice(-this.maxEvents); + } + } + + /** + * Record a successful RPC call. + */ + recordSuccess(endpointUrl: string, chainId: number, durationMs: number): void { + const key = this.metricKey(endpointUrl, chainId); + const metrics = this.getOrCreateMetrics(key); + metrics.totalCalls++; + metrics.successes++; + metrics.responseTimes.push(durationMs); + + // Keep last 1000 response times for percentile calculation + if (metrics.responseTimes.length > 1000) { + metrics.responseTimes = metrics.responseTimes.slice(-1000); + } + + } + + /** + * Record a failed RPC call. + */ + recordFailure(endpointUrl: string, chainId: number, error: string, durationMs: number): void { + const key = this.metricKey(endpointUrl, chainId); + const metrics = this.getOrCreateMetrics(key); + metrics.totalCalls++; + metrics.failures++; + metrics.responseTimes.push(durationMs); + metrics.lastError = error; + metrics.lastErrorAt = Date.now(); + + if (metrics.responseTimes.length > 1000) { + metrics.responseTimes = metrics.responseTimes.slice(-1000); + } + + } + + /** + * Record a call that was skipped due to open circuit. + */ + recordSkippedCall(endpointUrl: string, chainId: number, circuitState: string): void { + const key = this.metricKey(endpointUrl, chainId); + const metrics = this.getOrCreateMetrics(key); + metrics.totalCalls++; + metrics.skipped++; + metrics.lastError = `Skipped (circuit ${circuitState})`; + metrics.lastErrorAt = Date.now(); + } + + /** + * Record a degraded response (served from cache or fallback). + */ + recordDegradedResponse(chainId: number, method: string, mode: string): void { + this.degradedResponses.push({ chainId, method, mode, timestamp: Date.now() }); + if (this.degradedResponses.length > 1000) { + this.degradedResponses = this.degradedResponses.slice(-1000); + } + } + + /** + * Update or register a circuit state snapshot. + */ + registerCircuit(snapshot: CircuitStateSnapshot): void { + const key = this.circuitKey(snapshot.chainId, snapshot.endpointUrl); + this.circuits.set(key, snapshot); + } + + /** + * Get the full dashboard of all RPC circuit states and metrics. + */ + getDashboard(query: RpcDashboardQuery = {}): RpcMonitorDashboard { + const now = Date.now(); + const chainGroups = this.groupCircuitsByChain(); + + const chains: ChainHealthSummary[] = []; + + for (const [chainIdStr, chainCircuits] of chainGroups) { + const chainId = Number(chainIdStr); + const healthSummary = this.computeChainHealth(chainId, chainCircuits); + chains.push(healthSummary); + } + + // Apply chain filter + const filteredChains = query.chainId + ? chains.filter((c) => c.chainId === query.chainId) + : chains; + + // Apply endpoint filter to circuits within chains + const endpointFiltered = query.endpointUrl + ? filteredChains.map((ch) => ({ + ...ch, + circuits: ch.circuits.filter((c) => c.endpointUrl.includes(query.endpointUrl!)), + })) + : filteredChains; + + const totalChains = endpointFiltered.length; + const healthyChains = endpointFiltered.filter((c) => c.overallHealth === 'healthy').length; + const degradedChains = endpointFiltered.filter((c) => c.overallHealth === 'degraded').length; + const unhealthyChains = endpointFiltered.filter((c) => c.overallHealth === 'unhealthy').length; + + const globalMetrics = this.computeGlobalMetrics(); + + const eventLimit = query.eventLimit ?? 100; + const recentEvents = this.events.slice(-eventLimit); + + return { + timestamp: new Date(now).toISOString(), + totalChains, + healthyChains, + degradedChains, + unhealthyChains, + globalMetrics, + chains: endpointFiltered, + recentEvents, + }; + } + + /** + * Get health summary for a specific chain. + */ + getChainHealth(chainId: number): ChainHealthSummary | null { + const dashboard = this.getDashboard({ chainId }); + return dashboard.chains[0] ?? null; + } + + /** + * Get recent circuit events. + */ + getRecentEvents(limit = 50): CircuitBreakerEvent[] { + return this.events.slice(-limit); + } + + /** + * Get Prometheus-formatted metrics for all circuits. + */ + getPrometheusMetrics(): string { + const lines: string[] = []; + + for (const snapshot of this.circuits.values()) { + const labels = `chain="${snapshot.chainId}",endpoint="${snapshot.endpointLabel}",url="${snapshot.endpointUrl}"`; + + lines.push(`# HELP rpc_circuit_state Current circuit breaker state (0=CLOSED, 1=OPEN, 2=HALF_OPEN)`); + lines.push(`# TYPE rpc_circuit_state gauge`); + const stateValue = snapshot.state === 'CLOSED' ? 0 : snapshot.state === 'OPEN' ? 1 : 2; + lines.push(`rpc_circuit_state{${labels}} ${stateValue}`); + + lines.push(`# HELP rpc_circuit_consecutive_failures Consecutive failures count`); + lines.push(`# TYPE rpc_circuit_consecutive_failures gauge`); + lines.push(`rpc_circuit_consecutive_failures{${labels}} ${snapshot.consecutiveFailures}`); + + lines.push(`# HELP rpc_circuit_total_failures Total failures since last reset`); + lines.push(`# TYPE rpc_circuit_total_failures counter`); + lines.push(`rpc_circuit_total_failures{${labels}} ${snapshot.totalFailures}`); + + lines.push(`# HELP rpc_circuit_total_successes Total successes since last reset`); + lines.push(`# TYPE rpc_circuit_total_successes counter`); + lines.push(`rpc_circuit_total_successes{${labels}} ${snapshot.totalSuccesses}`); + + lines.push(`# HELP rpc_circuit_downtime_ms Cumulative downtime in ms`); + lines.push(`# TYPE rpc_circuit_downtime_ms gauge`); + lines.push(`rpc_circuit_downtime_ms{${labels}} ${snapshot.cumulativeDowntimeMs}`); + + // Per-endpoint call metrics + const key = this.circuitKey(snapshot.chainId, snapshot.endpointUrl); + const metric = this.callMetrics.get(key); + if (metric) { + lines.push(`# HELP rpc_call_total Total RPC calls`); + lines.push(`# TYPE rpc_call_total counter`); + lines.push(`rpc_call_total{${labels}} ${metric.totalCalls}`); + + lines.push(`# HELP rpc_call_successes Successful RPC calls`); + lines.push(`# TYPE rpc_call_successes counter`); + lines.push(`rpc_call_successes{${labels}} ${metric.successes}`); + + lines.push(`# HELP rpc_call_failures Failed RPC calls`); + lines.push(`# TYPE rpc_call_failures counter`); + lines.push(`rpc_call_failures{${labels}} ${metric.failures}`); + } + } + + return lines.join('\n'); + } + + /** + * Get total aggregated metrics across all chains. + */ + getAggregatedMetrics(): RpcMonitorMetrics { + return this.computeGlobalMetrics(); + } + + /** + * Reset metrics for all circuits. + */ + resetAllMetrics(): void { + this.callMetrics.clear(); + this.degradedResponses = []; + } + + // ── Private ─────────────────────────────────────────────────────────────── + + private metricKey(endpointUrl: string, chainId: number): string { + return `${chainId}:${endpointUrl}`; + } + + private circuitKey(chainId: number, endpointUrl: string): string { + return `${chainId}:${endpointUrl}`; + } + + private getOrCreateMetrics(key: string): { + totalCalls: number; + successes: number; + failures: number; + skipped: number; + degraded: number; + responseTimes: number[]; + lastError: string | null; + lastErrorAt: number | null; + } { + if (!this.callMetrics.has(key)) { + this.callMetrics.set(key, { + totalCalls: 0, + successes: 0, + failures: 0, + skipped: 0, + degraded: 0, + responseTimes: [], + lastError: null, + lastErrorAt: null, + }); + } + return this.callMetrics.get(key)!; + } + + + + private groupCircuitsByChain(): Map { + const groups = new Map(); + for (const snapshot of this.circuits.values()) { + const key = String(snapshot.chainId); + if (!groups.has(key)) { + groups.set(key, []); + } + groups.get(key)!.push(snapshot); + } + return groups; + } + + private computeChainHealth( + chainId: number, + circuits: CircuitStateSnapshot[], + ): ChainHealthSummary { + const totalEndpoints = circuits.length; + let openEndpoints = 0; + let halfOpenEndpoints = 0; + let closedEndpoints = 0; + let degradedEndpoints = 0; + + for (const c of circuits) { + if (c.state === 'OPEN') openEndpoints++; + else if (c.state === 'HALF_OPEN') halfOpenEndpoints++; + else if (c.state === 'CLOSED') closedEndpoints++; + + if (c.consecutiveFailures > 0) degradedEndpoints++; + } + + const healthyEndpoints = closedEndpoints; + + let overallHealth: 'healthy' | 'degraded' | 'unhealthy'; + if (openEndpoints > 0 && closedEndpoints === 0) { + overallHealth = 'unhealthy'; + } else if (openEndpoints > 0 || halfOpenEndpoints > 0 || degradedEndpoints > 0) { + overallHealth = 'degraded'; + } else { + overallHealth = 'healthy'; + } + + // Compute per-chain metrics + const chainKey = String(chainId); + let totalCalls = 0; + let successes = 0; + let failures = 0; + let skipped = 0; + let degraded = 0; + const allResponseTimes: number[] = []; + + for (const [key, metrics] of this.callMetrics) { + if (key.startsWith(chainKey + ':')) { + totalCalls += metrics.totalCalls; + successes += metrics.successes; + failures += metrics.failures; + skipped += metrics.skipped; + degraded += metrics.degraded; + allResponseTimes.push(...metrics.responseTimes); + } + } + + // Count degraded responses for this chain + const chainDegradedCount = this.degradedResponses.filter( + (d) => d.chainId === chainId, + ).length; + + return { + chainId, + chainName: circuits[0]?.endpointLabel ?? `Chain ${chainId}`, + totalEndpoints, + healthyEndpoints, + degradedEndpoints, + openEndpoints, + halfOpenEndpoints, + closedEndpoints, + overallHealth, + metrics: this.computeMetrics(totalCalls, successes, failures, skipped, degraded + chainDegradedCount, allResponseTimes), + circuits, + }; + } + + private computeMetrics( + totalCalls: number, + successes: number, + failures: number, + skipped: number, + degraded: number, + responseTimes: number[], + ): RpcMonitorMetrics { + const sorted = [...responseTimes].sort((a, b) => a - b); + const len = sorted.length; + + const avg = len > 0 ? sorted.reduce((a, b) => a + b, 0) / len : 0; + const p50 = len > 0 ? sorted[Math.floor(len * 0.5)] : 0; + const p95 = len > 0 ? sorted[Math.floor(len * 0.95)] : 0; + const p99 = len > 0 ? sorted[Math.floor(len * 0.99)] : 0; + + return { + totalCalls, + successfulCalls: successes, + failedCalls: failures, + skippedCalls: skipped, + degradedResponses: degraded, + avgResponseTimeMs: Math.round(avg), + p50ResponseTimeMs: p50, + p95ResponseTimeMs: p95, + p99ResponseTimeMs: p99, + }; + } + + private computeGlobalMetrics(): RpcMonitorMetrics { + let totalCalls = 0; + let totalSuccesses = 0; + let totalFailures = 0; + let totalSkipped = 0; + let totalDegraded = 0; + const allResponseTimes: number[] = []; + + for (const metrics of this.callMetrics.values()) { + totalCalls += metrics.totalCalls; + totalSuccesses += metrics.successes; + totalFailures += metrics.failures; + totalSkipped += metrics.skipped; + totalDegraded += metrics.degraded; + allResponseTimes.push(...metrics.responseTimes); + } + + totalDegraded += this.degradedResponses.length; + + return this.computeMetrics( + totalCalls, + totalSuccesses, + totalFailures, + totalSkipped, + totalDegraded, + allResponseTimes, + ); + } +} + +/** Singleton instance. */ +export const rpcMonitorService = new RpcMonitorService(); diff --git a/backend/services/rpc/rpcProviderFallback.ts b/backend/services/rpc/rpcProviderFallback.ts new file mode 100644 index 00000000..4f0fd774 --- /dev/null +++ b/backend/services/rpc/rpcProviderFallback.ts @@ -0,0 +1,467 @@ +/** + * RPC Provider with Fallback — Issue #RPC-CB + * + * Wraps the circuit breaker pattern around actual RPC calls with: + * - Configurable timeout per call + * - Automatic fallback to alternative providers on failure + * - Graceful degradation (cache_stale / degraded / fail_fast) + * - Request-level abort via AbortController + * + * Usage: + * const provider = new RpcProviderFallback(ethereumConfig, rpcMonitorService); + * const result = await provider.call('eth_blockNumber', [], { method: 'POST' }); + * // Or for read calls that can tolerate stale data: + * const result = await provider.callWithFallback('eth_call', [tx], { method: 'POST', dataType: 'read' }); + */ + +import { + type RpcChainConfig, + type RpcEndpointConfig, + resolveEndpointUrl, + DEFAULT_ENDPOINT_TIMEOUT_MS, +} from './rpcConfig'; +import { + CircuitBreaker, + RpcTimeoutError, + AllProvidersFailedError, +} from './circuitBreaker'; +import type { RpcMonitorService } from './rpcMonitorService'; + +// ───────────────────────────────────────────────────────────────────────────── +// Types +// ───────────────────────────────────────────────────────────────────────────── + +export interface RpcCallOptions { + /** HTTP method. Default: 'POST' */ + method?: string; + /** Optional custom headers beyond what the endpoint provides. */ + headers?: Record; + /** + * Type of data being requested. + * 'read' — Read-only call, can fall back to cached data. + * 'write' — Transaction submission, must attempt all providers. + * 'query' — General query, follows standard fallback logic. + */ + dataType?: 'read' | 'write' | 'query'; + /** Custom timeout override for this call (ms). */ + timeoutMs?: number; + /** AbortSignal for cancelling the request externally. */ + signal?: AbortSignal; +} + +export interface RpcCallResult { + /** Parsed JSON response from the RPC provider. */ + data: T; + /** Which endpoint URL served this request. */ + usedEndpoint: string; + /** Which provider (index in the endpoints array) served this request. */ + usedProviderIndex: number; + /** Total time taken including fallback attempts. */ + durationMs: number; + /** Whether this was served from a degraded/cached source. */ + degraded: boolean; + /** Error details if a fallback occurred. */ + fallbackErrors?: { endpoint: string; error: string; durationMs: number }[]; +} + +// ───────────────────────────────────────────────────────────────────────────── +// JSON-RPC Request/Response types +// ───────────────────────────────────────────────────────────────────────────── + +interface JsonRpcRequest { + jsonrpc: '2.0'; + method: string; + params: unknown[]; + id: number; +} + +interface JsonRpcResponse { + jsonrpc: '2.0'; + id: number; + result?: T; + error?: { code: number; message: string; data?: unknown }; +} + +let jsonRpcIdCounter = 1; +function nextId(): number { + return jsonRpcIdCounter++; +} + +// ───────────────────────────────────────────────────────────────────────────── +// RPC Provider Fallback +// ───────────────────────────────────────────────────────────────────────────── + +export class RpcProviderFallback { + private readonly chainConfig: RpcChainConfig; + private readonly circuitBreakers: Map = new Map(); + private readonly monitor: RpcMonitorService | null; + + /** Cache of last successful responses per JSON-RPC method (for degraded mode). */ + private lastSuccessfulCache = new Map(); + private cacheTimestamps = new Map(); + + constructor( + chainConfig: RpcChainConfig, + monitor?: RpcMonitorService, + ) { + this.chainConfig = chainConfig; + this.monitor = monitor ?? null; + + // Create a circuit breaker for each endpoint + for (const endpoint of chainConfig.endpoints) { + const cb = new CircuitBreaker(endpoint); + cb.on('stateChange', (event) => { + this.monitor?.recordCircuitEvent(event); + }); + this.circuitBreakers.set(endpoint.url, cb); + } + } + + // ── Public API ──────────────────────────────────────────────────────────── + + /** + * Execute an RPC call on the chain, using circuit breaker and automatic + * fallback across configured providers. + * + * @param method JSON-RPC method (e.g. 'eth_blockNumber', 'eth_call') + * @param params Parameters array + * @param options Call options + * @returns The RPC result + * @throws CircuitOpenError — all circuits are open + * @throws AllProvidersFailedError — all providers exhausted + * @throws RpcTimeoutError — request timed out + */ + async call( + method: string, + params: unknown[], + options: RpcCallOptions = {}, + ): Promise> { + const startTime = Date.now(); + const fallbackErrors: { endpoint: string; error: string; durationMs: number }[] = []; + const cacheKey = this.cacheKey(method, params); + + // Determine which endpoints to try based on data type + const endpointsToTry = this.getEligibleEndpoints(options.dataType ?? 'query'); + + if (endpointsToTry.length === 0) { + // All circuits are open — attempt degradation + return this.handleDegraded(cacheKey, method, Date.now() - startTime); + } + + for (let i = 0; i < endpointsToTry.length; i++) { + const endpoint = endpointsToTry[i]; + const cb = this.circuitBreakers.get(endpoint.url); + + // Check circuit breaker — allowRequest throws CircuitOpenError if fully open + if (!cb) { + fallbackErrors.push({ + endpoint: endpoint.url, + error: 'No circuit breaker registered', + durationMs: 0, + }); + continue; + } + + try { + cb.allowRequest(); + } catch (err) { + const errorMsg = err instanceof Error ? err.message : String(err); + fallbackErrors.push({ + endpoint: endpoint.url, + error: errorMsg, + durationMs: 0, + }); + this.monitor?.recordSkippedCall(endpoint.url, this.chainConfig.chainId, cb.getState()); + continue; + } + + const isProbe = cb.getState() === 'HALF_OPEN'; + const attemptStart = Date.now(); + + try { + const result = await this.executeSingleCall( + endpoint, + method, + params, + options, + ); + + cb.recordSuccess(isProbe); + this.cacheResponse(cacheKey, result); + + const elapsed = Date.now() - startTime; + this.monitor?.recordSuccess(endpoint.url, this.chainConfig.chainId, elapsed); + + return { + data: result, + usedEndpoint: endpoint.url, + usedProviderIndex: this.chainConfig.endpoints.indexOf(endpoint), + durationMs: elapsed, + degraded: false, + fallbackErrors: fallbackErrors.length > 0 ? fallbackErrors : undefined, + }; + } catch (err) { + const attemptDuration = Date.now() - attemptStart; + cb.recordFailure(isProbe); + + const errorMsg = err instanceof Error ? err.message : String(err); + fallbackErrors.push({ + endpoint: endpoint.url, + error: errorMsg, + durationMs: attemptDuration, + }); + + this.monitor?.recordFailure(endpoint.url, this.chainConfig.chainId, errorMsg, attemptDuration); + + // Continue to next provider + } + } + + // All providers exhausted — handle degradation + return this.handleDegraded(cacheKey, method, Date.now() - startTime); + } + + /** + * Convenience method for read calls — uses 'read' data type. + */ + async callRead( + method: string, + params: unknown[], + options: Omit = {}, + ): Promise> { + return this.call(method, params, { ...options, dataType: 'read' }); + } + + /** + * Convenience method for write (transaction) calls — uses 'write' data type, + * which will try all available endpoints even if partially degraded. + */ + async callWrite( + method: string, + params: unknown[], + options: Omit = {}, + ): Promise> { + return this.call(method, params, { ...options, dataType: 'write' }); + } + + /** + * Returns all circuit breaker snapshots for this chain's endpoints. + */ + getCircuitStates(): ReturnType[] { + return Array.from(this.circuitBreakers.values()).map((cb) => cb.snapshot()); + } + + /** + * Manually reset the circuit breaker for a specific endpoint URL. + */ + manualResetEndpoint(endpointUrl: string): boolean { + const cb = this.circuitBreakers.get(endpointUrl); + if (!cb) return false; + cb.manualReset(); + return true; + } + + /** + * Manually reset all circuit breakers for this chain. + */ + manualResetAll(): void { + for (const cb of this.circuitBreakers.values()) { + cb.manualReset(); + } + } + + /** + * Manually open the circuit for a specific endpoint URL. + */ + manualOpenEndpoint(endpointUrl: string): boolean { + const cb = this.circuitBreakers.get(endpointUrl); + if (!cb) return false; + cb.manualOpen(); + return true; + } + + /** Update the chain configuration (hot-reload). */ + updateChainConfig(config: RpcChainConfig): void { + this.chainConfig.endpoints = config.endpoints; + // Add circuit breakers for any new endpoints + for (const endpoint of config.endpoints) { + if (!this.circuitBreakers.has(endpoint.url)) { + const cb = new CircuitBreaker(endpoint); + cb.on('stateChange', (event) => { + this.monitor?.recordCircuitEvent(event); + }); + this.circuitBreakers.set(endpoint.url, cb); + } + } + } + + /** Get the chain ID. */ + getChainId(): number { + return this.chainConfig.chainId; + } + + /** Get the chain name. */ + getChainName(): string { + return this.chainConfig.chainName; + } + + /** Get the degradation policy for this chain. */ + getDegradationPolicy(): RpcChainConfig['onCircuitOpen'] { + return this.chainConfig.onCircuitOpen; + } + + // ── Private ─────────────────────────────────────────────────────────────── + + /** + * Execute a single RPC call against an endpoint with timeout. + */ + private async executeSingleCall( + endpoint: RpcEndpointConfig, + method: string, + params: unknown[], + options: RpcCallOptions, + ): Promise { + const url = resolveEndpointUrl(endpoint.url); + const timeoutMs = options.timeoutMs ?? endpoint.timeoutMs ?? DEFAULT_ENDPOINT_TIMEOUT_MS; + + const controller = new AbortController(); + const timeoutId = setTimeout(() => controller.abort(), timeoutMs); + + // Combine options.signal with our timeout + const combinedSignal = options.signal + ? this.combineSignals(options.signal, controller.signal) + : controller.signal; + + // Handle abort from the caller + if (options.signal?.aborted) { + clearTimeout(timeoutId); + throw new RpcTimeoutError(url, endpoint.chainId, timeoutMs); + } + + const requestBody: JsonRpcRequest = { + jsonrpc: '2.0', + method, + params, + id: nextId(), + }; + + const headers: Record = { + 'Content-Type': 'application/json', + ...endpoint.headers, + ...options.headers, + }; + + try { + const response = await fetch(url, { + method: options.method ?? 'POST', + headers, + body: JSON.stringify(requestBody), + signal: combinedSignal, + }); + + clearTimeout(timeoutId); + + if (!response.ok) { + throw new Error(`HTTP ${response.status}: ${response.statusText}`); + } + + const json = (await response.json()) as JsonRpcResponse; + + if (json.error) { + throw new Error(`JSON-RPC error ${json.error.code}: ${json.error.message}`); + } + + return json.result as T; + } catch (err) { + clearTimeout(timeoutId); + + if (err instanceof Error && err.name === 'AbortError') { + throw new RpcTimeoutError(url, endpoint.chainId, timeoutMs); + } + + throw err; + } + } + + /** + * Returns endpoints eligible for the given data type (respects circuit state). + * Callers must also check allowRequest() on each endpoint's circuit breaker. + */ + private getEligibleEndpoints(dataType: 'read' | 'write' | 'query'): RpcEndpointConfig[] { + const allEndpoints = this.chainConfig.endpoints; + + return allEndpoints.filter((ep) => { + // Filter by read/write capability + if (dataType === 'read' && !ep.supportsRead) return false; + if (dataType === 'write' && !ep.supportsWrite) return false; + return true; + }); + } + + /** + * Handle degradation when all providers fail. + */ + private async handleDegraded( + cacheKey: string, + method: string, + totalDurationMs: number, + ): Promise> { + + // Attempt cache_stale or degraded response + if (this.chainConfig.onAllFailed === 'cache_stale') { + const cached = this.lastSuccessfulCache.get(cacheKey); + if (cached !== undefined) { + this.monitor?.recordDegradedResponse(this.chainConfig.chainId, method, 'cache_stale'); + + return { + data: cached as T, + usedEndpoint: 'degraded_cache', + usedProviderIndex: -1, + durationMs: totalDurationMs, + degraded: true, + fallbackErrors: [ + { endpoint: 'all', error: 'All providers exhausted, serving stale cache', durationMs: totalDurationMs }, + ], + }; + } + } + + // If all providers failed and no cache fallback, throw + this.monitor?.recordDegradedResponse(this.chainConfig.chainId, method, 'fail_fast'); + + throw new AllProvidersFailedError( + this.chainConfig.chainId, + this.chainConfig.endpoints.map((e) => e.url), + ); + } + + private cacheKey(method: string, params: unknown[]): string { + return `${method}:${JSON.stringify(params)}`; + } + + private cacheResponse(key: string, data: unknown): void { + this.lastSuccessfulCache.set(key, data); + this.cacheTimestamps.set(key, Date.now()); + + // Prune cache if it grows too large (keep last 500 entries) + if (this.lastSuccessfulCache.size > 500) { + const oldestKey = this.cacheTimestamps.entries().next().value?.[0]; + if (oldestKey) { + this.lastSuccessfulCache.delete(oldestKey); + this.cacheTimestamps.delete(oldestKey); + } + } + } + + private combineSignals(...signals: AbortSignal[]): AbortSignal { + const controller = new AbortController(); + for (const signal of signals) { + if (signal.aborted) { + controller.abort(signal.reason); + return controller.signal; + } + signal.addEventListener('abort', () => controller.abort(signal.reason), { once: true }); + } + return controller.signal; + } +} diff --git a/backend/services/shared/apiResponse.ts b/backend/services/shared/apiResponse.ts index a8b1b587..4331cb56 100644 --- a/backend/services/shared/apiResponse.ts +++ b/backend/services/shared/apiResponse.ts @@ -162,7 +162,12 @@ export type ErrorCode = | 'PAYMENT_GATEWAY_ERROR' | 'PAYMENT_GATEWAY_FALLBACK_FAILED' | 'PAYMENT_GATEWAY_CONFIG_INVALID' - | 'PAYMENT_REFUND_PARTIAL_FAILED'; + | 'PAYMENT_REFUND_PARTIAL_FAILED' + // ── RPC / Circuit Breaker (RPC timeout & circuit breaker feature) ───────── + | 'RPC_TIMEOUT' + | 'RPC_CIRCUIT_OPEN' + | 'RPC_ALL_PROVIDERS_FAILED' + | 'RPC_PROVIDER_NOT_FOUND'; /** * Maps each error code to the HTTP status code that should be sent to the @@ -242,6 +247,11 @@ export const ERROR_HTTP_STATUS_MAP: Record = { PAYMENT_GATEWAY_FALLBACK_FAILED: 502, PAYMENT_GATEWAY_CONFIG_INVALID: 422, PAYMENT_REFUND_PARTIAL_FAILED: 422, + // RPC / Circuit Breaker + RPC_TIMEOUT: 504, + RPC_CIRCUIT_OPEN: 503, + RPC_ALL_PROVIDERS_FAILED: 503, + RPC_PROVIDER_NOT_FOUND: 404, }; // ─────────────────────────────────────────────────────────────────────────────