Start offchain monitors for static sources

chain/ethereum/tests/manifest.rs

@@ -4,8 +4,7 @@ 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::data_source::{offchain, DataSource};
 use graph::prelude::{
     anyhow, async_trait, serde_yaml, tokio, DeploymentHash, Entity, Link, Logger, SubgraphManifest,
     SubgraphManifestValidationError, UnvalidatedSubgraphManifest,
@@ -28,6 +27,7 @@ const GQL_SCHEMA_FULLTEXT: &str = include_str!("full-text.graphql");
 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 = "{}";
+const FILE_CID: &str = "bafkreigkhuldxkyfkoaye4rgcqcwr45667vkygd45plwq6hawy7j4rbdky";
 
 #[derive(Default, Debug, Clone)]
 struct TextResolver {
@@ -84,7 +84,7 @@ async fn resolve_manifest(
     resolver.add("/ipfs/Qmschema", &GQL_SCHEMA);
     resolver.add("/ipfs/Qmabi", &ABI);
     resolver.add("/ipfs/Qmmapping", &MAPPING_WITH_IPFS_FUNC_WASM);
-    resolver.add("/ipfs/Qmfile", &FILE);
+    resolver.add(FILE_CID, &FILE);
 
     let resolver: Arc<dyn LinkResolverTrait> = Arc::new(resolver);
 
@@ -130,7 +130,8 @@ specVersion: 0.0.2
 
 #[tokio::test]
 async fn ipfs_manifest() {
-    const YAML: &str = "
+    let yaml = format!(
+        "
 schema:
   file:
     /: /ipfs/Qmschema
@@ -139,7 +140,7 @@ dataSources:
     kind: file/ipfs
     source:
       file:
-        /: /ipfs/Qmfile
+        /: {}
     mapping:
       apiVersion: 0.0.6
       language: wasm/assemblyscript
@@ -149,9 +150,11 @@ dataSources:
         /: /ipfs/Qmmapping
       handler: handleFile
 specVersion: 0.0.7
-";
+",
+        FILE_CID
+    );
 
-    let manifest = resolve_manifest(YAML, SPEC_VERSION_0_0_7).await;
+    let manifest = resolve_manifest(&yaml, SPEC_VERSION_0_0_7).await;
 
     assert_eq!("Qmmanifest", manifest.id.as_str());
     assert_eq!(manifest.data_sources.len(), 1);
@@ -161,7 +164,7 @@ specVersion: 0.0.7
     };
     assert_eq!(
         data_source.source,
-        Some(offchain::Source::Ipfs(Link::from("/ipfs/Qmfile")))
+        Some(offchain::Source::Ipfs(FILE_CID.parse().unwrap()))
     );
 }
 

core/src/polling_monitor/mod.rs

@@ -14,7 +14,7 @@ use graph::util::monitored::MonitoredVecDeque as VecDeque;
 use tokio::sync::{mpsc, watch};
 use tower::{Service, ServiceExt};
 
-use self::metrics::PollingMonitorMetrics;
+pub use self::metrics::PollingMonitorMetrics;
 
 /// Spawn a monitor that actively polls a service. Whenever the service has capacity, the monitor
 /// pulls object ids from the queue and polls the service. If the object is not present or in case

core/src/subgraph/context.rs

@@ -1,10 +1,21 @@
-use crate::subgraph::SubgraphInstance;
+use crate::{
+    polling_monitor::{
+        ipfs_service::IpfsService, spawn_monitor, PollingMonitor, PollingMonitorMetrics,
+    },
+    subgraph::SubgraphInstance,
+};
+use bytes::Bytes;
+use cid::Cid;
 use graph::{
     blockchain::Blockchain,
     components::store::DeploymentId,
-    prelude::{CancelGuard, RuntimeHostBuilder},
+    data_source::offchain,
+    env::ENV_VARS,
+    ipfs_client::IpfsClient,
+    prelude::{CancelGuard, DeploymentHash, MetricsRegistry, RuntimeHostBuilder},
+    slog::Logger,
+    tokio::{select, spawn, sync::mpsc},
 };
-
 use std::collections::HashMap;
 use std::sync::{Arc, RwLock};
 
@@ -18,4 +29,59 @@ where
     pub instance: SubgraphInstance<C, T>,
     pub instances: SharedInstanceKeepAliveMap,
     pub filter: C::TriggerFilter,
+    pub offchain_monitor: OffchainMonitor,
+    pub offchain_monitor_rx: mpsc::Receiver<(offchain::Source, Bytes)>,
+}
+
+pub struct OffchainMonitor {
+    ipfs_monitor: PollingMonitor<Cid>,
+}
+
+impl OffchainMonitor {
+    pub fn new(
+        logger: Logger,
+        registry: Arc<dyn MetricsRegistry>,
+        subgraph_hash: &DeploymentHash,
+        client: IpfsClient,
+    ) -> (Self, mpsc::Receiver<(offchain::Source, Bytes)>) {
+        let (ipfs_monitor_tx, mut ipfs_monitor_rx) = mpsc::channel(10);
+        let ipfs_service = IpfsService::new(
+            client,
+            ENV_VARS.mappings.max_ipfs_file_bytes.unwrap_or(1 << 20) as u64,
+            ENV_VARS.mappings.ipfs_timeout,
+            10,
+        );
+        let ipfs_monitor = spawn_monitor(
+            ipfs_service,
+            ipfs_monitor_tx,
+            logger,
+            PollingMonitorMetrics::new(registry, subgraph_hash),
+        );
+        let (tx, rx) = mpsc::channel(10);
+        spawn(async move {
+            loop {
+                select! {
+                    msg = ipfs_monitor_rx.recv() => {
+                        match msg {
+                            Some((cid, bytes)) => {
+                                if let Err(_) = tx.send((offchain::Source::Ipfs(cid), bytes)).await {
+                                    break;
+                                }
+                            },
+                            None => break,
+                        }
+                    }
+                }
+            }
+        });
+        (Self { ipfs_monitor }, rx)
+    }
+
+    pub fn monitor(&self, source: offchain::Source) {
+        match source {
+            offchain::Source::Ipfs(cid) => {
+                self.ipfs_monitor.monitor(cid);
+            }
+        }
+    }
 }

core/src/subgraph/instance.rs

@@ -11,6 +11,8 @@ use graph::{
 };
 use std::collections::HashMap;
 
+use super::context::OffchainMonitor;
+
 pub struct SubgraphInstance<C: Blockchain, T: RuntimeHostBuilder<C>> {
     subgraph_id: DeploymentHash,
     network: String,
@@ -40,6 +42,7 @@ where
         host_builder: T,
         trigger_processor: Box<dyn TriggerProcessor<C, T>>,
         host_metrics: Arc<HostMetrics>,
+        offchain_monitor: &OffchainMonitor,
     ) -> Result<Self, Error> {
         let subgraph_id = manifest.id.clone();
         let network = manifest.network_name();
@@ -71,14 +74,21 @@ where
                 Some(ref module_bytes) => module_bytes,
             };
 
+            // Create services for static offchain data sources
+            if let DataSource::Offchain(ds) = &ds {
+                if let Some(source) = &ds.source {
+                    offchain_monitor.monitor(source.clone());
+                }
+            }
+
             let host = this.new_host(
                 logger.cheap_clone(),
                 ds,
                 module_bytes,
                 templates.cheap_clone(),
                 host_metrics.cheap_clone(),
             )?;
-            this.hosts.push(Arc::new(host))
+            this.hosts.push(Arc::new(host));
         }
 
         Ok(this)

