refactor: simplify waker notification paths

AGENTS.md

@@ -35,7 +35,8 @@ Declare restricted visibility at the module boundary and use `pub` for items in
 
 - Allow normal executor `Waker::clone` inside short state critical sections. Custom clone panic recovery and reentrancy are not general guarantees; do not require them in reviews unless an explicit local contract does.
 - Reuse borrowed-waker registration and avoid redundant clones.
-- Keep wake callbacks and replaced or cancelled waker destruction outside primitive locks. If a batch wake panics, attempt the remaining wakes and propagate the first panic.
+- Keep wake callbacks and replaced or cancelled waker destruction outside primitive locks.
+- Executor waker operations, including cloning, waking, and dropping, are expected not to panic; do not add custom panic recovery.
 
 Decision: [#257](https://github.com/apache/asyncband/pull/257).
 

CHANGELOG.md

@@ -27,6 +27,10 @@ All notable changes to this project will be documented in this file.
 
 * Add bounded MPMC `reserve` and `try_reserve` methods returning a borrowed `Permit`, so callers can wait for capacity before constructing a value; sends and reservations receive capacity in wait-queue order, and unused permits release it.
 
+### Notable changes
+
+* Executor waker operations, including cloning, waking, and dropping, are expected not to panic; recovery from panicking waker callbacks is no longer supported.
+
 ## v0.7.3 (2026-09-29)
 
 ### New features

asyncband/src/completion/mod.rs

@@ -121,11 +121,6 @@ impl<T> Completer<T> {
     ///
     /// Returns `value` if all observers were already dropped. A successful completion does not
     /// guarantee that an observer will remain alive long enough to read the value.
-    ///
-    /// # Panics
-    ///
-    /// Panics if notifying a waiting observer panics. The value remains committed, and notification
-    /// is still attempted for every other pending observer before the panic resumes.
     pub fn complete(mut self, value: T) -> Result<(), T> {
         let Some(shared) = self.shared.upgrade() else {
             return Err(value);

asyncband/src/event/auto_reset.rs

@@ -102,11 +102,6 @@ impl AutoResetEvent {
     /// A stored signal is available to a future wait. Further sets coalesce while it remains
     /// unassigned. Creating a wait future does not register it; registration happens when it is
     /// first polled without a stored signal.
-    ///
-    /// # Panics
-    ///
-    /// Panics if waking a selected task panics. Its signal remains assigned and can still be
-    /// consumed by polling that wait or passed on by dropping it.
     pub fn set(&self) {
         let waker = self.state.lock().signal();
         if let Some(waker) = waker {

asyncband/src/event/manual_reset.rs

@@ -97,11 +97,6 @@ impl ManualResetEvent {
     ///
     /// The event remains set until [`reset`](Self::reset) is called. Calling `set` while it is
     /// already set has no effect.
-    ///
-    /// # Panics
-    ///
-    /// Panics if waking a selected task panics. The event remains set, and waking is still
-    /// attempted for every other selected task before the panic resumes.
     pub fn set(&self) {
         let mut wakers = WakerBatch::new();
         {

asyncband/src/internal/mod.rs

@@ -15,8 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::panic;
-use std::panic::AssertUnwindSafe;
 use std::task::Waker;
 
 /// Retains the current task waker, returning any replaced registration for unlocked destruction.
@@ -35,32 +33,12 @@ pub fn register_waker(slot: &mut Option<Waker>, waker: &Waker) -> Option<Waker>
     }
 }
 
-/// 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.
+/// Wakes every waker.
 #[inline]
 // A no-feature or blocking-only build has no primitive that fans notifications out.
 #[allow(dead_code)]
-pub(crate) 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() {
+pub(crate) fn wake_all(wakers: impl Iterator<Item = Waker>) {
+    for waker in wakers {
         waker.wake();
     }
 }

asyncband/src/internal/semaphore.rs

@@ -162,88 +162,62 @@ impl Semaphore {
     pub fn notify_all(&self) {
         let mut waiters = self.waiters.lock();
         let mut wakers = WakerBatch::new();
-        loop {
-            match waiters.unlink_first_waiter(|node| {
-                node.permits = 0;
-                true
-            }) {
-                None => break,
-                Some((id, waiter)) => {
-                    let remove_now = waiter.waker.is_none();
-                    if let Some(waker) = waiter.waker.take() {
-                        wakers.push(waker);
-                    }
-                    if remove_now {
-                        waiters.remove_unlinked_waiter(id);
-                    }
-                }
+        while let Some((id, waiter)) = waiters.unlink_first_waiter(|node| {
+            node.permits = 0;
+            true
+        }) {
+            if let Some(waker) = waiter.waker.take() {
+                wakers.push(waker);
+            } else {
+                waiters.remove_unlinked_waiter(id);
             }
         }
         drop(waiters);
         wake_all(&mut wakers);
     }
 
-    fn insert_permits_with_lock(
-        &self,
+    fn insert_permits_with_lock<'a>(
+        &'a self,
         mut rem: usize,
-        waiters: MutexGuard<'_, WaitList<WaitNode>>,
+        mut waiters: MutexGuard<'a, WaitList<WaitNode>>,
     ) {
         let mut batch = WakerBatch::new();
-        let mut lock = Some(waiters);
-
-        // One iterator covers the entire release. If a callback panics, `wake_all` keeps pulling
-        // batches during unwinding, so the remaining permits are still distributed and notified.
-        wake_all(std::iter::from_fn(|| {
-            loop {
-                if let Some(waker) = batch.next() {
-                    return Some(waker);
-                }
-                if rem == 0 {
-                    return None;
-                }
 
-                let mut waiters = lock.take().unwrap_or_else(|| self.waiters.lock());
-                while !batch.will_spill() {
-                    match waiters.unlink_first_waiter(|node| {
-                        if node.permits <= rem {
-                            rem -= node.permits;
-                            node.permits = 0;
-                            true
-                        } else {
-                            node.permits -= rem;
-                            rem = 0;
-                            false
-                        }
-                    }) {
-                        None => break,
-                        Some((id, waiter)) => {
-                            let remove_now = waiter.waker.is_none();
-                            if let Some(waker) = waiter.waker.take() {
-                                batch.push(waker);
-                            }
-                            if remove_now {
-                                waiters.remove_unlinked_waiter(id);
-                            }
-                        }
-                    }
+        loop {
+            while rem > 0 && !batch.will_spill() {
+                let Some((id, waiter)) = waiters.unlink_first_waiter(|node| {
+                    let acquired = node.permits.min(rem);
+                    node.permits -= acquired;
+                    rem -= acquired;
+                    node.permits == 0
+                }) else {
+                    break;
+                };
+                if let Some(waker) = waiter.waker.take() {
+                    batch.push(waker);
+                } else {
+                    waiters.remove_unlinked_waiter(id);
                 }
+            }
 
-                if rem > 0 && waiters.is_empty() {
-                    // Retire the remainder before the overflow check so unwinding cannot retry it.
-                    let added = std::mem::take(&mut rem);
-                    // The lock serializes additions; concurrent operations can only remove permits.
-                    let current = self.permits.load(Ordering::Relaxed);
-                    assert!(
-                        current.checked_add(added).is_some(),
-                        "number of added permits ({added}) would overflow usize::MAX (prev: {current})"
-                    );
-                    self.permits.fetch_add(added, Ordering::Release);
-                }
+            if rem > 0 && waiters.is_empty() {
+                // The lock serializes additions; concurrent operations can only remove permits.
+                let current = self.permits.load(Ordering::Relaxed);
+                assert!(
+                    current.checked_add(rem).is_some(),
+                    "number of added permits ({rem}) would overflow usize::MAX (prev: {current})"
+                );
+                self.permits.fetch_add(rem, Ordering::Release);
+                rem = 0;
+            }
 
-                // Neither wake callbacks nor destruction of the taken waker run under this lock.
-                drop(waiters);
+            drop(waiters);
+            wake_all(&mut batch);
+            if rem == 0 {
+                return;
             }
-        }));
+            waiters = self.waiters.lock();
+        }
     }
 }
 
@@ -411,8 +385,6 @@ fn acquired_or_enqueue(
 
 #[cfg(test)]
 mod tests {
-    use std::panic;
-    use std::panic::AssertUnwindSafe;
     use std::sync::Arc;
     use std::sync::atomic::AtomicUsize;
     use std::sync::atomic::Ordering;
@@ -442,87 +414,35 @@ mod tests {
     }
 
     #[test]
-    fn release_distributes_permits_to_all_waiters() {
-        const WAITER_COUNT: usize = 35;
+    fn release_distributes_permits_across_multiple_batches() {
+        const WAITER_COUNT: usize = WakerBatch::STACK_SIZE * 2 + 1;
 
         let semaphore = Semaphore::new(0);
-        let counter = Arc::new(WakeCounter(AtomicUsize::new(0)));
-        let waker = Waker::from(counter.clone());
+        let counters = (0..WAITER_COUNT)
+            .map(|_| Arc::new(WakeCounter(AtomicUsize::new(0))))
+            .collect::<Vec<_>>();
+        let wakers = counters
+            .iter()
+            .map(|counter| Waker::from(counter.clone()))
+            .collect::<Vec<_>>();
         let mut acquires = (0..WAITER_COUNT)
             .map(|_| semaphore.poll_acquire(1))
             .collect::<Vec<_>>();
 
-        for acquire in &mut acquires {
-            assert!(acquire.poll_once(&waker).is_pending());
+        for (acquire, waker) in acquires.iter_mut().zip(&wakers) {
+            assert!(acquire.poll_once(waker).is_pending());
         }
         assert_eq!(semaphore.waiters.lock().occupied_len(), WAITER_COUNT);
 
-        semaphore.release(WAITER_COUNT);
-        assert_eq!(counter.0.load(Ordering::Relaxed), WAITER_COUNT);
-
-        for acquire in &mut acquires {
-            assert!(acquire.poll_once(&waker).is_ready());
-        }
-        assert_eq!(semaphore.waiters.lock().occupied_len(), 0);
-    }
-
-    #[test]
-    fn panicking_wakes_preserve_permits_and_the_first_panic() {
-        const WAITER_COUNT: usize = 65;
-
-        struct TrackedWake {
-            count: AtomicUsize,
-            panic_message: Option<&'static str>,
-        }
-
-        impl Wake for TrackedWake {
-            fn wake(self: Arc<Self>) {
-                self.count.fetch_add(1, Ordering::Relaxed);
-                if let Some(message) = self.panic_message {
-                    panic::panic_any(message);
-                }
-            }
+        semaphore.release(WAITER_COUNT + 2);
+        for counter in counters {
+            assert_eq!(counter.0.load(Ordering::Relaxed), 1);
         }
+        assert_eq!(semaphore.available_permits(), 2);
 
-        let semaphore = Semaphore::new(0);
-        let trackers = (0..WAITER_COUNT)
-            .map(|index| {
-                Arc::new(TrackedWake {
-                    count: AtomicUsize::new(0),
-                    panic_message: if index == 0 {
-                        Some("first wake panic")
-                    } else if index == WAITER_COUNT / 2 {
-                        Some("later wake panic")
-                    } else {
-                        None
-                    },
-                })
-            })
-            .collect::<Vec<_>>();
-        let mut acquires = trackers
-            .iter()
-            .map(|tracker| {
-                let mut acquire = semaphore.poll_acquire(1);
-                assert!(
-                    acquire
-                        .poll_once(&Waker::from(tracker.clone()))
-                        .is_pending()
-                );
-                acquire
-            })
-            .collect::<Vec<_>>();
-
-        let payload = panic::catch_unwind(AssertUnwindSafe(|| semaphore.release(WAITER_COUNT + 2)))
-            .expect_err("the original wake panic must reach the caller");
-        assert_eq!(payload.downcast_ref::<&str>(), Some(&"first wake panic"));
-
-        for tracker in trackers {
-            assert_eq!(tracker.count.load(Ordering::Relaxed), 1);
-        }
         for acquire in &mut acquires {
             assert!(acquire.poll_once(Waker::noop()).is_ready());
         }
-        assert_eq!(semaphore.available_permits(), 2);
         assert_eq!(semaphore.waiters.lock().occupied_len(), 0);
     }
 }

asyncband/src/lib.rs

@@ -93,6 +93,9 @@
 //! remain with the caller. Await Asyncband futures inside any executor that polls standard Rust
 //! futures, and compose those runtime services around them.
 //!
+//! Executor waker operations, including cloning, waking, and dropping, are expected not to panic.
+//! Recovery from panicking waker operations is not part of the public API contract.
+//!
 //! # Async first, blocking by adaptation
 //!
 //! Async and synchronous primitives have different optimization constraints. Asyncband designs its