chain/ethereum,graph,node: moved BlockIngestor to chain/ethereum and removed IngestorAdapter (#3131)

The IngestorAdapter was actually protocol specific and having the abstraction in the Blockchain trait was not used expect for Ethereum chain.

Removed `IngestorAdapter` from block chain and moved the chain agnostic `BlockIngestor` to `graph-chain-ethereum` crate.

Co-authored-by: Matthieu Vachon <matt@streamingfast.io>

Author: Matthieu Vachon

chain/ethereum/src/chain.rs

@@ -14,17 +14,15 @@ use graph::{
         },
         firehose_block_stream::FirehoseBlockStream,
         polling_block_stream::PollingBlockStream,
-        Block, BlockHash, BlockPtr, Blockchain, ChainHeadUpdateListener,
-        IngestorAdapter as IngestorAdapterTrait, IngestorError, TriggerFilter as _,
+        Block, BlockPtr, Blockchain, ChainHeadUpdateListener, IngestorError, TriggerFilter as _,
     },
     cheap_clone::CheapClone,
     components::store::DeploymentLocator,
     firehose,
-    log::factory::{ComponentLoggerConfig, ElasticComponentLoggerConfig},
     prelude::{
-        async_trait, error, lazy_static, o, serde_json as json, web3::types::H256, BlockNumber,
-        ChainStore, EthereumBlockWithCalls, Future01CompatExt, Logger, LoggerFactory,
-        MetricsRegistry, NodeId, SubgraphStore,
+        async_trait, lazy_static, o, serde_json as json, BlockNumber, ChainStore,
+        EthereumBlockWithCalls, Future01CompatExt, Logger, LoggerFactory, MetricsRegistry, NodeId,
+        SubgraphStore,
     },
 };
 use prost::Message;
@@ -72,7 +70,6 @@ pub struct Chain {
     registry: Arc<dyn MetricsRegistry>,
     firehose_endpoints: Arc<FirehoseEndpoints>,
     eth_adapters: Arc<EthereumNetworkAdapters>,
-    ancestor_count: BlockNumber,
     chain_store: Arc<dyn ChainStore>,
     call_cache: Arc<dyn EthereumCallCache>,
     subgraph_store: Arc<dyn SubgraphStore>,
@@ -99,7 +96,6 @@ impl Chain {
         firehose_endpoints: FirehoseEndpoints,
         eth_adapters: EthereumNetworkAdapters,
         chain_head_update_listener: Arc<dyn ChainHeadUpdateListener>,
-        ancestor_count: BlockNumber,
         reorg_threshold: BlockNumber,
         is_ingestible: bool,
     ) -> Self {
@@ -110,7 +106,6 @@ impl Chain {
             registry,
             firehose_endpoints: Arc::new(firehose_endpoints),
             eth_adapters: Arc::new(eth_adapters),
-            ancestor_count,
             chain_store,
             call_cache,
             subgraph_store,
@@ -121,6 +116,12 @@ impl Chain {
     }
 }
 
+impl Chain {
+    pub fn cheapest_adapter(&self) -> Arc<EthereumAdapter> {
+        self.eth_adapters.cheapest().unwrap().clone()
+    }
+}
+
 #[async_trait]
 impl Blockchain for Chain {
     const KIND: BlockchainKind = BlockchainKind::Ethereum;
@@ -145,8 +146,6 @@ impl Blockchain for Chain {
 
     type NodeCapabilities = crate::capabilities::NodeCapabilities;
 
-    type IngestorAdapter = IngestorAdapter;
-
     type RuntimeAdapter = RuntimeAdapter;
 
     fn triggers_adapter(
@@ -299,29 +298,6 @@ impl Blockchain for Chain {
         )))
     }
 
-    fn ingestor_adapter(&self) -> Arc<Self::IngestorAdapter> {
-        let eth_adapter = self.eth_adapters.cheapest().unwrap().clone();
-        let logger = self
-            .logger_factory
-            .component_logger(
-                "BlockIngestor",
-                Some(ComponentLoggerConfig {
-                    elastic: Some(ElasticComponentLoggerConfig {
-                        index: String::from("block-ingestor-logs"),
-                    }),
-                }),
-            )
-            .new(o!("provider" => eth_adapter.provider().to_string()));
-
-        let adapter = IngestorAdapter {
-            eth_adapter,
-            logger,
-            ancestor_count: self.ancestor_count,
-            chain_store: self.chain_store.clone(),
-        };
-        Arc::new(adapter)
-    }
-
     fn chain_store(&self) -> Arc<dyn ChainStore> {
         self.chain_store.clone()
     }
@@ -624,79 +600,3 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
         }
     }
 }
