gunnar-server 2.0.0

//! **Replication to a second gunnar**: the inverse of `gunnar
//! mirror`.
//!
//! After a successful receive-pack this server pushes the repository on to
//! another gunnar — or to any SSH git remote — so that a second copy of every
//! repository exists on hardware you control, kept current without a human
//! running anything.
//!
//! ```text
//!   git push ──▶ gunnar A ──▶ event bus ──▶ this subscriber ──▶ queue ──▶ send-pack ──▶ gunnar B
//!                    │                                                       │
//!                    └── the client is already done here ────────────────────┘
//! ```
//!
//! # It is a subscriber, and that is the whole architecture
//!
//! Nothing in the push path calls into this module. `receive-pack` publishes
//! `RefUpdated` / `RefDeleted` to the event bus exactly as it did before replication, and
//! [`Replicator::spawn`] subscribes. The push path therefore cannot be made
//! slower, or made to fail, by anything here. That is the hard requirement,
//! because **a push that succeeded locally has succeeded**. A dead standby that
//! could refuse a push would be a standby that takes the primary down with it.
//!
//! Three things follow from being a subscriber, and each one is load-bearing:
//!
//! * `EventBus::publish` is a `broadcast::send`: it never blocks and never
//!   fails, with or without a subscriber attached.
//! * a replication that cannot keep up **lags the bus**, and the bus answers a
//!   lagging subscriber with an `EventGap` rather than with memory. That gap is
//!   handled below ([`TargetRunner`]'s resync), because a dropped `RefUpdated`
//!   is a ref that silently never replicates.
//! * the work runs on its own tasks. `gunnar-client`'s send-pack is blocking, so
//!   each attempt goes through `spawn_blocking` and never occupies a runtime
//!   worker that git traffic needs.
//!
//! # The queue bound, and why it is shaped the way it is
//!
//! **A replication job is "make target T's copy of repository R equal ours",
//! not "apply this ref update".** That is the decision the whole bound rests on.
//! The job carries no payload from the event, so a thousand pushes to one
//! repository against an unreachable standby collapse into **one** pending job.
//! That kills the unbounded-growth shape by construction, rather than by a
//! policy somebody has to remember to obey.
//!
//! What remains bounded by number is the count of *distinct repositories* with
//! work outstanding, per target:
//!
//! * **[`DEFAULT_QUEUE_CAPACITY`] = 1024 repositories per target.** A forge with
//!   more than a thousand repositories all changed since a standby was last
//!   reachable is not a queueing problem, it is an outage, and the answer to an
//!   outage is a resync rather than a bigger queue. At 1024 the queue is a
//!   `JobKey` and a timestamp each, on the order of a hundred kilobytes; the
//!   same firehose with one job per *push* would be unbounded, which is the
//!   shape that OOM'd a machine in this tree on 2026-08-03.
//! * **Hitting the bound REFUSES the new job and marks the target
//!   `needs_full_resync`.** Not drop-oldest: the oldest entry is the repository
//!   that has been waiting longest and is therefore the most stale, and silently
//!   abandoning it is exactly the "replication that stops while looking healthy"
//!   this module exists to prevent. Refusing the newest at least leaves the
//!   queue a coherent, still-draining set, and the refused repository is
//!   re-enqueued by its very next push anyway.
//! * Either way the operator is told, twice: a `ReplicationFailed` event with
//!   `needs_full_resync = true`, and a counter
//!   ([`ReplicationCounters::refused_queue_full`]) that only ever goes up.
//!
//! `needs_full_resync` is sticky. It is cleared only by a resync that actually
//! enqueued every matching repository ([`Replicator::resync`]), never by a
//! later job happening to succeed: one repository catching up says nothing
//! about the one that was dropped.
//!
//! # Failure is loud, and a REMOTE one is never final
//!
//! A failed attempt is re-queued with exponential backoff ([`backoff`]) capped
//! at [`DEFAULT_RETRY_MAX`], and a **remote** one retries forever. Giving up on
//! those after N attempts would produce precisely the rotting standby described
//! above: the queue would be empty, the counters quiet, and the copy months out
//! of date. Every attempt that fails emits a `ReplicationFailed` carrying the
//! attempt number and the delay before the next one, so "still trying" and
//! "not trying" are distinguishable from the outside.
//!
//! # A LOCAL failure is bounded, because retrying it changes nothing
//!
//! 🔴 MEASURED on the two-appliance drill, 2026-08-30: a repository whose
//! pack **this server's own store** could not index was re-queued forever at the
//! 300 s cap. The standby was healthy, the network was fine, and the identical
//! object-selection call was made every five minutes for 21 minutes against a
//! store whose answer could not change until somebody fixed it. The `EventGap`
//! leg of that drill never converged, and the reason sat in a log nobody reads.
//!
//! So an attempt now says **where** it failed ([`Fault`]). A `Fault::Remote` —
//! the dial, the advertisement, the send-pack, the standby's own refusal — is
//! the transient case above and retries without limit. A `Fault::Local` — the
//! catalogue, the store handle, `refs`, the object selection, `emit_pack` — is
//! this box failing to read its own repository, and it is bounded at
//! [`MAX_LOCAL_ATTEMPTS`]. On the last one the job leaves the queue and the
//! target is marked `needs_full_resync` (the module's existing name for *"a hole
//! retrying will not close"* — one implementation, not a second flag), `last_error` carries
//! the store's own sentence prefixed by how many attempts bought nothing, and a
//! `replication.stalled` row says all of it once instead of every five minutes
//! forever.
//!
//! Nothing about that is final. A stall is cleared by **new information**: the
//! repository's next push re-enqueues it, and so does
//! [`Replicator::resync`]. What is gone is the futile loop, not the copy.
//!
//! # What this module may and may not link — and it IS on the appliance now
//!
//! This module used to be `#[cfg(feature = "client")]`, and that header used to
//! say: *"a real gap … closing it means extracting send-pack (which needs only
//! `gunnar-pktline`, `gunnar-pack` and `gix-hash`) out of the porcelain half of
//! `gunnar-client`, which is a change to a Cargo manifest and to
//! `tests/feature_split.rs`, not to this file."*
//!
//! **That is exactly what happened on 2026-08-30, and this file did not
//! change.** `gunnar-client 1.1.4` and `nornir-git 0.1.9` each moved their gix
//! porcelain behind a default-on `porcelain` feature; `gunnar-server` gained a
//! `replication` feature that takes the WIRE half of gunnar-client alone —
//! send-pack, the russh dialler, and `plan_ref_sync` — and the appliance row in
//! `.nornir/build-variants.toml` names it. MEASURED, `cargo tree -p
//! gunnar-server -e normal`, unique packages:
//!
//! ```text
//!   appliance                          365
//!   appliance + replication            369    +4  gix-packetline, gunnar-client,
//!                                                 nornir-git, nornir-hash
//!   appliance + the old `client`       493  +128  gix, ~57 gix-*, and the
//!                                                 reqwest/rustls/icu tail
//! ```
//!
//! What is still forbidden here is the thing that ban was always about: **gix
//! porcelain**. `xtask guard` bans the `gix` facade crate by name in an `S`
//! graph, and `tests/feature_split.rs` asserts the appliance links no `gix`
//! *while linking `gunnar-client`* — because an absence proved over a graph with
//! no client edge in it proves nothing. So this module may say `gunnar_client::`
//! and `nornir_git::ssh`, and may not open a repository.

use std::collections::{BTreeMap, HashMap, VecDeque};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use anyhow::{Context, Result};
use gunnar_core::RepoRef;
use gunnar_grpc::{EventBus, EventFilter};
use gunnar_proto::{event, EventKind, RefUpdate, RepoPath};
use serde::{Deserialize, Serialize};

use crate::replication_view::RefCoverage;
use crate::stores::ServerStores;

/// The tracing target every line in this module is on.
pub const TARGET_REPLICATION: &str = "gunnar::replication";

/// Where a server's replication configuration lives, under the data root.
///
/// A JSON file rather than a table in the control repository, for the same
/// reason `gunnar mirror`'s registry is one: it is a handful of rows describing
/// where copies go, an operator must be able to read it with `cat` during an
/// incident, and the repositories on disk are the real data.
pub const REPLICATION_CONFIG: &str = "replication.json";

/// The `known_hosts` file the replicator uses when a target does not name one.
///
/// Under the data root and **not** `~/.ssh/known_hosts`: a server's identity
/// decisions must not depend on whose home directory it happens to run as.
pub const REPLICATION_KNOWN_HOSTS: &str = "replication_known_hosts";

/// How many distinct repositories may have work outstanding for one target.
/// See the module docs for why this number and why refusing is the right
/// answer at it.
pub const DEFAULT_QUEUE_CAPACITY: usize = 1024;

/// The first retry delay. Doubles per consecutive failure.
pub const DEFAULT_RETRY_BASE: Duration = Duration::from_secs(2);

/// The retry delay never grows past this. Five minutes: long enough that a
/// standby down overnight costs a few dozen attempts rather than a few
/// hundred thousand, short enough that one comes back within five minutes of
/// the mirror being reachable again.
pub const DEFAULT_RETRY_MAX: Duration = Duration::from_secs(300);

/// How many times a **local** failure is retried before the job stops.
///
/// Three, not one: a store handle can lose a race with a repository that is
/// being created or reopened, and one bad moment must not stall a standby. Four
/// or forty would buy the same nothing the drill's 21 minutes bought — the call is
/// deterministic in the store's own state, so if it failed three times with
/// nothing else changing it will fail the fourth time too. See [`Fault`].
pub const MAX_LOCAL_ATTEMPTS: u32 = 3;

/// How long an idle worker waits on its `Notify` before looking again.
///
/// A backstop, not a poll interval: an enqueue wakes the worker immediately.
/// It exists so a notification lost to a race costs an hour of staleness rather
/// than a standby that never moves again.
const IDLE_WAIT: Duration = Duration::from_secs(3600);

/// **How often a standby that is refusing connections is probed.**
///
/// Five seconds. It is the answer to a measured number: after
/// a standby outage, 40 tiny pushes took **4 minutes 59 seconds** to catch up,
/// and that wait was bounded by [`DEFAULT_RETRY_MAX`] rather than by the bytes.
/// The backoff is right for a standby that is *there and failing* — it stops a
/// broken peer being hammered — and wrong for one that has come *back*, because
/// the condition it is waiting out has already ended.
///
/// A TCP connect every five seconds to one address is nothing. It is only done
/// while a target is already failing and its next attempt is further away than
/// this, so a healthy replicator never opens a probe at all.
pub const RECONNECT_PROBE: Duration = Duration::from_secs(5);

/// How long a reachability probe waits for the connect itself.
const PROBE_TIMEOUT: Duration = Duration::from_secs(2);

// ───────────────────────────────────────────────────────────────────────────
// configuration
// ───────────────────────────────────────────────────────────────────────────

/// Which repositories a target takes (replication: "per repository or per
/// namespace").
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "scope", rename_all = "snake_case")]
pub enum ReplicationScope {
    /// Every repository on this server.
    Server,
    /// Every repository in one namespace (multi-tenancy).
    Namespace {
        /// The namespace.
        namespace: String,
    },
    /// Exactly one repository.
    Repo {
        /// Its namespace.
        namespace: String,
        /// Its name.
        repo: String,
    },
}

impl ReplicationScope {
    /// Does this scope take `path`?
    pub fn covers(&self, path: &RepoRef) -> bool {
        match self {
            ReplicationScope::Server => true,
            ReplicationScope::Namespace { namespace } => path.namespace() == namespace,
            ReplicationScope::Repo { namespace, repo } => {
                path.namespace() == namespace && path.name() == repo
            }
        }
    }
}

/// One standby.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ReplicationTarget {
    /// The operator's name for it. Appears in every event, log line and status
    /// row, so an incident names *which* standby rather than a URL.
    pub name: String,
    /// What it takes.
    #[serde(flatten)]
    pub scope: ReplicationScope,
    /// Where it is. Either a **base** — `ssh://git@standby:2222` — onto which
    /// `/<namespace>/<repo>.git` is appended, or, for a `repo`-scoped target, a
    /// complete URL ending in `.git`.
    pub remote: String,
    /// The private key this server authenticates to the standby with.
    ///
    /// A server has no `ssh-agent` and no interactive user; a deploy key on
    /// disk is what it has. Absent, the push falls back to
    /// `gunnar-client`'s key ladder, which on a server is almost never what was
    /// meant, so a target without one warns at start-up.
    #[serde(default)]
    pub identity: Option<PathBuf>,
    /// Override the `known_hosts` file for this target.
    #[serde(default)]
    pub known_hosts: Option<PathBuf>,
    /// Refuse an unknown host key instead of trusting it on first use.
    ///
    /// Off by default so a standby can be brought up without a provisioning
    /// step; on is the right setting for anything that outlives an afternoon.
    #[serde(default)]
    pub strict_host_key: bool,
    /// **Which repositories, by visibility** (the browse replica). `All`
    /// is a standby: every covered repository. `PublicOnly` is the browse
    /// replica's feed: a `Private` repository is never enqueued for it, and one
    /// that flips Public → Private is DELETED there — so the replica physically
    /// holds no private data and its anonymous surface cannot leak what is not
    /// on it. Named `visibility` because [`Self::scope`] was already taken by
    /// the namespace/repository axis; `--scope public-only` on the CLI is the
    /// operator's spelling.
    ///
    /// `serde(default)` = `All`, so a `replication.json` written before this
    /// field existed reads as the standby it was.
    #[serde(default)]
    pub visibility: VisibilityScope,
    /// **The replica's control plane, `host:port`**, required for a
    /// `PublicOnly` target and refused by name when missing
    /// (`replication_wire::target_from_proto`): the Public → Private delete is
    /// `Repos.Delete{purge}` over this address, signed with [`Self::identity`]
    /// — the same deploy key the send-pack authenticates with, which the
    /// replica holds as a `Machine` with namespace admin because push-create
    /// already needs that. The same door brings a repository BACK: a push-create
    /// is sha1 by construction, so a repository of another object format is
    /// `Repos.Create`d here first, in its own format, and pushed after
    /// (2026-09-22). `None` for a plain standby, which never forgets — and
    /// which therefore cannot receive a sha256 repository it does not hold.
    #[serde(default)]
    pub control: Option<String>,
}

/// Which repositories a target carries, by visibility. See
/// [`ReplicationTarget::visibility`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
pub enum VisibilityScope {
    /// Every covered repository — a standby.
    #[default]
    All,
    /// Only `Public` repositories — a browse replica's feed.
    PublicOnly,
}

impl VisibilityScope {
    /// Does a target with this scope carry a repository of `visibility`?
    pub fn carries(self, visibility: gunnar_core::Visibility) -> bool {
        match self {
            VisibilityScope::All => true,
            VisibilityScope::PublicOnly => visibility == gunnar_core::Visibility::Public,
        }
    }
}

impl ReplicationTarget {
    /// **`host:port` to probe for reachability**, out of [`Self::remote`].
    ///
    /// Only the authority is taken, and a missing port becomes SSH's 22 — the
    /// same default `gunnar-client` connects on. `None` when the remote is not
    /// a shape this can read, which makes the reconnect wake simply not apply
    /// rather than guessing an address and probing something else.
    ///
    /// This is a REACHABILITY probe and never an authentication one: a TCP
    /// accept says the box is back and its listener is up, which is exactly
    /// the condition whose absence the backoff was waiting out. The real
    /// answer is still the next send-pack.
    pub fn probe_addr(&self) -> Option<String> {
        let after_scheme = self.remote.split("://").nth(1)?;
        let authority = after_scheme.split(['/', '?', '#']).next()?;
        let hostport = authority.rsplit('@').next()?;
        if hostport.is_empty() {
            return None;
        }
        // An IPv6 literal is bracketed; anything after the bracket is a port.
        if let Some(rest) = hostport.strip_prefix('[') {
            let (host, tail) = rest.split_once(']')?;
            let port = tail.strip_prefix(':').unwrap_or("22");
            return Some(format!("[{host}]:{port}"));
        }
        match hostport.split_once(':') {
            Some((host, port)) if !host.is_empty() && !port.is_empty() => {
                Some(format!("{host}:{port}"))
            }
            _ => Some(format!("{hostport}:22")),
        }
    }

    /// The URL this target's copy of `path` lives at.
    pub fn url_for(&self, path: &RepoRef) -> String {
        if self.remote.ends_with(".git") {
            return self.remote.clone();
        }
        format!(
            "{}/{}/{}.git",
            self.remote.trim_end_matches('/'),
            path.namespace(),
            path.name()
        )
    }
}

/// **Whether this box takes client writes.** A `Primary` serves
/// `git-receive-pack` as always; a `Standby` refuses a direct client push and
/// takes only the replication push from its primary.
///
/// It is the fact the target roster cannot express. `demote` clears the roster
/// — "I ship no second copy" — but a box with an empty roster is still, by
/// itself, a writable box: it will accept a `git push` and hold a write the
/// new primary never sees. That is the split-brain from the receiving side, and
/// the role is what closes it. Persisted in `replication.json` so it survives
/// the reboot after a power cut.
///
/// Default `Primary`, so a config written before this field existed — and every
/// box that never touches it — deserialises to the writable box it already was.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
pub enum Role {
    /// Takes client writes. The default and every pre-failover server.
    #[default]
    Primary,
    /// Read-only to clients; accepts only its primary's replication push.
    Standby,
    /// **The browse replica**: a [`Role::Standby`]
    /// that ALSO serves anonymous reads of `Public` repositories, and that
    /// REFUSES `Mirror.SetRole PRIMARY` and every control-plane mutator by
    /// name. A standby is a warm spare an operator may promote; a browse box
    /// must never be promoted by accident — not by hand, not by the passive
    /// heartbeat (`standby_watch` stops on it). Fed by a `PublicOnly` target,
    /// so it physically holds no private data. The way out is `SetRole
    /// STANDBY`, which only ever narrows what the box serves.
    Browse,
}

impl Role {
    /// Exactly [`Role::Standby`]. For "does this box refuse client writes"
    /// use [`Role::refuses_client_writes`], which a browse replica also does.
    pub fn is_standby(self) -> bool {
        matches!(self, Role::Standby)
    }

    /// Exactly [`Role::Browse`].
    pub fn is_browse(self) -> bool {
        matches!(self, Role::Browse)
    }

    /// **Is a direct client `git push` refused on this box?** True for a
    /// standby and for a browse replica; the primary's replication push (a
    /// `Machine`) is exempt in both — see `gunnar_ssh::Control::write_frozen_for`.
    pub fn refuses_client_writes(self) -> bool {
        matches!(self, Role::Standby | Role::Browse)
    }

    /// The proto spelling, one place (one implementation).
    pub fn to_proto(self) -> gunnar_proto::ReplicationRole {
        match self {
            Role::Primary => gunnar_proto::ReplicationRole::Primary,
            Role::Standby => gunnar_proto::ReplicationRole::Standby,
            Role::Browse => gunnar_proto::ReplicationRole::Browse,
        }
    }

    /// The proto's answer, one place. An unknown value is `Primary`: the safe
    /// direction — a garbled request makes a box writable, never silently
    /// read-only (`Mirror.SetRole` says the same).
    pub fn from_proto(role: gunnar_proto::ReplicationRole) -> Role {
        match role {
            gunnar_proto::ReplicationRole::Primary => Role::Primary,
            gunnar_proto::ReplicationRole::Standby => Role::Standby,
            gunnar_proto::ReplicationRole::Browse => Role::Browse,
        }
    }
}

/// Everything `gunnar serve` needs to know about replication.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ReplicationConfig {
    /// **This box's role**: a writable `Primary`, a read-only `Standby` or a
    /// read-only, anonymous-serving, never-promotable `Browse` replica.
    ///
    /// `#[serde(default)]` = `Primary`, so an old `replication.json` with no
    /// `role` key is read as the writable box it was written by.
    #[serde(default)]
    pub role: Role,
    /// The standbys.
    #[serde(default)]
    pub targets: Vec<ReplicationTarget>,
    /// Distinct repositories that may be outstanding per target.
    #[serde(default = "default_queue_capacity")]
    pub queue_capacity: usize,
    /// First retry delay, milliseconds.
    #[serde(default = "default_retry_base_ms")]
    pub retry_base_ms: u64,
    /// Retry delay ceiling, milliseconds.
    #[serde(default = "default_retry_max_ms")]
    pub retry_max_ms: u64,
    /// **How long a [`RefCoverage`] walk is reused**, milliseconds. Zero is
    /// "walk every time", which is what a test that changes a repository's refs
    /// and then asks about them needs.
    #[serde(default = "default_ref_coverage_ttl_ms")]
    pub ref_coverage_ttl_ms: u64,
    /// **Do not subscribe to the bus.** Never read from the file (it is
    /// `serde(skip)`), and never a mode an operator can reach.
    ///
    /// It exists for one reason: measure, never assert. The headline test of this feature
    /// asserts that a push to one gunnar appears on a second, and a test like
    /// that is only worth anything if it can be made to fail *by breaking the
    /// subscriber specifically*. `replication.rs`'s
    /// `nothing_replicates_when_the_subscriber_is_not_started` is that red
    /// half, kept in the suite rather than performed once and described.
    #[serde(skip)]
    pub subscriber_disabled: bool,
}

