diff --git a/crates/e2e/tests/e2e/pool_indexer.rs b/crates/e2e/tests/e2e/pool_indexer.rs index 73afccf9c9..990b3cd098 100644 --- a/crates/e2e/tests/e2e/pool_indexer.rs +++ b/crates/e2e/tests/e2e/pool_indexer.rs @@ -22,6 +22,7 @@ use { MetricsConfig, NetworkConfig, NetworkName, + UniswapV3Config, }, serde::Deserialize, sqlx::PgPool, @@ -204,18 +205,21 @@ fn pool_indexer_config( name: NetworkName::new("mainnet"), chain_id: 1, rpc_url: "http://127.0.0.1:8545".parse().unwrap(), - factories: factories - .into_iter() - .map(|address| FactoryConfig { - address, - deploy_block: 0, - }) - .collect(), - chunk_size: 1000, + uniswap_v3: Some(UniswapV3Config { + factories: factories + .into_iter() + .map(|address| FactoryConfig { + address, + deploy_block: 0, + }) + .collect(), + chunk_size: 1000, + }), poll_interval_secs: 1, use_latest: true, fetch_concurrency: 8, prefetch_concurrency: 50, + balancer_v2: None, }, api: ApiConfig { bind_address: SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, POOL_INDEXER_PORT)), diff --git a/crates/pool-indexer/src/config.rs b/crates/pool-indexer/src/config.rs index 629bb4876d..11ef3a34f5 100644 --- a/crates/pool-indexer/src/config.rs +++ b/crates/pool-indexer/src/config.rs @@ -3,6 +3,7 @@ use { anyhow::{Context, Result}, serde::Deserialize, std::{ + collections::HashSet, fmt, net::{Ipv4Addr, SocketAddr, SocketAddrV4}, num::NonZeroU32, @@ -16,10 +17,14 @@ const fn default_max_connections() -> NonZeroU32 { NonZeroU32::new(10).expect("non-zero literal") } -const fn default_chunk_size() -> u64 { +const fn default_uniswap_v3_chunk_size() -> u64 { 500 } +const fn default_balancer_v2_chunk_size() -> u64 { + 100_000 +} + const fn default_poll_interval_secs() -> u64 { 3 } @@ -82,26 +87,26 @@ pub struct NetworkConfig { pub chain_id: u64, #[serde(deserialize_with = "configs::deserialize_env::deserialize_url_from_env")] pub rpc_url: Url, - /// Uniswap V3 factories to index; non-empty and unique, enforced at parse. - #[serde(deserialize_with = "serde_ext::deserialize_nonempty_unique_vec")] - pub factories: Vec, - /// Blocks per `eth_getLogs` chunk during catch-up. - #[serde(default = "default_chunk_size")] - pub chunk_size: u64, /// Interval for polling for new blocks during live indexing. #[serde(default = "default_poll_interval_secs")] pub poll_interval_secs: u64, /// Number of `eth_getLogs` chunks fetched in parallel during live indexing. #[serde(default = "default_fetch_concurrency")] pub fetch_concurrency: usize, - /// `symbol()` / `decimals()` token-metadata RPC calls in flight during - /// the backfill passes. + /// Token-metadata RPC calls (`decimals()`, plus `symbol()` for Uniswap V3) + /// in flight during the backfill/enrich passes. #[serde(default = "default_prefetch_concurrency")] pub prefetch_concurrency: usize, /// Use `latest` instead of `finalized` as the indexing head. Set by tests /// against Anvil, which doesn't simulate finality. #[serde(skip)] pub use_latest: bool, + /// Uniswap V3 pools to index. Set on its own or alongside `balancer_v2`. + #[serde(default)] + pub uniswap_v3: Option, + /// Balancer V2 pools to index. Set on its own or alongside `uniswap_v3`. + #[serde(default)] + pub balancer_v2: Option, } impl NetworkConfig { @@ -109,29 +114,121 @@ impl NetworkConfig { Duration::from_secs(self.poll_interval_secs) } - pub fn indexer_config(&self, factory: Address) -> IndexerConfig { + pub fn indexer_config(&self, chunk_size: u64, factory: Address) -> IndexerConfig { IndexerConfig { network: self.name.clone(), chain_id: self.chain_id, factory_address: factory, - chunk_size: self.chunk_size, + chunk_size, use_latest: self.use_latest, fetch_concurrency: self.fetch_concurrency, prefetch_concurrency: self.prefetch_concurrency, } } + + /// Cross-field checks: index at least one protocol, and every factory + /// address (across both) is unique — checkpoints are keyed by factory + /// address in the shared `pool_indexer_checkpoints`, so a repeated address + /// would drive two indexer loops onto one row. + fn validate(&self) -> Result<()> { + anyhow::ensure!( + self.uniswap_v3.is_some() || self.balancer_v2.is_some(), + "network {}: configure at least one of `uniswap-v3` or `balancer-v2`", + self.name, + ); + if let Some(balancer) = &self.balancer_v2 { + anyhow::ensure!( + balancer.factory_count() > 0, + "network {}: balancer-v2 requires at least one factory", + self.name, + ); + } + let uniswap_factories = self + .uniswap_v3 + .iter() + .flat_map(|u| &u.factories) + .map(|f| f.address); + let balancer_factories = self + .balancer_v2 + .iter() + .flat_map(|b| b.factories()) + .map(|f| f.address); + let mut seen = HashSet::new(); + for factory in uniswap_factories.chain(balancer_factories) { + anyhow::ensure!( + seen.insert(factory), + "network {}: factory {factory} configured more than once", + self.name, + ); + } + Ok(()) + } } -/// The factory and the block it was deployed at. The indexer cold-seeds by -/// replaying on-chain events from `deploy_block`, then live-indexes. +/// A pool factory and the block it was deployed at; the indexer cold-seeds by +/// replaying `PoolCreated` from `deploy_block`, then live-indexes. Shared by +/// the Uniswap V3 and Balancer V2 configs. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Deserialize)] #[serde(rename_all = "kebab-case", deny_unknown_fields)] pub struct FactoryConfig { pub address: Address, - /// Block the factory was deployed at; on-chain cold-seed scans from here + /// Block the factory was deployed at; on-chain cold-seed scans from here. pub deploy_block: u64, } +/// Uniswap V3 discovery config: the factories whose `PoolCreated` events the +/// indexer scans. Non-empty and unique, enforced at parse. +#[derive(Debug, Deserialize)] +#[serde(rename_all = "kebab-case", deny_unknown_fields)] +pub struct UniswapV3Config { + #[serde(deserialize_with = "serde_ext::deserialize_nonempty_unique_vec")] + pub factories: Vec, + /// Blocks per `eth_getLogs` chunk. Small: the indexer fetches every pool's + /// events chain-wide, so a wide range blows past RPC log-response caps. + #[serde(default = "default_uniswap_v3_chunk_size")] + pub chunk_size: u64, +} + +/// Balancer V2 indexer config. Pools are created by per-type factories and +/// registered with a single Vault; the pool type is implied by which group a +/// factory is listed under. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "kebab-case", deny_unknown_fields)] +pub struct BalancerV2Config { + pub vault: Address, + /// Blocks per `eth_getLogs` chunk. Large: `PoolCreated` is factory-filtered + /// (few logs), so the cold-seed scan stays a handful of calls. + #[serde(default = "default_balancer_v2_chunk_size")] + pub chunk_size: u64, + #[serde(default)] + pub weighted: Vec, + #[serde(default)] + pub weighted_v3plus: Vec, + #[serde(default)] + pub stable: Vec, + #[serde(default)] + pub liquidity_bootstrapping: Vec, + #[serde(default)] + pub composable_stable: Vec, +} + +impl BalancerV2Config { + /// All configured factories, across every pool type. + fn factories(&self) -> impl Iterator { + self.weighted + .iter() + .chain(&self.weighted_v3plus) + .chain(&self.stable) + .chain(&self.liquidity_bootstrapping) + .chain(&self.composable_stable) + } + + /// Total factories configured across all pool types. + pub fn factory_count(&self) -> usize { + self.factories().count() + } +} + /// Subset of [`NetworkConfig`] handed to [`UniswapV3Indexer`] at runtime. #[derive(Debug, Clone)] pub struct IndexerConfig { @@ -190,6 +287,7 @@ impl Configuration { let content = std::fs::read_to_string(path) .with_context(|| format!("reading config file {}", path.display()))?; let parsed: Self = toml::from_str(&content).context("parsing config file")?; + parsed.network.validate()?; Ok(parsed) } } diff --git a/crates/pool-indexer/src/run.rs b/crates/pool-indexer/src/run.rs index 71e173b071..44d85d59f8 100644 --- a/crates/pool-indexer/src/run.rs +++ b/crates/pool-indexer/src/run.rs @@ -47,17 +47,19 @@ pub async fn bootstrap(config: Configuration) { // Seed every factory concurrently, like the serve path. let mut factory_set = JoinSet::new(); - for factory in network.factories.iter().copied() { - let indexer = UniswapV3Indexer::new( - provider.clone(), - db.clone(), - &network.indexer_config(factory.address), - ); - let db = db.clone(); - let network = network.clone(); - factory_set.spawn(async move { - bootstrap_factory(&db, &indexer, &network, &factory).await; - }); + if let Some(uniswap_v3) = &network.uniswap_v3 { + for factory in uniswap_v3.factories.iter().copied() { + let indexer = UniswapV3Indexer::new( + provider.clone(), + db.clone(), + &network.indexer_config(uniswap_v3.chunk_size, factory.address), + ); + let db = db.clone(); + let network = network.clone(); + factory_set.spawn(async move { + bootstrap_factory(&db, &indexer, &network, &factory).await; + }); + } } while let Some(result) = factory_set.join_next().await { result.expect("bootstrap task panicked"); @@ -72,7 +74,11 @@ pub async fn run(config: Configuration) { let startup = Arc::new(Some(AtomicBool::new(false))); let barrier = Arc::new(StartupBarrier::new( startup.clone(), - config.network.factories.len(), + config + .network + .uniswap_v3 + .as_ref() + .map_or(0, |u| u.factories.len()), )); // Abort the metrics task when `run` exits, so tests can rebind the port. @@ -153,7 +159,12 @@ fn build_api_state(db: &PgPool, network: &NetworkConfig) -> Arc { Arc::new(AppState { db: db.clone(), network: network.name.clone(), - factories: network.factories.iter().map(|f| f.address).collect(), + factories: network + .uniswap_v3 + .iter() + .flat_map(|u| &u.factories) + .map(|f| f.address) + .collect(), }) } @@ -161,51 +172,52 @@ async fn run_network_indexer(db: PgPool, network: NetworkConfig, barrier: Arc