graph,store,core - add active_assigments() for unpaused subgraphs and wire it in subgraph startup

Author: incrypto32

core/src/subgraph/registrar.rs

@@ -226,7 +226,7 @@ where
         let logger = self.logger.clone();
         let node_id = self.node_id.clone();
 
-        future::result(self.store.assignments(&self.node_id))
+        future::result(self.store.active_assignments(&self.node_id))
             .map_err(|e| anyhow!("Error querying subgraph assignments: {}", e))
             .and_then(move |deployments| {
                 // This operation should finish only after all subgraphs are

graph/src/components/store/traits.rs

@@ -106,6 +106,9 @@ pub trait SubgraphStore: Send + Sync + 'static {
 
     fn assignments(&self, node: &NodeId) -> Result<Vec<DeploymentLocator>, StoreError>;
 
+    /// Returns assignments that are not paused
+    fn active_assignments(&self, node: &NodeId) -> Result<Vec<DeploymentLocator>, StoreError>;
+
     /// Return `true` if a subgraph `name` exists, regardless of whether the
     /// subgraph has any deployments attached to it
     fn subgraph_exists(&self, name: &SubgraphName) -> Result<bool, StoreError>;

node/src/manager/commands/assign.rs

@@ -74,7 +74,7 @@ pub fn pause_or_resume(
     primary: ConnectionPool,
     sender: &NotificationSender,
     search: &DeploymentSearch,
-    pause: bool,
+    should_pause: bool,
 ) -> Result<(), Error> {
     let locator = search.locate_unique(&primary)?;
 
@@ -86,18 +86,22 @@ pub fn pause_or_resume(
         .ok_or_else(|| anyhow!("failed to locate site for {locator}"))?;
 
     let change = match conn.assignment_status(&site)? {
-        Some((_, paused)) => {
-            if paused == pause {
-                println!("deployment {locator} is already {paused}");
-                vec![]
-            } else {
+        Some((_, is_paused)) => {
+            if should_pause {
+                if is_paused {
+                    println!("deployment {locator} is already paused");
+                    return Ok(());
+                }
                 println!("pausing {locator}");
                 conn.pause_subgraph(&site)?
+            } else {
+                println!("resuming {locator}");
+                conn.resume_subgraph(&site)?
             }
         }
         None => {
-            println!("resuming {locator}");
-            conn.resume_subgraph(&site)?
+            println!("deployment {locator} not found");
+            return Ok(());
         }
     };
     println!("Operation completed");

store/postgres/src/primary.rs

@@ -588,6 +588,22 @@ mod queries {
             .collect::<Result<Vec<Site>, _>>()
     }
 
+    // All assignments for a node that are currently not paused
+    pub(super) fn active_assignments(
+        conn: &PgConnection,
+        node: &NodeId,
+    ) -> Result<Vec<Site>, StoreError> {
+        ds::table
+            .inner_join(a::table.on(a::id.eq(ds::id)))
+            .filter(a::node_id.eq(node.as_str()))
+            .filter(a::paused_at.is_null())
+            .select(ds::all_columns)
+            .load::<Schema>(conn)?
+            .into_iter()
+            .map(Site::try_from)
+            .collect::<Result<Vec<Site>, _>>()
+    }
+
     pub(super) fn fill_assignments(
         conn: &PgConnection,
         infos: &mut [status::Info],
@@ -1803,6 +1819,10 @@ impl Mirror {
         self.read(|conn| queries::assignments(conn, node))
     }
 
+    pub fn active_assignments(&self, node: &NodeId) -> Result<Vec<Site>, StoreError> {
+        self.read(|conn| queries::active_assignments(conn, node))
+    }
+
     pub fn assigned_node(&self, site: &Site) -> Result<Option<NodeId>, StoreError> {
         self.read(|conn| queries::assigned_node(conn, site))
     }

store/postgres/src/subgraph_store.rs

@@ -1304,6 +1304,12 @@ impl SubgraphStoreTrait for SubgraphStore {
             .map(|sites| sites.iter().map(|site| site.into()).collect())
     }
 
+    fn active_assignments(&self, node: &NodeId) -> Result<Vec<DeploymentLocator>, StoreError> {
+        self.mirror
+            .active_assignments(node)
+            .map(|sites| sites.iter().map(|site| site.into()).collect())
+    }
+
     fn subgraph_exists(&self, name: &SubgraphName) -> Result<bool, StoreError> {
         self.mirror.subgraph_exists(name)
     }