gunnar_server/

purge_queue.rs

//! **One account purge at a time, and the lockout always comes off.**
//!
//! `Accounts.Purge` empties a user's whole account: every repository, every
//! soft-delete tombstone still holding bytes, every orphan store under the
//! account's directory subtree, and the namespace rows themselves. That is
//! minutes of pure unlink I/O on a large account, so it is queued rather than
//! run inline, for exactly the reason [`crate::gc_queue`] gives for a gc.
//!
//! # Why this is a SECOND LINE and not more jobs on the gc one
//!
//! The mechanism is shared — both kinds run on [`crate::job_queue`], which is
//! one worker, one FIFO, dedup by subject and a finished job kept for a poller.
//! What is **not** shared is the line, and that is the ruling this module has
//! always argued for. It survives the extraction unchanged:
//!
//! **A purge holds a total lockout on a customer's account for the whole time
//! it is queued.** A gc holds nothing — it is maintenance, it is re-derivable,
//! and it can wait behind anything without costing anybody anything. Put them
//! in one queue and a user is locked out of their account while the box works
//! through an hour of somebody else's repository maintenance, because a gc that
//! arrived first is ahead of them. The queue a job waits in is part of what the
//! job costs, and these two jobs cost different things while waiting.
//!
//! So `CoreControlPlane` holds two [`crate::job_queue::JobQueue`]s and each has
//! its own jera lane. What used to be two copies of the queue is now two
//! instances of one, which is the same two lines and half the code.
//!
//! # The property this module exists for
//!
//! The gate goes on **before** the job is enqueued, in the RPC handler, so
//! there is no window in which a purge is accepted and the user can still push.
//! **That is deliberately not the queue's business** — the queue cannot promise
//! it, because the promise is about what is true when `Accounts.Purge`
//! *returns*, which is before the queue has done anything at all. The gate is
//! taken by the caller, under an id the caller takes from
//! [`PurgeQueue::next_job_id`] first.
//!
//! It comes off **here**, in [`JobKind::settle`], and it comes off on **every**
//! path out of the runner:
//!
//! * the runner returns `Ok`   → released;
//! * the runner returns `Err`  → released, and the job is FAILED with the
//!   reason and the list of stores still on the disk;
//! * the runner **panics**     → caught, released, the job is FAILED saying so,
//!   **and the worker survives**. An unwinding worker takes the queue with it,
//!   so the account after the panicking one would sit QUEUED for ever behind a
//!   thread that no longer exists — one account's bug bricking the next one's.
//!
//! The catch and the always-runs settle are now [`crate::job_queue`]'s, which
//! is the whole gain from the extraction: a gc, and a mirror, and anything
//! after them get the same two guarantees without a second person having to
//! rediscover why they matter.
//!
//! What this module cannot cover is the process dying between two unlinks.
//! That is what the lease in `gunnar_core::gate` is for: a purge's gate carries
//! a deadline, the runner re-stamps it as it goes, and a gate whose deadline
//! has passed is not a gate. Belt here, braces there, because the outcome they
//! both prevent — an account shut by a row nobody remembers, held by a process
//! that is gone — is worse than an unfinished purge. The boot sweep that
//! releases every lease a killed process left behind is
//! `CoreControlPlane::release_stale_purge_leases`, called once at start-up: it
//! is about rows in the identity table and not about jobs, so it is not the
//! queue's and did not move here.
//!
//! # What this module is not
//!
//! It is not the purge. That is [`crate::admin::purge_account`], reached
//! through a runner handed in at construction, so the queue can be tested with
//! no disk under it and there is one implementation of what emptying an account
//! means (one implementation). The release is handed in the same way and is
//! `AuthTable::lift_gate_held_by` in the real server — *held_by*, so a purge
//! finishing cannot lift a suspension an operator put on while it ran.

use std::sync::Arc;
use std::time::Duration;

use crate::job_queue::{Job, JobKind, JobQueue, JobState, Reporter};

