chain, core, graph, store: Filter call handlers from failed txns

Author: tilacog

Cargo.lock

@@ -28,6 +28,7 @@ ethabi = { git = "https://github.com/graphprotocol/ethabi.git", branch = "master
 
 # We have a couple custom patches to web3.
 web3 = { git = "https://github.com/graphprotocol/rust-web3", branch = "master" }
+itertools = "0.10.0"
 
 graph-runtime-wasm = { path = "../../runtime/wasm" }
 graph-runtime-derive = { path = "../../runtime/derive" }

chain/ethereum/Cargo.toml

@@ -24,6 +24,7 @@ use graph::{components::ethereum::EthereumNetworkIdentifier, prelude::*};
 use crate::{data_source::DataSource, Chain};
 
 pub type EventSignature = H256;
+pub type FunctionSelector = [u8; 4];
 
 #[derive(Clone, Debug)]
 pub struct EthereumContractCall {
@@ -270,7 +271,8 @@ impl EthereumLogFilter {
 pub(crate) struct EthereumCallFilter {
     // Each call filter has a map of filters keyed by address, each containing a tuple with
     // start_block and the set of function signatures
-    pub contract_addresses_function_signatures: HashMap<Address, (BlockNumber, HashSet<[u8; 4]>)>,
+    pub contract_addresses_function_signatures:
+        HashMap<Address, (BlockNumber, HashSet<FunctionSelector>)>,
 }
 
 impl EthereumCallFilter {
@@ -355,12 +357,12 @@ impl EthereumCallFilter {
     }
 }
 
-impl FromIterator<(BlockNumber, Address, [u8; 4])> for EthereumCallFilter {
+impl FromIterator<(BlockNumber, Address, FunctionSelector)> for EthereumCallFilter {
     fn from_iter<I>(iter: I) -> Self
     where
-        I: IntoIterator<Item = (BlockNumber, Address, [u8; 4])>,
+        I: IntoIterator<Item = (BlockNumber, Address, FunctionSelector)>,
     {
-        let mut lookup: HashMap<Address, (BlockNumber, HashSet<[u8; 4]>)> = HashMap::new();
+        let mut lookup: HashMap<Address, (BlockNumber, HashSet<FunctionSelector>)> = HashMap::new();
         iter.into_iter()
             .for_each(|(start_block, address, function_signature)| {
                 if !lookup.contains_key(&address) {
@@ -387,7 +389,7 @@ impl From<EthereumBlockFilter> for EthereumCallFilter {
                 .contract_addresses
                 .into_iter()
                 .map(|(start_block_opt, address)| (address, (start_block_opt, HashSet::default())))
-                .collect::<HashMap<Address, (BlockNumber, HashSet<[u8; 4]>)>>(),
+                .collect::<HashMap<Address, (BlockNumber, HashSet<FunctionSelector>)>>(),
         }
     }
 }

chain/ethereum/src/adapter.rs

@@ -1,10 +1,8 @@
-use std::collections::HashSet;
-use std::iter::FromIterator;
-use std::sync::Arc;
-
 use anyhow::{Context, Error};
 use graph::data::subgraph::UnifiedMappingApiVersion;
-use graph::prelude::{EthereumCallCache, LightEthereumBlock, LightEthereumBlockExt};
+use graph::prelude::{
+    EthereumCallCache, LightEthereumBlock, LightEthereumBlockExt, StopwatchMetrics,
+};
 use graph::{
     blockchain::{
         block_stream::{
@@ -23,6 +21,9 @@ use graph::{
         LoggerFactory, MetricsRegistry, NodeId, SubgraphStore,
     },
 };
+use std::collections::HashSet;
+use std::iter::FromIterator;
+use std::sync::Arc;
 
 use crate::data_source::DataSourceTemplate;
 use crate::data_source::UnresolvedDataSourceTemplate;
@@ -142,6 +143,7 @@ impl Blockchain for Chain {
         loc: &DeploymentLocator,
         capabilities: &Self::NodeCapabilities,
         unified_api_version: UnifiedMappingApiVersion,
+        stopwatch_metrics: StopwatchMetrics,
     ) -> Result<Arc<Self::TriggersAdapter>, Error> {
         let eth_adapter = self.eth_adapters.cheapest_with(capabilities)?.clone();
         let logger = self
@@ -154,8 +156,9 @@ impl Blockchain for Chain {
             logger,
             ethrpc_metrics,
             eth_adapter,
+            stopwatch_metrics,
             chain_store: self.chain_store.cheap_clone(),
-            _unified_api_version: unified_api_version,
+            unified_api_version,
         };
         Ok(Arc::new(adapter))
     }
@@ -182,7 +185,12 @@ impl Blockchain for Chain {
         let requirements = filter.node_capabilities();
 
         let triggers_adapter = self
-            .triggers_adapter(&deployment, &requirements, unified_api_version.clone())
+            .triggers_adapter(
+                &deployment,
+                &requirements,
+                unified_api_version.clone(),
+                metrics.stopwatch.clone(),
+            )
             .expect(&format!(
                 "no adapter for network {} with capabilities {}",
                 self.name, requirements
@@ -330,9 +338,10 @@ impl Manifest<Chain> for DummyManifest {
 pub struct TriggersAdapter {
     logger: Logger,
     ethrpc_metrics: Arc<SubgraphEthRpcMetrics>,
+    stopwatch_metrics: StopwatchMetrics,
     chain_store: Arc<dyn ChainStore>,
     eth_adapter: Arc<EthereumAdapter>,
-    _unified_api_version: UnifiedMappingApiVersion,
+    unified_api_version: UnifiedMappingApiVersion,
 }
 
 #[async_trait]
@@ -348,9 +357,11 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
             self.logger.clone(),
             self.chain_store.clone(),
             self.ethrpc_metrics.clone(),
+            self.stopwatch_metrics.clone(),
             from,
             to,
             filter,
+            self.unified_api_version.clone(),
         )
         .await
     }
@@ -378,9 +389,11 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
                     logger.clone(),
                     self.chain_store.clone(),
                     self.ethrpc_metrics.clone(),
+                    self.stopwatch_metrics.clone(),
                     block_number,
                     block_number,
                     filter,
+                    self.unified_api_version.clone(),
                 )
                 .await?;
                 assert!(blocks.len() == 1);
@@ -392,7 +405,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
                     &filter.log,
                     &full_block.ethereum_block,
                 ));
-                triggers.append(&mut parse_call_triggers(&filter.call, &full_block));
+                triggers.append(&mut parse_call_triggers(&filter.call, &full_block)?);
                 triggers.append(&mut parse_block_triggers(filter.block.clone(), &full_block));
                 Ok(BlockWithTriggers::new(block, triggers))
             }