moqtap_proxy/

release_timer.rs

//! A process-wide release wheel: one dedicated OS thread, one min-heap of
//! deadlines, one `CancellationToken` per armed deadline.
//!
//! `tokio::time::sleep` cannot be used for release timing. Tokio parks its
//! runtime on a wait with a millisecond timeout, and on Windows every
//! *interruptible* wait is quantised to the ~15.6 ms system tick.
//! `std::thread::sleep` is
//! the exception — std implements it with a high-resolution waitable timer,
//! so it lands within ~0.3 ms — but it cannot be woken early.
//!
//! This module recovers *accurate *and* interruptible* in plain safe code: the
//! thread sleeps in [`SLICE`]-bounded steps and re-reads the heap between them,
//! and parks on an untimed `Condvar` — which *is* woken promptly — when there
//! is nothing to wait for.
//!
//! Nothing here is `pub`, the whole module is safe code, and there is no
//! `#[cfg]` outside its test module: one implementation runs on every
//! platform. That is what lets this file move into `quinn-netem`, give
//! each impaired socket its own wheel, or be replaced wholesale, without
//! any of it being a breaking change.

use std::cmp::{Ordering, Reverse};
use std::collections::BinaryHeap;
use std::sync::{Arc, Condvar, Mutex, OnceLock, Weak};
use std::thread::JoinHandle;
use std::time::{Duration, Instant};
use tokio_util::sync::CancellationToken;

use crate::instrument::TimerBackend;

/// How long the release thread may sleep without re-reading the heap.
///
/// This is a responsiveness knob, not an accuracy one: it sets how fast a
/// newly armed *nearer* deadline is noticed, never the accuracy of one the
/// wheel already holds. The last sleep before a deadline is
/// `min(remaining, SLICE) == remaining`, so accuracy comes from
/// `std::thread::sleep` whatever this constant is.
const SLICE: Duration = Duration::from_micros(500);
/// Heap length below which the sweep never runs.
///
/// Below this many entries the residue is bounded and cheap (~40 bytes a
/// slot), and rebuilding a short heap on a timer would trade that bounded
/// residue for an unbounded `retain` rate.
const SWEEP_MIN_LEN: usize = 64;
/// Minimum interval between two sweeps.
const SWEEP_INTERVAL: Duration = Duration::from_millis(250);

/// A registered deadline. Cheap to clone (one `Arc` + one token clone).
///
/// Dropping every clone abandons the deadline: the wheel drops the slot at
/// its own instant, or at the next sweep, whichever comes first. There is
/// no unregister call.
#[derive(Clone, Debug)]
pub(crate) struct Deadline {
    entry: Arc<Entry>,
}

impl Deadline {
    /// The token the wheel cancels once `Instant::now() >= at`.
    ///
    /// Level-triggered: awaiting it after the wheel has fired resolves
    /// immediately, so the engine may create and drop `token().cancelled()`
    /// on every `select!` iteration without a lost-wakeup race and without
    /// re-registering with the wheel.
    pub(crate) fn token(&self) -> &CancellationToken {
        &self.entry.token
    }

    /// A deadline that has already passed. Allocates one token, never
    /// touches the wheel, never starts the release thread.
    fn already_due() -> Self {
        let token = CancellationToken::new();
        token.cancel();
        Self { entry: Arc::new(Entry { token }) }
    }
}

/// Register `at` with the process-wide wheel.
///
/// **The sole lazy-initialisation trigger for the release thread.** When
/// `at <= Instant::now()` it returns an already-cancelled [`Deadline`]
/// without constructing the wheel, so `Delay { by: 0 }`, a delay swallowed
/// by the monotonic clamp, and every session that never delays all leave
/// [`crate::instrument::release_timer_started`] `false`.
pub(crate) fn arm_at(at: Instant) -> Deadline {
    if at <= Instant::now() {
        return Deadline::already_due();
    }
    shared().arm(at)
}

/// The process-wide wheel, constructed on first use. Never dropped: it lives in
/// a `static OnceLock<ReleaseTimer>`, so its thread is never joined and the
/// *atexit joins a thread that holds a lock* hazard cannot arise.
fn shared() -> &'static ReleaseTimer {
    static SHARED: OnceLock<ReleaseTimer> = OnceLock::new();
    SHARED.get_or_init(|| {
        let timer = ReleaseTimer::new();
        // Published here and not in `ReleaseTimer::new`, which also runs
        // for owned wheels (this module's tests, any per-socket owner)
        // that are not the process-wide one.
        crate::instrument::note_release_timer_started(timer.backend());
        timer
    })
}

/// Whether the process-wide wheel has been constructed.
///
/// Read by this module's tests; every non-test caller in the crate wants
/// [`backend`], which answers the same question and says which one.
#[allow(dead_code)]
pub(crate) fn started() -> bool {
    crate::instrument::release_timer_started()
}

/// Which primitive the process-wide wheel's thread waits on, or `None` if
/// it was never constructed. This is what
/// [`crate::instrument::release_timer_backend`] returns.
pub(crate) fn backend() -> Option<TimerBackend> {
    crate::instrument::release_timer_backend()
}

