Subgraph Compositions: Validations

graph/src/data/subgraph/mod.rs

@@ -577,7 +577,7 @@ pub struct BaseSubgraphManifest<C, S, D, T> {
 #[derive(Debug, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct IndexerHints {
-    prune: Option<Prune>,
+    pub prune: Option<Prune>,
 }
 
 impl IndexerHints {

graph/src/data_source/mod.rs

@@ -339,7 +339,7 @@ impl<C: Blockchain> UnresolvedDataSource<C> {
                 .await
                 .map(DataSource::Onchain),
             Self::Subgraph(unresolved) => unresolved
-                .resolve(resolver, logger, manifest_idx)
+                .resolve::<C>(resolver, logger, manifest_idx)
                 .await
                 .map(DataSource::Subgraph),
             Self::Offchain(_unresolved) => {

graph/src/data_source/subgraph.rs

@@ -2,12 +2,16 @@ use crate::{
     blockchain::{block_stream::EntitySourceOperation, Block, Blockchain},
     components::{link_resolver::LinkResolver, store::BlockNumber},
     data::{
-        subgraph::{calls_host_fn, SPEC_VERSION_1_3_0},
+        subgraph::{
+            calls_host_fn, SubgraphManifest, UnresolvedSubgraphManifest, LATEST_VERSION,
+            SPEC_VERSION_1_3_0,
+        },
         value::Word,
     },
     data_source::{self, common::DeclaredCall},
     ensure,
     prelude::{CheapClone, DataSourceContext, DeploymentHash, Link},
+    schema::TypeKind,
 };
 use anyhow::{anyhow, Context, Error, Result};
 use futures03::{stream::FuturesOrdered, TryStreamExt};
@@ -211,8 +215,62 @@ pub struct UnresolvedMapping {
 }
 
 impl UnresolvedDataSource {
+    fn validate_mapping_entities<C: Blockchain>(
+        mapping_entities: &[String],
+        source_manifest: &SubgraphManifest<C>,
+    ) -> Result<(), Error> {
+        for entity in mapping_entities {
+            let type_kind = source_manifest.schema.kind_of_declared_type(&entity);
+
+            match type_kind {
+                Some(TypeKind::Interface) => {
+                    return Err(anyhow!(
+                        "Entity {} is an interface and cannot be used as a mapping entity",
+                        entity
+                    ));
+                }
+                Some(TypeKind::Aggregation) => {
+                    return Err(anyhow!(
+                        "Entity {} is an aggregation and cannot be used as a mapping entity",
+                        entity
+                    ));
+                }
+                None => {
+                    return Err(anyhow!("Entity {} not found in source manifest", entity));
+                }
+                Some(TypeKind::Object) => {}
+            }
+        }
+        Ok(())
+    }
+
+    async fn resolve_source_manifest<C: Blockchain>(
+        &self,
+        resolver: &Arc<dyn LinkResolver>,
+        logger: &Logger,
+    ) -> Result<Arc<SubgraphManifest<C>>, Error> {
+        let source_raw = resolver
+            .cat(logger, &self.source.address.to_ipfs_link())
+            .await
+            .context("Failed to resolve source subgraph manifest")?;
+
+        let source_raw: serde_yaml::Mapping = serde_yaml::from_slice(&source_raw)
+            .context("Failed to parse source subgraph manifest as YAML")?;
+
+        let deployment_hash = self.source.address.clone();
+
+        let source_manifest = UnresolvedSubgraphManifest::<C>::parse(deployment_hash, source_raw)
+            .context("Failed to parse source subgraph manifest")?;
+
+        source_manifest
+            .resolve(resolver, logger, LATEST_VERSION.clone())
+            .await
+            .context("Failed to resolve source subgraph manifest")
+            .map(Arc::new)
+    }
+
     #[allow(dead_code)]
-    pub(super) async fn resolve(
+    pub(super) async fn resolve<C: Blockchain>(
         self,
         resolver: &Arc<dyn LinkResolver>,
         logger: &Logger,
@@ -224,7 +282,38 @@ impl UnresolvedDataSource {
             "source" => format_args!("{:?}", &self.source),
         );
 
-        let kind = self.kind;
+        let kind = self.kind.clone();
+        let source_manifest = self.resolve_source_manifest::<C>(resolver, logger).await?;
+        let source_spec_version = &source_manifest.spec_version;
+
+        if source_spec_version < &SPEC_VERSION_1_3_0 {
+            return Err(anyhow!(
+                "Source subgraph manifest spec version {} is not supported, minimum supported version is {}",
+                source_spec_version,
+                SPEC_VERSION_1_3_0
+            ));
+        }
+
+        let pruning_enabled = match source_manifest.indexer_hints.as_ref() {
+            None => false,
+            Some(hints) => hints.prune.is_some(),
+        };
+
+        if pruning_enabled {
+            return Err(anyhow!(
+                "Pruning is enabled for source subgraph, which is not supported"
+            ));
+        }
+
+        let mapping_entities: Vec<String> = self
+            .mapping
+            .handlers
+            .iter()
+            .map(|handler| handler.entity.clone())
+            .collect();
+
+        Self::validate_mapping_entities(&mapping_entities, &source_manifest)?;
+
         let source = Source {
             address: self.source.address,
             start_block: self.source.start_block,

store/test-store/tests/chain/ethereum/manifest.rs

@@ -37,6 +37,32 @@ const GQL_SCHEMA: &str = r#"
   type TestEntity @entity { id: ID! }
 "#;
 const GQL_SCHEMA_FULLTEXT: &str = include_str!("full-text.graphql");
+const SOURCE_SUBGRAPH_MANIFEST: &str = "
+dataSources: []
+schema:
+  file:
+    /: /ipfs/QmSourceSchema
+specVersion: 1.3.0
+";
+
+const SOURCE_SUBGRAPH_SCHEMA: &str = "
+type TestEntity @entity { id: ID! }
+type User @entity { id: ID! }
+type Profile @entity { id: ID! }
+
+type TokenData @entity(timeseries: true) {
+    id: Int8!
+    timestamp: Timestamp!
+    amount: BigDecimal!
+}
+
+type TokenStats @aggregation(intervals: [\"hour\", \"day\"], source: \"TokenData\") {
+    id: Int8!
+    timestamp: Timestamp!
+    totalAmount: BigDecimal! @aggregate(fn: \"sum\", arg: \"amount\")
+}
+";
+
 const MAPPING_WITH_IPFS_FUNC_WASM: &[u8] = include_bytes!("ipfs-on-ethereum-contracts.wasm");
 const ABI: &str = "[{\"type\":\"function\", \"inputs\": [{\"name\": \"i\",\"type\": \"uint256\"}],\"name\":\"get\",\"outputs\": [{\"type\": \"address\",\"name\": \"o\"}]}]";
 const FILE: &str = "{}";
@@ -83,23 +109,33 @@ impl LinkResolverTrait for TextResolver {
     }
 }
 
-async fn resolve_manifest(
+async fn try_resolve_manifest(
     text: &str,
     max_spec_version: Version,
-) -> SubgraphManifest<graph_chain_ethereum::Chain> {
+) -> Result<SubgraphManifest<graph_chain_ethereum::Chain>, anyhow::Error> {
     let mut resolver = TextResolver::default();
     let id = DeploymentHash::new("Qmmanifest").unwrap();
 
     resolver.add(id.as_str(), &text);
     resolver.add("/ipfs/Qmschema", &GQL_SCHEMA);
     resolver.add("/ipfs/Qmabi", &ABI);
     resolver.add("/ipfs/Qmmapping", &MAPPING_WITH_IPFS_FUNC_WASM);
+    resolver.add("/ipfs/QmSource", &SOURCE_SUBGRAPH_MANIFEST);
+    resolver.add("/ipfs/QmSource2", &SOURCE_SUBGRAPH_MANIFEST);
+    resolver.add("/ipfs/QmSourceSchema", &SOURCE_SUBGRAPH_SCHEMA);
     resolver.add(FILE_CID, &FILE);
 
     let resolver: Arc<dyn LinkResolverTrait> = Arc::new(resolver);
 
-    let raw = serde_yaml::from_str(text).unwrap();
-    SubgraphManifest::resolve_from_raw(id, raw, &resolver, &LOGGER, max_spec_version)
+    let raw = serde_yaml::from_str(text)?;
+    Ok(SubgraphManifest::resolve_from_raw(id, raw, &resolver, &LOGGER, max_spec_version).await?)
+}
+
+async fn resolve_manifest(
+    text: &str,
+    max_spec_version: Version,
+) -> SubgraphManifest<graph_chain_ethereum::Chain> {
+    try_resolve_manifest(text, max_spec_version)
         .await
         .expect("Parsing simple manifest works")
 }
@@ -184,7 +220,7 @@ dataSources:
         - Gravatar
     network: mainnet
     source: 
-      address: 'QmSWWT2yrTFDZSL8tRyoHEVrcEKAUsY2hj2TMQDfdDZU8h'
+      address: 'QmSource'
       startBlock: 9562480
     mapping:
       apiVersion: 0.0.6
@@ -195,7 +231,7 @@ dataSources:
         /: /ipfs/Qmmapping
       handlers:
         - handler: handleEntity
-          entity: User
+          entity: TestEntity
 specVersion: 1.3.0
 ";
 
@@ -214,6 +250,42 @@ specVersion: 1.3.0
     }
 }
 
+#[tokio::test]
+async fn subgraph_ds_manifest_aggregations_should_fail() {
+    let yaml = "
+schema:
+  file:
+    /: /ipfs/Qmschema
+dataSources:
+  - name: SubgraphSource
+    kind: subgraph
+    entities:
+        - Gravatar
+    network: mainnet
+    source: 
+      address: 'QmSource'
+      startBlock: 9562480
+    mapping:
+      apiVersion: 0.0.6
+      language: wasm/assemblyscript
+      entities:
+        - TestEntity
+      file:
+        /: /ipfs/Qmmapping
+      handlers:
+        - handler: handleEntity
+          entity: TokenStats # This is an aggregation and should fail
+specVersion: 1.3.0
+";
+
+    let result = try_resolve_manifest(yaml, SPEC_VERSION_1_3_0).await;
+    assert!(result.is_err());
+    let err = result.unwrap_err();
+    assert!(err
+        .to_string()
+        .contains("Entity TokenStats is an aggregation and cannot be used as a mapping entity"));
+}
+
 #[tokio::test]
 async fn graft_manifest() {
     const YAML: &str = "
@@ -1506,7 +1578,7 @@ dataSources:
         - Gravatar
     network: mainnet
     source: 
-      address: 'QmSWWT2yrTFDZSL8tRyoHEVrcEKAUsY2hj2TMQDfdDZU8h'
+      address: 'QmSource'
       startBlock: 9562480
     mapping:
       apiVersion: 0.0.6
@@ -1537,6 +1609,8 @@ dataSources:
             resolver.add("/ipfs/Qmabi", &ABI);
             resolver.add("/ipfs/Qmschema", &GQL_SCHEMA);
             resolver.add("/ipfs/Qmmapping", &MAPPING_WITH_IPFS_FUNC_WASM);
+            resolver.add("/ipfs/QmSource", &SOURCE_SUBGRAPH_MANIFEST);
+            resolver.add("/ipfs/QmSourceSchema", &SOURCE_SUBGRAPH_SCHEMA);
 
             let resolver: Arc<dyn LinkResolverTrait> = Arc::new(resolver);
 

tests/integration-tests/source-subgraph/subgraph.yaml

@@ -1,4 +1,4 @@
-specVersion: 0.0.8
+specVersion: 1.3.0
 schema:
   file: ./schema.graphql
 dataSources:

tests/integration-tests/subgraph-data-sources/subgraph.yaml

@@ -6,7 +6,7 @@ dataSources:
     name: Contract
     network: test
     source:
-      address: 'Qmaqf8cRxfxbduZppSHKG9DMuX5JZPMoGuwGb2DQuo48sq'
+      address: 'QmaKaj4gCYo4TmGq27tgqwrsBLwNncHGvR6Q9e6wDBYo8M'
       startBlock: 0
     mapping:
       apiVersion: 0.0.7