core,graph,node,store: send StoreEvents for Pause and Resume and handle it

Author: incrypto32

core/src/subgraph/registrar.rs

@@ -171,13 +171,20 @@ where
                     match operation {
                         EntityChangeOperation::Set => {
                             store
-                                .assigned_node(&deployment)
+                                .assignment_status(&deployment)
                                 .map_err(|e| {
                                     anyhow!("Failed to get subgraph assignment entity: {}", e)
                                 })
                                 .map(|assigned| -> Box<dyn Stream<Item = _, Error = _> + Send> {
-                                    if let Some(assigned) = assigned {
+                                    if let Some((assigned,is_paused)) = assigned {
                                         if assigned == node_id {
+
+                                            if is_paused{
+                                                // Subgraph is paused, so we don't start it
+                                                debug!(logger, "Deployment assignee is this node, but it is paused, so we don't start it"; "assigned_to" => assigned, "node_id" => &node_id,"paused" => is_paused);
+                                                return Box::new(stream::empty());
+                                            }
+
                                             // Start subgraph on this node
                                             debug!(logger, "Deployment assignee is this node, broadcasting add event"; "assigned_to" => assigned, "node_id" => &node_id);
                                             Box::new(stream::once(Ok(AssignmentEvent::Add {

graph/src/components/store/traits.rs

@@ -95,6 +95,15 @@ pub trait SubgraphStore: Send + Sync + 'static {
 
     fn assigned_node(&self, deployment: &DeploymentLocator) -> Result<Option<NodeId>, StoreError>;
 
+    /// Returns Option<(node_id,is_paused)> where `node_id` is the node that
+    /// the subgraph is assigned to, and `is_paused` is true if the
+    /// subgraph is paused.
+    /// Returns None if the deployment does not exist.
+    fn assignment_status(
+        &self,
+        deployment: &DeploymentLocator,
+    ) -> Result<Option<(NodeId, bool)>, StoreError>;
+
     fn assignments(&self, node: &NodeId) -> Result<Vec<DeploymentLocator>, StoreError>;
 
     /// Return `true` if a subgraph `name` exists, regardless of whether the

node/src/manager/commands/assign.rs

@@ -85,15 +85,23 @@ pub fn pause_or_resume(
         .locate_site(locator.clone())?
         .ok_or_else(|| anyhow!("failed to locate site for {locator}"))?;
 
-    if pause {
-        println!("pausing {locator}");
-        conn.pause_subgraph(&site)?;
-        println!("paused {locator}")
-    } else {
-        println!("resuming {locator}");
-        conn.resume_subgraph(&site)?;
-        println!("resumed {locator}")
-    }
+    let change = match conn.assignment_status(&site)? {
+        Some((_, paused)) => {
+            if paused == pause {
+                println!("deployment {locator} is already {paused}");
+                vec![]
+            } else {
+                println!("pausing {locator}");
+                conn.pause_subgraph(&site)?
+            }
+        }
+        None => {
+            println!("resuming {locator}");
+            conn.resume_subgraph(&site)?
+        }
+    };
+    println!("Operation completed");
+    conn.send_store_event(sender, &StoreEvent::new(change))?;
 
     Ok(())
 }

store/postgres/src/primary.rs

@@ -405,6 +405,7 @@ pub fn make_dummy_site(deployment: DeploymentHash, namespace: Namespace, network
 /// mirrored through `Mirror::refresh_tables` and must be queries, i.e.,
 /// read-only
 mod queries {
+    use diesel::data_types::PgTimestamp;
     use diesel::dsl::{any, exists, sql};
     use diesel::pg::PgConnection;
     use diesel::prelude::{
@@ -626,6 +627,36 @@ mod queries {
             .transpose()
     }
 
+    /// Returns Option<(node_id,is_paused)> where `node_id` is the node that
+    /// the subgraph is assigned to, and `is_paused` is true if the
+    /// subgraph is paused.
+    /// Returns None if the deployment does not exist.
+    pub(super) fn assignment_status(
+        conn: &PgConnection,
+        site: &Site,
+    ) -> Result<Option<(NodeId, bool)>, StoreError> {
+        a::table
+            .filter(a::id.eq(site.id))
+            .select((a::node_id, a::paused_at))
+            .first::<(String, Option<PgTimestamp>)>(conn)
+            .optional()?
+            .map(|(node, ts)| {
+                let node_id = NodeId::new(&node).map_err(|()| {
+                    constraint_violation!(
+                        "invalid node id `{}` in assignment for `{}`",
+                        node,
+                        site.deployment
+                    )
+                })?;
+
+                match ts {
+                    Some(_) => Ok((node_id, true)),
+                    None => Ok((node_id, false)),
+                }
+            })
+            .transpose()
+    }
+
     pub(super) fn version_info(
         conn: &PgConnection,
         version: &str,
@@ -981,7 +1012,7 @@ impl<'a> Connection<'a> {
         }
     }
 
-    pub fn pause_subgraph(&self, site: &Site) -> Result<(), StoreError> {
+    pub fn pause_subgraph(&self, site: &Site) -> Result<Vec<EntityChange>, StoreError> {
         use subgraph_deployment_assignment as a;
 
         let conn = self.conn.as_ref();
@@ -991,7 +1022,11 @@ impl<'a> Connection<'a> {
             .execute(conn)?;
         match updates {
             0 => Err(StoreError::DeploymentNotFound(site.deployment.to_string())),
-            1 => Ok(()),
+            1 => {
+                let change =
+                    EntityChange::for_assignment(site.into(), EntityChangeOperation::Removed);
+                Ok(vec![change])
+            }
             _ => {
                 // `id` is the primary key of the subgraph_deployment_assignment table,
                 // and we can therefore only update no or one entry
@@ -1000,7 +1035,7 @@ impl<'a> Connection<'a> {
         }
     }
 
-    pub fn resume_subgraph(&self, site: &Site) -> Result<(), StoreError> {
+    pub fn resume_subgraph(&self, site: &Site) -> Result<Vec<EntityChange>, StoreError> {
         use subgraph_deployment_assignment as a;
 
         let conn = self.conn.as_ref();
@@ -1010,7 +1045,10 @@ impl<'a> Connection<'a> {
             .execute(conn)?;
         match updates {
             0 => Err(StoreError::DeploymentNotFound(site.deployment.to_string())),
-            1 => Ok(()),
+            1 => {
+                let change = EntityChange::for_assignment(site.into(), EntityChangeOperation::Set);
+                Ok(vec![change])
+            }
             _ => {
                 // `id` is the primary key of the subgraph_deployment_assignment table,
                 // and we can therefore only update no or one entry
@@ -1148,6 +1186,14 @@ impl<'a> Connection<'a> {
         queries::assigned_node(self.conn.as_ref(), site)
     }
 
+    /// Returns Option<(node_id,is_paused)> where `node_id` is the node that
+    /// the subgraph is assigned to, and `is_paused` is true if the
+    /// subgraph is paused.
+    /// Returns None if the deployment does not exist.
+    pub fn assignment_status(&self, site: &Site) -> Result<Option<(NodeId, bool)>, StoreError> {
+        queries::assignment_status(self.conn.as_ref(), site)
+    }
+
     /// Create a copy of the site `src` in the shard `shard`, but mark it as
     /// not active. If there already is a site in `shard`, return that
     /// instead.
@@ -1761,6 +1807,14 @@ impl Mirror {
         self.read(|conn| queries::assigned_node(conn, site))
     }
 
+    /// Returns Option<(node_id,is_paused)> where `node_id` is the node that
+    /// the subgraph is assigned to, and `is_paused` is true if the
+    /// subgraph is paused.
+    /// Returns None if the deployment does not exist.
+    pub fn assignment_status(&self, site: &Site) -> Result<Option<(NodeId, bool)>, StoreError> {
+        self.read(|conn| queries::assignment_status(conn, site))
+    }
+
     pub fn find_active_site(&self, subgraph: &DeploymentHash) -> Result<Option<Site>, StoreError> {
         self.read(|conn| queries::find_active_site(conn, subgraph))
     }

store/postgres/src/subgraph_store.rs

@@ -1286,6 +1286,18 @@ impl SubgraphStoreTrait for SubgraphStore {
         self.mirror.assigned_node(site.as_ref())
     }
 
+    /// Returns Option<(node_id,is_paused)> where `node_id` is the node that
+    /// the subgraph is assigned to, and `is_paused` is true if the
+    /// subgraph is paused.
+    /// Returns None if the deployment does not exist.
+    fn assignment_status(
+        &self,
+        deployment: &DeploymentLocator,
+    ) -> Result<Option<(NodeId, bool)>, StoreError> {
+        let site = self.find_site(deployment.id.into())?;
+        self.mirror.assignment_status(site.as_ref())
+    }
+
     fn assignments(&self, node: &NodeId) -> Result<Vec<DeploymentLocator>, StoreError> {
         self.mirror
             .assignments(node)