Support Connection Retries

rosetta-docker/Cargo.toml

@@ -17,6 +17,7 @@ rosetta-client = { version = "0.4.0", path = "../rosetta-client" }
 rosetta-core = { version = "0.4.0", path = "../rosetta-core" }
 surf = { version = "2.3.2", default-features = false, features = ["h1-client-no-tls"] }
 tokio = "1.26.0"
+tokio-retry = "0.3"
 
 [dev-dependencies]
 tokio = { version = "1.26.0", features = ["macros"] }

rosetta-docker/src/lib.rs

@@ -9,6 +9,10 @@ use futures::stream::StreamExt;
 use rosetta_client::{Client, Signer, Wallet};
 use rosetta_core::{BlockchainClient, BlockchainConfig};
 use std::time::Duration;
+use tokio_retry::{
+    strategy::{jitter, ExponentialBackoff},
+    RetryIf,
+};
 
 pub struct Env {
     config: BlockchainConfig,
@@ -31,8 +35,22 @@ impl Env {
         builder.stop_container(&builder.node_name(&config)).await?;
         builder.delete_network(&builder.network_name()).await?;
         let network = builder.create_network().await?;
-        let node = builder.run_node(&config, &network).await?;
-        let connector = builder.run_connector(&config, &network).await?;
+        let node = match builder.run_node(&config, &network).await {
+            Ok(node) => node,
+            Err(e) => {
+                network.delete().await?;
+                return Err(e);
+            }
+        };
+        let connector = match builder.run_connector(&config, &network).await {
+            Ok(connector) => connector,
+            Err(e) => {
+                let opts = ContainerStopOpts::builder().build();
+                let _ = node.stop(&opts).await;
+                network.delete().await?;
+                return Err(e);
+            }
+        };
         Ok(Self {
             config,
             network,
@@ -77,6 +95,7 @@ struct EnvBuilder<'a> {
 impl<'a> EnvBuilder<'a> {
     pub fn new(prefix: &'a str) -> Result<Self> {
         let version = ApiVersion::new(1, Some(41), None);
+        // TODO: Support custom connections #138
         #[cfg(unix)]
         let docker = Docker::unix_versioned("/var/run/docker.sock", version);
         #[cfg(not(unix))]
@@ -225,7 +244,7 @@ impl<'a> EnvBuilder<'a> {
             opts = opts.expose(PublishPort::tcp(port), port);
         }
         let container = self.run_container(name, &opts.build(), network).await?;
-        //wait_for_http(&format!("http://127.0.0.1:{}", config.node_port)).await?;
+        // wait_for_http(&format!("http://127.0.0.1:{}", config.node_port)).await?;
         tokio::time::sleep(Duration::from_secs(30)).await;
         Ok(container)
     }
@@ -239,7 +258,7 @@ impl<'a> EnvBuilder<'a> {
         let link = self.node_name(config);
         let opts = ContainerCreateOpts::builder()
             .name(&name)
-            .image(format!("analoglabs/connector-{}", config.blockchain))
+            .image(format!("analoglabs/connector-{}:latest", config.blockchain))
             .command(vec![
                 format!("--network={}", config.network),
                 format!("--addr=0.0.0.0:{}", config.connector_port),
@@ -255,8 +274,20 @@ impl<'a> EnvBuilder<'a> {
             )
             .build();
         let container = self.run_container(name, &opts, network).await?;
-        wait_for_http(&format!("http://127.0.0.1:{}", config.connector_port)).await?;
-        Ok(container)
+
+        match wait_for_http(
+            &format!("http://127.0.0.1:{}", config.connector_port),
+            &container,
+        )
+        .await
+        {
+            Ok(()) => Ok(container),
+            Err(err) => {
+                log::error!("connector failed to start: {}", err);
+                let _ = container.stop(&ContainerStopOpts::builder().build()).await;
+                Err(err)
+            }
+        }
     }
 }
 
@@ -286,17 +317,30 @@ async fn health(container: &Container) -> Result<Option<Health>> {
     }))
 }
 
-async fn wait_for_http(url: &str) -> Result<()> {
-    loop {
-        match surf::get(url).await {
-            Ok(_) => {
-                break;
-            }
-            Err(err) => {
-                log::error!("{}", err);
-                tokio::time::sleep(Duration::from_millis(500)).await;
+async fn wait_for_http(url: &str, container: &Container) -> Result<()> {
+    let retry_strategy = ExponentialBackoff::from_millis(2)
+        .factor(100)
+        .max_delay(Duration::from_secs(10))
+        .map(jitter) // add jitter to delays
+        .take(20); // limit to 20 retries
+
+    RetryIf::spawn(
+        retry_strategy,
+        || async move {
+            match surf::get(url).await {
+                Ok(_) => Ok(()),
+                Err(err) => {
+                    if health(container).await.is_err() {
+                        return Err("container exited".to_string());
+                    }
+                    log::error!("url: {} error: {}", url, err);
+                    Err(err.to_string())
+                }
             }
-        }
-    }
-    Ok(())
+        },
+        // Retry Condition
+        |error_message: &String| !error_message.starts_with("container exited"),
+    )
+    .await
+    .map_err(|err| anyhow::format_err!("{}", err))
 }

rosetta-server/Cargo.toml

@@ -24,6 +24,7 @@ serde_json = "1.0.94"
 sled = "0.34.7"
 tide = { version = "0.16.0", default-features = false, features = ["h1-server", "logger"] }
 tokio = { version = "1.26.0", features = ["full"] }
+tokio-retry = "0.3"
 
 [build-dependencies]
 anyhow = "1.0.69"

rosetta-server/src/lib.rs

@@ -18,6 +18,10 @@ use std::time::Duration;
 use tide::http::headers::HeaderValue;
 use tide::security::{CorsMiddleware, Origin};
 use tide::{Body, Request, Response};
+use tokio_retry::{
+    strategy::{jitter, ExponentialBackoff},
+    Retry,
+};
 
 pub use rosetta_core::*;
 
@@ -41,7 +45,26 @@ pub async fn main<T: BlockchainClient>() -> Result<()> {
 
     log::info!("connecting to {}", &opts.node_addr);
     let config = T::create_config(&opts.network)?;
-    let client = T::new(config, &opts.node_addr).await?;
+
+    let client = {
+        // TODO: Allow configuring the retry strategy and retry count
+        let retry_strategy = ExponentialBackoff::from_millis(2)
+            .factor(100)
+            .max_delay(Duration::from_secs(5))
+            .map(jitter) // add jitter to delays
+            .take(30); // limit to 30 retries
+        Retry::spawn(retry_strategy, || async {
+            match T::new(config.clone(), &opts.node_addr).await {
+                Ok(client) => Ok(client),
+                Err(err) => {
+                    log::error!("{}", err);
+                    Err(err)
+                }
+            }
+        })
+        .await?
+    };
+
     let indexer = Arc::new(Indexer::new(&opts.path, client)?);
 
     let cors = CorsMiddleware::new()