gunnar-server 2.0.0

//! **The account purge queue, and the one property it exists to guarantee: the
//! lockout comes OFF.**
//!
//! `PurgeQueue` is shaped like `GcQueue` — one worker, FIFO, deduplicated by
//! subject — and the shared properties are asserted the same way, on counters
//! about work that actually happened rather than on sleeps. Those tests are
//! here rather than borrowed from `gc_queue.rs` because the two queues are two
//! objects and a property proved about one is not a property of the other.
//!
//! What is genuinely new, and what the whole module is for:
//!
//! **A purge holds a total lockout on a customer's account while it runs.** If
//! it can end without releasing that lockout, the account is bricked: shut, and
//! shut by a row nobody remembers writing, held by a process that no longer
//! exists. The lease in `gunnar_core::gate` is the backstop for the case where
//! the whole server dies; these tests are the near cases, and every one of them
//! was seen red against a queue that released the gate on the success path
//! only:
//!
//! * the purge returns an error  → gate released, 0 released before the fix;
//! * the purge PANICS            → gate released, and the worker survives to
//!   take the next job; before the fix the panic unwound the worker thread, the
//!   account stayed shut, and every subsequent purge sat QUEUED for ever behind
//!   a thread that was gone.
//!
//! The assertions are on the release closure's own call log — the applied
//! output — never on the job's `gate_released` flag alone, which a queue could
//! set without calling anything.

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

use gunnar_server::purge_queue::{GateRelease, PurgeOrder, PurgeQueue, PurgeRunner, PurgeState};

/// Wait for `cond`, failing rather than hanging.
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}");
}

/// Every `(account, job_id)` the queue asked to have unlocked, in order.
#[derive(Clone, Default)]
struct ReleaseLog(Arc<Mutex<Vec<(String, String)>>>);

impl ReleaseLog {
    fn release(&self) -> GateRelease {
        let log = Arc::clone(&self.0);
        Arc::new(move |account: &str, job_id: &str| {
            log.lock()
                .unwrap()
                .push((account.to_owned(), job_id.to_owned()));
            Ok(true)
        })
    }

    fn accounts(&self) -> Vec<String> {
        self.0
            .lock()
            .unwrap()
            .iter()
            .map(|(account, _)| account.clone())
            .collect()
    }
}

/// **A purge that fails still unlocks the account.**
///
/// The failure mode this guards is not exotic: one unreadable directory in an
/// account of two hundred is enough. A purge that gave up there and left the
/// gate on would leave a paying customer locked out of an account that still
/// has their data in it, for a reason no listing explains.
#[test]
fn a_purge_that_fails_still_releases_the_lockout() {
    let log = ReleaseLog::default();
    let runner: PurgeRunner = Arc::new(|_order, progress| {
        progress.found(3);
        progress.reclaimed(1024);
        progress.failed("acme/two.git: Permission denied".to_owned());
        Err("1 of 3 stores could not be reclaimed".to_owned())
    });
    let queue = PurgeQueue::new(runner, log.release());

    let job = queue.enqueue(PurgeOrder::new("acme", queue.next_job_id(), "an unpaid invoice"));
    wait_until("the purge to settle", || {
        queue
            .status("acme")
            .is_some_and(|job| !job.state.is_live())
    });
    let settled = queue.status("acme").expect("the job is still held");

    assert_eq!(settled.state, PurgeState::Failed, "{settled:?}");
    assert_eq!(
        log.accounts(),
        vec!["acme".to_owned()],
        "the queue must have ASKED for the lockout to come off — asserted on the release \
         closure's own log, because a job that merely set a flag would be green on the flag"
    );
    assert!(
        settled.gate_released,
        "and it must say so on the job, because that is the field an operator reads"
    );
    assert_eq!(settled.job_id, job.job_id, "the same job settled");
    assert_eq!(
        settled.failures,
        vec!["acme/two.git: Permission denied".to_owned()],
        "a failed purge must name what is still on the disk; that list is the recovery plan"
    );
    // Partial progress is kept, not discarded. What it did reclaim is gone and
    // reporting 0 would send an operator looking for bytes that are not there.
    assert_eq!(settled.bytes_reclaimed, 1024);
    assert_eq!(settled.stores_found, 3);
}

