feat!: define error codes in an enum and expose them in new variants

Cargo.toml

@@ -34,8 +34,8 @@ postcard = { workspace = true, features = ["alloc", "use-std"], optional = true
 noq = { workspace = true, optional = true }
 # used as a buffer for serialization when using rpc
 smallvec = { version = "1.14.0", features = ["write"], optional = true }
-# used for the Debug impl of rpc::Handler
-derive_more = { version = "2", features = ["debug"], optional = true }
+# used for the Debug impl of rpc::Handler and the Display impl of rpc::ErrorCode
+derive_more = { version = "2", features = ["debug", "display"], optional = true }
 # used in the test utils to generate noq endpoints
 rustls = { version = "0.23.5", default-features = false, features = ["std"], optional = true }
 # used in the test utils to generate noq endpoints

src/channel.rs

@@ -2,17 +2,46 @@
 
 use std::io;
 
+#[cfg(feature = "rpc")]
+use n0_error::e;
 use n0_error::stack_error;
 
+#[cfg(feature = "rpc")]
+use crate::rpc::{self, ErrorCode};
+
 pub mod mpsc;
 pub mod none;
 pub mod oneshot;
 
-/// Error when sending a oneshot or mpsc message. For local communication,
-/// the only thing that can go wrong is that the receiver has been dropped.
+/// Error when sending a oneshot or mpsc message.
+///
+/// For a local channel, the only error is [`ReceiverClosed`].
+///
+/// For a remote channel, the receiver is on the other side of the connection:
+/// the server for the updates of a request, and the client for a response. A
+/// send fails for one of two reasons:
+///
+/// * This sender found the message bad: it is too large
+///   ([`MaxSizeExceeded`]), or it does not encode (`EncodeFailed`).
+/// * Something other than this message ended the channel:
+///   * The receiver closed it ([`ReceiverClosed`]).
+///   * The receiver stopped the stream with an irpc error code ([`Remote`]).
+///     For example, the server rejected the request, or the receiver could not
+///     decode an earlier message.
+///   * The connection failed ([`Io`]).
 ///
-/// For rpc communication, there can be any number of errors, so this is a
-/// generic io error.
+/// If the server rejects the request and also closes the connection, the close
+/// usually arrives first, and the sender gets [`Io`] instead of [`Remote`].
+/// [`Self::remote_code`] returns the code in both cases.
+///
+/// After an error, the channel is closed. Later sends get the same error if
+/// the receiver ended the channel, and [`Io`] with
+/// [`io::ErrorKind::BrokenPipe`] otherwise.
+///
+/// [`ReceiverClosed`]: Self::ReceiverClosed
+/// [`MaxSizeExceeded`]: Self::MaxSizeExceeded
+/// [`Remote`]: Self::Remote
+/// [`Io`]: Self::Io
 #[stack_error(derive, add_meta, from_sources)]
 #[non_exhaustive]
 pub enum SendError {
@@ -23,13 +52,31 @@ pub enum SendError {
     /// with code 0, for example because it was dropped.
     #[error("Receiver closed")]
     ReceiverClosed,
-    /// The message exceeded the maximum allowed message size (see [`MAX_MESSAGE_SIZE`]).
+    /// The receiver stopped the stream with an irpc error code, and the channel is closed.
+    ///
+    /// The `code` says why, see [`ErrorCode`], and the [`rpc`] module for an overview.
+    ///
+    /// [`ErrorCode`]: crate::rpc::ErrorCode
+    /// [`rpc`]: crate::rpc
+    #[cfg(feature = "rpc")]
+    #[error("Receiver closed the channel: {code}")]
+    Remote { code: crate::rpc::ErrorCode },
+    /// This message is larger than [`MAX_MESSAGE_SIZE`], so this sender closed the channel.
     ///
     /// [`MAX_MESSAGE_SIZE`]: crate::rpc::MAX_MESSAGE_SIZE
-    #[error("Maximum message size exceeded")]
-    MaxMessageSizeExceeded,
-    /// The underlying io error. This can occur for remote communication,
-    /// due to a network error or serialization error.
+    #[error("Message too large")]
+    MaxSizeExceeded,
+    /// This message does not encode, so this sender closed the channel.
+    #[cfg(feature = "rpc")]
+    #[error("Message does not encode")]
+    EncodeFailed {
+        #[error(std_err)]
+        source: postcard::Error,
+    },
+    /// An io error, for example because the connection was lost.
+    ///
+    /// After a send was cancelled, later sends get an io error with
+    /// [`io::ErrorKind::BrokenPipe`].
     #[error("Io error")]
     Io {
         #[error(std_err)]
@@ -41,10 +88,52 @@ impl From<SendError> for io::Error {
     fn from(e: SendError) -> Self {
         match e {
             SendError::ReceiverClosed { .. } => io::Error::new(io::ErrorKind::BrokenPipe, e),
-            SendError::MaxMessageSizeExceeded { .. } => {
-                io::Error::new(io::ErrorKind::InvalidData, e)
-            }
             SendError::Io { source, .. } => source,
+            #[cfg(feature = "rpc")]
+            SendError::Remote { .. } => io::Error::new(io::ErrorKind::ConnectionReset, e),
+            _ => io::Error::new(io::ErrorKind::InvalidData, e),
+        }
+    }
+}
+
+#[cfg(feature = "rpc")]
+impl SendError {
+    /// Returns the irpc error code that the peer sent, if any.
+    ///
+    /// This is the code of a [`Remote`](Self::Remote) error, or the code that
+    /// the peer closed the connection with.
+    pub fn remote_code(&self) -> Option<ErrorCode> {
+        match self {
+            Self::Remote { code, .. } => Some(*code),
+            Self::Io { source, .. } => rpc::io_close_code(source),
+            _ => None,
+        }
+    }
+
+    /// Returns the error for a receiver that stopped the stream with `code`.
+    pub(crate) fn from_error_code(code: ErrorCode) -> Self {
+        match code {
+            // A receiver that is dropped stops the stream with code 0.
+            ErrorCode::Closed => e!(SendError::ReceiverClosed),
+            code => e!(SendError::Remote { code }),
+        }
+    }
+
+    /// Returns the error for a send after an earlier error closed the sender.
+    ///
+    /// `code` is the code that the receiver stopped the stream with, if it did.
+    pub(crate) fn closed(code: Option<ErrorCode>) -> Self {
+        code.map(Self::from_error_code)
+            .unwrap_or_else(|| io::Error::from(io::ErrorKind::BrokenPipe).into())
+    }
+}
+
+#[cfg(feature = "rpc")]
+impl From<noq::WriteError> for SendError {
+    fn from(err: noq::WriteError) -> Self {
+        match rpc::stop_code(&err) {
+            Some(code) => SendError::from_error_code(code),
+            None => e!(SendError::Io, err.into()),
         }
     }
 }

src/channel/mpsc.rs

@@ -7,36 +7,92 @@ use std::{fmt::Debug, future::Future, io, marker::PhantomData, pin::Pin, sync::A
 use n0_error::{e, stack_error};
 
 use super::SendError;
+#[cfg(feature = "rpc")]
+use crate::rpc::{self, ErrorCode};
 
-/// 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.
+/// Error when receiving an mpsc message.
 ///
-/// For rpc communication, there can be any number of errors, so this is a
-/// generic io error.
-#[stack_error(derive, add_meta, from_sources)]
+/// A local channel has no errors.
+///
+/// For a remote channel, the sender is on the other side of the connection:
+/// the server for a response stream, and the client for the updates of a request.
+///
+/// If the server rejects the request and also closes the connection, the close
+/// usually arrives first, and the receiver gets [`Io`] instead of [`Remote`].
+/// [`Self::remote_code`] returns the code in both cases.
+///
+/// [`Io`]: Self::Io
+/// [`Remote`]: Self::Remote
+#[stack_error(derive, add_meta)]
 #[non_exhaustive]
 pub enum RecvError {
-    /// The message exceeded the maximum allowed message size (see [`MAX_MESSAGE_SIZE`]).
+    /// The sender reset the stream with an irpc error code, and the channel is closed.
+    ///
+    /// The `code` says why, see [`ErrorCode`], and the [`rpc`] module for an overview.
+    ///
+    /// [`ErrorCode`]: crate::rpc::ErrorCode
+    /// [`rpc`]: crate::rpc
+    #[cfg(feature = "rpc")]
+    #[error("Sender closed the channel: {code}")]
+    Remote { code: crate::rpc::ErrorCode },
+    /// The message is larger than [`MAX_MESSAGE_SIZE`], so this receiver closed the channel.
     ///
     /// [`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("Message too large")]
+    MaxSizeExceeded,
+    /// The message does not decode, so this receiver closed the channel.
+    #[cfg(feature = "rpc")]
+    #[error("Message does not decode")]
+    DecodeFailed {
+        #[error(std_err)]
+        source: postcard::Error,
+    },
+    /// An io error, for example because the connection was lost.
+    ///
+    /// A stream that ends in the middle of a message also gives an io error.
     #[error("Io error")]
     Io {
         #[error(std_err)]
         source: io::Error,
     },
 }
 
+/// Converts an io error from reading the stream of a remote channel.
+///
+/// A reset with an irpc error code becomes [`RecvError::Remote`], and any
+/// other io error becomes [`RecvError::Io`].
+impl From<io::Error> for RecvError {
+    fn from(err: io::Error) -> Self {
+        #[cfg(feature = "rpc")]
+        if let Some(code) = rpc::reset_code(&err) {
+            return e!(RecvError::Remote { code });
+        }
+        e!(RecvError::Io, err)
+    }
+}
+
 impl From<RecvError> for io::Error {
     fn from(e: RecvError) -> Self {
         match e {
             RecvError::Io { source, .. } => source,
-            RecvError::MaxMessageSizeExceeded { .. } => {
-                io::Error::new(io::ErrorKind::InvalidData, e)
-            }
+            #[cfg(feature = "rpc")]
+            RecvError::Remote { .. } => io::Error::new(io::ErrorKind::ConnectionReset, e),
+            _ => io::Error::new(io::ErrorKind::InvalidData, e),
+        }
+    }
+}
+
+#[cfg(feature = "rpc")]
+impl RecvError {
+    /// Returns the irpc error code that the peer sent, if any.
+    ///
+    /// This is the code of a [`Remote`](Self::Remote) error, or the code that
+    /// the peer closed the connection with.
+    pub fn remote_code(&self) -> Option<ErrorCode> {
+        match self {
+            Self::Remote { code, .. } => Some(*code),
+            Self::Io { source, .. } => rpc::io_close_code(source),
+            _ => None,
         }
     }
 }

src/channel/oneshot.rs

@@ -6,14 +6,24 @@ use n0_error::{e, stack_error};
 use n0_future::future::Boxed as BoxFuture;
 
 use super::SendError;
+#[cfg(feature = "rpc")]
+use crate::rpc::{self, ErrorCode};
 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.
+/// Error when receiving a oneshot message.
 ///
-/// For rpc communication, there can be any number of errors, so this is a
-/// generic io error.
-#[stack_error(derive, add_meta, from_sources)]
+/// For a local channel, the only error is [`SenderClosed`]. For a remote
+/// channel, the sender is on the other side of the connection, for example
+/// the server for a response.
+///
+/// If the server rejects the request and also closes the connection, the close
+/// usually arrives first, and the receiver gets [`Io`] instead of [`Remote`].
+/// [`Self::remote_code`] returns the code in both cases.
+///
+/// [`SenderClosed`]: Self::SenderClosed
+/// [`Io`]: Self::Io
+/// [`Remote`]: Self::Remote
+#[stack_error(derive, add_meta)]
 #[non_exhaustive]
 pub enum RecvError {
     /// The sender has been closed.
@@ -23,28 +33,74 @@ pub enum RecvError {
     /// without sending, for example because it was dropped.
     #[error("Sender closed")]
     SenderClosed,
-    /// The message exceeded the maximum allowed message size (see [`MAX_MESSAGE_SIZE`]).
+    /// The sender reset the stream with an irpc error code, and the channel is closed.
+    ///
+    /// The `code` says why, see [`ErrorCode`], and the [`rpc`] module for an overview.
+    ///
+    /// [`ErrorCode`]: crate::rpc::ErrorCode
+    /// [`rpc`]: crate::rpc
+    #[cfg(feature = "rpc")]
+    #[error("Sender closed the channel: {code}")]
+    Remote { code: crate::rpc::ErrorCode },
+    /// The message is larger than [`MAX_MESSAGE_SIZE`], so this receiver closed the channel.
     ///
     /// [`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("Message too large")]
+    MaxSizeExceeded,
+    /// The message does not decode, so this receiver closed the channel.
+    #[cfg(feature = "rpc")]
+    #[error("Message does not decode")]
+    DecodeFailed {
+        #[error(std_err)]
+        source: postcard::Error,
+    },
+    /// An io error, for example because the connection was lost.
+    ///
+    /// A stream that ends in the middle of a message also gives an io error.
     #[error("Io error")]
     Io {
         #[error(std_err)]
         source: io::Error,
     },
 }
 
+/// Converts an io error from reading the stream of a remote channel.
+///
+/// A reset with an irpc error code becomes [`RecvError::Remote`], and any
+/// other io error becomes [`RecvError::Io`].
+impl From<io::Error> for RecvError {
+    fn from(err: io::Error) -> Self {
+        #[cfg(feature = "rpc")]
+        if let Some(code) = rpc::reset_code(&err) {
+            return e!(RecvError::Remote { code });
+        }
+        e!(RecvError::Io, err)
+    }
+}
+
 impl From<RecvError> for io::Error {
     fn from(e: RecvError) -> Self {
         match e {
             RecvError::Io { source, .. } => source,
             RecvError::SenderClosed { .. } => io::Error::new(io::ErrorKind::BrokenPipe, e),
-            RecvError::MaxMessageSizeExceeded { .. } => {
-                io::Error::new(io::ErrorKind::InvalidData, e)
-            }
+            #[cfg(feature = "rpc")]
+            RecvError::Remote { .. } => io::Error::new(io::ErrorKind::ConnectionReset, e),
+            _ => io::Error::new(io::ErrorKind::InvalidData, e),
+        }
+    }
+}
+
+#[cfg(feature = "rpc")]
+impl RecvError {
+    /// Returns the irpc error code that the peer sent, if any.
+    ///
+    /// This is the code of a [`Remote`](Self::Remote) error, or the code that
+    /// the peer closed the connection with.
+    pub fn remote_code(&self) -> Option<ErrorCode> {
+        match self {
+            Self::Remote { code, .. } => Some(*code),
+            Self::Io { source, .. } => rpc::io_close_code(source),
+            _ => None,
         }
     }
 }