Revert "fix: Allow grafts to add data sources (#3989)"

This reverts commit 3b3b458.

Author: leoyvens

core/src/subgraph/instance_manager.rs

@@ -8,7 +8,7 @@ use graph::blockchain::Blockchain;
 use graph::blockchain::NodeCapabilities;
 use graph::blockchain::{BlockchainKind, TriggerFilter};
 use graph::components::subgraph::ProofOfIndexingVersion;
-use graph::data::subgraph::{UnresolvedSubgraphManifest, SPEC_VERSION_0_0_6};
+use graph::data::subgraph::SPEC_VERSION_0_0_6;
 use graph::prelude::{SubgraphInstanceManager as SubgraphInstanceManagerTrait, *};
 use graph::{blockchain::BlockchainMap, components::store::DeploymentLocator};
 use graph_runtime_wasm::module::ToAscPtr;
@@ -176,49 +176,47 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
             .writable(logger.clone(), deployment.id)
             .await?;
 
-        let raw_yaml = serde_yaml::to_string(&manifest).unwrap();
-        let manifest = UnresolvedSubgraphManifest::parse(deployment.hash.cheap_clone(), manifest)?;
-
-        // Make sure the `raw_yaml` is present on both this subgraph and the graft base.
-        self.subgraph_store
-            .set_manifest_raw_yaml(&deployment.hash, raw_yaml)
-            .await?;
-        if let Some(graft) = &manifest.graft {
-            let file_bytes = self
-                .link_resolver
-                .cat(&logger, &graft.base.to_ipfs_link())
-                .await?;
-            let yaml = String::from_utf8(file_bytes)?;
-            self.subgraph_store
-                .set_manifest_raw_yaml(&graft.base, yaml)
-                .await?;
-        }
-
-        info!(logger, "Resolve subgraph files using IPFS");
-
-        // Allow for infinite retries for subgraph definition files.
-        let link_resolver = Arc::from(self.link_resolver.with_retries());
-        let mut manifest = manifest
-            .resolve(&link_resolver, &logger, ENV_VARS.max_spec_version.clone())
-            .await?;
-
-        info!(logger, "Successfully resolved subgraph files using IPFS");
-
-        let manifest_idx_and_name: Vec<(u32, String)> = manifest.template_idx_and_name().collect();
-
         // Start the subgraph deployment before reading dynamic data
         // sources; if the subgraph is a graft or a copy, starting it will
         // do the copying and dynamic data sources won't show up until after
         // that is done
         store.start_subgraph_deployment(&logger).await?;
 
-        // Dynamic data sources are loaded by appending them to the manifest.
-        //
-        // Refactor: Preferrably we'd avoid any mutation of the manifest.
-        let (manifest, static_data_sources) = {
-            let data_sources = load_dynamic_data_sources(store.clone(), logger.clone(), &manifest)
-                .await
-                .context("Failed to load dynamic data sources")?;
+        let (manifest, manifest_idx_and_name, static_data_sources) = {
+            info!(logger, "Resolve subgraph files using IPFS");
+
+            let mut manifest = SubgraphManifest::resolve_from_raw(
+                deployment.hash.cheap_clone(),
+                manifest,
+                // Allow for infinite retries for subgraph definition files.
+                &Arc::from(self.link_resolver.with_retries()),
+                &logger,
+                ENV_VARS.max_spec_version.clone(),
+            )
+            .await
+            .context("Failed to resolve subgraph from IPFS")?;
+
+            // We cannot include static data sources in the map because a static data source and a
+            // template may have the same name in the manifest.
+            let ds_len = manifest.data_sources.len() as u32;
+            let manifest_idx_and_name: Vec<(u32, String)> = manifest
+                .templates
+                .iter()
+                .map(|t| t.name().to_owned())
+                .enumerate()
+                .map(|(idx, name)| (ds_len + idx as u32, name))
+                .collect();
+
+            let data_sources = load_dynamic_data_sources(
+                store.clone(),
+                logger.clone(),
+                &manifest,
+                manifest_idx_and_name.clone(),
+            )
+            .await
+            .context("Failed to load dynamic data sources")?;
+
+            info!(logger, "Successfully resolved subgraph files using IPFS");
 
             let static_data_sources = manifest.data_sources.clone();
 
@@ -231,7 +229,7 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
                 manifest.data_sources.len()
             );
 
-            (manifest, static_data_sources)
+            (manifest, manifest_idx_and_name, static_data_sources)
         };
 
         let static_filters =

