From 448fb04daece29e485bd1f32151b276e1f8c3a50 Mon Sep 17 00:00:00 2001
From: Frando <frando@unbiskant.org>
Date: Fri, 9 Oct 2026 15:01:35 +0200
Subject: [PATCH 1/4] refactor!: define the error codes on a public ErrorCode
 enum

The codes were public constants, and code compared stream codes against
them one by one. `ErrorCode` defines each code once, as its
discriminant, with `to_code` and `from_code`, and converts into the
`VarInt` that noq takes. The code and its docs use it everywhere
instead of the constants, also for code 0. The behavior is the same.
---
 src/rpc.rs               | 165 ++++++++++++++++++++++++++-------------
 tests/invalid_request.rs |  18 ++---
 tests/mpsc_channel.rs    |   4 +-
 3 files changed, 117 insertions(+), 70 deletions(-)

diff --git a/src/rpc.rs b/src/rpc.rs
index ee5129c4..8826c702 100644
--- a/src/rpc.rs
+++ b/src/rpc.rs
@@ -9,9 +9,9 @@
 //! Each request has its own stream. If a request is bad, the server stops and
 //! resets the streams of that request:
 //!
-//! * with [`ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED`] if the request is larger
+//! * with [`ErrorCode::MaxMessageSizeExceeded`] if the request is larger
 //!   than [`MAX_MESSAGE_SIZE`],
-//! * with [`ERROR_CODE_DECODE_FAILED`] if the request does not decode. For
+//! * with [`ErrorCode::DecodeFailed`] if the request does not decode. For
 //!   example, a client of a newer protocol version sent a request type that
 //!   the server does not know. A stream that ends before its request is
 //!   complete is also a request that does not decode.
@@ -27,22 +27,20 @@
 //! next request.
 //!
 //! A client that drops a request before it wrote all of it resets the stream
-//! with [`ERROR_CODE_ABORTED`]. For example, it dropped the future of
+//! with [`ErrorCode::Aborted`]. For example, it dropped the future of
 //! [`Client::rpc`] while it wrote a large request. The
 //! server skips a request whose stream was reset. It resets its side of the
 //! stream with code 0, and reads the next request.
 //!
 //! A remote channel receiver also stops its stream with
-//! [`ERROR_CODE_DECODE_FAILED`] if a message does not decode. Its sender then
+//! [`ErrorCode::DecodeFailed`] if a message does not decode. Its sender then
 //! gets an error on the next send.
 //!
 //! # Error codes
 //!
-//! irpc reserves the codes 0 to 15 for streams and connections. Code 0 means
-//! no error, for example a normal close. irpc also uses
-//! [`ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED`], [`ERROR_CODE_ENCODE_FAILED`],
-//! [`ERROR_CODE_DECODE_FAILED`], and [`ERROR_CODE_ABORTED`], and can add more
-//! codes in this range in a later version.
+//! irpc closes connections, and stops and resets streams, with the codes of
+//! [`ErrorCode`]. irpc reserves the codes 0 to 15, and can add more codes in
+//! this range in a later version.
 //!
 //! An application should use codes from 16 upward, so that a peer can tell its
 //! codes apart from the codes of irpc. irpc does not check this. A code below
