ntex-rt 4.5.0

use std::pin::Pin;
use std::sync::atomic::{AtomicBool, AtomicUsize};
use std::sync::{Mutex, mpsc};
use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
use std::{cell::Cell, cell::RefCell, future::poll_fn, rc::Rc, thread};

use super::*;

/// Iterations of a stress test, fewer under Miri.
const fn iters(n: usize, miri: usize) -> usize {
    if cfg!(miri) { miri } else { n }
}

/// Wall clock limit of a wait, Miri runs much slower.
const WAIT: Duration = Duration::from_secs(if cfg!(miri) { 600 } else { 30 });

#[derive(Debug)]
struct NoopNotify;

impl Notify for NoopNotify {
    fn notify(&self) -> io::Result<()> {
        Ok(())
    }
}

struct WakeOnDrop(Rc<RefCell<Option<Waker>>>);

impl Drop for WakeOnDrop {
    fn drop(&mut self) {
        if let Some(w) = self.0.borrow_mut().take() {
            w.wake();
        }
    }
}

#[derive(Debug, Default)]
struct CountNotify(Arc<std::sync::atomic::AtomicUsize>);

impl Notify for CountNotify {
    fn notify(&self) -> io::Result<()> {
        self.0.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        Ok(())
    }
}

#[derive(Debug)]
struct FailNotify {
    calls: Arc<AtomicUsize>,
    failures: usize,
    kind: io::ErrorKind,
}

impl Notify for FailNotify {
    fn notify(&self) -> io::Result<()> {
        if self.calls.fetch_add(1, Ordering::Relaxed) < self.failures {
            Err(self.kind.into())
        } else {
            Ok(())
        }
    }
}

struct PollDriver;

impl Driver for PollDriver {
    fn handle(&self) -> Box<dyn Notify> {
        Box::new(NoopNotify)
    }

    fn run(&self, rt: &Runtime) -> io::Result<()> {
        while rt.poll() != PollResult::Ready {}
        Ok(())
    }
}

/// STATEMENT: Scheduler notifications retry interruptions and coalesce queued work.
///
/// HOW:
/// - Submit pairs of tasks locally and remotely with the first two notifications interrupted.
/// - Check three initial calls, one call next round, and both tasks completed per poll.
#[test]
fn interrupted_scheduler_notifications() {
    for remote in [false, true] {
        let calls = Arc::new(AtomicUsize::new(0));
        let done = Arc::new(AtomicUsize::new(0));
        let rt = Runtime::new(Box::new(FailNotify {
            calls: calls.clone(),
            failures: 2,
            kind: io::ErrorKind::Interrupted,
        }));
        for round in 0..2 {
            let (handle, done2) = (rt.handle(), done.clone());
            let submit = move || {
                for _ in 0..2 {
                    let done = done2.clone();
                    handle.spawn_detached(async move {
                        done.fetch_add(1, Ordering::Relaxed);
                    });
                }
            };
            if remote {
                thread::spawn(submit).join().unwrap();
            } else {
                submit();
            }
            assert_eq!(calls.load(Ordering::Relaxed), 3 + round);
            assert_eq!(rt.poll(), PollResult::Pending);
            assert_eq!(done.load(Ordering::Relaxed), 2 * (round + 1));
        }
    }
}

/// STATEMENT: The `block_on` completion notification retries interrupted driver calls.
///
/// HOW:
/// - Mark the runtime awake and block on a future returning 42 with two injected interruptions.
/// - Check the result and exactly three notification attempts.
#[test]
fn interrupted_block_on_notification() {
    let calls = Arc::new(AtomicUsize::new(0));
    let rt = Runtime::new(Box::new(FailNotify {
        calls: calls.clone(),
        failures: 2,
        kind: io::ErrorKind::Interrupted,
    }));
    rt.awake();
    assert_eq!(rt.block_on(async { 42 }, &PollDriver), 42);
    assert_eq!(calls.load(Ordering::Relaxed), 3);
}

/// STATEMENT: Explicit `Handle::notify` returns driver errors without retrying them.
///
/// HOW:
/// - Inject `Interrupted` and `Other` errors through the notifier.
/// - Check the returned error kind and a single driver call for each case.
#[test]
fn handle_notify_returns_errors() {
    for kind in [io::ErrorKind::Interrupted, io::ErrorKind::Other] {
        let calls = Arc::new(AtomicUsize::new(0));
        let rt = Runtime::new(Box::new(FailNotify {
            calls: calls.clone(),
            failures: usize::MAX,
            kind,
        }));
        assert_eq!(rt.handle().notify().unwrap_err().kind(), kind);
        assert_eq!(calls.load(Ordering::Relaxed), 1);
    }
}

#[cfg(not(any(miri, ntex_loom)))]
/// STATEMENT: Unrecoverable scheduler notification failures abort rather than lose work.
///
/// HOW:
/// - Run local spawn, remote spawn, and `block_on` completion with a failing notifier in children.
/// - Require termination before the deadline and the notification failure diagnostic.
/// - Check `SIGABRT` on Unix, or a failing non-panic exit elsewhere.
#[test]
fn failed_scheduler_notification_aborts() {
    use std::process::{Command, Stdio};

    const CHILD: &str = "NTEX_RT_NOTIFY_ABORT_CHILD";
    if let Some(path) = std::env::var_os(CHILD) {
        struct Logger;
        impl log::Log for Logger {
            fn enabled(&self, _: &log::Metadata<'_>) -> bool {
                true
            }
            fn log(&self, record: &log::Record<'_>) {
                eprintln!("{}", record.args());
            }
            fn flush(&self) {}
        }
        log::set_logger(&Logger).unwrap();
        log::set_max_level(log::LevelFilter::Error);
        let rt = Runtime::new(Box::new(FailNotify {
            calls: Arc::new(AtomicUsize::new(0)),
            failures: usize::MAX,
            kind: io::ErrorKind::Other,
        }));
        match path.to_str().unwrap() {
            "local" => rt.spawn_detached(async {}),
            "remote" => {
                let handle = rt.handle();
                thread::spawn(move || handle.spawn_detached(async {}))
                    .join()
                    .unwrap();
            }
            "block_on" => {
                rt.awake();
                rt.block_on(async {}, &PollDriver);
            }
            _ => panic!("unknown notification path"),
        }
        panic!("notification failure did not abort");
    }

    for path in ["local", "remote", "block_on"] {
        let mut child = Command::new(std::env::current_exe().unwrap())
            .args([
                "--exact",
                "rt::tests::failed_scheduler_notification_aborts",
                "--nocapture",
            ])
            .env(CHILD, path)
            .stdout(Stdio::null())
            .stderr(Stdio::piped())
            .spawn()
            .unwrap();
        let start = Instant::now();
        while child.try_wait().unwrap().is_none() {
            if start.elapsed() > WAIT {
                child.kill().unwrap();
                child.wait().unwrap();
                panic!("{path}: notification failure did not terminate the process");
            }
            thread::sleep(Duration::from_millis(10));
        }
        let output = child.wait_with_output().unwrap();
        let stderr = String::from_utf8_lossy(&output.stderr);
        assert!(
            stderr.contains("Failed to notify runtime driver:"),
            "{stderr}"
        );
        #[cfg(unix)]
        {
            use std::os::unix::process::ExitStatusExt;
            assert_eq!(output.status.signal(), Some(libc::SIGABRT), "{stderr}");
        }
        #[cfg(not(unix))]
        {
            assert!(!output.status.success(), "{stderr}");
            assert_ne!(output.status.code(), Some(101), "{stderr}");
        }
    }
}

/// STATEMENT: Local scheduling skips notifications while awake and coalesces them while idle.
///
/// HOW:
/// - Mark the runtime awake, spawn a task, and check no notification before draining it.
/// - Spawn two tasks while idle and check one notification and a `Pending` poll result.
#[test]
fn awake_skips_local_notify() {
    use std::sync::atomic::Ordering;

    let cnt = Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let rt = Runtime::new(Box::new(CountNotify(cnt.clone())));

    // the driver is awake and polls the runtime before blocking
    rt.awake();
    drop(rt.spawn(async {}));
    assert_eq!(cnt.load(Ordering::Relaxed), 0);
    assert_eq!(rt.poll(), PollResult::Pending);

    // the runtime is idle, the driver may block
    drop(rt.spawn(async {}));
    drop(rt.spawn(async {}));
    assert_eq!(cnt.load(Ordering::Relaxed), 1);
    assert_eq!(rt.poll(), PollResult::Pending);
}

/// STATEMENT: Remote submissions notify once per batch until polling clears the wake flag.
///
/// HOW:
/// - Submit and drain two successive batches of 100 tasks from another thread.
/// - Check one new notification per batch and no additional notification during draining.
#[test]
fn remote_schedule_notifies_once() {
    let cnt = Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let rt = Runtime::new(Box::new(CountNotify(cnt.clone())));
    let spawn = |n| {
        let hnd = rt.handle();
        std::thread::spawn(move || {
            for _ in 0..n {
                hnd.spawn(async {});
            }
        })
        .join()
        .unwrap();
    };

    spawn(100);
    assert_eq!(cnt.load(Ordering::Relaxed), 1);
    while rt.poll() == PollResult::PollAgain {}

    // `poll` clears the flag
    spawn(100);
    assert_eq!(cnt.load(Ordering::Relaxed), 2);
    while rt.poll() == PollResult::PollAgain {}
    assert_eq!(cnt.load(Ordering::Relaxed), 2);
}

