gunnar_server/

mirror_queue.rs

//! **One mirror sync at a time, for the whole box.**
//!
//! `Mirror.Sync` exists so a console button can pull a repository off another
//! forge. The moment it exists, ten owners can press ten buttons inside one
//! second, and a sync is the most expensive button on the box: it is a whole
//! `git clone` over somebody else's network, into this box's storage engine,
//! with the pack streaming through the whole time.
//!
//! # Why this is the third queue and not a third caller of one queue
//!
//! [`crate::job_queue`] argues it in full and this is the mirror's half of that
//! argument: **the line a job waits in is part of what the job costs.** A gc
//! waits holding nothing. A purge waits holding a total lockout on an account.
//! A mirror waits holding a repository that is *not yet usable* — somebody
//! asked for a copy of an upstream and until the sync runs there is nothing
//! there to clone. Putting it behind a compaction that is only holding the disk
//! would be the same mistake in a third colour, so it gets its own worker and
//! its own line.
//!
//! # What is here and what is not
//!
//! Not the fetch. That is [`crate::mirror_fetch`], reached through a runner
//! handed in at construction — the same shape [`crate::gc_queue`] uses, and for
//! the same two reasons: one writer for what a sync *is* (one implementation), and a queue
//! that can be tested with no network and no store under it.
//!
//! What is here is the mirror's answers to [`JobKind`]'s questions: a sync's
//! subject is a repository path, a sync's order carries the upstream as well
//! (so the worker does not have to re-read a registration that may have been
//! edited while the job waited), and a sync's report is objects, bytes and the
//! references that moved.
//!
//! # `progress`, and the honest shape of it
//!
//! A gc and a purge report none — each is one indivisible act as far as a
//! caller is concerned, and a fabricated `0/1` would be a progress bar that
//! lies. A mirror cannot do without one: a fetch of a large repository is
//! minutes of network with a console watching.
//!
//! **Both halves are real and live, and they are published from opposite sides
//! of the storage seam.** `gunnar-wire`'s `FetchProgress` publishes
//! `objects_total` the instant the packfile header arrives — twelve bytes, read
//! before an object is stored — so `(0, N)` is known at the *start* of the
//! transfer. The numerator is the ingest's, because the only code that can
//! count objects without inflating them a second time is the code that is
//! already splitting the pack: gunnar has exactly one pack decoder (one implementation), and
//! `fetch_pack`'s module doc refuses to become the second one for this exact
//! reason.
//!
//! # 🔴 What this used to say, and what closed it
//!
//! Until 2026-09-08 this paragraph read *"the numerator has exactly two
//! values"*, and the pair really did go `(0, N) → (N, N)` — one publish at the
//! header and one at the finish, with nothing in between. The reason given was
//! that znippy "takes a pack in one call and reports what it indexed when it
//! returns", which was true of the *seam* and never true of the engine: znippy
//! splits the pack entry by entry in `pack_walk`, and has known the running
//! count all along. It simply had nobody to tell.
//!
//! It has now. [`GitOps::put_pack_observed`](git_storage_trait::GitOps::put_pack_observed)
//! is one defaulted method on the storage contract — no existing implementer
//! changed, and an engine that has no entry loop keeps publishing nothing and
//! says so by name through `publishes_pack_progress()`. znippy overrides it and
//! publishes **once per object** from the walk it was already doing; nothing
//! new is decompressed and no second decoder exists.
//! [`crate::mirror_fetch`] passes the sink straight through to the same
//! `Reporter` the wire's two figures go to.
//!
//! So the pair now steps: `(0, N)`, then a value per object as the engine takes
//! them in, ending at `(N, N)` — and `fetch_pack`'s header-versus-receipt
//! cross-check at the end is now also the proof that the live count was honest.
//! [`MirrorReport::bytes`] is still the figure that moves while the *network*
//! is the bottleneck, and the object count is the one that moves while
//! *storage* is; between them a console has something live for both halves of a
//! sync, and neither is fabricated.

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

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

/// How long a finished sync stays readable. The shared default: ten minutes is
/// far longer than any console poll interval and far shorter than a session.
pub const FINISHED_RETENTION: Duration = <Mirror as JobKind>::FINISHED_RETENTION;

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

/// Where a sync is. Mirrors `gunnar.v1.MirrorJobState`; 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 the gc and the purge, so
/// three status RPCs cannot drift apart about what `Done` means.
pub type MirrorState = JobState;

