node, graph, chains: BlockchainBuilder to simplify instantiating chains (#4408)

neysofu · web-flow · commit 54d4974c0551 · 2023-03-06T12:21:16.000+01:00
* node, graph, chains: BuildableBlockchain abstraction

* chains: replace Chain::new with Chain::build

* chains, node, graph: BlockchainBuilder pattern
diff --git a/chain/arweave/src/chain.rs b/chain/arweave/src/chain.rs
@@ -1,8 +1,10 @@
 use graph::blockchain::client::ChainClient;
-use graph::blockchain::{Block, BlockchainKind, EmptyNodeCapabilities};
+use graph::blockchain::{
+    BasicBlockchainBuilder, Block, BlockchainBuilder, BlockchainKind, EmptyNodeCapabilities,
+};
 use graph::cheap_clone::CheapClone;
 use graph::data::subgraph::UnifiedMappingApiVersion;
-use graph::firehose::{FirehoseEndpoint, FirehoseEndpoints};
+use graph::firehose::FirehoseEndpoint;
 use graph::prelude::MetricsRegistry;
 use graph::{
     blockchain::{
@@ -44,20 +46,14 @@ impl std::fmt::Debug for Chain {
     }
 }
 
-impl Chain {
-    pub fn new(
-        logger_factory: LoggerFactory,
-        name: String,
-        chain_store: Arc<dyn ChainStore>,
-        firehose_endpoints: FirehoseEndpoints,
-        metrics_registry: Arc<dyn MetricsRegistry>,
-    ) -> Self {
+impl BlockchainBuilder<Chain> for BasicBlockchainBuilder {
+    fn build(self) -> Chain {
         Chain {
-            logger_factory,
-            name,
-            client: Arc::new(ChainClient::<Self>::new_firehose(firehose_endpoints)),
-            chain_store,
-            metrics_registry,
+            logger_factory: self.logger_factory,
+            name: self.name,
+            client: Arc::new(ChainClient::<Chain>::new_firehose(self.firehose_endpoints)),
+            chain_store: self.chain_store,
+            metrics_registry: self.metrics_registry,
         }
     }
 }
diff --git a/chain/cosmos/src/chain.rs b/chain/cosmos/src/chain.rs
@@ -2,6 +2,7 @@ use std::sync::Arc;
 
 use graph::blockchain::block_stream::FirehoseCursor;
 use graph::blockchain::client::ChainClient;
+use graph::blockchain::{BasicBlockchainBuilder, BlockchainBuilder};
 use graph::cheap_clone::CheapClone;
 use graph::data::subgraph::UnifiedMappingApiVersion;
 use graph::prelude::MetricsRegistry;
@@ -16,7 +17,7 @@ use graph::{
         IngestorError, RuntimeAdapter as RuntimeAdapterTrait,
     },
     components::store::DeploymentLocator,
-    firehose::{self, FirehoseEndpoint, FirehoseEndpoints, ForkStep},
+    firehose::{self, FirehoseEndpoint, ForkStep},
     prelude::{async_trait, o, BlockNumber, ChainStore, Error, Logger, LoggerFactory},
 };
 use prost::Message;
@@ -42,20 +43,14 @@ impl std::fmt::Debug for Chain {
     }
 }
 
-impl Chain {
-    pub fn new(
-        logger_factory: LoggerFactory,
-        name: String,
-        chain_store: Arc<dyn ChainStore>,
-        firehose_endpoints: FirehoseEndpoints,
-        metrics_registry: Arc<dyn MetricsRegistry>,
-    ) -> Self {
+impl BlockchainBuilder<Chain> for BasicBlockchainBuilder {
+    fn build(self) -> Chain {
         Chain {
-            logger_factory,
-            name,
-            client: Arc::new(ChainClient::new_firehose(firehose_endpoints)),
-            chain_store,
-            metrics_registry,
+            logger_factory: self.logger_factory,
+            name: self.name,
+            client: Arc::new(ChainClient::new_firehose(self.firehose_endpoints)),
+            chain_store: self.chain_store,
+            metrics_registry: self.metrics_registry,
         }
     }
 }
diff --git a/chain/near/src/chain.rs b/chain/near/src/chain.rs
@@ -1,8 +1,8 @@
 use graph::blockchain::client::ChainClient;
-use graph::blockchain::BlockchainKind;
+use graph::blockchain::{BasicBlockchainBuilder, BlockchainBuilder, BlockchainKind};
 use graph::cheap_clone::CheapClone;
 use graph::data::subgraph::UnifiedMappingApiVersion;
-use graph::firehose::{FirehoseEndpoint, FirehoseEndpoints};
+use graph::firehose::FirehoseEndpoint;
 use graph::prelude::{MetricsRegistry, TryFutureExt};
 use graph::{
     anyhow::Result,
@@ -105,22 +105,15 @@ impl std::fmt::Debug for Chain {
     }
 }
 
-impl Chain {
-    pub fn new(
-        logger_factory: LoggerFactory,
-        name: String,
-        chain_store: Arc<dyn ChainStore>,
-        firehose_endpoints: FirehoseEndpoints,
-        metrics_registry: Arc<dyn MetricsRegistry>,
-        block_stream_builder: Arc<dyn BlockStreamBuilder<Self>>,
-    ) -> Self {
+impl BlockchainBuilder<Chain> for BasicBlockchainBuilder {
+    fn build(self) -> Chain {
         Chain {
-            logger_factory,
-            name,
-            client: Arc::new(ChainClient::new_firehose(firehose_endpoints)),
-            chain_store,
-            metrics_registry,
-            block_stream_builder,
+            logger_factory: self.logger_factory,
+            name: self.name,
+            chain_store: self.chain_store,
+            client: Arc::new(ChainClient::new_firehose(self.firehose_endpoints)),
+            metrics_registry: self.metrics_registry,
+            block_stream_builder: Arc::new(NearStreamBuilder {}),
         }
     }
 }
diff --git a/graph/src/blockchain/builder.rs b/graph/src/blockchain/builder.rs
@@ -0,0 +1,25 @@
+use super::Blockchain;
+use crate::{
+    components::{metrics::MetricsRegistry, store::ChainStore},
+    firehose::FirehoseEndpoints,
+    prelude::LoggerFactory,
+};
+use std::sync::Arc;
+
+/// An implementor of [`BlockchainBuilder`] for chains that don't require
+/// particularly fancy builder logic.
+pub struct BasicBlockchainBuilder {
+    pub logger_factory: LoggerFactory,
+    pub name: String,
+    pub chain_store: Arc<dyn ChainStore>,
+    pub firehose_endpoints: FirehoseEndpoints,
+    pub metrics_registry: Arc<dyn MetricsRegistry>,
+}
+
+/// Something that can build a [`Blockchain`].
+pub trait BlockchainBuilder<C>
+where
+    C: Blockchain,
+{
+    fn build(self) -> C;
+}
diff --git a/graph/src/blockchain/mod.rs b/graph/src/blockchain/mod.rs
@@ -3,6 +3,7 @@
 //! trait which is the centerpiece of this module.
 
 pub mod block_stream;
+mod builder;
 pub mod client;
 mod empty_node_capabilities;
 pub mod firehose_block_ingestor;
@@ -43,6 +44,7 @@ use std::{
 use web3::types::H256;
 
 pub use block_stream::{ChainHeadUpdateListener, ChainHeadUpdateStream, TriggersAdapter};
+pub use builder::{BasicBlockchainBuilder, BlockchainBuilder};
 pub use empty_node_capabilities::EmptyNodeCapabilities;
 pub use types::{BlockHash, BlockPtr, ChainIdentifier};
 
diff --git a/node/src/main.rs b/node/src/main.rs
@@ -7,7 +7,10 @@ use ethereum::{
 use git_testament::{git_testament, render_testament};
 use graph::blockchain::client::ChainClient;
 use graph::blockchain::firehose_block_ingestor::{FirehoseBlockIngestor, Transforms};
-use graph::blockchain::{Block as BlockchainBlock, Blockchain, BlockchainKind, BlockchainMap};
+use graph::blockchain::{
+    BasicBlockchainBuilder, Block as BlockchainBlock, Blockchain, BlockchainBuilder,
+    BlockchainKind, BlockchainMap,
+};
 use graph::components::store::BlockStore;
 use graph::data::graphql::effort::LoadManager;
 use graph::env::EnvVars;
@@ -40,7 +43,6 @@ use graph_server_json_rpc::JsonRpcServer;
 use graph_server_metrics::PrometheusMetricsServer;
 use graph_server_websocket::SubscriptionServer as GraphQLSubscriptionServer;
 use graph_store_postgres::{register_jobs as register_store_jobs, ChainHeadUpdateListener, Store};
-use near::NearStreamBuilder;
 use std::collections::BTreeMap;
 use std::io::{BufRead, BufReader};
 use std::path::Path;
@@ -297,7 +299,7 @@ async fn main() {
 
         let network_store = store_builder.network_store(network_identifiers);
 
-        let arweave_chains = arweave_networks_as_chains(
+        let arweave_chains = networks_as_chains::<arweave::Chain>(
             &mut blockchain_map,
             &logger,
             &arweave_networks,
@@ -320,7 +322,7 @@ async fn main() {
             metrics_registry.clone(),
         );
 
-        let near_chains = near_networks_as_chains(
+        let near_chains = networks_as_chains::<near::Chain>(
             &mut blockchain_map,
             &logger,
             &near_networks,
@@ -329,7 +331,7 @@ async fn main() {
             metrics_registry.clone(),
         );
 
-        let cosmos_chains = cosmos_networks_as_chains(
+        let cosmos_chains = networks_as_chains::<cosmos::Chain>(
             &mut blockchain_map,
             &logger,
             &cosmos_networks,
@@ -613,15 +615,19 @@ async fn main() {
     futures::future::pending::<()>().await;
 }
 
-/// Return the hashmap of Arweave chains and also add them to `blockchain_map`.
-fn arweave_networks_as_chains(
+/// Return the hashmap of chains and also add them to `blockchain_map`.
+fn networks_as_chains<C>(
     blockchain_map: &mut BlockchainMap,
     logger: &Logger,
     firehose_networks: &FirehoseNetworks,
     store: &Store,
     logger_factory: &LoggerFactory,
     metrics_registry: Arc<MetricsRegistry>,
-) -> HashMap<String, FirehoseChain<arweave::Chain>> {
+) -> HashMap<String, FirehoseChain<C>>
+where
+    C: Blockchain,
+    BasicBlockchainBuilder: BlockchainBuilder<C>,
+{
     let chains: Vec<_> = firehose_networks
         .networks
         .iter()
@@ -633,7 +639,9 @@ fn arweave_networks_as_chains(
                 .or_else(|| {
                     error!(
                         logger,
-                        "No store configured for Arweave chain {}; ignoring this chain", chain_id
+                        "No store configured for {} chain {}; ignoring this chain",
+                        C::KIND,
+                        chain_id
                     );
                     None
                 })
@@ -642,21 +650,24 @@ fn arweave_networks_as_chains(
             (
                 chain_id.clone(),
                 FirehoseChain {
-                    chain: Arc::new(arweave::Chain::new(
-                        logger_factory.clone(),
-                        chain_id.clone(),
-                        chain_store,
-                        endpoints.clone(),
-                        metrics_registry.clone(),
-                    )),
+                    chain: Arc::new(
+                        BasicBlockchainBuilder {
+                            logger_factory: logger_factory.clone(),
+                            name: chain_id.clone(),
+                            chain_store,
+                            firehose_endpoints: endpoints.clone(),
+                            metrics_registry: metrics_registry.clone(),
+                        }
+                        .build(),
+                    ),
                     firehose_endpoints: endpoints.clone(),
                 },
             )
         })
         .collect();
 
     for (chain_id, firehose_chain) in chains.iter() {
-        blockchain_map.insert::<arweave::Chain>(chain_id.clone(), firehose_chain.chain.clone())
+        blockchain_map.insert::<C>(chain_id.clone(), firehose_chain.chain.clone())
     }
 
     HashMap::from_iter(chains)
@@ -764,106 +775,6 @@ fn ethereum_networks_as_chains(
     HashMap::from_iter(chains)
 }
 
-fn cosmos_networks_as_chains(
-    blockchain_map: &mut BlockchainMap,
-    logger: &Logger,
-    firehose_networks: &FirehoseNetworks,
-    store: &Store,
-    logger_factory: &LoggerFactory,
-    metrics_registry: Arc<MetricsRegistry>,
-) -> HashMap<String, FirehoseChain<cosmos::Chain>> {
-    let chains: Vec<_> = firehose_networks
-        .networks
-        .iter()
-        .filter_map(|(network_name, firehose_endpoints)| {
-            store
-                .block_store()
-                .chain_store(network_name)
-                .map(|chain_store| (network_name, chain_store, firehose_endpoints))
-                .or_else(|| {
-                    error!(
-                        logger,
-                        "No store configured for Cosmos chain {}; ignoring this chain",
-                        network_name
-                    );
-                    None
-                })
-        })
-        .map(|(network_name, chain_store, firehose_endpoints)| {
-            (
-                network_name.clone(),
-                FirehoseChain {
-                    chain: Arc::new(cosmos::Chain::new(
-                        logger_factory.clone(),
-                        network_name.clone(),
-                        chain_store,
-                        firehose_endpoints.clone(),
-                        metrics_registry.clone(),
-                    )),
-                    firehose_endpoints: firehose_endpoints.clone(),
-                },
-            )
-        })
-        .collect();
-
-    for (network_name, firehose_chain) in chains.iter() {
-        blockchain_map.insert::<cosmos::Chain>(network_name.clone(), firehose_chain.chain.clone())
-    }
-
-    HashMap::from_iter(chains)
-}
-
-/// Return the hashmap of NEAR chains and also add them to `blockchain_map`.
-fn near_networks_as_chains(
-    blockchain_map: &mut BlockchainMap,
-    logger: &Logger,
-    firehose_networks: &FirehoseNetworks,
-    store: &Store,
-    logger_factory: &LoggerFactory,
-    metrics_registry: Arc<MetricsRegistry>,
-) -> HashMap<String, FirehoseChain<near::Chain>> {
-    let chains: Vec<_> = firehose_networks
-        .networks
-        .iter()
-        .filter_map(|(chain_id, endpoints)| {
-            store
-                .block_store()
-                .chain_store(chain_id)
-                .map(|chain_store| (chain_id, chain_store, endpoints))
-                .or_else(|| {
-                    error!(
-                        logger,
-                        "No store configured for NEAR chain {}; ignoring this chain", chain_id
-                    );
-                    None
-                })
-        })
-        .map(|(chain_id, chain_store, endpoints)| {
-            (
-                chain_id.clone(),
-                FirehoseChain {
-                    chain: Arc::new(near::Chain::new(
-                        logger_factory.clone(),
-                        chain_id.clone(),
-                        chain_store,
-                        endpoints.clone(),
-                        metrics_registry.clone(),
-                        Arc::new(NearStreamBuilder {}),
-                    )),
-                    firehose_endpoints: endpoints.clone(),
-                },
-            )
-        })
-        .collect();
-
-    for (chain_id, firehose_chain) in chains.iter() {
-        blockchain_map
-            .insert::<graph_chain_near::Chain>(chain_id.clone(), firehose_chain.chain.clone())
-    }
-
-    HashMap::from_iter(chains)
-}
-
 fn start_block_ingestor(
     logger: &Logger,
     logger_factory: &LoggerFactory,
