exp: concurrent health check

Cargo.toml

@@ -29,6 +29,7 @@ sqlx = { version = "0.6", features = [ "runtime-actix-native-tls" , "postgres",
 deadpool-redis = "0.11.0"
 chrono = "0.4.23"
 uuid = "1.2.2"
+tokio = "1.23.0"
 
 [dev-dependencies]
 serial_test = "0.9.0"

src/components/app.rs

@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use super::{
     configuration::Config, database::DatabaseComponent, health::HealthComponent,
     redis::RedisComponent, synapse::SynapseComponent,
@@ -36,9 +38,9 @@ impl AppComponents {
             panic!("Unable connecting to redis {:?}", err)
         }
 
-        // TODO: Should we refactor HealthComponent to avoid cloning structs?
-        health.register_component(Box::new(db.clone()), "database".to_string());
-        health.register_component(Box::new(redis.clone()), "redis".to_string());
+        // TODO: Should we refactor HealthComponent to avoid cloning structs or should we refactor components for a cheapier cloning?
+        health.register_component(Arc::new(db.clone()), "database".to_string());
+        health.register_component(Arc::new(redis.clone()), "redis".to_string());
 
         Self {
             config,

src/components/health.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashMap, fmt};
+use std::{collections::HashMap, fmt, sync::Arc};
 
 use async_trait::async_trait;
 
@@ -13,7 +13,7 @@ pub trait Healthy {
 }
 
 struct ComponentToCheck {
-    component: Box<dyn Healthy + Send + Sync>,
+    component: Arc<dyn Healthy + Send + Sync>,
     name: String,
 }
 
@@ -31,20 +31,32 @@ pub struct HealthComponent {
 }
 
 impl HealthComponent {
-    pub fn register_component(&mut self, component: Box<dyn Healthy + Send + Sync>, name: String) {
+    pub fn register_component(&mut self, component: Arc<dyn Healthy + Send + Sync>, name: String) {
         self.components_to_check
             .push(ComponentToCheck { component, name });
     }
 
     #[tracing::instrument(name = "Calculate components status")]
     pub async fn calculate_status(&self) -> HashMap<String, ComponentHealthStatus> {
         let mut result = HashMap::new();
+        let (tx, mut rx) = tokio::sync::mpsc::channel(self.components_to_check.len());
 
-        // TODO: Parallelize this checks
         for component in self.components_to_check.as_slice() {
-            let healthy = component.component.is_healthy().await;
+            let tx_cloned = tx.clone();
+            let component_cloned = component.component.clone();
+            let component_name = component.name.clone();
+            log::debug!("About to check: {}", component_name);
+            tokio::spawn(async move {
+                let healthy = component_cloned.is_healthy().await;
+                tx_cloned.send((component_name, healthy)).await.unwrap()
+            });
+            log::debug!("Spawned: {}", component.name);
+        }
+        drop(tx);
+
+        while let Some((name, healthy)) = rx.recv().await {
             result.insert(
-                component.name.to_string(),
+                name,
                 ComponentHealthStatus {
                     status: if healthy {
                         PASS.to_string()
@@ -54,7 +66,6 @@ impl HealthComponent {
                 },
             );
         }
-
         result
     }
 }