/// How long a finished purge stays readable. The same ten minutes
/// [`crate::gc_queue::FINISHED_RETENTION`] uses, and for the same reason: far
/// longer than any console poll, far shorter than a session.
///
/// Its own constant rather than a re-export, because the two queues holding the
/// same number today is a coincidence and not a contract — a purge's result is
/// the more valuable of the two to still have when somebody looks back. The
/// runner takes it per kind ([`JobKind::FINISHED_RETENTION`]) for that reason.
pub const FINISHED_RETENTION: Duration = <Purge as JobKind>::FINISHED_RETENTION;

/// Finished jobs kept regardless of age, most recent first.
pub const FINISHED_KEPT: usize = <Purge as JobKind>::FINISHED_KEPT;

/// Where a purge is. [`JobState`]'s four words, shared with every other kind of
/// job so `PurgeStatus` and `GcStatus` cannot drift apart about what `Done`
/// means.
pub type PurgeState = JobState;

/// One account purge, running or completed.
///
/// A [`Job`] of kind [`Purge`]: the queue's fields (`job_id`, `state`,
/// `queue_position`, `error`, the timestamps) plus a purge's [`PurgeReport`],
/// which dereferences flat — `job.stores_purged` reads the report and
/// `job.state` the queue. The account is `job.key`.
pub type PurgeJob = Job<Purge>;

/// What a purge found, reclaimed, and gave back.
#[derive(Debug, Clone, Default)]
pub struct PurgeReport {
    /// Stores the walk found. **0 until the job starts.**
    pub stores_found: u32,
    pub stores_purged: u32,
    pub bytes_reclaimed: u64,
    pub principal_removed: bool,
    /// **Whether the lockout this job took has been released**, as observed —
    /// set from what the release closure actually answered, never optimistically.
    pub gate_released: bool,
    /// One line per store that could not be reclaimed. The recovery list.
    pub failures: Vec<String>,
}

/// **What a purge was asked for**, carried on the job so the worker can hand it
/// back to the runner minutes later.
///
/// The queue holds the whole order rather than the account name alone because
/// `remove_principal` and `expect_empty` change what a purge *is*, and a queue
/// that kept them in a side map keyed by account would be a second place the
/// two could disagree — with the more destructive reading winning whenever they
/// did.
#[derive(Debug, Clone)]
pub struct PurgeOrder {
    /// The account: a principal id, which is also its namespace.
    pub account: String,
    /// **The job id, minted before the gate was taken**, because the gate's
    /// holder has to be this job or the release could not tell this lockout
    /// from anybody else's. See [`PurgeQueue::next_job_id`].
    pub job_id: String,
    /// The principal that asked, for the audit trail.
    pub principal_id: String,
    /// Why, shown to the locked-out user and written into every reclaim's log
    /// line.
    pub reason: String,
    /// Also remove the principal, its keys, grants, tokens and memberships.
    pub remove_principal: bool,
    /// Refuse rather than destroy unless the account holds no live
    /// repositories.
    pub expect_empty: bool,
}

impl PurgeOrder {
    /// The ordinary order: empty the account, keep the principal.
    pub fn new(
        account: impl Into<String>,
        job_id: impl Into<String>,
        reason: impl Into<String>,
    ) -> Self {
        PurgeOrder {
            account: account.into(),
            job_id: job_id.into(),
            principal_id: String::new(),
            reason: reason.into(),
            remove_principal: false,
            expect_empty: false,
        }
    }

    /// Record who asked.
    pub fn by(mut self, principal_id: impl Into<String>) -> Self {
        self.principal_id = principal_id.into();
        self
    }

    /// Take the principal with the data.
    pub fn removing_principal(mut self, remove: bool) -> Self {
        self.remove_principal = remove;
        self
    }

    /// Refuse rather than destroy if anything is still live.
    pub fn expecting_empty(mut self, expect: bool) -> Self {
        self.expect_empty = expect;
        self
    }
}

/// **What a purge actually is**, handed in so this module does not have to know.
///
/// Takes the order and a [`Progress`] to report through, and returns `Ok` when
/// the account is empty. The real one is
/// [`crate::account_purge::purge_account`].
///
/// It reports through a handle rather than returning a summary because a purge
/// of a large account runs for minutes and a console is polling the whole time:
/// a return value can only be read once the walk is over, which is precisely
/// when nobody needs it any more.
pub type PurgeRunner = Arc<dyn Fn(&PurgeOrder, &Progress) -> Result<(), String> + Send + Sync>;