fn default_queue_capacity() -> usize {
    DEFAULT_QUEUE_CAPACITY
}
fn default_retry_base_ms() -> u64 {
    DEFAULT_RETRY_BASE.as_millis() as u64
}
fn default_retry_max_ms() -> u64 {
    DEFAULT_RETRY_MAX.as_millis() as u64
}
fn default_ref_coverage_ttl_ms() -> u64 {
    REF_COVERAGE_TTL.as_millis() as u64
}

impl Default for ReplicationConfig {
    fn default() -> Self {
        ReplicationConfig {
            role: Role::Primary,
            targets: Vec::new(),
            queue_capacity: DEFAULT_QUEUE_CAPACITY,
            retry_base_ms: default_retry_base_ms(),
            retry_max_ms: default_retry_max_ms(),
            ref_coverage_ttl_ms: default_ref_coverage_ttl_ms(),
            subscriber_disabled: false,
        }
    }
}

impl ReplicationConfig {
    /// Read `<root>/replication.json`, or the empty configuration when it is
    /// absent.
    ///
    /// A missing file is not an error — most servers replicate nowhere — but a
    /// **malformed** one is. A server that silently replicated nowhere because
    /// a comma was wrong would be the failure this whole module is written to
    /// prevent, arriving before the first push.
    pub fn load(root: impl AsRef<Path>) -> Result<Self> {
        let path = root.as_ref().join(REPLICATION_CONFIG);
        let text = match std::fs::read_to_string(&path) {
            Ok(text) => text,
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
                return Ok(ReplicationConfig::default())
            }
            Err(err) => {
                return Err(err).with_context(|| format!("reading {}", path.display()));
            }
        };
        // **A power cut during `save` used to leave this file ZERO bytes**:
        // the old `save` wrote a temp file and renamed it with no fsync, so on
        // xfs the rename's directory entry was journaled while the data blocks
        // were not, and an interrupted save left `replication.json` present but
        // empty. Measured on the appliance KVM 2026-08-31: a 0-byte file here
        // failed the parse below, and that error unwound `Server::start` past a
        // live control-plane runtime — which, as PID 1, is the `Attempted to
        // kill init` kernel panic. `save` is fixed to fsync, but a box crashed
        // by an OLDER build still has the 0-byte file, and it must come up
        // SERVING rather than refuse to boot on it. An empty — or all-whitespace
        // — file is unambiguously that interrupted-save artifact and never
        // something an operator or `serde_json` wrote (the smallest thing `save`
        // emits is a `{ … }` object), so treat it as "no replication configured"
        // and let the next `apply` rewrite it. A NON-empty malformed file is a
        // different thing — an operator's typo — and stays a hard error.
        if text.trim().is_empty() {
            tracing::warn!(
                target: TARGET_REPLICATION,
                event = "replication.config_empty",
                path = %path.display(),
                "replication.json is empty — the signature of a power cut during a save. \
                 Treating it as no replication configured and coming up serving; it is \
                 rewritten on the next apply. Reconfigure standbys over the control plane if \
                 this box had any."
            );
            return Ok(ReplicationConfig::default());
        }
        serde_json::from_str(&text).with_context(|| {
            format!(
                "parsing {}; gunnar will not start replicating nowhere because of a syntax error",
                path.display()
            )
        })
    }

    /// Write the configuration back, atomically.
    ///
    /// **Durable rename.** The fsync-before-rename dance is shared with
    /// `webhooks.json` through [`crate::durable::write_atomic`] (one implementation), because
    /// the two files have the exact same power-cut hazard and one subtly wrong
    /// copy is what left a 0-byte `replication.json` panicking PID 1 on boot.
    pub fn save(&self, root: impl AsRef<Path>) -> Result<()> {
        let text = serde_json::to_string_pretty(self)?;
        crate::durable::write_atomic(root.as_ref(), REPLICATION_CONFIG, text.as_bytes())
    }

    /// [`ReplicationConfig::save`], **and a copy on the slow disk**.
    ///
    /// # Why the roster is the first thing that goes off-stripe
    ///
    /// `/data` is four disks striped with no redundancy: one member dies and
    /// all of it is gone, repositories and `replication.json` together. The
    /// roster is what says where the copies of those repositories are — so on
    /// the one morning it is worth anything, it is on the disk that just died.
    /// It is under 100 MB (it is under 100 *bytes* on most boxes), which makes
    /// it the cheapest thing on the whole list and the only one that changes
    /// what happens when something breaks.
    ///
    /// This is the mirrored form and [`ReplicationConfig::save`] is the plain
    /// one; the SERVER always uses this. The plain form stays because a caller
    /// with a bare path and no server (a test, a tool) has no slow root to give
    /// and must not be made to invent one.
    pub fn save_mirrored(
        &self,
        root: impl AsRef<Path>,
        slow: &crate::slow::SlowRoot,
    ) -> Result<()> {
        let text = serde_json::to_string_pretty(self)?;
        crate::durable::write_atomic_mirrored(
            root.as_ref(),
            slow,
            REPLICATION_CONFIG,
            text.as_bytes(),
        )
    }

    /// Add or replace a target by name (`Mirror.SetReplication`'s body).
    pub fn set_target(&mut self, target: ReplicationTarget) {
        match self.targets.iter_mut().find(|t| t.name == target.name) {
            Some(existing) => *existing = target,
            None => self.targets.push(target),
        }
    }

    /// Remove a target by name. True when one was there.
    pub fn remove_target(&mut self, name: &str) -> bool {
        let before = self.targets.len();
        self.targets.retain(|t| t.name != name);
        before != self.targets.len()
    }

    /// **Drop every target at once — demotion.** Returns how many were cleared.
    ///
    /// One write, so a recovered old-primary stops being a primary atomically
    /// rather than one `remove_target` at a time (which on a box with a long
    /// roster leaves it half a primary in between). Idempotent: an empty roster
    /// clears to the same place and returns 0.
    pub fn clear_targets(&mut self) -> usize {
        let cleared = self.targets.len();
        self.targets.clear();
        cleared
    }

    fn retry_base(&self) -> Duration {
        Duration::from_millis(self.retry_base_ms.max(1))
    }

    fn retry_max(&self) -> Duration {
        Duration::from_millis(self.retry_max_ms.max(self.retry_base_ms.max(1)))
    }
}

// ───────────────────────────────────────────────────────────────────────────
// counters
// ───────────────────────────────────────────────────────────────────────────

/// The counters, read.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ReplicationCounters {
    /// Jobs that entered a queue.
    pub enqueued: u64,
    /// Enqueues that landed on a repository already pending, and cost nothing.
    /// The number that says the bound is doing its job.
    pub coalesced: u64,
    /// Enqueues refused because a target's queue was full.
    pub refused_queue_full: u64,
    /// Pushes attempted, including retries.
    pub attempts: u64,
    /// Attempts that the standby accepted.
    pub succeeded: u64,
    /// Attempts that failed. **Never resets.**
    pub failed: u64,
    /// References actually moved on a standby.
    pub refs_replicated: u64,
    /// Times the event-stream subscription fell behind and lost events.
    pub bus_gaps: u64,
    /// Jobs that STOPPED because the two ends cannot hold the same repository
    /// — today, differing object formats. **Never resets.** Non-zero means a
    /// named repository has no second copy and no push will give it one; the
    /// fix is on the standby, and then `replication resync`.
    pub stopped_incompatible: u64,
    /// Jobs that STOPPED because this server could not read its own repository
    /// [`MAX_LOCAL_ATTEMPTS`] times running. **Never resets**, and a non-zero
    /// value is an operator's cue that a standby is missing a repository for a
    /// reason no amount of waiting will fix.
    pub stalled_local: u64,
    /// Copies a `PublicOnly` target DELETED because the repository went
    /// Private (the browse replica). Counted apart from `succeeded` so a
    /// forget is never read as a sync.
    pub forgotten: u64,
    /// Jobs DROPPED because the repository they name was deleted. Counted
    /// apart from `failed`, because they are not failures: the work stopped
    /// existing. A non-zero value beside a zero `stalled_local` is the shape
    /// this counter was added for — deletes are happening and they are no
    /// longer stalling anybody.
    pub cancelled: u64,
}

#[derive(Debug, Default)]
struct Counters {
    enqueued: AtomicU64,
    coalesced: AtomicU64,
    refused_queue_full: AtomicU64,
    attempts: AtomicU64,
    succeeded: AtomicU64,
    failed: AtomicU64,
    refs_replicated: AtomicU64,
    bus_gaps: AtomicU64,
    stopped_incompatible: AtomicU64,
    stalled_local: AtomicU64,
    /// Copies a `PublicOnly` target DELETED because the repository went
    /// Private (the browse replica).
    forgotten: AtomicU64,
    /// Jobs dropped because the repository was deleted.
    cancelled: AtomicU64,
}

impl Counters {
    fn snapshot(&self) -> ReplicationCounters {
        let read = |c: &AtomicU64| c.load(Ordering::Relaxed);
        ReplicationCounters {
            enqueued: read(&self.enqueued),
            coalesced: read(&self.coalesced),
            refused_queue_full: read(&self.refused_queue_full),
            attempts: read(&self.attempts),
            succeeded: read(&self.succeeded),
            failed: read(&self.failed),
            refs_replicated: read(&self.refs_replicated),
            bus_gaps: read(&self.bus_gaps),
            stopped_incompatible: read(&self.stopped_incompatible),
            stalled_local: read(&self.stalled_local),
            forgotten: read(&self.forgotten),
            cancelled: read(&self.cancelled),
        }
    }
}

// ───────────────────────────────────────────────────────────────────────────
// the queue
// ───────────────────────────────────────────────────────────────────────────

/// One repository, awaiting convergence onto one target.
#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct JobKey {
    /// The namespace.
    pub namespace: String,
    /// The repository.
    pub repo: String,
}

impl JobKey {
    fn path(&self) -> Result<RepoRef> {
        RepoRef::new(&self.namespace, &self.repo).with_context(|| {
            format!(
                "{}/{} is not a legal repository path",
                self.namespace, self.repo
            )
        })
    }

    fn proto(&self) -> RepoPath {
        RepoPath::new(&self.namespace, &self.repo)
    }
}

impl std::fmt::Display for JobKey {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}/{}", self.namespace, self.repo)
    }
}

#[derive(Debug, Clone)]
struct Job {
    key: JobKey,
    /// How many times this job has already failed. 0 for a fresh one.
    attempt: u32,
    /// Not before this instant. The backoff.
    ready_at: Instant,
    /// **When this repository first became outstanding for this target**, and
    /// it survives every requeue and every coalesce.
    ///
    /// `ready_at` is a backoff and moves forward on each failure; `attempt`
    /// resets when a push re-creates a coalesced entry. Neither can answer "how
    /// long has this standby been behind", and that is the only question a dead
    /// standby answers differently from a healthy one within the first minute.
    first_seen: Instant,
}

/// What [`TargetRunner::enqueue`] did. Returned rather than logged so the
/// caller, and a test, can tell the three apart.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Enqueued {
    /// A new pending job.
    Accepted,
    /// The repository was already pending. Nothing was added, and nothing was
    /// lost: the pending job will push whatever the repository holds when it
    /// runs.
    Coalesced,
    /// The queue is at its bound. The job was **not** taken and the target is
    /// now marked as needing a full resync.
    RefusedQueueFull,
}

/// The bound, and the coalescing that makes it hold.
#[derive(Debug)]
struct Queue {
    capacity: usize,
    order: VecDeque<JobKey>,
    jobs: HashMap<JobKey, Job>,
}

impl Queue {
    fn new(capacity: usize) -> Self {
        Queue {
            capacity: capacity.max(1),
            order: VecDeque::new(),
            jobs: HashMap::new(),
        }
    }

    fn len(&self) -> usize {
        self.jobs.len()
    }

    /// The instant the longest-outstanding job first became outstanding.
    fn oldest_first_seen(&self) -> Option<Instant> {
        self.jobs.values().map(|job| job.first_seen).min()
    }

    /// Every outstanding job, in queue order, emptying this queue.
    fn drain_jobs(&mut self) -> Vec<Job> {
        let mut out = Vec::with_capacity(self.jobs.len());
        while let Some(key) = self.order.pop_front() {
            if let Some(job) = self.jobs.remove(&key) {
                out.push(job);
            }
        }
        self.jobs.clear();
        out
    }

    /// Take jobs a previous queue was holding. Silently drops what does not
    /// fit — the caller marks the target as needing a full resync when it does,
    /// because a dropped job is a repository that would otherwise wait for a
    /// push that may never come.
    ///
    /// Returns how many did not fit.
    fn adopt(&mut self, jobs: Vec<Job>) -> usize {
        let mut lost = 0;
        for job in jobs {
            if self.jobs.contains_key(&job.key) {
                continue;
            }
            if self.jobs.len() >= self.capacity {
                lost += 1;
                continue;
            }
            self.order.push_back(job.key.clone());
            self.jobs.insert(job.key.clone(), job);
        }
        lost
    }

    /// **Forget one outstanding job**, for a repository that has been deleted.
    ///
    /// The ordering entry is left to [`Self::take_ready`], which already skips
    /// a key with no job behind it ("stale ordering entry for a job already
    /// taken") — one rule for a dangling name, not two.
    ///
    /// Returns whether there was anything to forget.
    fn remove(&mut self, key: &JobKey) -> bool {
        self.jobs.remove(key).is_some()
    }

    /// Take a job, or say why not.
    fn push(&mut self, key: JobKey, ready_at: Instant) -> Enqueued {
        if let Some(existing) = self.jobs.get_mut(&key) {
            // The pending job already means "converge this repository", which
            // covers whatever just happened to it. Pull its readiness forward,
            // though: a repository being actively pushed to should not sit
            // behind an old backoff.
            if ready_at < existing.ready_at {
                existing.ready_at = ready_at;
            }
            return Enqueued::Coalesced;
        }
        if self.jobs.len() >= self.capacity {
            return Enqueued::RefusedQueueFull;
        }
        self.order.push_back(key.clone());
        self.jobs.insert(
            key.clone(),
            Job {
                key,
                attempt: 0,
                ready_at,
                first_seen: ready_at,
            },
        );
        Enqueued::Accepted
    }

    /// Put a failed job back, with its next attempt's delay already applied.
    /// Re-queueing never hits the bound: the job was already counted.
    fn requeue(&mut self, job: Job) {
        if self.jobs.contains_key(&job.key) {
            // A push arrived while this attempt was in flight and re-created
            // the entry. That entry is newer; keep it.
            return;
        }
        self.order.push_back(job.key.clone());
        self.jobs.insert(job.key.clone(), job);
    }

    /// **Pull every pending job's backoff to `now`.**
    ///
    /// Used by the reconnect wake: when a standby that was refusing
    /// connections starts accepting them again, the reason every job is
    /// waiting has gone away, and the wait itself is an artefact of the
    /// failure rather than a property of the work.
    ///
    /// Returns how many jobs were pulled forward, so the log line says
    /// something applied rather than "woke up".
    fn ready_now(&mut self, now: Instant) -> usize {
        let mut moved = 0;
        for job in self.jobs.values_mut() {
            if job.ready_at > now {
                job.ready_at = now;
                moved += 1;
            }
        }
        moved
    }

    /// The first job whose backoff has elapsed, or the instant to wake at.
    fn take_ready(&mut self, now: Instant) -> std::result::Result<Job, Option<Instant>> {
        let mut rotated = 0;
        let mut earliest: Option<Instant> = None;
        while rotated < self.order.len() {
            let key = self.order[0].clone();
            let ready_at = match self.jobs.get(&key) {
                Some(job) => job.ready_at,
                None => {
                    // Stale ordering entry for a job already taken.
                    self.order.pop_front();
                    continue;
                }
            };
            if ready_at <= now {
                self.order.pop_front();
                return Ok(self.jobs.remove(&key).expect("just looked it up"));
            }
            earliest = Some(match earliest {
                Some(e) if e <= ready_at => e,
                _ => ready_at,
            });
            self.order.rotate_left(1);
            rotated += 1;
        }
        Err(earliest)
    }
}

/// The delay before attempt `attempt + 1`, given `attempt` consecutive
/// failures.
///
/// `base · 2^(attempt−1)`, capped at `max`. Exposed and pure so the doubling
/// can be asserted by value rather than by watching a clock. A backoff test
/// that only measured "the second attempt came later" would pass against a
/// constant delay.
pub fn backoff(attempt: u32, base: Duration, max: Duration) -> Duration {
    // One implementation, in `crate::durable`, shared with the webhook worker
    // (one implementation). Kept as a re-exporting wrapper so this module's public API and
    // its red-first `backoff_doubles_and_then_stops_at_the_cap` test are
    // unchanged.
    crate::durable::backoff(attempt, base, max)
}

/// **Resolve when `addr` accepts a TCP connection.** Never returns otherwise.
///
/// Only ever raced against a sleep, so "never returns" is bounded by whatever
/// it is racing. It probes every [`RECONNECT_PROBE`] with a
/// [`PROBE_TIMEOUT`] on the connect, so a host that black-holes packets costs
/// one socket rather than a stuck task.
///
/// A DNS name that does not resolve is a failure like any other and is retried;
/// a standby named by a hostname that comes back with the box is the ordinary
/// case, not an exception.
async fn reachable_again(addr: &str) {
    loop {
        tokio::time::sleep(RECONNECT_PROBE).await;
        if let Ok(Ok(stream)) =
            tokio::time::timeout(PROBE_TIMEOUT, tokio::net::TcpStream::connect(addr)).await
        {
            drop(stream);
            return;
        }
    }
}

// ───────────────────────────────────────────────────────────────────────────
// per-target state
// ───────────────────────────────────────────────────────────────────────────

/// What one standby looks like right now: the body of
/// `Mirror.ReplicationStatus`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TargetStatus {
    /// The target's name.
    pub name: String,
    /// Where it is.
    pub remote: String,
    /// **Repositories with work outstanding, INCLUDING the one an attempt is
    /// in flight for.**
    ///
    /// 🔴 It used to be the queue length alone, and a job the worker had picked
    /// up is not on the queue. MEASURED 2026-08-31: a standby hard-powered-off
    /// with a push outstanding read `pending 0  failures 0` for the whole of
    /// the kernel's connect timeout — every number on the row said healthy.
    pub pending: usize,
    /// The repository an attempt is in flight for, if any.
    pub in_flight: Option<String>,
    /// How long the OLDEST outstanding repository has been outstanding.
    /// `None` when nothing is outstanding.
    pub outstanding_for: Option<Duration>,
    /// Consecutive failures for the job at the head of the queue. 0 when the
    /// last attempt succeeded.
    pub consecutive_failures: u32,
    /// The last failure's message, if there has been one since the last
    /// success.
    pub last_error: Option<String>,
    /// When a push last landed on this standby.
    pub last_success_unix_ms: Option<i64>,
    /// The standby has a hole retrying will not close. See the module docs.
    pub needs_full_resync: bool,
}

/// **What one repository's second copy looks like, on one standby.**
///
/// The answer to *"does `alice/edda` exist on `standby-fi-hel1`?"*, which
/// nothing on the control plane could give before: `ReplicationStandby` carries
/// aggregates only, and an aggregate that says `synced 105 · failed 0` while 21
/// of 25 repositories have no second copy is the exact shape of alarm this
/// module exists to refuse — one an admin reads as "I am safe".
///
/// # It is held IN MEMORY, and the wire says so
///
/// This is what the running server has observed. A restart resets it to *"no
/// record"* for every repository, which under-claims rather than over-claims —
/// the safe direction — and `ReplicationStatusResponse::observed_since_unix_ms`
/// carries the instant the record begins so a console can say *"no record since
/// the server started at 05:12"* rather than the false *"never replicated"*.
/// Persisting it is a separate change and does not move this shape.
#[derive(Debug, Clone, Default)]
struct RepoRecord {
    /// When a push to this standby last landed for this repository. 0 = never
    /// observed.
    last_success_unix_ms: i64,
    /// How many mirrored refs the standby held after that success — the number
    /// to compare against the primary's own ref count.
    refs_at_last_success: u64,
    /// The last failure since the last success.
    last_error: Option<String>,
    /// Retrying cannot help: the two ends cannot hold the same repository, or
    /// this server cannot read its own. A human has to act.
    stopped: bool,
}

/// **One repository's second copy on one standby**, joined against the
/// catalogue so that a repository which has NEVER been replicated has a row.
///
/// That join is the point: a list built from what the replicator has done can
/// only ever show successes, and the question an admin is asking is which
/// repositories are *missing*.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RepoCopy {
    /// The standby.
    pub standby: String,
    /// The repository's namespace.
    pub namespace: String,
    /// The repository.
    pub repo: String,
    /// 0 = this server has no record of a successful replication.
    pub last_success_unix_ms: i64,
    /// Mirrored refs the standby held at that success.
    pub refs_at_last_success: u64,
    /// Work is outstanding for it right now — queued, or an attempt in flight.
    pub outstanding: bool,
    /// The last failure since the last success. Empty = none.
    pub last_error: String,
    /// Retrying cannot help. See [`RepoRecord::stopped`].
    pub stopped: bool,
    /// This server holds it as **sha1**, and a standby PUSH-CREATES sha256.
    pub sha1: bool,
    /// **What a mirror of this repository does NOT carry.** `None` when this
    /// server could not read its own refs — which is a different answer from
    /// zero and must not be printed as one.
    pub refs: Option<RefCoverage>,
}

