Exponential backoff retry on catalog watcher

1 files changed, 49 insertions, 25 deletions

diff --git a/src/catalog.rs b/src/catalog.rs
index 0f0ecf5..cad7f5b 100644
--- a/src/catalog.rs
+++ b/src/catalog.rs
@@ -3,6 +3,7 @@
 //! See <https://developer.hashicorp.com/consul/api-docs/catalog>
 //! for the full definition of the API.
 
+use std::cmp;
 use std::collections::HashMap;
 use std::fmt::Write;
 use std::sync::Arc;
@@ -11,7 +12,7 @@ use std::time::Duration;
 use anyhow::Result;
 use futures::future::BoxFuture;
 use futures::stream::futures_unordered::FuturesUnordered;
-use futures::{FutureExt, StreamExt};
+use futures::{FutureExt, StreamExt, TryFutureExt};
 use log::*;
 use serde::{Deserialize, Serialize};
 use tokio::select;
@@ -19,7 +20,7 @@ use tokio::sync::watch;
 
 use crate::{Consul, WithIndex};
 
-/// Node summary, as specified in response to "list nodes" API calls in 
+/// Node summary, as specified in response to "list nodes" API calls in
 /// <https://developer.hashicorp.com/consul/api-docs/catalog#list-nodes>
 #[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(rename_all = "PascalCase")]
@@ -126,11 +127,8 @@ impl Consul {
         &self,
         last_index: Option<usize>,
     ) -> Result<WithIndex<ServiceList>> {
-        self.get_with_index::<ServiceList>(
-            format!("{}/v1/catalog/services", self.url),
-            last_index,
-        )
-        .await
+        self.get_with_index::<ServiceList>(format!("{}/v1/catalog/services", self.url), last_index)
+            .await
     }
 
     /// The "list nodes for a service" API call of the Catalog api
@@ -166,10 +164,17 @@ impl Consul {
     /// Launches a background task that watches all services and the nodes that serve them,
     /// and make that info available in a tokio watch channel.
     /// The worker terminates when the channel is dropped.
-    pub fn watch_all_service_health(&self) -> watch::Receiver<AllServiceHealth> {
+    pub fn watch_all_service_health(
+        &self,
+        max_retry_interval: Duration,
+    ) -> watch::Receiver<AllServiceHealth> {
         let (tx, rx) = watch::channel(HashMap::new());
 
-        tokio::spawn(do_watch_all_service_health(self.clone(), tx));
+        tokio::spawn(do_watch_all_service_health(
+            self.clone(),
+            tx,
+            max_retry_interval,
+        ));
 
         rx
     }
@@ -194,10 +199,16 @@ impl Consul {
     }
 }
 
-async fn do_watch_all_service_health(consul: Consul, tx: watch::Sender<AllServiceHealth>) {
+async fn do_watch_all_service_health(
+    consul: Consul,
+    tx: watch::Sender<AllServiceHealth>,
+    max_retry_interval: Duration,
+) {
     let mut services = AllServiceHealth::new();
-    let mut service_watchers = FuturesUnordered::<BoxFuture<(String, Result<_>)>>::new();
-    let mut service_list: BoxFuture<Result<_>> = Box::pin(consul.catalog_service_list(None));
+    let mut service_watchers =
+        FuturesUnordered::<BoxFuture<(String, std::result::Result<_, (usize, _)>)>>::new();
+    let mut service_list: BoxFuture<std::result::Result<_, (usize, _)>> =
+        Box::pin(consul.catalog_service_list(None).map_err(|e| (1, e)));
 
     loop {
         select! {
@@ -211,18 +222,20 @@ async fn do_watch_all_service_health(consul: Consul, tx: watch::Sender<AllServic
 
                                 let service = service.to_string();
                                 service_watchers.push(Box::pin(async {
-                                    let res = consul.health_service_instances(&service, None).await;
+                                    let res = consul.health_service_instances(&service, None).await
+                                        .map_err(|e| (1, e));
                                     (service, res)
                                 }));
                             }
                         }
-                        service_list = Box::pin(consul.catalog_service_list(Some(list_index)));
+                        service_list = Box::pin(consul.catalog_service_list(Some(list_index)).map_err(|e| (1, e)));
                     }
-                    Err(e) => {
-                        warn!("Error listing services: {}", e);
-                        service_list = Box::pin(async {
-                            tokio::time::sleep(Duration::from_secs(30)).await;
-                            consul.catalog_service_list(None).await
+                    Err((err_count, e)) => {
+                        warn!("Error listing services: {} ({} consecutive errors)", e, err_count);
+                        let consul = &consul;
+                        service_list = Box::pin(async move {
+                            tokio::time::sleep(retry_to_time(err_count, max_retry_interval)).await;
+                            consul.catalog_service_list(None).await.map_err(|e| (err_count + 1, e))
                         });
                     }
                 }
@@ -235,7 +248,8 @@ async fn do_watch_all_service_health(consul: Consul, tx: watch::Sender<AllServic
 
                         let consul = &consul;
                         service_watchers.push(Box::pin(async move {
-                            let res = consul.health_service_instances(&service, Some(index)).await;
+                            let res = consul.health_service_instances(&service, Some(index)).await
+                                .map_err(|e| (1, e));
                             (service, res)
                         }));
 
@@ -243,11 +257,12 @@ async fn do_watch_all_service_health(consul: Consul, tx: watch::Sender<AllServic
                             break;
                         }
                     }
-                    Err(e) => {
-                        warn!("Error getting service {}: {}", service, e);
-                        service_watchers.push(Box::pin(async {
-                            tokio::time::sleep(Duration::from_secs(30)).await;
-                            let res = consul.health_service_instances(&service, None).await;
+                    Err((err_count, e)) => {
+                        warn!("Error getting service {}: {} ({} consecutive errors)", service, e, err_count);
+                        let consul = &consul;
+                        service_watchers.push(Box::pin(async move {
+                            tokio::time::sleep(retry_to_time(err_count, max_retry_interval)).await;
+                            let res = consul.health_service_instances(&service, None).await.map_err(|e| (err_count + 1, e));
                             (service, res)
                         }));
                     }
@@ -266,3 +281,12 @@ async fn some_or_pending<T>(value: Option<T>) -> T {
         None => futures::future::pending().await,
     }
 }
+
+fn retry_to_time(retries: usize, max_time: Duration) -> Duration {
+    // Exponential retry interval, starting at 2 seconds, maxing out at max_time,
+    // with exponential increase of *1.5 each time
+    cmp::min(
+        max_time,
+        Duration::from_secs_f64(2.0f64 * 1.5f64.powf(retries as f64)),
+    )
+}