server/lib/tests/the_twin_follows_at_once_when_certified.rs
//! ★ **A twin certified with a minute's tick follows within seconds, not at its next tick**
//! (FAST-REL, 2026-10-25). Two real holgers on 127.0.0.1 and 127.0.0.2, certified through the
//! RPCs (`nordisk_twin::certify` over holger's `GrpcBoxes`), with an estate's 60 s tick. Its
//! own test binary: the live follower is one per process (`twin_follow::set_live`).
use std::io::Write as _;
use std::time::Duration;
use traits::test_support::hold_port_on;
use holger_server_lib::twin_tls::nordisk_twin;
use holger_server_lib::twin_tls::{self, GrpcBoxes};
use holger_server_lib::{read_ron_config, wire_holger, Holger};
const TOKEN: &str = "console-secret";
struct Box_ {
ip: &'static str,
grpc: u16,
http: u16,
_cfg: tempfile::NamedTempFile,
_h: Holger,
}
impl Box_ {
fn ctl(&self) -> String {
format!("{}:{}", self.ip, self.grpc)
}
fn base(&self) -> String {
format!("http://{}:{}", self.ip, self.http)
}
}
/// Start a box, retrying on fresh ports when another process took one between
/// `free_port` and the bind (the suite shares its host with other builds).
fn start(data: &std::path::Path, ip: &'static str) -> Box_ {
let mut last = String::new();
for _ in 0..8 {
match try_start(data, ip) {
Ok(b) => return b,
Err(e)
if e.contains("in use")
|| e.contains("cannot open")
|| e.contains("failed to bind") =>
{
last = e
}
Err(e) => panic!("{e}"),
}
}
panic!("no free ports after 8 tries: {last}")
}
fn try_start(data: &std::path::Path, ip: &'static str) -> Result<Box_, String> {
let (grpc, http, door) = (hold_port_on(ip), hold_port_on(ip), hold_port_on(ip));
std::fs::create_dir_all(data.join("audit")).unwrap();
let ron = format!(
r#"(
repositories: [
(ron_name: "crates", ron_repo_type: "raw", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "storage", ron_exposed_endpoint: "default")),
ron_data_dir: Some("{data}/crates")),
],
exposed_endpoints: [
(ron_name: "default", ron_url: "{ip}:{grpc}", ron_http_url: Some("{ip}:{http}")),
],
storage_endpoints: [ (ron_name: "storage", ron_storage_type: "znippy", ron_path: "{data}") ],
audit: ( dir: "{data}/audit", enabled: true ),
auth: Some(( methods: [ StaticToken(subject: "console", token: Some("{TOKEN}")) ], roles: {{ "console": Admin }}, require_auth_for_reads: true )),
replication_door: Some(( ron_bind: Some("{ip}:{door}") )),
)"#,
data = data.display(),
);
let mut cfg = tempfile::NamedTempFile::new().unwrap();
cfg.write_all(ron.as_bytes()).unwrap();
let mut h: Holger = read_ron_config(cfg.path()).unwrap();
h.instantiate_backends().unwrap();
wire_holger(&mut h).unwrap();
h.start().map_err(|e| format!("{e:#}"))?;
Ok(Box_ {
ip,
grpc,
http,
_cfg: cfg,
_h: h,
})
}
fn cl() -> reqwest::Client {
reqwest::Client::new()
}
async fn wait_healthy(base: &str) {
for _ in 0..300 {
if let Ok(r) = cl().get(format!("{base}/healthz")).send().await {
if r.status() == 200 {
return;
}
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
panic!("never healthy at {base}");
}
async fn twin_door(b: &Box_) -> serde_json::Value {
serde_json::from_str(
&cl()
.get(format!("{}/-/twin", b.base()))
.send()
.await
.unwrap()
.text()
.await
.unwrap(),
)
.unwrap()
}
fn boxes(poll: u64) -> GrpcBoxes {
GrpcBoxes {
poll_seconds: poll,
reason: "the weaker box (test)".into(),
..GrpcBoxes::new(tokio::runtime::Handle::current(), Some(TOKEN.into()))
}
}
/// ★ **A twin certified with a minute's tick follows within seconds, not at its next tick**
/// (FAST-REL, 2026-10-25, Holger's fast under load): `BecomeTwin` starts the follow, whose first
/// cycle runs before the front has installed the twin's credential (`holger-twin-13`); the loop
/// then rested its whole tick (60 s in an estate) while the certificate landed seconds later,
/// and the pairing's `follow` step waited it out. Over the real RPCs, with the estate's tick:
/// the install wakes the follow, and the twin holds the primary's artifact at once.
#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
async fn a_twin_certified_with_a_minutes_tick_follows_within_seconds() {
let (pdata, tdata) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
let p = start(pdata.path(), "127.0.0.1");
wait_healthy(&p.base()).await;
let artifact: Vec<u8> = (0..20_000u32).map(|i| (i % 241) as u8).collect();
let put = cl()
.put(format!("{}/crates/tick-1.0.tar", p.base()))
.bearer_auth(TOKEN)
.body(artifact.clone())
.send()
.await
.unwrap();
assert!(
put.status().is_success(),
"the primary takes the write: {}",
put.status()
);
let t = start(tdata.path(), "127.0.0.2");
wait_healthy(&t.base()).await;
let target = nordisk_twin::certify::Target {
reader_ctl: t.ctl(),
source_ctl: p.ctl(),
become_twin_of: Some(p.ctl()),
principal: twin_tls::TWIN_PRINCIPAL.into(),
purpose: "twin".into(),
};
let b = boxes(60);
tokio::task::spawn_blocking(move || nordisk_twin::certify::once(&b, &target))
.await
.unwrap()
.expect("the front certifies the twin through the RPCs");
let certified = std::time::Instant::now();
let mut held = None;
while certified.elapsed() < Duration::from_secs(20) {
let r = cl()
.get(format!("{}/crates/tick-1.0.tar", t.base()))
.bearer_auth(TOKEN)
.send()
.await
.unwrap();
if r.status().is_success() {
assert_eq!(
r.bytes().await.unwrap().as_ref(),
&artifact[..],
"the same bytes"
);
held = Some(certified.elapsed());
break;
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
let took = held.unwrap_or_else(|| {
panic!("the twin, certified, still held nothing after 20 s of its 60 s tick")
});
assert!(
took < Duration::from_secs(20),
"followed {took:?} after its certificate"
);
let tj = twin_door(&t).await;
assert!(
tj["replication"]["follow"]
.as_str()
.unwrap_or_default()
.starts_with("mtls"),
"it follows over the door: {tj}"
);
}