diff --git a/src/channel/oneshot.rs b/src/channel/oneshot.rs index ef4a382..0c02c2d 100644 --- a/src/channel/oneshot.rs +++ b/src/channel/oneshot.rs @@ -18,6 +18,9 @@ use crate::util::FusedOneshotReceiver; pub enum RecvError { /// 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/rpc.rs b/src/rpc.rs index cbdf737..8954746 100644 --- a/src/rpc.rs +++ b/src/rpc.rs @@ -325,10 +325,10 @@ impl RemoteSender { 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)); diff --git a/tests/oneshot_channel.rs b/tests/oneshot_channel.rs index d1f57ae..1cde146 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<()> {