all: Split store functions for chain id mgmt out of ChainStore

Author: lutter

graph/src/blockchain/mock.rs

@@ -2,8 +2,9 @@ use crate::{
     bail,
     components::{
         link_resolver::LinkResolver,
+        network_provider::ChainName,
         store::{
-            BlockNumber, ChainHeadStore, DeploymentCursorTracker, DeploymentLocator,
+            BlockNumber, ChainHeadStore, ChainIdStore, DeploymentCursorTracker, DeploymentLocator,
             SourceableStore,
         },
         subgraph::InstanceDSTemplateInfo,
@@ -569,10 +570,20 @@ impl ChainStore for MockChainStore {
     fn chain_identifier(&self) -> Result<ChainIdentifier, Error> {
         unimplemented!()
     }
-    fn set_chain_identifier(&self, _ident: &ChainIdentifier) -> Result<(), Error> {
-        unimplemented!()
-    }
     fn as_head_store(self: Arc<Self>) -> Arc<dyn ChainHeadStore> {
         self.clone()
     }
 }
+
+impl ChainIdStore for MockChainStore {
+    fn chain_identifier(&self, _name: &ChainName) -> Result<ChainIdentifier, Error> {
+        unimplemented!()
+    }
+    fn set_chain_identifier(
+        &self,
+        _name: &ChainName,
+        _ident: &ChainIdentifier,
+    ) -> Result<(), Error> {
+        unimplemented!()
+    }
+}

graph/src/components/network_provider/chain_identifier_validator.rs

@@ -1,11 +1,11 @@
-use anyhow::anyhow;
+use std::sync::Arc;
+
 use thiserror::Error;
 
 use crate::blockchain::BlockHash;
 use crate::blockchain::ChainIdentifier;
 use crate::components::network_provider::ChainName;
-use crate::components::store::BlockStore;
-use crate::components::store::ChainStore;
+use crate::components::store::ChainIdStore;
 
 /// Additional requirements for stores that are necessary for provider checks.
 pub trait ChainIdentifierValidator: Send + Sync + 'static {
@@ -51,24 +51,29 @@ pub enum ChainIdentifierValidationError {
     Store(#[source] anyhow::Error),
 }
 
-impl<C, B> ChainIdentifierValidator for B
-where
-    C: ChainStore,
-    B: BlockStore<ChainStore = C>,
-{
+pub fn chain_id_validator(store: Arc<dyn ChainIdStore>) -> Arc<dyn ChainIdentifierValidator> {
+    Arc::new(ChainIdentifierStore::new(store))
+}
+
+pub(crate) struct ChainIdentifierStore {
+    store: Arc<dyn ChainIdStore>,
+}
+
+impl ChainIdentifierStore {
+    pub fn new(store: Arc<dyn ChainIdStore>) -> Self {
+        Self { store }
+    }
+}
+
+impl ChainIdentifierValidator for ChainIdentifierStore {
     fn validate_identifier(
         &self,
         chain_name: &ChainName,
         chain_identifier: &ChainIdentifier,
     ) -> Result<(), ChainIdentifierValidationError> {
-        let chain_store = self.chain_store(&chain_name).ok_or_else(|| {
-            ChainIdentifierValidationError::Store(anyhow!(
-                "unable to get store for chain '{chain_name}'"
-            ))
-        })?;
-
-        let store_identifier = chain_store
-            .chain_identifier()
+        let store_identifier = self
+            .store
+            .chain_identifier(chain_name)
             .map_err(|err| ChainIdentifierValidationError::Store(err))?;
 
         if store_identifier.is_default() {
@@ -108,14 +113,8 @@ where
         chain_name: &ChainName,
         chain_identifier: &ChainIdentifier,
     ) -> Result<(), ChainIdentifierValidationError> {
-        let chain_store = self.chain_store(&chain_name).ok_or_else(|| {
-            ChainIdentifierValidationError::Store(anyhow!(
-                "unable to get store for chain '{chain_name}'"
-            ))
-        })?;
-
-        chain_store
-            .set_chain_identifier(chain_identifier)
+        self.store
+            .set_chain_identifier(chain_name, chain_identifier)
             .map_err(|err| ChainIdentifierValidationError::Store(err))
     }
 }

graph/src/components/network_provider/genesis_hash_check.rs

@@ -6,13 +6,15 @@ use slog::error;
 use slog::warn;
 use slog::Logger;
 
+use crate::components::network_provider::chain_id_validator;
 use crate::components::network_provider::ChainIdentifierValidationError;
 use crate::components::network_provider::ChainIdentifierValidator;
 use crate::components::network_provider::ChainName;
 use crate::components::network_provider::NetworkDetails;
 use crate::components::network_provider::ProviderCheck;
 use crate::components::network_provider::ProviderCheckStatus;
 use crate::components::network_provider::ProviderName;
+use crate::components::store::ChainIdStore;
 
 /// Requires providers to have the same network version and genesis hash as one
 /// previously stored in the database.
@@ -26,6 +28,12 @@ impl GenesisHashCheck {
             chain_identifier_store,
         }
     }
+
+    pub fn from_id_store(id_store: Arc<dyn ChainIdStore>) -> Self {
+        Self {
+            chain_identifier_store: chain_id_validator(id_store),
+        }
+    }
 }
 
 #[async_trait]

graph/src/components/network_provider/mod.rs

@@ -5,6 +5,7 @@ mod network_details;
 mod provider_check;
 mod provider_manager;
 
+pub use self::chain_identifier_validator::chain_id_validator;
 pub use self::chain_identifier_validator::ChainIdentifierValidationError;
 pub use self::chain_identifier_validator::ChainIdentifierValidator;
 pub use self::extended_blocks_check::ExtendedBlocksCheck;

graph/src/components/store/traits.rs

@@ -9,6 +9,7 @@ use super::*;
 use crate::blockchain::block_stream::{EntitySourceOperation, FirehoseCursor};
 use crate::blockchain::{BlockTime, ChainIdentifier, ExtendedBlockPtr};
 use crate::components::metrics::stopwatch::StopwatchMetrics;
+use crate::components::network_provider::ChainName;
 use crate::components::server::index_node::VersionInfo;
 use crate::components::subgraph::SubgraphVersionSwitchingMode;
 use crate::components::transaction_receipt;
@@ -443,7 +444,7 @@ pub trait QueryStoreManager: Send + Sync + 'static {
     ) -> Result<Arc<dyn QueryStore + Send + Sync>, QueryExecutionError>;
 }
 
-pub trait BlockStore: Send + Sync + 'static {
+pub trait BlockStore: ChainIdStore + Send + Sync + 'static {
     type ChainStore: ChainStore;
 
     fn create_chain_store(
@@ -481,6 +482,19 @@ pub trait ChainHeadStore: Send + Sync {
     ) -> Result<(), Error>;
 }
 
+#[async_trait]
+pub trait ChainIdStore: Send + Sync + 'static {
+    /// Return the chain identifier for this store.
+    fn chain_identifier(&self, chain_name: &ChainName) -> Result<ChainIdentifier, Error>;
+
+    /// Update the chain identifier for this store.
+    fn set_chain_identifier(
+        &self,
+        chain_name: &ChainName,
+        ident: &ChainIdentifier,
+    ) -> Result<(), Error>;
+}
+
 /// Common trait for blockchain store implementations.
 #[async_trait]
 pub trait ChainStore: ChainHeadStore {
@@ -590,9 +604,6 @@ pub trait ChainStore: ChainHeadStore {
     /// Return the chain identifier for this store.
     fn chain_identifier(&self) -> Result<ChainIdentifier, Error>;
 
-    /// Update the chain identifier for this store.
-    fn set_chain_identifier(&self, ident: &ChainIdentifier) -> Result<(), Error>;
-
     /// Workaround for Rust issue #65991 that keeps us from using an
     /// `Arc<dyn ChainStore>` as an `Arc<dyn ChainHeadStore>`
     fn as_head_store(self: Arc<Self>) -> Arc<dyn ChainHeadStore>;

node/src/main.rs

@@ -290,7 +290,7 @@ async fn main_inner() {
         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(
+            provider_checks.push(Arc::new(network_provider::GenesisHashCheck::from_id_store(
                 block_store.clone(),
             )));
         }

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::network_provider::ChainIdentifierValidator;
 use graph::components::network_provider::ChainName;
+use graph::components::store::ChainIdStore;
 use graph::components::store::StoreError;
 use graph::prelude::BlockNumber;
 use graph::prelude::ChainStore as _;
@@ -164,7 +164,7 @@ pub fn remove(primary: ConnectionPool, store: Arc<BlockStore>, name: String) ->
 pub async fn update_chain_genesis(
     networks: &Networks,
     coord: Arc<PoolCoordinator>,
-    store: Arc<BlockStore>,
+    store: Arc<dyn ChainIdStore>,
     logger: &Logger,
     chain_id: ChainName,
     genesis_hash: BlockHash,
@@ -188,18 +188,14 @@ 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_identifier(
+    store.set_chain_identifier(
         &chain_id,
         &ChainIdentifier {
             net_version: ident.net_version.clone(),
             genesis_block_hash: genesis_hash,
         },
     )?;
 
-    // Update the primary public.chains
-    println!("Updating primary public.chains");
-    store.set_chain_identifier(chain_id, &ident)?;
-
     // Refresh the new values
     println!("Refresh mappings");
     crate::manager::commands::database::remap(&coord, None, None, false).await?;

node/src/manager/commands/provider_checks.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 use std::time::Duration;
 
+use graph::components::network_provider::chain_id_validator;
 use graph::components::network_provider::ChainIdentifierValidator;
 use graph::components::network_provider::ChainName;
 use graph::components::network_provider::ExtendedBlocksCheck;
@@ -36,9 +37,10 @@ pub async fn execute(
             .providers_unchecked(chain_name)
             .unique_by(|x| x.provider_name())
         {
+            let validator = chain_id_validator(store.clone());
             match tokio::time::timeout(
                 timeout,
-                run_checks(logger, chain_name, adapter, store.clone()),
+                run_checks(logger, chain_name, adapter, validator.clone()),
             )
             .await
             {
@@ -56,11 +58,9 @@ pub async fn execute(
             .providers_unchecked(chain_name)
             .unique_by(|x| x.provider_name())
         {
-            match tokio::time::timeout(
-                timeout,
-                run_checks(logger, chain_name, adapter, store.clone()),
-            )
-            .await
+            let validator = chain_id_validator(store.clone());
+            match tokio::time::timeout(timeout, run_checks(logger, chain_name, adapter, validator))
+                .await
             {
                 Ok(result) => {
                     errors.extend(result);
@@ -76,9 +76,10 @@ pub async fn execute(
             .providers_unchecked(chain_name)
             .unique_by(|x| x.provider_name())
         {
+            let validator = chain_id_validator(store.clone());
             match tokio::time::timeout(
                 timeout,
-                run_checks(logger, chain_name, adapter, store.clone()),
+                run_checks(logger, chain_name, adapter, validator.clone()),
             )
             .await
             {

node/src/manager/commands/run.rs

@@ -10,7 +10,7 @@ use crate::MetricsContext;
 use graph::anyhow::bail;
 use graph::cheap_clone::CheapClone;
 use graph::components::link_resolver::{ArweaveClient, FileSizeLimit};
-use graph::components::network_provider::ChainIdentifierValidator;
+use graph::components::network_provider::chain_id_validator;
 use graph::components::store::DeploymentLocator;
 use graph::components::subgraph::Settings;
 use graph::endpoint::EndpointMetrics;
@@ -98,8 +98,7 @@ pub async fn run(
         Vec::new();
 
     if env_vars.genesis_validation_enabled {
-        let store: Arc<dyn ChainIdentifierValidator> = network_store.block_store();
-
+        let store = chain_id_validator(network_store.block_store());
         provider_checks.push(Arc::new(
             graph::components::network_provider::GenesisHashCheck::new(store),
         ));

store/postgres/src/block_store.rs

@@ -10,13 +10,16 @@ use diesel::{
     r2d2::{ConnectionManager, PooledConnection},
     sql_query, ExpressionMethods as _, PgConnection, RunQueryDsl,
 };
-use graph::components::network_provider::ChainName;
 use graph::{
     blockchain::ChainIdentifier,
     components::store::{BlockStore as BlockStoreTrait, QueryPermit},
     prelude::{error, info, BlockNumber, BlockPtr, Logger, ENV_VARS},
     slog::o,
 };
+use graph::{
+    components::{network_provider::ChainName, store::ChainIdStore},
+    prelude::ChainStore as _,
+};
 use graph::{internal_error, prelude::CheapClone};
 use graph::{prelude::StoreError, util::timed_cache::TimedCache};
 
@@ -605,3 +608,41 @@ impl BlockStoreTrait for BlockStore {
             .map_err(anyhow::Error::from)
     }
 }
+
+impl ChainIdStore for BlockStore {
+    fn chain_identifier(&self, chain_name: &ChainName) -> Result<ChainIdentifier, anyhow::Error> {
+        let chain_store = self
+            .chain_store(&chain_name)
+            .ok_or_else(|| anyhow!("unable to get store for chain '{chain_name}'"))?;
+
+        chain_store.chain_identifier()
+    }
+
+    fn set_chain_identifier(
+        &self,
+        chain_name: &ChainName,
+        ident: &ChainIdentifier,
+    ) -> Result<(), anyhow::Error> {
+        use primary::chains as c;
+
+        // Update the block shard first since that contains a copy from the primary
+        let chain_store = self
+            .chain_store(&chain_name)
+            .ok_or_else(|| anyhow!("unable to get store for chain '{chain_name}'"))?;
+
+        chain_store.set_chain_identifier(ident)?;
+
+        // Update the master copy in the primary
+        let primary_pool = self.pools.get(&*PRIMARY_SHARD).unwrap();
+        let mut conn = primary_pool.get()?;
+
+        diesel::update(c::table.filter(c::name.eq(chain_name.as_str())))
+            .set((
+                c::genesis_block_hash.eq(ident.genesis_block_hash.hash_hex()),
+                c::net_version.eq(&ident.net_version),
+            ))
+            .execute(&mut conn)?;
+
+        Ok(())
+    }
+}