refactor: Modify error types (#69)

mpsc::Receiver::recv returns a Result<Option<T>, RecvError>. RecvError
has a case `SenderClosed`. So we have 2 different ways to report that
the remote has closed the sender normally. Ok(None) and
Err(RecvError::SenderClosed). This is confusing. SenderClosed isn't
really an error, it is just what you get if the other side closes the
stream.

So this PR creates separate RecvError for oneshot and mpsc.

In oneshot we still need the SenderClosed case since we expect exactly 1
item, and therefore the receiver being closed before that item is an
error.

In mpsc we remove the SenderClosed error and just report Ok(None) in
that case.

Author: rklaehn

irpc-iroh/src/lib.rs

@@ -8,7 +8,7 @@ use iroh::{
     protocol::{AcceptError, ProtocolHandler},
 };
 use irpc::{
-    channel::RecvError,
+    channel::oneshot,
     rpc::{
         Handler, RemoteConnection, RemoteService, ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED,
         MAX_MESSAGE_SIZE,
@@ -257,7 +257,7 @@ pub async fn read_request_raw<R: DeserializeOwned + 'static>(
             ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED.into(),
             b"request exceeded max message size",
         );
-        return Err(RecvError::MaxMessageSizeExceeded.into());
+        return Err(oneshot::RecvError::MaxMessageSizeExceeded.into());
     }
     let mut buf = vec![0; size as usize];
     recv.read_exact(&mut buf)

src/lib.rs

@@ -371,13 +371,47 @@ pub mod channel {
 
     /// Oneshot channel, similar to tokio's oneshot channel
     pub mod oneshot {
-        use std::{fmt::Debug, future::Future, pin::Pin, task};
+        use std::{fmt::Debug, future::Future, io, pin::Pin, task};
 
         use n0_future::future::Boxed as BoxFuture;
 
-        use super::{RecvError, SendError};
+        use super::SendError;
         use crate::util::FusedOneshotReceiver;
 
+        /// Error when receiving a oneshot or mpsc message. For local communication,
+        /// the only thing that can go wrong is that the sender has been closed.
+        ///
+        /// For rpc communication, there can be any number of errors, so this is a
+        /// generic io error.
+        #[derive(Debug, thiserror::Error)]
+        pub enum RecvError {
+            /// The sender has been closed. This is the only error that can occur
+            /// for local communication.
+            #[error("sender closed")]
+            SenderClosed,
+            /// The message exceeded the maximum allowed message size (see [`MAX_MESSAGE_SIZE`]).
+            ///
+            /// [`MAX_MESSAGE_SIZE`]: crate::rpc::MAX_MESSAGE_SIZE
+            #[error("maximum message size exceeded")]
+            MaxMessageSizeExceeded,
+            /// An io error occurred. This can occur for remote communication,
+            /// due to a network error or deserialization error.
+            #[error("io error: {0}")]
+            Io(#[from] io::Error),
+        }
+
+        impl From<RecvError> for io::Error {
+            fn from(e: RecvError) -> Self {
+                match e {
+                    RecvError::Io(e) => e,
+                    RecvError::SenderClosed => io::Error::new(io::ErrorKind::BrokenPipe, e),
+                    RecvError::MaxMessageSizeExceeded => {
+                        io::Error::new(io::ErrorKind::InvalidData, e)
+                    }
+                }
+            }
+        }
+
         /// Create a local oneshot sender and receiver pair.
         ///
         /// This is currently using a tokio channel pair internally.
@@ -586,9 +620,38 @@ pub mod channel {
     ///
     /// For the rpc case, the send side can not be cloned, hence mpsc instead of mpsc.
     pub mod mpsc {
-        use std::{fmt::Debug, future::Future, marker::PhantomData, pin::Pin, sync::Arc};
+        use std::{fmt::Debug, future::Future, io, marker::PhantomData, pin::Pin, sync::Arc};
 
-        use super::{RecvError, SendError};
+        use super::SendError;
+
+        /// Error when receiving a oneshot or mpsc message. For local communication,
+        /// the only thing that can go wrong is that the sender has been closed.
+        ///
+        /// For rpc communication, there can be any number of errors, so this is a
+        /// generic io error.
+        #[derive(Debug, thiserror::Error)]
+        pub enum RecvError {
+            /// The message exceeded the maximum allowed message size (see [`MAX_MESSAGE_SIZE`]).
+            ///
+            /// [`MAX_MESSAGE_SIZE`]: crate::rpc::MAX_MESSAGE_SIZE
+            #[error("maximum message size exceeded")]
+            MaxMessageSizeExceeded,
+            /// An io error occurred. This can occur for remote communication,
+            /// due to a network error or deserialization error.
+            #[error("io error: {0}")]
+            Io(#[from] io::Error),
+        }
+
+        impl From<RecvError> for io::Error {
+            fn from(e: RecvError) -> Self {
+                match e {
+                    RecvError::Io(e) => e,
+                    RecvError::MaxMessageSizeExceeded => {
+                        io::Error::new(io::ErrorKind::InvalidData, e)
+                    }
+                }
+            }
+        }
 
         /// Create a local mpsc sender and receiver pair, with the given buffer size.
         ///
@@ -1067,38 +1130,6 @@ pub mod channel {
             }
         }
     }
-
-    /// Error when receiving a oneshot or mpsc message. For local communication,
-    /// the only thing that can go wrong is that the sender has been closed.
-    ///
-    /// For rpc communication, there can be any number of errors, so this is a
-    /// generic io error.
-    #[derive(Debug, thiserror::Error)]
-    pub enum RecvError {
-        /// The sender has been closed. This is the only error that can occur
-        /// for local communication.
-        #[error("sender closed")]
-        SenderClosed,
-        /// The message exceeded the maximum allowed message size (see [`MAX_MESSAGE_SIZE`]).
-        ///
-        /// [`MAX_MESSAGE_SIZE`]: crate::rpc::MAX_MESSAGE_SIZE
-        #[error("maximum message size exceeded")]
-        MaxMessageSizeExceeded,
-        /// An io error occurred. This can occur for remote communication,
-        /// due to a network error or deserialization error.
-        #[error("io error: {0}")]
-        Io(#[from] io::Error),
-    }
-
-    impl From<RecvError> for io::Error {
-        fn from(e: RecvError) -> Self {
-            match e {
-                RecvError::Io(e) => e,
-                RecvError::SenderClosed => io::Error::new(io::ErrorKind::BrokenPipe, e),
-                RecvError::MaxMessageSizeExceeded => io::Error::new(io::ErrorKind::InvalidData, e),
-            }
-        }
-    }
 }
 
 /// A wrapper for a message with channels to send and receive it.
@@ -1682,8 +1713,10 @@ pub enum Error {
     Request(#[from] RequestError),
     #[error("send error: {0}")]
     Send(#[from] channel::SendError),
-    #[error("recv error: {0}")]
-    Recv(#[from] channel::RecvError),
+    #[error("mpsc recv error: {0}")]
+    MpscRecv(#[from] channel::mpsc::RecvError),
+    #[error("oneshot recv error: {0}")]
+    OneshotRecv(#[from] channel::oneshot::RecvError),
     #[cfg(feature = "rpc")]
     #[error("recv error: {0}")]
     Write(#[from] rpc::WriteError),
@@ -1697,7 +1730,8 @@ impl From<Error> for io::Error {
         match e {
             Error::Request(e) => e.into(),
             Error::Send(e) => e.into(),
-            Error::Recv(e) => e.into(),
+            Error::MpscRecv(e) => e.into(),
+            Error::OneshotRecv(e) => e.into(),
             #[cfg(feature = "rpc")]
             Error::Write(e) => e.into(),
         }
@@ -1772,7 +1806,7 @@ pub mod rpc {
         channel::{
             mpsc::{self, DynReceiver, DynSender},
             none::NoSender,
-            oneshot, RecvError, SendError,
+            oneshot, SendError,
         },
         util::{now_or_never, AsyncReadVarintExt, WriteVarintExt},
         LocalSender, RequestError, RpcMessage, Service,
@@ -1970,16 +2004,13 @@ pub mod rpc {
     impl<T: DeserializeOwned> From<quinn::RecvStream> for oneshot::Receiver<T> {
         fn from(mut read: quinn::RecvStream) -> Self {
             let fut = async move {
-                let size = read
-                    .read_varint_u64()
-                    .await?
-                    .ok_or(RecvError::Io(io::Error::new(
-                        io::ErrorKind::UnexpectedEof,
-                        "failed to read size",
-                    )))?;
+                let size = read.read_varint_u64().await?.ok_or(io::Error::new(
+                    io::ErrorKind::UnexpectedEof,
+                    "failed to read size",
+                ))?;
                 if size > MAX_MESSAGE_SIZE {
                     read.stop(ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED.into()).ok();
-                    return Err(RecvError::MaxMessageSizeExceeded);
+                    return Err(oneshot::RecvError::MaxMessageSizeExceeded);
                 }
                 let rest = read
                     .read_to_end(size as usize)
@@ -2076,7 +2107,12 @@ pub mod rpc {
         fn recv(
             &mut self,
         ) -> Pin<
-            Box<dyn Future<Output = std::result::Result<Option<T>, RecvError>> + Send + Sync + '_>,
+            Box<
+                dyn Future<Output = std::result::Result<Option<T>, mpsc::RecvError>>
+                    + Send
+                    + Sync
+                    + '_,
+            >,
         > {
             Box::pin(async {
                 let read = &mut self.recv;
@@ -2087,7 +2123,7 @@ pub mod rpc {
                     self.recv
                         .stop(ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED.into())
                         .ok();
-                    return Err(RecvError::MaxMessageSizeExceeded);
+                    return Err(mpsc::RecvError::MaxMessageSizeExceeded);
                 }
                 let mut buf = vec![0; size as usize];
                 read.read_exact(&mut buf)
@@ -2375,7 +2411,7 @@ pub mod rpc {
                 ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED.into(),
                 b"request exceeded max message size",
             );
-            return Err(RecvError::MaxMessageSizeExceeded.into());
+            return Err(mpsc::RecvError::MaxMessageSizeExceeded.into());
         }
         let mut buf = vec![0; size as usize];
         recv.read_exact(&mut buf)

tests/mpsc_channel.rs

@@ -6,7 +6,10 @@ use std::{
 };
 
 use irpc::{
-    channel::{mpsc, RecvError, SendError},
+    channel::{
+        mpsc::{self, Receiver, RecvError},
+        SendError,
+    },
     util::AsyncWriteVarintExt,
 };
 use quinn::Endpoint;
@@ -122,7 +125,7 @@ async fn vec_receiver(server: Endpoint) -> Result<(), RecvError> {
         .accept_bi()
         .await
         .map_err(|e| RecvError::Io(e.into()))?;
-    let mut recv = mpsc::Receiver::<Vec<u8>>::from(recv);
+    let mut recv = Receiver::<Vec<u8>>::from(recv);
     while recv.recv().await?.is_some() {}
     Err(RecvError::Io(io::ErrorKind::UnexpectedEof.into()))
 }
@@ -145,7 +148,7 @@ async fn mpsc_max_message_size_send() -> TestResult<()> {
     let Err(cause) = server.await? else {
         panic!("server should have failed due to max message size");
     };
-    assert!(matches!(cause, RecvError::Io(e) if e.kind() == ErrorKind::ConnectionReset));
+    assert!(matches!(cause, mpsc::RecvError::Io(e) if e.kind() == ErrorKind::ConnectionReset));
     Ok(())
 }
 
@@ -165,24 +168,24 @@ async fn mpsc_max_message_size_recv() -> TestResult<()> {
     let Err(cause) = server.await? else {
         panic!("server should have failed due to max message size");
     };
-    assert!(matches!(cause, RecvError::MaxMessageSizeExceeded));
+    assert!(matches!(cause, mpsc::RecvError::MaxMessageSizeExceeded));
     Ok(())
 }
 
-async fn noser_receiver(server: Endpoint) -> Result<(), RecvError> {
+async fn noser_receiver(server: Endpoint) -> Result<(), mpsc::RecvError> {
     let conn = server
         .accept()
         .await
         .unwrap()
         .await
-        .map_err(|e| RecvError::Io(e.into()))?;
+        .map_err(|e| mpsc::RecvError::Io(e.into()))?;
     let (_, recv) = conn
         .accept_bi()
         .await
-        .map_err(|e| RecvError::Io(e.into()))?;
+        .map_err(|e| mpsc::RecvError::Io(e.into()))?;
     let mut recv = mpsc::Receiver::<NoSer>::from(recv);
     while recv.recv().await?.is_some() {}
-    Err(RecvError::Io(io::ErrorKind::UnexpectedEof.into()))
+    Err(mpsc::RecvError::Io(io::ErrorKind::UnexpectedEof.into()))
 }
 
 /// Checks that a serialization error is caught and propagated to the receiver.
@@ -203,7 +206,7 @@ async fn mpsc_serialize_error_send() -> TestResult<()> {
     let Err(cause) = server.await? else {
         panic!("server should have failed due to serialization error");
     };
-    assert!(matches!(cause, RecvError::Io(e) if e.kind() == ErrorKind::ConnectionReset));
+    assert!(matches!(cause, mpsc::RecvError::Io(e) if e.kind() == ErrorKind::ConnectionReset));
     Ok(())
 }
 
@@ -220,6 +223,6 @@ async fn mpsc_serialize_error_recv() -> TestResult<()> {
     let Err(cause) = server.await? else {
         panic!("server should have failed due to serialization error");
     };
-    assert!(matches!(cause, RecvError::Io(e) if e.kind() == ErrorKind::InvalidData));
+    assert!(matches!(cause, mpsc::RecvError::Io(e) if e.kind() == ErrorKind::InvalidData));
     Ok(())
 }

tests/oneshot_channel.rs

@@ -3,7 +3,10 @@
 use std::io::{self, ErrorKind};
 
 use irpc::{
-    channel::{oneshot, RecvError, SendError},
+    channel::{
+        oneshot::{self, RecvError},
+        SendError,
+    },
     util::AsyncWriteVarintExt,
 };
 use quinn::Endpoint;