chain, graph : move subgraph trigger scanning back to TriggersAdapter…

…Wrapper

chain/arweave/src/chain.rs

@@ -27,11 +27,13 @@ use graph::{
  prelude::{async_trait, o, BlockNumber, ChainStore, Error, Logger, LoggerFactory},
 };
 use prost::Message;
+use std::collections::HashSet;
 use std::sync::Arc;
 
 use crate::adapter::TriggerFilter;
 use crate::data_source::{DataSourceTemplate, UnresolvedDataSourceTemplate};
 use crate::trigger::{self, ArweaveTrigger};
+use crate::Block as ArweaveBlock;
 use crate::{
  codec,
  data_source::{DataSource, UnresolvedDataSource},
@@ -138,7 +140,7 @@ impl Blockchain for Chain {
 
  let firehose_mapper = Arc::new(FirehoseMapper {
  adapter,
- filter: filter.filter.clone(),
+ filter: filter.chain_filter.clone(),
  });
 
  Ok(Box::new(FirehoseBlockStream::new(
@@ -198,7 +200,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
  &self,
  _from: BlockNumber,
  _to: BlockNumber,
- _filter: &Arc<TriggerFilterWrapper<Chain>>,
+ _filter: &TriggerFilter,
  ) -> Result<(Vec<BlockWithTriggers<Chain>>, BlockNumber), Error> {
  panic!("Should never be called since not used by FirehoseBlockStream")
  }
@@ -266,6 +268,14 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
  number: block.number.saturating_sub(1),
  }))
  }
