feat(scanner): persistent RPC connection via keep-alive tuning + heartbeat

.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

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"

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

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),

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<typeof app.listen> | null = null
 let x402HttpServer: ReturnType<typeof app.listen> | null = null
 let reResolveTimer: ReturnType<typeof setInterval> | null = null
 let retryTimer: ReturnType<typeof setInterval> | null = null
+let rpcHeartbeatTimer: ReturnType<typeof setInterval> | null = null
+let rpcHeartbeatInFlight = false
+let latestBlockCache: CachedLatestBlock | null = null
 const sseClients: Set<Response> = new Set()
 let dashboardWsServer: WebSocketServer | null = null
 let dashboardWsHeartbeatTimer: ReturnType<typeof setInterval> | 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<void> {
         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<void> {
       })
       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<void> => {
+        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) {

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<string, Promise<void>>()
 
@@ -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: {

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<string, unknown>

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"