From bf18ed452d6aae438e84ae008a01a74776abdc19 Mon Sep 17 00:00:00 2001
From: LeoniePhiline <22329650+LeoniePhiline@users.noreply.github.com>
Date: Fri, 27 Mar 2026 01:13:26 +0100
Subject: [PATCH] sync: fix panic in `Chan::recv_many` when called with
 non-empty vector on closed channel (#7991)

`Chan::recv_many` intends to assert that no slots
have been consumed when exiting with `Ready` via
the `rx_closed` code path.

Instead of asserting no items were added to the
buffer, it asserted buffer emptiness, incorrectly
making assumptions about the provided buffer.

When `recv_many` was called on an empty channel
with idle semaphore after the receiver was closed,
the method would panic.

The branch coverage had been previously missing.

This changeset corrects the assertion
and adds tests covering the code path.

Fixes #7990.
---
 tokio/src/sync/mpsc/chan.rs | 14 +++++---------
 tokio/tests/sync_mpsc.rs    | 24 ++++++++++++++++++++++++
 2 files changed, 29 insertions(+), 9 deletions(-)

diff --git a/tokio/src/sync/mpsc/chan.rs b/tokio/src/sync/mpsc/chan.rs
index 20e2c90d4..7f7f8b16d 100644
--- a/tokio/src/sync/mpsc/chan.rs
+++ b/tokio/src/sync/mpsc/chan.rs
@@ -306,13 +306,11 @@ impl<T, S: Semaphore> Rx<T, S> {
                             return Ready(Some(value));
                         }
                         Some(Read::Closed) => {
-                            // TODO: This check may not be required as it most
-                            // likely can only return `true` at this point. A
-                            // channel is closed when all tx handles are
+                            // A channel is closed when all tx handles are
                             // dropped. Dropping a tx handle releases memory,
                             // which ensures that if dropping the tx handle is
                             // visible, then all messages sent are also visible.
-                            assert!(self.inner.semaphore.is_idle());
+                            debug_assert!(self.inner.semaphore.is_idle());
                             coop.made_progress();
                             return Ready(None);
                         }
@@ -380,13 +378,11 @@ impl<T, S: Semaphore> Rx<T, S> {
                                 if number_added > 0 {
                                     self.inner.semaphore.add_permits(number_added);
                                 }
-                                // TODO: This check may not be required as it most
-                                // likely can only return `true` at this point. A
-                                // channel is closed when all tx handles are
+                                // A channel is closed when all tx handles are
                                 // dropped. Dropping a tx handle releases memory,
                                 // which ensures that if dropping the tx handle is
                                 // visible, then all messages sent are also visible.
-                                assert!(self.inner.semaphore.is_idle());
+                                debug_assert!(self.inner.semaphore.is_idle());
                                 coop.made_progress();
                                 return Ready(number_added);
                             }
@@ -415,7 +411,7 @@ impl<T, S: Semaphore> Rx<T, S> {
             try_recv!();
 
             if rx_fields.rx_closed && self.inner.semaphore.is_idle() {
-                assert!(buffer.is_empty());
+                debug_assert_eq!(buffer.len(), initial_length);
                 coop.made_progress();
                 Ready(0usize)
             } else {
diff --git a/tokio/tests/sync_mpsc.rs b/tokio/tests/sync_mpsc.rs
index 3ebac7390..048b94eb5 100644
--- a/tokio/tests/sync_mpsc.rs
+++ b/tokio/tests/sync_mpsc.rs
@@ -358,6 +358,30 @@ async fn send_recv_many_unbounded_capacity() {
     assert_eq!(expected, buffer);
 }
 
+#[maybe_tokio_test]
+async fn recv_many_with_non_empty_buffer_bounded_rx_closed_and_idle() {
+    let (_tx, mut rx) = mpsc::channel::<i32>(1);
+
+    let mut buffer: Vec<i32> = vec![1];
+
+    rx.close();
+
+    assert_eq!(0, rx.recv_many(&mut buffer, 1).await);
+    assert_eq!(vec![1], buffer);
+}
+
+#[maybe_tokio_test]
+async fn recv_many_with_non_empty_buffer_unbounded_rx_closed_and_idle() {
+    let (_tx, mut rx) = mpsc::unbounded_channel::<i32>();
+
+    let mut buffer: Vec<i32> = vec![1];
+
+    rx.close();
+
+    assert_eq!(0, rx.recv_many(&mut buffer, 1).await);
+    assert_eq!(vec![1], buffer);
+}
+
 #[tokio::test]
 #[cfg(feature = "full")]
 async fn async_send_recv_unbounded() {