nordisk/gunnar
crates/gunnar-server/src/mirror.rs
//! **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)
.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]
The file goes on; the console shows all of it.