Add retry connections to chain

src/bins/rmb-relay.rs

@@ -145,8 +145,6 @@ async fn app(args: Args) -> Result<()> {
 
     let redis_cache = RedisCache::new(pool.clone(), "twin");
 
-    redis_cache.flush().await?;
-
     let twins = SubstrateTwinDB::<RedisCache>::new(args.substrate.clone(), redis_cache.clone())
         .await
         .context("cannot create substrate twin db object")?;
@@ -176,7 +174,7 @@ async fn app(args: Args) -> Result<()> {
         .await
         .unwrap();
 
-    let l = events::Listener::new(args.substrate[0].as_str(), redis_cache).await?;
+    let mut l = events::Listener::new(args.substrate, redis_cache).await?;
     tokio::spawn(async move {
         if let Err(e) = l.listen().await {
             log::error!("failed to listen to events: {:#}", e);

src/cache/memory.rs

@@ -45,6 +45,12 @@ where
             Some(v) => Ok(Some(v.clone())),
         }
     }
+    async fn flush(&self) -> Result<()> {
+        let mut mem = self.mem.write().await;
+        mem.clear();
+
+        Ok(())
+    }
 }
 
 #[cfg(test)]

src/cache/mod.rs

@@ -11,6 +11,7 @@ use std::marker::{Send, Sync};
 pub trait Cache<T>: Send + Sync + 'static {
     async fn set<S: ToString + Send + Sync>(&self, id: S, obj: T) -> Result<()>;
     async fn get<S: ToString + Send + Sync>(&self, id: S) -> Result<Option<T>>;
+    async fn flush(&self) -> Result<()>;
 }
 
 #[async_trait]
@@ -31,6 +32,12 @@ where
             None => Ok(None),
         }
     }
+    async fn flush(&self) -> Result<()> {
+        match self {
+            Some(cache) => cache.flush().await,
+            None => Ok(()),
+        }
+    }
 }
 
 #[derive(Clone, Copy)]
@@ -47,4 +54,7 @@ where
     async fn get<S: ToString + Send + Sync>(&self, _id: S) -> Result<Option<T>> {
         Ok(None)
     }
+    async fn flush(&self) -> Result<()> {
+        Ok(())
+    }
 }

src/cache/redis.rs

@@ -39,12 +39,6 @@ impl RedisCache {
 
         Ok(conn)
     }
-    pub async fn flush(&self) -> Result<()> {
-        let mut conn = self.get_connection().await?;
-        cmd("DEL").arg(&self.prefix).query_async(&mut *conn).await?;
-
-        Ok(())
-    }
 }
 
 #[async_trait]
@@ -83,6 +77,12 @@ where
             None => Ok(None),
         }
     }
+    async fn flush(&self) -> Result<()> {
+        let mut conn = self.get_connection().await?;
+        cmd("DEL").arg(&self.prefix).query_async(&mut *conn).await?;
+
+        Ok(())
+    }
 }
 
 #[cfg(test)]

src/events/events.rs

@@ -1,3 +1,5 @@
+use std::collections::LinkedList;
+
 use crate::{cache::Cache, tfchain::tfchain, twin::Twin};
 use anyhow::Result;
 use futures::StreamExt;
@@ -11,17 +13,77 @@ where
 {
     cache: C,
     api: OnlineClient<PolkadotConfig>,
+    substrate_urls: LinkedList<String>,
 }
 
 impl<C> Listener<C>
 where
     C: Cache<Twin> + Clone,
 {
-    pub async fn new(url: &str, cache: C) -> Result<Self> {
-        let api = OnlineClient::<PolkadotConfig>::from_url(url).await?;
-        Ok(Listener { api, cache })
+    pub async fn new(substrate_urls: Vec<String>, cache: C) -> Result<Self> {
+        let mut urls = LinkedList::new();
+        for url in substrate_urls {
+            urls.push_back(url);
+        }
+
+        let api = Self::connect(&mut urls).await?;
+
+        cache.flush().await?;
+        Ok(Listener {
+            api,
+            cache,
+            substrate_urls: urls,
+        })
+    }
+
+    async fn connect(urls: &mut LinkedList<String>) -> Result<OnlineClient<PolkadotConfig>> {
+        let trials = urls.len() * 2;
+        for _ in 0..trials {
+            let url = match urls.front() {
+                Some(url) => url,
+                None => anyhow::bail!("substrate urls list is empty"),
+            };
+
+            match OnlineClient::<PolkadotConfig>::from_url(url).await {
+                Ok(client) => return Ok(client),
+                Err(err) => {
+                    log::error!(
+                        "failed to create substrate client with url \"{}\": {}",
+                        url,
+                        err
+                    );
+                }
+            }
+
+            if let Some(front) = urls.pop_front() {
+                urls.push_back(front);
+            }
+        }
+
+        anyhow::bail!("failed to connect to substrate using the provided urls")
+    }
+
+    pub async fn listen(&mut self) -> Result<()> {
+        loop {
+            // always flush in case some blocks were finalized before reconnecting
+            self.cache.flush().await?;
+            match self.handle_events().await {
+                Err(err) => {
+                    if let Some(subxt::Error::Rpc(_)) = err.downcast_ref::<subxt::Error>() {
+                        self.api = Self::connect(&mut self.substrate_urls).await?;
+                    } else {
+                        return Err(err);
+                    }
+                }
+                Ok(_) => {
+                    // reconnect here too?
+                    self.api = Self::connect(&mut self.substrate_urls).await?;
+                }
+            }
+        }
     }
-    pub async fn listen(&self) -> Result<()> {
+
+    async fn handle_events(&self) -> Result<()> {
         log::info!("started chain events listener");
         let mut blocks_sub = self.api.blocks().subscribe_finalized().await?;
         while let Some(block) = blocks_sub.next().await {