chore: add error priority

Signed-off-by: iGxnon <[email protected]>

curp/src/client_new/unary.rs

@@ -23,9 +23,9 @@ use crate::{
     rpc::{
         self,
         connect::{BypassedConnect, ConnectApi},
-        ConfChange, CurpError, FetchClusterRequest, FetchClusterResponse, FetchReadStateRequest,
-        Member, ProposeConfChangeRequest, ProposeId, ProposeRequest, Protocol, PublishRequest,
-        ReadState, ShutdownRequest, WaitSyncedRequest,
+        ConfChange, CurpError, CurpErrorPriority, FetchClusterRequest, FetchClusterResponse,
+        FetchReadStateRequest, Member, ProposeConfChangeRequest, ProposeId, ProposeRequest,
+        Protocol, PublishRequest, ReadState, ShutdownRequest, WaitSyncedRequest,
     },
 };
 
@@ -254,7 +254,7 @@ impl<C: Command> Unary<C> {
             .await;
         let super_quorum = super_quorum(size);
 
-        let mut err = None;
+        let mut err: Option<CurpError> = None;
         let mut execute_result: Option<C::ER> = None;
         let mut ok_cnt = 0;
 
@@ -263,10 +263,16 @@ impl<C: Command> Unary<C> {
                 Ok(resp) => resp.into_inner(),
                 Err(e) => {
                     warn!("propose cmd({propose_id}) to server({id}) error: {e:?}");
-                    if e.return_early() {
+                    if e.priority() == CurpErrorPriority::ReturnImmediately {
                         return Err(e);
                     }
-                    err = Some(e);
+                    if let Some(old_err) = err.as_ref() {
+                        if old_err.priority() <= e.priority() {
+                            err = Some(e);
+                        }
+                    } else {
+                        err = Some(e);
+                    }
                     continue;
                 }
             };
@@ -384,7 +390,7 @@ impl<C: Command> ClientApi for Unary<C> {
                 futures::future::Either::Left((fast_result, slow_round)) => match fast_result {
                     Ok(er) => er.map(|e| (e, None)),
                     Err(err) => {
-                        if err.return_early() {
+                        if err.priority() > CurpErrorPriority::Low {
                             return Err(err);
                         }
                         // fallback to slow round if fast round failed
@@ -395,7 +401,7 @@ impl<C: Command> ClientApi for Unary<C> {
                 futures::future::Either::Right((slow_result, fast_round)) => match slow_result {
                     Ok(er) => er.map(|(asr, e)| (e, Some(asr))),
                     Err(err) => {
-                        if err.return_early() {
+                        if err.priority() > CurpErrorPriority::Low {
                             return Err(err);
                         }
                         let fr = fast_round.await?;
@@ -406,7 +412,7 @@ impl<C: Command> ClientApi for Unary<C> {
         } else {
             let (fr, sr) = futures::future::join(fast_round, slow_round).await;
             if let Err(err) = fr {
-                if err.return_early() {
+                if err.priority() > CurpErrorPriority::Low {
                     return Err(err);
                 }
             }
@@ -515,17 +521,23 @@ impl<C: Command> ClientApi for Unary<C> {
         let mut max_term = 0;
         let mut res = None;
         let mut ok_cnt = 0;
-        let mut err = None;
+        let mut err: Option<CurpError> = None;
 
         while let Some((id, resp)) = responses.next().await {
             let inner = match resp {
                 Ok(r) => r,
                 Err(e) => {
                     warn!("fetch cluster from {} failed, {:?}", id, e);
-                    if e.return_early() {
+                    if e.priority() == CurpErrorPriority::ReturnImmediately {
                         return Err(e);
                     }
-                    err = Some(e);
+                    if let Some(old_err) = err.as_ref() {
+                        if old_err.priority() <= e.priority() {
+                            err = Some(e);
+                        }
+                    } else {
+                        err = Some(e);
+                    }
                     continue;
                 }
             };

curp/src/rpc/mod.rs

@@ -635,24 +635,36 @@ impl CurpError {
         Self::Internal(reason.into())
     }
 
-    /// Errors that should return early to the retry layer when we
-    /// got the error at propose stage
-    pub(crate) fn return_early(&self) -> bool {
-        matches!(
-            *self,
+    /// Get the priority of the error
+    pub(crate) fn priority(&self) -> CurpErrorPriority {
+        match *self {
             CurpError::Duplicated(_)
-                | CurpError::ShuttingDown(_)
-                | CurpError::InvalidConfig(_)
-                | CurpError::NodeAlreadyExists(_)
-                | CurpError::NodeNotExists(_)
-                | CurpError::LearnerNotCatchUp(_)
-                | CurpError::ExpiredClientId(_)
-                | CurpError::WrongClusterVersion(_)
-                | CurpError::Redirect(_)
-        )
+            | CurpError::ShuttingDown(_)
+            | CurpError::InvalidConfig(_)
+            | CurpError::NodeAlreadyExists(_)
+            | 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 = 1,
+    /// High priority
+    High = 2,
+    /// Should be returned immediately if any server response it
+    ReturnImmediately = 3,
+}
+
 impl<E: std::error::Error + 'static> From<E> for CurpError {
     #[inline]
     fn from(value: E) -> Self {