core/src/subgraph/instance_manager.rs

@@ -7,11 +7,13 @@ use graph::blockchain::block_stream::BlockStreamMetrics;
 use graph::blockchain::Blockchain;
 use graph::blockchain::NodeCapabilities;
 use graph::blockchain::{BlockchainKind, TriggerFilter};
+use graph::ipfs_client::IpfsClient;
 use graph::prelude::{SubgraphInstanceManager as SubgraphInstanceManagerTrait, *};
 use graph::{blockchain::BlockchainMap, components::store::DeploymentLocator};
 use graph_runtime_wasm::RuntimeHostBuilder;
 use tokio::task;
 
+use super::context::OffchainMonitor;
 use super::SubgraphTriggerProcessor;
 
 pub struct SubgraphInstanceManager<S: SubgraphStore> {
@@ -22,6 +24,7 @@ pub struct SubgraphInstanceManager<S: SubgraphStore> {
     manager_metrics: SubgraphInstanceManagerMetrics,
     instances: SharedInstanceKeepAliveMap,
     link_resolver: Arc<dyn LinkResolver>,
+    ipfs_client: IpfsClient,
     static_filters: bool,
 }
 
@@ -133,6 +136,7 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
         chains: Arc<BlockchainMap>,
         metrics_registry: Arc<dyn MetricsRegistry>,
         link_resolver: Arc<dyn LinkResolver>,
+        ipfs_client: IpfsClient,
         static_filters: bool,
     ) -> Self {
         let logger = logger_factory.component_logger("SubgraphInstanceManager", None);
@@ -146,6 +150,7 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
             metrics_registry,
             instances: SharedInstanceKeepAliveMap::default(),
             link_resolver,
+            ipfs_client,
             static_filters,
         }
     }
@@ -290,6 +295,13 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
             stopwatch_metrics,
         ));
 
+        let (offchain_monitor, offchain_monitor_rx) = OffchainMonitor::new(
+            logger.cheap_clone(),
+            registry.cheap_clone(),
+            &manifest.id,
+            self.ipfs_client.cheap_clone(),
+        );
+
         // Initialize deployment_head with current deployment head. Any sort of trouble in
         // getting the deployment head ptr leads to initializing with 0
         let deployment_head = store.block_ptr().map(|ptr| ptr.number).unwrap_or(0) as f64;