-
-pub struct IngestorAdapter {
-    logger: Logger,
-    ancestor_count: i32,
-    eth_adapter: Arc<EthereumAdapter>,
-    chain_store: Arc<dyn ChainStore>,
-}
-
-#[async_trait]
-impl IngestorAdapterTrait<Chain> for IngestorAdapter {
-    fn logger(&self) -> &Logger {
-        &self.logger
-    }
-
-    fn ancestor_count(&self) -> BlockNumber {
-        self.ancestor_count
-    }
-
-    async fn latest_block(&self) -> Result<BlockPtr, IngestorError> {
-        self.eth_adapter
-            .latest_block_header(&self.logger)
-            .compat()
-            .await
-            .map(|block| block.into())
-    }
-
-    async fn ingest_block(
-        &self,
-        block_hash: &BlockHash,
-    ) -> Result<Option<BlockHash>, IngestorError> {
-        // TODO: H256::from_slice can panic
-        let block_hash = H256::from_slice(block_hash.as_slice());
-
-        // Get the fully populated block
-        let block = self
-            .eth_adapter
-            .block_by_hash(&self.logger, block_hash)
-            .compat()
-            .await?
-            .ok_or_else(|| IngestorError::BlockUnavailable(block_hash))?;
-        let ethereum_block = self
-            .eth_adapter
-            .load_full_block(&self.logger, block)
-            .await?;
-
-        // We need something that implements `Block` to store the block; the
-        // store does not care whether the block is final or not
-        let ethereum_block = BlockFinality::NonFinal(EthereumBlockWithCalls {
-            ethereum_block,
-            calls: None,
-        });
-
-        // Store it in the database and try to advance the chain head pointer
-        self.chain_store
-            .upsert_block(Arc::new(ethereum_block))
-            .await?;
-
-        self.chain_store
-            .cheap_clone()
-            .attempt_chain_head_update(self.ancestor_count)
-            .await
-            .map(|missing| missing.map(|h256| h256.into()))
-            .map_err(|e| {
-                error!(self.logger, "failed to update chain head");
-                IngestorError::Unknown(e)
-            })
-    }
-
-    fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error> {
-        self.chain_store.chain_head_ptr()
-    }
-
-    fn cleanup_cached_blocks(&self) -> Result<Option<(i32, usize)>, Error> {
-        self.chain_store.cleanup_cached_blocks(self.ancestor_count)
-    }
-}

graph/src/blockchain/block_ingestor.rs chain/ethereum/src/ingestor.rsgraph/src/blockchain/block_ingestor.rs renamed to chain/ethereum/src/ingestor.rs

