node: Allow connecting to only one chain

'graphman chain ingest' is taking forever since it connects to all chains,
but we only need an adapter for a single chain

Author: lutter

node/src/bin/manager.rs

@@ -1028,7 +1028,19 @@ impl Context {
         self,
         chain_name: &str,
     ) -> anyhow::Result<(Arc<ChainStore>, Arc<EthereumAdapter>)> {
-        let networks = self.networks().await?;
+        let logger = self.logger.clone();
+        let registry = self.metrics_registry();
+        let metrics = Arc::new(EndpointMetrics::mock());
+        let networks = Networks::from_config_for_chain(
+            logger,
+            &self.config,
+            registry,
+            metrics,
+            &[],
+            chain_name,
+        )
+        .await?;
+
         let chain_store = self.chain_store(chain_name)?;
         let ethereum_adapter = networks
             .ethereum_rpcs(chain_name.into())

node/src/chain.rs

@@ -48,10 +48,39 @@ pub enum ProviderNetworkStatus {
     },
 }
 
+pub trait ChainFilter: Send + Sync {
+    fn filter(&self, chain_name: &str) -> bool;
+}
+
+pub struct AnyChainFilter;
+
+impl ChainFilter for AnyChainFilter {
+    fn filter(&self, _: &str) -> bool {
+        true
+    }
+}
+
+pub struct OneChainFilter {
+    chain_name: String,
+}
+
+impl OneChainFilter {
+    pub fn new(chain_name: String) -> Self {
+        Self { chain_name }
+    }
+}
+
+impl ChainFilter for OneChainFilter {
+    fn filter(&self, chain_name: &str) -> bool {
+        self.chain_name == chain_name
+    }
+}
+
 pub fn create_substreams_networks(
     logger: Logger,
     config: &Config,
     endpoint_metrics: Arc<EndpointMetrics>,
+    chain_filter: &dyn ChainFilter,
 ) -> Vec<AdapterConfiguration> {
     debug!(
         logger,
@@ -63,7 +92,13 @@ pub fn create_substreams_networks(
     let mut networks_by_kind: BTreeMap<(BlockchainKind, ChainName), Vec<Arc<FirehoseEndpoint>>> =
         BTreeMap::new();
 
-    for (name, chain) in &config.chains.chains {
+    let filtered_chains = config
+        .chains
+        .chains
+        .iter()
+        .filter(|(name, _)| chain_filter.filter(name));
+
+    for (name, chain) in filtered_chains {
         let name: ChainName = name.as_str().into();
         for provider in &chain.providers {
             if let ProviderDetails::Substreams(ref firehose) = provider.details {
@@ -113,6 +148,7 @@ pub fn create_firehose_networks(
     logger: Logger,
     config: &Config,
     endpoint_metrics: Arc<EndpointMetrics>,
+    chain_filter: &dyn ChainFilter,
 ) -> Vec<AdapterConfiguration> {
     debug!(
         logger,
@@ -124,7 +160,13 @@ pub fn create_firehose_networks(
     let mut networks_by_kind: BTreeMap<(BlockchainKind, ChainName), Vec<Arc<FirehoseEndpoint>>> =
         BTreeMap::new();
 
-    for (name, chain) in &config.chains.chains {
+    let filtered_chains = config
+        .chains
+        .chains
+        .iter()
+        .filter(|(name, _)| chain_filter.filter(name));
+
+    for (name, chain) in filtered_chains {
         let name: ChainName = name.as_str().into();
         for provider in &chain.providers {
             let logger = logger.cheap_clone();
@@ -179,18 +221,20 @@ pub fn create_firehose_networks(
 
 /// Parses all Ethereum connection strings and returns their network names and
 /// `EthereumAdapter`.
-pub async fn create_all_ethereum_networks(
+pub async fn create_ethereum_networks(
     logger: Logger,
     registry: Arc<MetricsRegistry>,
     config: &Config,
     endpoint_metrics: Arc<EndpointMetrics>,
+    chain_filter: &dyn ChainFilter,
 ) -> anyhow::Result<Vec<AdapterConfiguration>> {
     let eth_rpc_metrics = Arc::new(ProviderEthRpcMetrics::new(registry));
     let eth_networks_futures = config
         .chains
         .chains
         .iter()
         .filter(|(_, chain)| chain.protocol == BlockchainKind::Ethereum)
+        .filter(|(name, _)| chain_filter.filter(name))
         .map(|(name, _)| {
             create_ethereum_networks_for_chain(
                 &logger,

node/src/network_setup.rs

@@ -30,8 +30,8 @@ use graph_store_postgres::{BlockStore, ChainHeadUpdateListener};
 use std::{any::Any, cmp::Ordering, sync::Arc, time::Duration};
 
 use crate::chain::{
-    create_all_ethereum_networks, create_firehose_networks, create_substreams_networks,
-    networks_as_chains,
+    create_ethereum_networks, create_firehose_networks, create_substreams_networks,
+    networks_as_chains, AnyChainFilter, ChainFilter, OneChainFilter,
 };
 
 #[derive(Debug, Clone)]
@@ -183,31 +183,38 @@ impl Networks {
             .await
     }
 
-    pub async fn from_config(
+    async fn from_config_inner(
         logger: Logger,
         config: &crate::config::Config,
         registry: Arc<MetricsRegistry>,
         endpoint_metrics: Arc<EndpointMetrics>,
         provider_checks: &[Arc<dyn ProviderCheck>],
+        chain_filter: &dyn ChainFilter,
     ) -> Result<Networks> {
         if config.query_only(&config.node) {
             return Ok(Networks::noop());
         }
 
-        let eth = create_all_ethereum_networks(
+        let eth = create_ethereum_networks(
             logger.cheap_clone(),
             registry,
             &config,
             endpoint_metrics.cheap_clone(),
+            chain_filter,
         )
         .await?;
         let firehose = create_firehose_networks(
             logger.cheap_clone(),
             &config,
             endpoint_metrics.cheap_clone(),
+            chain_filter,
+        );
+        let substreams = create_substreams_networks(
+            logger.cheap_clone(),
+            &config,
+            endpoint_metrics,
+            chain_filter,
         );
-        let substreams =
-            create_substreams_networks(logger.cheap_clone(), &config, endpoint_metrics);
         let adapters: Vec<_> = eth
             .into_iter()
             .chain(firehose.into_iter())
@@ -217,6 +224,44 @@ impl Networks {
         Ok(Networks::new(&logger, adapters, provider_checks))
     }
 
+    pub async fn from_config_for_chain(
+        logger: Logger,
+        config: &crate::config::Config,
+        registry: Arc<MetricsRegistry>,
+        endpoint_metrics: Arc<EndpointMetrics>,
+        provider_checks: &[Arc<dyn ProviderCheck>],
+        chain_name: &str,
+    ) -> Result<Networks> {
+        let filter = OneChainFilter::new(chain_name.to_string());
+        Self::from_config_inner(
+            logger,
+            config,
+            registry,
+            endpoint_metrics,
+            provider_checks,
+            &filter,
+        )
+        .await
+    }
+
+    pub async fn from_config(
+        logger: Logger,
+        config: &crate::config::Config,
+        registry: Arc<MetricsRegistry>,
+        endpoint_metrics: Arc<EndpointMetrics>,
+        provider_checks: &[Arc<dyn ProviderCheck>],
+    ) -> Result<Networks> {
+        Self::from_config_inner(
+            logger,
+            config,
+            registry,
+            endpoint_metrics,
+            provider_checks,
+            &AnyChainFilter,
+        )
+        .await
+    }
+
     fn new(
         logger: &Logger,
         adapters: Vec<AdapterConfiguration>,