Fix error: Migration backend error: Canceled

src/bin/coordinator.rs

@@ -53,11 +53,14 @@ fn gen_conf() -> CoordinatorConfig {
     let thread_number = s.get::<usize>("thread_number").unwrap_or_else(|_| 4);
     let thread_number = max(1, thread_number);
 
+    let proxy_timeout = s.get::<usize>("proxy_timeout").unwrap_or_else(|_| 2);
+
     CoordinatorConfig {
         address,
         broker_addresses: Arc::new(ArcSwap::new(Arc::new(broker_address_list))),
         reporter_id,
         thread_number,
+        proxy_timeout,
     }
 }
 
@@ -74,7 +77,7 @@ fn gen_service(
         http_client,
     ));
 
-    let timeout = Duration::new(2, 0);
+    let timeout = Duration::new(config.proxy_timeout as u64, 0);
     let pool_size = 2;
     let client_factory = PooledRedisClientFactory::new(pool_size, timeout);
 

src/common/utils.rs

@@ -4,6 +4,7 @@ use crate::protocol::{BinSafeStr, RespVec};
 use crc16::{State, XMODEM};
 use futures::{stream, Stream};
 use std::cmp::min;
+use std::fmt;
 use std::net::{SocketAddr, ToSocketAddrs};
 use std::str;
 
@@ -246,6 +247,26 @@ pub fn byte_to_uppercase(b: u8) -> u8 {
     }
 }
 
+pub struct RetryError<T> {
+    inner: T,
+}
+
+impl<T: fmt::Debug> fmt::Debug for RetryError<T> {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "RetryError({:?})", self.inner)
+    }
+}
+
+impl<T> RetryError<T> {
+    pub fn new(inner: T) -> Self {
+        Self { inner }
+    }
+
+    pub fn into_inner(self) -> T {
+        self.inner
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/coordinator/service.rs

@@ -30,6 +30,7 @@ pub struct CoordinatorConfig {
     pub broker_addresses: BrokerAddresses,
     pub reporter_id: String,
     pub thread_number: usize,
+    pub proxy_timeout: usize,
 }
 
 impl CoordinatorConfig {

src/migration/scan_task.rs

@@ -596,7 +596,12 @@ where
             );
         }
 
-        self.cmd_handler.handle_cmd_task(cmd_task);
+        if let Err(retry_err) = self.cmd_handler.handle_cmd_task(cmd_task) {
+            return Err(ClusterSendError::Retry(BlockingHintTask::new(
+                retry_err.into_inner(),
+                false,
+            )));
+        }
         Ok(())
     }
 

src/proxy/backend.rs

@@ -566,3 +566,59 @@ impl Error for BackendError {
         }
     }
 }
+
+pub enum SenderBackendError<T> {
+    Io(io::Error),
+    NodeNotFound,
+    InvalidProtocol,
+    InvalidAddress,
+    Canceled,
+    InvalidState,
+    Retry(T),
+}
+
+impl<T> SenderBackendError<T> {
+    pub fn from_backend_error(err: BackendError) -> Self {
+        match err {
+            BackendError::Io(io_err) => SenderBackendError::Io(io_err),
+            BackendError::NodeNotFound => SenderBackendError::NodeNotFound,
+            BackendError::InvalidProtocol => SenderBackendError::InvalidProtocol,
+            BackendError::InvalidAddress => SenderBackendError::InvalidAddress,
+            BackendError::Canceled => SenderBackendError::Canceled,
+            BackendError::InvalidState => SenderBackendError::InvalidState,
+        }
+    }
+}
+
+impl<T: fmt::Debug> fmt::Debug for SenderBackendError<T> {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            Self::Io(io_err) => write!(f, "BackendError::Io({:?})", io_err),
+            Self::NodeNotFound => write!(f, "backendError::NodeNotFound"),
+            Self::InvalidProtocol => write!(f, "backendError::InvalidProtocol"),
+            Self::InvalidAddress => write!(f, "backendError::InvalidAddress"),
+            Self::Canceled => write!(f, "backendError::Canceled"),
+            Self::InvalidState => write!(f, "backendError::InvalidState"),
+            Self::Retry(task) => write!(f, "BackendError::Retry({:?})", task),
+        }
+    }
+}
+
+impl<T: fmt::Debug> fmt::Display for SenderBackendError<T> {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "{:?}", self)
+    }
+}
+
+impl<T: fmt::Debug> Error for SenderBackendError<T> {
+    fn description(&self) -> &str {
+        "sender backend error"
+    }
+
+    fn cause(&self) -> Option<&dyn Error> {
+        match self {
+            Self::Io(err) => Some(err),
+            _ => None,
+        }
+    }
+}

src/proxy/cluster.rs

@@ -582,6 +582,7 @@ pub enum ClusterSendError<T: CmdTask> {
         slot: usize,
         address: String,
     },
+    Retry(T),
 }
 
 impl<T: CmdTask> fmt::Display for ClusterSendError<T> {
@@ -605,6 +606,7 @@ impl<T: CmdTask> fmt::Debug for ClusterSendError<T> {
             Self::ActiveRedirection { slot, address, .. } => {
                 format!("ClusterSendError::Moved({} {})", slot, address)
             }
+            Self::Retry(_) => "ClusterSendError::Retry".to_string(),
         };
         write!(f, "{}", s)
     }
