liminal-server 0.28.9

//! Flapjack's post 108 (Cambium record b4e939ab, 25 October): a stop that lands while the start
//! thread admits what it held must never let an admission run after the stop, and must wait for
//! the one already running, so the stop's connection snapshot sees every connection it spawned.

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, mpsc};

use super::HeldAdmissions;

type Admission = Box<dyn FnOnce() + Send>;

#[test]
fn a_stop_during_the_admissions_waits_for_the_running_one_and_runs_none_after()
-> Result<(), Box<dyn std::error::Error>> {
    let held: Arc<HeldAdmissions<Admission>> = Arc::new(HeldAdmissions::default());
    held.starting();
    let ran = Arc::new(AtomicUsize::new(0));
    let (entered, entering) = mpsc::channel();
    let (release, released) = mpsc::channel::<()>();
    let first = Arc::clone(&ran);
    assert!(
        held.hold(Box::new(move || {
            // A closed channel means the test has already failed; the admission still ends.
            if entered.send(()).is_ok() && released.recv().is_err() {
                eprintln!("the test released nothing: the admission ends");
            }
            first.fetch_add(1, Ordering::SeqCst);
        }))
        .is_none()
    );
    let second = Arc::clone(&ran);
    assert!(
        held.hold(Box::new(move || {
            second.fetch_add(10, Ordering::SeqCst);
        }))
        .is_none()
    );

    assert!(held.admitting(), "the start published with admissions held");
    let start = {
        let held = Arc::clone(&held);
        std::thread::spawn(move || held.admit_held(|admit| admit()))
    };
    entering.recv()?;
    let (stopped_tx, stopped) = mpsc::channel();
    let stop = {
        let held = Arc::clone(&held);
        std::thread::spawn(move || {
            held.stop();
            if stopped_tx.send(()).is_err() {
                eprintln!("the test stopped listening before the stop returned");
            }
        })
    };
    assert!(
        held.wait_stopped(std::time::Duration::from_secs(30)),
        "the stop never began"
    );
    // The stop has begun and the admission is still inside its call: the stop must not return.
    assert!(
        stopped
            .recv_timeout(std::time::Duration::from_millis(100))
            .is_err(),
        "the stop returned while an admission it must count was still running"
    );
    release.send(())?;
    stopped.recv()?;
    assert_eq!(
        ran.load(Ordering::SeqCst),
        1,
        "the running admission finished and the second never ran"
    );
    start.join().map_err(|_| "the start thread panicked")?;
    stop.join().map_err(|_| "the stop panicked")?;
    assert!(
        held.hold(Box::new(|| {})).is_none(),
        "an admission after the stop is dropped"
    );
    Ok(())
}

#[test]
fn admissions_held_while_admitting_run_in_arrival_order_then_admissions_run_at_once() {
    let held: HeldAdmissions<usize> = HeldAdmissions::default();
    held.starting();
    assert!(held.hold(0).is_none());
    assert!(held.hold(1).is_none());
    assert!(held.admitting());
    let mut order = Vec::new();
    held.admit_held(|admission| {
        if admission == 0 {
            // one arriving while the start admits joins the end of the queue
            assert!(held.hold(2).is_none());
        }
        order.push(admission);
    });
    assert_eq!(order, [0, 1, 2]);
    assert_eq!(
        held.hold(3),
        Some(3),
        "after the admissions an admission runs at once"
    );
}

#[test]
fn a_stop_before_the_start_publishes_admits_nothing() {
    let held: HeldAdmissions<usize> = HeldAdmissions::default();
    held.starting();
    assert!(held.hold(0).is_none());
    held.stop();
    assert!(!held.admitting());
    let mut ran = Vec::new();
    held.admit_held(|admission| ran.push(admission));
    assert!(ran.is_empty());
}