gunnar-server 3.0.0

//! **What `Mirror.Sync` actually does**: one fetch from an upstream, into this
//! box's own storage engine, and one compare-and-swap that moves the
//! references.
//!
//! # There is one client, and two ways to reach a remote
//!
//! [`gunnar_wire::fetch_pack`] is the whole protocol — the advertisement, the
//! `ls-refs`, the want/have negotiation, the sideband demultiplex, the packfile
//! streamed into a [`PackIngest`](gunnar_wire::PackIngest). It takes a
//! `&mut dyn BufRead` and a `&mut dyn Write` and cannot tell what is behind
//! them, which is the seam this module fills twice and only twice:
//!
//! | remote | the pair | who built it |
//! |---|---|---|
//! | `ssh://`, `git+ssh://`, `git@host:path` | a russh channel's two halves | `nornir-git`, through `gunnar_client::connect_upload_pack` — the **same dialler** replication pushes over |
//! | `https://`, `http://` | a buffered request/response adapter | [`crate::mirror_http`] |
//!
//! Neither is a second protocol implementation, and that is the point. A bug in
//! negotiation is one bug, in one file, in `gunnar-wire`.
//!
//! # This module writes the references, and `fetch_pack` deliberately does not
//!
//! `fetch_pack`'s module doc states the rule: it returns a **plan**
//! (`FetchReport::updates`, each entry a name, the old value and the new one)
//! and applies nothing, because applying a whole mirror's plan with no gate, no
//! atomicity across the batch and no way for an operator to see it first is not
//! something a protocol crate should be able to do.
//!
//! So the apply is here, and it is **one [`GitOps::put_refs_cas`]** with the
//! plan's own `old` as every expectation. That is not tidiness:
//!
//! * A mirror is not the only writer. `git push` reaches the same repository
//!   over SSH and over smart HTTP, and a fetch that read a tip, spent four
//!   minutes on the network and then wrote unconditionally would **silently
//!   destroy** whatever landed in between. The `old` in the plan is read from
//!   the same `GitOps::refs()` snapshot the `have` list came from, so the CAS
//!   is checking exactly the value the fetch reasoned about.
//! * A CAS failure is a **FAILED job that names the reference**, never a skip.
//!   "Synced" about a repository where one branch quietly did not move is the
//!   answer that makes a mirror untrustworthy, and it reads identically to a
//!   mirror that worked.
//! * Every edit or none. A half-applied plan leaves the repository at a state
//!   no upstream ever had.
//!
//! # …and the batch goes through branch protection first
//!
//! A mirror moves references, and until 2026-09-11 it was the only writer on
//! this box that moved them without asking [`BranchProtectionGate`]. That is
//! not a smaller version of a push: `Mirror.Sync` needs **Write** and
//! `Policy.SetProtection` needs **Admin**, so a principal who may not configure
//! a protection rule could register a mirror against the repository it protects
//! and have the upstream's tips applied over it — a non-fast-forward onto a ref
//! whose rule says `allow_force_push: false`, applied by the one path that
//! never looked. See [`Protection`] for what is decided here and what is
//! refused as undecidable.
//!
//! # The credential is never in this module's output
//!
//! The value is fetched from the sealed secret store, held for the length of
//! one dial, and never formatted. An SSH key is materialised into a
//! `0600` temporary file because `nornir-git` authenticates from a path and
//! this tree does not fork a second dialler to change that (one implementation); the file is
//! unlinked when the fetch returns, on every path including a panic, because
//! [`tempfile::NamedTempFile`] unlinks on `Drop`. A token becomes an
//! `Authorization` header value inside [`crate::mirror_http`] and nothing else.
//! Errors that mention a URL go through [`scrub`] first.

use std::io::{BufRead, Write};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use gix_hash::ObjectId;
use gunnar_core::MirrorCredentialKind;
use gunnar_store::git_storage_trait::{GitOps, RefCas};
use gunnar_wire::fetch_pack::{self, FetchPack, FetchProgress, FetchReport, LocalRefs};

use crate::job_queue::Reporter;
use crate::mirror_queue::{Mirror, MirrorOrder};
use crate::stores::ServerStores;

/// The tracing target a sync's lines carry — the same one every other git
/// transfer on this box uses, so `RUST_LOG=gunnar::git=info` selects a mirror
/// as well as a clone.
const TARGET_GIT: &str = "gunnar::git";

/// **The `known_hosts` a mirror's SSH dial trusts**, under the data root.
///
/// Not `~/.ssh/known_hosts`, and this is the same ruling replication's
/// `REPLICATION_KNOWN_HOSTS` makes for the push direction: a server's identity
/// is the box, not a home directory. The appliance has no home directory at
/// all, so the default would be a file that cannot be written and a
/// trust-on-first-use that silently never remembers — every sync a first
/// contact, which is the same as no host-key checking with extra steps.
///
/// A separate file from replication's, because they are separate rosters: the
/// standbys this box pushes to and the upstreams it pulls from are different
/// sets of machines, and a pin acquired for one is not consent for the other.
pub const MIRROR_KNOWN_HOSTS: &str = "mirror_known_hosts";

/// **The file that makes a mirror's SSH dial refuse an unknown host key**,
/// under the data root beside [`MIRROR_KNOWN_HOSTS`].
///
/// Without it a mirror dials trust-on-first-use: an upstream this box has never
/// seen is recorded and trusted, and only a *changed* key is refused. That is
/// the right default for a developer's laptop and a defensible one for a server
/// being brought up without a provisioning step — it is what replication's own
/// `strict_host_key` defaults to, and its comment already says that on "is the
/// right setting for anything that outlives an afternoon". The mirror side had
/// no way to say on at all: the policy was a literal in [`dial_ssh`].
///
/// A file rather than a field on `MirrorRecord`, for the reason
/// [`MIRROR_KNOWN_HOSTS`] is a file under the data root: the roster of hosts
/// this box will pull from is the box's, not one repository's, and the pin it
/// is checked against is already per-box. Create it once, and every mirror on
/// the box requires a `known_hosts` entry that somebody put there.
pub const MIRROR_STRICT_HOST_KEY: &str = "mirror_strict_host_key";

/// Does this box refuse an unknown upstream host key? See
/// [`MIRROR_STRICT_HOST_KEY`].
///
/// The presence of the file is read at each dial rather than cached, so an
/// operator who creates it does not have to restart the server to mean it.
pub fn strict_host_keys(data_root: &std::path::Path) -> bool {
    data_root.join(MIRROR_STRICT_HOST_KEY).exists()
}

/// **The file that lets this box's mirrors dial an address that is not
/// publicly routable**, under the data root beside [`MIRROR_KNOWN_HOSTS`].
///
/// A file and not an environment variable, for the reason `wg.rs` writes out at
/// length: an env var is process-global, so a rule that can only be exercised
/// by `set_var` is a rule whose tests flake the moment `cargo test` runs them
/// on two threads. A file under the data root is per-box, survives a restart,
/// is visible to an operator looking at the box, and each test gets its own.
///
/// It exists because refusing every private destination outright would be a
/// product regression, not a fix: mirroring from a forge on the operator's own
/// LAN is an ordinary thing for an appliance to do. What must not be ordinary
/// is a **low-privileged** caller getting it for free — see
/// [`check_destination`].
pub const MIRROR_ALLOW_PRIVATE: &str = "mirror_allow_private_destinations";

/// **The credential a mirror dials with**, in the one shape it exists in
/// memory.
///
/// Deliberately not `Debug`, `Clone` or `Serialize`. The compiler is the only
/// mechanism that reliably stops a secret being formatted into a log line by
/// somebody adding a `{:?}` three months from now, so the derive that would
/// allow it is simply absent.
pub enum Credential {
    /// No credential. A public upstream needs none, and a mirror of one works.
    None,
    /// An OpenSSH private key, for an `ssh://` upstream.
    SshKey(Vec<u8>),
    /// A token, for an `https://` upstream. `username` empty means `Bearer`.
    Token { username: String, secret: String },
}

impl Credential {
    fn kind(&self) -> MirrorCredentialKind {
        match self {
            Credential::None => MirrorCredentialKind::None,
            Credential::SshKey(_) => MirrorCredentialKind::SshKey,
            Credential::Token { .. } => MirrorCredentialKind::Token,
        }
    }
}

/// **Which transport a remote URL names.**
///
/// Decided by the URL and by nothing else — not by which credential happens to
/// be stored, and not by trying one and falling back. A fallback would turn "I
/// gave you the wrong kind of key" into "your upstream is down", and those go
/// to two different people.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Wire {
    /// `ssh://`, `git+ssh://`, or the scp-like `git@host:ns/repo.git`.
    Ssh,
    /// `https://` or `http://`.
    Http,
}

impl Wire {
    /// The word an operator sees in a refusal.
    fn name(self) -> &'static str {
        match self {
            Wire::Ssh => "ssh",
            Wire::Http => "smart http",
        }
    }
}

/// **What transport `url` names, or a refusal that says what gunnar can dial.**
///
/// The scp-like form is recognised the way git recognises it: a `:` before any
/// `/`, with no scheme. A URL that is neither is refused rather than guessed
/// at — a mirror registered against a path on the local disk would otherwise
/// look like it was configured and then fail on the first sync with a
/// connection error.
pub fn classify(url: &str) -> Result<Wire, String> {
    let url = url.trim();
    if url.starts_with("ssh://") || url.starts_with("git+ssh://") {
        return Ok(Wire::Ssh);
    }
    if url.starts_with("https://") || url.starts_with("http://") {
        return Ok(Wire::Http);
    }
    if url.contains("://") {
        return Err(format!(
            "{} names a scheme gunnar cannot mirror from. The two transports are `ssh://` \
             and `https://`",
            scrub(url)
        ));
    }
    // `git@host:ns/repo.git` — a colon before any slash, and a host in front
    // of it.
    match (url.find(':'), url.find('/')) {
        (Some(colon), slash) if colon > 0 && slash.is_none_or(|slash| colon < slash) => {
            Ok(Wire::Ssh)
        }
        _ => Err(format!(
            "{url:?} is not a remote gunnar can mirror from. Write it as \
             `ssh://git@host/ns/repo.git`, `git@host:ns/repo.git` or `https://host/ns/repo.git`"
        )),
    }
}

// ── where a mirror may dial ──────────────────────────────────────────────────

