From 729cf3ed837aaa12c16d7d113939eb3d0af1201d Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 17 Jun 2026 20:35:55 +0000 Subject: [PATCH] fix(scanner): auto-restart block sync on crash instead of stopping MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Catch-up sync ran inside a single try/catch alongside the WebSocket startup. When a block failed (e.g. a sustained RPC outage that exhausted the per-request retries), runCatchup re-threw, the outer catch logged "API will run without scanner", and the real-time WS path never even started — sync stopped for good until a process restart. Wrap catch-up in a supervisor that restarts it with exponential backoff. Progress is checkpointed per block, so each restart resumes from the last synced block with no lost or duplicated work. Once catch-up completes, the (self-healing) WS subscription path starts as before. The supervisor runs detached so a slow/retrying catch-up never blocks startup, and a shutdown flag stops the loop cleanly on SIGINT/SIGTERM. Backoff is configurable via SCANNER_RESTART_BASE_DELAY_MS (5s) and SCANNER_RESTART_MAX_DELAY_MS (60s). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01BEG3HfBuwNhijzL168Rkvv --- .env.example | 4 + README.md | 2 + src/env.ts | 5 ++ src/index.ts | 134 ++++++++++++++++++++------------ src/services/catchup.service.ts | 49 ++++++++++++ test/catchup.test.ts | 113 ++++++++++++++++++++++++++- 6 files changed, 255 insertions(+), 52 deletions(-) diff --git a/.env.example b/.env.example index 7876c62..6a77044 100644 --- a/.env.example +++ b/.env.example @@ -26,6 +26,10 @@ SCANNER_FAILURE_WINDOW_MS=300000 SCANNER_COOLDOWN_MS=60000 WS_RECONNECT_BASE_DELAY_MS=1000 WS_RECONNECT_MAX_DELAY_MS=30000 +# Catch-up auto-restart backoff (when block sync crashes, it restarts from the +# last synced block instead of stopping) +SCANNER_RESTART_BASE_DELAY_MS=5000 +SCANNER_RESTART_MAX_DELAY_MS=60000 # Metadata resolution IPFS_GATEWAY_URLS=https://ipfs.io/ipfs/|https://cloudflare-ipfs.com/ipfs/|https://dweb.link/ipfs/ diff --git a/README.md b/README.md index e45be5c..1e34b49 100644 --- a/README.md +++ b/README.md @@ -382,6 +382,8 @@ Copy `.env.example` to `.env` and configure: | `SCANNER_TX_CONCURRENCY` | `8` | Concurrent transaction processing limit | | `WS_RECONNECT_BASE_DELAY_MS` | `1000` | WebSocket reconnect backoff base | | `WS_RECONNECT_MAX_DELAY_MS` | `30000` | WebSocket reconnect backoff ceiling | +| `SCANNER_RESTART_BASE_DELAY_MS` | `5000` | Backoff base before restarting catch-up after a crash | +| `SCANNER_RESTART_MAX_DELAY_MS` | `60000` | Backoff ceiling between catch-up restart attempts | | `IPFS_GATEWAY_URLS` | `https://ipfs.io/ipfs/|https://cloudflare-ipfs.com/ipfs/|https://dweb.link/ipfs/` | Ordered IPFS gateway list for metadata resolution | | `METADATA_HTTP_TIMEOUT_MS` | `10000` | HTTP metadata fetch timeout | | `METADATA_IPFS_TIMEOUT_MS` | `30000` | IPFS gateway fetch timeout per attempt | diff --git a/src/env.ts b/src/env.ts index daf049f..deb73ed 100644 --- a/src/env.ts +++ b/src/env.ts @@ -42,6 +42,11 @@ export const env = { SCANNER_COOLDOWN_MS: parseInt(process.env.SCANNER_COOLDOWN_MS || '60000', 10), WS_RECONNECT_BASE_DELAY_MS: parseInt(process.env.WS_RECONNECT_BASE_DELAY_MS || '1000', 10), WS_RECONNECT_MAX_DELAY_MS: parseInt(process.env.WS_RECONNECT_MAX_DELAY_MS || '30000', 10), + // Block-sync supervisor: when catch-up crashes, wait then restart it (resuming + // from the persisted checkpoint) instead of giving up. Backoff grows + // exponentially from base up to max between restart attempts. + SCANNER_RESTART_BASE_DELAY_MS: parseInt(process.env.SCANNER_RESTART_BASE_DELAY_MS || '5000', 10), + SCANNER_RESTART_MAX_DELAY_MS: parseInt(process.env.SCANNER_RESTART_MAX_DELAY_MS || '60000', 10), // Metadata resolution IPFS_GATEWAY_URLS: ( diff --git a/src/index.ts b/src/index.ts index 422aa07..b5ba05c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -6,11 +6,11 @@ import { env } from './env' import { log } from './utils/logger' import { MongoPoolManager } from './utils/mongoPoolManager' import type { MongoConnectionParams } from './utils/mongodb' -import { getChainConfig } from './config/chains' +import { getChainConfig, type ChainConfig } from './config/chains' import { buildSearchX402Endpoints } from './config/x402-search' import { ScannerService } from './services/scanner.service' import { WsSubscriptionService } from './services/ws-subscription.service' -import { runCatchup } from './services/catchup.service' +import { runCatchupWithRestart } from './services/catchup.service' import { reResolveStaleMetadata, retryFailedResolutions } from './services/agent-metadata.service' import { createHealthRouter } from './controllers/health.controller' import { createAgentsRouter } from './controllers/agents.controller' @@ -93,6 +93,7 @@ if (x402App) { // --- State --- let scanner: ScannerService | null = null let wsService: WsSubscriptionService | null = null +let scannerStopping = false let httpServer: ReturnType | null = null let x402HttpServer: ReturnType | null = null let reResolveTimer: ReturnType | null = null @@ -291,6 +292,78 @@ async function stopDashboardWsServer(): Promise { }) } +// --- Scanner sync supervisor --- +// Drives catch-up sync and, once caught up, the real-time WebSocket subscriptions. +// Catch-up can crash mid-sync (e.g. a sustained RPC outage that exhausts the +// per-request retries). Because progress is checkpointed per block, we wait and +// restart from the persisted block instead of giving up — block sync keeps +// retrying rather than stopping for good. The WS path self-heals on its own +// (reconnect + RPC cooldown), so it is started once after catch-up succeeds. +async function startScannerSyncWithRestart( + scannerService: ScannerService, + chainConfig: ChainConfig, +): Promise { + const latestBlock = await runCatchupWithRestart( + scannerService, + chainConfig.chainId, + env.CATCHUP_BATCH_SIZE, + { + baseDelayMs: env.SCANNER_RESTART_BASE_DELAY_MS, + maxDelayMs: env.SCANNER_RESTART_MAX_DELAY_MS, + shouldStop: () => scannerStopping, + }, + ) + + // latestBlock is null only when shutdown interrupted catch-up before it completed. + if (scannerStopping || latestBlock === null) return + + // Real-time subscriptions for ongoing blocks/logs. + wsService = new WsSubscriptionService({ + wsUrl: chainConfig.wsUrl, + chainId: chainConfig.chainId, + scanner: scannerService, + baseDelayMs: env.WS_RECONNECT_BASE_DELAY_MS, + maxDelayMs: env.WS_RECONNECT_MAX_DELAY_MS, + failureWindowMs: env.SCANNER_FAILURE_WINDOW_MS, + cooldownMs: env.SCANNER_COOLDOWN_MS, + onEvent: (type, data) => { + broadcastSSE(type, data) + }, + }) + await wsService.start() + log({ level: 'info', msg: 'WebSocket subscription service started' }) + + reResolveTimer = setInterval(async () => { + try { + const count = await reResolveStaleMetadata( + chainConfig.chainId, + env.METADATA_RE_RESOLVE_MAX_AGE_MS, + env.METADATA_RE_RESOLVE_BATCH_SIZE, + ) + if (count > 0) { + log({ level: 'info', msg: `Re-resolved ${count} stale agent metadata entries` }) + } + } catch (error) { + log({ level: 'error', msg: 'Periodic metadata re-resolution failed', error }) + } + }, env.METADATA_RE_RESOLVE_INTERVAL_MS) + + retryTimer = setInterval(async () => { + try { + const count = await retryFailedResolutions( + chainConfig.chainId, + env.METADATA_RETRY_MAX_AGE_MS, + env.METADATA_RETRY_BATCH_SIZE, + ) + if (count > 0) { + log({ level: 'info', msg: `Retried ${count} failed metadata resolutions` }) + } + } catch (error) { + log({ level: 'error', msg: 'Failed metadata retry job error', error }) + } + }, env.METADATA_RETRY_INTERVAL_MS) +} + // --- Startup --- async function startServer(): Promise { // 1. Initialize MongoDB @@ -364,56 +437,12 @@ async function startServer(): Promise { chainId: chainConfig.chainId, })) - // 5. Run catch-up from last synced block to latest - log({ level: 'info', msg: 'Starting catch-up sync...' }) - const latestBlock = await runCatchup(scanner, chainConfig.chainId, env.CATCHUP_BATCH_SIZE) - log({ level: 'info', msg: `Catch-up complete. Latest block: ${latestBlock}` }) - - // 6. Start WebSocket subscriptions for real-time events - wsService = new WsSubscriptionService({ - wsUrl: chainConfig.wsUrl, - chainId: chainConfig.chainId, - scanner, - baseDelayMs: env.WS_RECONNECT_BASE_DELAY_MS, - maxDelayMs: env.WS_RECONNECT_MAX_DELAY_MS, - failureWindowMs: env.SCANNER_FAILURE_WINDOW_MS, - cooldownMs: env.SCANNER_COOLDOWN_MS, - onEvent: (type, data) => { - broadcastSSE(type, data) - }, + // 5. Run catch-up + real-time sync in the background, restarting on crash. + // Detached so a long-running (or repeatedly retrying) catch-up never + // blocks startup — the API is already serving above. + void startScannerSyncWithRestart(scanner, chainConfig).catch((error) => { + log({ level: 'error', msg: 'Scanner sync supervisor exited unexpectedly', error }) }) - await wsService.start() - log({ level: 'info', msg: 'WebSocket subscription service started' }) - - reResolveTimer = setInterval(async () => { - try { - const count = await reResolveStaleMetadata( - chainConfig.chainId, - env.METADATA_RE_RESOLVE_MAX_AGE_MS, - env.METADATA_RE_RESOLVE_BATCH_SIZE, - ) - if (count > 0) { - log({ level: 'info', msg: `Re-resolved ${count} stale agent metadata entries` }) - } - } catch (error) { - log({ level: 'error', msg: 'Periodic metadata re-resolution failed', error }) - } - }, env.METADATA_RE_RESOLVE_INTERVAL_MS) - - retryTimer = setInterval(async () => { - try { - const count = await retryFailedResolutions( - chainConfig.chainId, - env.METADATA_RETRY_MAX_AGE_MS, - env.METADATA_RETRY_BATCH_SIZE, - ) - if (count > 0) { - log({ level: 'info', msg: `Retried ${count} failed metadata resolutions` }) - } - } catch (error) { - log({ level: 'error', msg: 'Failed metadata retry job error', error }) - } - }, env.METADATA_RETRY_INTERVAL_MS) } catch (error) { log({ level: 'error', msg: 'Scanner initialization failed', error }) log({ level: 'warn', msg: 'API will run without scanner. Existing data remains queryable.' }) @@ -429,6 +458,9 @@ signals.forEach((signal) => { process.on(signal, async () => { log({ level: 'info', msg: `Received ${signal}. Starting graceful shutdown...` }) + // Stop the catch-up restart loop from scheduling further attempts. + scannerStopping = true + if (wsService) { wsService.stop() log({ level: 'info', msg: 'WebSocket subscription service stopped' }) diff --git a/src/services/catchup.service.ts b/src/services/catchup.service.ts index 445c890..1cf1361 100644 --- a/src/services/catchup.service.ts +++ b/src/services/catchup.service.ts @@ -1,5 +1,6 @@ import { log } from '../utils/logger' import { env } from '../env' +import { sleep } from '../utils/retry' import { getChainSyncState } from '../repositories/chain-state.repository' import type { ScannerService } from './scanner.service' @@ -49,6 +50,54 @@ export async function runCatchup( return latestBlock } +export interface CatchupRestartOptions { + /** Initial delay before the first restart attempt. */ + baseDelayMs: number + /** Maximum delay between restart attempts (exponential backoff ceiling). */ + maxDelayMs: number + /** Returns true when sync should stop scheduling further restarts (e.g. shutdown). */ + shouldStop?: () => boolean +} + +/** + * Run catch-up, automatically restarting it after a crash with exponential + * backoff. Catch-up progress is checkpointed per block, so each restart resumes + * from the last synced block — no work is lost or repeated. Without this, a + * single failed block would abort sync permanently. + * + * Resolves with the latest block reached once catch-up completes, or null if it + * stopped (via shouldStop) before completing. + */ +export async function runCatchupWithRestart( + scanner: ScannerService, + chainId: number, + batchSize: number, + options: CatchupRestartOptions, +): Promise { + const shouldStop = options.shouldStop ?? (() => false) + let restartDelayMs = options.baseDelayMs + + while (!shouldStop()) { + try { + log({ level: 'info', msg: 'Starting catch-up sync...' }) + const latestBlock = await runCatchup(scanner, chainId, batchSize) + log({ level: 'info', msg: `Catch-up complete. Latest block: ${latestBlock}` }) + return latestBlock + } catch (error) { + if (shouldStop()) return null + log({ + level: 'error', + msg: `Catch-up sync crashed; restarting in ${restartDelayMs}ms (resumes from last synced block)`, + error, + }) + await sleep(restartDelayMs) + restartDelayMs = Math.min(restartDelayMs * 2, options.maxDelayMs) + } + } + + return null +} + /** * Fill gap blocks between lastSyncedBlock and a target block. */ diff --git a/test/catchup.test.ts b/test/catchup.test.ts index 5102368..6a5a416 100644 --- a/test/catchup.test.ts +++ b/test/catchup.test.ts @@ -13,10 +13,16 @@ vi.mock('../src/repositories/chain-state.repository', () => ({ getChainSyncState: vi.fn(), })) -import { runCatchup, fillGap } from '../src/services/catchup.service' +vi.mock('../src/utils/retry', () => ({ + sleep: vi.fn().mockResolvedValue(undefined), +})) + +import { runCatchup, runCatchupWithRestart, fillGap } from '../src/services/catchup.service' import { getChainSyncState } from '../src/repositories/chain-state.repository' +import { sleep } from '../src/utils/retry' const mockGetChainSyncState = vi.mocked(getChainSyncState) +const mockSleep = vi.mocked(sleep) function createMockScanner(latestBlock: number) { return { @@ -94,6 +100,111 @@ describe('runCatchup', () => { }) }) +describe('runCatchupWithRestart', () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + const restartOptions = { baseDelayMs: 10, maxDelayMs: 100 } + + it('returns the latest block without restarting when catch-up succeeds', async () => { + mockGetChainSyncState.mockResolvedValue({ + chainId: 1, + network: 'mainnet', + lastSyncedBlock: 10, + lastSyncedBlockHash: '0x', + updatedAt: Date.now(), + }) + + const scanner = createMockScanner(12) + const result = await runCatchupWithRestart(scanner as any, 1, 10, restartOptions) + + expect(result).toBe(12) + expect(scanner.processBlock).toHaveBeenCalledTimes(2) + expect(mockSleep).not.toHaveBeenCalled() + }) + + it('restarts after a crash and resumes from the persisted checkpoint', async () => { + mockGetChainSyncState.mockResolvedValue({ + chainId: 1, + network: 'mainnet', + lastSyncedBlock: 10, + lastSyncedBlockHash: '0x', + updatedAt: Date.now(), + }) + + const scanner = createMockScanner(12) + // First attempt: block 11 fails -> runCatchup rethrows. + // Second attempt: resumes from the checkpoint and completes. + scanner.processBlock.mockRejectedValueOnce(new Error('rpc timeout')) + + const result = await runCatchupWithRestart(scanner as any, 1, 10, restartOptions) + + expect(result).toBe(12) + expect(mockSleep).toHaveBeenCalledTimes(1) + expect(mockSleep).toHaveBeenCalledWith(10) + // 1 failed (11) + 2 succeeded (11, 12) on the retry + expect(scanner.processBlock).toHaveBeenCalledTimes(3) + expect(scanner.getLatestBlockNumber).toHaveBeenCalledTimes(2) + }) + + it('grows the restart delay exponentially up to the ceiling', async () => { + mockGetChainSyncState.mockResolvedValue({ + chainId: 1, + network: 'mainnet', + lastSyncedBlock: 10, + lastSyncedBlockHash: '0x', + updatedAt: Date.now(), + }) + + const scanner = createMockScanner(12) + scanner.processBlock + .mockRejectedValueOnce(new Error('boom')) + .mockRejectedValueOnce(new Error('boom')) + .mockRejectedValueOnce(new Error('boom')) + + const result = await runCatchupWithRestart(scanner as any, 1, 10, { baseDelayMs: 10, maxDelayMs: 25 }) + + expect(result).toBe(12) + // 10 -> 20 -> capped at 25 + expect(mockSleep.mock.calls.map((c) => c[0])).toEqual([10, 20, 25]) + }) + + it('does not run when shouldStop is already true', async () => { + const scanner = createMockScanner(12) + const result = await runCatchupWithRestart(scanner as any, 1, 10, { + ...restartOptions, + shouldStop: () => true, + }) + + expect(result).toBeNull() + expect(scanner.getLatestBlockNumber).not.toHaveBeenCalled() + expect(scanner.processBlock).not.toHaveBeenCalled() + }) + + it('stops after a crash without restarting when shouldStop flips to true', async () => { + mockGetChainSyncState.mockResolvedValue({ + chainId: 1, + network: 'mainnet', + lastSyncedBlock: 10, + lastSyncedBlockHash: '0x', + updatedAt: Date.now(), + }) + + const scanner = createMockScanner(12) + scanner.processBlock.mockRejectedValue(new Error('boom')) + + const shouldStop = vi.fn().mockReturnValueOnce(false).mockReturnValue(true) + const result = await runCatchupWithRestart(scanner as any, 1, 10, { + ...restartOptions, + shouldStop, + }) + + expect(result).toBeNull() + expect(mockSleep).not.toHaveBeenCalled() + }) +}) + describe('fillGap', () => { beforeEach(() => { vi.clearAllMocks()