Refactor/curp client retry

curp/src/client_new/mod.rs

@@ -4,6 +4,9 @@
 /// Unary rpc client
 mod unary;
 
+/// Retry layer
+mod retry;
+
 use std::{collections::HashMap, sync::Arc};
 
 use async_trait::async_trait;
@@ -71,20 +74,20 @@
 
     /// Fetch leader id
     #[inline]
     async fn fetch_leader_id(&self, linearizable: bool) -> Result<ServerId, Self::Error> {
         if linearizable {
             let resp = self.fetch_cluster(true).await?;
             return Ok(resp.leader_id.unwrap_or_else(|| {
                 unreachable!("linearizable fetch cluster should return a leader id")
             }));
         }
         let resp = self.fetch_cluster(false).await?;
         if let Some(id) = resp.leader_id {
             return Ok(id);
         }
         // fallback to linearizable fetch
         self.fetch_leader_id(true).await
     }
 }
 
 /// Update leader state
@@ -95,7 +98,7 @@
 }
 
 /// Client builder to build a client
 #[derive(Debug, Clone)]
 pub struct ClientBuilder {
     /// local server id
     local_server_id: Option<ServerId>,
@@ -111,30 +114,30 @@
     /// Create a client builder
     #[inline]
     #[must_use]
     pub fn new(config: ClientConfig) -> Self {
         Self {
             local_server_id: None,
             config,
             all_members: None,
             cluster_version: None,
         }
     }
 
     /// Set the local server id
     #[inline]
     #[must_use]
     pub fn local_server_id(&mut self, id: ServerId) -> &mut Self {
         self.local_server_id = Some(id);
         self
     }
 
     /// Set the initial cluster version
     #[inline]
     #[must_use]
     pub fn cluster_version(&mut self, cluster_version: u64) -> &mut Self {
         self.cluster_version = Some(cluster_version);
         self
     }
 
     /// Fetch initial all members from some endpoints if you do not know the whole members
     ///
@@ -142,47 +145,47 @@
     ///
     /// Return `tonic::Status` for connection failure or some server errors.
     #[inline]
     pub async fn fetch_all_members(
         &mut self,
         addrs: Vec<String>,
     ) -> Result<&mut Self, tonic::Status> {
         let propose_timeout = *self.config.propose_timeout();
         let mut futs: FuturesUnordered<_> = addrs
             .into_iter()
             .map(|mut addr| {
                 if !addr.starts_with("http://") {
                     addr.insert_str(0, "http://");
                 }
                 async move {
                     let mut protocol_client = ProtocolClient::connect(addr).await.map_err(|e| {
                         tonic::Status::cancelled(format!("cannot connect to addr, error: {e}"))
                     })?;
                     let mut req = tonic::Request::new(FetchClusterRequest::default());
                     req.set_timeout(propose_timeout);
                     let fetch_cluster_res = protocol_client.fetch_cluster(req).await?.into_inner();
                     Ok::<FetchClusterResponse, tonic::Status>(fetch_cluster_res)
                 }
             })
             .collect();
         let mut err = tonic::Status::invalid_argument("addrs is empty");
         while let Some(r) = futs.next().await {
             match r {
                 Ok(r) => {
                     self.cluster_version = Some(r.cluster_version);
                     self.all_members = Some(r.into_members_addrs());
                     return Ok(self);
                 }
                 Err(e) => err = e,
             }
         }
         Err(err)
     }
 
     /// Set the initial all members
     #[inline]
     #[must_use]
     pub fn set_all_members(&mut self, all_members: HashMap<ServerId, Vec<String>>) -> &mut Self {
         self.all_members = Some(all_members);
         self
     }
 }

curp/src/client_new/retry.rs