/// An owned release wheel: one OS thread, one heap, one parker.
///
/// The process-wide wheel is one of these in a `OnceLock`. Instances
/// created directly (this module's tests today, per-socket owners later)
/// join their thread on drop — unless the drop is itself running on that
/// thread, which the `Drop` impl below explains.
pub(crate) struct ReleaseTimer {
    inner: Arc<Inner>,
    thread: Option<JoinHandle<()>>,
}

impl ReleaseTimer {
    /// Resolve a backend and start the release thread.
    ///
    /// Honours `MOQTAP_RELEASE_TIMER` (`slice` | `condvar`), read once,
    /// here. Unset or unrecognised means `slice`. **Cannot fail**, and
    /// that is a property of the design rather than an omission: there is
    /// no OS object to fail to create, so there is no fallback ladder and
    /// no error path to test.
    pub(crate) fn new() -> Self {
        let parker = match std::env::var("MOQTAP_RELEASE_TIMER").as_deref() {
            Ok("condvar") => Parker::Condvar,
            // An unrecognised value resolves to the sliced sleeper rather
            // than panicking: this is a diagnostic knob read inside a
            // forwarding task, and `backend()` reports what was actually
            // chosen, so a typo is observable without being fatal.
            _ => Parker::SleepSlice,
        };
        let inner = Arc::new(Inner {
            state: Mutex::new(State {
                heap: BinaryHeap::new(),
                seq: 0,
                stopping: false,
                wakes: 0,
                last_sweep: Instant::now(),
            }),
            cv: Condvar::new(),
            parker,
        });
        let thread_inner = Arc::clone(&inner);
        let thread = std::thread::Builder::new()
            .name("moqtap-release".to_owned())
            .spawn(move || run(thread_inner))
            .expect("release timer thread");
        Self { inner, thread: Some(thread) }
    }

    /// Register `at`. `at <= now` yields an already-cancelled [`Deadline`]
    /// without touching the heap or waking the thread.
    pub(crate) fn arm(&self, at: Instant) -> Deadline {
        if at <= Instant::now() {
            return Deadline::already_due();
        }
        let entry = Arc::new(Entry { token: CancellationToken::new() });
        {
            let mut g = self.inner.state.lock().expect("release wheel state");
            g.seq += 1;
            let seq = g.seq;
            g.heap.push(Reverse(Slot { at, seq, entry: Arc::downgrade(&entry) }));
        } // guard released before notify
        self.inner.cv.notify_one(); // a no-op when nobody waits
        Deadline { entry }
    }

    /// Which primitive this wheel's thread waits on.
    pub(crate) fn backend(&self) -> TimerBackend {
        match self.inner.parker {
            Parker::SleepSlice => TimerBackend::SleepSlice,
            Parker::Condvar => TimerBackend::Condvar,
        }
    }

    /// Heap length, including slots whose [`Deadline`] was dropped and
    /// that have not been popped or swept yet.
    #[cfg(test)]
    pub(crate) fn debug_len(&self) -> usize {
        self.inner.state.lock().expect("release wheel state").heap.len()
    }

    /// Loop iterations the release thread has completed — one per wake.
    /// Lets a test assert that an empty wheel costs nothing: the count must
    /// not move across an idle window.
    #[cfg(test)]
    pub(crate) fn debug_wakes(&self) -> u64 {
        self.inner.state.lock().expect("release wheel state").wakes
    }
}

/// Sets `stopping`, wakes the thread, and joins it — except when the drop
/// is itself running *on* that thread, where it detaches instead. Pending
/// deadlines are abandoned, not fired: a dropped wheel has no listeners
/// left.
///
/// # Why the join is conditional
///
/// From the release thread the join is not a slow path, it is a permanent
/// deadlock. Step 3 of [`run`] cancels tokens *outside* the lock and *on*
/// the release thread, so every waker registered on a deadline — which is
/// what awaiting `token().cancelled()` installs — runs there too. A waker
/// that drops the last handle to the wheel enters this function on the
/// release thread, and `join()` then waits for the thread executing it.
/// `stopping` and `notify_all` do not rescue it: the thread is inside the
/// cancel loop and can never return to the top of `run` to read the flag.
///
/// Detaching there is not a compromise, it is the only outcome that
/// terminates. `stopping` is set and `notify_all` sent before the branch,
/// so the detached thread finishes the cancel loop it is in, sees the flag
/// on its next iteration and returns on its own: a bounded, self-clearing
/// thread rather than an unbounded hang. The handle already carries the
/// id `spawn` gave the thread, so no separate field is needed to
/// recognise it.
impl Drop for ReleaseTimer {
    fn drop(&mut self) {
        {
            let mut g = self.inner.state.lock().expect("release wheel state");
            g.stopping = true;
        }
        self.inner.cv.notify_all();
        if let Some(handle) = self.thread.take() {
            if handle.thread().id() == std::thread::current().id() {
                return; // detach: joining here would join this thread to itself
            }
            let _ = handle.join();
        }
    }
}

// ── internals ──────────────────────────────────────────────────────────

/// What a [`Deadline`] and its heap slot share.
#[derive(Debug)]
struct Entry {
    token: CancellationToken,
}