/// STATEMENT: Concurrent remote submissions cannot strand work when the driver goes idle.
///
/// HOW:
/// - Submit counted tasks from four producer threads to a condition-variable notifier.
/// - Poll and wait like a blocking driver until every task runs; fail any timed-out wait.
/// - Join all producers after reaching the expected completion count.
#[test]
fn remote_schedule_no_lost_wakeup() {
    use std::sync::{Condvar, Mutex, atomic::AtomicUsize};
    use std::time::Duration;

    #[derive(Debug, Default)]
    struct Event(Mutex<bool>, Condvar);

    #[derive(Debug)]
    struct EventNotify(Arc<Event>);

    impl Notify for EventNotify {
        fn notify(&self) -> io::Result<()> {
            *self.0.0.lock().unwrap() = true;
            self.0.1.notify_one();
            Ok(())
        }
    }

    const THREADS: usize = 4;
    const TASKS: usize = iters(10_000, 4);
    const LOST: Duration = if cfg!(miri) { WAIT } else { Duration::from_secs(5) };

    let ev = Arc::new(Event::default());
    let rt = Runtime::new(Box::new(EventNotify(ev.clone())));
    let cnt = Arc::new(AtomicUsize::new(0));
    let threads: Vec<_> = (0..THREADS)
        .map(|_| {
            let (hnd, cnt) = (rt.handle(), cnt.clone());
            std::thread::spawn(move || {
                for _ in 0..TASKS {
                    let cnt = cnt.clone();
                    hnd.spawn(async move {
                        cnt.fetch_add(1, Ordering::Relaxed);
                    });
                }
            })
        })
        .collect();

    // blocks like a driver, a lost wakeup times out
    while cnt.load(Ordering::Relaxed) < THREADS * TASKS {
        if rt.poll() == PollResult::Pending {
            let guard = ev.0.lock().unwrap();
            // the last tasks may have run in this poll
            let (mut guard, res) =
                ev.1.wait_timeout_while(guard, LOST, |n| {
                    !*n && cnt.load(Ordering::Relaxed) < THREADS * TASKS
                })
                .unwrap();
            assert!(!res.timed_out(), "lost wakeup");
            *guard = false;
        }
    }
    for t in threads {
        t.join().unwrap();
    }
}

/// STATEMENT: A remote task waking a local task makes the runtime request another poll.
///
/// HOW:
/// - Poll a local future once to save its waker, then remotely submit a task that wakes it.
/// - Check `PollAgain` followed by `Pending` and the local future's completion flag.
#[test]
fn local_task_woken_by_remote_task() {
    use std::sync::{Mutex, atomic::AtomicBool, atomic::Ordering};

    let rt = Runtime::new(Box::new(NoopNotify));
    let waker = Arc::new(Mutex::new(None::<Waker>));
    let done = Rc::new(AtomicBool::new(false));
    let (waker2, done2) = (waker.clone(), done.clone());
    drop(rt.spawn(poll_fn(move |cx| {
        if waker2.lock().unwrap().replace(cx.waker().clone()).is_some() {
            done2.store(true, Ordering::Relaxed);
            Poll::Ready(())
        } else {
            Poll::Pending
        }
    })));
    assert_eq!(rt.poll(), PollResult::Pending);

    let hnd = rt.handle();
    std::thread::spawn(move || {
        hnd.spawn(async move {
            waker.lock().unwrap().clone().unwrap().wake();
        });
    })
    .join()
    .unwrap();

    // the remote task schedules the local task, it must be polled again
    assert_eq!(rt.poll(), PollResult::PollAgain);
    assert_eq!(rt.poll(), PollResult::Pending);
    assert!(done.load(Ordering::Relaxed));
}

/// STATEMENT: `Runtime` spawn entrypoints do not populate the owner thread-ID cache.
///
/// HOW:
/// - Clear the TLS ID and spawn normal, unchecked borrowing, and detached local futures.
/// - Check the cache stays empty, restore it, and verify all results and side effects.
/// - Drain the runtime and check no task allocations remain.
#[test]
#[cfg(not(ntex_loom))]
fn runtime_spawn_skips_owner_lookup() {
    let borrowed = Cell::new(false);
    let rt = Runtime::new(Box::new(NoopNotify));
    let output = Rc::new(42);
    let detached = Rc::new(Cell::new(false));
    let done = detached.clone();

    // All Runtime entrypoints already know the owner, even with a cold TLS cache.
    let id = THREAD_ID.with(Cell::take);
    let task = rt.spawn(async move { output });
    let unchecked = unsafe { rt.spawn_unchecked(async { borrowed.set(true) }) };
    rt.spawn_detached(async move { done.set(true) });
    assert!(THREAD_ID.with(Cell::get).is_none());
    THREAD_ID.with(|slot| slot.set(id));

    assert_eq!(rt.poll(), PollResult::Pending);
    assert_eq!(*block_on_thread(task).unwrap(), 42);
    block_on_thread(unchecked).unwrap();
    assert!(borrowed.get());
    assert!(detached.get());
    drain(&rt);
    assert_eq!(rt.queue.task_count(), 0, "task allocations retained");
}

/// STATEMENT: A running task can schedule and await another task on the current runtime.
///
/// HOW:
/// - Spawn a detached parent that uses `Runtime::with_current` to spawn a child returning 1.
/// - Poll with the runtime installed as current and check the parent records the child result.
#[test]
fn schedule_while_running() {
    let rt = Runtime::new(Box::new(NoopNotify));
    let done = Rc::new(RefCell::new(0));
    let done2 = done.clone();
    drop(rt.spawn(async move {
        // schedules into the local queue from a running task
        let h = Runtime::with_current(|rt| rt.spawn(async { 1 }));
        *done2.borrow_mut() = h.await.unwrap();
    }));
    CURRENT_RUNTIME.set(&rt, || while rt.poll() == PollResult::PollAgain {});
    assert_eq!(*done.borrow(), 1);
}

/// STATEMENT: Event intervals bound local polling and still drain overflowed remote work.
///
/// HOW:
/// - For intervals zero and one, check an empty poll and two tasks split across two polls.
/// - Remotely submit 130 counted tasks with interval one to overflow the fixed queue.
/// - Check 129 tasks run before `PollAgain` and the final task runs before `Pending`.
#[test]
fn event_interval() {
    use std::sync::atomic::{AtomicUsize, Ordering};

    for val in [0, 1] {
        let rt = Runtime::builder()
            .event_interval(val)
            .build(Box::new(NoopNotify));
        assert_eq!(rt.poll(), PollResult::Pending);

        drop(rt.spawn(async {}));
        drop(rt.spawn(async {}));
        assert_eq!(rt.poll(), PollResult::PollAgain);
        assert_eq!(rt.poll(), PollResult::Pending);
    }

    // tasks scheduled from other threads overflow the fixed queue
    let rt = Runtime::builder()
        .event_interval(1)
        .build(Box::new(NoopNotify));
    let cnt = Arc::new(AtomicUsize::new(0));
    let hnd = rt.handle();
    let cnt2 = cnt.clone();
    std::thread::spawn(move || {
        for _ in 0..130 {
            let cnt = cnt2.clone();
            hnd.spawn(async move {
                cnt.fetch_add(1, Ordering::Relaxed);
            });
        }
    })
    .join()
    .unwrap();
    assert_eq!(rt.poll(), PollResult::PollAgain);
    assert_eq!(cnt.load(Ordering::Relaxed), 129);
    // Untouched tasks reclaim directly instead of consuming the FIFO budget.
    assert_eq!(rt.poll(), PollResult::Pending);
    assert_eq!(cnt.load(Ordering::Relaxed), 130);
}

/// STATEMENT: Runtime time is live while idle and fixed at turn start while awake.
///
/// HOW:
/// - Sleep between idle time reads, then mark the runtime awake and check its time freezes.
/// - Check the default clock remains live when the runtime is not installed as current.
/// - Run two tasks under `CURRENT_RUNTIME` and check both clocks report turn start.
/// - Check idle time advances after those tasks drain.
#[test]
fn now() {
    let rt = Runtime::new(Box::new(NoopNotify));

    // idle runtime returns the current time
    let t = Instant::now();
    assert!(rt.now() >= t);
    std::thread::sleep(Duration::from_millis(5));
    assert!(rt.now() >= t + Duration::from_millis(5));

    // tasks observe the turn start
    rt.awake();
    let start = rt.now();
    std::thread::sleep(Duration::from_millis(5));
    assert_eq!(rt.now(), start);
    // the runtime is not current on this thread
    assert!(crate::rt::now() >= start + Duration::from_millis(5));

    let seen = Rc::new(RefCell::new(Vec::new()));
    for _ in 0..2 {
        let seen = seen.clone();
        drop(rt.spawn(async move {
            seen.borrow_mut()
                .push((Runtime::with_current(Runtime::now), crate::rt::now()));
            std::thread::sleep(Duration::from_millis(5));
        }));
    }
    CURRENT_RUNTIME.set(&rt, || assert_eq!(rt.poll(), PollResult::Pending));
    assert_eq!(*seen.borrow(), vec![(start, start), (start, start)]);

    // idle again after the queue is drained
    assert!(rt.now() > start);
}

/// STATEMENT: Local scheduling while idle starts a fresh turn and its busy-time accounting.
///
/// HOW:
/// - End a delayed awake turn, then spawn a task while idle and record the new turn time.
/// - Check `awake` preserves that time and the new turn starts after the previous delay.
/// - Bound the added busy duration by elapsed times immediately before and after polling.
#[test]
fn local_schedule_while_idle_starts_turn() {
    let sleep = Duration::from_millis(20);
    let rt = Runtime::new(Box::new(NoopNotify));
    rt.awake();
    let start = rt.now();
    std::thread::sleep(sleep);
    assert_eq!(rt.poll(), PollResult::Pending);
    let busy = rt.busy_duration();

    // the driver wakes a task before blocking, the next turn starts here
    drop(rt.spawn(async {}));
    let turn_start = rt.now();
    rt.awake();
    assert_eq!(rt.now(), turn_start);
    assert!(turn_start >= start + sleep);
    let before_poll = turn_start.elapsed();
    assert_eq!(rt.poll(), PollResult::Pending);
    let after_poll = turn_start.elapsed();
    let total = rt.busy_duration();
    assert!(total >= busy + before_poll);
    assert!(total <= busy + after_poll);
}

