all: Fold the reachable parts of registrar::start_subgraph into provider

Author: lutter

core/src/subgraph/provider.rs

@@ -1,5 +1,5 @@
-use std::collections::HashSet;
 use std::sync::Mutex;
+use std::{collections::HashSet, time::Instant};
 
 use async_trait::async_trait;
 
@@ -67,28 +67,28 @@ impl<I: SubgraphInstanceManager> SubgraphAssignmentProvider<I> {
 
 #[async_trait]
 impl<I: SubgraphInstanceManager> SubgraphAssignmentProviderTrait for SubgraphAssignmentProvider<I> {
-    async fn start(
-        &self,
-        loc: DeploymentLocator,
-        stop_block: Option<BlockNumber>,
-    ) -> Result<(), SubgraphAssignmentProviderError> {
+    async fn start(&self, loc: DeploymentLocator, stop_block: Option<BlockNumber>) {
         let logger = self.logger_factory.subgraph_logger(&loc);
 
         // If subgraph ID already in set
         if !self.deployment_registry.insert(loc.id) {
             info!(logger, "Subgraph deployment is already running");
 
-            return Err(SubgraphAssignmentProviderError::AlreadyRunning(
-                loc.hash.clone(),
-            ));
+            return;
         }
 
+        let start_time = Instant::now();
+
         self.instance_manager
             .cheap_clone()
             .start_subgraph(loc, stop_block)
             .await;
 
-        Ok(())
+        debug!(
+            logger,
+            "Subgraph started";
+            "start_ms" => start_time.elapsed().as_millis()
+        );
     }
 
     async fn stop(

core/src/subgraph/registrar.rs

@@ -1,5 +1,4 @@
 use std::collections::HashSet;
-use std::time::Instant;
 
 use async_trait::async_trait;
 use graph::blockchain::Blockchain;
@@ -12,7 +11,6 @@ use graph::data::subgraph::schema::DeploymentCreate;
 use graph::data::subgraph::Graft;
 use graph::data::value::Word;
 use graph::futures03;
-use graph::futures03::future::FutureExt;
 use graph::futures03::future::TryFutureExt;
 use graph::futures03::stream;
 use graph::futures03::stream::TryStreamExt;
@@ -238,7 +236,6 @@ where
     }
 
     async fn start_assigned_subgraphs(&self) -> Result<(), Error> {
-        let provider = self.provider.clone();
         let logger = self.logger.clone();
         let node_id = self.node_id.clone();
 
@@ -258,9 +255,12 @@ where
         let (sender, receiver) = futures03::channel::mpsc::channel::<()>(1);
         for id in deployments {
             let sender = sender.clone();
-            let logger = logger.clone();
+            let provider = self.provider.cheap_clone();
 
-            graph::spawn(start_subgraph(id, provider.clone(), logger).map(move |()| drop(sender)));
+            graph::spawn(async move {
+                provider.start(id, None).await;
+                drop(sender)
+            });
         }
         drop(sender);
         let _: Vec<_> = receiver.collect().await;
@@ -473,7 +473,7 @@ async fn handle_assignment_event(
             deployment,
             node_id: _,
         } => {
-            start_subgraph(deployment, provider.clone(), logger).await;
+            provider.start(deployment, None).await;
             Ok(())
         }
         AssignmentEvent::Remove {
@@ -486,39 +486,6 @@ async fn handle_assignment_event(
     }
 }
 
-async fn start_subgraph(
-    deployment: DeploymentLocator,
-    provider: Arc<impl SubgraphAssignmentProviderTrait>,
-    logger: Logger,
-) {
-    let logger = logger
-        .new(o!("subgraph_id" => deployment.hash.to_string(), "sgd" => deployment.id.to_string()));
-
-    trace!(logger, "Start subgraph");
-
-    let start_time = Instant::now();
-    let result = provider.start(deployment.clone(), None).await;
-
-    debug!(
-        logger,
-        "Subgraph started";
-        "start_ms" => start_time.elapsed().as_millis()
-    );
-
-    match result {
-        Ok(()) => (),
-        Err(SubgraphAssignmentProviderError::AlreadyRunning(_)) => (),
-        Err(e) => {
-            // Errors here are likely an issue with the subgraph.
-            error!(
-                logger,
-                "Subgraph instance failed to start";
-                "error" => e.to_string()
-            );
-        }
-    }
-}
-
 /// Resolves the subgraph's earliest block
 async fn resolve_start_block(
     manifest: &SubgraphManifest<impl Blockchain>,

graph/src/components/subgraph/provider.rs

@@ -5,11 +5,7 @@ use crate::{components::store::DeploymentLocator, prelude::*};
 /// Common trait for subgraph providers.
 #[async_trait]
 pub trait SubgraphAssignmentProvider: Send + Sync + 'static {
-    async fn start(
-        &self,
-        deployment: DeploymentLocator,
-        stop_block: Option<BlockNumber>,
-    ) -> Result<(), SubgraphAssignmentProviderError>;
+    async fn start(&self, deployment: DeploymentLocator, stop_block: Option<BlockNumber>);
     async fn stop(
         &self,
         deployment: DeploymentLocator,

node/src/manager/commands/run.rs

@@ -216,8 +216,7 @@ pub async fn run(
 
     let locator = locate(subgraph_store.as_ref(), &hash)?;
 
-    SubgraphAssignmentProvider::start(subgraph_provider.as_ref(), locator, Some(stop_block))
-        .await?;
+    SubgraphAssignmentProvider::start(subgraph_provider.as_ref(), locator, Some(stop_block)).await;
 
     loop {
         tokio::time::sleep(Duration::from_millis(1000)).await;

tests/src/fixture/mod.rs

@@ -286,8 +286,7 @@ impl TestContext {
 
         self.provider
             .start(self.deployment.clone(), Some(stop_block.number))
-            .await
-            .expect("unable to start subgraph");
+            .await;
 
         debug!(self.logger, "TEST: syncing to {}", stop_block.number);
 
@@ -305,10 +304,7 @@ impl TestContext {
         // In case the subgraph has been previously started.
         self.provider.stop(self.deployment.clone()).await.unwrap();
 
-        self.provider
-            .start(self.deployment.clone(), None)
-            .await
-            .expect("unable to start subgraph");
+        self.provider.start(self.deployment.clone(), None).await;
 
         wait_for_sync(
             &self.logger,