fix: add backoff, restart metric and crash tracking to ingester restarts

src/__tests__/status.test.ts

@@ -52,6 +52,7 @@ vi.mock('../config', () => ({
 
 import { registerRESTRoutes } from '../api/rest'
 import { registerNetworkSelector } from '../middleware/network'
+import { recordCrash, resetIngesterState } from '../ingester/restart'
 
 async function buildApp() {
   const app = Fastify({ logger: false })
@@ -70,6 +71,7 @@ const rowsByNetwork: Record<string, { last_ledger: number; last_processed_at: Da
 describe('GET /status', () => {
   beforeEach(() => {
     vi.clearAllMocks()
+    resetIngesterState()
     // Return a row only for the network the query actually filtered on; an
     // unfiltered SELECT would therefore hand a network the other one's row.
     mockQuery.mockImplementation(async (_sql: string, params: unknown[]) => ({
@@ -85,14 +87,14 @@ describe('GET /status', () => {
     expect(testnet.json().network).toBe('testnet')
     expect(testnet.json().lastIndexedLedger).toBe(111111)
     expect(testnet.json().watchedPairs).toEqual(['XLM:native/USDC:GTEST'])
-    expect(mockQuery.mock.calls[0][1]).toEqual(['testnet'])
+    expect(mockQuery.mockCalls[0][1]).toEqual(['testnet'])
 
     const mainnet = await app.inject({ method: 'GET', url: '/status?network=mainnet' })
     expect(mainnet.statusCode).toBe(200)
     expect(mainnet.json().network).toBe('mainnet')
     expect(mainnet.json().lastIndexedLedger).toBe(222222)
     expect(mainnet.json().watchedPairs).toEqual(['XLM:native/USDC:GMAIN'])
-    expect(mockQuery.mock.calls[1][1]).toEqual(['mainnet'])
+    expect(mockQuery.mockCalls[1][1]).toEqual(['mainnet'])
   })
 
   it("does not let one network's indexer rows bleed into the other", async () => {
@@ -117,7 +119,7 @@ describe('GET /status', () => {
     expect(body.ingestLagSeconds).toBeLessThan(60)
   })
 
-  it('returns nulls — never another network\'s data — when a network has not ingested', async () => {
+  it('returns nulls — never another network \'s data — when a network has not ingested', async () => {
     mockQuery.mockResolvedValue({ rows: [] })
     const app = await buildApp()
 
@@ -138,4 +140,20 @@ describe('GET /status', () => {
     expect(res.statusCode).toBe(400)
     expect(mockQuery).not.toHaveBeenCalled()
   })
+
+  it('surfaces the last crash per ingester and network', async () => {
+    recordCrash('prices', 'mainnet', new Error('boom'))
+    const app = await buildApp()
+
+    const res = await app.inject({ method: 'GET', url: '/status?network=mainnet' })
+    const body = res.json()
+
+    expect(body.network).toBe('mainnet')
+    expect(body.crashes).toBeDefined()
+    expect(body.crashes.length).toBe(1)
+    expect(body.crashes[0].name).toBe('prices')
+    expect(body.crashes[0].network).toBe('mainnet')
+    expect(body.crashes[0].message).toBe('boom')
+    expect(typeof body.crashes[0].at).toBe('string')
+  })
 })

src/index.ts

@@ -24,6 +24,7 @@ import {
   PRICE_POINTS_RETENTION_DAYS,
 } from './jobs/pricePointsPartitionJob'
 import { loadPersistedPairs, getActivePairs } from './pairsRegistry'
+import { ingesterRestarts } from './metrics'
 import { markBootComplete } from './health'
 
 const log = createLogger('startup')
@@ -187,11 +188,74 @@ async function main() {
   // the (venue, network) pair: a crash in, say, the Soroswap ingester on
   // mainnet only restarts that one instance and cannot affect SDEX/AMM or the
   // ingesters running on testnet.
+  //
+  // A crash-looping ingester must not be invisible: restarts are counted, the
+  // delay grows exponentially up to a ceiling, and the last crash is recorded
+  // so /status can surface it. A sustained healthy run resets the backoff so a
+  // one-off blip does not permanently slow restarts.
+  const INGESTER_RESTART_BASE_MS = 10_000
+  const INGESTER_RESTART_MAX_MS = 5 * 60_000
+  const INGESTER_HEALTHY_RESET_MS = 5 * 60_000
+
+  type IngesterKey = string
+  const ingesterKey = (name: string, network: NetworkName): IngesterKey => `${name}/${network}`
+
+  interface IngesterState {
+    failures: number
+    lastCrashAt: Date | null
+    lastCrashMessage: string | null
+  }
+  const ingesterState = new Map<IngesterKey, IngesterState>()
+
+  const getIngesterState = (name: string, network: NetworkName): IngesterState => {
+    const key = ingesterKey(name, network)
+    let state = ingesterState.get(key)
+    if (!state) {
+      state = { failures: 0, lastCrashAt: null, lastCrashMessage: null }
+      ingesterState.set(key, state)
+    }
+    return state
+  }
+
   const restartIngester = (name: string, network: NetworkName, fn: () => Promise<void>) => {
-    fn().catch(err => {
-      log.error({ ingester: name, network, err }, 'ingester crashed; restarting in 10s')
-      setTimeout(() => restartIngester(name, network, fn), 10_000)
-    })
+    const state = getIngesterState(name, network)
+    let healthyTimer: NodeJS.Timeout | undefined
+    let runId = 0
+
+    const run = () => {
+      fn().then(() => {
+        // fn() resolving means the ingester ran to completion. Treat a run
+        // that lasted at least INGESTER_HEALTHY_RESET_MS as healthy and reset
+        // the backoff; a shorter run keeps the accumulated failure count.
+        if (healthyTimer) clearTimeout(healthyTimer)
+        state.failures = 0
+      }).catch(err => {
+        if (healthyTimer) clearTimeout(healthyTimer)
+        state.failures += 1
+        state.lastCrashAt = new Date()
+        state.lastCrashMessage = (err as Error).message
+        ingesterRestarts.inc({ ingester: name, network })
+
+        const delay = Math.min(
+          INGESTER_RESTART_BASE_MS * 2 ** (state.failures - 1),
+          INGESTER_RESTART_MAX_MS,
+        )
+        log.error(
+          { err, ingester: name, network, failures: state.failures, delayMs: delay },
+          'ingester crashed, restarting after backoff',
+        )
+        setTimeout(run, delay)
+      })
+
+      // If the ingester stays up past the healthy threshold, reset the
+      // backoff even though fn() has not resolved (infinite loops never do).
+      healthyTimer = setTimeout(() => {
+        state.failures = 0
+      }, INGESTER_HEALTHY_RESET_MS)
+      healthyTimer.unref()
+    }
+
+    run()
   }
 
   const enabledNetworks = getEnabledNetworks()