/// STATEMENT: Busy duration counts task and awake time, excludes idle time, and publishes each poll.
///
/// HOW:
/// - Sleep in a task and between `awake` and `poll`; check both increase the handle's busy duration.
/// - Sleep while idle and bound any subsequent increase by the poll's own elapsed time.
/// - Leave a backlog with interval one and check time in and between busy polls is counted.
#[test]
fn busy_duration() {
    let sleep = Duration::from_millis(20);
    let rt = Runtime::builder()
        .event_interval(1)
        .build(Box::new(NoopNotify));
    let hnd = rt.handle();
    assert_eq!(rt.poll(), PollResult::Pending);

    // task time is counted
    drop(rt.spawn(async move { std::thread::sleep(sleep) }));
    assert_eq!(rt.poll(), PollResult::Pending);
    let busy = hnd.busy_duration().unwrap();
    assert!(busy >= sleep);

    // time between `awake` and `poll` is counted
    rt.awake();
    std::thread::sleep(sleep);
    assert_eq!(rt.poll(), PollResult::Pending);
    assert!(hnd.busy_duration().unwrap() >= busy + sleep);

    // idle time is not counted
    let busy = rt.busy_duration();
    std::thread::sleep(sleep);
    assert_eq!(rt.busy_duration(), busy);
    let poll_start = Instant::now();
    assert_eq!(rt.poll(), PollResult::Pending);
    let poll_elapsed = poll_start.elapsed();
    assert!(rt.busy_duration() <= busy + poll_elapsed);

    // published on each poll while the runtime stays busy
    drop(rt.spawn(async move { std::thread::sleep(sleep) }));
    drop(rt.spawn(async {}));
    let busy = rt.busy_duration();
    assert_eq!(rt.poll(), PollResult::PollAgain);
    let busy2 = rt.busy_duration();
    assert!(busy2 >= busy + sleep);

    // time between busy polls is counted
    std::thread::sleep(sleep);
    assert_eq!(rt.poll(), PollResult::Pending);
    assert!(rt.busy_duration() >= busy2 + sleep);
}

/// STATEMENT: Handles expose in-progress busy time and clear `current_turn` after polling ends.
///
/// HOW:
/// - Run a sleeping task on another thread, keeping its turn open with a channel.
/// - Check the handle's busy duration and current turn include elapsed running time.
/// - Release and join the owner; check busy time includes the sleep and `current_turn` is `None`.
#[test]
fn busy_duration_during_turn() {
    let sleep = Duration::from_millis(50);
    let (tx, rx) = std::sync::mpsc::channel();
    let (done_tx, done_rx) = std::sync::mpsc::channel::<()>();
    let th = thread::spawn(move || {
        let rt = Runtime::new(Box::new(NoopNotify));
        let hnd = rt.handle();
        drop(rt.spawn(async move {
            tx.send(hnd).unwrap();
            thread::sleep(sleep);
            // keep the turn running until the other thread checked it
            let _ = done_rx.recv();
        }));
        assert_eq!(rt.poll(), PollResult::Pending);
        assert_eq!(rt.current_turn(), None);
    });

    // the task is running, its turn is visible from another thread
    let hnd = rx.recv().unwrap();
    thread::sleep(sleep / 2);
    assert!(hnd.busy_duration().unwrap() >= sleep / 2);
    assert!(hnd.current_turn().unwrap() >= sleep / 2);
    done_tx.send(()).unwrap();

    th.join().unwrap();
    assert!(hnd.busy_duration().unwrap() >= sleep);
    assert_eq!(hnd.current_turn(), None);
}

/// STATEMENT: Current-task metadata reports local and remote spawn sites only while tasks run.
///
/// HOW:
/// - Hold a locally spawned task inside poll and inspect its file and line from another thread.
/// - Check metadata clears after polling, then remotely spawn a task that reports its own location.
/// - Verify the remote spawn site and that no current task remains after completion.
#[test]
fn current_task() {
    let (tx, rx) = std::sync::mpsc::channel();
    let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
    let (polled_tx, polled_rx) = std::sync::mpsc::channel::<()>();
    let (spawned_tx, spawned_rx) = std::sync::mpsc::channel::<()>();
    let th = thread::spawn(move || {
        let rt = Runtime::new(Box::new(NoopNotify));
        let hnd = rt.handle();
        assert_eq!(hnd.current_task(), None);

        let line = line!() + 1;
        drop(rt.spawn(async move {
            tx.send((hnd, line)).unwrap();
            release_rx.recv().unwrap();
        }));
        assert_eq!(rt.poll(), PollResult::Pending);
        polled_tx.send(()).unwrap();

        spawned_rx.recv().unwrap();
        assert_eq!(rt.poll(), PollResult::Pending);
    });

    // the running task is visible from another thread
    let (hnd, line) = rx.recv().unwrap();
    let task = hnd.current_task().unwrap();
    assert_eq!((task.file(), task.line()), (file!(), line));
    release_tx.send(()).unwrap();

    polled_rx.recv().unwrap();
    assert_eq!(hnd.current_task(), None);

    // tasks spawned from other threads record their location too
    let (tx, rx) = std::sync::mpsc::channel();
    let hnd2 = hnd.clone();
    let line = line!() + 1;
    hnd.spawn(async move {
        tx.send(hnd2.current_task()).unwrap();
    });
    spawned_tx.send(()).unwrap();
    th.join().unwrap();
    let task = rx.recv().unwrap().unwrap();
    assert_eq!((task.file(), task.line()), (file!(), line));
    assert_eq!(hnd.current_task(), None);
}

/// STATEMENT: Concurrent busy-duration reads never regress as busy turns end.
///
/// HOW:
/// - Repeatedly run sleeping tasks and idle gaps on an owner thread.
/// - Read `busy_duration` through its handle until completion and compare every sample to the last.
#[test]
fn busy_duration_is_monotonic() {
    let done = Arc::new(AtomicBool::new(false));
    let (tx, rx) = std::sync::mpsc::channel();
    let done2 = done.clone();
    let th = thread::spawn(move || {
        let rt = Runtime::new(Box::new(NoopNotify));
        tx.send(rt.handle()).unwrap();
        for _ in 0..100 {
            drop(rt.spawn(async { thread::sleep(Duration::from_micros(500)) }));
            assert_eq!(rt.poll(), PollResult::Pending);
            thread::sleep(Duration::from_micros(200));
        }
        done2.store(true, Ordering::Release);
    });

    // a busy period is not counted twice when it ends
    let hnd = rx.recv().unwrap();
    let mut last = Duration::ZERO;
    while !done.load(Ordering::Acquire) {
        let busy = hnd.busy_duration().unwrap();
        assert!(busy >= last, "{busy:?} < {last:?}");
        last = busy;
    }
    th.join().unwrap();
}

/// STATEMENT: Poll counters and histogram count active polls but not stopped-runtime polls.
///
/// HOW:
/// - Check initial counters, an empty poll, and two interval-limited polls with one `PollAgain`.
/// - Verify handle counters and histogram count match three polls and one backlog result.
/// - Stop the runtime and check `Ready` leaves counters and histogram unchanged.
#[test]
fn poll_count() {
    let rt = Runtime::builder()
        .event_interval(1)
        .build(Box::new(NoopNotify));
    let hnd = rt.handle();
    assert_eq!((rt.poll_count(), rt.poll_again_count()), (0, 0));

    assert_eq!(rt.poll(), PollResult::Pending);
    assert_eq!((rt.poll_count(), rt.poll_again_count()), (1, 0));

    // one task per poll, the first poll leaves a backlog
    drop(rt.spawn(async {}));
    drop(rt.spawn(async {}));
    assert_eq!(rt.poll(), PollResult::PollAgain);
    assert_eq!(rt.poll(), PollResult::Pending);
    assert_eq!(hnd.poll_count(), Some(3));
    assert_eq!(hnd.poll_again_count(), Some(1));
    assert_eq!(hnd.poll_histogram().unwrap().count(), 3);

    // a stopped runtime does not run tasks
    rt.stop.set(true);
    assert_eq!(rt.poll(), PollResult::Ready);
    assert_eq!((rt.poll_count(), rt.poll_again_count()), (3, 1));
    assert_eq!(rt.poll_histogram().count(), 3);
}

/// STATEMENT: Shutdown also drops a waiting task woken by another queued task's destructor.
///
/// HOW:
/// - Poll a detached pending task to retain its waker, then queue a `WakeOnDrop` guard.
/// - Drop the runtime and check the saved waker is consumed and its slot has no task-held clones.
#[test]
fn schedule_while_clearing() {
    let rt = Runtime::new(Box::new(NoopNotify));
    let waker = Rc::new(RefCell::new(None));
    let waker2 = waker.clone();
    drop(rt.spawn(poll_fn(move |cx| {
        *waker2.borrow_mut() = Some(cx.waker().clone());
        Poll::<()>::Pending
    })));
    assert_eq!(rt.poll(), PollResult::Pending);
    assert!(waker.borrow().is_some());

    // queued task wakes the waiting task while the queue is cleared
    let guard = WakeOnDrop(waker.clone());
    drop(rt.spawn(async move {
        let _g = guard;
    }));
    drop(rt.spawn(async {}));
    drop(rt);
    assert!(waker.borrow().is_none());
    // the woken task is dropped too, releasing its clone of `waker`
    assert_eq!(Rc::strong_count(&waker), 1);
}

/// STATEMENT: Self-wake totals count wakes during poll, not later external wakes.
///
/// HOW:
/// - Run a future that wakes itself five times with warning tracking enabled and disabled.
/// - Check runtime and handle totals, and no tracking clock when the threshold is zero.
/// - Wake its saved waker outside poll, complete it, and check the self-wake total stays five.
#[test]
fn self_wake() {
    for count in [0, 10_000] {
        let rt = Runtime::builder()
            .self_wake_threshold(count, Duration::from_secs(1))
            .build(Box::new(NoopNotify));
        let waker = Rc::new(RefCell::new(None::<Waker>));
        let waker2 = waker.clone();
        let mut polls = 0;
        let h = rt.spawn(poll_fn(move |cx| {
            polls += 1;
            match polls {
                1..=5 => {
                    cx.waker().wake_by_ref();
                    Poll::Pending
                }
                6 => {
                    *waker2.borrow_mut() = Some(cx.waker().clone());
                    Poll::Pending
                }
                _ => Poll::Ready(()),
            }
        }));
        CURRENT_RUNTIME.set(&rt, || while rt.poll() == PollResult::PollAgain {});
        assert_eq!(rt.self_wake_count(), 5);
        assert_eq!(rt.handle().self_wake_count(), Some(5));
        if count == 0 {
            assert_eq!(rt.queue.self_wake_clock.get(), None);
        }

        // a wake from outside of the poll is not a self-wake
        waker.borrow_mut().take().unwrap().wake();
        CURRENT_RUNTIME.set(&rt, || while rt.poll() == PollResult::PollAgain {});
        assert!(h.is_finished());
        assert_eq!(rt.self_wake_count(), 5);
    }
}