/// **The default** for [`ReplicationConfig::ref_coverage_ttl_ms`]: how long a
/// [`RefCoverage`] walk is reused before it is taken again.
///
/// Counting refs means opening every repository's reference database, and
/// `Admin.Stats` is a page the console can poll. One walk a minute is
/// invisible against a number measured in refs that change on a push, and it is
/// what keeps a status call from being an unbounded cost on a server with a
/// thousand repositories.
pub const REF_COVERAGE_TTL: Duration = Duration::from_secs(60);

#[derive(Debug, Default)]
struct TargetState {
    consecutive_failures: u32,
    last_error: Option<String>,
    last_success_unix_ms: Option<i64>,
    needs_full_resync: bool,
    /// The job the worker has taken off the queue and is pushing right now,
    /// with the instant it first became outstanding. See [`TargetStatus::pending`].
    in_flight: Option<(JobKey, Instant)>,
}

/// One standby's queue, worker and state.
struct TargetRunner {
    target: ReplicationTarget,
    queue: Mutex<Queue>,
    state: Mutex<TargetState>,
    /// What this standby has been observed to hold, per repository. Keyed by
    /// repository, so it is bounded by the catalogue rather than by traffic;
    /// an entry for a repository that has since been deleted is simply never
    /// joined against and costs a `JobKey`.
    history: Mutex<HashMap<JobKey, RepoRecord>>,
    wake: tokio::sync::Notify,
    counters: Arc<Counters>,
    retry_base: Duration,
    retry_max: Duration,
    bus: Arc<EventBus>,
}

impl TargetRunner {
    fn lock_queue(&self) -> std::sync::MutexGuard<'_, Queue> {
        self.queue.lock().unwrap_or_else(|p| p.into_inner())
    }

    fn lock_state(&self) -> std::sync::MutexGuard<'_, TargetState> {
        self.state.lock().unwrap_or_else(|p| p.into_inner())
    }

    fn lock_history(&self) -> std::sync::MutexGuard<'_, HashMap<JobKey, RepoRecord>> {
        self.history.lock().unwrap_or_else(|p| p.into_inner())
    }

    /// **Drop this target's outstanding job for a repository that is gone**,
    /// and the per-repository row that went with it.
    ///
    /// Never blocks and never fails: a job that is not queued is already
    /// cancelled, and one already in flight is caught by the worker's own
    /// [`Fault::Gone`] arm — a delete that lands mid-attempt cannot be undone
    /// from here, and does not need to be.
    ///
    /// The history row goes too. It is keyed by repository, it is what
    /// `Mirror.ReplicationStatus` lists per repository, and a row for a
    /// repository that no longer exists is a line an operator would have to
    /// learn to ignore.
    fn cancel(&self, key: &JobKey) -> bool {
        let dropped = self.lock_queue().remove(key);
        self.lock_history().remove(key);
        if dropped {
            self.counters.cancelled.fetch_add(1, Ordering::Relaxed);
        }
        dropped
    }

    /// Offer this target a repository. Never blocks; never fails.
    fn enqueue(&self, key: JobKey) -> Enqueued {
        let outcome = self.lock_queue().push(key.clone(), Instant::now());
        match outcome {
            Enqueued::Accepted => {
                self.counters.enqueued.fetch_add(1, Ordering::Relaxed);
                self.wake.notify_one();
            }
            Enqueued::Coalesced => {
                self.counters.coalesced.fetch_add(1, Ordering::Relaxed);
                self.wake.notify_one();
            }
            Enqueued::RefusedQueueFull => {
                self.counters
                    .refused_queue_full
                    .fetch_add(1, Ordering::Relaxed);
                self.lock_state().needs_full_resync = true;
                tracing::error!(
                    target: TARGET_REPLICATION,
                    event = "replication.queue_full",
                    standby = %self.target.name,
                    repo = %key,
                    capacity = self.lock_queue().capacity,
                    "the replication queue for this standby is full; the repository was NOT \
                     queued and the standby now needs a full resync. Retrying will not close \
                     this hole; run `gunnar replication resync`."
                );
                self.fail_event(
                    &key,
                    0,
                    Duration::ZERO,
                    "the replication queue is full",
                    true,
                );
            }
        }
        outcome
    }

    /// Publish a `ReplicationFailed` on the event stream.
    fn fail_event(
        &self,
        key: &JobKey,
        attempt: u32,
        retry_in: Duration,
        reason: &str,
        needs_full_resync: bool,
    ) {
        self.bus.publish(event::Payload::ReplicationFailed(
            gunnar_proto::ReplicationFailed {
                repo: Some(key.proto()),
                target: self.target.name.clone(),
                remote_url: key
                    .path()
                    .map(|p| self.target.url_for(&p))
                    .unwrap_or_else(|_| self.target.remote.clone()),
                attempt,
                retry_in_ms: retry_in.as_millis() as u64,
                reason: reason.to_owned(),
                needs_full_resync,
            },
        ));
    }

    fn status(&self) -> TargetStatus {
        let (queued, queued_oldest) = {
            let queue = self.lock_queue();
            (queue.len(), queue.oldest_first_seen())
        };
        let state = self.lock_state();
        let now = Instant::now();
        let oldest = match (queued_oldest, state.in_flight.as_ref()) {
            (Some(q), Some((_, f))) => Some(q.min(*f)),
            (Some(q), None) => Some(q),
            (None, Some((_, f))) => Some(*f),
            (None, None) => None,
        };
        TargetStatus {
            name: self.target.name.clone(),
            remote: self.target.remote.clone(),
            pending: queued + usize::from(state.in_flight.is_some()),
            in_flight: state.in_flight.as_ref().map(|(key, _)| key.to_string()),
            outstanding_for: oldest.map(|at| now.saturating_duration_since(at)),
            consecutive_failures: state.consecutive_failures,
            last_error: state.last_error.clone(),
            last_success_unix_ms: state.last_success_unix_ms,
            needs_full_resync: state.needs_full_resync,
        }
    }
}

// ───────────────────────────────────────────────────────────────────────────
// the replicator
// ───────────────────────────────────────────────────────────────────────────

/// **What one standby's backfill queued**, per [`Replicator::resync_scoped`].
///
/// Counted rather than summed into one number because the three outcomes mean
/// different things to an operator: `queued` is new work, `already_pending` is
/// work that was already going to happen and cost nothing, and `refused` is a
/// hole the call did **not** close.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ResyncReport {
    /// The standby.
    pub standby: String,
    /// Repositories this target's scope covers, of those considered.
    pub in_scope: u64,
    /// Newly queued.
    pub queued: u64,
    /// Already outstanding; nothing was added and nothing was lost.
    pub already_pending: u64,
    /// Refused by the queue bound. Non-zero means the backfill is INCOMPLETE.
    pub refused: u64,
    /// Repositories in scope this server holds as **sha1**. See
    /// [`Fault::Incompatible`].
    pub sha1_repos: Vec<String>,
    /// Repositories the caller excluded that this target covers. Reported
    /// rather than silently dropped: an exclusion that matched nothing is a
    /// typo, and it produces the same command with a very different outcome.
    pub excluded: Vec<String>,
    /// `Private` repositories this `PublicOnly` target covers and, by design,
    /// did not queue (the browse replica). Zero for a standby.
    pub private_skipped: u64,
}

/// A [`RefCoverage`] walk and when it was taken.
type CoverageWalk = (Instant, Arc<BTreeMap<String, RefCoverage>>);

/// The replicator: one subscriber, one queue and one worker per standby.
pub struct Replicator {
    stores: Arc<ServerStores>,
    bus: Arc<EventBus>,
    config: ReplicationConfig,
    runners: Vec<Arc<TargetRunner>>,
    counters: Arc<Counters>,
    /// **When this server's per-repository replication record begins.**
    ///
    /// [`RepoRecord`] is in memory, so a row saying "never replicated" may mean
    /// "not since this instant" rather than "not ever". A console that could
    /// not tell those apart would print the second sentence about a standby
    /// holding a perfectly good copy, which is the same species of lie as an
    /// aggregate that reads healthy over 21 missing repositories.
    observed_since_unix_ms: i64,
    /// The last [`RefCoverage`] walk and when it was taken. See
    /// [`REF_COVERAGE_TTL`].
    coverage: Mutex<Option<CoverageWalk>>,
}

/// The tasks [`Replicator::spawn`] started, aborted when this is dropped.
pub struct ReplicationTasks {
    tasks: Vec<tokio::task::JoinHandle<()>>,
}

impl ReplicationTasks {
    /// Stop the subscriber and every worker.
    pub fn abort(&self) {
        for task in &self.tasks {
            task.abort();
        }
    }

    /// How many tasks are running. A `spawn` that started nothing is a
    /// replicator that will never move a byte, and it must be visible.
    pub fn len(&self) -> usize {
        self.tasks.len()
    }

    /// True when nothing was started.
    pub fn is_empty(&self) -> bool {
        self.tasks.is_empty()
    }
}

impl Drop for ReplicationTasks {
    fn drop(&mut self) {
        self.abort();
    }
}

impl Replicator {
    /// Build a replicator. Nothing runs until [`Replicator::spawn`].
    pub fn new(
        stores: Arc<ServerStores>,
        bus: Arc<EventBus>,
        config: ReplicationConfig,
    ) -> Arc<Self> {
        Self::new_carrying(stores, bus, config, None)
    }

    /// **Build a replicator that inherits what the last one had done.**
    ///
    /// 🔴 THE ALARM THAT LIED, measured on the reference appliance 2026-08-31: re-adding
    /// an existing standby **only to turn on `--strict-host-key`** put its row
    /// back to `synced 0 · last success NEVER · STALE` while the standby was
    /// fully in sync. `SetReplication` is idempotent by name, so an operator
    /// changing one flag is the ordinary case — and it made a healthy standby
    /// print the console's alarm string, which is worse than no alarm at all
    /// because it teaches an operator to ignore the one that matters.
    ///
    /// The cause was structural rather than a bug in a line: the counters and
    /// every target's state live on the `Replicator`, and `apply` tears the
    /// whole object down and builds a new one. So the fix is here, at
    /// construction, where the previous object is still in hand.
    ///
    /// What carries, and the rule that decides it — **the NAME carries the
    /// work, the REMOTE carries the history**:
    ///
    /// * the server-wide counters always, because `failed` is documented as
    ///   never resetting and an operator reading it after a reconfigure is
    ///   asking the same question as before it;
    /// * a target's **queue** whenever the name survives: a pending job means
    ///   "converge this repository", which is true of whatever URL it now
    ///   points at, and dropping it during a reconfigure is a repository
    ///   silently waiting for a push;
    /// * a target's **success history** only when the name AND the remote both
    ///   survive. A changed URL is a *different standby*, and `last success
    ///   3 s ago` about a box that has never been dialled is exactly the lie
    ///   this function exists to stop — so that case resets the history and
    ///   sets `needs_full_resync`, which is the truth about a new peer.
    pub fn new_carrying(
        stores: Arc<ServerStores>,
        bus: Arc<EventBus>,
        config: ReplicationConfig,
        previous: Option<&Replicator>,
    ) -> Arc<Self> {
        let counters = match previous {
            Some(previous) => Arc::clone(&previous.counters),
            None => Arc::new(Counters::default()),
        };
        let retry_base = config.retry_base();
        let retry_max = config.retry_max();
        let runners = config
            .targets
            .iter()
            .map(|target| {
                if target.identity.is_none() {
                    tracing::warn!(
                        target: TARGET_REPLICATION,
                        event = "replication.no_identity",
                        standby = %target.name,
                        "this replication target names no identity file, so the push will fall \
                         back to the client key ladder (~/.ssh). A server should authenticate to \
                         its standby with an explicit deploy key."
                    );
                }
                let mut queue = Queue::new(config.queue_capacity);
                let mut state = TargetState::default();
                let mut history: HashMap<JobKey, RepoRecord> = HashMap::new();
                if let Some(old) = previous
                    .iter()
                    .flat_map(|p| p.runners.iter())
                    .find(|r| r.target.name == target.name)
                {
                    let same_peer = old.target.remote == target.remote;
                    let mut carried = old.lock_queue().drain_jobs();
                    let mut old_state = old.lock_state();
                    // The previous worker was aborted mid-attempt, so its
                    // in-flight repository is on no queue anywhere. Put it back
                    // or a reconfigure during a backfill silently drops it.
                    if let Some((key, first_seen)) = old_state.in_flight.take() {
                        carried.push(Job {
                            key,
                            attempt: 0,
                            ready_at: Instant::now(),
                            first_seen,
                        });
                    }
                    let lost = queue.adopt(carried);
                    state = TargetState {
                        needs_full_resync: old_state.needs_full_resync
                            || lost > 0
                            || !same_peer,
                        ..TargetState::default()
                    };
                    if same_peer {
                        state.consecutive_failures = old_state.consecutive_failures;
                        state.last_error = old_state.last_error.clone();
                        state.last_success_unix_ms = old_state.last_success_unix_ms;
                        // What this box was observed to hold, repository by
                        // repository — the console's "which of these have a
                        // second copy". Same rule as the aggregate above: the
                        // NAME carries the work, the REMOTE carries the history.
                        history = old.lock_history().clone();
                    } else {
                        tracing::warn!(
                            target: TARGET_REPLICATION,
                            event = "replication.retargeted",
                            standby = %target.name,
                            was = %old.target.remote,
                            now = %target.remote,
                            "this standby's name was kept and its URL changed, so it is a                              DIFFERENT box: its success history is cleared rather than carried,                              and it needs a full resync"
                        );
                    }
                }
                Arc::new(TargetRunner {
                    target: target.clone(),
                    queue: Mutex::new(queue),
                    state: Mutex::new(state),
                    history: Mutex::new(history),
                    wake: tokio::sync::Notify::new(),
                    counters: Arc::clone(&counters),
                    retry_base,
                    retry_max,
                    bus: Arc::clone(&bus),
                })
            })
            .collect();
        Arc::new(Replicator {
            stores,
            bus,
            config,
            runners,
            counters,
            // A reconfigure does not restart the observation: the previous
            // replicator's per-repository history is carried into this one for
            // every target whose peer did not change.
            observed_since_unix_ms: previous
                .map(|p| p.observed_since_unix_ms)
                .unwrap_or_else(gunnar_grpc::bus::now_unix_ms),
            // Not carried. A reconfigure is cheap and the walk behind this is
            // a fact about the repositories rather than about a target, so
            // taking it again costs one walk and cannot report the old scope.
            coverage: Mutex::new(None),
        })
    }

    /// ★ **Refs this server holds that a mirror does not carry**, per
    /// repository, keyed `namespace/repo`. See [`RefCoverage`].
    ///
    /// **`max_age` is the caller's, because the two callers are not asking the
    /// same question.** An operator running `replication list` wants the
    /// number for the push they just made and passes [`Duration::ZERO`]; a
    /// console polling `Admin.Stats` wants a cheap answer and passes
    /// [`ReplicationConfig::ref_coverage_ttl_ms`]. The walk opens every
    /// repository's reference database, so a polled surface paying for one on
    /// every frame is how a status page becomes a load.
    ///
    /// A repository this server cannot read is **absent** from the map rather
    /// than present with a zero, because those are different answers and the
    /// row prints them differently.
    pub fn ref_coverage_by_repo(&self, max_age: Duration) -> Arc<BTreeMap<String, RefCoverage>> {
        let mut slot = match self.coverage.lock() {
            Ok(slot) => slot,
            Err(poisoned) => poisoned.into_inner(),
        };
        if let Some((taken, coverage)) = slot.as_ref() {
            if !max_age.is_zero() && taken.elapsed() < max_age {
                return Arc::clone(coverage);
            }
        }
        let fresh = Arc::new(self.walk_ref_coverage());
        *slot = Some((Instant::now(), Arc::clone(&fresh)));
        fresh
    }

    /// The server-wide sum of [`Self::ref_coverage_by_repo`] — the number the
    /// summary line and the console's replication panel print, off the same
    /// walk as the per-repository rows (one implementation).
    pub fn ref_coverage(&self, max_age: Duration) -> RefCoverage {
        let mut total = RefCoverage::default();
        for coverage in self.ref_coverage_by_repo(max_age).values() {
            total.add(*coverage);
        }
        total
    }

    /// How stale the POLLED surfaces may let this number get. See the field.
    pub fn ref_coverage_max_age(&self) -> Duration {
        Duration::from_millis(self.config.ref_coverage_ttl_ms)
    }

    /// **The last ref-coverage walk's server-wide sum, or zeros — and NEVER a
    /// walk of its own.**
    ///
    /// This is what the POLLED surfaces read (`Admin.Stats`, the appliance
    /// band): they draw on a timer, and a reader that walked would be the exact
    /// "re-open every store every 60 s from a polling caller" the cold-status
    /// measurement is about. So
    /// they peek the cache and no more. The walk that fills the cache is driven
    /// ONLY on demand — `replication list --refs` / `Mirror.ReplicationStatus`
    /// with `include_refs` — so the number a console shows is the last one an
    /// operator asked for, refreshed the moment anyone asks again, and no
    /// surface silently runs up the store-open count to keep it warm.
    ///
    /// Zeros before the first walk. `refs_total == 0` is how every printer
    /// already spells "no ref-coverage figure to show", so an un-warmed cache
    /// reads as "not computed yet", not as "all mirrored".
    pub fn ref_coverage_cached(&self) -> RefCoverage {
        let slot = match self.coverage.lock() {
            Ok(slot) => slot,
            Err(poisoned) => poisoned.into_inner(),
        };
        let mut total = RefCoverage::default();
        if let Some((_taken, coverage)) = slot.as_ref() {
            for c in coverage.values() {
                total.add(*c);
            }
        }
        total
    }

    /// The walk behind [`Self::ref_coverage_by_repo`].
    fn walk_ref_coverage(&self) -> BTreeMap<String, RefCoverage> {
        use gunnar_client::mirror::is_mirrored_ref;
        use gunnar_store::git_storage_trait::GitOps;

        let repos = match self.stores.catalog().list(None) {
            Ok(repos) => repos,
            Err(err) => {
                tracing::warn!(
                    target: TARGET_REPLICATION,
                    %err,
                    "could not list repositories to count the refs a mirror does not carry"
                );
                return BTreeMap::new();
            }
        };
        let mut out = BTreeMap::new();
        for repo in repos {
            let path = repo.path();
            let key = format!("{}/{}", path.namespace(), path.name());
            let rows = match self.stores.repo_store(&repo).and_then(|store| {
                let git = store.serve();
                GitOps::refs(git).context("reading this repository's refs")
            }) {
                Ok(rows) => rows,
                Err(err) => {
                    // Absent, not zero: "we could not look" and "there are
                    // none outside refs/heads and refs/tags" are opposite
                    // answers to an operator asking what is not replicated.
                    tracing::warn!(
                        target: TARGET_REPLICATION,
                        repo = %key,
                        %err,
                        "could not count this repository's refs; its unmirrored count is \
                         reported as unknown rather than as zero"
                    );
                    continue;
                }
            };
            let mut coverage = RefCoverage::default();
            for row in rows {
                coverage.total += 1;
                if !is_mirrored_ref(&row.name) {
                    coverage.unmirrored += 1;
                }
            }
            out.insert(key, coverage);
        }
        out
    }

    /// The configuration this replicator was built from.
    pub fn config(&self) -> &ReplicationConfig {
        &self.config
    }

    /// When this server's per-repository record begins. See the field.
    pub fn observed_since_unix_ms(&self) -> i64 {
        self.observed_since_unix_ms
    }

    /// The counters, now.
    pub fn counters(&self) -> ReplicationCounters {
        self.counters.snapshot()
    }

    /// Every standby's state, for `Mirror.ReplicationStatus`.
    pub fn status(&self) -> Vec<TargetStatus> {
        self.runners.iter().map(|r| r.status()).collect()
    }

