crates/gunnar-server/tests/replication_follows_wg0.rs
//! **The mutual replication listener follows the WireGuard tunnel** (owner,
//! 2026-09-22, amended 2026-09-24): `:50052` answers on the outer address (eth1)
//! for the process lifetime and ALSO on `wg0`'s address while the tunnel is up;
//! `:50051` does not move. A tunnel raised at RUNTIME opens the second door with
//! no restart, and `twin certify` hands the twin `<tunnel>,<utility>`: it dials
//! the tunnel first and falls back to the outer address.
//!
//! No root and no `wg0` here: the "outer" address is `127.0.0.1` and the
//! "tunnel" is `127.0.0.2` (both on `lo`), published through the same
//! [`gunnar_server::tunnel_addr`] a real `wg::establish` publishes through. What
//! is proven is the listener's rule and the pairing's plumbing; the tunnel's own
//! packets are proven on the bench (growbench TD8 over the HubRelay).
//!
//! ONE test function: the tunnel address is process-wide, so the steps are a
//! sequence, not independent tests racing on it.
use std::net::{IpAddr, Ipv4Addr, SocketAddr, TcpStream};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use gunnar_core::{NewRepo, Principal, PrincipalKind, RepoRef};
use gunnar_grpc::{ControlPlane, EventBus, ServerConfig, ServiceContext};
use gunnar_server::stores::ServerStores;
use gunnar_server::twin::{self, Follower, GrpcSource};
use gunnar_server::{serve, tunnel_addr, twin_tls, CoreControlPlane};
/// The twin's channel on a LOADED box: its first `Mirror.LogHead` takes longer
/// than `twin::HEAD_WITHIN`, which `GrpcSource` reports as STALLED — the one
/// thing a slow box does to a follower — and every later call is answered.
/// The follower hands a stalled connection back (`Ended::Redial`) for its
/// supervisor to dial again; a test thread that only runs it once ends there.
struct StalledOnce {
inner: GrpcSource,
stalled: AtomicBool,
}
impl twin::LogSource for StalledOnce {
fn alternatives(&self) -> bool {
self.inner.alternatives()
}
fn head(&self) -> anyhow::Result<(u64, u64)> {
self.inner.head()
}
fn head_and_identity(&self) -> anyhow::Result<(u64, u64, Option<String>)> {
if !self.stalled.swap(true, Ordering::SeqCst) {
anyhow::bail!(
"{} the primary said nothing to Mirror.LogHead within {} s on this connection",
twin::STALLED,
twin::HEAD_WITHIN.as_secs()
);
}
self.inner.head_and_identity()
}
fn fetch(
&self,
from: u64,
to: u64,
max_bytes: u64,
) -> anyhow::Result<Vec<gunnar_core::txlog::Entry>> {
self.inner.fetch(from, to, max_bytes)
}
fn snapshot(&self, into: &mut twin::snapshot::Stager) -> anyhow::Result<bool> {
self.inner.snapshot(into)
}
}
/// How long step 3 gives the twin to reach the primary's head. An OUTER bound
/// only: the wait is on the twin's own polls (see [`Told`]), so a healthy run
/// ends the moment the twin asks again after applying the head, and this is
/// reached only when the follower is truly stuck (it says what it had applied).
const FOLLOW_WITHIN: Duration = Duration::from_secs(120);
/// The twin's source, telling the test what the twin has applied each time it
/// asks the primary for its head. `Follower::run` asks again AT ONCE after a
/// poll that applied (T1), so the ask that follows the last apply is the event
/// step 3 waits on — not a clock (as tests/twin_mtls_pairing P1 waits).
struct Told {
inner: Box<dyn twin::LogSource>,
twin: Arc<ServerStores>,
tell: std::sync::mpsc::Sender<u64>,
}
impl twin::LogSource for Told {
fn alternatives(&self) -> bool {
self.inner.alternatives()
}
fn head(&self) -> anyhow::Result<(u64, u64)> {
self.inner.head()
}
fn head_and_identity(&self) -> anyhow::Result<(u64, u64, Option<String>)> {
// `Follower::applied_seq`: a step appends and applies before the next ask.
let _ = self.tell.send(self.twin.txlog().head_seq());
self.inner.head_and_identity()
}
fn fetch(
&self,
from: u64,
to: u64,
max_bytes: u64,
) -> anyhow::Result<Vec<gunnar_core::txlog::Entry>> {
self.inner.fetch(from, to, max_bytes)
}
fn snapshot(&self, into: &mut twin::snapshot::Stager) -> anyhow::Result<bool> {
self.inner.snapshot(into)
}
}
fn reachable(at: SocketAddr) -> bool {
TcpStream::connect_timeout(&at, Duration::from_millis(300)).is_ok()
}
fn wait(what: &str, secs: u64, mut f: impl FnMut() -> bool) {
let t0 = Instant::now();
while !f() {
assert!(
t0.elapsed() < Duration::from_secs(secs),
"timed out waiting for: {what}"
);
std::thread::sleep(Duration::from_millis(50));
}
}
#[test]
fn the_replication_listener_stays_on_eth1_adds_wg0_at_runtime_and_certify_hands_the_twin_both_addresses(
) {
let dir = tempfile::tempdir().expect("tempdir");
let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
stores
.catalog()
.create_namespace("alice")
.expect("namespace");
stores
.create_repo(
NewRepo::new(RepoRef::new("alice", "one").expect("legal")),
false,
)
.expect("repo");
let pki = stores.pki().expect("pki");
let issuer = pki
.create("gunnar issuer", gunnar_core::DEFAULT_ISSUER_DAYS)
.expect("issuer");
let auth = stores.auth();
auth.add_principal(Principal::new("twin-box", "twin-box", PrincipalKind::Machine).unwrap())
.expect("twin principal");
auth.add_principal(Principal::new("twin", "twin", PrincipalKind::Group).unwrap())
.expect("twin group");
auth.add_to_group("twin", "twin-box").expect("membership");
let hosts = vec![
gunnar_grpc::client_signer::REPLICATION_SERVER_NAME.to_owned(),
"127.0.0.1".to_owned(),
"127.0.0.2".to_owned(),
];
let leaf = pki
.issue_server_identity(&hosts, gunnar_core::DEFAULT_LEAF_DAYS)
.expect("leaf");
let trust = gunnar_grpc::Trust {
client_ca: issuer.cert_pem.clone().into_bytes(),
cert: format!("{}{}", leaf.cert_pem, issuer.cert_pem).into_bytes(),
key: leaf.key_pem.expect("key").into_bytes(),
};
let plane: Arc<dyn ControlPlane> = Arc::new(CoreControlPlane::new(Arc::clone(&stores)));
let ctx = ServiceContext::new(
Arc::clone(&plane),
Arc::new(EventBus::new(16)),
ServerConfig::default().insecure_allow_all_for_local_development(),
);
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.expect("rt");
// ── 1. No tunnel: the door is on the outer address, and records no host. ──
tunnel_addr::publish(None);
let std_listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
let base = std_listener.local_addr().expect("addr");
std_listener.set_nonblocking(true).expect("nonblocking");
let wg0: Ipv4Addr = "127.0.0.2".parse().unwrap();
let on_wg0 = SocketAddr::new(IpAddr::V4(wg0), base.port());
assert_eq!(
tunnel_addr::replication_extra(base, false, tunnel_addr::current()),
None,
"no tunnel ⇒ only the derived (eth1) address"
);
let (stop_tx, stop_rx) = tokio::sync::watch::channel(false);
let root = stores.root().to_path_buf();
let (ctx2, trust2, root2) = (ctx.clone(), trust.clone(), root.clone());
rt.block_on(async move {
let listener = tokio::net::TcpListener::from_std(std_listener).expect("tokio listener");
serve::record_replication_door(&root2, base, None).expect("record");
tokio::spawn(serve::replication_door(
listener,
base,
false,
false,
ctx2,
|| 0,
trust2,
root2,
stop_rx,
));
});
assert!(reachable(base), "the door answers on the outer address");
assert!(
!reachable(on_wg0),
"and not on the tunnel address, which is not up"
);
assert_eq!(twin_tls::replication_port(&root), Some(base.port()));
assert_eq!(
twin_tls::replication_host(&root),
None,
"no tunnel ⇒ no host recorded; the twin dials the host it joined"
);
// ── 2. The tunnel comes up at RUNTIME: a second door opens on it, no restart,
// and the outer one STAYS (a tunnel with no handshake must not take it). ──
tunnel_addr::publish(Some(wg0));
wait("the door to open on the tunnel address", 10, || {
twin_tls::replication_host(&root).as_deref() == Some("127.0.0.2,127.0.0.1")
});
wait("the door to answer on the tunnel address", 10, || {
reachable(on_wg0)
});
assert!(reachable(base), "the outer address keeps answering");
assert_eq!(
twin_tls::replication_port(&root),
Some(base.port()),
"same port on both"
);
// ── 3. certify hands the twin the tunnel address; the twin follows THROUGH it. ──
let twin_dir = tempfile::tempdir().expect("twin dir");
let twin_stores = Arc::new(ServerStores::open(twin_dir.path()).expect("twin stores"));
// Joined at the OUTER address, on a port nothing serves: following can only
// succeed through the host and port the pairing handed over.
twin::become_twin(&twin_stores, "127.0.0.1:1").expect("a fresh box becomes a twin");
let csr = twin_tls::request(twin_stores.root()).expect("csr");
let (cert, ca, port, host) = plane
.issue_twin_certificate("twin-box", &csr)
.expect("the primary certifies its twin");
assert_eq!(
(host.as_str(), port),
("127.0.0.2,127.0.0.1", base.port()),
"certify hands the TUNNEL address first, the outer one next"
);
let twin_plane = CoreControlPlane::new(Arc::clone(&twin_stores));
twin_plane
.install_twin_certificate(&cert, &ca, port, &host)
.expect("the twin installs, with the host");
assert_eq!(
twin_tls::load_host(twin_stores.root()).as_deref(),
Some("127.0.0.2,127.0.0.1")
);
let key = twin::twin_private_key(twin_stores.root()).expect("key");
let source = GrpcSource::connect("127.0.0.1:1", key, twin_stores.root())
.expect("the paired twin builds its channel to the tunnel address");
let stop = Arc::new(AtomicBool::new(false));
let (s, st) = (Arc::clone(&stop), Arc::clone(&twin_stores));
// A loaded box: the first LogHead on this channel is STALLED (slower than HEAD_WITHIN).
let source = StalledOnce {
inner: source,
stalled: AtomicBool::new(false),
};
let (tell, told) = std::sync::mpsc::channel::<u64>();
let follower = std::thread::spawn(move || {
let mut source: Option<Box<dyn twin::LogSource>> = Some(Box::new(source));
loop {
let inner = source.take().unwrap_or_else(|| {
let key = twin::twin_private_key(st.root()).expect("the twin's key");
Box::new(
GrpcSource::connect("127.0.0.1:1", key, st.root())
.expect("the paired twin re-dials its channel to the tunnel address"),
)
});
let told = Told {
inner,
twin: Arc::clone(&st),
tell: tell.clone(),
};
match Follower::new(Arc::clone(&st), Box::new(told)).run(Duration::from_millis(100), &s)
{
// What the supervisor does (twin::start_follower_every): a hand-back — a
// LogHead slower than HEAD_WITHIN on a loaded box — is dialled again at once.
twin::Ended::Redial => continue,
ended => return ended,
}
}
});
let want = stores.txlog().head_seq();
assert!(want > 0);
let deadline = Instant::now() + FOLLOW_WITHIN;
let mut applied = 0;
while applied < want {
match told.recv_timeout(deadline.saturating_duration_since(Instant::now())) {
Ok(seq) => applied = seq,
Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
panic!("the twin's follower stopped ({:?}) having applied {applied} of {want}", follower.join())
}
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => panic!(
"the twin did not follow over the tunnel address within {FOLLOW_WITHIN:?}: applied {} of {want}",
twin_stores.txlog().head_seq()
),
}
}
stop.store(true, Ordering::SeqCst);
assert_eq!(
follower.join().expect("follower"),
twin::Ended::Stopped,
"the follower ends because it was stopped, not on its own"
);
// ── 4. A renewal from a primary with no tunnel clears the host again. ──
let (cert, ca, port, _) = plane
.issue_twin_certificate("twin-box", &twin_tls::request(twin_stores.root()).unwrap())
.unwrap();
twin_plane
.install_twin_certificate(&cert, &ca, port, "")
.expect("renewal without a host");
assert_eq!(
twin_tls::load_host(twin_stores.root()),
None,
"an empty host clears the recorded one"
);
let _ = stop_tx.send(true);
tunnel_addr::publish(None);
}
/// A host that is not an IP address is refused before anything is installed.
#[test]
fn a_replication_host_that_is_not_an_ip_is_refused() {
let dir = tempfile::tempdir().unwrap();
let stores = Arc::new(ServerStores::open(dir.path()).unwrap());
twin::become_twin(&stores, "127.0.0.1:1").unwrap();
let err = format!(
"{:#}",
twin_tls::install_at(stores.root(), "x", "y", 50052, Some("wg0.example")).unwrap_err()
);
assert!(err.contains("not an IP address"), "{err}");
assert!(!stores.root().join(twin_tls::TLS_CERT).exists());
}