fix rmb calls timeout

src/bins/rmb-relay.rs

@@ -11,6 +11,7 @@ use rmb::relay::{
     limiter::{FixedWindowOptions, Limiters},
 };
 use rmb::twin::SubstrateTwinDB;
+use tokio::sync::oneshot;
 
 /// A peer requires only which rely to connect to, and
 /// which identity (mnemonics)
@@ -92,7 +93,7 @@ fn set_limits() -> Result<()> {
     Ok(())
 }
 
-async fn app(args: Args) -> Result<()> {
+async fn app(args: Args, tx: tokio::sync::oneshot::Sender<()>) -> Result<()> {
     if args.workers == 0 {
         anyhow::bail!("number of workers cannot be zero");
     }
@@ -172,10 +173,32 @@ async fn app(args: Args) -> Result<()> {
 
     let mut l = events::Listener::new(args.substrate, redis_cache).await?;
     tokio::spawn(async move {
-        l.listen()
-            .await
-            .context("failed to listen to chain events")
-            .unwrap();
+        let max_retries = 7; // max wait is 2^7 = 128 seconds ( 2 minutes)
+        let mut attempt = 0;
+        let mut backoff = Duration::from_secs(1);
+
+        loop {
+            match l.listen().await.context("failed to listen to chain events") {
+                Ok(_) => break,
+                Err(e) => {
+                    attempt += 1;
+                    if attempt > max_retries {
+                        log::error!("Listener failed after {} attempts: {:?}", attempt, e);
+                        let _ = tx.send(());
+                        break;
+                    } else {
+                        log::warn!(
+                            "Listener failed on attempt {}: {:?}. Retrying in {:?}...",
+                            attempt,
+                            e,
+                            backoff
+                        );
+                        tokio::time::sleep(backoff).await;
+                        backoff *= 2;
+                    }
+                }
+            }
+        }
     });
 
     r.start(&args.listen).await.unwrap();
@@ -185,8 +208,21 @@ async fn app(args: Args) -> Result<()> {
 #[tokio::main]
 async fn main() {
     let args = Args::parse();
-    if let Err(e) = app(args).await {
-        eprintln!("{:#}", e);
-        std::process::exit(1);
+    let (tx, rx) = oneshot::channel();
+    let app_handle = tokio::spawn(async move {
+        if let Err(e) = app(args, tx).await {
+            eprintln!("{:#}", e);
+            std::process::exit(1);
+        }
+    });
+
+    tokio::select! {
+        _ = app_handle => {
+            log::info!("Application is closing successfully.");
+        }
+        _ = rx => {
+            log::error!("Listener shutdown signal received. Exiting application.");
+            std::process::exit(1);
+        }
     }
 }