/// STATEMENT: Self-wake streak reporting honors count, duration, disablement, and turn-clock refreshes.
///
/// HOW:
/// - Feed synthetic task statistics through thresholds and check counts and the `REPORTED` bit.
/// - Check long-duration and disabled thresholds suppress reporting without losing global totals.
/// - Check streak start times refresh after `awake`, a subsequent busy poll, and a new `block_on`.
#[test]
fn self_wake_streak() {
    #[derive(Default)]
    struct Stats {
        wakes: Cell<u32>,
        start: Cell<u32>,
    }

    impl TaskStats for &Stats {
        fn location(self) -> SpawnLocation {
            Location::caller()
        }
        fn self_wakes(self) -> u32 {
            self.wakes.get()
        }
        fn set_self_wakes(self, wakes: u32) {
            self.wakes.set(wakes);
        }
        fn streak_start(self) -> u32 {
            self.start.get()
        }
        fn set_streak_start(self, start: u32) {
            self.start.set(start);
        }
    }

    fn streak(rt: &Runtime, n: u32) -> u32 {
        let stats = Stats::default();
        for _ in 0..n {
            rt.queue.self_wake(&stats, stats.wakes.get());
        }
        stats.wakes.get()
    }

    fn start(rt: &Runtime) -> u32 {
        let stats = Stats::default();
        rt.queue.self_wake(&stats, 0);
        stats.start.get()
    }

    fn now(rt: &Runtime) -> u32 {
        u32::try_from(
            rt.queue
                .poll_start
                .get()
                .saturating_duration_since(rt.queue.epoch)
                .as_millis(),
        )
        .unwrap()
    }

    struct BusyDriver;

    impl Driver for BusyDriver {
        fn handle(&self) -> Box<dyn Notify> {
            Box::new(NoopNotify)
        }

        fn run(&self, rt: &Runtime) -> io::Result<()> {
            let start = Instant::now();
            while rt.poll() != PollResult::Ready {
                assert!(start.elapsed() < WAIT, "timeout");
            }
            Ok(())
        }
    }

    let rt = Runtime::builder()
        .self_wake_threshold(3, Duration::ZERO)
        .build(Box::new(NoopNotify));
    assert_eq!(streak(&rt, 2), 2);
    assert_eq!(streak(&rt, 3), 3 | REPORTED);
    assert_eq!(streak(&rt, 5), 5 | REPORTED);
    assert_eq!(rt.self_wake_count(), 10);

    // the streak is shorter than the duration
    let rt = Runtime::builder()
        .self_wake_threshold(3, Duration::from_hours(1))
        .build(Box::new(NoopNotify));
    assert_eq!(streak(&rt, 5), 5);

    // disabled
    let rt = Runtime::builder()
        .self_wake_threshold(0, Duration::ZERO)
        .build(Box::new(NoopNotify));
    assert_eq!(streak(&rt, 5), 0);
    assert_eq!(rt.self_wake_count(), 5);

    let rt = Runtime::new(Box::new(NoopNotify));
    let before = start(&rt);
    thread::sleep(Duration::from_millis(2));
    rt.awake();
    assert!(now(&rt) > before);
    assert_eq!(start(&rt), now(&rt), "streak after awake");

    let task = rt.spawn(poll_fn(|cx| {
        cx.waker().wake_by_ref();
        Poll::<()>::Pending
    }));
    let before = start(&rt);
    thread::sleep(Duration::from_millis(2));
    assert_eq!(rt.poll(), PollResult::PollAgain);
    assert!(now(&rt) > before);
    assert_eq!(start(&rt), now(&rt), "streak after the next poll");
    task.cancel();
    assert_eq!(rt.poll(), PollResult::Pending);

    // Already awake, so the root spawn does not refresh the clock.
    rt.awake();
    let before = start(&rt);
    thread::sleep(Duration::from_millis(2));
    rt.block_on(
        async {
            assert!(now(&rt) > before);
            assert_eq!(start(&rt), now(&rt), "streak in a new block_on");
        },
        &BusyDriver,
    );
}

/// STATEMENT: A warning logger may reenter polling without duplicate streak reports or retained tasks.
///
/// HOW:
/// - Install a scoped warning counter whose first log callback polls the current runtime.
/// - Run self-waking tasks that either wake again or complete during the nested poll.
/// - Check one warning, exact poll and self-wake totals, and zero remaining task allocations.
#[test]
fn reentrant_self_wake_logger() {
    scoped_tls::scoped_thread_local!(static WARNINGS: Cell<usize>);

    struct Logger;

    impl log::Log for Logger {
        fn enabled(&self, metadata: &log::Metadata<'_>) -> bool {
            metadata.level() == log::Level::Warn
                && metadata.target() == "ntex_rt::rt"
                && WARNINGS.is_set()
        }

        fn log(&self, record: &log::Record<'_>) {
            if self.enabled(record.metadata()) {
                WARNINGS.with(|warnings| {
                    let count = warnings.get();
                    warnings.set(count + 1);
                    if count == 0 {
                        CURRENT_RUNTIME.with(|rt| {
                            rt.poll();
                        });
                    }
                });
            }
        }

        fn flush(&self) {}
    }

    log::set_logger(&Logger).unwrap();
    log::set_max_level(log::LevelFilter::Warn);

    // The nested poll either wakes again or completes and frees the task.
    for ready_at in [3, 2] {
        let rt = Runtime::builder()
            .event_interval(1)
            .self_wake_threshold(1, Duration::ZERO)
            .build(Box::new(NoopNotify));
        let warnings = Cell::new(0);
        let polls = Rc::new(Cell::new(0u32));
        let count = polls.clone();
        rt.spawn_detached(poll_fn(move |cx| {
            let n = count.get() + 1;
            count.set(n);
            if n < ready_at {
                cx.waker().wake_by_ref();
                Poll::Pending
            } else {
                Poll::Ready(())
            }
        }));

        CURRENT_RUNTIME.set(&rt, || {
            WARNINGS.set(&warnings, || {
                rt.poll();
                assert_eq!(warnings.get(), 1, "one warning per streak");
                assert_eq!(rt.poll(), PollResult::Pending);
            });
        });
        assert_eq!(polls.get(), ready_at);
        assert_eq!(rt.self_wake_count(), u64::from(ready_at - 1));
        assert_eq!(rt.queue.task_count(), 0, "task allocation retained");
    }
}

struct DropFlag(Arc<AtomicBool>);

impl Drop for DropFlag {
    fn drop(&mut self) {
        self.0.store(true, Ordering::SeqCst);
    }
}

type Slot = Arc<std::sync::Mutex<Option<Waker>>>;

fn poll_until(rt: &Runtime, cond: impl Fn() -> bool) {
    let start = Instant::now();
    CURRENT_RUNTIME.set(rt, || {
        while !cond() {
            assert!(start.elapsed() < WAIT, "timeout");
            if rt.poll() == PollResult::Pending {
                thread::yield_now();
            }
        }
    });
}

fn block_on_thread<F: Future>(fut: F) -> F::Output {
    struct ThreadWaker(thread::Thread);

    impl std::task::Wake for ThreadWaker {
        fn wake(self: Arc<Self>) {
            self.0.unpark();
        }
    }

    let waker = Waker::from(Arc::new(ThreadWaker(thread::current())));
    let mut cx = std::task::Context::from_waker(&waker);
    let mut fut = std::pin::pin!(fut);
    loop {
        if let Poll::Ready(v) = fut.as_mut().poll(&mut cx) {
            return v;
        }
        thread::park();
    }
}

/// Stores its waker until polled `n` times.
fn wait_polls(
    slot: Slot,
    polls: Arc<std::sync::atomic::AtomicUsize>,
    n: usize,
) -> impl Future<Output = usize> {
    poll_fn(move |cx| {
        let cnt = polls.fetch_add(1, Ordering::SeqCst) + 1;
        if cnt >= n {
            Poll::Ready(cnt)
        } else {
            *slot.lock().unwrap() = Some(cx.waker().clone());
            Poll::Pending
        }
    })
}

/// STATEMENT: Standard task wakers support remote reference wakes, cloning, dropping, and owned wakes.
///
/// HOW:
/// - Poll a task waiting for four polls and move each saved waker through worker threads.
/// - Exercise `wake_by_ref`, clone/drop plus wake, and a final consuming wake.
/// - Check each intermediate poll count and join the local handle with result four.
#[test]
fn remote_wakers() {
    let rt = Runtime::new(Box::new(NoopNotify));
    let slot = Slot::default();
    let polls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let h = rt.spawn(wait_polls(slot.clone(), polls.clone(), 4));
    poll_until(&rt, || polls.load(Ordering::SeqCst) == 1);

    // wake by ref, the thread keeps the waker
    let w = slot.lock().unwrap().take().unwrap();
    let th = thread::spawn(move || {
        w.wake_by_ref();
        w
    });
    let w = th.join().unwrap();
    poll_until(&rt, || polls.load(Ordering::SeqCst) == 2);

    // clone, drop and wake by value
    let w2 = slot.lock().unwrap().take().unwrap();
    thread::spawn(move || {
        let c = w2.clone();
        drop(w2);
        drop(w);
        c.wake();
    })
    .join()
    .unwrap();
    poll_until(&rt, || polls.load(Ordering::SeqCst) == 3);

    // The standard waker crosses threads; the join remains local.
    let w = slot.lock().unwrap().take().unwrap();
    thread::spawn(move || w.wake()).join().unwrap();
    poll_until(&rt, || h.is_finished());
    assert_eq!(block_on_thread(h).unwrap(), 4);
}

