node, store: Do not fail startup when genesis block for a chain failed

Instead of crashing the node when the genesis block for a chain changed,
log an error and do not start a block ingestor for that chain.

Author: lutter

node/src/main.rs

@@ -661,17 +661,25 @@ fn start_block_ingestor(
     eth_networks
         .networks
         .iter()
-        .for_each(|(network_name, eth_adapters)| {
+        .filter_map(|(network_name, eth_adapters)| {
+            let chain_store = block_store
+                .chain_store(network_name)
+                .expect("network with name");
+            if chain_store.is_ingestible() {
+                Some((network_name, eth_adapters, chain_store))
+            } else {
+                None
+            }
+        })
+        .for_each(|(network_name, eth_adapters, chain_store)| {
             info!(
                 logger,
                 "Starting block ingestor for network";
                 "network_name" => &network_name
             );
             let eth_adapter = eth_adapters.cheapest().unwrap(); //Safe to unwrap since it cannot be empty
             let block_ingestor = BlockIngestor::new(
-                block_store
-                    .chain_store(network_name)
-                    .expect("network with name"),
+                chain_store,
                 eth_adapter.clone(),
                 *ANCESTOR_COUNT,
                 network_name.to_string(),

store/postgres/src/block_store.rs

@@ -23,6 +23,14 @@ use crate::{subgraph_store::PRIMARY_SHARD, Shard};
 #[cfg(debug_assertions)]
 pub const FAKE_NETWORK_SHARED: &str = "fake_network_shared";
 
+/// The status of a chain: whether we can only read from the chain, or
+/// whether it is ok to ingest from it, too
+#[derive(Copy, Clone)]
+pub enum ChainStatus {
+    ReadOnly,
+    Ingestible,
+}
+
 mod primary {
     use std::str::FromStr;
 
@@ -214,57 +222,77 @@ impl BlockStore {
             }
         }
 
+        /// Check that the configuration for `chain` hasn't changed so that
+        /// it is ok to ingest from it
+        fn chain_ingestible(
+            logger: &Logger,
+            chain: &primary::Chain,
+            shard: &Shard,
+            ident: &Option<EthereumNetworkIdentifier>,
+        ) -> bool {
+            if &chain.shard != shard {
+                error!(
+                    logger,
+                    "the chain {} is stored in shard {} but is configured for shard {}",
+                    chain.name,
+                    chain.shard,
+                    shard
+                );
+                return false;
+            }
+            if let Some(ident) = ident {
+                if chain.net_version != ident.net_version {
+                    error!(logger,
+                        "the net version for chain {} has changed from {} to {} since the last time we ran",
+                        chain.name,
+                        chain.net_version,
+                        ident.net_version
+                    );
+                    return false;
+                }
+                if &chain.genesis_block != &format!("{:x}", ident.genesis_block_hash) {
+                    error!(logger,
+                        "the genesis block hash for chain {} has changed from {} to {:x} since the last time we ran",
+                        chain.name,
+                        chain.genesis_block,
+                        ident.genesis_block_hash
+                    );
+                    return false;
+                }
+            }
+            return true;
+        }
+
         // For each configured chain, add a chain store
         for (chain_name, idents, shard) in chains {
             let ident = reduce_idents(&chain_name, idents)?;
-            let (chain, create) = match (
+            match (
                 existing_chains
                     .iter()
                     .find(|chain| chain.name == chain_name),
                 ident,
             ) {
                 (Some(chain), ident) => {
-                    if chain.shard != shard {
-                        return Err(StoreError::Unknown(anyhow!(
-                            "the chain {} is stored in shard {} but is configured for shard {}",
-                            chain.name,
-                            chain.shard,
-                            shard
-                        )));
-                    }
-                    if let Some(ident) = ident {
-                        if chain.net_version != ident.net_version {
-                            return Err(StoreError::Unknown(anyhow!(
-                                "the net version for chain {} has changed from {} to {} since the last time we ran",
-                                chain.name,
-                                chain.net_version,
-                                ident.net_version
-                            )));
-                        }
-                        if &chain.genesis_block != &format!("{:x}", ident.genesis_block_hash) {
-                            return Err(StoreError::Unknown(anyhow!(
-                                "the genesis block hash for chain {} has changed from {} to {:x} since the last time we ran",
-                                chain.name,
-                                chain.genesis_block,
-                                ident.genesis_block_hash
-                            )));
-                        }
-                    }
-                    (chain.clone(), false)
+                    let status = if chain_ingestible(&block_store.logger, chain, &shard, &ident) {
+                        ChainStatus::Ingestible
+                    } else {
+                        ChainStatus::ReadOnly
+                    };
+                    block_store.add_chain_store(&chain, status, false)?;
+                }
+                (None, Some(ident)) => {
+                    let chain =
+                        primary::add_chain(&block_store.primary, &chain_name, &ident, &shard)?;
+                    block_store.add_chain_store(&chain, ChainStatus::Ingestible, true)?;
                 }
-                (None, Some(ident)) => (
-                    primary::add_chain(&block_store.primary, &chain_name, &ident, &shard)?,
-                    true,
-                ),
                 (None, None) => {
-                    return Err(StoreError::Unknown(anyhow!(
+                    error!(
+                        &block_store.logger,
                         " the chain {} is new but we could not get a network identifier for it",
                         chain_name
-                    )));
+                    );
                 }
             };
-
-            block_store.add_chain_store(&chain, create)?;
         }
 
         // There might be chains we have in the database that are not yet/
@@ -280,14 +308,15 @@ impl BlockStore {
             .iter()
             .filter(|chain| !configured_chains.contains(&chain.name))
         {
-            block_store.add_chain_store(&chain, false)?;
+            block_store.add_chain_store(&chain, ChainStatus::ReadOnly, false)?;
         }
         Ok(block_store)
     }
 
     fn add_chain_store(
         &self,
         chain: &primary::Chain,
+        status: ChainStatus,
         create: bool,
     ) -> Result<Arc<ChainStore>, StoreError> {
         let pool = self
@@ -301,6 +330,7 @@ impl BlockStore {
             chain.name.clone(),
             chain.storage.clone(),
             &ident,
+            status,
             self.chain_head_update_listener.clone(),
             sender,
             pool,
@@ -343,7 +373,7 @@ impl BlockStore {
         // of the configured chains
         let conn = self.primary.get()?;
         primary::find_chain(&conn, chain)?
-            .map(|chain| self.add_chain_store(&chain, false))
+            .map(|chain| self.add_chain_store(&chain, ChainStatus::ReadOnly, false))
             .transpose()
     }
 

store/postgres/src/chain_store.rs

@@ -21,6 +21,7 @@ use graph::prelude::{
 };
 
 use crate::{
+    block_store::ChainStatus,
     chain_head_listener::{ChainHeadUpdateListener, ChainHeadUpdateSender},
     connection_pool::ConnectionPool,
 };
@@ -1060,6 +1061,7 @@ pub struct ChainStore {
     pub chain: String,
     storage: data::Storage,
     genesis_block_ptr: EthereumBlockPointer,
+    status: ChainStatus,
     chain_head_update_listener: Arc<ChainHeadUpdateListener>,
     chain_head_update_sender: ChainHeadUpdateSender,
 }
@@ -1069,6 +1071,7 @@ impl ChainStore {
         chain: String,
         storage: data::Storage,
         net_identifier: &EthereumNetworkIdentifier,
+        status: ChainStatus,
         chain_head_update_listener: Arc<ChainHeadUpdateListener>,
         chain_head_update_sender: ChainHeadUpdateSender,
         pool: ConnectionPool,
@@ -1078,13 +1081,18 @@ impl ChainStore {
             chain,
             storage,
             genesis_block_ptr: (net_identifier.genesis_block_hash, 0 as u64).into(),
+            status,
             chain_head_update_listener,
             chain_head_update_sender,
         };
 
         store
     }
 
+    pub fn is_ingestible(&self) -> bool {
+        matches!(self.status, ChainStatus::Ingestible)
+    }
+
     fn get_conn(&self) -> Result<PooledConnection<ConnectionManager<PgConnection>>, Error> {
         self.conn.get().map_err(Error::from)
     }