/// **The host and port a mirror URL names**, in the three shapes
/// [`classify`] admits.
///
/// Deliberately not a URL parser: the only question asked of this string is
/// *which machine would be dialled*, and the answer is the authority. A
/// userinfo half is dropped — `git@host` names `host` — and an IPv6 literal
/// keeps its brackets off so it can be parsed as an address.
pub(crate) fn authority(url: &str) -> Result<(String, u16), String> {
    let url = url.trim();
    let (default_port, rest) = if let Some(rest) = url.strip_prefix("ssh://") {
        (22, rest)
    } else if let Some(rest) = url.strip_prefix("git+ssh://") {
        (22, rest)
    } else if let Some(rest) = url.strip_prefix("https://") {
        (443, rest)
    } else if let Some(rest) = url.strip_prefix("http://") {
        (80, rest)
    } else {
        // The scp-like `git@host:ns/repo.git`. `classify` has already decided
        // this shape is ssh; the colon here separates a PATH, never a port.
        // The userinfo goes first so a `@` inside it cannot be read as a host,
        // and a bracketed literal is taken whole — `[::1]:path` is all colons
        // and splitting on the first one would yield `[`.
        let rest = url.rsplit('@').next().unwrap_or_default();
        let host = match rest.strip_prefix('[') {
            Some(inside) => inside.split(']').next().unwrap_or_default(),
            None => rest.split(':').next().unwrap_or_default(),
        };
        return non_empty(host, url).map(|host| (host, 22));
    };
    let authority = rest.split(['/', '?', '#']).next().unwrap_or_default();
    let authority = authority.rsplit('@').next().unwrap_or_default();
    // `[::1]:9418` and `[::1]`, before the bare-host split, because an IPv6
    // literal is all colons.
    if let Some(rest) = authority.strip_prefix('[') {
        let Some((host, tail)) = rest.split_once(']') else {
            return Err(format!("{} names an unterminated IPv6 literal", scrub(url)));
        };
        let port = match tail.strip_prefix(':') {
            None => default_port,
            Some(digits) => digits
                .parse()
                .map_err(|_| format!("{} names a port that is not a number", scrub(url)))?,
        };
        return non_empty(host, url).map(|host| (host, port));
    }
    match authority.split_once(':') {
        None => non_empty(authority, url).map(|host| (host, default_port)),
        Some((host, digits)) => {
            let port = digits
                .parse()
                .map_err(|_| format!("{} names a port that is not a number", scrub(url)))?;
            non_empty(host, url).map(|host| (host, port))
        }
    }
}

fn non_empty(host: &str, url: &str) -> Result<String, String> {
    if host.is_empty() {
        return Err(format!("{} names no host to dial", scrub(url)));
    }
    Ok(host.to_owned())
}

/// **Refuse a destination this server can reach and its caller cannot** — the
/// pure half, with the addresses handed in.
///
/// `Mirror.Add` needs Admin and `Mirror.Sync` needs Write, but *every enrolled
/// principal is intrinsically Admin in its own namespace*
/// (`AuthTable::owns_account`), so `Mirror.Add{create_repo}` against
/// `alice/anything` is available to anybody with a key. Before this check the
/// only thing standing between that caller and an arbitrary request from the
/// server's own network position was `remote_url.trim().is_empty()`: the cloud
/// metadata endpoint at `169.254.169.254`, a control plane on `127.0.0.1`, a
/// printer on `10.0.0.0/8` — all dialled by the appliance, with the appliance's
/// routes, and the outcome partly readable back through `Mirror.Status`.
///
/// **Every** resolved address has to be routable, not merely the first: a name
/// that resolves to one public address and one loopback address is the
/// rebinding shape, and taking the best of the two is how a check like this
/// gets to be decorative.
///
/// The ranges are written out rather than taken from `IpAddr::is_global`, which
/// is unstable, and each line is the reason it is there.
pub(crate) fn address_verdict(host: &str, addrs: &[std::net::IpAddr]) -> Result<(), String> {
    if addrs.is_empty() {
        return Err(format!("{host} resolves to no address at all"));
    }
    for addr in addrs {
        if let Some(why) = not_routable(*addr) {
            return Err(format!(
                "{host} resolves to {addr}, which is {why}. A mirror dials from the server, so a \
                 destination the caller could not reach on their own is the server's network \
                 position being borrowed. Create `{MIRROR_ALLOW_PRIVATE}` under this box's data \
                 root to allow it deliberately."
            ));
        }
    }
    Ok(())
}

/// Why `addr` is not somewhere an arbitrary caller should be able to point this
/// server, or `None` when it is an ordinary public address.
fn not_routable(addr: std::net::IpAddr) -> Option<&'static str> {
    use std::net::IpAddr;
    match addr {
        IpAddr::V4(v4) => {
            let [a, b, c, _] = v4.octets();
            match () {
                _ if v4.is_unspecified() => Some("the unspecified address"),
                _ if v4.is_loopback() => Some("loopback — this server itself"),
                _ if v4.is_private() => Some("a private network (RFC 1918)"),
                _ if v4.is_link_local() => {
                    Some("link-local (RFC 3927), the range cloud metadata services answer on")
                }
                _ if v4.is_broadcast() => Some("the broadcast address"),
                _ if v4.is_multicast() => Some("multicast"),
                _ if v4.is_documentation() => Some("a documentation range (RFC 5737)"),
                // 100.64.0.0/10, carrier-grade NAT: the provider's own network,
                // not the caller's.
                _ if a == 100 && (64..128).contains(&b) => Some("carrier-grade NAT (RFC 6598)"),
                // 192.0.0.0/24, IETF protocol assignments.
                _ if a == 192 && b == 0 && c == 0 => Some("an IETF protocol assignment"),
                // 198.18.0.0/15, benchmarking.
                _ if a == 198 && (b == 18 || b == 19) => Some("a benchmarking range (RFC 2544)"),
                // 240.0.0.0/4, reserved.
                _ if a >= 240 => Some("a reserved range"),
                _ => None,
            }
        }
        // A v4 address written the long way is the same address. Checked
        // first, or `::ffff:127.0.0.1` walks straight past every arm below it.
        IpAddr::V6(v6) if v6.to_ipv4_mapped().is_some() => {
            not_routable(IpAddr::V4(v6.to_ipv4_mapped().expect("just matched")))
        }
        IpAddr::V6(v6) => {
            let segments = v6.segments();
            match () {
                _ if v6.is_unspecified() => Some("the unspecified address"),
                _ if v6.is_loopback() => Some("loopback — this server itself"),
                _ if v6.is_multicast() => Some("multicast"),
                // fc00::/7, unique local.
                _ if segments[0] & 0xfe00 == 0xfc00 => Some("a unique-local address (RFC 4193)"),
                // fe80::/10, link-local.
                _ if segments[0] & 0xffc0 == 0xfe80 => Some("link-local (RFC 4291)"),
                // 2001:db8::/32, documentation.
                _ if segments[0] == 0x2001 && segments[1] == 0x0db8 => {
                    Some("a documentation range (RFC 3849)")
                }
                _ => None,
            }
        }
    }
}

/// [`address_verdict`] with the resolution and the opt-out file read — the
/// edge, and the only part of this that touches the world.
///
/// # The window this does not close, said rather than glossed
///
/// The name is resolved here and again by `ureq` or by the ssh dialler a moment
/// later, so a resolver that answers differently the second time (DNS
/// rebinding) is not stopped by this. Closing that needs the dial pinned to the
/// address that was checked, which is a change inside `ureq`'s and
/// `nornir-git`'s connect paths rather than in this file. What this does close
/// is the whole of the *static* case — a URL that simply names an internal
/// address — which is what `Mirror.Add` accepted unchecked.
pub fn check_destination(url: &str, data_root: &std::path::Path) -> Result<(), String> {
    use std::net::ToSocketAddrs as _;
    if data_root.join(MIRROR_ALLOW_PRIVATE).exists() {
        return Ok(());
    }
    let (host, port) = authority(url)?;
    // A literal needs no resolver, and asking one for it would let a DNS server
    // answer a question that has no question in it.
    if let Ok(addr) = host.parse::<std::net::IpAddr>() {
        return address_verdict(&host, &[addr]);
    }
    let addrs: Vec<std::net::IpAddr> = (host.as_str(), port)
        .to_socket_addrs()
        .map_err(|err| format!("{host} could not be resolved: {err}"))?
        .map(|socket| socket.ip())
        .collect();
    address_verdict(&host, &addrs)
}

/// **Refuse a credential that cannot be offered over the transport the URL
/// names — by name, before anything is dialled.**
///
/// The failure this prevents is the quiet one. An SSH key against an `https://`
/// remote is not "authentication failed": there is nowhere in an HTTP request
/// to put a private key, so the request goes out anonymous and the upstream
/// answers 401 — and the operator, who can see that a credential IS stored,
/// concludes the key is wrong and rotates it. Forever. The same in reverse: a
/// token has no place in an SSH publickey exchange, so the dial falls through
/// to whatever else is offerable and fails on authentication with no mention of
/// the token at all.
///
/// A stored credential of no kind is not a mismatch — a public upstream mirrors
/// fine with nothing, and that path must keep working.
pub fn check_credential(wire: Wire, credential: &Credential) -> Result<(), String> {
    match (wire, credential) {
        (_, Credential::None) => Ok(()),
        (Wire::Ssh, Credential::SshKey(_)) => Ok(()),
        (Wire::Http, Credential::Token { .. }) => Ok(()),
        (wire, held) => Err(format!(
            "this mirror's remote is dialled over {}, and the credential stored for it is {}. \
             {} cannot carry it, so the fetch would go out unauthenticated and be refused as \
             though no credential were set. Store {} instead, with `Mirror.SetCredential`.",
            wire.name(),
            held.kind().name(),
            wire.name(),
            match wire {
                Wire::Ssh => "an ssh private key",
                Wire::Http => "a token",
            }
        )),
    }
}

// ── the gate ─────────────────────────────────────────────────────────────────

/// **The repository's branch protection, over the batch a mirror is about to
/// apply.**
///
/// One implementation: this is `gunnar_policy::BranchProtectionGate`, the same
/// object `receive_pack`'s [`crate::transport::PolicyGate`] evaluates, reading
/// the same document through the same [`crate::transport::load_policy`]. A
/// second copy of "what does `allow_force_push` mean" is precisely the pair
/// that drifts, and the three independent ways past branch protection this
/// repository has had are all the same shape: a second writer that does what
/// the gated writer does, without the gate.
///
/// # What a mirror can decide, and the one thing it cannot
///
/// `allow_create`, `allow_delete` and `allow_force_push` are decidable from the
/// batch and the objects, which are already in the store when [`apply`] runs.
/// `require_signature` is decidable too — but only against the set of objects
/// the fetch introduced, so a repository carrying any protection rule makes the
/// ingest ask for that set ([`crate::storage::ObjectSet::Required`]) instead of
/// paying nothing for it, exactly as a push does when a gate is live.
///
/// `allowed_principals` is the one that cannot be decided here, and it is
/// refused rather than skipped. The caller's identity does not cross the
/// control-plane seam on this verb — `mirror_sync(&self, path)` carries a
/// repository and nothing about who asked — so there is no principal to compare
/// against the list. "I could not tell" is not "it is fine", so a rule that
/// names its principals refuses the whole sync by name, and the operator either
/// drops the rule or stops mirroring onto the refs it covers.
struct Protection {
    gate: gunnar_policy::BranchProtectionGate,
    keys: gunnar_policy::StaticKeyDirectory,
    repo: gunnar_policy::RepoId,
}

