darkbio_clock/

paused.rs

// clock-rs: virtual clock for testing blocking code
// Copyright 2026 Dark Bio AG. All rights reserved.
//
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

//! Test clocks, which move only when their owner advances them.

use crate::primitives::{Condvar, Mutex, MutexGuard};
use crate::{Clock, Signal};
#[cfg(feature = "crossbeam")]
use crossbeam_channel::{Receiver, RecvTimeoutError, Sender, TryRecvError};
use std::collections::{BTreeMap, HashSet};
use std::fmt;
use std::sync::{Arc, PoisonError};
use std::time::{Duration, Instant, SystemTime};

/// Owns a clock that moves only when advanced, for tests.
///
/// [`Self::clock`] hands out handles that read and sleep on its time. Only the
/// owner moves it, through methods that take `&mut self`, so each test clock
/// has one driver.
#[cfg_attr(docsrs, doc(cfg(feature = "test-clock")))]
pub struct TestClock {
    /// State shared with every clock handle.
    paused: Arc<Paused>,
}

impl TestClock {
    /// Creates a stopped clock at the current real monotonic and wall times.
    pub fn new() -> Self {
        let now = Instant::now();
        Self {
            paused: Arc::new(Paused {
                start: now,
                state: Mutex::new(PausedState {
                    now,
                    wall: WallAnchor {
                        time: SystemTime::now(),
                        instant: now,
                    },
                    sequence: 0,
                    registered: 0,
                    notified: 0,
                    deadlines: BTreeMap::new(),
                    #[cfg(feature = "crossbeam")]
                    timers: BTreeMap::new(),
                    #[cfg(feature = "crossbeam")]
                    fired: Vec::new(),
                }),
                changed: Condvar::new(),
                #[cfg(test)]
                before_wait: std::sync::Mutex::new(None),
                #[cfg(test)]
                before_rewait: std::sync::Mutex::new(None),
                #[cfg(test)]
                before_parked_wait: std::sync::Mutex::new(None),
                #[cfg(all(test, feature = "crossbeam"))]
                after_timer_receive: std::sync::Mutex::new(None),
            }),
        }
    }

    /// Returns a handle that reads and sleeps on this clock.
    pub fn clock(&self) -> Clock {
        Clock {
            paused: Some(self.paused.clone()),
        }
    }

    /// Moves both times forward by `by`, waking the sleeps and deadline waits
    /// it reaches and firing its due timers.
    ///
    /// Returns once the reached waits are woken and the due timers hold their
    /// messages, without waiting for any thread to act. Other waits keep
    /// waiting, since an advance never counts as a notification. A zero
    /// advance does nothing.
    ///
    /// # Panics
    ///
    /// Panics if either time would overflow, before changing either one.
    pub fn advance(&mut self, by: Duration) {
        self.advance_with(|now| {
            now.checked_add(by)
                .expect("clock advance overflows Instant")
        });
    }

    /// Moves monotonic time to `target` and wall time by the same amount,
    /// waking the sleeps and deadline waits it reaches and firing its due timers.
    ///
    /// Returns once the reached waits are woken and the due timers hold their
    /// messages, without waiting for any thread to act. Other waits keep
    /// waiting, since an advance never counts as a notification. Advancing to
    /// the current time does nothing.
    ///
    /// # Panics
    ///
    /// Panics if `target` is before now or wall time would overflow.
    /// Neither time changes after a panic.
    pub fn advance_to(&mut self, target: Instant) {
        self.advance_with(|now| {
            assert!(target >= now, "clock cannot go backwards");
            target
        });
    }

    /// Sets wall time forwards or backwards without moving monotonic time.
    ///
    /// This wakes no waits and fires no timers, since deadlines use monotonic time.
    pub fn set_system_time(&mut self, time: SystemTime) {
        let mut state = self.paused.lock();
        state.wall = WallAnchor {
            time,
            instant: state.now,
        };
    }

    /// Blocks until at least `count` clock sleeps or condvar waits are
    /// registered, timed or not.
    ///
    /// A wait stays registered from the moment it starts waiting until it
    /// returns. So the count includes a thread that a notification or an
    /// advance has woken but that has not run yet. Threads blocked in crossbeam
    /// receives, selects or mutexes, or busy outside clock waits, do not count.
    /// An earlier registration counts too, so the count proves no progress on
    /// its own. Nothing bounds the wait, so run tests under a runner with a
    /// per-test timeout, since `cargo test` alone never stops a hung test.
    pub fn wait_registered(&self, count: usize) {
        let mut state = self.paused.lock();
        while state.registered < count {
            state = self
                .paused
                .changed
                .wait(state)
                .unwrap_or_else(PoisonError::into_inner);
        }
    }

