windows-threadpool-sys 0.2.0

// Copyright (c) 2026 Mike Grier
//! Unit tests for periodic thread-pool timers.
//!
//! The important assertions here are about repetition and teardown ordering.
//! Cadence itself is only loosely checked, because the pool coalesces timers and
//! a loaded machine can delay any tick.

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::{Duration, Instant, SystemTime};

use crate::callback_env::CallbackEnviron;
use crate::pool::ThreadpoolPool;
use crate::timer::ThreadpoolPeriodicTimer;
use crate::timer::tests::Fires;

fn counting_timer(period: Duration) -> (ThreadpoolPeriodicTimer, Arc<Fires>) {
    let fires = Fires::new();
    let recorder = Arc::clone(&fires);
    let timer = ThreadpoolPeriodicTimer::new(period, move |_tick| recorder.record(), None)
        .expect("create timer");
    (timer, fires)
}

// --- creation ---

#[test]
fn new_periodic_timer_succeeds() {
    assert!(ThreadpoolPeriodicTimer::new(Duration::from_millis(1), |_| {}, None).is_ok());
}

#[test]
fn new_with_env_succeeds() {
    let mut env = CallbackEnviron::new();
    assert!(ThreadpoolPeriodicTimer::new(Duration::from_millis(1), |_| {}, Some(&mut env)).is_ok());
}

/// A zero period would describe a timer that never repeats, which is what
/// `ThreadpoolTimer` is for. Accepting it would recreate exactly the ambiguity that
/// splitting the types removed.
#[test]
fn a_zero_period_is_rejected() {
    let error = ThreadpoolPeriodicTimer::new(Duration::ZERO, |_| {}, None)
        .expect_err("a zero period must be rejected");
    assert_eq!(error.kind(), std::io::ErrorKind::InvalidInput);
}

/// The boundary is at one millisecond, because the pool takes the period as
/// whole milliseconds and anything shorter rounds down to zero -- which the pool
/// reads as "do not repeat".
#[test]
fn a_sub_millisecond_period_is_rejected() {
    for micros in [1u64, 100, 500, 999] {
        let error = ThreadpoolPeriodicTimer::new(Duration::from_micros(micros), |_| {}, None)
            .expect_err("a sub-millisecond period must be rejected");
        assert_eq!(
            error.kind(),
            std::io::ErrorKind::InvalidInput,
            "period of {micros}us"
        );
    }
}

/// A fractional period would be truncated by the pool, so the timer would tick
/// sooner than `period()` reports. Rejecting keeps the two in agreement.
#[test]
fn a_fractional_millisecond_period_is_rejected() {
    for micros in [1_500u64, 2_001, 10_500, 999_999] {
        let error = ThreadpoolPeriodicTimer::new(Duration::from_micros(micros), |_| {}, None)
            .expect_err("a fractional-millisecond period must be rejected");
        assert_eq!(
            error.kind(),
            std::io::ErrorKind::InvalidInput,
            "period of {micros}us"
        );
    }
    // A sub-millisecond remainder in nanoseconds is the same defect.
    let error = ThreadpoolPeriodicTimer::new(
        Duration::from_millis(5) + Duration::from_nanos(1),
        |_| {},
        None,
    )
    .expect_err("a period with a nanosecond remainder must be rejected");
    assert_eq!(error.kind(), std::io::ErrorKind::InvalidInput);
}

/// Beyond `u32::MAX` milliseconds the period would be capped, making the timer
/// tick far more often than asked.
#[test]
fn a_period_beyond_the_maximum_is_rejected() {
    let error = ThreadpoolPeriodicTimer::new(
        ThreadpoolPeriodicTimer::MAX_PERIOD + Duration::from_millis(1),
        |_| {},
        None,
    )
    .expect_err("a period beyond the maximum must be rejected");
    assert_eq!(error.kind(), std::io::ErrorKind::InvalidInput);

    assert!(
        ThreadpoolPeriodicTimer::new(ThreadpoolPeriodicTimer::MAX_PERIOD, |_| {}, None).is_ok(),
        "the advertised maximum period must be accepted"
    );
}