@@ -78,29 +76,73 @@ use crate::{
 /// Default max message size (16 MiB).
 pub const MAX_MESSAGE_SIZE: u64 = 1024 * 1024 * 16;
 
-/// Error code on streams and connections if a message is larger than [`MAX_MESSAGE_SIZE`].
+/// An error code that irpc uses on streams and connections.
 ///
-/// See [Error codes] for the codes that irpc reserves.
+/// An application should use codes from 16 upward, see [Error codes].
 ///
 /// [Error codes]: self#error-codes
-pub const ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED: u32 = 1;
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+#[repr(u32)]
+#[non_exhaustive]
+pub enum ErrorCode {
+    /// No error, for example a normal close or a dropped receiver.
+    Closed = 0,
+    /// A message is larger than [`MAX_MESSAGE_SIZE`].
+    ///
+    /// A server also uses this code for a request that is too large, see
+    /// [Bad requests].
+    ///
+    /// [Bad requests]: self#bad-requests
+    MaxMessageSizeExceeded = 1,
+    /// A message does not encode.
+    EncodeFailed = 2,
+    /// A message does not decode.
+    ///
+    /// A server also uses this code for a request that does not decode, see
+    /// [Bad requests].
+    ///
+    /// [Bad requests]: self#bad-requests
+    DecodeFailed = 3,
+    /// The client dropped the request before it wrote all of it.
+    ///
+    /// The server skips such a request, see [Bad requests].
+    ///
+    /// [Bad requests]: self#bad-requests
+    Aborted = 4,
+}
 
-/// Error code on streams if the sender could not encode a message.
-pub const ERROR_CODE_ENCODE_FAILED: u32 = 2;
+impl ErrorCode {
+    /// Every error code, for [`Self::from_code`].
+    const ALL: [Self; 5] = [
+        Self::Closed,
+        Self::MaxMessageSizeExceeded,
+        Self::EncodeFailed,
+        Self::DecodeFailed,
+        Self::Aborted,
+    ];
+
+    /// Returns the number of this error code.
+    pub const fn to_code(self) -> u32 {
+        self as u32
+    }
 
-/// Error code on streams and connections if the receiver could not decode a message.
-///
-/// A server stops and resets the streams of a request that does not decode,
-/// and by default closes the connection with this code. A remote channel
-/// receiver stops its stream if a message does not decode.
-pub const ERROR_CODE_DECODE_FAILED: u32 = 3;
+    /// Returns the error code for `code`, or `None` if irpc does not use `code`.
+    pub fn from_code(code: u32) -> Option<Self> {
+        Self::ALL.into_iter().find(|c| c.to_code() == code)
+    }
 
-/// Error code on a request stream if the client drops the request before it is written.
-///
-/// The server skips such a request, see [Bad requests].
-///
-/// [Bad requests]: self#bad-requests
-pub const ERROR_CODE_ABORTED: u32 = 4;
+    fn from_varint(code: VarInt) -> Option<Self> {
+        u32::try_from(code.into_inner())
+            .ok()
+            .and_then(Self::from_code)
+    }
+}
+
+impl From<ErrorCode> for VarInt {
+    fn from(code: ErrorCode) -> Self {
+        VarInt::from_u32(code.to_code())
+    }
+}
 
 /// Error when reading a request with [`read_request`].
 ///
@@ -144,11 +186,11 @@ pub enum ReadRequestError {
 
 impl ReadRequestError {
     /// Returns the irpc error code for a bad request, or `None` for a connection error.
-    fn error_code(&self) -> Option<u32> {
+    fn error_code(&self) -> Option<ErrorCode> {
         match self {
-            Self::MaxMessageSizeExceeded { .. } => Some(ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED),
+            Self::MaxMessageSizeExceeded { .. } => Some(ErrorCode::MaxMessageSizeExceeded),
             Self::InvalidRequest { .. } | Self::DecodeFailed { .. } => {
-                Some(ERROR_CODE_DECODE_FAILED)
+                Some(ErrorCode::DecodeFailed)
             }
             Self::Connection { .. } => None,
         }
@@ -213,14 +255,14 @@ impl From<WriteError> for io::Error {
 
 impl From<noq::WriteError> for SendError {
     fn from(err: noq::WriteError) -> Self {
-        match err {
-            noq::WriteError::Stopped(code)
-                if code == ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED.into() =>
-            {
-                e!(SendError::MaxMessageSizeExceeded)
-            }
+        let code = match &err {
+            noq::WriteError::Stopped(code) => ErrorCode::from_varint(*code),
+            _ => None,
+        };
+        match code {
+            Some(ErrorCode::MaxMessageSizeExceeded) => e!(SendError::MaxMessageSizeExceeded),
             // A receiver that is dropped stops the stream with code 0.
-            noq::WriteError::Stopped(code) if code == 0u32.into() => e!(SendError::ReceiverClosed),
+            Some(ErrorCode::Closed) => e!(SendError::ReceiverClosed),
             _ => e!(SendError::Io, io::Error::from(err)),
         }
     }
@@ -251,11 +293,11 @@ pub trait RemoteConnection: Send + Sync + Debug + 'static {
 /// A connection to a remote service that can be used to send the initial message.
 ///
 /// If it drops before the message is written, it resets the stream with
-/// [`ERROR_CODE_ABORTED`].
+/// [`ErrorCode::Aborted`].
 #[derive(Debug)]
 pub struct RemoteSender<S>(ResetOnDrop, noq::RecvStream, std::marker::PhantomData<S>);
 
-/// A send stream that resets with [`ERROR_CODE_ABORTED`] if it drops before [`Self::into_inner`].
+/// A send stream that resets with [`ErrorCode::Aborted`] if it drops before [`Self::into_inner`].
 #[derive(Debug)]
 struct ResetOnDrop(Option<noq::SendStream>);
 
@@ -272,7 +314,7 @@ impl ResetOnDrop {
 impl Drop for ResetOnDrop {
     fn drop(&mut self) {
         if let Some(send) = &mut self.0 {
-            send.reset(ERROR_CODE_ABORTED.into()).ok();
+            send.reset(ErrorCode::Aborted.into()).ok();
         }
     }
 }
@@ -338,7 +380,7 @@ impl<T: DeserializeOwned> From<noq::RecvStream> for oneshot::Receiver<T> {
                 return Err(e!(oneshot::RecvError::SenderClosed));
             };
             if size > MAX_MESSAGE_SIZE {
-                read.stop(ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED.into()).ok();
+                read.stop(ErrorCode::MaxMessageSizeExceeded.into()).ok();
                 return Err(e!(oneshot::RecvError::MaxMessageSizeExceeded));
             }
             let rest = read
@@ -346,7 +388,7 @@ impl<T: DeserializeOwned> From<noq::RecvStream> for oneshot::Receiver<T> {
                 .await
                 .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
             let msg: T = postcard::from_bytes(&rest).map_err(|e| {
-                read.stop(ERROR_CODE_DECODE_FAILED.into()).ok();
+                read.stop(ErrorCode::DecodeFailed.into()).ok();
                 io::Error::new(io::ErrorKind::InvalidData, e)
             })?;
             Ok(msg)
@@ -385,7 +427,7 @@ impl<T: RpcMessage> From<noq::SendStream> for oneshot::Sender<T> {
                 let size = match postcard::experimental::serialized_size(&value) {
                     Ok(size) => size,
                     Err(e) => {
-                        writer.reset(ERROR_CODE_ENCODE_FAILED.into()).ok();
+                        writer.reset(ErrorCode::EncodeFailed.into()).ok();
                         return Err(e!(
                             SendError::Io,
                             io::Error::new(io::ErrorKind::InvalidData, e,)
@@ -393,15 +435,13 @@ impl<T: RpcMessage> From<noq::SendStream> for oneshot::Sender<T> {
                     }
                 };
                 if size as u64 > MAX_MESSAGE_SIZE {
-                    writer
-                        .reset(ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED.into())
-                        .ok();
+                    writer.reset(ErrorCode::MaxMessageSizeExceeded.into()).ok();
                     return Err(e!(SendError::MaxMessageSizeExceeded));
                 }
                 // write via a small buffer to avoid allocation for small values
                 let mut buf = SmallVec::<[u8; 128]>::new();
                 if let Err(e) = buf.write_length_prefixed(value) {
-                    writer.reset(ERROR_CODE_ENCODE_FAILED.into()).ok();
+                    writer.reset(ErrorCode::EncodeFailed.into()).ok();
                     return Err(e.into());
                 }
                 writer.write_all(&buf).await?;
@@ -445,7 +485,7 @@ impl<T: RpcMessage> DynReceiver<T> for NoqReceiver<T> {
             };
             if size > MAX_MESSAGE_SIZE {
                 self.recv
-                    .stop(ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED.into())
+                    .stop(ErrorCode::MaxMessageSizeExceeded.into())
                     .ok();
                 return Err(e!(mpsc::RecvError::MaxMessageSizeExceeded));
             }
@@ -454,7 +494,7 @@ impl<T: RpcMessage> DynReceiver<T> for NoqReceiver<T> {
                 .await
                 .map_err(|e| io::Error::new(io::ErrorKind::UnexpectedEof, e))?;
             let msg: T = postcard::from_bytes(&buf).map_err(|e| {
-                read.stop(ERROR_CODE_DECODE_FAILED.into()).ok();
+                read.stop(ErrorCode::DecodeFailed.into()).ok();
                 io::Error::new(io::ErrorKind::InvalidData, e)
             })?;
             Ok(Some(msg))
@@ -481,7 +521,7 @@ impl<T: RpcMessage> NoqSenderInner<T> {
         let size = match postcard::experimental::serialized_size(&value) {
             Ok(size) => size,
             Err(e) => {
-                self.send.reset(ERROR_CODE_ENCODE_FAILED.into()).ok();
+                self.send.reset(ErrorCode::EncodeFailed.into()).ok();
                 return Err(e!(
                     SendError::Io,
                     io::Error::new(io::ErrorKind::InvalidData, e)
@@ -490,13 +530,13 @@ impl<T: RpcMessage> NoqSenderInner<T> {
         };
         if size as u64 > MAX_MESSAGE_SIZE {
             self.send
-                .reset(ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED.into())
+                .reset(ErrorCode::MaxMessageSizeExceeded.into())
                 .ok();
             return Err(e!(SendError::MaxMessageSizeExceeded));
         }
         self.buffer.clear();
         if let Err(e) = self.buffer.write_length_prefixed(value) {
-            self.send.reset(ERROR_CODE_ENCODE_FAILED.into()).ok();
+            self.send.reset(ErrorCode::EncodeFailed.into()).ok();
             return Err(e.into());
         }
         Ok(())
@@ -688,7 +728,7 @@ impl HandlerError {
     /// closes a connection with code 0 for a normal close, which
     /// [`Handler::from_sender`] also uses when its receiver is gone. For a bad
     /// request, irpc closes the connection with
-    /// [`ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED`] or [`ERROR_CODE_DECODE_FAILED`].
+    /// [`ErrorCode::MaxMessageSizeExceeded`] or [`ErrorCode::DecodeFailed`].
     ///
     /// # Examples
     ///
@@ -889,7 +929,7 @@ impl<S: RemoteService> Handler<S> {
                 local_sender.send_raw(msg).await.map_err(|err| {
                     // The receiver is gone. Close as a dropped connection does.
                     warn!("handler stopped: {err:#}");
-                    HandlerError::close_connection(0, b"")
+                    HandlerError::close_connection(ErrorCode::Closed.to_code(), b"")
                 })
             }
         })
@@ -1008,7 +1048,7 @@ async fn read_request_inner<S: Service>(
         let (mut send, mut recv) = match connection.accept_bi().await {
             Ok(streams) => streams,
             Err(ConnectionError::ApplicationClosed(cause))
-                if cause.error_code.into_inner() == 0 =>
+                if ErrorCode::from_varint(cause.error_code) == Some(ErrorCode::Closed) =>
             {
                 trace!("remote side closed connection {cause:?}");
                 return Ok(None);
@@ -1024,7 +1064,7 @@ async fn read_request_inner<S: Service>(
             ReadFrame::Reset => {
                 debug!("skipped request: the client reset its stream");
                 // A drop finishes the stream, which looks like an empty response.
-                send.reset(0u32.into()).ok();
+                send.reset(ErrorCode::Closed.into()).ok();
                 continue;
             }
             // The next `accept_bi` returns why.
@@ -1058,7 +1098,7 @@ impl<S> ReadFrame<S> {
     /// Returns the frame for a stream that failed before the request was complete.
     fn from_read_err(err: &noq::ReadError) -> Self {
         match err {
-            // Any reset, not only `ERROR_CODE_ABORTED`: the update sender of a
+            // Any reset, not only `ErrorCode::Aborted`: the update sender of a
             // request resets the stream if an update does not encode. noq drops
             // unread data on a reset, so this reset can arrive before the request.
             noq::ReadError::Reset(_) => Self::Reset,
@@ -1112,3 +1152,16 @@ async fn read_request_frame<S: Service>(recv: &mut noq::RecvStream) -> ReadFrame
         Err(err) => ReadFrame::DecodeFailed(err),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn error_code_round_trips() {
+        for code in ErrorCode::ALL {
+            assert_eq!(ErrorCode::from_code(code.to_code()), Some(code));
+        }
+        assert_eq!(ErrorCode::from_code(16), None);
+    }
+}
diff --git a/tests/invalid_request.rs b/tests/invalid_request.rs
index c24a622a..ebf0b7cb 100644
--- a/tests/invalid_request.rs
+++ b/tests/invalid_request.rs
@@ -6,10 +6,7 @@ use std::{pin::pin, time::Duration};
 use irpc::{
     Client, WithChannels,
     noq::listen,
-    rpc::{
-        ERROR_CODE_DECODE_FAILED, ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED, Handler, MAX_MESSAGE_SIZE,
-        ReadRequestError, read_request,
-    },
+    rpc::{ErrorCode, Handler, MAX_MESSAGE_SIZE, ReadRequestError, read_request},
 };
 use n0_future::{future::poll_once, task::AbortOnDropHandle};
 use noq::ConnectionError;
@@ -90,7 +87,7 @@ async fn bad_request_closes_connection() -> TestResult<()> {
     let ConnectionError::ApplicationClosed(close) = conn.closed().await else {
         panic!("server closes the connection");
     };
-    assert_eq!(close.error_code, ERROR_CODE_DECODE_FAILED.into());
+    assert_eq!(close.error_code, ErrorCode::DecodeFailed.into());
     Ok(())
 }
 
@@ -107,10 +104,7 @@ async fn too_large_request_closes_connection() -> TestResult<()> {
     let ConnectionError::ApplicationClosed(close) = conn.closed().await else {
         panic!("server closes the connection");
     };
-    assert_eq!(
-        close.error_code,
-        ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED.into()
-    );
+    assert_eq!(close.error_code, ErrorCode::MaxMessageSizeExceeded.into());
     Ok(())
 }
 
@@ -139,7 +133,7 @@ async fn dropped_rpc_keeps_connection() -> TestResult<()> {
     let client = Client::<client::EchoProtocol>::boxed(conn);
 
     // The request is larger than the flow control window, so one poll cannot
-    // write all of it. The drop then resets the stream with `ERROR_CODE_ABORTED`.
+    // write all of it. The drop then resets the stream with `ErrorCode::Aborted`.
     let mut rpc = Box::pin(client.rpc(client::Echo("a".repeat(8 * 1024 * 1024))));
     assert!(poll_once(&mut rpc).await.is_none(), "the write is not done");
     drop(rpc);
@@ -160,7 +154,7 @@ async fn truncated_request_closes_connection() -> TestResult<()> {
     let ConnectionError::ApplicationClosed(close) = conn.closed().await else {
         panic!("server closes the connection");
     };
-    assert_eq!(close.error_code, ERROR_CODE_DECODE_FAILED.into());
+    assert_eq!(close.error_code, ErrorCode::DecodeFailed.into());
     Ok(())
 }
 
@@ -189,7 +183,7 @@ async fn reset_before_read_skips_request() -> TestResult<()> {
     let (tx, _rx) = client
         .client_streaming(client::Upload("a".into()), 1)
         .await?;
-    // The update does not encode, so the sender resets with `ERROR_CODE_ENCODE_FAILED`.
+    // The update does not encode, so the sender resets with `ErrorCode::EncodeFailed`.
     tx.send(NoSer(1))
         .await
         .expect_err("odd numbers do not encode");
diff --git a/tests/mpsc_channel.rs b/tests/mpsc_channel.rs
index d0d4f567..3af0535f 100644
--- a/tests/mpsc_channel.rs
+++ b/tests/mpsc_channel.rs
@@ -12,7 +12,7 @@ use irpc::{
         mpsc::{self, Receiver, RecvError},
         oneshot,
     },
-    rpc::{ERROR_CODE_DECODE_FAILED, Handler},
+    rpc::{ErrorCode, Handler},
     rpc_requests,
     util::AsyncWriteVarintExt,
 };
@@ -421,7 +421,7 @@ async fn mpsc_decode_error_stops_stream() -> TestResult<()> {
     // an odd number does not decode as `NoSer`
     send.write_length_prefixed(1u64).await?;
     let code = timeout(Duration::from_secs(5), send.stopped()).await??;
-    assert_eq!(code, Some(ERROR_CODE_DECODE_FAILED.into()));
+    assert_eq!(code, Some(ErrorCode::DecodeFailed.into()));
     conn.close(0u32.into(), b"");
     server.await??;
     Ok(())

From 8c07261df722b629a083b96491b0e3546e0208ee Mon Sep 17 00:00:00 2001
From: Frando <frando@unbiskant.org>
Date: Fri, 9 Oct 2026 15:01:43 +0200
Subject: [PATCH 2/4] feat!: split the error codes into request and message
 codes, and carry them in Remote variants

irpc used the same codes for a bad request and for a bad channel message,
so a client could not tell a rejected request from a rejected response,
and every reset or stop from the peer reached the user as an `Io` error.

The error codes are now split into request codes (`RequestAborted`,
`RequestMaxSizeExceeded`, `RequestEncodeFailed`, `RequestDecodeFailed`)
and message codes (`MessageMaxSizeExceeded`, `MessageEncodeFailed`,
`MessageDecodeFailed`), and `ErrorCode` implements `Display`.
`SendError`, `oneshot::RecvError`, `mpsc::RecvError`, and `WriteError`
get a `Remote { code }` variant for a code that the peer reset or stopped
the stream with, and keep their own variants only for errors that this
side found: `MaxSizeExceeded`, `EncodeFailed`, and `DecodeFailed`.
`ReadRequestError` has `MaxSizeExceeded`, `Malformed`, and `DecodeFailed`.
`remote_code()` returns the code of a `Remote` error.

The server rejects a bad request with the request code on both streams and
when it closes the connection. A client that cannot encode its request
resets the stream with the request code. When the server skips a request
whose stream the client reset, it resets its side with `RequestAborted`,
so a client with an mpsc response gets an error instead of a clean end.
After a remote send fails, later sends get the same error if the receiver
ended the channel, and `Io(BrokenPipe)` if this sender closed it.
---
 Cargo.toml               |   4 +-
 src/channel.rs           |  61 ++++-
 src/channel/mpsc.rs      |  51 +++-
 src/channel/oneshot.rs   |  50 +++-
 src/lib.rs               |  14 +-
 src/rpc.rs               | 524 ++++++++++++++++++++++++++-------------
 tests/invalid_request.rs | 145 ++++++++++-
 tests/mpsc_channel.rs    |  98 +++++---
 tests/oneshot_channel.rs |  26 +-
 9 files changed, 697 insertions(+), 276 deletions(-)

diff --git a/Cargo.toml b/Cargo.toml
index 306c16dd..e9bbe1d7 100644
--- a/Cargo.toml
+++ b/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
diff --git a/src/channel.rs b/src/channel.rs
index de8f4899..d66921d8 100644
--- a/src/channel.rs
+++ b/src/channel.rs
@@ -8,11 +8,31 @@ 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 rpc communication, there can be any number of errors, so this is a
-/// generic io error.
+/// 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`]).
+///
+/// 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 +43,30 @@ 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`].
+    ///
+    /// [`ErrorCode`]: crate::rpc::ErrorCode
+    #[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 +78,10 @@ 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),
         }
     }
 }
diff --git a/src/channel/mpsc.rs b/src/channel/mpsc.rs
index 80f82a21..6bf87e93 100644
--- a/src/channel/mpsc.rs
+++ b/src/channel/mpsc.rs
@@ -8,21 +8,48 @@ use n0_error::{e, stack_error};
 
 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.
+/// Error when receiving an mpsc message.
 ///
-/// For rpc communication, there can be any number of errors, so this is a
-/// generic io error.
+/// A local channel has no errors: `recv` returns `Ok(None)` when all senders
+/// are gone. A remote channel does the same when its sender closes the stream.
+///
+/// For a remote channel, the sender is on the other side of the connection:
+/// the server for a response, and the client for the updates of a request.
+///
+/// If the server rejects the request and also closes the connection, the close
+/// can arrive first, and the receiver gets [`Io`] instead of [`Remote`].
+///
+/// [`Io`]: Self::Io
+/// [`Remote`]: Self::Remote
 #[stack_error(derive, add_meta, from_sources)]
 #[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`]. For example, the server rejected
+    /// or skipped the request, see [Bad requests] and [Abandoned requests].
+    ///
+    /// [`ErrorCode`]: crate::rpc::ErrorCode
+    /// [Bad requests]: crate::rpc#bad-requests
+    /// [Abandoned requests]: crate::rpc#abandoned-requests
+    #[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)]
@@ -34,9 +61,9 @@ 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),
         }
     }
 }
diff --git a/src/channel/oneshot.rs b/src/channel/oneshot.rs
index 7543da1f..1dc8f7ab 100644
--- a/src/channel/oneshot.rs
+++ b/src/channel/oneshot.rs
@@ -8,11 +8,18 @@ use n0_future::future::Boxed as BoxFuture;
 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.
+/// Error when receiving a oneshot message.
 ///
-/// For rpc communication, there can be any number of errors, so this is a
-/// generic io error.
+/// 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
+/// can arrive first, and the receiver gets [`Io`] instead of [`Remote`].
+///
+/// [`SenderClosed`]: Self::SenderClosed
+/// [`Io`]: Self::Io
+/// [`Remote`]: Self::Remote
 #[stack_error(derive, add_meta, from_sources)]
 #[non_exhaustive]
 pub enum RecvError {
@@ -23,13 +30,32 @@ 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`]. For example, the server rejected
+    /// or skipped the request, see [Bad requests] and [Abandoned requests].
+    ///
+    /// [`ErrorCode`]: crate::rpc::ErrorCode
+    /// [Bad requests]: crate::rpc#bad-requests
+    /// [Abandoned requests]: crate::rpc#abandoned-requests
+    #[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)]
@@ -42,9 +68,9 @@ impl From<RecvError> for io::Error {
         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),
         }
     }
 }
diff --git a/src/lib.rs b/src/lib.rs
index 165ce0df..bcdae49d 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -769,9 +769,9 @@ impl<S: Service> Client<S> {
                 #[cfg(not(feature = "rpc"))]
                 Request::Remote(_request) => unreachable!(),
                 #[cfg(feature = "rpc")]
-                Request::Remote(request) => {
+                Request::Remote(mut request) => {
                     // see https://www.iroh.computer/blog/0rtt-api#connect-side
-                    let buf = rpc::prepare_write::<S>(msg)?;
+                    let buf = request.prepare(msg)?;
                     let (_tx, _rx) = request.write_raw(&buf).await?;
                     if this.0.zero_rtt_rejected().await {
                         // 0rtt was not accepted, the data is lost, send it again!
@@ -808,9 +808,9 @@ impl<S: Service> Client<S> {
                 #[cfg(not(feature = "rpc"))]
                 Request::Remote(_request) => unreachable!(),
                 #[cfg(feature = "rpc")]
-                Request::Remote(request) => {
+                Request::Remote(mut request) => {
                     // see https://www.iroh.computer/blog/0rtt-api#connect-side
-                    let buf = rpc::prepare_write::<S>(msg)?;
+                    let buf = request.prepare(msg)?;
                     let (_tx, rx) = request.write_raw(&buf).await?;
                     if this.0.zero_rtt_rejected().await {
                         // 0rtt was not accepted, the data is lost, send it again!
@@ -856,9 +856,9 @@ impl<S: Service> Client<S> {
                 #[cfg(not(feature = "rpc"))]
                 Request::Remote(_request) => unreachable!(),
                 #[cfg(feature = "rpc")]
-                Request::Remote(request) => {
+                Request::Remote(mut request) => {
                     // see https://www.iroh.computer/blog/0rtt-api#connect-side
-                    let buf = rpc::prepare_write::<S>(msg)?;
+                    let buf = request.prepare(msg)?;
                     let (_tx, rx) = request.write_raw(&buf).await?;
                     if this.0.zero_rtt_rejected().await {
                         // 0rtt was not accepted, the data is lost, send it again!
@@ -955,7 +955,7 @@ pub enum Error {
     #[error("Oneshot recv error")]
     OneshotRecv { source: channel::oneshot::RecvError },
     #[cfg(feature = "rpc")]
-    #[error("Recv error")]
+    #[error("Write error")]
     Write { source: rpc::WriteError },
 }
 
diff --git a/src/rpc.rs b/src/rpc.rs
index 8826c702..1d52381a 100644
--- a/src/rpc.rs
+++ b/src/rpc.rs
@@ -4,17 +4,23 @@
 //! streams. The [`crate::noq`] module has the transport with dial by socket
 //! address. The [`crate::iroh`] module has the transport with dial by endpoint id.
 //!
-//! # Bad requests
+//! # Requests and channel messages
+//!
+//! Each request has its own pair of streams. The client writes the request as
+//! the first message on its stream. After the request, the streams carry the
+//! channel messages of the request: the response, and for some requests,
+//! updates from the client. The name of each error code says what the code is
+//! about: a `Request` code is about the request, and a `Message` code is about
+//! a channel message, see [`ErrorCode`].
 //!
-//! Each request has its own stream. If a request is bad, the server stops and
-//! resets the streams of that request:
+//! # Bad requests
 //!
-//! * with [`ErrorCode::MaxMessageSizeExceeded`] if the request is larger
-//!   than [`MAX_MESSAGE_SIZE`],
-//! * with [`ErrorCode::DecodeFailed`] if the request does not decode. For
-//!   example, a client of a newer protocol version sent a request type that
-//!   the server does not know. A stream that ends before its request is
-//!   complete is also a request that does not decode.
+//! A request is bad if it is larger than [`MAX_MESSAGE_SIZE`], or if it does
+//! not decode. For example, a client of a newer protocol version sent a request
+//! type that the server does not know. A stream that ends before its request is
+//! complete also counts as a request that does not decode. The server stops and
+//! resets the streams of a bad request with
+//! [`ErrorCode::RequestMaxSizeExceeded`] or [`ErrorCode::RequestDecodeFailed`].
 //!
 //! By default, a [`Handler`] then closes the connection with the same code, as
 //! for any other protocol violation. A protocol that adds request types at the
@@ -22,19 +28,37 @@
 //! reads the next request, so an old server rejects the new requests and keeps
 //! serving the old ones on the same connection.
 //!
+//! The client gets the code in a `Remote` error from the receiver of the
+//! response, for example [`oneshot::RecvError::Remote`] from [`Client::rpc`].
+//! If the server closes the connection, the close can arrive first, and the
+//! client gets the connection error.
+//!
 //! [`read_request`] returns a bad request as a [`ReadRequestError`]. A server
 //! with its own loop decides what to do: close the connection, or read the
 //! next request.
 //!
+//! # Abandoned requests
+//!
 //! A client that drops a request before it wrote all of it resets the stream
-//! with [`ErrorCode::Aborted`]. For example, it dropped the future of
-//! [`Client::rpc`] while it wrote a large request. The
-//! server skips a request whose stream was reset. It resets its side of the
-//! stream with code 0, and reads the next request.
+//! with [`ErrorCode::RequestAborted`]. For example, it dropped the future of
+//! [`Client::rpc`] while it wrote a large request. A client that cannot send a
+//! request resets the stream with [`ErrorCode::RequestMaxSizeExceeded`] or
+//! [`ErrorCode::RequestEncodeFailed`]. The server skips a request whose stream
+//! was reset. It resets its side of the stream with
+//! [`ErrorCode::RequestAborted`], so the receiver of the response gets an
+//! error, and reads the next request.
+//!
+//! # Channel messages
 //!
-//! A remote channel receiver also stops its stream with
-//! [`ErrorCode::DecodeFailed`] if a message does not decode. Its sender then
-//! gets an error on the next send.
+//! A sender resets its stream with [`ErrorCode::MessageMaxSizeExceeded`] if a
+//! message is larger than [`MAX_MESSAGE_SIZE`], and with
+//! [`ErrorCode::MessageEncodeFailed`] if a message does not encode. A receiver
+//! stops its stream with [`ErrorCode::MessageMaxSizeExceeded`] if a message is
+//! larger than [`MAX_MESSAGE_SIZE`], and with [`ErrorCode::MessageDecodeFailed`]
+//! if a message does not decode. The channel then ends. The side that found the
+//! error gets its own variant, for example [`mpsc::RecvError::DecodeFailed`],
+//! and the other side gets the code in a `Remote` error, for example
+//! [`SendError::Remote`].
 //!
 //! # Error codes
 //!
@@ -56,6 +80,7 @@ use n0_future::future::Boxed as BoxFuture;
 use noq::{ConnectionError, VarInt};
 use serde::de::DeserializeOwned;
 use smallvec::SmallVec;
+use tokio::io::AsyncReadExt;
 use tracing::{debug, trace, warn};
 
 #[cfg(feature = "iroh")]
@@ -78,47 +103,82 @@ pub const MAX_MESSAGE_SIZE: u64 = 1024 * 1024 * 16;
 
 /// An error code that irpc uses on streams and connections.
 ///
-/// An application should use codes from 16 upward, see [Error codes].
+/// A `Request` code is about the request, and a `Message` code is about a
+/// channel message, see [Requests and channel messages]. An application should
+/// use codes from 16 upward, see [Error codes].
+///
+/// # Examples
+///
+/// ```
+/// use irpc::rpc::ErrorCode;
+///
+/// let code = ErrorCode::RequestDecodeFailed.to_code();
+/// assert_eq!(
+///     ErrorCode::from_code(code),
+///     Some(ErrorCode::RequestDecodeFailed)
+/// );
+/// ```
 ///
+/// [Requests and channel messages]: self#requests-and-channel-messages
 /// [Error codes]: self#error-codes
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+#[derive(Debug, derive_more::Display, Clone, Copy, PartialEq, Eq, Hash)]
 #[repr(u32)]
 #[non_exhaustive]
 pub enum ErrorCode {
     /// No error, for example a normal close or a dropped receiver.
+    #[display("closed")]
     Closed = 0,
-    /// A message is larger than [`MAX_MESSAGE_SIZE`].
+    /// The client reset the stream of the request before the server read the request.
     ///
-    /// A server also uses this code for a request that is too large, see
-    /// [Bad requests].
+    /// For example, the client dropped the request before it wrote all of it.
+    /// The server skips such a request, and resets its side of the stream with
+    /// this code, see [Abandoned requests].
     ///
-    /// [Bad requests]: self#bad-requests
-    MaxMessageSizeExceeded = 1,
-    /// A message does not encode.
-    EncodeFailed = 2,
-    /// A message does not decode.
+    /// [Abandoned requests]: self#abandoned-requests
+    #[display("request aborted")]
+    RequestAborted = 1,
+    /// The request is larger than [`MAX_MESSAGE_SIZE`].
     ///
-    /// A server also uses this code for a request that does not decode, see
-    /// [Bad requests].
+    /// The server rejects such a request, see [Bad requests]. A client that
+    /// finds its request too large resets the stream with this code.
     ///
     /// [Bad requests]: self#bad-requests
-    DecodeFailed = 3,
-    /// The client dropped the request before it wrote all of it.
+    #[display("request too large")]
+    RequestMaxSizeExceeded = 2,
+    /// The request does not encode.
+    ///
+    /// A client that cannot encode its request resets the stream with this code.
+    #[display("request does not encode")]
+    RequestEncodeFailed = 3,
+    /// The request does not decode, or its stream ended before it was complete.
     ///
-    /// The server skips such a request, see [Bad requests].
+    /// The server rejects such a request, see [Bad requests].
     ///
     /// [Bad requests]: self#bad-requests
-    Aborted = 4,
+    #[display("request does not decode")]
+    RequestDecodeFailed = 4,
+    /// A channel message is larger than [`MAX_MESSAGE_SIZE`].
+    #[display("message too large")]
+    MessageMaxSizeExceeded = 5,
+    /// A channel message does not encode.
+    #[display("message does not encode")]
+    MessageEncodeFailed = 6,
+    /// A channel message does not decode.
+    #[display("message does not decode")]
+    MessageDecodeFailed = 7,
 }
 
 impl ErrorCode {
     /// Every error code, for [`Self::from_code`].
-    const ALL: [Self; 5] = [
+    const ALL: [Self; 8] = [
         Self::Closed,
-        Self::MaxMessageSizeExceeded,
-        Self::EncodeFailed,
-        Self::DecodeFailed,
-        Self::Aborted,
+        Self::RequestAborted,
+        Self::RequestMaxSizeExceeded,
+        Self::RequestEncodeFailed,
+        Self::RequestDecodeFailed,
+        Self::MessageMaxSizeExceeded,
+        Self::MessageEncodeFailed,
+        Self::MessageDecodeFailed,
     ];
 
     /// Returns the number of this error code.
@@ -146,24 +206,26 @@ impl From<ErrorCode> for VarInt {
 
 /// Error when reading a request with [`read_request`].
 ///
-/// For [`MaxMessageSizeExceeded`], [`InvalidRequest`], and [`DecodeFailed`],
-/// irpc resets the streams of the request, so its client gets an error. The
-/// connection is still usable, so a server can read the next request.
+/// For [`MaxSizeExceeded`], [`Malformed`], and
+/// [`DecodeFailed`], irpc stops and resets the streams of the request
+/// with [`ErrorCode::RequestMaxSizeExceeded`] or
+/// [`ErrorCode::RequestDecodeFailed`], so its client gets an error. The connection is still
+/// usable, so a server can read the next request.
 ///
-/// [`MaxMessageSizeExceeded`]: Self::MaxMessageSizeExceeded
-/// [`InvalidRequest`]: Self::InvalidRequest
+/// [`MaxSizeExceeded`]: Self::MaxSizeExceeded
+/// [`Malformed`]: Self::Malformed
 /// [`DecodeFailed`]: Self::DecodeFailed
 #[stack_error(derive, add_meta)]
 #[non_exhaustive]
 pub enum ReadRequestError {
     /// The request is larger than [`MAX_MESSAGE_SIZE`].
-    #[error("Maximum message size exceeded")]
-    MaxMessageSizeExceeded,
+    #[error("Request too large")]
+    MaxSizeExceeded,
     /// The request is not framed correctly.
     ///
     /// For example, the stream ended before the request was complete.
-    #[error("Invalid request")]
-    InvalidRequest {
+    #[error("Request malformed")]
+    Malformed {
         #[error(std_err)]
         source: io::Error,
     },
@@ -188,9 +250,9 @@ impl ReadRequestError {
     /// Returns the irpc error code for a bad request, or `None` for a connection error.
     fn error_code(&self) -> Option<ErrorCode> {
         match self {
-            Self::MaxMessageSizeExceeded { .. } => Some(ErrorCode::MaxMessageSizeExceeded),
-            Self::InvalidRequest { .. } | Self::DecodeFailed { .. } => {
-                Some(ErrorCode::DecodeFailed)
+            Self::MaxSizeExceeded { .. } => Some(ErrorCode::RequestMaxSizeExceeded),
+            Self::Malformed { .. } | Self::DecodeFailed { .. } => {
+                Some(ErrorCode::RequestDecodeFailed)
             }
             Self::Connection { .. } => None,
         }
@@ -211,33 +273,55 @@ impl From<ReadRequestError> for io::Error {
 #[stack_error(derive, add_meta, from_sources)]
 #[non_exhaustive]
 pub enum WriteError {
-    /// Error writing to the stream with noq
-    #[error("Error writing to stream")]
-    Noq {
+    /// The request is larger than [`MAX_MESSAGE_SIZE`], so this client did not send it.
+    #[error("Request too large")]
+    MaxSizeExceeded,
+    /// The request does not encode.
+    #[error("Request does not encode")]
+    EncodeFailed {
         #[error(std_err)]
-        source: noq::WriteError,
+        source: postcard::Error,
     },
-    /// The message exceeded the maximum allowed message size (see [`MAX_MESSAGE_SIZE`]).
-    #[error("Maximum message size exceeded")]
-    MaxMessageSizeExceeded,
-    /// Generic IO error, e.g. when serializing the message or when using
-    /// other transports.
-    #[error("Error serializing")]
+    /// The server stopped the stream with an irpc error code while this client wrote the request.
+    ///
+    /// For example, the server rejected the request, see [Bad requests].
+    ///
+    /// [Bad requests]: self#bad-requests
+    #[error("Server rejected the request: {code}")]
+    Remote { code: ErrorCode },
+    /// An io error, for example because the connection was lost.
+    #[error("Io error")]
     Io {
         #[error(std_err)]
         source: io::Error,
     },
 }
 
-impl From<postcard::Error> for WriteError {
-    fn from(value: postcard::Error) -> Self {
-        e!(Self::Io, io::Error::new(io::ErrorKind::InvalidData, value))
+impl WriteError {
+    /// Returns the code that this client resets its stream with, if it found the error itself.
+    fn reset_code(&self) -> Option<ErrorCode> {
+        match self {
+            Self::MaxSizeExceeded { .. } => Some(ErrorCode::RequestMaxSizeExceeded),
+            Self::EncodeFailed { .. } => Some(ErrorCode::RequestEncodeFailed),
+            _ => None,
+        }
     }
 }
 
-impl From<postcard::Error> for SendError {
-    fn from(value: postcard::Error) -> Self {
-        e!(Self::Io, io::Error::new(io::ErrorKind::InvalidData, value))
+/// Returns the irpc error code that the receiver stopped the stream with, if any.
+fn stop_code(err: &noq::WriteError) -> Option<ErrorCode> {
+    match err {
+        noq::WriteError::Stopped(code) => ErrorCode::from_varint(*code),
+        _ => None,
+    }
+}
+
+impl From<noq::WriteError> for WriteError {
+    fn from(err: noq::WriteError) -> Self {
+        match stop_code(&err).filter(|code| *code != ErrorCode::Closed) {
+            Some(code) => e!(WriteError::Remote { code }),
+            None => e!(WriteError::Io, err.into()),
+        }
     }
 }
 
@@ -245,27 +329,76 @@ impl From<WriteError> for io::Error {
     fn from(e: WriteError) -> Self {
         match e {
             WriteError::Io { source, .. } => source,
-            WriteError::MaxMessageSizeExceeded { .. } => {
-                io::Error::new(io::ErrorKind::InvalidData, e)
-            }
-            WriteError::Noq { source, .. } => source.into(),
+            WriteError::Remote { .. } => io::Error::new(io::ErrorKind::ConnectionReset, e),
+            _ => io::Error::new(io::ErrorKind::InvalidData, e),
         }
     }
 }
 
 impl From<noq::WriteError> for SendError {
     fn from(err: noq::WriteError) -> Self {
-        let code = match &err {
-            noq::WriteError::Stopped(code) => ErrorCode::from_varint(*code),
+        match stop_code(&err) {
+            Some(code) => SendError::from_error_code(code),
+            None => e!(SendError::Io, err.into()),
+        }
+    }
+}
+
+impl WriteError {
+    /// Returns the code of a [`Remote`](Self::Remote) error.
+    pub fn remote_code(&self) -> Option<ErrorCode> {
+        match self {
+            Self::Remote { code, .. } => Some(*code),
             _ => None,
-        };
+        }
+    }
+}
+
+impl oneshot::RecvError {
+    /// Returns the code of a [`Remote`](Self::Remote) error.
+    pub fn remote_code(&self) -> Option<ErrorCode> {
+        match self {
+            Self::Remote { code, .. } => Some(*code),
+            _ => None,
+        }
+    }
+}
+
+impl mpsc::RecvError {
+    /// Returns the code of a [`Remote`](Self::Remote) error.
+    pub fn remote_code(&self) -> Option<ErrorCode> {
+        match self {
+            Self::Remote { code, .. } => Some(*code),
+            _ => None,
+        }
+    }
+}
+
+impl SendError {
+    /// Returns the code of a [`Remote`](Self::Remote) error.
+    pub fn remote_code(&self) -> Option<ErrorCode> {
+        match self {
+            Self::Remote { code, .. } => Some(*code),
+            _ => None,
+        }
+    }
+
+    /// Returns the error for a receiver that stopped the stream with `code`.
+    fn from_error_code(code: ErrorCode) -> Self {
         match code {
-            Some(ErrorCode::MaxMessageSizeExceeded) => e!(SendError::MaxMessageSizeExceeded),
             // A receiver that is dropped stops the stream with code 0.
-            Some(ErrorCode::Closed) => e!(SendError::ReceiverClosed),
-            _ => e!(SendError::Io, io::Error::from(err)),
+            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.
+    fn closed(code: Option<ErrorCode>) -> Self {
+        code.map(Self::from_error_code)
+            .unwrap_or_else(|| io::Error::from(io::ErrorKind::BrokenPipe).into())
+    }
 }
 
 /// Trait to abstract over a client connection to a remote service.
@@ -293,28 +426,39 @@ pub trait RemoteConnection: Send + Sync + Debug + 'static {
 /// A connection to a remote service that can be used to send the initial message.
 ///
 /// If it drops before the message is written, it resets the stream with
-/// [`ErrorCode::Aborted`].
+/// [`ErrorCode::RequestAborted`].
 #[derive(Debug)]
 pub struct RemoteSender<S>(ResetOnDrop, noq::RecvStream, std::marker::PhantomData<S>);
 
-/// A send stream that resets with [`ErrorCode::Aborted`] if it drops before [`Self::into_inner`].
+/// A send stream that resets with [`ErrorCode::RequestAborted`] if it drops before [`Self::into_inner`].
 #[derive(Debug)]
 struct ResetOnDrop(Option<noq::SendStream>);
 
 impl ResetOnDrop {
     fn get_mut(&mut self) -> &mut noq::SendStream {
-        self.0.as_mut().expect("only `into_inner` takes the stream")
+        self.0
+            .as_mut()
+            .expect("only `into_inner` and `reset` take the stream")
     }
 
     fn into_inner(mut self) -> noq::SendStream {
-        self.0.take().expect("only `into_inner` takes the stream")
+        self.0
+            .take()
+            .expect("only `into_inner` and `reset` take the stream")
+    }
+
+    /// Resets the stream with `code`, instead of [`ErrorCode::RequestAborted`] on drop.
+    fn reset(&mut self, code: ErrorCode) {
+        if let Some(mut send) = self.0.take() {
+            send.reset(code.into()).ok();
+        }
     }
 }
 
 impl Drop for ResetOnDrop {
     fn drop(&mut self) {
         if let Some(send) = &mut self.0 {
-            send.reset(ErrorCode::Aborted.into()).ok();
+            send.reset(ErrorCode::RequestAborted.into()).ok();
         }
     }
 }
@@ -324,9 +468,7 @@ impl Drop for ResetOnDrop {
 /// When `S::SPAN_PROPAGATION` is true, the message is wrapped in a tuple with
 /// span context: `(Option<SpanContextCarrier>, msg)`.
 /// When false, the message is serialized directly.
-pub(crate) fn prepare_write<S: Service>(
-    msg: impl Into<S>,
-) -> Result<SmallVec<[u8; 128]>, WriteError> {
+fn prepare_write<S: Service>(msg: impl Into<S>) -> Result<SmallVec<[u8; 128]>, WriteError> {
     let msg = msg.into();
     let mut buf = SmallVec::<[u8; 128]>::new();
 
@@ -335,13 +477,13 @@ pub(crate) fn prepare_write<S: Service>(
         let span_ctx = crate::span_propagation::SpanContextCarrier::from_current();
         let payload = (span_ctx, msg);
         if postcard::experimental::serialized_size(&payload)? as u64 > MAX_MESSAGE_SIZE {
-            return Err(e!(WriteError::MaxMessageSizeExceeded));
+            return Err(e!(WriteError::MaxSizeExceeded));
         }
         buf.write_length_prefixed(&payload)?;
     } else {
         // Original wire format without span context
         if postcard::experimental::serialized_size(&msg)? as u64 > MAX_MESSAGE_SIZE {
-            return Err(e!(WriteError::MaxMessageSizeExceeded));
+            return Err(e!(WriteError::MaxSizeExceeded));
         }
         buf.write_length_prefixed(&msg)?;
     }
@@ -354,14 +496,33 @@ impl<S: Service> RemoteSender<S> {
         Self(ResetOnDrop(Some(send)), recv, PhantomData)
     }
 
+    /// Writes the request `msg`, and returns the streams of the request.
+    ///
+    /// If `msg` is too large or does not encode, this resets the stream with the
+    /// code of the error.
     pub async fn write(
-        self,
+        mut self,
         msg: impl Into<S>,
     ) -> std::result::Result<(noq::SendStream, noq::RecvStream), WriteError> {
-        let buf = prepare_write(msg)?;
+        let buf = self.prepare(msg)?;
         self.write_raw(&buf).await
     }
 
+    /// Encodes `msg` for this request.
+    ///
+    /// If `msg` is too large or does not encode, this resets the stream with the
+    /// code of the error, so the server learns why the request is missing.
+    pub(crate) fn prepare(
+        &mut self,
+        msg: impl Into<S>,
+    ) -> std::result::Result<SmallVec<[u8; 128]>, WriteError> {
+        prepare_write(msg).inspect_err(|err| {
+            if let Some(code) = err.reset_code() {
+                self.0.reset(code);
+            }
+        })
+    }
+
     pub(crate) async fn write_raw(
         self,
         buf: &[u8],
@@ -376,20 +537,20 @@ impl<T: DeserializeOwned> From<noq::RecvStream> for oneshot::Receiver<T> {
     fn from(mut read: noq::RecvStream) -> Self {
         let fut = async move {
             // A sender that is dropped without sending finishes the stream.
-            let Some(size) = read.read_varint_u64().await? else {
+            let Some(size) = read.read_varint_u64().await.map_err(oneshot_recv_error)? else {
                 return Err(e!(oneshot::RecvError::SenderClosed));
             };
             if size > MAX_MESSAGE_SIZE {
-                read.stop(ErrorCode::MaxMessageSizeExceeded.into()).ok();
-                return Err(e!(oneshot::RecvError::MaxMessageSizeExceeded));
+                read.stop(ErrorCode::MessageMaxSizeExceeded.into()).ok();
+                return Err(e!(oneshot::RecvError::MaxSizeExceeded));
             }
-            let rest = read
-                .read_to_end(size as usize)
+            let mut buf = vec![0; size as usize];
+            AsyncReadExt::read_exact(&mut read, &mut buf)
                 .await
-                .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
-            let msg: T = postcard::from_bytes(&rest).map_err(|e| {
-                read.stop(ErrorCode::DecodeFailed.into()).ok();
-                io::Error::new(io::ErrorKind::InvalidData, e)
+                .map_err(oneshot_recv_error)?;
+            let msg: T = postcard::from_bytes(&buf).map_err(|err| {
+                read.stop(ErrorCode::MessageDecodeFailed.into()).ok();
+                e!(oneshot::RecvError::DecodeFailed, err)
             })?;
             Ok(msg)
         };
@@ -397,6 +558,36 @@ impl<T: DeserializeOwned> From<noq::RecvStream> for oneshot::Receiver<T> {
     }
 }
 
+/// Returns the noq read error in `err`, if the stream failed.
+fn read_error(err: &io::Error) -> Option<&noq::ReadError> {
+    err.get_ref()?.downcast_ref()
+}
+
+/// Returns the irpc error code that the sender reset the stream with, if any, other than 0.
+fn reset_code(err: &io::Error) -> Option<ErrorCode> {
+    match read_error(err)? {
+        noq::ReadError::Reset(code) => ErrorCode::from_varint(*code),
+        _ => None,
+    }
+    .filter(|code| *code != ErrorCode::Closed)
+}
+
+/// Returns the error for a oneshot stream that failed.
+fn oneshot_recv_error(err: io::Error) -> oneshot::RecvError {
+    match reset_code(&err) {
+        Some(code) => e!(oneshot::RecvError::Remote { code }),
+        None => err.into(),
+    }
+}
+
+/// Returns the error for an mpsc stream that failed.
+fn mpsc_recv_error(err: io::Error) -> mpsc::RecvError {
+    match reset_code(&err) {
+        Some(code) => e!(mpsc::RecvError::Remote { code }),
+        None => err.into(),
+    }
+}
+
 impl From<noq::RecvStream> for crate::channel::none::NoReceiver {
     fn from(read: noq::RecvStream) -> Self {
         drop(read);
@@ -427,21 +618,18 @@ impl<T: RpcMessage> From<noq::SendStream> for oneshot::Sender<T> {
                 let size = match postcard::experimental::serialized_size(&value) {
                     Ok(size) => size,
                     Err(e) => {
-                        writer.reset(ErrorCode::EncodeFailed.into()).ok();
-                        return Err(e!(
-                            SendError::Io,
-                            io::Error::new(io::ErrorKind::InvalidData, e,)
-                        ));
+                        writer.reset(ErrorCode::MessageEncodeFailed.into()).ok();
+                        return Err(e!(SendError::EncodeFailed, e));
                     }
                 };
                 if size as u64 > MAX_MESSAGE_SIZE {
-                    writer.reset(ErrorCode::MaxMessageSizeExceeded.into()).ok();
-                    return Err(e!(SendError::MaxMessageSizeExceeded));
+                    writer.reset(ErrorCode::MessageMaxSizeExceeded.into()).ok();
+                    return Err(e!(SendError::MaxSizeExceeded));
                 }
                 // write via a small buffer to avoid allocation for small values
                 let mut buf = SmallVec::<[u8; 128]>::new();
                 if let Err(e) = buf.write_length_prefixed(value) {
-                    writer.reset(ErrorCode::EncodeFailed.into()).ok();
+                    writer.reset(ErrorCode::MessageEncodeFailed.into()).ok();
                     return Err(e.into());
                 }
                 writer.write_all(&buf).await?;
@@ -480,22 +668,22 @@ impl<T: RpcMessage> DynReceiver<T> for NoqReceiver<T> {
     ) -> Pin<Box<dyn Future<Output = Result<Option<T>, mpsc::RecvError>> + Send + Sync + '_>> {
         Box::pin(async {
             let read = &mut self.recv;
-            let Some(size) = read.read_varint_u64().await? else {
+            let Some(size) = read.read_varint_u64().await.map_err(mpsc_recv_error)? else {
                 return Ok(None);
             };
             if size > MAX_MESSAGE_SIZE {
                 self.recv
-                    .stop(ErrorCode::MaxMessageSizeExceeded.into())
+                    .stop(ErrorCode::MessageMaxSizeExceeded.into())
                     .ok();
-                return Err(e!(mpsc::RecvError::MaxMessageSizeExceeded));
+                return Err(e!(mpsc::RecvError::MaxSizeExceeded));
             }
             let mut buf = vec![0; size as usize];
-            read.read_exact(&mut buf)
+            AsyncReadExt::read_exact(read, &mut buf)
                 .await
-                .map_err(|e| io::Error::new(io::ErrorKind::UnexpectedEof, e))?;
-            let msg: T = postcard::from_bytes(&buf).map_err(|e| {
-                read.stop(ErrorCode::DecodeFailed.into()).ok();
-                io::Error::new(io::ErrorKind::InvalidData, e)
+                .map_err(mpsc_recv_error)?;
+            let msg: T = postcard::from_bytes(&buf).map_err(|err| {
+                read.stop(ErrorCode::MessageDecodeFailed.into()).ok();
+                e!(mpsc::RecvError::DecodeFailed, err)
             })?;
             Ok(Some(msg))
         })
@@ -521,22 +709,19 @@ impl<T: RpcMessage> NoqSenderInner<T> {
         let size = match postcard::experimental::serialized_size(&value) {
             Ok(size) => size,
             Err(e) => {
-                self.send.reset(ErrorCode::EncodeFailed.into()).ok();
-                return Err(e!(
-                    SendError::Io,
-                    io::Error::new(io::ErrorKind::InvalidData, e)
-                ));
+                self.send.reset(ErrorCode::MessageEncodeFailed.into()).ok();
+                return Err(e!(SendError::EncodeFailed, e));
             }
         };
         if size as u64 > MAX_MESSAGE_SIZE {
             self.send
-                .reset(ErrorCode::MaxMessageSizeExceeded.into())
+                .reset(ErrorCode::MessageMaxSizeExceeded.into())
                 .ok();
-            return Err(e!(SendError::MaxMessageSizeExceeded));
+            return Err(e!(SendError::MaxSizeExceeded));
         }
         self.buffer.clear();
         if let Err(e) = self.buffer.write_length_prefixed(value) {
-            self.send.reset(ErrorCode::EncodeFailed.into()).ok();
+            self.send.reset(ErrorCode::MessageEncodeFailed.into()).ok();
             return Err(e.into());
         }
         Ok(())
@@ -580,22 +765,8 @@ impl<T: RpcMessage> NoqSenderInner<T> {
 
 enum NoqSenderState<T> {
     Open(NoqSenderInner<T>),
-    Closed(CloseReason),
-}
-
-#[derive(Debug, Clone, Copy)]
-enum CloseReason {
-    ReceiverClosed,
-    Other,
-}
-
-impl CloseReason {
-    fn error(self) -> SendError {
-        match self {
-            Self::ReceiverClosed => e!(SendError::ReceiverClosed),
-            Self::Other => io::Error::from(io::ErrorKind::BrokenPipe).into(),
-        }
-    }
+    /// Closed by an error, with the code that the receiver stopped the stream with if it did.
+    Closed(Option<ErrorCode>),
 }
 
 impl<T> NoqSenderState<T> {
@@ -604,15 +775,15 @@ impl<T> NoqSenderState<T> {
     /// A cancelled send never puts it back, so the sender stays closed: the
     /// stream can hold part of a message.
     fn take_for_send(&mut self) -> Self {
-        std::mem::replace(self, Self::Closed(CloseReason::Other))
+        std::mem::replace(self, Self::Closed(None))
     }
 
-    /// Returns the state after a send, so that later sends fail with the same kind of error.
+    /// Returns the state after a send, so that later sends fail with the same error.
     fn after_send<R>(sender: NoqSenderInner<T>, res: &Result<R, SendError>) -> Self {
         match res {
             Ok(_) => Self::Open(sender),
-            Err(SendError::ReceiverClosed { .. }) => Self::Closed(CloseReason::ReceiverClosed),
-            Err(_) => Self::Closed(CloseReason::Other),
+            Err(SendError::ReceiverClosed { .. }) => Self::Closed(Some(ErrorCode::Closed)),
+            Err(err) => Self::Closed(err.remote_code()),
         }
     }
 }
@@ -636,9 +807,9 @@ impl<T: RpcMessage> DynSender<T> for NoqSender<T> {
                     *guard = NoqSenderState::after_send(sender, &res);
                     res
                 }
-                NoqSenderState::Closed(reason) => {
-                    *guard = NoqSenderState::Closed(reason);
-                    Err(reason.error())
+                NoqSenderState::Closed(code) => {
+                    *guard = NoqSenderState::Closed(code);
+                    Err(SendError::closed(code))
                 }
             }
         })
@@ -657,9 +828,9 @@ impl<T: RpcMessage> DynSender<T> for NoqSender<T> {
                     *guard = NoqSenderState::after_send(sender, &res);
                     res
                 }
-                NoqSenderState::Closed(reason) => {
-                    *guard = NoqSenderState::Closed(reason);
-                    Err(reason.error())
+                NoqSenderState::Closed(code) => {
+                    *guard = NoqSenderState::Closed(code);
+                    Err(SendError::closed(code))
                 }
             }
         })
@@ -727,8 +898,8 @@ impl HandlerError {
     /// The code should be 16 or higher, see [Error codes]. irpc
     /// closes a connection with code 0 for a normal close, which
     /// [`Handler::from_sender`] also uses when its receiver is gone. For a bad
-    /// request, irpc closes the connection with
-    /// [`ErrorCode::MaxMessageSizeExceeded`] or [`ErrorCode::DecodeFailed`].
+    /// request, irpc closes the connection with [`ErrorCode::RequestMaxSizeExceeded`]
+    /// or [`ErrorCode::RequestDecodeFailed`].
     ///
     /// # Examples
     ///
@@ -993,8 +1164,8 @@ pub trait IncomingRemoteConnection: crate::sealed::Sealed {
 /// Returns `None` if the remote closed the connection with code 0, or if this
 /// side closed it. Skips a request that the client abandoned.
 ///
-/// Returns [`ReadRequestError::MaxMessageSizeExceeded`],
-/// [`ReadRequestError::InvalidRequest`], or [`ReadRequestError::DecodeFailed`]
+/// Returns [`ReadRequestError::MaxSizeExceeded`],
+/// [`ReadRequestError::Malformed`], or [`ReadRequestError::DecodeFailed`]
 /// for a bad request. The connection is still open after these errors, see
 /// [Bad requests].
 ///
@@ -1048,7 +1219,7 @@ async fn read_request_inner<S: Service>(
         let (mut send, mut recv) = match connection.accept_bi().await {
             Ok(streams) => streams,
             Err(ConnectionError::ApplicationClosed(cause))
-                if ErrorCode::from_varint(cause.error_code) == Some(ErrorCode::Closed) =>
+                if cause.error_code == ErrorCode::Closed.into() =>
             {
                 trace!("remote side closed connection {cause:?}");
                 return Ok(None);
@@ -1058,13 +1229,14 @@ async fn read_request_inner<S: Service>(
         };
         let err = match read_request_frame::<S>(&mut recv).await {
             ReadFrame::Request(carrier, msg) => return Ok(Some((msg, carrier, recv, send))),
-            ReadFrame::MaxMessageSizeExceeded => e!(ReadRequestError::MaxMessageSizeExceeded),
-            ReadFrame::InvalidRequest(source) => e!(ReadRequestError::InvalidRequest, source),
+            ReadFrame::MaxSizeExceeded => e!(ReadRequestError::MaxSizeExceeded),
+            ReadFrame::Malformed(source) => e!(ReadRequestError::Malformed, source),
             ReadFrame::DecodeFailed(source) => e!(ReadRequestError::DecodeFailed, source),
-            ReadFrame::Reset => {
-                debug!("skipped request: the client reset its stream");
-                // A drop finishes the stream, which looks like an empty response.
-                send.reset(ErrorCode::Closed.into()).ok();
+            ReadFrame::Reset(reason) => {
+                debug!(?reason, "skipped request: the client reset its stream");
+                // A dropped stream finishes, which a client with an mpsc
+                // response would take for the end of the response.
+                send.reset(ErrorCode::RequestAborted.into()).ok();
                 continue;
             }
             // The next `accept_bi` returns why.
@@ -1083,13 +1255,13 @@ enum ReadFrame<S> {
     /// A complete request.
     Request(Option<crate::span_propagation::SpanContextCarrier>, S),
     /// The request is larger than [`MAX_MESSAGE_SIZE`].
-    MaxMessageSizeExceeded,
+    MaxSizeExceeded,
     /// The stream ended before the request was complete, or the size prefix is too long.
-    InvalidRequest(io::Error),
+    Malformed(io::Error),
     /// The request does not decode.
     DecodeFailed(postcard::Error),
-    /// The client reset the stream before the request was complete.
-    Reset,
+    /// The client reset the stream before the request was complete, with the irpc code if it sent one.
+    Reset(Option<ErrorCode>),
     /// The connection was lost before the request was complete.
     ConnectionLost,
 }
@@ -1098,14 +1270,14 @@ impl<S> ReadFrame<S> {
     /// Returns the frame for a stream that failed before the request was complete.
     fn from_read_err(err: &noq::ReadError) -> Self {
         match err {
-            // Any reset, not only `ErrorCode::Aborted`: the update sender of a
+            // Any reset, not only `ErrorCode::RequestAborted`: the update sender of a
             // request resets the stream if an update does not encode. noq drops
             // unread data on a reset, so this reset can arrive before the request.
-            noq::ReadError::Reset(_) => Self::Reset,
+            noq::ReadError::Reset(code) => Self::Reset(ErrorCode::from_varint(*code)),
             noq::ReadError::ConnectionLost(_) => Self::ConnectionLost,
             // A server cannot get these: irpc stops a request stream only after
             // its read failed, and only a client gets `ZeroRttRejected`.
-            noq::ReadError::ClosedStream | noq::ReadError::ZeroRttRejected => Self::Reset,
+            noq::ReadError::ClosedStream | noq::ReadError::ZeroRttRejected => Self::Reset(None),
         }
     }
 }
@@ -1115,31 +1287,28 @@ async fn read_request_frame<S: Service>(recv: &mut noq::RecvStream) -> ReadFrame
     let size = match recv.read_varint_u64().await {
         Ok(Some(size)) => size,
         Ok(None) => {
-            return ReadFrame::InvalidRequest(io::Error::new(
+            return ReadFrame::Malformed(io::Error::new(
                 io::ErrorKind::UnexpectedEof,
                 "stream ended before the request",
             ));
         }
         Err(err) => {
-            let read_err = err
-                .get_ref()
-                .and_then(|err| err.downcast_ref::<noq::ReadError>());
-            return match read_err {
+            return match read_error(&err) {
                 Some(read_err) => ReadFrame::from_read_err(read_err),
                 // The varint is longer than a u64, or the stream ended inside it.
-                None => ReadFrame::InvalidRequest(err),
+                None => ReadFrame::Malformed(err),
             };
         }
     };
     if size > MAX_MESSAGE_SIZE {
-        return ReadFrame::MaxMessageSizeExceeded;
+        return ReadFrame::MaxSizeExceeded;
     }
     let mut buf = vec![0; size as usize];
     match recv.read_exact(&mut buf).await {
         Ok(()) => {}
         Err(noq::ReadExactError::ReadError(err)) => return ReadFrame::from_read_err(&err),
         Err(err @ noq::ReadExactError::FinishedEarly(_)) => {
-            return ReadFrame::InvalidRequest(io::Error::new(io::ErrorKind::UnexpectedEof, err));
+            return ReadFrame::Malformed(io::Error::new(io::ErrorKind::UnexpectedEof, err));
         }
     }
     let decoded = if S::SPAN_PROPAGATION {
@@ -1164,4 +1333,15 @@ mod tests {
         }
         assert_eq!(ErrorCode::from_code(16), None);
     }
+
+    #[test]
+    fn write_error_maps_stop_codes() {
+        let err = WriteError::from(noq::WriteError::Stopped(
+            ErrorCode::RequestDecodeFailed.into(),
+        ));
+        assert_eq!(err.remote_code(), Some(ErrorCode::RequestDecodeFailed));
+        assert_eq!(io::Error::from(err).kind(), io::ErrorKind::ConnectionReset);
+        let err = WriteError::from(noq::WriteError::Stopped(16u32.into()));
+        assert!(matches!(err, WriteError::Io { .. }), "{err:?}");
+    }
 }
diff --git a/tests/invalid_request.rs b/tests/invalid_request.rs
index ebf0b7cb..4c133cec 100644
--- a/tests/invalid_request.rs
+++ b/tests/invalid_request.rs
@@ -5,8 +5,9 @@ use std::{pin::pin, time::Duration};
 
 use irpc::{
     Client, WithChannels,
+    channel::oneshot,
     noq::listen,
-    rpc::{ErrorCode, Handler, MAX_MESSAGE_SIZE, ReadRequestError, read_request},
+    rpc::{ErrorCode, Handler, MAX_MESSAGE_SIZE, ReadRequestError, WriteError, read_request},
 };
 use n0_future::{future::poll_once, task::AbortOnDropHandle};
 use noq::ConnectionError;
@@ -35,6 +36,9 @@ mod server {
         #[rpc(tx = oneshot::Sender<()>, rx = mpsc::Receiver<NoSer>)]
         #[wrap(Upload)]
         Upload(String),
+        #[rpc(tx = mpsc::Sender<()>, rx = mpsc::Receiver<NoSer>)]
+        #[wrap(Exchange)]
+        Exchange(String),
     }
 }
 
@@ -57,9 +61,15 @@ mod client {
         #[rpc(tx = oneshot::Sender<()>, rx = mpsc::Receiver<NoSer>)]
         #[wrap(Upload)]
         Upload(String),
+        #[rpc(tx = mpsc::Sender<()>, rx = mpsc::Receiver<NoSer>)]
+        #[wrap(Exchange)]
+        Exchange(String),
         #[rpc(tx = oneshot::Sender<String>)]
         #[wrap(Shout)]
         Shout(String),
+        #[rpc(tx = oneshot::Sender<()>, rx = mpsc::Receiver<u32>)]
+        #[wrap(ShoutStream)]
+        ShoutStream(String),
     }
 }
 
@@ -87,7 +97,7 @@ async fn bad_request_closes_connection() -> TestResult<()> {
     let ConnectionError::ApplicationClosed(close) = conn.closed().await else {
         panic!("server closes the connection");
     };
-    assert_eq!(close.error_code, ErrorCode::DecodeFailed.into());
+    assert_eq!(close.error_code, ErrorCode::RequestDecodeFailed.into());
     Ok(())
 }
 
@@ -104,7 +114,7 @@ async fn too_large_request_closes_connection() -> TestResult<()> {
     let ConnectionError::ApplicationClosed(close) = conn.closed().await else {
         panic!("server closes the connection");
     };
-    assert_eq!(close.error_code, ErrorCode::MaxMessageSizeExceeded.into());
+    assert_eq!(close.error_code, ErrorCode::RequestMaxSizeExceeded.into());
     Ok(())
 }
 
@@ -117,14 +127,87 @@ async fn skip_bad_requests_keeps_connection() -> TestResult<()> {
     // A client on this one connection, so the second request shows that it still works.
     let client = Client::<client::EchoProtocol>::boxed(conn);
 
-    client
+    let err = client
         .rpc(client::Shout("a".into()))
         .await
         .expect_err("server does not know the request");
+    assert!(
+        matches!(
+            err,
+            irpc::Error::OneshotRecv {
+                source: oneshot::RecvError::Remote {
+                    code: ErrorCode::RequestDecodeFailed,
+                    ..
+                },
+                ..
+            }
+        ),
+        "{err:?}"
+    );
     assert_eq!(client.rpc(client::Echo("b".into())).await?, "b");
     Ok(())
 }
 
+/// The sender of the updates of a bad request gets the code of the rejection, as the receiver of its response does.
+#[tokio::test]
+async fn bad_request_update_sender() -> TestResult<()> {
+    let (server, client_endpoint, server_addr) = create_connected_endpoints()?;
+    let handler = echo_handler().skip_bad_requests(true);
+    let _server = AbortOnDropHandle::new(tokio::spawn(listen(server, handler)));
+    let conn = client_endpoint.connect(server_addr, "localhost")?.await?;
+    let client = Client::<client::EchoProtocol>::boxed(conn);
+
+    let (tx, rx) = client
+        .client_streaming(client::ShoutStream("a".into()), 1)
+        .await?;
+    let err = rx.await.expect_err("server does not know the request");
+    assert_eq!(err.remote_code(), Some(ErrorCode::RequestDecodeFailed));
+    let err = tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            if let Err(err) = tx.send(1).await {
+                break err;
+            }
+        }
+    })
+    .await?;
+    assert_eq!(err.remote_code(), Some(ErrorCode::RequestDecodeFailed));
+    Ok(())
+}
+
+/// A client that cannot send a request resets its stream with the reason.
+#[tokio::test]
+async fn too_large_rpc_resets_with_reason() -> TestResult<()> {
+    let (server, client_endpoint, server_addr) = create_connected_endpoints()?;
+    let connecting = client_endpoint.connect(server_addr, "localhost")?;
+    let accepting = async { server.accept().await.expect("endpoint is open").await };
+    let (client_conn, server_conn) = tokio::try_join!(connecting, accepting)?;
+    let client = Client::<client::EchoProtocol>::boxed(client_conn);
+
+    let large = "a".repeat(MAX_MESSAGE_SIZE as usize + 1);
+    let err = client
+        .rpc(client::Echo(large))
+        .await
+        .expect_err("the request is too large");
+    assert!(
+        matches!(
+            err,
+            irpc::Error::Write {
+                source: WriteError::MaxSizeExceeded { .. },
+                ..
+            }
+        ),
+        "{err:?}"
+    );
+    let (_send, mut recv) = server_conn.accept_bi().await?;
+    assert_eq!(
+        recv.read(&mut [0]).await,
+        Err(noq::ReadError::Reset(
+            ErrorCode::RequestMaxSizeExceeded.into()
+        ))
+    );
+    Ok(())
+}
+
 #[tokio::test]
 async fn dropped_rpc_keeps_connection() -> TestResult<()> {
     let (server, client_endpoint, server_addr) = create_connected_endpoints()?;
@@ -133,7 +216,7 @@ async fn dropped_rpc_keeps_connection() -> TestResult<()> {
     let client = Client::<client::EchoProtocol>::boxed(conn);
 
     // The request is larger than the flow control window, so one poll cannot
-    // write all of it. The drop then resets the stream with `ErrorCode::Aborted`.
+    // write all of it. The drop then resets the stream with `ErrorCode::RequestAborted`.
     let mut rpc = Box::pin(client.rpc(client::Echo("a".repeat(8 * 1024 * 1024))));
     assert!(poll_once(&mut rpc).await.is_none(), "the write is not done");
     drop(rpc);
@@ -154,7 +237,7 @@ async fn truncated_request_closes_connection() -> TestResult<()> {
     let ConnectionError::ApplicationClosed(close) = conn.closed().await else {
         panic!("server closes the connection");
     };
-    assert_eq!(close.error_code, ErrorCode::DecodeFailed.into());
+    assert_eq!(close.error_code, ErrorCode::RequestDecodeFailed.into());
     Ok(())
 }
 
@@ -180,14 +263,55 @@ async fn reset_before_read_skips_request() -> TestResult<()> {
     let conn = client_endpoint.connect(server_addr, "localhost")?.await?;
     let client = Client::<client::EchoProtocol>::boxed(conn.clone());
 
-    let (tx, _rx) = client
+    let (tx, rx) = client
         .client_streaming(client::Upload("a".into()), 1)
         .await?;
-    // The update does not encode, so the sender resets with `ErrorCode::EncodeFailed`.
+    // The update does not encode, so the sender resets with `ErrorCode::MessageEncodeFailed`.
     tx.send(NoSer(1))
         .await
         .expect_err("odd numbers do not encode");
     assert_eq!(client.rpc(client::Echo("b".into())).await?, "b");
+    // The server skipped the upload and reset its side of the stream.
+    let err = rx.await.expect_err("server skips the upload");
+    assert_eq!(err.remote_code(), Some(ErrorCode::RequestAborted));
+    conn.close(0u32.into(), b"");
+    server.await??;
+    Ok(())
+}
+
+/// Like [`reset_before_read_skips_request`], but the response is an mpsc channel.
+#[tokio::test]
+async fn reset_before_read_fails_mpsc_response() -> TestResult<()> {
+    let (server, client_endpoint, server_addr) = create_connected_endpoints()?;
+    let server = tokio::spawn(async move {
+        let conn = server.accept().await.expect("endpoint is open").await?;
+        // Read only after the reset arrived: noq then has dropped the request.
+        while conn.stats().frame_rx.reset_stream == 0 {
+            tokio::time::sleep(Duration::from_millis(1)).await;
+        }
+        let Some(server::EchoMessage::Echo(msg)) =
+            read_request::<server::EchoProtocol>(&conn).await?
+        else {
+            panic!("server skips the exchange");
+        };
+        msg.tx.send(msg.inner.0).await?;
+        conn.closed().await;
+        TestResult::Ok(())
+    });
+    let conn = client_endpoint.connect(server_addr, "localhost")?.await?;
+    let client = Client::<client::EchoProtocol>::boxed(conn.clone());
+
+    let (tx, mut rx) = client
+        .bidi_streaming(client::Exchange("a".into()), 1, 1)
+        .await?;
+    // The update does not encode, so the sender resets with `ErrorCode::MessageEncodeFailed`.
+    tx.send(NoSer(1))
+        .await
+        .expect_err("odd numbers do not encode");
+    assert_eq!(client.rpc(client::Echo("b".into())).await?, "b");
+    // The server skipped the exchange, so the response fails instead of ending.
+    let err = rx.recv().await.expect_err("server skips the exchange");
+    assert_eq!(err.remote_code(), Some(ErrorCode::RequestAborted));
     conn.close(0u32.into(), b"");
     server.await??;
     Ok(())
@@ -207,10 +331,7 @@ async fn read_request_returns_truncated_request() -> TestResult<()> {
     let err = read_request::<server::EchoProtocol>(&server_conn)
         .await
         .expect_err("the request is not complete");
-    assert!(
-        matches!(err, ReadRequestError::InvalidRequest { .. }),
-        "{err:?}"
-    );
+    assert!(matches!(err, ReadRequestError::Malformed { .. }), "{err:?}");
     Ok(())
 }
 
diff --git a/tests/mpsc_channel.rs b/tests/mpsc_channel.rs
index 3af0535f..508dc5d5 100644
--- a/tests/mpsc_channel.rs
+++ b/tests/mpsc_channel.rs
@@ -288,13 +288,16 @@ async fn mpsc_max_message_size_send() -> TestResult<()> {
     let Err(cause) = send.send(vec![0u8; 1024 * 1024 * 32]).await else {
         panic!("client should have failed due to max message size");
     };
-    assert!(matches!(cause, SendError::MaxMessageSizeExceeded { .. }));
+    assert!(matches!(cause, SendError::MaxSizeExceeded { .. }));
+    // later sends get a broken pipe, also for a message that is small enough
+    assert!(matches!(
+        send.send(vec![0u8; 1]).await,
+        Err(SendError::Io { source, .. }) if source.kind() == io::ErrorKind::BrokenPipe
+    ));
     let Err(cause) = server.await? else {
         panic!("server should have failed due to max message size");
     };
-    assert!(
-        matches!(cause, mpsc::RecvError::Io { source, .. } if source.kind() == ErrorKind::ConnectionReset)
-    );
+    assert_eq!(cause.remote_code(), Some(ErrorCode::MessageMaxSizeExceeded));
     Ok(())
 }
 
@@ -309,13 +312,11 @@ async fn mpsc_max_message_size_try_send() -> TestResult<()> {
     let Err(cause) = send.try_send(vec![0u8; 1024 * 1024 * 32]).await else {
         panic!("client should have failed due to max message size");
     };
-    assert!(matches!(cause, SendError::MaxMessageSizeExceeded { .. }));
+    assert!(matches!(cause, SendError::MaxSizeExceeded { .. }));
     let Err(cause) = server.await? else {
         panic!("server should have failed due to max message size");
     };
-    assert!(
-        matches!(cause, mpsc::RecvError::Io { source, .. } if source.kind() == ErrorKind::ConnectionReset)
-    );
+    assert_eq!(cause.remote_code(), Some(ErrorCode::MessageMaxSizeExceeded));
     Ok(())
 }
 
@@ -335,10 +336,7 @@ 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,
-        mpsc::RecvError::MaxMessageSizeExceeded { .. }
-    ));
+    assert!(matches!(cause, mpsc::RecvError::MaxSizeExceeded { .. }));
     Ok(())
 }
 
@@ -372,15 +370,11 @@ async fn mpsc_serialize_error_send() -> TestResult<()> {
     let Err(cause) = send.send(NoSer(1)).await else {
         panic!("client should have failed due to serialization error");
     };
-    assert!(
-        matches!(cause, SendError::Io { source, .. } if source.kind() == ErrorKind::InvalidData)
-    );
+    assert!(matches!(cause, SendError::EncodeFailed { .. }), "{cause:?}");
     let Err(cause) = server.await? else {
         panic!("server should have failed due to serialization error");
     };
-    assert!(
-        matches!(cause, mpsc::RecvError::Io { source, .. } if source.kind() == ErrorKind::ConnectionReset)
-    );
+    assert_eq!(cause.remote_code(), Some(ErrorCode::MessageEncodeFailed));
     Ok(())
 }
 
@@ -398,30 +392,74 @@ async fn mpsc_serialize_error_recv() -> TestResult<()> {
         panic!("server should have failed due to serialization error");
     };
     assert!(
-        matches!(cause, mpsc::RecvError::Io { source, .. } if source.kind() == ErrorKind::InvalidData)
+        matches!(cause, mpsc::RecvError::DecodeFailed { .. }),
+        "{cause:?}"
+    );
+    Ok(())
+}
+
+/// Receives a message that does not decode, and keeps the connection until the client closes it.
+async fn noser_receiver_kept(server: Endpoint) -> TestResult<()> {
+    let conn = server.accept().await.unwrap().await?;
+    let (_, recv) = conn.accept_bi().await?;
+    let mut recv = Receiver::<NoSer>::from(recv);
+    assert!(recv.recv().await.is_err());
+    // keep the connection, so that the stop reaches the client
+    conn.closed().await;
+    Ok(())
+}
+
+/// Checks that a sender gets `MessageDecodeFailed` from the remote receiver if it cannot decode a message.
+#[tokio::test]
+async fn mpsc_decode_error_reaches_sender() -> TestResult<()> {
+    let (server, client, server_addr) = create_connected_endpoints()?;
+    let server = tokio::spawn(noser_receiver_kept(server));
+    let conn = client.connect(server_addr, "localhost")?.await?;
+    let (send, _) = conn.open_bi().await?;
+    // An odd number encodes as `u64`, but does not decode as `NoSer`.
+    let send = mpsc::Sender::<u64>::from(send);
+    let clone = send.clone();
+    let err = timeout(Duration::from_secs(5), async {
+        loop {
+            if let Err(err) = send.send(1).await {
+                break err;
+            }
+        }
+    })
+    .await?;
+    assert_eq!(err.remote_code(), Some(ErrorCode::MessageDecodeFailed));
+    // later sends, also on a clone that did not see the stop
+    assert_eq!(
+        send.send(2)
+            .await
+            .expect_err("the sender is closed")
+            .remote_code(),
+        Some(ErrorCode::MessageDecodeFailed)
     );
+    assert_eq!(
+        clone
+            .try_send(2)
+            .await
+            .expect_err("the sender is closed")
+            .remote_code(),
+        Some(ErrorCode::MessageDecodeFailed)
+    );
+    conn.close(0u32.into(), b"");
+    server.await??;
     Ok(())
 }
 
-/// Checks that a receiver stops the stream with code 3 if a message does not decode.
+/// Checks that a receiver stops the stream with `MessageDecodeFailed` if a message does not decode.
 #[tokio::test]
 async fn mpsc_decode_error_stops_stream() -> TestResult<()> {
     let (server, client, server_addr) = create_connected_endpoints()?;
-    let server = tokio::spawn(async move {
-        let conn = server.accept().await.unwrap().await?;
-        let (_, recv) = conn.accept_bi().await?;
-        let mut recv = Receiver::<NoSer>::from(recv);
-        assert!(recv.recv().await.is_err());
-        // keep the connection, so that the stop reaches the client
-        conn.closed().await;
-        TestResult::Ok(())
-    });
+    let server = tokio::spawn(noser_receiver_kept(server));
     let conn = client.connect(server_addr, "localhost")?.await?;
     let (mut send, _) = conn.open_bi().await?;
     // an odd number does not decode as `NoSer`
     send.write_length_prefixed(1u64).await?;
     let code = timeout(Duration::from_secs(5), send.stopped()).await??;
-    assert_eq!(code, Some(ErrorCode::DecodeFailed.into()));
+    assert_eq!(code, Some(ErrorCode::MessageDecodeFailed.into()));
     conn.close(0u32.into(), b"");
     server.await??;
     Ok(())
diff --git a/tests/oneshot_channel.rs b/tests/oneshot_channel.rs
index 1cde1469..720c55d4 100644
--- a/tests/oneshot_channel.rs
+++ b/tests/oneshot_channel.rs
@@ -1,12 +1,13 @@
 #![cfg(feature = "noq_endpoint_setup")]
 
-use std::io::{self, ErrorKind};
+use std::io;
 
 use irpc::{
     channel::{
         SendError,
         oneshot::{self, RecvError},
     },
+    rpc::ErrorCode,
     util::AsyncWriteVarintExt,
 };
 use n0_error::e;
@@ -65,13 +66,11 @@ async fn oneshot_max_message_size_send() -> TestResult<()> {
     let Err(cause) = send.send(vec![0u8; 1024 * 1024 * 32]).await else {
         panic!("client should have failed due to max message size");
     };
-    assert!(matches!(cause, SendError::MaxMessageSizeExceeded { .. }));
+    assert!(matches!(cause, SendError::MaxSizeExceeded { .. }));
     let Err(cause) = server.await? else {
         panic!("server should have failed due to max message size");
     };
-    assert!(
-        matches!(cause, RecvError::Io { source, .. } if source.kind() == ErrorKind::ConnectionReset)
-    );
+    assert_eq!(cause.remote_code(), Some(ErrorCode::MessageMaxSizeExceeded));
     Ok(())
 }
 
@@ -89,7 +88,7 @@ async fn oneshot_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, RecvError::MaxSizeExceeded { .. }));
     Ok(())
 }
 
@@ -109,7 +108,7 @@ async fn noser_receiver(server: Endpoint) -> Result<(), RecvError> {
     Err(e!(RecvError::Io, io::ErrorKind::UnexpectedEof.into()))
 }
 
-/// Checks that trying to send a message that cannot be serialized results in an error on the sender side and a connection reset on the receiver side.
+/// Checks that a message that does not encode gives `MessageEncodeFailed` on both sides, as an error code for the receiver.
 #[tokio::test]
 async fn oneshot_serialize_error_send() -> TestResult<()> {
     let (server, client, server_addr) = create_connected_endpoints()?;
@@ -121,16 +120,11 @@ async fn oneshot_serialize_error_send() -> TestResult<()> {
     let Err(cause) = send.send(NoSer(1)).await else {
         panic!("client should have failed due to serialization error");
     };
-    assert!(
-        matches!(cause, SendError::Io { source, .. } if source.kind() == ErrorKind::InvalidData)
-    );
+    assert!(matches!(cause, SendError::EncodeFailed { .. }), "{cause:?}");
     let Err(cause) = server.await? else {
         panic!("server should have failed due to serialization error");
     };
-    println!("Server error: {cause:?}");
-    assert!(
-        matches!(cause, RecvError::Io { source, .. } if source.kind() == ErrorKind::ConnectionReset)
-    );
+    assert_eq!(cause.remote_code(), Some(ErrorCode::MessageEncodeFailed));
     Ok(())
 }
 
@@ -147,8 +141,6 @@ async fn oneshot_serialize_error_recv() -> TestResult<()> {
         panic!("server should have failed due to serialization error");
     };
     println!("Server error: {cause:?}");
-    assert!(
-        matches!(cause, RecvError::Io { source, .. } if source.kind() == ErrorKind::InvalidData)
-    );
+    assert!(matches!(cause, RecvError::DecodeFailed { .. }), "{cause:?}");
     Ok(())
 }

From d94c4f0461d25ae4da96409e3a832921383129f3 Mon Sep 17 00:00:00 2001
From: Frando <frando@unbiskant.org>
Date: Mon, 12 Oct 2026 13:33:12 +0200
Subject: [PATCH 3/4] refactor: encode requests and channel messages in one
 function

The client encoded a request, and the oneshot and mpsc senders encoded a
message, each with its own copy of the size check, the encoding, and the
reset. A message that passed the size check but then failed to encode
gave an `Io` error, and a request in that case reset its stream with
`RequestAborted` instead of `RequestEncodeFailed`. `encode` now measures
a value once, checks its size, and encodes it, and `encode_message` also
resets the stream of a channel message that does not encode.
---
 src/rpc.rs  | 121 ++++++++++++++++++++++++++--------------------------
 src/util.rs |   1 -
 2 files changed, 61 insertions(+), 61 deletions(-)

diff --git a/src/rpc.rs b/src/rpc.rs
index 1d52381a..01def4b0 100644
--- a/src/rpc.rs
+++ b/src/rpc.rs
@@ -78,7 +78,7 @@ use std::{
 use n0_error::{e, stack_error};
 use n0_future::future::Boxed as BoxFuture;
 use noq::{ConnectionError, VarInt};
-use serde::de::DeserializeOwned;
+use serde::{Serialize, de::DeserializeOwned};
 use smallvec::SmallVec;
 use tokio::io::AsyncReadExt;
 use tracing::{debug, trace, warn};
@@ -470,25 +470,66 @@ impl Drop for ResetOnDrop {
 /// When false, the message is serialized directly.
 fn prepare_write<S: Service>(msg: impl Into<S>) -> Result<SmallVec<[u8; 128]>, WriteError> {
     let msg = msg.into();
-    let mut buf = SmallVec::<[u8; 128]>::new();
-
-    if S::SPAN_PROPAGATION {
+    let mut buf = SmallVec::new();
+    let res = if S::SPAN_PROPAGATION {
         // Include span context in wire format
         let span_ctx = crate::span_propagation::SpanContextCarrier::from_current();
-        let payload = (span_ctx, msg);
-        if postcard::experimental::serialized_size(&payload)? as u64 > MAX_MESSAGE_SIZE {
-            return Err(e!(WriteError::MaxSizeExceeded));
-        }
-        buf.write_length_prefixed(&payload)?;
+        encode(&(span_ctx, msg), &mut buf)
     } else {
         // Original wire format without span context
-        if postcard::experimental::serialized_size(&msg)? as u64 > MAX_MESSAGE_SIZE {
-            return Err(e!(WriteError::MaxSizeExceeded));
-        }
-        buf.write_length_prefixed(&msg)?;
+        encode(&msg, &mut buf)
+    };
+    res.map_err(|err| match err {
+        EncodeError::MaxSizeExceeded => e!(WriteError::MaxSizeExceeded),
+        EncodeError::Failed(source) => e!(WriteError::EncodeFailed, source),
+    })?;
+    Ok(buf)
+}
+
+/// Why a message did not encode.
+enum EncodeError {
+    /// The message is larger than [`MAX_MESSAGE_SIZE`].
+    MaxSizeExceeded,
+    /// The message does not encode.
+    Failed(postcard::Error),
+}
+
+/// Encodes `value` with its size prefix into `buf`.
+fn encode(value: &impl Serialize, buf: &mut SmallVec<[u8; 128]>) -> Result<(), EncodeError> {
+    let size = postcard::experimental::serialized_size(value).map_err(EncodeError::Failed)? as u64;
+    if size > MAX_MESSAGE_SIZE {
+        return Err(EncodeError::MaxSizeExceeded);
     }
+    buf.clear();
+    buf.write_varint_u64(size)
+        .expect("a buffer does not fail to write");
+    postcard::to_io(value, buf).map_err(EncodeError::Failed)?;
+    Ok(())
+}
 
-    Ok(buf)
+/// Encodes the channel message `value` with its size prefix into `buf`.
+///
+/// Resets `send` if `value` is too large or does not encode, so the receiver
+/// gets an error instead of the end of the stream.
+fn encode_message(
+    value: &impl Serialize,
+    buf: &mut SmallVec<[u8; 128]>,
+    send: &mut noq::SendStream,
+) -> Result<(), SendError> {
+    encode(value, buf).map_err(|err| {
+        let (code, err) = match err {
+            EncodeError::MaxSizeExceeded => (
+                ErrorCode::MessageMaxSizeExceeded,
+                e!(SendError::MaxSizeExceeded),
+            ),
+            EncodeError::Failed(source) => (
+                ErrorCode::MessageEncodeFailed,
+                e!(SendError::EncodeFailed, source),
+            ),
+        };
+        send.reset(code.into()).ok();
+        err
+    })
 }
 
 impl<S: Service> RemoteSender<S> {
@@ -615,23 +656,9 @@ impl<T: RpcMessage> From<noq::SendStream> for oneshot::Sender<T> {
     fn from(mut writer: noq::SendStream) -> Self {
         oneshot::Sender::Boxed(Box::new(move |value| {
             Box::pin(async move {
-                let size = match postcard::experimental::serialized_size(&value) {
-                    Ok(size) => size,
-                    Err(e) => {
-                        writer.reset(ErrorCode::MessageEncodeFailed.into()).ok();
-                        return Err(e!(SendError::EncodeFailed, e));
-                    }
-                };
-                if size as u64 > MAX_MESSAGE_SIZE {
-                    writer.reset(ErrorCode::MessageMaxSizeExceeded.into()).ok();
-                    return Err(e!(SendError::MaxSizeExceeded));
-                }
                 // write via a small buffer to avoid allocation for small values
-                let mut buf = SmallVec::<[u8; 128]>::new();
-                if let Err(e) = buf.write_length_prefixed(value) {
-                    writer.reset(ErrorCode::MessageEncodeFailed.into()).ok();
-                    return Err(e.into());
-                }
+                let mut buf = SmallVec::new();
+                encode_message(&value, &mut buf, &mut writer)?;
                 writer.write_all(&buf).await?;
                 Ok(())
             })
@@ -701,38 +728,12 @@ struct NoqSenderInner<T> {
 }
 
 impl<T: RpcMessage> NoqSenderInner<T> {
-    /// Encodes `value` into the buffer.
-    ///
-    /// Resets the stream if `value` is too large or does not encode, so the
-    /// receiver gets an error instead of the end of the stream.
-    fn encode(&mut self, value: T) -> Result<(), SendError> {
-        let size = match postcard::experimental::serialized_size(&value) {
-            Ok(size) => size,
-            Err(e) => {
-                self.send.reset(ErrorCode::MessageEncodeFailed.into()).ok();
-                return Err(e!(SendError::EncodeFailed, e));
-            }
-        };
-        if size as u64 > MAX_MESSAGE_SIZE {
-            self.send
-                .reset(ErrorCode::MessageMaxSizeExceeded.into())
-                .ok();
-            return Err(e!(SendError::MaxSizeExceeded));
-        }
-        self.buffer.clear();
-        if let Err(e) = self.buffer.write_length_prefixed(value) {
-            self.send.reset(ErrorCode::MessageEncodeFailed.into()).ok();
-            return Err(e.into());
-        }
-        Ok(())
-    }
-
     fn send(
         &mut self,
         value: T,
     ) -> Pin<Box<dyn Future<Output = Result<(), SendError>> + Send + Sync + '_>> {
-        Box::pin(async {
-            self.encode(value)?;
+        Box::pin(async move {
+            encode_message(&value, &mut self.buffer, &mut self.send)?;
             self.send.write_all(&self.buffer).await?;
             self.buffer.clear();
             Ok(())
@@ -743,9 +744,9 @@ impl<T: RpcMessage> NoqSenderInner<T> {
         &mut self,
         value: T,
     ) -> Pin<Box<dyn Future<Output = Result<bool, SendError>> + Send + Sync + '_>> {
-        Box::pin(async {
+        Box::pin(async move {
             // todo: move the non-async part out of the box. Will require a new return type.
-            self.encode(value)?;
+            encode_message(&value, &mut self.buffer, &mut self.send)?;
             let Some(n) = now_or_never(self.send.write(&self.buffer)) else {
                 return Ok(false);
             };
diff --git a/src/util.rs b/src/util.rs
index 053f96d0..4763861a 100644
--- a/src/util.rs
+++ b/src/util.rs
@@ -164,7 +164,6 @@ mod varint_util {
     /// helper to write a length-prefixed value.
     pub trait WriteVarintExt: std::io::Write {
         /// Write a varint
-        #[allow(dead_code)]
         fn write_varint_u64(&mut self, value: u64) -> io::Result<usize>;
         /// Write a value with a varint encoded length prefix.
         fn write_length_prefixed<T: Serialize>(&mut self, value: T) -> io::Result<()>;

From 96f5d23d17267b3e771fc4b23378ece31dcc7db8 Mon Sep 17 00:00:00 2001
From: Frando <frando@unbiskant.org>
Date: Mon, 12 Oct 2026 13:35:08 +0200
Subject: [PATCH 4/4] feat: return the code of a connection close from
 remote_code

When a default `Handler` rejects a request, it resets the streams of the
request and then closes the connection with the same code. The close
arrives first, so the client got an `Io` error, and `remote_code`
returned `None`. `remote_code` on the channel errors and on `WriteError`
now also returns the irpc code that the peer closed the connection with,
and the new `Error::remote_code` returns the code from any of the errors
in `Error`.

The code mapping impls of the channel errors now live in the channel
modules. `From<io::Error>` for the receive errors gives a `Remote` error
for a reset with an irpc code, so the receivers use `?`. This also
tightens the docs of the error codes and errors, and shortens the tests.
---
 src/channel.rs           |  54 ++++++++++++-
 src/channel/mpsc.rs      |  47 +++++++++---
 src/channel/oneshot.rs   |  42 ++++++++--
 src/lib.rs               |  21 +++++
 src/rpc.rs               | 161 ++++++++++++++-------------------------
 tests/invalid_request.rs |  86 ++++++---------------
 tests/mpsc_channel.rs    |  19 ++---
 tests/oneshot_channel.rs |   1 -
 8 files changed, 231 insertions(+), 200 deletions(-)

diff --git a/src/channel.rs b/src/channel.rs
index d66921d8..5dac6087 100644
--- a/src/channel.rs
+++ b/src/channel.rs
@@ -2,8 +2,13 @@
 
 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;
@@ -25,6 +30,10 @@ pub mod oneshot;
 ///     decode an earlier message.
 ///   * The connection failed ([`Io`]).
 ///
+/// 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.
@@ -45,9 +54,10 @@ pub enum SendError {
     ReceiverClosed,
     /// The receiver stopped the stream with an irpc error code, and the channel is closed.
     ///
-    /// The code says why, see [`ErrorCode`].
+    /// 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 },
@@ -85,3 +95,45 @@ impl From<SendError> for io::Error {
         }
     }
 }
+
+#[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()),
+        }
+    }
+}
diff --git a/src/channel/mpsc.rs b/src/channel/mpsc.rs
index 6bf87e93..035cf600 100644
--- a/src/channel/mpsc.rs
+++ b/src/channel/mpsc.rs
@@ -7,31 +7,31 @@ 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 an mpsc message.
 ///
-/// A local channel has no errors: `recv` returns `Ok(None)` when all senders
-/// are gone. A remote channel does the same when its sender closes the stream.
+/// 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, and the client for the updates of a request.
+/// 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
-/// can arrive first, and the receiver gets [`Io`] instead of [`Remote`].
+/// 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, from_sources)]
+#[stack_error(derive, add_meta)]
 #[non_exhaustive]
 pub enum RecvError {
     /// The sender reset the stream with an irpc error code, and the channel is closed.
     ///
-    /// The code says why, see [`ErrorCode`]. For example, the server rejected
-    /// or skipped the request, see [Bad requests] and [Abandoned requests].
+    /// The `code` says why, see [`ErrorCode`], and the [`rpc`] module for an overview.
     ///
     /// [`ErrorCode`]: crate::rpc::ErrorCode
-    /// [Bad requests]: crate::rpc#bad-requests
-    /// [Abandoned requests]: crate::rpc#abandoned-requests
+    /// [`rpc`]: crate::rpc
     #[cfg(feature = "rpc")]
     #[error("Sender closed the channel: {code}")]
     Remote { code: crate::rpc::ErrorCode },
@@ -57,6 +57,20 @@ pub enum RecvError {
     },
 }
 
+/// 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 {
@@ -68,6 +82,21 @@ impl From<RecvError> for io::Error {
     }
 }
 
+#[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,
+        }
+    }
+}
+
 /// Create a local mpsc sender and receiver pair, with the given buffer size.
 ///
 /// This is currently using a tokio channel pair internally.
diff --git a/src/channel/oneshot.rs b/src/channel/oneshot.rs
index 1dc8f7ab..665213df 100644
--- a/src/channel/oneshot.rs
+++ b/src/channel/oneshot.rs
@@ -6,6 +6,8 @@ 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 message.
@@ -15,12 +17,13 @@ use crate::util::FusedOneshotReceiver;
 /// the server for a response.
 ///
 /// If the server rejects the request and also closes the connection, the close
-/// can arrive first, and the receiver gets [`Io`] instead of [`Remote`].
+/// 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, from_sources)]
+#[stack_error(derive, add_meta)]
 #[non_exhaustive]
 pub enum RecvError {
     /// The sender has been closed.
@@ -32,12 +35,10 @@ pub enum RecvError {
     SenderClosed,
     /// The sender reset the stream with an irpc error code, and the channel is closed.
     ///
-    /// The code says why, see [`ErrorCode`]. For example, the server rejected
-    /// or skipped the request, see [Bad requests] and [Abandoned requests].
+    /// The `code` says why, see [`ErrorCode`], and the [`rpc`] module for an overview.
     ///
     /// [`ErrorCode`]: crate::rpc::ErrorCode
-    /// [Bad requests]: crate::rpc#bad-requests
-    /// [Abandoned requests]: crate::rpc#abandoned-requests
+    /// [`rpc`]: crate::rpc
     #[cfg(feature = "rpc")]
     #[error("Sender closed the channel: {code}")]
     Remote { code: crate::rpc::ErrorCode },
@@ -63,6 +64,20 @@ pub enum RecvError {
     },
 }
 
+/// 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 {
@@ -75,6 +90,21 @@ impl From<RecvError> for io::Error {
     }
 }
 
+#[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,
+        }
+    }
+}
+
 /// Create a local oneshot sender and receiver pair.
 ///
 /// This is currently using a tokio channel pair internally.
diff --git a/src/lib.rs b/src/lib.rs
index bcdae49d..9c09b5a7 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -959,6 +959,27 @@ pub enum Error {
     Write { source: rpc::WriteError },
 }
 
+#[cfg(feature = "rpc")]
+impl Error {
+    /// Returns the irpc error code that the peer sent, if any.
+    ///
+    /// This is the code that the peer reset or stopped a stream with, or closed
+    /// the connection with, see [`rpc::ErrorCode`].
+    pub fn remote_code(&self) -> Option<rpc::ErrorCode> {
+        match self {
+            Self::Request {
+                source: RequestError::Connection { source, .. },
+                ..
+            } => rpc::close_code(source),
+            Self::Request { .. } => None,
+            Self::Send { source, .. } => source.remote_code(),
+            Self::MpscRecv { source, .. } => source.remote_code(),
+            Self::OneshotRecv { source, .. } => source.remote_code(),
+            Self::Write { source, .. } => source.remote_code(),
+        }
+    }
+}
+
 /// Type alias for a result with an irpc error type.
 pub type Result<T, E = Error> = std::result::Result<T, E>;
 
diff --git a/src/rpc.rs b/src/rpc.rs
index 01def4b0..85c402e1 100644
--- a/src/rpc.rs
+++ b/src/rpc.rs
@@ -9,9 +9,8 @@
 //! Each request has its own pair of streams. The client writes the request as
 //! the first message on its stream. After the request, the streams carry the
 //! channel messages of the request: the response, and for some requests,
-//! updates from the client. The name of each error code says what the code is
-//! about: a `Request` code is about the request, and a `Message` code is about
-//! a channel message, see [`ErrorCode`].
+//! updates from the client. A `Request` error code is about the request, and a
+//! `Message` error code is about a channel message, see [`ErrorCode`].
 //!
 //! # Bad requests
 //!
@@ -29,9 +28,12 @@
 //! serving the old ones on the same connection.
 //!
 //! The client gets the code in a `Remote` error from the receiver of the
-//! response, for example [`oneshot::RecvError::Remote`] from [`Client::rpc`].
-//! If the server closes the connection, the close can arrive first, and the
-//! client gets the connection error.
+//! response, for example [`oneshot::RecvError::Remote`] from [`Client::rpc`],
+//! and from the sender of its updates, [`SendError::Remote`].
+//! When the server also closes the connection, as a [`Handler`] does by
+//! default, the close usually arrives first, and the client gets an `Io` error
+//! with the connection error instead. In both cases, `remote_code` returns the
+//! code, for example [`crate::Error::remote_code`].
 //!
 //! [`read_request`] returns a bad request as a [`ReadRequestError`]. A server
 //! with its own loop decides what to do: close the connection, or read the
@@ -43,10 +45,12 @@
 //! with [`ErrorCode::RequestAborted`]. For example, it dropped the future of
 //! [`Client::rpc`] while it wrote a large request. A client that cannot send a
 //! request resets the stream with [`ErrorCode::RequestMaxSizeExceeded`] or
-//! [`ErrorCode::RequestEncodeFailed`]. The server skips a request whose stream
-//! was reset. It resets its side of the stream with
-//! [`ErrorCode::RequestAborted`], so the receiver of the response gets an
-//! error, and reads the next request.
+//! [`ErrorCode::RequestEncodeFailed`].
+//!
+//! When the client resets the stream of a request before the server read the
+//! request, the server skips the request and reads the next one. It resets its
+//! side of the stream with [`ErrorCode::RequestAborted`], so that the receiver
+//! of the response gets an error instead of the end of the stream.
 //!
 //! # Channel messages
 //!
@@ -58,7 +62,8 @@
 //! if a message does not decode. The channel then ends. The side that found the
 //! error gets its own variant, for example [`mpsc::RecvError::DecodeFailed`],
 //! and the other side gets the code in a `Remote` error, for example
-//! [`SendError::Remote`].
+//! [`SendError::Remote`]. A reset can drop messages that the receiver did not
+//! read yet.
 //!
 //! # Error codes
 //!
@@ -206,11 +211,11 @@ impl From<ErrorCode> for VarInt {
 
 /// Error when reading a request with [`read_request`].
 ///
-/// For [`MaxSizeExceeded`], [`Malformed`], and
-/// [`DecodeFailed`], irpc stops and resets the streams of the request
-/// with [`ErrorCode::RequestMaxSizeExceeded`] or
-/// [`ErrorCode::RequestDecodeFailed`], so its client gets an error. The connection is still
-/// usable, so a server can read the next request.
+/// For [`MaxSizeExceeded`], [`Malformed`], and [`DecodeFailed`], irpc stops
+/// and resets the streams of the request with
+/// [`ErrorCode::RequestMaxSizeExceeded`] or [`ErrorCode::RequestDecodeFailed`],
+/// so its client gets an error. The connection is still usable, so a server
+/// can read the next request.
 ///
 /// [`MaxSizeExceeded`]: Self::MaxSizeExceeded
 /// [`Malformed`]: Self::Malformed
@@ -268,15 +273,14 @@ impl From<ReadRequestError> for io::Error {
     }
 }
 
-/// Error that can occur when writing the initial message when doing a
-/// cross-process RPC.
+/// Error when a client writes a request.
 #[stack_error(derive, add_meta, from_sources)]
 #[non_exhaustive]
 pub enum WriteError {
     /// The request is larger than [`MAX_MESSAGE_SIZE`], so this client did not send it.
     #[error("Request too large")]
     MaxSizeExceeded,
-    /// The request does not encode.
+    /// The request does not encode, so this client did not send it.
     #[error("Request does not encode")]
     EncodeFailed {
         #[error(std_err)]
@@ -284,9 +288,7 @@ pub enum WriteError {
     },
     /// The server stopped the stream with an irpc error code while this client wrote the request.
     ///
-    /// For example, the server rejected the request, see [Bad requests].
-    ///
-    /// [Bad requests]: self#bad-requests
+    /// The `code` says why, see [`ErrorCode`], and the [`rpc`](self) module for an overview.
     #[error("Server rejected the request: {code}")]
     Remote { code: ErrorCode },
     /// An io error, for example because the connection was lost.
@@ -308,8 +310,29 @@ impl WriteError {
     }
 }
 
+/// Returns the irpc error code that the peer closed the connection with, if any, other than 0.
+pub(crate) fn close_code(err: &ConnectionError) -> Option<ErrorCode> {
+    match err {
+        ConnectionError::ApplicationClosed(close) => ErrorCode::from_varint(close.error_code),
+        _ => None,
+    }
+    .filter(|code| *code != ErrorCode::Closed)
+}
+
+/// Returns the irpc error code that the peer closed the connection with, if `err` comes from that close.
+pub(crate) fn io_close_code(err: &io::Error) -> Option<ErrorCode> {
+    let err = err.get_ref()?;
+    if let Some(noq::ReadError::ConnectionLost(err)) = err.downcast_ref() {
+        close_code(err)
+    } else if let Some(noq::WriteError::ConnectionLost(err)) = err.downcast_ref() {
+        close_code(err)
+    } else {
+        None
+    }
+}
+
 /// Returns the irpc error code that the receiver stopped the stream with, if any.
-fn stop_code(err: &noq::WriteError) -> Option<ErrorCode> {
+pub(crate) fn stop_code(err: &noq::WriteError) -> Option<ErrorCode> {
     match err {
         noq::WriteError::Stopped(code) => ErrorCode::from_varint(*code),
         _ => None,
@@ -335,70 +358,18 @@ impl From<WriteError> for io::Error {
     }
 }
 
-impl From<noq::WriteError> for SendError {
-    fn from(err: noq::WriteError) -> Self {
-        match stop_code(&err) {
-            Some(code) => SendError::from_error_code(code),
-            None => e!(SendError::Io, err.into()),
-        }
-    }
-}
-
 impl WriteError {
-    /// Returns the code of a [`Remote`](Self::Remote) error.
-    pub fn remote_code(&self) -> Option<ErrorCode> {
-        match self {
-            Self::Remote { code, .. } => Some(*code),
-            _ => None,
-        }
-    }
-}
-
-impl oneshot::RecvError {
-    /// Returns the code of a [`Remote`](Self::Remote) error.
-    pub fn remote_code(&self) -> Option<ErrorCode> {
-        match self {
-            Self::Remote { code, .. } => Some(*code),
-            _ => None,
-        }
-    }
-}
-
-impl mpsc::RecvError {
-    /// Returns the code of a [`Remote`](Self::Remote) error.
-    pub fn remote_code(&self) -> Option<ErrorCode> {
-        match self {
-            Self::Remote { code, .. } => Some(*code),
-            _ => None,
-        }
-    }
-}
-
-impl SendError {
-    /// Returns the code of a [`Remote`](Self::Remote) error.
+    /// 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, .. } => io_close_code(source),
             _ => None,
         }
     }
-
-    /// Returns the error for a receiver that stopped the stream with `code`.
-    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.
-    fn closed(code: Option<ErrorCode>) -> Self {
-        code.map(Self::from_error_code)
-            .unwrap_or_else(|| io::Error::from(io::ErrorKind::BrokenPipe).into())
-    }
 }
 
 /// Trait to abstract over a client connection to a remote service.
@@ -578,7 +549,7 @@ impl<T: DeserializeOwned> From<noq::RecvStream> for oneshot::Receiver<T> {
     fn from(mut read: noq::RecvStream) -> Self {
         let fut = async move {
             // A sender that is dropped without sending finishes the stream.
-            let Some(size) = read.read_varint_u64().await.map_err(oneshot_recv_error)? else {
+            let Some(size) = read.read_varint_u64().await? else {
                 return Err(e!(oneshot::RecvError::SenderClosed));
             };
             if size > MAX_MESSAGE_SIZE {
@@ -586,9 +557,7 @@ impl<T: DeserializeOwned> From<noq::RecvStream> for oneshot::Receiver<T> {
                 return Err(e!(oneshot::RecvError::MaxSizeExceeded));
             }
             let mut buf = vec![0; size as usize];
-            AsyncReadExt::read_exact(&mut read, &mut buf)
-                .await
-                .map_err(oneshot_recv_error)?;
+            AsyncReadExt::read_exact(&mut read, &mut buf).await?;
             let msg: T = postcard::from_bytes(&buf).map_err(|err| {
                 read.stop(ErrorCode::MessageDecodeFailed.into()).ok();
                 e!(oneshot::RecvError::DecodeFailed, err)
@@ -605,7 +574,7 @@ fn read_error(err: &io::Error) -> Option<&noq::ReadError> {
 }
 
 /// Returns the irpc error code that the sender reset the stream with, if any, other than 0.
-fn reset_code(err: &io::Error) -> Option<ErrorCode> {
+pub(crate) fn reset_code(err: &io::Error) -> Option<ErrorCode> {
     match read_error(err)? {
         noq::ReadError::Reset(code) => ErrorCode::from_varint(*code),
         _ => None,
@@ -613,22 +582,6 @@ fn reset_code(err: &io::Error) -> Option<ErrorCode> {
     .filter(|code| *code != ErrorCode::Closed)
 }
 
-/// Returns the error for a oneshot stream that failed.
-fn oneshot_recv_error(err: io::Error) -> oneshot::RecvError {
-    match reset_code(&err) {
-        Some(code) => e!(oneshot::RecvError::Remote { code }),
-        None => err.into(),
-    }
-}
-
-/// Returns the error for an mpsc stream that failed.
-fn mpsc_recv_error(err: io::Error) -> mpsc::RecvError {
-    match reset_code(&err) {
-        Some(code) => e!(mpsc::RecvError::Remote { code }),
-        None => err.into(),
-    }
-}
-
 impl From<noq::RecvStream> for crate::channel::none::NoReceiver {
     fn from(read: noq::RecvStream) -> Self {
         drop(read);
@@ -695,7 +648,7 @@ impl<T: RpcMessage> DynReceiver<T> for NoqReceiver<T> {
     ) -> Pin<Box<dyn Future<Output = Result<Option<T>, mpsc::RecvError>> + Send + Sync + '_>> {
         Box::pin(async {
             let read = &mut self.recv;
-            let Some(size) = read.read_varint_u64().await.map_err(mpsc_recv_error)? else {
+            let Some(size) = read.read_varint_u64().await? else {
                 return Ok(None);
             };
             if size > MAX_MESSAGE_SIZE {
@@ -705,9 +658,7 @@ impl<T: RpcMessage> DynReceiver<T> for NoqReceiver<T> {
                 return Err(e!(mpsc::RecvError::MaxSizeExceeded));
             }
             let mut buf = vec![0; size as usize];
-            AsyncReadExt::read_exact(read, &mut buf)
-                .await
-                .map_err(mpsc_recv_error)?;
+            AsyncReadExt::read_exact(read, &mut buf).await?;
             let msg: T = postcard::from_bytes(&buf).map_err(|err| {
                 read.stop(ErrorCode::MessageDecodeFailed.into()).ok();
                 e!(mpsc::RecvError::DecodeFailed, err)
diff --git a/tests/invalid_request.rs b/tests/invalid_request.rs
index 4c133cec..4a63645c 100644
--- a/tests/invalid_request.rs
+++ b/tests/invalid_request.rs
@@ -5,7 +5,6 @@ use std::{pin::pin, time::Duration};
 
 use irpc::{
     Client, WithChannels,
-    channel::oneshot,
     noq::listen,
     rpc::{ErrorCode, Handler, MAX_MESSAGE_SIZE, ReadRequestError, WriteError, read_request},
 };
@@ -36,9 +35,6 @@ mod server {
         #[rpc(tx = oneshot::Sender<()>, rx = mpsc::Receiver<NoSer>)]
         #[wrap(Upload)]
         Upload(String),
-        #[rpc(tx = mpsc::Sender<()>, rx = mpsc::Receiver<NoSer>)]
-        #[wrap(Exchange)]
-        Exchange(String),
     }
 }
 
@@ -90,10 +86,12 @@ async fn bad_request_closes_connection() -> TestResult<()> {
     let conn = client_endpoint.connect(server_addr, "localhost")?.await?;
     let client = Client::<client::EchoProtocol>::boxed(conn.clone());
 
-    client
+    let err = client
         .rpc(client::Shout("a".into()))
         .await
         .expect_err("server does not know the request");
+    // The close usually arrives before the reset of the stream.
+    assert_eq!(err.remote_code(), Some(ErrorCode::RequestDecodeFailed));
     let ConnectionError::ApplicationClosed(close) = conn.closed().await else {
         panic!("server closes the connection");
     };
@@ -131,19 +129,7 @@ async fn skip_bad_requests_keeps_connection() -> TestResult<()> {
         .rpc(client::Shout("a".into()))
         .await
         .expect_err("server does not know the request");
-    assert!(
-        matches!(
-            err,
-            irpc::Error::OneshotRecv {
-                source: oneshot::RecvError::Remote {
-                    code: ErrorCode::RequestDecodeFailed,
-                    ..
-                },
-                ..
-            }
-        ),
-        "{err:?}"
-    );
+    assert_eq!(err.remote_code(), Some(ErrorCode::RequestDecodeFailed));
     assert_eq!(client.rpc(client::Echo("b".into())).await?, "b");
     Ok(())
 }
@@ -241,20 +227,20 @@ async fn truncated_request_closes_connection() -> TestResult<()> {
     Ok(())
 }
 
-/// The update sender resets the stream of its request before the server reads it.
+/// The update senders reset the streams of their requests before the server reads them.
 #[tokio::test]
 async fn reset_before_read_skips_request() -> TestResult<()> {
     let (server, client_endpoint, server_addr) = create_connected_endpoints()?;
     let server = tokio::spawn(async move {
         let conn = server.accept().await.expect("endpoint is open").await?;
-        // Read only after the reset arrived: noq then has dropped the request.
-        while conn.stats().frame_rx.reset_stream == 0 {
+        // Read only after both resets arrived: noq then has dropped both requests.
+        while conn.stats().frame_rx.reset_stream < 2 {
             tokio::time::sleep(Duration::from_millis(1)).await;
         }
         let Some(server::EchoMessage::Echo(msg)) =
             read_request::<server::EchoProtocol>(&conn).await?
         else {
-            panic!("server skips the upload");
+            panic!("server skips the upload and the exchange");
         };
         msg.tx.send(msg.inner.0).await?;
         conn.closed().await;
@@ -263,54 +249,26 @@ async fn reset_before_read_skips_request() -> TestResult<()> {
     let conn = client_endpoint.connect(server_addr, "localhost")?.await?;
     let client = Client::<client::EchoProtocol>::boxed(conn.clone());
 
-    let (tx, rx) = client
+    let (upload_tx, upload_rx) = client
         .client_streaming(client::Upload("a".into()), 1)
         .await?;
-    // The update does not encode, so the sender resets with `ErrorCode::MessageEncodeFailed`.
-    tx.send(NoSer(1))
-        .await
-        .expect_err("odd numbers do not encode");
-    assert_eq!(client.rpc(client::Echo("b".into())).await?, "b");
-    // The server skipped the upload and reset its side of the stream.
-    let err = rx.await.expect_err("server skips the upload");
-    assert_eq!(err.remote_code(), Some(ErrorCode::RequestAborted));
-    conn.close(0u32.into(), b"");
-    server.await??;
-    Ok(())
-}
-
-/// Like [`reset_before_read_skips_request`], but the response is an mpsc channel.
-#[tokio::test]
-async fn reset_before_read_fails_mpsc_response() -> TestResult<()> {
-    let (server, client_endpoint, server_addr) = create_connected_endpoints()?;
-    let server = tokio::spawn(async move {
-        let conn = server.accept().await.expect("endpoint is open").await?;
-        // Read only after the reset arrived: noq then has dropped the request.
-        while conn.stats().frame_rx.reset_stream == 0 {
-            tokio::time::sleep(Duration::from_millis(1)).await;
-        }
-        let Some(server::EchoMessage::Echo(msg)) =
-            read_request::<server::EchoProtocol>(&conn).await?
-        else {
-            panic!("server skips the exchange");
-        };
-        msg.tx.send(msg.inner.0).await?;
-        conn.closed().await;
-        TestResult::Ok(())
-    });
-    let conn = client_endpoint.connect(server_addr, "localhost")?.await?;
-    let client = Client::<client::EchoProtocol>::boxed(conn.clone());
-
-    let (tx, mut rx) = client
+    let (exchange_tx, mut exchange_rx) = client
         .bidi_streaming(client::Exchange("a".into()), 1, 1)
         .await?;
-    // The update does not encode, so the sender resets with `ErrorCode::MessageEncodeFailed`.
-    tx.send(NoSer(1))
+    // The updates do not encode, so the senders reset with `ErrorCode::MessageEncodeFailed`.
+    upload_tx.send(NoSer(1)).await.expect_err("does not encode");
+    exchange_tx
+        .send(NoSer(1))
         .await
-        .expect_err("odd numbers do not encode");
+        .expect_err("does not encode");
     assert_eq!(client.rpc(client::Echo("b".into())).await?, "b");
-    // The server skipped the exchange, so the response fails instead of ending.
-    let err = rx.recv().await.expect_err("server skips the exchange");
+    // The server reset its side of both streams, so the mpsc response fails instead of ending.
+    let err = upload_rx.await.expect_err("server skips the upload");
+    assert_eq!(err.remote_code(), Some(ErrorCode::RequestAborted));
+    let err = exchange_rx
+        .recv()
+        .await
+        .expect_err("server skips the exchange");
     assert_eq!(err.remote_code(), Some(ErrorCode::RequestAborted));
     conn.close(0u32.into(), b"");
     server.await??;
diff --git a/tests/mpsc_channel.rs b/tests/mpsc_channel.rs
index 508dc5d5..63e47603 100644
--- a/tests/mpsc_channel.rs
+++ b/tests/mpsc_channel.rs
@@ -427,22 +427,13 @@ async fn mpsc_decode_error_reaches_sender() -> TestResult<()> {
         }
     })
     .await?;
-    assert_eq!(err.remote_code(), Some(ErrorCode::MessageDecodeFailed));
+    let decode_failed = Some(ErrorCode::MessageDecodeFailed);
+    assert_eq!(err.remote_code(), decode_failed);
     // later sends, also on a clone that did not see the stop
+    assert_eq!(send.send(2).await.unwrap_err().remote_code(), decode_failed);
     assert_eq!(
-        send.send(2)
-            .await
-            .expect_err("the sender is closed")
-            .remote_code(),
-        Some(ErrorCode::MessageDecodeFailed)
-    );
-    assert_eq!(
-        clone
-            .try_send(2)
-            .await
-            .expect_err("the sender is closed")
-            .remote_code(),
-        Some(ErrorCode::MessageDecodeFailed)
+        clone.try_send(2).await.unwrap_err().remote_code(),
+        decode_failed
     );
     conn.close(0u32.into(), b"");
     server.await??;
diff --git a/tests/oneshot_channel.rs b/tests/oneshot_channel.rs
index 720c55d4..e5456160 100644
--- a/tests/oneshot_channel.rs
+++ b/tests/oneshot_channel.rs
@@ -140,7 +140,6 @@ async fn oneshot_serialize_error_recv() -> TestResult<()> {
     let Err(cause) = server.await? else {
         panic!("server should have failed due to serialization error");
     };
-    println!("Server error: {cause:?}");
     assert!(matches!(cause, RecvError::DecodeFailed { .. }), "{cause:?}");
     Ok(())
 }