Refactor manifest data sources

chain/ethereum/src/chain.rs

@@ -245,6 +245,7 @@ impl Chain {
 #[async_trait]
 impl Blockchain for Chain {
     const KIND: BlockchainKind = BlockchainKind::Ethereum;
+    const ALIASES: &'static [&'static str] = &["ethereum/contract"];
 
     type Block = BlockFinality;
 

chain/ethereum/tests/manifest.rs

@@ -4,6 +4,8 @@ use std::time::Duration;
 
 use graph::data::subgraph::schema::SubgraphError;
 use graph::data::subgraph::{SPEC_VERSION_0_0_4, SPEC_VERSION_0_0_7};
+use graph::data_source::DataSource;
+use graph::offchain;
 use graph::prelude::{
     anyhow, async_trait, serde_yaml, tokio, DeploymentHash, Entity, Link, Logger, SubgraphManifest,
     SubgraphManifestValidationError, UnvalidatedSubgraphManifest,
@@ -152,8 +154,15 @@ specVersion: 0.0.7
     let manifest = resolve_manifest(YAML, SPEC_VERSION_0_0_7).await;
 
     assert_eq!("Qmmanifest", manifest.id.as_str());
-    assert_eq!(manifest.offchain_data_sources.len(), 1);
-    assert_eq!(manifest.data_sources.len(), 0);
+    assert_eq!(manifest.data_sources.len(), 1);
+    let data_source = match &manifest.data_sources[0] {
+        DataSource::Offchain(ds) => ds,
+        DataSource::Onchain(_) => unreachable!(),
+    };
+    assert_eq!(
+        data_source.source,
+        Some(offchain::Source::Ipfs(Link::from("/ipfs/Qmfile")))
+    );
 }
 
 #[tokio::test]
@@ -392,7 +401,12 @@ specVersion: 0.0.2
 ";
 
     let manifest = resolve_manifest(YAML, SPEC_VERSION_0_0_4).await;
-    let required_capabilities = NodeCapabilities::from_data_sources(&manifest.data_sources);
+    let onchain_data_sources = manifest
+        .data_sources
+        .iter()
+        .filter_map(|ds| ds.as_onchain().cloned())
+        .collect::<Vec<_>>();
+    let required_capabilities = NodeCapabilities::from_data_sources(&onchain_data_sources);
 
     assert_eq!("Qmmanifest", manifest.id.as_str());
     assert_eq!(true, required_capabilities.traces);

core/src/subgraph/instance.rs

@@ -1,18 +1,15 @@
 use futures01::sync::mpsc::Sender;
-use graph::components::subgraph::ProofOfIndexingVersion;
-use graph::data::subgraph::SPEC_VERSION_0_0_6;
-
-use std::collections::HashMap;
-
 use graph::{
     blockchain::Blockchain,
     components::{
         store::SubgraphFork,
-        subgraph::{MappingError, SharedProofOfIndexing},
+        subgraph::{MappingError, ProofOfIndexingVersion, SharedProofOfIndexing},
     },
-    prelude::ENV_VARS,
+    data::subgraph::SPEC_VERSION_0_0_6,
+    data_source::DataSource,
+    prelude::*,
 };
-use graph::{blockchain::DataSource, prelude::*};
+use std::collections::HashMap;
 
 pub struct SubgraphInstance<C: Blockchain, T: RuntimeHostBuilder<C>> {
     subgraph_id: DeploymentHash,
@@ -92,7 +89,7 @@ where
     fn new_host(
         &mut self,
         logger: Logger,
-        data_source: C::DataSource,
+        data_source: DataSource<C>,
         module_bytes: &Arc<Vec<u8>>,
         templates: Arc<Vec<C::DataSourceTemplate>>,
         host_metrics: Arc<HostMetrics>,
@@ -151,7 +148,7 @@ where
     pub(crate) fn add_dynamic_data_source(
         &mut self,
         logger: &Logger,
-        data_source: C::DataSource,
+        data_source: DataSource<C>,
         templates: Arc<Vec<C::DataSourceTemplate>>,
         metrics: Arc<HostMetrics>,
     ) -> Result<Option<Arc<T::Host>>, Error> {

core/src/subgraph/instance_manager.rs

@@ -220,7 +220,12 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
             (manifest, manifest_idx_and_name)
         };
 
-        let required_capabilities = C::NodeCapabilities::from_data_sources(&manifest.data_sources);
+        let onchain_data_sources = manifest
+            .data_sources
+            .iter()
+            .filter_map(|d| d.as_onchain().cloned())
+            .collect::<Vec<_>>();
+        let required_capabilities = C::NodeCapabilities::from_data_sources(&onchain_data_sources);
         let network = manifest.network_name();
 
         let chain = self
@@ -230,7 +235,7 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
             .clone();
 
         // Obtain filters from the manifest
-        let mut filter = C::TriggerFilter::from_data_sources(manifest.data_sources.iter());
+        let mut filter = C::TriggerFilter::from_data_sources(onchain_data_sources.iter());
 
         if self.static_filters {
             filter.extend_with_template(manifest.templates.clone().into_iter());

core/src/subgraph/loader.rs

@@ -1,18 +1,19 @@
 use std::time::Instant;
 
-use graph::blockchain::{Blockchain, DataSource, DataSourceTemplate as _};
+use graph::blockchain::{Blockchain, DataSource as _, DataSourceTemplate as _};
 use graph::components::store::WritableStore;
+use graph::data_source::DataSource;
 use graph::prelude::*;
 
 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<C::DataSource>, Error> {
+) -> Result<Vec<DataSource<C>>, Error> {
     let start_time = Instant::now();
 
-    let mut data_sources: Vec<C::DataSource> = vec![];
+    let mut data_sources: Vec<DataSource<C>> = vec![];
 
     for stored in store
         .load_dynamic_data_sources(manifest_idx_and_name)
@@ -33,7 +34,7 @@ pub async fn load_dynamic_data_sources<C: Blockchain>(
             "Assertion failure: new data source has lower creation block than existing ones"
         );
 
-        data_sources.push(ds);
+        data_sources.push(DataSource::Onchain(ds));
     }
 
     trace!(

core/src/subgraph/runner.rs

@@ -5,7 +5,7 @@ use crate::subgraph::state::IndexingState;
 use crate::subgraph::stream::new_block_stream;
 use atomic_refcell::AtomicRefCell;
 use graph::blockchain::block_stream::{BlockStreamEvent, BlockWithTriggers, FirehoseCursor};
-use graph::blockchain::{Block, Blockchain, DataSource, TriggerFilter as _};
+use graph::blockchain::{Block, Blockchain, DataSource as _, TriggerFilter as _};
 use graph::components::store::EntityKey;
 use graph::components::{
     store::ModificationsAndCache,
@@ -16,6 +16,7 @@ use graph::data::subgraph::{
     schema::{SubgraphError, SubgraphHealth, POI_OBJECT},
     SubgraphFeature,
 };
+use graph::data_source::DataSource;
 use graph::prelude::*;
 use graph::util::{backoff::ExponentialBackoff, lfu_cache::LfuCache};
 use std::convert::TryFrom;
@@ -465,7 +466,7 @@ where
             // Try to create a runtime host for the data source
             let host = self.ctx.instance.add_dynamic_data_source(
                 &self.logger,
-                data_source.clone(),
+                DataSource::Onchain(data_source.clone()),
                 self.inputs.templates.clone(),
                 self.metrics.host.clone(),
             )?;

graph/src/blockchain/mod.rs

@@ -81,6 +81,7 @@ pub trait Block: Send + Sync {
 // This is only `Debug` because some tests require that
 pub trait Blockchain: Debug + Sized + Send + Sync + Unpin + 'static {
     const KIND: BlockchainKind;
+    const ALIASES: &'static [&'static str] = &[];
 
     // The `Clone` bound is used when reprocessing a block, because `triggers_in_block` requires an
     // owned `Block`. It would be good to come up with a way to remove this bound.

graph/src/components/subgraph/host.rs

@@ -8,6 +8,7 @@ use futures::sync::mpsc;
 
 use crate::blockchain::TriggerWithHandler;
 use crate::components::store::SubgraphFork;
+use crate::data_source::DataSource;
 use crate::prelude::*;
 use crate::{blockchain::Blockchain, components::subgraph::SharedProofOfIndexing};
 use crate::{components::metrics::HistogramVec, runtime::DeterministicHostError};
@@ -151,7 +152,7 @@ pub trait RuntimeHostBuilder<C: Blockchain>: Clone + Send + Sync + 'static {
         &self,
         network_name: String,
         subgraph_id: DeploymentHash,
-        data_source: C::DataSource,
+        data_source: DataSource<C>,
         top_level_templates: Arc<Vec<C::DataSourceTemplate>>,
         mapping_request_sender: mpsc::Sender<Self::Req>,
         metrics: Arc<HostMetrics>,