fix(indexer): stop a manipulated pool from minting prices

A price used to propagate from any pool with enough depth on the priced side,
which let a thin pool with one inflated side mint a price for its other token
and carry it across the graph. Depth is now measured on the side that is
already credibly priced, and a pool-priced side may claim at most 100x that
depth.

Guard rails around it, all in one place: prices travel at most 3 hops from a
GT/anchor seed, |tick| beyond 700k is a broken init rather than a market,
stored prices must land in a plausibility band, and a TVL above $1B is treated
as corrupt instead of as a whale.

Adds a log tail so pool stats follow chain events instead of polling, a
demand-gated frontpage, and a last-good fallback so a failed refresh serves
the previous stat rather than a hole.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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>): void {
+let refreshQueue = Promise.resolve()
+
+function enqueueRefresh(fn: () => Promise<void>): Promise<void> {
+  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<void>, 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 <T,>(fn: () => Promise<T>): Promise<[T, number]> => {
   return [r, Date.now() - t0]
 }
 
-async function boot(): Promise<void> {
-  log('up33 lp-indexer starting —', usingPrivateRpc ? 'rpc: private (.env)' : 'rpc: public')
-  startApi()
-
+async function initialRefresh(): Promise<void> {
   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<void> {
   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<void> {
   })
   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<void> {
+  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)
 })