gunnar-server 3.0.0

//! **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();
}