tokio 1.53.1: Multiple notify_one calls on Tokio Notify store at most one permit when no waiters are present, while notify_waiters stores zero permits and wakes all registered waiters simultaneously

# This file is automatically @generated by Cargo.
# It is not intended for manual editing.
version = 4

[[package]]
name = "pin-project-lite"
version = "0.2.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"

[[package]]
name = "tokio"
version = "1.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed"
dependencies = [
 "pin-project-lite",
]

[[package]]
name = "tokio-sample"
version = "0.1.0"
dependencies = [
 "tokio",
]

[package]
name = "tokio-sample"
version = "0.1.0"
edition = "2021"

[dependencies]
tokio = { version = "1.53.1", default-features = false, features = ["sync"] }

# Tokio Notify Permit Saturation and Waiter Notification

### `search_known_solution` Result
`search_known_solution` returned existing samples covering other Tokio components:
- `sha256:8345acff9638c570b383f63d50f7d0ae8c2559955a43fd5ce4514722f7b91551` (Tokio broadcast channel lag and zero-receiver drop)
- `sha256:05d1890d1e0cc4aff8802744fc7c937ab95762e3615db73584bdff49b69fa791` (Tokio watch channel change tracking)
- `sha256:1ff6e5d349fd30f10ae07ff7eecbcd551ea33279ba3e1ed6b5e2a0c68f4b72d6` (Tokio MPSC receiver close with in-flight sender permits)
- `sha256:3114583fe7d663d0553a5731690052f41f1660c66158047230a4198a72e13ca9` (Tokio Runtime::Builder default drivers)
- `sha256:9b9720fef342edf7ef51c61e4781caeafe87bc3863b3fdfe9efd67fdb0a2caf8` (Tokio non-Send futures with LocalSet)

None covered `tokio::sync::Notify` permit absorption and waiter dispatch semantics.

### Pinned Release
- `tokio` 1.53.1 (`pkg:cargo/tokio@1.53.1`)

### Observed Behaviour
- Calling `notify_one()` on `tokio::sync::Notify` when no tasks are actively waiting stores at most one permit. Multiple consecutive `notify_one()` calls (e.g. 3 times) do not accumulate multiple permits: only the first subsequent `notified().await` resolves immediately, while any following `notified().await` calls remain `Poll::Pending`.
- Calling `notify_waiters()` when no tasks are actively waiting stores zero permits; subsequent `notified()` futures remain `Poll::Pending`.
- When multiple tasks are registered and waiting, `notify_waiters()` unparks all registered waiters concurrently, whereas `notify_one()` unparks exactly one registered waiter and leaves the others pending.

{"case":{"caseId":"case:sha256:ccec3e857262e6ebce5e8b370d07521f98f13790614176888c535167d7f160e1","contract":["Calling notify_one multiple times on Tokio Notify when no tasks are waiting stores at most one permit, so only the first subsequent notified future completes immediately while subsequent futures remain pending","Calling notify_waiters when no tasks are waiting stores zero permits, leaving subsequent notified futures pending","Calling notify_waiters wakes all currently registered waiting futures simultaneously","Calling notify_one wakes exactly one registered waiting future while leaving other registered waiters pending"],"goal":"Multiple notify_one calls on Tokio Notify store at most one permit when no waiters are present, while notify_waiters stores zero permits and wakes all registered waiters simultaneously","kind":"HOW","packages":["pkg:cargo/tokio@1.53.1"],"schemaVersion":1,"symbols":["tokio::sync::Notify::notify_one","tokio::sync::Notify::notified","tokio::sync::Notify::notify_waiters"]},"contractCommand":["cargo","test","--offline"],"environment":{"arch":"x64","ecosystem":"cargo","executionContext":"rust","language":"rust","os":"linux","packageManager":"cargo","runtime":"rust","schemaVersion":1},"license":"MIT-0","packages":["pkg:cargo/tokio@1.53.1"],"schemaVersion":1,"symbols":["tokio::sync::Notify::notify_one","tokio::sync::Notify::notified","tokio::sync::Notify::notify_waiters"],"verifierAdapter":"cargo@1"}

use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
use tokio::sync::Notify;

/// A lightweight waker that records the count of wake notifications.
struct WakeCounter {
    count: AtomicUsize,
}

impl WakeCounter {
    fn new() -> Arc<Self> {
        Arc::new(Self {
            count: AtomicUsize::new(0),
        })
    }