/// Whole-millisecond periods across the accepted range are reported exactly as
/// given, which is the property the rejections above exist to preserve.
#[test]
fn an_accepted_period_is_reported_exactly() {
    for millis in [1u64, 2, 15, 1_000, 60_000, u64::from(u32::MAX)] {
        let period = Duration::from_millis(millis);
        let timer =
            ThreadpoolPeriodicTimer::new(period, |_| {}, None).expect("a whole-millisecond period");
        assert_eq!(timer.period(), period, "period of {millis}ms");
    }
}

#[test]
fn the_shortest_accepted_period_is_one_millisecond() {
    assert_eq!(
        ThreadpoolPeriodicTimer::MIN_PERIOD,
        Duration::from_millis(1)
    );
    assert!(
        ThreadpoolPeriodicTimer::new(ThreadpoolPeriodicTimer::MIN_PERIOD, |_| {}, None).is_ok(),
        "the advertised minimum period must be accepted"
    );
}

/// The guard exists to keep a periodic timer from silently being a one-shot, so
/// this checks the behaviour rather than the guard: the shortest accepted period
/// must actually repeat.
#[test]
fn the_shortest_accepted_period_actually_repeats() {
    let fires = Fires::new();
    let counter = Arc::clone(&fires);
    let timer = ThreadpoolPeriodicTimer::new(
        ThreadpoolPeriodicTimer::MIN_PERIOD,
        move |_| {
            counter.record();
        },
        None,
    )
    .expect("create timer");

    timer.start_after(Duration::ZERO);
    // Fails rather than hangs if the timer only ever fires once, which is what
    // a period that rounded down to zero would do.
    fires.wait_for(3);
    timer.stop_and_drain();

    assert!(
        fires.count() >= 3,
        "a timer at the minimum period is not repeating"
    );
}

#[test]
fn new_timer_is_stopped() {
    let timer =
        ThreadpoolPeriodicTimer::new(Duration::from_millis(1), |_| {}, None).expect("create timer");
    assert!(!timer.is_running(), "a fresh timer must not be running");
}

#[test]
fn the_period_is_reported() {
    let timer = ThreadpoolPeriodicTimer::new(Duration::from_millis(25), |_| {}, None)
        .expect("create timer");
    assert_eq!(timer.period(), Duration::from_millis(25));
}

#[test]
fn drop_without_starting_is_clean() {
    let _timer =
        ThreadpoolPeriodicTimer::new(Duration::from_millis(1), |_| {}, None).expect("create timer");
}

// --- repetition ---

#[test]
fn start_ticks_repeatedly() {
    let (timer, fires) = counting_timer(Duration::from_millis(2));
    timer.start();
    fires.wait_for(5);
    timer.stop_and_drain();
    assert!(
        fires.count() >= 5,
        "a periodic timer must keep ticking; saw {}",
        fires.count()
    );
}

#[test]
fn start_after_controls_only_the_first_tick() {
    let (timer, fires) = counting_timer(Duration::from_millis(2));
    let started = Instant::now();
    timer.start_after(Duration::from_millis(80));
    fires.wait_for(1);
    let first = started.elapsed();
    // The first tick honours the explicit delay, not the period.
    assert!(
        first >= Duration::from_millis(60),
        "the first tick came after only {first:?}, well before the 80ms delay"
    );
    fires.wait_for(3);
    timer.stop_and_drain();
}

#[test]
fn a_running_timer_reports_running() {
    let (timer, fires) = counting_timer(Duration::from_millis(50));
    timer.start_after(Duration::from_millis(1));
    fires.wait_for(1);
    assert!(timer.is_running(), "ticking must not clear the schedule");
    timer.stop_and_drain();
    assert!(!timer.is_running());
}

#[test]
fn start_at_accepts_an_absolute_first_tick() {
    let (timer, fires) = counting_timer(Duration::from_millis(2));
    timer.start_at(SystemTime::now() + Duration::from_millis(10));
    fires.wait_for(3);
    timer.stop_and_drain();
}

#[test]
fn start_with_window_still_ticks() {
    let (timer, fires) = counting_timer(Duration::from_millis(2));
    timer.start_with_window(Duration::from_millis(1), Duration::from_millis(10));
    fires.wait_for(3);
    timer.stop_and_drain();
}

