From 699d9da5c591586087f71f030c17d7a04fb24cc5 Mon Sep 17 00:00:00 2001 From: Abdulrahman Abdulraheem <127429439+actvabdulrahman@users.noreply.github.com> Date: Sun, 18 Oct 2026 15:49:39 +0100 Subject: [PATCH 1/2] `feat: include Phoenix pools in /pools endpoint with venue tagging (#211)` --- openapi.yaml | 3 + src/api/rest.ts | 3 +- src/api/schemas.ts | 1 + src/ingest/venues/index.ts | 3 + src/ingest/venues/phoenix.ts | 153 +++++++++++++++++++++++++++++++++++ 5 files changed, 162 insertions(+), 1 deletion(-) create mode 100644 src/ingest/venues/phoenix.ts diff --git a/openapi.yaml b/openapi.yaml index eb49b630..26bc92ef 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -289,6 +289,8 @@ paths: type: number fee_bp: type: integer + venue: + type: string timestamp: type: string format: date-time @@ -2149,3 +2151,4 @@ components: properties: message: type: string + diff --git a/src/api/rest.ts b/src/api/rest.ts index 2f5fbdce..013029c0 100644 --- a/src/api/rest.ts +++ b/src/api/rest.ts @@ -225,7 +225,7 @@ export async function registerRESTRoutes(app: FastifyInstance) { WHERE ids.pool_id IS NOT NULL ) SELECT l.pool_id, l.asset_a, l.asset_b, - l.reserve_a::float, l.reserve_b::float, l.spot_price::float, l.fee_bp, l.timestamp + l.reserve_a::float, l.reserve_b::float, l.spot_price::float, l.fee_bp, l.timestamp, l.venue FROM ids CROSS JOIN LATERAL ( SELECT * FROM pool_snapshots s @@ -267,3 +267,4 @@ export async function registerRESTRoutes(app: FastifyInstance) { } ) } + diff --git a/src/api/schemas.ts b/src/api/schemas.ts index c7db8496..74c8956a 100644 --- a/src/api/schemas.ts +++ b/src/api/schemas.ts @@ -345,3 +345,4 @@ export function installResponseValidation(app: FastifyInstance): void { } }) } + diff --git a/src/ingest/venues/index.ts b/src/ingest/venues/index.ts index 0ea72a28..ef032935 100644 --- a/src/ingest/venues/index.ts +++ b/src/ingest/venues/index.ts @@ -2,6 +2,7 @@ import { sdexAdapter } from './sdex' import { ammAdapter } from './amm' import { soroswapAdapter } from './soroswap' import { aquariusAdapter } from './aquarius' +import { phoenixAdapter } from './phoenix' import type { VenueAdapter } from './types' export * from './types' @@ -10,6 +11,7 @@ export { sdexAdapter, startSDEXIngester } from './sdex' export { ammAdapter, startAMMIngester } from './amm' export { soroswapAdapter, startSoroswapIngester } from './soroswap' export { aquariusAdapter, startAquariusIngester } from './aquarius' +export { phoenixAdapter, startPhoenixIngester } from './phoenix' /** * Registry of all available venue adapters. @@ -22,3 +24,4 @@ export const venueAdapters: VenueAdapter[] = [ soroswapAdapter, aquariusAdapter, ] + diff --git a/src/ingest/venues/phoenix.ts b/src/ingest/venues/phoenix.ts new file mode 100644 index 00000000..d5ad526a --- /dev/null +++ b/src/ingest/venues/phoenix.ts @@ -0,0 +1,153 @@ +// TODO: Phoenix contract addresses not verified; using TODO placeholder discovery +/** + * Phoenix AMM Venue Adapter + * + * Indexes Phoenix liquidity pool (Stellar classic) reserves via the + * Phoenix AMM API, calculates spot prices from constant-product invariants, + * and stores them in price_points with source = 'Phoenix_amm'. + * + * Issue: #103, #218 + */ + +import { config, activeNetwork, getNetworkConfig, type NetworkName } from '../../config' +import { createLogger } from '../../logger' +import type { WatchedPair } from '../../types' +import type { VenueAdapter, VenuePricePoint } from './types' +import { + getLastPrice, + resetLastPrice, + writeVenuePricePoints, + startVenueAdapter, +} from './runner' +import { fetchWithTimeout } from '../../fetch' + +const log = createLogger('ingester:Phoenix') + +export function _resetLastPrice(): void { + resetLastPrice('Phoenix') +} + +export function _getLastPrice(network: NetworkName, pairKey: string): number | undefined { + return getLastPrice('Phoenix', network, pairKey) +} + +export interface PhoenixPool { + pool_hash: string + reserves: string[] + total_shares: string +} + +export interface PhoenixListResponse { + results?: PhoenixPool[] +} + +export async function fetchPhoenixPools( + pair: WatchedPair, + apiUrl: string = config.Phoenix.apiUrl +): Promise { + try { + const assetAStr = pair.assetA.issuer + ? `${pair.assetA.code}:${pair.assetA.issuer}` + : 'native' + const assetBStr = pair.assetB.issuer + ? `${pair.assetB.code}:${pair.assetB.issuer}` + : 'native' + + const params = new URLSearchParams() + params.append('assets[]', assetAStr) + params.append('assets[]', assetBStr) + + const res = await fetchWithTimeout(`${apiUrl}?${params.toString()}`) + if (!res.ok) return [] + const data = (await res.json()) as PhoenixListResponse + return data.results ?? [] + } catch (err) { + log.error({ pair: pair.pairKey, err }, 'failed to fetch pools') + return [] + } +} + +export function derivePhoenixPricePoints( + pools: PhoenixPool[], + pair: WatchedPair +): VenuePricePoint[] { + return pools.flatMap((pool) => { + const [r0Str, r1Str] = pool.reserves + if (!r0Str || !r1Str) return [] + + const r0 = parseFloat(r0Str) + const r1 = parseFloat(r1Str) + if (r0 <= 0 || r1 <= 0) return [] + + return [ + { + assetA: pair.assetA.code, + assetB: pair.assetB.code, + pairKey: pair.pairKey, + source: 'Phoenix_amm', + poolId: pool.pool_hash, + price: r1 / r0, + baseVolume: 0, + counterVolume: 0, + ledger: 0, + timestamp: new Date(), + }, + ] + }) +} + +export async function ingestPhoenixPair( + pair: WatchedPair, + network: NetworkName = activeNetwork +): Promise { + const pools = await fetchPhoenixPools(pair, getNetworkConfig(network).Phoenix.apiUrl) + if (pools.length === 0) return + + const points = derivePhoenixPricePoints(pools, pair) + if (points.length === 0) return + + await writeVenuePricePoints('Phoenix', points, pair, network) +} + +export const PhoenixAdapter: VenueAdapter = { + name: 'Phoenix', + kind: 'amm', + source: 'Phoenix_amm', + + isEnabled(network: NetworkName): boolean { + return getNetworkConfig(network).Phoenix.enabled + }, + + getPollIntervalMs(_network: NetworkName): number { + return config.indexer.pollIntervalMs + }, + + async discover(pair: WatchedPair, network: NetworkName): Promise { + const pools = await fetchPhoenixPools(pair, getNetworkConfig(network).Phoenix.apiUrl) + return pools.length > 0 ? [pools] : [] + }, + + async read(target: PhoenixPool[], _pair: WatchedPair, _network: NetworkName): Promise { + return target + }, + + derive(raw: PhoenixPool[], _target: PhoenixPool[], pair: WatchedPair, _network: NetworkName): VenuePricePoint[] { + return derivePhoenixPricePoints(raw, pair) + }, + + async write(points: VenuePricePoint[], _target: PhoenixPool[], pair: WatchedPair, network: NetworkName): Promise { + await writeVenuePricePoints('Phoenix', points, pair, network) + }, + + async ingestPair(pair: WatchedPair, network: NetworkName): Promise { + await ingestPhoenixPair(pair, network) + }, + + _resetLastPrice, + _getLastPrice, +} + +export async function startPhoenixIngester(network: NetworkName = activeNetwork): Promise { + await startVenueAdapter(PhoenixAdapter, network) +} + From c7ce01ee5eebf5ac5c8844d8598508c854a5df57 Mon Sep 17 00:00:00 2001 From: Abdulrahman Abdulraheem <127429439+actvabdulrahman@users.noreply.github.com> Date: Tue, 20 Oct 2026 08:53:32 +0100 Subject: [PATCH 2/2] `feat: incremental pool discovery with metrics and cycle tests (#220)` --- install_output.txt | Bin 0 -> 760 bytes openapi.json | 3 + package-lock.json | 26 +++---- src/__tests__/discoveryCycle.test.ts | 69 +++++++++++++++++ src/ingest/venues/aquarius.ts | 28 +++++++ src/ingest/venues/runner.ts | 108 +++++++++++++++++++++++++++ src/ingest/venues/types.ts | 15 ++++ src/metrics.ts | 21 ++++++ test_output.txt | Bin 0 -> 5498 bytes 9 files changed, 256 insertions(+), 14 deletions(-) create mode 100644 install_output.txt create mode 100644 src/__tests__/discoveryCycle.test.ts create mode 100644 test_output.txt diff --git a/install_output.txt b/install_output.txt new file mode 100644 index 0000000000000000000000000000000000000000..1c85d343b1c7f74f8c3f8fbba2cecabe97209920 GIT binary patch literal 760 zcmb7?Yfi&J421m~i977iRszJMf{0@v4$vipG-{fZJb=>!-#Bd=q#}e?S!catdpzE+ z&r>_GXwR&SFWABoyTfkrrnY9!xoPbTU$n|G(RyrQ-6SSC5);5w_!HZh>gc<&Yr8y( zjnx8?zsZq1}>_xWZzOp_yM={{~=c0b?`69-i}-rb_?bW zd`Wahu4qq=QEvsFmirl;2UhQ{f;tFPs$Db1NC)Y2<2NfJb6f4O0TmlO%9nyIZ9dk2 zB!^U$`s%&)dWGA5SG^;XDg-b>SL`|o{|fWM|CFU;W!$S$jTNIoo%>O(y+(%#O0U8` z{@^6uCHa1Gn{lq_s)UE$gXR)YP~75Vf~ohQTlF~qC$nJZtl<@;WZ1u?2cPqF@`tW| JpZ$f$e*?aDeDeSR literal 0 HcmV?d00001 diff --git a/openapi.json b/openapi.json index 94cb0a74..84c89487 100644 --- a/openapi.json +++ b/openapi.json @@ -442,6 +442,9 @@ "fee_bp": { "type": "integer" }, + "venue": { + "type": "string" + }, "timestamp": { "type": "string", "format": "date-time" diff --git a/package-lock.json b/package-lock.json index bf123327..63ba131f 100644 --- a/package-lock.json +++ b/package-lock.json @@ -365,7 +365,6 @@ "dev": true, "license": "MIT", "optional": true, - "peer": true, "dependencies": { "@emnapi/wasi-threads": "1.2.2", "tslib": "^2.4.0" @@ -378,19 +377,6 @@ "dev": true, "license": "MIT", "optional": true, - "peer": true, - "dependencies": { - "tslib": "^2.4.0" - } - }, - "node_modules/@emnapi/runtime": { - "version": "1.11.1", - "resolved": "https://registry.npmjs.org/@emnapi/runtime/-/runtime-1.11.1.tgz", - "integrity": "sha512-vgj7R3y3Wgx24IQaGPA/R6YFXLHVMOZ0uVEyIQPaWs+rd1AzfEMXlAC22FYwO1XkKR6NPsq7mUandH8oIRdZFw==", - "dev": true, - "license": "MIT", - "optional": true, - "peer": true, "dependencies": { "tslib": "^2.4.0" } @@ -2114,6 +2100,7 @@ "integrity": "sha512-8kzdPJ3FsNsVIurqBs7oodNnCEVbni9yUEkaHbgptDACOPW04jimGagZ51E6+lXUwJjgnBw+hyko/lkFWCldqw==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "undici-types": "~6.21.0" } @@ -2143,6 +2130,7 @@ "integrity": "sha512-38C0/Ddb7HcRG0Z4/DUem8x57d2p9jYgp18mkaYswEOQBGsI1CG4f/hjm0ZCeaJfWhSZ4k7jgs29V1Zom7Ki9A==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@bcoe/v8-coverage": "^1.0.2", "@vitest/utils": "4.1.5", @@ -3920,6 +3908,7 @@ "version": "2.3.3", "resolved": "https://registry.npmjs.org/fsevents/-/fsevents-2.3.3.tgz", "integrity": "sha512-5xoDfX+fL7faATnagmWPpbFtwh/R77WmMMqqHGS65C3vvB0YHrgF+B1YmZ3441tMj5n63k0212XNoJwzlhffQw==", + "dev": true, "hasInstallScript": true, "license": "MIT", "optional": true, @@ -4073,6 +4062,7 @@ "resolved": "https://registry.npmjs.org/graphql/-/graphql-16.13.2.tgz", "integrity": "sha512-5bJ+nf/UCpAjHM8i06fl7eLyVC9iuNAjm9qzkiu2ZGhM0VscSvS6WDPfAwkdkBuoXGM9FJSbKl6wylMwP9Ktig==", "license": "MIT", + "peer": true, "engines": { "node": "^12.22.0 || ^14.16.0 || ^16.0.0 || >=17.0.0" } @@ -4160,6 +4150,7 @@ "resolved": "https://registry.npmjs.org/hono/-/hono-4.13.5.tgz", "integrity": "sha512-O6+/eCYRkzzzy0rPWwKLiGBR1nFuUPZynnwjxN1MBA62NNqbT0wQEzQyK2gSO5yDIDB336sXQleAhOHrzlYyKw==", "license": "MIT", + "peer": true, "engines": { "node": ">=16.9.0" } @@ -5550,6 +5541,7 @@ "resolved": "https://registry.npmjs.org/pg/-/pg-8.20.0.tgz", "integrity": "sha512-ldhMxz2r8fl/6QkXnBD3CR9/xg694oT6DZQ2s6c/RI28OjtSOpxnPrUCGOBJ46RCUxcWdx3p6kw/xnDHjKvaRA==", "license": "MIT", + "peer": true, "dependencies": { "pg-connection-string": "^2.12.0", "pg-pool": "^3.13.0", @@ -5647,6 +5639,7 @@ "integrity": "sha512-QP88BAKvMam/3NxH6vj2o21R6MjxZUAd6nlwAS/pnGvN9IVLocLHxGYIzFhg6fUQ+5th6P4dv4eW9jX3DSIj7A==", "dev": true, "license": "MIT", + "peer": true, "engines": { "node": ">=12" }, @@ -5810,6 +5803,7 @@ "devOptional": true, "hasInstallScript": true, "license": "Apache-2.0", + "peer": true, "dependencies": { "@prisma/engines": "5.22.0" }, @@ -6949,6 +6943,7 @@ "integrity": "sha512-5C1sg4USs1lfG0GFb2RLXsdpXqBSEhAaA/0kPL01wxzpMqLILNxIxIOKiILz+cdg/pLnOUxFYOR5yhHU666wbw==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "esbuild": "~0.27.0", "get-tsconfig": "^4.7.5" @@ -7122,6 +7117,7 @@ "integrity": "sha512-t7g7GVRpMXjNpa67HaVWI/8BWtdVIQPCL2WoozXXA7LBGEFK4AkkKkHx2hAQf5x1GZSlcmEDPkVLSGahxnEEZw==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "lightningcss": "^1.32.0", "picomatch": "^4.0.4", @@ -7200,6 +7196,7 @@ "integrity": "sha512-9Xx1v3/ih3m9hN+SbfkUyy0JAs72ap3r7joc87XL6jwF0jGg6mFBvQ1SrwaX+h8BlkX6Hz9shdd1uo6AF+ZGpg==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@vitest/expect": "4.1.5", "@vitest/mocker": "4.1.5", @@ -7378,6 +7375,7 @@ "resolved": "https://registry.npmjs.org/zod/-/zod-4.5.4.tgz", "integrity": "sha512-sC95tT5iHHH9gtpj6A81kh+NEaRAUFN+qlUPDUbRfOMvNf5QCBqsb3WgvnpVtK5Y+4UfA6KqufotuTvMGiTlsA==", "license": "MIT", + "peer": true, "funding": { "url": "https://github.com/sponsors/colinhacks" } diff --git a/src/__tests__/discoveryCycle.test.ts b/src/__tests__/discoveryCycle.test.ts new file mode 100644 index 00000000..570f97ea --- /dev/null +++ b/src/__tests__/discoveryCycle.test.ts @@ -0,0 +1,69 @@ +import { describe, it, expect, vi, beforeEach } from "vitest" +import { aquariusAdapter } from "../ingest/venues/aquarius" +import { pools_added_total, pools_removed_total, register } from "../metrics" +import { getActivePairs } from "../pairsRegistry" + +vi.mock("../pairsRegistry", () => ({ + getActivePairs: vi.fn(), +})) + +vi.mock("../db", () => ({ upsertPricePoints: vi.fn().mockResolvedValue(undefined) })) +vi.mock("../webhookDispatcher", () => ({ dispatchPriceUpdate: vi.fn().mockResolvedValue(undefined) })) +vi.mock("../events", () => ({ publishPriceUpdate: vi.fn() })) +vi.mock("../config", () => ({ + config: { indexer: { pollIntervalMs: 5000 }, aquarius: { apiUrl: "https://test" } }, + activeNetwork: "testnet", + getNetworkConfig: () => ({ aquarius: { enabled: true, apiUrl: "https://test" } }), +})) + +describe("pool discovery cycle", () => { + beforeEach(() => { + vi.clearAllMocks() + pools_added_total.reset() + pools_removed_total.reset() + if ((aquariusAdapter as any).getWatchedSet) { + (aquariusAdapter as any).getWatchedSet().clear() + } + }) + + it("adds and removes pools across discovery cycles with metrics", async () => { + const pair = { pairKey: "XLM/USDC", assetA: { code: "XLM", issuer: null }, assetB: { code: "USDC", issuer: "GABC" } } + ;(getActivePairs as any).mockReturnValue([pair]) + + const originalFetch = global.fetch + let callCount = 0 + global.fetch = vi.fn().mockImplementation(() => { + callCount++ + if (callCount === 1) { + return Promise.resolve({ + ok: true, + json: async () => ({ results: [{ pool_hash: "pool1", reserves: ["100", "200"], total_shares: "10" }] }), + }) + } else { + return Promise.resolve({ + ok: true, + json: async () => ({ results: [{ pool_hash: "pool2", reserves: ["50", "100"], total_shares: "5" }] }), + }) + } + }) as any + + if (aquariusAdapter.runDiscoveryCycle) { + await aquariusAdapter.runDiscoveryCycle("testnet") + } + const watchedAfter1 = (aquariusAdapter as any).getWatchedSet ? (aquariusAdapter as any).getWatchedSet() : new Set() + expect(watchedAfter1.size).toBeGreaterThanOrEqual(1) + + if (aquariusAdapter.runDiscoveryCycle) { + await aquariusAdapter.runDiscoveryCycle("testnet") + } + const watchedAfter2 = (aquariusAdapter as any).getWatchedSet ? (aquariusAdapter as any).getWatchedSet() : new Set() + expect(watchedAfter2.size).toBeGreaterThanOrEqual(1) + + const addedMetric = await register.getSingleMetric("pools_added_total") + const removedMetric = await register.getSingleMetric("pools_removed_total") + expect(addedMetric).toBeDefined() + expect(removedMetric).toBeDefined() + + global.fetch = originalFetch + }) +}) diff --git a/src/ingest/venues/aquarius.ts b/src/ingest/venues/aquarius.ts index 9f267cc9..4cb563d6 100644 --- a/src/ingest/venues/aquarius.ts +++ b/src/ingest/venues/aquarius.ts @@ -17,8 +17,13 @@ import { resetLastPrice, writeVenuePricePoints, startVenueAdapter, + createRediscoveryState, + runDiscoveryCycle, + startRediscoveryLoop, + getTargetKey, } from './runner' import { fetchWithTimeout } from '../../fetch' +import { getActivePairs } from '../../pairsRegistry' const log = createLogger('ingester:aquarius') @@ -40,6 +45,8 @@ export interface AquariusListResponse { results?: AquariusPool[] } +const aquariusRediscoveryState = createRediscoveryState() + export async function fetchAquariusPools( pair: WatchedPair, apiUrl: string = config.aquarius.apiUrl @@ -144,6 +151,27 @@ export const aquariusAdapter: VenueAdapter = { _resetLastPrice, _getLastPrice, + + async startRediscovery(intervalMs: number, network: NetworkName = activeNetwork): Promise { + const discoverFn = async (pair: WatchedPair, net: NetworkName) => { + const pools = await fetchAquariusPools(pair, getNetworkConfig(net).aquarius.apiUrl) + return pools + } + await startRediscoveryLoop('Aquarius', aquariusAdapter, network, intervalMs, aquariusRediscoveryState, discoverFn) + }, + + getWatchedSet(): Set { + return aquariusRediscoveryState.watched + }, + + async runDiscoveryCycle(network: NetworkName = activeNetwork): Promise<{ added: number; removed: number }> { + const pairs = getActivePairs() + const discoverFn = async (pair: WatchedPair, net: NetworkName) => { + const pools = await fetchAquariusPools(pair, getNetworkConfig(net).aquarius.apiUrl) + return pools + } + return runDiscoveryCycle('Aquarius', network, pairs, discoverFn, aquariusRediscoveryState) + }, } export async function startAquariusIngester(network: NetworkName = activeNetwork): Promise { diff --git a/src/ingest/venues/runner.ts b/src/ingest/venues/runner.ts index 20cd3b83..5ac33e72 100644 --- a/src/ingest/venues/runner.ts +++ b/src/ingest/venues/runner.ts @@ -6,6 +6,7 @@ import { publishPriceUpdate } from '../../events' import { createLogger } from '../../logger' import type { WatchedPair } from '../../types' import type { VenueAdapter, VenuePricePoint } from './types' +import { pools_added_total, pools_removed_total, pools_discovered_total } from '../../metrics' const log = createLogger('ingester:runner') @@ -80,6 +81,113 @@ export async function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)) } +// ── Rediscovery helpers ────────────────────────────────────────────────────── + +export interface RediscoveryState { + watched: Set + watchedTargets: Map +} + +export function createRediscoveryState(): RediscoveryState { + return { + watched: new Set(), + watchedTargets: new Map(), + } +} + +export function getTargetKey(target: any): string { + if (target == null) return '' + if (typeof target === 'string') return target + if (target.pool_hash) return target.pool_hash + if (target.poolId) return target.poolId + if (target.poolAddress) return target.poolAddress + if (target.id) return target.id + try { + return JSON.stringify(target) + } catch { + return String(target) + } +} + +export async function runDiscoveryCycle( + venueName: string, + network: NetworkName, + pairs: WatchedPair[], + discoverFn: (pair: WatchedPair, network: NetworkName) => Promise, + state: RediscoveryState +): Promise<{ added: number; removed: number }> { + const currentSet = new Set() + const currentMap = new Map() + + for (const pair of pairs) { + try { + const discovered = await discoverFn(pair, network) + if (Array.isArray(discovered)) { + for (const item of discovered) { + const key = getTargetKey(item) + if (key) { + currentSet.add(key) + currentMap.set(key, item) + } + } + } + } catch (err) { + log.error({ venue: venueName, pair: pair.pairKey, err }, 'discovery cycle error') + } + } + + let added = 0 + let removed = 0 + + for (const key of currentSet) { + if (!state.watched.has(key)) { + added++ + state.watched.add(key) + if (currentMap.has(key)) { + state.watchedTargets.set(key, currentMap.get(key) as T) + } + } + } + + for (const key of Array.from(state.watched)) { + if (!currentSet.has(key)) { + removed++ + state.watched.delete(key) + state.watchedTargets.delete(key) + } + } + + pools_discovered_total.inc({ venue: venueName, network }, currentSet.size) + if (added > 0) pools_added_total.inc({ venue: venueName, network }, added) + if (removed > 0) pools_removed_total.inc({ venue: venueName, network }, removed) + + log.debug({ venue: venueName, network, added, removed, total: currentSet.size }, 'discovery cycle complete') + + return { added, removed } +} + +export async function startRediscoveryLoop( + venueName: string, + adapter: VenueAdapter, + network: NetworkName, + intervalMs: number, + state: RediscoveryState, + discoverFn: (pair: WatchedPair, network: NetworkName) => Promise +): Promise { + const pairs = getActivePairs() + await runDiscoveryCycle(venueName, network, pairs, discoverFn, state) + const timer = setInterval(async () => { + try { + const currentPairs = getActivePairs() + await runDiscoveryCycle(venueName, network, currentPairs, discoverFn, state) + } catch (err) { + log.error({ venue: venueName, network, err }, 'rediscovery loop error') + } + }, intervalMs) + ;(global as any).__rediscoveryTimers = (global as any).__rediscoveryTimers || [] + ;(global as any).__rediscoveryTimers.push(timer) +} + // ── Scheduling, Retry & Error Handling Runner ───────────────────────────────── /** diff --git a/src/ingest/venues/types.ts b/src/ingest/venues/types.ts index 2fa4601d..5a56dc6a 100644 --- a/src/ingest/venues/types.ts +++ b/src/ingest/venues/types.ts @@ -76,4 +76,19 @@ export interface VenueAdapter { * Gets the last seen price for threshold/webhook tracking. */ _getLastPrice?(network: NetworkName, pairKey: string): number | undefined + + /** + * Start periodic rediscovery of targets for watched pairs. + */ + startRediscovery?(intervalMs: number, network?: NetworkName): Promise + + /** + * Get current watched set (for testing) + */ + getWatchedSet?(): Set + + /** + * Run a discovery cycle manually (for testing) + */ + runDiscoveryCycle?(network?: NetworkName): Promise<{ added: number; removed: number }> } diff --git a/src/metrics.ts b/src/metrics.ts index 39a3a505..6f3a31d7 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -59,6 +59,27 @@ export const last_trade_timestamp = new Gauge({ registers: [register] }) +export const pools_discovered_total = new Counter({ + name: 'pools_discovered_total', + help: 'Total number of pools discovered across all venues', + labelNames: ['venue', 'network'], + registers: [register] +}) + +export const pools_added_total = new Counter({ + name: 'pools_added_total', + help: 'Total number of pools added during discovery cycles', + labelNames: ['venue', 'network'], + registers: [register] +}) + +export const pools_removed_total = new Counter({ + name: 'pools_removed_total', + help: 'Total number of pools removed during discovery cycles', + labelNames: ['venue', 'network'], + registers: [register] +}) + export const http_requests_total = new Counter({ name: 'http_requests_total', help: 'Total number of HTTP requests served, by method, route template and status class', diff --git a/test_output.txt b/test_output.txt new file mode 100644 index 0000000000000000000000000000000000000000..c18625bc962f8acee043ae8fa06415def25d32d0 GIT binary patch literal 5498 zcmd6r=}sG06vxl!N_~f^BVk7kHehxsRZ<9PswxXkKarF;*aW9|sqLmkguYT9r*G5t z_rJ&2o*AG78`SD*#=gthmwV3q{nwZFg?(d*9b3~nHn0b_U}an4SzvX@>VS7V^}$-> zwZ!`ND=^xNPhz|Fqdf&Ou@ANkmPMWwW?Qyt74`;T?DMWzo$=Nh;HcO$@HB0kofdFS zR<^Bc2kf<(b)YGT$AS8HtiEOTL)aOlXm^VV=lZ}^Y@ZRiAB0)O-mzD=o*mgQ$Xnyw z5jfhcIpsqj%3~<~8^C$VR(VfwCH9f|BF`a^t)RJv-3@D1WNrf`eMkpB{^a)ndD;x= zPq-Sa4N@GGbKE6?ui@x~_!tIm9a94g^WbR0Lo4Vmk1t{GclbL1?$}O%U(WC*T}@=F zz+DfBPS8*5LM^NZ9^4Q2!5cTm_&RVZxywO5?QZf& z-w(koKX!d=1`El;b*!$L&V0>@Ii5oB$5-~8z3xc!ZPs^2GzazyT{xxp*r_s}MY@HV zuZt|7E28S327Yc59-kE*d+rC+OoaNDzH|@+*5hWp7A?$)^s`scb2q z*ItQtja|ih2aJlGJ$r5U_}%4{G$YS^g#RA!+hf*Z%`;7;Bd&JH1joU5nAhgZdhb0zQd470_aJgyGj2DIX7mt5~pchW=DCefbQOE^UNKkl7W}H;=tS411n<295Vn zQD!XT4YFX98zFko5dV($n`cjcn3i38@;>>h{L}rZF@|}-9h#7xm2Vf2dy#hn#AIG& zu2b$I(W9^bTR?)7z+2RK2PxV?hkdj=3;r`Wp9Q{vg`13hxao~8Jqt$nw7kUfCHbUc z<)&nob*CAHP`k(m&5Xp;;=5R4Wu2Hi-Nw`6pWVjF7OK$1Q=NaeY3Z!ovw5d$Ryn8- z*UGLjf*PY-JI|WSHftPB>+@&hj6&AGR<*WCy|hXWt}?RJH^@S&+f+fR`jj6jzvuLRIq`NWr^A=}PA2yDZf^kFk;WXD>p< zRHH9b<&^G|5>E6(Y^wXx9j+BwSY9oERu`w*N%wg_tPA%={g>oQCIdRAIj1 z-6M))=PxZsi>agD%xH6Slu_Pr-fKI>RX)p1te3V>*5-M;I9Hw(Z`u>pDOQnWkxc1+ zA0u~>HIJ#dqOSl;<0#asyiXlYS{AREYplhu4T*ha_Zb&e_~~=%jPiL^+~xC0I^&h$ z!26U^j(_o&hnz^FfB76E9b5!a%-A_^9ce)|_3e1+T#~oS*sEeY=?U}owdNh=Tj1A={vH9OrFt|2_&DmRQ0kP-YdK(7T?CE-VJ+x(rNWo`o>rd zURMH#PUu@^tnU(bH#oBd=0Ui#)$0zy)dRjd+FKv(sp{+Dy}D&2_Yx=6!^VBZvn=7A fzxLMP(ebL*&^JARpyE_wa|uCy*| literal 0 HcmV?d00001