fix: a sender gets ReceiverClosed if the remote receiver is dropped

src/channel.rs

@@ -18,6 +18,9 @@ pub mod oneshot;
 pub enum SendError {
     /// 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`]).

src/rpc.rs

@@ -211,6 +211,8 @@ impl From<noq::WriteError> 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)),
         }
     }

tests/mpsc_channel.rs

@@ -6,15 +6,19 @@ use std::{
 };
 
 use irpc::{
+    Client, WithChannels,
     channel::{
         SendError,
         mpsc::{self, Receiver, RecvError},
+        oneshot,
     },
-    rpc::ERROR_CODE_DECODE_FAILED,
+    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;
 
@@ -120,6 +124,92 @@ 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::<Vec<u8>>::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::<Vec<u8>>::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(())
+}
+
+/// 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<u32>)]
+        #[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::<Proto>::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::<Proto>::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()