add block tracking

Author: Zoran Cvetkov

core/src/subgraph/instance_manager.rs

@@ -212,11 +212,11 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
         max_spec_version: Version,
         is_runner_test: bool,
     ) -> anyhow::Result<Vec<(DeploymentHash, Arc<dyn SourceableStore>)>> {
-        let mut writable_stores = Vec::new();
+        let mut sourceable_stores = Vec::new();
         let subgraph_store = self.subgraph_store.clone();
 
         if is_runner_test {
-            return Ok(writable_stores);
+            return Ok(sourceable_stores);
         }
 
         for hash in hashes {
@@ -235,7 +235,7 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
                 .active_locator(&hash)?
                 .ok_or_else(|| anyhow!("no active deployment for hash {}", hash))?;
 
-            let readable_store = subgraph_store
+            let sourceable_store = subgraph_store
                 .clone()
                 .sourceable(
                     logger.clone(),
@@ -244,10 +244,10 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
                 )
                 .await?;
 
-            writable_stores.push((loc.hash, readable_store));
+            sourceable_stores.push((loc.hash, sourceable_store));
         }
 
-        Ok(writable_stores)
+        Ok(sourceable_stores)
     }
 
     pub async fn build_subgraph_runner<C>(

graph/src/components/store/traits.rs

@@ -269,27 +269,6 @@ impl<T: ?Sized + ReadStore> ReadStore for Arc<T> {
     }
 }
 
-pub trait SourceableStore: Send + Sync + '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>;
-}
-
-// This silly impl is needed until https://github.com/rust-lang/rust/issues/65991 is stable.
-impl<T: ?Sized + SourceableStore> SourceableStore for Arc<T> {
-    fn get_range(
-        &self,
-        entity_type: &EntityType,
-        block_range: Range<BlockNumber>,
-    ) -> Result<BTreeMap<BlockNumber, Vec<Entity>>, StoreError> {
-        (**self).get_range(entity_type, block_range)
-    }
-}
-
 pub trait DeploymentCursorTracker: Sync + Send + 'static {
     fn input_schema(&self) -> InputSchema;
 
@@ -316,6 +295,27 @@ impl<T: ?Sized + DeploymentCursorTracker> DeploymentCursorTracker for Arc<T> {
     }
 }
 
+pub trait SourceableStore: DeploymentCursorTracker {
+    /// 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>;
+}
+
+// This silly impl is needed until https://github.com/rust-lang/rust/issues/65991 is stable.
+impl<T: ?Sized + SourceableStore> SourceableStore for Arc<T> {
+    fn get_range(
+        &self,
+        entity_type: &EntityType,
+        block_range: Range<BlockNumber>,
+    ) -> Result<BTreeMap<BlockNumber, Vec<Entity>>, StoreError> {
+        (**self).get_range(entity_type, block_range)
+    }
+}
+
 /// A view of the store for indexing. All indexing-related operations need
 /// to go through this trait. Methods in this trait will never return a
 /// `StoreError::DatabaseUnavailable`. Instead, they will retry the

store/postgres/src/writable.rs

@@ -1572,19 +1572,19 @@ impl ReadStore for WritableStore {
 pub struct SourceableStore {
     site: Arc<Site>,
     store: Arc<DeploymentStore>,
-    _writable: Arc<dyn store::WritableStore>,
+    writable: Arc<dyn store::WritableStore>,
 }
 
 impl SourceableStore {
     pub fn new(
         site: Arc<Site>,
         store: Arc<DeploymentStore>,
-        _writable: Arc<dyn store::WritableStore>,
+        writable: Arc<dyn store::WritableStore>,
     ) -> Self {
         Self {
             site,
             store,
-            _writable,
+            writable,
         }
     }
 }
@@ -1599,6 +1599,19 @@ impl store::SourceableStore for SourceableStore {
             .get_range(self.site.clone(), entity_type, block_range)
     }
 }
+impl DeploymentCursorTracker for SourceableStore {
+    fn input_schema(&self) -> InputSchema {
+        ReadStore::input_schema(&self.writable)
+    }
+
+    fn block_ptr(&self) -> Option<BlockPtr> {
+        self.writable.block_ptr()
+    }
+
+    fn firehose_cursor(&self) -> FirehoseCursor {
+        self.writable.firehose_cursor()
+    }
+}
 
 impl DeploymentCursorTracker for WritableStore {
     fn block_ptr(&self) -> Option<BlockPtr> {