impl Protection {
    /// The rules covering `repo`, or `None` when it has none configured.
    ///
    /// Read **before** the fetch, because whether any rule exists is what
    /// decides if the ingest pays for an object-set enumeration — the same
    /// ordering, for the same reason, as `receive_pack`'s.
    ///
    /// `Err` is a policy document that could not be read, and it REFUSES the
    /// sync — the same fail-closed rule the push path has (GUN-007). A mirror
    /// that synced with its gates disabled because the policy did not parse
    /// would be the bypass GUN-032 closed, back through the side door.
    fn open(
        stores: &ServerStores,
        row: &gunnar_core::Repo,
        git_dir: &std::path::Path,
    ) -> std::result::Result<Option<Self>, String> {
        let config = crate::transport::load_policy(
            stores,
            row,
            &git_dir.join(crate::transport::POLICY_FILE),
        )
        .map_err(|why| format!("refusing to mirror into {}: {why}", row.key()))?;
        if config.protection.is_empty() {
            return Ok(None);
        }
        Ok(Some(Protection {
            gate: gunnar_policy::BranchProtectionGate::new(config.protection.clone()),
            keys: crate::transport::key_directory(stores),
            repo: gunnar_policy::RepoId::new(&row.namespace, &row.name),
        }))
    }

    /// **Refuse the batch, or let it through.** `Err` is the refusal a FAILED
    /// job carries, and nothing has been written when it is returned.
    ///
    /// `new_objects` is the fetch's own receipt. An ingest that did not
    /// enumerate hands back the reason rather than an empty set, and an empty
    /// set here would be a silent always-allow for `require_signature` — so the
    /// `Err` arm refuses, quoting why.
    fn check(
        &self,
        git: &dyn gunnar_store::git_storage_trait::GitServe,
        updates: &[gunnar_policy::RefUpdate],
        new_objects: std::result::Result<&[ObjectId], &str>,
    ) -> Result<(), String> {
        use gunnar_policy::Gate as _;

        // The field with no answer on this path, refused first and by name.
        for update in updates {
            for rule in self.gate.rules() {
                if !rule.pattern.matches_ref(&update.name) {
                    continue;
                }
                if rule.allowed_principals.is_some() {
                    return Err(format!(
                        "{} is protected by {}, which names the principals that may move it. A \
                         mirror sync carries no principal — the control-plane verb takes a \
                         repository and not a caller — so that rule cannot be satisfied and this \
                         sync is refused rather than applied unchecked. Nothing was moved.",
                        update.name, rule.pattern
                    ));
                }
                break;
            }
        }

        let new_objects = new_objects.map_err(|reason| {
            format!(
                "this repository has branch protection and the fetched pack's object set could \
                 not be enumerated ({reason}), so the gate could not decide. Nothing was moved."
            )
        })?;
        // Never consulted: every rule that could read it has already refused
        // above. Named rather than left as a plausible-looking value.
        let principal = gunnar_policy::Principal::machine("mirror");
        let ctx = gunnar_policy::PushContext::new(
            &principal,
            &self.repo,
            updates,
            new_objects,
            git,
            &self.keys,
        );
        match self.gate.check(&ctx) {
            Ok(gunnar_policy::Verdict::Allow) => Ok(()),
            Ok(gunnar_policy::Verdict::Reject(rejection)) => Err(format!(
                "branch protection refused this mirror's reference batch: {}. Nothing was moved.",
                rejection.reason
            )),
            // A gate that could not decide refuses; it does not allow.
            Err(err) => Err(format!(
                "branch protection could not decide this mirror's reference batch: {err}. \
                 Nothing was moved."
            )),
        }
    }
}

// ── the local half ───────────────────────────────────────────────────────────

/// **What this repository already holds**, as `fetch_pack`'s have-source.
///
/// One `GitOps::refs()` call, one snapshot, taken inside `FetchPack::fetch` so
/// the `have` list and every `old` in the plan come from the same read — a
/// reference that moves under the fetch cannot make the plan promise an old
/// value the store never held.
///
/// `HEAD` and the pseudo-refs are not rows in `refs()` at all, and a purely
/// symbolic row under `refs/` carries no oid and is skipped: it is not a thing
/// to offer as a `have`, and `fetch_pack`'s own doc says an unborn ref should
/// simply be absent.
struct StoreRefs<'a> {
    git: &'a dyn gunnar_store::git_storage_trait::GitServe,
    /// Objects this store holds that no reference names yet — an interrupted
    /// import's ([`already_here`]). Offered as `have`s under names the plan
    /// never matches, so they change what is sent and nothing that is written.
    extra_haves: Vec<ObjectId>,
}

impl LocalRefs for StoreRefs<'_> {
    fn tips(&self) -> gunnar_wire::Result<Vec<(String, ObjectId)>> {
        let rows = GitOps::refs(self.git).map_err(|err| {
            gunnar_wire::Error::Store(gunnar_store::Error::backend(format!(
                "reading this repository's refs to offer them as haves: {err:#}"
            )))
        })?;
        let mut out = Vec::with_capacity(rows.len());
        for row in rows {
            if row.name == "HEAD" {
                continue;
            }
            let Some(raw) = row.oid.as_deref() else {
                continue;
            };
            let Ok(oid) = gix_hash::oid::try_from_bytes(raw) else {
                return Err(gunnar_wire::Error::Store(gunnar_store::Error::backend(
                    format!("{} holds an oid this store cannot decode", row.name),
                )));
            };
            out.push((row.name, oid.to_owned()));
        }
        for (n, oid) in self.extra_haves.iter().enumerate() {
            out.push((format!("refs/gunnar/resume/{n}"), *oid));
        }
        Ok(out)
    }
}

/// **The queue's progress field, written from the wire and from the walk.**
///
/// Three publishers, one job record, and each of the three is the only place on
/// the system that knows its own number:
///
/// | published by | what | when |
/// |---|---|---|
/// | [`FetchProgress::objects_total`] | the denominator | once, the instant the packfile header is read |
/// | [`FetchProgress::pack_bytes`] | the byte count | as the stream is consumed, while the network is the bottleneck |
/// | [`PackProgress::indexed`] | the numerator | once per object, as the engine's pack walk splits them |
///
/// # Why the numerator arrives from the other side of the seam
///
/// `gunnar-wire` reads bytes 8..12 of the header and can publish `(0, N)`
/// before an object is stored — twelve bytes, and it is the reason the
/// denominator is live at all. It cannot publish the numerator: counting
/// objects as they pass means inflating every one of them, which is a second
/// pack decoder, and gunnar has exactly one (one implementation, and `fetch_pack`'s own
/// module doc refuses it in those words). The one decoder is the storage
/// engine's walk, which has to split every entry anyway, so the count is free
/// exactly there and nowhere else. This type is where the two halves meet.
///
/// # `total` is remembered here, and that is not a second counter
///
/// [`PackProgress::indexed`] carries its own `total` — the count the engine
/// read from the same header — and it is *checked* against the one the wire
/// announced rather than trusted, because two readings of one field that
/// disagree mean the ingest is being handed a different pack from the one that
/// was negotiated. `fetch_pack` makes the same comparison at the end, on the
/// receipt; this is the same fact asserted while there is still a transfer to
/// abandon.
struct QueueProgress<'a> {
    report: &'a Reporter<Mirror>,
    total: AtomicU64,
    /// A second sink, for a caller that has to see the *series* rather than the
    /// job's latest reading. See [`sync_once_watching`].
    watching: Option<&'a dyn git_storage_trait::PackProgress>,
}

impl QueueProgress<'_> {
    /// **The one place a sync's progress pair is written.**
    ///
    /// All three publishers go through here — the header's `(0, N)`, the
    /// engine's per-object readings, and [`finish`]'s `(N, N)` — so the job and
    /// any watcher see the same values in the same order. Two routes to the
    /// same field would be two places the series could differ, and the series
    /// is the whole subject.
    fn publish(&self, done: u64, total: u64) {
        self.report.progress(done, total);
        if let Some(watcher) = self.watching {
            watcher.indexed(done, total);
        }
    }
}

impl FetchProgress for QueueProgress<'_> {
    fn objects_total(&self, objects: u32) {
        self.total.store(u64::from(objects), Ordering::SeqCst);
        self.publish(0, u64::from(objects));
    }

    fn pack_bytes(&self, streamed: u64) {
        self.report.report(|mirror| mirror.bytes = streamed);
    }
}

impl git_storage_trait::PackProgress for QueueProgress<'_> {
    fn indexed(&self, done: u64, total: u64) {
        // The wire's denominator wins. An engine that read a different total
        // from the same twelve bytes is a bug worth a log line, not a reason to
        // renumber a progress bar mid-flight and leave a console watching a
        // percentage that jumped sideways.
        let announced = self.total.load(Ordering::SeqCst);
        let total = if announced == 0 || announced == total {
            total
        } else {
            tracing::warn!(
                target: TARGET_GIT,
                announced,
                indexed_total = total,
                "the packfile header the wire read and the one the store read declare \
                 different object counts; the wire's is the one being rendered"
            );
            announced
        };
        self.publish(done, total);
    }
}

// ── the fetch ────────────────────────────────────────────────────────────────

/// **One sync, start to finish.** The [`crate::mirror_queue::MirrorRunner`]
/// body.
///
/// Errors are `String` because they are going onto a job a console will read,
/// not up a call stack — the same reason `gc_queue`'s runner returns one.
pub fn sync_once(
    stores: &Arc<ServerStores>,
    order: &MirrorOrder,
    credential: &Credential,
    report: &Reporter<Mirror>,
) -> Result<(), String> {
    sync_once_watching(stores, order, credential, report, None)
}

