From f6e1ebf2fb1dedc82fa296559381456e69f8d625 Mon Sep 17 00:00:00 2001 From: Simplicity team Date: Sat, 17 Oct 2026 18:07:12 +0100 Subject: [PATCH 1/4] fix: A crash-looping ingester is invisible (#331) --- src/__tests__/status.test.ts | 20 +++++++- src/index.ts | 88 ++++++++++++++++++++++++++++++++++-- 2 files changed, 103 insertions(+), 5 deletions(-) diff --git a/src/__tests__/status.test.ts b/src/__tests__/status.test.ts index 065bfe17..9c4995e2 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[]) => ({ @@ -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 18f372c4..28de2eab 100644 --- a/src/index.ts +++ b/src/index.ts @@ -10,12 +10,65 @@ import { redis } from './redis' import { pgPool } from './db' import { buildApp } from './app' import { fanOutManager } from './ws/fanout' +import { registry } from './metrics' import { venueAdapters, startVenueAdapter } from './ingest/venues' import { startSnapshotIngester } from './ingesters/snapshot' import { createAggregateQueue, startAggregateWorker, scheduleAggregateRefresh } from './jobs/aggregateRefresh' import { createSnapshotRetentionQueue, startSnapshotRetentionWorker, scheduleSnapshotRetention, pruneAllNetworks, SNAPSHOT_RETENTION_DAYS } from './jobs/snapshotRetention' import { loadPersistedPairs, getActivePairs } from './pairsRegistry' +import { Counter } from 'prom-client' + +// ── Ingester restart bookkeeping ──────────────────────────────────────────── +// A crash-looping ingester used to be invisible: it restarted every ten +// seconds forever, with no backoff, no metric, and no mention on /status. The +// state below makes that visible and bounded. + +const INGESTER_RESTART_BASE_MS = 10_000 +const INGESTER_RESTART_MAX_MS = 5 * 60_000 +// How long an ingester must run without crashing before its backoff resets, +// so a nightly blip does not permanently slow restarts. +const INGESTER_HEALTHY_RESET_MS = 10 * 60_000 + +export const ingesterRestartCounter = new Counter({ + name: 'lens_ingester_restarts_total', + help: 'Total number of times an ingester has crashed and been restarted', + labelNames: ['ingester', 'network'] as const, + registers: [registry], +}) + +interface IngesterCrashRecord { + time: Date + message: string +} + +interface IngesterState { + consecutiveFailures: number + lastCrash: IngesterCrashRecord | null +} + +const ingesterStates = new Map() + +const ingesterStateKey = (name: string, network: NetworkName) => `${name}/${network}` + +const getIngesterState = (name: string, network: NetworkName): IngesterState => { + const key = ingesterStateKey(name, network) + let state = ingesterStates.get(key) + if (!state) { + state = { consecutiveFailures: 0, lastCrash: null } + ingesterStates.set(key, state) + } + return state +} + +export const getIngesterLastCrash = (name: string, network: NetworkName): IngesterCrashRecord | null => + ingesterStates.get(ingesterStateKey(name, network))?.lastCrash ?? null + +export const getAllIngesterLastCrashes = (): Array<{ ingester: string; network: NetworkName; lastCrash: IngesterCrashRecord | null }> => + Array.from(ingesterStates.entries()).map(([key, state]) => { + const [ingester, network] = key.split('/') as [string, NetworkName] + return { ingester, network, lastCrash: state.lastCrash } + }) async function main() { // ── Ensure DB schema is up-to-date ──────────────────────────────────────── @@ -141,10 +194,37 @@ async function main() { // mainnet only restarts that one instance and cannot affect SDEX/AMM or the // ingesters running on testnet. const restartIngester = (name: string, network: NetworkName, fn: () => Promise) => { - fn().catch(err => { - console.error(`[lens] ${name}/${network} ingester crashed, restarting in 10s:`, err.message) - setTimeout(() => restartIngester(name, network, fn), 10_000) - }) + const state = getIngesterState(name, network) + const startedAt = Date.now() + fn() + .then(() => { + // A clean return is not a crash, but it also is not a healthy run — + // the ingesters are infinite loops, so returning means it stopped. + // Treat it as a failure so the backoff still applies. + throw new Error('ingester returned unexpectedly') + }) + .catch(err => { + const message = (err as Error)?.message ?? String(err) + state.lastCrash = { time: new Date(), message } + ingesterRestartCounter.inc({ ingester: name, network }) + + // Reset the backoff if the previous run was healthy for long enough. + const ranFor = Date.now() - startedAt + if (ranFor >= INGESTER_HEALTHY_RESET_MS) { + state.consecutiveFailures = 0 + } + state.consecutiveFailures += 1 + + const delay = Math.min( + INGESTER_RESTART_BASE_MS * 2 ** (state.consecutiveFailures - 1), + INGESTER_RESTART_MAX_MS, + ) + console.error( + `[lens] ${name}/${network} ingester crashed (failure #${state.consecutiveFailures}), restarting in ${delay}ms:`, + message, + ) + setTimeout(() => restartIngester(name, network, fn), delay) + }) } const enabledNetworks = getEnabledNetworks() From c5dfad4243dc4bf2ac4909984a10735fc7d58577 Mon Sep 17 00:00:00 2001 From: Simplicity team Date: Sat, 17 Oct 2026 20:02:53 +0100 Subject: [PATCH 2/4] fix: A crash-looping ingester is invisible (#331) --- src/index.ts | 116 ++++++++++++++------------------------------------- 1 file changed, 32 insertions(+), 84 deletions(-) diff --git a/src/index.ts b/src/index.ts index 28de2eab..33313c64 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,6 +1,5 @@ import 'dotenv/config' import { execSync } from 'child_process' - if (!process.env.DIRECT_DATABASE_URL && process.env.DATABASE_URL) { process.env.DIRECT_DATABASE_URL = process.env.DATABASE_URL } @@ -10,65 +9,14 @@ import { redis } from './redis' import { pgPool } from './db' import { buildApp } from './app' import { fanOutManager } from './ws/fanout' -import { registry } from './metrics' import { venueAdapters, startVenueAdapter } from './ingest/venues' import { startSnapshotIngester } from './ingesters/snapshot' import { createAggregateQueue, startAggregateWorker, scheduleAggregateRefresh } from './jobs/aggregateRefresh' import { createSnapshotRetentionQueue, startSnapshotRetentionWorker, scheduleSnapshotRetention, pruneAllNetworks, SNAPSHOT_RETENTION_DAYS } from './jobs/snapshotRetention' import { loadPersistedPairs, getActivePairs } from './pairsRegistry' -import { Counter } from 'prom-client' - -// ── Ingester restart bookkeeping ──────────────────────────────────────────── -// A crash-looping ingester used to be invisible: it restarted every ten -// seconds forever, with no backoff, no metric, and no mention on /status. The -// state below makes that visible and bounded. - -const INGESTER_RESTART_BASE_MS = 10_000 -const INGESTER_RESTART_MAX_MS = 5 * 60_000 -// How long an ingester must run without crashing before its backoff resets, -// so a nightly blip does not permanently slow restarts. -const INGESTER_HEALTHY_RESET_MS = 10 * 60_000 - -export const ingesterRestartCounter = new Counter({ - name: 'lens_ingester_restarts_total', - help: 'Total number of times an ingester has crashed and been restarted', - labelNames: ['ingester', 'network'] as const, - registers: [registry], -}) - -interface IngesterCrashRecord { - time: Date - message: string -} - -interface IngesterState { - consecutiveFailures: number - lastCrash: IngesterCrashRecord | null -} - -const ingesterStates = new Map() - -const ingesterStateKey = (name: string, network: NetworkName) => `${name}/${network}` -const getIngesterState = (name: string, network: NetworkName): IngesterState => { - const key = ingesterStateKey(name, network) - let state = ingesterStates.get(key) - if (!state) { - state = { consecutiveFailures: 0, lastCrash: null } - ingesterStates.set(key, state) - } - return state -} - -export const getIngesterLastCrash = (name: string, network: NetworkName): IngesterCrashRecord | null => - ingesterStates.get(ingesterStateKey(name, network))?.lastCrash ?? null - -export const getAllIngesterLastCrashes = (): Array<{ ingester: string; network: NetworkName; lastCrash: IngesterCrashRecord | null }> => - Array.from(ingesterStates.entries()).map(([key, state]) => { - const [ingester, network] = key.split('/') as [string, NetworkName] - return { ingester, network, lastCrash: state.lastCrash } - }) +import { ingesterRestarts, recordIngesterCrash } from './metrics' async function main() { // ── Ensure DB schema is up-to-date ──────────────────────────────────────── @@ -193,38 +141,38 @@ 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. - const restartIngester = (name: string, network: NetworkName, fn: () => Promise) => { - const state = getIngesterState(name, network) + // + // A crash-looping ingester must not be invisible: restarts are counted in a + // Prometheus counter, the last crash is recorded for /status, and the delay + // grows exponentially (capped) so a permanently broken venue stops hammering + // its upstream every ten seconds. The backoff resets once an ingester has + // stayed up for HEALTHY_RUN_MS, so a nightly blip does not permanently slow + // restarts. + const BASE_RESTART_DELAY_MS = 10_000 + const MAX_RESTART_DELAY_MS = 5 * 60_000 + const HEALTHY_RUN_MS = 5 * 60_000 + + const restartIngester = ( + name: string, + network: NetworkName, + fn: () => Promise, + attempt = 0, + ) => { const startedAt = Date.now() - fn() - .then(() => { - // A clean return is not a crash, but it also is not a healthy run — - // the ingesters are infinite loops, so returning means it stopped. - // Treat it as a failure so the backoff still applies. - throw new Error('ingester returned unexpectedly') - }) - .catch(err => { - const message = (err as Error)?.message ?? String(err) - state.lastCrash = { time: new Date(), message } - ingesterRestartCounter.inc({ ingester: name, network }) - - // Reset the backoff if the previous run was healthy for long enough. - const ranFor = Date.now() - startedAt - if (ranFor >= INGESTER_HEALTHY_RESET_MS) { - state.consecutiveFailures = 0 - } - state.consecutiveFailures += 1 - - const delay = Math.min( - INGESTER_RESTART_BASE_MS * 2 ** (state.consecutiveFailures - 1), - INGESTER_RESTART_MAX_MS, - ) - console.error( - `[lens] ${name}/${network} ingester crashed (failure #${state.consecutiveFailures}), restarting in ${delay}ms:`, - message, - ) - setTimeout(() => restartIngester(name, network, fn), delay) - }) + fn().catch(err => { + const message = err instanceof Error ? err.message : String(err) + const delay = Math.min(BASE_RESTART_DELAY_MS * 2 ** attempt, MAX_RESTART_DELAY_MS) + const nextAttempt = Date.now() - startedAt >= HEALTHY_RUN_MS ? 0 : attempt + 1 + + ingesterRestarts.inc({ ingester: name, network }) + recordIngesterCrash(name, network, message) + + console.error( + `[lens] ${name}/${network} ingester crashed, restarting in ${Math.round(delay / 1000)}s:`, + message, + ) + setTimeout(() => restartIngester(name, network, fn, nextAttempt), delay) + }) } const enabledNetworks = getEnabledNetworks() From e69b2da19dc3e458c9f4dd7beacd7f71f404b3c8 Mon Sep 17 00:00:00 2001 From: Simplicity team Date: Sat, 17 Oct 2026 20:14:44 +0100 Subject: [PATCH 3/4] fix: A crash-looping ingester is invisible (#331) --- src/index.ts | 106 +++++++++++++++++++++++++++++++++++---------------- 1 file changed, 73 insertions(+), 33 deletions(-) diff --git a/src/index.ts b/src/index.ts index 33313c64..b2deb94c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,5 +1,6 @@ import 'dotenv/config' import { execSync } from 'child_process' + if (!process.env.DIRECT_DATABASE_URL && process.env.DATABASE_URL) { process.env.DIRECT_DATABASE_URL = process.env.DATABASE_URL } @@ -15,8 +16,7 @@ import { startSnapshotIngester } from './ingesters/snapshot' import { createAggregateQueue, startAggregateWorker, scheduleAggregateRefresh } from './jobs/aggregateRefresh' import { createSnapshotRetentionQueue, startSnapshotRetentionWorker, scheduleSnapshotRetention, pruneAllNetworks, SNAPSHOT_RETENTION_DAYS } from './jobs/snapshotRetention' import { loadPersistedPairs, getActivePairs } from './pairsRegistry' - -import { ingesterRestarts, recordIngesterCrash } from './metrics' +import { registry, ingesterRestarts } from './metrics' async function main() { // ── Ensure DB schema is up-to-date ──────────────────────────────────────── @@ -142,37 +142,77 @@ async function main() { // 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 in a - // Prometheus counter, the last crash is recorded for /status, and the delay - // grows exponentially (capped) so a permanently broken venue stops hammering - // its upstream every ten seconds. The backoff resets once an ingester has - // stayed up for HEALTHY_RUN_MS, so a nightly blip does not permanently slow - // restarts. - const BASE_RESTART_DELAY_MS = 10_000 - const MAX_RESTART_DELAY_MS = 5 * 60_000 - const HEALTHY_RUN_MS = 5 * 60_000 - - const restartIngester = ( - name: string, - network: NetworkName, - fn: () => Promise, - attempt = 0, - ) => { - const startedAt = Date.now() - fn().catch(err => { - const message = err instanceof Error ? err.message : String(err) - const delay = Math.min(BASE_RESTART_DELAY_MS * 2 ** attempt, MAX_RESTART_DELAY_MS) - const nextAttempt = Date.now() - startedAt >= HEALTHY_RUN_MS ? 0 : attempt + 1 - - ingesterRestarts.inc({ ingester: name, network }) - recordIngesterCrash(name, network, message) - - console.error( - `[lens] ${name}/${network} ingester crashed, restarting in ${Math.round(delay / 1000)}s:`, - message, - ) - setTimeout(() => restartIngester(name, network, fn, nextAttempt), delay) - }) + // 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 + } + + // Exposed so /status (src/api/rest.ts) can surface the last crash per + // (ingester, network) without reaching into the private state map. + ;(globalThis as { __lensIngesterState?: Map }).__lensIngesterState = ingesterState + + const restartIngester = (name: string, network: NetworkName, fn: () => Promise) => { + 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, + ) + console.error( + `[lens] ${name}/${network} ingester crashed (failure #${state.failures}), restarting in ${Math.round(delay / 1000)}s:`, + (err as Error).message, + ) + 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() From a7480a904874b1f78ba09071063fe00c3a3b6e14 Mon Sep 17 00:00:00 2001 From: Simplicity team Date: Tue, 20 Oct 2026 14:59:17 +0100 Subject: [PATCH 4/4] fix(review): address maintainer feedback on #479 --- src/__tests__/status.test.ts | 4 ++-- src/index.ts | 12 ++++-------- 2 files changed, 6 insertions(+), 10 deletions(-) diff --git a/src/__tests__/status.test.ts b/src/__tests__/status.test.ts index 9c4995e2..b006b353 100644 --- a/src/__tests__/status.test.ts +++ b/src/__tests__/status.test.ts @@ -87,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 () => { diff --git a/src/index.ts b/src/index.ts index 7dbbb0ed..5c8c9feb 100644 --- a/src/index.ts +++ b/src/index.ts @@ -24,7 +24,7 @@ import { PRICE_POINTS_RETENTION_DAYS, } from './jobs/pricePointsPartitionJob' import { loadPersistedPairs, getActivePairs } from './pairsRegistry' -import { registry, ingesterRestarts } from './metrics' +import { ingesterRestarts } from './metrics' import { markBootComplete } from './health' const log = createLogger('startup') @@ -217,10 +217,6 @@ async function main() { return state } - // Exposed so /status (src/api/rest.ts) can surface the last crash per - // (ingester, network) without reaching into the private state map. - ;(globalThis as { __lensIngesterState?: Map }).__lensIngesterState = ingesterState - const restartIngester = (name: string, network: NetworkName, fn: () => Promise) => { const state = getIngesterState(name, network) let healthyTimer: NodeJS.Timeout | undefined @@ -244,9 +240,9 @@ async function main() { INGESTER_RESTART_BASE_MS * 2 ** (state.failures - 1), INGESTER_RESTART_MAX_MS, ) - console.error( - `[lens] ${name}/${network} ingester crashed (failure #${state.failures}), restarting in ${Math.round(delay / 1000)}s:`, - (err as Error).message, + log.error( + { err, ingester: name, network, failures: state.failures, delayMs: delay }, + 'ingester crashed, restarting after backoff', ) setTimeout(run, delay) })