offchain: add match_and_decode, persist normally, require source

core/src/subgraph/context.rs

@@ -67,11 +67,7 @@ impl OffchainMonitor {
     }
 
     pub fn add_data_source(&mut self, ds: offchain::DataSource) -> Result<(), Error> {
-        let source = match &ds.source {
-            Some(source) => source,
-            None => anyhow::bail!("Failed to add offchain data source (missing source)"),
-        };
-        match source {
+        match ds.source {
             offchain::Source::Ipfs(cid) => self.ipfs_monitor.monitor(cid.clone()),
         };
         self.data_sources.push(ds);

core/src/subgraph/instance.rs

@@ -76,9 +76,7 @@ where
 
             // Create services for static offchain data sources
             if let DataSource::Offchain(ds) = &ds {
-                if ds.source.is_some() {
-                    offchain_monitor.add_data_source(ds.clone())?;
-                }
+                offchain_monitor.add_data_source(ds.clone())?;
             }
 
             let host = this.new_host(

core/src/subgraph/runner.rs

@@ -16,7 +16,7 @@ use graph::data::subgraph::{
     schema::{SubgraphError, SubgraphHealth, POI_OBJECT},
     SubgraphFeature,
 };
-use graph::data_source::{self, offchain, DataSource, TriggerData};
+use graph::data_source::{offchain, DataSource, TriggerData};
 use graph::prelude::*;
 use graph::util::{backoff::ExponentialBackoff, lfu_cache::LfuCache};
 use std::convert::TryFrom;
@@ -223,18 +223,15 @@ where
             let (data_sources, runtime_hosts) =
                 self.create_dynamic_data_sources(block_state.drain_created_data_sources())?;
 
-            let mut onchain_data_sources = Vec::new();
-            for ds in data_sources {
-                match ds {
-                    DataSource::Onchain(ds) => onchain_data_sources.push(ds),
-                    DataSource::Offchain(ds) => {
-                        self.ctx.offchain_monitor.add_data_source(ds.clone())?
-                    }
+            for ds in &data_sources {
+                if let DataSource::Offchain(ds) = ds {
+                    self.ctx.offchain_monitor.add_data_source(ds.clone())?
                 }
             }
-            let data_sources = onchain_data_sources;
 
-            let filter = C::TriggerFilter::from_data_sources(data_sources.iter());
+            let filter = C::TriggerFilter::from_data_sources(
+                data_sources.iter().filter_map(DataSource::as_onchain),
+            );
 
             // Reprocess the triggers from this block that match the new data sources
             let block_with_triggers = self
@@ -260,10 +257,7 @@ where
 
             // Add entity operations for the new data sources to the block state
             // and add runtimes for the data sources to the subgraph instance.
-            self.persist_dynamic_data_sources(
-                &mut block_state.entity_cache,
-                data_sources.into_iter().map(DataSource::Onchain).collect(),
-            );
+            self.persist_dynamic_data_sources(&mut block_state.entity_cache, data_sources);
 
             // Process the triggers in each host in the same order the
             // corresponding data sources have been created.
@@ -586,30 +580,16 @@ where
         source: offchain::Source,
         data: bytes::Bytes,
     ) -> Result<(), Error> {
-        let matching_data_sources = self
-            .ctx
-            .offchain_monitor
-            .data_sources
-            .iter()
-            .filter(|ds| ds.source.as_ref() == Some(&source));
-
-        let mut block_state =
-            BlockState::<C>::new(self.inputs.store.cheap_clone(), LfuCache::new());
+        let block_state = BlockState::<C>::new(self.inputs.store.cheap_clone(), LfuCache::new());
         let entity_cache = std::mem::take(&mut self.state.entity_lfu_cache);
 
-        self.persist_dynamic_data_sources(
-            &mut block_state.entity_cache,
-            matching_data_sources
-                .cloned()
-                .map(data_source::DataSource::Offchain)
-                .collect(),
-        );
+        // TODO: Set per-file causality region
+        // let causality_region = match &source {
+        //     offchain::Source::Ipfs(cid) => format!("ipfs/{}", cid.to_string()),
+        // };
 
-        let causality_region = match &source {
-            offchain::Source::Ipfs(cid) => format!("ipfs/{}", cid.to_string()),
-        };
         let trigger = TriggerData::Offchain(offchain::TriggerData {
-            source: source.clone(),
+            source,
             data: Arc::new(data),
         });
 
@@ -622,7 +602,7 @@ where
                 &trigger,
                 block_state,
                 &None,
-                &causality_region,
+                "IPFS TODO",
                 &self.inputs.debug_fork,
                 &self.metrics.subgraph,
             )

graph/src/data_source/mod.rs

@@ -103,14 +103,15 @@ impl<C: Blockchain> DataSource<C> {
         block: &Arc<C::Block>,
         logger: &Logger,
     ) -> Result<Option<TriggerWithHandler<MappingTrigger<C>>>, Error> {
-        match self {
-            Self::Onchain(ds) => ds
-                .match_and_decode(trigger.as_onchain().unwrap().clone(), block, logger)
+        match (self, trigger) {
+            (Self::Onchain(ds), TriggerData::Onchain(trigger)) => ds
+                .match_and_decode(trigger, block, logger)
                 .map(|t| t.map(|t| t.map(MappingTrigger::Onchain))),
-            Self::Offchain(ds) => Ok(Some(TriggerWithHandler::new(
-                MappingTrigger::Offchain(trigger.as_offchain().unwrap().clone()),
-                ds.mapping.handler.clone(),
-            ))),
+            (Self::Offchain(ds), TriggerData::Offchain(trigger)) => {
+                Ok(ds.match_and_decode(trigger))
+            }
+            (Self::Onchain(_), TriggerData::Offchain(_))
+            | (Self::Offchain(_), TriggerData::Onchain(_)) => Ok(None),
         }
     }
 
@@ -323,20 +324,6 @@ pub enum TriggerData<C: Blockchain> {
 }
 
 impl<C: Blockchain> TriggerData<C> {
-    fn as_onchain(&self) -> Option<&C::TriggerData> {
-        match self {
-            Self::Onchain(trigger) => Some(trigger),
-            Self::Offchain(_) => None,
-        }
-    }
-
-    fn as_offchain(&self) -> Option<&offchain::TriggerData> {
-        match self {
-            Self::Onchain(_) => None,
-            Self::Offchain(trigger) => Some(trigger),
-        }
-    }
-
     pub fn error_context(&self) -> String {
         match self {
             Self::Onchain(trigger) => trigger.error_context(),

graph/src/data_source/offchain.rs

@@ -9,20 +9,22 @@ use crate::{
     data_source,
     prelude::{DataSourceContext, Link},
 };
-use anyhow::{self, Error};
+use anyhow::{self, Context, Error};
 use cid::Cid;
 use serde::Deserialize;
 use slog::{info, Logger};
 use std::{fmt, sync::Arc};
 
+use super::TriggerWithHandler;
+
 pub const OFFCHAIN_KINDS: &'static [&'static str] = &["file/ipfs"];
 
 #[derive(Clone, Debug)]
 pub struct DataSource {
     pub kind: String,
     pub name: String,
     pub manifest_idx: u32,
-    pub source: Option<Source>,
+    pub source: Source,
     pub mapping: Mapping,
     pub context: Arc<Option<DataSourceContext>>,
     pub creation_block: Option<BlockNumber>,
@@ -46,7 +48,7 @@ impl<C: Blockchain> TryFrom<DataSourceTemplateInfo<C>> for DataSource {
             kind: template.kind.clone(),
             name: template.name.clone(),
             manifest_idx: template.manifest_idx,
-            source: Some(Source::Ipfs(source.parse()?)),
+            source: Source::Ipfs(source.parse()?),
             mapping: template.mapping.clone(),
             context: Arc::new(info.context),
             creation_block: Some(info.creation_block),
@@ -55,18 +57,32 @@ impl<C: Blockchain> TryFrom<DataSourceTemplateInfo<C>> for DataSource {
 }
 
 impl DataSource {
+    pub fn match_and_decode<C: Blockchain>(
+        &self,
+        trigger: &TriggerData,
+    ) -> Option<TriggerWithHandler<super::MappingTrigger<C>>> {
+        if self.source != trigger.source {
+            return None;
+        }
+
+        Some(TriggerWithHandler::new(
+            data_source::MappingTrigger::Offchain(trigger.clone()),
+            self.mapping.handler.clone(),
+        ))
+    }
+
     pub fn as_stored_dynamic_data_source(&self) -> StoredDynamicDataSource {
-        let param = self.source.as_ref().map(|source| match source {
+        let param = match self.source {
             Source::Ipfs(link) => Bytes::from(link.to_bytes()),
-        });
+        };
         let context = self
             .context
             .as_ref()
             .as_ref()
             .map(|ctx| serde_json::to_value(&ctx).unwrap());
         StoredDynamicDataSource {
             manifest_idx: self.manifest_idx,
-            param,
+            param: Some(param),
             context,
             creation_block: self.creation_block,
         }
@@ -76,11 +92,8 @@ impl DataSource {
         template: &DataSourceTemplate,
         stored: StoredDynamicDataSource,
     ) -> Result<Self, Error> {
-        let source = stored.param.and_then(|bytes| {
-            Cid::try_from(bytes.as_slice().to_vec())
-                .ok()
-                .map(Source::Ipfs)
-        });
+        let param = stored.param.context("no param on stored data source")?;
+        let source = Source::Ipfs(Cid::try_from(param.as_slice().to_vec())?);
         let context = Arc::new(stored.context.map(serde_json::from_value).transpose()?);
         Ok(Self {
             kind: template.kind.clone(),
@@ -113,7 +126,7 @@ pub struct Mapping {
 pub struct UnresolvedDataSource {
     pub kind: String,
     pub name: String,
-    pub source: Option<UnresolvedSource>,
+    pub source: UnresolvedSource,
     pub mapping: UnresolvedMapping,
 }
 
@@ -145,10 +158,7 @@ impl UnresolvedDataSource {
             "source" => format_args!("{:?}", &self.source),
         );
         let source = match self.kind.as_str() {
-            "file/ipfs" => self
-                .source
-                .map(|src| src.file.link.parse().map(Source::Ipfs))
-                .transpose()?,
+            "file/ipfs" => Source::Ipfs(self.source.file.link.parse()?),
             _ => {
                 anyhow::bail!(
                     "offchain data source has invalid `kind`, expected `file/ipfs` but found {}",
@@ -194,7 +204,6 @@ pub struct UnresolvedDataSourceTemplate {
     pub mapping: UnresolvedMapping,
 }
 
-
 #[derive(Clone, Debug)]
 pub struct DataSourceTemplate {
     pub kind: String,