crates/gunnar-server/src/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()
}
}