migrate2watch deployment_status

Author: taslimmuhammed

common/src/subgraph_client/client.rs

@@ -1,7 +1,6 @@
 // Copyright 2023-, Edge & Node, GraphOps, and Semiotic Labs.
 // SPDX-License-Identifier: Apache-2.0
 
-
 use super::monitor::{monitor_deployment_status, DeploymentStatus};
 use anyhow::anyhow;
 use axum::body::Bytes;
@@ -161,19 +160,7 @@ impl DeploymentClient {
         &self,
         variables: T::Variables,
     ) -> Result<ResponseResult<T::ResponseData>, anyhow::Error> {
-        if let Some(ref status_rx) = self.status {
-                //change
-            while status_rx.borrow().is_none() {dbg!("looping");};
-            let some_ref = status_rx.borrow();
-            let deployment_status = some_ref.as_ref().unwrap();
-
-            if !deployment_status.synced || &deployment_status.health != "healthy" {
-                return Err(anyhow!(
-                    "Deployment `{}` is not ready or healthy to be queried",
-                    self.query_url
-                ));
-            }
-        }
+        self.check_deployment_status()?;
 
         let body = T::build_query(variables);
         let reqwest_response = self
@@ -194,20 +181,7 @@ impl DeploymentClient {
     }
 
     pub async fn query_raw(&self, body: Bytes) -> Result<reqwest::Response, anyhow::Error> {
-        if let Some(ref status_rx) = self.status {
-                //change
-            while status_rx.borrow().is_none() {dbg!("looping");};
-            let some_ref = status_rx.borrow();
-            let deployment_status = some_ref.as_ref().unwrap();
-
-            if !deployment_status.synced || &deployment_status.health != "healthy" {
-                return Err(anyhow!(
-                    "Deployment `{}` is not ready or healthy to be queried",
-                    self.query_url
-                ));
-            }
-        }
-
+        self.check_deployment_status()?;
         Ok(self
             .http_client
             .post(self.query_url.as_ref())
@@ -217,6 +191,27 @@ impl DeploymentClient {
             .send()
             .await?)
     }
+    fn check_deployment_status(&self)->Result<(), anyhow::Error>{
+        if let Some(ref status_rx) = self.status {
+            let status_ref = status_rx.borrow();
+            let status = status_ref.as_ref();
+            match status{
+                Some(deployment_status)=>{
+                   if !deployment_status.synced || &deployment_status.health != "healthy" {
+                        return Err(anyhow!(
+                            "Deployment `{}` is not ready or healthy to be queried",
+                            self.query_url
+                        ));
+                    }
+                }
+                None => return Err(anyhow!(
+                    "Deployment `{}` is not available",
+                    self.query_url
+                ))
+            }
+        }
+        Ok(())
+    }
 }
 
 /// Client for a subgraph that can fall back from a local deployment to a remote query URL
@@ -299,6 +294,7 @@ impl SubgraphClient {
 #[cfg(test)]
 mod test {
     use std::str::FromStr;
+    use std::time::Duration;
 
     use serde_json::json;
     use wiremock::matchers::{method, path};
@@ -445,6 +441,9 @@ mod test {
             .unwrap(),
         );
 
+        // Waiting for status to propegate
+        tokio::time::sleep(Duration::from_millis(1000)).await;
+        
         // Query the subgraph
         let data = client
             .query::<UserQuery, _>(user_query::Variables {})
@@ -525,6 +524,9 @@ mod test {
             .unwrap(),
         );
 
+        // Waiting for status to propegate
+        tokio::time::sleep(Duration::from_millis(100)).await;
+
         // Query the subgraph
         let data = client
             .query::<UserQuery, _>(user_query::Variables {})
@@ -605,6 +607,9 @@ mod test {
             .unwrap(),
         );
 
+        // Waiting for status to propegate
+        tokio::time::sleep(Duration::from_millis(100)).await;
+
         // Query the subgraph
         let data = client
             .query::<UserQuery, _>(user_query::Variables {})
@@ -614,4 +619,4 @@ mod test {
 
         assert_eq!(data.user.name, "remote".to_string());
     }
-}
+}

common/src/subgraph_client/monitor.rs

@@ -68,6 +68,7 @@ pub fn monitor_deployment_status(
             });
             match result{
                 Ok(deployment_status) => {
+                    dbg!("Sending......");
                     tx
                     .send(Some(deployment_status))
                     .expect("Failed to update deployment_status channel");
@@ -77,7 +78,7 @@ pub fn monitor_deployment_status(
                         "Error querying deployment status for `{}`: {}",
                         deployment, err
                     );
-                    sleep(Duration::from_secs(15)).await
+                    sleep(Duration::from_secs(15)).await;
                 },
             }
         }
@@ -122,7 +123,7 @@ mod tests {
             .await;
 
             let mut status_rx = monitor_deployment_status(deployment, status_url);
-            status_rx.changed().await;
+            status_rx.changed().await.unwrap();
             let status_ref = status_rx.borrow();
             assert_eq!(
                 status_ref.as_ref().unwrap().clone(),
@@ -161,7 +162,7 @@ mod tests {
             .await;
 
             let mut status_rx = monitor_deployment_status(deployment, status_url);
-            status_rx.changed().await;
+            status_rx.changed().await.unwrap();
             let status_ref = status_rx.borrow();
             assert_eq!(
                 status_ref.as_ref().unwrap().clone(),
@@ -200,7 +201,7 @@ mod tests {
             .await;
 
             let mut status_rx = monitor_deployment_status(deployment, status_url);
-            status_rx.changed().await;
+            status_rx.changed().await.unwrap();
             let status_ref = status_rx.borrow();
             assert_eq!(
                 status_ref.as_ref().unwrap().clone(),
@@ -239,7 +240,7 @@ mod tests {
             .await;
 
         let mut status_rx = monitor_deployment_status(deployment, status_url);
-        status_rx.changed().await;
+        status_rx.changed().await.unwrap();
         let status_ref = status_rx.borrow();
         assert_eq!(
             status_ref.as_ref().unwrap().clone(),