/// **A purge that PANICS still unlocks the account, and the worker lives.**
///
/// Two failures in one, and the second is the nastier: an unwinding worker
/// thread takes the queue with it, so every account purged after the first
/// panic sits QUEUED for ever behind a thread that no longer exists — a second
/// account bricked by the first one's bug.
#[test]
fn a_purge_that_panics_releases_the_lockout_and_the_worker_survives() {
    let log = ReleaseLog::default();
    let runner: PurgeRunner = Arc::new(|order: &PurgeOrder, progress| {
        if order.account == "boom" {
            panic!("a deliberate panic inside a purge");
        }
        progress.found(1);
        progress.reclaimed(7);
        Ok(())
    });
    let queue = PurgeQueue::new(runner, log.release());

    queue.enqueue(PurgeOrder::new("boom", queue.next_job_id(), "a purge that will panic"));
    wait_until("the panicking purge to settle", || {
        queue
            .status("boom")
            .is_some_and(|job| !job.state.is_live())
    });
    let settled = queue.status("boom").expect("the job is still held");
    assert_eq!(
        settled.state,
        PurgeState::Failed,
        "a panic is a failure and must be reported as one, not left RUNNING for ever"
    );
    assert!(
        settled.gate_released,
        "a panicking purge must not leave the account locked: {settled:?}"
    );
    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:?}"
    );

    // The worker took the next job, which is the half a `catch_unwind` around
    // the runner buys that a `Drop` guard on its own would not.
    queue.enqueue(PurgeOrder::new("after", queue.next_job_id(), "the next account"));
    wait_until("the job after the panic to settle", || {
        queue
            .status("after")
            .is_some_and(|job| !job.state.is_live())
    });
    let after = queue.status("after").expect("the job is still held");
    assert_eq!(
        after.state,
        PurgeState::Done,
        "the worker must survive a panicking purge; a dead worker leaves every later account \
         QUEUED for ever: {after:?}"
    );
    assert_eq!(log.accounts(), vec!["boom".to_owned(), "after".to_owned()]);
}

/// A release that itself fails is reported, not swallowed — and it is the one
/// state that must be loud, because the account is empty AND shut.
#[test]
fn a_release_that_fails_is_reported_on_the_job() {
    let release: GateRelease = Arc::new(|_account, _job| Err("the identity table is gone".into()));
    let runner: PurgeRunner = Arc::new(|_order, progress| {
        progress.found(1);
        progress.reclaimed(1);
        Ok(())
    });
    let queue = PurgeQueue::new(runner, release);

    queue.enqueue(PurgeOrder::new("acme", queue.next_job_id(), "a purge whose unlock will fail"));
    wait_until("the purge to settle", || {
        queue
            .status("acme")
            .is_some_and(|job| !job.state.is_live())
    });
    let settled = queue.status("acme").expect("the job is still held");

    assert!(
        !settled.gate_released,
        "the unlock did not happen, so the job must not claim it did: {settled:?}"
    );
    assert_eq!(
        settled.state,
        PurgeState::Failed,
        "a purge that emptied the account but could not unlock it is NOT a success — the \
         customer cannot use what is left of their account: {settled:?}"
    );
    assert!(
        settled
            .error
            .as_deref()
            .is_some_and(|err| err.contains("identity table")),
        "and the reason must survive to the operator: {settled:?}"
    );
}

/// **A gate somebody else holds is left alone.** The release closure answers
/// `false` — "there was nothing of mine to lift" — and that is not a failure.
#[test]
fn a_lockout_held_by_somebody_else_is_left_in_place_and_is_not_an_error() {
    let release: GateRelease = Arc::new(|_account, _job| Ok(false));
    let runner: PurgeRunner = Arc::new(|_order, progress| {
        progress.found(2);
        progress.reclaimed(2048);
        Ok(())
    });
    let queue = PurgeQueue::new(runner, release);

    queue.enqueue(PurgeOrder::new("acme", queue.next_job_id(), "a purge under an operator's suspension"));
    wait_until("the purge to settle", || {
        queue
            .status("acme")
            .is_some_and(|job| !job.state.is_live())
    });
    let settled = queue.status("acme").expect("the job is still held");

    assert_eq!(
        settled.state,
        PurgeState::Done,
        "the purge did what it was asked; somebody else's suspension is not its failure: \
         {settled:?}"
    );
    assert!(
        !settled.gate_released,
        "the account is still shut, on the other holder's authority, and the job must say so \
         rather than claim an unlock it did not make: {settled:?}"
    );
}