    /// Blocks until at least `count` registered waits are parked, with their
    /// deadlines unreached or absent, and no notification on this clock is
    /// outstanding.
    ///
    /// A notification stays outstanding until every wait it ended has
    /// returned. So a woken thread that has not run yet holds the barrier, on
    /// whichever condvar of this clock it waits.
    ///
    /// The count must cover every worker the test expects to park, and the
    /// test must control the work they receive. Threads blocked in crossbeam
    /// receives, selects or mutexes, or busy outside clock waits, are invisible
    /// to it, and a worker that exits never counts again. Nothing bounds the
    /// wait, so run tests under a runner with a per-test timeout.
    pub fn wait_parked(&self, count: usize) {
        let mut state = self.paused.lock();
        while !state.parked(count) {
            // Let a test release a held worker once this barrier must wait
            #[cfg(test)]
            {
                let hook = self.paused.before_parked_wait.lock().unwrap().take();
                if let Some(hook) = hook {
                    drop(state);
                    hook();
                    state = self.paused.lock();
                    continue;
                }
            }
            state = self
                .paused
                .changed
                .wait(state)
                .unwrap_or_else(PoisonError::into_inner);
        }
    }

    /// Blocks until at least `count` timers are armed on this clock and unfired.
    ///
    /// A waiting receive's timer counts, and so does a timer whose receiver was
    /// dropped. A timer armed earlier counts too, so the count proves no
    /// progress on its own. Nothing bounds the wait, so run tests under a runner
    /// with a per-test timeout, since `cargo test` alone never stops a hung test.
    #[cfg(feature = "crossbeam")]
    #[cfg_attr(docsrs, doc(cfg(all(feature = "test-clock", feature = "crossbeam"))))]
    pub fn wait_timers(&self, count: usize) {
        let mut state = self.paused.lock();
        while state.timers.len() < count {
            state = self
                .paused
                .changed
                .wait(state)
                .unwrap_or_else(PoisonError::into_inner);
        }
    }

    /// Returns the earliest deadline among this clock's registered sleeps,
    /// deadline waits and unfired timers, or `None` when there is none.
    ///
    /// A timed wait stays listed until it stops waiting, so one an advance
    /// reaches stays listed until its thread runs, and the result is then the
    /// current time. Await an advance's effect before reading the next
    /// deadline. A timer leaves the list when it fires.
    pub fn next_deadline(&self) -> Option<Instant> {
        // Compare the earliest registered wait with the earliest unfired timer, and report
        // one already reached at the current time, so that advancing to it stays valid
        let state = self.paused.lock();
        let deadline = state.deadlines.keys().next().map(|&(deadline, _)| deadline);
        #[cfg(feature = "crossbeam")]
        let deadline = deadline
            .into_iter()
            .chain(state.timers.keys().next().map(|&(deadline, _)| deadline))
            .min();
        deadline.map(|deadline| deadline.max(state.now))
    }

    /// Validates both new times, publishes them together with the due timers'
    /// messages, then wakes the reached waits.
    fn advance_with(&mut self, next: impl FnOnce(Instant) -> Instant) {
        // Check the new times before taking the lock, so a failed check panics with
        // no lock held. Only the owner advances, so nothing changes them in between.
        let (now, wall) = {
            let state = self.paused.lock();
            (state.now, state.wall)
        };
        let next = next(now);
        if next == now {
            return;
        }
        wall.at(next).expect("clock advance overflows SystemTime");

        // Publish the time and collect each reached wait's signal once, in the same
        // lock hold that registrations take. The wakes follow deadline and
        // registration order, since the address set only filters out repeats.
        let mut state = self.paused.lock();
        state.now = next;
        let mut seen = HashSet::new();
        let signals: Vec<_> = state
            .deadlines
            .range(..=(next, u64::MAX))
            .filter(|(_, signal)| seen.insert(Arc::as_ptr(signal)))
            .map(|(_, signal)| signal.clone())
            .collect();

        // Deliver every due timer before another thread can read the new time
        #[cfg(feature = "crossbeam")]
        {
            while let Some(entry) = state
                .timers
                .first_entry()
                .filter(|entry| entry.key().0 <= next)
            {
                let timer = entry.remove();
                state.fire(timer);
            }
        }
        drop(state);

        // Wake outside the clock lock, since registration takes the signal lock first
        for signal in signals {
            signal.wake();
        }
    }
}