/// A heap slot. `Weak`, so a dropped [`Deadline`] costs a skipped pop
/// rather than a token nobody holds.
struct Slot {
    at: Instant,
    seq: u64,
    entry: Weak<Entry>,
}

// `Weak` is neither `Ord` nor `Eq`, so these cannot be derived. Order is
// `(at, seq)`, `seq` monotonic, so equal instants pop in registration
// order. The heap is a `BinaryHeap<Reverse<Slot>>`, i.e. a *min*-heap:
// inverting either half of this pair turns the wheel into a LIFO that
// serves the farthest deadline first, and nothing but a wrong-looking
// timing test would say so — which is why
// `the_heap_is_a_min_heap_and_ties_break_by_registration` below tests it
// purely.
impl PartialEq for Slot {
    fn eq(&self, o: &Self) -> bool {
        (self.at, self.seq) == (o.at, o.seq)
    }
}
impl Eq for Slot {}
impl Ord for Slot {
    fn cmp(&self, o: &Self) -> Ordering {
        (self.at, self.seq).cmp(&(o.at, o.seq))
    }
}
impl PartialOrd for Slot {
    fn partial_cmp(&self, o: &Self) -> Option<Ordering> {
        Some(self.cmp(o))
    }
}

/// Everything the release thread and its registrars share under one lock.
struct State {
    /// A *min*-heap of armed deadlines.
    heap: BinaryHeap<Reverse<Slot>>,
    /// Monotonic registration counter, the tie-break in [`Slot`]'s `Ord`.
    seq: u64,
    /// Set by [`ReleaseTimer`]'s `Drop`; the thread returns on its next
    /// wake.
    stopping: bool,
    /// Loop iterations. Always compiled — it is one `u64` increment under
    /// a lock the thread already holds — and read only by `debug_wakes`.
    /// The `+=` counts as a read, so `dead_code` does not fire on it in a
    /// non-test build.
    wakes: u64,
    /// When the heap was last rebuilt to drop abandoned slots.
    last_sweep: Instant,
}

/// The shared half of a wheel: what the thread and its registrars both
/// hold an `Arc` to.
struct Inner {
    state: Mutex<State>,
    cv: Condvar,
    parker: Parker,
}

/// How the thread waits when a deadline is armed.
///
/// **Exactly one implementation ships.** `SleepSlice` is the backend on
/// every platform. `Condvar` exists because [`TimerBackend::Condvar`] and
/// `ImpairmentKind::CoarseReleaseTimer` are public API and must be
/// reachable by a test: forcing it on Windows produces a genuinely
/// tick-bound wheel, which is exactly what that impairment reports, so the fault
/// injection is not a simulation. **No `#[cfg]`: both arms compile on
/// every platform**, and the choice is made once, at construction, from
/// `MOQTAP_RELEASE_TIMER`. This enum is the seam for adding a platform
/// timer without touching the wheel.
enum Parker {
    /// `std::thread::sleep(min(remaining, SLICE))`, re-reading the heap
    /// between slices.
    SleepSlice,
    /// `Condvar::wait_timeout` for the whole remainder: interruptible
    /// everywhere, high-resolution only on Unix.
    Condvar,
}

/// What the thread should do next. **Pure** — no clock read, no lock — so
/// `the_wait_plan_never_overshoots` pins the never-overshoot rule without
/// timing anything.
#[derive(Debug, PartialEq, Eq)]
enum Wait {
    /// The head is due now; fire it before waiting again.
    Fire,
    /// Nothing is armed; park on the condvar with no timeout.
    Idle,
    /// Sleep this long, then re-read the heap.
    Sleep(Duration),
}

/// Decide the next wait from a clock reading and the head of the heap.
///
/// Never returns a `Sleep` that runs past `next`: that, plus the
/// `slot.at <= now` re-check in [`run`], is the whole never-fire-early
/// guarantee.
fn plan(now: Instant, next: Option<Instant>, slice: Duration) -> Wait {
    match next {
        None => Wait::Idle,
        Some(at) => match at.checked_duration_since(now) {
            None => Wait::Fire,                   // overdue
            Some(d) if d.is_zero() => Wait::Fire, // due exactly now
            Some(d) => Wait::Sleep(d.min(slice)), // never sleeps past `at`
        },
    }
}

