diff --git a/indexer/api.ts b/indexer/api.ts index 3db7066..6da8f4e 100644 --- a/indexer/api.ts +++ b/indexer/api.ts @@ -3,8 +3,8 @@ // strings). Served same-origin in production (nginx /api → this) and through // the vite dev/preview proxy locally. import { createServer, type IncomingMessage, type ServerResponse } from 'node:http' -import { PORT, log, now } from './config' -import { db, kvGet, poolCounts } from './store' +import { PORT, TUNE, log } from './config' +import { db, kvGet, notePoolsHit, poolCounts } from './store' const JSONH = { 'content-type': 'application/json; charset=utf-8' } @@ -54,8 +54,13 @@ function poolsWhere(params: Params): { where: string; args: (string | number)[] return { where: clauses.length ? 'WHERE ' + clauses.join(' AND ') : '', args } } +// TVL ranking sinks corrupt figures to the bottom instead of filtering them: +// this chain's whole TVL is ~8 orders under maxPoolTvlUsd, so anything above it +// is a pricing artefact, not a whale — but it must stay reachable by address or +// symbol search so it can still be diagnosed. (2026-07-22: without this guard +// 120 of the 120 rows the POOLS tab fetches were fabricated.) const ORDER: Record = { - tvl: 'ORDER BY (s.tvl_usd IS NULL), s.tvl_usd DESC', + tvl: `ORDER BY (s.tvl_usd IS NULL OR s.tvl_usd >= ${TUNE.maxPoolTvlUsd}), s.tvl_usd DESC`, vol: 'ORDER BY (st.vol24h_usd IS NULL), st.vol24h_usd DESC', created: 'ORDER BY (p.created_block IS NULL), p.created_block DESC, p.pair_index DESC', } @@ -118,7 +123,7 @@ function getPools(params: Params) { } const totals = Object.fromEntries(poolCounts().map((c) => [c.proto, c.n])) - return { ready: kvGet('ready') === '1', asof: now(), totals, count, pools, tokens } + return { ready: kvGet('ready') === '1', asof: Number(kvGet('snapshot_asof')) || null, totals, count, pools, tokens } } function getTokens(params: Params) { @@ -141,13 +146,23 @@ function getHealth() { const tokens = (db.prepare('SELECT COUNT(*) AS n FROM tokens').get() as { n: number }).n const priced = (db.prepare('SELECT COUNT(*) AS n FROM tokens WHERE price_usd > 0').get() as { n: number }).n const tvl = (db.prepare('SELECT COUNT(*) AS n FROM pool_state WHERE tvl_usd IS NOT NULL').get() as { n: number }).n + // pricing-health canary: a healthy chain has zero of these + const corrupt = ( + db.prepare('SELECT COUNT(*) AS n FROM pool_state WHERE tvl_usd >= ?').get(TUNE.maxPoolTvlUsd) as { n: number } + ).n + const stateFreshness = db.prepare('SELECT MIN(updated) AS oldest, MAX(updated) AS newest FROM pool_state').get() + const statsFreshness = db.prepare('SELECT MAX(updated) AS newest FROM pool_stats').get() return { ready: kvGet('ready') === '1', - asof: now(), + asof: Number(kvGet('snapshot_asof')) || null, + lastBootError: kvGet('boot_error') || null, pools: totals, tokens, pricedTokens: priced, tvlPools: tvl, + corruptTvlPools: corrupt, + stateFreshness, + statsFreshness, v3Cursor: Number(kvGet('v3_cursor') ?? 0), v2Count: Number(kvGet('v2_count') ?? 0), rssMb: Math.round(process.memoryUsage.rss() / 1e6), @@ -166,7 +181,10 @@ export function startApi(): void { } let body: unknown let cache = 'public, max-age=10' - if (url.pathname === '/api/pools') body = getPools(url.searchParams) + if (url.pathname === '/api/pools') { + notePoolsHit() // open the frontpage-sweep demand gate + body = getPools(url.searchParams) + } else if (url.pathname === '/api/tokens') body = getTokens(url.searchParams) else if (url.pathname === '/api/health') { body = getHealth() diff --git a/indexer/config.ts b/indexer/config.ts index e16fc03..485864a 100644 --- a/indexer/config.ts +++ b/indexer/config.ts @@ -15,18 +15,47 @@ export const PORT = Number(process.env.INDEXER_PORT || 8787) export const DB_PATH = process.env.INDEXER_DB || fileURLToPath(new URL('./data/index.db', import.meta.url)) +/** positive-number env override for a tuning knob; unset/0/garbage → default */ +const envMs = (key: string, def: number): number => { + const v = Number(process.env[key]) + return Number.isFinite(v) && v > 0 ? v : def +} + +// Cadence philosophy: DISCOVERY and the dust sweeps are slow — they only keep +// the long tail honest, and none of it reaches a user until it ranks onto the +// POOLS tab. The `frontpage` tier is the fast one: it sweeps just the top-N +// pools by TVL (exactly what /api/pools?sort=tvl serves) so the on-screen data +// is timely — and only while someone is actually polling (gated on the last +// /api/pools hit) so an idle site costs nothing. All knobs take env overrides +// (docker-compose env) so cadence can be tuned without a code rebuild. export const TUNE = { - tailMs: 10_000, // factory tail + v2 allPairsLength poll - hotSweepMs: 60_000, // state refresh for hot pools - fullSweepMs: 3_600_000, // state refresh for ACTIVE pools (≥$100 TVL or <48h old) - censusMs: 21_600_000, // 6h full-catalog dust census (~114k pools and growing) - statsMs: 300_000, // GeckoTerminal enrichment cycle + // --- discovery + background sweeps (slow) --- + tailMs: envMs('TAIL_MS', 1_800_000), // factory tail + v2 allPairsLength poll (new pools may lag up to 30min) + hotSweepMs: envMs('HOT_SWEEP_MS', 600_000), // "might rise onto the homepage" set (≥$10k TVL / GT-active / <1h) + fullSweepMs: envMs('ACTIVE_SWEEP_MS', 14_400_000), // ACTIVE pools (≥$100 TVL or <48h old), every 4h + censusMs: envMs('CENSUS_MS', 86_400_000), // full-catalog dust census (~114k pools and growing), daily + statsMs: envMs('STATS_MS', 300_000), // GeckoTerminal enrichment cycle (external HTTP, off the RPC queue) + // --- frontpage tier: the pools the POOLS tab actually shows, swept fast --- + frontpageMs: envMs('FRONTPAGE_MS', 15_000), // sweep cadence for the on-screen top-N + frontpageN: envMs('FRONTPAGE_N', 64), // how many top-TVL pools count as "on the homepage" + frontpageGateMs: envMs('FRONTPAGE_GATE_MS', 90_000), // skip the sweep if /api/pools has been idle this long + // --- event tail: reread only the pools that emitted a swap/liquidity event --- + logtailMs: envMs('LOGTAIL_MS', 12_000), // how often to pull new logs and reread the pools that changed + logtailMaxBlocks: envMs('LOGTAIL_MAX_BLOCKS', 5_000), // cap the span per pull so a backlog catches up in bounded chunks + // --- fixed knobs --- gtPaceMs: 2_600, // ≥2.6s between GT calls (free tier: 30/min) batch: 400, // calls per multicall aggregate batchGapMs: 40, // pause between aggregates (gentle on the RPC) - hotTvlUsd: 10_000, // pools at/above this TVL refresh every hotSweepMs - minDepthUsd: 300, // min priced-side USD depth to propagate a price through a pool - gtFreshSecs: 1_800, // GT prices younger than this are never overwritten by propagation + hotTvlUsd: 10_000, // pools at/above this TVL are always in the hot set + minDepthUsd: 300, // min CREDIBLE-side USD depth to propagate a price through a pool + maxUncredibleRatio: 25, // a pool-priced side may claim at most this × the credible side before TVL clamps to 2× credible + // --- pricing safety rails (see state.ts header) --- + maxPriceHops: 3, // how far a price may travel from its GT/anchor seed + maxAbsTick: 700_000, // |tick| past this is a broken init, not a market (v3 MAX_TICK = 887272) + minTokenUsd: 1e-18, // plausibility band for any price we store, serve or propagate + maxTokenUsd: 1e9, + maxSideOverDepth: 100, // a pool-priced side may claim at most this × the credible depth behind its price + maxPoolTvlUsd: 1e9, // above this a TVL figure is corrupt, not a whale: never ranked, never swept fast } /** repo-root .env `RPC` (SECRET — never log/print it). Fallback: key-free public RPC. */ @@ -34,7 +63,7 @@ export function rpcUrl(): string { const env = process.env.RPC?.trim() if (env) return env try { - const text = readFileSync(new URL('../.env', import.meta.url), 'utf8') + const text = readFileSync(new URL('../../.env', import.meta.url), 'utf8') const m = text.match(/^\s*RPC\s*=\s*(\S+)\s*$/m) if (m) return m[1] } catch { @@ -43,6 +72,16 @@ export function rpcUrl(): string { return PUBLIC_RPC } +/** Extra Alchemy key is indexer-only; shared paid/public RPC remains fallback. */ +export function rpcUrls(): string[] { + const extraKey = process.env.EXTRA_ALCHEMY_RPC_KEY?.trim() + const dedicated = extraKey + ? `https://robinhood-mainnet.g.alchemy.com/v2/${extraKey}` + : undefined + const shared = rpcUrl() + return dedicated && dedicated !== shared ? [dedicated, shared] : [shared] +} + export const now = () => Math.floor(Date.now() / 1000) /** terminal-style timestamped log line */ diff --git a/indexer/logtail.ts b/indexer/logtail.ts new file mode 100644 index 0000000..968df05 --- /dev/null +++ b/indexer/logtail.ts @@ -0,0 +1,49 @@ +// Event-driven state refresh. A pool's reserves change only when it emits an +// event, so we don't reread on a timer — we ask the chain which pools changed. +// One eth_getLogs (v2 Sync, v3 Swap/Mint/Burn, any address) since the last block +// we processed yields exactly the pools that traded; we reread only those. Cost +// tracks trading activity, not catalog size — this is what lets the 6-hourly +// census over 100k+ dust pools, and the timed hot/active sweeps, go away. +// +// Two deliberate choices: +// - reread state, don't apply the logs' own deltas: whatever the chain returns +// now is the truth, so a reorg needs no unwinding and a missed log self-heals +// on the pool's next event (the daily census is the longstop). +// - gated on the same POOLS-tab demand signal as the frontpage sweep — the only +// consumer of fresh state is a watching user, and a getLogs every 12s costs +// 75 CU. Idle → no polling. On resume we jump the cursor to head instead of +// replaying the idle backlog: freshness wants the current state, not history. +import { numberToHex, toEventSelector } from 'viem' +import { TUNE, log } from './config' +import { pc } from './rpc' +import { computeTvlFor, sweepState } from './state' +import { kvGet, kvSet, poolsHitAgeMs } from './store' + +export const TOPICS = [ + toEventSelector('Sync(uint112,uint112)'), // univ2 reserves (fires on every mint/burn/swap) + toEventSelector('Swap(address,address,int256,int256,uint160,uint128,int24)'), // univ3 price + liquidity + toEventSelector('Mint(address,address,int24,int24,uint128,uint256,uint256)'), // univ3 liquidity add + toEventSelector('Burn(address,int24,int24,uint128,uint256,uint256)'), // univ3 liquidity remove +] + +/** reread every catalog pool that emitted a tracked event since the last run */ +export async function logtail(): Promise { + if (poolsHitAgeMs() > TUNE.frontpageGateMs) return // nobody watching — don't spend getLogs + const head = Number(await pc.getBlockNumber()) + const cursor = Number(kvGet('logtail_block') ?? 0) + const from = cursor && head - cursor <= TUNE.logtailMaxBlocks ? cursor + 1 : head // skip idle backlog + if (from > head) return + + const logs = (await pc.request({ + method: 'eth_getLogs', + params: [{ fromBlock: numberToHex(from), toBlock: numberToHex(head), topics: [TOPICS] }], + })) as { address: string }[] + + const dirty = [...new Set(logs.map((l) => l.address.toLowerCase()))] + if (dirty.length) { + const swept = await sweepState(dirty) // sweepState ignores addresses not in the catalog + computeTvlFor(dirty) + log(`[logtail] blocks ${from}-${head}: ${dirty.length} pools emitted, ${swept} in catalog`) + } + kvSet('logtail_block', String(head)) +} diff --git a/indexer/main.ts b/indexer/main.ts index 0c30ec8..3e4d642 100644 --- a/indexer/main.ts +++ b/indexer/main.ts @@ -3,24 +3,42 @@ // Run: `npm run indexer` (tsx). Data lives in indexer/data/index.db (SQLite). // // Boot: backfill → token meta → full state sweep → GT cycle → reprice → ready. -// Loops: tail 10s · hot sweep 60s · full sweep 60min · GT stats 5min. +// Loops: logtail 12s (reread only pools that emitted events) · frontpage 15s +// (on-screen top-N by TVL, gated on API demand) · tail 30min · hot 10min +// · active 4h · census 24h · GT stats 5min (off the RPC queue). // The API starts listening immediately; `ready:false` in responses tells the // frontend to keep using its client-side fallback until the first pass lands. -import { log, PORT, TUNE } from './config' -import { usingPrivateRpc } from './rpc' +import { log, now, PORT, sleep, TUNE } from './config' +import { safeError, usingPrivateRpc } from './rpc' import { backfillV3, syncV2, tailV3 } from './catalog' import { computeTvlFor, ensureTokenMeta, reprice, sweepState } from './state' import { gtCycle } from './stats' -import { activeAddrs, allPoolAddrs, db, hotAddrs, kvGet, kvSet, poolCounts } from './store' +import { logtail } from './logtail' +import { activeAddrs, allPoolAddrs, db, frontpageAddrs, hotAddrs, kvGet, kvSet, poolCounts, poolsHitAgeMs } from './store' import { startApi } from './api' -/** setTimeout-chained loop — never overlaps itself, logs failures and keeps going */ -function loop(name: string, ms: number, fn: () => Promise): void { +let refreshQueue = Promise.resolve() + +function enqueueRefresh(fn: () => Promise): Promise { + const run = refreshQueue.then(fn, fn) + refreshQueue = run + return run +} + +/** + * setTimeout-chained loop. `queued` (default) routes the tick through the shared + * RPC queue so state sweeps neither overlap nor get dropped. Stats runs + * off-queue: it hits GeckoTerminal over HTTP (not the RPC) with ~30 calls paced + * ≥2.6s apart, so enqueuing it would pin the queue for over a minute and starve + * the 15s frontpage sweep. It touches the DB only in synchronous transactions, + * which JS's single thread already serializes against the sweeps' writes. + */ +function loop(name: string, ms: number, fn: () => Promise, queued = true): void { const tick = async () => { try { - await fn() + await (queued ? enqueueRefresh(fn) : fn()) } catch (e) { - log(`[${name}] error:`, String(e).slice(0, 200)) + log(`[${name}] error:`, safeError(e)) } setTimeout(tick, ms) } @@ -33,10 +51,7 @@ const timed = async (fn: () => Promise): Promise<[T, number]> => { return [r, Date.now() - t0] } -async function boot(): Promise { - log('up33 lp-indexer starting —', usingPrivateRpc ? 'rpc: private (.env)' : 'rpc: public') - startApi() - +async function initialRefresh(): Promise { const [addedV3, msV3] = await timed(backfillV3) if (addedV3 > 0 || !kvGet('v3_boot_logged')) { log(`[catalog] univ3 backfill done: +${addedV3} pools (${(msV3 / 1000).toFixed(0)}s)`) @@ -50,16 +65,32 @@ async function boot(): Promise { if (metaN) log(`[tokens] metadata fetched for ${metaN} tokens (${(msMeta / 1000).toFixed(0)}s)`) const all = allPoolAddrs() - const [, msSweep] = await timed(() => sweepState(all)) - log(`[sweep] full ${all.length} pools (${(msSweep / 1000).toFixed(0)}s)`) + const [swept, msSweep] = await timed(() => sweepState(all)) + log(`[sweep] full ${swept}/${all.length} pools (${(msSweep / 1000).toFixed(0)}s)`) + if (all.length && !swept) throw new Error('initial state sweep updated no pools') - await gtCycle().catch((e) => log('[stats] gt cycle failed:', String(e).slice(0, 120))) + await gtCycle().catch((e) => log('[stats] gt cycle failed:', safeError(e))) const pr = reprice() log(`[price] ${pr.priced} tokens priced · tvl on ${pr.tvlPools} pools`) kvSet('ready', '1') + kvSet('snapshot_asof', String(now())) + kvSet('boot_error', '') log(`READY — http://localhost:${PORT}/api/health`) +} +function startLoops(): void { + // primary freshness: reread pools the instant a trade touches them. + loop('logtail', TUNE.logtailMs, logtail) + // keep the pools users are looking at fresh even absent a trade, but only + // while they're looking (demand gate) so an idle site spends no RPC. + loop('frontpage', TUNE.frontpageMs, async () => { + if (poolsHitAgeMs() > TUNE.frontpageGateMs) return + const top = frontpageAddrs(TUNE.frontpageN) + if (!top.length) return + await sweepState(top) + computeTvlFor(top) + }) loop('tail', TUNE.tailMs, async () => { const fresh = [...(await tailV3()), ...(await syncV2())] if (fresh.length) { @@ -76,20 +107,46 @@ async function boot(): Promise { }) loop('active', TUNE.fullSweepMs, async () => { const addrs = activeAddrs() - const [, ms] = await timed(() => sweepState(addrs)) + const [swept, ms] = await timed(() => sweepState(addrs)) const p = reprice() - log(`[sweep] active ${addrs.length} pools (${(ms / 1000).toFixed(0)}s) · ${p.priced} tokens priced · tvl on ${p.tvlPools}`) + log(`[sweep] active ${swept}/${addrs.length} pools (${(ms / 1000).toFixed(0)}s) · ${p.priced} tokens priced · tvl on ${p.tvlPools}`) }) loop('census', TUNE.censusMs, async () => { const addrs = allPoolAddrs() - const [, ms] = await timed(() => sweepState(addrs)) + const [swept, ms] = await timed(() => sweepState(addrs)) const p = reprice() - log(`[sweep] census ${addrs.length} pools (${(ms / 1000).toFixed(0)}s) · tvl on ${p.tvlPools}`) - }) - loop('stats', TUNE.statsMs, async () => { - await gtCycle() - reprice() + log(`[sweep] census ${swept}/${addrs.length} pools (${(ms / 1000).toFixed(0)}s) · tvl on ${p.tvlPools}`) + if (addrs.length && !swept) throw new Error('census state sweep updated no pools') + kvSet('snapshot_asof', String(now())) }) + loop( + 'stats', + TUNE.statsMs, + async () => { + await gtCycle() + reprice() + }, + false, // off the RPC queue — paced GT HTTP must not block the frontpage sweep + ) +} + +async function boot(): Promise { + log('up33 lp-indexer starting —', usingPrivateRpc ? 'rpc: private (.env)' : 'rpc: public') + startApi() + startLoops() + + for (;;) { + try { + await enqueueRefresh(initialRefresh) + return + } catch (e) { + const error = safeError(e) + kvSet('boot_error', error) + const retryMs = 60_000 // fixed short retry — independent of the (slow) hot cadence + log(`[boot] error; retrying in ${retryMs / 1_000}s:`, error) + await sleep(retryMs) + } + } } process.on('SIGINT', () => { @@ -103,6 +160,6 @@ process.on('SIGTERM', () => { }) boot().catch((e) => { - log('FATAL boot:', e) + log('FATAL boot:', safeError(e)) process.exit(1) }) diff --git a/indexer/pricing.test.ts b/indexer/pricing.test.ts new file mode 100644 index 0000000..45ec3ac --- /dev/null +++ b/indexer/pricing.test.ts @@ -0,0 +1,202 @@ +import assert from 'node:assert/strict' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { after, test } from 'node:test' + +const dir = mkdtempSync(join(tmpdir(), 'lp-terminal-pricing-')) +process.env.INDEXER_DB = join(dir, 'test.db') + +const { reprice, tvlOf, v3Price1Per0, v3SpotOk, weightedMedian } = await import('./state') +const { db, insertPool, setTokenPrice, upsertState, upsertTokenMeta } = await import('./store') + +after(() => { + db.close() + rmSync(dir, { recursive: true, force: true }) +}) + +test('v3 spot price comes from sqrtPriceX96, decimals-adjusted', () => { + // the autopsied P/USDG pool (P 18dec, USDG 6dec): sqrt implies ≈ $66.7/P — + // the balance ratio the old code used had priced it $733,344 (11,000× off) + const p = v3Price1Per0('647112078155313432359389', 18, 6) + assert.ok(p !== null && Math.abs(p - 66.71) / 66.71 < 0.01) + // same-decimals identity: sqrt = 2^96 → price exactly 1 + assert.equal(v3Price1Per0(String(2 ** 96), 18, 18), 1) + // absent/degenerate states never price + assert.equal(v3Price1Per0(null, 18, 6), null) + assert.equal(v3Price1Per0('0', 18, 6), null) +}) + +test('a v3 pool only votes on price when its slot0 is a market', () => { + assert.equal(v3SpotOk('12345', 180_896), true) + // no in-range liquidity: slot0 is wherever the last poke left it, and the + // next dust swap moves it anywhere for free + assert.equal(v3SpotOk('0', 180_896), false) + assert.equal(v3SpotOk(null, 180_896), false) + // parked at the boundary (MAX_TICK 887272) — a broken init, not a market. + // This is the pool that minted the $3.5e50 quote on 2026-07-22. + assert.equal(v3SpotOk('12345', 887_271), false) + assert.equal(v3SpotOk('12345', -887_271), false) + assert.equal(v3SpotOk('12345', null), false) + assert.equal(v3SpotOk('not-a-number', 0), false) +}) + +test('a token price is the depth-weighted median of its quotes', () => { + // one manipulated pool must outweigh every honest pool, not just out-claim it + assert.equal(weightedMedian([{ usd: 5, depth: 1_000 }]), 5) + assert.equal( + weightedMedian([ + { usd: 1e9, depth: 400 }, // manipulated, thin + { usd: 2, depth: 10_000 }, // honest, deep + ]), + 2, + ) + // three-way: the middle of the depth mass wins, not the biggest number + assert.equal( + weightedMedian([ + { usd: 100, depth: 500 }, + { usd: 2, depth: 900 }, + { usd: 2.1, depth: 800 }, + ]), + 2.1, + ) +}) + +test('tvl clamps a runaway pool-priced side to the credible side', () => { + // the autopsied case: $387 credible USDG vs $10.85M pool-priced P → $774 approx + assert.deepEqual(tvlOf(10_850_000, false, 387, true), { tvl: 774, approx: true }) + // both credible: plain sum, exact + assert.deepEqual(tvlOf(5_000, true, 5_200, true), { tvl: 10_200, approx: false }) + // pool-priced side under the ratio passes through (honest near-edge pools) + assert.deepEqual(tvlOf(9_000, true, 600, false), { tvl: 9_600, approx: false }) + // one side unpriced: 2× the priced side, approximate + assert.deepEqual(tvlOf(null, false, 1_000, true), { tvl: 2_000, approx: true }) + // junk-vs-junk: the smaller side bounds the claim + assert.deepEqual(tvlOf(1e9, false, 100, false), { tvl: 200, approx: true }) + // nothing priced: no figure + assert.deepEqual(tvlOf(null, false, null, false), { tvl: null, approx: false }) +}) + +test('tvl caps a side at the credible depth that established its price', () => { + // two junk sides inflated by the SAME factor pass the ratio test (that is how + // a $1.4k pool reported $2.95e49) — the depth cap is what actually bounds it + assert.deepEqual(tvlOf(1e24, false, 1e24, false, 1e5, 1e5), { tvl: 200_000, approx: true }) + // a side under its cap is untouched + assert.deepEqual(tvlOf(4_000, false, 4_100, false, 1e5, 1e5), { tvl: 8_100, approx: false }) + // the cap applies before the ratio test, and the ratio test still wins when + // it is the tighter bound + assert.deepEqual(tvlOf(1e9, false, 100, false, 1e6, Infinity), { tvl: 200, approx: true }) +}) + +// --------------------------------------------------------------------------- +// End-to-end: the 2026-07-22 poisoning, replayed against the real pass. +// +// One v3 pool parked at MAX_TICK with zero in-range liquidity, holding $1,464 +// of real USDG, priced its counter-token at $3.5e50; that price then out-bid +// every honest quote (its fabricated "depth" was 1e19× bigger) and spread until +// 120 of the 120 rows the POOLS tab fetches were fabricated. +// --------------------------------------------------------------------------- + +const A = (n: number) => '0x' + n.toString(16).padStart(40, '0') +const [USDG, WETH, GME, KITTY, FAKE, JUNK] = [1, 2, 3, 4, 5, 6].map(A) +const MAX_SQRT_RATIO = 1461446703485210103287273052203988822378723970341n +const sqrtX96 = (p1per0: number, dec0: number, dec1: number) => + BigInt(Math.round(Math.sqrt(p1per0 * 10 ** (dec1 - dec0)) * 2 ** 96)) +const units = (x: number, dec: number) => BigInt(Math.round(x * 10 ** Math.min(dec, 15))) * 10n ** BigInt(Math.max(dec - 15, 0)) + +const priceOf = (addr: string) => + (db.prepare('SELECT price_usd FROM tokens WHERE address = ?').get(addr) as { price_usd: number | null }).price_usd +const poolTvl = (addr: string) => + db.prepare('SELECT tvl_usd, tvl_approx FROM pool_state WHERE address = ?').get(addr) as { + tvl_usd: number | null + tvl_approx: number + } + +test('reprice: a boundary-tick pool with no in-range liquidity cannot mint a price', () => { + for (const [a, sym, dec] of [ + [USDG, 'USDG', 6], + [WETH, 'WETH', 18], + [GME, 'GME', 18], + [KITTY, 'KITTY', 18], + [FAKE, 'FAKE', 18], + [JUNK, 'JUNK', 18], + ] as const) + upsertTokenMeta(a, sym, dec, true) + setTokenPrice(USDG, 1, 100_000, 'gt') + setTokenPrice(WETH, 2_000, 100_000, 'gt') + + // patient zero: 0.239 GME + 1408 USDG, slot0 pinned at MAX_TICK, L = 0 + const gmeUsdg = A(0x100) + insertPool({ address: gmeUsdg, proto: 'univ3', token0: GME, token1: USDG, feePpm: 10_000, tickSpacing: 200 }) + upsertState(gmeUsdg, { + sqrtPrice: MAX_SQRT_RATIO, + tick: 887_271, + liquidity: 0n, + reserve0: units(0.239, 18), + reserve1: units(1_408, 6), + }) + + // honest: 5 WETH + 1e7 KITTY at 2,000,000 KITTY/WETH → KITTY = $0.001 + const wethKitty = A(0x101) + insertPool({ address: wethKitty, proto: 'univ3', token0: WETH, token1: KITTY, feePpm: 3_000, tickSpacing: 60 }) + upsertState(wethKitty, { + sqrtPrice: sqrtX96(2e6, 18, 18), + tick: 180_896, + liquidity: 36_819_258_015_569_838_458_222n, + reserve0: units(5, 18), + reserve1: units(1e7, 18), + }) + + // thin dissenter on the same token, 1000× off: $400 of USDG vs $10k of WETH + const usdgKitty = A(0x102) + insertPool({ address: usdgKitty, proto: 'univ2', token0: USDG, token1: KITTY, feePpm: 3_000 }) + upsertState(usdgKitty, { reserve0: units(400, 6), reserve1: units(4e8, 18), totalSupply: 1n }) + + // hop 2, and then a pool where FAKE's balance alone would claim $1e9 + const kittyFake = A(0x103) + insertPool({ address: kittyFake, proto: 'univ2', token0: KITTY, token1: FAKE, feePpm: 3_000 }) + upsertState(kittyFake, { reserve0: units(1e6, 18), reserve1: units(1e3, 18), totalSupply: 1n }) + const fakeJunk = A(0x104) + insertPool({ address: fakeJunk, proto: 'univ2', token0: FAKE, token1: JUNK, feePpm: 3_000 }) + upsertState(fakeJunk, { reserve0: units(1e9, 18), reserve1: units(1e6, 18), totalSupply: 1n }) + + reprice() + + // 1. the boundary pool never priced GME, and its TVL is 2× the credible side + assert.equal(priceOf(GME), null) + const gmeTvl = poolTvl(gmeUsdg) + assert.ok(gmeTvl.tvl_usd !== null && Math.abs(gmeTvl.tvl_usd - 2_816) < 1, `got ${gmeTvl.tvl_usd}`) + assert.equal(gmeTvl.tvl_approx, 1) + + // 2. the deep honest quote sets KITTY; the thin 1000×-off pool does not move it + const kitty = priceOf(KITTY) + assert.ok(kitty !== null && Math.abs(kitty - 0.001) / 0.001 < 0.01, `got ${kitty}`) + + // 3. hop 2 prices FAKE, but its depth is capped by KITTY's — so the pool where + // FAKE holds a nominal $1e9 reports $200k, not $2e9 + const fake = priceOf(FAKE) + assert.ok(fake !== null && Math.abs(fake - 1) < 0.01, `got ${fake}`) + const junkTvl = poolTvl(fakeJunk) + assert.ok(junkTvl.tvl_usd !== null && junkTvl.tvl_usd <= 250_000, `got ${junkTvl.tvl_usd}`) + assert.equal(junkTvl.tvl_approx, 1) + + // 4. the invariant the frontpage actually depends on + const worst = db.prepare('SELECT MAX(tvl_usd) AS m FROM pool_state').get() as { m: number | null } + assert.ok((worst.m ?? 0) < 1e9, `a pool reported ${worst.m}`) + const dearest = db.prepare('SELECT MAX(price_usd) AS m FROM tokens').get() as { m: number | null } + assert.ok((dearest.m ?? 0) <= 1e9, `a token priced ${dearest.m}`) +}) + +test('reprice: a poisoned price does not survive the next pass', () => { + // simulate the live DB state: KITTY pinned at a fantasy price with the + // fabricated depth that used to make it unbeatable + setTokenPrice(KITTY, 3.5e50, 1e49, 'pool') + reprice() + const kitty = priceOf(KITTY) + assert.ok(kitty !== null && Math.abs(kitty - 0.001) / 0.001 < 0.01, `got ${kitty}`) + + // an out-of-band GT seed is dropped too, whatever depth it claims + setTokenPrice(WETH, 1e30, 1e9, 'gt') + reprice() + assert.equal(priceOf(WETH), null) +}) diff --git a/indexer/rpc.ts b/indexer/rpc.ts index 718db1d..88e3003 100644 --- a/indexer/rpc.ts +++ b/indexer/rpc.ts @@ -1,5 +1,5 @@ -import { createPublicClient, defineChain, http, type PublicClient } from 'viem' -import { log, PUBLIC_RPC, TUNE, rpcUrl, sleep } from './config' +import { createPublicClient, defineChain, fallback, http, type PublicClient } from 'viem' +import { log, PUBLIC_RPC, TUNE, rpcUrls, sleep } from './config' // duplicated from src/config/chain.ts — that module imports src/config/env.ts // (import.meta.env, vite-only) so it can't be loaded under node @@ -11,20 +11,26 @@ const robinhood = defineChain({ contracts: { multicall3: { address: '0xcA11bde05977b3631167028862bE2a173976CA11' } }, }) -const url = rpcUrl() -export const usingPrivateRpc = url !== PUBLIC_RPC +const urls = rpcUrls() +export const usingPrivateRpc = urls.some((url) => url !== PUBLIC_RPC) // timeout is deliberately tight: a healthy 400-call aggregate answers in 2-4s // (measured 2026-07-16); a stalled attempt should fail fast and retry, not // pin the whole boot for 30s. Bad chunks degrade to sub-chunks in mc(). export const pc: PublicClient = createPublicClient({ chain: robinhood, - transport: http(url, { timeout: 10_000, retryCount: 2, retryDelay: 400 }), + transport: fallback( + urls.map((url) => http(url, { timeout: 10_000 })), + { retryCount: 2, retryDelay: 400 }, + ), }) /** error text safe to log — the RPC url (secret) is redacted */ -const redact = (e: unknown) => - String(e instanceof Error ? `${e.name}: ${e.message.split('\n')[0]}` : e) - .replaceAll(url, '') +export const safeError = (e: unknown) => + urls + .reduce( + (text, url) => text.replaceAll(url, ''), + String(e instanceof Error ? `${e.name}: ${e.message.split('\n')[0]}` : e), + ) .slice(0, 120) // loose call shape — abi fragments come from parseAbi, results are narrowed by ok() @@ -49,7 +55,7 @@ export async function mc(calls: Call[]): Promise { try { out.push(...(await agg(chunk))) } catch (e) { - log('[rpc] chunk failed, retrying:', redact(e)) + log('[rpc] chunk failed, retrying:', safeError(e)) await sleep(600) try { out.push(...(await agg(chunk))) @@ -59,7 +65,7 @@ export async function mc(calls: Call[]): Promise { try { out.push(...(await agg(part))) } catch (e2) { - log(`[rpc] dropped ${part.length}-call sub-chunk:`, redact(e2)) + log(`[rpc] dropped ${part.length}-call sub-chunk:`, safeError(e2)) out.push(...part.map(() => ({ status: 'failure' as const }))) } } diff --git a/indexer/rpcConfig.test.ts b/indexer/rpcConfig.test.ts new file mode 100644 index 0000000..94c94ba --- /dev/null +++ b/indexer/rpcConfig.test.ts @@ -0,0 +1,21 @@ +import assert from 'node:assert/strict' +import test from 'node:test' +import { rpcUrls } from './config' + +test('indexer uses its dedicated RPC before the shared paid RPC', () => { + const previousExtraKey = process.env.EXTRA_ALCHEMY_RPC_KEY + const previousRpc = process.env.RPC + try { + process.env.EXTRA_ALCHEMY_RPC_KEY = 'extra-test-key' + process.env.RPC = 'https://shared.invalid' + assert.deepEqual(rpcUrls(), [ + 'https://robinhood-mainnet.g.alchemy.com/v2/extra-test-key', + 'https://shared.invalid', + ]) + } finally { + if (previousExtraKey === undefined) delete process.env.EXTRA_ALCHEMY_RPC_KEY + else process.env.EXTRA_ALCHEMY_RPC_KEY = previousExtraKey + if (previousRpc === undefined) delete process.env.RPC + else process.env.RPC = previousRpc + } +}) diff --git a/indexer/state.test.ts b/indexer/state.test.ts new file mode 100644 index 0000000..4681969 --- /dev/null +++ b/indexer/state.test.ts @@ -0,0 +1,85 @@ +import assert from 'node:assert/strict' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { after, test } from 'node:test' + +const dir = mkdtempSync(join(tmpdir(), 'lp-terminal-indexer-')) +process.env.INDEXER_DB = join(dir, 'test.db') + +const { db, insertPool, upsertState } = await import('./store') +const { pc } = await import('./rpc') +const { sweepState } = await import('./state') + +after(() => { + db.close() + rmSync(dir, { recursive: true, force: true }) +}) + +test('partial pool reads preserve the last-good state', async () => { + const v2 = '0x0000000000000000000000000000000000000001' + const v3 = '0x0000000000000000000000000000000000000002' + const complete = '0x0000000000000000000000000000000000000003' + const v3Token0 = '0x0000000000000000000000000000000000000020' + const v3Token1 = '0x0000000000000000000000000000000000000021' + insertPool({ + address: v2, + proto: 'univ2', + token0: '0x0000000000000000000000000000000000000010', + token1: '0x0000000000000000000000000000000000000011', + feePpm: 3_000, + }) + insertPool({ address: v3, proto: 'univ3', token0: v3Token0, token1: v3Token1, feePpm: 500, tickSpacing: 10 }) + insertPool({ + address: complete, + proto: 'univ2', + token0: '0x0000000000000000000000000000000000000030', + token1: '0x0000000000000000000000000000000000000031', + feePpm: 3_000, + }) + upsertState(v2, { reserve0: 11n, reserve1: 12n, totalSupply: 13n }) + upsertState(v3, { sqrtPrice: 14n, tick: 15, liquidity: 16n, reserve0: 17n, reserve1: 18n }) + upsertState(complete, { reserve0: 31n, reserve1: 32n, totalSupply: 33n }) + + const read = (address: string) => + db + .prepare('SELECT sqrt_price, tick, liquidity, reserve0, reserve1, total_supply, updated FROM pool_state WHERE address = ?') + .get(address) + const v2Before = read(v2) + const v3Before = read(v3) + + const original = pc.multicall + Object.defineProperty(pc, 'multicall', { + configurable: true, + value: async ({ contracts }: { contracts: { address: string; functionName: string }[] }) => + contracts.map((call) => { + if (call.address === v2 && call.functionName === 'getReserves') + return { status: 'success', result: [21n, 22n, 0] } + if (call.address === v2 && call.functionName === 'totalSupply') return { status: 'failure' } + if (call.address === v3 && call.functionName === 'slot0') return { status: 'success', result: [24n, 25] } + if (call.address === v3 && call.functionName === 'liquidity') return { status: 'success', result: 26n } + if (call.address === v3Token0) return { status: 'success', result: 27n } + if (call.address === v3Token1) return { status: 'failure' } + if (call.address === complete && call.functionName === 'getReserves') + return { status: 'success', result: [41n, 42n, 0] } + if (call.address === complete && call.functionName === 'totalSupply') return { status: 'success', result: 43n } + throw new Error(`unexpected call: ${call.address} ${call.functionName}`) + }), + }) + try { + assert.equal(await sweepState([v2, v3, complete]), 1) + assert.deepEqual(read(v2), v2Before) + assert.deepEqual(read(v3), v3Before) + const { updated: _updated, ...completeAfter } = read(complete) as Record + assert.deepEqual(completeAfter, { + sqrt_price: null, + tick: null, + liquidity: null, + reserve0: '41', + reserve1: '42', + total_supply: '43', + }) + } finally { + Object.defineProperty(pc, 'multicall', { configurable: true, value: original }) + } +}) diff --git a/indexer/state.ts b/indexer/state.ts index f7d612d..1d614cc 100644 --- a/indexer/state.ts +++ b/indexer/state.ts @@ -5,18 +5,43 @@ // are the TVL basis, matching how GT/dexscreener report "reserve". // univ2: getReserves + totalSupply. // -// Pricing is a waterfall: GeckoTerminal token prices are ground truth while -// fresh (stats.ts seeds them, depth = pool reserve/2); everything else comes -// from anchor propagation — a token gets priced through the deepest pool that -// pairs it against an already-priced token, requiring ≥ TUNE.minDepthUsd of -// priced-side depth so dust pools can't set prices. TVL then = sum of priced -// sides (single-priced-side pools: 2× that side, flagged approximate). +// Pricing: GeckoTerminal token prices (stats.ts, depth = pool reserve/2) plus +// the USDG anchor are the only CREDIBLE seeds; every other price is propagated +// from them through pools, using the pool's SPOT — v2 reserve ratio, v3 slot0 +// sqrt price. v3 BALANCES are liquidity-shape artifacts, never a price +// (measured: a near-edge pool priced a token 11,000× off spot, turning a ~$1.4k +// pool into $10.85M of "TVL"). +// +// Four rails keep a manipulated pool from minting a price (2026-07-22 autopsy: +// ONE v3 pool parked at MAX_TICK with zero in-range liquidity, holding $1,464 +// of real USDG, priced its counter-token at $3.5e50 — which then poisoned 138 +// tokens and put 120 fake rows on the POOLS frontpage, burying every real one): +// +// 1. spot must be usable — a v3 pool with no in-range liquidity, or parked at +// a boundary tick, has a slot0 anyone can move for free: not a price. +// 2. propagation is a BFS by HOP from the seeds. A token settles at the +// shortest hop that can price it and never re-prices from a longer path, so +// a chain of junk pools can't out-shout the WETH pool next door. +// 3. depth (the weight behind a quote) is CREDIBLE dollars: past hop 1 it is +// capped by the depth backing the known side. The old `balance × its price` +// let a fabricated price fabricate its own authority — the bigger the lie, +// the more it outranked every honest quote, permanently. +// 4. within a hop, a token's quotes are combined by DEPTH-WEIGHTED MEDIAN, so +// one manipulated pool has to outweigh every honest pool to move a price. +// +// Nothing pool-derived is inherited across passes (loadSeeds reads GT/anchor +// only) — a poisoned price cannot survive a reprice. TVL = sum of priced sides, +// bounded by tvlOf(): one-sided → 2× that side; a pool-priced side that dwarfs +// the credible side → clamped to 2× credible; and any pool-priced side is +// capped at maxSideOverDepth × the credible depth that established its price. +// Every bounded figure flags tvl_approx. import { erc20Abi, formatUnits } from 'viem' import { uniV2PairAbi, uniV3PoolAbi } from '../src/abi' import { ADDR, TUNE, log, now } from './config' import { mc, ok, type Call } from './rpc' import { allTokens, + clearDerivedPrices, db, missingMetaTokens, setTokenPrice, @@ -71,6 +96,8 @@ export async function sweepState(addrs: string[]): Promise { for (let i = 0; i < addrs.length; i += 5_000) { done += await sweepSlice(addrs.slice(i, i + 5_000)) } + if (done < addrs.length) + log(`[sweep] ${done}/${addrs.length} pools updated; kept last-good state for ${addrs.length - done}`) return done } @@ -95,6 +122,7 @@ async function sweepSlice(addrs: string[]): Promise { } const res = await mc(calls) let i = 0 + let done = 0 tx(() => { for (const p of rows) { if (p.proto === 'univ3') { @@ -102,32 +130,58 @@ async function sweepSlice(addrs: string[]): Promise { const liq = ok(res[i++]) const b0 = ok(res[i++]) const b1 = ok(res[i++]) - if (!s0) continue + if (!s0 || liq === undefined || b0 === undefined || b1 === undefined) continue upsertState(p.address, { sqrtPrice: s0[0], tick: s0[1], - liquidity: liq ?? 0n, - reserve0: b0 ?? 0n, - reserve1: b1 ?? 0n, + liquidity: liq, + reserve0: b0, + reserve1: b1, }) + done++ } else { const rs = ok(res[i++]) const ts = ok(res[i++]) - if (!rs) continue - upsertState(p.address, { reserve0: rs[0], reserve1: rs[1], totalSupply: ts ?? 0n }) + if (!rs || ts === undefined) continue + upsertState(p.address, { reserve0: rs[0], reserve1: rs[1], totalSupply: ts }) + done++ } } }) - return rows.length + return done } -type PriceEntry = { usd: number; depth: number; src: string; updated: number } +/** `hops` = distance from a credible seed: 0 = GT/anchor, 1..n = propagated */ +type PriceEntry = { usd: number; depth: number; src: string; hops: number } +/** a price outside this band is a broken pool, not a market */ +export const plausibleUsd = (x: number | null | undefined): x is number => + x != null && Number.isFinite(x) && x >= TUNE.minTokenUsd && x <= TUNE.maxTokenUsd + +/** + * Seeds for a pricing pass: GT + anchor ONLY. Pool-derived prices are rebuilt + * from scratch every time — inheriting one (with the depth it claimed) is what + * made poisoning permanent, since nothing honest could ever outbid it again. + */ +const loadSeeds = (): Map => { + const m = new Map() + for (const t of allTokens()) + if (t.price_src && t.price_src !== 'pool' && plausibleUsd(t.price_usd)) + m.set(t.address, { usd: t.price_usd!, depth: t.price_depth_usd, src: t.price_src, hops: 0 }) + return m +} + +/** every stored price, for TVL arithmetic only (computeTvlFor — no propagation) */ const loadPrices = (): Map => { const m = new Map() for (const t of allTokens()) - if (t.price_usd != null && t.price_usd > 0) - m.set(t.address, { usd: t.price_usd, depth: t.price_depth_usd, src: t.price_src ?? '?', updated: t.price_updated ?? 0 }) + if (plausibleUsd(t.price_usd)) + m.set(t.address, { + usd: t.price_usd!, + depth: t.price_depth_usd, + src: t.price_src ?? '?', + hops: t.price_src === 'pool' ? 1 : 0, + }) return m } @@ -138,66 +192,193 @@ type StateRow = { token1: string reserve0: string reserve1: string + sqrt_price: string | null + liquidity: string | null + tick: number | null } const statesQ = () => db .prepare( - `SELECT p.address, p.proto, p.token0, p.token1, s.reserve0, s.reserve1 + `SELECT p.address, p.proto, p.token0, p.token1, s.reserve0, s.reserve1, s.sqrt_price, s.liquidity, s.tick FROM pools p JOIN pool_state s ON s.address = p.address`, ) .all() as StateRow[] +const Q96 = 2 ** 96 +/** univ3 spot from slot0: HUMAN token1 per token0 (null when absent/degenerate) */ +export function v3Price1Per0(sqrtPrice: string | null, dec0: number, dec1: number): number | null { + if (!sqrtPrice) return null + const r = Number(sqrtPrice) / Q96 + const p = r * r * 10 ** (dec0 - dec1) + return Number.isFinite(p) && p > 0 ? p : null +} + /** - * Full pricing pass: propagate USD prices from GT/anchor seeds through pools, - * then recompute every pool's TVL. Pure JS over in-memory rows (~35k pools), - * runs after full sweeps and after each GT cycle. + * Is a univ3 pool's slot0 a price at all? Two ways it isn't: + * - no in-range liquidity: slot0 is wherever the last poke left it and the + * next dust swap moves it anywhere, for free. + * - parked at a boundary tick (|tick| → MAX_TICK 887272): a pool initialized + * at the extreme, which is how the $3.5e50 quote got minted. + * Such pools still get TVL from the OTHER side's price — they just don't vote. + */ +export function v3SpotOk(liquidity: string | null, tick: number | null): boolean { + if (tick === null || Math.abs(tick) > TUNE.maxAbsTick) return false + try { + return liquidity !== null && BigInt(liquidity) > 0n + } catch { + return false + } +} + +/** HUMAN token1-per-token0 spot, or null when this pool may not set a price */ +const spotOf = (s: StateRow, b0: number, b1: number, decs: Map): number | null => { + if (s.proto !== 'univ3') return b0 > 0 && b1 > 0 ? b1 / b0 : null + if (!v3SpotOk(s.liquidity, s.tick)) return null + return v3Price1Per0(s.sqrt_price, decs.get(s.token0) ?? 18, decs.get(s.token1) ?? 18) +} + +/** + * Depth-weighted median of a token's candidate quotes. A single manipulated + * pool now has to bring more than half the credible depth backing a token to + * move its price — with "deepest wins" it only had to claim a bigger number. + */ +export function weightedMedian(quotes: readonly { usd: number; depth: number }[]): number { + if (quotes.length === 1) return quotes[0].usd + const sorted = [...quotes].sort((a, b) => a.usd - b.usd) + const half = sorted.reduce((a, q) => a + q.depth, 0) / 2 + let acc = 0 + for (const q of sorted) { + acc += q.depth + if (acc >= half) return q.usd + } + return sorted[sorted.length - 1].usd +} + +/** + * TVL from two priced sides. `credible` = the side's price came from GT or the + * USDG anchor rather than pool propagation. Three bounds, all flagging approx: + * - `cap0/cap1`: a pool-priced side may not claim more than + * maxSideOverDepth × the credible depth that established its price. This is + * the absolute ceiling — without it a $1.4k pool's fabricated quote turned + * into a $2.95e49 "TVL" that no ratio test could catch. + * - a side claiming more than maxUncredibleRatio × its credible counterparty + * is capped at 2× the credible side. + * - junk-vs-junk pools are bounded by the smaller side. + */ +export function tvlOf( + u0: number | null, + cred0: boolean, + u1: number | null, + cred1: boolean, + cap0 = Infinity, + cap1 = Infinity, +): { tvl: number | null; approx: boolean } { + let capped = false + if (u0 != null && u0 > cap0) { + u0 = cap0 + capped = true + } + if (u1 != null && u1 > cap1) { + u1 = cap1 + capped = true + } + const bounded = (r: { tvl: number | null; approx: boolean }) => (capped ? { ...r, approx: true } : r) + if (u0 == null && u1 == null) return { tvl: null, approx: false } + if (u0 == null || u1 == null) return { tvl: (u0 ?? u1)! * 2, approx: true } + if (cred0 !== cred1) { + const cu = cred0 ? u0 : u1 + const ju = cred0 ? u1 : u0 + if (ju > cu * TUNE.maxUncredibleRatio) return { tvl: cu * 2, approx: true } + } else if (!cred0 && !cred1) { + const lo = Math.min(u0, u1) + if (Math.max(u0, u1) > lo * TUNE.maxUncredibleRatio) return { tvl: lo * 2, approx: true } + } + return bounded({ tvl: u0 + u1, approx: false }) +} + +const credible = (e?: PriceEntry) => !!e && e.src !== 'pool' +/** ceiling on what a pool-priced side may claim; credible seeds are unbounded */ +const capOf = (e?: PriceEntry) => (e && e.src === 'pool' ? e.depth * TUNE.maxSideOverDepth : Infinity) + +/** at most this many quotes retained per token per hop (memory bound) */ +const MAX_QUOTES = 32 + +/** + * Full pricing pass: BFS USD prices out from the GT/anchor seeds, then + * recompute every pool's TVL. Pure JS over in-memory rows (~230k pools), runs + * after full sweeps and after each GT cycle. Pool-derived prices are discarded + * and rebuilt from scratch on every call — see the rails in the file header. */ export function reprice(): { priced: number; tvlPools: number } { const decs = new Map(allTokens().map((t) => [t.address, t.decimals])) - const prices = loadPrices() + const prices = loadSeeds() // bootstrap anchor before the first GT cycle: USDG ≈ $1 (GT overwrites it) if (!prices.has(ADDR.USDG.toLowerCase())) - prices.set(ADDR.USDG.toLowerCase(), { usd: 1, depth: 1, src: 'anchor', updated: now() }) + prices.set(ADDR.USDG.toLowerCase(), { usd: 1, depth: 1, src: 'anchor', hops: 0 }) - const states = statesQ() + // Balances and spot are decoded ONCE — every pool is walked maxPriceHops + // times plus once more for TVL, and formatUnits+BigInt per visit was the + // pass's whole GC bill (the container's heap cap is 448MB). const human = (raw: string, addr: string) => Number(formatUnits(BigInt(raw), decs.get(addr) ?? 18)) - const gtFresh = (e: PriceEntry) => e.src === 'gt' && now() - e.updated < TUNE.gtFreshSecs + const states = statesQ().map((s) => { + const b0 = human(s.reserve0, s.token0) + const b1 = human(s.reserve1, s.token1) + return { s, b0, b1, spot: spotOf(s, b0, b1, decs) } + }) - const dirty = new Map() - for (let round = 0; round < 3; round++) { - let changed = 0 - for (const s of states) { - const b0 = human(s.reserve0, s.token0) - const b1 = human(s.reserve1, s.token1) - for (const [known, other, kb, ob] of [ - [s.token0, s.token1, b0, b1], - [s.token1, s.token0, b1, b0], + // hop 1 = priced directly against a seed, hop 2 = against a hop-1 token, ... + // A token settles at the FIRST hop that can price it and is never revisited, + // so a long chain of junk pools can never displace a nearer honest quote. + const derived = new Map() + for (let hop = 1; hop <= TUNE.maxPriceHops; hop++) { + const quotes = new Map() + for (const { s, b0, b1, spot: p1per0 } of states) { + if (p1per0 === null) continue + for (const [known, other, kb, ob, knownIs0] of [ + [s.token0, s.token1, b0, b1, true], + [s.token1, s.token0, b1, b0, false], ] as const) { const kp = prices.get(known) - if (!kp || ob <= 0) continue - const depth = kb * kp.usd - if (depth < TUNE.minDepthUsd) continue - const existing = prices.get(other) - if (existing && (gtFresh(existing) || existing.depth >= depth)) continue - const e: PriceEntry = { usd: depth / ob, depth, src: 'pool', updated: now() } - prices.set(other, e) - dirty.set(other, e) - changed++ + if (!kp || kp.hops !== hop - 1 || ob <= 0 || prices.has(other)) continue + // credible dollars only: past the seeds a quote is worth no more than + // the depth backing the side it came from, so an inflated price can no + // longer inflate its own authority + const depth = kp.hops === 0 ? kb * kp.usd : Math.min(kb * kp.usd, kp.depth) + if (!(depth >= TUNE.minDepthUsd)) continue + const usd = knownIs0 ? kp.usd / p1per0 : kp.usd * p1per0 + if (!plausibleUsd(usd)) continue + const list = quotes.get(other) + if (!list) quotes.set(other, [{ usd, depth }]) + else if (list.push({ usd, depth }) > MAX_QUOTES * 2) { + list.sort((a, b) => b.depth - a.depth) + list.length = MAX_QUOTES + } } } - if (!changed) break + if (!quotes.size) break + for (const [addr, qs] of quotes) { + const e: PriceEntry = { + usd: weightedMedian(qs), + depth: qs.reduce((m, q) => Math.max(m, q.depth), 0), + src: 'pool', + hops: hop, + } + prices.set(addr, e) + derived.set(addr, e) + } } let tvlPools = 0 tx(() => { - for (const [addr, e] of dirty) setTokenPrice(addr, e.usd, e.depth, e.src) - for (const s of states) { + clearDerivedPrices() // no pool-derived price outlives the pass that made it + for (const [addr, e] of derived) setTokenPrice(addr, e.usd, e.depth, e.src) + for (const { s, b0, b1 } of states) { const p0 = prices.get(s.token0) const p1 = prices.get(s.token1) - const u0 = p0 ? human(s.reserve0, s.token0) * p0.usd : null - const u1 = p1 ? human(s.reserve1, s.token1) * p1.usd : null - const tvl = u0 != null && u1 != null ? u0 + u1 : u0 != null ? u0 * 2 : u1 != null ? u1 * 2 : null - setTvl(s.address, tvl, tvl != null && (u0 == null || u1 == null)) + const u0 = p0 ? b0 * p0.usd : null + const u1 = p1 ? b1 * p1.usd : null + const { tvl, approx } = tvlOf(u0, credible(p0), u1, credible(p1), capOf(p0), capOf(p1)) + setTvl(s.address, tvl, approx) if (tvl != null) tvlPools++ } }) @@ -210,7 +391,7 @@ export function computeTvlFor(addrs: string[]): void { const decs = new Map(allTokens().map((t) => [t.address, t.decimals])) const prices = loadPrices() const q = db.prepare( - `SELECT p.address, p.proto, p.token0, p.token1, s.reserve0, s.reserve1 + `SELECT p.address, p.proto, p.token0, p.token1, s.reserve0, s.reserve1, s.sqrt_price, s.liquidity, s.tick FROM pools p JOIN pool_state s ON s.address = p.address WHERE p.address = ?`, ) tx(() => { @@ -222,8 +403,8 @@ export function computeTvlFor(addrs: string[]): void { const p1 = prices.get(s.token1) const u0 = p0 ? human(s.reserve0, s.token0) * p0.usd : null const u1 = p1 ? human(s.reserve1, s.token1) * p1.usd : null - const tvl = u0 != null && u1 != null ? u0 + u1 : u0 != null ? u0 * 2 : u1 != null ? u1 * 2 : null - setTvl(s.address, tvl, tvl != null && (u0 == null || u1 == null)) + const { tvl, approx } = tvlOf(u0, credible(p0), u1, credible(p1), capOf(p0), capOf(p1)) + setTvl(s.address, tvl, approx) } }) } diff --git a/indexer/stats.ts b/indexer/stats.ts index d8f5192..a213067 100644 --- a/indexer/stats.ts +++ b/indexer/stats.ts @@ -8,6 +8,7 @@ // NOTE: GT has no UP33 dex entry — UP33 pool stats stay on the frontend's // existing dexscreener path; this indexer only serves the Uniswap catalog. import { GT, TUNE, log, sleep } from './config' +import { plausibleUsd } from './state' import { poolRow, setTokenPrice, upsertStats } from './store' const LISTS = [ @@ -60,15 +61,17 @@ function ingest(p: GtPool): boolean { const h24 = a.transactions?.h24 const txns = h24 ? (h24.buys ?? 0) + (h24.sells ?? 0) : null upsertStats(addr, num(a.volume_usd?.h24), txns, reserve, 'geckoterminal') - // token price seeds: ground truth while fresh; depth = half the pool's reserve + // Token price seeds — these are the CREDIBLE roots the whole pricing graph + // hangs off, and they're the one input the propagation rails can't second- + // guess, so an implausible GT quote is dropped rather than seeded. const depth = (reserve ?? 0) / 2 if (depth > 0) { const base = tokenOfId(p.relationships?.base_token?.data?.id) const quote = tokenOfId(p.relationships?.quote_token?.data?.id) const bp = num(a.base_token_price_usd) const qp = num(a.quote_token_price_usd) - if (base && bp && bp > 0) setTokenPrice(base, bp, depth, 'gt') - if (quote && qp && qp > 0) setTokenPrice(quote, qp, depth, 'gt') + if (base && plausibleUsd(bp)) setTokenPrice(base, bp, depth, 'gt') + if (quote && plausibleUsd(qp)) setTokenPrice(quote, qp, depth, 'gt') } return true } diff --git a/indexer/store.ts b/indexer/store.ts index 5c570a4..dcfd662 100644 --- a/indexer/store.ts +++ b/indexer/store.ts @@ -4,7 +4,7 @@ import { mkdirSync } from 'node:fs' import { dirname } from 'node:path' import { DatabaseSync } from 'node:sqlite' -import { DB_PATH, now } from './config' +import { DB_PATH, TUNE, now } from './config' mkdirSync(dirname(DB_PATH), { recursive: true }) export const db = new DatabaseSync(DB_PATH) @@ -47,7 +47,7 @@ CREATE TABLE IF NOT EXISTS pool_state ( reserve1 TEXT NOT NULL DEFAULT '0', total_supply TEXT, -- univ2 LP supply tvl_usd REAL, - tvl_approx INTEGER NOT NULL DEFAULT 0, -- 1 = only one side priced (tvl = 2× that side) + tvl_approx INTEGER NOT NULL DEFAULT 0, -- 1 = bounded figure: one side unpriced (2× priced side) or junk-side clamp (see tvlOf) updated INTEGER NOT NULL ); @@ -129,6 +129,17 @@ const priceQ = db.prepare(` export const setTokenPrice = (addr: string, usd: number, depthUsd: number, src: string) => void priceQ.run(addr.toLowerCase(), usd, depthUsd, src, now()) +// Drop every pool-derived price (reprice rebuilds them from the credible seeds +// on the same pass) plus anything outside the plausibility band, whatever its +// source — a GT quote can be garbage too, and a stored fantasy price is how one +// broken pool poisoned 138 tokens. Runs inside reprice's transaction, so the +// API never observes the gap. +const clearDerivedQ = db.prepare( + `UPDATE tokens SET price_usd = NULL, price_depth_usd = 0, price_src = NULL, price_updated = NULL + WHERE price_src = 'pool' OR (price_usd IS NOT NULL AND (price_usd < ? OR price_usd > ?))`, +) +export const clearDerivedPrices = () => void clearDerivedQ.run(TUNE.minTokenUsd, TUNE.maxTokenUsd) + export type TokenRow = { address: string symbol: string @@ -184,16 +195,40 @@ const upStatsQ = db.prepare(` export const upsertStats = (addr: string, vol24h: number | null, txns24h: number | null, liqUsd: number | null, source: string) => void upStatsQ.run(addr.toLowerCase(), vol24h, txns24h, liqUsd, source, now()) +/** + * Frontpage set: the top-N pools by TVL — exactly what /api/pools?sort=tvl + * returns to the POOLS tab, so sweeping these keeps every on-screen row fresh. + * The maxPoolTvlUsd ceiling matches the API's ranking guard: a corrupt figure + * must not be able to spend the fast tier's whole RPC budget on itself. + */ +export const frontpageAddrs = (n: number): string[] => + ( + db + .prepare('SELECT address FROM pool_state WHERE tvl_usd IS NOT NULL AND tvl_usd < ? ORDER BY tvl_usd DESC LIMIT ?') + .all(TUNE.maxPoolTvlUsd, n) as { address: string }[] + ).map((r) => r.address) + +// ---- demand gate ---- +// The frontpage sweep only spends RPC while the POOLS tab is actually being +// polled: the API stamps `lastPoolsHit` on every /api/pools request, the sweep +// loop checks its age. Idle site → no fast sweeps → no cost. Starts "closed" +// (lastPoolsHit = 0) so nothing sweeps until the first request lands. +let lastPoolsHit = 0 +export const notePoolsHit = (): void => { + lastPoolsHit = Date.now() +} +export const poolsHitAgeMs = (): number => Date.now() - lastPoolsHit + /** hot set: real TVL, or GT-visible activity, or freshly created */ export const hotAddrs = (): string[] => ( db .prepare( - `SELECT address FROM pool_state WHERE tvl_usd >= ? + `SELECT address FROM pool_state WHERE tvl_usd BETWEEN ? AND ? UNION SELECT address FROM pool_stats WHERE vol24h_usd > 0 UNION SELECT address FROM pools WHERE added_ts > ?`, ) - .all(10_000, now() - 3_600) as { address: string }[] + .all(TUNE.hotTvlUsd, TUNE.maxPoolTvlUsd, now() - 3_600) as { address: string }[] ).map((r) => r.address) /** diff --git a/scripts/logtail-smoke.ts b/scripts/logtail-smoke.ts new file mode 100644 index 0000000..a905c40 --- /dev/null +++ b/scripts/logtail-smoke.ts @@ -0,0 +1,27 @@ +// Proves the event-tail signal on Robinhood Chain: one eth_getLogs over the +// last SPAN blocks surfaces every pool that traded, grouped by event. +// Run: npm run smoke:logtail +import { numberToHex } from 'viem' +import { pc } from '../indexer/rpc' +import { TOPICS } from '../indexer/logtail' + +const SPAN = 1_500 +const NAMES = ['v2 Sync', 'v3 Swap', 'v3 Mint', 'v3 Burn'] + +const head = Number(await pc.getBlockNumber()) +const from = head - SPAN + 1 +const logs = (await pc.request({ + method: 'eth_getLogs', + params: [{ fromBlock: numberToHex(from), toBlock: numberToHex(head), topics: [TOPICS] }], +})) as { address: string; topics: string[] }[] + +const pools = new Set() +const byTopic = new Map() +for (const l of logs) { + pools.add(l.address.toLowerCase()) + byTopic.set(l.topics[0], (byTopic.get(l.topics[0]) ?? 0) + 1) +} + +console.log(`blocks ${from}-${head} (${SPAN}): ${logs.length} logs, ${pools.size} distinct pools`) +TOPICS.forEach((t, i) => console.log(` ${NAMES[i].padEnd(8)} ${byTopic.get(t) ?? 0}`)) +console.log('sample pools:', [...pools].slice(0, 8).join(' ')) diff --git a/src/lib/poolStatFallback.test.ts b/src/lib/poolStatFallback.test.ts new file mode 100644 index 0000000..11cbae0 --- /dev/null +++ b/src/lib/poolStatFallback.test.ts @@ -0,0 +1,36 @@ +import assert from 'node:assert/strict' +import test from 'node:test' +import type { PoolStat } from './poolstats' +import { poolStatWithFallback } from './poolStatFallback' + +const fallback: PoolStat = { + vol24hUsd: 130_000, + liqUsd: 900_000, + source: 'dexscreener', +} + +test('fills missing volume without replacing chain liquidity', () => { + const primary: PoolStat = { vol24hUsd: null, liqUsd: 1_000_000, source: 'chain' } + + assert.deepEqual(poolStatWithFallback(primary, fallback), { + vol24hUsd: 130_000, + liqUsd: 1_000_000, + source: 'dexscreener', + }) +}) + +test('preserves valid zero volume and only fills missing liquidity', () => { + const primary: PoolStat = { vol24hUsd: 0, liqUsd: null, source: 'chain' } + + assert.deepEqual(poolStatWithFallback(primary, fallback), { + vol24hUsd: 0, + liqUsd: 900_000, + source: 'chain', + }) +}) + +test('uses whichever complete stat is available', () => { + assert.equal(poolStatWithFallback(undefined, undefined), undefined) + assert.equal(poolStatWithFallback(fallback, undefined), fallback) + assert.equal(poolStatWithFallback(undefined, fallback), fallback) +}) diff --git a/src/lib/poolStatFallback.ts b/src/lib/poolStatFallback.ts new file mode 100644 index 0000000..b16eeec --- /dev/null +++ b/src/lib/poolStatFallback.ts @@ -0,0 +1,10 @@ +import type { PoolStat } from './poolstats' + +export function poolStatWithFallback(primary?: PoolStat, fallback?: PoolStat): PoolStat | undefined { + if (!primary || !fallback) return primary ?? fallback + return { + vol24hUsd: primary.vol24hUsd ?? fallback.vol24hUsd, + liqUsd: primary.liqUsd ?? fallback.liqUsd, + source: primary.vol24hUsd == null && fallback.vol24hUsd != null ? fallback.source : primary.source, + } +} diff --git a/src/lib/poolstats.ts b/src/lib/poolstats.ts index 7df6aff..bae294f 100644 --- a/src/lib/poolstats.ts +++ b/src/lib/poolstats.ts @@ -5,6 +5,7 @@ // side (≈$) or the WETH side × WETH price derived from DexScreener. import { ADDR } from '../config/addresses' import { ENV } from '../config/env' +import { pickDsTokenUsd, type DsPair } from './tokenPrice' import type { Pool, V2Pool } from '../types' export type PoolStat = { @@ -15,9 +16,8 @@ export type PoolStat = { // in same-origin proxy mode (server deploys) these route through nginx so // users behind restrictive networks keep TVL/volume/USD features -const DS_BASE = ENV.proxied - ? '/dexscreener/latest/dex/pairs/robinhood/' - : 'https://api.dexscreener.com/latest/dex/pairs/robinhood/' +const DS_ROOT = ENV.proxied ? '/dexscreener' : 'https://api.dexscreener.com' +const DS_BASE = DS_ROOT + '/latest/dex/pairs/robinhood/' const V2_SUBGRAPH = (ENV.proxied ? '/goldsky' : 'https://api.goldsky.com') + '/api/public/project_cmhef02640198x7p2cz2w70u8/subgraphs/up-robinhood-v2-mainnet/0.1.0/gn' @@ -25,7 +25,7 @@ const V2_SUBGRAPH = const WETH = ADDR.WETH.toLowerCase() const USDG = ADDR.USDG.toLowerCase() -async function fetchDexscreener( +export async function fetchDexscreener( addrs: string[], ): Promise<{ stats: Record; wethUsd: number | null }> { const stats: Record = {} @@ -56,6 +56,17 @@ async function fetchDexscreener( return { stats, wethUsd } } +/** venue USD price of a token — its most-liquid robinhood pair on dexscreener + * (see tokenPrice.ts for why aggregator unit-quotes are not used) */ +export async function fetchDsTokenUsd(token: string, signal?: AbortSignal): Promise { + const r = await fetch(`${DS_ROOT}/latest/dex/tokens/${token}`, { signal }) + if (!r.ok) throw new Error(`dexscreener ${r.status}`) + const j = (await r.json()) as { pairs?: DsPair[] } + const price = pickDsTokenUsd(j?.pairs ?? [], token) + if (price === null) throw new Error('no liquid dexscreener pair prices this token') + return price +} + async function fetchV2Subgraph( v2Pools: V2Pool[], wethUsd: number | null, diff --git a/src/lib/tokenPrice.test.ts b/src/lib/tokenPrice.test.ts new file mode 100644 index 0000000..bb002a9 --- /dev/null +++ b/src/lib/tokenPrice.test.ts @@ -0,0 +1,72 @@ +import assert from 'node:assert/strict' +import test from 'node:test' +import { pickDsTokenUsd, type DsPair } from './tokenPrice' + +const UP = '0x57C0E45cB534413D1C20A4240955d6bB250BB4F1' +const WETH = '0x0Bd7D308f8E1639FAb988df18A8011f41EAcAD73' + +// shaped like the live dexscreener rows that motivated this module: two deep +// UP/WETH pools around $0.140 and an $8-liquidity v4 pool marked $0.1585 +const pairs: DsPair[] = [ + { + chainId: 'robinhood', + priceUsd: '0.1405', + priceNative: '0.00007610', + liquidity: { usd: 277107 }, + baseToken: { address: UP }, + quoteToken: { address: WETH }, + }, + { + chainId: 'robinhood', + priceUsd: '0.1402', + priceNative: '0.00007595', + liquidity: { usd: 247016 }, + baseToken: { address: UP }, + quoteToken: { address: WETH }, + }, + { + chainId: 'robinhood', + priceUsd: '0.1585', + liquidity: { usd: 8.25 }, + baseToken: { address: UP }, + quoteToken: { address: WETH }, + }, + // same token on another chain must never leak in + { + chainId: 'base', + priceUsd: '9.99', + liquidity: { usd: 5_000_000 }, + baseToken: { address: UP }, + quoteToken: { address: WETH }, + }, +] + +test('prices from the most-liquid robinhood pair, not the best-looking mid', () => { + assert.equal(pickDsTokenUsd(pairs, UP), 0.1405) + // case-insensitive address match + assert.equal(pickDsTokenUsd(pairs, UP.toLowerCase()), 0.1405) +}) + +test('quote-side tokens derive USD via priceUsd / priceNative', () => { + const weth = pickDsTokenUsd(pairs, WETH) + assert.ok(weth !== null && Math.abs(weth - 0.1405 / 0.0000761) < 0.01) +}) + +test('dust pools below the liquidity floor never price a token', () => { + const dustOnly = pairs.filter((p) => Number(p.liquidity?.usd) < 100) + assert.equal(pickDsTokenUsd(dustOnly, UP), null) +}) + +test('tolerates garbage rows without throwing', () => { + const garbage: DsPair[] = [ + {}, + { chainId: 'robinhood' }, + { chainId: 'robinhood', liquidity: { usd: 'nope' }, baseToken: { address: UP }, priceUsd: '1' }, + { chainId: 'robinhood', liquidity: { usd: 5000 }, baseToken: { address: UP }, priceUsd: '0' }, + // quote-side without priceNative cannot derive — must be skipped + { chainId: 'robinhood', liquidity: { usd: 5000 }, quoteToken: { address: WETH }, priceUsd: '0.14' }, + ] + assert.equal(pickDsTokenUsd(garbage, UP), null) + assert.equal(pickDsTokenUsd(garbage, WETH), null) + assert.equal(pickDsTokenUsd([], UP), null) +}) diff --git a/src/lib/tokenPrice.ts b/src/lib/tokenPrice.ts new file mode 100644 index 0000000..5eff575 --- /dev/null +++ b/src/lib/tokenPrice.ts @@ -0,0 +1,38 @@ +// Venue USD pricing from dexscreener pair rows (pure — fetch lives in poolstats). +// Why not an aggregator unit-quote: selling 1 token routes through whichever +// venue shows the best mid, and on a thin chain that is systematically a stale +// dust pool nobody arbitrages (measured on UP: kyber $0.150 via an $8-liquidity +// v4 pool vs $0.140 on the two $250k pools). Every token except ETH ran high. +// The most-liquid pair's trade-derived price is the honest mark. + +export type DsPair = { + chainId?: string + priceUsd?: string | number + priceNative?: string | number + liquidity?: { usd?: number | string } + baseToken?: { address?: string } + quoteToken?: { address?: string } +} + +/** dust pools sit at stale prices — never price a token off one */ +export const DS_MIN_LIQ_USD = 100 + +/** USD price of `token` from its most-liquid robinhood pair, or null when no + * pair above the dust floor prices it. Quote-side tokens derive via + * priceUsd / priceNative (USD per quote unit). */ +export function pickDsTokenUsd(pairs: DsPair[], token: string): number | null { + const t = token.toLowerCase() + let best: { liq: number; price: number } | null = null + for (const p of pairs) { + if (p?.chainId !== 'robinhood') continue + const liq = Number(p?.liquidity?.usd) + if (!Number.isFinite(liq) || liq < DS_MIN_LIQ_USD) continue + const pu = Number(p?.priceUsd) + const pn = Number(p?.priceNative) + let price: number | null = null + if (p?.baseToken?.address?.toLowerCase() === t && pu > 0) price = pu + else if (p?.quoteToken?.address?.toLowerCase() === t && pu > 0 && pn > 0) price = pu / pn + if (price !== null && (!best || liq > best.liq)) best = { liq, price } + } + return best?.price ?? null +}