@@ -0,0 +1,247 @@
+#![allow(unused)] // TODO: remove
+
+use std::{marker::PhantomData, ops::SubAssign, sync::Arc, time::Duration};
+
+use async_trait::async_trait;
+use curp_external_api::cmd::Command;
+use futures::Future;
+use tracing::warn;
+
+use crate::{
+    members::ServerId,
+    rpc::{
+        connect::ConnectApi, ConfChange, CurpError, FetchClusterResponse, Member, ReadState,
+        Redirect,
+    },
+};
+
+use super::{ClientApi, LeaderStateUpdate, ProposeResponse};
+
+/// Backoff config
+#[derive(Debug, Clone)]
+pub(super) enum BackoffConfig {
+    /// A fixed delay backoff
+    Fixed,
+    /// A exponential delay backoff
+    Exponential {
+        /// Control the max delay of exponential
+        max_delay: Duration,
+    },
+}
+
+/// Retry config to control the retry policy
+#[derive(Debug, Clone)]
+pub(super) struct RetryConfig {
+    /// Backoff config
+    backoff: BackoffConfig,
+    /// Initial delay
+    delay: Duration,
+    /// Retry count
+    count: usize,
+}
+
+/// Backoff tool
+#[derive(Debug)]
+struct Backoff {
+    /// The retry config
+    config: RetryConfig,
+    /// Current delay
+    cur_delay: Duration,
+    /// Total RPC count
+    count: usize,
+}
+
+impl RetryConfig {
+    /// Create a fixed retry config
+    fn new_fixed(delay: Duration, count: usize) -> Self {
+        assert!(count > 0, "retry count should be larger than 0");
+        Self {
+            backoff: BackoffConfig::Fixed,
+            delay,
+            count,
+        }
+    }
+
+    /// Create a exponential retry config
+    fn new_exponential(delay: Duration, max_delay: Duration, count: usize) -> Self {
+        assert!(count > 0, "retry count should be larger than 0");
+        Self {
+            backoff: BackoffConfig::Exponential { max_delay },
+            delay,
+            count,
+        }
+    }
+
+    /// Create a backoff process
+    fn init_backoff(&self) -> Backoff {
+        Backoff {
+            config: self.clone(),
+            cur_delay: self.delay,
+            count: self.count,
+        }
+    }
+}
+
+impl Backoff {
+    /// Get the next delay duration, None means the end.
+    fn next_delay(&mut self) -> Option<Duration> {
+        if self.count == 0 {
+            return None;
+        }
+        self.count.sub_assign(1);
+        let mut cur = self.cur_delay;
+        if let BackoffConfig::Exponential { max_delay } = self.config.backoff {
+            self.cur_delay = self
+                .cur_delay
+                .checked_mul(2)
+                .unwrap_or(self.cur_delay)
+                .min(max_delay);
+        }
+        Some(cur)
+    }
+}
+
+/// The retry client automatically retry the requests of the inner client api
+/// which raises the [`tonic::Status`] error
+#[derive(Debug)]
+pub(super) struct Retry<Api> {
+    /// Inner client
+    inner: Api,
+    /// Retry config
+    config: RetryConfig,
+}
+
+impl<Api> Retry<Api>
+where
+    Api: ClientApi<Error = CurpError> + LeaderStateUpdate + Send + Sync + 'static,
+{
+    /// Create a retry client
+    fn new(inner: Api, config: RetryConfig) -> Self {
+        Self { inner, config }
+    }
+
+    /// Takes a function f and run retry.
+    async fn retry<'a, R, F>(&'a self, f: impl Fn(&'a Api) -> F) -> Result<R, tonic::Status>
+    where
+        F: Future<Output = Result<R, CurpError>>,
+    {
+        let mut backoff = self.config.init_backoff();
+        while let Some(delay) = backoff.next_delay() {
+            let err = match f(&self.inner).await {
+                Ok(res) => return Ok(res),
+                Err(err) => err,
+            };
+
+            match err {
+                // some errors that should not retry
+                CurpError::Duplicated(_)
+                | CurpError::ShuttingDown(_)
+                | CurpError::InvalidConfig(_)
+                | CurpError::NodeNotExists(_)
+                | CurpError::NodeAlreadyExists(_)
+                | CurpError::LearnerNotCatchUp(_) => {
+                    return Err(tonic::Status::from(err));
+                }
+
+                // some errors that could have a retry
+                CurpError::ExpiredClientId(_)
+                | CurpError::KeyConflict(_)
+                | CurpError::RpcTransport(_)
+                | CurpError::Internal(_) => {}
+
+                // update the cluster state if got WrongClusterVersion
+                CurpError::WrongClusterVersion(_) => {
+                    // the inner client should automatically update cluster state when fetch_cluster
+                    if let Err(e) = self.inner.fetch_cluster(false).await {
+                        warn!("fetch cluster failed, error {e:?}");
+                    }
+                }
+
+                // update the leader state if got Redirect
+                CurpError::Redirect(Redirect { leader_id, term }) => {
+                    self.inner.update_leader(leader_id, term);
+                }
+            }
+
+            warn!("retry on {} seconds later", delay.as_secs_f32());
+            tokio::time::sleep(delay).await;
+        }
+
+        Err(tonic::Status::deadline_exceeded("request timeout"))
+    }
+}
+
+#[async_trait]
+impl<Api> ClientApi for Retry<Api>
+where
+    Api: ClientApi<Error = CurpError> + LeaderStateUpdate + Send + Sync + 'static,
+{
+    /// The client error
+    type Error = tonic::Status;
+
+    /// Inherit the command type
+    type Cmd = Api::Cmd;
+
+    /// Get the local connection when the client is on the server node.
+    async fn local_connect(&self) -> Option<Arc<dyn ConnectApi>> {
+        self.inner.local_connect().await
+    }
+
+    /// Send propose to the whole cluster, `use_fast_path` set to `false` to fallback into ordered
+    /// requests (event the requests are commutative).
+    async fn propose(
+        &self,
+        cmd: &Self::Cmd,
+        use_fast_path: bool,
+    ) -> Result<ProposeResponse<Self::Cmd>, tonic::Status> {
+        self.retry::<_, _>(|client| client.propose(cmd, use_fast_path))
+            .await
+    }
+
+    /// Send propose configuration changes to the cluster
+    async fn propose_conf_change(
+        &self,
+        changes: Vec<ConfChange>,
+    ) -> Result<Vec<Member>, tonic::Status> {
+        self.retry::<_, _>(|client| {
+            let changes_c = changes.clone();
+            async move { client.propose_conf_change(changes_c).await }
+        })
+        .await
+    }
+
+    /// Send propose to shutdown cluster
+    async fn propose_shutdown(&self) -> Result<(), tonic::Status> {
+        self.retry::<_, _>(ClientApi::propose_shutdown).await
+    }
+
+    /// Send propose to publish a node id and name
+    async fn propose_publish(
+        &self,
+        node_id: ServerId,
+        node_name: String,
+    ) -> Result<(), Self::Error> {
+        self.retry::<_, _>(|client| {
+            let name_c = node_name.clone();
+            async move { client.propose_publish(node_id, name_c).await }
+        })
+        .await
+    }
+
+    /// Send fetch read state from leader
+    async fn fetch_read_state(&self, cmd: &Self::Cmd) -> Result<ReadState, tonic::Status> {
+        self.retry::<_, _>(|client| client.fetch_read_state(cmd))
+            .await
+    }
+
+    /// Send fetch cluster requests to all servers (That's because initially, we didn't
+    /// know who the leader is.)
+    /// Note: The fetched cluster may still be outdated if `linearizable` is false
+    async fn fetch_cluster(
+        &self,
+        linearizable: bool,
+    ) -> Result<FetchClusterResponse, tonic::Status> {
+        self.retry::<_, _>(|client| client.fetch_cluster(linearizable))
+            .await
+    }
+}

curp/src/rpc/mod.rs

@@ -636,8 +636,8 @@
     }
 
     /// Get the priority of the error
     pub(crate) fn priority(&self) -> CurpErrorPriority {
         match *self {
             CurpError::Duplicated(_)
             | CurpError::ShuttingDown(_)
             | CurpError::InvalidConfig(_)
@@ -645,23 +645,30 @@
             | CurpError::NodeNotExists(_)
             | CurpError::LearnerNotCatchUp(_)
             | CurpError::ExpiredClientId(_)
             | CurpError::Redirect(_) => CurpErrorPriority::ReturnImmediately,
             CurpError::WrongClusterVersion(_) => CurpErrorPriority::High,
             CurpError::RpcTransport(_) | CurpError::Internal(_) | CurpError::KeyConflict(_) => {
                 CurpErrorPriority::Low
             }
         }
     }
 }
 
 /// The priority of curp error, indicate which error should be handled in retry layer
 #[derive(Debug, PartialEq, Eq, PartialOrd, Ord)]
 pub(crate) enum CurpErrorPriority {
-    /// Low priority
+    /// Low priority, in multiple sequenced RPCs, if preceding RPCs returns
+    /// a low-priority error, will not exit prematurely. In concurrent RPCs,
+    /// a low-priority error returned may be overridden by a higher-priority error.
     Low = 1,
-    /// High priority
+    /// High priority, in multiple sequenced RPCs, if preceding RPCs returns
+    /// a high-priority error, it will exit early, preventing next RPCs from
+    /// proceeding. In concurrent RPCs, high-priority errors will override
+    /// low-priority errors.
     High = 2,
-    /// Should be returned immediately if any server response it
+    /// Should be returned immediately if any server response it. If requests are
+    /// sent to multiple servers, the request that does not have received a
+    /// response will be terminated immediately.
     ReturnImmediately = 3,
 }
 
@@ -783,7 +790,7 @@
     pub(crate) fn is_conflict_with_cmd(&self, c: &C) -> bool {
         match self.inner {
             PoolEntryInner::Command(ref cmd) => cmd.is_conflict(c),
             PoolEntryInner::ConfChange(ref _conf_change) => true,
         }
     }
 }
@@ -794,7 +801,7 @@
 {
     fn is_conflict(&self, other: &Self) -> bool {
         let PoolEntryInner::Command(ref cmd1) = self.inner else {
             return true;
         };
         let PoolEntryInner::Command(ref cmd2) = other.inner else {
             return true;