#[test]
fn restarting_replaces_the_previous_schedule() {
    let (timer, fires) = counting_timer(Duration::from_millis(2));
    timer.start_after(Duration::from_secs(3_600));
    timer.start_after(Duration::from_millis(1));
    fires.wait_for(3);
    timer.stop_and_drain();
}

// --- the overlap contract ---

/// The property that distinguishes this type: the pool queues ticks on schedule
/// regardless of whether the previous tick has finished, so a slow callback runs
/// concurrently with itself.
///
/// This is asserted as a capability rather than a guarantee -- the pool is free
/// to schedule as it sees fit -- so the test tolerates not observing overlap
/// while still failing if the timer stops ticking altogether.
#[test]
fn a_slow_callback_may_overlap_itself() {
    let concurrent = Arc::new(AtomicUsize::new(0));
    let peak = Arc::new(AtomicUsize::new(0));
    let fires = Fires::new();

    let in_flight = Arc::clone(&concurrent);
    let high_water = Arc::clone(&peak);
    let recorder = Arc::clone(&fires);

    // Ticks are scheduled far faster than the callback can complete.
    let timer = ThreadpoolPeriodicTimer::new(
        Duration::from_millis(2),
        move |_tick| {
            let now = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
            high_water.fetch_max(now, Ordering::SeqCst);
            std::thread::sleep(Duration::from_millis(25));
            in_flight.fetch_sub(1, Ordering::SeqCst);
            recorder.record();
        },
        None,
    )
    .expect("create timer");

    timer.start_after(Duration::from_millis(1));
    fires.wait_for(4);
    timer.stop_and_drain();

    let observed = peak.load(Ordering::SeqCst);
    assert!(observed >= 1, "the timer must have ticked at all");
    // Document what actually happened; overlap is expected but not mandated.
    if observed == 1 {
        eprintln!(
            "note: the pool did not overlap ticks in this run, though the callback \
             outlasts the period and may do so"
        );
    }
}

// --- stopping ---

#[test]
fn stop_prevents_further_ticks() {
    let (timer, fires) = counting_timer(Duration::from_millis(2));
    timer.start_after(Duration::from_millis(1));
    fires.wait_for(3);
    timer.stop_and_drain();

    let settled = fires.count();
    std::thread::sleep(Duration::from_millis(60));
    assert_eq!(
        fires.count(),
        settled,
        "no tick may run after stop and drain"
    );
}

#[test]
fn stopping_a_stopped_timer_is_a_no_op() {
    let (timer, fires) = counting_timer(Duration::from_millis(2));
    timer.stop();
    timer.stop();
    assert!(!timer.is_running());
    assert_eq!(fires.count(), 0);
}

#[test]
fn a_timer_can_be_restarted_after_stopping() {
    let (timer, fires) = counting_timer(Duration::from_millis(2));
    timer.start_after(Duration::from_millis(1));
    fires.wait_for(2);
    timer.stop_and_drain();
    let after_stop = fires.count();

    timer.start_after(Duration::from_millis(1));
    fires.wait_for(after_stop + 2);
    timer.stop_and_drain();
    assert!(fires.count() > after_stop);
}

/// A tick can stop its own timer, which is how "tick until done" is written.
#[test]
fn a_tick_can_stop_the_timer() {
    let ticks = Arc::new(AtomicUsize::new(0));
    let counter = Arc::clone(&ticks);
    let fires = Fires::new();
    let recorder = Arc::clone(&fires);

    let timer = ThreadpoolPeriodicTimer::new(
        Duration::from_millis(2),
        move |tick| {
            recorder.record();
            if counter.fetch_add(1, Ordering::SeqCst) >= 2 {
                tick.stop();
            }
        },
        None,
    )
    .expect("create timer");

    timer.start_after(Duration::from_millis(1));
    // The timer stops itself; wait for that rather than for a tick count.
    let deadline = Instant::now() + Duration::from_secs(30);
    while timer.is_running() && Instant::now() < deadline {
        std::thread::yield_now();
    }
    assert!(
        !timer.is_running(),
        "the tick should have stopped the timer"
    );

    timer.stop_and_drain();
    let settled = fires.count();
    std::thread::sleep(Duration::from_millis(50));
    assert_eq!(fires.count(), settled);
    assert!(settled >= 3);
}

