crates/gunnar-server/tests/twin_heartbeat.rs
//! **The dual heartbeat, over a real loopback gRPC listener** (2026-09-15).
//!
//! A primary serves `Mirror.LogHead` / `Admin.Health` on 127.0.0.1; a twin
//! (a fresh data root made a twin by `become_twin`) follows it with the
//! appliance's own `GrpcSource` + reporter. Measured, in order:
//!
//! * H-ok — the primary's `Admin.Health` carries one twin row, `ok`, lag 0,
//! the twin's volume (total > 0, fill > 0); the twin's plane reports its
//! primary `ok`.
//! * H-twin-killed — the follower is stopped: the primary reads `silent`
//! within the bound plus two polls, and not before the bound.
//! * H-main-killed — the follower runs again, the primary's listener is shut
//! down: the twin's `Admin.Health` / `Bootstrap.Status` primary reads
//! `silent` within the bound plus two polls.
use std::net::SocketAddr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use gunnar_core::{NewRepo, RepoRef};
use gunnar_grpc::heartbeat::{Bounds, Registry};
use gunnar_grpc::{EventBus, ServerConfig, ServiceContext};
use gunnar_server::stores::ServerStores;
use gunnar_server::twin::{self, Follower, GrpcSource};
use gunnar_server::CoreControlPlane;
const SILENT_MS: u64 = 1_500;
const POLL: Duration = Duration::from_millis(100);
fn free_addr() -> SocketAddr {
let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
l.local_addr().expect("addr")
}
struct Primary {
rt: tokio::runtime::Runtime,
addr: SocketAddr,
stop: Option<tokio::sync::oneshot::Sender<()>>,
_dir: tempfile::TempDir,
}
impl Primary {
fn start() -> Primary {
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 plane: Arc<dyn gunnar_grpc::ControlPlane> =
Arc::new(CoreControlPlane::new(Arc::clone(&stores)));
let ctx = ServiceContext::new(
plane,
Arc::new(EventBus::new(16)),
ServerConfig::default().insecure_allow_all_for_local_development(),
)
.with_heartbeats(Arc::new(Registry::new(Bounds {
silent_after_ms: SILENT_MS,
behind_polls: 5,
})));
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.expect("rt");
let addr = free_addr();
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;
});
let p = Primary {
rt,
addr,
stop: Some(tx),
_dir: dir,
};
let t0 = Instant::now();
while std::net::TcpStream::connect(addr).is_err() {
assert!(
t0.elapsed() < Duration::from_secs(10),
"the primary never listened"
);
std::thread::sleep(Duration::from_millis(20));
}
p
}
fn health(&self) -> gunnar_proto::HealthResponse {
let at = format!("http://{}", self.addr);
self.rt.block_on(async move {
let mut c = gunnar_proto::admin_client::AdminClient::connect(at)
.await
.expect("connect");
c.health(gunnar_proto::HealthRequest {})
.await
.expect("Admin.Health")
.into_inner()
})
}
fn kill(&mut self) {
if let Some(tx) = self.stop.take() {
let _ = tx.send(());
}
let t0 = Instant::now();
while std::net::TcpStream::connect(self.addr).is_ok() {
assert!(
t0.elapsed() < Duration::from_secs(10),
"the primary never stopped listening"
);
std::thread::sleep(Duration::from_millis(20));
}
}
}
struct Running {
stop: Arc<AtomicBool>,
handle: std::thread::JoinHandle<()>,
}
fn follow(stores: &Arc<ServerStores>, primary: SocketAddr) -> Running {
let key = twin::twin_private_key(stores.root()).expect("the twin's key");
let reporting = Arc::clone(stores);
let source = GrpcSource::connect(&primary.to_string(), key, stores.root())
.expect("source")
.with_reporter(move || twin::measure_report(&reporting));
let stop = Arc::new(AtomicBool::new(false));
let (s, st) = (Arc::clone(&stop), Arc::clone(stores));
let handle = std::thread::spawn(move || {
Follower::new(st, Box::new(source)).run(POLL, &s);
});
Running { stop, handle }
}
impl Running {
fn kill(self) {
self.stop.store(true, Ordering::SeqCst);
self.handle.join().expect("follower thread");
}
}
fn wait<T>(what: &str, within: Duration, mut probe: impl FnMut() -> Option<T>) -> (T, Duration) {
let t0 = Instant::now();
loop {
if let Some(v) = probe() {
return (v, t0.elapsed());
}
assert!(t0.elapsed() < within, "{what}: not within {within:?}");
std::thread::sleep(Duration::from_millis(25));
}
}
#[test]
fn dual_heartbeat_twin_killed_main_reads_silent_and_main_killed_twin_reads_silent() {
let mut primary = Primary::start();
let twin_dir = tempfile::tempdir().expect("tempdir");
let twin_stores = Arc::new(ServerStores::open(twin_dir.path()).expect("twin stores"));
twin::become_twin(&twin_stores, &primary.addr.to_string()).expect("a fresh box becomes a twin");
let twin_plane = CoreControlPlane::new(Arc::clone(&twin_stores));
let bound = Duration::from_millis(SILENT_MS) + 2 * POLL + Duration::from_millis(500);
// ── H-ok ──
let running = follow(&twin_stores, primary.addr);
let (row, _) = wait(
"the primary sees its twin ok",
Duration::from_secs(20),
|| {
primary
.health()
.twins
.into_iter()
.find(|t| t.state == "ok" && t.lag_entries == 0 && t.head_seq > 0)
},
);
// The fill is a whole ‰ of the volume the tempdir sits on: under 1 ‰ it reads 0 (oden 2026-10-06: /tmp a 100 GiB
// tmpfs with 50 MiB used), so it is judged against the bytes it rides with, not against > 0.
assert!(
row.disk_total_bytes > 0 && row.disk_used_bytes > 0,
"the twin's volume rides on the poll: {row:?}"
);
assert!(
row.fill_permille > 0 || row.disk_used_bytes * 1000 < row.disk_total_bytes,
"the twin's fill rides on the poll: {row:?}"
);
assert!(row.address.starts_with("127.0.0.1:"), "{row:?}");
assert_eq!(row.applied_seq, row.head_seq);
let p = gunnar_grpc::ControlPlane::primary_heartbeat(&twin_plane, SILENT_MS)
.expect("a twin reports its primary");
assert_eq!(
(p.state.as_str(), p.address.as_str()),
("ok", primary.addr.to_string().as_str()),
"{p:?}"
);
println!(
"ROW H-ok primary sees twin ok lag {} fill {}‰ of {} B; twin sees primary ok",
row.lag_entries, row.fill_permille, row.disk_total_bytes
);
// ── H-twin-killed ──
running.kill();
let killed = Instant::now();
let (row, took) = wait("the primary reads its killed twin silent", bound, || {
primary
.health()
.twins
.into_iter()
.find(|t| t.state == "silent")
});
assert!(
killed.elapsed() >= Duration::from_millis(SILENT_MS) - POLL * 2,
"silent before the bound: {:?}",
killed.elapsed()
);
assert!(row.why.contains("no poll"), "{row:?}");
println!(
"ROW H-twin-killed primary reads silent after {took:?} (bound {SILENT_MS} ms): {}",
row.why
);
// ── H-main-killed ──
let running = follow(&twin_stores, primary.addr);
wait(
"the primary sees its twin ok again",
Duration::from_secs(20),
|| primary.health().twins.into_iter().find(|t| t.state == "ok"),
);
primary.kill();
let killed = Instant::now();
let (p, took) = wait("the twin reads its killed primary silent", bound, || {
gunnar_grpc::ControlPlane::primary_heartbeat(&twin_plane, SILENT_MS)
.filter(|p| p.state == "silent")
});
assert!(
killed.elapsed() >= Duration::from_millis(SILENT_MS) - POLL * 2,
"silent before the bound: {:?}",
killed.elapsed()
);
assert!(
p.last_answer_unix_ms > 0 && p.why.contains("no answer"),
"{p:?}"
);
println!(
"ROW H-main-killed twin reads primary silent after {took:?} (bound {SILENT_MS} ms): {}",
p.why
);
running.kill();
}