@@ -620,6 +622,7 @@ impl<T: CmdTask> Error for ClusterSendError<T> {
             Self::SlotNotCovered => None,
             Self::MigrationError => None,
             Self::ActiveRedirection { .. } => None,
+            Self::Retry(_) => None,
         }
     }
 }
@@ -649,6 +652,7 @@ impl<T: CmdTask> ClusterSendError<T> {
                 slot,
                 address,
             },
+            Self::Retry(task) => ClusterSendError::Retry(f(task)),
         }
     }
 }

src/proxy/manager.rs

@@ -17,7 +17,7 @@ use crate::common::config::ClusterConfig;
 use crate::common::proto::ProxyClusterMeta;
 use crate::common::response;
 use crate::common::track::TrackedFutureRegistry;
-use crate::common::utils::gen_moved;
+use crate::common::utils::{gen_moved, RetryError};
 use crate::migration::manager::{MigrationManager, MigrationMap, SwitchError};
 use crate::migration::task::MgrSubCmd;
 use crate::migration::task::SwitchArg;
@@ -304,7 +304,7 @@ impl<F: RedisClientFactory, C: ConnFactory<Pkt = RespPacket>> MetaManager<F, C>
     pub fn send(&self, cmd_ctx: CmdCtx) {
         let max_redirections = self.config.max_redirections;
         let default_redirection_address = self.config.default_redirection_address.as_ref();
-        send_cmd_ctx(
+        loop_send_cmd_ctx(
             &self.meta_map,
             cmd_ctx,
             max_redirections,
@@ -329,6 +329,11 @@ impl<F: RedisClientFactory, C: ConnFactory<Pkt = RespPacket>> MetaManager<F, C>
                         b"unexpected active redirection".to_vec(),
                     )));
                 }
+                ClusterSendError::Retry(task) => {
+                    task.set_resp_result(Ok(Resp::Error(
+                        b"unexpected retry error on sync task".to_vec(),
+                    )));
+                }
                 other_err => {
                     error!("Failed to process sync task {:?}", other_err);
                 }
@@ -373,15 +378,42 @@ impl<F: RedisClientFactory, C: ConnFactory<Pkt = RespPacket>> MetaManager<F, C>
     }
 }
 
-pub fn send_cmd_ctx<C: ConnFactory<Pkt = RespPacket>>(
+pub fn loop_send_cmd_ctx<C: ConnFactory<Pkt = RespPacket>>(
     meta_map: &SharedMetaMap<C>,
     cmd_ctx: CmdCtx,
     max_redirections: Option<NonZeroUsize>,
     default_redirection_address: Option<&String>,
 ) {
+    let mut cmd_ctx = cmd_ctx;
+    const MAX_RETRY_NUM: usize = 10;
+    for i in 0..MAX_RETRY_NUM {
+        cmd_ctx = match send_cmd_ctx(
+            meta_map,
+            cmd_ctx,
+            max_redirections,
+            default_redirection_address,
+        ) {
+            Ok(()) => return,
+            Err(retry_err) => retry_err.into_inner(),
+        };
+        if i + 1 == MAX_RETRY_NUM {
+            let resp = Resp::Error(b"cmd exceeds retry limit".to_vec());
+            cmd_ctx.set_resp_result(Ok(resp));
+            return;
+        }
+        info!("retry send cmd_ctx");
+    }
+}
+
+pub fn send_cmd_ctx<C: ConnFactory<Pkt = RespPacket>>(
+    meta_map: &SharedMetaMap<C>,
+    cmd_ctx: CmdCtx,
+    max_redirections: Option<NonZeroUsize>,
+    default_redirection_address: Option<&String>,
+) -> Result<(), RetryError<CmdCtx>> {
     let meta_map = meta_map.lease();
     let mut cmd_ctx = match meta_map.migration_map.send(cmd_ctx) {
-        Ok(()) => return,
+        Ok(()) => return Ok(()),
         Err(e) => match e {
             ClusterSendError::SlotNotFound(cmd_ctx) => cmd_ctx,
             ClusterSendError::ActiveRedirection {
@@ -397,11 +429,12 @@ pub fn send_cmd_ctx<C: ConnFactory<Pkt = RespPacket>>(
                     address,
                     max_redirections,
                 );
-                return;
+                return Ok(());
             }
+            ClusterSendError::Retry(cmd_ctx) => return Err(RetryError::new(cmd_ctx.into_inner())),
             err => {
                 error!("migration send task failed: {:?}", err);
-                return;
+                return Ok(());
             }
         },
     };
@@ -448,11 +481,16 @@ pub fn send_cmd_ctx<C: ConnFactory<Pkt = RespPacket>>(
                     address,
                     max_redirections,
                 );
-                return;
+                return Ok(());
+            }
+            ClusterSendError::Retry(cmd_ctx) => {
+                return Err(RetryError::new(cmd_ctx.into_inner()));
             }
             err => warn!("Failed to forward cmd_ctx: {:?}", err),
         }
     }
+
+    Ok(())
 }
 
 fn send_cmd_ctx_to_remote_directly<C: ConnFactory<Pkt = RespPacket>>(
@@ -520,7 +558,7 @@ impl<C: ConnFactory<Pkt = RespPacket>> CmdTaskSender for BlockingTaskRetrySender
     type Task = CmdCtx;
 
     fn send(&self, cmd_task: Self::Task) -> Result<(), BackendError> {
-        send_cmd_ctx(
+        loop_send_cmd_ctx(
             &self.meta_map,
             cmd_task,
             self.max_redirections,