diff --git a/.env.example b/.env.example index 6a77044..4647e52 100644 --- a/.env.example +++ b/.env.example @@ -12,6 +12,12 @@ ETH_RPC_URL=http://10.1.200.113:8545 ETH_WS_URL=ws://10.1.200.113:8546 # Max time (ms) for a single JSON-RPC request before timing out (slow nodes / large blocks) ETH_RPC_TIMEOUT_MS=120000 +# How long (ms) idle keep-alive sockets to the node stay open — keeps one persistent +# connection between request bursts instead of dialing a new one each time +ETH_RPC_FREE_SOCKET_TIMEOUT_MS=300000 +# RPC heartbeat interval (ms): keeps the connection warm and the latest-block cache +# fresh (served by /v1/health). Must be shorter than ETH_RPC_FREE_SOCKET_TIMEOUT_MS. +ETH_RPC_HEARTBEAT_INTERVAL_MS=15000 # Chain CHAIN_ID=1 diff --git a/package.json b/package.json index 053948f..29f8fa6 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@missionsquad/agentindex-api", - "version": "1.5.2", + "version": "1.5.3", "description": "AgentIndex blockchain scanner API", "main": "lib/index.js", "types": "lib/index.d.ts", @@ -30,7 +30,7 @@ "@missionsquad/x402-proxy": "^0.1.0", "cors": "^2.8.5", "dotenv": "^16.4.7", - "evmdecoder": "^0.0.70", + "evmdecoder": "^0.0.71", "express": "^4.21.2", "mongodb": "^6.12.0", "ws": "^8.18.0" diff --git a/src/controllers/health.controller.ts b/src/controllers/health.controller.ts index d51f1fb..825bb76 100644 --- a/src/controllers/health.controller.ts +++ b/src/controllers/health.controller.ts @@ -1,25 +1,28 @@ import { Router, Request, Response } from 'express' import { getChainSyncState } from '../repositories/chain-state.repository' -import type { ScannerService } from '../services/scanner.service' import type { HealthResponse } from '../types/api' import { env } from '../env' -export function createHealthRouter(scanner: ScannerService | null): Router { +export interface CachedLatestBlock { + value: number + /** Epoch ms when the RPC heartbeat last refreshed the value. */ + at: number +} + +export function createHealthRouter( + getCachedLatestBlock: (() => CachedLatestBlock | null) | null, +): Router { const router = Router() router.get('/', async (_req: Request, res: Response) => { try { const chainId = env.CHAIN_ID const syncState = await getChainSyncState(chainId) - let latestBlock: number | null = null - - try { - if (scanner) { - latestBlock = await scanner.getLatestBlockNumber() - } - } catch { - // Scanner may not be initialized - } + + // Served from the scanner's RPC heartbeat cache. Health probes must never + // issue live RPC calls: during an RPC outage every probe would stall for + // the full request timeout and pile more requests onto the failing node. + const latestBlock = getCachedLatestBlock?.()?.value ?? null const lastSynced = syncState?.lastSyncedBlock ?? null const syncLag = latestBlock !== null && lastSynced !== null diff --git a/src/env.ts b/src/env.ts index deb73ed..60c254d 100644 --- a/src/env.ts +++ b/src/env.ts @@ -28,6 +28,16 @@ export const env = { // out. Applied to evmdecoder's HTTP transport (node-fetch request timeout AND the // keep-alive agent socket timeout). Raise for slow nodes / large blocks. ETH_RPC_TIMEOUT_MS: parseInt(process.env.ETH_RPC_TIMEOUT_MS || '120000', 10), + // How long (ms) idle keep-alive sockets to the RPC node stay open. Long values + // let the scanner reuse one persistent connection between request bursts instead + // of dialing a new TCP connection each time — new-connection churn adds latency + // and can trip stateful middleboxes (NAT/firewall/IDS) on the path to the node. + // Requires evmdecoder >= 0.0.71. + ETH_RPC_FREE_SOCKET_TIMEOUT_MS: parseInt(process.env.ETH_RPC_FREE_SOCKET_TIMEOUT_MS || '300000', 10), + // Interval (ms) of the RPC heartbeat that refreshes the cached latest block + // number (served by /v1/health) and keeps the keep-alive connection warm. + // Must be shorter than ETH_RPC_FREE_SOCKET_TIMEOUT_MS. + ETH_RPC_HEARTBEAT_INTERVAL_MS: parseInt(process.env.ETH_RPC_HEARTBEAT_INTERVAL_MS || '15000', 10), // Chain CHAIN_ID: parseInt(process.env.CHAIN_ID || '1', 10), diff --git a/src/index.ts b/src/index.ts index b5ba05c..a18afd2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -13,6 +13,7 @@ import { WsSubscriptionService } from './services/ws-subscription.service' import { runCatchupWithRestart } from './services/catchup.service' import { reResolveStaleMetadata, retryFailedResolutions } from './services/agent-metadata.service' import { createHealthRouter } from './controllers/health.controller' +import type { CachedLatestBlock } from './controllers/health.controller' import { createAgentsRouter } from './controllers/agents.controller' import { createReputationRouter } from './controllers/reputation.controller' import { createAddressesRouter } from './controllers/addresses.controller' @@ -98,6 +99,9 @@ let httpServer: ReturnType | null = null let x402HttpServer: ReturnType | null = null let reResolveTimer: ReturnType | null = null let retryTimer: ReturnType | null = null +let rpcHeartbeatTimer: ReturnType | null = null +let rpcHeartbeatInFlight = false +let latestBlockCache: CachedLatestBlock | null = null const sseClients: Set = new Set() let dashboardWsServer: WebSocketServer | null = null let dashboardWsHeartbeatTimer: ReturnType | null = null @@ -143,7 +147,7 @@ if (x402App) { } // --- API Routes --- -app.use('/v1/health', createHealthRouter(scanner)) +app.use('/v1/health', createHealthRouter(null)) app.use('/v1/agents', createAgentsRouter()) app.use('/v1/reputation', createReputationRouter()) app.use('/v1/address', createAddressesRouter()) @@ -411,6 +415,7 @@ async function startServer(): Promise { abiDirectory: env.ABI_DIRECTORY, txConcurrency: env.SCANNER_TX_CONCURRENCY, rpcTimeoutMs: env.ETH_RPC_TIMEOUT_MS, + freeSocketTimeoutMs: env.ETH_RPC_FREE_SOCKET_TIMEOUT_MS, onEventFactsPersisted: (eventFacts) => { void publishDashboardActivityFromPersistedEventFacts(eventFacts).catch((error) => { log({ level: 'warn', msg: 'Failed to publish dashboard activity websocket events', error }) @@ -419,13 +424,35 @@ async function startServer(): Promise { }) await scanner.initialize() - // Re-register health router with initialized scanner - // (The initial registration used null scanner) + // Re-register health router with the latest-block cache accessor + // (The initial registration had no cache to read from) app._router.stack = app._router.stack.filter( (layer: { route?: { path?: string } }) => !layer.route || !layer.route.path?.startsWith('/v1/health'), ) - app.use('/v1/health', createHealthRouter(scanner)) + app.use('/v1/health', createHealthRouter(() => latestBlockCache)) + + // RPC heartbeat: refreshes the latest-block cache and, by polling more + // often than the keep-alive idle timeouts, keeps one warm persistent + // connection to the RPC node. New-connection churn is what trips stateful + // middleboxes between this container and the node, so the connection must + // never sit idle long enough to be torn down. + const runRpcHeartbeat = async (): Promise => { + if (rpcHeartbeatInFlight) return + rpcHeartbeatInFlight = true + try { + const value = await scanner!.getLatestBlockNumber() + latestBlockCache = { value, at: Date.now() } + } catch (error) { + log({ level: 'warn', msg: 'RPC heartbeat failed to fetch latest block', error }) + } finally { + rpcHeartbeatInFlight = false + } + } + void runRpcHeartbeat() + rpcHeartbeatTimer = setInterval(() => { + void runRpcHeartbeat() + }, env.ETH_RPC_HEARTBEAT_INTERVAL_MS) // Re-register transactions router with decoder access for real-time decode app._router.stack = app._router.stack.filter( @@ -476,6 +503,11 @@ signals.forEach((signal) => { retryTimer = null } + if (rpcHeartbeatTimer !== null) { + clearInterval(rpcHeartbeatTimer) + rpcHeartbeatTimer = null + } + if (sseClients.size > 0) { log({ level: 'info', msg: `Closing ${sseClients.size} SSE client(s)` }) for (const client of sseClients) { diff --git a/src/services/scanner.service.ts b/src/services/scanner.service.ts index 8030712..60282b3 100644 --- a/src/services/scanner.service.ts +++ b/src/services/scanner.service.ts @@ -30,6 +30,7 @@ interface ScannerOptions { abiDirectory?: string txConcurrency?: number rpcTimeoutMs?: number + freeSocketTimeoutMs?: number onEventFactsPersisted?: (eventFacts: EventFact[]) => void } @@ -51,6 +52,7 @@ export class ScannerService { private readonly abiDirectory: string | undefined private readonly txConcurrency: number private readonly rpcTimeoutMs: number + private readonly freeSocketTimeoutMs: number private readonly onEventFactsPersisted?: (eventFacts: EventFact[]) => void private readonly inFlightTx = new Map>() @@ -61,6 +63,7 @@ export class ScannerService { this.abiDirectory = opts.abiDirectory this.txConcurrency = Math.max(1, opts.txConcurrency ?? 8) this.rpcTimeoutMs = Math.max(1000, opts.rpcTimeoutMs ?? 120_000) + this.freeSocketTimeoutMs = Math.max(1000, opts.freeSocketTimeoutMs ?? 300_000) this.onEventFactsPersisted = opts.onEventFactsPersisted } @@ -72,6 +75,9 @@ export class ScannerService { // Configurable RPC request timeout. evmdecoder >= 0.0.70 propagates this to // both the node-fetch request timeout and the keep-alive agent socket timeout. timeout: this.rpcTimeoutMs, + // Hold idle keep-alive sockets open so consecutive request bursts reuse one + // persistent connection instead of dialing a new one per burst. + freeSocketTimeout: this.freeSocketTimeoutMs, }, }, abi: { diff --git a/test/controllers.test.ts b/test/controllers.test.ts index 6902a6e..444dea9 100644 --- a/test/controllers.test.ts +++ b/test/controllers.test.ts @@ -195,9 +195,7 @@ describe('Health Controller', () => { beforeEach(() => resetMocks()) it('GET / returns health status', async () => { - const router = createHealthRouter({ - getLatestBlockNumber: vi.fn().mockResolvedValue(105), - } as any) + const router = createHealthRouter(() => ({ value: 105, at: Date.now() })) const res = await invokeRoute(router, 'get', '/') const body = res.body as Record diff --git a/yarn.lock b/yarn.lock index 506cae6..a56572e 100644 --- a/yarn.lock +++ b/yarn.lock @@ -4652,10 +4652,10 @@ events@^3.3.0: resolved "https://registry.npmjs.org/events/-/events-3.3.0.tgz#31a95ad0a924e2d2c419a813aeb2c4e878ea7400" integrity sha512-mQw+2fkQbALzQ7V0MY0IqdnXNOeTtP4r0lN9z7AAawCXgqea7bDii20AYrIBrFd/Hx0M2Ocz6S111CaFkUcb0Q== -evmdecoder@^0.0.70: - version "0.0.70" - resolved "https://registry.yarnpkg.com/evmdecoder/-/evmdecoder-0.0.70.tgz#f824c4b0aa24aa142422d27583c041334f8fd499" - integrity sha512-HTWPr8hR0rvHioySoUU0YKXbIL0EUT95rGdvTY/EhXPuKZ0idQrLPXquwN7aOIdM8Hg6FHb7j7wSZFHArLfnjQ== +evmdecoder@^0.0.71: + version "0.0.71" + resolved "https://registry.npmjs.org/evmdecoder/-/evmdecoder-0.0.71.tgz#bea916ad84d0aebc139799cdb303b62b8e039dde" + integrity sha512-YDFOm9WPXG5TNsbo8aDvW6sk2JeB5oAOLJXiEu7SduYPeP/Wr1zlqTSQGGYZ0/9pUE52B8o3maOTXcm6jxieug== dependencies: "@splunkdlt/async-tasks" "^1.1.0" "@splunkdlt/cache" "^1.0.3"