    /// ★ **WHICH REPOSITORIES HAVE A SECOND COPY**, one row per repository per
    /// standby that covers it.
    ///
    /// The question the control plane could not answer. `ReplicationStandby`
    /// carries aggregates — `synced`, `failed`, `pending`, `last success` — and
    /// on the reference appliance on 2026-08-31 those aggregates read healthy while **21
    /// of 25 repositories had no second copy at all**. An admin reading that
    /// console concludes they are safe. This is the row that makes it possible
    /// to say otherwise, and to say it per repository.
    ///
    /// # It is driven from the CATALOGUE, not from what the replicator has done
    ///
    /// That is the load-bearing half. A list assembled from successes can only
    /// ever show repositories that HAVE a copy, and the question is which ones
    /// do not. So the catalogue is the outer loop, every repository each
    /// target's scope covers gets a row, and a repository the replicator has
    /// never touched appears with `last_success_unix_ms: 0` — the same "0 means
    /// NEVER" the standby row already uses.
    ///
    /// The record behind it is in memory (see [`RepoRecord`]), so a restarted
    /// server answers "no record" for repositories the standby really does
    /// hold. That under-claims, which is the safe direction, and the caller
    /// publishes the instant the record begins so a console can say which of
    /// the two it is looking at.
    pub fn repo_copies(
        &self,
        coverage: Option<&BTreeMap<String, RefCoverage>>,
    ) -> Result<Vec<RepoCopy>> {
        let repos = self
            .stores
            .catalog()
            .list(None)
            .context("listing repositories to report their second copies")?;
        // **The ref-coverage join is the CALLER's**, and it is `None` unless the
        // operator asked for it. Counting the refs a mirror does not carry means
        // opening every repository's reference database, and on a cold box that
        // is one refold per repository — the O(all repos × cold-store
        // open) a plain status call must never be. So the expensive
        // walk is opt-in and happens ONCE in the caller, which passes the map in
        // here; `None` leaves every row's `refs` absent, which the wire and the
        // console already print as "not asked / could not read" rather than as
        // zero. The copy STATE below — synced/queued/never, the verdict, the
        // pending clock — is the operational signal and it costs no store open.
        let empty = BTreeMap::new();
        let coverage = coverage.unwrap_or(&empty);
        let mut out = Vec::new();
        for runner in &self.runners {
            let outstanding: std::collections::HashSet<JobKey> = {
                let queue = runner.lock_queue();
                let state = runner.lock_state();
                queue
                    .jobs
                    .keys()
                    .cloned()
                    .chain(state.in_flight.iter().map(|(key, _)| key.clone()))
                    .collect()
            };
            let history = runner.lock_history();
            for repo in &repos {
                let path = repo.path();
                if !runner.target.scope.covers(&path) {
                    continue;
                }
                let key = JobKey {
                    namespace: path.namespace().to_owned(),
                    repo: path.name().to_owned(),
                };
                let record = history.get(&key).cloned().unwrap_or_default();
                out.push(RepoCopy {
                    standby: runner.target.name.clone(),
                    refs: coverage
                        .get(&format!("{}/{}", key.namespace, key.repo))
                        .copied(),
                    namespace: key.namespace.clone(),
                    repo: key.repo.clone(),
                    last_success_unix_ms: record.last_success_unix_ms,
                    refs_at_last_success: record.refs_at_last_success,
                    outstanding: outstanding.contains(&key),
                    last_error: record.last_error.unwrap_or_default(),
                    stopped: record.stopped,
                    sha1: repo.hash_kind == gunnar_core::HashKind::Sha1,
                });
            }
        }
        Ok(out)
    }

    /// Offer one repository to every target whose scope covers it.
    ///
    /// Public because the subscriber is not the only caller: a resync and a
    /// future `Mirror.Sync` push both want exactly this, and a second copy of
    /// the scope match would be a second thing to keep in step (one implementation).
    pub fn enqueue(&self, path: &RepoRef) -> Vec<(String, Enqueued)> {
        let key = JobKey {
            namespace: path.namespace().to_owned(),
            repo: path.name().to_owned(),
        };
        // **The visibility filter** (the browse replica): a `PublicOnly` target is never
        // offered a `Private` repository. Read once per push, not per target;
        // a repository the catalog cannot find is treated as Private, the
        // direction that leaks nothing. A `PublicOnly` runner with nothing to
        // carry is simply not in the answer — the same as a scope miss.
        let visibility = self.visibility_of(path);
        self.runners
            .iter()
            .filter(|r| r.target.scope.covers(path))
            .filter(|r| r.target.visibility.carries(visibility))
            .map(|r| (r.target.name.clone(), r.enqueue(key.clone())))
            .collect()
    }

    /// **Public → Private: make every `PublicOnly` target FORGET `path`.**
    ///
    /// The other half of the visibility filter. `enqueue` keeps a private
    /// repository off a browse replica from now on; this removes the copy it
    /// already holds. Only `PublicOnly` targets are told — a standby keeps
    /// everything by definition. The job is the same queue entry a push makes
    /// (coalescing with one already outstanding); the WORKER decides at take
    /// time whether to ship or forget by reading the catalog then, so a flip
    /// and a flip-back that race the queue settle on the truth at execution.
    pub fn enqueue_forget(&self, path: &RepoRef) -> Vec<(String, Enqueued)> {
        let key = JobKey {
            namespace: path.namespace().to_owned(),
            repo: path.name().to_owned(),
        };
        self.runners
            .iter()
            .filter(|r| r.target.scope.covers(path))
            .filter(|r| r.target.visibility == VisibilityScope::PublicOnly)
            .map(|r| (r.target.name.clone(), r.enqueue(key.clone())))
            .collect()
    }

    /// **A DELETED repository: drop the work that was queued for it**, on
    /// every target that covers it, whatever its visibility scope.
    ///
    /// # The defect, measured 2026-09-20 on the live estate
    ///
    /// ```text
    /// 12:34 repo deleted        proof/f13-1789907688270 the repository was deleted
    /// 12:34 replication failed  proof/f13-1789907688270 is not in the catalogue, so it
    ///                           cannot be replicated — a full resync is needed
    /// ```
    ///
    /// A fixture created a repository, a push enqueued it, the fixture deleted
    /// it, and the queue still held a job naming it. The worker took that job,
    /// [`replicate_once`] asked the catalog for a row that no longer exists,
    /// and the miss was classified [`Fault::Local`] — "this box cannot read its
    /// own repository" — which retries [`MAX_LOCAL_ATTEMPTS`] times and then
    /// STALLS the standby with the sticky `needs_full_resync`. So a fixture
    /// that cleaned up after itself correctly put the replication subsystem
    /// into a state demanding a full resync of every repository on the box.
    ///
    /// A deleted repository is not a repository this server failed to read. It
    /// is work that no longer exists, and the answer to work that no longer
    /// exists is to drop it, not to retry it five times and raise an alarm.
    ///
    /// Unlike [`enqueue_forget`](Self::enqueue_forget), this is **not**
    /// restricted to `PublicOnly` targets: the stuck job was on a full standby,
    /// and a full standby's queue holds exactly the same moot entry. Nothing is
    /// deleted on the standby by this call — whether a full standby should drop
    /// its copy of a deleted repository is a separate ruling, and cancelling
    /// the local queue entry does not pre-empt it. Today a full standby keeps
    /// what it was given, which is what "standby" has always meant here.
    ///
    /// Returns, per target, whether a job was actually dropped.
    pub fn cancel(&self, path: &RepoRef) -> Vec<(String, bool)> {
        let key = JobKey {
            namespace: path.namespace().to_owned(),
            repo: path.name().to_owned(),
        };
        self.runners
            .iter()
            .filter(|r| r.target.scope.covers(path))
            .map(|r| (r.target.name.clone(), r.cancel(&key)))
            .collect()
    }

    /// The catalog's word on `path`'s visibility, Private when it has none.
    fn visibility_of(&self, path: &RepoRef) -> gunnar_core::Visibility {
        match self.stores.catalog().get(path) {
            Ok(Some(repo)) => repo.visibility,
            Ok(None) => gunnar_core::Visibility::Private,
            Err(err) => {
                tracing::warn!(
                    target: TARGET_REPLICATION,
                    repo = %path,
                    %err,
                    "the catalog could not be read for a replication visibility check; \
                     treating the repository as Private"
                );
                gunnar_core::Visibility::Private
            }
        }
    }

    /// Enqueue **every** catalogued repository each target covers.
    ///
    /// This is what closes a `needs_full_resync` hole, and the flag is cleared
    /// only for targets whose whole set fitted; a resync that itself hit the
    /// bound has not resynced anything.
    ///
    /// The event-bus-gap path's spelling of [`Self::resync_scoped`]: every
    /// standby, every repository. One writer (one implementation) — the operator's
    /// `Mirror.ResyncReplication` and a lost event take exactly the same code
    /// through the same queue, so a backfill cannot behave differently from the
    /// gap recovery that has been exercised since replication landed.
    pub fn resync(&self) -> Result<usize> {
        Ok(self
            .resync_scoped(None, None, &[])?
            .iter()
            .map(|report| report.queued)
            .sum::<u64>() as usize)
    }

    /// **BACKFILL: queue what a standby covers, whether or not anybody pushes
    /// to it again.**
    ///
    /// # The gap this closes, measured
    ///
    /// Replication is a subscriber to the push bus, so a standby is a copy of
    /// what is pushed *after* it exists. MEASURED on the reference appliance
    /// 2026-08-31: a target added against 22 existing repositories left the
    /// standby holding **0**; a repository created while no target existed was
    /// still absent 60 s after one was added; and one further commit to it made
    /// the standby hold its *whole* history. So a repository's history rides
    /// along on its NEXT push, and a repository nobody pushes to again never
    /// arrives. `alice/linux-mirror` had **3818 refs on the primary and 0 on
    /// the standby**, and 21 of 25 repositories were in the same position.
    ///
    /// # What it does NOT do, and why that is the whole design
    ///
    /// It **enqueues**. It does not copy anything itself, does not open a
    /// repository, does not touch the push path and does not block. Every byte
    /// still moves on the existing worker, through the existing bound queue,
    /// under the existing backoff — which is what makes a backfill of 8 GB safe
    /// on a box that must keep serving git (the standing rule: nobody waits on
    /// background work). It is also why it is **cheap to repeat**: a job that
    /// is already outstanding coalesces, and a repository the standby already
    /// has costs one advertisement and an empty command list. Re-running it is
    /// the resume.
    ///
    /// * `standby` — `None` is every configured target. `Some(name)` is one, and
    ///   a name that is not configured is an **error**: "resynced" about a
    ///   standby that was never there reads exactly like "resynced" about the
    ///   one you meant.
    /// * `only` — `None` is every repository the target's scope covers.
    ///   `Some(path)` is one repository, which is how a repository too large to
    ///   ship tonight stays a separate decision from the other twenty-four.
    /// * `except` — the other half of that decision, and the one an operator
    ///   actually reaches for: *everything but this*. On the reference appliance 24 of
    ///   25 repositories are a few hundred megabytes together and one is ~8.7 GB
    ///   / 13.9 M objects that projects 182–235 GiB against a 224.9 GiB `/data`.
    ///   Ignored when `only` is given.
    ///
    /// `needs_full_resync` is cleared only by a call that covered a target's
    /// **whole** set and had nothing refused. A single-repository resync never
    /// clears it: one repository catching up says nothing about the one that
    /// was dropped.
    pub fn resync_scoped(
        &self,
        standby: Option<&str>,
        only: Option<&RepoRef>,
        except: &[RepoRef],
    ) -> Result<Vec<ResyncReport>> {
        if let Some(name) = standby {
            anyhow::ensure!(
                self.runners.iter().any(|r| r.target.name == name),
                "no replication target named {name:?}; configured: [{}]",
                self.runners
                    .iter()
                    .map(|r| r.target.name.as_str())
                    .collect::<Vec<_>>()
                    .join(", ")
            );
        }
        let repos = match only {
            None => self
                .stores
                .catalog()
                .list(None)
                .context("listing repositories for a replication resync")?,
            Some(path) => vec![self
                .stores
                .catalog()
                .get(path)
                .with_context(|| format!("looking {path} up for a replication resync"))?
                .with_context(|| {
                    format!("{path} is not a repository on this server, so it cannot be resynced")
                })?],
        };
        let mut out = Vec::new();
        for runner in &self.runners {
            if standby.is_some_and(|name| runner.target.name != name) {
                continue;
            }
            let mut report = ResyncReport {
                standby: runner.target.name.clone(),
                ..ResyncReport::default()
            };
            for repo in &repos {
                let path = repo.path();
                if !runner.target.scope.covers(&path) {
                    continue;
                }
                if only.is_none()
                    && except
                        .iter()
                        .any(|e| e.namespace() == path.namespace() && e.name() == path.name())
                {
                    report.excluded.push(path.to_string());
                    continue;
                }
                // **The visibility filter, on the backfill path too** (the
                // browse replica): a `PublicOnly` target's resync never queues a
                // Private repository as a SHIP. Counted, so the operator's
                // `in_scope` is what the target will actually hold.
                //
                // It queues it as a FORGET instead, and that is the difference
                // between a resync that derives visibility and one that trusts
                // a one-shot event. The `VisibilityChanged` event fires exactly
                // once; on a full queue it was dropped, and nothing afterwards
                // ever revisited the question, so a repository made Private
                // stayed readable on the browse replica with no state anywhere
                // saying so. Every Private repository in scope being offered as
                // a forget is derivable from the catalog alone — it survives a
                // restart, needs no history, and is what makes the resync a
                // repair rather than a retry.
                //
                // It is cheap to repeat: the worker re-reads the catalog at
                // take time and forgets, `forget_once` treats the replica's
                // `NotFound` as success, and a replica that never held the
                // repository answers one 404. It is not free — one RPC per
                // private repository in scope per full resync — and that is the
                // price of the guarantee.
                if !runner.target.visibility.carries(repo.visibility) {
                    report.private_skipped += 1;
                    let key = JobKey {
                        namespace: path.namespace().to_owned(),
                        repo: path.name().to_owned(),
                    };
                    match runner.enqueue(key) {
                        Enqueued::Accepted | Enqueued::Coalesced => {}
                        // Counted as a refusal so the resync does not clear
                        // `needs_full_resync` on the strength of a queue that
                        // would not take the one job that mattered.
                        Enqueued::RefusedQueueFull => report.refused += 1,
                    }
                    continue;
                }
                report.in_scope += 1;
                // **Named, not silently dropped.** A standby push-creates
                // sha256, so a sha1 repository replicates to nothing until it
                // is pre-created there with `--hash sha1`. It is still queued —
                // the operator may already have done it, and the attempt is the
                // only thing that can find out — but the operator is told which
                // ones the question applies to before the queue answers it.
                if repo.hash_kind == gunnar_core::HashKind::Sha1 {
                    report.sha1_repos.push(path.to_string());
                }
                let key = JobKey {
                    namespace: path.namespace().to_owned(),
                    repo: path.name().to_owned(),
                };
                match runner.enqueue(key) {
                    Enqueued::Accepted => report.queued += 1,
                    Enqueued::Coalesced => report.already_pending += 1,
                    Enqueued::RefusedQueueFull => report.refused += 1,
                }
            }
            tracing::info!(
                target: TARGET_REPLICATION,
                event = "replication.resync",
                standby = %runner.target.name,
                scope = %match only {
                    Some(path) => path.to_string(),
                    None => "every covered repository".to_owned(),
                },
                in_scope = report.in_scope,
                excluded = report.excluded.len(),
                queued = report.queued,
                already_pending = report.already_pending,
                refused = report.refused,
                sha1 = report.sha1_repos.len(),
                private_skipped = report.private_skipped,
                "a replication resync queued a standby's backfill"
            );
            // An exclusion is a hole this call deliberately did not close, so
            // it must not clear the flag either — same rule as `only`.
            if only.is_none() && report.refused == 0 && report.excluded.is_empty() {
                let mut state = runner.lock_state();
                if state.needs_full_resync {
                    tracing::info!(
                        target: TARGET_REPLICATION,
                        event = "replication.resync_complete",
                        standby = %runner.target.name,
                        "every covered repository was queued; the standby no longer needs a \
                         full resync"
                    );
                    state.needs_full_resync = false;
                }
            }
            out.push(report);
        }
        Ok(out)
    }

    /// Start the event-stream subscriber and one worker per standby.
    ///
    /// Must be called from inside a tokio runtime. Returns the handles; drop
    /// them and everything stops.
    pub fn spawn(self: &Arc<Self>) -> ReplicationTasks {
        let mut tasks = Vec::new();
        if self.runners.is_empty() {
            return ReplicationTasks { tasks };
        }
        for runner in &self.runners {
            let runner = Arc::clone(runner);
            let me = Arc::clone(self);
            // A32-14, stays a task: the replicator's workers; their handles are kept and aborted at its stop.
            tasks.push(tokio::spawn(async move { worker(me, runner).await }));
        }
        if self.config.subscriber_disabled {
            tracing::error!(
                target: TARGET_REPLICATION,
                event = "replication.subscriber_disabled",
                "the replication subscriber is DISABLED. Nothing will ever be enqueued from a \
                 push. This is a test seam and is not reachable from configuration."
            );
        } else {
            let me = Arc::clone(self);
            let bus = Arc::clone(&self.bus);
            // A32-14, stays a task: the replicator's bus subscriber; kept with the workers.
            tasks.push(tokio::spawn(async move { subscribe(me, bus).await }));
        }
        tracing::info!(
            target: TARGET_REPLICATION,
            event = "replication.started",
            standbys = self.runners.len(),
            queue_capacity = self.config.queue_capacity,
            tasks = tasks.len(),
            "replication is running"
        );
        ReplicationTasks { tasks }
    }
}

/// **The live replication slot: what makes a standby configurable on a box with
/// no shell.**
///
/// The 2026-08-30 drill measured replication working and then said the thing
/// that mattered: it was not in the appliance binary. Compiling it in
/// (`replication`, 2026-08-30) is only half an operator surface. The other half
/// is that the reference appliance is an OS-less image running as PID 1 — **no shell,
/// no `vi`, no `scp`** — so "edit `replication.json` and restart" is not a
/// procedure anybody can carry out. A target has to be settable over the
/// control plane, from a laptop, and it has to take effect without a reboot.
///
/// This is the object that makes that possible. It owns the replicator's whole
/// lifecycle:
///
/// * [`ReplicationSlot::config`] answers with the RUNNING configuration when
///   there is one and the on-disk file otherwise, so a `list` before any target
///   exists reads the file rather than inventing an empty answer;
/// * [`ReplicationSlot::apply`] writes [`REPLICATION_CONFIG`] **first**, then
///   swaps the running replicator. Disk first is deliberate: a box that
///   acknowledged `SetReplication` and then lost power must come back
///   replicating, and the only thing that survives a power cut is the file;
/// * dropping the previous [`ReplicationTasks`] aborts its workers, so a removed
///   target stops pushing rather than lingering as an orphaned task nobody can
///   see.
///
/// # The runtime it spawns on is not the one that called it
///
/// [`Replicator::spawn`] uses `tokio::spawn`, so it lands on whatever runtime is
/// current. A `SetReplication` arrives on the CONTROL runtime, and the workers
/// must not run there: replication does a blocking send-pack, and the control
/// plane is the surface an operator uses while a box is in trouble. The slot
/// therefore carries the git runtime's [`Handle`](tokio::runtime::Handle) and
/// enters it around the spawn, which is the same runtime `Server::start` used
/// and for the same reason.
///
/// # It is still a subscriber, and nothing here changes that
///
/// Every path below runs on the control plane or at startup. The push path does
/// not reach this type at all, so replication's hard requirement — a push that
/// succeeded locally has succeeded, whatever replication is doing — survives a
/// live reconfiguration untouched.
pub struct ReplicationSlot {
    stores: Arc<ServerStores>,
    bus: Arc<EventBus>,
    /// The runtime replication workers belong on. See the type's header.
    runtime: tokio::runtime::Handle,
    running: Mutex<Option<(Arc<Replicator>, ReplicationTasks)>>,
}

impl ReplicationSlot {
    /// An empty slot over `stores`, whose workers will run on `runtime`.
    pub fn new(
        stores: Arc<ServerStores>,
        bus: Arc<EventBus>,
        runtime: tokio::runtime::Handle,
    ) -> Arc<Self> {
        Arc::new(ReplicationSlot {
            stores,
            bus,
            runtime,
            running: Mutex::new(None),
        })
    }

    /// The running replicator, if any target is configured.
    pub fn replicator(&self) -> Option<Arc<Replicator>> {
        self.running
            .lock()
            .expect("the replication slot lock is not poisoned")
            .as_ref()
            .map(|(replicator, _)| Arc::clone(replicator))
    }

    /// **The configuration an operator is asking about.**
    ///
    /// The running one when a replicator exists; otherwise the file. Reading the
    /// file in the second case is the load-bearing half: a server with no
    /// targets runs no replicator at all (that is replication's "a server that
    /// replicates nowhere pays nothing"), and answering such a box with a
    /// default-constructed config would report `retry_max` and `queue_capacity`
    /// it does not actually have.
    pub fn config(&self) -> Result<ReplicationConfig> {
        if let Some(replicator) = self.replicator() {
            return Ok(replicator.config().clone());
        }
        ReplicationConfig::load(self.stores.root()).context("reading the replication configuration")
    }

    /// **Persist `config` and make it the running one.** Disk first.
    ///
    /// Returns the number of standby workers now running. Zero is a legitimate
    /// answer and means every target was removed: the replicator is torn down
    /// rather than left running with an empty target list, because a subscriber
    /// with nothing to send is a subscriber that can still lag the bus.
    pub fn apply(&self, config: ReplicationConfig) -> Result<usize> {
        // 1. DISK FIRST. A power cut between here and the swap loses a running
        //    replicator, which the next boot rebuilds from this file; a power
        //    cut the other way round would lose the operator's decision.
        //    A twin keeps no per-box mirror (C7, `crate::twin::forget_mirrors`): the file alone.
        let saved = if self.stores.is_twin() {
            config.save(self.stores.root())
        } else {
            config.save_mirrored(self.stores.root(), self.stores.slow())
        };
        saved.context("writing the replication configuration")?;

        // 2. The SHARED write-gate, from the SAME config that just hit disk, so
        //    the SSH/HTTP front doors freeze or thaw in lockstep with the role
        //    an operator wrote. Disk first (above) then memory here matches the
        //    roster's own order: a crash between the two rebuilds the gate from
        //    the file at the next boot (`ServerStores::open`).
        self.stores
            .writes_frozen()
            .store(config.role.refuses_client_writes(), std::sync::atomic::Ordering::SeqCst);
        // The browse flag the control plane's mutator refusal and the front
        // door's wording read: same order, same reason.
        self.stores
            .browse()
            .store(config.role.is_browse(), std::sync::atomic::Ordering::SeqCst);

        let mut running = self
            .running
            .lock()
            .expect("the replication slot lock is not poisoned");
        // Dropping the old ReplicationTasks aborts its workers (see its `Drop`).
        // Taken BEFORE the new one is built so two replicators are never
        // subscribed to the bus at the same time, which would double every push.
        // Kept, not dropped: the new replicator inherits this one's counters,
        // queues and per-target history. See `Replicator::new_carrying` — an
        // operator changing one flag on a healthy standby used to reset its row
        // to the console's alarm string.
        let previous = running.take().map(|(replicator, tasks)| {
            tasks.abort();
            replicator
        });
        if config.targets.is_empty() {
            tracing::info!(
                target: TARGET_REPLICATION,
                event = "replication.stopped",
                "the last replication target was removed; the replicator is stopped"
            );
            return Ok(0);
        }
        let replicator = Replicator::new_carrying(
            Arc::clone(&self.stores),
            Arc::clone(&self.bus),
            config,
            previous.as_deref(),
        );
        let tasks = {
            // See the type's header: the workers belong on the git runtime, not
            // on whichever runtime answered the RPC.
            let _guard = self.runtime.enter();
            // A32-14, stays a task: the replicator's workers on the git runtime; their handles are kept and stopped with it.
            replicator.spawn()
        };
        anyhow::ensure!(
            !tasks.is_empty(),
            "a replicator with standbys configured started no tasks; it would be a replication \
             that silently never runs, which is the failure replication is about"
        );
        let standbys = replicator.runners.len();
        *running = Some((replicator, tasks));
        Ok(standbys)
    }

