crates/gunnar-server/tests/fence_before_recharge.rs
//! **Fence before recharge** (R-13, owner 2026-10-10): over the real wire.
//!
//! The old primary is ALIVE and its twin still follows it. A fresh main is laid at
//! another address. Recharging it from the twin now would leave TWO writable
//! primaries, so:
//!
//! * an order with no `fence` is refused `not_fenced`, and the main stays empty;
//! * an order that names the old primary as its `fence` fences it first
//! (`Mirror.SetRole(STANDBY)`, `DemoteReplication`) — it then reports
//! `serving=fenced` — and recharges the main from the twin.
//!
//! A fenced primary still answers its twin's polls, so the twin keeps saying
//! `following`: the proof of the fence travels with the call, it cannot be read
//! off the twin.
#![cfg(feature = "replication")]
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::{Duration, Instant};
use gunnar_core::ssh_key::PrivateKey;
use gunnar_core::{NewRepo, RepoRef};
use gunnar_grpc::{EventBus, ServerConfig, ServiceContext};
use gunnar_repair::exec;
use gunnar_repair::Order;
use gunnar_server::repair::{self, ServerEstate};
use gunnar_server::stores::ServerStores;
use gunnar_server::twin;
use gunnar_server::CoreControlPlane;
fn free_addr() -> SocketAddr {
let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
l.local_addr().expect("addr")
}
fn wait<T>(what: &str, within: Duration, mut probe: impl FnMut() -> Option<T>) -> T {
let t0 = Instant::now();
loop {
if let Some(v) = probe() {
return v;
}
assert!(t0.elapsed() < within, "{what}: not within {within:?}");
std::thread::sleep(Duration::from_millis(25));
}
}
struct Served {
rt: tokio::runtime::Runtime,
stop: Option<tokio::sync::oneshot::Sender<()>>,
addr: SocketAddr,
}
/// A control plane with a replication slot, as `Server::start` builds it: the fence
/// (`Mirror.SetRole`, `DemoteReplication`) needs one.
fn serve(stores: &Arc<ServerStores>, addr: SocketAddr, restore_keys: Vec<String>) -> Served {
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.expect("rt");
let bus = Arc::new(EventBus::new(16));
let slot = gunnar_server::mirror::ReplicationSlot::new(
Arc::clone(stores),
Arc::clone(&bus),
rt.handle().clone(),
);
let plane: Arc<dyn gunnar_grpc::ControlPlane> = Arc::new(
CoreControlPlane::new(Arc::clone(stores))
.with_restore_keys(restore_keys)
.with_replication(slot),
);
let ctx = ServiceContext::new(
plane,
bus,
ServerConfig::default().insecure_allow_all_for_local_development(),
);
let (tx, rx) = tokio::sync::oneshot::channel::<()>();
rt.spawn(async move {
let _ = gunnar_grpc::router(ctx, || 0)
.serve_with_shutdown(addr, async {
let _ = rx.await;
})
.await;
});
wait("the box listens", Duration::from_secs(10), || {
std::net::TcpStream::connect(addr).ok()
});
Served {
rt,
stop: Some(tx),
addr,
}
}
impl Served {
fn kill(mut self) {
if let Some(tx) = self.stop.take() {
let _ = tx.send(());
}
let addr = self.addr;
wait("the box stops listening", Duration::from_secs(10), || {
std::net::TcpStream::connect(addr).is_err().then_some(())
});
}
fn head(&self) -> u64 {
let url = format!("http://{}", self.addr);
self.rt.block_on(async move {
let mut c = gunnar_proto::bootstrap_client::BootstrapClient::connect(url)
.await
.expect("connect");
c.status(gunnar_proto::BootstrapStatusRequest {})
.await
.expect("Bootstrap.Status")
.into_inner()
.log_head_seq
})
}
}
fn write_repo(stores: &ServerStores, ns: &str, name: &str) {
if stores
.catalog()
.namespaces()
.unwrap()
.iter()
.all(|n| n.name != ns)
{
stores.catalog().create_namespace(ns).expect("namespace");
}
stores
.create_repo(NewRepo::new(RepoRef::new(ns, name).expect("legal")), false)
.expect("repo");
}
fn repair_key() -> (PrivateKey, String) {
let pem = gunnar_core::mint_ed25519().expect("mint");
let key = PrivateKey::from_openssh(std::str::from_utf8(&pem).unwrap()).expect("parse");
let line = key.public_key().to_openssh().expect("public line");
(key, line)
}
fn recharge(main: SocketAddr, twin: SocketAddr, fence: Option<SocketAddr>) -> Order {
Order::RechargeFromTwin {
main: main.to_string(),
twin: twin.to_string(),
accept_lost: false,
epoch: 1,
case: Some("case-r13".into()),
fence: fence.map(|a| a.to_string()),
fenced: None,
}
}
#[test]
fn a_recharge_beside_a_live_primary_is_refused_until_the_order_fences_it() {
let (old_addr, twin_addr, main_addr) = (free_addr(), free_addr(), free_addr());
// ── an estate: a primary with data, ALIVE, its twin following it ──
let pd = tempfile::tempdir().unwrap();
let primary = Arc::new(ServerStores::open(pd.path()).expect("primary"));
write_repo(&primary, "alice", "one");
write_repo(&primary, "bob", "two");
let served_primary = serve(&primary, old_addr, Vec::new());
let td = tempfile::tempdir().unwrap();
let twin_stores = Arc::new(ServerStores::open(td.path()).expect("twin stores"));
twin::become_twin(&twin_stores, &old_addr.to_string()).expect("a fresh box becomes a twin");
twin::start_follower_every(Arc::clone(&twin_stores), Duration::from_millis(200));
wait(
"the twin levels with the primary",
Duration::from_secs(30),
|| {
let live = primary.txlog().head_seq();
twin::status(td.path())
.filter(|s| {
s.state == twin::State::Following
&& s.applied_seq == live
&& s.primary_head_seq == live
&& twin_stores.txlog().head_seq() == live
})
.map(|_| ())
},
);
let head = twin_stores.txlog().head_seq();
let served_twin = serve(&twin_stores, twin_addr, Vec::new());
assert_eq!(
repair::health(&primary, &old_addr.to_string())
.get("serving")
.map(String::as_str),
Some("yes")
);
// ── someone lays a fresh main elsewhere and asks for a recharge ──
let (key, key_line) = repair_key();
let md = tempfile::tempdir().unwrap();
let main = Arc::new(ServerStores::open(md.path()).expect("the new main"));
let served_main = serve(&main, main_addr, vec![key_line]);
let estate = ServerEstate {
key,
own: main_addr.to_string(),
entitlements: Default::default(),
};
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.expect("rt");
let empty = served_main.head();
let bare = rt.block_on(exec::execute(
&estate,
1,
&recharge(main_addr, twin_addr, None),
&|| false,
));
assert!(!bare.ok, "{bare:?}");
assert!(bare.detail.starts_with("not_fenced"), "{}", bare.detail);
assert!(
bare.detail.contains(&head.to_string()),
"the primary's head, as the twin heard it: {}",
bare.detail
);
assert!(
repair::data_dead(&main),
"nothing was streamed into the new main"
);
assert_eq!(served_main.head(), empty, "its log did not move");
// ── the order names the old primary: fenced first, then recharged ──
let fenced = rt.block_on(exec::execute(
&estate,
2,
&recharge(main_addr, twin_addr, Some(old_addr)),
&|| false,
));
assert!(fenced.ok, "{fenced:?}");
assert!(
fenced.detail.contains(&format!("fenced {old_addr}"))
&& fenced.detail.contains("recharged"),
"{}",
fenced.detail
);
assert_eq!(
repair::health(&primary, &old_addr.to_string())
.get("serving")
.map(String::as_str),
Some("fenced"),
"the old primary says it is fenced"
);
assert_eq!(
served_main.head(),
head,
"the new main holds what the twin held"
);
assert!(!repair::data_dead(&main));
// The fenced primary still answers its twin: the twin cannot tell, so the proof
// had to travel with the call.
std::thread::sleep(Duration::from_millis(600));
assert_eq!(
twin::status(td.path()).map(|s| s.state),
Some(twin::State::Following)
);
println!("ROW fence-before-recharge: bare order refused not_fenced; fence {old_addr} then recharged {main_addr} to {head}");
twin::stop_follower();
served_main.kill();
served_twin.kill();
served_primary.kill();
}