address PR comments

Author: Zoran Cvetkov

chain/arweave/src/chain.rs

@@ -11,7 +11,7 @@ use graph::components::store::{DeploymentCursorTracker, SourceableStore};
 use graph::data::subgraph::UnifiedMappingApiVersion;
 use graph::env::EnvVars;
 use graph::firehose::FirehoseEndpoint;
-use graph::prelude::{DeploymentHash, MetricsRegistry};
+use graph::prelude::MetricsRegistry;
 use graph::substreams::Clock;
 use graph::{
     blockchain::{
@@ -121,7 +121,7 @@ impl Blockchain for Chain {
         deployment: DeploymentLocator,
         store: impl DeploymentCursorTracker,
         start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
+        _source_subgraph_stores: Vec<Arc<dyn SourceableStore>>,
         filter: Arc<TriggerFilterWrapper<Self>>,
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error> {

chain/cosmos/src/chain.rs

@@ -2,7 +2,7 @@ use graph::blockchain::firehose_block_ingestor::FirehoseBlockIngestor;
 use graph::blockchain::{BlockIngestor, NoopDecoderHook, TriggerFilterWrapper};
 use graph::components::adapter::ChainId;
 use graph::env::EnvVars;
-use graph::prelude::{DeploymentHash, MetricsRegistry};
+use graph::prelude::MetricsRegistry;
 use graph::substreams::Clock;
 use std::collections::HashSet;
 use std::convert::TryFrom;
@@ -114,7 +114,7 @@ impl Blockchain for Chain {
         deployment: DeploymentLocator,
         store: impl DeploymentCursorTracker,
         start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
+        _source_subgraph_stores: Vec<Arc<dyn SourceableStore>>,
         filter: Arc<TriggerFilterWrapper<Self>>,
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error> {

chain/ethereum/src/chain.rs

@@ -12,7 +12,7 @@ use graph::data::subgraph::UnifiedMappingApiVersion;
 use graph::firehose::{FirehoseEndpoint, ForkStep};
 use graph::futures03::compat::Future01CompatExt;
 use graph::prelude::{
-    BlockHash, ComponentLoggerConfig, DeploymentHash, ElasticComponentLoggerConfig, EthereumBlock,
+    BlockHash, ComponentLoggerConfig, ElasticComponentLoggerConfig, EthereumBlock,
     EthereumCallCache, LightEthereumBlock, LightEthereumBlockExt, MetricsRegistry,
 };
 use graph::schema::InputSchema;
@@ -128,7 +128,7 @@ impl BlockStreamBuilder<Chain> for EthereumStreamBuilder {
         chain: &Chain,
         deployment: DeploymentLocator,
         start_blocks: Vec<BlockNumber>,
-        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
+        source_subgraph_stores: Vec<Arc<dyn SourceableStore>>,
         subgraph_current_block: Option<BlockPtr>,
         filter: Arc<TriggerFilterWrapper<Chain>>,
         unified_api_version: UnifiedMappingApiVersion,
@@ -150,7 +150,7 @@ impl BlockStreamBuilder<Chain> for EthereumStreamBuilder {
         chain: &Chain,
         deployment: DeploymentLocator,
         start_blocks: Vec<BlockNumber>,
-        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
+        source_subgraph_stores: Vec<Arc<dyn SourceableStore>>,
         subgraph_current_block: Option<BlockPtr>,
         filter: Arc<TriggerFilterWrapper<Chain>>,
         unified_api_version: UnifiedMappingApiVersion,
@@ -437,7 +437,7 @@ impl Blockchain for Chain {
         deployment: DeploymentLocator,
         store: impl DeploymentCursorTracker,
         start_blocks: Vec<BlockNumber>,
-        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
+        source_subgraph_stores: Vec<Arc<dyn SourceableStore>>,
         filter: Arc<TriggerFilterWrapper<Self>>,
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error> {

chain/near/src/chain.rs

@@ -13,7 +13,7 @@ use graph::data::subgraph::UnifiedMappingApiVersion;
 use graph::env::EnvVars;
 use graph::firehose::FirehoseEndpoint;
 use graph::futures03::TryFutureExt;
-use graph::prelude::{DeploymentHash, MetricsRegistry};
+use graph::prelude::MetricsRegistry;
 use graph::schema::InputSchema;
 use graph::substreams::{Clock, Package};
 use graph::{
@@ -152,7 +152,7 @@ impl BlockStreamBuilder<Chain> for NearStreamBuilder {
         _chain: &Chain,
         _deployment: DeploymentLocator,
         _start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
+        _source_subgraph_stores: Vec<Arc<dyn SourceableStore>>,
         _subgraph_current_block: Option<BlockPtr>,
         _filter: Arc<TriggerFilterWrapper<Chain>>,
         _unified_api_version: UnifiedMappingApiVersion,
@@ -232,7 +232,7 @@ impl Blockchain for Chain {
         deployment: DeploymentLocator,
         store: impl DeploymentCursorTracker,
         start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
+        _source_subgraph_stores: Vec<Arc<dyn SourceableStore>>,
         filter: Arc<TriggerFilterWrapper<Self>>,
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error> {

chain/starknet/src/chain.rs

@@ -23,8 +23,8 @@ use graph::{
     firehose::{self, FirehoseEndpoint, ForkStep},
     futures03::future::TryFutureExt,
     prelude::{
-        async_trait, BlockHash, BlockNumber, ChainStore, DeploymentHash, Error, Logger,
-        LoggerFactory, MetricsRegistry,
+        async_trait, BlockHash, BlockNumber, ChainStore, Error, Logger, LoggerFactory,
+        MetricsRegistry,
     },
     schema::InputSchema,
     slog::o,
@@ -116,7 +116,7 @@ impl Blockchain for Chain {
         deployment: DeploymentLocator,
         store: impl DeploymentCursorTracker,
         start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
+        _source_subgraph_stores: Vec<Arc<dyn SourceableStore>>,
         filter: Arc<TriggerFilterWrapper<Self>>,
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error> {
@@ -240,7 +240,7 @@ impl BlockStreamBuilder<Chain> for StarknetStreamBuilder {
         _chain: &Chain,
         _deployment: DeploymentLocator,
         _start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
+        _source_subgraph_stores: Vec<Arc<dyn SourceableStore>>,
         _subgraph_current_block: Option<BlockPtr>,
         _filter: Arc<TriggerFilterWrapper<Chain>>,
         _unified_api_version: UnifiedMappingApiVersion,

chain/substreams/src/block_stream.rs

@@ -11,7 +11,7 @@ use graph::{
     },
     components::store::{DeploymentLocator, SourceableStore},
     data::subgraph::UnifiedMappingApiVersion,
-    prelude::{async_trait, BlockNumber, BlockPtr, DeploymentHash},
+    prelude::{async_trait, BlockNumber, BlockPtr},
     schema::InputSchema,
     slog::o,
 };
@@ -104,7 +104,7 @@ impl BlockStreamBuilderTrait<Chain> for BlockStreamBuilder {
         _chain: &Chain,
         _deployment: DeploymentLocator,
         _start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
+        _source_subgraph_stores: Vec<Arc<dyn SourceableStore>>,
         _subgraph_current_block: Option<BlockPtr>,
         _filter: Arc<TriggerFilterWrapper<Chain>>,
         _unified_api_version: UnifiedMappingApiVersion,

chain/substreams/src/chain.rs

@@ -9,9 +9,7 @@ use graph::blockchain::{
 use graph::components::adapter::ChainId;
 use graph::components::store::{DeploymentCursorTracker, SourceableStore};
 use graph::env::EnvVars;
-use graph::prelude::{
-    BlockHash, CheapClone, DeploymentHash, Entity, LoggerFactory, MetricsRegistry,
-};
+use graph::prelude::{BlockHash, CheapClone, Entity, LoggerFactory, MetricsRegistry};
 use graph::schema::EntityKey;
 use graph::{
     blockchain::{
@@ -142,7 +140,7 @@ impl Blockchain for Chain {
         deployment: DeploymentLocator,
         store: impl DeploymentCursorTracker,
         _start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
+        _source_subgraph_stores: Vec<Arc<dyn SourceableStore>>,
         filter: Arc<TriggerFilterWrapper<Self>>,
         _unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error> {

core/src/subgraph/inputs.rs

@@ -6,7 +6,7 @@ use graph::{
     },
     data::subgraph::{SubgraphFeature, UnifiedMappingApiVersion},
     data_source::DataSourceTemplate,
-    prelude::{BlockNumber, DeploymentHash},
+    prelude::BlockNumber,
 };
 use std::collections::BTreeSet;
 use std::sync::Arc;
@@ -16,7 +16,7 @@ pub struct IndexingInputs<C: Blockchain> {
     pub features: BTreeSet<SubgraphFeature>,
     pub start_blocks: Vec<BlockNumber>,
     pub end_blocks: BTreeSet<BlockNumber>,
-    pub source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
+    pub source_subgraph_stores: Vec<Arc<dyn SourceableStore>>,
     pub stop_block: Option<BlockNumber>,
     pub store: Arc<dyn WritableStore>,
     pub debug_fork: Option<Arc<dyn SubgraphFork>>,

core/src/subgraph/instance_manager.rs

@@ -203,26 +203,25 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
         }
     }
 
-    pub async fn hashes_to_read_store<C: Blockchain>(
+    pub async fn get_sourceable_stores<C: Blockchain>(
         &self,
         hashes: Vec<DeploymentHash>,
         is_runner_test: bool,
-    ) -> anyhow::Result<Vec<(DeploymentHash, Arc<dyn SourceableStore>)>> {
-        let mut sourceable_stores = Vec::new();
-        let subgraph_store = self.subgraph_store.clone();
-
+    ) -> anyhow::Result<Vec<Arc<dyn SourceableStore>>> {
         if is_runner_test {
-            return Ok(sourceable_stores);
+            return Ok(Vec::new());
         }
 
+        let mut sourceable_stores = Vec::new();
+        let subgraph_store = self.subgraph_store.clone();
+
         for hash in hashes {
             let loc = subgraph_store
                 .active_locator(&hash)?
                 .ok_or_else(|| anyhow!("no active deployment for hash {}", hash))?;
 
             let sourceable_store = subgraph_store.clone().sourceable(loc.id.clone()).await?;
-
-            sourceable_stores.push((loc.hash, sourceable_store));
+            sourceable_stores.push(sourceable_store);
         }
 
         Ok(sourceable_stores)
@@ -469,21 +468,21 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
 
         let decoder = Box::new(Decoder::new(decoder_hook));
 
-        let subgraph_data_source_read_stores = self
-            .hashes_to_read_store::<C>(subgraph_ds_source_deployments, is_runner_test)
+        let subgraph_data_source_stores = self
+            .get_sourceable_stores::<C>(subgraph_ds_source_deployments, is_runner_test)
             .await?;
 
         let triggers_adapter = Arc::new(TriggersAdapterWrapper::new(
             triggers_adapter,
-            subgraph_data_source_read_stores.clone(),
+            subgraph_data_source_stores.clone(),
         ));
 
         let inputs = IndexingInputs {
             deployment: deployment.clone(),
             features,
             start_blocks,
             end_blocks,
-            source_subgraph_stores: subgraph_data_source_read_stores,
+            source_subgraph_stores: subgraph_data_source_stores,
             stop_block,
             store,
             debug_fork,

graph/src/blockchain/block_stream.rs

@@ -149,7 +149,7 @@ pub trait BlockStreamBuilder<C: Blockchain>: Send + Sync {
         chain: &C,
         deployment: DeploymentLocator,
         start_blocks: Vec<BlockNumber>,
-        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
+        source_subgraph_stores: Vec<Arc<dyn SourceableStore>>,
         subgraph_current_block: Option<BlockPtr>,
         filter: Arc<TriggerFilterWrapper<C>>,
         unified_api_version: UnifiedMappingApiVersion,
@@ -160,7 +160,7 @@ pub trait BlockStreamBuilder<C: Blockchain>: Send + Sync {
         chain: &C,
         deployment: DeploymentLocator,
         start_blocks: Vec<BlockNumber>,
-        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
+        source_subgraph_stores: Vec<Arc<dyn SourceableStore>>,
         subgraph_current_block: Option<BlockPtr>,
         filter: Arc<TriggerFilterWrapper<C>>,
         unified_api_version: UnifiedMappingApiVersion,
@@ -320,13 +320,13 @@ impl<C: Blockchain> BlockWithTriggers<C> {
 /// logic for each chain, increasing code repetition.
 pub struct TriggersAdapterWrapper<C: Blockchain> {
     pub adapter: Arc<dyn TriggersAdapter<C>>,
-    pub source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
+    pub source_subgraph_stores: Vec<Arc<dyn SourceableStore>>,
 }
 
 impl<C: Blockchain> TriggersAdapterWrapper<C> {
     pub fn new(
         adapter: Arc<dyn TriggersAdapter<C>>,
-        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
+        source_subgraph_stores: Vec<Arc<dyn SourceableStore>>,
     ) -> Self {
         Self {
             adapter,