gunnar_server/

gc_queue.rs

//! **One gc at a time, for the whole box.**
//!
//! `Repos.Gc` exists so a console button can maintain a repository. The moment
//! it exists, ten repository owners can press ten buttons inside one second,
//! and a gc is not cheap: it seals the live logs, compacts the archive, and
//! unlinks the entire lazily-exploded read cache — hundreds of thousands of
//! files on a large repository, all of it I/O. Ten of those at once is not ten
//! times the work, it is a box that has stopped answering.
//!
//! # Why the server and not the caller
//!
//! A console can serialise **itself**. It cannot serialise the CLI, a second
//! console, a script, or the next client anyone writes — and every one of those
//! holds a credential that reaches the same disk. A rule that only one
//! participant obeys is not an invariant; it is a convention that holds until
//! the first person who did not read it. The server is the only place that sees
//! every request, so it is the only place the rule can actually live.
//!
//! # The two properties, and why they are tested rather than argued
//!
//! - **Never two at once.** A single jera lane takes jobs off one queue, one at a time.
//!   The queue counts the runs that were *in flight together* and remembers the
//!   peak, so `tests/gc_queue.rs` asserts on an observed maximum rather than on
//!   a sleep that happened not to overlap. Seen red against a
//!   thread-per-request queue, which reported a peak of 10 against ten requests.
//! - **A double press is one job.** A request naming a repository that is
//!   already queued or running returns the job that is already there. Impatient
//!   double-clicking is the normal case for a button, not the exception, and a
//!   queue that took it literally would run the expensive thing twice for no
//!   reason. Asserted on the number of times the runner was actually CALLED,
//!   which a queue that merely deduplicated its bookkeeping could not fake.
//!
//! Both properties now live in [`crate::job_queue`] and are asserted here, on
//! `GcQueue`, unchanged: a property proved about the runner is a property of
//! every kind on it, but `tests/gc_queue.rs` is what says it holds for **this**
//! door, and that file did not move.
//!
//! # What a finished job is for
//!
//! The worker does not drop a job when it finishes it. A console polls, and a
//! poller that was between requests at the instant the gc completed would
//! otherwise get "no such job" — indistinguishable from "you never asked". So a
//! completed job is kept for [`FINISHED_RETENTION`], and beyond that a bounded
//! number of the most recent are kept anyway, so a busy server forgets by age
//! and a quiet one does not forget at all.
//!
//! # What this module is, and is not
//!
//! It is not the maintenance. That is [`crate::admin::gc_one`], which the CLI
//! verb also calls, and this module reaches it through a runner handed in at
//! construction (one implementation: one writer, and a queue that could be tested without a
//! disk under it). The runner is the only thing here that knows what a gc is.
//!
//! It is not the queue either, any more. The jera lane, the FIFO, the
//! dedup, the retention and the instruments are [`crate::job_queue`], shared
//! with the purge queue and with anything added later; what is left here is
//! **the gc's answers** to that runner's questions — what a gc's subject is (a
//! repository path), what a gc produces (bytes), and what a gc is (the runner).
//! One thing a gc gained by moving: the worker now catches a panicking gc and
//! reports it as a FAILED job instead of dying with it, which the purge queue
//! had paid for separately and the gc queue never had.

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

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

/// How long a finished job stays readable. Ten minutes is far longer than any
/// console poll interval and far shorter than a session, so a button that
/// completed is still reporting its bytes when whoever pressed it looks back,
/// and a server that has run thousands is not holding all of them.
pub const FINISHED_RETENTION: Duration = <Gc as JobKind>::FINISHED_RETENTION;

/// Finished jobs kept regardless of age, most recent first.
///
/// The floor under [`FINISHED_RETENTION`]: a server that has done nothing for
/// an hour should still be able to say what its last gc freed. Bounded, because
/// "keep everything" is how a long-lived process grows a leak with a good
/// excuse.
pub const FINISHED_KEPT: usize = <Gc as JobKind>::FINISHED_KEPT;

