crates/gunnar-server/src/replication_wire.rs
//! **Replication's configuration, over the wire and back.**
//!
//! One file, one direction each way, and it is deliberately not part of
//! [`crate::mirror`]. `mirror.rs` is the replicator: a subscriber, a queue and a
//! send-pack, and it knows nothing about gRPC. `control_plane.rs` is the
//! control plane and knows nothing about replication's types. This is the seam
//! between them, so neither has to grow a dependency on the other's vocabulary.
//!
//! # Why the operator surface exists at all
//!
//! The 2026-08-30 drill measured replication working end to end and then said
//! the thing that mattered: it was not in the appliance binary. Compiling it in
//! is half the job. The other half is that the reference appliance is an **OS-less
//! image running as PID 1** — no shell, no editor, no `scp` — so
//! "edit `<data-root>/replication.json` and restart" is not a procedure anyone
//! can carry out on the box that ships. A feature that ships and cannot be
//! turned on has not shipped.
//!
//! So `Mirror.SetReplication` / `RemoveReplication` / `ReplicationStatus`, and
//! this file is the translation under them.
//!
//! # What is NOT translated, and why
//!
//! [`ReplicationConfig`]'s tuning — `queue_capacity`, `retry_base_ms`,
//! `retry_max_ms`, `subscriber_disabled` — has no field on the wire. Those are
//! a deployment's shape, not a standby's identity, and the one that is a test
//! seam (`subscriber_disabled`) must never be reachable from a network call:
//! `mirror.rs` says so by name. `set_target` on a config read back from the
//! slot therefore preserves whatever they are, rather than resetting them to
//! defaults on every `SetReplication` — which is exactly what building a fresh
//! `ReplicationConfig` here would have done.
use std::path::{Path, PathBuf};
use gunnar_grpc::{ControlError, ControlResult};
use crate::mirror::{ReplicationScope, ReplicationTarget, Replicator, VisibilityScope};
/// **The deploy key the appliance pushes to its standby with**, under the data
/// root beside `replication.json`.
///
/// It is minted ON THE BOX and the private half never leaves it. That is not a
/// nicety, it is the only design that works here: an appliance with no shell
/// cannot be handed a file, and a private key travelling through a control-plane
/// RPC would be a private key in a caller's shell history, in a terminal
/// scrollback and in whatever logs that call passed through.
///
/// So the exchange runs the other way. `SetReplication` mints this key if it is
/// not there, and answers with its PUBLIC half; the operator authorises that one
/// line on the standby, exactly as they would any other deploy key. `0600`,
/// because it is a private key and the appliance is the only thing that has ever
/// held it.
pub const REPLICATION_KEY: &str = "replication_key";
/// Mint the replication deploy key if it does not exist, and return its path
/// and its public OpenSSH line.
///
/// Idempotent and stable: a second `SetReplication` reuses the same key, so an
/// operator who has already authorised it on the standby does not have to do it
/// again for every target they add.
pub fn ensure_identity(root: &Path) -> ControlResult<(PathBuf, String)> {
let path = root.join(REPLICATION_KEY);
let bytes = match std::fs::read(&path) {
Ok(bytes) => bytes,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
let minted = gunnar_core::mint_ed25519().map_err(|err| {
ControlError::Internal(format!("minting the replication deploy key: {err}"))
})?;
// Write-then-rename, and 0600 BEFORE the bytes are visible under the
// final name: a private key that is briefly world-readable is a
// private key that was world-readable.
let tmp = root.join(format!("{REPLICATION_KEY}.tmp"));
std::fs::write(&tmp, &minted).map_err(|err| {
ControlError::Internal(format!("writing {}: {err}", tmp.display()))
})?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(&tmp, std::fs::Permissions::from_mode(0o600)).map_err(
|err| ControlError::Internal(format!("chmod 600 {}: {err}", tmp.display())),
)?;
}
std::fs::rename(&tmp, &path).map_err(|err| {
ControlError::Internal(format!("renaming {} into place: {err}", tmp.display()))
})?;
tracing::info!(
target: crate::mirror::TARGET_REPLICATION,
event = "replication.identity_minted",
path = %path.display(),
"minted this server's replication deploy key; its public half must be authorised \
on the standby"
);
minted
}
Err(err) => {
return Err(ControlError::Internal(format!(
"reading {}: {err}",
path.display()
)))
}
};
let key = gunnar_core::HostKey::from_openssh(bytes).map_err(|err| {
ControlError::Internal(format!("{} is not an OpenSSH key: {err}", path.display()))
})?;
let public = key.public_openssh().map_err(|err| {
ControlError::Internal(format!("encoding the replication key's public half: {err}"))
})?;
Ok((path, public))
}
/// An empty string is "not set", never a path of length zero.
///
/// proto3 has no `null` for a scalar, so the wire cannot tell "the operator did
/// not name a key" from "the operator named the empty path". Treating `""` as
/// `None` is the only reading that is ever useful: an empty `identity_path`
/// falls back to the client key ladder, which is what a target with no key does
/// today, and an empty `known_hosts` means the server's own.
fn opt_path(s: String) -> Option<PathBuf> {
let s = s.trim();
(!s.is_empty()).then(|| PathBuf::from(s))
}
fn path_str(p: &Option<PathBuf>) -> String {
p.as_ref()
.map(|p| p.display().to_string())
.unwrap_or_default()
}
/// Wire → [`ReplicationTarget`], refusing everything a standby cannot be.
///
/// Every refusal here is an `InvalidArgument` naming the field, because the
/// caller is an operator at a terminal and the whole value of the verb is that
/// it tells them what is wrong while they can still fix it. The alternative —
/// accepting a target that can never connect — is a standby that reports
/// `failed` forever into a log nobody reads, which is the failure this feature
/// was rebuilt out of.
pub fn target_from_proto(t: gunnar_proto::ReplicationTarget) -> ControlResult<ReplicationTarget> {
let name = t.name.trim().to_owned();
if name.is_empty() {
return Err(ControlError::InvalidArgument(
"name is required: every event, log line and status row identifies a standby by \
its name rather than by its URL"
.into(),
));
}
let remote = t.remote.trim().to_owned();
if remote.is_empty() {
return Err(ControlError::InvalidArgument("remote is required".into()));
}
// SSH ONLY, refused by name. The push half is `gunnar-client`'s send-pack
// over its russh transport; it has no HTTP transport and the appliance has
// no `http` feature to give it one. An `https://` here would be accepted,
// stored, and then fail on every attempt forever.
if !(remote.starts_with("ssh://") || remote.starts_with("git+ssh://")) {
return Err(ControlError::InvalidArgument(format!(
"remote {remote:?} is not an ssh url. Replication is a send-pack over SSH; \
the server's client half has no HTTP transport to dial anything else with"
)));
}
let scope = match t.scope.and_then(|s| s.of) {
None | Some(gunnar_proto::replication_scope::Of::Server(_)) => ReplicationScope::Server,
Some(gunnar_proto::replication_scope::Of::Namespace(ns)) => {
let ns = ns.trim().to_owned();
if ns.is_empty() {
return Err(ControlError::InvalidArgument(
"a namespace-scoped target with an empty namespace would take nothing".into(),
));
}
ReplicationScope::Namespace { namespace: ns }
}
Some(gunnar_proto::replication_scope::Of::Repo(path)) => {
path.validate()
.map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
ReplicationScope::Repo {
namespace: path.namespace,
repo: path.repo,
}
}
};
// **PublicOnly needs a control address** (the browse replica): the Public → Private
// delete has nowhere to go without one, and a replica that could not be
// told to forget would keep private data — refused at add time, by name,
// rather than discovered at the first flip.
let control = Some(t.control.trim().to_owned()).filter(|c| !c.is_empty());
let visibility = if t.public_only {
if control.is_none() {
return Err(ControlError::InvalidArgument(
"a public-only target needs `control` (the replica's control-plane host:port): \
a repository that goes Private is deleted there over that address, and \
without it the replica would keep a copy it must not"
.into(),
));
}
// The identity requirement is NOT checked here — see
// [`require_public_only_identity`], which runs after the mint.
VisibilityScope::PublicOnly
} else {
VisibilityScope::All
};
Ok(ReplicationTarget {
name,
scope,
remote,
identity: opt_path(t.identity_path),
known_hosts: opt_path(t.known_hosts_path),
strict_host_key: t.strict_host_key,
visibility,
control,
})
}
/// **A public-only target must end up with a deploy key — checked AFTER the
/// mint, never before it.**
///
/// The requirement is right: a repository that goes Private is DELETED on the
/// replica over `control`, and that call is signed with the same deploy key the
/// send-pack authenticates with. A browse replica that could not be told to
/// forget would keep private data.
///
/// It used to be enforced inside [`target_from_proto`], which is one statement
/// too early. MEASURED 2026-09-10 against a running appliance:
///
/// ```text
/// $ gunnar-cli control replication add public ssh://git@…:2222 \
/// --scope public-only --control …:50051
/// gunnar: Client specified an invalid argument: invalid argument: a public-only
/// target needs `identity_path`: the delete on a Private flip is signed with the
/// same deploy key the send-pack authenticates with
/// ```
///
/// — refused for the absence of a key the server MINTS on the very next line
/// (`control_plane::replication_set` calls [`ensure_identity`] the moment
/// `identity` is `None`, for the reason recorded on [`REPLICATION_KEY`]: an
/// appliance with no shell cannot be handed a file). So the one target shape
/// that most needs the minted key was the one shape the mint could not reach,
/// and the documented one-line recipe for a browse replica could not be run at
/// all on the box it was written for.
///
/// Moving it here keeps the guarantee — nothing is stored without a key — and
/// lets the mint satisfy it. What is left is a genuine failure: the box could
/// not produce a key at all.
pub fn require_public_only_identity(t: &ReplicationTarget) -> ControlResult<()> {
if t.visibility == VisibilityScope::PublicOnly && t.identity.is_none() {
return Err(ControlError::InvalidArgument(
"a public-only target needs a deploy key and this box has none: the delete on a \
Private flip is signed with the same key the send-pack authenticates with. The \
server mints one itself unless `identity_path` names another, so this means the \
mint failed"
.into(),
));
}
Ok(())
}
/// [`ReplicationTarget`] → wire.
pub fn target_to_proto(t: &ReplicationTarget) -> gunnar_proto::ReplicationTarget {
use gunnar_proto::replication_scope::Of;
let of = match &t.scope {
ReplicationScope::Server => Of::Server(true),
ReplicationScope::Namespace { namespace } => Of::Namespace(namespace.clone()),
ReplicationScope::Repo { namespace, repo } => Of::Repo(gunnar_proto::RepoPath {
namespace: namespace.clone(),
repo: repo.clone(),
}),
};
gunnar_proto::ReplicationTarget {
name: t.name.clone(),
scope: Some(gunnar_proto::ReplicationScope { of: Some(of) }),
remote: t.remote.clone(),
identity_path: path_str(&t.identity),
known_hosts_path: path_str(&t.known_hosts),
strict_host_key: t.strict_host_key,
public_only: t.visibility == VisibilityScope::PublicOnly,
control: t.control.clone().unwrap_or_default(),
}
}
/// The RUNNING state, straight off [`crate::replication_view::report`].
///
/// **Through the one reader and not around it.** That function is what the
/// console band on the box's own screen and `Admin.Stats` already draw from,
/// including the staleness verdict and the "never synced" zero. Building a
/// second view here out of `Replicator::status` would have been a twin, and two
/// readers of the same counters are two chances to disagree about whether a
/// standby has ever worked — which is the one question this whole surface
/// exists to answer.
pub fn running_to_proto(
replicator: &Replicator,
include_refs: bool,
out: &mut gunnar_proto::ReplicationStatusResponse,
) {
let report = crate::replication_view::report(replicator);
out.configured = report.configured;
out.observed_since_unix_ms = replicator.observed_since_unix_ms();
// **The ref-coverage walk is opt-in, and it is taken ONCE here.**
//
// `include_refs` is false for a plain `replication list` and for every
// status a mutating verb prints back: those want the verdict, the pending
// clock and the per-repository copy state, none of which opens a store. It
// is true only for `replication list --refs`, when the operator has asked
// what a mirror does not carry — and then this is the single place the
// reference databases are opened, one FRESH walk whose per-repository rows
// and server-wide sum both come off it (one implementation), rather than the old code's
// walk on every call that made a cold box unresponsive.
let coverage = include_refs.then(|| replicator.ref_coverage_by_repo(std::time::Duration::ZERO));
// **Per repository, and driven from the catalogue.** A failure to read the
// catalogue leaves this empty rather than failing the whole status call:
// the aggregates are still worth answering, and an empty list is
// distinguishable from a list of zeroes.
out.repos = match replicator.repo_copies(coverage.as_deref()) {
Ok(copies) => copies
.into_iter()
.map(|c| gunnar_proto::RepoReplication {
repo: Some(gunnar_proto::RepoPath::new(&c.namespace, &c.repo)),
// `None` stays `None`: "we could not read this repository's
// refs" and "none of them are unmirrored" are opposite answers.
refs_total: c.refs.map(|r| r.total),
refs_unmirrored: c.refs.map(|r| r.unmirrored),
standby: c.standby,
last_success_unix_ms: c.last_success_unix_ms,
refs_at_last_success: c.refs_at_last_success,
outstanding: c.outstanding,
last_error: c.last_error,
stopped: c.stopped,
sha1: c.sha1,
})
.collect(),
Err(err) => {
tracing::warn!(
target: crate::mirror::TARGET_REPLICATION,
%err,
"could not list repositories for the per-repository replication report; the \
aggregates below are still answered"
);
Vec::new()
}
};
// The server-wide summary is the sum of the same walk the rows came off
// (one implementation), or left at zero when refs were not asked for — which every
// printer already renders as "no ref figure", not as "nothing unmirrored".
if let Some(coverage) = coverage.as_deref() {
let mut total = crate::replication_view::RefCoverage::default();
for c in coverage.values() {
total.add(*c);
}
out.refs_total = total.total;
out.refs_unmirrored = total.unmirrored;
}
out.enqueued = report.enqueued;
out.synced = report.synced;
out.failed = report.failed;
out.bus_gaps = report.bus_gaps;
out.standbys = report
.standbys
.into_iter()
.map(|s| gunnar_proto::ReplicationStandby {
verdict: s.verdict().word().to_owned(),
name: s.name,
remote: s.remote,
pending: s.pending,
consecutive_failures: s.consecutive_failures,
last_error: s.last_error,
last_success_unix_ms: s.last_success_unix_ms,
needs_full_resync: s.needs_full_resync,
stale: s.stale,
in_flight: s.in_flight,
outstanding_for_ms: s.outstanding_for_ms,
behind: s.behind,
})
.collect();
}
#[cfg(test)]
mod tests {
use super::*;
fn public_only_wire(identity: &str) -> gunnar_proto::ReplicationTarget {
gunnar_proto::ReplicationTarget {
name: "public".into(),
scope: Some(gunnar_proto::ReplicationScope {
of: Some(gunnar_proto::replication_scope::Of::Server(true)),
}),
remote: "ssh://git@replica:2222".into(),
identity_path: identity.into(),
known_hosts_path: String::new(),
strict_host_key: false,
public_only: true,
control: "replica:50051".into(),
}
}
/// **A public-only target with no `identity_path` must PARSE**, because the
/// server mints the key immediately afterwards.
///
/// Red-first, and the red is what shipped. MEASURED 2026-09-10 against a
/// running server, typing the one-line recipe the `replication browse`
/// help text itself prints:
///
/// ```text
/// invalid argument: a public-only target needs `identity_path`: the delete
/// on a Private flip is signed with the same deploy key the send-pack
/// authenticates with
/// ```
///
/// `control_plane::replication_set` calls [`ensure_identity`] the moment
/// `identity` is `None` — one statement after this parse — precisely
/// because an appliance has no shell and cannot be handed a key file. So
/// the target shape that needs the minted key most was the only shape the
/// mint could not reach, and a browse replica could not be configured on
/// the box this control plane exists for.
#[test]
fn a_public_only_target_parses_without_an_identity_because_the_mint_follows() {
let parsed = target_from_proto(public_only_wire("")).expect(
"a public-only target with no identity must parse: the server mints one on the \
next line, and refusing here makes that mint unreachable",
);
assert_eq!(parsed.visibility, VisibilityScope::PublicOnly);
assert!(parsed.identity.is_none(), "nothing invented an identity");
}
/// And the guarantee still holds: nothing is stored public-only without a
/// key. The check simply moved to where a key exists.
#[test]
fn the_identity_requirement_still_bites_after_the_mint() {
let mut minted = target_from_proto(public_only_wire("")).expect("parses");
let err = require_public_only_identity(&minted)
.expect_err("a public-only target with no key must be refused somewhere");
let text = format!("{err:?}");
assert!(
text.contains("mint failed"),
"the refusal has to say what actually went wrong now — the operator did not \
forget a flag, the box could not make a key:\n{text}"
);
minted.identity = Some(std::path::PathBuf::from("/data/replication_key"));
require_public_only_identity(&minted).expect("a minted key satisfies it");
}
/// The OTHER public-only requirement is unmoved: `control` is a thing the
/// operator types and nothing can mint, so it is still refused at parse.
#[test]
fn a_public_only_target_without_a_control_address_is_still_refused_at_parse() {
let mut wire = public_only_wire("/data/replication_key");
wire.control = String::new();
let err = target_from_proto(wire).expect_err("no control address");
assert!(
format!("{err:?}").contains("needs `control`"),
"a replica that cannot be told to forget a repository that went Private would \
keep private data"
);
}
}