@@ -309,6 +321,7 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
             host_builder,
             tp,
             host_metrics.clone(),
+            &offchain_monitor,
         )?;
 
         let inputs = IndexingInputs {
@@ -331,6 +344,8 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
             instance,
             instances: self.instances.cheap_clone(),
             filter,
+            offchain_monitor,
+            offchain_monitor_rx,
         };
 
         let metrics = RunnerMetrics {

graph/Cargo.toml

@@ -10,6 +10,7 @@ async-stream = "0.3"
 atomic_refcell = "0.1.8"
 bigdecimal = { version = "0.1.0", features = ["serde"] }
 bytes = "1.0.1"
+cid = "0.8.3"
 diesel = { version = "1.4.8", features = ["postgres", "serde_json", "numeric", "r2d2", "chrono"] }
 diesel_derives = "1.4"
 chrono = "0.4.22"
@@ -69,4 +70,4 @@ maplit = "1.0.2"
 structopt = { version = "0.3" }
 
 [build-dependencies]
-tonic-build = { version = "0.7.2", features = ["prost"] }
+tonic-build = { version = "0.7.2", features = ["prost"] }

graph/src/data/subgraph/mod.rs

@@ -38,9 +38,9 @@ use crate::{
         subgraph::features::validate_subgraph_features,
     },
     data_source::{
-        DataSource, DataSourceTemplate, UnresolvedDataSource, UnresolvedDataSourceTemplate,
+        offchain::OFFCHAIN_KINDS, DataSource, DataSourceTemplate, UnresolvedDataSource,
+        UnresolvedDataSourceTemplate,
     },
-    offchain::OFFCHAIN_KINDS,
     prelude::{r, CheapClone, ENV_VARS},
 };
 

graph/src/data_source.rs → graph/src/data_source/mod.rs

@@ -1,3 +1,5 @@
+pub mod offchain;
+
 use crate::{
     blockchain::{
         Blockchain, DataSource as _, DataSourceTemplate as _, UnresolvedDataSource as _,
@@ -8,7 +10,7 @@ use crate::{
         store::{BlockNumber, StoredDynamicDataSource},
         subgraph::DataSourceTemplateInfo,
     },
-    offchain::{self, OFFCHAIN_KINDS},
+    data_source::offchain::OFFCHAIN_KINDS,
     prelude::{CheapClone as _, DataSourceContext},
 };
 use anyhow::Error;

graph/src/offchain/mod.rs → graph/src/data_source/offchain.rs

@@ -9,8 +9,8 @@ use crate::{
     data_source,
     prelude::{DataSourceContext, Link},
 };
-
 use anyhow::{self, Error};
+use cid::Cid;
 use serde::Deserialize;
 use slog::{info, Logger};
 use std::sync::Arc;
@@ -46,7 +46,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(Link::from(source))),
+            source: Some(Source::Ipfs(source.parse()?)),
             mapping: template.mapping.clone(),
             context: Arc::new(info.context),
             creation_block: Some(info.creation_block),
@@ -57,7 +57,7 @@ impl<C: Blockchain> TryFrom<DataSourceTemplateInfo<C>> for DataSource {
 impl DataSource {
     pub fn as_stored_dynamic_data_source(&self) -> StoredDynamicDataSource {
         let param = self.source.as_ref().map(|source| match source {
-            Source::Ipfs(link) => Bytes::from(link.link.as_bytes()),
+            Source::Ipfs(link) => Bytes::from(link.to_bytes()),
         });
         let context = self
             .context
@@ -77,9 +77,9 @@ impl DataSource {
         stored: StoredDynamicDataSource,
     ) -> Result<Self, Error> {
         let source = stored.param.and_then(|bytes| {
-            String::from_utf8(bytes.as_slice().to_vec())
+            Cid::try_from(bytes.as_slice().to_vec())
                 .ok()
-                .map(|link| Source::Ipfs(Link::from(link)))
+                .map(Source::Ipfs)
         });
         let context = Arc::new(stored.context.map(serde_json::from_value).transpose()?);
         Ok(Self {
@@ -96,7 +96,7 @@ impl DataSource {
 
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub enum Source {
-    Ipfs(Link),
+    Ipfs(Cid),
 }
 
 #[derive(Clone, Debug)]
@@ -145,7 +145,10 @@ impl UnresolvedDataSource {
             "source" => format_args!("{:?}", &self.source),
         );
         let source = match self.kind.as_str() {
-            "file/ipfs" => self.source.map(|src| Source::Ipfs(src.file)),
+            "file/ipfs" => self
+                .source
+                .map(|src| src.file.link.parse().map(Source::Ipfs))
+                .transpose()?,
             _ => {
                 anyhow::bail!(
                     "offchain data source has invalid `kind`, expected `file/ipfs` but found {}",

graph/src/lib.rs

@@ -22,8 +22,6 @@ pub mod data_source;
 
 pub mod blockchain;
 
-pub mod offchain;
-
 pub mod runtime;
 
 pub mod firehose;