Exponential backoff retry on catalog watcher

Cargo.toml

@@ -2,7 +2,7 @@
 name = "df-consul"
 description = "Deuxfleurs' async Rust bindings for (a subset of) the Consul HTTP API"
 authors = [ "Alex Auvolat <alex@adnab.me>" ]
-version = "0.3.0"
+version = "0.3.2"
 edition = "2021"
 license = "MIT"
 repository = "https://git.deuxfleurs.fr/Deuxfleurs/df-consul"
@@ -20,3 +20,4 @@ futures = "0.3.25"
 
 [dev-dependencies]
 tokio = { version = "1.23", features = ["rt", "rt-multi-thread", "macros"] }
+pretty_env_logger = "0.4.0"

examples/test.rs

@@ -1,8 +1,12 @@
+use std::time::Duration;
+
 use df_consul::*;
 
 #[tokio::main]
 async fn main() {
-    let config = ConsulConfig {
+    pretty_env_logger::init();
+
+    let config = Config {
         addr: "http://localhost:8500".into(),
         ca_cert: None,
         tls_skip_verify: false,
@@ -41,13 +45,4 @@ async fn main() {
                 .unwrap()
         );
     }
-
-    println!("== WATCHING EVERYTHING ==");
-    let mut watch = consul.watch_all_service_health();
-    loop {
-        if watch.changed().await.is_err() {
-            break;
-        }
-        println!("\n{:?}", watch.borrow_and_update());
-    }
 }

examples/watch_test.rs

@@ -0,0 +1,27 @@
+use std::time::Duration;
+
+use df_consul::*;
+
+#[tokio::main]
+async fn main() {
+    pretty_env_logger::init();
+
+    let config = Config {
+        addr: "http://localhost:8500".into(),
+        ca_cert: None,
+        tls_skip_verify: false,
+        client_cert: None,
+        client_key: None,
+    };
+
+    let consul = Consul::new(config, "").unwrap();
+
+    println!("== WATCHING EVERYTHING ==");
+    let mut watch = consul.watch_all_service_health(Duration::from_secs(30));
+    loop {
+        if watch.changed().await.is_err() {
+            break;
+        }
+        println!("\n{:?}", watch.borrow_and_update());
+    }
+}

src/catalog.rs

@@ -1,3 +1,9 @@
+//! Contains structures to interact with the catalog API
+//!
+//! 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;
@@ -6,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;
@@ -14,35 +20,44 @@ use tokio::sync::watch;
 
 use crate::{Consul, WithIndex};
 
+/// 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")]
-pub struct ConsulNode {
+pub struct Node {
     pub node: String,
     pub address: String,
     pub meta: HashMap<String, String>,
 }
 
+/// One of the services returned in a CatalogNode
 #[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(rename_all = "PascalCase")]
-pub struct ConsulService {
+pub struct Service {
     pub service: String,
     pub address: String,
     pub port: u16,
     pub tags: Vec<String>,
 }
 
+/// Full node info, as specified in response to "retrieve map of services for a node" API call in
+/// <https://developer.hashicorp.com/consul/api-docs/catalog#retrieve-map-of-services-for-a-node>
 #[derive(Serialize, Deserialize, Debug)]
 #[serde(rename_all = "PascalCase")]
-pub struct ConsulCatalogNode {
-    pub node: ConsulNode,
-    pub services: HashMap<String, ConsulService>,
+pub struct CatalogNode {
+    pub node: Node,
+    pub services: HashMap<String, Service>,
 }
 
-pub type ConsulServiceList = HashMap<String, Vec<String>>;
+/// Concise service list, as specified in response to "list services" API call in
+/// <https://developer.hashicorp.com/consul/api-docs/catalog#list-services>
+pub type ServiceList = HashMap<String, Vec<String>>;
 
+/// Node serving a service, as specified in response to "list nodes for a service" API call in
+/// <https://developer.hashicorp.com/consul/api-docs/catalog#list-nodes-for-service>
 #[derive(Serialize, Deserialize, Debug)]
 #[serde(rename_all = "PascalCase")]
-pub struct ConsulServiceNode {
+pub struct ServiceNode {
     pub node: String,
     pub address: String,
     pub node_meta: HashMap<String, String>,
@@ -52,17 +67,21 @@ pub struct ConsulServiceNode {
     pub service_port: u16,
 }
 
+/// Node serving a service with health info,
+/// as specified in response to "list service instances for a service" health API call in
+/// <https://developer.hashicorp.com/consul/api-docs/health#list-service-instances-for-service>
 #[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(rename_all = "PascalCase")]
-pub struct ConsulHealthServiceNode {
-    pub node: ConsulNode,
-    pub service: ConsulService,
-    pub checks: Vec<ConsulHealthCheck>,
+pub struct HealthServiceNode {
+    pub node: Node,
+    pub service: Service,
+    pub checks: Vec<HealthCheck>,
 }
 
+/// A health check as returned in HealthServiceNode
 #[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(rename_all = "PascalCase")]
-pub struct ConsulHealthCheck {
+pub struct HealthCheck {
     pub node: String,
     #[serde(rename = "CheckID")]
     pub check_id: String,
@@ -73,42 +92,53 @@ pub struct ConsulHealthCheck {
     pub type_: String,
 }
 
-pub type AllServiceHealth = HashMap<String, Arc<[ConsulHealthServiceNode]>>;
+/// Map containing all services and their associated nodes, with health checks,
+/// returned by `watch_all_service_health`
+pub type AllServiceHealth = HashMap<String, Arc<[HealthServiceNode]>>;
 
 impl Consul {
+    /// The "list nodes" API call of the Catalog API
+    ///
+    /// <https://developer.hashicorp.com/consul/api-docs/catalog#list-nodes>
     pub async fn catalog_node_list(
         &self,
         last_index: Option<usize>,
-    ) -> Result<WithIndex<Vec<ConsulNode>>> {
+    ) -> Result<WithIndex<Vec<Node>>> {
         self.get_with_index(format!("{}/v1/catalog/nodes", self.url), last_index)
             .await
     }
 
+    /// The "retrieve map of services for a node" API call of the Catalog API
+    ///
+    /// <https://developer.hashicorp.com/consul/api-docs/catalog#retrieve-map-of-services-for-a-node>
     pub async fn catalog_node(
         &self,
         host: &str,
         last_index: Option<usize>,
-    ) -> Result<WithIndex<Option<ConsulCatalogNode>>> {
+    ) -> Result<WithIndex<Option<CatalogNode>>> {
         self.get_with_index(format!("{}/v1/catalog/node/{}", self.url, host), last_index)
             .await
     }
 
+    /// The "list services" API call of the Catalog api
+    ///
+    /// <https://developer.hashicorp.com/consul/api-docs/catalog#list-services>
     pub async fn catalog_service_list(
         &self,
         last_index: Option<usize>,
-    ) -> Result<WithIndex<ConsulServiceList>> {
-        self.get_with_index::<ConsulServiceList>(
-            format!("{}/v1/catalog/services", self.url),
-            last_index,
-        )
-        .await
+    ) -> Result<WithIndex<ServiceList>> {
+        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
+    ///
+    /// <https://developer.hashicorp.com/consul/api-docs/catalog#list-nodes-for-service>
     pub async fn catalog_service_nodes(
         &self,
         service: &str,
         last_index: Option<usize>,
-    ) -> Result<WithIndex<Vec<ConsulServiceNode>>> {
+    ) -> Result<WithIndex<Vec<ServiceNode>>> {
         self.get_with_index(
             format!("{}/v1/catalog/service/{}", self.url, service),
             last_index,
@@ -116,11 +146,14 @@ impl Consul {
         .await
     }
 
+    /// The "list service instances for a service" API call of the Health api
+    ///
+    /// <https://developer.hashicorp.com/consul/api-docs/health#list-service-instances-for-service>
     pub async fn health_service_instances(
         &self,
         service: &str,
         last_index: Option<usize>,
-    ) -> Result<WithIndex<Vec<ConsulHealthServiceNode>>> {
+    ) -> Result<WithIndex<Vec<HealthServiceNode>>> {
         self.get_with_index(
             format!("{}/v1/health/service/{}", self.url, service),
             last_index,
@@ -128,10 +161,20 @@ impl Consul {
         .await
     }
 
-    pub fn watch_all_service_health(&self) -> watch::Receiver<AllServiceHealth> {
+    /// 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,
+        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
     }
@@ -156,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! {
@@ -173,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))
                         });
                     }
                 }
@@ -197,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)
                         }));
 
@@ -205,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)
                         }));
                     }
@@ -228,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)),
+    )
+}

