fetch block from DB

Author: Zoran Cvetkov

graph/src/components/store/traits.rs

@@ -293,17 +293,24 @@ impl<T: ?Sized + DeploymentCursorTracker> DeploymentCursorTracker for Arc<T> {
     }
 }
 
-pub trait SourceableStore: DeploymentCursorTracker {
+#[async_trait]
+pub trait SourceableStore: Sync + Send + 'static {
     /// Returns all versions of entities of the given entity_type that were
     /// changed in the given block_range.
     fn get_range(
         &self,
         entity_type: &EntityType,
         block_range: Range<BlockNumber>,
     ) -> Result<BTreeMap<BlockNumber, Vec<Entity>>, StoreError>;
+
+    fn input_schema(&self) -> InputSchema;
+
+    /// Get a pointer to the most recently processed block in the subgraph.
+    async fn block_ptr(&self) -> Result<Option<BlockPtr>, StoreError>;
 }
 
 // This silly impl is needed until https://github.com/rust-lang/rust/issues/65991 is stable.
+#[async_trait]
 impl<T: ?Sized + SourceableStore> SourceableStore for Arc<T> {
     fn get_range(
         &self,
@@ -312,6 +319,14 @@ impl<T: ?Sized + SourceableStore> SourceableStore for Arc<T> {
     ) -> Result<BTreeMap<BlockNumber, Vec<Entity>>, StoreError> {
         (**self).get_range(entity_type, block_range)
     }
+
+    fn input_schema(&self) -> InputSchema {
+        (**self).input_schema()
+    }
+
+    async fn block_ptr(&self) -> Result<Option<BlockPtr>, StoreError> {
+        (**self).block_ptr().await
+    }
 }
 
 /// A view of the store for indexing. All indexing-related operations need

store/postgres/src/subgraph_store.rs

@@ -1545,16 +1545,12 @@ impl SubgraphStoreTrait for SubgraphStore {
         let site = self.find_site(deployment)?;
         let store = self.for_site(&site)?;
         let input_schema = self.input_schema(&site.deployment)?;
-        let block_ptr = store.block_ptr(site.clone()).await?;
-        let block_cursor = store.block_cursor(site.clone()).await?;
-        let s = Arc::new(SourceableStore::new(
+
+        Ok(Arc::new(SourceableStore::new(
             site,
             store.clone(),
-            block_ptr,
-            block_cursor,
             input_schema,
-        ));
-        Ok(s as Arc<dyn store::SourceableStore>)
+        )))
     }
 
     async fn stop_subgraph(&self, loc: &DeploymentLocator) -> Result<(), StoreError> {

store/postgres/src/writable.rs

@@ -5,6 +5,7 @@ use std::sync::{Mutex, RwLock, TryLockError as RwLockError};
 use std::time::Instant;
 use std::{collections::BTreeMap, sync::Arc};
 
+use async_trait::async_trait;
 use graph::blockchain::block_stream::FirehoseCursor;
 use graph::blockchain::BlockTime;
 use graph::components::store::{Batch, DeploymentCursorTracker, DerivedEntityQuery, ReadStore};
@@ -1572,29 +1573,20 @@ impl ReadStore for WritableStore {
 pub struct SourceableStore {
     site: Arc<Site>,
     store: Arc<DeploymentStore>,
-    block_ptr: Option<BlockPtr>,
-    block_cursor: FirehoseCursor,
     input_schema: InputSchema,
 }
 
 impl SourceableStore {
-    pub fn new(
-        site: Arc<Site>,
-        store: Arc<DeploymentStore>,
-        block_ptr: Option<BlockPtr>,
-        block_cursor: FirehoseCursor,
-        input_schema: InputSchema,
-    ) -> Self {
+    pub fn new(site: Arc<Site>, store: Arc<DeploymentStore>, input_schema: InputSchema) -> Self {
         Self {
             site,
             store,
-            block_ptr,
-            block_cursor,
             input_schema,
         }
     }
 }
 
+#[async_trait]
 impl store::SourceableStore for SourceableStore {
     fn get_range(
         &self,
@@ -1604,18 +1596,13 @@ impl store::SourceableStore for SourceableStore {
         self.store
             .get_range(self.site.clone(), entity_type, block_range)
     }
-}
-impl DeploymentCursorTracker for SourceableStore {
+
     fn input_schema(&self) -> InputSchema {
         self.input_schema.cheap_clone()
     }
 
-    fn block_ptr(&self) -> Option<BlockPtr> {
-        self.block_ptr.clone()
-    }
-
-    fn firehose_cursor(&self) -> FirehoseCursor {
-        self.block_cursor.clone()
+    async fn block_ptr(&self) -> Result<Option<BlockPtr>, StoreError> {
+        self.store.block_ptr(self.site.cheap_clone()).await
     }
 }