/// [`sync_once`], with a **second** place the object count is published to.
///
/// # Why this door exists, and why it is not a test hook
///
/// [`Reporter::progress`] *overwrites*, by design — its own doc says a progress
/// that accumulated would be a second counter to keep right — so the job record
/// holds the latest reading and never the series. That is correct for a console,
/// which only ever wants the latest reading, and it means the one thing that
/// cannot be established from the job record afterwards is **whether the count
/// moved at all**. A publisher that fires `(0, N)` and then `(N, N)` and a
/// publisher that fires five hundred times in between leave the job in exactly
/// the same state, so from the outside they are the same publisher.
///
/// That is measure-never-assert in its plainest form: a stub is
/// indistinguishable from an
/// implementation unless something can see it work — and it is precisely the
/// defect this seam was opened to close: `(0, N) → (N, N)` was the shape for as
/// long as no engine published from inside its pack walk. So the series gets a
/// door, and the acceptance test stands at it. There is one other caller,
/// [`sync_once`], which passes `None`.
///
/// The sink is *additional*: the job is written exactly as before, in the same
/// order, whether or not anybody is watching. A test that changed what the
/// production path publishes would be testing itself.
pub fn sync_once_watching(
    stores: &Arc<ServerStores>,
    order: &MirrorOrder,
    credential: &Credential,
    report: &Reporter<Mirror>,
    watching: Option<&dyn git_storage_trait::PackProgress>,
) -> Result<(), String> {
    let wire = classify(&order.upstream)?;
    check_credential(wire, credential)?;
    // Checked again on the worker, and not only at the door where
    // `Mirror.Add` checked it: a registration made before this rule existed,
    // or a name whose A record has moved since, reaches the dial through here
    // and nowhere else.
    check_destination(&order.upstream, stores.root())?;

    // Resolved here, on the worker, and not at enqueue time: a repository can
    // be deleted while its sync waits, and discovering that when the work
    // starts is a FAILED job saying so, which is the truth.
    let repo_ref = gunnar_core::resolve_repo_path(&order.path).map_err(|err| {
        format!(
            "`{}` is not a valid <namespace>/<name>: {err:#}",
            order.path
        )
    })?;
    let existing = stores
        .catalog()
        .get(&repo_ref)
        .map_err(|err| format!("reading the catalog row for {}: {err:#}", order.path))?;
    // **An import's repository is made HERE, by the first sync that reaches
    // the upstream** (owner order 2026-10-18; `MirrorRecord::creates_repo`).
    let to_create = existing.is_none()
        && stores
            .mirrors()
            .get(&repo_ref)
            .map_err(|err| format!("reading the registration for {}: {err:#}", order.path))?
            .is_some_and(|record| record.creates_repo);
    if existing.is_none() && !to_create {
        return Err(format!("{} is no longer a live repository", order.path));
    }
    // **An encrypted repository is never fetched into** (wiring tracer B
    // 2026-10-08): the ingest below runs no encrypted gate (`PushIngest::new`),
    // so an upstream's plaintext would be stored and framed for the twin.
    // `Mirror.Add` / `Repos.SetEncrypted` refuse making such a mirror; this is
    // a registration older than that rule, or a flag set below the door.
    if existing.as_ref().is_some_and(|row| row.encrypted) {
        return Err(format!(
            "{} is encrypted, so it is not fetched into: a mirror copies another server's \
             plaintext, and an encrypted repository holds only what its members sealed \
             (remove the mirror)",
            order.path
        ));
    }
    // **The namespace's room, asked of the same gates a push is held to**
    // (2026-10-02). A suspended namespace or one already at its cap fetches
    // nothing — refused here, before the dial, in the gate's own words — and
    // the fetch is held to the tightest ceiling in the chain as it streams.
    let ceiling = room_for(&stores.plan_gates(), repo_ref.namespace()).map_err(|(why, line)| {
        report.report(|r| r.refused = why);
        line
    })?;

    let auth = match credential {
        Credential::Token { username, secret } if username.is_empty() => {
            Some(format!("Bearer {secret}"))
        }
        Credential::Token { username, secret } => {
            Some(format!("Basic {}", base64_basic(username, secret)))
        }
        _ => None,
    };

    let row = match existing {
        Some(row) => row,
        None => create_from_upstream(stores, &repo_ref, order, wire, credential, auth.clone())?,
    };
    let fetched = fetch_into(
        stores, &row, order, wire, credential, auth, ceiling, report, watching,
    )
    .and_then(|advertised| landed(stores, &row, order, &advertised).map(|()| advertised));
    match &fetched {
        Ok(advertised) => after_landing(stores, &repo_ref, &row, order, advertised),
        Err(_) if to_create => discard_created(stores, &repo_ref, order),
        Err(_) => {}
    }
    fetched.map(|_| ())
}

/// **A sync is DONE only when what the upstream advertised is here**
/// (import-big, 2026-10-26).
///
/// Every branch and tag the upstream advertised must be a reference in this
/// repository when the sync says it is done, and an upstream that advertised
/// none is not an import that worked: it is an empty import, said as a
/// failure. Read back from the store, after the apply, rather than inferred
/// from the plan — the plan is what was meant, this is what landed. Before it,
/// a fetch whose plan came out empty applied nothing and settled DONE over a
/// repository with no references at all.
fn landed(
    stores: &Arc<ServerStores>,
    row: &gunnar_core::model::Repo,
    order: &MirrorOrder,
    advertised: &[gunnar_wire::upload_pack::AdvertisedRef],
) -> Result<(), String> {
    let wanted: Vec<&str> = advertised
        .iter()
        .filter(|r| fetch_pack::mirror(&r.name) && r.target.is_some())
        .map(|r| r.name.as_str())
        .collect();
    let upstream = scrub(&order.upstream);
    if wanted.is_empty() {
        return Err(format!(
            "{upstream} has no branches or tags, so there was nothing to import into {}. \
             Check the address, or push a first commit there and sync again.",
            order.path
        ));
    }
    let store = stores
        .repo_store(row)
        .map_err(|err| format!("opening {} to check what landed: {err:#}", order.path))?;
    let held = GitOps::refs(store.serve()).map_err(|err| {
        format!(
            "reading {}'s references to check what landed: {err:#}",
            order.path
        )
    })?;
    let missing: Vec<&str> = wanted
        .iter()
        .copied()
        .filter(|name| !held.iter().any(|r| r.name == *name && r.oid.is_some()))
        .collect();
    if let Some(first) = missing.first() {
        return Err(format!(
            "the fetch from {upstream} ended without its references landing: {} of {} \
             (first {first}) are not in {}. Nothing is reported as imported that is not \
             here; sync again.",
            missing.len(),
            wanted.len(),
            order.path
        ));
    }
    Ok(())
}

/// **Ask the upstream what it is, then make the repository to match.**
///
/// One `ls-refs` at the upstream's own object format (`ls_remote_any`): the
/// format becomes the repository's, and `HEAD`'s `symref-target` its default
/// branch. Nothing is created when the upstream cannot be reached or refuses
/// (a private repository without its token), so a failed first import leaves
/// no repository behind.
fn create_from_upstream(
    stores: &Arc<ServerStores>,
    repo_ref: &gunnar_core::RepoRef,
    order: &MirrorOrder,
    wire: Wire,
    credential: &Credential,
    auth: Option<String>,
) -> Result<gunnar_core::model::Repo, String> {
    let advertisement = ls_remote(stores, order, wire, credential, auth)?;
    let hash = match advertisement.hash_kind {
        gix_hash::Kind::Sha1 => gunnar_core::HashKind::Sha1,
        _ => gunnar_core::HashKind::Sha256,
    };
    let mut new = gunnar_core::NewRepo::new(repo_ref.clone())
        .hash_kind(hash)
        .visibility(gunnar_core::Visibility::Private);
    new.description = format!("mirror of {}", scrub(&order.upstream));
    if let Some(branch) = advertised_head(&advertisement.refs) {
        new = new.default_branch(branch);
    }
    tracing::info!(
        target: TARGET_GIT,
        event = "mirror.create_from_upstream",
        repo = %order.path,
        hash = ?hash,
        "an import's repository is made at the upstream's object format and default branch"
    );
    let row = stores
        .create_repo(new, false)
        .map_err(|err| format!("creating {} for the import: {err:#}", order.path))?;
    stores
        .repo_store(&row)
        .map_err(|err| format!("opening {} to mirror into it: {err:#}", order.path))?;
    Ok(row)
}

/// **One `ls-refs` against the upstream**, at its own object format.
fn ls_remote(
    stores: &Arc<ServerStores>,
    order: &MirrorOrder,
    wire: Wire,
    credential: &Credential,
    auth: Option<String>,
) -> Result<gunnar_wire::fetch_pack::Advertisement, String> {
    let listed =
        |err: gunnar_wire::Error| format!("asking {} what it holds: {err}", scrub(&order.upstream));
    match wire {
        Wire::Http => {
            let (mut reader, mut writer) = open_http(&order.upstream, auth)?;
            gunnar_wire::ls_remote_any(&mut reader, &mut writer).map_err(listed)
        }
        Wire::Ssh => dial_ssh(
            &order.upstream,
            credential,
            &stores.root().join(MIRROR_KNOWN_HOSTS),
            strict_host_keys(stores.root()),
            |reader, writer| gunnar_wire::ls_remote_any(reader, writer),
        )?
        .map_err(listed),
    }
}

/// **What an interrupted import already brought in**, offered back to the
/// upstream as `have`s (import-big, 2026-10-26).
///
/// A repository that holds no branch and no tag is either new or an import
/// that stopped between its pack landing and its references being written —
/// a restart in the index, the longest step of a big import. The pack's bytes
/// were durable before that step, and the indexer resumes them on its own after
/// the restart; what was missing is only the reference write. So the resumed
/// sync asks the upstream what it advertises, lets the index catch up, and
/// offers every advertised tip this store already has as a `have`: the
/// upstream then sends nothing it sent before, and a 6.6 GB kernel pack is not
/// fetched (or stored) a second time. A new repository has nothing, and pays
/// one `ls-refs` for finding out.
#[allow(clippy::too_many_arguments)]
fn already_here(
    stores: &Arc<ServerStores>,
    git: &dyn gunnar_store::git_storage_trait::GitServe,
    git_dir: &std::path::Path,
    order: &MirrorOrder,
    wire: Wire,
    credential: &Credential,
    auth: Option<String>,
    report: &Reporter<Mirror>,
) -> Result<Vec<ObjectId>, String> {
    let held =
        GitOps::refs(git).map_err(|err| format!("reading {}'s references: {err:#}", order.path))?;
    if held
        .iter()
        .any(|r| fetch_pack::mirror(&r.name) && r.oid.is_some())
    {
        return Ok(Vec::new());
    }
    let advertisement = ls_remote(stores, order, wire, credential, auth)?;
    // An interrupted import's pack is indexed here: held to a cgroup's memory as the
    // first try was (IMPORT-2); its object count is not known again, so no refusal.
    let held = crate::import_room::MemoryLimit::read()
        .and_then(|l| CgroupHold::start(stores, git_dir, &l));
    index_landed(git, git_dir, stores.engine(), report, held.as_ref()).map_err(|err| {
        format!(
            "{} holds objects this server could not index: {err}",
            order.path
        )
    })?;
    let mut out = Vec::new();
    for advertised in &advertisement.refs {
        let Some(target) = advertised.target else {
            continue;
        };
        if !fetch_pack::mirror(&advertised.name) || out.contains(&target) {
            continue;
        }
        if GitOps::has(git, target.as_bytes()).unwrap_or(false) {
            out.push(target);
        }
    }
    if !out.is_empty() {
        tracing::info!(
            target: TARGET_GIT,
            event = "mirror.sync.resumed",
            repo = %order.path,
            haves = out.len(),
            "an import that stopped after its pack landed resumes without fetching it again"
        );
    }
    Ok(out)
}

/// `HEAD`'s branch, as the upstream advertised it, when it is a branch.
fn advertised_head(refs: &[gunnar_wire::upload_pack::AdvertisedRef]) -> Option<String> {
    refs.iter()
        .find(|r| r.name == "HEAD")
        .and_then(|r| r.symref_target.clone())
        .filter(|target| {
            target.starts_with("refs/heads/") && gunnar_store::refname::validate(target).is_ok()
        })
}

/// A first import that failed after its repository was made: the repository
/// goes again, so nothing empty is left behind. Best effort — a removal that
/// fails is logged, and the registration still says the repository is to come.
fn discard_created(
    stores: &Arc<ServerStores>,
    repo_ref: &gunnar_core::RepoRef,
    order: &MirrorOrder,
) {
    let removed = stores
        .catalog()
        .delete(repo_ref)
        .map_err(|err| format!("{err:#}"))
        .and_then(|_| {
            stores
                .purge_repo(repo_ref)
                .map_err(|err| format!("{err:#}"))
        });
    if let Err(err) = removed {
        tracing::warn!(
            target: TARGET_GIT,
            repo = %order.path,
            %err,
            "a failed first import could not remove the repository it had made"
        );
    }
}