/// STATEMENT: Concurrent remote waker operations neither lose progress nor repoll completed work.
///
/// HOW:
/// - Run a future with a fixed poll target while four threads wake, wake by reference, and clone/drop.
/// - Poll until finished, stop and join the workers, and check the result and poll count equal the target.
#[test]
fn remote_wakers_stress() {
    const POLLS: usize = iters(5000, 10);

    let rt = Runtime::new(Box::new(NoopNotify));
    let slot = Slot::default();
    let polls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let stop = Arc::new(AtomicBool::new(false));
    let h = rt.spawn(wait_polls(slot.clone(), polls.clone(), POLLS));

    let wakers: Vec<_> = (0..4)
        .map(|i| {
            let slot = slot.clone();
            let stop = stop.clone();
            thread::spawn(move || {
                while !stop.load(Ordering::SeqCst) {
                    let w = slot.lock().unwrap().clone();
                    match (w, i % 3) {
                        (Some(w), 0) => w.wake(),
                        (Some(w), 1) => w.wake_by_ref(),
                        (Some(w), _) => drop(w.clone()),
                        (None, _) => thread::yield_now(),
                    }
                }
            })
        })
        .collect();
    poll_until(&rt, || h.is_finished());
    stop.store(true, Ordering::SeqCst);
    for th in wakers {
        th.join().unwrap();
    }
    assert_eq!(block_on_thread(h).unwrap(), POLLS);
    assert_eq!(polls.load(Ordering::SeqCst), POLLS);
    poll_until(&rt, || true);
}

/// STATEMENT: Dropping the last external waker cancels an otherwise unreferenced pending future.
///
/// HOW:
/// - Detach a future that retains a drop flag and publishes its only waker.
/// - Drop that waker locally and remotely in separate cases, polling until the future's flag is set.
#[test]
fn last_waker_drops_future() {
    let rt = Runtime::new(Box::new(NoopNotify));
    for remote in [false, true] {
        let slot = Slot::default();
        let dropped = Arc::new(AtomicBool::new(false));
        let flag = DropFlag(dropped.clone());
        let slot2 = slot.clone();
        drop(rt.spawn(async move {
            let _flag = flag;
            poll_fn(|cx| {
                *slot2.lock().unwrap() = Some(cx.waker().clone());
                Poll::<()>::Pending
            })
            .await;
        }));
        poll_until(&rt, || slot.lock().unwrap().is_some());
        assert!(!dropped.load(Ordering::SeqCst));

        let w = slot.lock().unwrap().take();
        if remote {
            thread::spawn(move || drop(w)).join().unwrap();
        } else {
            drop(w);
        }
        poll_until(&rt, || dropped.load(Ordering::SeqCst));
    }
}

/// STATEMENT: Local cancellation still drops a pending future after a remote wake is queued.
///
/// HOW:
/// - Poll a drop-tracked local future until it saves a waker.
/// - Consume the waker on another thread, cancel the join handle, and poll until the drop flag is set.
#[test]
fn local_cancel_after_remote_wake() {
    let rt = Runtime::new(Box::new(NoopNotify));
    let dropped = Arc::new(AtomicBool::new(false));
    let flag = DropFlag(dropped.clone());
    let polled = Rc::new(Cell::new(false));
    let polled2 = polled.clone();
    let slot = Slot::default();
    let slot2 = slot.clone();
    let h = rt.spawn(async move {
        let _flag = flag;
        polled2.set(true);
        poll_fn(|cx| {
            *slot2.lock().unwrap() = Some(cx.waker().clone());
            Poll::<()>::Pending
        })
        .await;
    });
    poll_until(&rt, || polled.get());

    let w = slot.lock().unwrap().take().unwrap();
    thread::spawn(move || w.wake()).join().unwrap();
    h.cancel();
    poll_until(&rt, || dropped.load(Ordering::SeqCst));
}

/// STATEMENT: A task cancelling itself during poll is dropped without running its self-wake again.
///
/// HOW:
/// - Give a pending future its own join handle and have its first poll wake and cancel itself.
/// - Poll until its drop flag is set and check the future was polled exactly once.
#[test]
fn cancel_while_running() {
    let rt = Runtime::new(Box::new(NoopNotify));
    let dropped = Arc::new(AtomicBool::new(false));
    let flag = DropFlag(dropped.clone());
    let handle = Rc::new(RefCell::new(None::<JoinHandle<()>>));
    let handle2 = handle.clone();
    let polls = Rc::new(Cell::new(0));
    let polls2 = polls.clone();
    let h = rt.spawn(async move {
        let _flag = flag;
        poll_fn(|cx| {
            polls2.set(polls2.get() + 1);
            if let Some(h) = handle2.borrow_mut().take() {
                cx.waker().wake_by_ref();
                h.cancel();
            }
            Poll::<()>::Pending
        })
        .await;
    });
    *handle.borrow_mut() = Some(h);
    poll_until(&rt, || dropped.load(Ordering::SeqCst));
    assert_eq!(polls.get(), 1);
}

/// STATEMENT: A task poll panic drops its future, fails its join, and leaves the runtime usable.
///
/// HOW:
/// - Catch a panic from polling a drop-tracked future and check its flag and finished handle.
/// - Check joining fails, then spawn and run a second task to verify continued progress.
#[test]
fn panic_in_poll() {
    let rt = Runtime::new(Box::new(NoopNotify));
    let dropped = Arc::new(AtomicBool::new(false));
    let flag = DropFlag(dropped.clone());
    let h = rt.spawn(async move {
        let _flag = flag;
        panic!("task panic");
    });
    let res = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        CURRENT_RUNTIME.set(&rt, || rt.poll())
    }));
    assert!(res.is_err());
    assert!(dropped.load(Ordering::SeqCst));
    assert!(h.is_finished());
    assert!(block_on_thread(h).is_err());

    // the runtime keeps running tasks
    let done = Rc::new(Cell::new(false));
    let done2 = done.clone();
    drop(rt.spawn(async move { done2.set(true) }));
    poll_until(&rt, || done.get());
}

/// STATEMENT: Runtime shutdown drops both locally queued and remotely submitted futures.
///
/// HOW:
/// - Queue local and remote futures carrying separate drop flags without polling the runtime.
/// - Drop the runtime and check both flags and the local handle's finished state.
#[test]
fn drop_runtime_with_tasks() {
    let dropped = Arc::new(AtomicBool::new(false));
    let remote = Arc::new(AtomicBool::new(false));
    let rt = Runtime::new(Box::new(NoopNotify));
    let flag = DropFlag(dropped.clone());
    let h = rt.spawn(async move {
        let _flag = flag;
        poll_fn(|_| Poll::<()>::Pending).await;
    });
    let flag = DropFlag(remote.clone());
    let hnd = rt.handle();
    thread::spawn(move || {
        hnd.spawn(async move {
            let _flag = flag;
        });
    })
    .join()
    .unwrap();
    drop(rt);
    assert!(dropped.load(Ordering::SeqCst));
    assert!(remote.load(Ordering::SeqCst));
    assert!(h.is_finished());
}

/// Pending future that borrows the caller's frame and sets a flag on drop.
struct Borrowing<'a>(&'a Cell<bool>, Option<Waker>);

impl Future for Borrowing<'_> {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        assert!(!self.0.get());
        self.1 = Some(cx.waker().clone());
        Poll::Pending
    }
}

impl Drop for Borrowing<'_> {
    fn drop(&mut self) {
        self.0.set(true);
    }
}

/// STATEMENT: Reentrant polling followed by self-waking is handled by the running poll, not notify.
///
/// HOW:
/// - Use a notifier that reenters poll and records notifications during an active task.
/// - Run a task that polls the runtime then self-wakes on its first poll and completes on its second.
/// - Check two polls, one self-wake, no active-task notification, and a cleared current-task pointer.
#[test]
fn reentrant_notify_self_wake() {
    #[derive(Debug, Default)]
    struct ReenterNotify(AtomicUsize);

    impl Notify for ReenterNotify {
        fn notify(&self) -> io::Result<()> {
            if CURRENT_RUNTIME.is_set() {
                CURRENT_RUNTIME.with(|rt| {
                    if !rt.queue.current.load(Ordering::Relaxed).is_null() {
                        self.0.fetch_add(1, Ordering::Relaxed);
                    }
                    rt.poll();
                });
            }
            Ok(())
        }
    }

    #[derive(Debug)]
    struct Shared(Arc<ReenterNotify>);

    impl Notify for Shared {
        fn notify(&self) -> io::Result<()> {
            self.0.notify()
        }
    }

    let notify = Arc::new(ReenterNotify::default());
    let rt = Runtime::new(Box::new(Shared(notify.clone())));
    let polls = Rc::new(Cell::new(0));
    let p = polls.clone();
    rt.spawn_detached(poll_fn(move |cx| {
        p.set(p.get() + 1);
        if p.get() == 1 {
            CURRENT_RUNTIME.with(|rt| {
                rt.poll();
            });
            cx.waker().wake_by_ref();
            Poll::Pending
        } else {
            Poll::Ready(())
        }
    }));
    CURRENT_RUNTIME.set(&rt, || {
        rt.poll();
        rt.poll();
    });
    assert_eq!(polls.get(), 2);
    assert_eq!(notify.0.load(Ordering::Relaxed), 0);
    assert_eq!(rt.self_wake_count(), 1);
    assert!(rt.queue.current.load(Ordering::Relaxed).is_null());
}

/// STATEMENT: Backtraces captured inside `block_on` stop before task implementation frames.
///
/// HOW:
/// - Capture and resolve a backtrace inside a `block_on` future using `PollDriver`.
/// - Require a resolved representation and check it does not contain `task.rs`.
#[cfg(not(miri))]
#[test]
fn block_on_backtrace_cutoff() {
    let rt = Runtime::new(Box::new(NoopNotify));
    let repr = rt.block_on(
        async { ntex_error::Backtrace::with_current().resolve().repr() },
        &PollDriver,
    );
    let repr = repr.unwrap();
    assert!(!repr.contains("task.rs"), "{repr}");
}

