fix: a oneshot receiver gets SenderClosed if the remote sender is dropped

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`]).

src/rpc.rs

@@ -325,10 +325,10 @@ impl<S: Service> RemoteSender<S> {
 impl<T: DeserializeOwned> From<noq::RecvStream> for oneshot::Receiver<T> {
     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));

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::<Vec<u8>>::from(recv).await;
+        TestResult::Ok(res)
+    });
+    let conn = client.connect(server_addr, "localhost")?.await?;
+    let (send, _) = conn.open_bi().await?;
+    drop(oneshot::Sender::<Vec<u8>>::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<()> {