gunnar-server 3.0.0

//! **The two properties `GcQueue` exists for, asserted on what was observed.**
//!
//! A gc unlinks the whole exploded read cache and is pure I/O. The queue is
//! there so ten console buttons pressed at once are ten gcs run one after
//! another instead of one box that stopped answering, and so an impatient
//! double-click is one gc instead of two.
//!
//! Neither gate is timing-based, because a timing-based test on a shared box is
//! not evidence — it is a coin toss that usually lands the right way up. Both
//! read **counters the queue keeps about work that actually happened**:
//!
//! - `peak_in_flight()` — the highest number of runner calls that were ever
//!   overlapping. The sleeps below only widen the window a violation would have
//!   to fit into; the assertion is on the peak, and a queue that ran everything
//!   sequentially by luck still reports 1.
//! - `runs()` — how many times the runner was CALLED. A dedup that only
//!   collapsed job ids while running the work twice would be green on ids and
//!   red here.
//!
//! Both were seen red against a deliberately wrong queue before they were
//! trusted green: see the commit message for the exact numbers.

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

use gunnar_server::gc_queue::{GcQueue, GcRunner, GcState};

/// Wait for `cond`, failing the test rather than hanging for ever.
///
/// A bounded wait and not a fixed sleep: the assertion after it is what decides
/// the test, and this only makes "the queue never got there" a failure with a
/// message instead of a hung suite.
fn wait_until(what: &str, mut cond: impl FnMut() -> bool) {
    let deadline = Instant::now() + Duration::from_secs(30);
    while Instant::now() < deadline {
        if cond() {
            return;
        }
        std::thread::sleep(Duration::from_millis(5));
    }
    panic!("timed out waiting for {what}");
}

/// A runner that records how many calls overlapped, and dwells long enough that
/// two running together would be seen.
fn overlapping_runner(dwell: Duration) -> (GcRunner, Arc<AtomicUsize>) {
    let peak = Arc::new(AtomicUsize::new(0));
    let live = Arc::new(AtomicUsize::new(0));
    let observed = Arc::clone(&peak);
    let runner: GcRunner = Arc::new(move |_path: &str| {
        let now = live.fetch_add(1, Ordering::SeqCst) + 1;
        peak.fetch_max(now, Ordering::SeqCst);
        std::thread::sleep(dwell);
        live.fetch_sub(1, Ordering::SeqCst);
        Ok(4096)
    });
    (runner, observed)
}

/// **Ten requests at once are ten gcs, one at a time.**
///
/// The assertion is an observed maximum, twice over: the queue's own
/// `peak_in_flight()` and a peak the *runner* counts for itself, so a queue
/// that instrumented its bookkeeping correctly while actually running two would
/// still be red.
///
/// And every job must have COMPLETED with its bytes. Without that, a queue that
/// ran nothing at all would pass a peak-of-1 assertion perfectly.
#[test]
fn ten_concurrent_requests_never_run_two_gcs_at_once() {
    let repos: Vec<String> = (0..10).map(|n| format!("acct/repo{n}")).collect();
    let (runner, runner_peak) = overlapping_runner(Duration::from_millis(20));
    let queue = GcQueue::new(runner);

    // Enqueued from ten threads, released together, so the requests genuinely
    // race rather than arriving in a tidy line.
    let start = Arc::new(std::sync::Barrier::new(repos.len()));
    let handles: Vec<_> = repos
        .iter()
        .cloned()
        .map(|path| {
            let queue = queue.clone();
            let start = Arc::clone(&start);
            std::thread::spawn(move || {
                start.wait();
                queue.enqueue(&path)
            })
        })
        .collect();
    for handle in handles {
        handle.join().expect("an enqueueing thread panicked");
    }

    wait_until("every gc to finish", || {
        repos
            .iter()
            .all(|path| matches!(queue.status(path).map(|j| j.state), Some(GcState::Done)))
    });

    assert_eq!(
        queue.runs(),
        repos.len() as u64,
        "every request must actually have run a gc; a queue that dropped work \
         would pass the overlap assertion below for the wrong reason"
    );
    assert_eq!(
        queue.peak_in_flight(),
        1,
        "two gcs ran at the same time. The whole point of this queue is that \
         they cannot: a gc is hundreds of thousands of unlinks and the box has \
         one disk."
    );
    assert_eq!(
        runner_peak.load(Ordering::SeqCst),
        1,
        "the RUNNER saw overlapping calls, whatever the queue's own counter says"
    );
    for path in &repos {
        let job = queue.status(path).expect("a finished job stays readable");
        assert_eq!(job.bytes_reclaimed, 4096, "{path} reported no bytes");
        assert_eq!(job.queue_position, 0, "a finished job is not waiting");
    }
    queue.stop();
}

