node: add optional extended block checks for providers

Author: isum

graph/src/env/mod.rs

@@ -226,6 +226,14 @@ pub struct EnvVars {
     ///
     /// If not specified, the graphman server will not start.
     pub graphman_server_auth_token: Option<String>,
+
+    /// If enabled for a chain, requires providers to support extended block details.
+    ///
+    /// Providers that do not support extended block details for enabled chains are
+    /// considered invalid and will not be used.
+    ///
+    /// Defaults to an empty list;
+    pub firehose_require_extended_blocks_for_chains: Vec<String>,
 }
 
 impl EnvVars {
@@ -311,6 +319,10 @@ impl EnvVars {
             genesis_validation_enabled: inner.genesis_validation_enabled.0,
             genesis_validation_timeout: Duration::from_secs(inner.genesis_validation_timeout),
             graphman_server_auth_token: inner.graphman_server_auth_token,
+            firehose_require_extended_blocks_for_chains:
+                Self::firehose_require_extended_blocks_for_chains(
+                    inner.firehose_require_extended_blocks_for_chains,
+                ),
         })
     }
 
@@ -335,6 +347,14 @@ impl EnvVars {
     pub fn log_gql_cache_timing(&self) -> bool {
         self.log_query_timing_contains("cache") && self.log_gql_timing()
     }
+
+    fn firehose_require_extended_blocks_for_chains(s: Option<String>) -> Vec<String> {
+        s.unwrap_or_default()
+            .split(",")
+            .map(|x| x.trim().to_string())
+            .filter(|x| !x.is_empty())
+            .collect()
+    }
 }
 
 impl Default for EnvVars {
@@ -462,6 +482,8 @@ struct Inner {
     genesis_validation_timeout: u64,
     #[envconfig(from = "GRAPHMAN_SERVER_AUTH_TOKEN")]
     graphman_server_auth_token: Option<String>,
+    #[envconfig(from = "GRAPH_NODE_FIREHOSE_REQUIRE_EXTENDED_BLOCKS_FOR_CHAINS")]
+    firehose_require_extended_blocks_for_chains: Option<String>,
 }
 
 #[derive(Clone, Debug)]

node/src/bin/manager.rs

@@ -4,7 +4,7 @@ use git_testament::{git_testament, render_testament};
 use graph::bail;
 use graph::blockchain::BlockHash;
 use graph::cheap_clone::CheapClone;
-use graph::components::adapter::ChainId;
+use graph::components::network_provider::ChainName;
 use graph::endpoint::EndpointMetrics;
 use graph::env::ENV_VARS;
 use graph::log::logger_with_levels;
@@ -1006,11 +1006,12 @@ impl Context {
         ))
     }
 