/// STATEMENT: Nested `block_on` is rejected for both the current runtime and a different runtime.
///
/// HOW:
/// - Attempt both nesting cases inside an outer `block_on` and catch the rejection panic.
/// - Check each rejection and successfully block on another future using the outer runtime afterward.
#[test]
fn block_on_rejects_nesting() {
    struct BusyDriver;

    impl Driver for BusyDriver {
        fn handle(&self) -> Box<dyn Notify> {
            Box::new(NoopNotify)
        }

        fn run(&self, rt: &Runtime) -> io::Result<()> {
            while rt.poll() != PollResult::Ready {}
            Ok(())
        }
    }

    for same in [true, false] {
        let rt = Runtime::new(Box::new(NoopNotify));
        let nested = rt.block_on(
            async {
                std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                    if same {
                        Runtime::with_current(|rt| rt.block_on(async {}, &BusyDriver));
                    } else {
                        Runtime::new(Box::new(NoopNotify)).block_on(async {}, &BusyDriver);
                    }
                }))
                .is_err()
            },
            &BusyDriver,
        );
        assert!(nested);
        // the runtime is usable after the rejected call
        assert_eq!(rt.block_on(async { 1 }, &BusyDriver), 1);
    }
}

/// STATEMENT: A driver error makes `block_on` unwind only after dropping its borrowing future.
///
/// HOW:
/// - Block on a frame-borrowing pending future using a driver that immediately returns an error.
/// - Catch the panic, check the future's drop flag, and poll the runtime afterward.
#[test]
fn block_on_drops_future_on_driver_error() {
    struct ErrorDriver;

    impl Driver for ErrorDriver {
        fn handle(&self) -> Box<dyn Notify> {
            Box::new(NoopNotify)
        }

        fn run(&self, _: &Runtime) -> io::Result<()> {
            Err(io::Error::other("driver stopped"))
        }
    }

    let rt = Runtime::new(Box::new(NoopNotify));
    let dropped = Cell::new(false);
    let res = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        rt.block_on(Borrowing(&dropped, None), &ErrorDriver);
    }));
    assert!(res.is_err());
    assert!(dropped.get());
    rt.poll();
}

/// STATEMENT: A driver panic drops `block_on`'s borrowing future even after it has been polled.
///
/// HOW:
/// - Use a driver that polls the borrowing pending future before panicking.
/// - Catch the unwind, check the future's drop flag, and poll the runtime afterward.
#[test]
fn block_on_drops_future_on_driver_panic() {
    struct PanicDriver;

    impl Driver for PanicDriver {
        fn handle(&self) -> Box<dyn Notify> {
            Box::new(NoopNotify)
        }

        fn run(&self, rt: &Runtime) -> io::Result<()> {
            rt.poll();
            panic!("driver panic");
        }
    }

    let rt = Runtime::new(Box::new(NoopNotify));
    let dropped = Cell::new(false);
    let res = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        rt.block_on(Borrowing(&dropped, None), &PanicDriver);
    }));
    assert!(res.is_err());
    assert!(dropped.get());
    rt.poll();
}

#[derive(Debug, Default, Clone)]
struct Counter(Arc<AtomicUsize>);

impl Counter {
    fn get(&self) -> usize {
        self.0.load(Ordering::SeqCst)
    }

    fn inc(&self) {
        self.0.fetch_add(1, Ordering::SeqCst);
    }

    fn token(&self) -> Token {
        Token(self.clone())
    }
}

/// Increments the counter on drop.
#[derive(Debug)]
struct Token(Counter);

impl Drop for Token {
    fn drop(&mut self) {
        self.0.inc();
    }
}

#[derive(Default)]
struct CountWaker(AtomicUsize);

impl std::task::Wake for CountWaker {
    fn wake(self: Arc<Self>) {
        self.0.fetch_add(1, Ordering::SeqCst);
    }
}

fn drain(rt: &Runtime) {
    CURRENT_RUNTIME.set(rt, || while rt.poll() == PollResult::PollAgain {});
}

