fix(scanner): auto-restart block sync on crash instead of stopping

.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/

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 |

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: (

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<typeof app.listen> | null = null
 let x402HttpServer: ReturnType<typeof app.listen> | null = null
 let reResolveTimer: ReturnType<typeof setInterval> | null = null
@@ -291,6 +292,78 @@ async function stopDashboardWsServer(): Promise<void> {
   })
 }
 
+// --- 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<void> {
+  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<void> {
   // 1. Initialize MongoDB
@@ -364,56 +437,12 @@ async function startServer(): Promise<void> {
         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' })

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<number | null> {
+  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.
  */

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