diff --git a/crates/pool-indexer/src/db/balancer_v2.rs b/crates/pool-indexer/src/db/balancer_v2.rs new file mode 100644 index 0000000000..c30674771d --- /dev/null +++ b/crates/pool-indexer/src/db/balancer_v2.rs @@ -0,0 +1,116 @@ +use { + crate::{db::bytes_to_addr, indexer::balancer_v2::NewBalancerPool}, + alloy_primitives::Address, + anyhow::{Context, Result}, + bigdecimal::BigDecimal, + sqlx::{PgPool, Postgres, Row, Transaction}, +}; + +/// Inserts discovered pools and their tokens. Pools are written before tokens +/// to satisfy the FK; both `ON CONFLICT DO NOTHING` so re-indexing a pool is a +/// no-op. +pub async fn insert_pools( + tx: &mut Transaction<'_, Postgres>, + factory: &Address, + pools: &[NewBalancerPool], +) -> Result<()> { + if pools.is_empty() { + return Ok(()); + } + + let mut pool_ids: Vec<&[u8]> = Vec::with_capacity(pools.len()); + let mut addresses: Vec<&[u8]> = Vec::with_capacity(pools.len()); + let mut pool_types: Vec<&str> = Vec::with_capacity(pools.len()); + let mut created_blocks: Vec = Vec::with_capacity(pools.len()); + let mut tok_pool_ids: Vec<&[u8]> = Vec::new(); + let mut positions: Vec = Vec::new(); + let mut tokens: Vec<&[u8]> = Vec::new(); + let mut decimals: Vec> = Vec::new(); + let mut weights: Vec> = Vec::new(); + for pool in pools { + pool_ids.push(pool.pool_id.as_slice()); + addresses.push(pool.address.as_slice()); + pool_types.push(pool.pool_type.as_str()); + created_blocks.push(pool.created_block.cast_signed()); + for token in &pool.tokens { + tok_pool_ids.push(pool.pool_id.as_slice()); + positions.push(i32::try_from(token.position).unwrap_or(i32::MAX)); + tokens.push(token.address.as_slice()); + decimals.push(token.decimals.map(i16::from)); + weights.push(token.weight.clone()); + } + } + + sqlx::query( + "INSERT INTO balancer_v2_pools (pool_id, address, factory, pool_type, created_block) + SELECT t.pid, t.addr, $1, t.ptype, t.cblk + FROM UNNEST($2::BYTEA[], $3::BYTEA[], $4::TEXT[], $5::INT8[]) + AS t(pid, addr, ptype, cblk) + ON CONFLICT (pool_id) DO NOTHING", + ) + .bind(factory.as_slice()) + .bind(pool_ids) + .bind(addresses) + .bind(pool_types) + .bind(created_blocks) + .execute(&mut **tx) + .await + .context("insert balancer pools")?; + + sqlx::query( + "INSERT INTO balancer_v2_pool_tokens (pool_id, position, token, decimals, weight) + SELECT t.pid, t.pos, t.tok, t.dec, t.wgt + FROM UNNEST($1::BYTEA[], $2::INT4[], $3::BYTEA[], $4::INT2[], $5::NUMERIC[]) + AS t(pid, pos, tok, dec, wgt) + ON CONFLICT (pool_id, position) DO NOTHING", + ) + .bind(tok_pool_ids) + .bind(positions) + .bind(tokens) + .bind(decimals) + .bind(weights) + .execute(&mut **tx) + .await + .context("insert balancer pool tokens")?; + + Ok(()) +} + +/// Distinct token addresses with no `decimals` recorded yet. +pub async fn get_tokens_missing_decimals(pool: &PgPool) -> Result> { + let rows = + sqlx::query("SELECT DISTINCT token FROM balancer_v2_pool_tokens WHERE decimals IS NULL") + .fetch_all(pool) + .await + .context("get_tokens_missing_decimals")?; + + rows.into_iter() + .map(|r| bytes_to_addr(r.get("token"))) + .collect() +} + +/// Sets `decimals` for every token row matching one of the inputs. Pass `-1` +/// for "tried, failed" so the next backfill's `IS NULL` filter still skips it. +pub async fn batch_set_token_decimals( + tx: &mut Transaction<'_, Postgres>, + entries: &[(Address, i16)], +) -> Result<()> { + if entries.is_empty() { + return Ok(()); + } + let tokens: Vec<&[u8]> = entries.iter().map(|(t, _)| t.as_slice()).collect(); + let decimals: Vec = entries.iter().map(|(_, d)| *d).collect(); + + sqlx::query( + "UPDATE balancer_v2_pool_tokens p + SET decimals = i.dec + FROM UNNEST($1::BYTEA[], $2::INT2[]) AS i(tok, dec) + WHERE p.token = i.tok AND p.decimals IS NULL", + ) + .bind(tokens) + .bind(decimals) + .execute(&mut **tx) + .await + .context("batch_set_token_decimals")?; + Ok(()) +} diff --git a/crates/pool-indexer/src/db/mod.rs b/crates/pool-indexer/src/db/mod.rs index c24144d0e0..433ed0f816 100644 --- a/crates/pool-indexer/src/db/mod.rs +++ b/crates/pool-indexer/src/db/mod.rs @@ -1 +1,46 @@ +pub mod balancer_v2; pub mod uniswap_v3; + +use { + alloy_primitives::Address, + anyhow::{Context, Result}, + sqlx::{PgPool, Postgres, Row, Transaction}, +}; + +/// Decodes a Postgres `BYTEA` column into an [`Address`]. +pub(crate) fn bytes_to_addr(b: Vec) -> Result
{ + Address::try_from(b.as_slice()).context("invalid address bytes") +} + +/// Highest block scanned for a factory's `PoolCreated` events. Shared by both +/// indexers: `pool_indexer_checkpoints` is keyed by factory address, which is +/// unique across protocols, so their rows never collide. +pub async fn get_checkpoint(pool: &PgPool, factory: &Address) -> Result> { + let row = sqlx::query( + "SELECT block_number FROM pool_indexer_checkpoints WHERE contract_address = $1", + ) + .bind(factory.as_slice()) + .fetch_optional(pool) + .await + .context("get_checkpoint")?; + + Ok(row.map(|r| r.get::("block_number").cast_unsigned())) +} + +pub async fn set_checkpoint( + tx: &mut Transaction<'_, Postgres>, + factory: &Address, + block_number: u64, +) -> Result<()> { + sqlx::query( + "INSERT INTO pool_indexer_checkpoints (contract_address, block_number) + VALUES ($1, $2) + ON CONFLICT (contract_address) DO UPDATE SET block_number = EXCLUDED.block_number", + ) + .bind(factory.as_slice()) + .bind(block_number.cast_signed()) + .execute(&mut **tx) + .await + .context("set_checkpoint")?; + Ok(()) +} diff --git a/crates/pool-indexer/src/db/uniswap_v3.rs b/crates/pool-indexer/src/db/uniswap_v3.rs index fa0273f662..cb195428e0 100644 --- a/crates/pool-indexer/src/db/uniswap_v3.rs +++ b/crates/pool-indexer/src/db/uniswap_v3.rs @@ -1,5 +1,8 @@ use { - crate::indexer::uniswap_v3::{LiquidityUpdateData, NewPoolData, PoolStateData, TickDeltaData}, + crate::{ + db::bytes_to_addr, + indexer::uniswap_v3::{LiquidityUpdateData, NewPoolData, PoolStateData, TickDeltaData}, + }, alloy_primitives::Address, anyhow::{Context, Result}, bigdecimal::BigDecimal, @@ -9,10 +12,6 @@ use { std::collections::BTreeSet, }; -fn bytes_to_addr(b: Vec) -> Result
{ - Address::try_from(b.as_slice()).context("invalid address bytes") -} - fn address_bytes_list(addresses: &[Address]) -> Vec<&[u8]> { addresses.iter().map(|address| address.as_slice()).collect() } @@ -21,36 +20,6 @@ fn decode_pool_rows(rows: Vec) -> Result> { rows.into_iter().map(PoolRow::try_from).collect() } -pub async fn get_checkpoint(pool: &PgPool, contract: &Address) -> Result> { - let row = sqlx::query( - "SELECT block_number FROM pool_indexer_checkpoints WHERE contract_address = $1", - ) - .bind(contract.as_slice()) - .fetch_optional(pool) - .await - .context("get_checkpoint")?; - - Ok(row.map(|r| r.get::("block_number").cast_unsigned())) -} - -pub async fn set_checkpoint( - tx: &mut Transaction<'_, Postgres>, - contract: &Address, - block_number: u64, -) -> Result<()> { - sqlx::query( - "INSERT INTO pool_indexer_checkpoints (contract_address, block_number) - VALUES ($1, $2) - ON CONFLICT (contract_address) DO UPDATE SET block_number = EXCLUDED.block_number", - ) - .bind(contract.as_slice()) - .bind(block_number.cast_signed()) - .execute(&mut **tx) - .await - .context("set_checkpoint")?; - Ok(()) -} - pub async fn insert_pools( tx: &mut Transaction<'_, Postgres>, factory: &Address, diff --git a/crates/pool-indexer/src/indexer/balancer_v2.rs b/crates/pool-indexer/src/indexer/balancer_v2.rs new file mode 100644 index 0000000000..1a794a0ab8 --- /dev/null +++ b/crates/pool-indexer/src/indexer/balancer_v2.rs @@ -0,0 +1,591 @@ +use { + crate::{ + config::{BalancerV2Config, FactoryConfig, NetworkName}, + db::{balancer_v2 as db, get_checkpoint, set_checkpoint}, + }, + alloy_primitives::{Address, B256, U256}, + alloy_provider::Provider, + alloy_rpc_types_eth::{BlockNumberOrTag, Log}, + alloy_sol_types::SolEvent, + anyhow::{Context, Result}, + bigdecimal::BigDecimal, + contracts::{ + BalancerV2BasePool, + BalancerV2BasePoolFactory::BalancerV2BasePoolFactory::PoolCreated, + BalancerV2Vault, + BalancerV2WeightedPool, + }, + ethrpc::AlloyProvider, + futures::{StreamExt, TryStreamExt}, + number::conversions::ufixed18_to_big_decimal, + sqlx::PgPool, + std::collections::HashMap, + tracing::instrument, +}; + +const BACKFILL_BATCH_SIZE: usize = 500; + +/// Balancer V2 pool type, implied by the factory group a pool was discovered +/// under. Stored as [`Self::as_str`] and served verbatim by the API; the +/// weighted V0/V3-plus distinction is recovered from the factory address, so +/// it isn't a separate type. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum PoolType { + Weighted, + Stable, + ComposableStable, + LiquidityBootstrapping, +} + +impl PoolType { + /// String stored in `balancer_v2_pools.pool_type`; matches the schema's + /// `CHECK` constraint. + pub fn as_str(self) -> &'static str { + match self { + PoolType::Weighted => "Weighted", + PoolType::Stable => "Stable", + PoolType::ComposableStable => "ComposableStable", + PoolType::LiquidityBootstrapping => "LiquidityBootstrapping", + } + } + + /// Whether pools of this type expose `getNormalizedWeights`. Only weighted + /// pools carry static normalized weights; stable/composable-stable/LBP + /// weights are absent or computed on-chain, so they aren't fetched here. + pub fn has_weights(self) -> bool { + matches!(self, PoolType::Weighted) + } +} + +/// Flattens a [`BalancerV2Config`] into `(pool_type, factory)` pairs across all +/// factory groups. The pool type is implied by the group; both weighted groups +/// map to [`PoolType::Weighted`]. +pub fn configured_factories(config: &BalancerV2Config) -> Vec<(PoolType, FactoryConfig)> { + let groups: [(PoolType, &[FactoryConfig]); 5] = [ + (PoolType::Weighted, &config.weighted), + (PoolType::Weighted, &config.weighted_v3plus), + (PoolType::Stable, &config.stable), + ( + PoolType::LiquidityBootstrapping, + &config.liquidity_bootstrapping, + ), + (PoolType::ComposableStable, &config.composable_stable), + ]; + groups + .into_iter() + .flat_map(|(pool_type, factories)| factories.iter().map(move |f| (pool_type, *f))) + .collect() +} + +/// Config for one Balancer V2 factory discovery loop. +pub struct IndexerConfig { + pub network: NetworkName, + pub vault: Address, + pub factory: Address, + pub pool_type: PoolType, + pub deploy_block: u64, + pub chunk_size: u64, + pub use_latest: bool, + pub fetch_concurrency: usize, + pub enrich_concurrency: usize, +} + +/// A token within a discovered pool, in `Vault.getPoolTokens` order. +pub struct NewPoolToken { + pub position: usize, + pub address: Address, + pub decimals: Option, + /// Normalized weight as a Bfp (1e18) fraction; `Some` only for weighted + /// pools. + pub weight: Option, +} + +/// A pool discovered from a factory `PoolCreated` event, enriched with on-chain +/// metadata. +pub struct NewBalancerPool { + pub pool_id: B256, + pub address: Address, + pub pool_type: PoolType, + pub created_block: u64, + pub tokens: Vec, +} + +#[derive(Clone, Copy, Debug)] +struct ChunkRange { + start: u64, + end: u64, +} + +/// Discovers Balancer V2 pools created by one factory and persists their +/// static metadata. Dynamic state (balances, amp, swap fee, ...) is fetched +/// on-chain by the driver, not here. +pub struct BalancerV2Indexer { + provider: AlloyProvider, + db: PgPool, + network: NetworkName, + vault: Address, + factory: Address, + factory_label: String, + pool_type: PoolType, + deploy_block: u64, + chunk_size: u64, + finality_tag: BlockNumberOrTag, + fetch_concurrency: usize, + enrich_concurrency: usize, +} + +impl BalancerV2Indexer { + pub fn new(provider: AlloyProvider, db: PgPool, config: IndexerConfig) -> Self { + Self { + provider, + db, + network: config.network, + vault: config.vault, + factory: config.factory, + factory_label: format!("{:#x}", config.factory), + pool_type: config.pool_type, + deploy_block: config.deploy_block, + chunk_size: config.chunk_size, + finality_tag: if config.use_latest { + BlockNumberOrTag::Latest + } else { + BlockNumberOrTag::Finalized + }, + fetch_concurrency: config.fetch_concurrency, + enrich_concurrency: config.enrich_concurrency, + } + } + + /// Per-factory live-discovery loop. + pub async fn run(self, poll_interval: std::time::Duration) -> ! { + let mut interval = tokio::time::interval(poll_interval); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + interval.tick().await; + if let Err(err) = self.run_once().await { + crate::metrics::Metrics::get() + .indexer_errors + .with_label_values(&[self.network.as_str(), self.factory_label.as_str()]) + .inc(); + tracing::error!(?err, factory = %self.factory, "balancer indexer error, retrying"); + } + } + } + + /// Scans `PoolCreated` from the factory's deploy block to the finalized + /// head, then returns. Idempotent: skips if a checkpoint already exists, so + /// re-running on a seeded DB is a fast no-op. + pub async fn bootstrap(&self) -> Result<()> { + if get_checkpoint(&self.db, &self.factory).await?.is_some() { + tracing::info!(factory = %self.factory, "existing checkpoint, skipping bootstrap"); + return Ok(()); + } + // `run_once` resumes at `checkpoint + 1`, so start one before the + // deploy block to scan the deploy block itself. + let start = self.deploy_block.saturating_sub(1); + let mut tx = self.db.begin().await.context("begin checkpoint tx")?; + set_checkpoint(&mut tx, &self.factory, start).await?; + tx.commit().await.context("commit checkpoint tx")?; + + loop { + let finalized = self.finalized_block().await?; + if self.last_indexed_block().await? >= finalized { + tracing::info!(block = finalized, factory = %self.factory, "balancer bootstrap caught up"); + return Ok(()); + } + self.run_once().await?; + } + } + + async fn run_once(&self) -> Result<()> { + let finalized = self.finalized_block().await?; + let last = self.last_indexed_block().await?; + let lag = finalized.saturating_sub(last); + crate::metrics::Metrics::get() + .indexer_lag_blocks + .with_label_values(&[self.network.as_str(), self.factory_label.as_str()]) + .set(i64::try_from(lag).unwrap_or(i64::MAX)); + if last >= finalized { + return Ok(()); + } + + // Fetch chunks in parallel, commit in order. + futures::stream::iter(self.pending_chunks(last, finalized)) + .map(|chunk| async move { + let logs = self.fetch_pool_created(chunk.start, chunk.end).await?; + Ok::<_, anyhow::Error>((chunk, logs)) + }) + .buffered(self.fetch_concurrency) + .try_for_each(|(chunk, logs)| self.commit_chunk(chunk, logs, finalized)) + .await?; + Ok(()) + } + + async fn finalized_block(&self) -> Result { + Ok(self + .provider + .get_block_by_number(self.finality_tag) + .await + .context("get finalized block")? + .context("no finalized block")? + .header + .number) + } + + async fn last_indexed_block(&self) -> Result { + Ok(get_checkpoint(&self.db, &self.factory).await?.unwrap_or(0)) + } + + fn pending_chunks(&self, last: u64, finalized: u64) -> Vec { + let mut chunks = Vec::new(); + let mut next_start = last + 1; + while next_start <= finalized { + let end = (next_start + self.chunk_size - 1).min(finalized); + chunks.push(ChunkRange { + start: next_start, + end, + }); + next_start = end + 1; + } + chunks + } + + /// `PoolCreated` is emitted by the factory only, so filter by the factory + /// address — the query stays tiny even over large ranges. + async fn fetch_pool_created(&self, from: u64, to: u64) -> Result> { + super::bisecting_get_logs( + &self.provider, + from, + to, + vec![self.factory], + vec![PoolCreated::SIGNATURE_HASH], + ) + .await + } + + #[instrument(skip(self, logs), fields(chunk_start = chunk.start, chunk_end = chunk.end))] + async fn commit_chunk(&self, chunk: ChunkRange, logs: Vec, target: u64) -> Result<()> { + let pools: Vec = + futures::stream::iter(pool_addresses(self.factory, &logs)) + .map(|(pool, block)| self.enrich_pool(pool, block)) + .buffer_unordered(self.enrich_concurrency) + .filter_map(|res| async move { + match res { + Ok(pool) => pool, + Err(err) => { + tracing::warn!(?err, "balancer pool enrichment failed; skipping"); + None + } + } + }) + .collect() + .await; + + let network = self.network.as_str(); + let factory = self.factory_label.as_str(); + crate::metrics::Metrics::get() + .events_applied + .with_label_values(&[network, factory, "new_pool"]) + .inc_by(pools.len() as u64); + + let mut tx = self.db.begin().await.context("begin transaction")?; + db::insert_pools(&mut tx, &self.factory, &pools).await?; + set_checkpoint(&mut tx, &self.factory, chunk.end).await?; + tx.commit().await.context("commit transaction")?; + + let metrics = crate::metrics::Metrics::get(); + metrics + .indexed_block + .with_label_values(&[network, factory]) + .set(i64::try_from(chunk.end).unwrap_or(i64::MAX)); + let lag = target.saturating_sub(chunk.end); + metrics + .indexer_lag_blocks + .with_label_values(&[network, factory]) + .set(i64::try_from(lag).unwrap_or(i64::MAX)); + Ok(()) + } + + /// Enriches a freshly-discovered pool with its on-chain metadata: + /// `getPoolId` → `Vault.getPoolTokens` → per-token `decimals` → + /// `getNormalizedWeights` (weighted pools). Returns `None` if a required + /// call fails (the pool is retried on the next pass — the checkpoint isn't + /// advanced past it until it inserts). + async fn enrich_pool( + &self, + pool: Address, + created_block: u64, + ) -> Result> { + let pool_id = match BalancerV2BasePool::Instance::new(pool, self.provider.clone()) + .getPoolId() + .call() + .await + { + Ok(id) => id, + Err(err) => { + tracing::warn!(%pool, ?err, "getPoolId failed; skipping pool"); + return Ok(None); + } + }; + + let tokens = BalancerV2Vault::Instance::new(self.vault, self.provider.clone()) + .getPoolTokens(pool_id.0.into()) + .call() + .await + .context("getPoolTokens")? + .tokens; + + let weights = if self.pool_type.has_weights() { + match BalancerV2WeightedPool::Instance::new(pool, self.provider.clone()) + .getNormalizedWeights() + .call() + .await + { + Ok(weights) => Some(weights), + Err(err) => { + tracing::warn!(%pool, ?err, "getNormalizedWeights failed; skipping pool"); + return Ok(None); + } + } + } else { + None + }; + + let decimals: HashMap = futures::stream::iter(tokens.clone()) + .map(|token| async move { (token, super::fetch_decimals(&self.provider, token).await) }) + .buffer_unordered(self.enrich_concurrency) + .filter_map(|(token, decimals)| async move { decimals.map(|d| (token, d)) }) + .collect() + .await; + + Ok(Some(assemble_pool( + pool, + pool_id, + self.pool_type, + created_block, + tokens, + &decimals, + weights, + ))) + } +} + +/// Decodes factory `PoolCreated` events into `(pool address, created block)`. +/// Logs are already factory-filtered; the emitter check is defense-in-depth. +fn pool_addresses(factory: Address, logs: &[Log]) -> Vec<(Address, u64)> { + logs.iter() + .filter_map(|log| { + let topic = log.topic0()?; + if *topic != PoolCreated::SIGNATURE_HASH || log.address() != factory { + return None; + } + let decoded = PoolCreated::decode_log(&log.inner).ok()?; + Some((decoded.data.pool, log.block_number.unwrap_or_default())) + }) + .collect() +} + +/// Builds a pool row from its enrichment results. Tokens keep their +/// `getPoolTokens` order via `position`; `weights[i]` aligns with `tokens[i]` +/// and is converted from 1e18 fixed-point to a decimal fraction. Pure (no I/O) +/// so the mapping is unit-testable. +fn assemble_pool( + pool: Address, + pool_id: B256, + pool_type: PoolType, + created_block: u64, + tokens: Vec
, + decimals: &HashMap, + weights: Option>, +) -> NewBalancerPool { + let weights = weights.unwrap_or_default(); + let tokens = tokens + .into_iter() + .enumerate() + .map(|(position, address)| NewPoolToken { + position, + address, + decimals: decimals.get(&address).copied(), + weight: weights.get(position).map(ufixed18_to_big_decimal), + }) + .collect(); + NewBalancerPool { + pool_id, + address: pool, + pool_type, + created_block, + tokens, + } +} + +/// Periodically fills `decimals` on pool tokens whose discovery-time +/// `decimals()` call failed (or was never made). `-1` is the "tried, failed" +/// sentinel so a known-broken token isn't probed every pass. +pub(crate) async fn backfill_decimals( + provider: AlloyProvider, + db: PgPool, + network: NetworkName, + concurrency: usize, + poll_interval: std::time::Duration, +) { + let mut interval = tokio::time::interval(poll_interval); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + interval.tick().await; + if let Err(err) = run_decimals_backfill_pass(&provider, &db, &network, concurrency).await { + tracing::error!(?err, "balancer token decimals backfill pass failed"); + } + } +} + +async fn run_decimals_backfill_pass( + provider: &AlloyProvider, + db: &PgPool, + network: &NetworkName, + concurrency: usize, +) -> Result<()> { + let tokens = db::get_tokens_missing_decimals(db).await?; + let network = network.as_str(); + crate::metrics::Metrics::get() + .backfill_pending + .with_label_values(&[network, "decimals"]) + .set(i64::try_from(tokens.len()).unwrap_or(-1)); + if tokens.is_empty() { + return Ok(()); + } + let total = tokens.len(); + tracing::info!(total, "backfilling balancer token decimals"); + + let mut stream = futures::stream::iter(tokens) + .map(|token| async move { + // `None` → `-1` is the "tried, failed" sentinel; the next pass's + // `IS NULL` filter skips it. + let decimals = super::fetch_decimals(provider, token) + .await + .map(i16::from) + .unwrap_or(-1); + (token, decimals) + }) + .buffer_unordered(concurrency) + .ready_chunks(BACKFILL_BATCH_SIZE); + + let mut updated = 0usize; + while let Some(batch) = stream.next().await { + match write_decimals_batch(db, &batch).await { + Ok(()) => { + for (_, decimals) in &batch { + updated += 1; + let result = if *decimals < 0 { "empty" } else { "ok" }; + crate::metrics::Metrics::get() + .backfilled + .with_label_values(&[network, "decimals", result]) + .inc(); + } + } + Err(err) => tracing::warn!(?err, "failed to backfill balancer decimals batch"), + } + } + tracing::info!(updated, total, "balancer token decimals backfill complete"); + Ok(()) +} + +async fn write_decimals_batch(db: &PgPool, entries: &[(Address, i16)]) -> Result<()> { + let mut tx = db.begin().await.context("begin decimals batch tx")?; + db::batch_set_token_decimals(&mut tx, entries).await?; + tx.commit().await.context("commit decimals batch tx")?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + const POOL: Address = Address::repeat_byte(0x11); + const TOKEN0: Address = Address::repeat_byte(0x01); + const TOKEN1: Address = Address::repeat_byte(0x02); + // 0.5 in Balancer's 1e18 fixed point. + const HALF: u64 = 500_000_000_000_000_000; + + #[test] + fn assembles_weighted_pool_with_fractional_weights() { + let pool_id = B256::repeat_byte(0x22); + // TOKEN1 deliberately missing from the decimals cache. + let decimals = HashMap::from([(TOKEN0, 18u8)]); + let weights = Some(vec![U256::from(HALF), U256::from(HALF)]); + + let pool = assemble_pool( + POOL, + pool_id, + PoolType::Weighted, + 100, + vec![TOKEN0, TOKEN1], + &decimals, + weights, + ); + + assert_eq!(pool.pool_id, pool_id); + assert_eq!(pool.address, POOL); + assert_eq!(pool.created_block, 100); + assert_eq!(pool.tokens.len(), 2); + + assert_eq!(pool.tokens[0].position, 0); + assert_eq!(pool.tokens[0].address, TOKEN0); + assert_eq!(pool.tokens[0].decimals, Some(18)); + assert_eq!( + pool.tokens[0].weight.as_ref().unwrap().to_string(), + "0.500000000000000000" + ); + + assert_eq!(pool.tokens[1].position, 1); + assert_eq!(pool.tokens[1].decimals, None); + assert!(pool.tokens[1].weight.is_some()); + } + + #[test] + fn assembles_non_weighted_pool_without_weights() { + let decimals = HashMap::from([(TOKEN0, 6u8)]); + let pool = assemble_pool( + POOL, + B256::repeat_byte(0x22), + PoolType::Stable, + 100, + vec![TOKEN0], + &decimals, + None, + ); + assert_eq!(pool.tokens.len(), 1); + assert_eq!(pool.tokens[0].decimals, Some(6)); + assert!(pool.tokens[0].weight.is_none()); + } + + #[test] + fn configured_factories_maps_groups_to_pool_types() { + let f = |b| FactoryConfig { + address: Address::ZERO, + deploy_block: b, + }; + let config = BalancerV2Config { + vault: Address::ZERO, + chunk_size: 100_000, + weighted: vec![f(1)], + weighted_v3plus: vec![f(2)], + stable: vec![f(3)], + liquidity_bootstrapping: vec![f(4)], + composable_stable: vec![f(5)], + }; + let got: Vec<_> = configured_factories(&config) + .into_iter() + .map(|(t, factory)| (t, factory.deploy_block)) + .collect(); + assert_eq!( + got, + vec![ + (PoolType::Weighted, 1), + (PoolType::Weighted, 2), + (PoolType::Stable, 3), + (PoolType::LiquidityBootstrapping, 4), + (PoolType::ComposableStable, 5), + ] + ); + } +} diff --git a/crates/pool-indexer/src/indexer/mod.rs b/crates/pool-indexer/src/indexer/mod.rs index c24144d0e0..b6afd520ab 100644 --- a/crates/pool-indexer/src/indexer/mod.rs +++ b/crates/pool-indexer/src/indexer/mod.rs @@ -1 +1,159 @@ +pub mod balancer_v2; pub mod uniswap_v3; + +use { + alloy_primitives::{Address, B256}, + alloy_provider::Provider, + alloy_rpc_types_eth::{Filter, FilterSet, Log}, + alloy_transport::RpcError, + anyhow::Result, + contracts::ERC20, + ethrpc::{AlloyProvider, alloy::errors::ContractErrorExt}, + std::time::Duration, +}; + +/// Retries `f` while the error is a transient transport failure +/// (`is_node_error`). Contract reverts and decoding failures bail out +/// immediately. On giveup, hands `on_giveup` the accumulated errors and +/// returns `None`. +pub(crate) async fn retry_node_call( + f: impl Fn() -> Fut, + on_giveup: impl FnOnce(&[alloy_contract::Error]), +) -> Option +where + Fut: std::future::Future>, +{ + match shared::retry::retry_with_sleep_if(f, |err: &alloy_contract::Error| err.is_node_error()) + .await + { + Ok(v) => Some(v), + Err(errors) => { + on_giveup(&errors); + None + } + } +} + +/// Fetches a token's `decimals()`, retrying transient node errors. Returns +/// `None` if the call ultimately fails (revert / decode / giveup). +pub(crate) async fn fetch_decimals(provider: &AlloyProvider, token: Address) -> Option { + retry_node_call( + || async move { + ERC20::Instance::new(token, provider.clone()) + .decimals() + .call() + .await + }, + |errors| tracing::warn!(%token, ?errors, "fetch_decimals gave up"), + ) + .await +} + +/// True if the server-side JSON-RPC payload rejected `eth_getLogs` for +/// being too wide / returning too many logs / exceeding a response-size +/// cap / hitting the server's query timeout. Substrings cover the +/// rejections empirically seen on OVH and Alchemy mainnet. Transport-level +/// errors (HTTP timeouts, DNS, connection resets) live in other `RpcError` +/// variants and short-circuit to false, so client-side noise can't trigger +/// pointless bisection. +pub(crate) fn is_range_too_large(err: &alloy_transport::TransportError) -> bool { + let RpcError::ErrorResp(payload) = err else { + return false; + }; + let msg = payload.message.to_lowercase(); + msg.contains("max block range") + || msg.contains("max results") + || msg.contains("log response size exceeded") + || msg.contains("query timeout exceeded") + || msg.contains("response is too big") +} + +/// Bisecting bound — substring matching on RPC error messages is necessarily +/// approximate, and a misclassified error would otherwise burn `log2(range)` +/// RPC calls before the recursion bottoms out at `to == from`. 8 halvings = +/// 256× resolution; for the indexer's ~1k-block chunks that means giving up +/// around ~4-block ranges, well past where range-size could plausibly still +/// be the cause. +const MAX_BISECTION_DEPTH: u32 = 8; + +/// Retry transient `eth_getLogs` failures (timeout, reset, throttle) with +/// backoff, capped by a per-call timeout, so one blip can't abort a long +/// cold-seed scan. Range-size rejections are bisected, not retried. +const GETLOGS_TIMEOUT: Duration = Duration::from_secs(45); +const MAX_GETLOGS_RETRIES: u32 = 6; +const GETLOGS_RETRY_BACKOFF: Duration = Duration::from_secs(2); + +/// Fetches logs for `[from, to]` filtered by the given contract addresses +/// and `topic0` event signatures, sequentially bisecting the block range on +/// "too large" rejections until each sub-range is tractable. An empty +/// `addresses` list means "any contract". Bisection depth is capped by +/// [`MAX_BISECTION_DEPTH`]. +pub(crate) fn bisecting_get_logs( + provider: &AlloyProvider, + from: u64, + to: u64, + addresses: Vec
, + topics: Vec, +) -> futures::future::BoxFuture<'_, Result>> { + bisecting_get_logs_with_depth(provider, from, to, addresses, topics, 0) +} + +fn bisecting_get_logs_with_depth( + provider: &AlloyProvider, + from: u64, + to: u64, + addresses: Vec
, + topics: Vec, + depth: u32, +) -> futures::future::BoxFuture<'_, Result>> { + Box::pin(async move { + let filter = Filter::new() + .address(addresses.clone()) + .event_signature(FilterSet::from_iter(topics.clone())) + .from_block(from) + .to_block(to); + + let mut attempt = 0u32; + loop { + let err = match tokio::time::timeout(GETLOGS_TIMEOUT, provider.get_logs(&filter)).await + { + Ok(Ok(logs)) => return Ok(logs), + // Range-size rejection: bisect (below), not retried. + Ok(Err(err)) if is_range_too_large(&err) => { + if to <= from || depth >= MAX_BISECTION_DEPTH { + return Err( + anyhow::Error::new(err).context(format!("get_logs({from}..={to})")) + ); + } + break; + } + Ok(Err(err)) => anyhow::Error::new(err).context(format!("get_logs({from}..={to})")), + Err(_elapsed) => anyhow::anyhow!("get_logs({from}..={to}) timed out"), + }; + // Transient failure: retry with backoff, then give up. + if attempt >= MAX_GETLOGS_RETRIES { + return Err(err); + } + attempt += 1; + tracing::warn!(%err, attempt, from, to, "get_logs failed, retrying"); + tokio::time::sleep(GETLOGS_RETRY_BACKOFF * attempt).await; + } + + let mid = (from + to) / 2; + tracing::debug!(from, to, mid, depth, "range too large, bisecting"); + let mut left = bisecting_get_logs_with_depth( + provider, + from, + mid, + addresses.clone(), + topics.clone(), + depth + 1, + ) + .await?; + let right = + bisecting_get_logs_with_depth(provider, mid + 1, to, addresses, topics, depth + 1) + .await?; + left.extend(right); + Ok(left) + }) +} diff --git a/crates/pool-indexer/src/indexer/uniswap_v3.rs b/crates/pool-indexer/src/indexer/uniswap_v3.rs index 72bbbae44a..6a6c6f3745 100644 --- a/crates/pool-indexer/src/indexer/uniswap_v3.rs +++ b/crates/pool-indexer/src/indexer/uniswap_v3.rs @@ -1,20 +1,19 @@ use { crate::{ config::{IndexerConfig, NetworkName}, - db::uniswap_v3 as db, + db::{get_checkpoint, set_checkpoint, uniswap_v3 as db}, }, - alloy_primitives::{Address, B256, aliases::U160}, + alloy_primitives::{Address, aliases::U160}, alloy_provider::Provider, - alloy_rpc_types_eth::{BlockNumberOrTag, Filter, FilterSet, Log}, + alloy_rpc_types_eth::{BlockNumberOrTag, Log}, alloy_sol_types::SolEvent, - alloy_transport::RpcError, anyhow::{Context, Result}, contracts::{ ERC20, IUniswapV3Factory::IUniswapV3Factory::PoolCreated, UniswapV3Pool::UniswapV3Pool::{Burn, Initialize, Mint, Swap}, }, - ethrpc::{AlloyProvider, alloy::errors::ContractErrorExt}, + ethrpc::AlloyProvider, futures::{StreamExt, TryStreamExt}, itertools::Itertools, sqlx::PgPool, @@ -153,7 +152,7 @@ impl UniswapV3Indexer { /// Bails if a checkpoint already exists; overwriting would silently /// regress and re-index history. pub async fn catch_up(&self, from_block: u64) -> Result<()> { - if db::get_checkpoint(&self.db, &self.factory).await?.is_some() { + if get_checkpoint(&self.db, &self.factory).await?.is_some() { anyhow::bail!( "catch_up called but checkpoint already exists for chain {} factory {}", self.chain_id, @@ -161,7 +160,7 @@ impl UniswapV3Indexer { ); } let mut tx = self.db.begin().await.context("begin checkpoint tx")?; - db::set_checkpoint(&mut tx, &self.factory, from_block).await?; + set_checkpoint(&mut tx, &self.factory, from_block).await?; tx.commit().await.context("commit checkpoint tx")?; self.catch_up_to_finalized().await @@ -228,9 +227,7 @@ impl UniswapV3Indexer { } async fn last_indexed_block(&self) -> Result { - Ok(db::get_checkpoint(&self.db, &self.factory) - .await? - .unwrap_or(0)) + Ok(get_checkpoint(&self.db, &self.factory).await?.unwrap_or(0)) } fn pending_chunks(&self, last_indexed_block: u64, finalized_block: u64) -> Vec { @@ -263,7 +260,7 @@ impl UniswapV3Indexer { // // The `WHERE EXISTS (… uniswap_v3_pools …)` clauses in the batch // writers stay as defense-in-depth. - bisecting_get_logs( + super::bisecting_get_logs( &self.provider, from, to, @@ -360,7 +357,7 @@ impl UniswapV3Indexer { db::upsert_pool_states(&mut tx, &self.factory, &changes.pool_states).await?; db::batch_update_pool_liquidity(&mut tx, &self.factory, &changes.liquidity_updates).await?; db::batch_update_ticks(&mut tx, &self.factory, &changes.tick_deltas).await?; - db::set_checkpoint(&mut tx, &self.factory, chunk.end).await?; + set_checkpoint(&mut tx, &self.factory, chunk.end).await?; tx.commit().await.context("commit transaction")?; Ok(()) @@ -371,7 +368,7 @@ impl UniswapV3Indexer { async fn prefetch_decimals(&self, logs: &[Log]) -> DecimalsCache { futures::stream::iter(pool_created_token_addresses(self.factory, logs)) .map(|token| async move { - let dec = fetch_decimals(&self.provider, token).await; + let dec = super::fetch_decimals(&self.provider, token).await; (token, dec) }) .buffer_unordered(self.prefetch_concurrency) @@ -381,28 +378,6 @@ impl UniswapV3Indexer { } } -/// Retries `f` while the error is a transient transport failure -/// (`is_node_error`). Contract reverts and decoding failures bail out -/// immediately. On giveup, hands `on_giveup` the accumulated errors and -/// returns `None`. -async fn retry_node_call( - f: impl Fn() -> Fut, - on_giveup: impl FnOnce(&[alloy_contract::Error]), -) -> Option -where - Fut: std::future::Future>, -{ - match shared::retry::retry_with_sleep_if(f, |err: &alloy_contract::Error| err.is_node_error()) - .await - { - Ok(v) => Some(v), - Err(errors) => { - on_giveup(&errors); - None - } - } -} - /// Unique addresses with a `Mint` or `Burn` in the chunk. We pre-load /// their `(tick, liquidity)` so the post-event liquidity can be derived /// from the in-event delta without an extra RPC. @@ -433,19 +408,6 @@ fn pool_event_emitters(logs: &[Log]) -> Vec
{ .collect() } -async fn fetch_decimals(provider: &AlloyProvider, token: Address) -> Option { - retry_node_call( - || async move { - ERC20::Instance::new(token, provider.clone()) - .decimals() - .call() - .await - }, - |errors| tracing::warn!(%token, ?errors, "fetch_decimals gave up"), - ) - .await -} - /// Periodically fills `token{0,1}_symbol` on newly-indexed pools. /// /// Tokens whose `symbol()` call fails (revert / decode / empty) are stored @@ -586,7 +548,7 @@ async fn run_decimals_backfill_pass( .map(|token| async move { // `None` → `-1` is the "tried and failed" sentinel; the // `IS NULL` filter on the next pass skips it. - let dec = fetch_decimals(provider, token) + let dec = super::fetch_decimals(provider, token) .await .map(i16::from) .unwrap_or(-1); @@ -655,115 +617,6 @@ async fn fetch_symbol(provider: &AlloyProvider, token: Address) -> Option bool { - let RpcError::ErrorResp(payload) = err else { - return false; - }; - let msg = payload.message.to_lowercase(); - msg.contains("max block range") - || msg.contains("max results") - || msg.contains("log response size exceeded") - || msg.contains("query timeout exceeded") - || msg.contains("response is too big") -} - -/// Bisecting bound — substring matching on RPC error messages is necessarily -/// approximate, and a misclassified error would otherwise burn `log2(range)` -/// RPC calls before the recursion bottoms out at `to == from`. 8 halvings = -/// 256× resolution; for the indexer's ~1k-block chunks that means giving up -/// around ~4-block ranges, well past where range-size could plausibly still -/// be the cause. -const MAX_BISECTION_DEPTH: u32 = 8; - -/// Retry transient `eth_getLogs` failures (timeout, reset, throttle) with -/// backoff, capped by a per-call timeout, so one blip can't abort a long -/// cold-seed scan. Range-size rejections are bisected, not retried. -const GETLOGS_TIMEOUT: Duration = Duration::from_secs(45); -const MAX_GETLOGS_RETRIES: u32 = 6; -const GETLOGS_RETRY_BACKOFF: Duration = Duration::from_secs(2); - -/// Fetches logs for `[from, to]` filtered by the given contract addresses -/// and `topic0` event signatures, sequentially bisecting the block range on -/// "too large" rejections until each sub-range is tractable. An empty -/// `addresses` list means "any contract". Bisection depth is capped by -/// [`MAX_BISECTION_DEPTH`]. -pub(crate) fn bisecting_get_logs( - provider: &AlloyProvider, - from: u64, - to: u64, - addresses: Vec
, - topics: Vec, -) -> futures::future::BoxFuture<'_, Result>> { - bisecting_get_logs_with_depth(provider, from, to, addresses, topics, 0) -} - -fn bisecting_get_logs_with_depth( - provider: &AlloyProvider, - from: u64, - to: u64, - addresses: Vec
, - topics: Vec, - depth: u32, -) -> futures::future::BoxFuture<'_, Result>> { - Box::pin(async move { - let filter = Filter::new() - .address(addresses.clone()) - .event_signature(FilterSet::from_iter(topics.clone())) - .from_block(from) - .to_block(to); - - let mut attempt = 0u32; - loop { - let err = match tokio::time::timeout(GETLOGS_TIMEOUT, provider.get_logs(&filter)).await - { - Ok(Ok(logs)) => return Ok(logs), - // Range-size rejection: bisect (below), not retried. - Ok(Err(err)) if is_range_too_large(&err) => { - if to <= from || depth >= MAX_BISECTION_DEPTH { - return Err( - anyhow::Error::new(err).context(format!("get_logs({from}..={to})")) - ); - } - break; - } - Ok(Err(err)) => anyhow::Error::new(err).context(format!("get_logs({from}..={to})")), - Err(_elapsed) => anyhow::anyhow!("get_logs({from}..={to}) timed out"), - }; - // Transient failure: retry with backoff, then give up. - if attempt >= MAX_GETLOGS_RETRIES { - return Err(err); - } - attempt += 1; - tracing::warn!(%err, attempt, from, to, "get_logs failed, retrying"); - tokio::time::sleep(GETLOGS_RETRY_BACKOFF * attempt).await; - } - - let mid = (from + to) / 2; - tracing::debug!(from, to, mid, depth, "range too large, bisecting"); - let mut left = bisecting_get_logs_with_depth( - provider, - from, - mid, - addresses.clone(), - topics.clone(), - depth + 1, - ) - .await?; - let right = - bisecting_get_logs_with_depth(provider, mid + 1, to, addresses, topics, depth + 1) - .await?; - left.extend(right); - Ok(left) - }) -} - /// Collects the unique set of token addresses from all `PoolCreated` events /// emitted by `factory` in `logs`. fn pool_created_token_addresses( diff --git a/crates/pool-indexer/src/run.rs b/crates/pool-indexer/src/run.rs index 44d85d59f8..0810445a91 100644 --- a/crates/pool-indexer/src/run.rs +++ b/crates/pool-indexer/src/run.rs @@ -2,8 +2,11 @@ use { crate::{ api::AppState, arguments::Arguments, - config::{Configuration, NetworkConfig}, - indexer::uniswap_v3::UniswapV3Indexer, + config::{BalancerV2Config, Configuration, FactoryConfig, NetworkConfig}, + indexer::{ + balancer_v2::{self, BalancerV2Indexer, PoolType}, + uniswap_v3::UniswapV3Indexer, + }, }, alloy_provider::Provider, clap::Parser, @@ -61,6 +64,21 @@ pub async fn bootstrap(config: Configuration) { }); } } + if let Some(balancer) = &network.balancer_v2 { + for (pool_type, factory) in balancer_v2::configured_factories(balancer) { + let indexer = BalancerV2Indexer::new( + provider.clone(), + db.clone(), + balancer_indexer_config(&network, balancer, pool_type, factory), + ); + factory_set.spawn(async move { + indexer + .bootstrap() + .await + .expect("balancer bootstrap failed"); + }); + } + } while let Some(result) = factory_set.join_next().await { result.expect("bootstrap task panicked"); } @@ -78,7 +96,12 @@ pub async fn run(config: Configuration) { .network .uniswap_v3 .as_ref() - .map_or(0, |u| u.factories.len()), + .map_or(0, |u| u.factories.len()) + + config + .network + .balancer_v2 + .as_ref() + .map_or(0, |b| b.factory_count()), )); // Abort the metrics task when `run` exits, so tests can rebind the port. @@ -178,10 +201,11 @@ async fn run_network_indexer(db: PgPool, network: NetworkConfig, barrier: Arc balancer_v2::IndexerConfig { + balancer_v2::IndexerConfig { + network: network.name.clone(), + vault: balancer.vault, + factory: factory.address, + pool_type, + deploy_block: factory.deploy_block, + chunk_size: balancer.chunk_size, + use_latest: network.use_latest, + fetch_concurrency: network.fetch_concurrency, + enrich_concurrency: network.prefetch_concurrency, + } +} + async fn run_factory_indexer( db: PgPool, indexer: UniswapV3Indexer, @@ -258,7 +325,7 @@ async fn bootstrap_factory( network: &NetworkConfig, factory: &crate::config::FactoryConfig, ) { - let checkpoint = crate::db::uniswap_v3::get_checkpoint(db, &factory.address) + let checkpoint = crate::db::get_checkpoint(db, &factory.address) .await .expect("failed to read checkpoint"); match checkpoint {