diff --git a/src/__tests__/status.test.ts b/src/__tests__/status.test.ts index 065bfe17..b006b353 100644 --- a/src/__tests__/status.test.ts +++ b/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 { 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') + }) }) diff --git a/src/index.ts b/src/index.ts index 5cf4a54f..5c8c9feb 100644 --- a/src/index.ts +++ b/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() + + 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) => { - 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()