/// **What a landed sync owes the mirror besides its references**: the
/// upstream's default branch (when it holds that branch now) and whether the
/// upstream uses Git LFS. Neither fails the sync — the references landed.
fn after_landing(
    stores: &Arc<ServerStores>,
    repo_ref: &gunnar_core::RepoRef,
    row: &gunnar_core::model::Repo,
    order: &MirrorOrder,
    advertised: &[gunnar_wire::upload_pack::AdvertisedRef],
) {
    let mut row = row.clone();
    if let Some(branch) = advertised_head(advertised) {
        let held = stores
            .repo_store(&row)
            .ok()
            .and_then(|store| GitOps::refs(store.serve()).ok())
            .is_some_and(|refs| refs.iter().any(|r| r.name == branch && r.oid.is_some()));
        if held && branch != row.default_branch {
            match stores.catalog().set_default_branch(repo_ref, &branch) {
                Ok(moved) => {
                    row = moved;
                    if let Err(err) = stores.repoint_head(&row) {
                        tracing::warn!(target: TARGET_GIT, repo = %order.path, %err, "pointing HEAD at the upstream's default branch");
                    }
                }
                Err(err) => {
                    tracing::warn!(target: TARGET_GIT, repo = %order.path, %err, "taking the upstream's default branch");
                }
            }
        }
    }
    let lfs = uses_lfs(stores, &row);
    if let Err(err) = stores.mirrors().set_lfs_detected(repo_ref, lfs) {
        tracing::warn!(target: TARGET_GIT, repo = %order.path, %err, "recording whether the upstream uses LFS");
    }
}

/// **Does the default branch route files through Git LFS?** A `filter=lfs`
/// in its root `.gitattributes` is the sign every LFS repository carries; the
/// files it names arrive as pointer files, and their content is not mirrored.
fn uses_lfs(stores: &Arc<ServerStores>, row: &gunnar_core::model::Repo) -> bool {
    use gunnar_store::git_storage_trait::ObjType;
    use gunnar_store::object;
    let Ok(store) = stores.repo_store(row) else {
        return false;
    };
    let git = store.serve();
    let hash = store.hash_kind();
    let Some(tip) = GitOps::refs(git)
        .ok()
        .and_then(|refs| refs.into_iter().find(|r| r.name == row.default_branch))
        .and_then(|r| r.oid)
        .and_then(|raw| ObjectId::try_from(&raw[..]).ok())
    else {
        return false;
    };
    let read = |id: &ObjectId, want: ObjType| match git.read(id.as_bytes()) {
        Ok(Some((kind, bytes))) if kind == want => Some(bytes),
        _ => None,
    };
    let Some(commit) =
        read(&tip, ObjType::Commit).and_then(|bytes| object::commit(&bytes, hash).ok())
    else {
        return false;
    };
    let Some(tree) = read(&commit.tree, ObjType::Tree) else {
        return false;
    };
    let Ok(Some(entry)) = object::tree_entry_named(&tree, hash, b".gitattributes") else {
        return false;
    };
    read(&entry.oid, ObjType::Blob).is_some_and(|attributes| {
        String::from_utf8_lossy(&attributes)
            .lines()
            .any(|line| !line.trim_start().starts_with('#') && line.contains("filter=lfs"))
    })
}

/// **The fetch and the apply**, into a repository that exists. Answers what
/// the upstream advertised, for [`after_landing`].
#[allow(clippy::too_many_arguments)]
fn fetch_into(
    stores: &Arc<ServerStores>,
    row: &gunnar_core::model::Repo,
    order: &MirrorOrder,
    wire: Wire,
    credential: &Credential,
    auth: Option<String>,
    ceiling: Option<gunnar_wire::PackCeiling>,
    report: &Reporter<Mirror>,
    watching: Option<&dyn git_storage_trait::PackProgress>,
) -> Result<Vec<gunnar_wire::upload_pack::AdvertisedRef>, String> {
    let row = row.clone();
    let store = stores
        .repo_store(&row)
        .map_err(|err| format!("opening {} to mirror into it: {err:#}", order.path))?;
    let git = store.serve();
    let hash_kind = store.hash_kind();

    // Opened BEFORE the ingest, and that ordering is load-bearing for the same
    // reason it is on the push path: whether this repository has a protection
    // rule at all is what decides if the fetch pays for an object-set
    // enumeration.
    let protection = Protection::open(stores, &row, store.git_dir())?;

    let extra_haves = already_here(
        stores,
        git,
        store.git_dir(),
        order,
        wire,
        credential,
        auth.clone(),
        report,
    )?;
    let local = StoreRefs { git, extra_haves };
    let progress = QueueProgress {
        report,
        total: AtomicU64::new(0),
        watching,
    };
    // The ingest a PUSH lands through, unchanged. A fetched pack and a pushed
    // pack are the same bytes (`fetch_pack`'s module doc), so this is the same
    // storage authority and not a second one.
    //
    // The object set is asked for exactly when something reads it. A repository
    // with a protection rule has a gate that does (`require_signature` walks the
    // commits this fetch introduced), so the enumeration is `Required` and an
    // engine that cannot produce one must refuse rather than let the gate decide
    // on nothing. With no rule configured nobody reads it, and `NotNeeded` keeps
    // the fetch from waiting on index work no answer depends on.
    //
    // The last argument is the numerator's whole route: the ingest hands it to
    // the engine, the engine hands it to the pack walk, and the walk publishes
    // once per object into `progress` above — which is the same value the wire
    // is publishing the denominator and the byte count into. One sink, three
    // publishers, no second decoder anywhere on the path.
    let ingest = crate::transport::PushIngest::new(
        store.git_dir().to_path_buf(),
        hash_kind,
        stores.engine(),
        if protection.is_some() {
            crate::storage::ObjectSet::Required
        } else {
            crate::storage::ObjectSet::NotNeeded
        },
        Some(&progress),
        // A fetched pack is a transaction exactly as a pushed one is: framed before it lands.
        Some(crate::transport::PackFrame::of(stores, &row)),
    )
    // …and held to the data set's room the way a push is (cov-13).
    .with_space(stores.space_ledger());

    let mut fetch = FetchPack::new(hash_kind, &fetch_pack::mirror, &local).with_progress(&progress);
    if let Some(ceiling) = ceiling {
        fetch = fetch.with_ceiling(ceiling);
    }

    let outcome = match wire {
        Wire::Http => {
            let (mut reader, mut writer) = open_http(&order.upstream, auth)?;
            fetch.fetch(&ingest, &mut reader, &mut writer)
        }
        Wire::Ssh => dial_ssh(
            &order.upstream,
            credential,
            &stores.root().join(MIRROR_KNOWN_HOSTS),
            strict_host_keys(stores.root()),
            |reader, writer| fetch.fetch(&ingest, reader, writer),
        )?,
    }
    .map_err(|err| {
        // Cut at the namespace's ceiling: the plan's refusal, as a field.
        if matches!(err, gunnar_wire::Error::OverCeiling(_)) {
            report.report(|r| r.refused = crate::mirror_queue::Refusal::PackCap);
        }
        format!(
            "fetching {} from {}: {err}",
            order.path,
            scrub(&order.upstream)
        )
    })?;

    // **Index the fetched pack ONCE, here, before a single read by oid**
    // (import-big, 2026-10-26) — the line `receive_pack::apply` has carried
    // since 2026-08-30 and this path never got.
    //
    // The znippy engine stores a pack's bytes before its oids exist and indexes
    // them behind the ack. `apply` reads by oid (the branch-protection gate, and
    // `put_refs_cas`, whose dangling-target check asks `has` for every new tip),
    // so without this its first read fell into the engine's read-time fallback:
    // a whole-pack resolve on this thread beside the indexer resolving the same
    // bytes. MEASURED on oden 2026-10-26, torvalds/linux from GitHub (6.60 GB,
    // 11 986 735 objects): the import landed only after the two resolvers had
    // driven the process to a 19.4 GB peak — past a 16 GB Mini, which is where
    // the owner's import of it died with nothing to say so.
    //
    // A refusal is the backend's own sentence (its memory watchdog gives up by
    // name, with what was available and what was needed): the pack is stored
    // and could not be indexed on this box, which is "this repository does not
    // fit here", never a sync that waits for ever or reads around it.
    if outcome.receipt.is_some() {
        // **IMPORT-2: the index must fit the memory this server may really use**: the box's
        // MemTotal, or the cgroup's memory.max when that is smaller (a container, a
        // MemoryMax= unit, where /proc/meminfo is the host's). Refused by name before
        // the index when it does not fit with the margin; under a cgroup the store's
        // memory guard reads the cgroup's memory, kept fresh while the index runs.
        let objects = progress.total.load(Ordering::SeqCst);
        let limit = crate::import_room::MemoryLimit::read();
        if let Some(limit) = &limit {
            crate::import_room::import_room(&order.path, objects, limit)?;
        }
        let held = limit
            .as_ref()
            .and_then(|l| CgroupHold::start(stores, store.git_dir(), l));
        index_landed(git, store.git_dir(), stores.engine(), report, held.as_ref()).map_err(
            |err| {
                format!(
                "{} was fetched and stored, and this server could not index it: {err}. Nothing \
                 was moved. The repository is larger than this server can take in; a server \
                 with more memory imports it.",
                order.path
            )
            },
        )?;
    }
    apply(git, order, &outcome, &progress, protection.as_ref())?;
    Ok(outcome.advertised)
}

/// **`GitServe::index_now`, with the index's own progress on the job.**
///
/// For a kernel-sized pack the index is minutes after the last byte arrived,
/// and a job that sat at `N/N` for all of them would read as stalled. So while
/// it runs, the engine's index row for this repository (`Engine::index_progress`,
/// the same row the dashboard reads) is published as the job's progress: the
/// objects indexed of the pack's total. When it returns the pair is the walk's
/// again (`finish`).
fn index_landed(
    git: &dyn gunnar_store::git_storage_trait::GitServe,
    git_dir: &std::path::Path,
    engine: crate::storage::Engine,
    report: &Reporter<Mirror>,
    held: Option<&CgroupHold>,
) -> Result<usize, String> {
    let indexed = indexed_with_progress(
        || git.index_now().map_err(|err| format!("{err:#}")),
        || {
            if let Some(held) = held {
                held.refresh();
            }
            if let Some((_, row)) = engine
                .index_progress()
                .into_iter()
                .find(|(dir, row)| dir == git_dir && row.active && row.objects_total > 0)
            {
                report.progress(row.objects_done, row.objects_total);
            }
        },
    );
    if let Ok(packs) = &indexed {
        tracing::info!(
            target: TARGET_GIT,
            event = "mirror.sync.indexed",
            packs,
            "the fetched pack was indexed once, before the references were checked"
        );
    }
    indexed
}

/// **The cgroup's memory, held under one import's index** (IMPORT-2).
///
/// Only when the limit is a cgroup's: the store's memory guard is pointed at a
/// meminfo-shaped file of the cgroup's memory (`import_room::cgroup_meminfo`), refreshed
/// every half second by the index's progress poller, with the import's margin as the
/// floor. Dropped after the index: the file goes and the store gets the process-wide
/// guard back, so a later push into the same repository reads no stale file.
struct CgroupHold {
    engine: crate::storage::Engine,
    git_dir: std::path::PathBuf,
    cgroup: std::path::PathBuf,
    file: std::path::PathBuf,
}