impl Default for TestClock {
    /// Creates a stopped clock at the current real monotonic and wall times.
    fn default() -> Self {
        Self::new()
    }
}

impl fmt::Debug for TestClock {
    /// Shows the advance, wall time, registered waits and armed timers.
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        // Snapshot the counts before calling the formatter's writer
        let state = self.paused.lock();
        let advanced = state.now - self.paused.start;
        let system_time = state.system_time();
        let registered = state.registered;
        #[cfg(feature = "crossbeam")]
        let timers = state.timers.len();
        drop(state);

        // Format without the clock lock, since the writer may read this clock
        let mut debug = f.debug_struct("TestClock");
        debug
            .field("advanced", &advanced)
            .field("system_time", &system_time)
            .field("registered", &registered);
        #[cfg(feature = "crossbeam")]
        debug.field("timers", &timers);
        debug.finish()
    }
}

/// A test clock's state, shared by its owner, handles and waiters.
pub(crate) struct Paused {
    /// Time the clock started at, to show how far it has advanced.
    start: Instant,
    /// Current times, registered wait count, deadlines and timers.
    pub(crate) state: Mutex<PausedState>,
    /// Wakes drivers when registrations, notifications or armed timers change.
    pub(crate) changed: Condvar,
    /// One-shot test hook, run before a wait's registration with no clock or
    /// signal lock held.
    #[cfg(test)]
    pub(crate) before_wait: std::sync::Mutex<Option<BeforeWait>>,
    /// One-shot test hook, run like `before_wait` before a registered wait waits again
    /// after a spurious wakeup.
    #[cfg(test)]
    pub(crate) before_rewait: std::sync::Mutex<Option<BeforeWait>>,
    /// Test hook before `wait_parked` blocks, run once with no clock lock held.
    #[cfg(test)]
    pub(crate) before_parked_wait: std::sync::Mutex<Option<BeforeParkedWait>>,
    /// One-shot test hook, run when a receive's timer wins, before the receiver
    /// is checked again, with no crate lock held.
    #[cfg(all(test, feature = "crossbeam"))]
    after_timer_receive: std::sync::Mutex<Option<TimerHook>>,
}

/// Mutable part of a test clock.
pub(crate) struct PausedState {
    /// Current monotonic time.
    now: Instant,
    /// Wall time as last set, with the monotonic instant it was set at.
    wall: WallAnchor,
    /// Last sequence number given to a timed wait or timer, which a `u64`
    /// cannot outgrow within any run.
    sequence: u64,
    /// Clock waits registered until their calls return, including woken waits.
    pub(crate) registered: usize,
    /// Notifications whose affected waits have not finished retiring.
    notified: usize,
    /// Registered timed waits by deadline and registration order, each with
    /// its signal and unlisted when its wait returns.
    deadlines: BTreeMap<(Instant, u64), Arc<Signal>>,
    /// Unfired timers by deadline and arming order.
    #[cfg(feature = "crossbeam")]
    timers: BTreeMap<(Instant, u64), Timer>,
    /// Senders of delivered public timers, kept so their channels stay connected
    /// while the clock lives.
    #[cfg(feature = "crossbeam")]
    fired: Vec<Sender<Instant>>,
}

impl PausedState {
    /// Reports whether at least `count` waits are parked with no outstanding notification.
    pub(crate) fn parked(&self, count: usize) -> bool {
        self.notified == 0 && self.unreached() >= count
    }

    /// Counts registered waits whose deadlines have not been reached.
    fn unreached(&self) -> usize {
        self.registered - self.deadlines.range(..=(self.now, u64::MAX)).count()
    }

    /// Keys a timed wait or timer by its deadline, ordering equal deadlines by
    /// when they were listed.
    fn next_key(&mut self, deadline: Instant) -> (Instant, u64) {
        self.sequence += 1;
        (deadline, self.sequence)
    }

    /// Returns the wall time at the current monotonic time.
    fn system_time(&self) -> SystemTime {
        self.wall
            .at(self.now)
            .expect("wall time fits, since every change checks it first")
    }

    /// Lists a timer until an advance reaches it, or delivers it at once if it
    /// is already due.
    #[cfg(feature = "crossbeam")]
    fn arm_timer(
        &mut self,
        deadline: Instant,
        retain: bool,
    ) -> ((Instant, u64), Receiver<Instant>) {
        // Give the timer a sequence number and room for its only message
        let key = self.next_key(deadline);
        let (sender, receiver) = crossbeam_channel::bounded(1);
        let timer = Timer {
            deadline,
            sender,
            retain,
        };

        // List or deliver it in the lock hold that read the time, so no advance slips between
        if deadline > self.now {
            self.timers.insert(key, timer);
        } else {
            self.fire(timer);
        }
        (key, receiver)
    }