    /// **Stop replicating**, without touching the file.
    ///
    /// Called from `Server::shutdown` and from `Server`'s `Drop`. The
    /// configuration deliberately stays on disk: a stopped server that comes
    /// back must come back replicating to the same standbys, and a shutdown that
    /// silently un-configured them would be a durability promise revoked by a
    /// reboot.
    pub fn stop(&self) {
        if let Some((_, tasks)) = self
            .running
            .lock()
            .expect("the replication slot lock is not poisoned")
            .take()
        {
            tasks.abort();
        }
    }
}

/// The event-stream subscriber. **This is the only coupling to the push path.**
async fn subscribe(replicator: Arc<Replicator>, bus: Arc<EventBus>) {
    use tokio_stream::StreamExt;

    // **`RepoDeleted` is in the list, and it was the hole.** A browse replica
    // is told about a repository that went Private and, until 2026-09-11, about
    // nothing at all when the repository was DELETED: `Repos.Delete` published
    // its event, this filter did not ask for it, and the arm below fell into
    // `_ => {}`. The primary forgot the repository and the public replica kept
    // serving it, indefinitely, with no state anywhere recording the
    // divergence. Deleting a repository is the strongest instruction an
    // operator can give about who may read it.
    let filter = EventFilter {
        kinds: vec![
            EventKind::RefUpdated,
            EventKind::RefDeleted,
            EventKind::VisibilityChanged,
            EventKind::RepoDeleted,
        ],
        ..EventFilter::default()
    };
    let mut stream = Box::pin(bus.subscribe(filter));
    while let Some(next) = stream.next().await {
        let event = match next {
            Ok(event) => event,
            Err(status) => {
                tracing::warn!(
                    target: TARGET_REPLICATION,
                    %status,
                    "the replication subscription yielded an error; continuing"
                );
                continue;
            }
        };
        match event.payload.as_ref() {
            Some(event::Payload::Gap(gap)) => {
                // Dropped events are dropped ref updates, and a ref update that
                // never reached this task is a reference that would silently
                // never replicate. Every covered repository is re-queued, and
                // the flag stays set if that did not fit.
                replicator.counters.bus_gaps.fetch_add(1, Ordering::Relaxed);
                for runner in &replicator.runners {
                    runner.lock_state().needs_full_resync = true;
                }
                tracing::error!(
                    target: TARGET_REPLICATION,
                    event = "replication.bus_gap",
                    dropped = gap.dropped,
                    "the replication subscriber fell behind the event bus and lost events; \
                     re-queueing every covered repository"
                );
                if let Err(err) = replicator.resync() {
                    tracing::error!(
                        target: TARGET_REPLICATION,
                        %err,
                        "the resync that a bus gap requires could not be started; the standbys \
                         remain marked as needing one"
                    );
                }
            }
            Some(event::Payload::RefUpdated(update)) => {
                enqueue_from(&replicator, update.repo.as_ref());
            }
            Some(event::Payload::RefDeleted(deleted)) => {
                enqueue_from(&replicator, deleted.repo.as_ref());
            }
            // **A visibility flip** (the browse replica). Public → Private: every
            // `PublicOnly` target must drop its copy. Private → Public: the
            // repository is offered as a push would offer it — `enqueue`
            // re-reads the catalog, sees Public, and the browse replica gets
            // it. `from` is not consulted: the catalog's present truth decides.
            Some(event::Payload::VisibilityChanged(changed)) => {
                let private = changed.to == gunnar_proto::Visibility::Private as i32;
                let Some(repo) = changed.repo.as_ref() else { continue };
                match RepoRef::new(&repo.namespace, &repo.repo) {
                    Ok(path) if private => {
                        forget_from(&replicator, &path, "a repository went Private");
                    }
                    Ok(_) => enqueue_from(&replicator, Some(repo)),
                    Err(err) => tracing::warn!(
                        target: TARGET_REPLICATION,
                        %err,
                        "a VisibilityChanged event named a path this server cannot resolve"
                    ),
                }
            }
            // **A deleted repository is a forget too**, and for a stronger
            // reason than a visibility flip: there is no catalog row left for a
            // later resync to derive anything from, so if this event is not
            // acted on, nothing on this box will ever mention the repository
            // again while the browse replica goes on serving it.
            Some(event::Payload::RepoDeleted(deleted)) => {
                let Some(repo) = deleted.repo.as_ref() else { continue };
                match RepoRef::new(&repo.namespace, &repo.repo) {
                    Ok(path) => {
                        // **First cancel, then forget.** Two different things
                        // happen to a deleted repository and they must happen
                        // in this order. The queued job to SEND it is moot on
                        // every target that covers it, whatever its visibility
                        // scope, and a moot job left in the queue is the one
                        // that asked for a full resync. The forget — telling a
                        // browse replica to delete its copy — is a NEW job, for
                        // `PublicOnly` targets only, and enqueuing it after the
                        // cancel is what keeps the cancel from eating it.
                        cancel_from(&replicator, &path, "a repository was deleted");
                        forget_from(&replicator, &path, "a repository was deleted");
                    }
                    Err(err) => tracing::warn!(
                        target: TARGET_REPLICATION,
                        %err,
                        "a RepoDeleted event named a path this server cannot resolve"
                    ),
                }
            }
            _ => {}
        }
    }
    tracing::warn!(
        target: TARGET_REPLICATION,
        "the event stream ended; replication is no longer being fed by pushes"
    );
}

/// **Tell every public-only replica to drop its copy**, and do not let the
/// instruction evaporate when a queue is full.
///
/// The forget used to be a single `enqueue_forget` whose outcome went into a
/// log line and nowhere else. `Enqueued::RefusedQueueFull` is a real answer on
/// a bounded queue, and on that answer the instruction was simply gone: no
/// retry, no flag, no second event ever — and a repository an operator made
/// Private stayed readable on the browse replica for ever. So a refusal raises
/// `needs_full_resync` on that runner, which is the same state a lost event
/// raises and the thing a resync exists to clear; `resync_scoped` now derives
/// the forgets from the catalog rather than waiting to be told again.
fn forget_from(replicator: &Arc<Replicator>, path: &RepoRef, why: &str) {
    for (standby, outcome) in replicator.enqueue_forget(path) {
        if outcome == Enqueued::RefusedQueueFull {
            if let Some(runner) = replicator.runners.iter().find(|r| r.target.name == standby) {
                runner.lock_state().needs_full_resync = true;
            }
            tracing::error!(
                target: TARGET_REPLICATION,
                event = "replication.forget_refused",
                %standby,
                repo = %path,
                "{why}, and the public-only replica's queue was full, so the instruction to \
                 forget it could not be queued. The standby is marked as needing a full \
                 resync, which is what re-derives the forget; until then the replica may \
                 still be serving this repository"
            );
            continue;
        }
        tracing::info!(
            target: TARGET_REPLICATION,
            event = "replication.forget_enqueued",
            %standby,
            repo = %path,
            outcome = ?outcome,
            "{why}; the public-only replica is told to forget it"
        );
    }
}

/// **Drop the outstanding work for a repository that no longer exists**, and
/// say per standby whether there was any. See [`Replicator::cancel`].
fn cancel_from(replicator: &Arc<Replicator>, path: &RepoRef, why: &str) {
    for (standby, dropped) in replicator.cancel(path) {
        if !dropped {
            continue;
        }
        tracing::info!(
            target: TARGET_REPLICATION,
            event = "replication.cancelled",
            %standby,
            repo = %path,
            "{why}; the replication job that was still queued for it is dropped rather than \
             left to fail five times and ask for a full resync"
        );
    }
}

fn enqueue_from(replicator: &Arc<Replicator>, repo: Option<&RepoPath>) {
    let Some(repo) = repo else { return };
    let path = match RepoRef::new(&repo.namespace, &repo.repo) {
        Ok(path) => path,
        Err(err) => {
            tracing::warn!(
                target: TARGET_REPLICATION,
                namespace = %repo.namespace,
                repo = %repo.repo,
                %err,
                "a ref event named a repository path this server cannot resolve; not replicated"
            );
            return;
        }
    };
    for (standby, outcome) in replicator.enqueue(&path) {
        tracing::debug!(
            target: TARGET_REPLICATION,
            event = "replication.enqueue",
            %standby,
            repo = %path,
            outcome = ?outcome,
            "a push enqueued a replication job"
        );
    }
}

/// One standby's drain loop.
async fn worker(replicator: Arc<Replicator>, runner: Arc<TargetRunner>) {
    loop {
        let now = Instant::now();
        // The guard is taken and released inside this block. Holding a
        // `std::sync::MutexGuard` across the `.await` below would make this
        // future non-`Send` and would not compile. That is the right
        // constraint; it is spelled out here so nobody widens the scope.
        let taken = {
            let mut queue = runner.lock_queue();
            queue.take_ready(now)
        };
        let job = match taken {
            Ok(job) => job,
            Err(earliest) => {
                // Sleep until the earliest backoff expires, or until something
                // is enqueued, whichever comes first. Never a poll loop.
                let wait = earliest
                    .map(|at| at.saturating_duration_since(now))
                    .unwrap_or(IDLE_WAIT);
                // **Wake on RECONNECT rather than sitting out the backoff.**
                //
                // MEASURED: after a standby outage, 40 tiny pushes
                // took 4 min 59 s to catch up, bounded by `DEFAULT_RETRY_MAX`
                // and not by the bytes. The backoff is the right answer for a
                // standby that is there and failing; it is the wrong one for a
                // standby that has come back, because the condition it is
                // waiting out has ended. So while a target is failing and its
                // next attempt is further off than a probe interval, race the
                // sleep against "does it accept a TCP connection yet".
                let failing = runner.lock_state().consecutive_failures > 0;
                let probe = failing
                    .then(|| runner.target.probe_addr())
                    .flatten()
                    .filter(|_| wait > RECONNECT_PROBE);
                match probe {
                    None => {
                        let _ = tokio::time::timeout(wait, runner.wake.notified()).await;
                    }
                    Some(addr) => {
                        tokio::select! {
                            _ = tokio::time::sleep(wait) => {}
                            _ = runner.wake.notified() => {}
                            _ = reachable_again(&addr) => {
                                let moved = runner.lock_queue().ready_now(Instant::now());
                                tracing::info!(
                                    target: TARGET_REPLICATION,
                                    event = "replication.reconnected",
                                    standby = %runner.target.name,
                                    addr = %addr,
                                    jobs = moved,
                                    "the standby is accepting connections again; {moved} \
                                     queued repository(ies) were pulled off the backoff \
                                     instead of waiting it out"
                                );
                            }
                        }
                    }
                }
                continue;
            }
        };

        runner.counters.attempts.fetch_add(1, Ordering::Relaxed);
        // **The job is outstanding from here until the attempt returns.** It is
        // no longer on the queue, so without this the row would say `pending 0`
        // for however long the attempt takes — which on a box that has been
        // powered off is the whole of the kernel's connect timeout.
        runner.lock_state().in_flight = Some((job.key.clone(), job.first_seen));
        let stores = Arc::clone(&replicator.stores);
        let target = runner.target.clone();
        let key = job.key.clone();
        let known_hosts_default = replicator.stores.root().join(REPLICATION_KNOWN_HOSTS);

        // send-pack is blocking. It runs on the blocking pool so a slow or
        // hung standby occupies a thread that git traffic never wanted.
        // **Ship or forget**, decided NOW from the catalog (the browse replica): a
        // `PublicOnly` target holding a job for a repository that is Private
        // at this instant deletes its copy instead of updating it. Decided at
        // take time rather than enqueue time so a flip that raced the queue
        // settles on the present truth, and so the same queue entry serves
        // both — there is no second queue to fall out of step.
        let forget = target.visibility == VisibilityScope::PublicOnly
            && !target
                .visibility
                .carries(replicator.visibility_of(&key.path().unwrap_or_else(|_| {
                    // An illegal path cannot be in the catalog; `replicate_once`
                    // below reports it as the local fault it is.
                    RepoRef::new("_", "_").expect("a legal placeholder")
                })));
        let outcome = if forget {
            forget_once(&target, &key).await.map(|()| Replicated {
                url: target.control.clone().unwrap_or_default(),
                updates: Vec::new(),
                up_to_date: 0,
                objects_sent: 0,
                pack_bytes: 0,
                forgotten: true,
            })
        } else {
            let attempt = {
                let stores = Arc::clone(&stores);
                let target = target.clone();
                let key = key.clone();
                let known_hosts_default = known_hosts_default.clone();
                move || {
                    replicate_once(&stores, &target, &key, &known_hosts_default)
                }
            };
            let first = tokio::task::spawn_blocking(attempt).await;
            match first {
                Err(join) => Err(AttemptError::from(anyhow::anyhow!(
                    "the replication task panicked: {join}"
                ))),
                // **The standby does not hold it, and a push cannot create it
                // in this object format.** The push-create prospect is sha1 by
                // construction (`push_create_spec`, 2026-09-20: git advertises
                // the object format before the client has said a word), so a
                // sha256 repository that the browse replica FORGOT — Public →
                // Private purges its copy — could never come back by itself.
                // MEASURED 2026-09-22 (`browse_replica.rs`, Private → Public):
                // `replication.incompatible … nordisk/gunnar is Sha256 but
                // browse is Sha1`, stopped, forever. So: create it there, in
                // its own format, over the control plane the forget already
                // uses, and push again. One extra round trip, once per return.
                Ok(Err(AttemptError {
                    create_first: Some(create),
                    ..
                })) => match create_once(&target, &key, *create).await {
                    Err(err) => Err(err),
                    Ok(()) => {
                        tracing::info!(
                            target: TARGET_REPLICATION,
                            event = "replication.precreated",
                            standby = %target.name,
                            repo = %key,
                            control = %target.control.clone().unwrap_or_default(),
                            "the standby did not hold the repository and a push-create would \
                             have been the wrong object format; it was created over the \
                             control plane in the repository's own format and the push \
                             follows"
                        );
                        let stores = Arc::clone(&stores);
                        let target = target.clone();
                        let key = key.clone();
                        match tokio::task::spawn_blocking(move || {
                            replicate_once(&stores, &target, &key, &known_hosts_default)
                        })
                        .await
                        {
                            Ok(outcome) => outcome,
                            Err(join) => Err(AttemptError::from(anyhow::anyhow!(
                                "the replication task panicked: {join}"
                            ))),
                        }
                    }
                },
                Ok(outcome) => outcome,
            }
        };

        match outcome {
            Ok(report) => {
                runner.counters.succeeded.fetch_add(1, Ordering::Relaxed);
                runner
                    .counters
                    .refs_replicated
                    .fetch_add(report.updates.len() as u64, Ordering::Relaxed);
                {
                    let mut state = runner.lock_state();
                    state.consecutive_failures = 0;
                    state.last_error = None;
                    state.last_success_unix_ms = Some(gunnar_grpc::bus::now_unix_ms());
                    state.in_flight = None;
                }
                runner.lock_history().insert(
                    job.key.clone(),
                    RepoRecord {
                        last_success_unix_ms: gunnar_grpc::bus::now_unix_ms(),
                        // Every mirrored ref the standby holds after this
                        // attempt: the ones it already agreed with plus the
                        // ones this push moved. It is the number an admin
                        // compares against the primary's own ref count, which
                        // is how "3818 here, 0 there" was ever said out loud.
                        refs_at_last_success: (report.up_to_date + report.updates.len()) as u64,
                        last_error: None,
                        stopped: false,
                    },
                );
                if report.forgotten {
                    runner.counters.forgotten.fetch_add(1, Ordering::Relaxed);
                    tracing::info!(
                        target: TARGET_REPLICATION,
                        event = "replication.forgot",
                        standby = %runner.target.name,
                        repo = %job.key,
                        control = %report.url,
                        "the public-only replica deleted its copy of a repository that is now \
                         Private"
                    );
                    continue;
                }
                tracing::info!(
                    target: TARGET_REPLICATION,
                    event = "replication.synced",
                    standby = %runner.target.name,
                    repo = %job.key,
                    refs = report.updates.len(),
                    up_to_date = report.up_to_date,
                    objects = report.objects_sent,
                    pack_bytes = report.pack_bytes,
                    "replicated to a standby"
                );
                runner
                    .bus
                    .publish(event::Payload::MirrorSynced(gunnar_proto::MirrorSynced {
                        repo: Some(job.key.proto()),
                        remote_url: report.url,
                        refs_changed: report.updates.len() as u32,
                        updates: report.updates,
                    }));
            }
            // **The repository was DELETED while this job was queued.** Not a
            // failure at all: the work stopped existing. It is handled before
            // `failed` is counted, before the history row is written and before
            // `needs_full_resync` can be raised, because every one of those
            // would be a lasting mark left by an operator doing exactly what
            // they meant to do. MEASURED 2026-09-20: `proof/f13-…` was deleted
            // by the fixture that made it, and the replication subsystem then
            // demanded a full resync of the whole standby.
            Err(AttemptError { fault: Fault::Gone, err, .. }) => {
                runner.counters.cancelled.fetch_add(1, Ordering::Relaxed);
                runner.lock_history().remove(&job.key);
                {
                    let mut state = runner.lock_state();
                    state.in_flight = None;
                    // The standby is not behind: there is nothing to be behind
                    // ON. Leaving `consecutive_failures` and `last_error` alone
                    // would make the next status read blame the standby for a
                    // repository the operator removed.
                    state.consecutive_failures = 0;
                    state.last_error = None;
                }
                tracing::info!(
                    target: TARGET_REPLICATION,
                    event = "replication.cancelled",
                    standby = %runner.target.name,
                    repo = %job.key,
                    reason = %format!("{err:#}"),
                    "the repository was deleted after this job was queued; the job is dropped. \
                     No retry, no resync: a delete is not a failure to replicate."
                );
                continue;
            }
            Err(AttemptError { fault, err, .. }) => {
                let attempt = job.attempt + 1;
                runner.counters.failed.fetch_add(1, Ordering::Relaxed);
                let reason = format!("{err:#}");
                // **The one case where trying again is not trying.** This box
                // could not read its own repository, `MAX_LOCAL_ATTEMPTS` times
                // running, with nothing on the retry path able to change the
                // answer. The drill watched exactly this go round every 300 s for
                // 21 minutes. The job stops, and the stop is NAMED — see the
                // module docs.
                let decision = after_failure(fault, attempt, runner.retry_base, runner.retry_max);
                let stalled = decision == AfterFailure::Stop;
                let delay = match decision {
                    AfterFailure::Retry(d) => d,
                    AfterFailure::Stop => Duration::ZERO,
                };
                {
                    let mut history = runner.lock_history();
                    let record = history.entry(job.key.clone()).or_default();
                    record.last_error = Some(reason.clone());
                    record.stopped = stalled;
                }
                {
                    let mut state = runner.lock_state();
                    state.in_flight = None;
                    state.consecutive_failures = attempt;
                    state.last_error = Some(match (stalled, fault) {
                        (true, Fault::Incompatible) => format!(
                            "STOPPED, and no retry can change it: {reason}"
                        ),
                        (true, _) => format!(
                            "STALLED after {attempt} attempts, this server cannot read its own \
                             repository and retrying will not change that: {reason}"
                        ),
                        (false, _) => reason.clone(),
                    });
                    if stalled {
                        // The module's existing name for "a hole retrying will
                        // not close", sticky, already on the wire and in
                        // `Mirror.ReplicationStatus` — not a second flag.
                        state.needs_full_resync = true;
                    }
                }
                if stalled {
                    if fault == Fault::Incompatible {
                        runner
                            .counters
                            .stopped_incompatible
                            .fetch_add(1, Ordering::Relaxed);
                        tracing::error!(
                            target: TARGET_REPLICATION,
                            event = "replication.incompatible",
                            standby = %runner.target.name,
                            repo = %job.key,
                            error = %reason,
                            "replication STOPPED for this repository and retrying cannot help: \
                             this server and the standby cannot hold the same repository. The \
                             standby does NOT have it and will not get it from a push either."
                        );
                        runner.fail_event(&job.key, attempt, Duration::ZERO, &reason, true);
                        continue;
                    }
                    runner.counters.stalled_local.fetch_add(1, Ordering::Relaxed);
                    tracing::error!(
                        target: TARGET_REPLICATION,
                        event = "replication.stalled",
                        standby = %runner.target.name,
                        repo = %job.key,
                        attempts = attempt,
                        error = %reason,
                        "replication STOPPED for this repository: this server could not read it \
                         {attempt} times running, and no retry can change that. The standby is \
                         not at fault and is otherwise up to date. This repository's next push \
                         re-enqueues it, and so does `gunnar control replication resync`; until \
                         then the standby does NOT have it."
                    );
                    // `retry_in = 0` is the honest number for a job that is not
                    // coming back, and `needs_full_resync` is what tells a
                    // subscriber this one is terminal rather than slow.
                    runner.fail_event(&job.key, attempt, Duration::ZERO, &reason, true);
                    continue;
                }
                tracing::error!(
                    target: TARGET_REPLICATION,
                    event = "replication.failed",
                    standby = %runner.target.name,
                    repo = %job.key,
                    attempt,
                    fault = ?fault,
                    retry_in_ms = delay.as_millis() as u64,
                    error = %reason,
                    "replication to a standby failed; it will be retried. The originating push \
                     is unaffected and remains accepted."
                );
                runner.fail_event(&job.key, attempt, delay, &reason, false);
                runner.lock_queue().requeue(Job {
                    key: job.key,
                    attempt,
                    ready_at: Instant::now() + delay,
                    // NOT `now`: the repository has been outstanding since it
                    // was first enqueued, and a clock that restarts on every
                    // failure is a clock that never says a standby is behind.
                    first_seen: job.first_seen,
                });
                runner.wake.notify_one();
            }
        }
    }
}