async fn yield_once() {
    let mut yielded = false;
    poll_fn(|cx| {
        if yielded {
            Poll::Ready(())
        } else {
            yielded = true;
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    })
    .await;
}

/// STATEMENT: Remote joins receive results when completion races join-waker registration.
///
/// HOW:
/// - Repeatedly submit numbered futures from a worker, varying their number of yields.
/// - Block on each remote join while the owner polls, and check every result and final completion count.
#[test]
fn remote_join_result_race() {
    const N: usize = iters(2000, 50);

    let rt = Runtime::new(Box::new(NoopNotify));
    let handle = rt.handle();
    let done = Counter::default();
    let done2 = done.clone();
    let th = thread::spawn(move || {
        for i in 0..N {
            // completes while the joining thread registers its waker
            let h = handle.spawn(async move {
                for _ in 0..i % 4 {
                    yield_once().await;
                }
                i
            });
            assert_eq!(block_on_thread(h).unwrap(), i);
            done2.inc();
        }
    });
    poll_until(&rt, || done.get() == N);
    th.join().unwrap();
}

/// STATEMENT: Remote join results are reclaimed once without retaining owner task allocations.
///
/// HOW:
/// - Drop remote handles before completion, after owner shutdown, or consume their result after shutdown.
/// - Drain the owner and check task allocations are gone while retained results remain undropped.
/// - Shut down, release the worker, and check each result token is dropped exactly once.
#[test]
fn remote_join_result_reclamation() {
    for path in 0..3 {
        let rt = Runtime::new(Box::new(NoopNotify));
        let out = Counter::default();
        let value = out.token();
        let handle = rt.handle();
        let (spawned_tx, spawned) = mpsc::channel();
        let (stopped, stopped_rx) = mpsc::channel::<()>();
        let th = thread::spawn(move || {
            let mut task = Some(handle.spawn(async move { value }));
            drop(handle);
            if path == 0 {
                drop(task.take());
            }
            spawned_tx.send(()).unwrap();
            stopped_rx.recv().unwrap();
            if let Some(task) = task {
                if path == 1 {
                    drop(task);
                } else {
                    drop(block_on_thread(task).unwrap());
                }
            }
        });
        spawned.recv().unwrap();
        drain(&rt);
        if path != 0 {
            assert_eq!(out.get(), 0);
        }
        assert_eq!(rt.queue.task_count(), 0);
        drop(rt);
        stopped.send(()).unwrap();
        th.join().unwrap();
        assert_eq!(out.get(), 1);
    }
}

/// STATEMENT: Remote cancellation destroys queued and already-polled futures on their owner thread.
///
/// HOW:
/// - Cancel a remotely submitted task before polling; check its body never runs and its drop thread.
/// - Start a second pending task, signal its remote handle to cancel, and check owner-thread destruction.
#[test]
fn remote_join_cancel() {
    // cancelled while queued, the future is dropped on the owner unpolled
    let rt = Runtime::new(Box::new(NoopNotify));
    let owner = thread::current().id();
    let polled = Arc::new(AtomicBool::new(false));
    let dropped_on = Arc::new(Mutex::new(None));
    let handle = rt.handle();
    let (polled2, dropped2) = (polled.clone(), dropped_on.clone());
    thread::spawn(move || {
        let guard = DropThread(dropped2);
        let h = handle.spawn(async move {
            let _guard = guard;
            polled2.store(true, Ordering::SeqCst);
        });
        assert!(!h.is_finished());
        h.cancel();
    })
    .join()
    .unwrap();
    drain(&rt);
    assert!(!polled.load(Ordering::SeqCst));
    assert_eq!(*dropped_on.lock().unwrap(), Some(owner));

    // cancelled while running
    let dropped_on = Arc::new(Mutex::new(None));
    let handle = rt.handle();
    let dropped2 = dropped_on.clone();
    let (started_tx, started) = mpsc::channel();
    let (cancel_tx, cancel_rx) = mpsc::channel::<()>();
    let th = thread::spawn(move || {
        let guard = DropThread(dropped2);
        let h = handle.spawn(async move {
            let _guard = guard;
            started_tx.send(()).unwrap();
            std::future::pending::<()>().await;
        });
        cancel_rx.recv().unwrap();
        h.cancel();
    });
    poll_until(&rt, || started.try_recv().is_ok());
    assert!(dropped_on.lock().unwrap().is_none());
    cancel_tx.send(()).unwrap();
    th.join().unwrap();
    poll_until(&rt, || dropped_on.lock().unwrap().is_some());
    assert_eq!(*dropped_on.lock().unwrap(), Some(owner));
}

/// STATEMENT: A remote join handle observes completion and retains the task's result.
///
/// HOW:
/// - Submit a task returning seven and check its remote handle is unfinished before owner polling.
/// - Drain the owner, signal the worker, then check `is_finished` and the joined result.
#[test]
fn remote_join_is_finished() {
    let rt = Runtime::new(Box::new(NoopNotify));
    let handle = rt.handle();
    let (spawned_tx, spawned) = mpsc::channel();
    let (ran, ran_rx) = mpsc::channel::<()>();
    let th = thread::spawn(move || {
        let h = handle.spawn(async { 7 });
        assert!(!h.is_finished());
        spawned_tx.send(()).unwrap();
        ran_rx.recv().unwrap();
        assert!(h.is_finished());
        assert_eq!(block_on_thread(h).unwrap(), 7);
    });
    spawned.recv().unwrap();
    drain(&rt);
    ran.send(()).unwrap();
    th.join().unwrap();
}

/// STATEMENT: `Handle::spawn` on the owner uses the local queue and executes on that same thread.
///
/// HOW:
/// - Spawn through a handle and inspect the empty remote queue and nonempty local queue.
/// - Poll to completion and compare the returned thread ID with the owner thread ID.
#[test]
fn local_handle_spawn() {
    let rt = Runtime::new(Box::new(NoopNotify));
    let handle = rt.handle();
    let h = handle.spawn(async { thread::current().id() });
    // spawned directly on the owner, not through the wake queue
    assert!(rt.queue.wake_queue.is_empty());
    assert!(!rt.queue.local_queue.is_empty());
    poll_until(&rt, || h.is_finished());
    assert_eq!(block_on_thread(h).unwrap(), thread::current().id());
}

struct DropThread(Arc<Mutex<Option<thread::ThreadId>>>);

impl Drop for DropThread {
    fn drop(&mut self) {
        *self.0.lock().unwrap() = Some(thread::current().id());
    }
}

/// STATEMENT: Joined, detached, dropped, and cancelled task outputs are each destroyed once.
///
/// HOW:
/// - Complete token-producing tasks and check outputs stay live until dropping or cancelling their handles.
/// - Detach a yielding task and wait for its output drop.
/// - Consume a joined output, drop it, drain, and check one destruction in every path.
#[test]
fn unconsumed_output_dropped() {
    let rt = Runtime::new(Box::new(NoopNotify));
    for cancel in [false, true] {
        let out = Counter::default();
        let tok = out.token();
        let h = rt.spawn(async move { tok });
        poll_until(&rt, || h.is_finished());
        assert_eq!(out.get(), 0);
        if cancel {
            h.cancel();
        } else {
            drop(h);
        }
        assert_eq!(out.get(), 1);
    }

    // detached before completion
    let out = Counter::default();
    let tok = out.token();
    drop(rt.spawn(async move {
        yield_once().await;
        tok
    }));
    poll_until(&rt, || out.get() == 1);

    // consumed by the join handle
    let out = Counter::default();
    let tok = out.token();
    let h = rt.spawn(async move { tok });
    poll_until(&rt, || h.is_finished());
    drop(block_on_thread(h).unwrap());
    drain(&rt);
    assert_eq!(out.get(), 1);
}

/// STATEMENT: Shutdown drops an unactivated remote submission without polling its future.
///
/// HOW:
/// - Remotely submit a future with separate poll and drop flags, then drop the owner runtime.
/// - Check the drop flag is set and the poll flag remains clear.
#[test]
fn remote_shutdown_before_run() {
    let rt = Runtime::new(Box::new(NoopNotify));
    let dropped = Arc::new(AtomicBool::new(false));
    let polled = Arc::new(AtomicBool::new(false));
    let hnd = rt.handle();
    let flag = DropFlag(dropped.clone());
    let polled2 = polled.clone();
    thread::spawn(move || {
        hnd.spawn(async move {
            let _flag = flag;
            polled2.store(true, Ordering::SeqCst);
        });
    })
    .join()
    .unwrap();
    drop(rt);
    assert!(dropped.load(Ordering::SeqCst));
    assert!(!polled.load(Ordering::SeqCst));
}

/// STATEMENT: Cancellation racing remote completion drops the future and every created output once.
///
/// HOW:
/// - Repeatedly park a token-producing future, remotely wake it, poll once, and cancel its handle.
/// - Join the waking thread and poll until the drop flag is set and created and dropped outputs match.
#[test]
fn cancel_complete_race() {
    let rt = Runtime::new(Box::new(NoopNotify));
    let created = Counter::default();
    let out = Counter::default();
    for _ in 0..iters(500, 10) {
        let dropped = Arc::new(AtomicBool::new(false));
        let flag = DropFlag(dropped.clone());
        let slot = Slot::default();
        let (slot2, created2, out2) = (slot.clone(), created.clone(), out.clone());
        let h = rt.spawn(async move {
            let _flag = flag;
            let mut first = true;
            poll_fn(|cx| {
                if first {
                    first = false;
                    *slot2.lock().unwrap() = Some(cx.waker().clone());
                    Poll::Pending
                } else {
                    Poll::Ready(())
                }
            })
            .await;
            created2.inc();
            out2.token()
        });
        poll_until(&rt, || slot.lock().unwrap().is_some());

        let w = slot.lock().unwrap().take().unwrap();
        let t1 = thread::spawn(move || w.wake());
        rt.poll();
        h.cancel();
        poll_until(&rt, || t1.is_finished());
        t1.join().unwrap();
        // the output is dropped once, whichever side wins
        poll_until(&rt, || {
            dropped.load(Ordering::SeqCst) && created.get() == out.get()
        });
    }
}

/// STATEMENT: Stale local and remote wakers do not repoll a completed task or retain its output.
///
/// HOW:
/// - Complete a two-poll task while keeping extra wakers, then wake them locally and remotely.
/// - Drain and check the poll count stays two; drop the join handle and check one output drop.
/// - Release the remaining wakers, including a remote consuming wake, and drain cleanup.
#[test]
fn stale_wakers() {
    let rt = Runtime::new(Box::new(NoopNotify));
    let slot = Slot::default();
    let polls = Arc::new(AtomicUsize::new(0));
    let out = Counter::default();
    let tok = out.token();
    let (slot2, polls2) = (slot.clone(), polls.clone());
    let h = rt.spawn(async move {
        wait_polls(slot2, polls2, 2).await;
        tok
    });
    poll_until(&rt, || polls.load(Ordering::SeqCst) == 1);

    let w = slot.lock().unwrap().take().unwrap();
    let (w2, w3) = (w.clone(), w.clone());
    w.wake();
    poll_until(&rt, || h.is_finished());

    // wakes after completion do not run the task
    w2.wake_by_ref();
    let w3 = thread::spawn(move || {
        w3.wake_by_ref();
        w3
    })
    .join()
    .unwrap();
    drain(&rt);
    assert_eq!(polls.load(Ordering::SeqCst), 2);

    // the output is dropped with the join handle
    drop(h);
    assert_eq!(out.get(), 1);
    // the last waker, dropped on another thread, frees the task
    drop(w2);
    thread::spawn(move || w3.wake()).join().unwrap();
    drain(&rt);
}

/// STATEMENT: An unclaimed output containing its own task waker does not form a retained cycle.
///
/// HOW:
/// - Produce an output containing a cloned task waker and a drop flag.
/// - Detach or drop before completion, or detach, cancel, or drop after completion.
/// - Drain each case and check the output's drop flag.
#[test]
#[allow(deprecated, reason = "verify legacy detach releases unclaimed outputs")]
fn unclaimed_output_with_own_waker() {
    // the output holds a waker of its task, a cycle unless it is dropped
    for case in 0..5 {
        let rt = Runtime::new(Box::new(NoopNotify));
        let dropped = Arc::new(AtomicBool::new(false));
        let mut flag = Some(DropFlag(dropped.clone()));
        let h = rt.spawn(poll_fn(move |cx| {
            Poll::Ready((cx.waker().clone(), flag.take().unwrap()))
        }));
        match case {
            0 => h.detach(),
            1 => drop(h),
            2 => {
                drain(&rt);
                h.detach();
            }
            3 => {
                drain(&rt);
                h.cancel();
            }
            _ => {
                drain(&rt);
                drop(h);
            }
        }
        drain(&rt);
        assert!(dropped.load(Ordering::SeqCst), "case {case}");
    }
}

struct ReDrop {
    waker: Rc<RefCell<Option<Waker>>>,
    other: Option<JoinHandle<()>>,
    hnd: Handle,
    spawned: Arc<AtomicBool>,
    dropped: Arc<AtomicBool>,
}

impl Drop for ReDrop {
    fn drop(&mut self) {
        // wakes itself, drops a handle and spawns a task
        if let Some(w) = self.waker.borrow_mut().take() {
            w.wake();
        }
        drop(self.other.take());
        let flag = DropFlag(self.spawned.clone());
        self.hnd.spawn(async move {
            let _flag = flag;
        });
        self.dropped.store(true, Ordering::SeqCst);
    }
}

/// STATEMENT: Cancellation and shutdown tolerate destructors that wake, release handles, and spawn.
///
/// HOW:
/// - Poll a task whose destructor wakes itself, drops another handle, and submits a guarded task.
/// - Trigger cancellation or queued shutdown and check all three destructor flags become set.
#[test]
fn reentrant_drop() {
    for clear in [false, true] {
        let rt = Runtime::new(Box::new(NoopNotify));
        let other_dropped = Arc::new(AtomicBool::new(false));
        let spawned = Arc::new(AtomicBool::new(false));
        let dropped = Arc::new(AtomicBool::new(false));
        let flag = DropFlag(other_dropped.clone());
        let other = rt.spawn(async move {
            let _flag = flag;
            poll_fn(|_| Poll::<()>::Pending).await;
        });
        let waker = Rc::new(RefCell::new(None::<Waker>));
        let redrop = ReDrop {
            waker: waker.clone(),
            other: Some(other),
            hnd: rt.handle(),
            spawned: spawned.clone(),
            dropped: dropped.clone(),
        };
        let waker2 = waker.clone();
        let h = rt.spawn(async move {
            let _redrop = redrop;
            poll_fn(|cx| {
                *waker2.borrow_mut() = Some(cx.waker().clone());
                Poll::<()>::Pending
            })
            .await;
        });
        poll_until(&rt, || waker.borrow().is_some());

        if clear {
            // queued when the runtime is dropped
            drop(h);
            waker.borrow().as_ref().unwrap().wake_by_ref();
            drop(rt);
        } else {
            h.cancel();
            poll_until(&rt, || {
                spawned.load(Ordering::SeqCst) && other_dropped.load(Ordering::SeqCst)
            });
        }
        assert!(dropped.load(Ordering::SeqCst));
        assert!(other_dropped.load(Ordering::SeqCst));
        assert!(spawned.load(Ordering::SeqCst));
    }
}

/// Wakes its task and polls the runtime on drop.
struct PollOnDrop {
    waker: Rc<RefCell<Option<Waker>>>,
    drops: Counter,
}

impl Drop for PollOnDrop {
    fn drop(&mut self) {
        self.drops.inc();
        let waker = self.waker.borrow_mut().take();
        if let Some(w) = waker {
            w.wake();
        }
        Runtime::with_current(|rt| while rt.poll() == PollResult::PollAgain {});
    }
}

/// STATEMENT: Reentrant polling from a dropping future cannot poll or destroy that future again.
///
/// HOW:
/// - Poll a pending task whose guard wakes its task and polls the current runtime on drop.
/// - Cancel it or drop the runtime with it queued, then check exactly one future poll and guard drop.
#[test]
fn reentrant_poll_in_drop() {
    for clear in [false, true] {
        let rt = Runtime::new(Box::new(NoopNotify));
        let drops = Counter::default();
        let polls = Counter::default();
        let waker = Rc::new(RefCell::new(None::<Waker>));
        let guard = PollOnDrop {
            waker: waker.clone(),
            drops: drops.clone(),
        };
        let (waker2, polls2) = (waker.clone(), polls.clone());
        let h = rt.spawn(async move {
            let _guard = guard;
            poll_fn(|cx| {
                polls2.inc();
                *waker2.borrow_mut() = Some(cx.waker().clone());
                Poll::<()>::Pending
            })
            .await;
        });
        poll_until(&rt, || waker.borrow().is_some());

        if clear {
            // queued when the runtime is dropped
            drop(h);
            waker.borrow().as_ref().unwrap().wake_by_ref();
            drop(rt);
        } else {
            h.cancel();
            drain(&rt);
        }
        // the future is neither polled nor dropped again
        assert_eq!(drops.get(), 1);
        assert_eq!(polls.get(), 1);
    }
}

/// Waker that runs a callback when it is cloned or released.
struct Hook {
    refs: Cell<usize>,
    on_clone: Cell<Option<Box<dyn FnOnce()>>>,
    on_release: Cell<Option<Box<dyn FnOnce()>>>,
}

impl Hook {
    const VTABLE: RawWakerVTable =
        RawWakerVTable::new(Self::clone, Self::release, |_| (), Self::release);

    fn waker(on_clone: Option<Box<dyn FnOnce()>>, on_release: Option<Box<dyn FnOnce()>>) -> Waker {
        let hook = Box::into_raw(Box::new(Hook {
            refs: Cell::new(1),
            on_clone: Cell::new(on_clone),
            on_release: Cell::new(on_release),
        }));
        unsafe { Waker::from_raw(RawWaker::new(hook.cast(), &Self::VTABLE)) }
    }

    unsafe fn clone(p: *const ()) -> RawWaker {
        let hook = unsafe { &*p.cast::<Hook>() };
        hook.refs.set(hook.refs.get() + 1);
        let cloned = unsafe { Waker::from_raw(RawWaker::new(p, &Self::VTABLE)) };
        if let Some(f) = hook.on_clone.take() {
            f();
        }
        std::mem::forget(cloned);
        RawWaker::new(p, &Self::VTABLE)
    }

    unsafe fn release(p: *const ()) {
        let hook = unsafe { &*p.cast::<Hook>() };
        hook.refs.set(hook.refs.get() - 1);
        if hook.refs.get() == 0 {
            if let Some(f) = hook.on_release.take() {
                f();
            }
            drop(unsafe { Box::from_raw(p.cast_mut().cast::<Hook>()) });
        }
    }
}

/// STATEMENT: A join observes completion triggered while cloning or releasing a replacement waker.
///
/// HOW:
/// - Register an initial join waker on a pending task and leave its last reference in the handle.
/// - Replace it with hooks that complete and drain during old-waker release or new-waker clone.
/// - Check the replacement poll immediately returns the successful result in both cases.
#[test]
fn join_completes_in_waker_callback() {
    // the target task completes while the join handle replaces its waker
    for in_clone in [false, true] {
        let rt = Runtime::new(Box::new(NoopNotify));
        CURRENT_RUNTIME.set(&rt, || {
            let ready = Rc::new(Cell::new(false));
            let waker = Rc::new(RefCell::new(None::<Waker>));
            let (ready2, waker2) = (ready.clone(), waker.clone());
            let mut h = rt.spawn(poll_fn(move |cx| {
                if ready2.get() {
                    Poll::Ready(1)
                } else {
                    *waker2.borrow_mut() = Some(cx.waker().clone());
                    Poll::Pending
                }
            }));
            while rt.poll() == PollResult::PollAgain {}

            let complete: Box<dyn FnOnce()> = Box::new(move || {
                ready.set(true);
                waker.borrow_mut().take().unwrap().wake();
                Runtime::with_current(|rt| while rt.poll() == PollResult::PollAgain {});
            });
            let (on_clone, on_release) = if in_clone {
                (Some(complete), None)
            } else {
                (None, Some(complete))
            };

            let first = Hook::waker(None, on_release);
            let res = Pin::new(&mut h).poll(&mut Context::from_waker(&first));
            assert!(res.is_pending());
            // the join handle keeps the last reference
            drop(first);

            let second = Hook::waker(on_clone, None);
            let res = Pin::new(&mut h).poll(&mut Context::from_waker(&second));
            assert!(matches!(res, Poll::Ready(Ok(1))));
        });
    }
}

/// STATEMENT: Completion during a panicking join-waker clone preserves the result and releases storage.
///
/// HOW:
/// - Register a join waker, then replace it with a hook that drains the task and panics during cloning.
/// - Catch the panic, check completion, and retrieve the `Rc` result through the original handle.
/// - Drop the hook waker, check its drop flag, drain, and check zero task allocations.
#[test]
fn join_completion_survives_waker_clone_panic() {
    let rt = Runtime::new(Box::new(NoopNotify));
    CURRENT_RUNTIME.set(&rt, || {
        let mut task = rt.spawn(async { Rc::new(42) });
        assert!(
            Pin::new(&mut task)
                .poll(&mut Context::from_waker(Waker::noop()))
                .is_pending()
        );
        let dropped = Arc::new(AtomicBool::new(false));
        let flag = DropFlag(dropped.clone());
        let waker = Hook::waker(
            Some(Box::new(|| {
                Runtime::with_current(drain);
                panic!("clone after completion");
            })),
            Some(Box::new(move || drop(flag))),
        );
        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            Pin::new(&mut task).poll(&mut Context::from_waker(&waker))
        }));
        assert!(result.is_err());
        assert!(task.is_finished());
        assert_eq!(*block_on_thread(task).unwrap(), 42);
        drop(waker);
        assert!(dropped.load(Ordering::SeqCst));
        drain(&rt);
        assert_eq!(rt.queue.task_count(), 0);
    });
}