impl CgroupHold {
    fn start(
        stores: &Arc<ServerStores>,
        git_dir: &std::path::Path,
        limit: &crate::import_room::MemoryLimit,
    ) -> Option<CgroupHold> {
        let cgroup = limit.cgroup_dir()?.to_path_buf();
        let dir = stores.root().join("import-memory");
        std::fs::create_dir_all(&dir).ok()?;
        let name: String = git_dir
            .to_string_lossy()
            .chars()
            .map(|c| if c.is_ascii_alphanumeric() { c } else { '_' })
            .collect();
        let held = CgroupHold {
            engine: stores.engine(),
            git_dir: git_dir.to_path_buf(),
            cgroup,
            file: dir.join(format!("{name}.meminfo")),
        };
        if !held.refresh() {
            return None;
        }
        let floor = limit.bytes / 100 * crate::import_room::MARGIN_PERCENT;
        if !held
            .engine
            .hold_index_memory(&held.git_dir, Some((&held.file, floor)))
        {
            let _ = std::fs::remove_file(&held.file);
            return None;
        }
        tracing::info!(
            target: TARGET_GIT,
            event = "mirror.import.memory_held",
            repo = %held.git_dir.display(),
            limit = limit.bytes,
            floor,
            "the import's index reads this server's cgroup memory ({}) and pauses under {}",
            limit.words(),
            crate::budget::human(floor)
        );
        Some(held)
    }

    /// Write the cgroup's memory now (a temp file and a rename, so the guard never
    /// reads half a file).
    fn refresh(&self) -> bool {
        let Some(c) = crate::import_room::cgroup_memory_in(&self.cgroup) else {
            return false;
        };
        let tmp = self.file.with_extension("meminfo.tmp");
        std::fs::write(&tmp, crate::import_room::cgroup_meminfo(&c)).is_ok()
            && std::fs::rename(&tmp, &self.file).is_ok()
    }
}

impl Drop for CgroupHold {
    fn drop(&mut self) {
        self.engine.hold_index_memory(&self.git_dir, None);
        let _ = std::fs::remove_file(&self.file);
    }
}

/// **`index` on this thread while `progress` is asked every half second on a scoped one**
/// ([`index_landed`]). The poller stops when the index ends, however it ends: a panic in
/// the index is an `Err` naming it (RA4-9).
fn indexed_with_progress(
    index: impl FnOnce() -> Result<usize, String>,
    progress: impl Fn() + Sync,
) -> Result<usize, String> {
    let done = std::sync::atomic::AtomicBool::new(false);
    std::thread::scope(|scope| {
        // A32-14, stays a scoped thread: it reads the index row for the length of one
        // `index_now` on the mirror job's own lane and is joined before this returns.
        scope.spawn(|| {
            while !done.load(Ordering::SeqCst) {
                progress();
                for _ in 0..10 {
                    if done.load(Ordering::SeqCst) {
                        return;
                    }
                    std::thread::sleep(std::time::Duration::from_millis(50));
                }
            }
        });
        // RA4-9: a panic in the index is caught here, so `done` is set however it ends and the
        // scope (which waits on the poller) returns; the mirror job then fails by name.
        let indexed = std::panic::catch_unwind(std::panic::AssertUnwindSafe(index));
        done.store(true, Ordering::SeqCst);
        indexed.unwrap_or_else(|panic| {
            let words = panic
                .downcast_ref::<&str>()
                .map(|s| s.to_string())
                .or_else(|| panic.downcast_ref::<String>().cloned())
                .unwrap_or_else(|| "a panic with no words".into());
            Err(format!("the index panicked: {words}"))
        })
    })
}

/// **How much a mirror sync into `namespace` may fetch, decided by the front
/// doors' own plan gates** ([`ServerStores::install_plan_gates`]).
///
/// The push path's questions, in its order (`gunnar_ssh::server`'s
/// `ask_plan`): the gate's verdict on a `Write`, then — for a write that grows
/// data, which a fetch is — `pack_quota` (is the namespace already at its
/// cap?) and the TIGHTEST `pack_ceiling` in the chain. The first `Refuse` is
/// the answer, verbatim. The ceiling is bound to the namespace's shared pack
/// ledger, marked here before any gate measured, so a mirror fetch and a push
/// landing at the same moment share the room instead of each spending it.
///
/// `Ok(None)` is "nothing caps it": a chain with no gate that answers, which is
/// the community appliance's ordinary state.
///
/// A refusal comes back with WHICH question refused it — the plan's verdict
/// on a write ([`crate::mirror_queue::Refusal::Plan`]) or the pack room
/// ([`crate::mirror_queue::Refusal::PackCap`]) — so the job can say so as a
/// field (`MirrorJob.refused_reason`).
pub(crate) fn room_for(
    gates: &[Arc<dyn gunnar_ssh::PlanGate>],
    namespace: &str,
) -> Result<Option<gunnar_wire::PackCeiling>, (crate::mirror_queue::Refusal, String)> {
    use gunnar_ssh::{PlanAction, PlanVerdict};
    if gates.is_empty() {
        return Ok(None);
    }
    // Taken BEFORE any gate measures, for the reason the push path gives:
    // only one of the two roundings is a quota that leaks.
    let mark = gunnar_lfs::AllowanceLedger::shared_pack().mark(namespace);
    let mut tightest: Option<gunnar_ssh::PackCeiling> = None;
    for gate in gates {
        if let PlanVerdict::Refuse(line) = gate.verdict(namespace, PlanAction::Write) {
            return Err((crate::mirror_queue::Refusal::Plan, line));
        }
        if let PlanVerdict::Refuse(line) = gate.pack_quota(namespace) {
            return Err((crate::mirror_queue::Refusal::PackCap, line));
        }
        if let Some(asked) = gate.pack_ceiling(namespace) {
            if tightest
                .as_ref()
                .is_none_or(|held| asked.bytes < held.bytes)
            {
                tightest = Some(asked);
            }
        }
    }
    Ok(tightest.map(|ceiling| gunnar_wire::PackCeiling {
        bytes: ceiling.bytes,
        refusal: ceiling.refusal,
        ledger: Some(
            Arc::new(crate::transport::PackMark { mark }) as Arc<dyn gunnar_wire::PackLedger>
        ),
    }))
}

/// **Apply the plan, in one transaction, against the values the fetch
/// reasoned about.**
///
/// See the module doc for why this is `put_refs_cas` and why a rejection is a
/// failure rather than a skip.
fn apply(
    git: &dyn gunnar_store::git_storage_trait::GitServe,
    order: &MirrorOrder,
    outcome: &FetchReport,
    progress: &QueueProgress<'_>,
    protection: Option<&Protection>,
) -> Result<(), String> {
    let zero = "0".repeat(outcome.hash_kind.len_in_hex());

    // Owned first, because `RefCas` borrows its oids and the batch has to
    // outlive every one of them.
    // (name, old, new, peeled)
    type OwnedUpdate = (String, Option<Vec<u8>>, Option<Vec<u8>>, Option<Vec<u8>>);
    let mut owned: Vec<OwnedUpdate> = Vec::new();
    let mut updates: Vec<(String, String, String)> = Vec::new();

    for plan in &outcome.updates {
        // **The name came off somebody else's wire, so it goes through the ref
        // name rules before it is written.**
        //
        // `plan.name` is the string an upstream's `ls-refs` answered with,
        // carried untouched from `fetch_pack`, and the only filter it passed on
        // the way is `fetch_pack::mirror`'s prefix test — which admits
        // `refs/heads/../../etc/passwd` exactly as readily as `refs/heads/main`.
        // Receive-pack puts every client-supplied name through
        // `gunnar_store::refname::validate` at the same point in its own path
        // (`receive_pack.rs`'s `funny refname`), and a fetch is the same
        // untrusted input arriving through a different door.
        //
        // It is checked HERE rather than relied on from the storage engine. The
        // memory backend does validate in `put_refs_cas`, so a name like this
        // would be refused there today — but that is a property of one engine,
        // and "a rule one backend enforces by accident is not a rule" is the
        // sentence `refname` itself is written under. The engines are swappable
        // (`GUNNAR_ENGINE`), and this is the one caller that can guarantee the
        // check ran.
        //
        // Only the names that came from the upstream. The prune half below
        // reads its names out of this repository's OWN refs, and a bad name
        // already in the store is one a prune should be able to delete rather
        // than one it must refuse to touch.
        gunnar_store::refname::validate(plan.name.as_bytes()).map_err(|err| {
            format!(
                "the upstream advertised a reference name gunnar will not write: {err}. Nothing \
                 was moved. A name is not a path, and this one would not have been a name."
            )
        })?;
        // The peel the remote already sent, rather than a peel computed here.
        // `ls-refs` was asked with `peel`, so an annotated tag arrived with the
        // commit it resolves to; recomputing it would be an object read per tag
        // for an answer already in hand (the tag re-peel, from the other side).
        let peeled = outcome
            .advertised
            .iter()
            .find(|advertised| advertised.name == plan.name)
            .and_then(|advertised| advertised.peeled)
            .map(|oid| oid.as_bytes().to_vec());
        updates.push((
            plan.name.clone(),
            plan.old
                .map(|old| old.to_string())
                .unwrap_or_else(|| zero.clone()),
            plan.new.to_string(),
        ));
        owned.push((
            plan.name.clone(),
            plan.old.map(|old| old.as_bytes().to_vec()),
            Some(plan.new.as_bytes().to_vec()),
            peeled,
        ));
    }

    if order.prune {
        // **What the upstream no longer has.** Only over the same namespace the
        // ref plan covers (`refs/heads/*`, `refs/tags/*`): a prune that reached
        // wider would delete this box's own `refs/gunnar/*` bookkeeping on the
        // grounds that an upstream never advertised it.
        let held = GitOps::refs(git)
            .map_err(|err| format!("reading {}'s refs to prune them: {err:#}", order.path))?;
        for row in held {
            if !fetch_pack::mirror(&row.name) {
                continue;
            }
            if outcome
                .advertised
                .iter()
                .any(|advertised| advertised.name == row.name)
            {
                continue;
            }
            let Some(old) = row.oid else { continue };
            updates.push((row.name.clone(), hex(&old), zero.clone()));
            owned.push((row.name, Some(old), None, None));
        }
    }

    if owned.is_empty() {
        finish(outcome, progress, Vec::new());
        return Ok(());
    }

    // **The gate, between the plan and the write.** Before this line the batch
    // below was the one reference write on the box that no gate had seen; a
    // repository with no protection rule configured has no `Protection` at all,
    // so a mirror of an unprotected repository pays nothing for this.
    if let Some(protection) = protection {
        let gated: Vec<gunnar_policy::RefUpdate> = owned
            .iter()
            .map(|(name, old, new, _)| {
                // The null-oid spelling never appears here: `owned` already
                // carries absence as `None`, which is what `RefUpdate`'s three
                // constructors take. `try_from_bytes` cannot fail on a value the
                // store or the wire produced, and an oid this store cannot
                // decode is the one `StoreRefs::tips` already refuses by name.
                let decode = |raw: &Vec<u8>| {
                    gix_hash::oid::try_from_bytes(raw)
                        .map(gix_hash::oid::to_owned)
                        .map_err(|err| {
                            format!("{name} holds an oid this server cannot decode: {err}")
                        })
                };
                Ok(gunnar_policy::RefUpdate {
                    name: name.clone(),
                    old: old.as_ref().map(&decode).transpose()?,
                    new: new.as_ref().map(&decode).transpose()?,
                })
            })
            .collect::<Result<_, String>>()?;
        let new_objects = match outcome.receipt.as_ref() {
            // Nothing was fetched, so nothing was introduced. That is a measured
            // empty set and not an unasked question — see `FetchReport::receipt`.
            None => Ok(Vec::new()),
            Some(receipt) => receipt.new_objects.clone(),
        };
        protection.check(git, &gated, new_objects.as_deref().map_err(String::as_str))?;
    }

    let batch: Vec<RefCas<'_>> = owned
        .iter()
        .map(|(name, old, new, peeled)| RefCas {
            name: name.clone(),
            old: old.as_deref(),
            new: new.as_deref(),
            peeled: peeled.clone(),
        })
        .collect();

    // ONE call. Every edit or none — a loop with a break leaves what it already
    // applied applied, and a half-mirrored repository is a state no upstream
    // ever had.
    GitOps::put_refs_cas(git, &batch).map_err(|err| {
        format!(
            "the fetch succeeded and the reference update did not: {err:#}. Nothing was moved. \
             The usual cause is a push that landed on {} while the fetch was running; the \
             expected values came from the snapshot the fetch negotiated against, and \
             overwriting a concurrent push is the one thing a mirror must never do. Sync again.",
            order.path
        )
    })?;

    tracing::info!(
        target: TARGET_GIT,
        event = "mirror.sync.applied",
        repo = %order.path,
        refs = updates.len(),
        objects = outcome.objects_done,
        bytes = outcome.pack_bytes,
        "a mirror sync moved this repository's references"
    );

    finish(outcome, progress, updates);
    Ok(())
}