/// What one successful replication did.
struct Replicated {
    url: String,
    updates: Vec<RefUpdate>,
    up_to_date: usize,
    objects_sent: u32,
    pack_bytes: u64,
    /// This attempt DELETED the target's copy (the browse replica) rather than updating
    /// it; `url` is then the control address the delete went to.
    forgotten: bool,
}

/// One attempt: open the repository **through the storage engine**, open the
/// channel, converge the standby.
///
/// The send-pack is **`gunnar-client`'s**, which is proved against the real
/// `git receive-pack` binary over pipes (`gunnar-client/tests/real_git_push.rs`).
/// A second implementation here would be a second thing to keep correct against
/// a protocol whose sharp edges are already written down (one implementation). What this
/// function owns is the two halves a send-pack cannot get from a `gix`
/// repository on this server: *what refs do I have* and *what objects go in the
/// pack*.
///
/// # 🔴 REPLICATION WAS BROKEN OUTRIGHT until 2026-08-30, and it said so
///
/// This opened the repository with `gix::open(&dir)` and handed the result to
/// `gunnar_client::push_mirror`. Since `e31ab8f4` put the server back on
/// znippy-git storage there is nothing there for `gix` to open: a push is
/// appended verbatim into `gunnar-git.znippy/objects.pack` and the refs are
/// rows in a log, so the directory has no `refs/`, no `objects/` and — on a
/// repository this build created — no `HEAD` either. MEASURED on the reference desktop
/// 2026-08-30 by `tests/replication.rs`, five attempts, five failures, every
/// one of them:
///
/// ```text
///   opening …/nordisk/gunnar.git to replicate it:
///   "…/nordisk/gunnar.git" does not appear to be a git repository:
///   Missing HEAD at '.git/HEAD'
/// ```
///
/// So **every configured standby has been receiving nothing at all**, loudly —
/// `ReplicationFailed` on the event stream, a climbing `failed` counter, a
/// `last_error` in `Mirror.Status` — which is why the "a dead standby is not
/// silent" and backoff tests stayed green while the two that assert the standby
/// actually *receives* timed out. A failure path that works perfectly is not a
/// replication.
///
/// The engine has both halves the old call needed from `gix`:
///
/// * the refs, from `GitOps::refs` — the same fold the advertisement is built
///   off, so the replicator and a clone of this server cannot disagree about
///   what it holds;
/// * the objects, from `gunnar_wire::objects_to_send(want, have)` — `want` is
///   what the commands move to, `have` is everything the standby advertised,
///   which is the whole of what a v0 push knows about the far end. This was
///   `GitOps::reachable(want, have)` until 2026-08-31, and asking the store
///   directly meant the tip of an annotated tag was never peeled: six of
///   the reference appliance's 25 repositories got a pack their standby could not
///   close. The selection is `upload_pack`'s now, so a repository that can be
///   cloned off this server can be replicated off it;
/// * and the pack itself, from `GitServe::emit_pack`, which is the same
///   emitter a clone out of this server streams (`gunnar_wire::upload_pack`).
///   The engine emits its stored entries rather than re-deflating them.
///
/// The handle comes from `ServerStores::repo_store`, which is the **cache the
/// transport is already using**. The old comment worried that a second handle
/// would fight the transport for the repository; taking the shared one is what
/// removes that worry rather than working around it.
/// **Where an attempt failed**, and therefore whether trying again can help.
///
/// The only two answers that matter to a retry policy. See the module docs and
/// [`MAX_LOCAL_ATTEMPTS`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Fault {
    /// The standby, the network, or the wire between them. Transient by
    /// default: a box comes back, a cable is replugged, a disk is grown.
    Remote,
    /// **This** server, reading **its own** repository: the catalogue, the store
    /// handle, `refs`, the object selection, `emit_pack`. Deterministic in
    /// state nothing on the retry path can change.
    Local,
    /// **The two ends cannot hold the same repository at all**, and no number
    /// of attempts changes that. Today that is exactly one thing: the object
    /// formats differ.
    ///
    /// 🔴 MEASURED 2026-08-31 on the reference appliance: `alice/replica-sha1` is
    /// sha1 and a standby **push-creates sha256**, so every attempt died on
    /// *"git cannot replicate between object formats"* and every one of them was
    /// retried, forever, at the 300 s ceiling. That is the drill's futile loop
    /// wearing a different hat, and the fix is the same one: stop, say which
    /// repository and what to do about it, and let a resync bring it back once
    /// somebody has. `alice/linux-mirror` is sha1, so this is not a curiosity.
    Incompatible,
    /// **The repository is GONE** — deleted between the enqueue and the
    /// attempt, so there is no catalog row to replicate from.
    ///
    /// Not [`Local`](Self::Local), which is what it used to be classified as
    /// and which cost a standby a full resync every time a fixture tidied up
    /// after itself (see [`Replicator::cancel`]). "This box cannot read its own
    /// repository" and "this box no longer has that repository" are different
    /// sentences with different remedies: the first is an alarm, the second is
    /// the operator getting exactly what they asked for.
    Gone,
}

/// **What the worker does with a failed attempt** — the whole retry policy, as
/// a function of two values and nothing else.
///
/// It is out here rather than inline in [`worker`] so that it can be asserted
/// at the boundary without a standby, a store or a clock: the defect the drill
/// hit was a POLICY defect (everything retried forever) and a policy that can
/// only be observed by watching a five-minute timer is a policy nobody checks.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AfterFailure {
    /// Re-queue, this far out.
    Retry(Duration),
    /// Stop. Nothing on this path can change the answer — see
    /// [`MAX_LOCAL_ATTEMPTS`].
    Stop,
}

/// [`AfterFailure`], decided.
///
/// A remote fault always retries, at any attempt count: a standby that is down
/// comes back, and a queue that gave up on it is the rotting standby the module
/// docs refuse. A local fault retries [`MAX_LOCAL_ATTEMPTS`] − 1 times and then
/// stops, because this box reading its own repository is deterministic in state
/// the retry path cannot touch.
pub fn after_failure(
    fault: Fault,
    attempt: u32,
    retry_base: Duration,
    retry_max: Duration,
) -> AfterFailure {
    // Not even once: the object formats are what they are, and the second
    // attempt reads the same advertisement as the first. A deleted repository
    // is the same shape of answer — the catalog will not grow the row back —
    // but the WORKER treats the two very differently: `Incompatible` raises
    // `needs_full_resync`, `Gone` drops the job and says nothing is wrong.
    if fault == Fault::Incompatible || fault == Fault::Gone {
        return AfterFailure::Stop;
    }
    if fault == Fault::Local && attempt >= MAX_LOCAL_ATTEMPTS {
        return AfterFailure::Stop;
    }
    AfterFailure::Retry(backoff(attempt, retry_base, retry_max))
}

/// One failed attempt: what went wrong, and which side of the wire it was on.
#[derive(Debug)]
struct AttemptError {
    fault: Fault,
    err: anyhow::Error,
    /// **The standby does not hold the repository, and a push cannot create
    /// it in this object format.** The `Repos.Create` that would, ready to
    /// send over the target's control plane; the worker sends it and pushes
    /// once more. `None` for every other failure. See [`AttemptError::absent`].
    create_first: Option<Box<gunnar_proto::CreateRepoRequest>>,
}

impl AttemptError {
    /// This box could not read its own repository.
    fn local(err: anyhow::Error) -> Self {
        Self { fault: Fault::Local, err, create_first: None }
    }

    /// The two ends cannot hold the same repository. See [`Fault::Incompatible`].
    fn incompatible(err: anyhow::Error) -> Self {
        Self { fault: Fault::Incompatible, err, create_first: None }
    }

    /// The standby advertised the push-create PROSPECT — nothing held, and
    /// the other object format, because a prospect is sha1 by construction
    /// (`push_create_spec`, 2026-09-20). Not a fault of either end: the
    /// repository has to be created there first, in its own format, which
    /// `create` says how to do. Classified [`Fault::Incompatible`] so that if
    /// the worker cannot act on it (no control address, no identity) the job
    /// stops with the same word it always used, rather than retrying forever.
    fn absent(create: gunnar_proto::CreateRepoRequest, err: anyhow::Error) -> Self {
        Self { fault: Fault::Incompatible, err, create_first: Some(Box::new(create)) }
    }

    /// The repository was deleted. See [`Fault::Gone`].
    fn gone(err: anyhow::Error) -> Self {
        Self { fault: Fault::Gone, err, create_first: None }
    }
}

/// Everything that is not explicitly local is remote — the safe default,
/// because an unclassified failure that retries forever loses nothing but time,
/// while an unclassified failure that stops loses a copy.
///
/// The blanket is over `E: Into<anyhow::Error>` so that every wire error
/// (`gunnar_pktline`, `std::io`) keeps its `?`. The only failures that are
/// *not* remote are the ones somebody wrote [`AttemptError::local`] on, which is
/// the direction the safety has to point.
impl<E: Into<anyhow::Error>> From<E> for AttemptError {
    fn from(err: E) -> Self {
        Self { fault: Fault::Remote, err: err.into(), create_first: None }
    }
}

/// **The client half of the control plane's mutual TLS**, for the one
/// control-plane call this module makes.
///
/// The same three variables `gunnar control bootstrap` writes and
/// `gunnar-cli`'s own `ClientTrust` reads. Read from the environment rather
/// than added to `ReplicationTarget`, for the reason the CLI gives: this is
/// per-*deployment* material, not per-target, and an appliance already takes it
/// through `EnvironmentFile`. All three or none — two out of three is a
/// misconfiguration that would otherwise surface as a handshake failure with no
/// hint about which piece is missing.
const CONTROL_CLIENT_CERT_ENV: &str = "GUNNAR_CLIENT_CERT";
const CONTROL_CLIENT_KEY_ENV: &str = "GUNNAR_CLIENT_KEY";
const CONTROL_CA_ENV: &str = "GUNNAR_CA";

struct ControlTrust {
    ca: Vec<u8>,
    cert: Vec<u8>,
    key: Vec<u8>,
}

impl ControlTrust {
    /// `None` when none of the three is set — the plaintext deployment, which
    /// is still the shipped default and still works.
    fn from_env() -> std::result::Result<Option<ControlTrust>, String> {
        let read = |name: &str| std::env::var_os(name).filter(|value| !value.is_empty());
        let (cert, key, ca) = (
            read(CONTROL_CLIENT_CERT_ENV),
            read(CONTROL_CLIENT_KEY_ENV),
            read(CONTROL_CA_ENV),
        );
        if cert.is_none() && key.is_none() && ca.is_none() {
            return Ok(None);
        }
        let want = |value: Option<std::ffi::OsString>, name: &str| {
            let path = value.ok_or_else(|| {
                format!(
                    "mutual TLS to a replica's control plane needs all three of \
                     {CONTROL_CLIENT_CERT_ENV}, {CONTROL_CLIENT_KEY_ENV} and {CONTROL_CA_ENV}; \
                     {name} is not set. `gunnar control bootstrap` writes all three files."
                )
            })?;
            let path = PathBuf::from(path);
            std::fs::read(&path).map_err(|err| format!("reading {}: {err}", path.display()))
        };
        Ok(Some(ControlTrust {
            cert: want(cert, CONTROL_CLIENT_CERT_ENV)?,
            key: want(key, CONTROL_CLIENT_KEY_ENV)?,
            ca: want(ca, CONTROL_CA_ENV)?,
        }))
    }
}

/// **What to dial, and whether that dial must carry TLS**, out of a target's
/// `control` and whether client material is configured.
///
/// The scheme follows the material, exactly as `gunnar-cli`'s `connect` decides
/// it: an operator who put the three files on the box meant them to be used,
/// and dialling `http://` with a certificate in hand would silently drop it.
/// An explicit scheme in `control` wins, because writing one is an instruction.
///
/// `https://` with nothing configured is an **error**, not a plaintext
/// fallback. What travels on this channel is a single-use sshsig credential for
/// a `Machine` with namespace admin on the replica, and an on-path observer who
/// gets it can replay it — the control-plane credential binds neither method
/// nor channel. Quietly downgrading it is the failure this refusal exists for.
fn control_url(control: &str, have_trust: bool) -> std::result::Result<(String, bool), String> {
    let url = if control.contains("://") {
        control.to_owned()
    } else if have_trust {
        format!("https://{control}")
    } else {
        format!("http://{control}")
    };
    let tls = url.starts_with("https://");
    if tls && !have_trust {
        return Err(format!(
            "{url} is an https:// control address and no client certificate is configured on \
             this server. Set {CONTROL_CLIENT_CERT_ENV}, {CONTROL_CLIENT_KEY_ENV} and \
             {CONTROL_CA_ENV} to the files `gunnar control bootstrap` wrote, or name the \
             address without a scheme to dial it in the clear."
        ));
    }
    Ok((url, tls))
}

/// **Delete a repository on a `PublicOnly` replica** (the browse replica):
/// `Repos.Delete{purge}` over [`ReplicationTarget::control`], signed with the
/// target's identity key — the same deploy key the send-pack authenticates
/// with, which the replica registered as a `Machine` with namespace admin
/// because push-create already needed it. The one write a browse box's
/// control plane accepts from anyone, and only from a `Machine`.
///
/// `NotFound` is success: the replica never had it, or already forgot. Every
/// other failure is [`Fault::Remote`] — the worker retries with backoff until
/// the replica answers, because a copy that should be gone and is not is the
/// one leak this design exists to prevent. A target without a control address
/// is [`Fault::Incompatible`]: `target_from_proto` refuses that at add time,
/// so seeing it here means a hand-edited `replication.json`, and the fix is
/// the operator's.
async fn forget_once(target: &ReplicationTarget, key: &JobKey) -> Result<(), AttemptError> {
    let (mut repos, credential) = control_plane_client(target, FORGET_METHOD).await?;
    let path = key.path().map_err(AttemptError::local)?;
    let mut request = tonic::Request::new(gunnar_proto::DeleteRepoRequest {
        path: Some(key.proto()),
        purge: true,
        ..Default::default()
    });
    request
        .metadata_mut()
        .insert(gunnar_grpc::CREDENTIAL_METADATA_KEY, credential);
    match repos.delete(request).await {
        Ok(_) => Ok(()),
        Err(status) if status.code() == tonic::Code::NotFound => {
            tracing::debug!(
                target: TARGET_REPLICATION,
                standby = %target.name,
                repo = %path,
                "the replica did not hold the repository; nothing to forget"
            );
            Ok(())
        }
        Err(status) => Err(AttemptError::from(anyhow::anyhow!(
            "{}'s control plane refused Repos.Delete for {path}: {} {}",
            target.name,
            status.code(),
            status.message()
        ))),
    }
}

/// **Create a repository on a standby, in this repository's object format**,
/// before pushing it there: `Repos.Create` over [`ReplicationTarget::control`],
/// signed as the delete is, with the same `Machine` deploy key. The forget's
/// other half (2026-09-22): what the forget purged, only a push can bring
/// back, and a push-create is sha1 by construction (`push_create_spec`), so
/// without this a sha256 repository that went Private and Public again stayed
/// off the browse replica for good. `request` was built by [`replicate_once`]
/// from the primary's catalog row.
///
/// `AlreadyExists` is [`Fault::Incompatible`]: the standby holds a repository
/// of that name after all — empty, in the other format — and no push or
/// create changes that; the operator deletes it there. Every other refusal is
/// [`Fault::Remote`], retried like the delete's, because a standby whose
/// control plane is down is a standby that comes back.
async fn create_once(
    target: &ReplicationTarget,
    key: &JobKey,
    request: gunnar_proto::CreateRepoRequest,
) -> Result<(), AttemptError> {
    let (mut repos, credential) = control_plane_client(target, CREATE_METHOD).await?;
    let path = key.path().map_err(AttemptError::local)?;
    let hash_kind = request.hash_kind;
    let mut request = tonic::Request::new(request);
    request
        .metadata_mut()
        .insert(gunnar_grpc::CREDENTIAL_METADATA_KEY, credential);
    match repos.create(request).await {
        Ok(_) => Ok(()),
        Err(status) if status.code() == tonic::Code::AlreadyExists => {
            Err(AttemptError::incompatible(anyhow::anyhow!(
                "{path} could not be created on {} in its own object format ({}): the standby \
                 already holds an EMPTY repository of that name in the other format, and git \
                 cannot replicate between object formats. Delete it there and `gunnar control \
                 replication resync {}`",
                target.name,
                match gunnar_proto::HashKind::try_from(hash_kind) {
                    Ok(gunnar_proto::HashKind::Sha1) => "sha1",
                    _ => "sha256",
                },
                target.name,
            )))
        }
        Err(status) => Err(AttemptError::from(anyhow::anyhow!(
            "{}'s control plane refused Repos.Create for {path}: {} {}",
            target.name,
            status.code(),
            status.message()
        ))),
    }
}

/// The replica-side method [`forget_once`] calls, as the credential binds it.
const FORGET_METHOD: &str = "Repos.Delete";
/// The replica-side method [`create_once`] calls, as the credential binds it.
const CREATE_METHOD: &str = "Repos.Create";

/// **The target's control plane, dialled and signed**: the channel and the
/// one-shot sshsig credential, for the two writes this module makes there
/// ([`forget_once`], [`create_once`]). One dial, one trust story, one place
/// that says the credential went out in the clear.
///
/// The credential is bound to `method` — the one call it is minted for — so a
/// copy lifted off the wire is worth that call and nothing else on the replica,
/// and a 5.3.3 replica, which refuses the method-blind v1 shape, serves it.
async fn control_plane_client(
    target: &ReplicationTarget,
    method: &str,
) -> Result<
    (
        gunnar_proto::repos_client::ReposClient<tonic::transport::Channel>,
        tonic::metadata::MetadataValue<tonic::metadata::Ascii>,
    ),
    AttemptError,
