Skip to content

closes #429: Add timeout and circuit breaker for external blockchain RPC calls #830

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sherifatolanike wants to merge 1 commit into
base: main
Choose a base branch
from
Open

closes #429: Add timeout and circuit breaker for external blockchain RPC calls #830

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
155 changes: 153 additions & 2 deletions backend/server.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -76,8 +81,8 @@ async function ensurePlanCache(pool: Pool): Promise<PlanCacheBootstrap> {
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.
Expand Down Expand Up @@ -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<number, RpcProviderFallback>();

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<RunningServer> {
const pool = options.pool ?? (await getPool());
Expand All @@ -166,6 +205,9 @@ export async function startServer(options: StartServerOptions = {}): Promise<Run

const rateLimitMw = buildRateLimitMiddleware();

// Initialize all RPC providers at startup so the dashboard is ready immediately
registerAllChainProviders();

const server = http.createServer(async (req, res) => {
const url = new URL(req.url ?? '/', `http://${req.headers.host ?? 'localhost'}`);
const { pathname } = url;
Expand All @@ -187,6 +229,110 @@ export async function startServer(options: StartServerOptions = {}): Promise<Run
return;
}

// -----------------------------------------------------------------
// RPC Circuit Breaker Dashboard GET /rpc/dashboard
// -----------------------------------------------------------------
if (pathname === '/rpc/dashboard' && method === 'GET') {
const chainIdParam = url.searchParams.get('chainId');
const endpointUrl = url.searchParams.get('endpointUrl');
const eventLimit = url.searchParams.get('eventLimit');

const query: { chainId?: number; endpointUrl?: string; eventLimit?: number } = {};
if (chainIdParam) query.chainId = Number(chainIdParam);
if (endpointUrl) query.endpointUrl = endpointUrl;
if (eventLimit) query.eventLimit = Number(eventLimit);

const dashboard = rpcMonitorService.getDashboard(query);
sendJson(res, 200, dashboard);
return;
}

// -----------------------------------------------------------------
// RPC Chain Health GET /rpc/health/:chainId
// -----------------------------------------------------------------
const rpcHealthMatch = pathname.match(/^\/rpc\/health\/(\d+)$/);
if (rpcHealthMatch && method === 'GET') {
const chainId = Number(rpcHealthMatch[1]);
const health = rpcMonitorService.getChainHealth(chainId);
if (!health) {
sendJson(res, 404, { error: `No health data for chain ${chainId}` });
return;
}
sendJson(res, 200, health);
return;
}

// -----------------------------------------------------------------
// RPC Circuit Reset POST /rpc/reset
// -----------------------------------------------------------------
if (pathname === '/rpc/reset' && method === 'POST') {
const body = (await readJsonBody(req)) as {
chainId: number;
endpointUrl?: string;
action: 'reset' | 'open' | 'reset_all' | 'reset_metrics';
};

if (!body.chainId || !body.action) {
sendJson(res, 400, { error: 'chainId and action are required' });
return;
}

const provider = rpcProviders.get(body.chainId);
if (!provider) {
sendJson(res, 404, { error: `No RPC provider for chain ${body.chainId}` });
return;
}

switch (body.action) {
case 'reset':
if (!body.endpointUrl) {
sendJson(res, 400, { error: 'endpointUrl is required for reset action' });
return;
}
provider.manualResetEndpoint(body.endpointUrl);
sendJson(res, 200, { success: true, action: 'reset', endpointUrl: body.endpointUrl });
return;
case 'open':
if (!body.endpointUrl) {
sendJson(res, 400, { error: 'endpointUrl is required for open action' });
return;
}
provider.manualOpenEndpoint(body.endpointUrl);
sendJson(res, 200, { success: true, action: 'open', endpointUrl: body.endpointUrl });
return;
case 'reset_all':
provider.manualResetAll();
sendJson(res, 200, { success: true, action: 'reset_all', chainId: body.chainId });
return;
case 'reset_metrics':
rpcMonitorService.resetAllMetrics();
sendJson(res, 200, { success: true, action: 'reset_metrics' });
return;
default:
sendJson(res, 400, { error: `Unknown action: ${body.action}` });
return;
}
}

// -----------------------------------------------------------------
// RPC Prometheus metrics GET /metrics/rpc
// -----------------------------------------------------------------
if (pathname === '/metrics/rpc' && method === 'GET') {
res.writeHead(200, { 'Content-Type': 'text/plain; version=0.0.4; charset=utf-8' });
res.end(rpcMonitorService.getPrometheusMetrics());
return;
}

// -----------------------------------------------------------------
// RPC Events GET /rpc/events
// -----------------------------------------------------------------
if (pathname === '/rpc/events' && method === 'GET') {
const limit = url.searchParams.get('limit');
const events = rpcMonitorService.getRecentEvents(limit ? Number(limit) : 50);
sendJson(res, 200, { events, count: events.length });
return;
}

// -----------------------------------------------------------------
// Prometheus metrics (bypass rate limiting)
// -----------------------------------------------------------------
Expand Down Expand Up @@ -364,6 +510,11 @@ export async function startServer(options: StartServerOptions = {}): Promise<Run
console.info(`[Server] GraphQL → POST /graphql`);
console.info(`[Server] Plans → /plans`);
console.info(`[Server] Metrics → GET /metrics/plan-cache`);
console.info(`[Server] RPC → GET /rpc/dashboard`);
console.info(`[Server] RPC → GET /rpc/health/:chainId`);
console.info(`[Server] RPC → POST /rpc/reset`);
console.info(`[Server] RPC → GET /rpc/events`);
console.info(`[Server] RPC → GET /metrics/rpc`);
console.info(`[Server] RateLimit → GET /rate-limits/analytics`);
console.info(`[Server] RateLimit → GET /rate-limits/status?apiKey=...`);
console.info(`[Server] RateLimit → POST /rate-limits/bypass`);
Expand Down
32 changes: 32 additions & 0 deletions backend/services/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -396,5 +396,37 @@ export type { IPaymentGateway, IPaymentRouter, PaymentRequest, PaymentResult, Re
export { buildRotationEmailHtml, buildRotationEmailText } from './notification/rotationEmailTemplate';
export type { RotationEmailData } from './notification/rotationEmailTemplate';

// ── RPC Circuit Breaker & Timeout (Issue #RPC-CB) ───────────────────────────
export {
RpcProviderFallback,
CircuitBreaker,
CircuitOpenError,
RpcTimeoutError,
AllProvidersFailedError,
RpcMonitorService,
rpcMonitorService,
DEFAULT_CIRCUIT_BREAKER_CONFIG,
DEFAULT_RPC_GLOBAL_CONFIG,
DEFAULT_CHAIN_ENDPOINTS,
DEFAULT_STELLAR_CHAIN_CONFIG,
resolveEndpointUrl,
} from './rpc';

export type {
CircuitBreakerConfig,
RpcEndpointConfig,
RpcChainConfig,
RpcGlobalConfig,
CircuitState,
CircuitStateSnapshot,
CircuitBreakerEvent,
RpcCallOptions,
RpcCallResult,
RpcMonitorMetrics,
ChainHealthSummary,
RpcMonitorDashboard,
RpcDashboardQuery,
} from './rpc';

// ── Monitoring — Lock Metrics (Issue #610) ────────────────────────────────────
export { collectLockMetrics, lockMetricsExporter } from '../monitoring/lockMetrics';
Loading