From 49717bf07f2aa1066b446796fceb65f2463ebb2b Mon Sep 17 00:00:00 2001 From: Frando Date: Fri, 9 Oct 2026 10:48:30 +0200 Subject: [PATCH 1/4] fix: try_send resets the stream for a message it cannot send `send` resets the stream if a message is too large or does not encode, but `try_send` only returned the error. The sender then closed, and the drop of the stream finished it, so the receiver saw a clean end instead of an error. Both now encode through one helper that resets the stream. --- src/rpc.rs | 59 +++++++++++++++++++++++-------------------- tests/mpsc_channel.rs | 21 +++++++++++++++ 2 files changed, 52 insertions(+), 28 deletions(-) diff --git a/src/rpc.rs b/src/rpc.rs index 89547467..137f29d5 100644 --- a/src/rpc.rs +++ b/src/rpc.rs @@ -465,33 +465,41 @@ struct NoqSenderInner { } impl NoqSenderInner { + /// 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(ERROR_CODE_ENCODE_FAILED.into()).ok(); + return Err(e!( + SendError::Io, + io::Error::new(io::ErrorKind::InvalidData, e) + )); + } + }; + if size as u64 > MAX_MESSAGE_SIZE { + self.send + .reset(ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED.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(); + return Err(e.into()); + } + Ok(()) + } + fn send( &mut self, value: T, ) -> Pin> + Send + Sync + '_>> { Box::pin(async { - let size = match postcard::experimental::serialized_size(&value) { - Ok(size) => size, - Err(e) => { - self.send.reset(ERROR_CODE_ENCODE_FAILED.into()).ok(); - return Err(e!( - SendError::Io, - io::Error::new(io::ErrorKind::InvalidData, e) - )); - } - }; - if size as u64 > MAX_MESSAGE_SIZE { - self.send - .reset(ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED.into()) - .ok(); - return Err(e!(SendError::MaxMessageSizeExceeded)); - } - let value = value; - self.buffer.clear(); - if let Err(e) = self.buffer.write_length_prefixed(value) { - self.send.reset(ERROR_CODE_ENCODE_FAILED.into()).ok(); - return Err(e.into()); - } + self.encode(value)?; self.send.write_all(&self.buffer).await?; self.buffer.clear(); Ok(()) @@ -503,13 +511,8 @@ impl NoqSenderInner { value: T, ) -> Pin> + Send + Sync + '_>> { Box::pin(async { - if postcard::experimental::serialized_size(&value)? as u64 > MAX_MESSAGE_SIZE { - return Err(e!(SendError::MaxMessageSizeExceeded)); - } // todo: move the non-async part out of the box. Will require a new return type. - let value = value; - self.buffer.clear(); - self.buffer.write_length_prefixed(value)?; + self.encode(value)?; let Some(n) = now_or_never(self.send.write(&self.buffer)) else { return Ok(false); }; diff --git a/tests/mpsc_channel.rs b/tests/mpsc_channel.rs index e6fe4b5b..d0d4f567 100644 --- a/tests/mpsc_channel.rs +++ b/tests/mpsc_channel.rs @@ -298,6 +298,27 @@ async fn mpsc_max_message_size_send() -> TestResult<()> { Ok(()) } +/// Checks that `try_send` resets the stream for a message that is too large, as `send` does. +#[tokio::test] +async fn mpsc_max_message_size_try_send() -> TestResult<()> { + let (server, client, server_addr) = create_connected_endpoints()?; + let server = tokio::spawn(vec_receiver(server)); + let conn = client.connect(server_addr, "localhost")?.await?; + let (send, _) = conn.open_bi().await?; + let send = mpsc::Sender::>::from(send); + 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 { .. })); + 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) + ); + Ok(()) +} + /// Checks that the max message size is enforced on receiver side. #[tokio::test] async fn mpsc_max_message_size_recv() -> TestResult<()> { From 92d262b33399f8d89a2802b1b9e2cf0141c2eca3 Mon Sep 17 00:00:00 2001 From: Frando Date: Fri, 9 Oct 2026 10:49:04 +0200 Subject: [PATCH 2/4] refactor: name how a send leaves the sender closed `mem::take` with a `Default` impl hid the state that a cancelled send leaves behind. `take_for_send` names it and says why the sender stays closed. --- src/rpc.rs | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/src/rpc.rs b/src/rpc.rs index 137f29d5..e6d05098 100644 --- a/src/rpc.rs +++ b/src/rpc.rs @@ -550,13 +550,15 @@ impl CloseReason { } } -impl Default for NoqSenderState { - fn default() -> Self { - Self::Closed(CloseReason::Other) +impl NoqSenderState { + /// Takes the state for a send, and leaves the sender closed until the send puts it back. + /// + /// 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)) } -} -impl NoqSenderState { /// Returns the state after a send, so that later sends fail with the same kind of error. fn after_send(sender: NoqSenderInner, res: &Result) -> Self { match res { @@ -579,7 +581,7 @@ impl DynSender for NoqSender { fn send(&self, value: T) -> Pin> + Send + '_>> { Box::pin(async { let mut guard = self.0.lock().await; - let sender = std::mem::take(guard.deref_mut()); + let sender = guard.take_for_send(); match sender { NoqSenderState::Open(mut sender) => { let res = sender.send(value).await; @@ -600,7 +602,7 @@ impl DynSender for NoqSender { ) -> Pin> + Send + '_>> { Box::pin(async { let mut guard = self.0.lock().await; - let sender = std::mem::take(guard.deref_mut()); + let sender = guard.take_for_send(); match sender { NoqSenderState::Open(mut sender) => { let res = sender.try_send(value).await; From 6409ebffec205107d299bc1c791683ac0b96207d Mon Sep 17 00:00:00 2001 From: Frando Date: Fri, 9 Oct 2026 10:49:16 +0200 Subject: [PATCH 3/4] docs: give ReceiverClosed and SenderClosed a one-line summary --- src/channel.rs | 8 ++++---- src/channel/oneshot.rs | 8 ++++---- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/src/channel.rs b/src/channel.rs index be8d048d..de8f4899 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -16,11 +16,11 @@ pub mod oneshot; #[stack_error(derive, add_meta, from_sources)] #[non_exhaustive] pub enum SendError { - /// The receiver has been closed. This is the only error that can occur - /// for local communication. + /// The receiver has been closed. /// - /// For remote communication, this means that the receiver stopped the - /// stream with code 0, for example because it was dropped. + /// This is the only error that can occur for local communication. For + /// remote communication, this means that the receiver stopped the stream + /// 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`]). diff --git a/src/channel/oneshot.rs b/src/channel/oneshot.rs index 0c02c2db..7543da1f 100644 --- a/src/channel/oneshot.rs +++ b/src/channel/oneshot.rs @@ -16,11 +16,11 @@ use crate::util::FusedOneshotReceiver; #[stack_error(derive, add_meta, from_sources)] #[non_exhaustive] pub enum RecvError { - /// The sender has been closed. This is the only error that can occur - /// for local communication. + /// The sender has been closed. /// - /// For remote communication, this means that the sender finished the - /// stream without sending, for example because it was dropped. + /// This is the only error that can occur for local communication. For + /// remote communication, this means that the sender finished the stream + /// without sending, for example because it was dropped. #[error("Sender closed")] SenderClosed, /// The message exceeded the maximum allowed message size (see [`MAX_MESSAGE_SIZE`]). From 68cbe94350eed56540e2d5b693303bc1eec0e2a4 Mon Sep 17 00:00:00 2001 From: Frando Date: Fri, 9 Oct 2026 13:00:16 +0200 Subject: [PATCH 4/4] docs: use reference links in the error handling docs Inline links with a target that differs from their text make the doc comments harder to read in the source. They now point to reference definitions at the end of each doc comment. --- src/rpc.rs | 36 +++++++++++++++++++++++++----------- 1 file changed, 25 insertions(+), 11 deletions(-) diff --git a/src/rpc.rs b/src/rpc.rs index e6d05098..99ad7ae4 100644 --- a/src/rpc.rs +++ b/src/rpc.rs @@ -28,7 +28,7 @@ //! //! 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 -//! [`Client::rpc`](crate::Client::rpc) while it wrote a large request. The +//! [`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. //! @@ -47,6 +47,8 @@ //! 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 //! 16 works, but a later version of irpc can give it a different meaning. +//! +//! [`Client::rpc`]: crate::Client::rpc use std::{ fmt::Debug, future::Future, io, marker::PhantomData, ops::DerefMut, pin::Pin, sync::Arc, }; @@ -78,7 +80,9 @@ pub const MAX_MESSAGE_SIZE: u64 = 1024 * 1024 * 16; /// Error code on streams and connections if a message is larger than [`MAX_MESSAGE_SIZE`]. /// -/// See [Error codes](self#error-codes) for the codes that irpc reserves. +/// See [Error codes] for the codes that irpc reserves. +/// +/// [Error codes]: self#error-codes pub const ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED: u32 = 1; /// Error code on streams if the sender could not encode a message. @@ -93,16 +97,20 @@ pub const ERROR_CODE_DECODE_FAILED: u32 = 3; /// 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](self#bad-requests). +/// The server skips such a request, see [Bad requests]. +/// +/// [Bad requests]: self#bad-requests pub const ERROR_CODE_ABORTED: u32 = 4; /// Error when reading a request with [`read_request`]. /// -/// For [`MaxMessageSizeExceeded`](Self::MaxMessageSizeExceeded), -/// [`InvalidRequest`](Self::InvalidRequest), and -/// [`DecodeFailed`](Self::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 [`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. +/// +/// [`MaxMessageSizeExceeded`]: Self::MaxMessageSizeExceeded +/// [`InvalidRequest`]: Self::InvalidRequest +/// [`DecodeFailed`]: Self::DecodeFailed #[stack_error(derive, add_meta)] #[non_exhaustive] pub enum ReadRequestError { @@ -676,7 +684,7 @@ impl HandlerError { /// all other requests on the connection, including streams that a handler moved /// into a spawned task. /// - /// The code should be 16 or higher, see [Error codes](self#error-codes). irpc + /// 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 @@ -710,6 +718,8 @@ impl HandlerError { /// } /// }); /// ``` + /// + /// [Error codes]: self#error-codes pub fn close_connection(code: u32, reason: impl AsRef<[u8]>) -> Self { Self { inner: HandlerErrorInner::CloseConnection { @@ -812,7 +822,9 @@ impl Handler { /// By default, a request that is too large or does not decode closes the /// connection. With `true`, only the request fails, and the handler reads /// the next request. Use it for a protocol that adds request types over - /// time, see [Bad requests](self#bad-requests). + /// time, see [Bad requests]. + /// + /// [Bad requests]: self#bad-requests pub fn skip_bad_requests(mut self, skip: bool) -> Self { self.skip_bad_requests = skip; self @@ -944,7 +956,9 @@ pub trait IncomingRemoteConnection: crate::sealed::Sealed { /// Returns [`ReadRequestError::MaxMessageSizeExceeded`], /// [`ReadRequestError::InvalidRequest`], or [`ReadRequestError::DecodeFailed`] /// for a bad request. The connection is still open after these errors, see -/// [Bad requests](self#bad-requests). +/// [Bad requests]. +/// +/// [Bad requests]: self#bad-requests pub async fn read_request( connection: &impl IncomingRemoteConnection, ) -> Result, ReadRequestError> {