From c9feb621e0e4a6223991c657d36a3540c719124f Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Fri, 9 Oct 2026 09:51:40 +0300 Subject: [PATCH 1/3] fix: later sends get ReceiverClosed too after the remote receiver is dropped --- src/rpc.rs | 32 ++++++++++++++++++++++------- tests/mpsc_channel.rs | 48 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 73 insertions(+), 7 deletions(-) diff --git a/src/rpc.rs b/src/rpc.rs index e979b8f..fefc4dc 100644 --- a/src/rpc.rs +++ b/src/rpc.rs @@ -530,10 +530,24 @@ impl NoqSenderInner { #[derive(Default)] enum NoqSenderState { Open(NoqSenderInner), + /// The receiver stopped the stream with code 0, see [`SendError::ReceiverClosed`]. + ReceiverClosed, + /// The stream failed for another reason, or a send was cancelled. #[default] Closed, } +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 { + Ok(_) => Self::Open(sender), + Err(SendError::ReceiverClosed { .. }) => Self::ReceiverClosed, + Err(_) => Self::Closed, + } + } +} + struct NoqSender(tokio::sync::Mutex>); impl Debug for NoqSender { @@ -550,11 +564,13 @@ impl DynSender for NoqSender { 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::ReceiverClosed => { + *guard = NoqSenderState::ReceiverClosed; + Err(e!(SendError::ReceiverClosed)) + } NoqSenderState::Closed => Err(io::Error::from(io::ErrorKind::BrokenPipe).into()), } }) @@ -570,11 +586,13 @@ impl DynSender for NoqSender { 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::ReceiverClosed => { + *guard = NoqSenderState::ReceiverClosed; + Err(e!(SendError::ReceiverClosed)) + } NoqSenderState::Closed => Err(io::Error::from(io::ErrorKind::BrokenPipe).into()), } }) @@ -585,7 +603,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::ReceiverClosed | NoqSenderState::Closed => {} } }) } diff --git a/tests/mpsc_channel.rs b/tests/mpsc_channel.rs index f22416a..4a47288 100644 --- a/tests/mpsc_channel.rs +++ b/tests/mpsc_channel.rs @@ -152,6 +152,54 @@ async fn mpsc_receiver_dropped() -> TestResult<()> { 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(()) +} + async fn vec_receiver(server: Endpoint) -> Result<(), RecvError> { let conn = server .accept() From ede7fdedb75755a6d67e65808011a998c969f20a Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Fri, 9 Oct 2026 09:57:01 +0300 Subject: [PATCH 2/3] refactor: make receiver_closed a flag on the closed state --- src/rpc.rs | 44 +++++++++++++++++++++++++++----------------- 1 file changed, 27 insertions(+), 17 deletions(-) diff --git a/src/rpc.rs b/src/rpc.rs index fefc4dc..4031fe6 100644 --- a/src/rpc.rs +++ b/src/rpc.rs @@ -527,14 +527,17 @@ impl NoqSenderInner { } } -#[derive(Default)] enum NoqSenderState { Open(NoqSenderInner), - /// The receiver stopped the stream with code 0, see [`SendError::ReceiverClosed`]. - ReceiverClosed, - /// The stream failed for another reason, or a send was cancelled. - #[default] - Closed, + Closed { receiver_closed: bool }, +} + +impl Default for NoqSenderState { + fn default() -> Self { + Self::Closed { + receiver_closed: false, + } + } } impl NoqSenderState { @@ -542,8 +545,9 @@ impl NoqSenderState { fn after_send(sender: NoqSenderInner, res: &Result) -> Self { match res { Ok(_) => Self::Open(sender), - Err(SendError::ReceiverClosed { .. }) => Self::ReceiverClosed, - Err(_) => Self::Closed, + Err(err) => Self::Closed { + receiver_closed: matches!(err, SendError::ReceiverClosed { .. }), + }, } } } @@ -567,11 +571,14 @@ impl DynSender for NoqSender { *guard = NoqSenderState::after_send(sender, &res); res } - NoqSenderState::ReceiverClosed => { - *guard = NoqSenderState::ReceiverClosed; - Err(e!(SendError::ReceiverClosed)) + NoqSenderState::Closed { receiver_closed } => { + *guard = NoqSenderState::Closed { receiver_closed }; + if receiver_closed { + Err(e!(SendError::ReceiverClosed)) + } else { + Err(io::Error::from(io::ErrorKind::BrokenPipe).into()) + } } - NoqSenderState::Closed => Err(io::Error::from(io::ErrorKind::BrokenPipe).into()), } }) } @@ -589,11 +596,14 @@ impl DynSender for NoqSender { *guard = NoqSenderState::after_send(sender, &res); res } - NoqSenderState::ReceiverClosed => { - *guard = NoqSenderState::ReceiverClosed; - Err(e!(SendError::ReceiverClosed)) + NoqSenderState::Closed { receiver_closed } => { + *guard = NoqSenderState::Closed { receiver_closed }; + if receiver_closed { + Err(e!(SendError::ReceiverClosed)) + } else { + Err(io::Error::from(io::ErrorKind::BrokenPipe).into()) + } } - NoqSenderState::Closed => Err(io::Error::from(io::ErrorKind::BrokenPipe).into()), } }) } @@ -603,7 +613,7 @@ impl DynSender for NoqSender { let mut guard = self.0.lock().await; match guard.deref_mut() { NoqSenderState::Open(sender) => sender.closed().await, - NoqSenderState::ReceiverClosed | NoqSenderState::Closed => {} + NoqSenderState::Closed { .. } => {} } }) } From 390eebe5ce8bf1638a8f3d660caecf25b494850c Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Fri, 9 Oct 2026 09:58:11 +0300 Subject: [PATCH 3/3] refactor: store why a sender closed as a CloseReason --- src/rpc.rs | 48 ++++++++++++++++++++++++++---------------------- 1 file changed, 26 insertions(+), 22 deletions(-) diff --git a/src/rpc.rs b/src/rpc.rs index 4031fe6..cbdf737 100644 --- a/src/rpc.rs +++ b/src/rpc.rs @@ -529,14 +529,27 @@ impl NoqSenderInner { enum NoqSenderState { Open(NoqSenderInner), - Closed { receiver_closed: bool }, + 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 Default for NoqSenderState { fn default() -> Self { - Self::Closed { - receiver_closed: false, - } + Self::Closed(CloseReason::Other) } } @@ -545,9 +558,8 @@ impl NoqSenderState { fn after_send(sender: NoqSenderInner, res: &Result) -> Self { match res { Ok(_) => Self::Open(sender), - Err(err) => Self::Closed { - receiver_closed: matches!(err, SendError::ReceiverClosed { .. }), - }, + Err(SendError::ReceiverClosed { .. }) => Self::Closed(CloseReason::ReceiverClosed), + Err(_) => Self::Closed(CloseReason::Other), } } } @@ -571,13 +583,9 @@ impl DynSender for NoqSender { *guard = NoqSenderState::after_send(sender, &res); res } - NoqSenderState::Closed { receiver_closed } => { - *guard = NoqSenderState::Closed { receiver_closed }; - if receiver_closed { - Err(e!(SendError::ReceiverClosed)) - } else { - Err(io::Error::from(io::ErrorKind::BrokenPipe).into()) - } + NoqSenderState::Closed(reason) => { + *guard = NoqSenderState::Closed(reason); + Err(reason.error()) } } }) @@ -596,13 +604,9 @@ impl DynSender for NoqSender { *guard = NoqSenderState::after_send(sender, &res); res } - NoqSenderState::Closed { receiver_closed } => { - *guard = NoqSenderState::Closed { receiver_closed }; - if receiver_closed { - Err(e!(SendError::ReceiverClosed)) - } else { - Err(io::Error::from(io::ErrorKind::BrokenPipe).into()) - } + NoqSenderState::Closed(reason) => { + *guard = NoqSenderState::Closed(reason); + Err(reason.error()) } } }) @@ -613,7 +617,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(_) => {} } }) }