Change: Remove RemoteError variant from StreamingError

openraft/src/error/decompose.rs

@@ -1,11 +1,9 @@
 use std::error::Error;
 
 use crate::error::into_ok::into_ok;
-use crate::error::Fatal;
 use crate::error::Infallible;
 use crate::error::RPCError;
 use crate::error::RaftError;
-use crate::error::StreamingError;
 use crate::error::Unreachable;
 use crate::RaftTypeConfig;
 
@@ -72,24 +70,3 @@ where
         }
     }
 }
-
-impl<C, R> DecomposeResult<C, R, StreamingError<C>> for Result<R, StreamingError<C, Fatal<C>>>
-where C: RaftTypeConfig
-{
-    type InnerError = Infallible;
-
-    /// `Fatal` is considered as `RPCError::Unreachable`.
-    fn decompose(self) -> Result<Result<R, Self::InnerError>, StreamingError<C>> {
-        match self {
-            Ok(r) => Ok(Ok(r)),
-            Err(e) => match e {
-                StreamingError::Closed(e) => Err(StreamingError::Closed(e)),
-                StreamingError::StorageError(e) => Err(StreamingError::StorageError(e)),
-                StreamingError::Timeout(e) => Err(StreamingError::Timeout(e)),
-                StreamingError::Unreachable(e) => Err(StreamingError::Unreachable(e)),
-                StreamingError::Network(e) => Err(StreamingError::Network(e)),
-                StreamingError::RemoteError(e) => Err(StreamingError::Unreachable(Unreachable::new(&e.source))),
-            },
-        }
-    }
-}

openraft/src/error/streaming_error.rs

@@ -1,10 +1,5 @@
-use std::error::Error;
-
-use crate::error::Fatal;
-use crate::error::Infallible;
 use crate::error::NetworkError;
 use crate::error::RPCError;
-use crate::error::RemoteError;
 use crate::error::ReplicationClosed;
 use crate::error::ReplicationError;
 use crate::error::Timeout;
@@ -16,13 +11,8 @@ use crate::StorageError;
 ///
 /// Thus this error includes storage error, network error, and remote error.
 #[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
-#[cfg_attr(
-    feature = "serde",
-    derive(serde::Serialize, serde::Deserialize),
-    serde(bound(serialize = "E: serde::Serialize")),
-    serde(bound(deserialize = "E: for <'d> serde::Deserialize<'d>"))
-)]
-pub enum StreamingError<C: RaftTypeConfig, E: Error = Infallible> {
+#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
+pub enum StreamingError<C: RaftTypeConfig> {
     /// The replication stream is closed intentionally.
     #[error(transparent)]
     Closed(#[from] ReplicationClosed),
@@ -42,24 +32,16 @@ pub enum StreamingError<C: RaftTypeConfig, E: Error = Infallible> {
     /// Failed to send the RPC request and should retry immediately.
     #[error(transparent)]
     Network(#[from] NetworkError),
-
-    /// Remote node returns an error.
-    #[error(transparent)]
-    RemoteError(#[from] RemoteError<C, E>),
 }
 
-impl<C: RaftTypeConfig> From<StreamingError<C, Fatal<C>>> for ReplicationError<C> {
-    fn from(e: StreamingError<C, Fatal<C>>) -> Self {
+impl<C: RaftTypeConfig> From<StreamingError<C>> for ReplicationError<C> {
+    fn from(e: StreamingError<C>) -> Self {
         match e {
             StreamingError::Closed(e) => ReplicationError::Closed(e),
             StreamingError::StorageError(e) => ReplicationError::StorageError(e),
             StreamingError::Timeout(e) => ReplicationError::RPCError(RPCError::Timeout(e)),
             StreamingError::Unreachable(e) => ReplicationError::RPCError(RPCError::Unreachable(e)),
             StreamingError::Network(e) => ReplicationError::RPCError(RPCError::Network(e)),
-            StreamingError::RemoteError(e) => {
-                // Fatal on remote error is considered as unreachable.
-                ReplicationError::RPCError(RPCError::Unreachable(Unreachable::new(&e.source)))
-            }
         }
     }
 }
@@ -74,22 +56,8 @@ impl<C: RaftTypeConfig> From<RPCError<C>> for StreamingError<C> {
                 unreachable!("PayloadTooLarge should not be converted to StreamingError")
             }
             RPCError::Network(e) => StreamingError::Network(e),
-            RPCError::RemoteError(e) => StreamingError::RemoteError(e),
-        }
-    }
-}
-
-impl<C: RaftTypeConfig> From<StreamingError<C>> for ReplicationError<C> {
-    fn from(e: StreamingError<C>) -> Self {
-        #[allow(unreachable_patterns)]
-        match e {
-            StreamingError::Closed(e) => ReplicationError::Closed(e),
-            StreamingError::StorageError(e) => ReplicationError::StorageError(e),
-            StreamingError::Timeout(e) => ReplicationError::RPCError(RPCError::Timeout(e)),
-            StreamingError::Unreachable(e) => ReplicationError::RPCError(RPCError::Unreachable(e)),
-            StreamingError::Network(e) => ReplicationError::RPCError(RPCError::Network(e)),
-            StreamingError::RemoteError(_e) => {
-                unreachable!("Infallible error should not be converted to ReplicationError")
+            RPCError::RemoteError(_e) => {
+                unreachable!("Infallible error should not be produced at all")
             }
         }
     }

openraft/src/network/snapshot_transport.rs

@@ -18,7 +18,6 @@ mod tokio_rt {
     use super::Chunked;
     use super::SnapshotTransport;
     use super::Streaming;
-    use crate::error::Fatal;
     use crate::error::InstallSnapshotError;
     use crate::error::RPCError;
     use crate::error::RaftError;
@@ -50,7 +49,7 @@ mod tokio_rt {
             mut snapshot: Snapshot<C>,
             mut cancel: impl Future<Output = ReplicationClosed> + OptionalSend + 'static,
             option: RPCOption,
-        ) -> Result<SnapshotResponse<C>, StreamingError<C, Fatal<C>>>
+        ) -> Result<SnapshotResponse<C>, StreamingError<C>>
         where
             Net: RaftNetwork<C> + ?Sized,
         {
@@ -264,7 +263,6 @@ use std::future::Future;
 
 use openraft_macros::add_async_trait;
 
-use crate::error::Fatal;
 use crate::error::InstallSnapshotError;
 use crate::error::RaftError;
 use crate::error::ReplicationClosed;
@@ -304,7 +302,7 @@ pub trait SnapshotTransport<C: RaftTypeConfig> {
         snapshot: Snapshot<C>,
         cancel: impl Future<Output = ReplicationClosed> + OptionalSend + 'static,
         option: RPCOption,
-    ) -> Result<SnapshotResponse<C>, StreamingError<C, Fatal<C>>>
+    ) -> Result<SnapshotResponse<C>, StreamingError<C>>
     where
         Net: RaftNetwork<C> + ?Sized;
 

openraft/src/network/v2/adapt_v1.rs

@@ -46,7 +46,7 @@ where
         use crate::network::snapshot_transport::Chunked;
         use crate::network::snapshot_transport::SnapshotTransport;
 
-        let resp = Chunked::send_snapshot(self, vote, snapshot, cancel, option).await.decompose_infallible()?;
+        let resp = Chunked::send_snapshot(self, vote, snapshot, cancel, option).await?;
         Ok(resp)
     }