core/src/subgraph/loader.rs

@@ -9,8 +9,8 @@ pub async fn load_dynamic_data_sources<C: Blockchain>(
     store: Arc<dyn WritableStore>,
     logger: Logger,
     manifest: &SubgraphManifest<C>,
+    manifest_idx_and_name: Vec<(u32, String)>,
 ) -> Result<Vec<DataSource<C>>, Error> {
-    let manifest_idx_and_name = manifest.template_idx_and_name().collect();
     let start_time = Instant::now();
 
     let mut data_sources: Vec<DataSource<C>> = vec![];

core/src/subgraph/registrar.rs

@@ -552,7 +552,6 @@ async fn create_subgraph_version<C: Blockchain, S: SubgraphStore>(
     version_switching_mode: SubgraphVersionSwitchingMode,
     resolver: &Arc<dyn LinkResolver>,
 ) -> Result<DeploymentLocator, SubgraphRegistrarError> {
-    let raw_string = serde_yaml::to_string(&raw).unwrap();
     let unvalidated = UnvalidatedSubgraphManifest::<C>::resolve(
         deployment,
         raw,
@@ -619,7 +618,7 @@ async fn create_subgraph_version<C: Blockchain, S: SubgraphStore>(
 
     // Apply the subgraph versioning and deployment operations,
     // creating a new subgraph deployment if one doesn't exist.
-    let deployment = DeploymentCreate::new(raw_string, &manifest, start_block)
+    let deployment = DeploymentCreate::new(&manifest, start_block)
         .graft(base_block)
         .debug(debug_fork);
     deployment_store

graph/src/components/store/traits.rs

@@ -152,13 +152,6 @@ pub trait SubgraphStore: Send + Sync + 'static {
 
     /// Find the deployment locators for the subgraph with the given hash
     fn locators(&self, hash: &str) -> Result<Vec<DeploymentLocator>, StoreError>;
-
-    /// This migrates subgraphs that existed before the raw_yaml column was added.
-    async fn set_manifest_raw_yaml(
-        &self,
-        hash: &DeploymentHash,
-        raw_yaml: String,
-    ) -> Result<(), StoreError>;
 }
 
 pub trait ReadStore: Send + Sync + 'static {

graph/src/data/subgraph/mod.rs

@@ -10,12 +10,13 @@ pub mod status;
 
 pub use features::{SubgraphFeature, SubgraphFeatureValidationError};
 
+use anyhow::ensure;
 use anyhow::{anyhow, Error};
 use futures03::{future::try_join3, stream::FuturesOrdered, TryStreamExt as _};
 use semver::Version;
 use serde::{de, ser};
 use serde_yaml;
-use slog::{info, Logger};
+use slog::{debug, info, Logger};
 use stable_hash::{FieldAddress, StableHash};
 use stable_hash_legacy::SequenceNumber;
 use std::{collections::BTreeSet, marker::PhantomData};
@@ -24,7 +25,6 @@ use wasmparser;
 use web3::types::Address;
 
 use crate::{
-    bail,
     blockchain::{BlockPtr, Blockchain, DataSource as _},
     components::{
         link_resolver::LinkResolver,
@@ -41,7 +41,6 @@ use crate::{
         offchain::OFFCHAIN_KINDS, DataSource, DataSourceTemplate, UnresolvedDataSource,
         UnresolvedDataSourceTemplate,
     },
-    ensure,
     prelude::{r, CheapClone, ENV_VARS},
 };
 
@@ -357,7 +356,7 @@ pub enum SubgraphManifestResolveError {
     #[error("subgraph is not valid YAML")]
     InvalidFormat,
     #[error("resolve error: {0}")]
-    ResolveError(#[from] anyhow::Error),
+    ResolveError(anyhow::Error),
 }
 
 /// Data source contexts are conveniently represented as entities.
@@ -497,7 +496,7 @@ pub struct BaseSubgraphManifest<C, S, D, T> {
 }
 
 /// SubgraphManifest with IPFS links unresolved
-pub type UnresolvedSubgraphManifest<C> = BaseSubgraphManifest<
+type UnresolvedSubgraphManifest<C> = BaseSubgraphManifest<
     C,
     UnresolvedSchema,
     UnresolvedDataSource<C>,
@@ -615,16 +614,35 @@ impl<C: Blockchain> SubgraphManifest<C> {
     /// Entry point for resolving a subgraph definition.
     pub async fn resolve_from_raw(
         id: DeploymentHash,
-        raw: serde_yaml::Mapping,
+        mut raw: serde_yaml::Mapping,
         resolver: &Arc<dyn LinkResolver>,
         logger: &Logger,
         max_spec_version: semver::Version,
     ) -> Result<Self, SubgraphManifestResolveError> {
-        let unresolved = UnresolvedSubgraphManifest::parse(id, raw)?;
+        // Inject the IPFS hash as the ID of the subgraph into the definition.
+        raw.insert("id".into(), id.to_string().into());
+
+        // Parse the YAML data into an UnresolvedSubgraphManifest
+        let unresolved: UnresolvedSubgraphManifest<C> = serde_yaml::from_value(raw.into())?;
+
+        debug!(logger, "Features {:?}", unresolved.features);
 
         let resolved = unresolved
             .resolve(resolver, logger, max_spec_version)
-            .await?;
+            .await
+            .map_err(SubgraphManifestResolveError::ResolveError)?;
+
+        if (resolved.spec_version < SPEC_VERSION_0_0_7)
+            && resolved
+                .data_sources
+                .iter()
+                .any(|ds| OFFCHAIN_KINDS.contains(&ds.kind()))
+        {
+            return Err(SubgraphManifestResolveError::ResolveError(anyhow!(
+                "Offchain data sources not supported prior to {}",
+                SPEC_VERSION_0_0_7
+            )));
+        }
 
         Ok(resolved)
     }
@@ -667,37 +685,15 @@ impl<C: Blockchain> SubgraphManifest<C> {
     ) -> Result<UnifiedMappingApiVersion, DifferentMappingApiVersions> {
         UnifiedMappingApiVersion::try_from_versions(self.api_versions())
     }
-
-    pub fn template_idx_and_name(&self) -> impl Iterator<Item = (u32, String)> + '_ {
-        // We cannot include static data sources in the map because a static data source and a
-        // template may have the same name in the manifest. Duplicated with
-        // `UnresolvedSubgraphManifest::template_idx_and_name`.
-        let ds_len = self.data_sources.len() as u32;
-        self.templates
-            .iter()
-            .map(|t| t.name().to_owned())
-            .enumerate()
-            .map(move |(idx, name)| (ds_len + idx as u32, name))
-    }
 }
 
 impl<C: Blockchain> UnresolvedSubgraphManifest<C> {
-    pub fn parse(
-        id: DeploymentHash,
-        mut raw: serde_yaml::Mapping,
-    ) -> Result<Self, SubgraphManifestResolveError> {
-        // Inject the IPFS hash as the ID of the subgraph into the definition.
-        raw.insert("id".into(), id.to_string().into());
-
-        serde_yaml::from_value(raw.into()).map_err(Into::into)
-    }
-
     pub async fn resolve(
         self,
         resolver: &Arc<dyn LinkResolver>,
         logger: &Logger,
         max_spec_version: semver::Version,
-    ) -> Result<SubgraphManifest<C>, SubgraphManifestResolveError> {
+    ) -> Result<SubgraphManifest<C>, anyhow::Error> {
         let UnresolvedSubgraphManifest {
             id,
             spec_version,
@@ -718,14 +714,14 @@ impl<C: Blockchain> UnresolvedSubgraphManifest<C> {
                 max_spec_version,
                 id,
                 spec_version
-            ).into());
+            ));
         }
 
         let ds_count = data_sources.len();
         if ds_count as u64 + templates.len() as u64 > u32::MAX as u64 {
-            return Err(
-                anyhow!("Subgraph has too many declared data sources and templates",).into(),
-            );
+            return Err(anyhow!(
+                "Subgraph has too many declared data sources and templates",
+            ));
         }
 
         let (schema, data_sources, templates) = try_join3(
@@ -758,17 +754,6 @@ impl<C: Blockchain> UnresolvedSubgraphManifest<C> {
             );
         }
 
-        if spec_version < SPEC_VERSION_0_0_7
-            && data_sources
-                .iter()
-                .any(|ds| OFFCHAIN_KINDS.contains(&ds.kind()))
-        {
-            bail!(
-                "Offchain data sources not supported prior to {}",
-                SPEC_VERSION_0_0_7
-            );
-        }
-
         Ok(SubgraphManifest {
             id,
             spec_version,

graph/src/data/subgraph/schema.rs

@@ -1,6 +1,6 @@
 //! Entity types that contain the graph-node state.
 
-use anyhow::{anyhow, bail, Error};
+use anyhow::{anyhow, Error};
 use hex;
 use lazy_static::lazy_static;
 use rand::rngs::OsRng;
@@ -110,12 +110,11 @@ pub struct DeploymentCreate {
 
 impl DeploymentCreate {
     pub fn new(
-        raw_manifest: String,
         source_manifest: &SubgraphManifest<impl Blockchain>,
         earliest_block: Option<BlockPtr>,
     ) -> Self {
         Self {
-            manifest: SubgraphManifestEntity::new(raw_manifest, source_manifest),
+            manifest: SubgraphManifestEntity::from(source_manifest),
             earliest_block: earliest_block.cheap_clone(),
             graft_base: None,
             graft_block: None,
@@ -164,52 +163,18 @@ pub struct SubgraphManifestEntity {
     pub repository: Option<String>,
     pub features: Vec<String>,
     pub schema: String,
-    pub raw_yaml: Option<String>,
 }
 
-impl SubgraphManifestEntity {
-    pub fn new(raw_yaml: String, manifest: &super::SubgraphManifest<impl Blockchain>) -> Self {
+impl<'a, C: Blockchain> From<&'a super::SubgraphManifest<C>> for SubgraphManifestEntity {
+    fn from(manifest: &'a super::SubgraphManifest<C>) -> Self {
         Self {
             spec_version: manifest.spec_version.to_string(),
             description: manifest.description.clone(),
             repository: manifest.repository.clone(),
             features: manifest.features.iter().map(|f| f.to_string()).collect(),
             schema: manifest.schema.document.clone().to_string(),
-            raw_yaml: Some(raw_yaml),
         }
     }
-
-    pub fn template_idx_and_name(&self) -> Result<Vec<(i32, String)>, Error> {
-        #[derive(Debug, Deserialize)]
-        struct MinimalDs {
-            name: String,
-        }
-        #[derive(Debug, Deserialize)]
-        #[serde(rename_all = "camelCase")]
-        struct MinimalManifest {
-            data_sources: Vec<MinimalDs>,
-            #[serde(default)]
-            templates: Vec<MinimalDs>,
-        }
-
-        let raw_yaml = match &self.raw_yaml {
-            Some(raw_yaml) => raw_yaml,
-            None => bail!("raw_yaml not present"),
-        };
-
-        let manifest: MinimalManifest = serde_yaml::from_str(raw_yaml)?;
-
-        let ds_len = manifest.data_sources.len() as i32;
-        let template_idx_and_name = manifest
-            .templates
-            .iter()
-            .map(|t| t.name.to_owned())
-            .enumerate()
-            .map(move |(idx, name)| (ds_len + idx as i32, name))
-            .collect();
-
-        Ok(template_idx_and_name)
-    }
 }
 
 #[derive(Clone, Debug)]

graph/src/util/error.rs

@@ -17,12 +17,3 @@ macro_rules! ensure {
         }
     };
 }
-
-// `bail!` from `anyhow`, but calling `from`.
-// For context see https://github.com/dtolnay/anyhow/issues/112#issuecomment-704549251.
-#[macro_export]
-macro_rules! bail {
-    ($($err:tt)*) => {
-        return Err(anyhow::anyhow!($($err)*).into());
-    };
-}