/// **Close the job's numbers out**, on both paths out of [`apply`].
///
/// One function and not two copies, because the pair below is the whole of what
/// a console reads when a sync ends and the two paths differ only in whether
/// any reference moved.
///
/// The `progress` call is the second half of the pair `FetchProgress` opened.
/// `objects_total` published `(0, N)` the instant the packfile header arrived;
/// **this publishes `(N, N)`**, and without it a finished job sits for ever at
/// nought per cent of a total it knows — which is worse than no progress bar,
/// because it reads as a transfer that stalled. `fetch_pack` has already
/// refused the transfer if the declared count and the indexed count disagree,
/// so `objects_done` is the same `N` and this cannot report a completion the
/// store did not make.
///
/// A fetch that wanted nothing — the ordinary answer for a poll of an idle
/// upstream — has no header, no `N`, and therefore **no progress at all**: the
/// pair stays `None`, and a caller reads `has_progress = false` rather than a
/// fabricated `0/0`.
fn finish(
    outcome: &FetchReport,
    progress: &QueueProgress<'_>,
    updates: Vec<(String, String, String)>,
) {
    if outcome.receipt.is_some() {
        let total = u64::from(outcome.objects_total);
        progress.publish(u64::from(outcome.objects_done), total);
    }
    progress.report.report(|mirror| {
        mirror.objects = u64::from(outcome.objects_done);
        mirror.bytes = outcome.pack_bytes;
        mirror.updates = updates;
    });
}

/// Lower-case hex, the spelling `gunnar.v1.RefUpdate` uses.
fn hex(raw: &[u8]) -> String {
    let mut out = String::with_capacity(raw.len() * 2);
    for byte in raw {
        use std::fmt::Write as _;
        let _ = write!(out, "{byte:02x}");
    }
    out
}

/// `base64(user:secret)`, the Basic-auth pair.
///
/// Written out rather than pulled in: the whole alphabet is 64 characters and
/// the encoder is nine lines, against a dependency that would be in the
/// appliance's graph for this one call. It is not a general base64 — it does
/// exactly the one thing, and `mirror_http`'s tests assert on the bytes it
/// produces.
fn base64_basic(user: &str, secret: &str) -> String {
    const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
    let raw = format!("{user}:{secret}");
    let raw = raw.as_bytes();
    let mut out = String::with_capacity(raw.len().div_ceil(3) * 4);
    for chunk in raw.chunks(3) {
        let b = [
            chunk[0],
            chunk.get(1).copied().unwrap_or(0),
            chunk.get(2).copied().unwrap_or(0),
        ];
        let packed = (u32::from(b[0]) << 16) | (u32::from(b[1]) << 8) | u32::from(b[2]);
        out.push(ALPHABET[(packed >> 18) as usize & 63] as char);
        out.push(ALPHABET[(packed >> 12) as usize & 63] as char);
        out.push(if chunk.len() > 1 {
            ALPHABET[(packed >> 6) as usize & 63] as char
        } else {
            '='
        });
        out.push(if chunk.len() > 2 {
            ALPHABET[packed as usize & 63] as char
        } else {
            '='
        });
    }
    out
}

// ── the ssh half ─────────────────────────────────────────────────────────────

/// **Dial `git-upload-pack` over SSH and hand the channel's two halves to
/// `run`.**
///
/// `gunnar_client::connect_upload_pack` is the fetch-side twin of the
/// `connect_receive_pack` replication already pushes with, over
/// `nornir-git`'s russh transport. There is no second dialler here and there
/// must not be: that transport owns the connect, the publickey exchange, the
/// `known_hosts` policy, the blocking bridge and — load-bearing — the `Drop`
/// that tears the channel down *inside* the driving runtime.
///
/// # The key is a file for the length of one dial, and then it is not
///
/// `nornir_git::ssh::SshAuth` authenticates from a **path**. gunnar's mirror
/// credential is sealed bytes in the control database, because a private key
/// sitting in the clear on `/data` is the thing that store exists to prevent.
/// The two are bridged by a [`tempfile::NamedTempFile`] created `0600` and
/// dropped — and therefore unlinked — before this function returns, on every
/// path out of it including a panic.
///
/// The alternative was an in-memory key API on `nornir-git`, which is the right
/// long-term shape, and it is a change to a crate outside this repository. It
/// is named here rather than left as a silent compromise.
#[cfg(feature = "replication")]
fn dial_ssh<T>(
    url: &str,
    credential: &Credential,
    known_hosts: &std::path::Path,
    strict: bool,
    run: impl FnOnce(&mut dyn BufRead, &mut dyn Write) -> T,
) -> Result<T, String> {
    use gunnar_client::ssh::{HostKeyPolicy, SshOptions};
    use std::io::BufReader;

    let mut options = SshOptions::new();
    // A server has no ssh-agent and no interactive user. Offering the agent
    // would be offering identities nobody configured for this mirror — the
    // silent substitution `gunnar_client::resolve_identity` exists to forbid.
    options.use_agent = false;

    // Held for the whole of `run`: dropping it unlinks the file, and the dial
    // reads it during `connect_upload_pack`.
    let key_file = match credential {
        Credential::SshKey(bytes) => {
            let mut file = tempfile::Builder::new()
                .prefix("gunnar-mirror-key-")
                .tempfile()
                .map_err(|err| format!("no room to materialise the mirror's ssh key: {err}"))?;
            #[cfg(unix)]
            {
                use std::os::unix::fs::PermissionsExt as _;
                std::fs::set_permissions(file.path(), std::fs::Permissions::from_mode(0o600))
                    .map_err(|err| {
                        format!("the mirror's ssh key could not be made private: {err}")
                    })?;
            }
            file.write_all(bytes)
                .and_then(|()| file.flush())
                .map_err(|err| format!("writing the mirror's ssh key: {err}"))?;
            options.key_path = Some(file.path().to_path_buf());
            Some(file)
        }
        // No key named means the ladder, which on a server finds nothing; that
        // is the honest behaviour for a public upstream over ssh and it is what
        // `gunnar-client` warns about at start-up.
        _ => None,
    };
    // Trust-on-first-use against a `known_hosts` under the data root is what
    // replication does when an operator has not said otherwise, and a
    // mirror is dialling the same kind of box. See [`MIRROR_KNOWN_HOSTS`] for
    // why the file is NOT `~/.ssh/known_hosts`.
    //
    // **Unless the box says otherwise**, which it could not before: the policy
    // was this literal and nothing else, so a first sync trusted whatever
    // answered on the upstream's address, on a path that is about to offer it a
    // private key. See [`MIRROR_STRICT_HOST_KEY`].
    options.host_key = if strict {
        HostKeyPolicy::StrictKnownHosts
    } else {
        HostKeyPolicy::TofuKnownHosts
    };
    options.known_hosts = Some(known_hosts.to_path_buf());

    let location = nornir_git::ssh::parse_ssh_url(url)
        .map_err(|err| format!("{} is not an ssh url: {err}", scrub(url)))?;
    let mut channel = gunnar_client::connect_upload_pack(&location, &options).map_err(|err| {
        // `{:#}` over an `anyhow::Error`, and not `{err}`: `gunnar_client::Error`
        // is a thiserror enum whose `Ssh` variant Displays as the bare word
        // "ssh" and carries the real sentence — which host key, which policy —
        // as its `source`. A FAILED job reading `opening git-upload-pack on …:
        // ssh` tells an operator nothing they can act on, and the reason was
        // one formatting flag away the whole time.
        format!(
            "opening git-upload-pack on {}: {:#}",
            scrub(url),
            anyhow::Error::new(err)
        )
    })?;
    let out = {
        let (reader, writer) = channel.io();
        let mut reader = BufReader::new(reader);
        run(&mut reader, writer)
    };
    drop(key_file);
    Ok(out)
}

/// **What a build with no ssh dialler says**, rather than what it does.
///
/// `replication` is the feature that carries `gunnar-client`'s wire half and
/// `nornir-git`'s russh transport; without it there is no dialler in the
/// binary. A mirror registered against an `ssh://` remote on such a build is a
/// registration that can never run, and it says so at the one place it can be
/// acted on — a FAILED job naming the flag — rather than failing as a
/// connection error that sends an operator to look at the network.
///
/// The smart-HTTP half is behind its own feature, `mirror-http` (A32-18), which every
/// appliance row names: [`NO_MIRROR_HTTP`] is its refusal.
#[cfg(not(feature = "replication"))]
fn dial_ssh<T>(
    url: &str,
    _credential: &Credential,
    _known_hosts: &std::path::Path,
    _strict: bool,
    _run: impl FnOnce(&mut dyn BufRead, &mut dyn Write) -> T,
) -> Result<T, String> {
    Err(format!(
        "{} is an ssh remote and this build carries no ssh dialler: it was compiled without \
         `--features replication`, which is what links `gunnar-client`'s russh transport. \
         Mirror from the upstream's `https://` URL, or use a build that has the feature.",
        scrub(url)
    ))
}

/// **Take any userinfo out of a URL before it is shown to anybody.**
///
/// `https://user:token@host/path` is a legal way to write a remote and an
/// operator may well have written one. Everything between `//` and the last
/// `@` of the authority is replaced, so a token that arrived in the URL cannot
/// leave through an error string.
///
/// Not a parser: this runs in an error path, and a URL this cannot make sense
/// of is returned with the whole authority elided rather than passed through.
pub(crate) fn scrub(url: &str) -> String {
    let Some((scheme, rest)) = url.split_once("://") else {
        return url.to_owned();
    };
    let (authority, path) = match rest.find('/') {
        Some(cut) => (&rest[..cut], &rest[cut..]),
        None => (rest, ""),
    };
    match authority.rfind('@') {
        Some(cut) => format!("{scheme}://<redacted>@{}{path}", &authority[cut + 1..]),
        None => url.to_owned(),
    }
}

