PollingBlockStream: Move sync status logic to instance manager (#3108)

* polling_block_stream: Move sync status logic to instance manager

* instance_manager: Add test for is_deployment_synced

* polling_block_stream: Remove last subgraph store dependency

This moves the time to sync metrics from the PollingBlockStream
to the instance_manager.

* polling_block_stream: Remove subgraph_store field/parameter

* instance_manager: Make sync status compare sub.block >= (chain.block - 1)

* store: Add chain head cache to ChainStore

* blockchain: Remove unused writable store dependency

Author: evaporei

chain/ethereum/src/chain.rs

@@ -232,7 +232,6 @@ 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>,
@@ -272,7 +271,6 @@ impl Blockchain for Chain {
         };
 
         Ok(Box::new(PollingBlockStream::new(
-            writable,
             chain_store,
             chain_head_update_stream,
             adapter,

chain/near/src/chain.rs

@@ -143,7 +143,6 @@ 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>,

core/src/subgraph/instance_manager.rs

@@ -30,6 +30,7 @@ use std::time::{Duration, Instant};
 use tokio::task;
 
 const MINUTE: Duration = Duration::from_secs(60);
+
 const BUFFERED_BLOCK_STREAM_SIZE: usize = 100;
 const BUFFERED_FIREHOSE_STREAM_SIZE: usize = 1;
 
@@ -481,7 +482,6 @@ async fn new_block_stream<C: Blockchain>(
 
             chain.new_polling_block_stream(
                 inputs.deployment.clone(),
-                inputs.store.clone(),
                 inputs.start_blocks.clone(),
                 start_block,
                 Arc::new(filter.clone()),
@@ -514,6 +514,7 @@ where
     let id_for_err = inputs.deployment.hash.clone();
     let mut should_try_unfail_deterministic = true;
     let mut should_try_unfail_non_deterministic = true;
+    let mut synced = false;
 
     // Exponential backoff that starts with two minutes and keeps
     // increasing its timeout exponentially until it reaches the ceiling.
@@ -532,6 +533,8 @@ where
             .await?
             .map_err(CancelableError::Error)
             .cancelable(&block_stream_canceler, || Err(CancelableError::Cancel));
+        let chain = inputs.chain.clone();
+        let chain_store = chain.chain_store();
 
         // Keep the stream's cancel guard around to be able to shut it down
         // when the subgraph deployment is unassigned
@@ -706,6 +709,20 @@ where
 
             match res {
                 Ok(needs_restart) => {
+                    // Once synced, no need to try to update the status again.
+                    if !synced && is_deployment_synced(&block_ptr, chain_store.cached_head_ptr()?) {
+                        // Updating the sync status is an one way operation.
+                        // This state change exists: not synced -> synced
+                        // This state change does NOT: synced -> not synced
+                        inputs.store.deployment_synced()?;
+
+                        // Stop trying to update the sync status.
+                        synced = true;
+
+                        // Stop recording time-to-sync metrics.
+                        ctx.block_stream_metrics.stopwatch.disable();
+                    }
+
                     // Keep trying to unfail subgraph for everytime it advances block(s) until it's
                     // health is not Failed anymore.
                     if should_try_unfail_non_deterministic {
@@ -1293,3 +1310,35 @@ fn persist_dynamic_data_sources<T: RuntimeHostBuilder<C>, C: Blockchain>(
     // Merge filters from data sources into the block stream builder
     ctx.state.filter.extend(data_sources.iter());
 }
+
+/// Checks if the Deployment BlockPtr is at least one block behind to the chain head.
+fn is_deployment_synced(deployment_head_ptr: &BlockPtr, chain_head_ptr: Option<BlockPtr>) -> bool {
+    matches!((deployment_head_ptr, &chain_head_ptr), (b1, Some(b2)) if b1.number >= (b2.number - 1))
+}
+
+#[test]
+fn test_is_deployment_synced() {
+    let block_0 = BlockPtr::try_from((
+        "bd34884280958002c51d3f7b5f853e6febeba33de0f40d15b0363006533c924f",
+        0,
+    ))
+    .unwrap();
+    let block_1 = BlockPtr::try_from((
+        "8511fa04b64657581e3f00e14543c1d522d5d7e771b54aa3060b662ade47da13",
+        1,
+    ))
+    .unwrap();
+    let block_2 = BlockPtr::try_from((
+        "b98fb783b49de5652097a989414c767824dff7e7fd765a63b493772511db81c1",
+        2,
+    ))
+    .unwrap();
+
+    assert!(!is_deployment_synced(&block_0, None));
+    assert!(!is_deployment_synced(&block_2, None));
+
+    assert!(!is_deployment_synced(&block_0, Some(block_2.clone())));
+
+    assert!(is_deployment_synced(&block_1, Some(block_2.clone())));
+    assert!(is_deployment_synced(&block_2, Some(block_2.clone())));
+}

graph/src/blockchain/mock.rs

@@ -308,7 +308,6 @@ 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>,

graph/src/blockchain/mod.rs

@@ -121,7 +121,6 @@ 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>,

graph/src/blockchain/polling_block_stream.rs

@@ -14,7 +14,6 @@ use super::block_stream::{
 use super::{Block, BlockPtr, Blockchain};
 
 use crate::components::store::BlockNumber;
-use crate::components::store::WritableStore;
 use crate::data::subgraph::UnifiedMappingApiVersion;
 use crate::prelude::*;
 #[cfg(debug_assertions)]
@@ -82,7 +81,6 @@ struct PollingBlockStreamContext<C>
 where
     C: Blockchain,
 {
-    subgraph_store: Arc<dyn WritableStore>,
     chain_store: Arc<dyn ChainStore>,
     adapter: Arc<C::TriggersAdapter>,
     node_id: NodeId,
@@ -107,7 +105,6 @@ where
 impl<C: Blockchain> Clone for PollingBlockStreamContext<C> {
     fn clone(&self) -> Self {
         Self {
-            subgraph_store: self.subgraph_store.cheap_clone(),
             chain_store: self.chain_store.cheap_clone(),
             adapter: self.adapter.clone(),
             node_id: self.node_id.clone(),
@@ -153,7 +150,6 @@ where
     C: Blockchain,
 {
     pub fn new(
-        subgraph_store: Arc<dyn WritableStore>,
         chain_store: Arc<dyn ChainStore>,
         chain_head_update_stream: ChainHeadUpdateStream,
         adapter: Arc<C::TriggersAdapter>,
@@ -175,7 +171,6 @@ where
             chain_head_update_stream,
             ctx: PollingBlockStreamContext {
                 current_block: start_block,
-                subgraph_store,
                 chain_store,
                 adapter,
                 node_id,
@@ -215,9 +210,6 @@ where
                     continue;
                 }
                 ReconciliationStep::Done => {
-                    // Reconciliation is complete, so try to mark subgraph as Synced
-                    ctx.update_subgraph_synced_status()?;
-
                     return Ok(NextBlocks::Done);
                 }
                 ReconciliationStep::Revert(from, to) => return Ok(NextBlocks::Revert(from, to)),
@@ -484,25 +476,6 @@ where
 
         Ok(ptr)
     }
-
-    /// Set subgraph deployment entity synced flag if and only if the subgraph block pointer is
-    /// caught up to the head block pointer.
-    fn update_subgraph_synced_status(&self) -> Result<(), StoreError> {
-        let head_ptr_opt = self.chain_store.chain_head_ptr()?;
-        let subgraph_ptr = self.current_block.clone();
-
-        if head_ptr_opt != subgraph_ptr || head_ptr_opt.is_none() || subgraph_ptr.is_none() {
-            // Not synced yet
-            Ok(())
-        } else {
-            // Synced
-
-            // Stop recording time-to-sync metrics.
-            self.metrics.stopwatch.disable();
-
-            self.subgraph_store.deployment_synced()
-        }
-    }
 }
 
 impl<C: Blockchain> BlockStream<C> for PollingBlockStream<C> {}

graph/src/components/store.rs

@@ -1170,6 +1170,9 @@ pub trait ChainStore: Send + Sync + 'static {
     /// The head block pointer will be None on initial set up.
     fn chain_head_ptr(&self) -> Result<Option<BlockPtr>, Error>;
 
+    /// In-memory time cached version of `chain_head_ptr`.
+    fn cached_head_ptr(&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.

store/postgres/src/chain_store.rs

@@ -5,6 +5,7 @@ use diesel::sql_types::Text;
 use diesel::{insert_into, update};
 use graph::blockchain::{Block, ChainIdentifier};
 use graph::prelude::web3::types::H256;
+use graph::util::timed_cache::TimedCache;
 use graph::{
     constraint_violation,
     prelude::{
@@ -19,6 +20,7 @@ use std::{
     convert::{TryFrom, TryInto},
     iter::FromIterator,
     sync::Arc,
+    time::Duration,
 };
 
 use graph::prelude::{
@@ -1176,6 +1178,7 @@ pub struct ChainStore {
     genesis_block_ptr: BlockPtr,
     status: ChainStatus,
     chain_head_update_sender: ChainHeadUpdateSender,
+    block_cache: TimedCache<&'static str, BlockPtr>,
 }
 
 impl ChainStore {
@@ -1194,6 +1197,7 @@ impl ChainStore {
             genesis_block_ptr: BlockPtr::new(net_identifier.genesis_block_hash.clone(), 0),
             status,
             chain_head_update_sender,
+            block_cache: TimedCache::new(Duration::from_secs(5)),
         };
 
         store
@@ -1408,11 +1412,22 @@ impl ChainStoreTrait for ChainStore {
                         (None, None) => None,
                         _ => unreachable!(),
                     })
-                    .and_then(|opt| opt)
+                    .and_then(|opt: Option<BlockPtr>| opt)
+                    .map(|head| {
+                        self.block_cache.set("head", Arc::new(head.clone()));
+                        head
+                    })
             })
             .map_err(Error::from)
     }
 
+    fn cached_head_ptr(&self) -> Result<Option<BlockPtr>, Error> {
+        match self.block_cache.get("head") {
+            Some(head) => Ok(Some(head.as_ref().clone())),
+            None => self.chain_head_ptr(),
+        }
+    }
+
     fn chain_head_cursor(&self) -> Result<Option<String>, Error> {
         use public::ethereum_networks::dsl::*;