timeout when trying to get net_identifiers at startup (#5568)

* timeout when trying to get net_identifiers at startup

* fix genesis validation

Author: mangas

graph/src/components/adapter.rs

@@ -290,7 +290,6 @@ impl<T: NetIdentifiable + Clone + 'static> ProviderManager<T> {
             .unwrap_or_default()
     }
 
-    #[cfg(debug_assertions)]
     pub async fn mark_all_valid(&self) {
         for (_, status) in self.inner.status.iter() {
             let mut s = status.write().await;

graph/src/firehose/endpoints.rs

@@ -121,7 +121,7 @@ impl GenesisDecoder for SubstreamsGenesisDecoder {
                     start_cursor: "".to_string(),
                     stop_block_num: 0,
                     final_blocks_only: true,
-                    production_mode: false,
+                    production_mode: true,
                     output_module: "map_blocks".to_string(),
                     modules: package.modules,
                     debug_initial_store_snapshot_for_modules: vec![],

node/src/bin/manager.rs

@@ -1009,7 +1009,7 @@ impl Context {
         let logger = self.logger.clone();
         let registry = self.metrics_registry();
         let metrics = Arc::new(EndpointMetrics::mock());
-        Networks::from_config(logger, &self.config, registry, metrics, block_store).await
+        Networks::from_config(logger, &self.config, registry, metrics, block_store, false).await
     }
 
     fn chain_store(self, chain_name: &str) -> anyhow::Result<Arc<ChainStore>> {

node/src/chain.rs

@@ -32,6 +32,7 @@ use graph::log::factory::LoggerFactory;
 use graph::prelude::anyhow;
 use graph::prelude::MetricsRegistry;
 use graph::slog::{debug, error, info, o, warn, Logger};
+use graph::tokio::time::timeout;
 use graph::url::Url;
 use graph::util::security::SafeDisplay;
 use graph_chain_ethereum::{self as ethereum, Transport};
@@ -432,13 +433,17 @@ pub async fn networks_as_chains(
         let chain_store = match store.chain_store(chain_id) {
             Some(c) => c,
             None => {
-                let ident = match networks.chain_identifier(&logger, chain_id).await {
-                    Ok(ident) => ident,
-                    Err(err) if !config.genesis_validation_enabled => {
-                        warn!(&logger, "unable to fetch genesis for {}. Err: {}.falling back to the default value because validation is disabled", chain_id, err);
+                let ident = match timeout(
+                    config.genesis_validation_timeout,
+                    networks.chain_identifier(&logger, chain_id),
+                )
+                .await
+                {
+                    Ok(Ok(ident)) => ident,
+                    err => {
+                        warn!(&logger, "unable to fetch genesis for {}. Err: {:?}.falling back to the default value", chain_id, err);
                         ChainIdentifier::default()
                     }
-                    err => err.expect("must be able to get chain identity to create a store"),
                 };
                 store
                     .create_chain_store(chain_id, ident)
@@ -708,10 +713,16 @@ mod test {
         let metrics_registry = Arc::new(MetricsRegistry::mock());
         let ident_validator = Arc::new(NoopIdentValidator);
 
-        let networks =
-            Networks::from_config(logger, &config, metrics_registry, metrics, ident_validator)
-                .await
-                .expect("can parse config");
+        let networks = Networks::from_config(
+            logger,
+            &config,
+            metrics_registry,
+            metrics,
+            ident_validator,
+            false,
+        )
+        .await
+        .expect("can parse config");
         let mut network_names = networks
             .adapters
             .iter()

node/src/main.rs

@@ -271,6 +271,7 @@ async fn main() {
             metrics_registry.cheap_clone(),
             endpoint_metrics,
             validator,
+            env_vars.genesis_validation_enabled,
         )
         .await
         .expect("unable to parse network configuration");

node/src/manager/commands/config.rs

@@ -177,6 +177,7 @@ pub async fn provider(
         registry,
         metrics,
         Arc::new(NoopIdentValidator),
+        false,
     )
     .await?;
     let network: ChainId = network.into();

node/src/manager/commands/run.rs

@@ -100,6 +100,7 @@ pub async fn run(
         metrics_registry.cheap_clone(),
         endpoint_metrics,
         ident_validator,
+        env_vars.genesis_validation_enabled,
     )
     .await
     .expect("unable to parse network configuration");

node/src/network_setup.rs

@@ -239,6 +239,7 @@ impl Networks {
         registry: Arc<MetricsRegistry>,
         endpoint_metrics: Arc<EndpointMetrics>,
         store: Arc<dyn IdentValidator>,
+        genesis_validation_enabled: bool,
     ) -> Result<Networks> {
         if config.query_only(&config.node) {
             return Ok(Networks::noop());
@@ -264,13 +265,19 @@ impl Networks {
             .chain(substreams.into_iter())
             .collect();
 
-        Ok(Networks::new(&logger, adapters, store))
+        Ok(Networks::new(
+            &logger,
+            adapters,
+            store,
+            genesis_validation_enabled,
+        ))
     }
 
     fn new(
         logger: &Logger,
         adapters: Vec<AdapterConfiguration>,
         validator: Arc<dyn IdentValidator>,
+        genesis_validation_enabled: bool,
     ) -> Self {
         let adapters2 = adapters.clone();
         let eth_adapters = adapters.iter().flat_map(|a| a.as_rpc()).cloned().map(
@@ -316,7 +323,7 @@ impl Networks {
             )
             .collect_vec();
 
-        Self {
+        let s = Self {
             adapters: adapters2,
             rpc_provider_manager: ProviderManager::new(
                 logger.clone(),
@@ -337,7 +344,22 @@ impl Networks {
                     .map(|(chain_id, endpoints)| (chain_id, endpoints)),
                 validator.cheap_clone(),
             ),
+        };
+
+        if !genesis_validation_enabled {
+            let (r, f, s) = (
+                s.rpc_provider_manager.clone(),
+                s.firehose_provider_manager.clone(),
+                s.substreams_provider_manager.clone(),
+            );
+            graph::spawn(async move {
+                r.mark_all_valid().await;
+                f.mark_all_valid().await;
+                s.mark_all_valid().await;
+            });
         }
+
+        s
     }
 
     pub async fn block_ingestors(