all: Reduce the ChainStore interface for basic blockchains

Author: lutter

chain/ethereum/src/chain.rs

@@ -585,7 +585,7 @@ impl Blockchain for Chain {
         let ingestor: Box<dyn BlockIngestor> = match self.chain_client().as_ref() {
             ChainClient::Firehose(_) => {
                 let ingestor = FirehoseBlockIngestor::<HeaderOnlyBlock, Self>::new(
-                    self.chain_store.cheap_clone(),
+                    self.chain_store.cheap_clone().as_head_store(),
                     self.chain_client(),
                     self.logger_factory
                         .component_logger("EthereumFirehoseBlockIngestor", None),

chain/near/src/chain.rs

@@ -8,7 +8,7 @@ use graph::blockchain::{
 };
 use graph::cheap_clone::CheapClone;
 use graph::components::network_provider::ChainName;
-use graph::components::store::{DeploymentCursorTracker, SourceableStore};
+use graph::components::store::{ChainHeadStore, DeploymentCursorTracker, SourceableStore};
 use graph::data::subgraph::UnifiedMappingApiVersion;
 use graph::env::EnvVars;
 use graph::firehose::FirehoseEndpoint;
@@ -29,7 +29,7 @@ use graph::{
     },
     components::store::DeploymentLocator,
     firehose::{self as firehose, ForkStep},
-    prelude::{async_trait, o, BlockNumber, ChainStore, Error, Logger, LoggerFactory},
+    prelude::{async_trait, o, BlockNumber, Error, Logger, LoggerFactory},
 };
 use prost::Message;
 use std::collections::BTreeSet;
@@ -165,7 +165,7 @@ pub struct Chain {
     logger_factory: LoggerFactory,
     name: ChainName,
     client: Arc<ChainClient<Self>>,
-    chain_store: Arc<dyn ChainStore>,
+    chain_head_store: Arc<dyn ChainHeadStore>,
     metrics_registry: Arc<MetricsRegistry>,
     block_stream_builder: Arc<dyn BlockStreamBuilder<Self>>,
     prefer_substreams: bool,
@@ -183,7 +183,7 @@ impl BlockchainBuilder<Chain> for BasicBlockchainBuilder {
         Chain {
             logger_factory: self.logger_factory,
             name: self.name,
-            chain_store: self.chain_store,
+            chain_head_store: self.chain_head_store,
             client: Arc::new(ChainClient::new_firehose(self.firehose_endpoints)),
             metrics_registry: self.metrics_registry,
             block_stream_builder: Arc::new(NearStreamBuilder {}),
@@ -276,7 +276,7 @@ impl Blockchain for Chain {
     }
 
     async fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error> {
-        self.chain_store.cheap_clone().chain_head_ptr().await
+        self.chain_head_store.cheap_clone().chain_head_ptr().await
     }
 
     async fn block_pointer_from_number(
@@ -302,7 +302,7 @@ impl Blockchain for Chain {
 
     async fn block_ingestor(&self) -> anyhow::Result<Box<dyn BlockIngestor>> {
         let ingestor = FirehoseBlockIngestor::<crate::HeaderOnlyBlock, Self>::new(
-            self.chain_store.cheap_clone(),
+            self.chain_head_store.cheap_clone(),
             self.chain_client(),
             self.logger_factory
                 .component_logger("NearFirehoseBlockIngestor", None),

chain/substreams/src/block_ingestor.rs

@@ -8,14 +8,14 @@ use graph::blockchain::{
     client::ChainClient, substreams_block_stream::SubstreamsBlockStream, BlockIngestor,
 };
 use graph::components::network_provider::ChainName;
+use graph::components::store::ChainHeadStore;
 use graph::prelude::MetricsRegistry;
 use graph::slog::trace;
 use graph::substreams::Package;
 use graph::tokio_stream::StreamExt;
 use graph::{
     blockchain::block_stream::BlockStreamEvent,
     cheap_clone::CheapClone,
-    components::store::ChainStore,
     prelude::{async_trait, error, info, DeploymentHash, Logger},
     util::backoff::ExponentialBackoff,
 };
@@ -26,7 +26,7 @@ const SUBSTREAMS_HEAD_TRACKER_BYTES: &[u8; 89935] = include_bytes!(
 );
 
 pub struct SubstreamsBlockIngestor {
-    chain_store: Arc<dyn ChainStore>,
+    chain_store: Arc<dyn ChainHeadStore>,
     client: Arc<ChainClient<super::Chain>>,
     logger: Logger,
     chain_name: ChainName,
@@ -35,7 +35,7 @@ pub struct SubstreamsBlockIngestor {
 
 impl SubstreamsBlockIngestor {
     pub fn new(
-        chain_store: Arc<dyn ChainStore>,
+        chain_store: Arc<dyn ChainHeadStore>,
         client: Arc<ChainClient<super::Chain>>,
         logger: Logger,
         chain_name: ChainName,

chain/substreams/src/chain.rs

@@ -7,7 +7,7 @@ use graph::blockchain::{
     NoopRuntimeAdapter, TriggerFilterWrapper,
 };
 use graph::components::network_provider::ChainName;
-use graph::components::store::{DeploymentCursorTracker, SourceableStore};
+use graph::components::store::{ChainHeadStore, DeploymentCursorTracker, SourceableStore};
 use graph::env::EnvVars;
 use graph::prelude::{BlockHash, CheapClone, Entity, LoggerFactory, MetricsRegistry};
 use graph::schema::EntityKey;
@@ -19,7 +19,7 @@ use graph::{
     },
     components::store::DeploymentLocator,
     data::subgraph::UnifiedMappingApiVersion,
-    prelude::{async_trait, BlockNumber, ChainStore},
+    prelude::{async_trait, BlockNumber},
     slog::Logger,
 };
 
@@ -65,7 +65,7 @@ impl blockchain::Block for Block {
 }
 
 pub struct Chain {
-    chain_store: Arc<dyn ChainStore>,
+    chain_head_store: Arc<dyn ChainHeadStore>,
     block_stream_builder: Arc<dyn BlockStreamBuilder<Self>>,
     chain_id: ChainName,
 
@@ -79,15 +79,15 @@ impl Chain {
         logger_factory: LoggerFactory,
         chain_client: Arc<ChainClient<Self>>,
         metrics_registry: Arc<MetricsRegistry>,
-        chain_store: Arc<dyn ChainStore>,
+        chain_store: Arc<dyn ChainHeadStore>,
         block_stream_builder: Arc<dyn BlockStreamBuilder<Self>>,
         chain_id: ChainName,
     ) -> Self {
         Self {
             logger_factory,
             client: chain_client,
             metrics_registry,
-            chain_store,
+            chain_head_store: chain_store,
             block_stream_builder,
             chain_id,
         }
@@ -168,7 +168,7 @@ impl Blockchain for Chain {
     }
 
     async fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error> {
-        self.chain_store.cheap_clone().chain_head_ptr().await
+        self.chain_head_store.cheap_clone().chain_head_ptr().await
     }
 
     async fn block_pointer_from_number(
@@ -195,7 +195,7 @@ impl Blockchain for Chain {
 
     async fn block_ingestor(&self) -> anyhow::Result<Box<dyn BlockIngestor>> {
         Ok(Box::new(SubstreamsBlockIngestor::new(
-            self.chain_store.cheap_clone(),
+            self.chain_head_store.cheap_clone(),
             self.client.cheap_clone(),
             self.logger_factory
                 .component_logger("SubstreamsBlockIngestor", None),
@@ -211,13 +211,13 @@ impl blockchain::BlockchainBuilder<super::Chain> for BasicBlockchainBuilder {
         let BasicBlockchainBuilder {
             logger_factory,
             name,
-            chain_store,
+            chain_head_store,
             firehose_endpoints,
             metrics_registry,
         } = self;
 
         Chain {
-            chain_store,
+            chain_head_store,
             block_stream_builder: Arc::new(crate::BlockStreamBuilder::new()),
             logger_factory,
             client: Arc::new(ChainClient::new_firehose(firehose_endpoints)),

graph/src/blockchain/builder.rs

@@ -2,8 +2,11 @@ use tonic::async_trait;
 
 use super::Blockchain;
 use crate::{
-    components::store::ChainStore, data::value::Word, env::EnvVars, firehose::FirehoseEndpoints,
-    prelude::LoggerFactory, prelude::MetricsRegistry,
+    components::store::ChainHeadStore,
+    data::value::Word,
+    env::EnvVars,
+    firehose::FirehoseEndpoints,
+    prelude::{LoggerFactory, MetricsRegistry},
 };
 use std::sync::Arc;
 
@@ -12,7 +15,7 @@ use std::sync::Arc;
 pub struct BasicBlockchainBuilder {
     pub logger_factory: LoggerFactory,
     pub name: Word,
-    pub chain_store: Arc<dyn ChainStore>,
+    pub chain_head_store: Arc<dyn ChainHeadStore>,
     pub firehose_endpoints: FirehoseEndpoints,
     pub metrics_registry: Arc<MetricsRegistry>,
 }

graph/src/blockchain/firehose_block_ingestor.rs

@@ -2,7 +2,7 @@ use std::{marker::PhantomData, sync::Arc, time::Duration};
 
 use crate::{
     blockchain::Block as BlockchainBlock,
-    components::store::ChainStore,
+    components::store::ChainHeadStore,
     firehose::{self, decode_firehose_block, HeaderOnly},
     prelude::{error, info, Logger},
     util::backoff::ExponentialBackoff,
@@ -40,7 +40,7 @@ pub struct FirehoseBlockIngestor<M, C: Blockchain>
 where
     M: prost::Message + BlockchainBlock + Default + 'static,
 {
-    chain_store: Arc<dyn ChainStore>,
+    chain_head_store: Arc<dyn ChainHeadStore>,
     client: Arc<ChainClient<C>>,
     logger: Logger,
     default_transforms: Vec<Transforms>,
@@ -54,13 +54,13 @@ where
     M: prost::Message + BlockchainBlock + Default + 'static,
 {
     pub fn new(
-        chain_store: Arc<dyn ChainStore>,
+        chain_head_store: Arc<dyn ChainHeadStore>,
         client: Arc<ChainClient<C>>,
         logger: Logger,
         chain_name: ChainName,
     ) -> FirehoseBlockIngestor<M, C> {
         FirehoseBlockIngestor {
-            chain_store,
+            chain_head_store,
             client,
             logger,
             phantom: PhantomData {},
@@ -78,7 +78,7 @@ where
         let mut backoff =
             ExponentialBackoff::new(Duration::from_millis(250), Duration::from_secs(30));
         loop {
-            match self.chain_store.clone().chain_head_cursor() {
+            match self.chain_head_store.clone().chain_head_cursor() {
                 Ok(cursor) => return cursor.unwrap_or_default(),
                 Err(e) => {
                     error!(self.logger, "Fetching chain head cursor failed: {:#}", e);
@@ -149,7 +149,7 @@ where
 
         trace!(self.logger, "Received new block to ingest {}", block.ptr());
 
-        self.chain_store
+        self.chain_head_store
             .clone()
             .set_chain_head(block, response.cursor.clone())
             .await

graph/src/blockchain/mock.rs

@@ -2,7 +2,10 @@ use crate::{
     bail,
     components::{
         link_resolver::LinkResolver,
-        store::{BlockNumber, DeploymentCursorTracker, DeploymentLocator, SourceableStore},
+        store::{
+            BlockNumber, ChainHeadStore, DeploymentCursorTracker, DeploymentLocator,
+            SourceableStore,
+        },
         subgraph::InstanceDSTemplateInfo,
     },
     data::subgraph::UnifiedMappingApiVersion,
@@ -471,6 +474,23 @@ pub struct MockChainStore {
     pub blocks: BTreeMap<BlockNumber, Vec<ExtendedBlockPtr>>,
 }
 
+#[async_trait]
+impl ChainHeadStore for MockChainStore {
+    async fn chain_head_ptr(self: Arc<Self>) -> Result<Option<BlockPtr>, Error> {
+        unimplemented!()
+    }
+    fn chain_head_cursor(&self) -> Result<Option<String>, Error> {
+        unimplemented!()
+    }
+    async fn set_chain_head(
+        self: Arc<Self>,
+        _block: Arc<dyn Block>,
+        _cursor: String,
+    ) -> Result<(), Error> {
+        unimplemented!()
+    }
+}
+
 #[async_trait]
 impl ChainStore for MockChainStore {
     async fn block_ptrs_by_numbers(
@@ -502,19 +522,6 @@ impl ChainStore for MockChainStore {
     ) -> Result<Option<H256>, Error> {
         unimplemented!()
     }
-    async fn chain_head_ptr(self: Arc<Self>) -> Result<Option<BlockPtr>, Error> {
-        unimplemented!()
-    }
-    fn chain_head_cursor(&self) -> Result<Option<String>, Error> {
-        unimplemented!()
-    }
-    async fn set_chain_head(
-        self: Arc<Self>,
-        _block: Arc<dyn Block>,
-        _cursor: String,
-    ) -> Result<(), Error> {
-        unimplemented!()
-    }
     async fn blocks(self: Arc<Self>, _hashes: Vec<BlockHash>) -> Result<Vec<Value>, Error> {
         unimplemented!()
     }
@@ -565,4 +572,7 @@ impl ChainStore for MockChainStore {
     fn set_chain_identifier(&self, _ident: &ChainIdentifier) -> Result<(), Error> {
         unimplemented!()
     }
+    fn as_head_store(self: Arc<Self>) -> Arc<dyn ChainHeadStore> {
+        self.clone()
+    }
 }

graph/src/components/store/traits.rs

@@ -454,9 +454,36 @@ pub trait BlockStore: Send + Sync + 'static {
     fn chain_store(&self, network: &str) -> Option<Arc<Self::ChainStore>>;
 }
 
+/// An interface for tracking the chain head in the store used by most chain
+/// implementations
+#[async_trait]
+pub trait ChainHeadStore: Send + Sync {
+    /// Get the current head block pointer for this chain.
+    /// Any changes to the head block pointer will be to a block with a larger block number, never
+    /// to a block with a smaller or equal block number.
+    ///
+    /// The head block pointer will be None on initial set up.
+    async fn chain_head_ptr(self: Arc<Self>) -> Result<Option<BlockPtr>, Error>;
+
+    /// Get the current head block cursor for this chain.
+    ///
+    /// The head block cursor will be None on initial set up.
+    fn chain_head_cursor(&self) -> Result<Option<String>, Error>;
+
+    /// This method does actually three operations:
+    /// - Upserts received block into blocks table
+    /// - Update chain head block into networks table
+    /// - Update chain head cursor into networks table
+    async fn set_chain_head(
+        self: Arc<Self>,
+        block: Arc<dyn Block>,
+        cursor: String,
+    ) -> Result<(), Error>;
+}
+
 /// Common trait for blockchain store implementations.
 #[async_trait]
-pub trait ChainStore: Send + Sync + 'static {
+pub trait ChainStore: ChainHeadStore {
     /// Get a pointer to this blockchain's genesis block.
     fn genesis_block_ptr(&self) -> Result<BlockPtr, Error>;
 
@@ -486,28 +513,6 @@ pub trait ChainStore: Send + Sync + 'static {
         ancestor_count: BlockNumber,
     ) -> Result<Option<H256>, Error>;
 
-    /// Get the current head block pointer for this chain.
-    /// Any changes to the head block pointer will be to a block with a larger block number, never
-    /// to a block with a smaller or equal block number.
-    ///
-    /// The head block pointer will be None on initial set up.
-    async fn chain_head_ptr(self: Arc<Self>) -> Result<Option<BlockPtr>, Error>;
-
-    /// Get the current head block cursor for this chain.
-    ///
-    /// The head block cursor will be None on initial set up.
-    fn chain_head_cursor(&self) -> Result<Option<String>, Error>;
-
-    /// This method does actually three operations:
-    /// - Upserts received block into blocks table
-    /// - Update chain head block into networks table
-    /// - Update chain head cursor into networks table
-    async fn set_chain_head(
-        self: Arc<Self>,
-        block: Arc<dyn Block>,
-        cursor: String,
-    ) -> Result<(), Error>;
-
     /// Returns the blocks present in the store.
     async fn blocks(
         self: Arc<Self>,
@@ -587,6 +592,10 @@ pub trait ChainStore: Send + Sync + 'static {
 
     /// Update the chain identifier for this store.
     fn set_chain_identifier(&self, ident: &ChainIdentifier) -> Result<(), Error>;
+
+    /// Workaround for Rust issue #65991 that keeps us from using an
+    /// `Arc<dyn ChainStore>` as an `Arc<dyn ChainHeadStore>`
+    fn as_head_store(self: Arc<Self>) -> Arc<dyn ChainHeadStore>;
 }
 
 pub trait EthereumCallCache: Send + Sync + 'static {