/// The release thread's body.
fn run(inner: Arc<Inner>) {
    let mut g = inner.state.lock().expect("release wheel state");
    loop {
        g.wakes += 1;
        if g.stopping {
            return;
        }
        let now = Instant::now();

        // 1. Collect everything due. `s.at <= now` is the whole
        //    never-fire-early guarantee; nothing else enforces it.
        let mut fired: Vec<Arc<Entry>> = Vec::new();
        while g.heap.peek().is_some_and(|Reverse(s)| s.at <= now) {
            let Reverse(slot) = g.heap.pop().expect("peeked a due slot");
            if let Some(e) = slot.entry.upgrade() {
                fired.push(e);
            }
        }

        // 2. Sweep abandoned slots. `strong_count()` does not create an
        //    `Arc`, so nothing can be dropped under the lock here.
        if g.heap.len() > SWEEP_MIN_LEN
            && now.saturating_duration_since(g.last_sweep) >= SWEEP_INTERVAL
        {
            g.heap.retain(|Reverse(s)| s.entry.strong_count() > 0);
            g.last_sweep = now;
        }

        let next = g.heap.peek().map(|Reverse(s)| s.at);

        // 3. Cancel *outside* the lock, then restart the iteration with a
        //    fresh clock. `cancel()` runs registered wakers; running them
        //    under the process-global wheel mutex would put scheduler code
        //    inside it, and a waker that re-armed would deadlock. The
        //    `Arc<Entry>`s are dropped out here too, for the same reason.
        if !fired.is_empty() {
            drop(g);
            for e in fired {
                e.token.cancel();
            }
            g = inner.state.lock().expect("release wheel state");
            continue; // `next` is now stale
        }

        // 4. Wait. The guard was NOT released above, so `next` is current
        //    and no arm can have slipped in between.
        match plan(Instant::now(), next, SLICE) {
            Wait::Fire => continue,
            Wait::Idle => {
                g = inner.cv.wait(g).expect("release wheel state");
            }
            Wait::Sleep(d) => match inner.parker {
                // A `notify_one` that lands here is lost on purpose: the
                // thread is not waiting, and it re-reads the heap within
                // `d <= SLICE` anyway.
                Parker::SleepSlice => {
                    drop(g);
                    std::thread::sleep(d);
                    g = inner.state.lock().expect("release wheel state");
                }
                // Forced-coarse: interruptible but tick-bound on Windows.
                // It waits the *whole* remainder rather than a slice —
                // slicing a `wait_timeout` would cost ~15 ms per slice.
                Parker::Condvar => {
                    let rest = next
                        .expect("Wait::Sleep implies an armed head")
                        .saturating_duration_since(Instant::now());
                    g = inner.cv.wait_timeout(g, rest).expect("release wheel state").0;
                }
            },
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::future::Future;
    // Aliased: `Ordering` in this module is `std::cmp::Ordering`, which
    // `Slot` needs.
    use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
    use std::task::{Context, Poll, Wake, Waker};

    // Every test here owns its wheel: it constructs `ReleaseTimer::new()`,
    // arms against that instance and drops it. None calls `arm_at`,
    // `shared()` or `started()` in a way that constructs the process-wide
    // wheel. `cargo test` runs these concurrently with the other modules'
    // unit tests in one process, so a shared heap would make `debug_len()`
    // a race — and an owned wheel is torn down through `Drop`, so every run
    // also exercises the instance API per-socket owners inherit, on both of
    // that impl's branches.

    /// Nearest-rank percentile of an already-sorted slice.
    fn pct(sorted: &[Duration], q: usize) -> Duration {
        assert!(!sorted.is_empty(), "no samples");
        let rank = (sorted.len() * q).div_ceil(100);
        sorted[rank.saturating_sub(1).min(sorted.len() - 1)]
    }

    /// Sort in place and return `(p50, p95, p99, max)`.
    fn quantiles(samples: &mut [Duration]) -> (Duration, Duration, Duration, Duration) {
        samples.sort_unstable();
        (pct(samples, 50), pct(samples, 95), pct(samples, 99), pct(samples, 100))
    }

    /// The invariant every other release guarantee rests on: a
    /// wheel that fires early is a wheel that reorders bytes. Structural,
    /// no timing budget, cannot flake.
    ///
    /// Arming one at a time also makes this the coverage of the
    /// empty-wheel `Condvar::wait` -> `notify_one` path, since the wheel
    /// is idle before every sample.
    #[tokio::test]
    async fn a_deadline_never_fires_early() {
        let timer = ReleaseTimer::new();
        for i in 0..300u64 {
            // Deterministic spacings over 0.2-5 ms; 1601 is coprime with
            // 4801, so 300 samples are 300 distinct durations.
            let micros = 200 + (i * 1601) % 4801;
            let at = Instant::now() + Duration::from_micros(micros);
            let d = timer.arm(at);
            d.token().cancelled().await;
            let observed = Instant::now();
            assert!(
                observed >= at,
                "sample {i} ({micros} us) fired {:?} early",
                at.saturating_duration_since(observed)
            );
        }
    }

    /// Block until the release thread has completed `n` more iterations.
    ///
    /// Bounded, and it returns rather than panicking when it gives up: a
    /// wheel that stops waking is what the callers' own assertions are
    /// about, and each of them says more about it than this could.
    fn wait_for_wakes(timer: &ReleaseTimer, n: u64) {
        let want = timer.debug_wakes() + n;
        let give_up = Instant::now() + Duration::from_secs(1);
        while timer.debug_wakes() < want && Instant::now() < give_up {
            std::thread::sleep(Duration::from_micros(200));
        }
    }

    /// The property `SLICE` exists to provide: a nearer deadline
    /// armed while the thread is already waiting on a far one is noticed
    /// within one slice.
    #[tokio::test]
    async fn an_earlier_deadline_registered_later_preempts_the_current_wait() {
        let timer = ReleaseTimer::new();
        let far = timer.arm(Instant::now() + Duration::from_secs(2));
        // The near deadline only arrives mid-wait if the thread is already
        // in its slice loop, and two completed iterations say it is: the
        // first is the wake that saw `far` armed, and a second can only
        // follow a slice that ran out.
        wait_for_wakes(&timer, 2);

        let start = Instant::now();
        let near_at = start + Duration::from_millis(5);
        let near =
            std::thread::scope(|s| s.spawn(|| timer.arm(near_at)).join().expect("arming thread"));
        near.token().cancelled().await;
        let elapsed = start.elapsed();

        assert!(!far.token().is_cancelled(), "the +2 s deadline fired before the +5 ms one");
        assert!(
            elapsed < Duration::from_millis(500),
            "the nearer deadline took {elapsed:?} to be noticed"
        );
    }

    /// An empty wheel costs nothing: the thread parks on an
    /// untimed `Condvar::wait`, not on a polling timeout.
    ///
    /// The one wake it allows is the post-fire iteration — the `continue`
    /// in step 3 of `run`, which cancels the fired tokens and then goes
    /// round once more to find the heap empty and park. So the last token
    /// can resolve one iteration before the thread is asleep, and it is
    /// exactly one iteration, because that one finds nothing armed and
    /// takes the untimed `wait`.
    #[tokio::test]
    async fn the_release_thread_parks_when_the_wheel_is_empty() {
        let timer = ReleaseTimer::new();
        let at = Instant::now() + Duration::from_millis(5);
        let armed: Vec<Deadline> = (0..100).map(|_| timer.arm(at)).collect();
        for d in &armed {
            d.token().cancelled().await;
        }

        /// The window the negative claim is made over. A window is the
        /// only evidence there is for "nothing happened", so this one is
        /// a duration and stays one.
        const IDLE: Duration = Duration::from_millis(200);

        let before = timer.debug_wakes();
        tokio::time::sleep(IDLE).await;
        let after = timer.debug_wakes();

        assert!(
            after - before <= 1,
            "the thread woke {} times over an idle {IDLE:?} window; at most one is the \
             post-fire iteration parking",
            after - before
        );
        assert_eq!(timer.debug_len(), 0, "fired slots stayed in the heap");
    }

    /// Abandoned slots are swept rather than carried to their own
    /// instants.
    ///
    /// The deadlines are far-dated on purpose: 30 s is the case the sweep
    /// exists for. It is `max_hold`'s default, and a torn-down `Hold` is
    /// exactly how a dead slot survives its own instant.
    ///
    /// This assumes the shipping [`Parker::SleepSlice`]. Under a forced
    /// `MOQTAP_RELEASE_TIMER=condvar` the thread waits out the *whole*
    /// remainder, so with only a +30 s head it does not wake to sweep at
    /// all and this test fails by design — which is one more reason no
    /// test in this binary may set that variable.
    #[tokio::test]
    async fn abandoned_far_deadlines_are_swept() {
        let timer = ReleaseTimer::new();
        let far = Instant::now() + Duration::from_secs(30);
        for _ in 0..10_000 {
            drop(timer.arm(far));
        }
        assert!(timer.debug_len() >= SWEEP_MIN_LEN, "the abandoned slots never reached the heap");

        // Give the thread a reason to wake.
        let wake = timer.arm(Instant::now() + Duration::from_millis(5));
        wake.token().cancelled().await;

        let give_up = Instant::now() + Duration::from_secs(1);
        loop {
            let len = timer.debug_len();
            if len < SWEEP_MIN_LEN {
                break;
            }
            assert!(Instant::now() < give_up, "{len} abandoned slots still in the heap after 1 s");
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    }

    /// The instance API per-socket owners inherit actually stops, rather
    /// than waiting out its deadlines.
    #[test]
    fn an_instance_timer_joins_its_thread_on_drop() {
        let timer = ReleaseTimer::new();
        let held = timer.arm(Instant::now() + Duration::from_secs(10));
        let t0 = Instant::now();
        drop(timer);
        let elapsed = t0.elapsed();
        drop(held);
        assert!(
            elapsed < Duration::from_millis(250),
            "dropping a wheel with a +10 s deadline took {elapsed:?}"
        );
    }

    /// A waker that tears the wheel down when it fires: it owns the only
    /// [`ReleaseTimer`] handle and drops it from inside `wake`.
    /// Records where it ran and whether the drop returned, so the test can tell
    /// "the drop is stuck" from *the waker never fired*.
    struct DropTheWheelOnWake {
        wheel: Mutex<Option<ReleaseTimer>>,
        fired_on: Mutex<Option<String>>,
        returned: AtomicBool,
    }

    impl Wake for DropTheWheelOnWake {
        fn wake(self: Arc<Self>) {
            self.wake_by_ref();
        }
        fn wake_by_ref(self: &Arc<Self>) {
            *self.fired_on.lock().expect("waker state") =
                Some(std::thread::current().name().unwrap_or("<unnamed>").to_owned());
            // The lock guard is a temporary and is released at the end of
            // this statement, so the wheel is dropped without it held.
            let wheel = self.wheel.lock().expect("waker state").take();
            drop(wheel); // the last handle: this runs `ReleaseTimer::drop`
            self.returned.store(true, AtomicOrdering::SeqCst);
        }
    }

    /// Tearing the wheel down from inside a deadline's own completion path
    /// returns instead of deadlocking.
    ///
    /// Step 3 of [`run`] cancels tokens on the release thread, so a waker
    /// registered on a deadline runs *there*, and a waker that drops the
    /// last handle to the wheel therefore enters `Drop` on the release
    /// thread. An unconditional `handle.join()` would then wait for the
    /// very thread executing it, forever: `stopping` is already set and
    /// `notify_all` already sent, and neither helps, because the thread is
    /// inside the cancel loop and cannot reach the top of `run` to read
    /// the flag.
    ///
    /// The waker is an ordinary [`Wake`] implementation registered by
    /// polling the same `cancelled()` future any awaiting task would, so
    /// nothing here is a shape only a test produces. Two assertions keep
    /// it honest: the first poll must return `Pending`, or no waker was
    /// registered and the drop never ran from a wake at all; and the wake
    /// must have landed on the release thread, or the self-join was not
    /// exercised even though the timing passed.
    ///
    /// The bound is a liveness limit, not an accuracy bound — a self-join
    /// never returns at all, so any window catches it and a generous one
    /// cannot redden because the box was busy. The joining branch is the
    /// one `an_instance_timer_joins_its_thread_on_drop` covers, and it
    /// still holds a 250 ms bound.
    #[test]
    fn dropping_the_wheel_from_inside_its_own_release_callback_returns() {
        /// Liveness limit. The fixed drop returns in microseconds; a
        /// self-joined one never does.
        const GIVE_UP: Duration = Duration::from_secs(10);

        let timer = ReleaseTimer::new();
        let deadline = timer.arm(Instant::now() + Duration::from_millis(50));
        let state = Arc::new(DropTheWheelOnWake {
            // The waker is now the only owner of the wheel.
            wheel: Mutex::new(Some(timer)),
            fired_on: Mutex::new(None),
            returned: AtomicBool::new(false),
        });

        let waker = Waker::from(Arc::clone(&state));
        let mut cx = Context::from_waker(&waker);
        let mut cancelled = Box::pin(deadline.token().cancelled());
        assert_eq!(
            cancelled.as_mut().poll(&mut cx),
            Poll::Pending,
            "the deadline resolved on the first poll, so no waker was registered and the \
             wheel would never be dropped from one",
        );

        let give_up = Instant::now() + GIVE_UP;
        while !state.returned.load(AtomicOrdering::SeqCst) {
            assert!(
                Instant::now() < give_up,
                "`ReleaseTimer::drop` did not return within {GIVE_UP:?}: {}",
                match state.fired_on.lock().expect("waker state").as_deref() {
                    Some(t) => format!("it is joining thread {t:?} from thread {t:?}"),
                    None => "the waker never fired at all".to_owned(),
                },
            );
            std::thread::sleep(Duration::from_millis(1));
        }

        assert_eq!(
            state.fired_on.lock().expect("waker state").as_deref(),
            Some("moqtap-release"),
            "the wake did not run on the release thread, so this test dropped the wheel from \
             somewhere the join was always safe and proves nothing",
        );
    }

    /// The test that makes this module's reason for existing falsifiable.
    /// Without it, "a dedicated wheel beats `tokio::time::sleep`" — the
    /// claim the whole file rests on — is never checked at all.
    ///
    /// The sampled quantity is `lateness = observed - deadline`: the
    /// requested 2 ms is subtracted from *both* arms before the percentile
    /// is taken.
    ///
    /// The assertion is on a *median* because the tokio arm's p50 is
    /// bounded below by the ~15.6 ms tick as a property of the timer, not
    /// of the machine, while the wheel arm's p95 rises under load.
    ///
    /// It runs by default, unlike the two calibration measurements at the
    /// bottom of this module, because it samples the wheel alone on a
    /// four-worker runtime — one arm, one token, one wake — rather than the
    /// whole read-hook-queue-release-write pipeline on a current-thread
    /// runtime that the same load starves. This test must never be
    /// `#[ignore]`d: it is the one the multi-threaded-runtime requirement
    /// rests on. If a runner ever deschedules the release thread past
    /// 2.79 ms more than half the time, weaken the factor to `x2` rather
    /// than delete or ignore the test.
    #[cfg(windows)]
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn the_wheel_beats_the_tokio_timer_on_lateness() {
        let timer = ReleaseTimer::new();
        let requested = Duration::from_millis(2);
        let mut wheel: Vec<Duration> = Vec::with_capacity(100);
        let mut tokio_timer: Vec<Duration> = Vec::with_capacity(100);

        for _ in 0..100 {
            let at = Instant::now() + requested;
            let d = timer.arm(at);
            d.token().cancelled().await;
            wheel.push(Instant::now().saturating_duration_since(at));

            let at = Instant::now() + requested;
            tokio::time::sleep(requested).await;
            tokio_timer.push(Instant::now().saturating_duration_since(at));
        }

        let (wheel_p50, ..) = quantiles(&mut wheel);
        let (tokio_p50, ..) = quantiles(&mut tokio_timer);
        assert!(
            wheel_p50 * 4 < tokio_p50,
            "wheel p50 lateness {wheel_p50:?}, tokio p50 lateness {tokio_p50:?}"
        );
    }

    /// Pure: no clock read inside the assertions, no thread.
    ///
    /// `t0 + 1 ms` is written on the *now* side rather than `t0 - 1 ms` on
    /// the deadline side because `Instant - Duration` panics near the
    /// clock's origin.
    #[test]
    fn the_wait_plan_never_overshoots() {
        let t0 = Instant::now();
        assert_eq!(plan(t0, None, SLICE), Wait::Idle);
        assert_eq!(plan(t0, Some(t0), SLICE), Wait::Fire);
        assert_eq!(plan(t0 + Duration::from_millis(1), Some(t0), SLICE), Wait::Fire,);
        // The *nearest deadline is already inside one slice* case, served by
        // the `min` and not by a branch.
        assert_eq!(
            plan(t0, Some(t0 + Duration::from_micros(100)), SLICE),
            Wait::Sleep(Duration::from_micros(100)),
        );
        assert_eq!(plan(t0, Some(t0 + Duration::from_millis(5)), SLICE), Wait::Sleep(SLICE),);
    }

    /// `Slot`'s `Ord` is hand-written (`Weak` is neither `Ord` nor
    /// `Eq`) and wrapped in `Reverse`, so one inverted comparison turns
    /// the wheel into a LIFO that serves the farthest deadline first — a
    /// defect that would otherwise show up only as a confusing timing
    /// failure.
    #[test]
    fn the_heap_is_a_min_heap_and_ties_break_by_registration() {
        let t0 = Instant::now();
        let ms = |n| t0 + Duration::from_millis(n);

        let mut heap: BinaryHeap<Reverse<Slot>> = BinaryHeap::new();
        for (at, seq) in [(ms(3), 1u64), (ms(1), 2), (ms(2), 3), (ms(1), 4), (ms(1), 5)] {
            heap.push(Reverse(Slot { at, seq, entry: Weak::new() }));
        }

        let mut popped: Vec<(Instant, u64)> = Vec::new();
        while let Some(Reverse(s)) = heap.pop() {
            popped.push((s.at, s.seq));
        }

        assert!(
            popped.windows(2).all(|w| w[0].0 <= w[1].0),
            "instants did not come out ascending: {popped:?}"
        );
        assert_eq!(
            popped,
            vec![(ms(1), 2), (ms(1), 4), (ms(1), 5), (ms(2), 3), (ms(3), 1)],
            "ties did not break by registration order"
        );
    }

    /// **Nothing armed is lost.** Every deadline the wheel
    /// accepts is eventually released, none of them early, and the sweep
    /// does not eat a live one on its way past.
    /// This is the completeness half of what the wheel guarantees, and it is
    /// stated the way it is *because* it has to survive a saturated box —
    /// unlike the calibration measurements at the bottom of this module. A
    /// descheduled release thread makes every sample **later**; it cannot make
    /// one vanish, and it cannot make one early. So the two assertions here are
    /// "it arrived at all" (with a give-up window 28x the deadline, which is a
    /// liveness limit and not an accuracy bound) and *it did not arrive before
    /// its instant*. Neither has a scheduler-dependent number in it.
    ///
    /// The shape is chosen so the sweep really runs **while live slots are
    /// in the heap**, which is the case every other test here misses:
    /// `a_deadline_never_fires_early` arms one at a time, so the heap never
    /// reaches [`SWEEP_MIN_LEN`];
    /// `the_release_thread_parks_when_the_wheel_is_empty`'s hundred slots
    /// are due 5 ms after construction, long before [`SWEEP_INTERVAL`]
    /// elapses; and `abandoned_far_deadlines_are_swept`'s ten thousand slots
    /// are all abandoned, so a sweep that dropped live ones too would look
    /// identical. Here 128 live and 128 abandoned slots share a heap, the
    /// deadlines are ~700 ms out so `SWEEP_INTERVAL` passes twice first,
    /// and the [`Parker::SleepSlice`] thread is awake every [`SLICE`]
    /// throughout, so the rebuild is guaranteed to execute over a heap
    /// that is half live.
    ///
    /// Arming runs farthest-first, so every later arm is *nearer* than the
    /// current head and the awaits then run in deadline order.
    #[tokio::test]
    async fn every_armed_deadline_is_eventually_released() {
        const LIVE: usize = 128;
        /// Interleaved with the live ones so a sweep cannot pass over a
        /// contiguous block of dead slots and miss the live ones.
        const ABANDONED: usize = 128;
        /// Far enough out that `SWEEP_INTERVAL` elapses twice before the
        /// first slot is due.
        const BASE: Duration = Duration::from_millis(700);
        /// Spread, so the batch is many wakes rather than one.
        const STEP: Duration = Duration::from_micros(200);
        /// A liveness limit, not an accuracy bound: a lost slot never
        /// arrives at all, so any window catches it and a generous one
        /// cannot go red because the box was busy.
        const GIVE_UP: Duration = Duration::from_secs(20);

        let timer = ReleaseTimer::new();
        let base = Instant::now() + BASE;

        // Farthest first: every subsequent arm is nearer than the head.
        let mut armed: Vec<(Instant, Deadline)> = Vec::with_capacity(LIVE);
        for i in 0..LIVE {
            let at = base + STEP * u32::try_from(LIVE - 1 - i).expect("LIVE fits in u32");
            armed.push((at, timer.arm(at)));
            if i < ABANDONED {
                // Dropped immediately: a slot the wheel must reap, sitting
                // between two it must not.
                drop(timer.arm(at + STEP / 2));
            }
        }
        assert!(
            timer.debug_len() > SWEEP_MIN_LEN,
            "the heap never reached the length that lets the sweep run, so this test would \
             pass with or without one: {} slots",
            timer.debug_len(),
        );

        // In deadline order, so each await really waits for its own slot.
        for (i, (at, d)) in armed.iter().rev().enumerate() {
            tokio::time::timeout(GIVE_UP, d.token().cancelled()).await.unwrap_or_else(|_| {
                panic!(
                    "deadline {i} of {LIVE} was armed and never released: {GIVE_UP:?} after a \
                     deadline {BASE:?} out it is still pending, with {} slot(s) in the heap",
                    timer.debug_len(),
                )
            });
            let observed = Instant::now();
            assert!(
                observed >= *at,
                "deadline {i} of {LIVE} fired {:?} early",
                at.saturating_duration_since(observed),
            );
        }

        assert_eq!(
            timer.debug_len(),
            0,
            "every slot was popped or swept, so the heap must be empty",
        );
    }

    /// Keeps the two process-wide read-backs compiled and exercised.
    ///
    /// Not a property test like the others here: `started()` and
    /// `backend()` are thin readers
    /// over `instrument`'s single monotonic atomic, and until `egress.rs`
    /// and `session.rs` call them they would otherwise be `dead_code`
    /// under `-D warnings`. Asserted in the monotone-safe direction only
    /// (`Some` backend implies started), because nothing in this binary
    /// constructs the process-wide wheel and a future test that does must
    /// not turn this into a race.
    #[test]
    fn the_process_wide_read_backs_agree() {
        if backend().is_some() {
            assert!(started(), "a backend was published without `started`");
        }
    }

    // ── calibration: opt-in ────────────────────────────────────────────
    //
    // The tests above are correctness properties — never early, never
    // lost, never reordered, never left parked — and hold on a loaded box,
    // because a starved release thread can only make a release later. What
    // follows measures how late, and under CPU saturation that is a property
    // of the machine as much as of the wheel: the wheel's lateness rises to
    // the tens of milliseconds `tokio::time::sleep` costs. So these are
    // `#[ignore]`d; run them on an otherwise idle machine:
    //
    // ```text
    // cargo test -p moqtap-proxy --lib -- --ignored --nocapture calibration
    // ```
    //
    // The default run still rejects the `tokio::time::sleep` implementation
    // this module replaces, through
    // `an_earlier_deadline_registered_later_preempts_the_current_wait` and
    // `a_deadline_never_fires_early`. The session-level companion is
    // `actions_timing::calibration_a_five_millisecond_delay_is_measured_as_five_milliseconds`.

    /// **Calibration measurement, not a gate. Requires a quiet machine.**
    /// Ignored by default; run with `-- --ignored --nocapture`.
    ///
    /// The standing proxy for a full calibration run, and the first
    /// measurement of the Unix backends. 200 deadlines at 3 ms on one
    /// instance; `p50 <= 5 ms` when
    /// the backend is high-resolution, `<= 60 ms` otherwise.
    #[tokio::test]
    #[ignore = "calibration measurement: asserts timing accuracy, which is not \
                measurable on a loaded box — run with --ignored on a quiet machine"]
    async fn calibration_release_error_p50_is_within_the_platform_budget() {
        let timer = ReleaseTimer::new();
        let mut lateness: Vec<Duration> = Vec::with_capacity(200);
        for _ in 0..200 {
            let at = Instant::now() + Duration::from_millis(3);
            let d = timer.arm(at);
            d.token().cancelled().await;
            lateness.push(Instant::now().saturating_duration_since(at));
        }
        assert_eq!(lateness.len(), 200, "lost samples");

        let bound = if timer.backend().is_high_resolution() {
            Duration::from_millis(5)
        } else {
            Duration::from_millis(60)
        };
        let (p50, p95, p99, max) = quantiles(&mut lateness);
        // Printed as well as asserted: a calibration run is worth reading
        // even when it is green, and `--nocapture` is how it is invoked.
        println!(
            "release lateness, backend {}, 200 deadlines at 3 ms: p50 {p50:?}, p95 {p95:?}, \
             p99 {p99:?}, max {max:?} (bound {bound:?})",
            timer.backend().as_str(),
        );
        assert!(
            p50 <= bound,
            "backend {} p50 {p50:?} exceeds {bound:?} (p95 {p95:?}, p99 {p99:?}, max {max:?})",
            timer.backend().as_str()
        );
    }
}