    /// Sends a timer's deadline, and keeps the sender of a delivered public timer.
    #[cfg(feature = "crossbeam")]
    fn fire(&mut self, timer: Timer) {
        // The capacity-1 channel has never been sent to, so this cannot block
        let delivered = timer.sender.send(timer.deadline).is_ok();

        // Keep a delivered public timer's sender, so its channel stays connected like
        // crossbeam's. Nothing reports a dropped receiver without sending, and a second
        // send would refill a consumed timer.
        if delivered && timer.retain {
            self.fired.push(timer.sender);
        }
    }
}

/// A wall time and the monotonic instant it was set at.
#[derive(Clone, Copy)]
struct WallAnchor {
    /// Wall time when the clock was created or last set.
    time: SystemTime,
    /// Monotonic time at that moment.
    instant: Instant,
}

impl WallAnchor {
    /// Returns the wall time at `instant`, or `None` if it does not fit.
    ///
    /// The whole span since the anchor is added at once, so a platform that
    /// rounds wall time, like Windows to 100 ns, rounds once and never per advance.
    fn at(&self, instant: Instant) -> Option<SystemTime> {
        self.time.checked_add(instant - self.instant)
    }
}

impl Paused {
    /// Replaces one signal's contribution to the outstanding notifications.
    pub(crate) fn notifications(&self, previous: usize, current: usize) {
        let mut state = self.lock();
        state.notified = state.notified - previous + current;
        self.changed.notify_all();
    }

    /// Returns armed timers and retained senders for the bookkeeping model.
    #[cfg(all(test, feature = "crossbeam", not(loom)))]
    pub(crate) fn timer_counts(&self) -> (usize, usize) {
        let state = self.lock();
        (state.timers.len(), state.fired.len())
    }

    /// Arms a public timer, whose channel stays connected after delivery.
    #[cfg(feature = "crossbeam")]
    pub(crate) fn at(&self, deadline: Instant) -> Receiver<Instant> {
        let mut state = self.lock();
        let (_, receiver) = state.arm_timer(deadline, true);
        self.changed.notify_all();
        receiver
    }

    /// Receives until the clock reaches `deadline`, where a message or a
    /// disconnection wins over expiry, as in crossbeam.
    ///
    /// The receiver is never checked under the clock lock, since a rendezvous
    /// receive can wait on a sender that reads the clock. Expiry is decided only
    /// at a time no advance changed since the check, and an advance delivers its
    /// due timers before anyone reads its time, so a timer due by the deadline
    /// holds its message by then.
    #[cfg(feature = "crossbeam")]
    pub(crate) fn recv_deadline<T>(
        &self,
        receiver: &Receiver<T>,
        deadline: Instant,
    ) -> Result<T, RecvTimeoutError> {
        // Check the receiver at a known time, and look again if an advance ran meanwhile
        let (key, timeout) = loop {
            let seen = self.now();
            match receiver.try_recv() {
                Ok(value) => return Ok(value),
                Err(TryRecvError::Disconnected) => return Err(RecvTimeoutError::Disconnected),
                Err(TryRecvError::Empty) => {}
            }
            let mut state = self.lock();
            if state.now != seen {
                continue;
            }

            // Decide expiry at the checked time, or list the timeout before unlocking so
            // that no advance slips between and wait_timers counts it
            if seen >= deadline {
                return Err(RecvTimeoutError::Timeout);
            }
            let timer = state.arm_timer(deadline, false);
            self.changed.notify_all();
            break timer;
        };

        // Unlist the timeout on every return from the adapter
        let _registration = ReceiveTimer { paused: self, key };

        // Wait on the receiver and the clock's timer, with no real timeout
        crossbeam_channel::select! {
            recv(receiver) -> result => result.map_err(RecvTimeoutError::from),
            recv(timeout) -> _ => {
                // Let tests make the receiver ready after the timer has won
                #[cfg(test)]
                {
                    let hook = self.after_timer_receive.lock().unwrap().take();
                    if let Some(hook) = hook {
                        hook();
                    }
                }

                // Wait out an advance still delivering, then recheck without the lock, since
                // select may pick the timeout before a message due at the same time
                drop(self.lock());
                receiver.try_recv().map_err(|err| match err {
                    TryRecvError::Empty => RecvTimeoutError::Timeout,
                    TryRecvError::Disconnected => RecvTimeoutError::Disconnected,
                })
            }
        }
    }