/// **Ten accounts at once are ten purges, one at a time.** Asserted on an
/// observed peak, as `gc_queue.rs` does: a purge is minutes of unlink I/O and
/// ten of them together is a box that has stopped answering.
#[test]
fn ten_concurrent_requests_never_run_two_purges_at_once() {
    let live = Arc::new(AtomicUsize::new(0));
    let runner_peak = Arc::new(AtomicUsize::new(0));
    let seen = Arc::clone(&runner_peak);
    let runner: PurgeRunner = Arc::new(move |_order, progress| {
        let now = live.fetch_add(1, Ordering::SeqCst) + 1;
        runner_peak.fetch_max(now, Ordering::SeqCst);
        std::thread::sleep(Duration::from_millis(20));
        live.fetch_sub(1, Ordering::SeqCst);
        progress.found(1);
        progress.reclaimed(1);
        Ok(())
    });
    let queue = PurgeQueue::new(runner, ReleaseLog::default().release());

    let accounts: Vec<String> = (0..10).map(|n| format!("acct{n}")).collect();
    let start = Arc::new(std::sync::Barrier::new(accounts.len()));
    let handles: Vec<_> = accounts
        .iter()
        .map(|account| {
            let queue = queue.clone();
            let account = account.clone();
            let start = Arc::clone(&start);
            std::thread::spawn(move || {
                let job_id = queue.next_job_id();
                start.wait();
                queue.enqueue(PurgeOrder::new(&account, job_id, "a concurrent purge"));
            })
        })
        .collect();
    for handle in handles {
        handle.join().expect("the enqueueing thread");
    }

    wait_until("every purge to finish", || {
        accounts
            .iter()
            .all(|a| queue.status(a).is_some_and(|job| job.state == PurgeState::Done))
    });

    assert_eq!(
        queue.peak_in_flight(),
        1,
        "two purges overlapped; a purge is minutes of unlink I/O and two at once is a box that \
         has stopped answering"
    );
    assert_eq!(
        seen.load(Ordering::SeqCst),
        1,
        "the RUNNER saw two of itself at once, so the queue's own instrument is not the \
         thing being trusted here"
    );
    assert_eq!(queue.runs(), 10, "every account must actually have been purged");
}

/// **Asking twice for one account is one purge.** Asserted on how many times
/// the runner was called, which a queue that only collapsed job ids could not
/// fake — and it matters more here than for a gc, because a second purge would
/// take a second lease on a gate the first one is about to release.
#[test]
fn a_second_request_for_a_running_account_is_the_same_job() {
    let gate = Arc::new(std::sync::Barrier::new(2));
    let held = Arc::clone(&gate);
    let runner: PurgeRunner = Arc::new(move |_order, progress| {
        held.wait();
        progress.found(1);
        progress.reclaimed(1);
        Ok(())
    });
    let queue = PurgeQueue::new(runner, ReleaseLog::default().release());

    let first = queue.enqueue(PurgeOrder::new("acme", queue.next_job_id(), "the first request"));
    wait_until("the purge to start", || {
        queue
            .status("acme")
            .is_some_and(|job| job.state == PurgeState::Running)
    });
    let second = queue.enqueue(PurgeOrder::new("acme", queue.next_job_id(), "the impatient second click"));
    assert_eq!(
        second.job_id, first.job_id,
        "a second request for an account already being purged must join the job that is there"
    );

    gate.wait();
    wait_until("the purge to finish", || {
        queue
            .status("acme")
            .is_some_and(|job| job.state == PurgeState::Done)
    });
    assert_eq!(
        queue.runs(),
        1,
        "the impatient click ran the purge a second time; the applied output is the only \
         place that shows it"
    );
}

/// A finished job is asked for AGAIN, and that is a new purge — not a refusal.
/// An account can accumulate new bytes after a purge (an ungate, a push, a
/// second cancellation), and a queue that answered "already done" would be
/// deciding it knows better.
#[test]
fn a_finished_account_may_be_purged_again() {
    let runner: PurgeRunner = Arc::new(|_order, progress| {
        progress.found(1);
        progress.reclaimed(1);
        Ok(())
    });
    let queue = PurgeQueue::new(runner, ReleaseLog::default().release());

    let first = queue.enqueue(PurgeOrder::new("acme", queue.next_job_id(), "the first purge"));
    wait_until("the first purge to finish", || {
        queue
            .status("acme")
            .is_some_and(|job| job.state == PurgeState::Done)
    });
    let second = queue.enqueue(PurgeOrder::new("acme", queue.next_job_id(), "the second purge"));
    assert_ne!(
        second.job_id, first.job_id,
        "a purge that finished is not the answer to a request made afterwards"
    );
    wait_until("the second purge to finish", || {
        queue
            .status("acme")
            .is_some_and(|job| job.job_id == second.job_id && job.state == PurgeState::Done)
    });
    assert_eq!(queue.runs(), 2);
}