> {
    use gunnar_proto::repos_client::ReposClient;
    let Some(control) = target.control.as_deref() else {
        return Err(AttemptError::incompatible(anyhow::anyhow!(
            "target {} names no control address, so a repository cannot be deleted there when \
             it goes Private, nor created there in its own object format before it is pushed; \
             re-add it with `--control host:port`",
            target.name
        )));
    };
    let Some(identity) = target.identity.as_deref() else {
        return Err(AttemptError::incompatible(anyhow::anyhow!(
            "target {} names no identity file; a control-plane write there must be signed \
             with the deploy key the replica knows",
            target.name
        )));
    };
    let key_bytes = std::fs::read(identity)
        .with_context(|| format!("reading the identity file {}", identity.display()))
        .map_err(AttemptError::local)?;
    let signer = gunnar_core::HostKey::from_openssh(key_bytes)
        .with_context(|| format!("{} is not an OpenSSH private key", identity.display()))
        .map_err(AttemptError::local)?;
    // **TLS, when this box has been given the material for it.** Before this
    // the channel was built with no `tls_config` at all and no way to ask for
    // one: the sshsig credential below — single-use, but for a `Machine` with
    // namespace admin on the replica, and not bound to the channel — went out
    // in the clear on every deployment, on the one path in the
    // constellation that had no way to do otherwise. `gunnar-cli` already
    // reads the same three files for the same control plane.
    let trust = ControlTrust::from_env().map_err(|err| {
        AttemptError::local(anyhow::anyhow!(
            "the client certificate for {}'s control plane could not be read: {err}",
            target.name
        ))
    })?;
    let (url, tls) = control_url(control, trust.is_some())
        .map_err(|err| AttemptError::incompatible(anyhow::anyhow!("{err}")))?;
    if !tls {
        // Not a refusal — a plaintext control plane is still the shipped
        // default and a loopback or private-link deployment is a real one — but
        // it is said every time rather than never, because the alternative is a
        // secret leaving the box with nothing anywhere recording that it did.
        tracing::warn!(
            target: TARGET_REPLICATION,
            event = "replication.forget_plaintext",
            standby = %target.name,
            control = %url,
            "the delete to this replica's control plane carries a signed credential over an \
             UNENCRYPTED channel. Set {CONTROL_CLIENT_CERT_ENV}, {CONTROL_CLIENT_KEY_ENV} and \
             {CONTROL_CA_ENV} on this server to dial it over mutual TLS"
        );
    }
    let endpoint = tonic::transport::Channel::from_shared(url.clone())
        .map_err(|err| anyhow::anyhow!("{url} is not a control-plane address: {err}"))
        .map_err(AttemptError::incompatible)?
        .connect_timeout(Duration::from_secs(10))
        .timeout(Duration::from_secs(60));
    let endpoint = match &trust {
        None => endpoint,
        Some(trust) => {
            // Pinned for the same reason the listener pins: this binary links
            // russh's aws-lc-rs as well as rustls' ring.
            gunnar_grpc::ensure_crypto_provider();
            endpoint
                .tls_config(
                    tonic::transport::ClientTlsConfig::new()
                        // The server's own issuer, and NOT the system trust
                        // store: this CA is gunnar's, and a control plane that
                        // would also accept a public CA's certificate is pinned
                        // to nothing.
                        .ca_certificate(tonic::transport::Certificate::from_pem(&trust.ca))
                        .identity(tonic::transport::Identity::from_pem(&trust.cert, &trust.key)),
                )
                .map_err(|err| {
                    AttemptError::local(anyhow::anyhow!(
                        "configuring mutual TLS for {}'s control plane: {err}",
                        target.name
                    ))
                })?
        }
    };
    let channel = endpoint
        .connect()
        .await
        .with_context(|| format!("dialing {}'s control plane at {url}", target.name))?;
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0);
    let header =
        control_credential(signer.private_key(), now, method).map_err(AttemptError::local)?;
    Ok((ReposClient::new(channel), header))
}

/// The `gunnar-sshsig-v2` header for one call to a replica's control plane:
/// fresh nonce, signed now, **bound to `method`** (`Service.Method`).
fn control_credential(
    key: &gunnar_core::ssh_key::PrivateKey,
    now_unix_ms: i64,
    method: &str,
) -> Result<tonic::metadata::MetadataValue<tonic::metadata::Ascii>> {
    let nonce = {
        use gunnar_core::ssh_key::rand_core::Rng as _;
        let mut bytes = [0u8; 24];
        gunnar_core::os_rng().fill_bytes(&mut bytes);
        bytes.iter().map(|b| format!("{b:02x}")).collect::<String>()
    };
    let credential = gunnar_grpc::sign_credential_for(key, now_unix_ms, &nonce, method)
        .map_err(|err| anyhow::anyhow!("signing the control-plane credential: {err}"))?;
    credential
        .to_header_value()
        .parse()
        .map_err(|err| anyhow::anyhow!("the credential is not a legal header value: {err}"))
}