@@ -1,9 +1,13 @@
-use std::{sync::Arc, time::Duration};
-
-use crate::{
-    blockchain::{Blockchain, IngestorAdapter, IngestorError},
-    prelude::{info, lazy_static, tokio, trace, warn, Error, LogCode, Logger},
+use crate::{chain::BlockFinality, EthereumAdapter, EthereumAdapterTrait};
+use graph::{
+    blockchain::{BlockHash, BlockPtr, IngestorError},
+    cheap_clone::CheapClone,
+    prelude::{
+        error, ethabi::ethereum_types::H256, info, lazy_static, tokio, trace, warn, ChainStore,
+        Error, EthereumBlockWithCalls, Future01CompatExt, LogCode, Logger,
+    },
 };
+use std::{sync::Arc, time::Duration};
 
 lazy_static! {
     // graph_node::config disallows setting this in a store with multiple
@@ -14,27 +18,27 @@ lazy_static! {
         .unwrap_or(false);
 }
 
-pub struct BlockIngestor<C>
-where
-    C: Blockchain,
-{
-    adapter: Arc<C::IngestorAdapter>,
+pub struct BlockIngestor {
     logger: Logger,
+    ancestor_count: i32,
+    eth_adapter: Arc<EthereumAdapter>,
+    chain_store: Arc<dyn ChainStore>,
     polling_interval: Duration,
 }
 
-impl<C> BlockIngestor<C>
-where
-    C: Blockchain,
-{
+impl BlockIngestor {
     pub fn new(
-        adapter: Arc<C::IngestorAdapter>,
+        logger: Logger,
+        ancestor_count: i32,
+        eth_adapter: Arc<EthereumAdapter>,
+        chain_store: Arc<dyn ChainStore>,
         polling_interval: Duration,
-    ) -> Result<BlockIngestor<C>, Error> {
-        let logger = adapter.logger().clone();
+    ) -> Result<BlockIngestor, Error> {
         Ok(BlockIngestor {
-            adapter,
             logger,
+            ancestor_count,
+            eth_adapter,
+            chain_store,
             polling_interval,
         })
     }
@@ -73,7 +77,7 @@ where
     }
 
     fn cleanup_cached_blocks(&self) {
-        match self.adapter.cleanup_cached_blocks() {
+        match self.chain_store.cleanup_cached_blocks(self.ancestor_count) {
             Ok(Some((min_block, count))) => {
                 if count > 0 {
                     info!(
@@ -97,12 +101,12 @@ where
         trace!(self.logger, "BlockIngestor::do_poll");
 
         // Get chain head ptr from store
-        let head_block_ptr_opt = self.adapter.chain_head_ptr()?;
+        let head_block_ptr_opt = self.chain_store.chain_head_ptr()?;
 
         // To check if there is a new block or not, fetch only the block header since that's cheaper
         // than the full block. This is worthwhile because most of the time there won't be a new
         // block, as we expect the poll interval to be much shorter than the block time.
-        let latest_block = self.adapter.latest_block().await?;
+        let latest_block = self.latest_block().await?;
 
         // If latest block matches head block in store, nothing needs to be done
         if Some(&latest_block) == head_block_ptr_opt.as_ref() {
@@ -122,7 +126,7 @@ where
                 let latest_number = latest_block.number;
                 let head_number = head_block_ptr.number;
                 let distance = latest_number - head_number;
-                let blocks_needed = (distance).min(self.adapter.ancestor_count());
+                let blocks_needed = (distance).min(self.ancestor_count);
                 let code = if distance >= 15 {
                     LogCode::BlockIngestionLagging
                 } else {
@@ -147,7 +151,7 @@ where
         // Might be a no-op if latest block is one that we have seen.
         // ingest_blocks will return a (potentially incomplete) list of blocks that are
         // missing.
-        let mut missing_block_hash = self.adapter.ingest_block(&latest_block.hash).await?;
+        let mut missing_block_hash = self.ingest_block(&latest_block.hash).await?;
 
         // Repeatedly fetch missing parent blocks, and ingest them.
         // ingest_blocks will continue to tell us about more missing parent
@@ -168,8 +172,58 @@ where
         //   iteration will have at most block number N-1.
         // - Therefore, the loop will iterate at most ancestor_count times.
         while let Some(hash) = missing_block_hash {
-            missing_block_hash = self.adapter.ingest_block(&hash).await?;
+            missing_block_hash = self.ingest_block(&hash).await?;
         }
         Ok(())
     }
+
+    async fn ingest_block(
+        &self,
+        block_hash: &BlockHash,
+    ) -> Result<Option<BlockHash>, IngestorError> {
+        // TODO: H256::from_slice can panic
+        let block_hash = H256::from_slice(block_hash.as_slice());
+
+        // Get the fully populated block
+        let block = self
+            .eth_adapter
+            .block_by_hash(&self.logger, block_hash)
+            .compat()
+            .await?
+            .ok_or_else(|| IngestorError::BlockUnavailable(block_hash))?;
+        let ethereum_block = self
+            .eth_adapter
+            .load_full_block(&self.logger, block)
+            .await?;
+
+        // We need something that implements `Block` to store the block; the
+        // store does not care whether the block is final or not
+        let ethereum_block = BlockFinality::NonFinal(EthereumBlockWithCalls {
+            ethereum_block,
+            calls: None,
+        });
+
+        // Store it in the database and try to advance the chain head pointer
+        self.chain_store
+            .upsert_block(Arc::new(ethereum_block))
+            .await?;
+
+        self.chain_store
+            .cheap_clone()
+            .attempt_chain_head_update(self.ancestor_count)
+            .await
+            .map(|missing| missing.map(|h256| h256.into()))
+            .map_err(|e| {
+                error!(self.logger, "failed to update chain head");
+                IngestorError::Unknown(e)
+            })
+    }
+
+    async fn latest_block(&self) -> Result<BlockPtr, IngestorError> {
+        self.eth_adapter
+            .latest_block_header(&self.logger)
+            .compat()
+            .await
+            .map(|block| block.into())
+    }
 }

chain/ethereum/src/lib.rs

@@ -3,6 +3,7 @@ mod capabilities;
 pub mod codec;
 mod data_source;
 mod ethereum_adapter;
+mod ingestor;
 pub mod runtime;
 mod transport;
 
@@ -26,6 +27,7 @@ pub use crate::adapter::{
 };
 pub use crate::chain::Chain;
 pub use crate::network::EthereumNetworks;
+pub use ingestor::{BlockIngestor, CLEANUP_BLOCKS};
 
 #[cfg(test)]
 mod tests;