crates/gunnar-cli/tests/replication.rs
#![cfg(all(feature = "server", feature = "client"))]
//! **Replication — a push to one gunnar appears on a second gunnar.**
//!
//! ```text
//! git ── ssh ──▶ gunnar A ── event bus ──▶ replicator ── ssh ──▶ gunnar B
//! │ │
//! └── "everything up-to-date", already └── asserted HERE
//! ```
//!
//! Two whole servers, both real: real russh listeners, real key tables, real
//! `fs` stores, real `git` as the client, and a real send-pack between them. The
//! headline test asserts the **ref and the file contents** on the standby, read
//! with the stock `git` binary out of the standby's own repository — not that
//! the replicator returned `Ok`, and not that gunnar A believes it replicated.
//!
//! # The four things "measure, never assert" makes this file responsible for
//!
//! 1. **The headline can be made red by breaking the subscriber specifically.**
//! `nothing_replicates_when_the_subscriber_is_not_started` is that red half,
//! kept in the suite instead of performed once and described in a commit
//! message. Without it, `a_push_to_one_gunnar_appears_on_the_second` would
//! pass against any mechanism that happened to copy the repository.
//! 2. **A dead standby does not fail the push**, and the failure is *loud*: a
//! `ReplicationFailed` on the event stream and a counter. Asserting the push
//! succeeded is not enough on its own — a replicator that did nothing at all
//! would pass that half.
//! 3. **The backoff really delays**, asserted both by the `retry_in_ms` the
//! events carry and by the wall clock between attempts. Either alone is
//! blind: the field could be computed and ignored, and a clock measurement
//! alone cannot tell doubling from a constant.
//! 4. **The queue bound really refuses.** Driven through the public
//! `Replicator::enqueue`, not through the private queue, so the counter, the
//! `needs_full_resync` flag and the event are all asserted on the path a push
//! actually takes.
mod common;
use std::sync::Arc;
use std::time::{Duration, Instant};
use common::{local_repo, tools_present, Harness};
use gunnar_cli::server::mirror::{
ReplicationConfig, ReplicationScope, ReplicationTarget, Replicator,
};
use gunnar_core::{HashKind, NewRepo, RepoRef};
use gunnar_grpc::{EventBus, EventFilter};
use gunnar_proto::{event, Event, ReplicationFailed};
use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver};
/// How long a test waits for replication to land on the standby.
///
/// Generous, because the path is: subprocess `git` → SSH → receive-pack →
/// bus → task wake-up → `spawn_blocking` → a second SSH connection →
/// receive-pack again. A wait that is too short becomes a flake that reads
/// exactly like the bug.
const WAIT: Duration = Duration::from_secs(30);
/// A port nothing can be listening on. `1` is privileged and unused, so a
/// connection to it is refused immediately and deterministically — which is
/// what "the mirror is down" has to mean for a test that must not be a race.
const DEAD_PORT: u16 = 1;
// ───────────────────────────────────────────────────────────────────────────
// helpers
// ───────────────────────────────────────────────────────────────────────────
/// A replication configuration pointing one gunnar at another.
fn to_standby(standby: &Harness, port: u16) -> ReplicationConfig {
ReplicationConfig {
targets: vec![ReplicationTarget {
name: "standby".into(),
scope: ReplicationScope::Server,
remote: format!("ssh://git@127.0.0.1:{port}"),
// The standby's own client key: a deploy key the standby's identity
// table already holds, with a Write grant on the repository. This
// is exactly what a real second gunnar needs provisioned.
identity: Some(standby.key_path.clone()),
// Never `~/.ssh/known_hosts` — a test must not write to the
// developer's home directory, and neither must a server.
known_hosts: Some(standby.dir.path().join("replication_known_hosts")),
strict_host_key: false,
visibility: Default::default(),
control: None,
}],
// Fast, so the retry tests finish. The production defaults are seconds
// and minutes; see `mirror.rs`.
retry_base_ms: 200,
retry_max_ms: 2_000,
..ReplicationConfig::default()
}
}
/// The oid a gunnar server advertises for `rev`, read with the **stock `git`
/// binary** over the wire.
///
/// Deliberately not through gunnar's own store: the question is whether a
/// second gunnar holds a repository git agrees is that repository, and gunnar
/// agreeing with gunnar would answer a different one.
///
/// # 🔴 This was `git rev-parse` inside `<git_dir>` until 2026-08-30
///
/// Which asked stock `git` to read a directory that, on the engine gunnar
/// ships, holds no `refs/`, no `packed-refs`, no `objects/` and no loose
/// object — a push is appended verbatim into `gunnar-git.znippy/objects.pack`
/// and the refs are rows in a log. So `rev-parse` failed on every call, on
/// **both** servers, and returned `None` whatever replication had done. Three
/// tests died of it (two waiting 30 s for a standby that had already received
/// the push, and one reporting that the PRIMARY did not hold the commit it had
/// just accepted) and a fourth —
/// `nothing_replicates_when_the_subscriber_is_not_started` — was passing
/// **vacuously**, because "the standby never got the ref" is what a broken
/// reader says too. The premise was wrong, not the server.
///
/// `ls-remote` keeps the property the old spelling was reaching for and
/// strengthens it: it is still stock `git`, still not gunnar reading its own
/// store, and it now asks through the door the world comes through rather than
/// over the filesystem behind it.
fn remote_ref(harness: &Harness, namespace: &str, repo: &str, rev: &str) -> Option<String> {
let url = harness.url(namespace, repo);
let out = harness.git(harness.dir.path(), &["ls-remote", &url, rev]);
if !out.ok {
return None;
}
out.stdout.lines().find_map(|line| {
let (oid, name) = line.split_once('\t')?;
(name.trim() == rev).then(|| oid.trim().to_owned())
})
}
/// Wait until `check` is true on the standby, or give up after [`WAIT`].
fn wait_for<T>(what: &str, mut check: impl FnMut() -> Option<T>) -> T {
let deadline = Instant::now() + WAIT;
loop {
if let Some(value) = check() {
return value;
}
assert!(
Instant::now() < deadline,
"waited {WAIT:?} for {what} and it never happened"
);
std::thread::sleep(Duration::from_millis(50));
}
}
/// The same wait, expected to time out — the shape a red demonstration needs.
/// Returns true when the thing under `check` NEVER happened within `patience`.
fn never_within<T>(patience: Duration, mut check: impl FnMut() -> Option<T>) -> bool {
let deadline = Instant::now() + patience;
while Instant::now() < deadline {
if check().is_some() {
return false;
}
std::thread::sleep(Duration::from_millis(50));
}
check().is_none()
}
/// A direct event-bus subscription, drained into a queue.
///
/// In-process rather than over `Events.Watch`: the bus is what the replicator
/// publishes to, and putting a gRPC listener, a port and an sshsig credential
/// between the assertion and the thing asserted would only add ways to be
/// flaky. `push_events.rs` owns the gRPC projection of this same stream.
struct BusTap {
events: UnboundedReceiver<Event>,
drain: Option<tokio::task::JoinHandle<()>>,
runtime: tokio::runtime::Handle,
}
impl Drop for BusTap {
fn drop(&mut self) {
if let Some(drain) = self.drain.take() {
drain.abort();
}
}
}
impl BusTap {
fn open(runtime: &tokio::runtime::Runtime, bus: Arc<EventBus>) -> BusTap {
let (tx, events) = unbounded_channel();
// Subscribe on the calling thread so every event published after this
// returns is already in this subscriber's ring — a subscription still
// being set up would make the test a race.
let stream = bus.subscribe(EventFilter::default());
let drain = runtime.spawn(async move {
let mut stream = Box::pin(stream);
while let Some(item) = tokio_stream::StreamExt::next(&mut stream).await {
match item {
Ok(event) => {
if tx.send(event).is_err() {
break;
}
}
Err(status) => {
eprintln!("the event stream ended: {status}");
break;
}
}
}
});
BusTap {
events,
drain: Some(drain),
runtime: runtime.handle().clone(),
}
}
/// Every `ReplicationFailed` delivered so far, draining what is queued.
fn drain_failures(&mut self) -> Vec<ReplicationFailed> {
let mut out = Vec::new();
while let Ok(event) = self.events.try_recv() {
if let Some(event::Payload::ReplicationFailed(failed)) = event.payload {
out.push(failed);
}
}
out
}
/// Collect `n` `ReplicationFailed` events, or fail.
///
/// Each one is paired with the bus's own `at_unix_ms`, because *when* the
/// events arrived is the only thing that can tell a backoff that is applied
/// from one that is merely computed and reported.
fn failures(&mut self, n: usize, what: &str) -> Vec<(i64, ReplicationFailed)> {
let mut seen = Vec::new();
let runtime = self.runtime.clone();
runtime.block_on(async {
let deadline = tokio::time::Instant::now() + WAIT;
while seen.len() < n {
let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
if remaining.is_zero() {
break;
}
match tokio::time::timeout(remaining, self.events.recv()).await {
Ok(Some(event)) => {
let at = event.at_unix_ms;
if let Some(event::Payload::ReplicationFailed(failed)) = event.payload {
seen.push((at, failed));
}
}
Ok(None) | Err(_) => break,
}
}
});
assert!(
seen.len() >= n,
"waited {WAIT:?} for {n} ReplicationFailed event(s) while {what}, and saw {}. \
An empty list means the failure path is SILENT — which is worse than a \
replication that never started, because the standby then rots looking healthy.",
seen.len()
);
seen
}
}
// ───────────────────────────────────────────────────────────────────────────
// the headline
// ───────────────────────────────────────────────────────────────────────────
/// **THE test.** A real `git push` to gunnar A, over real SSH, appears on
/// gunnar B — asserted as the ref *and* the file contents, read with `git` out
/// of B's own repository.
/// **A standby can be named over the CONTROL PLANE, on a box with no shell, and
/// before any repository exists.**
///
/// This is the half of replication that was missing on 2026-08-30. Replication was
/// proved to work, and then compiled into the appliance — and the appliance is
/// an OS-less image running as PID 1 with **no shell, no editor and no `scp`**.
/// Its configuration lives in `<data-root>/replication.json`. Until
/// `Mirror.SetReplication` existed there was nothing on that box able to write
/// that file, which is a feature that ships and cannot be turned on.
///
/// Five things are asserted, and each is a separate way the surface could be
/// useless:
///
/// 1. **The primary starts with NO target**, so the slot is genuinely empty and
/// nothing here is reconfiguring a replicator that was already running.
/// 2. **A target may name a repository that does not exist.** A fresh appliance
/// has no repositories, and an operator must be able to point the box at its
/// standby on day one; the control plane does not look the repository up,
/// and this is the assertion that says so if it ever starts to. The
/// fixture's absence is checked first, so the acceptance is not vacuous.
/// 3. **The server-wide standby is set** and comes back `configured`.
/// 4. **It takes effect with NO restart** — the push that follows lands on the
/// standby, and the oid is compared against the CLIENT's, not gunnar's.
/// 5. **It is on disk**, in `replication.json`, so it survives the reboot the
/// operator has no shell to prepare for — and removal is equally durable,
/// while removing a name that is not there is an error rather than a quiet
/// success.
///
/// 🔴 **Seen RED twice before it was believed**, 2026-08-30:
///
/// * with step 3's `replication_set` replaced by a plain `status`, step 4 fails
/// with *"waited 30s for a control-plane-configured push to reach the standby
/// and it never happened"* — so the push landing is caused by this surface and
/// not by something the harness does anyway;
/// * with the `config.save(...)` removed from `ReplicationSlot::apply`, step 5
/// fails with *"no …/data/replication.json after SetReplication"* — so the
/// durability half is asserted on a file that is really written.
#[test]
fn a_standby_can_be_named_over_the_control_plane_before_any_repository_exists() {
if !tools_present() {
eprintln!("skipping: git and ssh are required");
return;
}
use gunnar_grpc::ControlPlane;
let standby = Harness::start("nordisk", "gunnar", HashKind::Sha256);
let standby_port = standby.ssh_port;
// NO replication configuration. This is a box as it comes off the medium.
let primary = Harness::start("nordisk", "gunnar", HashKind::Sha256);
let slot = Arc::clone(
primary
.server
.as_ref()
.expect("the server is running")
.replication_slot(),
);
let plane = gunnar_cli::server::CoreControlPlane::new(Arc::clone(&primary.stores))
.with_replication(Arc::clone(&slot));
// 1. Nothing is configured, and the surface SAYS so rather than erroring.
let before = plane
.replication_status(false)
.expect("status on an empty slot");
assert!(
before.targets.is_empty() && !before.configured,
"a box with no standby reported one: {before:?}"
);
let known_hosts = primary.dir.path().join("replication_known_hosts");
let target = |name: &str, of| gunnar_proto::ReplicationTarget {
name: name.into(),
scope: Some(gunnar_proto::ReplicationScope { of: Some(of) }),
remote: format!("ssh://git@127.0.0.1:{standby_port}"),
identity_path: standby.key_path.display().to_string(),
known_hosts_path: known_hosts.display().to_string(),
strict_host_key: false,
public_only: false,
control: String::new(),
};
// 2. **BEFORE THE REPOSITORY EXISTS.** A target scoped to a repository this
// server has never heard of is accepted, because a fresh appliance has no
// repositories and an operator must be able to point it at its standby on
// day one. The control plane does not look the repository up — and if it
// ever starts to, this is the assertion that says so.
assert!(
primary
.stores
.catalog()
.get(&RepoRef::new("nordisk", "not-created-yet").unwrap())
.ok()
.flatten()
.is_none(),
"the fixture repository exists, so the next assertion would prove nothing"
);
let out = plane
.replication_set(target(
"future",
gunnar_proto::replication_scope::Of::Repo(gunnar_proto::RepoPath {
namespace: "nordisk".into(),
repo: "not-created-yet".into(),
}),
))
.expect("SetReplication for a repository that does not exist yet");
assert_eq!(out.targets.len(), 1, "the target did not stick: {out:?}");
plane
.replication_remove("future")
.expect("removing the day-one target");
// 2b. **AND THE BOX MINTS ITS OWN KEY.** An operator who names no identity
// cannot be asking for the client key ladder: on an image with no home
// directory that ladder finds nothing and every push fails on
// authentication forever. They are asking for "this box's own key", and
// until 2026-08-30 there was no way to put one on a machine with no
// shell. So the server mints one, keeps the private half, and answers
// with the public line to authorise on the standby.
let keyless = gunnar_proto::ReplicationTarget {
identity_path: String::new(),
..target("keyless", gunnar_proto::replication_scope::Of::Server(true))
};
let out = plane
.replication_set(keyless.clone())
.expect("SetReplication with no identity");
assert!(
out.identity_public_key.starts_with("ssh-ed25519 "),
"no public key came back, so the operator has nothing to authorise on the standby and \
every push will fail on auth: {:?}",
out.identity_public_key
);
let key_path = primary.stores.root().join("replication_key");
assert!(
key_path.is_file(),
"{} was not written, so the private half does not exist",
key_path.display()
);
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let mode = std::fs::metadata(&key_path).unwrap().permissions().mode() & 0o777;
assert_eq!(mode, 0o600, "the deploy key is mode {mode:o}, not 0600");
}
// The target on disk names the minted key, or the next boot pushes with
// nothing.
let json = std::fs::read_to_string(primary.stores.root().join("replication.json")).unwrap();
assert!(
json.contains("replication_key"),
"the stored target does not name the minted key:\n{json}"
);
// Stable: a second target reuses it, so an operator authorises ONE line
// rather than one per standby.
let again = plane
.replication_set(keyless)
.expect("SetReplication again");
assert_eq!(
again.identity_public_key, out.identity_public_key,
"the box minted a SECOND key, so the line already authorised on the standby is now wrong"
);
plane
.replication_remove("keyless")
.expect("removing the keyless target");
// 3. The server-wide standby, set with no restart of anything.
let out = plane
.replication_set(target(
"standby",
gunnar_proto::replication_scope::Of::Server(true),
))
.expect("SetReplication");
assert_eq!(out.targets.len(), 1, "the target did not stick: {out:?}");
assert!(
out.configured,
"SetReplication returned but nothing is running: {out:?}"
);
// 4. It is running NOW: push, with nothing restarted in between.
let work = primary.dir.path().join("work");
let head = local_repo(&primary, &work, HashKind::Sha256, "configured remotely");
primary
.git(
&work,
&[
"push",
&primary.url("nordisk", "gunnar"),
"main:refs/heads/main",
],
)
.expect_ok("git push to the primary");
let replicated = wait_for(
"a control-plane-configured push to reach the standby",
|| remote_ref(&standby, "nordisk", "gunnar", "refs/heads/main"),
);
assert_eq!(
replicated, head,
"the standby's refs/heads/main must be the commit the client pushed to a primary that \
was told about the standby over the control plane, with no restart"
);
// 5. And it survives a reboot, because it is on disk.
let path = primary.stores.root().join("replication.json");
let json = std::fs::read_to_string(&path)
.unwrap_or_else(|err| panic!("no {} after SetReplication: {err}", path.display()));
assert!(
json.contains("\"standby\"") && json.contains(&format!("127.0.0.1:{standby_port}")),
"replication.json does not name the standby, so a reboot loses it:\n{json}"
);
// And removing it by name stops replication and is equally durable.
let out = plane
.replication_remove("standby")
.expect("RemoveReplication by name");
assert!(
out.targets.is_empty(),
"the target survived removal: {out:?}"
);
assert!(
plane.replication_remove("standby").is_err(),
"removing a standby that is not there reported success, which reads exactly like \
removing the one you meant"
);
let json = std::fs::read_to_string(&path).expect("replication.json after removal");
assert!(
!json.contains("\"standby\""),
"the removed standby is still on disk, so a reboot brings it back:\n{json}"
);
}
/// **Demote: the failover primitive — one call drops EVERY standby.**
///
/// When a primary dies and its standby is promoted, the recovered old box comes
/// back UP and SERVING but its `replication.json` still lists the box
/// that is now the NEW primary — so its replicator resumes and force-clobbers
/// the new primary's post-failover writes (proven applied, ref for ref, in
/// `deploy/checks/failover-drill.sh`). `replication demote` is what the operator
/// runs on the recovered box to stop being a primary before it is re-attached
/// as the new primary's standby.
///
/// # 🔴 Seen red
///
/// Two things this catches, each of which a wrong implementation passes:
///
/// * a demote that returns `Ok` without clearing — the hollow version — leaves
/// the roster in place, so both `after.targets.is_empty()` and the on-disk
/// check go red on it;
/// * demote is NOT `remove`: with TWO standbys configured, `remove` by name
/// leaves the other and the box is still a primary. Only demote empties a
/// multi-standby roster, which is the whole point on a recovered box that may
/// have had several.
#[test]
fn demote_drops_every_standby_and_is_durable_and_idempotent() {
if !tools_present() {
eprintln!("skipping: git and ssh are required");
return;
}
use gunnar_grpc::ControlPlane;
let standby_a = Harness::start("nordisk", "gunnar", HashKind::Sha256);
let standby_b = Harness::start("nordisk", "gunnar", HashKind::Sha256);
let primary = Harness::start("nordisk", "gunnar", HashKind::Sha256);
let slot = Arc::clone(
primary
.server
.as_ref()
.expect("the server is running")
.replication_slot(),
);
let plane = gunnar_cli::server::CoreControlPlane::new(Arc::clone(&primary.stores))
.with_replication(Arc::clone(&slot));
let known_hosts = primary.dir.path().join("replication_known_hosts");
let target = |name: &str, key: &std::path::Path, port: u16| gunnar_proto::ReplicationTarget {
name: name.into(),
scope: Some(gunnar_proto::ReplicationScope {
of: Some(gunnar_proto::replication_scope::Of::Server(true)),
}),
remote: format!("ssh://git@127.0.0.1:{port}"),
identity_path: key.display().to_string(),
known_hosts_path: known_hosts.display().to_string(),
strict_host_key: false,
public_only: false,
control: String::new(),
};
// TWO standbys — a recovered box that fanned out to more than one.
plane
.replication_set(target("alpha", &standby_a.key_path, standby_a.ssh_port))
.expect("SetReplication alpha");
plane
.replication_set(target("beta", &standby_b.key_path, standby_b.ssh_port))
.expect("SetReplication beta");
let s = plane.replication_status(false).expect("status");
assert_eq!(s.targets.len(), 2, "two standbys did not stick: {s:?}");
// `remove` by name clears ONE — the box is still a primary of `beta`. This
// is the contrast that makes demote a distinct primitive rather than a
// rename of remove.
let s = plane
.replication_remove("alpha")
.expect("RemoveReplication alpha");
assert_eq!(
s.targets.len(),
1,
"remove cleared more than the one named standby: {s:?}"
);
// `demote` clears the REST in one call, with no name — atomic.
let after = plane
.replication_demote()
.expect("demote clears every remaining target");
assert!(
after.targets.is_empty(),
"demote left standbys behind — this box is still a primary: {after:?}"
);
assert!(
!after.configured,
"demote returned but a standby is still configured: {after:?}"
);
// Durable: gone from `replication.json`, so the reboot does not resurrect
// the primary role and re-clobber the new primary.
let path = primary.stores.root().join("replication.json");
let json = std::fs::read_to_string(&path).expect("replication.json after demote");
assert!(
!json.contains("\"alpha\"") && !json.contains("\"beta\""),
"a demoted standby is still on disk, so a reboot brings the primary role back:\n{json}"
);
// Idempotent, and never NotFound: "make sure I am not a primary" must not
// fail on a box that already is not one — unlike `remove`, which errors on a
// name that is not there.
let again = plane
.replication_demote()
.expect("demote on an already-empty roster is Ok, not an error");
assert!(
again.targets.is_empty(),
"the second demote resurrected a target: {again:?}"
);
}
/// **SetRole: the write-freeze the roster cannot express, proven APPLIED.**
///
/// `demote` clears the roster — "I ship no second copy" — but a box with an
/// empty roster is, by itself, still a writable box: it will accept a `git
/// push`. That is the split-brain from the RECEIVING side, and the role is what
/// closes it. A `STANDBY` freezes client writes; `PRIMARY` — the takeover flip —
/// thaws them.
///
/// The assertion is on the SHARED write-gate the SSH front door actually reads
/// (`ServerStores::writes_frozen`) and on the durable `replication.json`, NOT
/// on the proto the call echoes back. That distinction is the whole test: a
/// hollow `set_role` that returned the role it was handed without flipping the
/// gate would pass a proto round-trip and still take every client write — the
/// exact green a measured assertion forbids. Proven RED by stubbing
/// `replication_set_role` to return the snapshot without `apply`:
/// `writes_frozen()` stayed `false` and this failed on the freeze assertion.
#[test]
fn set_role_freezes_the_gate_the_front_door_reads_and_promote_thaws_it_durably() {
if !tools_present() {
eprintln!("skipping: git and ssh are required");
return;
}
use gunnar_grpc::ControlPlane;
use std::sync::atomic::Ordering::SeqCst;
let unit = Harness::start("nordisk", "gunnar", HashKind::Sha256);
let slot = Arc::clone(
unit.server
.as_ref()
.expect("the server is running")
.replication_slot(),
);
let plane = gunnar_cli::server::CoreControlPlane::new(Arc::clone(&unit.stores))
.with_replication(Arc::clone(&slot));
let frozen = || unit.stores.writes_frozen().load(SeqCst);
let standby_i32 = gunnar_proto::ReplicationRole::Standby as i32;
let primary_i32 = gunnar_proto::ReplicationRole::Primary as i32;
// A fresh box is a writable primary, and the gate the front door reads is
// thawed — this is every deployment before failover was built.
let s = plane.replication_status(false).expect("status");
assert_eq!(
s.role, primary_i32,
"a fresh box must report PRIMARY: {s:?}"
);
assert!(!frozen(), "a fresh box must not freeze client writes");
// STANDBY: the applied gate must FREEZE. Read off the same atomic the SSH
// dispatch consults, so this cannot be the proto echo lying.
let s = plane
.replication_set_role(gunnar_proto::ReplicationRole::Standby)
.expect("set standby");
assert_eq!(
s.role, standby_i32,
"set_role(standby) did not report STANDBY: {s:?}"
);
assert!(
frozen(),
"set_role(standby) returned but the write-gate the front door reads is still open — \
a client push would still land here"
);
// Durable: it survives the reboot a recovered box comes back from, so the box
// does not come back thawed and start taking writes it should refuse.
let json = std::fs::read_to_string(unit.stores.root().join("replication.json"))
.expect("replication.json after set_role");
assert!(
json.contains("Standby"),
"the STANDBY role is not on disk, so a reboot would thaw the freeze:\n{json}"
);
// PROMOTE — the takeover flip — must THAW, applied.
let s = plane
.replication_set_role(gunnar_proto::ReplicationRole::Primary)
.expect("promote");
assert_eq!(
s.role, primary_i32,
"set_role(primary) did not report PRIMARY: {s:?}"
);
assert!(
!frozen(),
"promote returned but the write-gate is still frozen — the promoted box would refuse \
the writes it now exists to take"
);
}
/// **Abdicate: freeze → flush → demote, in that order, proven APPLIED.**
///
/// The voluntary counterpart to `demote`. A crash leaves the standby to be
/// promoted and the old box to be demoted before it clobbers; abdication is the
/// orderly hand-off, and the order is what makes it one: the box FREEZES (stops
/// taking client writes) BEFORE it clears its roster, so nothing lands here in
/// the window `demote` alone left open.
///
/// Asserted on applied state: the shared write-gate the front door reads is
/// frozen, the roster is empty, and both facts are durable in `replication.json`.
/// Proven RED against a hollow abdicate that returned `Ok` without freezing:
/// `writes_frozen()` stayed `false` and this failed on the freeze assertion —
/// the box would have handed off and kept taking writes.
#[test]
fn abdicate_freezes_before_it_clears_the_roster_and_both_are_durable() {
if !tools_present() {
eprintln!("skipping: git and ssh are required");
return;
}
use gunnar_grpc::ControlPlane;
use std::sync::atomic::Ordering::SeqCst;
// A healthy primary→standby pair, so the FLUSH drains fast rather than
// sitting out the ceiling against a dead box.
let standby = Harness::start("nordisk", "gunnar", HashKind::Sha256);
let standby_port = standby.ssh_port;
let primary = Harness::start_replicating(
"nordisk",
"gunnar",
HashKind::Sha256,
to_standby(&standby, standby_port),
);
let slot = Arc::clone(
primary
.server
.as_ref()
.expect("the server is running")
.replication_slot(),
);
let plane = gunnar_cli::server::CoreControlPlane::new(Arc::clone(&primary.stores))
.with_replication(Arc::clone(&slot));
let frozen = || primary.stores.writes_frozen().load(SeqCst);
// Push once so the flush has real, already-drained work to reconcile.
let work = primary.dir.path().join("work");
let _head = local_repo(
&primary,
&work,
HashKind::Sha256,
"content before abdication\n",
);
primary
.git(
&work,
&[
"push",
&primary.url("nordisk", "gunnar"),
"main:refs/heads/main",
],
)
.expect_ok("git push to the primary");
let s = plane.replication_status(false).expect("status");
assert_eq!(s.targets.len(), 1, "a standby did not stick: {s:?}");
assert!(!frozen(), "the primary must start writable");
// Abdicate. Freeze → flush → demote, one call.
let after = plane.replication_abdicate().expect("abdicate");
// FROZEN: this box stops taking client writes (role STANDBY + the shared
// gate the front door reads is set).
assert_eq!(
after.role,
gunnar_proto::ReplicationRole::Standby as i32,
"abdicate did not turn this box into a standby: {after:?}"
);
assert!(
frozen(),
"abdicate returned but the write-gate is still open — a client could push to the box \
it just handed off from"
);
// DEMOTED: the roster is empty, so this box can never clobber the new
// primary — the split-brain `demote` closes, reached the orderly way.
assert!(
after.targets.is_empty(),
"abdicate left a standby on the roster — the abdicated box can still clobber: {after:?}"
);
assert!(
!after.configured,
"abdicate returned still configured: {after:?}"
);
// Both facts durable: a reboot must not bring back either the writable role
// or the roster.
let json = std::fs::read_to_string(primary.stores.root().join("replication.json"))
.expect("replication.json after abdicate");
assert!(
json.contains("Standby"),
"the STANDBY role is not on disk, so a reboot thaws the abdicated box:\n{json}"
);
assert!(
!json.contains("\"standby\""),
"a standby is still on disk, so a reboot brings the primary role back:\n{json}"
);
// Idempotent: abdicating an already-abdicated box lands in the same place.
let again = plane
.replication_abdicate()
.expect("a second abdicate is Ok, not an error");
assert!(
again.targets.is_empty(),
"the second abdicate resurrected a target: {again:?}"
);
assert_eq!(
again.role,
gunnar_proto::ReplicationRole::Standby as i32,
"the second abdicate un-froze the box: {again:?}"
);
}
#[test]
fn a_push_to_one_gunnar_appears_on_the_second() {
if !tools_present() {
eprintln!("skipping: git and ssh are required");
return;
}
// The standby comes up first: the primary's configuration needs its port.
let standby = Harness::start("nordisk", "gunnar", HashKind::Sha256);
let standby_port = standby.ssh_port;
let primary = Harness::start_replicating(
"nordisk",
"gunnar",
HashKind::Sha256,
to_standby(&standby, standby_port),
);
let work = primary.dir.path().join("work");
let head = local_repo(&primary, &work, HashKind::Sha256, "the primary's content\n");
primary
.git(
&work,
&[
"push",
&primary.url("nordisk", "gunnar"),
"main:refs/heads/main",
],
)
.expect_ok("git push to the primary");
let replicated = wait_for("the push to reach the standby", || {
remote_ref(&standby, "nordisk", "gunnar", "refs/heads/main")
});
// Applied output #1: the standby's ref names the oid `git` on the client
// said it pushed. Compared against the CLIENT's rev-parse, not against
// anything gunnar reports — both halves of gunnar can be wrong the same way.
assert_eq!(
replicated, head,
"the standby's refs/heads/main must be the commit the client pushed to the primary"
);
// Applied output #2: the OBJECTS came too. A ref pointing at an oid the
// standby does not have would satisfy the assertion above and be useless —
// and it is exactly what a replication that copied refs without a pack
// would produce.
//
// Read by CLONING the standby with stock `git`, which is the only way to
// make it hand the objects over on this engine — and a stronger question
// than `git show` in a directory ever was, because the pack has to be
// emitted, sent and indexed by a client nobody here wrote.
let checkout = standby.dir.path().join("replicated");
standby
.git(
standby.dir.path(),
&[
"clone",
"--quiet",
&standby.url("nordisk", "gunnar"),
checkout.to_str().expect("a utf-8 path"),
],
)
.expect_ok("clone the standby");
let content = standby.git(&checkout, &["show", "origin/main:hello.txt"]);
content.expect_ok("read the replicated blob out of a clone of the standby");
assert_eq!(
content.stdout, "the primary's content\n",
"the standby must hold the file contents, not merely a reference to them"
);
// …and the replicator says so too, with numbers rather than a boolean.
//
// Waited for rather than read once: the standby writes the reference and
// sends its report before the primary's worker resumes and does its
// bookkeeping, so the ref being visible above does not yet imply the
// counter has moved. Reading it immediately made this test flake under
// load (seen 2026-08-03, `succeeded: 0` with `attempts: 1`).
let counters = wait_for("the success counter to move", || {
let counters = primary.replicator().counters();
(counters.succeeded >= 1).then_some(counters)
});
assert!(
counters.refs_replicated >= 1,
"at least one reference must have been moved on the standby: {counters:?}"
);
assert_eq!(
counters.failed, 0,
"nothing should have failed against a healthy standby: {counters:?}"
);
let status = primary.replicator().status();
assert_eq!(status.len(), 1);
assert!(status[0].last_success_unix_ms.is_some());
assert!(!status[0].needs_full_resync);
}
/// ★ **AN ANNOTATED TAG REPLICATES WITH THE COMMIT IT POINTS AT.**
///
/// \U0001f534 MEASURED on the reference appliance 2026-08-31. A resync of 25 repositories put
/// 18 second copies on a standby that had none, and failed six — every one of
/// them on an annotated tag, every one with the same refusal raised by the
/// STANDBY's own `receive-pack`:
///
/// ```text
/// alice/skade refs/tags/v0.9.10: missing necessary objects:
/// object 74ee9228820638cc11b76bd973d2c5999f7df27c is not in this
/// repository (peeling a tag)
/// ```
///
/// The primary was not missing it: `git ls-remote` against the primary
/// advertised `refs/tags/v0.9.10` = `9495ef90…` **and** its peel
/// `74ee9228…`. So the pack the primary built carried the tag object and
/// not the commit under it.
///
/// # The shape, measured rather than assumed
///
/// A replication push sets `want` to the ref tips being moved, and for an
/// annotated tag that tip is the **tag object**. The store's `reachable`
/// contributes a tag's own id and stops, so the commit under it arrives only
/// when some *branch* tip's closure happens to contain it — which is why the
/// other nineteen repositories were fine.
///
/// Read off the operator's own clones, 2026-08-31:
///
/// * `skade` — `git branch -a --contains 74ee9228` names **no branch at all**;
/// * `edda` — **25** annotated tags name the same commit `2c7f9919…`, and
/// `--contains` finds it on one branch that the appliance does not mirror.
///
/// So the fixture here is that shape and not the easy one: the tagged commit is
/// on **no branch**. Written first as "tag the tip of `main`", it passed
/// against the defect, because `refs/heads/main` was in the same push and its
/// closure carried the commit the tag needed.
///
/// A clone of these repositories works, which is what hid this: `upload-pack`
/// peels its wants before it asks the store, and the replication path did not.
///
/// # What makes this the measurement
///
/// * the tag is **annotated**, asserted by its object id differing from the
/// commit's — a lightweight tag is a ref pointing straight at a commit and
/// could never fail this way. `local_repo` makes no tag at all, which is why
/// ten replication tests and a `--tags` push never saw this;
/// * the standby is read with stock `git` through its own door, and the peel is
/// read out of a **clone** — a ref advertised for an object the standby does
/// not hold is exactly what a refs-without-the-objects replication leaves;
/// * `git fsck` on that clone asks the connectivity question of something
/// nobody here wrote.
#[test]
fn an_annotated_tag_replicates_with_the_commit_it_points_at() {
if !tools_present() {
eprintln!("skipping: git and ssh are required");
return;
}
let standby = Harness::start("nordisk", "gunnar", HashKind::Sha256);
let standby_port = standby.ssh_port;
let primary = Harness::start_replicating(
"nordisk",
"gunnar",
HashKind::Sha256,
to_standby(&standby, standby_port),
);
let work = primary.dir.path().join("work");
let head = local_repo(&primary, &work, HashKind::Sha256, "on the branch\n");
// A commit on NO branch, tagged. `--orphan` gives it its own root, so it
// shares not one object with `main`, and the branch that carried it is
// deleted before the push: exactly `skade`'s `v0.9.10`.
primary
.git(&work, &["checkout", "--quiet", "--orphan", "released"])
.expect_ok("git checkout --orphan");
std::fs::write(work.join("hello.txt"), "released, and on no branch\n")
.expect("write the tagged file");
primary
.git(&work, &["add", "hello.txt"])
.expect_ok("git add on the orphan");
primary
.git(&work, &["commit", "--quiet", "-m", "the release commit"])
.expect_ok("git commit on the orphan");
let tagged = {
let out = primary.git(&work, &["rev-parse", "HEAD"]);
out.expect_ok("git rev-parse the orphan commit");
out.stdout.trim().to_owned()
};
primary
.git(
&work,
&[
"tag",
"-a",
"v0.9.10",
"-m",
"the release the appliance choked on",
],
)
.expect_ok("git tag -a");
let tag_object = {
let out = primary.git(&work, &["rev-parse", "refs/tags/v0.9.10"]);
out.expect_ok("git rev-parse refs/tags/v0.9.10");
out.stdout.trim().to_owned()
};
primary
.git(&work, &["checkout", "--quiet", "main"])
.expect_ok("git checkout main");
primary
.git(&work, &["branch", "-D", "released"])
.expect_ok("git branch -D released");
// The two premises, asserted rather than assumed. Either one silently false
// turns this back into the test that passed against the defect.
assert_ne!(
tag_object, tagged,
"refs/tags/v0.9.10 names the commit itself, so this is a LIGHTWEIGHT tag and the \
defect cannot show"
);
let contains = primary.git(&work, &["branch", "--contains", &tagged]);
assert!(
contains.stdout.trim().is_empty(),
"the tagged commit is on a branch ({}), so the branch's own closure would carry it \
and the tag edge would never be exercised",
contains.stdout.trim()
);
primary
.git(
&work,
&[
"push",
&primary.url("nordisk", "gunnar"),
"main:refs/heads/main",
"refs/tags/v0.9.10:refs/tags/v0.9.10",
],
)
.expect_ok("git push a branch and an annotated tag to the primary");
// The primary holds both, read through its own door first: "the standby is
// missing the tag" and "the tag was never pushed" are the same output, and
// one of them is a broken premise rather than the bug.
let on_primary = wait_for("the primary to advertise the annotated tag", || {
remote_ref(&primary, "nordisk", "gunnar", "refs/tags/v0.9.10")
});
assert_eq!(on_primary, tag_object);
let replicated = wait_for("the annotated tag to reach the standby", || {
remote_ref(&standby, "nordisk", "gunnar", "refs/tags/v0.9.10")
});
assert_eq!(
replicated, tag_object,
"the standby's refs/tags/v0.9.10 must be the tag OBJECT the client pushed"
);
// The objects. A clone is the only way to make this engine hand them over,
// and the peel is the object the appliance's standby said it did not have.
let checkout = standby.dir.path().join("tagged");
standby
.git(
standby.dir.path(),
&[
"clone",
"--quiet",
&standby.url("nordisk", "gunnar"),
checkout.to_str().expect("a utf-8 path"),
],
)
.expect_ok("clone the standby");
let peeled = standby.git(&checkout, &["rev-parse", "v0.9.10^{commit}"]);
peeled.expect_ok("peel the replicated tag in a clone of the standby");
assert_eq!(
peeled.stdout.trim(),
tagged,
"the tag on the standby does not peel to the commit that was tagged — this is \
`not in this repository (peeling a tag)`, one layer up"
);
let content = standby.git(&checkout, &["show", "v0.9.10:hello.txt"]);
content.expect_ok("read the tagged blob out of a clone of the standby");
assert_eq!(
content.stdout, "released, and on no branch\n",
"the standby holds the tag and its commit but not the tree beneath it"
);
standby
.git(&checkout, &["fsck", "--no-progress"])
.expect_ok("git fsck of a clone taken off the standby");
// And `main` still arrived: a fix that sent the whole repository for every
// push would satisfy everything above.
assert_eq!(
remote_ref(&standby, "nordisk", "gunnar", "refs/heads/main").as_deref(),
Some(head.as_str()),
"the branch in the same push did not reach the standby"
);
let counters = primary.replicator().counters();
assert_eq!(
counters.failed, 0,
"the replicator recorded a failure against a healthy standby: {counters:?}"
);
}
/// **The red half of the test above, kept in the suite.**
///
/// The same two servers, the same push — with the event-bus subscriber not started.
/// Nothing else changes: the replicator exists, its worker is running, its
/// queue is empty because no event ever reached it. The standby must stay
/// empty.
///
/// Without this, `a_push_to_one_gunnar_appears_on_the_second` proves only that
/// the standby ends up with the commit, which a test fixture, a shared temp
/// directory or a stray `git push` could also arrange. With it, the headline is
/// pinned to the subscriber.
#[test]
fn nothing_replicates_when_the_subscriber_is_not_started() {
if !tools_present() {
eprintln!("skipping: git and ssh are required");
return;
}
let standby = Harness::start("nordisk", "gunnar", HashKind::Sha256);
let standby_port = standby.ssh_port;
let broken = ReplicationConfig {
subscriber_disabled: true,
..to_standby(&standby, standby_port)
};
let primary = Harness::start_replicating("nordisk", "gunnar", HashKind::Sha256, broken);
let work = primary.dir.path().join("work");
local_repo(&primary, &work, HashKind::Sha256, "never replicated\n");
primary
.git(
&work,
&[
"push",
&primary.url("nordisk", "gunnar"),
"main:refs/heads/main",
],
)
.expect_ok("git push to the primary still succeeds");
assert!(
never_within(Duration::from_secs(3), || remote_ref(
&standby,
"nordisk",
"gunnar",
"refs/heads/main"
)),
"the standby received a ref although nothing was ever enqueued. Either something \
other than the event-bus subscriber is replicating — in which case the headline test is \
not measuring what it claims — or this seam does not disable it."
);
let counters = primary.replicator().counters();
assert_eq!(
(counters.enqueued, counters.attempts, counters.succeeded),
(0, 0, 0),
"with no subscriber nothing may be enqueued, attempted or succeed: {counters:?}"
);
}
/// A branch deleted on the primary is deleted on the standby.
///
/// The arm a replication that only ever *adds* refs would never reach — and the
/// difference between a copy and a mirror.
#[test]
fn a_branch_deleted_on_the_primary_is_deleted_on_the_standby() {
if !tools_present() {
eprintln!("skipping: git and ssh are required");
return;
}
let standby = Harness::start("nordisk", "gunnar", HashKind::Sha256);
let standby_port = standby.ssh_port;
let primary = Harness::start_replicating(
"nordisk",
"gunnar",
HashKind::Sha256,
to_standby(&standby, standby_port),
);
let work = primary.dir.path().join("work");
local_repo(&primary, &work, HashKind::Sha256, "content\n");
let url = primary.url("nordisk", "gunnar");
primary
.git(&work, &["push", &url, "main:refs/heads/main"])
.expect_ok("push main");
primary
.git(&work, &["push", &url, "main:refs/heads/doomed"])
.expect_ok("push a second branch");
wait_for("the doomed branch to reach the standby", || {
remote_ref(&standby, "nordisk", "gunnar", "refs/heads/doomed")
});
primary
.git(&work, &["push", &url, ":refs/heads/doomed"])
.expect_ok("delete the branch on the primary");
// No `assert!` around this: `wait_for` IS the assertion — it panics naming the
// thing that never happened once WAIT elapses. Wrapping it in `assert!(… == ())`
// compared unit to unit, which is true however replication behaves, and clippy
// rejects it as `==-comparison of unit values`. The guard is the call, not the
// comparison.
wait_for("the deletion to reach the standby", || {
remote_ref(&standby, "nordisk", "gunnar", "refs/heads/doomed")
.is_none()
.then_some(())
});
// The deletion must not have taken the rest with it.
assert!(
remote_ref(&standby, "nordisk", "gunnar", "refs/heads/main").is_some(),
"converging must delete only what the primary no longer has"
);
}
// ───────────────────────────────────────────────────────────────────────────
// the standby being down
// ───────────────────────────────────────────────────────────────────────────
/// **A dead standby must not fail the push**, and must not be silent about it.
///
/// Three assertions, and all three are needed. The push succeeding is the
/// requirement; the counter and the event are what tell a replicator that
/// *tried and failed* apart from one that never ran — and a replicator that
/// never ran would pass the first assertion perfectly.
#[test]
fn a_dead_standby_does_not_fail_the_push_and_is_not_silent() {
if !tools_present() {
eprintln!("skipping: git and ssh are required");
return;
}
// A standby is still started, only so its key and known-hosts file exist;
// the primary is pointed at a port nothing is listening on.
let corpse = Harness::start("nordisk", "gunnar", HashKind::Sha256);
let primary = Harness::start_replicating(
"nordisk",
"gunnar",
HashKind::Sha256,
to_standby(&corpse, DEAD_PORT),
);
let mut tap = BusTap::open(&primary.runtime, primary.bus());
let work = primary.dir.path().join("work");
let head = local_repo(&primary, &work, HashKind::Sha256, "still accepted\n");
let push = primary.git(
&work,
&[
"push",
&primary.url("nordisk", "gunnar"),
"main:refs/heads/main",
],
);
// THE requirement. A push that succeeded locally has succeeded, whatever
// the standby is doing.
push.expect_ok("the push must succeed although the standby is unreachable");
// And the primary really did accept it — otherwise "the push succeeded"
// would be a claim about an exit code and nothing else.
assert_eq!(
remote_ref(&primary, "nordisk", "gunnar", "refs/heads/main").as_deref(),
Some(head.as_str()),
"the primary must hold the commit it reported accepting"
);
let failures = tap.failures(1, "the standby is unreachable");
let first = &failures[0].1;
assert_eq!(first.target, "standby");
assert_eq!(
first.repo.as_ref().map(|r| r.to_string()).as_deref(),
Some("nordisk/gunnar"),
"the event must name the repository that did not replicate"
);
assert!(
first.remote_url.contains(&format!(":{DEAD_PORT}/")),
"the event must name the URL that was tried: {:?}",
first.remote_url
);
assert_eq!(first.attempt, 1, "the first failure is attempt 1");
assert!(
first.retry_in_ms > 0,
"a failure that will be retried must say when: {first:?}"
);
assert!(
!first.reason.is_empty(),
"a failure with no reason is a failure nobody can act on"
);
assert!(
!first.needs_full_resync,
"an ordinary connection failure is retried, not a resync"
);
let counters = primary.replicator().counters();
assert!(
counters.failed >= 1,
"the failure counter must have moved: {counters:?}"
);
assert_eq!(
counters.succeeded, 0,
"nothing can have succeeded against a dead port: {counters:?}"
);
let status = primary.replicator().status();
assert!(
status[0].last_error.is_some(),
"the standby's status must carry the last error: {status:?}"
);
assert!(status[0].consecutive_failures >= 1);
assert!(
status[0].pending >= 1,
"the job must still be queued for a retry, not dropped: {status:?}"
);
}
/// The backoff really backs off — asserted twice, because either assertion
/// alone is blind.
#[test]
fn the_retry_delay_doubles_and_the_clock_agrees() {
if !tools_present() {
eprintln!("skipping: git and ssh are required");
return;
}
let corpse = Harness::start("nordisk", "gunnar", HashKind::Sha256);
let primary = Harness::start_replicating(
"nordisk",
"gunnar",
HashKind::Sha256,
to_standby(&corpse, DEAD_PORT),
);
let mut tap = BusTap::open(&primary.runtime, primary.bus());
let work = primary.dir.path().join("work");
local_repo(&primary, &work, HashKind::Sha256, "content\n");
primary
.git(
&work,
&[
"push",
&primary.url("nordisk", "gunnar"),
"main:refs/heads/main",
],
)
.expect_ok("the push succeeds");
let failures = tap.failures(3, "the standby stays unreachable");
// Half one: the schedule, by VALUE. `retry_base_ms` is 200 in this
// configuration, so a constant delay — the thing a clock measurement alone
// cannot rule out — fails right here.
let scheduled: Vec<(u32, u64)> = failures
.iter()
.take(3)
.map(|(_, f)| (f.attempt, f.retry_in_ms))
.collect();
assert_eq!(
scheduled,
vec![(1, 200), (2, 400), (3, 800)],
"the attempt number and the delay before the next attempt must both grow"
);
// Half two: the clock — because a schedule that is computed, published and
// then ignored would pass half one perfectly.
//
// **The measurement is the DIFFERENCE between consecutive gaps, not the
// elapsed time**, and that is the whole of why this assertion can go red.
// Each gap is `connect_cost + scheduled_delay`; the connect cost is the same
// every time and cancels, leaving `400 ms − 200 ms = 200 ms` when the delay
// is applied and ≈ 0 when it is not. Measured 2026-08-03: the first version
// of this assertion compared *total elapsed* against 600 ms and **passed
// against a worker that retried immediately**, because three SSH connection
// attempts cost more than 600 ms on their own. It was a guard that could
// never have gone red, found by trying to make it.
let gap = |a: usize, b: usize| failures[b].0 - failures[a].0;
let first_gap = gap(0, 1);
let second_gap = gap(1, 2);
assert!(
second_gap - first_gap >= 100,
"the gap between attempts 2 and 3 ({second_gap} ms) is not meaningfully longer than the \
gap between 1 and 2 ({first_gap} ms). The schedule doubles from 200 ms to 400 ms, and \
the fixed connection cost cancels out of the difference, so anything under 100 ms here \
means the delay is computed and then ignored."
);
}
// ───────────────────────────────────────────────────────────────────────────
// the bound
// ───────────────────────────────────────────────────────────────────────────
/// The queue bound, driven through the **public** enqueue path: the counter,
/// the sticky `needs_full_resync` flag and the `ReplicationFailed` event are
/// all asserted where a push would produce them.
#[test]
fn the_queue_bound_refuses_and_says_so_on_the_stream() {
let dir = tempfile::tempdir().expect("a temp dir");
let stores = Arc::new(
gunnar_cli::server::ServerStores::open(dir.path().join("data")).expect("open the stores"),
);
stores.catalog().create_namespace("ns").expect("namespace");
for i in 0..6 {
stores
.catalog()
.create(NewRepo::new(
RepoRef::new("ns", &format!("r{i}")).expect("a legal path"),
))
.expect("create the repository");
}
let runtime = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.expect("a runtime");
let bus = Arc::new(EventBus::new(256));
let mut tap = BusTap::open(&runtime, Arc::clone(&bus));
let config = ReplicationConfig {
targets: vec![ReplicationTarget {
name: "standby".into(),
scope: ReplicationScope::Server,
remote: format!("ssh://git@127.0.0.1:{DEAD_PORT}"),
identity: None,
known_hosts: Some(dir.path().join("known_hosts")),
strict_host_key: false,
visibility: Default::default(),
control: None,
}],
queue_capacity: 2,
..ReplicationConfig::default()
};
// Deliberately NOT spawned: nothing drains, so the bound is what decides
// what the queue holds. A worker running here would make the test a race
// against its own drain.
let replicator = Replicator::new(Arc::clone(&stores), Arc::clone(&bus), config);
for i in 0..6 {
replicator.enqueue(&RepoRef::new("ns", &format!("r{i}")).unwrap());
}
let counters = replicator.counters();
assert_eq!(
counters.enqueued, 2,
"only the bound may be admitted: {counters:?}"
);
assert_eq!(
counters.refused_queue_full, 4,
"every repository past the bound must be REFUSED, and counted: {counters:?}"
);
let status = replicator.status();
assert_eq!(status[0].pending, 2, "the queue must not have grown");
assert!(
status[0].needs_full_resync,
"hitting the bound leaves a hole retrying will not close, and the standby must say so"
);
// …and it is on the event stream, not only in a counter nobody reads.
let mut failures = Vec::new();
let deadline = Instant::now() + Duration::from_secs(5);
while failures.len() < 4 && Instant::now() < deadline {
failures.extend(tap.drain_failures());
std::thread::sleep(Duration::from_millis(20));
}
assert_eq!(
failures.len(),
4,
"each refusal must be published, not merely counted"
);
assert!(
failures.iter().all(|f| f.needs_full_resync),
"a queue-full refusal must be flagged as needing a resync: {failures:?}"
);
assert!(
failures.iter().all(|f| f.retry_in_ms == 0),
"a refused job is not retried; a non-zero delay would promise one: {failures:?}"
);
// A resync clears the flag, because it re-queues everything — but only
// once there is room. While the queue is still full it must NOT clear.
replicator.resync().expect("resync");
assert!(
replicator.status()[0].needs_full_resync,
"a resync that itself hit the bound has resynced nothing and must leave the flag set"
);
runtime.shutdown_background();
}
/// ★ **Deleting a repository CANCELS the replication work outstanding for it,
/// and asks nobody for a full resync.**
///
/// 🔴 MEASURED 2026-09-20 on the live estate, in the appliance's own log:
///
/// ```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 made a repository, a push enqueued it for the standby, the fixture
/// deleted it, and the queued job outlived the repository. The worker took it,
/// the catalog had no row, the miss was classified as a LOCAL fault — "this
/// server cannot read its own repository" — and `MAX_LOCAL_ATTEMPTS` of those
/// set the sticky `needs_full_resync`. A test tidying up after itself put the
/// replication subsystem into the state that demands a resync of **every**
/// repository on the box.
///
/// Driven through the public path, deliberately: the delete is the
/// `RepoDeleted` event `Repos.Delete` publishes, on a real bus, with the
/// subscriber and the workers running — the counter, the flag and the queue are
/// all read where a real delete would leave them.
///
/// The standby is `DEAD_PORT`, and that is not incidental: it means every
/// attempt that is NOT about the missing catalog row fails REMOTELY and
/// retries, so a `needs_full_resync` seen here can only have come from the
/// local-fault arm this fixes.
#[test]
fn deleting_a_repository_cancels_its_replication_work_and_asks_for_no_resync() {
let dir = tempfile::tempdir().expect("a temp dir");
let stores = Arc::new(
gunnar_cli::server::ServerStores::open(dir.path().join("data")).expect("open the stores"),
);
stores
.catalog()
.create_namespace("proof")
.expect("namespace");
let doomed = RepoRef::new("proof", "f13-1789907688270").expect("a legal path");
stores
.catalog()
.create(NewRepo::new(doomed.clone()))
.expect("create the repository");
let runtime = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.expect("a runtime");
let bus = Arc::new(EventBus::new(256));
let config = ReplicationConfig {
targets: vec![ReplicationTarget {
name: "standby".into(),
scope: ReplicationScope::Server,
remote: format!("ssh://git@127.0.0.1:{DEAD_PORT}"),
identity: None,
known_hosts: Some(dir.path().join("known_hosts")),
strict_host_key: false,
visibility: Default::default(),
control: None,
}],
queue_capacity: 8,
// Short, so a remote failure that beats the delete comes back around
// inside the test's patience instead of sitting in a backoff.
retry_base_ms: 20,
retry_max_ms: 40,
..ReplicationConfig::default()
};
let replicator = Arc::new(Replicator::new(
Arc::clone(&stores),
Arc::clone(&bus),
config,
));
let _tasks = runtime.block_on(async { replicator.spawn() });
// A push enqueues it.
replicator.enqueue(&doomed);
// …and then the operator deletes it: the catalog row goes, and
// `Repos.Delete` publishes exactly this event.
stores
.catalog()
.delete(&doomed)
.expect("delete the repository");
bus.publish(event::Payload::RepoDeleted(gunnar_proto::RepoDeleted {
repo: Some(gunnar_proto::RepoPath {
namespace: "proof".into(),
repo: "f13-1789907688270".into(),
}),
principal_id: "operator".into(),
}));
// The job goes, one way or the other: the subscriber drops it out of the
// queue, or — if the worker already had it in flight — the worker's own
// `Fault::Gone` arm drops it when the catalog comes back empty. Both are
// the same verdict and both count it.
let cancelled = wait_for("the deleted repository's job to be cancelled", || {
let counters = replicator.counters();
(counters.cancelled > 0).then_some(counters)
});
assert_eq!(
cancelled.cancelled, 1,
"one delete, one cancellation: {cancelled:?}"
);
// THE ASSERTION. Give the worker room to have got it wrong — five local
// attempts at a 20 ms base is well inside this — and then read the flag.
std::thread::sleep(Duration::from_millis(600));
let status = replicator.status();
assert!(
!status[0].needs_full_resync,
"THE DEFECT: a repository the operator deleted demanded a full resync of the whole \
standby: {status:?}"
);
assert_eq!(
status[0].pending, 0,
"the cancelled job must be out of the queue, not merely unexecutable: {status:?}"
);
let counters = replicator.counters();
assert_eq!(
counters.stalled_local, 0,
"a deleted repository is not this server failing to read its own: {counters:?}"
);
assert!(
status[0].last_error.is_none(),
"the standby is not at fault for a repository the operator removed, and must not be \
left wearing its error: {status:?}"
);
runtime.shutdown_background();
}
/// Coalescing through the public path: many pushes to one repository against a
/// stalled standby cost one queue slot, which is what makes the bound a bound
/// rather than a delay.
#[test]
fn repeated_pushes_to_one_repository_do_not_grow_the_queue() {
let dir = tempfile::tempdir().expect("a temp dir");
let stores = Arc::new(
gunnar_cli::server::ServerStores::open(dir.path().join("data")).expect("open the stores"),
);
let bus = Arc::new(EventBus::new(16));
let config = ReplicationConfig {
targets: vec![ReplicationTarget {
name: "standby".into(),
scope: ReplicationScope::Namespace {
namespace: "ns".into(),
},
remote: format!("ssh://git@127.0.0.1:{DEAD_PORT}"),
identity: None,
known_hosts: Some(dir.path().join("known_hosts")),
strict_host_key: false,
visibility: Default::default(),
control: None,
}],
queue_capacity: 4,
..ReplicationConfig::default()
};
let replicator = Replicator::new(stores, bus, config);
let hot = RepoRef::new("ns", "hot").unwrap();
for _ in 0..10_000 {
replicator.enqueue(&hot);
}
let counters = replicator.counters();
assert_eq!(
counters.enqueued, 1,
"one repository, one slot: {counters:?}"
);
assert_eq!(counters.coalesced, 9_999, "{counters:?}");
assert_eq!(
counters.refused_queue_full, 0,
"coalescing must not be reported as a refusal: {counters:?}"
);
assert_eq!(replicator.status()[0].pending, 1);
assert!(
!replicator.status()[0].needs_full_resync,
"coalescing loses nothing, so it must never mark the standby as holed"
);
// A repository OUTSIDE the target's namespace is not this target's work.
let elsewhere = RepoRef::new("other", "repo").unwrap();
assert!(
replicator.enqueue(&elsewhere).is_empty(),
"a namespace-scoped target must not take a repository from another namespace"
);
assert_eq!(replicator.status()[0].pending, 1);
}
// ───────────────────────────────────────────────────────────────────────────
// BACKFILL — a standby is a copy of what is pushed AFTER it exists
// ───────────────────────────────────────────────────────────────────────────
/// Every ref a gunnar advertises, read with the **stock `git` binary** over the
/// wire. `None` when `ls-remote` itself failed.
///
/// The counting reader for the same door [`remote_ref`] uses, because "3818
/// refs here and 0 there" is the sentence a backfill is judged by and one named
/// ref cannot say it.
fn remote_refs(harness: &Harness, namespace: &str, repo: &str) -> Option<Vec<String>> {
let url = harness.url(namespace, repo);
let out = harness.git(harness.dir.path(), &["ls-remote", &url]);
if !out.ok {
return None;
}
Some(
out.stdout
.lines()
.filter_map(|line| {
line.split_once('\t')
.map(|(_, name)| name.trim().to_owned())
})
.collect(),
)
}
/// ★ **THE BACKFILL TEST. A repository written before the standby existed
/// arrives only when somebody asks for it.**
///
/// 🔴 MEASURED on the reference appliance 2026-08-31, three ways: a target added against
/// 22 existing repositories left the standby with **0**; a repository created
/// while no target existed was still absent 60 s after one was added; and one
/// further commit to it put its **whole history** on the standby. So a
/// repository's history rides along on its NEXT push, and a repository nobody
/// pushes to again never arrives. `alice/linux-mirror`: **primary 3818 refs,
/// standby 0**, with 21 of 25 repositories in the same position.
///
/// This test is that measurement, in the suite:
///
/// 1. the primary holds a repository, pushed while **no standby was
/// configured** — asserted through the primary's OWN door first, so the red
/// below cannot be a broken reader saying "absent";
/// 2. a standby is configured over the control plane and given time to do
/// nothing — **RED**: `ls-remote` against the standby returns the repository
/// with no refs at all;
/// 3. `Mirror.ResyncReplication` is called for that standby by name;
/// 4. **GREEN**: the same `ls-remote` returns the same ref list as the primary,
/// and a `git clone` off the standby produces the same object id and passes
/// `git fsck`.
///
/// Step 2 is the half that makes step 4 mean something. Without it the test
/// would pass against a replicator that enqueued on `apply`, against one that
/// enqueued on start-up, and against a fixture that copied the directory.
#[test]
fn a_repository_written_before_the_standby_existed_arrives_only_after_a_resync() {
if !tools_present() {
eprintln!("skipping: git and ssh are required");
return;
}
use gunnar_grpc::ControlPlane;
let standby = Harness::start("nordisk", "gunnar", HashKind::Sha256);
let standby_port = standby.ssh_port;
// **No replication configuration.** This is the appliance as it was on the
// day the repositories were written.
let primary = Harness::start("nordisk", "gunnar", HashKind::Sha256);
let work = primary.dir.path().join("work");
let head = local_repo(
&primary,
&work,
HashKind::Sha256,
"written before the standby existed\n",
);
primary
.git(
&work,
&[
"push",
&primary.url("nordisk", "gunnar"),
"main:refs/heads/main",
"main:refs/heads/release",
"--tags",
],
)
.expect_ok("git push to a primary with no standby");
// The reader is proved on the PRIMARY first. "The standby has no refs" and
// "this reader cannot read refs" are the same output, and one of them is
// the bug found earlier by having a test pass vacuously.
let on_primary = remote_refs(&primary, "nordisk", "gunnar")
.expect("ls-remote against the primary must work, or the red below proves nothing");
assert!(
on_primary.iter().any(|r| r == "refs/heads/main"),
"the primary does not hold what was just pushed to it: {on_primary:?}"
);
let slot = Arc::clone(
primary
.server
.as_ref()
.expect("the server is running")
.replication_slot(),
);
let plane = gunnar_cli::server::CoreControlPlane::new(Arc::clone(&primary.stores))
.with_replication(Arc::clone(&slot));
let known_hosts = primary.dir.path().join("replication_known_hosts");
let target = gunnar_proto::ReplicationTarget {
name: "standby".into(),
scope: Some(gunnar_proto::ReplicationScope {
of: Some(gunnar_proto::replication_scope::Of::Server(true)),
}),
remote: format!("ssh://git@127.0.0.1:{standby_port}"),
identity_path: standby.key_path.display().to_string(),
known_hosts_path: known_hosts.display().to_string(),
strict_host_key: false,
public_only: false,
control: String::new(),
};
let out = plane
.replication_set(target.clone())
.expect("SetReplication");
assert!(out.configured, "the replicator is not running: {out:?}");
// ── RED ────────────────────────────────────────────────────────────────
// The target is configured, the worker is running, the repository is right
// there in the catalogue — and NOTHING is enqueued, because replication is
// a subscriber to the push bus and no push happened.
assert!(
never_within(Duration::from_secs(5), || {
remote_ref(&standby, "nordisk", "gunnar", "refs/heads/main")
}),
"the standby acquired the repository without a push and without a resync, so this test \
cannot tell a backfill from whatever did that"
);
let red = remote_refs(&standby, "nordisk", "gunnar")
.expect("ls-remote against the standby must work");
assert!(
red.is_empty(),
"RED: the standby was expected to hold NO refs before the resync, and holds {red:?}"
);
// …and the control plane SAYS so, per repository. This is the row a web
// console draws: an aggregate that reads `synced 0 · failed 0` over a
// standby holding nothing is what lets an admin conclude they are safe.
let status = plane
.replication_status(false)
.expect("status before the resync");
let row = status
.repos
.iter()
.find(|r| r.standby == "standby" && r.repo.as_ref().is_some_and(|p| p.repo == "gunnar"))
.unwrap_or_else(|| {
panic!(
"the control plane must have a row for a repository with NO second copy — a \
list built from successes can only ever show the ones that do: {status:?}"
)
});
assert_eq!(
row.last_success_unix_ms, 0,
"a repository the standby has never received must read 0 = NEVER: {row:?}"
);
assert!(!row.stopped, "{row:?}");
assert!(
status.observed_since_unix_ms > 0,
"the record's start instant must be published, or `0 = never` cannot be told apart \
from `no record since the last restart`: {status:?}"
);
assert!(
plane
.replication_resync("no-such-standby", None, &[])
.err()
.is_some(),
"resyncing a standby that is not configured must be an error: \"resynced\" about a \
standby that was never there reads exactly like \"resynced\" about the one you meant"
);
// ── the verb ───────────────────────────────────────────────────────────
let report = plane
.replication_resync("standby", None, &[])
.expect("ResyncReplication for a configured standby");
let row = report
.standbys
.iter()
.find(|s| s.standby == "standby")
.unwrap_or_else(|| panic!("the resync said nothing about the standby: {report:?}"));
assert_eq!(row.in_scope, 1, "one repository is in scope: {row:?}");
assert_eq!(
row.queued, 1,
"it must be QUEUED, not merely counted: {row:?}"
);
assert_eq!(row.refused, 0, "{row:?}");
// ── GREEN ──────────────────────────────────────────────────────────────
let replicated = wait_for("the resync to put the repository on the standby", || {
remote_ref(&standby, "nordisk", "gunnar", "refs/heads/main")
});
assert_eq!(
replicated, head,
"the standby's refs/heads/main must be the commit the CLIENT pushed to the primary"
);
let mut green = wait_for("every ref to reach the standby", || {
let mut refs = remote_refs(&standby, "nordisk", "gunnar")?;
refs.sort();
(refs.len() >= on_primary.len()).then_some(refs)
});
green.sort();
let mut expected = on_primary.clone();
expected.sort();
assert_eq!(
green,
expected,
"GREEN: the standby must advertise the same refs as the primary. primary {} refs, \
standby {} refs",
expected.len(),
green.len()
);
// The objects came too, read by CLONING the standby with stock `git` — the
// only way to make this engine hand them over, and a question a
// refs-without-a-pack replication would fail.
let checkout = standby.dir.path().join("backfilled");
standby
.git(
standby.dir.path(),
&[
"clone",
"--quiet",
&standby.url("nordisk", "gunnar"),
checkout.to_str().expect("a utf-8 path"),
],
)
.expect_ok("clone the backfilled repository off the standby");
let cloned_head = standby.git(&checkout, &["rev-parse", "HEAD"]);
cloned_head.expect_ok("rev-parse in a clone of the standby");
assert_eq!(
cloned_head.stdout.trim(),
head,
"the clone's HEAD must be the same object id the client pushed to the primary"
);
standby
.git(&checkout, &["fsck", "--no-progress"])
.expect_ok("git fsck of a clone taken off the backfilled standby");
let content = standby.git(&checkout, &["show", "HEAD:hello.txt"]);
content.expect_ok("read the backfilled blob out of a clone of the standby");
assert_eq!(
content.stdout, "written before the standby existed\n",
"the standby must hold the file contents, not merely a reference to them"
);
// The per-repository row now says the copy exists, with the ref count a
// console compares against the primary's.
let after = wait_for("the per-repository row to record the copy", || {
let status = plane.replication_status(false).ok()?;
let row = status.repos.iter().find(|r| {
r.standby == "standby" && r.repo.as_ref().is_some_and(|p| p.repo == "gunnar")
})?;
(row.last_success_unix_ms > 0).then(|| row.clone())
});
// `refs_at_last_success` counts MIRRORED refs — `refs/heads/*` and
// `refs/tags/*` — and `git ls-remote` also prints `HEAD`, which is not one.
// Measured, not assumed: this assertion was written against the raw
// `ls-remote` count and came back 2 against 3.
let mirrored = expected.iter().filter(|r| *r != "HEAD").count();
assert_eq!(
after.refs_at_last_success as usize,
mirrored,
"the row must report the same number of MIRRORED refs the standby advertises \
({mirrored} of the {} lines ls-remote prints): {after:?}",
expected.len()
);
assert!(!after.stopped && after.last_error.is_empty(), "{after:?}");
// ── `--except`: the shape a real backfill is run in ────────────────────
//
// On the reference appliance 24 of the 25 repositories are a few hundred megabytes
// together and ONE — `alice/linux-mirror`, ~8.7 GB and 13.9 M objects —
// projects 182–235 GiB against a standby whose `/data` is 224.9 GiB. The
// safe twenty-four have to be shippable in one call, and the excluded one
// has to be NAMED rather than quietly passed over.
let excluded = plane
.replication_resync(
"standby",
None,
&[gunnar_proto::RepoPath {
namespace: "nordisk".into(),
repo: "gunnar".into(),
}],
)
.expect("a resync with an exclusion");
let row = &excluded.standbys[0];
assert_eq!(
row.in_scope, 0,
"the only repository was excluded, so nothing is in scope: {row:?}"
);
assert_eq!(
row.excluded,
vec!["nordisk/gunnar".to_owned()],
"an exclusion must be reported BY NAME — a typo that matched nothing and a real \
exclusion are the same command with very different outcomes: {row:?}"
);
// …and an exclusion that matched nothing says so by being absent from the
// list, rather than by looking identical to one that matched.
let missed = plane
.replication_resync(
"standby",
None,
&[gunnar_proto::RepoPath {
namespace: "nordisk".into(),
repo: "spelled-wrong".into(),
}],
)
.expect("a resync whose exclusion matches nothing");
assert!(missed.standbys[0].excluded.is_empty(), "{missed:?}");
assert_eq!(
missed.standbys[0].in_scope, 1,
"a mistyped exclusion must not silently exclude something else: {missed:?}"
);
// Running it again is the resume, and costs nothing: the standby already
// has everything, so the job is queued, finds an empty command list and
// reports up to date rather than shipping the pack twice.
let again = plane
.replication_resync("standby", None, &[])
.expect("a second resync");
let row = &again.standbys[0];
assert_eq!(
row.queued + row.already_pending,
1,
"a repeated resync must still cover the repository: {row:?}"
);
assert_eq!(row.refused, 0, "{row:?}");
// And one repository can be named on its own — the shape a repository too
// large for tonight's window is decided about with.
let one = plane
.replication_resync(
"standby",
Some(&gunnar_proto::RepoPath {
namespace: "nordisk".into(),
repo: "gunnar".into(),
}),
&[],
)
.expect("a single-repository resync");
assert_eq!(one.standbys[0].in_scope, 1, "{one:?}");
assert!(
plane
.replication_resync(
"standby",
Some(&gunnar_proto::RepoPath {
namespace: "nordisk".into(),
repo: "not-a-repository".into(),
}),
&[],
)
.is_err(),
"resyncing a repository this server does not have must be an error rather than a \
backfill of nothing — the ordinary way to get there is a typo"
);
}
/// ★ **Re-adding a healthy standby must not make it read as broken.**
///
/// 🔴 MEASURED on the reference appliance 2026-08-31: re-adding `standby-fi-hel1` **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 changing one flag is the ordinary operator move —
/// and it printed the console's alarm string about a healthy box, which is
/// worse than no alarm because it teaches an operator to ignore the real one.
///
/// Seen RED by restoring `Counters::default()` and `TargetState::default()` in
/// `Replicator::new_carrying`: `last success` comes back `None` and `synced`
/// comes back 0 on the assertions below.
///
/// The other direction is asserted too, and it is the half that keeps this from
/// being a lie of its own: a target whose **URL** changed is a DIFFERENT box,
/// and carrying `last success 3 s ago` onto a peer that has never been dialled
/// would be exactly the alarm-that-lies this test is about.
#[test]
fn re_adding_a_standby_keeps_what_it_had_already_done() {
if !tools_present() {
eprintln!("skipping: git and ssh are required");
return;
}
use gunnar_grpc::ControlPlane;
let standby = Harness::start("nordisk", "gunnar", HashKind::Sha256);
let standby_port = standby.ssh_port;
let primary = Harness::start("nordisk", "gunnar", HashKind::Sha256);
let slot = Arc::clone(
primary
.server
.as_ref()
.expect("the server is running")
.replication_slot(),
);
let plane = gunnar_cli::server::CoreControlPlane::new(Arc::clone(&primary.stores))
.with_replication(Arc::clone(&slot));
let known_hosts = primary.dir.path().join("replication_known_hosts");
let target = |strict: bool| gunnar_proto::ReplicationTarget {
name: "standby".into(),
scope: Some(gunnar_proto::ReplicationScope {
of: Some(gunnar_proto::replication_scope::Of::Server(true)),
}),
remote: format!("ssh://git@127.0.0.1:{standby_port}"),
identity_path: standby.key_path.display().to_string(),
known_hosts_path: known_hosts.display().to_string(),
strict_host_key: strict,
public_only: false,
control: String::new(),
};
plane
.replication_set(target(false))
.expect("SetReplication");
let work = primary.dir.path().join("work");
let head = local_repo(&primary, &work, HashKind::Sha256, "healthy\n");
primary
.git(
&work,
&[
"push",
&primary.url("nordisk", "gunnar"),
"main:refs/heads/main",
],
)
.expect_ok("git push to the primary");
let landed = wait_for("the push to reach the standby", || {
remote_ref(&standby, "nordisk", "gunnar", "refs/heads/main")
});
assert_eq!(landed, head);
let before = wait_for("the success to be recorded", || {
let out = plane.replication_status(false).ok()?;
(out.synced >= 1 && out.standbys.first()?.last_success_unix_ms > 0).then_some(out)
});
// **The operator move that lied**: the same standby, one flag changed.
let after = plane
.replication_set(target(true))
.expect("SetReplication again, with strict host keys on");
let row = after
.standbys
.first()
.unwrap_or_else(|| panic!("the standby vanished: {after:?}"));
assert!(
row.last_success_unix_ms > 0,
"a healthy standby read `last success NEVER` after one flag was changed: {after:?}"
);
assert_eq!(
row.last_success_unix_ms, before.standbys[0].last_success_unix_ms,
"the recorded success must be the SAME instant, not a fresh one: {after:?}"
);
assert!(
!row.stale,
"a fully-synced standby read STALE after a flag was changed: {after:?}"
);
assert!(
after.synced >= before.synced,
"the server-wide counters were reset by a reconfigure: {} -> {}",
before.synced,
after.synced
);
// …and the flag really did change, or the assertions above are about a
// call that did nothing.
assert!(
after.targets[0].strict_host_key,
"the reconfigure did not apply: {after:?}"
);
// The other direction: a NEW URL under the same name is a different box,
// and its history must NOT be carried.
let moved = plane
.replication_set(gunnar_proto::ReplicationTarget {
remote: format!("ssh://git@127.0.0.1:{DEAD_PORT}"),
..target(true)
})
.expect("SetReplication pointing the same name at another box");
let row = &moved.standbys[0];
assert_eq!(
row.last_success_unix_ms, 0,
"a standby whose URL changed has never taken a byte, and saying otherwise is the same \
lie this test exists about: {moved:?}"
);
assert!(
row.needs_full_resync,
"a standby pointed at a new box needs a full resync: {moved:?}"
);
}
// ───────────────────────────────────────────────────────────────────────────
// what a mirror does NOT carry
// ───────────────────────────────────────────────────────────────────────────
/// ★ **The green row that was 56% of the truth.**
///
/// 🔴 MEASURED on the live primary 2026-08-31, in the first restore drill —
/// repositories cloned back OFF the standby: across all 24 replicated
/// repositories **1292 refs exist and 567 of them are not on the standby**.
/// `alice/korp` alone: 144 refs, 94 of them outside `refs/heads/` and
/// `refs/tags/`. Most are `refs/remotes/origin/*` left by pushes from mirror
/// clones, but not all — `refs/superseded/wip-pre-reset-…` and
/// `refs/salvage/*` are places somebody parked work *because it
/// mattered*, and for some of them the primary is the only copy.
///
/// `gunnar_client::mirror::is_mirrored_ref` deciding to carry branches and tags
/// only is a product choice and is NOT changed here. What is changed is that
/// the surfaces said nothing: `replication list` printed `copied alice/korp
/// 50 refs` and the console drew a green row, and 50 of 144 is not what either
/// of them looked like.
///
/// **Both directions, because a counter that only prints when it is non-zero
/// cannot be told from one that is broken.** A repository whose refs are all
/// branches reports `0`, and one with a single salvage ref reports exactly `1`.
///
/// Seen RED by returning `RefCoverage::default()` from `walk_ref_coverage`:
/// the salvage repository then reports 0 unmirrored refs and 0 total, and the
/// `Some(0)` / `Some(1)` assertions on the wire fail — a blank where a number
/// belongs.
#[test]
fn the_rows_count_the_refs_a_mirror_does_not_carry() {
if !tools_present() {
eprintln!("skipping: git and ssh are required");
return;
}
use gunnar_grpc::ControlPlane;
let standby = Harness::start("nordisk", "gunnar", HashKind::Sha256);
let standby_port = standby.ssh_port;
let primary = Harness::start_replicating(
"nordisk",
"gunnar",
HashKind::Sha256,
ReplicationConfig {
// The POLLED path walks on every ask here. The operator's path
// (`ReplicationStatus`, asserted at the end of this test) is always
// fresh; this knob is what the console's polling would reuse, and a
// test that pushes a ref and then asks about it must not be reading
// last minute's answer.
ref_coverage_ttl_ms: 0,
..to_standby(&standby, standby_port)
},
);
// One repository whose refs are ALL branches, and one with a salvage ref
// beside its branch. Both are pushed with stock git, so the refs are real
// rows in the primary's own reference database rather than fixtures.
let work = primary.dir.path().join("work");
local_repo(&primary, &work, HashKind::Sha256, "mirrored\n");
primary
.git(
&work,
&[
"push",
&primary.url("nordisk", "gunnar"),
"main:refs/heads/main",
],
)
.expect_ok("push a branch");
let salvage = RepoRef::new("nordisk", "salvage").expect("a legal path");
let created = primary
.stores
.catalog()
.create(NewRepo::new(salvage.clone()).hash_kind(HashKind::Sha256))
.expect("create the second repository");
primary
.stores
.repo_store(&created)
.expect("materialise it on disk");
primary
.stores
.auth()
.grant(gunnar_core::Grant::repo(
"tester",
salvage.clone(),
gunnar_core::Level::Write,
))
.expect("grant the test principal write on it");
let url = primary.url("nordisk", "salvage");
primary
.git(&work, &["push", &url, "main:refs/heads/main"])
.expect_ok("push a branch to the salvage repository");
primary
.git(&work, &["push", &url, "main:refs/superseded/wip-pre-reset"])
.expect_ok("push a ref NO mirror will carry");
let replicator = primary.replicator();
let by_repo = replicator.ref_coverage_by_repo(std::time::Duration::ZERO);
let mirrored = by_repo
.get("nordisk/gunnar")
.unwrap_or_else(|| panic!("no coverage row for nordisk/gunnar: {by_repo:?}"));
assert_eq!(
(mirrored.total, mirrored.unmirrored),
(1, 0),
"a repository whose refs are all branches must report ZERO not mirrored — not a \
blank, or the number cannot be told from a broken counter"
);
let parked = by_repo
.get("nordisk/salvage")
.unwrap_or_else(|| panic!("no coverage row for nordisk/salvage: {by_repo:?}"));
assert_eq!(
(parked.total, parked.unmirrored),
(2, 1),
"the salvage ref is on this box only and nothing counted it: {parked:?}"
);
let server_wide = replicator.ref_coverage(std::time::Duration::ZERO);
assert_eq!(
(server_wide.total, server_wide.unmirrored),
(3, 1),
"the summary line's numbers are not the sum of the rows: {server_wide:?}"
);
assert_eq!(server_wide.mirrorable(), 2);
// …and all of it reaches the wire `replication list` prints from.
let slot = Arc::clone(
primary
.server
.as_ref()
.expect("the server is running")
.replication_slot(),
);
let plane = gunnar_cli::server::CoreControlPlane::new(Arc::clone(&primary.stores))
.with_replication(Arc::clone(&slot));
// `--refs`: the ref-coverage half is opt-in, so ask for it — this
// is the path that must still produce the correct shortfall.
let out = plane.replication_status(true).expect("ReplicationStatus");
assert_eq!(
(out.refs_total, out.refs_unmirrored),
(3, 1),
"the summary line carries no shortfall: {out:?}"
);
// …and a PLAIN status opens no ref database, so it carries no ref figure at
// all — zeros and absent per-repository counts, not a walk.
let plain = plane.replication_status(false).expect("ReplicationStatus");
assert_eq!(
(plain.refs_total, plain.refs_unmirrored),
(0, 0),
"a plain `replication list` must not compute ref coverage: {plain:?}"
);
assert!(
plain
.repos
.iter()
.all(|r| r.refs_total.is_none() && r.refs_unmirrored.is_none()),
"a plain status leaves per-repository ref counts absent: {plain:?}"
);
let row = |name: &str| {
out.repos
.iter()
.find(|r| r.repo.as_ref().is_some_and(|p| p.repo == name))
.unwrap_or_else(|| panic!("no per-repository row for {name}: {out:?}"))
.clone()
};
assert_eq!(
(row("gunnar").refs_unmirrored, row("gunnar").refs_total),
(Some(0), Some(1)),
"zero must be sent as a NUMBER; an absent field means \"could not read the refs\""
);
assert_eq!(
(row("salvage").refs_unmirrored, row("salvage").refs_total),
(Some(1), Some(2)),
);
}
/// ★ **A plain `replication list` opens ZERO repository stores; `--refs`
/// opens one per repository — and the count, not the printed text, is the
/// assertion.**
///
/// 🔴 The load that made a cold appliance unresponsive today: the old
/// `repo_copies` counted ref coverage on EVERY `Mirror.ReplicationStatus`,
/// opening every repository's reference database — one refold per
/// repository — so a single status query was O(all repos × cold-store open).
/// A test that only read the printed roster would pass the broken code, because
/// the roster it prints is correct; the cost is invisible in the output. So the
/// red is an OPEN COUNT, taken off [`ServerStores::store_opens`] at the one site
/// a store is actually opened, and it is load-independent: with N repositories
/// cold, a plain status must add 0 and `--refs` must add exactly N.
///
/// Seen RED by having `running_to_proto` ignore `include_refs` and walk anyway:
/// the plain-status assertion then reads N opens where it demands 0.
#[test]
fn a_plain_replication_status_opens_no_store_and_refs_opens_one_each() {
use gunnar_cli::server::mirror::ReplicationSlot;
use gunnar_grpc::ControlPlane;
const REPOS: u64 = 8;
let dir = tempfile::tempdir().expect("a temp dir");
let stores = Arc::new(
gunnar_cli::server::ServerStores::open(dir.path().join("data")).expect("open the stores"),
);
stores.catalog().create_namespace("ns").expect("namespace");
// Catalogued but NOT materialised: `create` writes a catalogue row and
// opens no store, so every repository is COLD — the cold appliance the
// incident was on. Opening one now is what the ref-coverage walk pays.
for i in 0..REPOS {
stores
.catalog()
.create(NewRepo::new(
RepoRef::new("ns", &format!("r{i}")).expect("a legal path"),
))
.expect("create the repository");
}
let runtime = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.expect("a runtime");
let bus = Arc::new(EventBus::new(256));
let slot = ReplicationSlot::new(
Arc::clone(&stores),
Arc::clone(&bus),
runtime.handle().clone(),
);
// One Server-scope target at a dead port: a worker is spawned but never
// connects and never drains, so nothing it does opens a store — the only
// opens in this test are the ones the status path itself pays.
slot.apply(ReplicationConfig {
targets: vec![ReplicationTarget {
name: "standby".into(),
scope: ReplicationScope::Server,
remote: format!("ssh://git@127.0.0.1:{DEAD_PORT}"),
identity: None,
known_hosts: Some(dir.path().join("known_hosts")),
strict_host_key: false,
visibility: Default::default(),
control: None,
}],
..ReplicationConfig::default()
})
.expect("apply the replication config");
let plane = gunnar_cli::server::CoreControlPlane::new(Arc::clone(&stores))
.with_replication(Arc::clone(&slot));
// PLAIN first, so the store-handle cache stays cold for the `--refs`
// measurement that follows. The green: a status glance opens nothing.
let before_plain = stores.store_opens();
let plain = plane.replication_status(false).expect("plain status");
let plain_opens = stores.store_opens() - before_plain;
assert_eq!(
plain_opens, 0,
"a plain `replication list` opened {plain_opens} store(s); it must open NONE — the \
verdict, pending and copy state need no reference database"
);
assert_eq!(
plain.repos.len(),
REPOS as usize,
"the per-repository copy rows are still there — it is only their ref counts that are \
absent on a plain status: {plain:?}"
);
assert!(
plain.repos.iter().all(|r| r.refs_total.is_none()),
"a plain status carries no per-repository ref count: {plain:?}"
);
assert_eq!((plain.refs_total, plain.refs_unmirrored), (0, 0));
// `--refs`: the operator asked, so ONE walk opens every repository's store
// exactly once. Cold cache from above, so the count is the whole catalogue.
let before_refs = stores.store_opens();
let refs = plane.replication_status(true).expect("status with --refs");
let refs_opens = stores.store_opens() - before_refs;
assert_eq!(
refs_opens, REPOS,
"`--refs` must open each repository's store exactly once ({REPOS}); it opened \
{refs_opens}"
);
// …and the opt-in number is real: every row now carries a count, and the
// empty repositories report ZERO unmirrored — a number, not an absence.
assert_eq!(
refs.repos.len(),
REPOS as usize,
"every covered repository has a row under --refs: {refs:?}"
);
assert!(
refs.repos
.iter()
.all(|r| r.refs_total == Some(0) && r.refs_unmirrored == Some(0)),
"these repositories have no refs at all, which is `Some(0)` — never absent: {refs:?}"
);
drop(runtime);
}
/// **I11's passive heartbeat, proven APPLIED and proven able to stay quiet.**
///
/// The crash path failover never had: a primary that dies without abdicating
/// tells nobody, so the standby has to notice by itself. The verb it calls has
/// existed since abdication landed (`SetRole(PRIMARY)`); this is the detection.
///
/// # Both halves, because either alone is a green that proves nothing
///
/// **GREEN — it takes over.** The watch is pointed at a port that was bound and
/// then released, which is a dead primary's address exactly: nothing is
/// listening and nothing ever will be. After the configured run of consecutive
/// failures the box must be writable — asserted on the SHARED write-gate the
/// SSH front door actually reads (`ServerStores::writes_frozen`) and on the
/// durable `replication.json`, never on a return value. A hollow watchdog that
/// logged "took over" without going through `ReplicationSlot::apply` would
/// leave the gate frozen and this fails.
///
/// **RED — it does NOT take over while the primary answers.** The same watch,
/// same threshold, pointed at a LIVE listener and left to run for many times
/// the threshold: the box must still be frozen. This is the half that matters,
/// because a detector that promotes unconditionally passes the green above
/// perfectly. Measured here at 30x the promote-after count in probes.
///
/// The clock is compressed (10 ms probes, 100 ms connect) so the test costs
/// milliseconds; the rule under test — consecutive failures against a
/// threshold — is the same one `standby_watch::Tally`'s unit tests drive
/// directly, and neither knows the wall-clock defaults.
#[test]
fn a_watching_standby_promotes_itself_when_the_primary_dies_and_not_while_it_answers() {
if !tools_present() {
eprintln!("skipping: git and ssh are required");
return;
}
use gunnar_cli::server::standby_watch::{watch, StandbyWatch};
use gunnar_grpc::ControlPlane;
use std::sync::atomic::Ordering::SeqCst;
let unit = Harness::start("nordisk", "gunnar", HashKind::Sha256);
let slot = Arc::clone(
unit.server
.as_ref()
.expect("the server is running")
.replication_slot(),
);
let plane = gunnar_cli::server::CoreControlPlane::new(Arc::clone(&unit.stores))
.with_replication(Arc::clone(&slot));
let frozen = || unit.stores.writes_frozen().load(SeqCst);
let role_on_disk = || {
std::fs::read_to_string(unit.stores.root().join("replication.json"))
.expect("replication.json")
};
// This box is a standby: read-only to clients, waiting on a primary.
plane
.replication_set_role(gunnar_proto::ReplicationRole::Standby)
.expect("set standby");
assert!(
frozen(),
"the standby must start frozen or the test proves nothing"
);
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("a runtime for the watch");
// ── RED HALF: a primary that ANSWERS must not be taken over ────────────
//
// A live listener on a real port. `watch` never returns here, so the
// timeout elapsing IS the assertion, taken together with the gate.
let alive = rt
.block_on(async { tokio::net::TcpListener::bind(("127.0.0.1", 0)).await })
.expect("bind a live primary");
let alive_addr = alive.local_addr().expect("live addr");
let quiet = StandbyWatch {
primary: alive_addr,
probe_interval: Duration::from_millis(10),
connect_timeout: Duration::from_millis(100),
promote_after: 2,
};
rt.block_on(async {
// A primary that ANSWERS: since S4/S15 a socket nobody reads is wedged,
// so the process behind it says its HTTP/2 SETTINGS back, as a server does.
let held = tokio::spawn(async move {
use tokio::io::AsyncWriteExt;
while let Ok((mut s, _)) = alive.accept().await {
tokio::spawn(async move {
let _ = s.write_all(&[0, 0, 0, 4, 0, 0, 0, 0, 0]).await;
tokio::time::sleep(Duration::from_secs(5)).await;
});
}
});
let ran =
tokio::time::timeout(Duration::from_millis(600), watch(Arc::clone(&slot), quiet)).await;
assert!(
ran.is_err(),
"the watch RETURNED while the primary was answering — it only returns after it \
promotes or when the box is no longer a standby, and neither happened"
);
held.abort();
});
assert!(
frozen(),
"the standby promoted itself while the primary was still answering on {alive_addr} — \
~60 probes against a threshold of 2, so this is not a threshold that is merely long"
);
assert!(
role_on_disk().contains("Standby"),
"the role on disk moved while the primary was answering:\n{}",
role_on_disk()
);
// ── GREEN HALF: a primary that is GONE is taken over ───────────────────
//
// Bind, take the address, release it. Nothing is listening there now, which
// is what a dead primary's control plane is.
let dead_addr = rt
.block_on(async {
let l = tokio::net::TcpListener::bind(("127.0.0.1", 0)).await?;
let a = l.local_addr()?;
drop(l);
Ok::<_, std::io::Error>(a)
})
.expect("a released port");
let dead = StandbyWatch {
primary: dead_addr,
probe_interval: Duration::from_millis(10),
connect_timeout: Duration::from_millis(100),
promote_after: 2,
};
rt.block_on(async {
tokio::time::timeout(Duration::from_secs(10), watch(Arc::clone(&slot), dead))
.await
.expect("the watch must return once it has taken over");
});
// APPLIED, both ways it has to be applied.
assert!(
!frozen(),
"the watch took over and the write-gate the SSH front door reads is STILL frozen — \
a client push would be refused by a box that just called itself the primary"
);
let json = role_on_disk();
assert!(
json.contains("Primary"),
"the takeover is not on disk, so the reboot this box comes back from would come \
up read-only with no primary anywhere:\n{json}"
);
// And it stops. A second run against the same (still dead) address must
// return immediately because this box is no longer a standby — not because
// it promoted twice.
rt.block_on(async {
tokio::time::timeout(Duration::from_secs(5), watch(Arc::clone(&slot), dead))
.await
.expect("a watch on a box that is already primary must return, not spin");
});
assert!(!frozen(), "the second watch re-froze a primary");
drop(rt);
}
/// **`demote` must not tell an operator the box stopped taking writes.**
///
/// Red-first, and the red is what shipped. The old line was
///
/// ```text
/// demoted — 1 standby target(s) cleared; this box is no longer a primary
/// role primary (takes client writes)
/// ```
///
/// — two lines, both true, printed one after the other, and contradicting each
/// other to anyone reading at the speed a failover is run at. MEASURED
/// 2026-09-10 driving the whole flip by hand: `demote` clears the roster and
/// leaves the box **writable**, exactly as its own documentation says it should
/// ("a demoted box still serves git and still takes writes. It has only stopped
/// shipping a second copy anywhere"). An operator who stops at the first line
/// leaves a box that will accept a `git push` nobody replicates, which is the
/// split brain the demote was run to prevent.
///
/// So the assertion is on the SENTENCE, because the sentence is the mechanism
/// here: nothing else stands between a correct verb and a wrong outcome.
#[test]
fn demote_says_what_it_cleared_and_never_that_the_box_went_read_only() {
use gunnar_cli::control::demote_report;
let primary = demote_report(1, gunnar_proto::ReplicationRole::Primary as i32);
assert!(
!primary.contains("no longer a primary"),
"the word `primary` means both `ships a copy somewhere` and `takes client writes`, \
and demote only ends the first. Saying it plainly is what made the next line \
contradict this one:\n{primary}"
);
assert!(
primary.contains("STILL TAKES CLIENT WRITES"),
"a box that is still writable after a demote has to say so, in the same breath:\n{primary}"
);
assert!(
primary.contains("replication standby"),
"and it has to name the verb that actually freezes it, or the operator has to go \
looking for it mid-failover:\n{primary}"
);
assert!(
primary.contains("1 standby target(s) cleared"),
"the count is the one fact that says whether this box really was still a primary \
or had already come back empty:\n{primary}"
);
// A box that is ALREADY frozen is not told to freeze itself: an instruction
// that is already satisfied reads as a step that failed.
let frozen = demote_report(0, gunnar_proto::ReplicationRole::Standby as i32);
assert!(
!frozen.contains("STILL TAKES CLIENT WRITES") && !frozen.contains("replication standby"),
"a standby was told to become a standby:\n{frozen}"
);
assert!(
frozen.contains("0 standby target(s) cleared"),
"the count is printed either way:\n{frozen}"
);
}