fn replicate_once(
    stores: &Arc<ServerStores>,
    target: &ReplicationTarget,
    key: &JobKey,
    known_hosts_default: &Path,
) -> std::result::Result<Replicated, AttemptError> {
    use gunnar_client::mirror::is_mirrored_ref;
    use gunnar_client::ssh::{HostKeyPolicy, SshOptions};
    use gunnar_store::git_storage_trait::{Caps, GitOps, GitServe};

    let path = key.path().map_err(AttemptError::local)?;
    // **A read fault and a missing row are two different answers.** The read
    // failing is this box unable to read its own catalog — local, retry, and
    // stall if it keeps up. The row being ABSENT is a repository that was
    // deleted after this job was queued, which no retry will bring back and
    // which nobody wants brought back. Before they were split, a delete cost
    // the standby a `needs_full_resync`.
    let repo_row = match stores.catalog().get(&path) {
        Ok(Some(row)) => row,
        Ok(None) => {
            return Err(AttemptError::gone(anyhow::anyhow!(
                "{path} is not in the catalogue: it was deleted after this replication job was \
                 queued, so there is nothing to replicate and the job is dropped"
            )))
        }
        Err(err) => {
            return Err(AttemptError::local(anyhow::Error::new(err).context(format!(
                "reading the catalogue for {path} to replicate it"
            ))))
        }
    };
    let store = stores
        .repo_store(&repo_row)
        .with_context(|| format!("opening {path} to replicate it"))
        .map_err(AttemptError::local)?;
    let git = store.serve();
    let hash_kind = store.hash_kind();

    // ── what this server holds ──────────────────────────────────────────────
    //
    // `HEAD` and the other pseudo-refs are not rows in `refs()` (the `GitServe` HEAD ruling), and
    // `is_mirrored_ref` drops anything outside `refs/heads/` and `refs/tags/`
    // — the same predicate `gunnar_client::mirrored_refs` applies to a gix
    // repository, so a mirror plan computed from either side is the same plan.
    let mut ours: Vec<(String, gix_hash::ObjectId)> = Vec::new();
    for row in GitOps::refs(git)
        .context("reading this server's refs to replicate them")
        .map_err(AttemptError::local)?
    {
        if !is_mirrored_ref(&row.name) {
            continue;
        }
        let Some(raw) = row.oid.as_deref() else {
            // A symbolic ref under `refs/` is not a thing to copy onto a mirror
            // as an object id; `gunnar_client::mirrored_refs` skips these too.
            continue;
        };
        let oid = gix_hash::oid::try_from_bytes(raw)
            .map_err(gix_error::Error::from).with_context(|| format!("{} holds an oid this store cannot decode", row.name))
            .map_err(AttemptError::local)?
            .to_owned();
        ours.push((row.name.clone(), oid));
    }
    ours.sort();

    let url = target.url_for(&path);
    let mut ssh = SshOptions::new();
    ssh.use_agent = false;
    ssh.key_path = target.identity.clone();
    ssh.known_hosts = Some(
        target
            .known_hosts
            .clone()
            .unwrap_or_else(|| known_hosts_default.to_path_buf()),
    );
    ssh.host_key = if target.strict_host_key {
        HostKeyPolicy::StrictKnownHosts
    } else {
        HostKeyPolicy::TofuKnownHosts
    };

    let loc = nornir_git::ssh::parse_ssh_url(&url)
        .map_err(|err| anyhow::anyhow!("{url} is not an ssh url: {err}"))?;
    let mut channel = gunnar_client::connect_receive_pack(&loc, &ssh)
        .with_context(|| format!("opening git-receive-pack on {} ({url})", target.name))?;

    let adv = {
        let (reader, _) = channel.io();
        gunnar_client::read_advertisement(reader)
            .with_context(|| format!("reading {}'s ref advertisement", target.name))?
    };
    if adv.hash_kind != hash_kind && adv.refs.iter().all(|r| r.oid.is_null()) {
        // **Nothing held, in the other format: the push-create PROSPECT.** A
        // standby that does not have the repository advertises an empty sha1
        // prospect (`push_create_spec`, 2026-09-20 — git says the object
        // format before the client has spoken), so a sha256 repository can
        // never arrive by push alone. The browse replica hits this every time
        // a repository it FORGOT (Public → Private purges its copy) goes
        // Public again: MEASURED 2026-09-22, `browse_replica.rs`, the last
        // step, `replication.incompatible` and a stop. The answer is not a
        // stop: the worker creates it over the control plane — the same
        // signed door the forget uses — with THIS repository's format, and
        // pushes again. The request is built here, where the catalog row is.
        //
        // A flush-pkt first: it is how a client says "never mind" to a
        // receive-pack waiting for commands, and the prospect is not written
        // until commands arrive, so nothing is left behind on the standby.
        {
            let (_, mut writer) = channel.io();
            gunnar_pktline::write_flush(&mut writer)?;
            std::io::Write::flush(&mut writer)?;
        }
        let create = gunnar_proto::CreateRepoRequest {
            path: Some(key.proto()),
            hash_kind: match hash_kind {
                gix_hash::Kind::Sha1 => gunnar_proto::HashKind::Sha1,
                _ => gunnar_proto::HashKind::Sha256,
            } as i32,
            default_branch: repo_row.default_branch.clone(),
            // The primary's row, as it is: on a `PublicOnly` target that is
            // Public (it would not be shipped otherwise), and on a full
            // standby whatever the primary says.
            visibility: match repo_row.visibility {
                gunnar_core::Visibility::Public => gunnar_proto::Visibility::Public,
                gunnar_core::Visibility::Private => gunnar_proto::Visibility::Private,
            } as i32,
            description: repo_row.description.clone(),
            adopt: false,
            // A mirror's repository carries what it is fed, never a server-written commit.
            readme: String::new(),
        };
        return Err(AttemptError::absent(
            create,
            anyhow::anyhow!(
                "{path} is {hash_kind:?} and {} does not hold it; a push would create it as \
                 {:?}, so it has to be created there first, in its own format",
                target.name,
                adv.hash_kind,
            ),
        ));
    }
    if adv.hash_kind != hash_kind {
        // A `?`-carrying `ensure!` cannot be used here: this function returns
        // `AttemptError`, and the classification is the point of that type.
        // `Incompatible` and NOT the `From` blanket's `Remote`: the standby is
        // up and answering, and the next attempt reads the same advertisement.
        return Err(AttemptError::incompatible(anyhow::anyhow!(
            "{path} is {hash_kind:?} but {} holds it as {:?}; git cannot replicate between \
             object formats. Delete it there and `gunnar control replication resync {}`: \
             the replicator then creates it in this repository's format before it pushes",
            target.name,
            adv.hash_kind,
            target.name,
        )));
    }

    // ── what the standby holds, and the plan between the two ────────────────
    let theirs: Vec<(String, gix_hash::ObjectId)> = adv
        .refs
        .iter()
        .filter_map(|r| {
            let name = std::str::from_utf8(r.name.as_slice()).ok()?;
            is_mirrored_ref(name).then(|| (name.to_owned(), r.oid))
        })
        .collect();
    let commands: Vec<gunnar_client::PushCommand> = gunnar_client::plan_ref_sync(&theirs, &ours)
        .into_iter()
        .map(|change| match change {
            gunnar_client::RefChange::Created { name, new } => {
                gunnar_client::PushCommand::create(name, new)
            }
            gunnar_client::RefChange::Updated { name, old, new } => {
                gunnar_client::PushCommand::update(name, old, new)
            }
            gunnar_client::RefChange::Deleted { name, old } => {
                gunnar_client::PushCommand::delete(name, old)
            }
        })
        .collect();
    let moving: std::collections::BTreeSet<&[u8]> =
        commands.iter().map(|c| c.name.as_slice()).collect();
    let up_to_date = ours
        .iter()
        .filter(|(name, _)| !moving.contains(name.as_bytes()))
        .count();

    if commands.is_empty() {
        // The remote is waiting for a command list. A flush-pkt is how a client
        // says "never mind"; closing the channel instead makes the standby's
        // receive-pack log a broken pipe for a replication that was simply
        // unnecessary.
        {
            let (_, mut writer) = channel.io();
            gunnar_pktline::write_flush(&mut writer)?;
            std::io::Write::flush(&mut writer)?;
        }
        return Ok(Replicated {
            url,
            updates: Vec::new(),
            up_to_date,
            objects_sent: 0,
            pack_bytes: 0,
            forgotten: false,
        });
    }

    // ── the objects, from the engine ────────────────────────────────────────
    //
    // Everything the standby advertised is a `have`: there is no negotiation
    // round in receive-pack, so that IS what the far end is known to hold.
    let all_deletes = commands.iter().all(gunnar_client::PushCommand::is_delete);
    let want: Vec<Vec<u8>> = commands
        .iter()
        .filter(|c| !c.is_delete())
        .map(|c| c.new.as_bytes().to_vec())
        .collect();
    let have: Vec<Vec<u8>> = adv
        .refs
        .iter()
        .filter(|r| !r.oid.is_null())
        .map(|r| r.oid.as_bytes().to_vec())
        .collect();
    let have_refs: Vec<&[u8]> = have.iter().map(Vec::as_slice).collect();
    let objects: Vec<Vec<u8>> = if all_deletes {
        Vec::new()
    } else {
        let want_refs: Vec<&[u8]> = want.iter().map(Vec::as_slice).collect();
        // **Through the serve path's selection, not the store's `reachable`.**
        //
        // 🔴 `c.new` for an annotated tag is the TAG OBJECT, and asking the
        // store for its closure got the tag and nothing under it on the engine
        // the appliance runs — so the standby refused the push with `missing
        // necessary objects: <commit> is not in this repository (peeling a
        // tag)` for six of the reference appliance's 25 repositories on 2026-08-31,
        // while `git clone` of the very same repositories worked. Clone works
        // because `select_objects` peels its tips first; this path did not peel
        // at all. One definition of "what a ref tip reaches" now, and it is the
        // one that was already right (one implementation) — see
        // [`gunnar_wire::objects_to_send`].
        gunnar_wire::objects_to_send(git, hash_kind, &want_refs, &have_refs)
            .context("selecting the objects the standby is missing")
            .map_err(AttemptError::local)?
    };
    let object_refs: Vec<&[u8]> = objects.iter().map(Vec::as_slice).collect();

    // `thin: false`, deliberately: a mirror must be able to stand alone the
    // instant it receives, and this end has no evidence about what the standby
    // can complete beyond the tips it advertised. `ofs_delta` is read off the
    // advertisement rather than assumed, because it is the same predicate the
    // capability line this send-pack writes is built from — using an encoding
    // the wire never declared is the defect `ofs_delta_agreed` exists for.
    let caps = Caps {
        thin: false,
        ofs_delta: adv.capabilities.has("ofs-delta"),
    };

    let mut pack_bytes: u64 = 0;
    let mut objects_sent: u32 = 0;
    // **The emitter's own failure is LOCAL and must not be laundered by the
    // wire.** `emit_pack` reads this server's store from inside send-pack's
    // writer closure, so its error reaches the caller wrapped in a transport
    // error that reads like the standby's fault. The flag is set where the
    // truth is known and read where the classification is made.
    let mut emit_failed = false;
    let report = {
        let emit = (!all_deletes).then_some(
            |w: &mut dyn std::io::Write| -> std::result::Result<(), gunnar_client::Error> {
                let stats = GitServe::emit_pack(git, &object_refs, &have_refs, &caps, w)
                    .map_err(|e| {
                        emit_failed = true;
                        gunnar_client::Error::Repo(e)
                    })?;
                pack_bytes = stats.bytes;
                objects_sent = stats.objects as u32;
                Ok(())
            },
        );
        gunnar_client::send_pack(
            &mut channel,
            &adv,
            &commands,
            hash_kind,
            emit,
            &gunnar_client::send_pack::options(),
        )
        .with_context(|| format!("replicating {path} to {} ({url})", target.name))
        .map_err(|e| {
            if emit_failed {
                AttemptError::local(e)
            } else {
                AttemptError::from(e)
            }
        })?
    };

    // A rejection is an answer rather than an error, but for replication it is
    // a FAILURE: the standby does not have what this server has. Turning it
    // into an error is what puts it on the retry path and on the event stream.
    if let Some(why) = report.failure_summary() {
        return Err(AttemptError::from(anyhow::anyhow!(
            "{} refused the replication push: {why}",
            target.name
        )));
    }

    Ok(Replicated {
        url,
        updates: commands
            .iter()
            .map(|c| RefUpdate {
                // Reference names are bytes on the wire; a gRPC message field
                // is a `String`. Lossy here is correct and safe: this is a
                // *report* about a push that has already happened, not a name
                // anything will be looked up by.
                ref_name: c.name.to_string(),
                old_oid: c.old.to_string(),
                new_oid: c.new.to_string(),
            })
            .collect(),
        up_to_date,
        objects_sent,
        pack_bytes,
        forgotten: false,
    })
}

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

    fn target(name: &str) -> ReplicationTarget {
        ReplicationTarget {
            name: name.to_owned(),
            scope: ReplicationScope::Server,
            remote: "ssh://git@standby:2222".into(),
            identity: None,
            known_hosts: None,
            strict_host_key: false,
            visibility: Default::default(),
            control: None,
        }
    }

    fn key(ns: &str, repo: &str) -> JobKey {
        JobKey {
            namespace: ns.to_owned(),
            repo: repo.to_owned(),
        }
    }

    /// **The browse replica's two control-plane writes sign v2, each bound to
    /// the call it makes** (stinky row 2, 2026-10-10).
    ///
    /// They were the last method-blind signer in the tree: a 5.3.3 replica
    /// refuses v1, so a primary still signing it could neither forget a
    /// repository that went Private nor create one. Proven against the real
    /// gate over the real key table: served at its own method, refused at
    /// `Access.AddKey` — a copy lifted off the wire is worth one call.
    #[test]
    fn the_replica_writes_sign_a_credential_bound_to_their_own_method() {
        let dir = tempfile::tempdir().unwrap();
        let stores = Arc::new(ServerStores::open(dir.path()).unwrap());
        let pem = gunnar_core::mint_ed25519().unwrap();
        let signer = gunnar_core::ssh_key::PrivateKey::from_openssh(&pem).unwrap();
        stores
            .auth()
            .bootstrap_admin(&signer.public_key().to_openssh().unwrap())
            .unwrap();
        let plane: Arc<dyn gunnar_grpc::ControlPlane> = Arc::new(
            crate::control_plane::CoreControlPlane::new(Arc::clone(&stores)),
        );
        let gate = gunnar_grpc::AuthGate::new(plane, gunnar_grpc::AuthMode::Required);

        for method in [FORGET_METHOD, CREATE_METHOD] {
            let now = gunnar_core::now_ms() as i64;
            let header = control_credential(&signer, now, method).unwrap();
            let text = header.to_str().unwrap();
            assert!(
                text.starts_with(gunnar_grpc::CREDENTIAL_SCHEME),
                "{method}: the header must be {}, got {text}",
                gunnar_grpc::CREDENTIAL_SCHEME
            );
            let mut md = tonic::metadata::MetadataMap::new();
            md.insert(gunnar_grpc::CREDENTIAL_METADATA_KEY, header.clone());
            assert!(
                gate.authenticate_bound(&md, now, Some(method)).is_ok(),
                "{method}: served at the method it was minted for"
            );
            let fresh = control_credential(&signer, now, method).unwrap();
            let mut md = tonic::metadata::MetadataMap::new();
            md.insert(gunnar_grpc::CREDENTIAL_METADATA_KEY, fresh);
            assert!(
                matches!(
                    gate.authenticate_bound(&md, now, Some("Access.AddKey")),
                    Err(gunnar_grpc::AuthFailure::MethodMismatch { .. })
                ),
                "{method}: refused anywhere else"
            );
        }
    }

    /// **The forget's channel carries TLS when the box has been given the
    /// material, and says so out loud when it has not.**
    ///
    /// The decision is asserted rather than the dial, because the dial is a
    /// network call and this is a rule about what is dialled. The bytes on that
    /// channel are a signed credential for a `Machine` with namespace admin on
    /// the replica; before this it went out in the clear on every deployment,
    /// with no way to configure otherwise.
    #[test]
    fn a_forget_dials_tls_when_the_box_has_a_client_certificate() {
        // The shipped default: nothing configured, plaintext, and no pretence
        // that it is anything else.
        assert_eq!(
            control_url("browse:50051", false),
            Ok(("http://browse:50051".to_owned(), false))
        );
        // Material on the box means the material is used. An operator who put
        // three files there meant them to be used, and a dial that dropped them
        // would be the silent downgrade.
        assert_eq!(
            control_url("browse:50051", true),
            Ok(("https://browse:50051".to_owned(), true))
        );
        // An explicit scheme is an instruction, in both directions.
        assert_eq!(
            control_url("http://browse:50051", true),
            Ok(("http://browse:50051".to_owned(), false))
        );
        assert_eq!(
            control_url("https://browse:50051", true),
            Ok(("https://browse:50051".to_owned(), true))
        );
        // …and `https://` with nothing configured is refused BY NAME rather
        // than dialled in the clear. A downgrade nobody is told about is how a
        // credential ends up on the wire of a network somebody thought was
        // encrypted.
        let err = control_url("https://browse:50051", false)
            .expect_err("https with no client certificate");
        assert!(err.contains(CONTROL_CLIENT_CERT_ENV), "{err}");
        assert!(err.contains(CONTROL_CLIENT_KEY_ENV), "{err}");
        assert!(err.contains(CONTROL_CA_ENV), "{err}");
    }

    /// RED-WHEN-BROKEN: the bound is asserted at the boundary, and the arm that
    /// refuses is asserted separately from the arm that accepts. A queue whose
    /// `push` always returned `Accepted` would pass a test that only counted
    /// what it inserted.
    #[test]
    fn the_queue_bound_refuses_rather_than_growing() {
        let mut queue = Queue::new(4);
        let now = Instant::now();
        for i in 0..4 {
            assert_eq!(
                queue.push(key("ns", &format!("r{i}")), now),
                Enqueued::Accepted,
                "the first {} must fit under a bound of 4",
                i + 1
            );
        }
        assert_eq!(queue.len(), 4);
        for i in 4..12 {
            assert_eq!(
                queue.push(key("ns", &format!("r{i}")), now),
                Enqueued::RefusedQueueFull,
                "past the bound every distinct repository must be REFUSED, not dropped-oldest"
            );
        }
        // The applied output: the queue never grew, and — the half a
        // drop-oldest implementation would fail — the ORIGINAL entries are
        // still the ones in it.
        assert_eq!(queue.len(), 4, "the bound must hold, not merely be logged");
        for i in 0..4 {
            assert!(
                queue.jobs.contains_key(&key("ns", &format!("r{i}"))),
                "r{i} was queued first and must not have been evicted by a later refusal"
            );
        }
    }

    /// The bound only holds because jobs coalesce. Assert the coalescing
    /// directly: 1000 pushes to one repository must be one pending job.
    #[test]
    fn a_thousand_pushes_to_one_repo_are_one_pending_job() {
        let mut queue = Queue::new(4);
        let now = Instant::now();
        assert_eq!(queue.push(key("ns", "hot"), now), Enqueued::Accepted);
        for _ in 0..1000 {
            assert_eq!(queue.push(key("ns", "hot"), now), Enqueued::Coalesced);
        }
        assert_eq!(
            queue.len(),
            1,
            "a job means `converge this repository`, so repeated pushes to it cost nothing"
        );
    }

    /// The identity-value trap. A backoff test sitting on one attempt
    /// cannot tell doubling from a constant, so every step is named by value.
    #[test]
    fn backoff_doubles_and_then_stops_at_the_cap() {
        let base = Duration::from_millis(100);
        let max = Duration::from_millis(800);
        assert_eq!(
            backoff(0, base, max),
            Duration::ZERO,
            "no failures, no wait"
        );
        assert_eq!(backoff(1, base, max), Duration::from_millis(100));
        assert_eq!(backoff(2, base, max), Duration::from_millis(200));
        assert_eq!(backoff(3, base, max), Duration::from_millis(400));
        assert_eq!(backoff(4, base, max), Duration::from_millis(800));
        assert_eq!(backoff(5, base, max), max, "capped, not doubling past max");
        assert_eq!(
            backoff(4000, base, max),
            max,
            "an attempt count that would overflow the shift must still be the cap"
        );
    }

    /// ★ **A repository that has been waiting keeps its clock through every
    /// failure.**
    ///
    /// 🔴 The number that made a dead standby look healthy: `ready_at` moves
    /// forward on every failure and `attempt` resets when a push re-creates a
    /// coalesced entry, so neither can answer *"how long has this standby been
    /// behind"* — and `last success` is a clock that has already stopped.
    /// `first_seen` is the one that keeps ticking, and it is asserted by value
    /// on both paths that could reset it.
    #[test]
    fn a_requeued_job_keeps_the_instant_it_first_became_outstanding() {
        let mut queue = Queue::new(4);
        let t0 = Instant::now();
        queue.push(key("ns", "waiting"), t0);
        assert_eq!(queue.oldest_first_seen(), Some(t0));

        // A failure: the job comes back with a backoff far in the future.
        let job = queue.take_ready(t0).expect("ready now");
        assert_eq!(job.first_seen, t0);
        queue.requeue(Job {
            key: job.key.clone(),
            attempt: 1,
            ready_at: t0 + Duration::from_secs(300),
            first_seen: job.first_seen,
        });
        assert_eq!(
            queue.oldest_first_seen(),
            Some(t0),
            "a retry must not restart the clock, or a standby that fails forever is never behind"
        );

        // A push arriving during the backoff coalesces onto the same entry and
        // must not restart it either.
        let later = t0 + Duration::from_secs(60);
        assert_eq!(queue.push(key("ns", "waiting"), later), Enqueued::Coalesced);
        assert_eq!(
            queue.oldest_first_seen(),
            Some(t0),
            "coalescing must not restart the clock"
        );

        // A DIFFERENT repository enqueued later does not become the oldest.
        queue.push(key("ns", "fresh"), later);
        assert_eq!(queue.oldest_first_seen(), Some(t0));
    }

    /// ★ **A queue handed to a new replicator keeps its work**, which is what
    /// makes `SetReplication` on an existing name safe to run during a backfill.
    #[test]
    fn a_queue_hands_its_outstanding_work_to_its_successor() {
        let mut old = Queue::new(4);
        let t0 = Instant::now();
        for i in 0..3 {
            old.push(key("ns", &format!("r{i}")), t0);
        }
        let carried = old.drain_jobs();
        assert_eq!(carried.len(), 3);
        assert_eq!(old.len(), 0, "draining must empty the queue it drained");

        let mut new = Queue::new(4);
        assert_eq!(new.adopt(carried), 0, "three jobs fit a bound of four");
        assert_eq!(new.len(), 3);
        assert_eq!(new.oldest_first_seen(), Some(t0), "the clocks carry too");

        // A SMALLER bound loses work, and says how much rather than silently
        // dropping it: the caller marks the target as needing a full resync.
        let mut small = Queue::new(2);
        let mut more = Queue::new(4);
        for i in 0..3 {
            more.push(key("ns", &format!("r{i}")), t0);
        }
        assert_eq!(
            small.adopt(more.drain_jobs()),
            1,
            "one of three cannot fit a bound of two, and the count is the alarm"
        );
        assert_eq!(small.len(), 2);
    }

    /// A job that is not yet due is not handed out, and the loop is told when
    /// to come back. Without this the worker would spin.
    #[test]
    fn a_job_in_backoff_is_not_ready_until_its_time() {
        let mut queue = Queue::new(4);
        let now = Instant::now();
        let later = now + Duration::from_secs(30);
        queue.push(key("ns", "waiting"), later);
        match queue.take_ready(now) {
            Err(Some(at)) => assert_eq!(at, later, "the loop must be told when to wake"),
            other => panic!("a job in backoff must not be handed out: {other:?}"),
        }
        let job = queue
            .take_ready(later)
            .expect("once its time comes, the job is ready");
        assert_eq!(job.key, key("ns", "waiting"));
        assert_eq!(queue.len(), 0);
    }

    /// ★ **A DELETED repository's queued job is dropped, and the queue is left
    /// able to hand out everything else.**
    ///
    /// 🔴 MEASURED 2026-09-20 on the live estate, two log rows one minute apart:
    ///
    /// ```text
    /// 12:34 repo deleted        proof/f13-1789907688270 the repository was deleted
    /// 12:34 replication failed  proof/f13-1789907688270 is not in the catalogue, so it
    ///                           cannot be replicated — a full resync is needed
    /// ```
    ///
    /// The delete was correct; the fixture was tidying up after itself. The
    /// queue kept the job anyway, the worker took it, the catalog had no row,
    /// the miss was called a LOCAL fault, and five of those stall the standby
    /// with the sticky `needs_full_resync`. A repository nobody has any more
    /// demanded a full resync of every repository somebody does.
    ///
    /// Seen RED against the old shape: `Queue` had no `remove`, so the
    /// cancelled key is still handed out by `take_ready`; and
    /// `after_failure(Fault::Gone, …)` does not compile, because the miss was
    /// spelled `Fault::Local` and nothing distinguished the two.
    #[test]
    fn a_deleted_repository_drops_its_job_and_is_not_a_failure() {
        let mut queue = Queue::new(4);
        let now = Instant::now();
        queue.push(key("proof", "f13-1789907688270"), now);
        queue.push(key("alice", "keeps-existing"), now);
        assert_eq!(queue.len(), 2);

        assert!(
            queue.remove(&key("proof", "f13-1789907688270")),
            "the job for the deleted repository was queued, so cancelling it drops something"
        );
        assert!(
            !queue.remove(&key("proof", "f13-1789907688270")),
            "cancelling twice is not an error and is not a second drop"
        );
        assert!(
            !queue.remove(&key("nobody", "never-queued")),
            "cancelling a repository with nothing outstanding drops nothing"
        );
        assert_eq!(queue.len(), 1, "only the deleted repository's job went");

        // The ordering entry for the cancelled key is still in `order`; the
        // NEXT job handed out must be the surviving one and not a dangling
        // name. This is the assertion that fails if `remove` ever tries to be
        // clever about `order`.
        let job = queue
            .take_ready(now)
            .expect("the repository that still exists is still due");
        assert_eq!(job.key, key("alice", "keeps-existing"));
        assert!(queue.take_ready(now).is_err(), "and nothing else is outstanding");

        // The policy half: a delete stops, like an incompatible pair — but the
        // worker's arms for the two are not the same, and only one of them
        // raises `needs_full_resync`.
        let base = Duration::from_millis(100);
        let max = Duration::from_millis(800);
        for attempt in [1u32, 2, 5, 1000] {
            assert_eq!(
                after_failure(Fault::Gone, attempt, base, max),
                AfterFailure::Stop,
                "no retry brings back a deleted repository, not even the first one (attempt \
                 {attempt})"
            );
        }
        assert_ne!(
            Fault::Gone,
            Fault::Local,
            "THE DEFECT: a deleted repository was classified as this box failing to read its \
             own, which is what asked for the resync"
        );
    }

    /// ★ **A remote fault retries forever; a local one stops, and stops by the
    /// third attempt.**
    ///
    /// 🔴 The two-appliance drill, 2026-08-30: a repository whose pack the PRIMARY's own
    /// store could not index was re-queued every 300 s for 21 minutes, against a
    /// healthy standby, making the identical `GitOps::reachable` call each time.
    /// The `EventGap` leg of that drill never converged and the reason sat in a
    /// log nobody reads.
    ///
    /// Seen RED by restoring the old policy — `AfterFailure::Retry(backoff(..))`
    /// unconditionally: every local arm below comes back `Retry`, including the
    /// thousandth.
    ///
    /// Both directions are asserted, because a policy that stopped on
    /// EVERYTHING would satisfy a test that only checked the local arm — and
    /// that is the rotting standby the module docs refuse.
    #[test]
    fn a_remote_fault_retries_forever_and_a_local_one_stops_by_the_third_attempt() {
        let base = Duration::from_millis(100);
        let max = Duration::from_millis(800);

        // Remote: never Stop, at any attempt count.
        for attempt in [1u32, 2, 3, 4, 10, 1000] {
            assert_eq!(
                after_failure(Fault::Remote, attempt, base, max),
                AfterFailure::Retry(backoff(attempt, base, max)),
                "a standby that is down must be retried at attempt {attempt}; giving up on it \
                 is the rotting standby this module refuses"
            );
        }

        // Local: the boundary named by value, both sides of it.
        assert_eq!(
            after_failure(Fault::Local, 1, base, max),
            AfterFailure::Retry(Duration::from_millis(100)),
            "one bad moment (a store handle racing a repository being created) must not \
             condemn durable bytes"
        );
        assert_eq!(
            after_failure(Fault::Local, 2, base, max),
            AfterFailure::Retry(Duration::from_millis(200))
        );
        assert_eq!(
            after_failure(Fault::Local, MAX_LOCAL_ATTEMPTS, base, max),
            AfterFailure::Stop,
            "three identical failures to read our OWN repository is the answer, not a phase"
        );
        assert_eq!(
            after_failure(Fault::Local, 40, base, max),
            AfterFailure::Stop,
            "and it stays stopped — a stall is cleared by new information (the next push, or \
             a resync), never by the clock"
        );
    }

    /// An unclassified failure must be REMOTE, so that the safe direction is
    /// the one a missing `AttemptError::local` falls into: retrying forever
    /// costs time, stopping wrongly costs a copy.
    #[test]
    fn an_unclassified_failure_is_remote() {
        let e: AttemptError = anyhow::anyhow!("something nobody classified").into();
        assert_eq!(e.fault, Fault::Remote);
        let io: AttemptError = std::io::Error::other("a wire read").into();
        assert_eq!(io.fault, Fault::Remote);
        assert_eq!(
            AttemptError::local(anyhow::anyhow!("our own store")).fault,
            Fault::Local
        );
        assert_eq!(
            AttemptError::incompatible(anyhow::anyhow!("sha1 vs sha256")).fault,
            Fault::Incompatible
        );
    }

    /// ★ **Two ends that cannot hold the same repository STOP — not even once
    /// retried.**
    ///
    /// 🔴 MEASURED on the reference appliance 2026-08-31: `alice/replica-sha1` is
    /// sha1, a standby PUSH-CREATES sha256, and every attempt died on *"git
    /// cannot replicate between object formats"* — and was retried, forever, at
    /// the 300 s ceiling. That is the drill's futile loop in a different hat: the
    /// second attempt reads the same advertisement as the first, so the answer
    /// cannot change until a human pre-creates the repository on the standby.
    /// `alice/linux-mirror` is sha1, so this is not a curiosity.
    ///
    /// Asserted at attempt **1**, which is the whole claim: `Fault::Local`
    /// deliberately allows two retries (a store handle can lose a race), and a
    /// policy that folded the two together would buy exactly the twenty-one
    /// minutes the drill watched.
    #[test]
    fn an_incompatible_object_format_stops_at_the_first_attempt() {
        let base = Duration::from_millis(100);
        let max = Duration::from_millis(800);
        for attempt in [1u32, 2, 3, 1000] {
            assert_eq!(
                after_failure(Fault::Incompatible, attempt, base, max),
                AfterFailure::Stop,
                "attempt {attempt} of a repository the standby cannot hold must STOP; the next \
                 attempt reads the same advertisement"
            );
        }
        // The other two policies are unchanged, or this test would pass against
        // a `Stop` that had swallowed them.
        assert_eq!(
            after_failure(Fault::Remote, 1, base, max),
            AfterFailure::Retry(Duration::from_millis(100))
        );
        assert_eq!(
            after_failure(Fault::Local, 1, base, max),
            AfterFailure::Retry(Duration::from_millis(100))
        );
    }

    #[test]
    fn scopes_select_what_they_say_they_do() {
        let repo = RepoRef::new("nordisk", "gunnar").unwrap();
        let other = RepoRef::new("other", "gunnar").unwrap();
        assert!(ReplicationScope::Server.covers(&repo));
        assert!(ReplicationScope::Server.covers(&other));

        let ns = ReplicationScope::Namespace {
            namespace: "nordisk".into(),
        };
        assert!(ns.covers(&repo));
        assert!(
            !ns.covers(&other),
            "a namespace scope must not take another"
        );

        let one = ReplicationScope::Repo {
            namespace: "nordisk".into(),
            repo: "gunnar".into(),
        };
        assert!(one.covers(&repo));
        assert!(!one.covers(&RepoRef::new("nordisk", "other").unwrap()));
    }

    #[test]
    fn a_base_url_gets_the_repository_path_and_a_full_one_does_not() {
        let path = RepoRef::new("nordisk", "gunnar").unwrap();
        let mut t = target("standby");
        assert_eq!(
            t.url_for(&path),
            "ssh://git@standby:2222/nordisk/gunnar.git"
        );
        t.remote = "ssh://git@standby:2222/".into();
        assert_eq!(
            t.url_for(&path),
            "ssh://git@standby:2222/nordisk/gunnar.git",
            "a trailing slash must not produce a doubled one"
        );
        t.remote = "ssh://git@standby:2222/elsewhere/copy.git".into();
        assert_eq!(
            t.url_for(&path),
            "ssh://git@standby:2222/elsewhere/copy.git",
            "an explicit .git URL is used verbatim"
        );
    }

    #[test]
    fn the_config_round_trips_and_a_missing_file_is_not_an_error() {
        let dir = tempfile::tempdir().unwrap();
        let empty = ReplicationConfig::load(dir.path()).expect("an absent config is empty");
        assert!(empty.targets.is_empty());
        assert_eq!(empty.queue_capacity, DEFAULT_QUEUE_CAPACITY);

        let mut config = ReplicationConfig::default();
        config.set_target(target("standby"));
        config.set_target(ReplicationTarget {
            scope: ReplicationScope::Namespace {
                namespace: "nordisk".into(),
            },
            ..target("nordisk-only")
        });
        config.save(dir.path()).expect("save");

        let back = ReplicationConfig::load(dir.path()).expect("load");
        assert_eq!(back.targets, config.targets);

        // Replacing by name, not appending a second row under the same name.
        let mut config = back;
        config.set_target(ReplicationTarget {
            remote: "ssh://git@elsewhere".into(),
            ..target("standby")
        });
        assert_eq!(config.targets.len(), 2);
        assert_eq!(
            config
                .targets
                .iter()
                .find(|t| t.name == "standby")
                .unwrap()
                .remote,
            "ssh://git@elsewhere"
        );
        assert!(config.remove_target("standby"));
        assert!(!config.remove_target("standby"), "removing twice is false");
    }

    /// A malformed config must stop the server rather than silently replicate
    /// nowhere. This is the arm that would otherwise be a permanent green.
    #[test]
    fn a_malformed_config_is_an_error_and_an_absent_one_is_not() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join(REPLICATION_CONFIG), b"{ not json").unwrap();
        let err = ReplicationConfig::load(dir.path()).expect_err("malformed JSON must not pass");
        assert!(
            format!("{err:#}").contains(REPLICATION_CONFIG),
            "the error must name the file an operator has to fix: {err:#}"
        );
    }

    /// **An EMPTY `replication.json` is the power-cut artifact, not an
    /// operator's malformed config — it degrades to "no replication" so the box
    /// boots, rather than being a fatal parse error that (as PID 1) is a kernel
    /// panic.**
    ///
    /// The distinction is exact and load-bearing: the test above proves a
    /// NON-empty malformed file is still fatal (an operator's typo), and this
    /// one proves a ZERO-BYTE file — which `serde_json` would also reject, and
    /// which only `save` interrupted by a power cut can produce — is NOT.
    ///
    /// # 🔴 Seen red
    ///
    /// Before `load` special-cased an empty file, this `expect` failed with the
    /// same `serde_json` "expected value" error the malformed test asserts —
    /// i.e. a 0-byte file was as fatal as a typo, which on the appliance was the
    /// `Attempted to kill init` wedge.
    #[test]
    fn an_empty_config_from_a_powercut_is_treated_as_absent_not_a_fatal_error() {
        let dir = tempfile::tempdir().unwrap();
        // The measured artifact: a 0-byte file left by an un-fsync'd rename.
        std::fs::write(dir.path().join(REPLICATION_CONFIG), b"").unwrap();
        let config = ReplicationConfig::load(dir.path())
            .expect("a 0-byte replication.json must load as an empty config, not fail the boot");
        assert!(
            config.targets.is_empty(),
            "an empty file must mean no replication configured"
        );
        // Whitespace-only (a partially-flushed write that landed only a newline)
        // is the same artifact and the same verdict.
        std::fs::write(dir.path().join(REPLICATION_CONFIG), b"\n  \n").unwrap();
        assert!(ReplicationConfig::load(dir.path())
            .expect("a whitespace-only replication.json must also load as empty")
            .targets
            .is_empty());
    }
}

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

    /// **The probe address, out of every remote shape a target can carry.**
    ///
    /// Getting this wrong would make the reconnect wake probe the wrong box,
    /// which is worse than not probing at all: it would pull a queue off its
    /// backoff because something *else* answered.
    #[test]
    fn a_probe_address_is_the_authority_and_nothing_else() {
        let t = |remote: &str| ReplicationTarget {
            name: "s".into(),
            scope: ReplicationScope::Server,
            remote: remote.into(),
            identity: None,
            known_hosts: None,
            strict_host_key: false,
            visibility: Default::default(),
            control: None,
        };
        assert_eq!(
            t("ssh://git@standby:2222").probe_addr().as_deref(),
            Some("standby:2222")
        );
        // No port is SSH's 22, the same default the client dials.
        assert_eq!(
            t("ssh://git@standby").probe_addr().as_deref(),
            Some("standby:22")
        );
        // The path never reaches the probe.
        assert_eq!(
            t("ssh://git@standby:2222/ns/repo.git")
                .probe_addr()
                .as_deref(),
            Some("standby:2222")
        );
        // No user is fine.
        assert_eq!(
            t("ssh://standby:2222").probe_addr().as_deref(),
            Some("standby:2222")
        );
        // IPv6 keeps its brackets, which is what `TcpStream::connect` wants.
        assert_eq!(
            t("ssh://git@[fd00::1]:2222").probe_addr().as_deref(),
            Some("[fd00::1]:2222")
        );
        assert_eq!(
            t("ssh://git@[fd00::1]").probe_addr().as_deref(),
            Some("[fd00::1]:22")
        );
        // A shape this cannot read probes NOTHING rather than guessing.
        assert_eq!(t("standby:2222").probe_addr(), None);
        assert_eq!(t("ssh://").probe_addr(), None);
    }

    /// **`ready_now` moves every waiting job and reports how many.**
    ///
    /// The count is what makes the reconnect log line applied output rather
    /// than "woke up", and it is the identity-value guard: a `ready_now` that
    /// did nothing would return 0 and this goes red.
    #[test]
    fn ready_now_pulls_every_backed_off_job_forward() {
        let mut q = Queue::new(8);
        let later = Instant::now() + Duration::from_secs(300);
        for name in ["a", "b", "c"] {
            q.push(
                JobKey {
                    namespace: "ns".into(),
                    repo: name.into(),
                },
                later,
            );
        }
        let now = Instant::now();
        // Nothing is ready while the backoff stands.
        assert!(
            q.take_ready(now).is_err(),
            "a backed-off queue yields nothing"
        );
        assert_eq!(q.ready_now(now), 3, "all three were waiting");
        assert_eq!(q.ready_now(now), 0, "already ready, nothing to move");
        // …and now they come out.
        for _ in 0..3 {
            q.take_ready(now).expect("ready after the reconnect wake");
        }
        assert!(q.take_ready(now).is_err(), "and the queue is empty");
    }
}