/// **How the lockout this job took is released.**
///
/// `(account, job_id) -> Ok(true)` when this job's gate was lifted, `Ok(false)`
/// when there was nothing of this job's to lift — an operator replaced the row
/// while the purge ran, or the lease expired and something else took it. The
/// real one is `AuthTable::lift_gate_held_by`.
///
/// `Ok(false)` is **not** a failure: the account is shut on somebody else's
/// authority, which is a correct outcome. `Err` is, and it is the loudest
/// state this queue can reach — the account is empty AND locked.
pub type GateRelease = Arc<dyn Fn(&str, &str) -> Result<bool, String> + Send + Sync>;

/// The handle a running purge reports through.
///
/// The shared runner's [`Reporter`], with a purge's vocabulary put on it: a
/// walk says `found`, `reclaimed`, `failed`, not "edit field 3 of the report".
/// Every method takes `&self` and writes into the job under the queue's lock,
/// so the console polling `PurgeStatus` sees a count that moves rather than a
/// job that sits at zero for four minutes and then jumps.
pub type Progress = Reporter<Purge>;

impl Progress {
    /// **How many stores the walk found.** Called once, when the walk has
    /// finished counting and before it starts unlinking.
    pub fn found(&self, stores: u32) {
        self.report(|job| job.stores_found = stores);
    }

    /// One store is gone, and this is what it was holding.
    pub fn reclaimed(&self, bytes: u64) {
        self.report(|job| {
            job.stores_purged = job.stores_purged.saturating_add(1);
            job.bytes_reclaimed = job.bytes_reclaimed.saturating_add(bytes);
        });
    }

    /// One store could NOT be reclaimed, and this is why.
    ///
    /// Recorded and the walk continues. Stopping at the first failure would
    /// leave the rest of an account that is being closed on the disk because
    /// one directory had bad permissions — and the operator would have to run
    /// the purge again anyway.
    pub fn failed(&self, note: String) {
        self.report(|job| job.failures.push(note));
    }

    /// The principal itself was removed.
    pub fn principal_removed(&self) {
        self.report(|job| job.principal_removed = true);
    }

    /// The account this purge is emptying, for a runner that wants it back.
    pub fn account(&self) -> &str {
        self.key()
    }
}

/// **The purge kind.** What a purge is, and — the part a gc has no equivalent
/// of — what a purge has to give back when it ends, however it ends.
pub struct Purge {
    run: PurgeRunner,
    release: GateRelease,
}

impl JobKind for Purge {
    const NAME: &'static str = "purge";

    type Order = PurgeOrder;
    type Report = PurgeReport;

    /// Keyed by ACCOUNT, because both questions the queue answers are per
    /// account: "is one already running for this account" on enqueue, and "how
    /// is mine going" on status.
    fn key(order: &PurgeOrder) -> &str {
        &order.account
    }

    fn run(&self, order: &PurgeOrder, progress: &Progress) -> Result<(), String> {
        (self.run)(order, progress)
    }

    /// **The release, on every path.**
    ///
    /// Unconditional, and the runner reaches it after a success, after a
    /// failure and after a panic — that is what [`JobKind::settle`] is for and
    /// it is the whole reason the release is the queue's job and not the
    /// runner's: a runner that returned early, failed, or blew up would each be
    /// a different missed unlock, and "release it in the runner's error path
    /// too" is a rule that holds until the next early return somebody adds.
    fn settle(
        &self,
        order: &PurgeOrder,
        outcome: Result<(), String>,
        progress: &Progress,
    ) -> (JobState, Option<String>) {
        let released = (self.release)(&order.account, &order.job_id);
        // What the closure ACTUALLY answered, never optimism: an `Err` is a
        // lockout still on, and a job claiming otherwise would send an operator
        // away from the one account that needs them.
        let lifted = matches!(released, Ok(true));
        progress.report(|job| job.gate_released = lifted);
        match (outcome, released) {
            // The ordinary end: emptied, and unlocked (or left locked by
            // whoever else holds the row, which is `Ok(false)` and is not this
            // job's failure).
            (Ok(()), Ok(_)) => (JobState::Done, None),
            // **Emptied but still locked.** The single worst outcome this queue
            // can produce and it must never read as success: the customer's
            // data is gone AND they cannot use what is left of the account.
            (Ok(()), Err(why)) => (
                JobState::Failed,
                Some(format!(
                    "the account was emptied but the lockout could NOT be lifted: {why}. The \
                     account is unusable until the gate is lifted by hand"
                )),
            ),
            (Err(why), Ok(_)) => (JobState::Failed, Some(why)),
            (Err(why), Err(unlock)) => (
                JobState::Failed,
                Some(format!(
                    "{why}; and the lockout could NOT be lifted either: {unlock}. The account is \
                     unusable until the gate is lifted by hand"
                )),
            ),
        }
    }

