diff --git a/CHANGELOG.md b/CHANGELOG.md
index 932d7e36..b1f286f7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -29,7 +29,9 @@ All notable changes to this project will be documented in this file.
 ### Bug fixes
 
 * Release cancelled wait registrations promptly and reclaim fulfilled `Semaphore::reduce_permits` debt nodes.
+* Preserve fan-out notifications when one registered waker panics.
 
 ### Improvements
 
 * Remove the `slab` dependency in favor of a focused internal waiter arena.
+* Describe disconnected channel states consistently in channel error messages.
diff --git a/asyncband/src/barrier/mod.rs b/asyncband/src/barrier/mod.rs
index 0ee1710d..a3c2af73 100644
--- a/asyncband/src/barrier/mod.rs
+++ b/asyncband/src/barrier/mod.rs
@@ -79,6 +79,7 @@ use std::task::Poll;
 use crate::internal::mutex::Mutex;
 use crate::internal::waitset::WaitSet;
 use crate::internal::waitset::WakerToken;
+use crate::internal::waitset::wake_all;
 
 /// A synchronization primitive for multiple tasks that need to wait for each other.
 ///
@@ -245,9 +246,7 @@ impl Barrier {
                 state.generation += 1;
                 let wakers = state.waiters.take_wakers();
                 drop(state);
-                for waker in wakers {
-                    waker.wake();
-                }
+                wake_all(wakers);
                 return BarrierWaitResult(true);
             }
 
diff --git a/asyncband/src/channel/broadcast/mod.rs b/asyncband/src/broadcast/mod.rs
similarity index 100%
rename from asyncband/src/channel/broadcast/mod.rs
rename to asyncband/src/broadcast/mod.rs
diff --git a/asyncband/src/channel/broadcast/mpmc/mod.rs b/asyncband/src/broadcast/mpmc/mod.rs
similarity index 100%
rename from asyncband/src/channel/broadcast/mpmc/mod.rs
rename to asyncband/src/broadcast/mpmc/mod.rs
diff --git a/asyncband/src/channel/broadcast/mpmc/unbounded/mod.rs b/asyncband/src/broadcast/mpmc/unbounded/mod.rs
similarity index 89%
rename from asyncband/src/channel/broadcast/mpmc/unbounded/mod.rs
rename to asyncband/src/broadcast/mpmc/unbounded/mod.rs
index 5eacfe18..9e17089c 100644
--- a/asyncband/src/channel/broadcast/mpmc/unbounded/mod.rs
+++ b/asyncband/src/broadcast/mpmc/unbounded/mod.rs
@@ -25,8 +25,9 @@
 //!
 //! This channel does not impose a capacity limit. A slow or stalled receiver can cause the
 //! buffer to grow without bound, because messages are retained until every active receiver has
-//! consumed them or the receiver is dropped. Use [`UnboundedSender::buffer_len`] to monitor the
-//! number of messages currently retained by the shared buffer.
+//! consumed them or the receiver is dropped. Use
+//! [`UnboundedSender::retained_message_count`] to monitor the number of messages currently retained
+//! by the channel.
 //!
 //! The buffer keeps the capacity a steady workload needs, so a channel that repeatedly fills and
 //! drains does not reallocate. Capacity grown for a one-off burst is released once a later cycle
@@ -80,11 +81,11 @@
 //! // One receiver draining the channel does not discard what the other has not read yet.
 //! assert_eq!(rx1.recv().await, Ok(1));
 //! assert_eq!(rx1.recv().await, Ok(2));
-//! assert_eq!(tx.buffer_len(), 2);
+//! assert_eq!(tx.retained_message_count(), 2);
 //!
 //! assert_eq!(rx2.recv().await, Ok(1));
 //! assert_eq!(rx2.recv().await, Ok(2));
-//! assert_eq!(tx.buffer_len(), 0);
+//! assert_eq!(tx.retained_message_count(), 0);
 //! # }
 //! ```
 
@@ -104,6 +105,7 @@ use crate::internal::arena::SlotId;
 use crate::internal::mutex::Mutex;
 use crate::internal::waitset::WaitSet;
 use crate::internal::waitset::WakerToken;
+use crate::internal::waitset::wake_all;
 
 #[cfg(test)]
 mod tests;
@@ -146,14 +148,14 @@ pub fn unbounded<T: Clone>() -> (UnboundedSender<T>, UnboundedReceiver<T>) {
 /// Error returned by [`UnboundedReceiver::recv`].
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum RecvError {
-    /// The sender has become disconnected, and there will never be any more data received on it.
+    /// All senders have been dropped, and this receiver has no remaining messages.
     Disconnected,
 }
 
 impl fmt::Display for RecvError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
-            RecvError::Disconnected => write!(f, "receiving on a closed channel"),
+            RecvError::Disconnected => write!(f, "receiving on a disconnected channel"),
         }
     }
 }
@@ -163,10 +165,9 @@ impl std::error::Error for RecvError {}
 /// Error returned by [`UnboundedReceiver::try_recv`].
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum TryRecvError {
-    /// This channel is currently empty, but the sender(s) have not yet disconnected, so data may
-    /// yet become available.
+    /// No message is currently available, but at least one sender remains.
     Empty,
-    /// The sender has become disconnected, and there will never be any more data received on it.
+    /// All senders have been dropped, and this receiver has no remaining messages.
     Disconnected,
 }
 
@@ -174,7 +175,7 @@ impl fmt::Display for TryRecvError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             TryRecvError::Empty => write!(f, "receiving on an empty channel"),
-            TryRecvError::Disconnected => write!(f, "receiving on a closed channel"),
+            TryRecvError::Disconnected => write!(f, "receiving on a disconnected channel"),
         }
     }
 }
@@ -364,10 +365,10 @@ struct Shared<T> {
     senders: AtomicUsize,
 }
 
-/// A sender handle to the broadcast channel.
+/// The sending side of an unbounded broadcast channel.
 ///
-/// The sender can be cloned to create multiple producers. When all senders are dropped,
-/// the channel is closed.
+/// The sender can be cloned to create multiple producers. Dropping the final sender disconnects
+/// the channel. Each receiver may drain its own buffered messages before observing disconnection.
 pub struct UnboundedSender<T> {
     shared: Arc<Shared<T>>,
 }
