chain, core, graph : use TriggersAdapterWrapper at top level

incrypto32 · incrypto32 · commit 686e7e19ae5d · 2024-09-12T12:56:00.000+05:30
diff --git a/chain/ethereum/src/chain.rs b/chain/ethereum/src/chain.rs
@@ -133,82 +133,40 @@ impl BlockStreamBuilder<Chain> for EthereumStreamBuilder {
         filter: Arc<TriggerFilterWrapper<Chain>>,
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Chain>>> {
-        let requirements = filter.filter.node_capabilities();
-        let adapter = chain
-            .triggers_adapter(&deployment, &requirements, unified_api_version.clone())
-            .unwrap_or_else(|_| {
-                panic!(
-                    "no adapter for network {} with capabilities {}",
-                    chain.name, requirements
-                )
-            });
-
-        let adapter = Arc::new(TriggersAdapterWrapper::new(adapter, source_subgraph_stores));
-
-        let logger = chain
-            .logger_factory
-            .subgraph_logger(&deployment)
-            .new(o!("component" => "BlockStream"));
-        let chain_store = chain.chain_store();
-        let chain_head_update_stream = chain
-            .chain_head_update_listener
-            .subscribe(chain.name.to_string(), logger.clone());
-
-        // Special case: Detect Celo and set the threshold to 0, so that eth_getLogs is always used.
-        // This is ok because Celo blocks are always final. And we _need_ to do this because
-        // some events appear only in eth_getLogs but not in transaction receipts.
-        // See also ca0edc58-0ec5-4c89-a7dd-2241797f5e50.
-        let chain_id = match chain.chain_client().as_ref() {
-            ChainClient::Rpc(adapter) => {
-                adapter
-                    .cheapest()
-                    .await
-                    .ok_or(anyhow!("unable to get eth adapter for chan_id call"))?
-                    .chain_id()
-                    .await?
-            }
-            _ => panic!("expected rpc when using polling blockstream"),
-        };
-        let reorg_threshold = match CELO_CHAIN_IDS.contains(&chain_id) {
-            false => chain.reorg_threshold,
-            true => 0,
-        };
-
-        Ok(Box::new(PollingBlockStream::new(
-            chain_store,
-            chain_head_update_stream,
-            adapter,
-            chain.node_id.clone(),
-            deployment.hash,
-            filter,
+        self.build_polling(
+            chain,
+            deployment,
             start_blocks,
-            reorg_threshold,
-            logger,
-            ENV_VARS.max_block_range_size,
-            ENV_VARS.target_triggers_per_block_range,
-            unified_api_version,
+            source_subgraph_stores,
             subgraph_current_block,
-        )))
+            filter,
+            unified_api_version,
+        )
+        .await
     }
 
     async fn build_polling(
         &self,
         chain: &Chain,
         deployment: DeploymentLocator,
         start_blocks: Vec<BlockNumber>,
+        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
         subgraph_current_block: Option<BlockPtr>,
         filter: Arc<TriggerFilterWrapper<Chain>>,
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Chain>>> {
         let requirements = filter.filter.node_capabilities();
-        let adapter = chain
-            .triggers_adapter(&deployment, &requirements, unified_api_version.clone())
-            .unwrap_or_else(|_| {
-                panic!(
-                    "no adapter for network {} with capabilities {}",
-                    chain.name, requirements
-                )
-            });
+        let adapter = TriggersAdapterWrapper::new(
+            chain
+                .triggers_adapter(&deployment, &requirements, unified_api_version.clone())
+                .unwrap_or_else(|_| {
+                    panic!(
+                        "no adapter for network {} with capabilities {}",
+                        chain.name, requirements
+                    )
+                }),
+            source_subgraph_stores,
+        );
 
         let logger = chain
             .logger_factory
@@ -242,7 +200,7 @@ impl BlockStreamBuilder<Chain> for EthereumStreamBuilder {
         Ok(Box::new(PollingBlockStream::new(
             chain_store,
             chain_head_update_stream,
-            adapter,
+            Arc::new(adapter),
             chain.node_id.clone(),
             deployment.hash,
             filter,
@@ -507,6 +465,7 @@ impl Blockchain for Chain {
                         self,
                         deployment,
                         start_blocks,
+                        source_subgraph_stores,
                         current_ptr,
                         filter,
                         unified_api_version,
diff --git a/chain/near/src/chain.rs b/chain/near/src/chain.rs
@@ -108,20 +108,6 @@ impl BlockStreamBuilder<Chain> for NearStreamBuilder {
             chain.metrics_registry.clone(),
         )))
     }
-
-    async fn build_subgraph_block_stream(
-        &self,
-        _chain: &Chain,
-        _deployment: DeploymentLocator,
-        _start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
-        _subgraph_current_block: Option<BlockPtr>,
-        _filter: Arc<TriggerFilterWrapper<Chain>>,
-        _unified_api_version: UnifiedMappingApiVersion,
-    ) -> Result<Box<dyn BlockStream<Chain>>> {
-        unimplemented!()
-    }
-
     async fn build_firehose(
         &self,
         chain: &Chain,
@@ -164,6 +150,7 @@ impl BlockStreamBuilder<Chain> for NearStreamBuilder {
         _chain: &Chain,
         _deployment: DeploymentLocator,
         _start_blocks: Vec<BlockNumber>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
         _subgraph_current_block: Option<BlockPtr>,
         _filter: Arc<TriggerFilterWrapper<Chain>>,
         _unified_api_version: UnifiedMappingApiVersion,
diff --git a/chain/starknet/src/chain.rs b/chain/starknet/src/chain.rs
@@ -197,19 +197,6 @@ impl BlockStreamBuilder<Chain> for StarknetStreamBuilder {
         unimplemented!()
     }
 
-    async fn build_subgraph_block_stream(
-        &self,
-        _chain: &Chain,
-        _deployment: DeploymentLocator,
-        _start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
-        _subgraph_current_block: Option<BlockPtr>,
-        _filter: Arc<TriggerFilterWrapper<Chain>>,
-        _unified_api_version: UnifiedMappingApiVersion,
-    ) -> Result<Box<dyn BlockStream<Chain>>> {
-        unimplemented!()
-    }
-
     async fn build_firehose(
         &self,
         chain: &Chain,
@@ -252,6 +239,7 @@ impl BlockStreamBuilder<Chain> for StarknetStreamBuilder {
         _chain: &Chain,
         _deployment: DeploymentLocator,
         _start_blocks: Vec<BlockNumber>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
         _subgraph_current_block: Option<BlockPtr>,
         _filter: Arc<TriggerFilterWrapper<Chain>>,
         _unified_api_version: UnifiedMappingApiVersion,
diff --git a/chain/substreams/src/block_stream.rs b/chain/substreams/src/block_stream.rs
@@ -99,24 +99,12 @@ impl BlockStreamBuilderTrait<Chain> for BlockStreamBuilder {
         unimplemented!()
     }
 
-    async fn build_subgraph_block_stream(
-        &self,
-        _chain: &Chain,
-        _deployment: DeploymentLocator,
-        _start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
-        _subgraph_current_block: Option<BlockPtr>,
-        _filter: Arc<TriggerFilterWrapper<Chain>>,
-        _unified_api_version: UnifiedMappingApiVersion,
-    ) -> Result<Box<dyn BlockStream<Chain>>> {
-        unimplemented!()
-    }
-
     async fn build_polling(
         &self,
         _chain: &Chain,
         _deployment: DeploymentLocator,
         _start_blocks: Vec<BlockNumber>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
         _subgraph_current_block: Option<BlockPtr>,
         _filter: Arc<TriggerFilterWrapper<Chain>>,
         _unified_api_version: UnifiedMappingApiVersion,
diff --git a/core/src/subgraph/inputs.rs b/core/src/subgraph/inputs.rs
@@ -1,5 +1,5 @@
 use graph::{
-    blockchain::{Blockchain, TriggersAdapter},
+    blockchain::{block_stream::TriggersAdapterWrapper, Blockchain},
     components::{
         store::{DeploymentLocator, SubgraphFork, WritableStore},
         subgraph::ProofOfIndexingVersion,
@@ -20,7 +20,7 @@ pub struct IndexingInputs<C: Blockchain> {
     pub stop_block: Option<BlockNumber>,
     pub store: Arc<dyn WritableStore>,
     pub debug_fork: Option<Arc<dyn SubgraphFork>>,
-    pub triggers_adapter: Arc<dyn TriggersAdapter<C>>,
+    pub triggers_adapter: Arc<TriggersAdapterWrapper<C>>,
     pub chain: Arc<C>,
     pub templates: Arc<Vec<DataSourceTemplate<C>>>,
     pub unified_api_version: UnifiedMappingApiVersion,
diff --git a/core/src/subgraph/instance_manager.rs b/core/src/subgraph/instance_manager.rs
@@ -6,7 +6,7 @@ use crate::subgraph::Decoder;
 use std::collections::BTreeSet;
 
 use crate::subgraph::runner::SubgraphRunner;
-use graph::blockchain::block_stream::BlockStreamMetrics;
+use graph::blockchain::block_stream::{BlockStreamMetrics, TriggersAdapterWrapper};
 use graph::blockchain::{Blockchain, BlockchainKind, DataSource, NodeCapabilities};
 use graph::components::metrics::gas::GasMetrics;
 use graph::components::store::WritableStore;
@@ -501,6 +501,11 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
             )
             .await?;
 
+        let triggers_adapter = Arc::new(TriggersAdapterWrapper::new(
+            triggers_adapter,
+            subgraph_data_source_writables.clone(),
+        ));
+
         let inputs = IndexingInputs {
             deployment: deployment.clone(),
             features,
diff --git a/graph/src/blockchain/block_stream.rs b/graph/src/blockchain/block_stream.rs
@@ -145,6 +145,7 @@ pub trait BlockStreamBuilder<C: Blockchain>: Send + Sync {
         chain: &C,
         deployment: DeploymentLocator,
         start_blocks: Vec<BlockNumber>,
+        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
         subgraph_current_block: Option<BlockPtr>,
         filter: Arc<TriggerFilterWrapper<C>>,
         unified_api_version: UnifiedMappingApiVersion,
@@ -159,7 +160,18 @@ pub trait BlockStreamBuilder<C: Blockchain>: Send + Sync {
         subgraph_current_block: Option<BlockPtr>,
         filter: Arc<TriggerFilterWrapper<C>>,
         unified_api_version: UnifiedMappingApiVersion,
-    ) -> Result<Box<dyn BlockStream<C>>>;
+    ) -> Result<Box<dyn BlockStream<C>>> {
+        self.build_polling(
+            chain,
+            deployment,
+            start_blocks,
+            source_subgraph_stores,
+            subgraph_current_block,
+            filter,
+            unified_api_version,
+        )
+        .await
+    }
 }
 
 #[derive(Debug, Clone)]
@@ -316,9 +328,8 @@ impl<C: Blockchain> TriggersAdapterWrapper<C> {
     }
 }
 
-#[async_trait]
-impl<C: Blockchain> TriggersAdapter<C> for TriggersAdapterWrapper<C> {
-    async fn ancestor_block(
+impl<C: Blockchain> TriggersAdapterWrapper<C> {
+    pub async fn ancestor_block(
         &self,
         ptr: BlockPtr,
         offset: BlockNumber,
@@ -328,7 +339,7 @@ impl<C: Blockchain> TriggersAdapter<C> for TriggersAdapterWrapper<C> {
     }
 
     // TODO: Do a proper implementation, this is a complete mock implementation
-    async fn scan_triggers(
+    pub async fn scan_triggers(
         &self,
         from: BlockNumber,
         to: BlockNumber,
@@ -337,7 +348,7 @@ impl<C: Blockchain> TriggersAdapter<C> for TriggersAdapterWrapper<C> {
         self.adapter.scan_triggers(from, to, filter).await
     }
 
-    async fn triggers_in_block(
+    pub async fn triggers_in_block(
         &self,
         logger: &Logger,
         block: C::Block,
@@ -346,15 +357,15 @@ impl<C: Blockchain> TriggersAdapter<C> for TriggersAdapterWrapper<C> {
         self.adapter.triggers_in_block(logger, block, filter).await
     }
 
-    async fn is_on_main_chain(&self, ptr: BlockPtr) -> Result<bool, Error> {
+    pub async fn is_on_main_chain(&self, ptr: BlockPtr) -> Result<bool, Error> {
         self.adapter.is_on_main_chain(ptr).await
     }
 
-    async fn parent_ptr(&self, block: &BlockPtr) -> Result<Option<BlockPtr>, Error> {
+    pub async fn parent_ptr(&self, block: &BlockPtr) -> Result<Option<BlockPtr>, Error> {
         self.adapter.parent_ptr(block).await
     }
 
-    async fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error> {
+    pub async fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error> {
         self.adapter.chain_head_ptr().await
     }
 }
diff --git a/graph/src/blockchain/polling_block_stream.rs b/graph/src/blockchain/polling_block_stream.rs
@@ -9,7 +9,7 @@ use std::time::Duration;
 
 use super::block_stream::{
     BlockStream, BlockStreamError, BlockStreamEvent, BlockWithTriggers, ChainHeadUpdateStream,
-    FirehoseCursor, TriggersAdapter, BUFFERED_BLOCK_STREAM_SIZE,
+    FirehoseCursor, TriggersAdapterWrapper, BUFFERED_BLOCK_STREAM_SIZE,
 };
 use super::{Block, BlockPtr, Blockchain, TriggerFilterWrapper};
 
@@ -79,7 +79,7 @@ where
     C: Blockchain,
 {
     chain_store: Arc<dyn ChainStore>,
-    adapter: Arc<dyn TriggersAdapter<C>>,
+    adapter: Arc<TriggersAdapterWrapper<C>>,
     node_id: NodeId,
     subgraph_id: DeploymentHash,
     // This is not really a block number, but the (unsigned) difference
@@ -146,7 +146,7 @@ where
     pub fn new(
         chain_store: Arc<dyn ChainStore>,
         chain_head_update_stream: ChainHeadUpdateStream,
-        adapter: Arc<dyn TriggersAdapter<C>>,
+        adapter: Arc<TriggersAdapterWrapper<C>>,
         node_id: NodeId,
         subgraph_id: DeploymentHash,
         filter: Arc<TriggerFilterWrapper<C>>,
diff --git a/tests/src/fixture/mod.rs b/tests/src/fixture/mod.rs