    /// Returns the clock's current monotonic time.
    pub(crate) fn now(&self) -> Instant {
        self.lock().now
    }

    /// Returns the clock's current wall time.
    pub(crate) fn system_time(&self) -> SystemTime {
        self.lock().system_time()
    }

    /// Reads the advance and wall time together for formatting.
    pub(crate) fn snapshot(&self) -> (Duration, SystemTime) {
        let state = self.lock();
        (state.now - self.start, state.system_time())
    }

    /// Counts a wait and lists its deadline until the guard drops, or returns
    /// `None` if the deadline has already been reached.
    ///
    /// Advances collect the waits to wake under the same lock, so a wait either
    /// registers in time to be woken or sees the new time. The caller holds its
    /// signal lock until it waits, so the wake cannot arrive before the wait.
    pub(crate) fn register(
        &self,
        deadline: Option<Instant>,
        signal: &Arc<Signal>,
    ) -> Option<Registration<'_>> {
        // Refuse a deadline that an advance has already reached
        let mut state = self.lock();
        if deadline.is_some_and(|deadline| deadline <= state.now) {
            return None;
        }

        // Count the wait, and list a timed one under its own key
        let key = deadline.map(|deadline| state.next_key(deadline));
        state.registered += 1;
        if let Some(key) = key {
            state.deadlines.insert(key, signal.clone());
        }

        // Wake drivers waiting for the count to grow
        self.changed.notify_all();
        Some(Registration { paused: self, key })
    }

    /// Takes the one-shot test hook before registration, or for a wait
    /// `again` after a spurious wakeup, for the caller to run with no lock held.
    #[cfg(test)]
    pub(crate) fn take_hook(&self, again: bool) -> Option<BeforeWait> {
        let hook = if again {
            &self.before_rewait
        } else {
            &self.before_wait
        };
        hook.lock().unwrap().take()
    }

    /// Locks the state, recovering it from poisoning, since no update under the
    /// lock can stop halfway.
    fn lock(&self) -> MutexGuard<'_, PausedState> {
        self.state.lock().unwrap_or_else(PoisonError::into_inner)
    }
}

/// A timer's one message, and whether its channel outlives delivery.
#[cfg(feature = "crossbeam")]
struct Timer {
    /// Deadline sent as the message, even when an advance overshoots it.
    deadline: Instant,
    /// Sender of the capacity-1 channel, sent to only once under the clock lock.
    sender: Sender<Instant>,
    /// Whether the clock keeps the sender after delivery, as for public timers.
    retain: bool,
}

/// Removes a receive adapter's timer when it returns, even before expiry.
#[cfg(feature = "crossbeam")]
struct ReceiveTimer<'a> {
    /// Clock holding the armed timer, if it has not fired yet.
    paused: &'a Paused,
    /// Key of this receive's timer in the clock's list.
    key: (Instant, u64),
}

#[cfg(feature = "crossbeam")]
impl Drop for ReceiveTimer<'_> {
    /// Unlists an unfired timer without retaining its sender.
    fn drop(&mut self) {
        self.paused.lock().timers.remove(&self.key);
    }
}

/// Counts one registered wait and lists its deadline until the wait returns.
pub(crate) struct Registration<'a> {
    /// Clock whose registration count includes this thread.
    paused: &'a Paused,
    /// Key of this wait's listed deadline, if it is timed.
    key: Option<(Instant, u64)>,
}

impl Drop for Registration<'_> {
    /// Removes this wait and its deadline from the clock's registrations.
    fn drop(&mut self) {
        // Uncount the wait and unlist its deadline
        let mut state = self.paused.lock();
        state.registered -= 1;
        if let Some(key) = self.key {
            state.deadlines.remove(&key);
        }

        // The test helper wait_unregistered waits for this count to drop
        #[cfg(test)]
        self.paused.changed.notify_all();
    }
}

/// Pauses a test wait before registration and exposes its real timer, if any.
#[cfg(test)]
pub(crate) type BeforeWait = Box<dyn FnOnce(Option<Instant>) + Send>;

/// Observes a parked wait barrier that cannot yet return.
#[cfg(test)]
pub(crate) type BeforeParkedWait = Box<dyn FnOnce() + Send>;

/// Observes a timer race with no crate lock held.
#[cfg(all(test, feature = "crossbeam"))]
type TimerHook = Box<dyn FnOnce() + Send>;

// The timer tests live in src/tests, loaded from here so that they keep this
// module's private items in reach
#[cfg(all(test, feature = "crossbeam", not(loom)))]
#[cfg_attr(coverage_nightly, coverage(off))]
#[path = "tests/paused.rs"]
mod tests;