cleanup WritableStore

Author: Zoran Cvetkov

core/src/subgraph/instance_manager.rs

@@ -16,7 +16,6 @@ use graph::data::value::Word;
 use graph::data_source::causality_region::CausalityRegionSeq;
 use graph::env::EnvVars;
 use graph::prelude::{SubgraphInstanceManager as SubgraphInstanceManagerTrait, *};
-use graph::semver::Version;
 use graph::{blockchain::BlockchainMap, components::store::DeploymentLocator};
 use graph_runtime_wasm::module::ToAscPtr;
 use graph_runtime_wasm::RuntimeHostBuilder;
@@ -206,10 +205,7 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
 
     pub async fn hashes_to_read_store<C: Blockchain>(
         &self,
-        logger: &Logger,
-        link_resolver: &Arc<dyn LinkResolver>,
         hashes: Vec<DeploymentHash>,
-        max_spec_version: Version,
         is_runner_test: bool,
     ) -> anyhow::Result<Vec<(DeploymentHash, Arc<dyn SourceableStore>)>> {
         let mut sourceable_stores = Vec::new();
@@ -220,29 +216,11 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
         }
 
         for hash in hashes {
-            let file_bytes = link_resolver
-                .cat(logger, &hash.to_ipfs_link())
-                .await
-                .map_err(SubgraphAssignmentProviderError::ResolveError)?;
-            let raw: serde_yaml::Mapping = serde_yaml::from_slice(&file_bytes)
-                .map_err(|e| SubgraphAssignmentProviderError::ResolveError(e.into()))?;
-            let manifest = UnresolvedSubgraphManifest::<C>::parse(hash.cheap_clone(), raw)?;
-            let manifest = manifest
-                .resolve(&link_resolver, &logger, max_spec_version.clone())
-                .await?;
-
             let loc = subgraph_store
                 .active_locator(&hash)?
                 .ok_or_else(|| anyhow!("no active deployment for hash {}", hash))?;
 
-            let sourceable_store = subgraph_store
-                .clone()
-                .sourceable(
-                    logger.clone(),
-                    loc.id.clone(),
-                    Arc::new(manifest.template_idx_and_name().collect()),
-                )
-                .await?;
+            let sourceable_store = subgraph_store.clone().sourceable(loc.id.clone()).await?;
 
             sourceable_stores.push((loc.hash, sourceable_store));
         }
@@ -492,13 +470,7 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
         let decoder = Box::new(Decoder::new(decoder_hook));
 
         let subgraph_data_source_read_stores = self
-            .hashes_to_read_store::<C>(
-                &logger,
-                &link_resolver,
-                subgraph_ds_source_deployments,
-                manifest.spec_version.clone(),
-                is_runner_test,
-            )
+            .hashes_to_read_store::<C>(subgraph_ds_source_deployments, is_runner_test)
             .await?;
 
         let triggers_adapter = Arc::new(TriggersAdapterWrapper::new(

graph/src/components/store/traits.rs

@@ -188,9 +188,7 @@ pub trait SubgraphStore: Send + Sync + 'static {
 
     async fn sourceable(
         self: Arc<Self>,
-        logger: Logger,
         deployment: DeploymentId,
-        manifest_idx_and_name: Arc<Vec<(u32, String)>>,
     ) -> Result<Arc<dyn SourceableStore>, StoreError>;
 
     /// Initiate a graceful shutdown of the writable that a previous call to

store/postgres/src/subgraph_store.rs

@@ -1539,18 +1539,21 @@ impl SubgraphStoreTrait for SubgraphStore {
 
     async fn sourceable(
         self: Arc<Self>,
-        logger: Logger,
         deployment: graph::components::store::DeploymentId,
-        manifest_idx_and_name: Arc<Vec<(u32, String)>>,
     ) -> Result<Arc<dyn store::SourceableStore>, StoreError> {
-        let writable = self
-            .clone()
-            .writable(logger, deployment, manifest_idx_and_name)
-            .await?;
         let deployment = deployment.into();
         let site = self.find_site(deployment)?;
         let store = self.for_site(&site)?;
-        let s = Arc::new(SourceableStore::new(site, store.clone(), writable));
+        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(
+            site,
+            store.clone(),
+            block_ptr,
+            block_cursor,
+            input_schema,
+        ));
         Ok(s as Arc<dyn store::SourceableStore>)
     }
 

store/postgres/src/writable.rs

@@ -1572,19 +1572,25 @@ impl ReadStore for WritableStore {
 pub struct SourceableStore {
     site: Arc<Site>,
     store: Arc<DeploymentStore>,
-    writable: Arc<dyn store::WritableStore>,
+    block_ptr: Option<BlockPtr>,
+    block_cursor: FirehoseCursor,
+    input_schema: InputSchema,
 }
 
 impl SourceableStore {
     pub fn new(
         site: Arc<Site>,
         store: Arc<DeploymentStore>,
-        writable: Arc<dyn store::WritableStore>,
+        block_ptr: Option<BlockPtr>,
+        block_cursor: FirehoseCursor,
+        input_schema: InputSchema,
     ) -> Self {
         Self {
             site,
             store,
-            writable,
+            block_ptr,
+            block_cursor,
+            input_schema,
         }
     }
 }
@@ -1601,15 +1607,15 @@ impl store::SourceableStore for SourceableStore {
 }
 impl DeploymentCursorTracker for SourceableStore {
     fn input_schema(&self) -> InputSchema {
-        ReadStore::input_schema(&self.writable)
+        self.input_schema.cheap_clone()
     }
 
     fn block_ptr(&self) -> Option<BlockPtr> {
-        self.writable.block_ptr()
+        self.block_ptr.clone()
     }
 
     fn firehose_cursor(&self) -> FirehoseCursor {
-        self.writable.firehose_cursor()
+        self.block_cursor.clone()
     }
 }
 

store/test-store/tests/postgres/writable.rs

@@ -113,7 +113,7 @@ where
             .expect("we can get a writable store");
         let sourceable = store
             .subgraph_store()
-            .sourceable(LOGGER.clone(), deployment.id, Arc::new(Vec::new()))
+            .sourceable(deployment.id)
             .await
             .expect("we can get a writable store");