// --- destruction ---

/// Dropping a running timer must terminate: `Drop` stops before draining, so the
/// timer cannot requeue itself forever.
// Dropping a started timer is this test's subject, so it must keep doing exactly
// that. Under `fail-fast` that drop is *required* to panic, so the armed
// behaviour is guarded by the child-process test instead.
#[cfg(not(feature = "fail-fast"))]
#[test]
fn drop_of_a_running_timer_terminates() {
    let started = Instant::now();
    {
        let (timer, fires) = counting_timer(Duration::from_millis(1));
        timer.start_after(Duration::from_millis(1));
        fires.wait_for(3);
        // Drop here, with the timer still running.
    }
    assert!(
        started.elapsed() < Duration::from_secs(10),
        "dropping a running periodic timer appears to have hung"
    );
}

/// Drop must wait for an executing tick before freeing the context.
// Dropping a started timer is this test's subject -- see the note above.
#[cfg(not(feature = "fail-fast"))]
#[test]
fn drop_waits_for_an_executing_tick() {
    let done = Arc::new(AtomicUsize::new(0));
    let flag = Arc::clone(&done);
    let started = Fires::new();
    let entered = Arc::clone(&started);

    {
        let timer = ThreadpoolPeriodicTimer::new(
            Duration::from_secs(3_600),
            move |_tick| {
                entered.record();
                std::thread::sleep(Duration::from_millis(30));
                flag.fetch_add(1, Ordering::SeqCst);
            },
            None,
        )
        .expect("create timer");
        // First tick almost immediately, then not again for an hour, so exactly
        // one tick is in flight when Drop runs.
        timer.start_after(Duration::from_millis(1));
        started.wait_for(1);
        // Drop here.
    }

    assert_eq!(
        done.load(Ordering::SeqCst),
        1,
        "Drop returned while a tick was still executing"
    );
}

// Dropping a started timer is this test's subject -- see the note above.
#[cfg(not(feature = "fail-fast"))]
#[test]
fn drop_while_started_but_not_yet_ticked_is_clean() {
    let (timer, fires) = counting_timer(Duration::from_secs(3_600));
    timer.start();
    drop(timer);
    assert_eq!(fires.count(), 0);
}

// --- callback state, environment, and thread-safety ---

#[test]
fn the_callback_may_own_heap_state() {
    let data = Arc::new(vec![2_u64, 4, 6]);
    let sum = Arc::new(AtomicUsize::new(0));
    let total = Arc::clone(&sum);
    let fires = Fires::new();
    let recorder = Arc::clone(&fires);

    let timer = ThreadpoolPeriodicTimer::new(
        Duration::from_millis(2),
        move |tick| {
            total.fetch_add(data.iter().sum::<u64>() as usize, Ordering::SeqCst);
            recorder.record();
            tick.stop();
        },
        None,
    )
    .expect("create timer");

    timer.start_after(Duration::from_millis(1));
    fires.wait_for(1);
    timer.stop_and_drain();
    assert!(sum.load(Ordering::SeqCst) >= 12);
}

#[test]
fn a_timer_runs_on_a_private_pool() {
    // Declared before the timer so it outlives it.
    let pool = ThreadpoolPool::new().expect("create pool");
    let mut env = CallbackEnviron::new();
    env.set_pool(&pool);

    let fires = Fires::new();
    let recorder = Arc::clone(&fires);
    let timer = ThreadpoolPeriodicTimer::new(
        Duration::from_millis(2),
        move |_tick| recorder.record(),
        Some(&mut env),
    )
    .expect("create timer");

    timer.start_after(Duration::from_millis(1));
    fires.wait_for(3);
    timer.stop_and_drain();
}

#[test]
fn a_periodic_timer_is_send_and_sync() {
    fn assert_send<T: Send>() {}
    fn assert_sync<T: Sync>() {}
    assert_send::<ThreadpoolPeriodicTimer>();
    assert_sync::<ThreadpoolPeriodicTimer>();
}

