sync: fix incorrect is_empty on mpsc block boundaries (#6603)

tokio/src/sync/mpsc/block.rs

@@ -168,14 +168,17 @@ impl<T> Block<T> {
         Some(Read::Value(value.assume_init()))
     }
 
-    /// Returns true if there is a value in the slot to be consumed
+    /// Returns true if *this* block has a value in the given slot.
     ///
-    /// # Safety
+    /// Always returns false when given an index from a different block.
-    ///
-    /// To maintain safety, the caller must ensure:
-    ///
-    /// * No concurrent access to the slot.
     pub(crate) fn has_value(&self, slot_index: usize) -> bool {
+        if slot_index < self.header.start_index {
+            return false;
+        }
+        if slot_index >= self.header.start_index + super::BLOCK_CAP {
+            return false;
+        }
+
         let offset = offset(slot_index);
         let ready_bits = self.header.ready_slots.load(Acquire);
         is_ready(ready_bits, offset)

tokio/tests/sync_mpsc.rs

@@ -1421,4 +1421,16 @@ async fn test_rx_unbounded_len_when_close_is_called_after_dropping_sender() {
     assert_eq!(rx.len(), 1);
 }
 
+// Regression test for https://github.com/tokio-rs/tokio/issues/6602
+#[tokio::test]
+async fn test_is_empty_32_msgs() {
+    let (sender, mut receiver) = mpsc::channel(33);
+
+    for value in 1..257 {
+        sender.send(value).await.unwrap();
+        receiver.recv().await.unwrap();
+        assert!(receiver.is_empty(), "{value}. len: {}", receiver.len());
+    }
+}
+
 fn is_debug<T: fmt::Debug>(_: &T) {}