sync: return Empty from try_recv() when mpsc is closed with outstanding permits (#8074)

tokio/src/sync/mpsc/chan.rs

@@ -436,7 +436,9 @@ impl<T, S: Semaphore> Rx<T, S> {
                         }
                         TryPopResult::Closed => return Err(TryRecvError::Disconnected),
                         // If close() was called, an empty queue should report Disconnected.
-                        TryPopResult::Empty if rx_fields.rx_closed => {
+                        TryPopResult::Empty
+                            if rx_fields.rx_closed && self.inner.semaphore.is_idle() =>
+                        {
                             return Err(TryRecvError::Disconnected)
                         }
                         TryPopResult::Empty => return Err(TryRecvError::Empty),

tokio/tests/sync_mpsc.rs

@@ -999,6 +999,19 @@ fn try_recv_after_receiver_close() {
     assert_eq!(Err(TryRecvError::Disconnected), rx.try_recv());
 }
 
+#[test]
+fn try_recv_after_receiver_close_with_permit() {
+    let (tx, mut rx) = mpsc::channel::<()>(5);
+
+    let permit = tx.try_reserve().unwrap();
+
+    assert_eq!(Err(TryRecvError::Empty), rx.try_recv());
+    rx.close();
+    assert_eq!(Err(TryRecvError::Empty), rx.try_recv());
+    drop(permit);
+    assert_eq!(Err(TryRecvError::Disconnected), rx.try_recv());
+}
+
 #[test]
 fn try_recv_close_while_empty_bounded() {
     let (tx, mut rx) = mpsc::channel::<()>(5);