crates/installer/nordisk-pairing/tests/two_readers_one_link.rs
//! **Two readers on one box's installer link** (estate-box W6,
//! `W6-two-readers-one-link`): two threads of ONE installer process, each
//! making complete log reads (connect, mutual TLS, request, answer) through
//! its own [`WgPath`] at the same time, while a capture follower reads the
//! same box every 2 s — the shape private-gunnar-ops' gate drives. The box's
//! door serves one reader at a time; none may lose its answer.
//!
//! The REAL box side, as `installer_link_goes_quiet.rs`:
//! [`installer_link::vm_link`] (a TUN, the dataplane pump), the bootstrap
//! listener serving the door, and the installer's real user-space end over
//! loopback UDP. It needs a private network namespace with `/dev/net/tun`,
//! so it is ignored by default:
//!
//! ```text
//! cargo test -p nordisk-pairing --features installer-wireguard --test two_readers_one_link --no-run
//! podman unshare unshare -n sh -c 'ip link set lo up; W6_SEPARATE_BOX=1 W6_STALL=6000:14000 \
//! <that test binary> --ignored --exact two_readers_on_one_link_neither_loses_its_answer --nocapture'
//! ```
//!
//! `W6_SEPARATE_BOX=1` runs the box as a process of its own (`w6_box`), and
//! `W6_STALL=<stop ms>:<every ms>` freezes it (SIGSTOP/SIGCONT) the way a
//! box starved of CPU on a loaded build host is; `W6_ITERATIONS` rounds
//! (12), `W6_QUIET=<s>` of silence before each, `W6_FOLLOW=0` without the
//! follower, `W6_TRACE=1` every DEBUG line of both ends' WireGuard.
//!
//! MEASURED on oden 2026-10-05, the box frozen 6 s of every 14 s: before
//! the door's turn (installer_link.rs `Turn`) rounds of 10-31 s and a read
//! lost in 3 of 5 runs, "reader A #0: the request: nothing from the VM
//! over the link" — the gate's own words; with it, every run green and the
//! frozen round 7.5 s.
#![cfg(feature = "installer-wireguard")]
use nordisk_pairing::bootlog::BootLog;
use nordisk_pairing::bootstrap::{self, Vm};
use nordisk_pairing::elect::Capacity;
use nordisk_pairing::installer::{self, Material};
use nordisk_pairing::installer_link::{self, LinkKeys, Path as _, WgPath};
use nordisk_pairing::join::{JoinReply, JoinRequest};
use nordisk_pairing::mtls::{self, InstallerCa, Issued};
use nordisk_pairing::seed::{Front, Kind, LinkSeed, Seed, TunnelKind};
use nordisk_pairing::signal::Signal;
use std::io::{Read, Write};
use std::net::{Ipv4Addr, SocketAddrV4};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
const ESTATE: &str = "gunnar.rs";
// ── a trace of every DEBUG+ event, W6_TRACE only (the investigation) ──
struct Say(&'static str);
struct Msg(String);
impl tracing::field::Visit for Msg {
fn record_debug(&mut self, f: &tracing::field::Field, v: &dyn std::fmt::Debug) {
self.0.push_str(&format!(" {}={:?}", f.name(), v));
}
}
impl tracing::Subscriber for Say {
fn enabled(&self, m: &tracing::Metadata<'_>) -> bool {
*m.level() <= tracing::Level::DEBUG
}
fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::span::Id {
tracing::span::Id::from_u64(1)
}
fn record(&self, _: &tracing::span::Id, _: &tracing::span::Record<'_>) {}
fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {}
fn event(&self, e: &tracing::Event<'_>) {
let mut m = Msg(String::new());
e.record(&mut m);
eprintln!("TRACE {} {} {} {}{}", std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_millis() % 100000, self.0, e.metadata().level(), e.metadata().target(), m.0);
}
fn enter(&self, _: &tracing::span::Id) {}
fn exit(&self, _: &tracing::span::Id) {}
}
fn trace_as(who: &'static str) {
if std::env::var_os("W6_TRACE").is_some() {
let _ = tracing::subscriber::set_global_default(Say(who));
}
}
struct TheBox {
identity: Issued,
}
impl Vm for TheBox {
fn hello(&self) -> Result<JoinRequest, String> {
Ok(JoinRequest { estate: ESTATE.into(), kind: Kind::Gunnar, tunnel_pubkey: "K".into(), capacity: Some(Capacity { cores: 2, ram_bytes: 1, data_bytes: 1, disk_mib_s: 1 }), csr_pem: String::new() })
}
fn fed(&self, _: &JoinReply) -> Result<(), String> {
Ok(())
}
fn status(&self) -> Signal {
Signal::of("main", TunnelKind::Wireguard, &nordisk_pairing::pair::Machine::default())
}
fn close(&self) -> Result<(), String> {
Ok(())
}
fn identity(&self) -> Issued {
self.identity.clone()
}
}
/// One complete read of the log door, as the gate's `door_probe::ask` makes
/// it: `Ok(status line)` or the words it failed with.
fn ask(endpoint: SocketAddrV4, keys: &LinkKeys, m: &Material) -> Result<String, String> {
let mut path = WgPath::new(endpoint, keys.clone())?;
let cfg = mtls::client_config(&m.ca.cert_pem, &m.me).map_err(|e| e.to_string())?;
let wire = path.connect(Duration::from_secs(20))?;
let conn = rustls::ClientConnection::new(cfg, mtls::vm_name()).map_err(|e| e.to_string())?;
let mut tls = rustls::StreamOwned::new(conn, wire);
let req = format!(
"GET {}?from=0 HTTP/1.1\r\nhost: {}\r\ncontent-type: application/toml\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
bootstrap::LOG_PATH,
mtls::VM_SERVER_NAME
);
tls.write_all(req.as_bytes()).map_err(|e| format!("the request: {e}"))?;
let _ = tls.flush();
let mut buf = Vec::new();
match tls.read_to_end(&mut buf) {
Ok(_) => {}
Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof && !buf.is_empty() => {}
Err(e) => return Err(format!("{e}")),
}
let text = String::from_utf8_lossy(&buf);
let status = text.lines().next().unwrap_or("").trim().to_string();
if status.contains(" 200") {
Ok(status)
} else {
Err(format!("`{status}`"))
}
}
/// **The box as a process of its own** (`W6_BOX_SEED=<seed file>`): its link,
/// its listener, the log door — so the reader test can STALL it the way an
/// overloaded box is stalled (SIGSTOP/SIGCONT), which a box in the reader's
/// own process cannot be. Prints `READY` once it listens; runs until killed.
#[test]
#[ignore = "the box half of two_readers_on_one_link_neither_loses_its_answer (W6_BOX_SEED)"]
fn w6_box() {
let Ok(seed_file) = std::env::var("W6_BOX_SEED") else { return };
trace_as("box");
let seed = Seed::parse(&std::fs::read_to_string(&seed_file).unwrap()).unwrap();
let link = installer_link::vm_link(&seed).unwrap_or_else(|e| panic!("REFUSED [link-netns] the box's link: {e}"));
let listener = std::net::TcpListener::bind((link.bind, bootstrap::PORT)).unwrap();
let log: &'static BootLog = Box::leak(Box::new(BootLog::new(10_000)));
for i in 0..400 {
log.push(&format!("gunnar: a line of the box's log, number {i}, long enough that the answer is several segments"));
}
let vm: Arc<dyn Vm> = Arc::new(TheBox { identity: Issued { cert_pem: seed.bootstrap_cert_pem.clone(), key_pem: seed.bootstrap_key_pem.clone() } });
println!("READY");
let _ = std::io::stdout().flush();
bootstrap::serve_for_life(vm.as_ref(), log, &seed.installer_ca_pem.clone(), listener, &|l| eprintln!("box: {l}"));
drop(link);
}
#[test]
#[ignore = "raises a TUN and brings lo up: run in a private network namespace (see the module doc)"]
fn two_readers_on_one_link_neither_loses_its_answer() {
trace_as("installer");
wg_appliance_core::addr::link_up("lo").unwrap_or_else(|e| panic!("REFUSED [link-netns] lo cannot be brought up here ({e}): run it in a private network namespace (module doc)"));
let ca = InstallerCa::mint(ESTATE, Duration::from_secs(3600)).unwrap();
let boot = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
let me = ca.issue_installer(Duration::from_secs(3600)).unwrap();
let (inst_private, inst_public) = wg_appliance_core::keys::generate_keypair();
let (vm_private, vm_public) = wg_appliance_core::keys::generate_keypair();
let port = {
let u = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
u.local_addr().unwrap().port()
};
let seed = Seed {
estate: ESTATE.into(),
tunnel: TunnelKind::Wireguard,
front: Front { door: "10.13.8.179:8474".into() },
kind: Kind::Gunnar,
installer_ca_pem: ca.cert_pem.clone(),
bootstrap_cert_pem: boot.cert_pem.clone(),
bootstrap_key_pem: boot.key_pem.clone(),
peers_expected: 3,
mullvad: None,
tailscale: None,
installer_link: Some(LinkSeed { installer_pubkey: wg_appliance_core::keys::encode_key(&inst_public), vm_private: wg_appliance_core::keys::encode_key(&vm_private), port, psk: None }),
};
// The box: in this process, or (W6_SEPARATE_BOX=1) a process of its own
// that W6_STALL=<stop ms>:<every ms> freezes the way an overloaded box is.
let separate = std::env::var("W6_SEPARATE_BOX").is_ok_and(|v| v == "1");
let mut child = None;
let mut link = None;
if separate {
let seed_file = std::env::temp_dir().join(format!("w6-seed-{}.toml", std::process::id()));
std::fs::write(&seed_file, seed.render().unwrap()).unwrap();
let mut c = std::process::Command::new(std::env::current_exe().unwrap())
.args(["--ignored", "--exact", "w6_box", "--nocapture"])
.env("W6_BOX_SEED", &seed_file)
.stdout(std::process::Stdio::piped())
.spawn()
.unwrap();
let mut out = std::io::BufReader::new(c.stdout.take().unwrap());
let mut line = String::new();
loop {
line.clear();
use std::io::BufRead as _;
if out.read_line(&mut line).unwrap() == 0 || line.contains("READY") {
break;
}
}
std::thread::spawn(move || std::io::copy(&mut out, &mut std::io::sink()));
child = Some(c);
} else {
let l = installer_link::vm_link(&seed).unwrap_or_else(|e| panic!("REFUSED [link-netns] the box's link cannot be raised here ({e}): run it in a private network namespace with /dev/net/tun (module doc)"));
let listener = std::net::TcpListener::bind((l.bind, bootstrap::PORT)).unwrap();
let log: &'static BootLog = Box::leak(Box::new(BootLog::new(10_000)));
for i in 0..400 {
log.push(&format!("gunnar: a line of the box's log, number {i}, long enough that the answer is several segments"));
}
let vm: Arc<dyn Vm> = Arc::new(TheBox { identity: boot.clone() });
let ca_pem = ca.cert_pem.clone();
std::thread::spawn(move || bootstrap::serve_for_life(vm.as_ref(), log, &ca_pem, listener, &|l| eprintln!("box: {l}")));
link = Some(l);
}
let stop_stall = Arc::new(AtomicBool::new(false));
if let (Some(c), Ok(spec)) = (child.as_ref(), std::env::var("W6_STALL")) {
let (stop_ms, every_ms) = spec.split_once(':').map(|(a, b)| (a.parse::<u64>().unwrap(), b.parse::<u64>().unwrap())).unwrap();
let pid = c.id().to_string();
let stop = stop_stall.clone();
std::thread::spawn(move || {
while !stop.load(Ordering::SeqCst) {
std::thread::sleep(Duration::from_millis(every_ms.saturating_sub(stop_ms)));
let _ = std::process::Command::new("kill").args(["-STOP", &pid]).status();
std::thread::sleep(Duration::from_millis(stop_ms));
let _ = std::process::Command::new("kill").args(["-CONT", &pid]).status();
}
});
}
let endpoint = SocketAddrV4::new(Ipv4Addr::LOCALHOST, port);
let keys = LinkKeys { installer_private: inst_private, vm_public, psk: None };
let m = Arc::new(Material { ca: ca.clone(), me: me.clone(), link: Some(keys.clone()) });
// The capture follower: the same box, every 2 s, for the whole test.
let stop = Arc::new(AtomicBool::new(false));
let follower = {
let (stop, keys, m) = (stop.clone(), keys.clone(), m.clone());
std::thread::spawn(move || {
let mut lost = Vec::new();
let follow = std::env::var("W6_FOLLOW").map(|v| v != "0").unwrap_or(true);
while follow && !stop.load(Ordering::SeqCst) {
if let Err(e) = installer::read_log(Box::new(WgPath::new(endpoint, keys.clone()).unwrap()), m.as_ref(), 0) {
lost.push(e);
}
std::thread::sleep(Duration::from_secs(2));
}
lost
})
};
const ROUNDS: usize = 4;
let iterations: usize = std::env::var("W6_ITERATIONS").ok().and_then(|v| v.parse().ok()).unwrap_or(12);
let mut red = Vec::new();
let mut said = Vec::new();
let quiet: u64 = std::env::var("W6_QUIET").ok().and_then(|v| v.parse().ok()).unwrap_or(0);
for it in 0..iterations {
std::thread::sleep(Duration::from_secs(quiet));
let read = |who: &str| -> (usize, Vec<String>, u128) {
let t = Instant::now();
let mut ok = 0;
let mut lost = Vec::new();
for i in 0..ROUNDS {
match ask(endpoint, &keys, &m) {
Ok(_) => ok += 1,
Err(e) => lost.push(format!("{who} #{i}: {e}")),
}
}
(ok, lost, t.elapsed().as_millis())
};
let (a, b) = std::thread::scope(|sc| {
let ha = sc.spawn(|| read("reader A"));
let b = read("reader B");
(ha.join().unwrap_or((0, vec!["reader A panicked".into()], 0)), b)
});
let line = format!("#{it}: reader A {}/{ROUNDS} in {} ms, reader B {}/{ROUNDS} in {} ms", a.0, a.2, b.0, b.2);
eprintln!("{line}");
if a.1.is_empty() && b.1.is_empty() {
said.push(line);
} else {
red.push(format!("{line}; lost: {}", a.1.iter().chain(b.1.iter()).cloned().collect::<Vec<_>>().join("; ")));
}
}
stop.store(true, Ordering::SeqCst);
stop_stall.store(true, Ordering::SeqCst);
let follower_lost = follower.join().unwrap();
drop(link);
if let Some(mut c) = child {
let _ = std::process::Command::new("kill").args(["-CONT", &c.id().to_string()]).status();
let _ = c.kill();
let _ = c.wait();
}
assert!(red.is_empty(), "{} of {iterations} rounds lost a reader:\n{}\n(fine: {})", red.len(), red.join("\n"), said.join("; "));
assert!(follower_lost.is_empty(), "the capture follower lost {} read(s): {}", follower_lost.len(), follower_lost.join("; "));
}