/// **Pressing the button twice is one job, and one gc.**
///
/// The first job is held inside the runner while the second request arrives, so
/// the second one meets a genuinely live job rather than a finished one. Both
/// halves are asserted: the same `job_id` back (what a console sees) and
/// `runs() == 1` (what the disk sees). The second is the one that cannot be
/// faked.
#[test]
fn a_double_press_does_not_enqueue_twice() {
    let release = Arc::new((Mutex::new(false), Condvar::new()));
    let entered = Arc::new(AtomicUsize::new(0));

    let held = Arc::clone(&release);
    let counted = Arc::clone(&entered);
    let runner: GcRunner = Arc::new(move |_path: &str| {
        counted.fetch_add(1, Ordering::SeqCst);
        let (lock, cv) = &*held;
        let mut go = lock.lock().unwrap();
        while !*go {
            go = cv.wait(go).unwrap();
        }
        Ok(77)
    });
    let queue = GcQueue::new(runner);

    let first = queue.enqueue("acct/impatient");
    wait_until("the first gc to start", || {
        entered.load(Ordering::SeqCst) == 1
    });

    // The second press, while the first is unambiguously still running.
    let second = queue.enqueue("acct/impatient");
    assert_eq!(
        second.job_id, first.job_id,
        "a second request for a repository already being gc'd must return the \
         job that is already there, not start another"
    );
    assert_eq!(second.state, GcState::Running);

    // A third, for good measure, while it is still held.
    let third = queue.enqueue("acct/impatient");
    assert_eq!(third.job_id, first.job_id);

    {
        let (lock, cv) = &*release;
        *lock.lock().unwrap() = true;
        cv.notify_all();
    }
    wait_until("the gc to finish", || {
        matches!(
            queue.status("acct/impatient").map(|j| j.state),
            Some(GcState::Done)
        )
    });

    assert_eq!(
        queue.runs(),
        1,
        "three requests for one repository ran {} gcs. Dedup that only \
         collapses job ids is not dedup.",
        queue.runs()
    );
    assert_eq!(
        queue
            .status("acct/impatient")
            .expect("still readable")
            .bytes_reclaimed,
        77
    );
    queue.stop();
}

/// A job that finished stays readable, so a console polling across the finish
/// sees the result rather than an empty answer it cannot tell from "never
/// asked".
///
/// And "never asked" is itself asserted, because the two must be different: a
/// `status` that answered `Some(default)` for an unknown repository would make
/// a button light up for a gc nobody requested.
#[test]
fn a_finished_job_stays_readable_and_an_unasked_one_is_absent() {
    let (runner, _) = overlapping_runner(Duration::from_millis(1));
    let queue = GcQueue::new(runner);

    assert!(
        queue.status("acct/never-asked").is_none(),
        "a repository nobody has asked about must have no job at all"
    );

    queue.enqueue("acct/done");
    wait_until("the gc to finish", || {
        matches!(
            queue.status("acct/done").map(|j| j.state),
            Some(GcState::Done)
        )
    });
    let job = queue.status("acct/done").expect("kept after finishing");
    assert_eq!(job.state, GcState::Done);
    assert_eq!(job.bytes_reclaimed, 4096);
    assert!(job.finished_at_unix_ms > 0, "a finished job carries when");
    assert!(
        queue.status("acct/never-asked").is_none(),
        "one repository's job must not answer for another's"
    );
    queue.stop();
}

/// A failing gc leaves a FAILED job carrying the reason, and does not wedge the
/// worker: the next repository still runs.
#[test]
fn a_failed_gc_is_reported_and_the_queue_carries_on() {
    let runner: GcRunner = Arc::new(|path: &str| {
        if path == "acct/broken" {
            Err("the store would not open".into())
        } else {
            Ok(9)
        }
    });
    let queue = GcQueue::new(runner);
    queue.enqueue("acct/broken");
    queue.enqueue("acct/fine");

    wait_until("both jobs to settle", || {
        matches!(
            queue.status("acct/broken").map(|j| j.state),
            Some(GcState::Failed)
        ) && matches!(
            queue.status("acct/fine").map(|j| j.state),
            Some(GcState::Done)
        )
    });

    let broken = queue.status("acct/broken").expect("kept");
    assert_eq!(
        broken.error.as_deref(),
        Some("the store would not open"),
        "a failed job must carry the reason; 'failed' alone is not actionable"
    );
    assert_eq!(broken.bytes_reclaimed, 0, "a failed gc reclaimed nothing");
    assert_eq!(queue.status("acct/fine").expect("kept").bytes_reclaimed, 9);
    queue.stop();
}