diff --git a/examples/auth.rs b/examples/auth.rs index 86ddba4c..76a8233a 100644 --- a/examples/auth.rs +++ b/examples/auth.rs @@ -136,10 +136,10 @@ mod storage { StorageMessage::Auth(msg) => { let WithChannels { inner, tx, .. } = msg; if authed { - conn.close(1u32.into(), b"invalid message"); + conn.close(400u32.into(), b"invalid message"); break; } else if inner.token != self.auth_token { - conn.close(1u32.into(), b"permission denied"); + conn.close(401u32.into(), b"permission denied"); break; } else { authed = true; @@ -148,7 +148,7 @@ mod storage { } msg => { if !authed { - conn.close(1u32.into(), b"permission denied"); + conn.close(401u32.into(), b"permission denied"); break; } else { self.handle_authenticated(msg).await; diff --git a/src/channel.rs b/src/channel.rs index 9d5f62ca..de8f4899 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -14,9 +14,13 @@ pub mod oneshot; /// For rpc communication, there can be any number of errors, so this is a /// generic io error. #[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. + /// + /// 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/mpsc.rs b/src/channel/mpsc.rs index 74138c61..80f82a21 100644 --- a/src/channel/mpsc.rs +++ b/src/channel/mpsc.rs @@ -14,6 +14,7 @@ use super::SendError; /// For rpc communication, there can be any number of errors, so this is a /// generic io error. #[stack_error(derive, add_meta, from_sources)] +#[non_exhaustive] pub enum RecvError { /// 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 921424a5..7543da1f 100644 --- a/src/channel/oneshot.rs +++ b/src/channel/oneshot.rs @@ -14,9 +14,13 @@ use crate::util::FusedOneshotReceiver; /// For rpc communication, there can be any number of errors, so this is a /// generic io error. #[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. + /// + /// 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`]). diff --git a/src/lib.rs b/src/lib.rs index 6350f815..5ba1f9d2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -949,6 +949,7 @@ impl ClientInner { /// Error when opening a request. When cross-process rpc is disabled, this is /// an empty enum since local requests can not fail. #[stack_error(derive, add_meta, from_sources)] +#[non_exhaustive] pub enum RequestError { /// Error in noq during connect #[cfg(feature = "rpc")] @@ -976,6 +977,7 @@ pub enum RequestError { /// Error type that subsumes all possible errors in this crate, for convenience. #[stack_error(derive, add_meta, from_sources)] +#[non_exhaustive] pub enum Error { #[error("Request error")] Request { source: RequestError }, diff --git a/src/rpc.rs b/src/rpc.rs index 10e6fe7b..99ad7ae4 100644 --- a/src/rpc.rs +++ b/src/rpc.rs @@ -3,6 +3,52 @@ //! The code in this module works with all connections that use [`noq`] //! 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 +//! +//! 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 +//! than [`MAX_MESSAGE_SIZE`], +//! * with [`ERROR_CODE_DECODE_FAILED`] 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. +//! +//! 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 +//! end of its enum can use [`Handler::skip_bad_requests`]: the handler then +//! reads the next request, so an old server rejects the new requests and keeps +//! serving the old ones on the same connection. +//! +//! [`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. +//! +//! 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`] 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 +//! 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. +//! +//! 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, }; @@ -32,15 +78,96 @@ use crate::{ /// Default max message size (16 MiB). pub const MAX_MESSAGE_SIZE: u64 = 1024 * 1024 * 16; -/// Error code on streams if the max message size was exceeded. +/// Error code on streams and connections if a message is larger than [`MAX_MESSAGE_SIZE`]. +/// +/// 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 tried to send an message that could not be postcard serialized. -pub const ERROR_CODE_INVALID_POSTCARD: u32 = 2; +/// Error code on streams if the sender could not encode a message. +pub const ERROR_CODE_ENCODE_FAILED: u32 = 2; + +/// 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; + +/// 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; + +/// 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. +/// +/// [`MaxMessageSizeExceeded`]: Self::MaxMessageSizeExceeded +/// [`InvalidRequest`]: Self::InvalidRequest +/// [`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, + /// The request is not framed correctly. + /// + /// For example, the stream ended before the request was complete. + #[error("Invalid request")] + InvalidRequest { + #[error(std_err)] + source: io::Error, + }, + /// The request does not decode. + /// + /// For example, a newer client sent a request type that this server does + /// not know. + #[error("Request does not decode")] + DecodeFailed { + #[error(std_err)] + source: postcard::Error, + }, + /// The connection failed. + #[error("Connection failed")] + Connection { + #[error(std_err)] + source: ConnectionError, + }, +} + +impl ReadRequestError { + /// Returns the irpc error code for a bad request, or `None` for a connection error. + fn error_code(&self) -> Option { + match self { + Self::MaxMessageSizeExceeded { .. } => Some(ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED), + Self::InvalidRequest { .. } | Self::DecodeFailed { .. } => { + Some(ERROR_CODE_DECODE_FAILED) + } + Self::Connection { .. } => None, + } + } +} + +impl From for io::Error { + fn from(err: ReadRequestError) -> Self { + match err { + ReadRequestError::Connection { source, .. } => source.into(), + err => io::Error::new(io::ErrorKind::InvalidData, err), + } + } +} /// Error that can occur when writing the initial message when doing a /// cross-process RPC. #[stack_error(derive, add_meta, from_sources)] +#[non_exhaustive] pub enum WriteError { /// Error writing to the stream with noq #[error("Error writing to stream")] @@ -92,6 +219,8 @@ impl From for SendError { { 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), _ => e!(SendError::Io, io::Error::from(err)), } } @@ -120,12 +249,33 @@ 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`]. #[derive(Debug)] -pub struct RemoteSender( - noq::SendStream, - noq::RecvStream, - std::marker::PhantomData, -); +pub struct RemoteSender(ResetOnDrop, noq::RecvStream, std::marker::PhantomData); + +/// A send stream that resets with [`ERROR_CODE_ABORTED`] if it drops before [`Self::into_inner`]. +#[derive(Debug)] +struct ResetOnDrop(Option); + +impl ResetOnDrop { + fn get_mut(&mut self) -> &mut noq::SendStream { + self.0.as_mut().expect("only `into_inner` takes the stream") + } + + fn into_inner(mut self) -> noq::SendStream { + self.0.take().expect("only `into_inner` takes the stream") + } +} + +impl Drop for ResetOnDrop { + fn drop(&mut self) { + if let Some(send) = &mut self.0 { + send.reset(ERROR_CODE_ABORTED.into()).ok(); + } + } +} /// Serialize a message for sending over the wire. /// @@ -159,7 +309,7 @@ pub(crate) fn prepare_write( impl RemoteSender { pub fn new(send: noq::SendStream, recv: noq::RecvStream) -> Self { - Self(send, recv, PhantomData) + Self(ResetOnDrop(Some(send)), recv, PhantomData) } pub async fn write( @@ -175,18 +325,18 @@ impl RemoteSender { buf: &[u8], ) -> std::result::Result<(noq::SendStream, noq::RecvStream), WriteError> { let RemoteSender(mut send, recv, _) = self; - send.write_all(buf).await?; - Ok((send, recv)) + send.get_mut().write_all(buf).await?; + Ok((send.into_inner(), recv)) } } impl From for oneshot::Receiver { fn from(mut read: noq::RecvStream) -> Self { let fut = async move { - let size = read.read_varint_u64().await?.ok_or(io::Error::new( - io::ErrorKind::UnexpectedEof, - "failed to read size", - ))?; + // A sender that is dropped without sending finishes the stream. + let Some(size) = read.read_varint_u64().await? else { + return Err(e!(oneshot::RecvError::SenderClosed)); + }; if size > MAX_MESSAGE_SIZE { read.stop(ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED.into()).ok(); return Err(e!(oneshot::RecvError::MaxMessageSizeExceeded)); @@ -195,8 +345,10 @@ impl From for oneshot::Receiver { .read_to_end(size as usize) .await .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; - let msg: T = postcard::from_bytes(&rest) - .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(); + io::Error::new(io::ErrorKind::InvalidData, e) + })?; Ok(msg) }; oneshot::Receiver::from(|| fut) @@ -233,7 +385,7 @@ impl From for oneshot::Sender { let size = match postcard::experimental::serialized_size(&value) { Ok(size) => size, Err(e) => { - writer.reset(ERROR_CODE_INVALID_POSTCARD.into()).ok(); + writer.reset(ERROR_CODE_ENCODE_FAILED.into()).ok(); return Err(e!( SendError::Io, io::Error::new(io::ErrorKind::InvalidData, e,) @@ -249,7 +401,7 @@ impl From for oneshot::Sender { // 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_INVALID_POSTCARD.into()).ok(); + writer.reset(ERROR_CODE_ENCODE_FAILED.into()).ok(); return Err(e.into()); } writer.write_all(&buf).await?; @@ -301,8 +453,10 @@ impl DynReceiver for NoqReceiver { read.read_exact(&mut buf) .await .map_err(|e| io::Error::new(io::ErrorKind::UnexpectedEof, e))?; - let msg: T = postcard::from_bytes(&buf) - .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; + let msg: T = postcard::from_bytes(&buf).map_err(|e| { + read.stop(ERROR_CODE_DECODE_FAILED.into()).ok(); + io::Error::new(io::ErrorKind::InvalidData, e) + })?; Ok(Some(msg)) }) } @@ -319,33 +473,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_INVALID_POSTCARD.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_INVALID_POSTCARD.into()).ok(); - return Err(e.into()); - } + self.encode(value)?; self.send.write_all(&self.buffer).await?; self.buffer.clear(); Ok(()) @@ -357,13 +519,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); }; @@ -381,11 +538,43 @@ impl NoqSenderInner { } } -#[derive(Default)] enum NoqSenderState { Open(NoqSenderInner), - #[default] - Closed, + 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(), + } + } +} + +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)) + } + + /// 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 { + Ok(_) => Self::Open(sender), + Err(SendError::ReceiverClosed { .. }) => Self::Closed(CloseReason::ReceiverClosed), + Err(_) => Self::Closed(CloseReason::Other), + } + } } struct NoqSender(tokio::sync::Mutex>); @@ -400,16 +589,17 @@ 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; - if res.is_ok() { - *guard = NoqSenderState::Open(sender); - } + *guard = NoqSenderState::after_send(sender, &res); res } - NoqSenderState::Closed => Err(io::Error::from(io::ErrorKind::BrokenPipe).into()), + NoqSenderState::Closed(reason) => { + *guard = NoqSenderState::Closed(reason); + Err(reason.error()) + } } }) } @@ -420,16 +610,17 @@ 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; - if res.is_ok() { - *guard = NoqSenderState::Open(sender); - } + *guard = NoqSenderState::after_send(sender, &res); res } - NoqSenderState::Closed => Err(io::Error::from(io::ErrorKind::BrokenPipe).into()), + NoqSenderState::Closed(reason) => { + *guard = NoqSenderState::Closed(reason); + Err(reason.error()) + } } }) } @@ -439,7 +630,7 @@ impl DynSender for NoqSender { let mut guard = self.0.lock().await; match guard.deref_mut() { NoqSenderState::Open(sender) => sender.closed().await, - NoqSenderState::Closed => {} + NoqSenderState::Closed(_) => {} } }) } @@ -493,6 +684,12 @@ 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]. 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 + /// [`ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED`] or [`ERROR_CODE_DECODE_FAILED`]. + /// /// # Examples /// /// ``` @@ -521,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 { @@ -542,6 +741,8 @@ impl HandlerError { /// A handler function can return `Err(HandlerError)` to close the connection. /// In all other cases, handle errors internally and return `Ok(())` from /// the handler function to read the next request from the connection. +/// A bad request also closes the connection, unless +/// [`Handler::skip_bad_requests`] is set. /// /// Often, the protocol type cannot be inferred from the message enum, /// so you may have to name it: `Handler::::sequential(..)`. @@ -574,11 +775,15 @@ impl HandlerError { pub struct Handler { #[debug(skip)] f: HandlerFn, + skip_bad_requests: bool, } impl Clone for Handler { fn clone(&self) -> Self { - Self { f: self.f.clone() } + Self { + f: self.f.clone(), + skip_bad_requests: self.skip_bad_requests, + } } } @@ -594,7 +799,10 @@ impl Handler { Fut: Future> + Send + 'static, { let f: HandlerFn = Arc::new(move |msg, rx, tx| Box::pin(f(msg, rx, tx))); - Self { f } + Self { + f, + skip_bad_requests: false, + } } /// Runs this handler on one request. @@ -609,14 +817,29 @@ impl Handler { (self.f)(request, rx, tx) } + /// Sets whether a bad request is skipped instead of closing the connection. + /// + /// 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]. + /// + /// [Bad requests]: self#bad-requests + pub fn skip_bad_requests(mut self, skip: bool) -> Self { + self.skip_bad_requests = skip; + self + } + /// Handles the requests of `connection` until the connection ends. + /// + /// Returns an error if the connection fails, or if a bad request closed it. pub async fn handle_connection( &self, connection: &impl IncomingRemoteConnection, ) -> io::Result<()> { debug!("connection accepted"); loop { - let Some((msg, carrier, rx, tx)) = read_request_inner::(connection).await? else { + let Some((msg, carrier, rx, tx)) = self.read_next(connection).await? else { return Ok(()); }; let fut = (self.f)(msg, rx, tx); @@ -632,6 +855,26 @@ impl Handler { } } } + + /// Reads the next request, and closes the connection or skips it on a bad request. + async fn read_next( + &self, + connection: &impl IncomingRemoteConnection, + ) -> io::Result>> { + loop { + match read_request_inner::(connection).await { + Ok(request) => return Ok(request), + Err(err) => match err.error_code() { + None => return Err(err.into()), // A connection error. + Some(_) if self.skip_bad_requests => debug!("skipped bad request: {err:#}"), + Some(code) => { + connection.close(code.into(), err.to_string().as_bytes()); + return Err(err.into()); + } + }, + } + } + } } impl Handler { @@ -706,9 +949,19 @@ pub trait IncomingRemoteConnection: crate::sealed::Sealed { /// Reads a request from a connection and converts it to a message enum. /// /// This combines `read_request_raw` with `RemoteService::with_remote_channels`. +/// +/// 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`] +/// for a bad request. The connection is still open after these errors, see +/// [Bad requests]. +/// +/// [Bad requests]: self#bad-requests pub async fn read_request( connection: &impl IncomingRemoteConnection, -) -> std::io::Result> { +) -> Result, ReadRequestError> { let Some((msg, carrier, rx, tx)) = read_request_inner::(connection).await? else { return Ok(None); }; @@ -728,65 +981,134 @@ pub async fn read_request( /// silently dropped. Use [`Handler::handle_connection`] (or [`read_request`]) if you need /// the propagated context to reach the generated handler spans. /// -/// Returns the parsed request and the stream pair if reading and parsing the request succeeded. -/// Returns None if the remote closed the connection with error code `0`. -/// Returns an error for all other failure cases. +/// Returns `None` in the same cases as [`read_request`]. pub async fn read_request_raw( connection: &impl IncomingRemoteConnection, -) -> std::io::Result> { +) -> Result, ReadRequestError> { Ok(read_request_inner::(connection) .await? .map(|(msg, _carrier, rx, tx)| (msg, rx, tx))) } +/// A request with its span context carrier and its streams. +type Request = ( + S, + Option, + noq::RecvStream, + noq::SendStream, +); + /// Internal: read a request and also return the propagated span context carrier. /// /// The carrier is `Some` iff `S::SPAN_PROPAGATION` is true and the remote sent one. async fn read_request_inner( connection: &impl IncomingRemoteConnection, -) -> std::io::Result< - Option<( - S, - Option, - noq::RecvStream, - noq::SendStream, - )>, -> { - let (send, mut recv) = match connection.accept_bi().await { - Ok((s, r)) => (s, r), - Err(ConnectionError::ApplicationClosed(cause)) if cause.error_code.into_inner() == 0 => { - trace!("remote side closed connection {cause:?}"); - return Ok(None); +) -> Result>, ReadRequestError> { + loop { + let (mut send, mut recv) = match connection.accept_bi().await { + Ok(streams) => streams, + Err(ConnectionError::ApplicationClosed(cause)) + if cause.error_code.into_inner() == 0 => + { + trace!("remote side closed connection {cause:?}"); + return Ok(None); + } + Err(ConnectionError::LocallyClosed) => return Ok(None), + Err(cause) => return Err(e!(ReadRequestError::Connection, cause)), + }; + let err = match read_request_frame::(&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::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(0u32.into()).ok(); + continue; + } + // The next `accept_bi` returns why. + ReadFrame::ConnectionLost => continue, + }; + if let Some(code) = err.error_code() { + recv.stop(code.into()).ok(); + send.reset(code.into()).ok(); + } + return Err(err); + } +} + +/// What the server reads from the stream of a request. +enum ReadFrame { + /// A complete request. + Request(Option, S), + /// The request is larger than [`MAX_MESSAGE_SIZE`]. + MaxMessageSizeExceeded, + /// The stream ended before the request was complete, or the size prefix is too long. + InvalidRequest(io::Error), + /// The request does not decode. + DecodeFailed(postcard::Error), + /// The client reset the stream before the request was complete. + Reset, + /// The connection was lost before the request was complete. + ConnectionLost, +} + +impl ReadFrame { + /// 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 + // 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::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, + } + } +} + +/// Reads and decodes a request. +async fn read_request_frame(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( + io::ErrorKind::UnexpectedEof, + "stream ended before the request", + )); } - Err(cause) => { - warn!("failed to accept bi stream {cause:?}"); - return Err(cause.into()); + Err(err) => { + let read_err = err + .get_ref() + .and_then(|err| err.downcast_ref::()); + return match read_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), + }; } }; - let size = recv - .read_varint_u64() - .await? - .ok_or_else(|| io::Error::new(io::ErrorKind::UnexpectedEof, "failed to read size"))?; if size > MAX_MESSAGE_SIZE { - connection.close( - ERROR_CODE_MAX_MESSAGE_SIZE_EXCEEDED.into(), - b"request exceeded max message size", - ); - return Err(e!(mpsc::RecvError::MaxMessageSizeExceeded).into()); + return ReadFrame::MaxMessageSizeExceeded; } let mut buf = vec![0; size as usize]; - recv.read_exact(&mut buf) - .await - .map_err(|e| io::Error::new(io::ErrorKind::UnexpectedEof, e))?; - - let (carrier, msg): (Option, S) = - if S::SPAN_PROPAGATION { - postcard::from_bytes(&buf).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))? - } else { - let msg = postcard::from_bytes(&buf) - .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; - (None, msg) - }; - - Ok(Some((msg, carrier, recv, send))) + 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)); + } + } + let decoded = if S::SPAN_PROPAGATION { + postcard::from_bytes(&buf) + } else { + postcard::from_bytes(&buf).map(|msg| (None, msg)) + }; + match decoded { + Ok((carrier, msg)) => ReadFrame::Request(carrier, msg), + Err(err) => ReadFrame::DecodeFailed(err), + } } diff --git a/src/rpc/iroh_impl.rs b/src/rpc/iroh_impl.rs index 2fae19c1..2fa41363 100644 --- a/src/rpc/iroh_impl.rs +++ b/src/rpc/iroh_impl.rs @@ -17,7 +17,7 @@ use tracing::{Instrument, debug, error_span, trace_span, warn}; use crate::{ RequestError, Service, - rpc::{Handler, IncomingRemoteConnection, RemoteConnection}, + rpc::{Handler, IncomingRemoteConnection, ReadRequestError, RemoteConnection}, }; impl RemoteConnection for Connection { @@ -137,6 +137,12 @@ async fn connect_and_open_bi( Ok((send, recv)) } +impl From for AcceptError { + fn from(err: ReadRequestError) -> Self { + AcceptError::from_err(err) + } +} + /// A [`ProtocolHandler`] for an irpc protocol. /// /// Can be added to an [`iroh::protocol::Router`] to handle incoming connections for an ALPN string. @@ -182,7 +188,6 @@ impl ProtocolHandler for IrohProtocol { /// Can be added to an [`iroh::protocol::Router`] to handle incoming connections for an ALPN string. /// /// For details about when it is safe to use 0rtt, see -/// For details about when it is safe to use 0rtt, see pub struct Iroh0RttProtocol { handler: Handler, request_id: AtomicU64, diff --git a/tests/invalid_request.rs b/tests/invalid_request.rs new file mode 100644 index 00000000..c24a622a --- /dev/null +++ b/tests/invalid_request.rs @@ -0,0 +1,281 @@ +#![cfg(all(feature = "noq_endpoint_setup", feature = "derive"))] +//! Checks what a server does with a bad request. + +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, + }, +}; +use n0_future::{future::poll_once, task::AbortOnDropHandle}; +use noq::ConnectionError; +use testresult::TestResult; + +#[allow(dead_code, reason = "this test uses only some of the helpers")] +mod common; +use common::*; + +/// The protocol of the server. +mod server { + use irpc::{ + channel::{mpsc, oneshot}, + rpc_requests, + }; + use serde::{Deserialize, Serialize}; + + use super::NoSer; + + #[rpc_requests(message = EchoMessage)] + #[derive(Debug, Serialize, Deserialize)] + pub enum EchoProtocol { + #[rpc(tx = oneshot::Sender)] + #[wrap(Echo)] + Echo(String), + #[rpc(tx = oneshot::Sender<()>, rx = mpsc::Receiver)] + #[wrap(Upload)] + Upload(String), + } +} + +/// A newer version of the protocol, with a request that the server does not know. +mod client { + use irpc::{ + channel::{mpsc, oneshot}, + rpc_requests, + }; + use serde::{Deserialize, Serialize}; + + use super::NoSer; + + #[rpc_requests(message = EchoMessage)] + #[derive(Debug, Serialize, Deserialize)] + pub enum EchoProtocol { + #[rpc(tx = oneshot::Sender)] + #[wrap(Echo)] + Echo(String), + #[rpc(tx = oneshot::Sender<()>, rx = mpsc::Receiver)] + #[wrap(Upload)] + Upload(String), + #[rpc(tx = oneshot::Sender)] + #[wrap(Shout)] + Shout(String), + } +} + +/// Returns a handler that answers `Echo`. +fn echo_handler() -> Handler { + Handler::sequential(|msg| async move { + if let server::EchoMessage::Echo(WithChannels { inner, tx, .. }) = msg { + tx.send(inner.0).await.ok(); + } + Ok(()) + }) +} + +#[tokio::test] +async fn bad_request_closes_connection() -> TestResult<()> { + let (server, client_endpoint, server_addr) = create_connected_endpoints()?; + let _server = AbortOnDropHandle::new(tokio::spawn(listen(server, echo_handler()))); + let conn = client_endpoint.connect(server_addr, "localhost")?.await?; + let client = Client::::boxed(conn.clone()); + + client + .rpc(client::Shout("a".into())) + .await + .expect_err("server does not know the request"); + let ConnectionError::ApplicationClosed(close) = conn.closed().await else { + panic!("server closes the connection"); + }; + assert_eq!(close.error_code, ERROR_CODE_DECODE_FAILED.into()); + Ok(()) +} + +#[tokio::test] +async fn too_large_request_closes_connection() -> TestResult<()> { + let (server, client_endpoint, server_addr) = create_connected_endpoints()?; + let _server = AbortOnDropHandle::new(tokio::spawn(listen(server, echo_handler()))); + let conn = client_endpoint.connect(server_addr, "localhost")?.await?; + + // An irpc client does not send a request this large, so write only its size prefix. + let (mut send, _recv) = conn.open_bi().await?; + send.write_all(&postcard::to_stdvec(&(MAX_MESSAGE_SIZE + 1))?) + .await?; + 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() + ); + Ok(()) +} + +#[tokio::test] +async fn skip_bad_requests_keeps_connection() -> 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?; + // A client on this one connection, so the second request shows that it still works. + let client = Client::::boxed(conn); + + client + .rpc(client::Shout("a".into())) + .await + .expect_err("server does not know the request"); + assert_eq!(client.rpc(client::Echo("b".into())).await?, "b"); + Ok(()) +} + +#[tokio::test] +async fn dropped_rpc_keeps_connection() -> TestResult<()> { + let (server, client_endpoint, server_addr) = create_connected_endpoints()?; + let _server = AbortOnDropHandle::new(tokio::spawn(listen(server, echo_handler()))); + let conn = client_endpoint.connect(server_addr, "localhost")?.await?; + let client = Client::::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`. + 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); + assert_eq!(client.rpc(client::Echo("b".into())).await?, "b"); + Ok(()) +} + +#[tokio::test] +async fn truncated_request_closes_connection() -> TestResult<()> { + let (server, client_endpoint, server_addr) = create_connected_endpoints()?; + let _server = AbortOnDropHandle::new(tokio::spawn(listen(server, echo_handler()))); + let conn = client_endpoint.connect(server_addr, "localhost")?.await?; + + let (mut send, _recv) = conn.open_bi().await?; + // The size prefix says 100 bytes, but only 3 follow. + send.write_all(&[100, 1, 2, 3]).await?; + send.finish()?; + let ConnectionError::ApplicationClosed(close) = conn.closed().await else { + panic!("server closes the connection"); + }; + assert_eq!(close.error_code, ERROR_CODE_DECODE_FAILED.into()); + Ok(()) +} + +/// The update sender resets the stream of its request before the server reads it. +#[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 { + tokio::time::sleep(Duration::from_millis(1)).await; + } + let Some(server::EchoMessage::Echo(msg)) = + read_request::(&conn).await? + else { + panic!("server skips the upload"); + }; + msg.tx.send(msg.inner.0).await?; + conn.closed().await; + TestResult::Ok(()) + }); + let conn = client_endpoint.connect(server_addr, "localhost")?.await?; + let client = Client::::boxed(conn.clone()); + + 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`. + tx.send(NoSer(1)) + .await + .expect_err("odd numbers do not encode"); + assert_eq!(client.rpc(client::Echo("b".into())).await?, "b"); + conn.close(0u32.into(), b""); + server.await??; + Ok(()) +} + +#[tokio::test] +async fn read_request_returns_truncated_request() -> 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 (mut send, _recv) = client_conn.open_bi().await?; + // The size prefix says 100 bytes, but only 3 follow. + send.write_all(&[100, 1, 2, 3]).await?; + send.finish()?; + let err = read_request::(&server_conn) + .await + .expect_err("the request is not complete"); + assert!( + matches!(err, ReadRequestError::InvalidRequest { .. }), + "{err:?}" + ); + Ok(()) +} + +/// A close in the middle of a request ends `read_request` as a close between requests does. +#[tokio::test] +async fn close_during_request_returns_none() -> 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 (mut send, _recv) = client_conn.open_bi().await?; + // The size prefix says 100 bytes, but only 3 follow. + send.write_all(&[100, 1, 2, 3]).await?; + let mut read = pin!(read_request::(&server_conn)); + while server_conn.stats().frame_rx.stream == 0 { + assert!(poll_once(&mut read).await.is_none()); + tokio::time::sleep(Duration::from_millis(1)).await; + } + assert!( + poll_once(&mut read).await.is_none(), + "server waits for the rest of the request" + ); + client_conn.close(0u32.into(), b""); + assert!(read.await?.is_none()); + Ok(()) +} + +#[tokio::test] +async fn read_request_returns_bad_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?; + let err = read_request::(&conn) + .await + .expect_err("server does not know the request"); + assert!( + matches!(err, ReadRequestError::DecodeFailed { .. }), + "{err:?}" + ); + // The connection is still open, so the server reads the next request. + let Some(server::EchoMessage::Echo(msg)) = + read_request::(&conn).await? + else { + panic!("client sends a second request"); + }; + msg.tx.send(msg.inner.0).await?; + conn.closed().await; + TestResult::Ok(()) + }); + let conn = client_endpoint.connect(server_addr, "localhost")?.await?; + let client = Client::::boxed(conn.clone()); + + client + .rpc(client::Shout("a".into())) + .await + .expect_err("server does not know the request"); + assert_eq!(client.rpc(client::Echo("b".into())).await?, "b"); + conn.close(0u32.into(), b""); + server.await??; + Ok(()) +} diff --git a/tests/mpsc_channel.rs b/tests/mpsc_channel.rs index 360d9375..d0d4f567 100644 --- a/tests/mpsc_channel.rs +++ b/tests/mpsc_channel.rs @@ -6,14 +6,19 @@ use std::{ }; use irpc::{ + Client, WithChannels, channel::{ SendError, mpsc::{self, Receiver, RecvError}, + oneshot, }, + rpc::{ERROR_CODE_DECODE_FAILED, Handler}, + rpc_requests, util::AsyncWriteVarintExt, }; use n0_error::e; use noq::Endpoint; +use serde::{Deserialize, Serialize}; use testresult::TestResult; use tokio::time::timeout; @@ -119,6 +124,140 @@ async fn mpsc_sender_clone_drop_error() -> TestResult<()> { Ok(()) } +/// Checks that a sender gets `ReceiverClosed` if the remote receiver is dropped, as for a local one. +#[tokio::test] +async fn mpsc_receiver_dropped() -> 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?; + drop(Receiver::>::from(recv)); + // keep the connection, so that the stop reaches the client + conn.closed().await; + TestResult::Ok(()) + }); + let conn = client.connect(server_addr, "localhost")?.await?; + let (send, _) = conn.open_bi().await?; + let send = mpsc::Sender::>::from(send); + let res = timeout(Duration::from_secs(5), async { + loop { + if let Err(err) = send.send(vec![1, 2, 3]).await { + break err; + } + } + }) + .await?; + assert!( + matches!(res, SendError::ReceiverClosed { .. }), + "expected ReceiverClosed, got {res:?}" + ); + conn.close(0u32.into(), b""); + server.await??; + Ok(()) +} + +/// Checks that after the remote receiver is dropped, every send on every clone +/// of a `Sender` gets `ReceiverClosed`, also on clones that did not see the stop. +#[tokio::test] +async fn mpsc_receiver_dropped_clones() -> 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?; + drop(Receiver::>::from(recv)); + // keep the connection, so that the stop reaches the client + conn.closed().await; + TestResult::Ok(()) + }); + let conn = client.connect(server_addr, "localhost")?.await?; + let (send, _) = conn.open_bi().await?; + let send1 = mpsc::Sender::>::from(send); + let send2 = send1.clone(); + // send on the first clone until the stop arrives + let err = timeout(Duration::from_secs(5), async { + loop { + if let Err(err) = send1.send(vec![1, 2, 3]).await { + break err; + } + } + }) + .await?; + assert!( + matches!(err, SendError::ReceiverClosed { .. }), + "expected ReceiverClosed, got {err:?}" + ); + // later sends on the same clone, and sends on a clone that never saw the stop + assert!(matches!( + send1.send(vec![1]).await, + Err(SendError::ReceiverClosed { .. }) + )); + assert!(matches!( + send2.send(vec![1]).await, + Err(SendError::ReceiverClosed { .. }) + )); + assert!(matches!( + send2.try_send(vec![1]).await, + Err(SendError::ReceiverClosed { .. }) + )); + conn.close(0u32.into(), b""); + server.await??; + Ok(()) +} + +/// Check that `ReceiverClosed` works well with the `Handler` as well. +#[tokio::test] +async fn mpsc_receiver_dropped_proto() -> TestResult<()> { + #[rpc_requests(message = ProtoMessage)] + #[derive(Debug, Serialize, Deserialize)] + enum Proto { + #[rpc(tx = oneshot::Sender<()>, rx = mpsc::Receiver)] + #[wrap(ClientStream)] + ClientStream(()), + } + let (server, client, server_addr) = create_connected_endpoints()?; + let server = tokio::spawn(async move { + let conn = server.accept().await.unwrap().await.unwrap(); + Handler::::sequential(|msg| async move { + match msg { + ProtoMessage::ClientStream(msg) => { + let WithChannels { tx, mut rx, .. } = msg; + tx.send(()).await.unwrap(); + let n = rx.recv().await.unwrap().unwrap(); + assert_eq!(n, 1); + drop(rx); + } + } + Ok(()) + }) + .handle_connection(&conn) + .await + .unwrap(); + }); + let client = Client::::noq(client, server_addr); + let (tx, rx) = client.client_streaming(ClientStream(()), 4).await.unwrap(); + assert!(matches!(rx.await.unwrap(), ())); + + // First send works. + let res = tx.send(1).await; + assert!(matches!(res, Ok(()))); + + // As soon as the Stopped frame arrives, our sends fail. + n0_future::time::timeout(Duration::from_secs(1), async { + loop { + let res = tx.send(2).await; + if matches!(res, Err(SendError::ReceiverClosed { .. })) { + break; + } + } + }) + .await + .unwrap(); + + drop(client); + server.await.unwrap(); + Ok(()) +} + async fn vec_receiver(server: Endpoint) -> Result<(), RecvError> { let conn = server .accept() @@ -159,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<()> { @@ -242,3 +402,27 @@ async fn mpsc_serialize_error_recv() -> TestResult<()> { ); Ok(()) } + +/// Checks that a receiver stops the stream with code 3 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::::from(recv); + assert!(recv.recv().await.is_err()); + // keep the connection, so that the stop reaches the client + conn.closed().await; + TestResult::Ok(()) + }); + 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(ERROR_CODE_DECODE_FAILED.into())); + conn.close(0u32.into(), b""); + server.await??; + Ok(()) +} diff --git a/tests/oneshot_channel.rs b/tests/oneshot_channel.rs index d1f57ae4..1cde1469 100644 --- a/tests/oneshot_channel.rs +++ b/tests/oneshot_channel.rs @@ -32,6 +32,27 @@ async fn vec_receiver(server: Endpoint) -> Result<(), RecvError> { Err(e!(RecvError::Io, io::ErrorKind::UnexpectedEof.into())) } +/// Checks that a receiver gets `SenderClosed` if the remote sender is dropped without sending, as for a local one. +#[tokio::test] +async fn oneshot_sender_dropped() -> 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 res = oneshot::Receiver::>::from(recv).await; + TestResult::Ok(res) + }); + let conn = client.connect(server_addr, "localhost")?.await?; + let (send, _) = conn.open_bi().await?; + drop(oneshot::Sender::>::from(send)); + let res = server.await??; + assert!( + matches!(res, Err(RecvError::SenderClosed { .. })), + "expected SenderClosed, got {res:?}" + ); + Ok(()) +} + /// Checks that the max message size is enforced on the sender side and that errors are propagated to the receiver side. #[tokio::test] async fn oneshot_max_message_size_send() -> TestResult<()> {