-    async fn networks(&self, block_store: Arc<BlockStore>) -> anyhow::Result<Networks> {
+    async fn networks(&self) -> anyhow::Result<Networks> {
         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, false).await
+
+        Networks::from_config(logger, &self.config, registry, metrics, &[]).await
     }
 
     fn chain_store(self, chain_name: &str) -> anyhow::Result<Arc<ChainStore>> {
@@ -1025,8 +1026,7 @@ impl Context {
         self,
         chain_name: &str,
     ) -> anyhow::Result<(Arc<ChainStore>, Arc<EthereumAdapter>)> {
-        let block_store = self.store().block_store();
-        let networks = self.networks(block_store).await?;
+        let networks = self.networks().await?;
         let chain_store = self.chain_store(chain_name)?;
         let ethereum_adapter = networks
             .ethereum_rpcs(chain_name.into())
@@ -1169,7 +1169,7 @@ async fn main() -> anyhow::Result<()> {
             match cmd {
                 CheckProviders {} => {
                     let store = ctx.store().block_store();
-                    let networks = ctx.networks(store.cheap_clone()).await?;
+                    let networks = ctx.networks().await?;
                     Ok(commands::config::check_provider_genesis(&networks, store).await)
                 }
                 Place { name, network } => {
@@ -1367,8 +1367,8 @@ async fn main() -> anyhow::Result<()> {
                 } => {
                     let store_builder = ctx.store_builder().await;
                     let store = ctx.store().block_store();
-                    let networks = ctx.networks(store.cheap_clone()).await?;
-                    let chain_id = ChainId::from(chain_name);
+                    let networks = ctx.networks().await?;
+                    let chain_id = ChainName::from(chain_name);
                     let block_hash = BlockHash::from_str(&block_hash)?;
                     commands::chain::update_chain_genesis(
                         &networks,

node/src/chain.rs

@@ -15,15 +15,12 @@ use graph::blockchain::{
     ChainIdentifier,
 };
 use graph::cheap_clone::CheapClone;
-use graph::components::adapter::ChainId;
+use graph::components::network_provider::ChainName;
 use graph::components::store::{BlockStore as _, ChainStore};
 use graph::data::store::NodeId;
 use graph::endpoint::EndpointMetrics;
 use graph::env::{EnvVars, ENV_VARS};
-use graph::firehose::{
-    FirehoseEndpoint, FirehoseGenesisDecoder, GenesisDecoder, SubgraphLimit,
-    SubstreamsGenesisDecoder,
-};
+use graph::firehose::{FirehoseEndpoint, SubgraphLimit};
 use graph::futures03::future::try_join_all;
 use graph::itertools::Itertools;
 use graph::log::factory::LoggerFactory;
@@ -63,11 +60,11 @@ pub fn create_substreams_networks(
         config.chains.ingestor,
     );
 
-    let mut networks_by_kind: BTreeMap<(BlockchainKind, ChainId), Vec<Arc<FirehoseEndpoint>>> =
+    let mut networks_by_kind: BTreeMap<(BlockchainKind, ChainName), Vec<Arc<FirehoseEndpoint>>> =
         BTreeMap::new();
 
     for (name, chain) in &config.chains.chains {
-        let name: ChainId = name.as_str().into();
+        let name: ChainName = name.as_str().into();
         for provider in &chain.providers {
             if let ProviderDetails::Substreams(ref firehose) = provider.details {
                 info!(
@@ -93,7 +90,6 @@ pub fn create_substreams_networks(
                         firehose.compression_enabled(),
                         SubgraphLimit::Unlimited,
                         endpoint_metrics.clone(),
-                        Box::new(SubstreamsGenesisDecoder {}),
                     )));
                 }
             }
@@ -124,11 +120,11 @@ pub fn create_firehose_networks(
         config.chains.ingestor,
     );
 
-    let mut networks_by_kind: BTreeMap<(BlockchainKind, ChainId), Vec<Arc<FirehoseEndpoint>>> =
+    let mut networks_by_kind: BTreeMap<(BlockchainKind, ChainName), Vec<Arc<FirehoseEndpoint>>> =
         BTreeMap::new();
 
     for (name, chain) in &config.chains.chains {
-        let name: ChainId = name.as_str().into();
+        let name: ChainName = name.as_str().into();
         for provider in &chain.providers {
             let logger = logger.cheap_clone();
             if let ProviderDetails::Firehose(ref firehose) = provider.details {
@@ -143,27 +139,6 @@ pub fn create_firehose_networks(
                     .entry((chain.protocol, name.clone()))
                     .or_insert_with(Vec::new);
 
-                let decoder: Box<dyn GenesisDecoder> = match chain.protocol {
-                    BlockchainKind::Arweave => {
-                        FirehoseGenesisDecoder::<graph_chain_arweave::Block>::new(logger)
-                    }
-                    BlockchainKind::Ethereum => {
-                        FirehoseGenesisDecoder::<graph_chain_ethereum::codec::Block>::new(logger)
-                    }
-                    BlockchainKind::Near => {
-                        FirehoseGenesisDecoder::<graph_chain_near::HeaderOnlyBlock>::new(logger)
-                    }
-                    BlockchainKind::Cosmos => {
-                        FirehoseGenesisDecoder::<graph_chain_cosmos::Block>::new(logger)
-                    }
-                    BlockchainKind::Substreams => {
-                        unreachable!("Substreams configuration should not be handled here");
-                    }
-                    BlockchainKind::Starknet => {
-                        FirehoseGenesisDecoder::<graph_chain_starknet::Block>::new(logger)
-                    }
-                };
-
                 // Create n FirehoseEndpoints where n is the size of the pool. If a
                 // subgraph limit is defined for this endpoint then each endpoint
                 // instance will have their own subgraph limit.
@@ -182,7 +157,6 @@ pub fn create_firehose_networks(
                         firehose.compression_enabled(),
                         firehose.limit_for(&config.node),
                         endpoint_metrics.cheap_clone(),
-                        decoder.box_clone(),
                     )));
                 }
             }
@@ -384,7 +358,7 @@ pub async fn networks_as_chains(
         async fn add_substreams<C: Blockchain>(
             networks: &Networks,
             config: &Arc<EnvVars>,
-            chain_id: ChainId,
+            chain_id: ChainName,
             blockchain_map: &mut BlockchainMap,
             logger_factory: LoggerFactory,
             chain_store: Arc<dyn ChainStore>,
@@ -608,7 +582,7 @@ pub async fn networks_as_chains(
 mod test {
     use crate::config::{Config, Opt};
     use crate::network_setup::{AdapterConfiguration, Networks};
-    use graph::components::adapter::{ChainId, NoopIdentValidator};
+    use graph::components::network_provider::ChainName;
     use graph::endpoint::EndpointMetrics;
     use graph::log::logger;
     use graph::prelude::{tokio, MetricsRegistry};
@@ -641,23 +615,15 @@ mod test {
         let metrics = Arc::new(EndpointMetrics::mock());
         let config = Config::load(&logger, &opt).expect("can create config");
         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,
-            false,
-        )
-        .await
-        .expect("can parse config");
+        let networks = Networks::from_config(logger, &config, metrics_registry, metrics, &[])
+            .await
+            .expect("can parse config");
         let mut network_names = networks
             .adapters
             .iter()
             .map(|a| a.chain_id())
-            .collect::<Vec<&ChainId>>();
+            .collect::<Vec<&ChainName>>();
         network_names.sort();
 
         let traces = NodeCapabilities {

node/src/config.rs

@@ -1,7 +1,7 @@
 use graph::{
     anyhow::Error,
     blockchain::BlockchainKind,
-    components::adapter::ChainId,
+    components::network_provider::ChainName,
     env::ENV_VARS,
     firehose::{SubgraphLimit, SUBGRAPHS_PER_CONN},
     itertools::Itertools,
@@ -104,7 +104,7 @@ fn validate_name(s: &str) -> Result<()> {
 }
 
 impl Config {
-    pub fn chain_ids(&self) -> Vec<ChainId> {
+    pub fn chain_ids(&self) -> Vec<ChainName> {
         self.chains
             .chains
             .keys()

node/src/main.rs

@@ -1,6 +1,5 @@
 use clap::Parser as _;
 use git_testament::{git_testament, render_testament};
-use graph::components::adapter::{IdentValidator, NoopIdentValidator};
 use graph::futures01::Future as _;
 use graph::futures03::compat::Future01CompatExt;
 use graph::futures03::future::TryFutureExt;
@@ -273,21 +272,36 @@ async fn main() {
     start_graphman_server(opt.graphman_port, graphman_server_config).await;
 
     let launch_services = |logger: Logger, env_vars: Arc<EnvVars>| async move {
+        use graph::components::network_provider;
+
         let block_store = network_store.block_store();
 
-        let validator: Arc<dyn IdentValidator> = if env_vars.genesis_validation_enabled {
-            network_store.block_store()
-        } else {
-            Arc::new(NoopIdentValidator {})
-        };
+        let mut provider_checks: Vec<Arc<dyn network_provider::ProviderCheck>> = Vec::new();
+
+        if env_vars.genesis_validation_enabled {
+            provider_checks.push(Arc::new(network_provider::GenesisHashCheck::new(
+                block_store.clone(),
+            )));
+        }
+
+        if !env_vars
+            .firehose_require_extended_blocks_for_chains
+            .is_empty()
+        {
+            provider_checks.push(Arc::new(network_provider::ExtendedBlocksCheck::new(
+                env_vars
+                    .firehose_require_extended_blocks_for_chains
+                    .iter()
+                    .map(|x| x.as_str().into()),
+            )));
+        }
 
         let network_adapters = Networks::from_config(
             logger.cheap_clone(),
             &config,
             metrics_registry.cheap_clone(),
             endpoint_metrics,
-            validator,
-            env_vars.genesis_validation_enabled,
+            &provider_checks,
         )
         .await
         .expect("unable to parse network configuration");

node/src/manager/commands/chain.rs

@@ -7,8 +7,8 @@ use graph::blockchain::BlockHash;
 use graph::blockchain::BlockPtr;
 use graph::blockchain::ChainIdentifier;
 use graph::cheap_clone::CheapClone;
-use graph::components::adapter::ChainId;
-use graph::components::adapter::IdentValidator;
+use graph::components::network_provider::ChainIdentifierStore;
+use graph::components::network_provider::ChainName;
 use graph::components::store::StoreError;
 use graph::prelude::BlockNumber;
 use graph::prelude::ChainStore as _;
@@ -161,7 +161,7 @@ pub async fn update_chain_genesis(
     coord: Arc<PoolCoordinator>,
     store: Arc<BlockStore>,
     logger: &Logger,
-    chain_id: ChainId,
+    chain_id: ChainName,
     genesis_hash: BlockHash,
     force: bool,
 ) -> Result<(), Error> {
@@ -183,7 +183,7 @@ pub async fn update_chain_genesis(
     // Update the local shard's genesis, whether or not it is the primary.
     // The chains table is replicated from the primary and keeps another genesis hash.
     // To keep those in sync we need to update the primary and then refresh the shard tables.
-    store.update_ident(
+    store.update_identifier(
         &chain_id,
         &ChainIdentifier {
             net_version: ident.net_version.clone(),

node/src/manager/commands/config.rs

@@ -1,11 +1,11 @@
 use std::{collections::BTreeMap, sync::Arc};
 
+use graph::components::network_provider::ChainIdentifierStore;
+use graph::components::network_provider::ChainName;
+use graph::components::network_provider::ProviderName;
 use graph::{
     anyhow::{bail, Context},
-    components::{
-        adapter::{ChainId, IdentValidator, IdentValidatorError, NoopIdentValidator, ProviderName},
-        subgraph::{Setting, Settings},
-    },
+    components::subgraph::{Setting, Settings},
     endpoint::EndpointMetrics,
     env::EnvVars,
     itertools::Itertools,
@@ -30,15 +30,15 @@ pub async fn check_provider_genesis(networks: &Networks, store: Arc<BlockStore>)
         let (_oks, errs): (Vec<_>, Vec<_>) = ids
             .into_iter()
             .map(|(provider, id)| {
-                id.map_err(IdentValidatorError::from)
-                    .and_then(|id| store.check_ident(chain_id, &id))
-                    .map_err(|e| (provider, e))
+                id.and_then(|id| store.validate_identifier(chain_id, &id).map_err(Into::into))
+                    .map_err(|e| (provider, e.to_string()))
             })
             .partition_result();
+
         let errs = errs
             .into_iter()
-            .dedup_by(|e1, e2| e1.eq(e2))
-            .collect::<Vec<(ProviderName, IdentValidatorError)>>();
+            .dedup_by(|e1, e2| e1 == e2)
+            .collect::<Vec<(ProviderName, String)>>();
 
         if errs.is_empty() {
             println!("chain_id: {}: status: OK", chain_id);
@@ -171,16 +171,8 @@ pub async fn provider(
 
     let metrics = Arc::new(EndpointMetrics::mock());
     let caps = caps_from_features(features)?;
-    let networks = Networks::from_config(
-        logger,
-        &config,
-        registry,
-        metrics,
-        Arc::new(NoopIdentValidator),
-        false,
-    )
-    .await?;
-    let network: ChainId = network.into();
+    let networks = Networks::from_config(logger, &config, registry, metrics, &[]).await?;
+    let network: ChainName = network.into();
     let adapters = networks.ethereum_rpcs(network.clone());
 
     let adapters = adapters.all_cheapest_with(&caps).await;