crates/gunnar-server/tests/job_queue_third_kind.rs
//! **A third kind of job, added without touching the runner.**
//!
//! `gc_queue` and `purge_queue` were two copies of one queue: a worker thread,
//! a FIFO, dedup by subject, finished jobs kept for a poller. The copies were
//! folded into [`gunnar_server::job_queue`], and the claim that made the fold
//! worth doing is that a *third* kind costs a `JobKind` impl and nothing else.
//!
//! This file is that claim, asserted rather than promised. It is a whole job
//! kind — `Mirror`, standing in for the `Mirror.Sync` that `gunnar-wire`'s
//! `fetch_pack` is being built for — declared **here, in a test crate**, which
//! is the strongest form the claim can take: a test crate cannot edit
//! `job_queue.rs` even by accident, so a green run is proof that the runner
//! needed no change.
//!
//! What the kind exercises, and why each one is here:
//!
//! * **A report of its own.** `MirrorReport` has nothing in common with a gc's
//! `bytes_reclaimed` or a purge's `stores_purged`, and the runner never sees
//! inside it.
//! * **`progress` moving mid-job.** The field gc and purge never had, and the
//! one a mirror cannot do without: a fetch of a large repository is minutes
//! of network with a console watching, and a job that sat at `Running` and
//! then jumped to `Done` would tell that console nothing. Asserted *while the
//! job is still running*, because a progress field only read after the finish
//! is a progress field that does not work.
//! * **Dedup on the repository**, on the applied output — how many times the
//! fetch was actually CALLED — because a mirror that ran twice for one
//! double-click would pull the same pack over the wire twice.
//! * **A failure that stays readable**, so a `MirrorStatus` RPC shaped like
//! `GcStatus` has something to render.
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Barrier, Mutex};
use std::time::{Duration, Instant};
use gunnar_server::job_queue::{JobKind, JobQueue, JobState, Reporter};
/// Wait for `cond`, failing the test rather than hanging for ever.
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}");
}
/// What a mirror was asked for. Keyed by repository, exactly as a gc is.
#[derive(Debug, Clone)]
struct MirrorOrder {
path: String,
upstream: String,
}
/// What a mirror produced. The kind's own fields; the runner never reads them.
#[derive(Debug, Clone, Default)]
struct MirrorReport {
objects: u64,
bytes: u64,
}
/// The kind. Holds what the work needs, the way `Gc` holds its runner and
/// `Purge` holds its runner and its gate release.
struct Mirror {
/// A stand-in for `fetch_pack`: told which repository, reports progress,
/// and is counted so dedup can be asserted on the applied output.
fetch: Fetch,
}
/// The stand-in `fetch_pack` a [`Mirror`] calls.
type Fetch = Arc<dyn Fn(&MirrorOrder, &Reporter<Mirror>) -> Result<(), String> + Send + Sync>;
impl JobKind for Mirror {
const NAME: &'static str = "mirror";
type Order = MirrorOrder;
type Report = MirrorReport;
fn key(order: &MirrorOrder) -> &str {
&order.path
}
fn run(&self, order: &MirrorOrder, report: &Reporter<Mirror>) -> Result<(), String> {
(self.fetch)(order, report)
}
}
/// **The whole point: a mirror job runs, reports progress as it goes, and the
/// runner was not touched to make it possible.**
#[test]
fn a_third_kind_with_progress_registers_without_touching_the_runner() {
// The fetch is held mid-job so `progress` is read while the job is still
// RUNNING. A progress field that is only correct once the job is over is
// not a progress field.
let held = Arc::new(Barrier::new(2));
let inside = Arc::clone(&held);
let calls = Arc::new(AtomicUsize::new(0));
let counted = Arc::clone(&calls);
let queue = JobQueue::new(Mirror {
fetch: Arc::new(move |order: &MirrorOrder, report: &Reporter<Mirror>| {
counted.fetch_add(1, Ordering::SeqCst);
assert_eq!(order.upstream, "ssh://upstream/acct/repo.git");
report.progress(3, 10);
// Wait for the assertions on the moving progress, then finish.
inside.wait();
report.progress(10, 10);
report.report(|r: &mut MirrorReport| {
r.objects = 10;
r.bytes = 4096;
});
Ok(())
}),
});
let job = queue.enqueue(MirrorOrder {
path: "acct/repo".to_owned(),
upstream: "ssh://upstream/acct/repo.git".to_owned(),
});
assert!(
job.job_id.starts_with("mirror-"),
"a kind's job ids carry its name so an operator reading one knows what \
it was: {job:?}"
);
wait_until("the mirror to report its first progress", || {
queue
.status("acct/repo")
.is_some_and(|job| job.progress == Some((3, 10)))
});
let running = queue.status("acct/repo").expect("the job is held");
assert_eq!(
running.state,
JobState::Running,
"progress must be readable WHILE the job runs, which is the only time \
anybody needs it: {running:?}"
);
// The impatient second click, while the fetch is unambiguously still going.
let second = queue.enqueue(MirrorOrder {
path: "acct/repo".to_owned(),
upstream: "ssh://upstream/acct/repo.git".to_owned(),
});
assert_eq!(
second.job_id, job.job_id,
"a second request for a repository already being mirrored must join the \
job that is there, not pull the same pack down twice"
);
held.wait();
wait_until("the mirror to finish", || {
queue
.status("acct/repo")
.is_some_and(|job| job.state == JobState::Done)
});
let done = queue
.status("acct/repo")
.expect("a finished job stays readable");
assert_eq!(done.progress, Some((10, 10)));
assert_eq!(
done.objects, 10,
"the kind's own report survives the runner"
);
assert_eq!(done.bytes, 4096);
assert_eq!(
queue.runs(),
1,
"two requests ran {} fetches; dedup that only collapses job ids is not \
dedup",
queue.runs()
);
assert_eq!(calls.load(Ordering::SeqCst), 1);
assert_eq!(
queue.peak_in_flight(),
1,
"the runner's one-at-a-time guarantee holds for a kind it has never \
heard of"
);
queue.stop();
}
/// A mirror that fails leaves a FAILED job carrying the reason, and the worker
/// takes the next repository — the same two properties the gc queue's failure
/// test asserts, inherited rather than re-implemented.
#[test]
fn a_failed_mirror_is_reported_and_the_queue_carries_on() {
let queue = JobQueue::new(Mirror {
fetch: Arc::new(|order: &MirrorOrder, report: &Reporter<Mirror>| {
if order.path == "acct/gone" {
return Err("the upstream refused the fetch".to_owned());
}
report.progress(1, 1);
report.report(|r: &mut MirrorReport| r.objects = 1);
Ok(())
}),
});
for path in ["acct/gone", "acct/fine"] {
queue.enqueue(MirrorOrder {
path: path.to_owned(),
upstream: "ssh://upstream/x.git".to_owned(),
});
}
wait_until("both mirrors to settle", || {
queue
.status("acct/gone")
.is_some_and(|job| job.state == JobState::Failed)
&& queue
.status("acct/fine")
.is_some_and(|job| job.state == JobState::Done)
});
let failed = queue.status("acct/gone").expect("kept");
assert_eq!(
failed.error.as_deref(),
Some("the upstream refused the fetch"),
"a failed job must carry the reason; 'failed' alone is not actionable"
);
assert_eq!(queue.status("acct/fine").expect("kept").objects, 1);
queue.stop();
}
/// A mirror that PANICS is caught, reported, and the worker survives to take
/// the next repository.
///
/// gc never had this and purge paid for it in a test of its own; a third kind
/// gets it from the runner without asking. `fetch_pack` parses bytes somebody
/// else wrote, so the day it panics is a question of when.
#[test]
fn a_panicking_mirror_does_not_take_the_worker_with_it() {
let log: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
let seen = Arc::clone(&log);
let queue = JobQueue::new(Mirror {
fetch: Arc::new(move |order: &MirrorOrder, _report: &Reporter<Mirror>| {
seen.lock().expect("the log").push(order.path.clone());
if order.path == "acct/boom" {
panic!("a deliberate panic inside a fetch");
}
Ok(())
}),
});
queue.enqueue(MirrorOrder {
path: "acct/boom".to_owned(),
upstream: "ssh://upstream/boom.git".to_owned(),
});
wait_until("the panicking mirror to settle", || {
queue
.status("acct/boom")
.is_some_and(|job| !job.state.is_live())
});
let settled = queue.status("acct/boom").expect("kept");
assert_eq!(settled.state, JobState::Failed);
assert!(
settled
.error
.as_deref()
.is_some_and(|err| err.contains("panicked")),
"the error must say a panic happened, or an operator reads `failed` \
with no cause: {settled:?}"
);
queue.enqueue(MirrorOrder {
path: "acct/after".to_owned(),
upstream: "ssh://upstream/after.git".to_owned(),
});
wait_until("the mirror after the panic to settle", || {
queue
.status("acct/after")
.is_some_and(|job| job.state == JobState::Done)
});
assert_eq!(
log.lock().expect("the log").as_slice(),
["acct/boom".to_owned(), "acct/after".to_owned()],
"the worker must survive a panicking job; a dead worker leaves every \
later repository QUEUED for ever"
);
queue.stop();
}