/// **An `https://` remote's two halves** ([`crate::mirror_http`], feature `mirror-http`).
#[cfg(feature = "mirror-http")]
fn open_http(
    url: &str,
    auth: Option<String>,
) -> Result<(crate::mirror_http::HttpRead, crate::mirror_http::HttpWrite), String> {
    crate::mirror_http::open(url, auth)
}

/// The refusal an `https://` remote gets in a build without `mirror-http` (A32-18): the
/// airgap default has no HTTP client, and says so where it can be acted on.
pub const NO_MIRROR_HTTP: &str = "this build carries no HTTP client: it was compiled without gunnar-server's `mirror-http` feature (the appliance rows name it)";

#[cfg(not(feature = "mirror-http"))]
fn open_http(url: &str, _auth: Option<String>) -> Result<(std::io::Empty, std::io::Sink), String> {
    Err(format!("{} is an https remote and {NO_MIRROR_HTTP}. Mirror from the upstream's ssh:// URL, or use a build that has the feature.", scrub(url)))
}

#[cfg(test)]
mod index_panic_tests {
    use super::*;

    /// ★ RA4-9 RED: **an index that panics ends the index step with a named refusal**, at
    /// once — the progress poller is stopped however the index ends, so the mirror job
    /// reaches FAILED instead of hanging for good.
    #[test]
    fn an_index_that_panics_is_a_named_failure_not_a_hung_job() {
        let (tx, rx) = std::sync::mpsc::channel();
        std::thread::spawn(move || {
            let r = std::panic::catch_unwind(|| {
                indexed_with_progress(|| panic!("the index broke"), || {})
            });
            let _ = tx.send(r);
        });
        let r = rx
            .recv_timeout(std::time::Duration::from_secs(5))
            .expect("the index step ended within 5 s: no hang");
        let said = r.expect("a panic in the index is an answer, not a panic up the mirror job");
        let why = said.expect_err("a refusal");
        assert!(
            why.contains("the index panicked") && why.contains("the index broke"),
            "{why}"
        );
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// **The userinfo scrub is the last thing between a token in a URL and the
    /// log.** Asserted on the bytes, because "it does not appear" is the whole
    /// property.
    /// ★ A32-18: the airgap default carries no HTTP client, so an `https://` remote is refused
    /// by name (and its token never shown), never dialled.
    #[cfg(not(feature = "mirror-http"))]
    #[test]
    fn an_https_remote_is_refused_by_name_in_a_build_without_mirror_http() {
        let e = open_http(
            "https://user:tok3n@forge.mock.invalid/ns/repo.git",
            Some("Bearer tok3n".into()),
        )
        .expect_err("refused");
        assert!(
            e.contains(NO_MIRROR_HTTP) && e.contains("mirror-http"),
            "{e}"
        );
        assert!(!e.contains("tok3n"), "{e}");
    }

    #[test]
    fn a_url_with_userinfo_never_shows_it() {
        let scrubbed = scrub("https://alice:ghp_SUPERSECRET@example.test/ns/repo.git");
        assert!(
            !scrubbed.contains("ghp_SUPERSECRET"),
            "the token survived the scrub: {scrubbed}"
        );
        assert!(
            !scrubbed.contains("alice"),
            "the whole userinfo goes, not just the password half: {scrubbed}"
        );
        assert_eq!(scrubbed, "https://<redacted>@example.test/ns/repo.git");
    }

    /// A URL with no userinfo is left exactly as the operator wrote it — an
    /// error that mangled the ordinary case would be worse than no scrub.
    #[test]
    fn a_plain_url_is_untouched() {
        let url = "https://example.test/ns/repo.git";
        assert_eq!(scrub(url), url);
        assert_eq!(
            scrub("ssh://git@example.test/ns/repo.git"),
            "ssh://<redacted>@example.test/ns/repo.git"
        );
    }

    /// **Both transports are recognised from the URL alone**, and the scp-like
    /// form is ssh — the shape every forge prints on its clone button.
    #[test]
    fn a_remote_url_names_its_transport() {
        assert_eq!(classify("ssh://git@host/ns/repo.git"), Ok(Wire::Ssh));
        assert_eq!(classify("git+ssh://git@host/ns/repo.git"), Ok(Wire::Ssh));
        assert_eq!(classify("git@host:ns/repo.git"), Ok(Wire::Ssh));
        assert_eq!(classify("https://host/ns/repo.git"), Ok(Wire::Http));
        assert_eq!(
            classify("http://127.0.0.1:9418/ns/repo.git"),
            Ok(Wire::Http)
        );
    }

    /// A remote gunnar cannot dial is refused when it is REGISTERED, not on the
    /// first sync — and the refusal names the two forms that work.
    #[test]
    fn a_remote_that_is_neither_is_refused_by_name() {
        let err = classify("/srv/git/ns/repo.git").expect_err("a local path is not a remote");
        assert!(err.contains("ssh://") && err.contains("https://"), "{err}");
        let err = classify("git://host/ns/repo.git").expect_err("the dumb git protocol");
        assert!(err.contains("scheme gunnar cannot mirror from"), "{err}");
    }

    /// **The mismatch is refused by name, in both directions.**
    ///
    /// The failure being prevented is not an error — it is a 401 (or an SSH
    /// authentication failure) against an operator who can see that a
    /// credential IS stored, and who will therefore rotate a key that was never
    /// the problem.
    #[test]
    fn a_credential_of_the_wrong_kind_is_refused_by_name() {
        let key = Credential::SshKey(b"-----BEGIN OPENSSH PRIVATE KEY-----".to_vec());
        let token = Credential::Token {
            username: String::new(),
            secret: "ghp_SECRET".to_owned(),
        };

        let err = check_credential(Wire::Http, &key).expect_err("a key cannot ride on http");
        assert!(
            err.contains("smart http") && err.contains("ssh private key"),
            "{err}"
        );
        assert!(
            !err.contains("BEGIN OPENSSH"),
            "the refusal quoted the key: {err}"
        );

        let err = check_credential(Wire::Ssh, &token).expect_err("a token cannot ride on ssh");
        assert!(err.contains("ssh") && err.contains("a token"), "{err}");
        assert!(
            !err.contains("ghp_SECRET"),
            "the refusal quoted the token: {err}"
        );

        // …and the matching pairs, plus the public case, are all fine.
        check_credential(Wire::Ssh, &key).expect("a key over ssh");
        check_credential(Wire::Http, &token).expect("a token over http");
        check_credential(Wire::Ssh, &Credential::None).expect("a public ssh upstream");
        check_credential(Wire::Http, &Credential::None).expect("a public http upstream");
    }

    /// **The authority, out of each of the three shapes** — and the default
    /// port with it, because the port is half of what gets dialled.
    #[test]
    fn a_remote_url_names_the_machine_it_would_dial() {
        for (url, want) in [
            ("https://host/ns/repo.git", ("host".to_owned(), 443)),
            ("http://host/ns/repo.git", ("host".to_owned(), 80)),
            ("http://host:9418/ns/repo.git", ("host".to_owned(), 9418)),
            ("ssh://git@host/ns/repo.git", ("host".to_owned(), 22)),
            ("ssh://git@host:2222/ns/repo.git", ("host".to_owned(), 2222)),
            ("git@host:ns/repo.git", ("host".to_owned(), 22)),
            // The userinfo is not the host, and a `@` inside it must not become
            // one either.
            (
                "https://user:pass@host/ns/repo.git",
                ("host".to_owned(), 443),
            ),
            // An IPv6 literal is all colons; the brackets are what separates the
            // address from the port, and splitting on the first colon would
            // yield `[`.
            ("ssh://git@[::1]:2222/ns/repo.git", ("::1".to_owned(), 2222)),
            (
                "https://[2606:4700::1111]/ns/repo.git",
                ("2606:4700::1111".to_owned(), 443),
            ),
            ("git@[::1]:ns/repo.git", ("::1".to_owned(), 22)),
        ] {
            assert_eq!(authority(url), Ok(want.clone()), "{url}");
        }
        assert!(authority("https:///ns/repo.git").is_err(), "no host at all");
    }

    /// **Every range a caller must not be able to point this server at**, and
    /// the two that must still work.
    ///
    /// Asserted on [`address_verdict`] rather than on [`check_destination`]
    /// because the addresses are handed in: no resolver runs, so this is the
    /// same answer on a laptop, on a build box with a corporate DNS and in a
    /// container with none.
    #[test]
    fn a_mirror_refuses_every_address_only_the_server_can_reach() {
        for addr in [
            "127.0.0.1",
            "0.0.0.0",
            "10.0.0.5",
            "172.16.0.1",
            "192.168.1.1",
            // The one the finding names: cloud instance metadata.
            "169.254.169.254",
            "100.64.0.1",
            "192.0.0.1",
            "198.18.0.1",
            "240.0.0.1",
            "255.255.255.255",
            "224.0.0.1",
            "::1",
            "::",
            "fc00::1",
            "fe80::1",
            "ff02::1",
            // …written the long way. Without the mapped-address arm this one
            // walks past every v6 test and is allowed.
            "::ffff:127.0.0.1",
            "::ffff:169.254.169.254",
        ] {
            let ip: std::net::IpAddr = addr.parse().expect("a literal");
            let err = address_verdict("upstream.invalid", &[ip])
                .expect_err("{addr} must not be dialled on a caller's say-so");
            assert!(
                err.contains(addr) || err.contains(&ip.to_string()),
                "{addr}: {err}"
            );
            assert!(err.contains(MIRROR_ALLOW_PRIVATE), "{addr}: {err}");
        }

        for addr in ["93.184.216.34", "8.8.8.8", "2606:4700:4700::1111"] {
            let ip: std::net::IpAddr = addr.parse().expect("a literal");
            address_verdict("upstream.invalid", &[ip])
                .unwrap_or_else(|err| panic!("{addr} is an ordinary public address: {err}"));
        }
    }

    /// **One bad answer out of several is a refusal.** A name that resolves to a
    /// routable address and a loopback address is the rebinding shape, and a
    /// check that took the first answer, or the best one, would pass it.
    #[test]
    fn one_private_answer_among_several_refuses_the_whole_name() {
        let addrs: Vec<std::net::IpAddr> = ["93.184.216.34", "127.0.0.1"]
            .iter()
            .map(|a| a.parse().expect("a literal"))
            .collect();
        let err = address_verdict("both.invalid", &addrs).expect_err("one of them is loopback");
        assert!(err.contains("127.0.0.1"), "{err}");

        // And nothing at all is not "fine": a name with no address is refused
        // rather than allowed through to a dial that would fail anyway.
        assert!(address_verdict("nowhere.invalid", &[]).is_err());
    }

    /// The Basic pair is encoded the way every HTTP client encodes it. Checked
    /// against a value anybody can verify by hand rather than against this
    /// function's own output.
    #[test]
    fn basic_auth_is_ordinary_base64() {
        assert_eq!(
            base64_basic("Aladdin", "open sesame"),
            "QWxhZGRpbjpvcGVuIHNlc2FtZQ=="
        );
        assert_eq!(base64_basic("a", "b"), "YTpi");
        assert_eq!(base64_basic("ab", "b"), "YWI6Yg==");
    }
}