feat: gracefully shutdown drainer if redis goes down (#2391)

Author: dracarys18

crates/drainer/src/lib.rs

@@ -10,7 +10,7 @@ use std::sync::{atomic, Arc};
 use common_utils::signals::get_allowed_signals;
 use diesel_models::kv;
 use error_stack::{IntoReport, ResultExt};
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, oneshot};
 
 use crate::{connection::pg_connection, services::Store};
 
@@ -36,9 +36,18 @@ pub async fn start_drainer(
                 "Failed while getting allowed signals".to_string(),
             ))?;
 
+    let (redis_error_tx, redis_error_rx) = oneshot::channel();
     let (tx, mut rx) = mpsc::channel(1);
     let handle = signal.handle();
-    let task_handle = tokio::spawn(common_utils::signals::signal_handler(signal, tx));
+    let task_handle = tokio::spawn(common_utils::signals::signal_handler(signal, tx.clone()));
+
+    let redis_conn_clone = store.redis_conn.clone();
+
+    // Spawn a task to monitor if redis is down or not
+    tokio::spawn(async move { redis_conn_clone.on_error(redis_error_tx).await });
+
+    //Spawns a task to send shutdown signal if redis goes down
+    tokio::spawn(redis_error_receiver(redis_error_rx, tx));
 
     let active_tasks = Arc::new(atomic::AtomicU64::new(0));
     'event: loop {
@@ -90,6 +99,20 @@ pub async fn start_drainer(
     Ok(())
 }
 
+pub async fn redis_error_receiver(rx: oneshot::Receiver<()>, shutdown_channel: mpsc::Sender<()>) {
+    match rx.await {
+        Ok(_) => {
+            logger::error!("The redis server failed ");
+            let _ = shutdown_channel.send(()).await.map_err(|err| {
+                logger::error!("Failed to send signal to the shutdown channel {err}")
+            });
+        }
+        Err(err) => {
+            logger::error!("Channel receiver error{err}");
+        }
+    }
+}
+
 async fn drainer_handler(
     store: Arc<Store>,
     stream_index: u8,