/// One sync, queued or completed.
///
/// A [`Job`] of kind [`Mirror`]: the queue's own fields (`job_id`, `state`,
/// `queue_position`, `progress`, `error`, the timestamps) plus a
/// [`MirrorReport`], which dereferences flat. The repository is `job.key`,
/// exactly as the queue keys on it.
pub type MirrorJob = Job<Mirror>;

/// **What a sync was asked for.**
///
/// The upstream travels on the order rather than being re-read by the worker,
/// and that is the rule [`JobKind::Order`] states: a registration can be
/// changed while its job waits, and a job that quietly fetched from a different
/// remote than the one the caller asked about would be a mirror nobody could
/// audit.
#[derive(Debug, Clone)]
pub struct MirrorOrder {
    /// `<namespace>/<name>` — the subject, and therefore the dedup key.
    pub path: String,
    /// Where the upstream is, as the registration held it when `Sync` was
    /// called.
    pub upstream: String,
    /// Delete local mirrored references the upstream no longer advertises.
    pub prune: bool,
}

/// **What one sync produced.**
///
/// The kind's own fields; the runner never reads them. Nothing here is
/// restated from the order — every number is measured by the fetch.
#[derive(Debug, Clone, Default)]
pub struct MirrorReport {
    /// Objects the packfile carried, as both the header declared and the ingest
    /// indexed — `fetch_pack` refuses the transfer if those two disagree, so
    /// there is one number here and not two.
    ///
    /// 0 on a mirror that was already up to date: no want, no pack, no walk
    /// charged to the remote. That is the ordinary answer for a poll.
    pub objects: u64,
    /// Packfile bytes taken off the wire, header and trailer included. **This
    /// is the figure that moves while the job runs.**
    pub bytes: u64,
    /// Every reference this sync moved, as `(name, old, new)` in hex, with an
    /// all-zeroes `new` for a deletion and an all-zeroes `old` for a creation —
    /// the `gunnar.v1.RefUpdate` convention, so the seam does no second
    /// encoding.
    pub updates: Vec<(String, String, String)>,
    /// **Why the sync was refused by the namespace's own plan**, when it was
    /// (2026-10-04) — `MirrorJob.refused_reason` on the wire, so a console
    /// reads a field instead of parsing the sentence in `error`.
    pub refused: Refusal,
}

/// Why a sync was REFUSED by the namespace's plan, as against failing.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum Refusal {
    /// Not refused by the plan: it succeeded, or it failed for another reason.
    #[default]
    None,
    /// The plan refuses writes at all (suspended, lapsed).
    Plan,
    /// The namespace is at its pack room, or the fetch was cut at the ceiling.
    PackCap,
}

/// **What a sync actually is**, handed in so this module does not have to know.
///
/// Takes the order and the job's reporter and either succeeds or returns the
/// sentence a console will show. The real one is
/// [`crate::mirror_fetch::sync_once`].
pub type MirrorRunner =
    Arc<dyn Fn(&MirrorOrder, &Reporter<Mirror>) -> Result<(), String> + Send + Sync>;

/// **The mirror kind.** Everything the shared runner needs to know about a
/// sync, and nothing else.
pub struct Mirror {
    run: MirrorRunner,
}

impl JobKind for Mirror {
    const NAME: &'static str = "mirror";

    type Order = MirrorOrder;
    type Report = MirrorReport;

    /// The repository. A second registration for the same repository replaces
    /// the first, so a repository has at most one upstream and the path is the
    /// whole subject.
    fn key(order: &MirrorOrder) -> &str {
        &order.path
    }

    fn run(&self, order: &MirrorOrder, report: &Reporter<Mirror>) -> Result<(), String> {
        (self.run)(order, report)
    }
}

/// The queue: one worker, one sync at a time, FIFO, deduplicated by repository.
///
/// A named type over [`JobQueue<Mirror>`] rather than an alias, so `Mirror.Sync`'s
/// door keeps speaking in orders instead of making every caller reach for the
/// generic.
#[derive(Clone)]
pub struct MirrorQueue(JobQueue<Mirror>);

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

    /// **Ask for `order.path` to be synced, and return without syncing it.**
    ///
    /// Returns the job now responsible for the repository: a new one, or the
    /// one that was already queued or running. Never enqueues a second sync for
    /// a repository that has one live — a double-clicked button would otherwise
    /// pull the same pack over somebody else's network twice.
    pub fn enqueue(&self, order: MirrorOrder) -> MirrorJob {
        self.0.enqueue(order)
    }

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

    /// The highest number of syncs that have ever been running together.
    ///
    /// **Must be 1 after any amount of traffic.** 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 syncs 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.
    pub fn stop(&self) {
        self.0.stop()
    }
}