From ec85b8ca07f6e12c89774e58886c16dd9fecc055 Mon Sep 17 00:00:00 2001 From: gideonpius7-design Date: Mon, 31 Aug 2026 07:01:59 +0000 Subject: [PATCH] feat: launch ingesters per enabled network in the bootstrap (#117) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit src/index.ts previously started exactly one instance of each venue ingester (SDEX, AMM, Soroswap, Snapshot, Aquarius), always bound to whatever network was active at startup. To index more than one network in the same process, each ingester now runs once per enabled network. - src/network/enabledNetworks.ts (new): getEnabledNetworks() reads the comma-separated ENABLED_NETWORKS env var (e.g. "testnet,mainnet"), trims, lowercases, dedupes and filters to known network names. When unset — or nothing valid remains — it falls back to [activeNetwork], so a single-network deployment starts exactly one instance of each ingester. - src/index.ts: loops over the enabled networks and starts each ingester with its network context. restartIngester now captures the specific (venue, network) pair, so a crash in one instance restarts only that instance and cannot affect the others. - src/ingesters/snapshot.ts: appendSnapshots() / startSnapshotIngester() take a network arg. price_snapshots rows are network-scoped via the (network, pair, ts) PK, so a single instance could only ever snapshot the active network — each enabled network now gets its own loop. - src/ingest/venues/aquarius.ts: startAquariusIngester() / ingestAquariusPair() take a network arg and resolve the per-network Aquarius block via getNetworkConfig(network). - .env.example: documents ENABLED_NETWORKS. - src/__tests__/enabledNetworks.test.ts (new): covers the fallback, explicit multi-network lists, whitespace/casing normalization, dedupe and filtering of unrecognized names. closes #117 Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_019ky4A59maw6dB2CptXyhdW --- .env.example | 7 ++++ src/__tests__/aquariusIngester.test.ts | 9 ++++- src/__tests__/enabledNetworks.test.ts | 51 ++++++++++++++++++++++++++ src/index.ts | 31 ++++++++++------ src/ingest/venues/aquarius.ts | 19 ++++++---- src/ingesters/snapshot.ts | 23 ++++++++---- src/network/enabledNetworks.ts | 30 +++++++++++++++ 7 files changed, 141 insertions(+), 29 deletions(-) create mode 100644 src/__tests__/enabledNetworks.test.ts create mode 100644 src/network/enabledNetworks.ts diff --git a/.env.example b/.env.example index 909f42f..af4fa31 100644 --- a/.env.example +++ b/.env.example @@ -23,6 +23,13 @@ PRICE_CACHE_TTL=10 # This drives all config.horizon / config.rpc / config.soroswap shortcut paths. STELLAR_NETWORK=testnet +# Comma-separated list of networks the ingesters run for, e.g. "testnet,mainnet". +# Each venue ingester (SDEX, AMM, Soroswap, Snapshot, Aquarius) is started once +# per network listed here, each instance independently fault-isolated. +# When unset (or nothing valid remains after parsing) it falls back to just +# STELLAR_NETWORK, so a single-network node starts one instance of each ingester. +ENABLED_NETWORKS=testnet + # ───────────────────────────────────────────────────────────────────────────── # Per-network Stellar configuration # diff --git a/src/__tests__/aquariusIngester.test.ts b/src/__tests__/aquariusIngester.test.ts index e4ad404..3e52c9e 100644 --- a/src/__tests__/aquariusIngester.test.ts +++ b/src/__tests__/aquariusIngester.test.ts @@ -12,7 +12,14 @@ const mocks = vi.hoisted(() => ({ }, })) -vi.mock('../config', () => ({ config: mocks.config })) +vi.mock('../config', () => ({ + config: mocks.config, + activeNetwork: 'testnet', + // startAquariusIngester / ingestAquariusPair resolve the per-network Aquarius + // block via getNetworkConfig(network); the mock returns the same shape for + // whichever network is asked for. + getNetworkConfig: () => mocks.config, +})) vi.mock('../pairsRegistry', () => mocks.pairsRegistry) vi.mock('../db', () => ({ upsertPricePoints: vi.fn().mockResolvedValue(undefined) })) vi.mock('../webhookDispatcher', () => ({ dispatchPriceUpdate: vi.fn().mockResolvedValue(undefined) })) diff --git a/src/__tests__/enabledNetworks.test.ts b/src/__tests__/enabledNetworks.test.ts new file mode 100644 index 0000000..5f7b919 --- /dev/null +++ b/src/__tests__/enabledNetworks.test.ts @@ -0,0 +1,51 @@ +import { describe, it, expect, afterEach, vi } from 'vitest' + +// activeNetwork is derived from STELLAR_NETWORK at config-module load; pin it so +// the fallback path is deterministic regardless of the ambient env. +vi.mock('../config', () => ({ activeNetwork: 'testnet' })) + +import { getEnabledNetworks } from '../network/enabledNetworks' + +const ORIGINAL = process.env.ENABLED_NETWORKS + +afterEach(() => { + if (ORIGINAL === undefined) delete process.env.ENABLED_NETWORKS + else process.env.ENABLED_NETWORKS = ORIGINAL +}) + +describe('getEnabledNetworks', () => { + it('falls back to the active network when ENABLED_NETWORKS is unset', () => { + delete process.env.ENABLED_NETWORKS + expect(getEnabledNetworks()).toEqual(['testnet']) + }) + + it('falls back to the active network when ENABLED_NETWORKS is blank', () => { + process.env.ENABLED_NETWORKS = ' ' + expect(getEnabledNetworks()).toEqual(['testnet']) + }) + + it('parses an explicit multi-network list, preserving first-seen order', () => { + process.env.ENABLED_NETWORKS = 'mainnet,testnet' + expect(getEnabledNetworks()).toEqual(['mainnet', 'testnet']) + }) + + it('trims whitespace and lowercases each entry', () => { + process.env.ENABLED_NETWORKS = ' Testnet , MAINNET ' + expect(getEnabledNetworks()).toEqual(['testnet', 'mainnet']) + }) + + it('de-duplicates repeated networks', () => { + process.env.ENABLED_NETWORKS = 'mainnet,mainnet,testnet,mainnet' + expect(getEnabledNetworks()).toEqual(['mainnet', 'testnet']) + }) + + it('filters out unrecognised network names and empty segments', () => { + process.env.ENABLED_NETWORKS = 'testnet,,futurenet,' + expect(getEnabledNetworks()).toEqual(['testnet']) + }) + + it('falls back to the active network when nothing valid remains', () => { + process.env.ENABLED_NETWORKS = 'futurenet,localnet' + expect(getEnabledNetworks()).toEqual(['testnet']) + }) +}) diff --git a/src/index.ts b/src/index.ts index 8e3e29c..6951718 100644 --- a/src/index.ts +++ b/src/index.ts @@ -8,7 +8,8 @@ if (!process.env.DIRECT_DATABASE_URL && process.env.DATABASE_URL) { import cors from '@fastify/cors' import compress from '@fastify/compress' import rateLimit from '@fastify/rate-limit' -import { config } from './config' +import { config, type NetworkName } from './config' +import { getEnabledNetworks } from './network/enabledNetworks' import { redis } from './redis' import { pgPool } from './db' import { registerRESTRoutes } from './api/rest' @@ -174,20 +175,26 @@ async function main() { } // ── Ingesters (run in background — infinite loops) ──────────────────────── - // Each ingester is independently fault-isolated via restartIngester. - // A crash in the Soroswap ingester cannot take down SDEX or AMM. - console.log('[lens] Starting ingesters...') - const restartIngester = (name: string, fn: () => Promise) => { + // Each ingester is independently fault-isolated via restartIngester, keyed by + // 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) => { fn().catch(err => { - console.error(`[lens] ${name} ingester crashed, restarting in 10s:`, err.message) - setTimeout(() => restartIngester(name, fn), 10_000) + console.error(`[lens] ${name}/${network} ingester crashed, restarting in 10s:`, err.message) + setTimeout(() => restartIngester(name, network, fn), 10_000) }) } - restartIngester('SDEX', startSDEXIngester) - restartIngester('AMM', startAMMIngester) - restartIngester('Soroswap', startSoroswapIngester) - restartIngester('Snapshot', startSnapshotIngester) - restartIngester('Aquarius', startAquariusIngester) + + const enabledNetworks = getEnabledNetworks() + console.log(`[lens] Starting ingesters for network(s): ${enabledNetworks.join(', ')}`) + for (const network of enabledNetworks) { + restartIngester('SDEX', network, () => startSDEXIngester(network)) + restartIngester('AMM', network, () => startAMMIngester(network)) + restartIngester('Soroswap', network, () => startSoroswapIngester(network)) + restartIngester('Snapshot', network, () => startSnapshotIngester(network)) + restartIngester('Aquarius', network, () => startAquariusIngester(network)) + } console.log(`[lens] Watching ${getActivePairs().length} pairs: ${getActivePairs().map(p => p.pairKey).join(', ')}`) } diff --git a/src/ingest/venues/aquarius.ts b/src/ingest/venues/aquarius.ts index 5e0fab3..5c6fb06 100644 --- a/src/ingest/venues/aquarius.ts +++ b/src/ingest/venues/aquarius.ts @@ -8,7 +8,7 @@ * Issue: #103 */ -import { config } from '../../config' +import { config, activeNetwork, getNetworkConfig, type NetworkName } from '../../config' import { getActivePairs } from '../../pairsRegistry' import { upsertPricePoints } from '../../db' import { dispatchPriceUpdate } from '../../webhookDispatcher' @@ -52,8 +52,11 @@ export async function fetchAquariusPools( } } -export async function ingestAquariusPair(pair: WatchedPair): Promise { - const pools = await fetchAquariusPools(pair) +export async function ingestAquariusPair( + pair: WatchedPair, + network: NetworkName = activeNetwork, +): Promise { + const pools = await fetchAquariusPools(pair, getNetworkConfig(network).aquarius.apiUrl) if (pools.length === 0) return const points = pools.flatMap(pool => { @@ -98,16 +101,16 @@ async function sleep(ms: number) { return new Promise(r => setTimeout(r, ms)) } -export async function startAquariusIngester(): Promise { - if (!config.aquarius.enabled) { - console.log('[aquarius] Aquarius is disabled on this network — ingester not started') +export async function startAquariusIngester(network: NetworkName = activeNetwork): Promise { + if (!getNetworkConfig(network).aquarius.enabled) { + console.log(`[aquarius] Aquarius is disabled on ${network} — ingester not started`) return } - console.log(`[aquarius] Starting Aquarius AMM ingester for ${getActivePairs().length} pairs`) + console.log(`[aquarius] Starting Aquarius AMM ingester for ${getActivePairs().length} pairs on ${network}`) while (true) { for (const pair of getActivePairs()) { - await ingestAquariusPair(pair) + await ingestAquariusPair(pair, network) } await sleep(config.indexer.pollIntervalMs) } diff --git a/src/ingesters/snapshot.ts b/src/ingesters/snapshot.ts index bc40758..de99775 100644 --- a/src/ingesters/snapshot.ts +++ b/src/ingesters/snapshot.ts @@ -1,5 +1,5 @@ import { pgPool } from '../db' -import { activeNetwork } from '../config' +import { activeNetwork, type NetworkName } from '../config' import { getActivePairs } from '../pairsRegistry' import { price_snapshots_total } from '../metrics' @@ -22,8 +22,15 @@ export function floorToMinute(date: Date): Date { * total base_volume traded during the minute that just closed. Pairs with no * price history yet are skipped (we don't fabricate a price). Returns the * number of rows inserted. + * + * Snapshot rows are network-scoped (the `(network, pair, ts)` primary key), so + * `network` selects both which `price_points` are read and which network the + * inserted `price_snapshots` rows are tagged with. */ -export async function appendSnapshots(now: Date = new Date()): Promise { +export async function appendSnapshots( + now: Date = new Date(), + network: NetworkName = activeNetwork, +): Promise { const pairs = getActivePairs() if (pairs.length === 0) return 0 @@ -53,7 +60,7 @@ export async function appendSnapshots(now: Date = new Date()): Promise { FROM latest l LEFT JOIN vol v ON v.pair_key = l.pair_key ON CONFLICT (network, pair, ts) DO NOTHING`, - [pairKeys, ts, activeNetwork] + [pairKeys, ts, network] ) const inserted = result.rowCount ?? 0 @@ -70,8 +77,8 @@ function sleep(ms: number) { * wall-clock minute boundary so snapshot timestamps land on :00 seconds and the * volume window cleanly covers the minute that just elapsed. */ -export async function startSnapshotIngester(): Promise { - console.log(`[snapshot] Starting 1-minute snapshot ingester for ${getActivePairs().length} pairs`) +export async function startSnapshotIngester(network: NetworkName = activeNetwork): Promise { + console.log(`[snapshot] Starting 1-minute snapshot ingester for ${getActivePairs().length} pairs on ${network}`) while (true) { // Sleep until the next minute boundary. @@ -79,10 +86,10 @@ export async function startSnapshotIngester(): Promise { await sleep(msToNextMinute) try { - const n = await appendSnapshots() - if (n > 0) console.log(`[snapshot] appended ${n} snapshot(s)`) + const n = await appendSnapshots(new Date(), network) + if (n > 0) console.log(`[snapshot] appended ${n} snapshot(s) on ${network}`) } catch (err) { - console.error('[snapshot] Error appending snapshots:', (err as Error).message) + console.error(`[snapshot] Error appending snapshots on ${network}:`, (err as Error).message) } } } diff --git a/src/network/enabledNetworks.ts b/src/network/enabledNetworks.ts new file mode 100644 index 0000000..9bf34b9 --- /dev/null +++ b/src/network/enabledNetworks.ts @@ -0,0 +1,30 @@ +import { activeNetwork, type NetworkName } from '../config' + +const KNOWN_NETWORKS: readonly NetworkName[] = ['testnet', 'mainnet'] + +/** + * The set of Stellar networks this process should ingest, driven by the + * comma-separated `ENABLED_NETWORKS` env var (e.g. `ENABLED_NETWORKS=testnet,mainnet`). + * + * Values are trimmed, lowercased, de-duplicated, and filtered to the known + * network names. When `ENABLED_NETWORKS` is unset — or nothing valid remains + * after filtering — this falls back to just the currently active network + * (`STELLAR_NETWORK`, via {@link activeNetwork}), so a single-network + * deployment starts exactly one instance of each ingester rather than one per + * network. + * + * Order follows first appearance in the env var; the fallback returns + * `[activeNetwork]`. + */ +export function getEnabledNetworks(): NetworkName[] { + const raw = process.env.ENABLED_NETWORKS + if (raw && raw.trim()) { + const parsed = raw + .split(',') + .map(s => s.trim().toLowerCase()) + .filter((s): s is NetworkName => (KNOWN_NETWORKS as readonly string[]).includes(s)) + const deduped = [...new Set(parsed)] + if (deduped.length > 0) return deduped + } + return [activeNetwork] +}