crates/installer/nordisk-pairing/tests/lake_boxes_pair.rs
//! **A Lake-shaped estate pairs end to end** (P3a, A28 H13).
//!
//! Until P3a nordisk-pairing knew three kinds (Front, Gunnar, Holger), so a third product (the
//! Lake: n ingestors, x Doris BE, y Doris FE) could not pair. Here each box names ITS OWN kind,
//! `Kind::Box(role)` (`fe`, `be`, `be`), and the REAL installer (`installer::run`) meets the
//! three boxes, each served by the real bootstrap listener on loopback and running the shared
//! loop (`boxloop::run`). The run closes only because every box's own signal says PAIRED. A box
//! kind is not elected: each box stands alone as its own primary and none becomes a twin, even
//! two boxes of the same kind. The box here is test-side (the Lake crates are not touched); the
//! tunnel is the in-process no-op, as in `holger_boxes_pair.rs`.
use nordisk_pairing::boxloop::{self, BoxDoors, Steps};
use nordisk_pairing::boxside::{self, Installed, Product};
use nordisk_pairing::door::Door;
use nordisk_pairing::elect::Capacity;
use nordisk_pairing::installer::{run, CloseWhen, Material, Plan, PlanVm, Sink};
use nordisk_pairing::installer_link::{DirectPath, Path};
use nordisk_pairing::join::{FrontTunnel, JoinReply};
use nordisk_pairing::mtls::InstallerCa;
use nordisk_pairing::seed::{Kind, Seed, TunnelKind};
use nordisk_pairing::signal::Signal;
use nordisk_pairing::tunnel::{NoTunnel, Tunnel};
use nordisk_pairing::Role;
use std::collections::BTreeMap;
use std::net::{Ipv4Addr, SocketAddrV4, TcpListener};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
const ESTATE: &str = "lake.test";
const H: Duration = Duration::from_secs(3600);
/// A Lake-shaped box's product half, as the installer sees it: its OWN kind and capacity.
struct LakeLike {
kind: Kind,
cores: u32,
fed_role: Mutex<Option<Role>>,
}
impl Product for LakeLike {
fn kind(&self) -> Kind {
self.kind
}
fn tunnel_pubkey(&self) -> String {
String::new()
}
fn capacity(&self) -> Result<Option<Capacity>, String> {
Ok(Some(Capacity { cores: self.cores, ram_bytes: 1 << 30, data_bytes: 1 << 30, disk_mib_s: 0 }))
}
fn offer(&self, _fed: Option<&JoinReply>) -> BTreeMap<String, String> {
BTreeMap::from([("version".to_string(), "test".to_string())])
}
fn fed(&self, reply: &JoinReply) {
*self.fed_role.lock().unwrap() = reply.role;
}
}
/// The box's steps: a primary is ready once fed; a box kind never becomes a twin (it is not
/// elected), so `become_twin` is recorded to prove it is never called.
struct LakeSteps {
became: Arc<Mutex<Option<Ipv4Addr>>>,
}
impl Steps for LakeSteps {
fn primary_ready(&self, _feed: &BTreeMap<String, String>) -> Result<bool, String> {
Ok(true)
}
fn become_twin(&self, primary: Ipv4Addr) -> Result<(), String> {
*self.became.lock().unwrap() = Some(primary);
Ok(())
}
fn follow_state(&self) -> Result<(u64, u64), String> {
Ok((3, 3))
}
}
struct Collect(Mutex<Vec<String>>);
impl Sink for Collect {
fn lines(&self, _vm: &str, _lines: &[String]) {}
fn status(&self, _vm: &str, _s: Option<&Signal>, _l: &str) {}
fn ledger(&self, _d: &Door) {}
fn say(&self, line: &str) {
self.0.lock().unwrap().push(line.to_string());
}
fn secret_offer(&self, _vm: &str, _o: &nordisk_pairing::bootstrap::SecretOffer) -> Result<(), String> {
Ok(())
}
}
struct OneBox {
installed: Arc<Installed<LakeLike>>,
at: SocketAddrV4,
root: std::path::PathBuf,
became: Arc<Mutex<Option<Ipv4Addr>>>,
rested: Arc<Mutex<Option<nordisk_pairing::pair::Machine>>>,
}
/// One Lake-shaped box of kind `kind`: its data root, the medium's seed, the bootstrap listener on
/// loopback, and the shared loop on its own thread.
fn boot(ca: &InstallerCa, name: &str, kind: Kind, cores: u32, done: Arc<AtomicBool>) -> OneBox {
let boot = ca.issue_bootstrap(H).unwrap();
let seed = Seed {
estate: ESTATE.into(),
tunnel: TunnelKind::Wireguard,
front: Default::default(),
kind,
installer_ca_pem: ca.cert_pem.clone(),
bootstrap_cert_pem: boot.cert_pem,
bootstrap_key_pem: boot.key_pem,
peers_expected: 4,
mullvad: None,
tailscale: None,
installer_link: None,
};
let root = std::env::temp_dir().join(format!("lake-boxes-pair-{}-{name}", std::process::id()));
let _ = std::fs::remove_dir_all(&root);
let installed = Arc::new(Installed::open(&root, seed.clone(), LakeLike { kind, cores, fed_role: Mutex::new(None) }).unwrap());
let l = TcpListener::bind("127.0.0.1:0").unwrap();
let std::net::SocketAddr::V4(at) = l.local_addr().unwrap() else { panic!() };
let log: &'static nordisk_pairing::bootlog::BootLog = Box::leak(Box::new(nordisk_pairing::bootlog::BootLog::new(100)));
log.push("korp-start: exec lake\nlake: first boot");
let (vm, ca_pem) = (installed.clone(), ca.cert_pem.clone());
std::thread::spawn(move || nordisk_pairing::bootstrap::serve(vm.as_ref(), log, &ca_pem, l, &|_| {}));
let became = Arc::new(Mutex::new(None));
let rested = Arc::new(Mutex::new(None));
let (i, b, r, root2) = (installed.clone(), became.clone(), rested.clone(), root.clone());
std::thread::spawn(move || {
let tunnel: boxloop::TunnelFactory = Box::new(|_, _| Ok(Box::new(NoTunnel::default()) as Box<dyn Tunnel>));
let doors = BoxDoors::new(i, LakeSteps { became: b }, tunnel);
let seed = seed.clone();
let m = boxloop::run(&root2, &seed, &doors, &|_| {}, &|d| {
if !done.load(Ordering::SeqCst) {
std::thread::sleep(d.min(Duration::from_millis(100)));
}
});
*r.lock().unwrap() = Some(m);
});
OneBox { installed, at, root, became, rested }
}
#[test]
fn three_lake_boxes_each_of_its_own_kind_pair_and_the_installer_closes_on_paired() {
let ca = InstallerCa::mint(ESTATE, H).unwrap();
let done = Arc::new(AtomicBool::new(false));
let (fe, be) = (Kind::named("fe").unwrap(), Kind::named("be").unwrap());
let boxes = [("fe", fe, 2, [127, 0, 0, 1]), ("be1", be, 4, [127, 0, 0, 2]), ("be2", be, 1, [127, 0, 0, 3])];
let booted: Vec<OneBox> = boxes.iter().map(|(name, kind, cores, _)| boot(&ca, name, *kind, *cores, done.clone())).collect();
let overlays = [nordisk_pairing::OVERLAY_BOX_A, nordisk_pairing::OVERLAY_BOX_B, Ipv4Addr::new(10, 44, 0, 4)];
let vms: Vec<PlanVm> = boxes
.iter()
.zip(overlays)
.map(|((name, kind, _, public), overlay)| PlanVm { name: (*name).into(), public: Ipv4Addr::from(*public), utility: Ipv4Addr::new(10, 13, 8, public[3] + 1), overlay, kind: *kind, feed: BTreeMap::new() })
.collect();
// The plan as an operator writes it: each VM tagged with its product's own kind.
let plan = Plan {
estate: ESTATE.into(),
tunnel: TunnelKind::Wireguard,
vms,
feed: BTreeMap::new(),
front: Some(FrontTunnel { tunnel_pubkey: "xTIBA5rboUvnH4htodjb6e697QjLERt1NAB4mZqp8Dg=".into(), endpoint: "10.13.8.1:51820".into() }),
port: 0,
link_port: 0,
budget_s: 90,
close_when: CloseWhen::Paired,
box_cert_valid_s: None,
front_waits_for: Vec::new(),
};
let m = Material { ca: ca.clone(), me: ca.issue_installer(H).unwrap(), link: None };
let paths: Vec<(PlanVm, Box<dyn Path>)> = plan.vms.iter().zip(&booted).map(|(v, b)| (v.clone(), Box::new(DirectPath { at: b.at }) as Box<dyn Path>)).collect();
let sink = Collect(Mutex::new(Vec::new()));
let out = run(&plan, &m, paths, &sink).unwrap_or_else(|e| panic!("{e}\n{:#?}", sink.0.lock().unwrap()));
done.store(true, Ordering::SeqCst);
assert!(out.vms.iter().all(|v| v.closed && v.admitted), "{out:?}");
assert!(sink.0.lock().unwrap().iter().any(|l| l.contains("every VM is paired")), "closed on PAIRED, not on fed: {:#?}", sink.0.lock().unwrap());
for ((name, kind, _, _), b) in boxes.iter().zip(&booted) {
assert!(b.installed.closed(), "{name}: the installer closed its listener");
assert_eq!(b.installed.product.kind(), *kind, "{name}: the box names its own kind");
assert_eq!(*b.installed.product.fed_role.lock().unwrap(), Some(Role::Primary), "{name}: a box kind stands alone, its own primary");
let s = Signal::parse(&std::fs::read_to_string(b.root.join(boxside::SIGNAL_REL)).unwrap()).unwrap();
assert!(s.paired && s.name == *name && s.role == Some(Role::Primary), "{name}: the signal the box wrote itself: {s:?}");
let v = out.vms.iter().find(|v| v.name == *name).unwrap();
assert!(v.signal.as_deref().is_some_and(|l| l.ends_with("PAIRED")), "{name}: the installer read PAIRED: {v:?}");
assert_eq!(*b.became.lock().unwrap(), None, "{name}: a box kind becomes nobody's twin");
}
for b in &booted {
for _ in 0..50 {
if b.rested.lock().unwrap().is_some() {
break;
}
std::thread::sleep(Duration::from_millis(20));
}
assert!(b.rested.lock().unwrap().as_ref().is_some_and(|m| m.paired()), "the loop rests PAIRED");
let _ = std::fs::remove_dir_all(&b.root);
}
}