fix(pubsub): retry connecting to backend (#2254)

* fix(`pubsub`): retry connecting to backend

* clippy

Author: yash-atreya

crates/pubsub/src/service.rs

@@ -31,6 +31,11 @@ pub(crate) struct PubSubService<T> {
 
     /// The request manager.
     pub(crate) in_flights: RequestManager,
+
+    /// Number of retries. Default is 10.
+    ///
+    /// Every retry is made at an interval of 3 seconds.
+    pub(crate) retries: u32,
 }
 
 impl<T: PubSubConnect> PubSubService<T> {
@@ -45,6 +50,7 @@ impl<T: PubSubConnect> PubSubService<T> {
             reqs,
             subs: SubscriptionManager::default(),
             in_flights: Default::default(),
+            retries: 10,
         };
         this.spawn();
         Ok(PubSubFrontend::new(tx))
@@ -190,6 +196,31 @@ impl<T: PubSubConnect> PubSubService<T> {
         Ok(())
     }
 
+    /// Attempt to reconnect with retries
+    async fn reconnect_with_retries(&mut self) -> TransportResult<()> {
+        let mut retry_count = 0;
+        loop {
+            match self.reconnect().await {
+                Ok(()) => break Ok(()),
+                Err(e) => {
+                    retry_count += 1;
+                    if retry_count >= self.retries {
+                        error!(
+                            "Reconnect failed after {} attempts, shutting down: {e}",
+                            retry_count
+                        );
+                        break Err(e);
+                    }
+                    warn!(
+                        "Reconnection attempt {}/{} failed: {}. Retrying in 3 seconds...",
+                        retry_count, self.retries, e
+                    );
+                    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+                }
+            }
+        }
+    }
+
     /// Spawn the service.
     pub(crate) fn spawn(mut self) {
         let fut = async move {
@@ -205,16 +236,14 @@ impl<T: PubSubConnect> PubSubService<T> {
                             if let Err(e) = self.handle_item(item) {
                                 break Err(e)
                             }
-                        } else if let Err(e) = self.reconnect().await {
-                            error!("Reconnect failed, shutting down: {e}");
+                        } else if let Err(e) = self.reconnect_with_retries().await {
                             break Err(e)
                         }
                     }
 
                     _ = &mut self.handle.error => {
                         error!("Pubsub service backend error.");
-                        if let Err(e) = self.reconnect().await {
-                            error!("Reconnect failed, shutting down: {e}");
+                        if let Err(e) = self.reconnect_with_retries().await {
                             break Err(e)
                         }
                     }