#[cfg(ntex_loom)]
/// STATEMENT: Reentrant join callbacks and destructor polling preserve lifecycle invariants under Loom.
///
/// HOW:
/// - Model completion during waker replacement and completion followed by a clone panic.
/// - Run destructor-driven reentrant polling in the same model, retaining each scenario's assertions.
#[test]
fn task_loom_reentrant_lifecycle() {
    loom::model(|| {
        join_completes_in_waker_callback();
        join_completion_survives_waker_clone_panic();
        reentrant_poll_in_drop();
    });
}

#[cfg(ntex_loom)]
/// STATEMENT: Panicking future destructors preserve cleanup and join failure under Loom instrumentation.
///
/// HOW:
/// - Run `panic_in_future_drop` inside `loom::model`.
/// - Reuse its ready and cancelled paths, drop-count checks, and failed-join assertions.
#[test]
fn task_loom_panicking_future_drop() {
    loom::model(panic_in_future_drop);
}

/// STATEMENT: Future-drop panics clean up other fields exactly once and leave handles failed and finished.
///
/// HOW:
/// - Run a ready future whose destructor panics, then repeat with cancellation before a second ready task.
/// - Catch each drop panic, check the companion flag, and check the surviving handle fails to join.
/// - Drain remaining work and check one destructor call per spawned future.
#[test]
fn panic_in_future_drop() {
    struct PanicOnDrop(Counter);

    impl Drop for PanicOnDrop {
        fn drop(&mut self) {
            self.0.inc();
            panic!("drop panic");
        }
    }

    struct PanicFut {
        _panic: PanicOnDrop,
        _flag: DropFlag,
        ready: bool,
    }

    impl Future for PanicFut {
        type Output = usize;

        fn poll(self: Pin<&mut Self>, _: &mut std::task::Context<'_>) -> Poll<usize> {
            if self.ready { Poll::Ready(1) } else { Poll::Pending }
        }
    }

    let rt = Runtime::new(Box::new(NoopNotify));
    for ready in [true, false] {
        let drops = Counter::default();
        let dropped = Arc::new(AtomicBool::new(false));
        let h = rt.spawn(PanicFut {
            _panic: PanicOnDrop(drops.clone()),
            _flag: DropFlag(dropped.clone()),
            ready,
        });
        let h = if ready {
            h
        } else {
            drain(&rt);
            h.cancel();
            rt.spawn(PanicFut {
                _panic: PanicOnDrop(drops.clone()),
                _flag: DropFlag(dropped.clone()),
                ready: true,
            })
        };
        let res = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| drain(&rt)));
        assert!(res.is_err());
        assert!(dropped.load(Ordering::SeqCst));
        if !ready {
            // the second task panics on the next run
            let res = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| drain(&rt)));
            assert!(res.is_err());
        }
        assert!(h.is_finished());
        assert!(block_on_thread(h).is_err());
        drain(&rt);
        assert_eq!(drops.get(), if ready { 1 } else { 2 });
    }
}

/// STATEMENT: Only the latest registered join waker is notified when a task completes.
///
/// HOW:
/// - Poll a pending join with two different counting wakers, replacing the first with the second.
/// - Wake the task, run until the second is notified, and check the first has zero notifications.
/// - Poll the join again and check the two-poll task's result.
#[test]
fn join_waker_replaced() {
    fn poll_with(h: &mut JoinHandle<usize>, w: &Arc<CountWaker>) -> Poll<usize> {
        let w = Waker::from(w.clone());
        let mut cx = std::task::Context::from_waker(&w);
        Pin::new(h).poll(&mut cx).map(Result::unwrap)
    }

    let rt = Runtime::new(Box::new(NoopNotify));
    {
        let slot = Slot::default();
        let polls = Arc::new(AtomicUsize::new(0));
        let mut h = rt.spawn(wait_polls(slot.clone(), polls.clone(), 2));
        poll_until(&rt, || polls.load(Ordering::SeqCst) == 1);

        let a = Arc::new(CountWaker::default());
        let b = Arc::new(CountWaker::default());
        assert!(poll_with(&mut h, &a).is_pending());
        assert!(poll_with(&mut h, &b).is_pending());

        slot.lock().unwrap().take().unwrap().wake();
        poll_until(&rt, || b.0.load(Ordering::SeqCst) == 1);
        assert_eq!(a.0.load(Ordering::SeqCst), 0);
        assert_eq!(poll_with(&mut h, &b), Poll::Ready(2));
    }
}