all: Narrow how much of the ChainStore can be accessed through Chain

Author: lutter

chain/ethereum/src/chain.rs

@@ -13,7 +13,7 @@ use graph::firehose::{FirehoseEndpoint, ForkStep};
 use graph::futures03::TryStreamExt;
 use graph::prelude::{
     retry, BlockHash, ComponentLoggerConfig, ElasticComponentLoggerConfig, EthereumBlock,
-    EthereumCallCache, LightEthereumBlock, LightEthereumBlockExt, MetricsRegistry,
+    EthereumCallCache, LightEthereumBlock, LightEthereumBlockExt, MetricsRegistry, StoreError,
 };
 use graph::schema::InputSchema;
 use graph::slog::{debug, error, trace, warn};
@@ -175,7 +175,7 @@ impl BlockStreamBuilder<Chain> for EthereumStreamBuilder {
             .logger_factory
             .subgraph_logger(&deployment)
             .new(o!("component" => "BlockStream"));
-        let chain_store = chain.chain_store();
+        let chain_store = chain.chain_store.cheap_clone();
         let chain_head_update_stream = chain
             .chain_head_update_listener
             .subscribe(chain.name.to_string(), logger.clone());
@@ -403,6 +403,13 @@ impl Chain {
         self.call_cache.clone()
     }
 
+    pub async fn block_number(
+        &self,
+        hash: &BlockHash,
+    ) -> Result<Option<(String, BlockNumber, Option<u64>, Option<BlockHash>)>, StoreError> {
+        self.chain_store.block_number(hash).await
+    }
+
     // TODO: This is only used to build the block stream which could prolly
     // be moved to the chain itself and return a block stream future that the
     // caller can spawn.
@@ -507,8 +514,8 @@ impl Blockchain for Chain {
         }
     }
 
-    fn chain_store(&self) -> Arc<dyn ChainStore> {
-        self.chain_store.clone()
+    async fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error> {
+        self.chain_store.cheap_clone().chain_head_ptr().await
     }
 
     async fn block_pointer_from_number(
@@ -615,7 +622,7 @@ impl Blockchain for Chain {
                     logger,
                     graph::env::ENV_VARS.reorg_threshold(),
                     self.chain_client(),
-                    self.chain_store().cheap_clone(),
+                    self.chain_store.cheap_clone(),
                     self.polling_ingestor_interval,
                     self.name.clone(),
                 )?)

chain/near/src/chain.rs

@@ -275,8 +275,8 @@ impl Blockchain for Chain {
         unimplemented!("This chain does not support Dynamic Data Sources. is_refetch_block_required always returns false, this shouldn't be called.")
     }
 
-    fn chain_store(&self) -> Arc<dyn ChainStore> {
-        self.chain_store.clone()
+    async fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error> {
+        self.chain_store.cheap_clone().chain_head_ptr().await
     }
 
     async fn block_pointer_from_number(

chain/substreams/src/chain.rs

@@ -167,8 +167,8 @@ impl Blockchain for Chain {
         unimplemented!("This chain does not support Dynamic Data Sources. is_refetch_block_required always returns false, this shouldn't be called.")
     }
 
-    fn chain_store(&self) -> Arc<dyn ChainStore> {
-        self.chain_store.clone()
+    async fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error> {
+        self.chain_store.cheap_clone().chain_head_ptr().await
     }
 
     async fn block_pointer_from_number(

core/src/subgraph/runner.rs

@@ -1141,7 +1141,7 @@ where
         if cached_head_ptr.is_none()
             || close_to_chain_head(&block_ptr, &cached_head_ptr, CAUGHT_UP_DISTANCE)
         {
-            self.state.cached_head_ptr = self.inputs.chain.chain_store().chain_head_ptr().await?;
+            self.state.cached_head_ptr = self.inputs.chain.chain_head_ptr().await?;
         }
         let is_caught_up =
             close_to_chain_head(&block_ptr, &self.state.cached_head_ptr, CAUGHT_UP_DISTANCE);
@@ -1463,7 +1463,7 @@ where
             && !self.inputs.store.is_deployment_synced()
             && !close_to_chain_head(
                 &block_ptr,
-                &self.inputs.chain.chain_store().chain_head_ptr().await?,
+                &self.inputs.chain.chain_head_ptr().await?,
                 // The "skip ptr updates timer" is ignored when a subgraph is at most 1000 blocks
                 // behind the chain head.
                 1000,

graph/src/blockchain/mock.rs

@@ -438,7 +438,7 @@ impl Blockchain for MockBlockchain {
         todo!()
     }
 
-    fn chain_store(&self) -> std::sync::Arc<dyn crate::components::store::ChainStore> {
+    async fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error> {
         todo!()
     }
 

graph/src/blockchain/mod.rs

@@ -31,7 +31,7 @@ use crate::{
     runtime::{gas::GasCounter, AscHeap, HostExportError},
 };
 use crate::{
-    components::store::{BlockNumber, ChainStore},
+    components::store::BlockNumber,
     prelude::{thiserror::Error, LinkResolver},
 };
 use anyhow::{anyhow, Context, Error};
@@ -196,7 +196,8 @@ pub trait Blockchain: Debug + Sized + Send + Sync + Unpin + 'static {
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error>;
 
-    fn chain_store(&self) -> Arc<dyn ChainStore>;
+    /// Return the pointer for the latest block that we are aware of
+    async fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error>;
 
     async fn block_pointer_from_number(
         &self,

server/index-node/src/resolver.rs

@@ -279,10 +279,9 @@ impl<S: Store> IndexNodeResolver<S> {
             );
             return Ok(r::Value::Null);
         };
-        let chain_store = chain.chain_store();
         let call_cache = chain.call_cache();
 
-        let (block_number, timestamp) = match chain_store.block_number(&block_hash).await {
+        let (block_number, timestamp) = match chain.block_number(&block_hash).await {
             Ok(Some((_, n, timestamp, _))) => (n, timestamp),
             Ok(None) => {
                 error!(