crates/gunnar-server/tests/gc_queue.rs
//! **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();
}