src/lib.rs

@@ -1,6 +1,6 @@
-mod catalog;
+pub mod catalog;
 mod kv;
-mod locking;
+pub mod locking;
 mod with_index;
 
 use std::fs::File;
@@ -10,14 +10,23 @@ use anyhow::{bail, Result};
 
 pub use with_index::WithIndex;
 
-pub struct ConsulConfig {
+/// Configuration parameters to talk to a Consul server
+pub struct Config {
+    /// HTTP address of the Consul server, with `http://` or `https://` prefix
     pub addr: String,
+    /// CA certificate of the Consul CA, when using TLS
     pub ca_cert: Option<String>,
-    pub tls_skip_verify: bool,
+    /// Client certificate for client auth when using TLS
     pub client_cert: Option<String>,
+    /// Client key for client auth when using TLS
     pub client_key: Option<String>,
+    /// Skip verification of consul server TLS certificates
+    pub tls_skip_verify: bool,
 }
 
+/// Client used to talk to a Consul server.
+/// All calls to the key/value API are automatically prefixed with an arbitrary string
+/// that is constructed at client creation.
 #[derive(Clone)]
 pub struct Consul {
     client: reqwest::Client,
@@ -27,7 +36,7 @@ pub struct Consul {
 }
 
 impl Consul {
-    pub fn new(config: ConsulConfig, kv_prefix: &str) -> Result<Self> {
+    pub fn new(config: Config, kv_prefix: &str) -> Result<Self> {
         let client = match (&config.client_cert, &config.client_key) {
             (Some(client_cert), Some(client_key)) => {
                 let mut client_cert_buf = vec![];

src/locking.rs

@@ -1,3 +1,8 @@
+//! Contains structures to interact with the locks/sessions API
+//!
+//! See <https://developer.hashicorp.com/consul/api-docs/session>
+//! for the full definition of the API.
+
 use anyhow::Result;
 use bytes::Bytes;
 use log::*;
@@ -5,37 +10,33 @@ use serde::{Deserialize, Serialize};
 
 use crate::Consul;
 
+/// Session creation request as specified in
+/// <https://developer.hashicorp.com/consul/api-docs/session#create-session>
 #[derive(Serialize, Deserialize, Debug)]
-pub struct ConsulSessionRequest {
-    #[serde(rename = "Name")]
+#[serde(rename_all = "PascalCase")]
+pub struct SessionRequest {
     pub name: String,
-
-    #[serde(rename = "Node")]
     pub node: Option<String>,
-
-    #[serde(rename = "LockDelay")]
     pub lock_delay: Option<String>,
-
     #[serde(rename = "TTL")]
     pub ttl: Option<String>,
-
-    #[serde(rename = "Behavior")]
     pub behavior: Option<String>,
 }
 
+/// (for internal use, mostly)
 #[derive(Serialize, Deserialize, Debug)]
-pub struct ConsulSessionResponse {
+pub struct SessionResponse {
     #[serde(rename = "ID")]
     pub id: String,
 }
 
 impl Consul {
-    pub async fn create_session(&self, req: &ConsulSessionRequest) -> Result<String> {
+    pub async fn create_session(&self, req: &SessionRequest) -> Result<String> {
         debug!("create_session {:?}", req);
 
         let url = format!("{}/v1/session/create", self.url);
         let http = self.client.put(&url).json(req).send().await?;
-        let resp: ConsulSessionResponse = http.json().await?;
+        let resp: SessionResponse = http.json().await?;
         Ok(resp.id)
     }
 

src/with_index.rs

@@ -3,12 +3,16 @@ use std::fmt::{Debug, Display};
 use anyhow::{bail, Result};
 use reqwest::Response;
 
+/// Wraps the returned value of an [API call with blocking
+/// possibility](https://developer.hashicorp.com/consul/api-docs/features/blocking) with the
+/// returned Consul index
 pub struct WithIndex<T> {
     value: T,
     index: usize,
 }
 
 impl<T> WithIndex<T> {
+    /// (for internal use, mostly)
     pub fn index_from(resp: &Response) -> Result<WithIndexBuilder<T>> {
         let index = match resp.headers().get("X-Consul-Index") {
             Some(v) => v.to_str()?.parse::<usize>()?,
@@ -20,10 +24,13 @@ impl<T> WithIndex<T> {
         })
     }
 
+    /// Returns the inner value, discarding the index
     pub fn into_inner(self) -> T {
         self.value
     }
 
+    /// Returns the Consul index, to be used in future calls to the same API endpoint to make them
+    /// blocking
     pub fn index(&self) -> usize {
         self.index
     }
@@ -60,12 +67,14 @@ impl<T: Display> Display for WithIndex<T> {
     }
 }
 
+/// (for internal use, mostly)
 pub struct WithIndexBuilder<T> {
     _phantom: std::marker::PhantomData<T>,
     index: usize,
 }
 
 impl<T> WithIndexBuilder<T> {
+    /// (for internal use, mostly)
     pub fn value(self, value: T) -> WithIndex<T> {
         WithIndex {
             value,