+
+ async fn load_blocks_by_numbers(
+ &self,
+ _logger: Logger,
+ _block_numbers: HashSet<BlockNumber>,
+ ) -> Result<Vec<ArweaveBlock>, Error> {
+ todo!()
+ }
 }
 
 pub struct FirehoseMapper {

chain/cosmos/src/chain.rs

@@ -4,6 +4,7 @@ use graph::components::adapter::ChainId;
 use graph::env::EnvVars;
 use graph::prelude::{DeploymentHash, MetricsRegistry};
 use graph::substreams::Clock;
+use std::collections::HashSet;
 use std::convert::TryFrom;
 use std::sync::Arc;
 
@@ -33,7 +34,7 @@ use crate::data_source::{
  DataSource, DataSourceTemplate, EventOrigin, UnresolvedDataSource, UnresolvedDataSourceTemplate,
 };
 use crate::trigger::CosmosTrigger;
-use crate::{codec, TriggerFilter};
+use crate::{codec, Block, TriggerFilter};
 
 pub struct Chain {
  logger_factory: LoggerFactory,
@@ -132,7 +133,7 @@ impl Blockchain for Chain {
 
  let firehose_mapper = Arc::new(FirehoseMapper {
  adapter,
- filter: filter.filter.clone(),
+ filter: filter.chain_filter.clone(),
  });
 
  Ok(Box::new(FirehoseBlockStream::new(
@@ -197,6 +198,14 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
  panic!("Should never be called since not used by FirehoseBlockStream")
  }
 
+ async fn load_blocks_by_numbers(
+ &self,
+ _logger: Logger,
+ _block_numbers: HashSet<BlockNumber>,
+ ) -> Result<Vec<Block>, Error> {
+ unimplemented!()
+ }
+
  async fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error> {
  unimplemented!()
  }
@@ -205,7 +214,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
  &self,
  _from: BlockNumber,
  _to: BlockNumber,
- _filter: &Arc<TriggerFilterWrapper<Chain>>,
+ _filter: &TriggerFilter,
  ) -> Result<(Vec<BlockWithTriggers<Chain>>, BlockNumber), Error> {
  panic!("Should never be called since not used by FirehoseBlockStream")
  }

chain/ethereum/src/adapter.rs

@@ -1111,9 +1111,9 @@ pub trait EthereumAdapter: Send + Sync + 'static {
 
  async fn load_blocks_by_numbers(
  &self,
- logger: Logger,
- chain_store: Arc<dyn ChainStore>,
- block_numbers: HashSet<BlockNumber>,
+ _logger: Logger,
+ _chain_store: Arc<dyn ChainStore>,
+ _block_numbers: HashSet<BlockNumber>,
  ) -> Box<dyn Stream<Item = Arc<LightEthereumBlock>, Error = Error> + Send>;
 
  /// Load Ethereum blocks in bulk, returning results as they come back as a Stream.

chain/ethereum/src/chain.rs

@@ -155,7 +155,7 @@ 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 requirements = filter.chain_filter.node_capabilities();
  let adapter = TriggersAdapterWrapper::new(
  chain
  .triggers_adapter(&deployment, &requirements, unified_api_version.clone())
@@ -480,7 +480,7 @@ impl Blockchain for Chain {
  store.firehose_cursor(),
  start_blocks,
  current_ptr,
- filter.filter.clone(),
+ filter.chain_filter.clone(),
  unified_api_version,
  )
  .await
@@ -717,7 +717,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
  &self,
  from: BlockNumber,
  to: BlockNumber,
- filter: &Arc<TriggerFilterWrapper<Chain>>,
+ filter: &TriggerFilter,
  ) -> Result<(Vec<BlockWithTriggers<Chain>>, BlockNumber), Error> {
  blocks_with_triggers(
  self.chain_client
@@ -735,6 +735,30 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
  .await
  }
 
+ async fn load_blocks_by_numbers(
+ &self,
+ logger: Logger,
+ block_numbers: HashSet<BlockNumber>,
+ ) -> Result<Vec<BlockFinality>> {
+ use graph::futures01::stream::Stream;
+
+ let adapter = self
+ .chain_client
+ .rpc()?
+ .cheapest_with(&self.capabilities)
+ .await?;
+
+ let blocks = adapter
+ .load_blocks_by_numbers(logger, self.chain_store.clone(), block_numbers)
+ .await
+ .map(|block| BlockFinality::Final(block))
+ .collect()
+ .compat()
+ .await?;
+
+ Ok(blocks)
+ }
+
  async fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error> {
  let chain_store = self.chain_store.clone();
  chain_store.chain_head_ptr().await
@@ -771,7 +795,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
  self.ethrpc_metrics.clone(),
  block_number,
  block_number,
- &Arc::new(TriggerFilterWrapper::<Chain>::new(filter.clone(), vec![])), // TODO(krishna): This is temporary until we take TriggerFilterWrapper as param in triggers_in_block
+ filter,
  self.unified_api_version.clone(),
  )
  .await?;

chain/ethereum/src/ethereum_adapter.rs

@@ -2,14 +2,12 @@ use futures03::{future::BoxFuture, stream::FuturesUnordered};
 use graph::blockchain::client::ChainClient;
 use graph::blockchain::BlockHash;
 use graph::blockchain::ChainIdentifier;
-use graph::blockchain::SubgraphFilter;
-use graph::blockchain::TriggerFilterWrapper;
+
 use graph::components::transaction_receipt::LightTransactionReceipt;
 use graph::data::store::ethereum::call;
 use graph::data::store::scalar;
 use graph::data::subgraph::UnifiedMappingApiVersion;
 use graph::data::subgraph::API_VERSION_0_0_7;
-use graph::data_source::subgraph;
 use graph::futures01::stream;
 use graph::futures01::Future;
 use graph::futures01::Stream;
@@ -21,10 +19,6 @@ use graph::prelude::ethabi::ParamType;
 use graph::prelude::ethabi::Token;
 use graph::prelude::tokio::try_join;
 use graph::prelude::web3::types::U256;
-use graph::prelude::DeploymentHash;
-use graph::prelude::Entity;
-use graph::prelude::Value;
-use graph::schema::InputSchema;
 use graph::slog::o;
 use graph::tokio::sync::RwLock;
 use graph::tokio::time::timeout;
@@ -65,6 +59,7 @@ use crate::chain::BlockFinality;
 use crate::trigger::LogRef;
 use crate::Chain;
 use crate::NodeCapabilities;
+use crate::TriggerFilter;
 use crate::{
  adapter::{
  ContractCall, ContractCallError, EthGetLogsFilter, EthereumAdapter as EthereumAdapterTrait,
@@ -1729,81 +1724,6 @@ impl EthereumAdapterTrait for EthereumAdapter {
  }
 }
 
-// TODO(krishna): Currently this is a mock implementation of subgraph triggers.
-// This will be replaced with the actual implementation which will use the filters to
-// query the database of the source subgraph and return the entity triggers.
-async fn subgraph_triggers(
- adapter: Arc<EthereumAdapter>,
- logger: Logger,
- chain_store: Arc<dyn ChainStore>,
- _subgraph_metrics: Arc<SubgraphEthRpcMetrics>,
- from: BlockNumber,
- to: BlockNumber,
- filter: &Arc<TriggerFilterWrapper<Chain>>,
- _unified_api_version: UnifiedMappingApiVersion,
-) -> Result<(Vec<BlockWithTriggers<crate::Chain>>, BlockNumber), Error> {
- let logger2 = logger.cheap_clone();
- let eth = adapter.clone();
- let to_ptr = eth.next_existing_ptr_to_number(&logger, to).await?;
- let to = to_ptr.block_number();
-
- let first_filter = filter.subgraph_filter.first().unwrap();
-
- let blocks = adapter
- .load_blocks_by_numbers(
- logger.cheap_clone(),
- chain_store.clone(),
- HashSet::from_iter(from..=to),
- )
- .await
- .and_then(move |block| {
- Ok(BlockWithTriggers::<Chain>::new_with_subgraph_triggers(
- BlockFinality::Final(block.clone()),
- vec![create_mock_subgraph_trigger(first_filter, &block)],
- &logger2,
- ))
- })
- .collect()
- .compat()
- .await?;
-
- Ok((blocks, to))
-}
-
-fn create_mock_subgraph_trigger(
- filter: &SubgraphFilter,
- block: &LightEthereumBlock,
-) -> subgraph::TriggerData {
- let mock_entity = create_mock_entity(block);
- subgraph::TriggerData {
- source: filter.subgraph.clone(),
- entity: mock_entity,
- entity_type: filter.entities.first().unwrap().clone(),
- }
-}
-
-fn create_mock_entity(block: &LightEthereumBlock) -> Entity {
- let id = DeploymentHash::new("test").unwrap();
- let data_schema = InputSchema::parse_latest(
- "type Block @entity { id: Bytes!, number: BigInt!, hash: Bytes! }",
- id.clone(),
- )
- .unwrap();
- let hash = Value::Bytes(scalar::Bytes::from(block.hash.unwrap().as_bytes().to_vec()));
- let data = data_schema
- .make_entity(vec![
- ("id".into(), hash.clone()),
- (
- "number".into(),
- Value::BigInt(scalar::BigInt::from(block.number())),
- ),
- ("hash".into(), hash),
- ])
- .unwrap();
-
- data
-}
-
 /// Returns blocks with triggers, corresponding to the specified range and filters; and the resolved
 /// `to` block, which is the nearest non-null block greater than or equal to the passed `to` block.
 /// If a block contains no triggers, there may be no corresponding item in the stream.
@@ -1825,33 +1745,13 @@ pub(crate) async fn blocks_with_triggers(
  subgraph_metrics: Arc<SubgraphEthRpcMetrics>,
  from: BlockNumber,
  to: BlockNumber,
- filter: &Arc<TriggerFilterWrapper<Chain>>,
+ filter: &TriggerFilter,
  unified_api_version: UnifiedMappingApiVersion,
 ) -> Result<(Vec<BlockWithTriggers<crate::Chain>>, BlockNumber), Error> {
  // Each trigger filter needs to be queried for the same block range
  // and the blocks yielded need to be deduped. If any error occurs
  // while searching for a trigger type, the entire operation fails.
  let eth = adapter.clone();
- let subgraph_filter = filter.subgraph_filter.clone();
-
- // TODO(krishna): In the initial implementation we do not allow any other datasource type
- // When using subgraph data sources, there if subgraph_filter is not empty, we can return
- // by just processing the subgraph triggers.
- if !subgraph_filter.is_empty() {
- return subgraph_triggers(
- adapter.clone(),
- logger.clone(),
- chain_store.clone(),
- subgraph_metrics.clone(),
- from,
- to,
- filter,
- unified_api_version,
- )
- .await;
- }
-
- let filter = filter.filter.clone();
  let call_filter = EthereumCallFilter::from(&filter.block);
 
  // Scan the block range to find relevant triggers

chain/near/src/chain.rs

@@ -32,10 +32,12 @@ use graph::{
  prelude::{async_trait, o, BlockNumber, ChainStore, Error, Logger, LoggerFactory},
 };
 use prost::Message;
+use std::collections::HashSet;
 use std::sync::Arc;
 
 use crate::adapter::TriggerFilter;
 use crate::codec::substreams_triggers::BlockAndReceipts;
+use crate::codec::Block;
 use crate::data_source::{DataSourceTemplate, UnresolvedDataSourceTemplate};
 use crate::trigger::{self, NearTrigger};
 use crate::{
@@ -243,7 +245,7 @@ impl Blockchain for Chain {
  deployment,
  store.firehose_cursor(),
  store.block_ptr(),
- filter.filter.clone(),
+ filter.chain_filter.clone(),
  )
  .await;
  }
@@ -255,7 +257,7 @@ impl Blockchain for Chain {
  store.firehose_cursor(),
  start_blocks,
  store.block_ptr(),
- filter.filter.clone(),
+ filter.chain_filter.clone(),
  unified_api_version,
  )
  .await
@@ -318,11 +320,19 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
  &self,
  _from: BlockNumber,
  _to: BlockNumber,
- _filter: &Arc<TriggerFilterWrapper<Chain>>,
+ _filter: &TriggerFilter,
  ) -> Result<(Vec<BlockWithTriggers<Chain>>, BlockNumber), Error> {
  panic!("Should never be called since not used by FirehoseBlockStream")
  }
 
+ async fn load_blocks_by_numbers(
+ &self,
+ _logger: Logger,
+ _block_numbers: HashSet<BlockNumber>,
+ ) -> Result<Vec<Block>> {
+ unimplemented!()
+ }
+
  async fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error> {
  unimplemented!()
  }