all: Defer getting manifest from IPFS when starting subgraph

The current code in `SubgraphAssignmentProvider.start` fetched the manifest
from IPFS before starting a subgraph. But the code calling `start`,
ultimately `SubgraphRegistrar.start_assigned_subgraphs` waited for all
subgraphs to start successfully before processing assignment events.

That could lead to a situation where a slow IPFS server, even if it was
slow for just one subgraph, could keep a node from processing assignment
events.

With these changes, interacting with IPFS is deferred to the future that is
spawned for running the subgraph so that slow IPFS can slow how long it
takes for a subgraph to start, but not the system overall.

Author: lutter

core/src/subgraph/instance_manager.rs

@@ -60,7 +60,6 @@ impl<S: SubgraphStore> SubgraphInstanceManagerTrait for SubgraphInstanceManager<
     async fn start_subgraph(
         self: Arc<Self>,
         loc: DeploymentLocator,
-        manifest: serde_yaml::Mapping,
         stop_block: Option<BlockNumber>,
     ) {
         let runner_index = self.subgraph_start_counter.fetch_add(1, Ordering::SeqCst);
@@ -78,6 +77,19 @@ impl<S: SubgraphStore> SubgraphInstanceManagerTrait for SubgraphInstanceManager<
             let deployment_status_metric = deployment_status_metric.clone();
 
             async move {
+                let link_resolver = self
+                    .link_resolver
+                    .for_manifest(&loc.hash.to_string())
+                    .map_err(SubgraphAssignmentProviderError::ResolveError)?;
+
+                let file_bytes = link_resolver
+                    .cat(&logger, &loc.hash.to_ipfs_link())
+                    .await
+                    .map_err(SubgraphAssignmentProviderError::ResolveError)?;
+
+                let manifest: serde_yaml::Mapping = serde_yaml::from_slice(&file_bytes)
+                    .map_err(|e| SubgraphAssignmentProviderError::ResolveError(e.into()))?;
+
                 match BlockchainKind::from_manifest(&manifest)? {
                     BlockchainKind::Ethereum => {
                         let runner = instance_manager

core/src/subgraph/provider.rs

@@ -44,14 +44,12 @@ impl DeploymentRegistry {
 pub struct SubgraphAssignmentProvider<I> {
     logger_factory: LoggerFactory,
     deployment_registry: DeploymentRegistry,
-    link_resolver: Arc<dyn LinkResolver>,
     instance_manager: Arc<I>,
 }
 
 impl<I: SubgraphInstanceManager> SubgraphAssignmentProvider<I> {
     pub fn new(
         logger_factory: &LoggerFactory,
-        link_resolver: Arc<dyn LinkResolver>,
         instance_manager: I,
         subgraph_metrics: Arc<SubgraphCountMetric>,
     ) -> Self {
@@ -61,7 +59,6 @@ impl<I: SubgraphInstanceManager> SubgraphAssignmentProvider<I> {
         // Create the subgraph provider
         SubgraphAssignmentProvider {
             logger_factory,
-            link_resolver: link_resolver.with_retries().into(),
             instance_manager: Arc::new(instance_manager),
             deployment_registry: DeploymentRegistry::new(subgraph_metrics),
         }
@@ -86,22 +83,9 @@ impl<I: SubgraphInstanceManager> SubgraphAssignmentProviderTrait for SubgraphAss
             ));
         }
 
-        let link_resolver = self
-            .link_resolver
-            .for_manifest(&loc.hash.to_string())
-            .map_err(SubgraphAssignmentProviderError::ResolveError)?;
-
-        let file_bytes = link_resolver
-            .cat(&logger, &loc.hash.to_ipfs_link())
-            .await
-            .map_err(SubgraphAssignmentProviderError::ResolveError)?;
-
-        let raw: serde_yaml::Mapping = serde_yaml::from_slice(&file_bytes)
-            .map_err(|e| SubgraphAssignmentProviderError::ResolveError(e.into()))?;
-
         self.instance_manager
             .cheap_clone()
-            .start_subgraph(loc, raw, stop_block)
+            .start_subgraph(loc, stop_block)
             .await;
 
         Ok(())

graph/src/components/subgraph/instance_manager.rs

@@ -13,7 +13,6 @@ pub trait SubgraphInstanceManager: Send + Sync + 'static {
     async fn start_subgraph(
         self: Arc<Self>,
         deployment: DeploymentLocator,
-        manifest: serde_yaml::Mapping,
         stop_block: Option<BlockNumber>,
     );
     async fn stop_subgraph(&self, deployment: DeploymentLocator);

node/src/launcher.rs

@@ -294,12 +294,8 @@ fn build_subgraph_registrar(
     );
 
     // Create IPFS-based subgraph provider
-    let subgraph_provider = IpfsSubgraphAssignmentProvider::new(
-        &logger_factory,
-        link_resolver.clone(),
-        subgraph_instance_manager,
-        sg_count,
-    );
+    let subgraph_provider =
+        IpfsSubgraphAssignmentProvider::new(&logger_factory, subgraph_instance_manager, sg_count);
 
     // Check version switching mode environment variable
     let version_switching_mode = ENV_VARS.subgraph_version_switching_mode;

node/src/manager/commands/run.rs

@@ -158,7 +158,6 @@ pub async fn run(
     // Create IPFS-based subgraph provider
     let subgraph_provider = Arc::new(IpfsSubgraphAssignmentProvider::new(
         &logger_factory,
-        link_resolver.cheap_clone(),
         subgraph_instance_manager,
         sg_metrics,
     ));

tests/src/fixture/mod.rs

@@ -574,7 +574,6 @@ pub async fn setup_inner<C: Blockchain>(
     // Create IPFS-based subgraph provider
     let subgraph_provider = Arc::new(IpfsSubgraphAssignmentProvider::new(
         &logger_factory,
-        link_resolver.cheap_clone(),
         subgraph_instance_manager.clone(),
         sg_count,
     ));