#[test]
fn a_timer_can_be_started_from_another_thread() {
    let (timer, fires) = counting_timer(Duration::from_millis(2));
    std::thread::scope(|scope| {
        let timer = &timer;
        scope.spawn(move || timer.start_after(Duration::from_millis(1)));
    });
    fires.wait_for(3);
    timer.stop_and_drain();
}

/// Whether a periodic `stop` discards a tick that is already queued.
///
/// **Measured on this type rather than inferred from the one-shot timer.**
/// `disarming_cancels_a_queued_tick_which_a_waits_disarm_does_not` pins the
/// answer for `ThreadpoolTimer`, and this type's documentation used to claim
/// the same behaviour by citing that test -- a claim about `SetThreadpoolTimer`
/// with a non-zero period, resting on a measurement of one with no period at
/// all. `M-T13.4` is that extrapolation being replaced by an observation.
///
/// The period is long enough that exactly one tick comes due inside the window:
/// the pool's only thread is held by a blocking work item, so the tick is
/// queued and cannot run, and no second tick is scheduled for ten seconds.
#[test]
fn stopping_a_periodic_timer_discards_a_tick_that_is_already_queued() {
    let mut outcomes = Vec::new();
    for stop_first in [false, true] {
        let pool = ThreadpoolPool::new().expect("create the private pool");
        pool.set_min_threads(1).expect("one thread minimum");
        pool.set_max_threads(1).expect("one thread maximum");
        let mut env = CallbackEnviron::new();
        env.set_pool(&pool);

        // `(entered, open)`. The first half is the handshake: submitting the
        // occupier does not prove the pool's only worker has picked it up, and
        // until it has, the thread is free to take the tick instead -- which
        // would run the tick, make the `ran == 0` precondition below false, and
        // fail this test for a reason that has nothing to do with what it
        // measures. Waiting for the occupier to report that it is *inside* its
        // callback establishes the single-worker occupancy rather than
        // inferring it from the order of two calls.
        let gate = Arc::new((
            std::sync::Mutex::new((false, false)),
            std::sync::Condvar::new(),
        ));
        let gate_for_work = Arc::clone(&gate);
        let occupier = crate::work::ThreadpoolWork::new(
            move || {
                let (lock, cvar) = &*gate_for_work;
                let mut state = lock.lock().unwrap_or_else(|p| p.into_inner());
                state.0 = true;
                cvar.notify_all();
                while !state.1 {
                    state = cvar.wait(state).unwrap_or_else(|p| p.into_inner());
                }
            },
            Some(&mut env),
        )
        .expect("create the occupying work item");
        occupier.submit();
        {
            let (lock, cvar) = &*gate;
            let mut state = lock.lock().unwrap_or_else(|p| p.into_inner());
            while !state.0 {
                state = cvar.wait(state).unwrap_or_else(|p| p.into_inner());
            }
        }

        let ran = Arc::new(AtomicUsize::new(0));
        let ran_for_callback = Arc::clone(&ran);
        let timer = ThreadpoolPeriodicTimer::new(
            Duration::from_secs(10),
            move |_tick| {
                ran_for_callback.fetch_add(1, Ordering::SeqCst);
            },
            Some(&mut env),
        )
        .expect("create the periodic timer");
        timer.start_after(Duration::from_millis(1));
        std::thread::sleep(Duration::from_millis(60));
        assert_eq!(
            ran.load(Ordering::SeqCst),
            0,
            "the pool's only thread is occupied, so the tick must still be queued"
        );

        if stop_first {
            timer.stop();
        }
        {
            let (lock, cvar) = &*gate;
            lock.lock().unwrap_or_else(|p| p.into_inner()).1 = true;
            cvar.notify_all();
        }
        timer.wait();
        std::thread::sleep(Duration::from_millis(100));
        outcomes.push(ran.load(Ordering::SeqCst));
        timer.stop_and_drain();
        occupier.stop_and_drain();
    }
    assert_eq!(
        outcomes[0], 1,
        "without a stop the queued tick runs, so the setup really does queue one"
    );
    assert_eq!(
        outcomes[1], 0,
        "with a stop the queued tick is discarded -- the same asymmetry with \
         waits that the one-shot timer shows, now measured on this type rather \
         than assumed from it"
    );
}