    /// Reported with `gate_released: false` on purpose: the caller has already
    /// taken a lockout under this id and has to know that nothing is going to
    /// take it off.
    fn shutdown_error(&self) -> String {
        "the purge queue is shutting down; the account is LOCKED and nothing here will unlock \
         it — lift the gate by hand"
            .to_owned()
    }

    /// No worker means the job will never run — and the account is already
    /// locked. Say both, on the job, rather than leave it QUEUED for ever.
    fn spawn_error(&self, err: &dyn std::fmt::Display) -> String {
        format!(
            "the purge worker could not be started: {err}. The account is LOCKED and nothing \
             here will unlock it — lift the gate by hand"
        )
    }
}

/// The queue: one worker, one purge at a time, FIFO, deduplicated by account.
///
/// A named type over [`JobQueue<Purge>`] rather than an alias, so
/// `Accounts.Purge`'s door keeps taking a [`PurgeOrder`] and taking its job id
/// out first, which is the shape the gate demands.
#[derive(Clone)]
pub struct PurgeQueue(JobQueue<Purge>);

impl PurgeQueue {
    /// A queue over `run`, releasing gates through `release`. No thread is
    /// started until something is enqueued.
    pub fn new(run: PurgeRunner, release: GateRelease) -> Self {
        PurgeQueue(JobQueue::new(Purge { run, release }))
    }

    /// **The job id the NEXT enqueue would use**, without enqueuing.
    ///
    /// The gate has to be taken before the job exists — that is what makes the
    /// lockout a fact by the time `Accounts.Purge` returns — and the gate's
    /// `holder` has to be the job id, or the release could not tell this job's
    /// lockout from anybody else's. So the id is minted here, handed to the
    /// gate, and handed back in on [`PurgeQueue::enqueue`] inside the order.
    ///
    /// Minting from the same counter as `enqueue`, so an id handed out here and
    /// never used simply leaves a gap. A gap in a job-id sequence costs
    /// nothing; two jobs sharing an id would make the release ambiguous.
    pub fn next_job_id(&self) -> String {
        self.0.next_job_id()
    }

    /// **Ask for an account to be emptied.**
    ///
    /// Returns the job now responsible for the account: a new one, or the one
    /// that was already queued or running — in which case this order's
    /// `job_id` is *not* used, and the caller must not assume its gate is the
    /// one that will be released. `Accounts.Purge` handles that by reading the
    /// gate back out of the table rather than trusting the one it wrote.
    pub fn enqueue(&self, order: PurgeOrder) -> PurgeJob {
        let job_id = order.job_id.clone();
        self.0.enqueue_as(order, job_id)
    }

    /// The job the queue is holding for `account`, if it still has one.
    pub fn status(&self, account: &str) -> Option<PurgeJob> {
        self.0.status(account)
    }

    /// The highest number of purges that have ever run together. **Must be 1.**
    pub fn peak_in_flight(&self) -> usize {
        self.0.peak_in_flight()
    }

    /// How many times the runner has been called.
    pub fn runs(&self) -> u64 {
        self.0.runs()
    }

    /// Stop the worker after the job it is on.
    pub fn stop(&self) {
        self.0.stop()
    }
}