all: Use a WritableStore much earlier when running a subgraph

lutter · lutter · commit 1486e8cc1f96 · 2022-01-18T09:40:32.000-08:00
diff --git a/chain/ethereum/src/chain.rs b/chain/ethereum/src/chain.rs
@@ -1,5 +1,6 @@
 use anyhow::{Context, Error};
 use graph::blockchain::BlockchainKind;
+use graph::components::store::WritableStore;
 use graph::data::subgraph::UnifiedMappingApiVersion;
 use graph::firehose::{FirehoseEndpoints, ForkStep};
 use graph::prelude::{
@@ -22,7 +23,6 @@ use graph::{
     prelude::{
         async_trait, lazy_static, o, serde_json as json, BlockNumber, ChainStore,
         EthereumBlockWithCalls, Future01CompatExt, Logger, LoggerFactory, MetricsRegistry, NodeId,
-        SubgraphStore,
     },
 };
 use prost::Message;
@@ -72,7 +72,6 @@ pub struct Chain {
     eth_adapters: Arc<EthereumNetworkAdapters>,
     chain_store: Arc<dyn ChainStore>,
     call_cache: Arc<dyn EthereumCallCache>,
-    subgraph_store: Arc<dyn SubgraphStore>,
     chain_head_update_listener: Arc<dyn ChainHeadUpdateListener>,
     reorg_threshold: BlockNumber,
     pub is_ingestible: bool,
@@ -92,7 +91,6 @@ impl Chain {
         registry: Arc<dyn MetricsRegistry>,
         chain_store: Arc<dyn ChainStore>,
         call_cache: Arc<dyn EthereumCallCache>,
-        subgraph_store: Arc<dyn SubgraphStore>,
         firehose_endpoints: FirehoseEndpoints,
         eth_adapters: EthereumNetworkAdapters,
         chain_head_update_listener: Arc<dyn ChainHeadUpdateListener>,
@@ -108,7 +106,6 @@ impl Chain {
             eth_adapters: Arc::new(eth_adapters),
             chain_store,
             call_cache,
-            subgraph_store,
             chain_head_update_listener,
             reorg_threshold,
             is_ingestible,
@@ -190,8 +187,8 @@ impl Blockchain for Chain {
     async fn new_firehose_block_stream(
         &self,
         deployment: DeploymentLocator,
+        writable: Arc<dyn WritableStore>,
         start_blocks: Vec<BlockNumber>,
-        firehose_cursor: Option<String>,
         filter: Arc<Self::TriggerFilter>,
         metrics: Arc<BlockStreamMetrics>,
         unified_api_version: UnifiedMappingApiVersion,
@@ -220,6 +217,7 @@ impl Blockchain for Chain {
             .new(o!("component" => "FirehoseBlockStream"));
 
         let firehose_mapper = Arc::new(FirehoseMapper {});
+        let firehose_cursor = writable.block_cursor()?;
 
         Ok(Box::new(FirehoseBlockStream::new(
             firehose_endpoint,
@@ -235,6 +233,7 @@ impl Blockchain for Chain {
     async fn new_polling_block_stream(
         &self,
         deployment: DeploymentLocator,
+        writable: Arc<dyn WritableStore>,
         start_blocks: Vec<BlockNumber>,
         subgraph_start_block: Option<BlockPtr>,
         filter: Arc<Self::TriggerFilter>,
@@ -259,12 +258,6 @@ impl Blockchain for Chain {
             .subgraph_logger(&deployment)
             .new(o!("component" => "BlockStream"));
         let chain_store = self.chain_store().clone();
-        let writable = self
-            .subgraph_store
-            .cheap_clone()
-            .writable(logger.clone(), deployment.id)
-            .await
-            .with_context(|| format!("no store for deployment `{}`", deployment.hash))?;
         let chain_head_update_stream = self
             .chain_head_update_listener
             .subscribe(self.name.clone(), logger.clone());
diff --git a/chain/near/src/chain.rs b/chain/near/src/chain.rs
@@ -1,5 +1,6 @@
 use graph::blockchain::BlockchainKind;
 use graph::cheap_clone::CheapClone;
+use graph::components::store::WritableStore;
 use graph::data::subgraph::UnifiedMappingApiVersion;
 use graph::firehose::FirehoseEndpoints;
 use graph::prelude::StopwatchMetrics;
@@ -100,8 +101,8 @@ impl Blockchain for Chain {
     async fn new_firehose_block_stream(
         &self,
         deployment: DeploymentLocator,
+        store: Arc<dyn WritableStore>,
         start_blocks: Vec<BlockNumber>,
-        firehose_cursor: Option<String>,
         filter: Arc<Self::TriggerFilter>,
         metrics: Arc<BlockStreamMetrics>,
         unified_api_version: UnifiedMappingApiVersion,
@@ -126,6 +127,7 @@ impl Blockchain for Chain {
             .new(o!("component" => "FirehoseBlockStream"));
 
         let firehose_mapper = Arc::new(FirehoseMapper {});
+        let firehose_cursor = store.block_cursor()?;
 
         Ok(Box::new(FirehoseBlockStream::new(
             firehose_endpoint,
@@ -141,6 +143,7 @@ impl Blockchain for Chain {
     async fn new_polling_block_stream(
         &self,
         _deployment: DeploymentLocator,
+        _writable: Arc<dyn WritableStore>,
         _start_blocks: Vec<BlockNumber>,
         _subgraph_start_block: Option<BlockPtr>,
         _filter: Arc<Self::TriggerFilter>,
diff --git a/core/src/subgraph/instance_manager.rs b/core/src/subgraph/instance_manager.rs
@@ -472,23 +472,20 @@ async fn new_block_stream<C: Blockchain>(
     };
 
     let block_stream = match is_firehose {
-        true => {
-            let firehose_cursor = inputs.store.block_cursor()?;
-
-            chain.new_firehose_block_stream(
-                inputs.deployment.clone(),
-                inputs.start_blocks.clone(),
-                firehose_cursor,
-                Arc::new(filter.clone()),
-                block_stream_metrics.clone(),
-                inputs.unified_api_version.clone(),
-            )
-        }
+        true => chain.new_firehose_block_stream(
+            inputs.deployment.clone(),
+            inputs.store.clone(),
+            inputs.start_blocks.clone(),
+            Arc::new(filter.clone()),
+            block_stream_metrics.clone(),
+            inputs.unified_api_version.clone(),
+        ),
         false => {
             let start_block = inputs.store.block_ptr()?;
 
             chain.new_polling_block_stream(
                 inputs.deployment.clone(),
+                inputs.store.clone(),
                 inputs.start_blocks.clone(),
                 start_block,
                 Arc::new(filter.clone()),
diff --git a/graph/src/blockchain/mock.rs b/graph/src/blockchain/mock.rs
@@ -6,7 +6,7 @@ use anyhow::Error;
 use async_trait::async_trait;
 use core::fmt;
 use serde::Deserialize;
-use std::convert::TryFrom;
+use std::{convert::TryFrom, sync::Arc};
 
 use super::{block_stream, HostFn, IngestorError, TriggerWithHandler};
 
@@ -296,8 +296,8 @@ impl Blockchain for MockBlockchain {
     async fn new_firehose_block_stream(
         &self,
         _deployment: crate::components::store::DeploymentLocator,
+        _store: Arc<dyn crate::components::store::WritableStore>,
         _start_blocks: Vec<crate::components::store::BlockNumber>,
-        _firehose_cursor: Option<String>,
         _filter: std::sync::Arc<Self::TriggerFilter>,
         _metrics: std::sync::Arc<block_stream::BlockStreamMetrics>,
         _unified_api_version: crate::data::subgraph::UnifiedMappingApiVersion,
@@ -308,6 +308,7 @@ impl Blockchain for MockBlockchain {
     async fn new_polling_block_stream(
         &self,
         _deployment: crate::components::store::DeploymentLocator,
+        _writable: Arc<dyn crate::components::store::WritableStore>,
         _start_blocks: Vec<crate::components::store::BlockNumber>,
         _subgraph_start_block: Option<BlockPtr>,
         _filter: std::sync::Arc<Self::TriggerFilter>,
diff --git a/graph/src/blockchain/mod.rs b/graph/src/blockchain/mod.rs
@@ -22,7 +22,7 @@ use crate::{
 };
 use crate::{
     components::{
-        store::{BlockNumber, ChainStore},
+        store::{BlockNumber, ChainStore, WritableStore},
         subgraph::DataSourceTemplateInfo,
     },
     prelude::{thiserror::Error, LinkResolver},
@@ -111,8 +111,8 @@ pub trait Blockchain: Debug + Sized + Send + Sync + Unpin + 'static {
     async fn new_firehose_block_stream(
         &self,
         deployment: DeploymentLocator,
+        store: Arc<dyn WritableStore>,
         start_blocks: Vec<BlockNumber>,
-        firehose_cursor: Option<String>,
         filter: Arc<Self::TriggerFilter>,
         metrics: Arc<BlockStreamMetrics>,
         unified_api_version: UnifiedMappingApiVersion,
@@ -121,6 +121,7 @@ pub trait Blockchain: Debug + Sized + Send + Sync + Unpin + 'static {
     async fn new_polling_block_stream(
         &self,
         deployment: DeploymentLocator,
+        writable: Arc<dyn WritableStore>,
         start_blocks: Vec<BlockNumber>,
         subgraph_start_block: Option<BlockPtr>,
         filter: Arc<Self::TriggerFilter>,
diff --git a/node/src/main.rs b/node/src/main.rs
@@ -479,7 +479,6 @@ fn ethereum_networks_as_chains(
                 registry.clone(),
                 chain_store.cheap_clone(),
                 chain_store,
-                store.subgraph_store(),
                 firehose_endpoints.map_or_else(|| FirehoseEndpoints::new(), |v| v.clone()),
                 eth_adapters.clone(),
                 chain_head_update_listener.clone(),
diff --git a/node/src/manager/commands/run.rs b/node/src/manager/commands/run.rs
@@ -102,7 +102,6 @@ pub async fn run(
         metrics_registry.clone(),
         chain_store.cheap_clone(),
         chain_store,
-        subgraph_store.clone(),
         firehose_endpoints.map_or_else(|| FirehoseEndpoints::new(), |v| v.clone()),
         eth_adapters,
         chain_head_update_listener,