@@ -375,8 +376,8 @@ pub struct UnboundedSender<T> {
 impl<T> Clone for UnboundedSender<T> {
     fn clone(&self) -> Self {
         // Relaxed is enough because this count publishes nothing on its own: receivers read it
-        // only to decide whether the channel is closed, and every message it could hide is
-        // published under `inner`, which a receiver holds before it observes the count.
+        // only to decide whether any sender remains, and every message it could hide is published
+        // under `inner`, which a receiver holds before it observes the count.
         self.shared.senders.fetch_add(1, Ordering::Relaxed);
         Self {
             shared: self.shared.clone(),
@@ -394,12 +395,9 @@ impl<T> Drop for UnboundedSender<T> {
     fn drop(&mut self) {
         match self.shared.senders.fetch_sub(1, Ordering::AcqRel) {
             1 => {
-                // If this is the last sender, we need to wake up the receiver so it can
-                // observe the disconnected state.
+                // Wake every parked receiver so it can observe the channel's disconnected state.
                 let wakers = self.shared.inner.lock().waiters.take_wakers();
-                for waker in wakers {
-                    waker.wake();
-                }
+                wake_all(wakers);
             }
             _ => {
                 // there are still other senders left, do nothing
@@ -461,16 +459,17 @@ impl<T> UnboundedSender<T> {
 
         // Notify all waiting receivers. An unsent message is dropped here too, once the lock is
         // released.
-        for waker in wakers {
-            waker.wake();
-        }
+        wake_all(wakers);
     }
 
-    /// Returns the number of messages currently retained by the shared buffer.
+    /// Returns the number of messages currently retained by the channel.
     ///
     /// This is not the number of messages any single receiver can still read. It is the shared
     /// backlog kept alive by the slowest active receiver.
     ///
+    /// The returned value is an instantaneous snapshot. It is suitable for diagnostics and soft
+    /// flow-control decisions, but concurrent sends and receives may change it immediately.
+    ///
     /// # Examples
     ///
     /// ```
@@ -478,36 +477,15 @@ impl<T> UnboundedSender<T> {
     ///
     /// let (tx, mut rx) = mpmc::unbounded();
     /// tx.send(10);
-    /// assert_eq!(tx.buffer_len(), 1);
+    /// assert_eq!(tx.retained_message_count(), 1);
     ///
     /// assert_eq!(rx.try_recv(), Ok(10));
-    /// assert_eq!(tx.buffer_len(), 0);
+    /// assert_eq!(tx.retained_message_count(), 0);
     /// ```
-    pub fn buffer_len(&self) -> usize {
+    pub fn retained_message_count(&self) -> usize {
         self.shared.inner.lock().buffer.len()
     }
 
-    /// Returns the number of active receivers.
-    ///
-    /// # Examples
-    ///
-    /// ```
-    /// use asyncband::broadcast::mpmc;
-    ///
-    /// let (tx, rx) = mpmc::unbounded::<i32>();
-    /// assert_eq!(tx.receiver_count(), 1);
-    ///
-    /// let rx2 = tx.subscribe();
-    /// assert_eq!(tx.receiver_count(), 2);
-    ///
-    /// drop(rx);
-    /// drop(rx2);
-    /// assert_eq!(tx.receiver_count(), 0);
-    /// ```
-    pub fn receiver_count(&self) -> usize {
-        self.shared.inner.lock().receivers.len()
-    }
-
     /// Creates a new receiver that starts receiving messages from the current tail of the channel.
     ///
     /// # Examples
@@ -536,7 +514,7 @@ impl<T> UnboundedSender<T> {
     }
 }
 
-/// A receiver handle to the broadcast channel.
+/// A receiver for an unbounded broadcast channel.
 ///
 /// Each receiver sees every message sent to the channel while the receiver is active.
 pub struct UnboundedReceiver<T> {
@@ -566,8 +544,8 @@ impl<T: Clone> UnboundedReceiver<T> {
     /// # Returns
     ///
     /// * `Ok(T)`: The next message.
-    /// * `Err(RecvError::Disconnected)`: All senders have been dropped and no more messages are
-    ///   available.
+    /// * `Err(RecvError::Disconnected)`: All senders have been dropped and this receiver has no
+    ///   remaining messages.
     ///
     /// # Cancel safety
     ///
@@ -601,8 +579,8 @@ impl<T: Clone> UnboundedReceiver<T> {
     ///
     /// * `Ok(T)`: The next message.
     /// * `Err(TryRecvError::Empty)`: No message is currently available.
-    /// * `Err(TryRecvError::Disconnected)`: All senders have been dropped and no more messages are
-    ///   available.
+    /// * `Err(TryRecvError::Disconnected)`: All senders have been dropped and this receiver has no
+    ///   remaining messages.
     ///
     /// # Examples
     ///
@@ -689,8 +667,12 @@ impl<T> UnboundedReceiver<T> {
 
     /// Returns the number of messages this receiver can still read.
     ///
-    /// This count is specific to this receiver, unlike [`UnboundedSender::buffer_len`], which
-    /// reports the shared backlog retained by the slowest active receiver.
+    /// This count is specific to this receiver, unlike
+    /// [`UnboundedSender::retained_message_count`], which reports the shared backlog retained by
+    /// the slowest active receiver.
+    ///
+    /// The returned value is an instantaneous snapshot. It is suitable for detecting that this
+    /// receiver is falling behind, but concurrent sends may change it immediately.
     ///
     /// # Examples
     ///
@@ -698,16 +680,16 @@ impl<T> UnboundedReceiver<T> {
     /// use asyncband::broadcast::mpmc;
     ///
     /// let (tx, mut rx) = mpmc::unbounded();
-    /// assert_eq!(rx.len(), 0);
+    /// assert_eq!(rx.unread_message_count(), 0);
     ///
     /// tx.send(10);
     /// tx.send(20);
-    /// assert_eq!(rx.len(), 2);
+    /// assert_eq!(rx.unread_message_count(), 2);
     ///
     /// assert_eq!(rx.try_recv(), Ok(10));
-    /// assert_eq!(rx.len(), 1);
+    /// assert_eq!(rx.unread_message_count(), 1);
     /// ```
-    pub fn len(&self) -> usize {
+    pub fn unread_message_count(&self) -> usize {
         let inner = self.shared.inner.lock();
         let head = *inner
             .receivers
@@ -715,23 +697,6 @@ impl<T> UnboundedReceiver<T> {
             .expect("active broadcast receiver must be registered");
         usize::try_from(inner.tail - head).expect("unread broadcast message count exceeds usize")
     }
-
-    /// Returns `true` if this receiver has no currently available messages.
-    ///
-    /// # Examples
-    ///
-    /// ```
-    /// use asyncband::broadcast::mpmc;
-    ///
-    /// let (tx, rx) = mpmc::unbounded();
-    /// assert!(rx.is_empty());
-    ///
-    /// tx.send(10);
-    /// assert!(!rx.is_empty());
-    /// ```
-    pub fn is_empty(&self) -> bool {
-        self.len() == 0
-    }
 }
 
 struct Recv<'a, T> {
@@ -765,7 +730,7 @@ impl<T: Clone> Future for Recv<'_, T> {
 
         // One critical section decides between all three outcomes. Senders append messages and
         // drain the wait set under this same lock, so registering here cannot miss a wake-up and
-        // cannot observe a closed channel that still has a message for this receiver.
+        // cannot report disconnection while a message remains for this receiver.
         let received = {
             let mut inner = receiver.shared.inner.lock();
 
diff --git a/asyncband/src/channel/broadcast/mpmc/unbounded/tests.rs b/asyncband/src/broadcast/mpmc/unbounded/tests.rs
similarity index 97%
rename from asyncband/src/channel/broadcast/mpmc/unbounded/tests.rs
rename to asyncband/src/broadcast/mpmc/unbounded/tests.rs
index e3f6746b..933ee021 100644
--- a/asyncband/src/channel/broadcast/mpmc/unbounded/tests.rs
+++ b/asyncband/src/broadcast/mpmc/unbounded/tests.rs
@@ -41,7 +41,7 @@ fn one_off_burst_allocation_is_returned_once_it_is_behind_us() {
     }
 
     // Draining evaluates the cycle that just peaked, so the burst allocation is still held.
-    assert_eq!(tx.buffer_len(), 0);
+    assert_eq!(tx.retained_message_count(), 0);
     assert!(tx.shared.inner.lock().buffer.capacity() >= burst);
 
     // The next cycle stays small, which is what releases the memory.
diff --git a/asyncband/src/channel/mod.rs b/asyncband/src/channel/mod.rs
deleted file mode 100644
index c2cdf588..00000000
--- a/asyncband/src/channel/mod.rs
+++ /dev/null
@@ -1,25 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements.  See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership.  The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License.  You may obtain a copy of the License at
-//
-//   http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied.  See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-#[cfg(feature = "broadcast")]
-pub mod broadcast;
-#[cfg(feature = "mpsc")]
-pub mod mpsc;
-#[cfg(feature = "oneshot")]
-pub mod oneshot;
-#[cfg(feature = "watch")]
-pub mod watch;
diff --git a/asyncband/src/internal/countdown.rs b/asyncband/src/internal/countdown.rs
index 2d6faff1..524a7923 100644
--- a/asyncband/src/internal/countdown.rs
+++ b/asyncband/src/internal/countdown.rs
@@ -23,6 +23,7 @@ use std::task::Poll;
 use crate::internal::mutex::Mutex;
 use crate::internal::waitset::WaitSet;
 use crate::internal::waitset::WakerToken;
+use crate::internal::waitset::wake_all;
 
 #[derive(Debug)]
 pub struct CountdownState {
@@ -66,9 +67,7 @@ impl CountdownState {
             waiters.take_wakers()
         };
 
-        for waker in wakers {
-            waker.wake();
-        }
+        wake_all(wakers);
     }
 
     /// Polls for zero, registering the current waker if the countdown is still active.
diff --git a/asyncband/src/internal/waitset.rs b/asyncband/src/internal/waitset.rs
index 8f8cd7ba..ddf8bf2e 100644
--- a/asyncband/src/internal/waitset.rs
+++ b/asyncband/src/internal/waitset.rs
@@ -16,12 +16,42 @@
 // under the License.
 
 use std::mem;
+use std::panic;
+use std::panic::AssertUnwindSafe;
 use std::task::Context;
 use std::task::Waker;
 
 use crate::internal::arena::Arena;
 use crate::internal::arena::SlotId;
 
+/// Wakes every waker while preserving the first panic.
+///
+/// If a wake callback panics, the remaining callbacks are still attempted during unwinding. Any
+/// later panic is suppressed so the first panic can continue to the caller.
+#[inline]
+pub fn wake_all(mut wakers: impl Iterator<Item = Waker>) {
+    struct WakeRemaining<'a, I: Iterator<Item = Waker>> {
+        wakers: &'a mut I,
+    }
+
+    impl<I: Iterator<Item = Waker>> Drop for WakeRemaining<'_, I> {
+        fn drop(&mut self) {
+            // This iterator is empty after normal completion. During unwinding, attempt every
+            // callback left after the one that panicked without replacing the original panic.
+            for waker in self.wakers.by_ref() {
+                let _ = panic::catch_unwind(AssertUnwindSafe(|| waker.wake()));
+            }
+        }
+    }
+
+    let remaining = WakeRemaining {
+        wakers: &mut wakers,
+    };
+    for waker in remaining.wakers.by_ref() {
+        waker.wake();
+    }
+}
+
 /// A single-owner token for a waker registered in a [`WaitSet`].
 ///
 /// This deliberately does not implement `Clone` or `Copy`: duplicating a token could let a stale
@@ -142,6 +172,14 @@ mod tests {
         }
     }
 
+    struct PanicWake;
+
+    impl Wake for PanicWake {
+        fn wake(self: Arc<Self>) {
+            panic!("wake failed");
+        }
+    }
+
     struct DropWake {
         dropped: Arc<AtomicBool>,
         wake_count: AtomicUsize,
@@ -200,6 +238,27 @@ mod tests {
         );
     }
 
+    #[test]
+    fn wake_all_notifies_remaining_waiters_after_a_panic() {
+        let mut waiters = WaitSet::new();
+        let panicking = Waker::from(Arc::new(PanicWake));
+        let tracker = Arc::new(TrackWake(AtomicUsize::new(0)));
+        let tracked = Waker::from(tracker.clone());
+        let mut first = None;
+        let mut second = None;
+        let mut third = None;
+
+        register(&mut waiters, &mut first, &panicking);
+        register(&mut waiters, &mut second, &panicking);
+        register(&mut waiters, &mut third, &tracked);
+
+        let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
+            wake_all(waiters.take_wakers());
+        }));
+        assert!(result.is_err());
+        assert_eq!(tracker.0.load(Ordering::Relaxed), 1);
+    }
+
     #[test]
     fn unregister_returns_the_waker_for_deferred_drop() {
         let mut waiters = WaitSet::new();
diff --git a/asyncband/src/lib.rs b/asyncband/src/lib.rs
index 4ee8c226..7217c775 100644
--- a/asyncband/src/lib.rs
+++ b/asyncband/src/lib.rs
@@ -112,7 +112,6 @@
 //!
 //! While incubation status is not necessarily a reflection of the completeness or stability of the
 //! code, it does indicate that the project has yet to be fully endorsed by the ASF.
-mod channel;
 mod internal;
 
 #[cfg(feature = "barrier")]
@@ -120,13 +119,13 @@ pub mod barrier;
 #[cfg(feature = "blocking")]
 pub mod blocking;
 #[cfg(feature = "broadcast")]
-pub use self::channel::broadcast;
+pub mod broadcast;
 #[cfg(feature = "condvar")]
 pub mod condvar;
 #[cfg(feature = "latch")]
 pub mod latch;
 #[cfg(feature = "mpsc")]
-pub use self::channel::mpsc;
+pub mod mpsc;
 #[cfg(feature = "mutex")]
 pub mod mutex;
 #[cfg(any(
@@ -137,7 +136,7 @@ pub mod mutex;
 ))]
 pub mod once;
 #[cfg(feature = "oneshot")]
-pub use self::channel::oneshot;
+pub mod oneshot;
 #[cfg(feature = "pool")]
 pub mod pool;
 #[cfg(feature = "rwlock")]
@@ -151,7 +150,7 @@ pub mod singleflight;
 #[cfg(feature = "waitgroup")]
 pub mod waitgroup;
 #[cfg(feature = "watch")]
-pub use self::channel::watch;
+pub mod watch;
 
 #[cfg(all(test, any(feature = "once-map", feature = "singleflight")))]
 mod test_support;
diff --git a/asyncband/src/channel/mpsc/bounded.rs b/asyncband/src/mpsc/bounded.rs
similarity index 92%
rename from asyncband/src/channel/mpsc/bounded.rs
rename to asyncband/src/mpsc/bounded.rs
index 08aefd42..c4d4bcf2 100644
--- a/asyncband/src/channel/mpsc/bounded.rs
+++ b/asyncband/src/mpsc/bounded.rs
@@ -94,13 +94,12 @@ impl<T> fmt::Debug for BoundedSender<T> {
 
 impl<T> Drop for BoundedSender<T> {
     fn drop(&mut self) {
-        // drop the sender; this closes the channel if it is the last sender
+        // Dropping the final underlying sender disconnects the channel.
         drop(self.sender.take());
 
         match self.state.senders.fetch_sub(1, Ordering::AcqRel) {
             1 => {
-                // If this is the last sender, we need to wake up the receiver so it can
-                // observe the disconnected state.
+                // Wake the receiver so it can observe the channel's disconnected state.
                 self.state.rx_waker.wake();
             }
             _ => {
@@ -172,12 +171,10 @@ impl<T> BoundedSender<T> {
     ///
     /// This method returns the [`Full`] error if the buffer of the channel is full.
     ///
-    /// This method returns the [`Disconnected`] error if the channel is currently empty, and there
-    /// are no outstanding [receivers].
+    /// This method returns the [`Disconnected`] error if the receiver has been dropped.
     ///
     /// [`Full`]: TrySendError::Full
     /// [`Disconnected`]: TrySendError::Disconnected
-    /// [receivers]: BoundedReceiver
     ///
     /// # Examples
     ///
@@ -287,13 +284,12 @@ impl<T> BoundedReceiver<T> {
 
     /// Receives the next value for this receiver and frees up a space in the buffer if successful.
     ///
-    /// This method returns `Err(RecvError::Disconnected)` if the channel has been closed and there
-    /// are no remaining messages in the channel's buffer. This indicates that no further values
-    /// can ever be received from this `Receiver`. The channel is closed when all senders have been
-    /// dropped.
+    /// This method returns `Err(RecvError::Disconnected)` after all senders have been dropped and
+    /// no buffered messages remain. At that point, this `Receiver` can never receive another
+    /// value.
     ///
-    /// If there are no messages in the channel's buffer, but the channel has not yet been closed,
-    /// this method will sleep until a message is sent or the channel is closed.
+    /// If the buffer is empty while a sender remains, this method sleeps until a message is sent or
+    /// the final sender is dropped.
     ///
     /// # Cancel safety
     ///
diff --git a/asyncband/src/channel/mpsc/error.rs b/asyncband/src/mpsc/error.rs
similarity index 82%
rename from asyncband/src/channel/mpsc/error.rs
rename to asyncband/src/mpsc/error.rs
index cf5d6492..3c10549e 100644
--- a/asyncband/src/channel/mpsc/error.rs
+++ b/asyncband/src/mpsc/error.rs
@@ -18,7 +18,7 @@
 use std::any::type_name;
 use std::fmt;
 
-/// An error returned when trying to send on a closed channel.
+/// An error returned when trying to send on a disconnected channel.
 ///
 /// Returned from [`UnboundedSender::send`] or [`BoundedSender::send`] if the
 /// corresponding [`UnboundedReceiver`] or [`BoundedReceiver`] has already been
@@ -53,7 +53,7 @@ impl<T> SendError<T> {
 
 impl<T> fmt::Display for SendError<T> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.write_str("sending on a closed channel")
+        f.write_str("sending on a disconnected channel")
     }
 }
 
@@ -68,10 +68,9 @@ impl<T> std::error::Error for SendError<T> {}
 /// Error returned by `try_send`.
 #[derive(Clone, PartialEq, Eq)]
 pub enum TrySendError<T> {
-    /// The channel is full, so data may not be sent at this time, but the receiver has not yet
-    /// disconnected.
+    /// The channel is full, so the message cannot be sent without waiting for capacity.
     Full(T),
-    /// The receiver has become disconnected, and there will never be any more data sent on it.
+    /// The receiver has been dropped, so the message can never be received.
     Disconnected(T),
 }
 
@@ -95,7 +94,7 @@ impl<T> fmt::Display for TrySendError<T> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.write_str(match self {
             TrySendError::Full(_) => "sending on a full channel",
-            TrySendError::Disconnected(_) => "sending on a closed channel",
+            TrySendError::Disconnected(_) => "sending on a disconnected channel",
         })
     }
 }
@@ -115,13 +114,13 @@ impl<T> std::error::Error for TrySendError<T> {}
 /// Error returned by `recv`.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum RecvError {
-    /// The sender has become disconnected, and there will never be any more data received on it.
+    /// All senders have been dropped, and no buffered messages remain.
     Disconnected,
 }
 
 impl fmt::Display for RecvError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.write_str("receiving on a closed channel")
+        f.write_str("receiving on a disconnected channel")
     }
 }
 
@@ -130,10 +129,9 @@ impl std::error::Error for RecvError {}
 /// Error returned by `try_recv`.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum TryRecvError {
-    /// This channel is currently empty, but the sender(s) have not yet disconnected, so data may
-    /// yet become available.
+    /// No message is currently available, but at least one sender remains.
     Empty,
-    /// The sender has become disconnected, and there will never be any more data received on it.
+    /// All senders have been dropped, and no buffered messages remain.
     Disconnected,
 }
 
@@ -141,7 +139,7 @@ impl fmt::Display for TryRecvError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.write_str(match self {
             TryRecvError::Empty => "receiving on an empty channel",
-            TryRecvError::Disconnected => "receiving on a closed channel",
+            TryRecvError::Disconnected => "receiving on a disconnected channel",
         })
     }
 }
diff --git a/asyncband/src/channel/mpsc/mod.rs b/asyncband/src/mpsc/mod.rs
similarity index 100%
rename from asyncband/src/channel/mpsc/mod.rs
rename to asyncband/src/mpsc/mod.rs
diff --git a/asyncband/src/channel/mpsc/unbounded.rs b/asyncband/src/mpsc/unbounded.rs
similarity index 91%
rename from asyncband/src/channel/mpsc/unbounded.rs
rename to asyncband/src/mpsc/unbounded.rs
index 3ee892a9..57552d2a 100644
--- a/asyncband/src/channel/mpsc/unbounded.rs
+++ b/asyncband/src/mpsc/unbounded.rs
@@ -88,13 +88,12 @@ impl<T> fmt::Debug for UnboundedSender<T> {
 
 impl<T> Drop for UnboundedSender<T> {
     fn drop(&mut self) {
-        // drop the sender; this closes the channel if it is the last sender
+        // Dropping the final underlying sender disconnects the channel.
         drop(self.sender.take());
 
         match self.state.senders.fetch_sub(1, Ordering::AcqRel) {
             1 => {
-                // If this is the last sender, we need to wake up the receiver so it can
-                // observe the disconnected state.
+                // Wake the receiver so it can observe the channel's disconnected state.
                 self.state.rx_waker.wake();
             }
             _ => {
@@ -186,13 +185,12 @@ impl<T> UnboundedReceiver<T> {
 
     /// Receives the next value for this receiver.
     ///
-    /// This method returns `Err(RecvError::Disconnected)` if the channel has been closed and there
-    /// are no remaining messages in the channel's buffer. This indicates that no further values
-    /// can ever be received from this `Receiver`. The channel is closed when all senders have been
-    /// dropped.
+    /// This method returns `Err(RecvError::Disconnected)` after all senders have been dropped and
+    /// no buffered messages remain. At that point, this `Receiver` can never receive another
+    /// value.
     ///
-    /// If there are no messages in the channel's buffer, but the channel has not yet been closed,
-    /// this method will sleep until a message is sent or the channel is closed.
+    /// If the buffer is empty while a sender remains, this method sleeps until a message is sent or
+    /// the final sender is dropped.
     ///
     /// # Cancel safety
     ///
diff --git a/asyncband/src/channel/oneshot/mod.rs b/asyncband/src/oneshot/mod.rs
similarity index 97%
rename from asyncband/src/channel/oneshot/mod.rs
rename to asyncband/src/oneshot/mod.rs
index 4ee05fa9..cb726c62 100644
--- a/asyncband/src/channel/oneshot/mod.rs
+++ b/asyncband/src/oneshot/mod.rs
@@ -133,8 +133,9 @@ const DISCONNECTED: u8 = 0b010;
 ///   returning to `EMPTY`, or the sender may move to `AWAKING` and take ownership of it. The sender
 ///   retains ownership of any message that it has not yet published.
 /// * `AWAKING`: the sender exclusively owns the published waker and any unpublished message while
-///   it publishes either a message or a disconnect. The receiver must not access either slot;
-///   cancellation may only transfer allocation cleanup to the sender by moving to `DISCONNECTED`.
+///   it publishes either a message or the channel's disconnected state. The receiver must not
+///   access either slot; cancellation may only transfer allocation cleanup to the sender by moving
+///   to `DISCONNECTED`.
 /// * `MESSAGE`: the sender has published an initialized message and no longer accesses the channel.
 ///   The receiver owns the message and the allocation.
 /// * `DISCONNECTED`: no message can subsequently be received. The transition that reaches or
@@ -322,9 +323,9 @@ impl<T> Channel<T> {
         // the initialized waker to the sender.
         let waker = unsafe { self.take_waker() };
 
-        // ORDERING: Release publishes the message or disconnect when this replaces AWAKING. The
-        // RMW's load half is Relaxed; if it reads a receiver-written DISCONNECTED, the conditional
-        // Acquire below completes the reverse allocation-ownership handoff.
+        // ORDERING: Release publishes the message or disconnected state when this replaces
+        // AWAKING. The RMW's load half is Relaxed; if it reads a receiver-written DISCONNECTED, the
+        // conditional Acquire below completes the reverse allocation-ownership handoff.
         let previous_state = self.state.swap(final_state, Ordering::Release);
         if matches!(previous_state, AWAKING) {
             (waker, true)
diff --git a/asyncband/src/channel/oneshot/receiver.rs b/asyncband/src/oneshot/receiver.rs
similarity index 96%
rename from asyncband/src/channel/oneshot/receiver.rs
rename to asyncband/src/oneshot/receiver.rs
index 7d232dc1..f1a0e1e9 100644
--- a/asyncband/src/channel/oneshot/receiver.rs
+++ b/asyncband/src/oneshot/receiver.rs
@@ -81,8 +81,8 @@ impl<T> Receiver<T> {
 
         // ORDERING: Relaxed is sufficient to enforce the method's contract.
         //
-        // Once true has been observed, it will remain true. However, if false is observed,
-        // the sender might have just disconnected but this thread has not observed it yet.
+        // Once true has been observed, it will remain true. However, if false is observed, the
+        // sender might just have been dropped without this thread observing it yet.
         matches!(channel.state.load(Ordering::Relaxed), DISCONNECTED)
     }
 
@@ -156,7 +156,7 @@ impl<T> Drop for Receiver<T> {
         //
         // ORDERING: This is a bidirectional ownership handoff. Release publishes the receiver's
         // last access when the sender must reclaim the allocation; Acquire receives a
-        // sender-published message or disconnect before receiver-side cleanup.
+        // sender-published message or disconnected state before receiver-side cleanup.
         match channel.state.swap(DISCONNECTED, Ordering::AcqRel) {
             // The sender has not sent anything, nor is it dropped. The sender is responsible for
             // deallocating the channel.
@@ -384,10 +384,10 @@ impl<T> Drop for Recv<T> {
 /// Error returned by [`Receiver::try_recv`].
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub enum TryRecvError {
-    /// This channel is currently empty, but the sender has not yet disconnected, so data may yet
-    /// become available.
+    /// No message is currently available, but the sender remains and may still send one.
     Empty,
-    /// The sender has become disconnected, and there will never be any more data received on it.
+    /// No message can become available because the sender was dropped without sending or the
+    /// message has already been received.
     Disconnected,
 }
 
@@ -395,7 +395,7 @@ impl fmt::Display for TryRecvError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.write_str(match self {
             TryRecvError::Empty => "receiving on an empty channel",
-            TryRecvError::Disconnected => "receiving on a closed channel",
+            TryRecvError::Disconnected => "receiving on a disconnected channel",
         })
     }
 }
@@ -409,13 +409,14 @@ impl std::error::Error for TryRecvError {}
 /// `try_recv` calls will return [`TryRecvError::Disconnected`] instead.
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub enum RecvError {
-    /// The sender has become disconnected, and there will never be any more data received on it.
+    /// No message can become available because the sender was dropped without sending or the
+    /// message has already been received.
     Disconnected,
 }
 
 impl fmt::Display for RecvError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.write_str("receiving on a closed channel")
+        f.write_str("receiving on a disconnected channel")
     }
 }
 
diff --git a/asyncband/src/channel/oneshot/sender.rs b/asyncband/src/oneshot/sender.rs
similarity index 92%
rename from asyncband/src/channel/oneshot/sender.rs
rename to asyncband/src/oneshot/sender.rs
index 58a712cf..f43996bf 100644
--- a/asyncband/src/channel/oneshot/sender.rs
+++ b/asyncband/src/oneshot/sender.rs
@@ -55,8 +55,8 @@ impl<T> Sender<T> {
         let channel_ptr = sender.channel_ptr;
 
         // SAFETY: The channel exists on the heap for the entire duration of this method, and we
-        // only ever acquire shared references to it. Note that if the receiver disconnects it
-        // does not free the channel.
+        // only ever acquire shared references to it. Dropping the receiver does not immediately
+        // free the channel.
         let channel = unsafe { channel_ptr.as_ref() };
 
         // Write the message into the channel on the heap.
@@ -106,7 +106,8 @@ impl<T> Sender<T> {
             // Moreover, since we just placed the message in the channel, the channel contains a
             // valid message.
             DISCONNECTED => {
-                // ORDERING: The RMW read DISCONNECTED from the receiver's Release endpoint drop.
+                // ORDERING: The RMW read DISCONNECTED from the receiver's Release-ordered drop
+                // transition.
                 // This Acquire completes the ownership handoff before SendError accesses the
                 // allocation.
                 fence(Ordering::Acquire);
@@ -118,21 +119,21 @@ impl<T> Sender<T> {
 
     /// Returns `true` if the channel is disconnected.
     ///
-    /// This occurs when the associated receiving endpoint is dropped.
+    /// This occurs when the receiver is dropped.
     ///
     /// If `true` is returned, a future call to [`send`](Sender::send) is guaranteed to return an
     /// error.
     pub fn is_disconnected(&self) -> bool {
         // SAFETY: The channel exists on the heap for the entire duration of this method, and we
-        // only ever acquire shared references to it. Note that if the receiver disconnects it
-        // does not free the channel.
+        // only ever acquire shared references to it. Dropping the receiver does not immediately
+        // free the channel.
         let channel = unsafe { self.channel_ptr.as_ref() };
 
         // ORDERING: Relaxed is sufficient for the method's contract: if this returns true, a
         // future call to send is guaranteed to return an error.
         //
-        // Once true has been observed, it will remain true. However, if false is observed,
-        // the receiver might have just disconnected but this thread has not observed it yet.
+        // Once true has been observed, it will remain true. However, if false is observed, the
+        // receiver might just have been dropped without this thread observing it yet.
         matches!(channel.state.load(Ordering::Relaxed), DISCONNECTED)
     }
 
@@ -153,14 +154,14 @@ impl<T> Drop for Sender<T> {
         //   alive, and thus didn't free the channel.
         let channel = unsafe { self.channel_ptr.as_ref() };
 
-        // Disconnect directly, or begin awakening a receiving task:
+        // Publish the channel's disconnected state directly, or begin awakening a receiving task:
         //
         // * EMPTY ^ 001 = DISCONNECTED
         // * RECEIVING ^ 001 = AWAKING
         // * DISCONNECTED ^ 001 = EMPTY (invalid), but this state is never observed
         //
-        // ORDERING: Release publishes a direct disconnect and orders it before the waiting path's
-        // final publication. The RMW's load half is Relaxed, so branches that consume
+        // ORDERING: Release publishes the disconnected state and orders it before the waiting
+        // path's final publication. The RMW's load half is Relaxed, so branches that consume
         // receiver-published resources use an Acquire fence.
         match channel.state.fetch_xor(0b001, Ordering::Release) {
             // The receiver is not waiting, nor is it dropped. The receiver is responsible for
@@ -184,7 +185,8 @@ impl<T> Drop for Sender<T> {
             }
             // The receiver was already dropped. We are responsible for freeing the channel.
             DISCONNECTED => {
-                // ORDERING: The RMW read DISCONNECTED from the receiver's Release endpoint drop.
+                // ORDERING: The RMW read DISCONNECTED from the receiver's Release-ordered drop
+                // transition.
                 // Acquire makes all preceding receiver accesses happen before deallocation.
                 fence(Ordering::Acquire);
                 // SAFETY: when the receiver switches the state to DISCONNECTED they have received
@@ -199,7 +201,7 @@ impl<T> Drop for Sender<T> {
     }
 }
 
-/// An error returned when trying to send on a closed channel. Returned from
+/// An error returned when trying to send on a disconnected channel. Returned from
 /// [`Sender::send`] if the corresponding [`Receiver`] has already been dropped.
 ///
 /// The message that could not be sent can be retrieved again with [`SendError::into_inner`].
@@ -254,7 +256,7 @@ impl<T> Drop for SendError<T> {
 
 impl<T> fmt::Display for SendError<T> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.write_str("sending on a closed channel")
+        f.write_str("sending on a disconnected channel")
     }
 }
 
diff --git a/asyncband/src/channel/oneshot/tests.rs b/asyncband/src/oneshot/tests.rs
similarity index 100%
rename from asyncband/src/channel/oneshot/tests.rs
rename to asyncband/src/oneshot/tests.rs
diff --git a/asyncband/src/channel/watch/error.rs b/asyncband/src/watch/error.rs
similarity index 88%
rename from asyncband/src/channel/watch/error.rs
rename to asyncband/src/watch/error.rs
index 1f0aa050..017d6820 100644
--- a/asyncband/src/channel/watch/error.rs
+++ b/asyncband/src/watch/error.rs
@@ -42,7 +42,7 @@ impl<T> SendError<T> {
 
 impl<T> fmt::Display for SendError<T> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.write_str("sending on a closed channel")
+        f.write_str("sending on a disconnected channel")
     }
 }
 
@@ -54,16 +54,16 @@ impl<T> fmt::Debug for SendError<T> {
 
 impl<T> std::error::Error for SendError<T> {}
 
-/// An error returned when every sender has disconnected and no unseen value remains.
+/// An error returned after all senders have been dropped and no unseen value remains.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum RecvError {
-    /// Every sender has disconnected, so the current value will never change again.
+    /// All senders have been dropped, so the current value will never change again.
     Disconnected,
 }
 
 impl fmt::Display for RecvError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.write_str("receiving on a closed channel")
+        f.write_str("receiving on a disconnected channel")
     }
 }
 
diff --git a/asyncband/src/channel/watch/mod.rs b/asyncband/src/watch/mod.rs
similarity index 91%
rename from asyncband/src/channel/watch/mod.rs
rename to asyncband/src/watch/mod.rs
index 2f2efdae..a5419049 100644
--- a/asyncband/src/channel/watch/mod.rs
+++ b/asyncband/src/watch/mod.rs
@@ -28,7 +28,7 @@
 //! version observed. Because snapshots do not retain the channel's internal lock, they may be kept
 //! or moved independently while senders continue publishing newer values.
 //!
-//! If every sender disconnects after publishing a final unseen value, each receiver can still
+//! If all senders are dropped after publishing a final unseen value, each receiver can still
 //! observe that value once before [`RecvError::Disconnected`] is reported.
 //!
 //! # Examples
@@ -65,6 +65,7 @@ pub use self::error::SendError;
 use crate::internal::mutex::Mutex;
 use crate::internal::waitset::WaitSet;
 use crate::internal::waitset::WakerToken;
+use crate::internal::waitset::wake_all;
 
 /// Creates a watch channel with an initial value.
 ///
@@ -107,7 +108,7 @@ struct State<T> {
     waiters: WaitSet,
 }
 
-/// A sending endpoint of a watch channel.
+/// The sending side of a watch channel.
 pub struct Sender<T> {
     shared: Arc<Shared<T>>,
 }
@@ -136,16 +137,11 @@ impl<T> Drop for Sender<T> {
     fn drop(&mut self) {
         let wakers = {
             let mut state = self.shared.state.lock();
-            state.senders = state
-                .senders
-                .checked_sub(1)
-                .expect("watch sender count underflowed");
+            state.senders -= 1;
             (state.senders == 0 && !state.waiters.is_empty()).then(|| state.waiters.take_wakers())
         };
         if let Some(wakers) = wakers {
-            for waker in wakers {
-                waker.wake();
-            }
+            wake_all(wakers);
         }
     }
 }
@@ -174,9 +170,7 @@ impl<T> Sender<T> {
             (wakers, replaced)
         };
         if let Some(wakers) = wakers {
-            for waker in wakers {
-                waker.wake();
-            }
+            wake_all(wakers);
         }
         drop(replaced);
         Ok(())
@@ -196,14 +190,9 @@ impl<T> Sender<T> {
             seen,
         }
     }
-
-    /// Returns the number of active receivers.
-    pub fn receiver_count(&self) -> usize {
-        self.shared.state.lock().receivers
-    }
 }
 
-/// A receiving endpoint of a watch channel.
+/// A receiver for a watch channel.
 ///
 /// Each receiver independently tracks the latest version it has observed.
 pub struct Receiver<T> {
@@ -235,10 +224,7 @@ impl<T> fmt::Debug for Receiver<T> {
 impl<T> Drop for Receiver<T> {
     fn drop(&mut self) {
         let mut state = self.shared.state.lock();
-        state.receivers = state
-            .receivers
-            .checked_sub(1)
-            .expect("watch receiver count underflowed");
+        state.receivers -= 1;
     }
 }
 
@@ -257,8 +243,8 @@ impl<T> Receiver<T> {
 
     /// Returns whether a version newer than the last observed version exists.
     ///
-    /// An unseen final version is reported before disconnection, even if every sender has already
-    /// been dropped.
+    /// An unseen final version is reported before disconnection, even if all senders have been
+    /// dropped.
     pub fn has_changed(&self) -> Result<bool, RecvError> {
         let state = self.shared.state.lock();
         if state.version != self.seen {
@@ -282,7 +268,7 @@ impl<T> Receiver<T> {
         .await
     }
 
-    /// Returns whether every sender has been dropped.
+    /// Returns whether all senders have been dropped.
     ///
     /// This does not mark the current version observed, so it may return `true` while a final
     /// unseen value is still available through [`Receiver::changed`].
diff --git a/asyncband/src/channel/watch/tests.rs b/asyncband/src/watch/tests.rs
similarity index 72%
rename from asyncband/src/channel/watch/tests.rs
rename to asyncband/src/watch/tests.rs
index 688a4723..3ffa6c2e 100644
--- a/asyncband/src/channel/watch/tests.rs
+++ b/asyncband/src/watch/tests.rs
@@ -25,30 +25,6 @@ fn send_panics_on_version_overflow() {
     tx.send(1).unwrap();
 }
 
-#[test]
-#[should_panic(expected = "watch sender count overflowed")]
-fn sender_clone_panics_on_count_overflow() {
-    let (tx, _rx) = channel(0);
-    tx.shared.state.lock().senders = usize::MAX;
-    let _ = tx.clone();
-}
-
-#[test]
-#[should_panic(expected = "watch receiver count overflowed")]
-fn receiver_clone_panics_on_count_overflow() {
-    let (tx, rx) = channel(0);
-    tx.shared.state.lock().receivers = usize::MAX;
-    let _ = rx.clone();
-}
-
-#[test]
-#[should_panic(expected = "watch receiver count overflowed")]
-fn subscribe_panics_on_count_overflow() {
-    let (tx, _rx) = channel(0);
-    tx.shared.state.lock().receivers = usize::MAX;
-    let _ = tx.subscribe();
-}
-
 #[test]
 fn concurrent_senders_commit_every_version() {
     const SENDERS: usize = 4;
diff --git a/tests-integration/tests/broadcast_mpmc_unbounded_test.rs b/tests-integration/tests/broadcast_mpmc_unbounded_test.rs
index 7617ac4b..5347b34c 100644
--- a/tests-integration/tests/broadcast_mpmc_unbounded_test.rs
+++ b/tests-integration/tests/broadcast_mpmc_unbounded_test.rs
@@ -55,8 +55,7 @@ impl Drop for Reentrant {
     fn drop(&mut self) {
         if let Some(channel) = &self.channel {
             // Deadlocks if the channel still holds its lock while dropping reclaimed messages.
-            let _ = channel.buffer_len();
-            let _ = channel.receiver_count();
+            let _ = channel.retained_message_count();
         }
     }
 }
@@ -147,7 +146,7 @@ fn test_try_recv() {
     assert_eq!(rx.try_recv(), Ok(10));
     assert_eq!(rx.try_recv(), Err(TryRecvError::Empty));
 
-    // Closed
+    // Disconnected
     drop(tx);
     assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected));
 }
@@ -165,33 +164,33 @@ async fn test_slow_receiver_keeps_every_message() {
     for i in 0..1024 {
         assert_eq!(rx1.recv().await, Ok(i));
     }
-    assert_eq!(tx.buffer_len(), 1024);
+    assert_eq!(tx.retained_message_count(), 1024);
 
     for i in 0..1024 {
         assert_eq!(rx2.recv().await, Ok(i));
     }
-    assert_eq!(tx.buffer_len(), 0);
+    assert_eq!(tx.retained_message_count(), 0);
 }
 
 #[tokio::test]
-async fn buffer_len_tracks_the_slowest_receiver() {
+async fn retained_message_count_tracks_the_slowest_receiver() {
     let (tx, mut rx1) = unbounded();
     let mut rx2 = tx.subscribe();
 
     tx.send(1);
     tx.send(2);
-    assert_eq!(tx.buffer_len(), 2);
+    assert_eq!(tx.retained_message_count(), 2);
 
     // Reclaiming waits for the slowest receiver, message by message.
     assert_eq!(rx1.recv().await, Ok(1));
-    assert_eq!(tx.buffer_len(), 2);
+    assert_eq!(tx.retained_message_count(), 2);
     assert_eq!(rx2.recv().await, Ok(1));
-    assert_eq!(tx.buffer_len(), 1);
+    assert_eq!(tx.retained_message_count(), 1);
 
     assert_eq!(rx1.recv().await, Ok(2));
-    assert_eq!(tx.buffer_len(), 1);
+    assert_eq!(tx.retained_message_count(), 1);
     assert_eq!(rx2.recv().await, Ok(2));
-    assert_eq!(tx.buffer_len(), 0);
+    assert_eq!(tx.retained_message_count(), 0);
 }
 
 #[tokio::test]
@@ -205,10 +204,10 @@ async fn test_dropping_a_lagging_receiver_releases_its_backlog() {
     for i in 0..128 {
         assert_eq!(rx1.recv().await, Ok(i));
     }
-    assert_eq!(tx.buffer_len(), 128);
+    assert_eq!(tx.retained_message_count(), 128);
 
     drop(rx2);
-    assert_eq!(tx.buffer_len(), 0);
+    assert_eq!(tx.retained_message_count(), 0);
 }
 
 #[tokio::test]
@@ -219,17 +218,17 @@ async fn resubscribe_keeps_the_original_receivers_backlog() {
     tx.send(2);
 
     let mut rx2 = rx.resubscribe();
-    assert_eq!(tx.buffer_len(), 2);
+    assert_eq!(tx.retained_message_count(), 2);
 
     tx.send(3);
 
     assert_eq!(rx2.recv().await, Ok(3));
-    assert_eq!(tx.buffer_len(), 3);
+    assert_eq!(tx.retained_message_count(), 3);
 
     assert_eq!(rx.recv().await, Ok(1));
     assert_eq!(rx.recv().await, Ok(2));
     assert_eq!(rx.recv().await, Ok(3));
-    assert_eq!(tx.buffer_len(), 0);
+    assert_eq!(tx.retained_message_count(), 0);
 }
 
 #[tokio::test]
@@ -239,7 +238,7 @@ async fn send_without_receivers_does_not_buffer() {
 
     tx.send(1);
     tx.send(2);
-    assert_eq!(tx.buffer_len(), 0);
+    assert_eq!(tx.retained_message_count(), 0);
 
     let mut rx = tx.subscribe();
     assert_eq!(rx.try_recv(), Err(TryRecvError::Empty));
@@ -249,33 +248,27 @@ async fn send_without_receivers_does_not_buffer() {
 }
 
 #[test]
-fn receiver_count_and_len_track_each_receiver() {
+fn unread_message_count_tracks_each_receiver() {
     let (tx, mut rx1) = unbounded();
-    assert_eq!(tx.receiver_count(), 1);
-    assert_eq!(rx1.len(), 0);
-    assert!(rx1.is_empty());
+    assert_eq!(rx1.unread_message_count(), 0);
 
     tx.send(1);
     tx.send(2);
-    assert_eq!(rx1.len(), 2);
-    assert!(!rx1.is_empty());
+    assert_eq!(rx1.unread_message_count(), 2);
 
     let mut rx2 = tx.subscribe();
-    assert_eq!(tx.receiver_count(), 2);
-    assert_eq!(rx2.len(), 0);
-    assert!(rx2.is_empty());
+    assert_eq!(rx2.unread_message_count(), 0);
 
     tx.send(3);
-    assert_eq!(rx1.len(), 3);
-    assert_eq!(rx2.len(), 1);
+    assert_eq!(rx1.unread_message_count(), 3);
+    assert_eq!(rx2.unread_message_count(), 1);
 
     assert_eq!(rx2.try_recv(), Ok(3));
-    assert_eq!(rx2.len(), 0);
+    assert_eq!(rx2.unread_message_count(), 0);
     drop(rx2);
-    assert_eq!(tx.receiver_count(), 1);
 
     assert_eq!(rx1.try_recv(), Ok(1));
-    assert_eq!(rx1.len(), 2);
+    assert_eq!(rx1.unread_message_count(), 2);
 }
 
 #[test]
@@ -331,7 +324,7 @@ fn panicking_clone_leaves_the_channel_consistent() {
     assert_eq!(rx1.try_recv().unwrap().value, 2);
     assert_eq!(rx2.try_recv().unwrap().value, 1);
     assert_eq!(rx2.try_recv().unwrap().value, 2);
-    assert_eq!(tx.buffer_len(), 0);
+    assert_eq!(tx.retained_message_count(), 0);
     assert_eq!(rx1.try_recv().unwrap_err(), TryRecvError::Empty);
 }
 
@@ -477,7 +470,7 @@ fn parked_recv_wakes_when_the_last_sender_drops() {
 }
 
 #[test]
-fn parked_recv_prefers_buffered_messages_over_disconnect() {
+fn parked_recv_prefers_buffered_messages_over_disconnection() {
     let (tx, mut rx) = unbounded();
     let tracker = Arc::new(TrackWake(AtomicUsize::new(0)));
     let waker = Waker::from(tracker);
@@ -495,7 +488,7 @@ fn parked_recv_prefers_buffered_messages_over_disconnect() {
 }
 
 #[tokio::test]
-async fn recv_drains_buffered_messages_before_reporting_disconnect() {
+async fn recv_drains_buffered_messages_before_reporting_disconnection() {
     let (tx, mut rx) = unbounded();
 
     tx.send(1);
@@ -508,7 +501,7 @@ async fn recv_drains_buffered_messages_before_reporting_disconnect() {
 }
 
 #[tokio::test]
-async fn recv_reports_disconnect_without_any_message() {
+async fn recv_reports_disconnection_without_any_message() {
     let (tx, mut rx) = unbounded::<()>();
     drop(tx);
     assert_eq!(rx.recv().await, Err(RecvError::Disconnected));
@@ -593,16 +586,22 @@ fn randomized_operations_track_the_reference_model() {
                 _ => {}
             }
 
-            assert_eq!(tx.receiver_count(), model.len(), "seed {seed}");
             let retained = model
                 .iter()
                 .map(|(_, cursor)| *cursor)
                 .min()
                 .map_or(0, |slowest| tail - slowest);
-            assert_eq!(tx.buffer_len(), retained as usize, "seed {seed}");
+            assert_eq!(
+                tx.retained_message_count(),
+                retained as usize,
+                "seed {seed}"
+            );
             for (receiver, cursor) in &model {
-                assert_eq!(receiver.len(), (tail - cursor) as usize, "seed {seed}");
-                assert_eq!(receiver.is_empty(), *cursor == tail, "seed {seed}");
+                assert_eq!(
+                    receiver.unread_message_count(),
+                    (tail - cursor) as usize,
+                    "seed {seed}"
+                );
             }
         }
     }
diff --git a/tests-integration/tests/mpsc_test.rs b/tests-integration/tests/mpsc_test.rs
index 72256a7a..ed132d36 100644
--- a/tests-integration/tests/mpsc_test.rs
+++ b/tests-integration/tests/mpsc_test.rs
@@ -105,44 +105,44 @@ async fn select_streams() {
 
     let mut rem = true;
     let mut msgs = vec![];
-    let mut rx1_closed = false;
-    let mut rx2_closed = false;
-    let mut rx3_closed = false;
-    let mut rx4_closed = false;
+    let mut rx1_disconnected = false;
+    let mut rx2_disconnected = false;
+    let mut rx3_disconnected = false;
+    let mut rx4_disconnected = false;
 
     while rem {
-        rem = !(rx1_closed && rx2_closed && rx3_closed && rx4_closed);
+        rem = !(rx1_disconnected && rx2_disconnected && rx3_disconnected && rx4_disconnected);
 
         tokio::select! {
-            result = rx1.recv(), if !rx1_closed => {
+            result = rx1.recv(), if !rx1_disconnected => {
                 match result {
                     Ok(x) => msgs.push(x),
-                    Err(RecvError::Disconnected) => rx1_closed = true,
+                    Err(RecvError::Disconnected) => rx1_disconnected = true,
                 }
             }
-            result = rx2.recv(), if !rx2_closed => {
+            result = rx2.recv(), if !rx2_disconnected => {
                 match result {
                     Ok(y) => msgs.push(y),
-                    Err(RecvError::Disconnected) => rx2_closed = true,
+                    Err(RecvError::Disconnected) => rx2_disconnected = true,
                 }
             }
-            result = rx3.recv(), if !rx3_closed => {
+            result = rx3.recv(), if !rx3_disconnected => {
                 match result {
                     Ok(z) => msgs.push(z),
-                    Err(RecvError::Disconnected) => rx3_closed = true,
+                    Err(RecvError::Disconnected) => rx3_disconnected = true,
                 }
             }
-            result = rx4.recv(), if !rx4_closed => {
+            result = rx4.recv(), if !rx4_disconnected => {
                 match result {
                     Ok(w) => msgs.push(w),
-                    Err(RecvError::Disconnected) => rx4_closed = true,
+                    Err(RecvError::Disconnected) => rx4_disconnected = true,
                 }
             }
             else => {
-                rx1_closed = true;
-                rx2_closed = true;
-                rx3_closed = true;
-                rx4_closed = true;
+                rx1_disconnected = true;
+                rx2_disconnected = true;
+                rx3_disconnected = true;
+                rx4_disconnected = true;
             }
         }
     }
@@ -201,7 +201,7 @@ fn try_recv_unbounded() {
 }
 
 #[test]
-fn try_recv_close_while_empty_unbounded() {
+fn try_recv_reports_disconnection_while_empty_unbounded() {
     let (tx, mut rx) = mpsc::unbounded::<()>();
 
     assert_eq!(Err(TryRecvError::Empty), rx.try_recv());
@@ -257,7 +257,7 @@ fn try_send_recv_bounded() {
 }
 
 #[tokio::test]
-async fn try_send_after_close_bounded() {
+async fn try_send_after_disconnection_bounded() {
     let (tx, rx) = mpsc::bounded(1);
 
     tx.try_send(1).unwrap();
@@ -267,7 +267,7 @@ async fn try_send_after_close_bounded() {
 }
 
 #[tokio::test]
-async fn send_after_close_bounded() {
+async fn send_after_disconnection_bounded() {
     let (tx, mut rx) = mpsc::bounded(1);
 
     tx.send(1).await.unwrap();
diff --git a/tests-integration/tests/oneshot_test/main.rs b/tests-integration/tests/oneshot_test/main.rs
index 0cb61520..6f6265c4 100644
--- a/tests-integration/tests/oneshot_test/main.rs
+++ b/tests-integration/tests/oneshot_test/main.rs
@@ -120,7 +120,7 @@ fn dropping_receiver_after_send_drops_message() {
 }
 
 #[test]
-fn dropping_unpolled_recv_closes_channel() {
+fn dropping_unpolled_recv_disconnects_channel() {
     let (sender, receiver) = oneshot::channel::<u128>();
     let receiver = receiver.into_future();
 
@@ -249,7 +249,7 @@ fn poll_with_different_wakers_across_threads() {
 }
 
 #[test]
-fn drop_pending_receiver_closes_channel_and_drops_waker() {
+fn drop_pending_receiver_disconnects_channel_and_drops_waker() {
     let (sender, receiver) = oneshot::channel::<u128>();
     let mut receiver = receiver.into_future();
 
@@ -300,7 +300,7 @@ fn concurrent_send_and_try_recv_to_completion() {
             Ok(999) => true,
             Ok(value) => panic!("unexpected value: {value}"),
             Err(TryRecvError::Empty) => false,
-            Err(TryRecvError::Disconnected) => panic!("unexpected disconnect"),
+            Err(TryRecvError::Disconnected) => panic!("unexpected channel disconnection"),
         });
     });
 
@@ -317,7 +317,7 @@ fn concurrent_drop_sender_and_try_recv_to_completion() {
     let (sender, receiver) = oneshot::channel::<i32>();
 
     let receiver_thread = spawn_named("receiver", move || {
-        spin_until("sender disconnect", || match receiver.try_recv() {
+        spin_until("channel disconnection", || match receiver.try_recv() {
             Ok(value) => panic!("unexpected value: {value}"),
             Err(TryRecvError::Empty) => false,
             Err(TryRecvError::Disconnected) => true,
@@ -367,7 +367,7 @@ fn concurrent_drop_sender_and_poll_to_completion() {
         let (waker, _waker_probe) = WakerProbe::new();
         let mut context = Context::from_waker(&waker);
 
-        spin_until("poll ready with disconnect", || {
+        spin_until("poll ready with disconnection", || {
             match Pin::new(&mut receiver).poll(&mut context) {
                 Poll::Ready(Err(oneshot::RecvError::Disconnected)) => true,
                 Poll::Ready(result) => panic!("unexpected result: {result:?}"),
diff --git a/tests-integration/tests/watch_test.rs b/tests-integration/tests/watch_test.rs
index 3aac616c..fcc38c8e 100644
--- a/tests-integration/tests/watch_test.rs
+++ b/tests-integration/tests/watch_test.rs
@@ -38,6 +38,14 @@ impl Wake for TrackWake {
     }
 }
 
+struct PanicWake;
+
+impl Wake for PanicWake {
+    fn wake(self: Arc<Self>) {
+        panic!("wake failed");
+    }
+}
+
 struct WakeCallback(Mutex<Option<Box<dyn FnOnce() + Send>>>);
 
 impl Wake for WakeCallback {
@@ -70,7 +78,7 @@ struct ReentrantDrop(Option<watch::Sender<ReentrantDrop>>);
 impl Drop for ReentrantDrop {
     fn drop(&mut self) {
         if let Some(sender) = &self.0 {
-            let _ = sender.receiver_count();
+            drop(sender.subscribe());
         }
     }
 }
@@ -168,16 +176,13 @@ fn final_unseen_value_is_reported_before_disconnection() {
 #[test]
 fn sending_without_receivers_returns_the_value_and_preserves_current() {
     let (tx, rx) = watch::channel(String::from("initial"));
-    assert_eq!(tx.receiver_count(), 1);
     drop(rx);
-    assert_eq!(tx.receiver_count(), 0);
 
     let error = tx.send(String::from("unsent")).unwrap_err();
     assert_eq!(error.as_inner(), "unsent");
     assert_eq!(error.into_inner(), "unsent");
 
     let mut replacement = tx.subscribe();
-    assert_eq!(tx.receiver_count(), 1);
     assert_eq!(&*replacement.borrow(), "initial");
     assert_eq!(replacement.has_changed(), Ok(false));
 
@@ -245,6 +250,28 @@ fn one_update_wakes_every_waiting_receiver_once() {
     );
 }
 
+#[test]
+fn panicking_waker_does_not_skip_other_waiters() {
+    let (tx, mut first) = watch::channel(0);
+    let mut second = first.clone();
+    let panicking = Waker::from(Arc::new(PanicWake));
+    let tracker = Arc::new(TrackWake(AtomicUsize::new(0)));
+    let tracked = Waker::from(tracker.clone());
+    let mut first_changed = Box::pin(first.changed());
+    let mut second_changed = Box::pin(second.changed());
+
+    assert!(poll_with(first_changed.as_mut(), &panicking).is_pending());
+    assert!(poll_with(second_changed.as_mut(), &tracked).is_pending());
+
+    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| tx.send(1)));
+    assert!(result.is_err());
+    assert_eq!(tracker.0.load(Ordering::Relaxed), 1);
+    assert_eq!(
+        poll_with(second_changed.as_mut(), &tracked),
+        Poll::Ready(Ok(Arc::new(1)))
+    );
+}
+
 #[test]
 fn only_the_last_sender_drop_wakes_a_waiter() {
     let (tx, mut rx) = watch::channel(());
@@ -316,7 +343,7 @@ fn wake_callbacks_run_outside_the_channel_lock() {
         let callback_sender = tx.clone();
         let waker = Waker::from(Arc::new(WakeCallback(Mutex::new(Some(Box::new(
             move || {
-                let _ = callback_sender.receiver_count();
+                drop(callback_sender.subscribe());
             },
         ))))));
         let mut changed = Box::pin(rx.changed());
@@ -359,7 +386,7 @@ fn replaced_wakers_are_dropped_outside_the_channel_lock() {
         let callback_sender = tx.clone();
         let old_waker = Waker::from(Arc::new(DropCallbackWake(Mutex::new(Some(Box::new(
             move || {
-                let _ = callback_sender.receiver_count();
+                drop(callback_sender.subscribe());
             },
         ))))));
         let mut changed = Box::pin(rx.changed());