crates/gunnar-server/tests/twin_follows_restored.rs
//! **T1e, natively: the twin FOLLOWS the primary restored from it — past the
//! head it held when the old primary died** (2026-09-21, T17).
//!
//! The KVM drill's T1e read the twin at head 262 after the restore and passed,
//! but the restored box's head was ALSO 262 — the head the twin held when the
//! old primary died — so that row could not tell "follows the restored box"
//! from "still holds what it had". The drill's twin log showed only failures
//! (a successful poll is not logged), which read as "no successful poll after
//! 17:39:10". The restored box's own log answers it: sixteen `Mirror.LogHead`
//! calls from `twin-box`, 17:39:30 → 17:40:00, every two seconds, all served
//! and signed, and none refused by the twin.
//!
//! This test closes the ambiguity on the real wire: a primary served over a
//! loopback gRPC listener, a twin following it with the appliance's own
//! `GrpcSource` (which pins the answering key and the log's name on its first
//! signed answer), the primary killed, a fresh box restored FROM THE TWIN,
//! restarted, served at the SAME address — and then a write on the restored
//! box that the dead primary never had. The twin must apply it, still pinned
//! to the same key and the same log, without being touched.
//!
//! What it does NOT prove, and why: the answering key is cached once per
//! process (`twin::answering_key`), so in one test process the restored box
//! signs with whatever the first box cached. That the restore REPLAYS the host
//! key is proven by `twin::tests::a_restored_box_comes_back_as_the_same_box_not_just_the_same_repositories`;
//! that a different key is refused, by the `check_answer` tests. What is
//! proven here is the log identity and the frames, end to end.
//!
//! To see it fail, delete the `pin_log_id` call in `twin::restore_finish`: the
//! restored box answers under a log name the twin never pinned and the twin
//! stops `diverged` instead of applying the new write.
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::{EventBus, ServerConfig, ServiceContext};
use gunnar_server::stores::ServerStores;
use gunnar_server::twin::{self, Follower, GrpcSource, State};
use gunnar_server::CoreControlPlane;
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")
}
/// A control plane for `stores`, served at `addr` until the sender fires.
struct Served {
_rt: tokio::runtime::Runtime,
stop: Option<tokio::sync::oneshot::Sender<()>>,
addr: SocketAddr,
}
fn serve(stores: &Arc<ServerStores>, addr: SocketAddr) -> Served {
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(),
);
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.expect("rt");
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: 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 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));
}
}
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");
}
#[test]
fn the_twin_follows_the_primary_restored_from_it_past_the_head_the_dead_primary_had() {
let addr = free_addr();
// ── the primary, and a twin following it over the wire ──
let pd = tempfile::tempdir().unwrap();
let primary = Arc::new(ServerStores::open(pd.path()).expect("primary"));
let real_id = twin::mint_log_id(pd.path()).expect("the primary's log name");
write_repo(&primary, "alice", "one");
write_repo(&primary, "alice", "two");
let served = serve(&primary, addr);
let td = tempfile::tempdir().unwrap();
let twin_stores = Arc::new(ServerStores::open(td.path()).expect("twin stores"));
twin::become_twin(&twin_stores, &addr.to_string()).expect("a fresh box becomes a twin");
let key = twin::twin_private_key(twin_stores.root()).expect("the twin's key");
let source = GrpcSource::connect(&addr.to_string(), key, twin_stores.root()).expect("source");
let stop = Arc::new(AtomicBool::new(false));
let follower = {
let (s, st) = (Arc::clone(&stop), Arc::clone(&twin_stores));
std::thread::spawn(move || {
Follower::new(st, Box::new(source)).run(POLL, &s);
})
};
let dead_head = primary.txlog().head_seq();
wait(
"the twin levels with the primary",
Duration::from_secs(20),
|| (twin_stores.txlog().head_seq() >= dead_head).then_some(()),
);
let pinned_key = wait(
"the twin pins the key its primary answers with",
Duration::from_secs(5),
|| twin::pinned_primary_key(td.path()),
);
assert_eq!(
twin::log_id(td.path()).as_deref(),
Some(real_id.as_str()),
"the twin took the primary's log name"
);
// ── the primary dies; the twin keeps polling an address nobody answers ──
served.kill();
drop(primary);
// What the dead primary gave its twin is what a restore brings back: the twin re-polls at
// once after it applies (T1), so it may hold the primary's last write past the head read
// above, and a restore is measured against the TWIN's head, not a head read earlier.
let dead_head = twin_stores.txlog().head_seq();
// ── a fresh box, restored FROM THE TWIN, then restarted, at the SAME address ──
let fd = tempfile::tempdir().unwrap();
let fresh = Arc::new(ServerStores::open(fd.path()).expect("fresh"));
let twins_name = twin::log_id(td.path()).expect("the twin knows its log's name");
let restored_to = twin::restore(
Arc::clone(twin_stores.txlog()),
Arc::clone(&fresh),
&twins_name,
)
.expect("restore from the twin");
assert_eq!(restored_to, dead_head);
drop(fresh);
let restored = Arc::new(ServerStores::open(fd.path()).expect("the restored box, restarted"));
assert_eq!(
twin::log_id(fd.path()).as_deref(),
Some(real_id.as_str()),
"the restored box IS the log the twin follows"
);
let served = serve(&restored, addr);
// ── a write the dead primary never had ──
write_repo(&restored, "alice", "after-the-restore");
let new_head = restored.txlog().head_seq();
assert!(
new_head > dead_head,
"the restored box moved past the dead primary's head ({dead_head} → {new_head})"
);
wait(
"the twin applies what only the restored box has",
Duration::from_secs(20),
|| {
// Applied, not only appended: the twin's log takes an entry before its stores do.
(twin_stores.txlog().head_seq() >= new_head
&& twin_stores
.catalog()
.get(&RepoRef::new("alice", "after-the-restore").unwrap())
.ok()
.flatten()
.is_some())
.then_some(())
},
);
let st = twin::status(td.path()).expect("a twin status");
assert_eq!(st.state, State::Following, "{st:?}");
assert!(st.caught_up(), "{st:?}");
assert_eq!(
twin::pinned_primary_key(td.path()).as_deref(),
Some(pinned_key.as_str()),
"the twin did not re-pin: it is the same box to it"
);
assert_eq!(
twin::log_id(td.path()).as_deref(),
Some(real_id.as_str()),
"the twin follows the same log"
);
assert!(
twin_stores
.catalog()
.get(&RepoRef::new("alice", "after-the-restore").unwrap())
.unwrap()
.is_some(),
"the repository created on the restored box is on the twin"
);
println!("ROW T1e-native the twin follows the restored primary past the dead head: dead {dead_head} → restored {new_head}, applied {}, log {real_id}", st.applied_seq);
stop.store(true, Ordering::SeqCst);
follower.join().expect("follower thread");
served.kill();
}