/// Where a job is. Mirrors `gunnar.v1.GcState`; the mapping to the wire lives
/// in `gunnar-grpc` so this crate does not depend on the protos.
///
/// The four words are [`JobState`]'s, shared with every other kind of job, so
/// `GcStatus` and `PurgeStatus` cannot drift apart about what `Done` means.
pub type GcState = JobState;

/// One maintenance run, queued or completed.
///
/// A [`Job`] of kind [`Gc`]: the queue's own fields (`job_id`, `state`,
/// `queue_position`, `error`, the timestamps, `progress`) plus a gc's
/// [`GcReport`], which dereferences flat — `job.bytes_reclaimed` reads the
/// report and `job.state` the queue, and a caller does not have to know which
/// half a field is in. The repository is `job.key`, exactly as the queue keys
/// on it.
pub type GcJob = Job<Gc>;

/// What a gc produced.
#[derive(Debug, Clone, Default)]
pub struct GcReport {
    /// Bytes that left the filesystem. Only meaningful when
    /// [`JobState::Done`]; 0 on a gc that failed, which reclaimed nothing.
    pub bytes_reclaimed: u64,
}

/// What a gc actually is, handed in so this module does not have to know.
///
/// Takes `<namespace>/<name>` and returns the bytes that left the filesystem.
/// The real one resolves the catalog row and calls [`crate::admin::gc_one`] —
/// the same function `gunnar gc` runs.
pub type GcRunner = Arc<dyn Fn(&str) -> Result<u64, String> + Send + Sync>;

/// **The gc kind.** Everything the shared runner needs to know about a gc, and
/// nothing else.
pub struct Gc {
    run: GcRunner,
}

impl JobKind for Gc {
    const NAME: &'static str = "gc";

    /// The repository, `<namespace>/<name>`. A gc has nothing else to be told:
    /// there are no options, so the order and the subject are the same string.
    type Order = String;
    type Report = GcReport;

    fn key(order: &String) -> &str {
        order
    }

    /// Call the runner and record what it freed.
    ///
    /// A gc reports nothing on the way through — no [`Reporter::progress`] —
    /// because as far as a caller is concerned it is one indivisible act, and a
    /// fabricated `0/1` would be a progress bar that lies. The bytes are known
    /// only at the end, which is when they are written.
    fn run(&self, path: &String, report: &Reporter<Gc>) -> Result<(), String> {
        let bytes = (self.run)(path)?;
        report.report(|gc| gc.bytes_reclaimed = bytes);
        Ok(())
    }
}

/// The queue: one worker, one job at a time, FIFO, deduplicated by repository.
///
/// A named type over [`JobQueue<Gc>`] rather than an alias, so `Repos.Gc`'s
/// door keeps speaking in repositories — `enqueue("acct/repo")` — instead of
/// making every caller build an order to say one string.
#[derive(Clone)]
pub struct GcQueue(JobQueue<Gc>);

impl GcQueue {
    /// A queue over `run`. No thread is started until something is enqueued.
    pub fn new(run: GcRunner) -> Self {
        GcQueue(JobQueue::new(Gc { run }))
    }

    /// **Ask for `path` to be maintained, and return without maintaining it.**
    ///
    /// Returns the job that is now responsible for the repository: a new one,
    /// or the one that was already queued or running. Never enqueues a second
    /// job for a repository that has one live.
    pub fn enqueue(&self, path: &str) -> GcJob {
        self.0.enqueue(path.to_owned())
    }

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

    /// The highest number of gcs that have ever been running together.
    ///
    /// **Must be 1 after any amount of traffic** — that is the invariant this
    /// module exists for, and this is how a test sees it rather than infers it.
    /// 0 before anything has run.
    pub fn peak_in_flight(&self) -> usize {
        self.0.peak_in_flight()
    }

    /// How many times the runner has been called — i.e. how many gcs actually
    /// happened. The applied output the dedup gate asserts on.
    pub fn runs(&self) -> u64 {
        self.0.runs()
    }

    /// Stop the worker after the job it is on. For tests and shutdown; a queue
    /// that is dropped without this simply leaves its thread parked on the
    /// condvar until the process ends.
    pub fn stop(&self) {
        self.0.stop()
    }
}