    fn waker(self: &Arc<Self>) -> Waker {
        fn clone_raw(data: *const ()) -> RawWaker {
            let arc = unsafe { Arc::from_raw(data as *const WakeCounter) };
            let cloned = Arc::clone(&arc);
            let _ = Arc::into_raw(arc);
            RawWaker::new(Arc::into_raw(cloned) as *const (), &VTABLE)
        }

        fn wake_raw(data: *const ()) {
            let arc = unsafe { Arc::from_raw(data as *const WakeCounter) };
            arc.count.fetch_add(1, Ordering::SeqCst);
        }

        fn wake_by_ref_raw(data: *const ()) {
            let arc = unsafe { Arc::from_raw(data as *const WakeCounter) };
            arc.count.fetch_add(1, Ordering::SeqCst);
            let _ = Arc::into_raw(arc);
        }

        fn drop_raw(data: *const ()) {
            unsafe { Arc::from_raw(data as *const WakeCounter) };
        }

        static VTABLE: RawWakerVTable =
            RawWakerVTable::new(clone_raw, wake_raw, wake_by_ref_raw, drop_raw);

        let cloned = Arc::clone(self);
        unsafe { Waker::from_raw(RawWaker::new(Arc::into_raw(cloned) as *const (), &VTABLE)) }
    }
}

/// Helper to poll a pinned future once with a custom waker.
fn poll_future<F: Future>(future: Pin<&mut F>, waker: &Waker) -> Poll<F::Output> {
    let mut cx = Context::from_waker(waker);
    future.poll(&mut cx)
}

/// Proves that `Notify::notify_one` stores at most 1 permit when no waiters are present,
/// that `Notify::notify_waiters` stores zero permits when no waiters are present,
/// and that `notify_waiters` wakes all registered waiters while `notify_one` wakes exactly one.
pub fn verify_notify_permit_and_waiter_semantics() {
    // 1. Multiple notify_one calls without waiters do not accumulate multiple permits:
    // Calling notify_one 3 times stores exactly 1 permit.
    let notify = Notify::new();
    notify.notify_one();
    notify.notify_one();
    notify.notify_one();

    let counter = WakeCounter::new();
    let waker = counter.waker();

    // First notified() future immediately consumes the single stored permit
    let mut fut1 = Box::pin(notify.notified());
    assert_eq!(
        poll_future(fut1.as_mut(), &waker),
        Poll::Ready(()),
        "first notified() must consume the single stored permit immediately"
    );

    // Second notified() future cannot resolve because only 1 permit was stored (no stacking)
    let mut fut2 = Box::pin(notify.notified());
    assert_eq!(
        poll_future(fut2.as_mut(), &waker),
        Poll::Pending,
        "second notified() must remain Pending because notify_one does not accumulate multiple permits"
    );

    // 2. notify_waiters without waiters stores zero permits
    let notify_no_waiters = Notify::new();
    notify_no_waiters.notify_waiters();

    let mut fut_unnotified = Box::pin(notify_no_waiters.notified());
    assert_eq!(
        poll_future(fut_unnotified.as_mut(), &waker),
        Poll::Pending,
        "notify_waiters without active waiters must not store any permit"
    );

    // 3. notify_waiters wakes all registered waiters simultaneously
    let notify_multi = Notify::new();
    let counter_a = WakeCounter::new();
    let counter_b = WakeCounter::new();
    let waker_a = counter_a.waker();
    let waker_b = counter_b.waker();

    let mut fut_a = Box::pin(notify_multi.notified());
    let mut fut_b = Box::pin(notify_multi.notified());

    // Register both waiters
    assert_eq!(poll_future(fut_a.as_mut(), &waker_a), Poll::Pending);
    assert_eq!(poll_future(fut_b.as_mut(), &waker_b), Poll::Pending);

    // notify_waiters notifies both registered tasks
    notify_multi.notify_waiters();
    assert_eq!(counter_a.count.load(Ordering::SeqCst), 1);
    assert_eq!(counter_b.count.load(Ordering::SeqCst), 1);

    assert_eq!(
        poll_future(fut_a.as_mut(), &waker_a),
        Poll::Ready(()),
        "waiter A must be ready after notify_waiters"
    );
    assert_eq!(
        poll_future(fut_b.as_mut(), &waker_b),
        Poll::Ready(()),
        "waiter B must be ready after notify_waiters"
    );

    // 4. notify_one wakes exactly one registered waiter
    let notify_single = Notify::new();
    let counter_1 = WakeCounter::new();
    let counter_2 = WakeCounter::new();
    let waker_1 = counter_1.waker();
    let waker_2 = counter_2.waker();

    let mut waiter_1 = Box::pin(notify_single.notified());
    let mut waiter_2 = Box::pin(notify_single.notified());

    assert_eq!(poll_future(waiter_1.as_mut(), &waker_1), Poll::Pending);
    assert_eq!(poll_future(waiter_2.as_mut(), &waker_2), Poll::Pending);

    notify_single.notify_one();
    assert_eq!(counter_1.count.load(Ordering::SeqCst), 1);
    assert_eq!(counter_2.count.load(Ordering::SeqCst), 0);

    assert_eq!(
        poll_future(waiter_1.as_mut(), &waker_1),
        Poll::Ready(()),
        "first waiter must be ready after notify_one"
    );
    assert_eq!(
        poll_future(waiter_2.as_mut(), &waker_2),
        Poll::Pending,
        "second waiter must remain Pending after notify_one"
    );
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_notify_permit_and_waiter_semantics() {
        verify_notify_permit_and_waiter_semantics();
    }
}