Add Balancer V2 config to pool-indexer

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)),

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,56 +87,148 @@ 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<FactoryConfig>,
-    /// 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<UniswapV3Config>,
+    /// Balancer V2 pools to index. Set on its own or alongside `uniswap_v3`.
+    #[serde(default)]
+    pub balancer_v2: Option<BalancerV2Config>,
 }
 
 impl NetworkConfig {
     pub fn poll_interval(&self) -> Duration {
         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<FactoryConfig>,
+    /// 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<FactoryConfig>,
+    #[serde(default)]
+    pub weighted_v3plus: Vec<FactoryConfig>,
+    #[serde(default)]
+    pub stable: Vec<FactoryConfig>,
+    #[serde(default)]
+    pub liquidity_bootstrapping: Vec<FactoryConfig>,
+    #[serde(default)]
+    pub composable_stable: Vec<FactoryConfig>,
+}
+
+impl BalancerV2Config {
+    /// All configured factories, across every pool type.
+    fn factories(&self) -> impl Iterator<Item = &FactoryConfig> {
+        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)
     }
 }

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,59 +159,65 @@ fn build_api_state(db: &PgPool, network: &NetworkConfig) -> Arc<AppState> {
     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(),
     })
 }
 
 async fn run_network_indexer(db: PgPool, network: NetworkConfig, barrier: Arc<StartupBarrier>) {
     tracing::info!(
         network = %network.name,
         chain_id = network.chain_id,
-        factories = network.factories.len(),
         "starting network indexer",
     );
 
     let provider = build_provider_checked(&network).await;
     let network = Arc::new(network);
 
-    // One task per factory. Provider + DB pool are shared; checkpoints are
-    // per-factory because they're keyed by `contract_address`.
     let mut factory_set = JoinSet::new();
-    for factory in network.factories.iter().copied() {
-        let indexer = UniswapV3Indexer::new(
+    if let Some(uniswap_v3) = &network.uniswap_v3 {
+        // One task per factory. Provider + DB pool are shared; checkpoints are
+        // per-factory because they're keyed by `contract_address`.
+        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),
+            );
+            factory_set.spawn(run_factory_indexer(
+                db.clone(),
+                indexer,
+                network.clone(),
+                factory,
+                barrier.clone(),
+            ));
+        }
+
+        // The symbol/decimals backfill scans every token missing the field, so
+        // one pair per process is enough (not per-factory). Spawned into the
+        // same JoinSet so a panic crashes the process via the same supervisor.
+        let backfill_concurrency = network.prefetch_concurrency;
+        let backfill_interval = network.poll_interval();
+        factory_set.spawn(crate::indexer::uniswap_v3::backfill_symbols(
             provider.clone(),
             db.clone(),
-            &network.indexer_config(factory.address),
-        );
-        factory_set.spawn(run_factory_indexer(
+            network.name.clone(),
+            backfill_concurrency,
+            backfill_interval,
+        ));
+        factory_set.spawn(crate::indexer::uniswap_v3::backfill_decimals(
+            provider.clone(),
             db.clone(),
-            indexer,
-            network.clone(),
-            factory,
-            barrier.clone(),
+            network.name.clone(),
+            backfill_concurrency,
+            backfill_interval,
         ));
     }
 
-    // The symbol/decimals backfill scans every token missing the field, so
-    // one pair per process is enough (not per-factory). Spawned into the
-    // same JoinSet so a panic crashes the process via the same supervisor.
-    let backfill_concurrency = network.prefetch_concurrency;
-    let backfill_interval = network.poll_interval();
-    factory_set.spawn(crate::indexer::uniswap_v3::backfill_symbols(
-        provider.clone(),
-        db.clone(),
-        network.name.clone(),
-        backfill_concurrency,
-        backfill_interval,
-    ));
-    factory_set.spawn(crate::indexer::uniswap_v3::backfill_decimals(
-        provider.clone(),
-        db.clone(),
-        network.name.clone(),
-        backfill_concurrency,
-        backfill_interval,
-    ));
-
     // Factory indexers + backfill are all infinite loops; any return is a
     // bug, so crash and let the orchestrator restart the pod.
     if let Some(result) = factory_set.join_next().await {