[WIP] Subgraph Datasources

chain/arweave/src/chain.rs

@@ -3,15 +3,15 @@ use graph::blockchain::client::ChainClient;
 use graph::blockchain::firehose_block_ingestor::FirehoseBlockIngestor;
 use graph::blockchain::{
     BasicBlockchainBuilder, Block, BlockIngestor, BlockchainBuilder, BlockchainKind,
-    EmptyNodeCapabilities, NoopDecoderHook, NoopRuntimeAdapter,
+    EmptyNodeCapabilities, NoopDecoderHook, NoopRuntimeAdapter, TriggerFilterWrapper,
 };
 use graph::cheap_clone::CheapClone;
 use graph::components::adapter::ChainId;
-use graph::components::store::DeploymentCursorTracker;
+use graph::components::store::{DeploymentCursorTracker, WritableStore};
 use graph::data::subgraph::UnifiedMappingApiVersion;
 use graph::env::EnvVars;
 use graph::firehose::FirehoseEndpoint;
-use graph::prelude::MetricsRegistry;
+use graph::prelude::{DeploymentHash, MetricsRegistry};
 use graph::substreams::Clock;
 use graph::{
     blockchain::{
@@ -27,11 +27,13 @@ use graph::{
     prelude::{async_trait, o, BlockNumber, ChainStore, Error, Logger, LoggerFactory},
 };
 use prost::Message;
+use std::collections::{HashMap, 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},
@@ -119,7 +121,8 @@ impl Blockchain for Chain {
         deployment: DeploymentLocator,
         store: impl DeploymentCursorTracker,
         start_blocks: Vec<BlockNumber>,
-        filter: Arc<Self::TriggerFilter>,
+        _source_subgraph_stores: HashMap<DeploymentHash, Arc<dyn WritableStore>>,
+        filter: Arc<TriggerFilterWrapper<Self>>,
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error> {
         let adapter = self
@@ -135,7 +138,10 @@ impl Blockchain for Chain {
             .subgraph_logger(&deployment)
             .new(o!("component" => "FirehoseBlockStream"));
 
-        let firehose_mapper = Arc::new(FirehoseMapper { adapter, filter });
+        let firehose_mapper = Arc::new(FirehoseMapper {
+            adapter,
+            filter: filter.chain_filter.clone(),
+        });
 
         Ok(Box::new(FirehoseBlockStream::new(
             deployment.hash,
@@ -199,6 +205,10 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
         panic!("Should never be called since not used by FirehoseBlockStream")
     }
 
+    async fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error> {
+        unimplemented!()
+    }
+
     async fn triggers_in_block(
         &self,
         logger: &Logger,
@@ -258,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

@@ -1,17 +1,18 @@
 use graph::blockchain::firehose_block_ingestor::FirehoseBlockIngestor;
-use graph::blockchain::{BlockIngestor, NoopDecoderHook};
+use graph::blockchain::{BlockIngestor, NoopDecoderHook, TriggerFilterWrapper};
 use graph::components::adapter::ChainId;
 use graph::env::EnvVars;
-use graph::prelude::MetricsRegistry;
+use graph::prelude::{DeploymentHash, MetricsRegistry};
 use graph::substreams::Clock;
+use std::collections::{HashMap, HashSet};
 use std::convert::TryFrom;
 use std::sync::Arc;
 
 use graph::blockchain::block_stream::{BlockStreamError, BlockStreamMapper, FirehoseCursor};
 use graph::blockchain::client::ChainClient;
 use graph::blockchain::{BasicBlockchainBuilder, BlockchainBuilder, NoopRuntimeAdapter};
 use graph::cheap_clone::CheapClone;
-use graph::components::store::DeploymentCursorTracker;
+use graph::components::store::{DeploymentCursorTracker, WritableStore};
 use graph::data::subgraph::UnifiedMappingApiVersion;
 use graph::{
     blockchain::{
@@ -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,
@@ -113,7 +114,8 @@ impl Blockchain for Chain {
         deployment: DeploymentLocator,
         store: impl DeploymentCursorTracker,
         start_blocks: Vec<BlockNumber>,
-        filter: Arc<Self::TriggerFilter>,
+        _source_subgraph_stores: HashMap<DeploymentHash, Arc<dyn WritableStore>>,
+        filter: Arc<TriggerFilterWrapper<Self>>,
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error> {
         let adapter = self
@@ -129,7 +131,10 @@ impl Blockchain for Chain {
             .subgraph_logger(&deployment)
             .new(o!("component" => "FirehoseBlockStream"));
 
-        let firehose_mapper = Arc::new(FirehoseMapper { adapter, filter });
+        let firehose_mapper = Arc::new(FirehoseMapper {
+            adapter,
+            filter: filter.chain_filter.clone(),
+        });
 
         Ok(Box::new(FirehoseBlockStream::new(
             deployment.hash,
@@ -193,6 +198,18 @@ 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!()
+    }
+
     async fn scan_triggers(
         &self,
         _from: BlockNumber,
@@ -467,9 +484,12 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
 
 #[cfg(test)]
 mod test {
-    use graph::prelude::{
-        slog::{o, Discard, Logger},
-        tokio,
+    use graph::{
+        blockchain::Trigger,
+        prelude::{
+            slog::{o, Discard, Logger},
+            tokio,
+        },
     };
 
     use super::*;
@@ -600,7 +620,10 @@ mod test {
             // they may not be in the same order
             for trigger in expected_triggers {
                 assert!(
-                    triggers.trigger_data.contains(&trigger),
+                    triggers.trigger_data.iter().any(|t| match t {
+                        Trigger::Chain(t) => t == &trigger,
+                        _ => false,
+                    }),
                     "Expected trigger list to contain {:?}, but it only contains: {:?}",
                     trigger,
                     triggers.trigger_data

chain/ethereum/src/adapter.rs

@@ -1109,6 +1109,13 @@ pub trait EthereumAdapter: Send + Sync + 'static {
         block_hash: H256,
     ) -> Box<dyn Future<Item = LightEthereumBlock, Error = Error> + Send>;
 
+    async fn load_blocks_by_numbers(
+        &self,
+        _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.
     /// May use the `chain_store` as a cache.
     async fn load_blocks(

chain/ethereum/src/chain.rs

@@ -3,15 +3,16 @@ use anyhow::{Context, Error};
 use graph::blockchain::client::ChainClient;
 use graph::blockchain::firehose_block_ingestor::{FirehoseBlockIngestor, Transforms};
 use graph::blockchain::{
-    BlockIngestor, BlockTime, BlockchainKind, ChainIdentifier, TriggersAdapterSelector,
+    BlockIngestor, BlockTime, BlockchainKind, ChainIdentifier, TriggerFilterWrapper,
+    TriggersAdapterSelector,
 };
 use graph::components::adapter::ChainId;
-use graph::components::store::DeploymentCursorTracker;
+use graph::components::store::{DeploymentCursorTracker, WritableStore};
 use graph::data::subgraph::UnifiedMappingApiVersion;
 use graph::firehose::{FirehoseEndpoint, ForkStep};
 use graph::futures03::compat::Future01CompatExt;
 use graph::prelude::{
-    BlockHash, ComponentLoggerConfig, ElasticComponentLoggerConfig, EthereumBlock,
+    BlockHash, ComponentLoggerConfig, DeploymentHash, ElasticComponentLoggerConfig, EthereumBlock,
     EthereumCallCache, LightEthereumBlock, LightEthereumBlockExt, MetricsRegistry,
 };
 use graph::schema::InputSchema;
@@ -36,7 +37,7 @@ use graph::{
     },
 };
 use prost::Message;
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::iter::FromIterator;
 use std::sync::Arc;
 use std::time::Duration;
@@ -61,6 +62,7 @@ use crate::{BufferedCallCache, NodeCapabilities};
 use crate::{EthereumAdapter, RuntimeAdapter};
 use graph::blockchain::block_stream::{
     BlockStream, BlockStreamBuilder, BlockStreamError, BlockStreamMapper, FirehoseCursor,
+    TriggersAdapterWrapper,
 };
 
 /// Celo Mainnet: 42220, Testnet Alfajores: 44787, Testnet Baklava: 62320
@@ -121,24 +123,50 @@ impl BlockStreamBuilder<Chain> for EthereumStreamBuilder {
         unimplemented!()
     }
 
+    async fn build_subgraph_block_stream(
+        &self,
+        chain: &Chain,
+        deployment: DeploymentLocator,
+        start_blocks: Vec<BlockNumber>,
+        source_subgraph_stores: HashMap<DeploymentHash, Arc<dyn WritableStore>>,
+        subgraph_current_block: Option<BlockPtr>,
+        filter: Arc<TriggerFilterWrapper<Chain>>,
+        unified_api_version: UnifiedMappingApiVersion,
+    ) -> Result<Box<dyn BlockStream<Chain>>> {
+        self.build_polling(
+            chain,
+            deployment,
+            start_blocks,
+            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: HashMap<DeploymentHash, Arc<dyn WritableStore>>,
         subgraph_current_block: Option<BlockPtr>,
-        filter: Arc<<Chain as Blockchain>::TriggerFilter>,
+        filter: Arc<TriggerFilterWrapper<Chain>>,
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Chain>>> {
-        let requirements = 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 requirements = filter.chain_filter.node_capabilities();
+        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
@@ -172,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,
@@ -409,17 +437,35 @@ impl Blockchain for Chain {
         deployment: DeploymentLocator,
         store: impl DeploymentCursorTracker,
         start_blocks: Vec<BlockNumber>,
-        filter: Arc<Self::TriggerFilter>,
+        source_subgraph_stores: HashMap<DeploymentHash, Arc<dyn WritableStore>>,
+        filter: Arc<TriggerFilterWrapper<Self>>,
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error> {
         let current_ptr = store.block_ptr();
+
+        if !filter.subgraph_filter.is_empty() {
+            return self
+                .block_stream_builder
+                .build_subgraph_block_stream(
+                    self,
+                    deployment,
+                    start_blocks,
+                    source_subgraph_stores,
+                    current_ptr,
+                    filter,
+                    unified_api_version,
+                )
+                .await;
+        }
+
         match self.chain_client().as_ref() {
             ChainClient::Rpc(_) => {
                 self.block_stream_builder
                     .build_polling(
                         self,
                         deployment,
                         start_blocks,
+                        source_subgraph_stores,
                         current_ptr,
                         filter,
                         unified_api_version,
@@ -434,7 +480,7 @@ impl Blockchain for Chain {
                         store.firehose_cursor(),
                         start_blocks,
                         current_ptr,
-                        filter,
+                        filter.chain_filter.clone(),
                         unified_api_version,
                     )
                     .await
@@ -689,6 +735,35 @@ 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
+    }
+
     async fn triggers_in_block(
         &self,
         logger: &Logger,