holger-estate-front/src/role.rs
//! **The front's role, for the life of the process** — from what the installer
//! fed (kept in `<data>/pairing/joined.toml`, so a reboot runs it again at once):
//!
//! * the estate's WireGuard relay (edda `estate-front::relay`) on
//! udp/[`nordisk_pairing::RELAY_PORT`], keyed with the front's own tunnel key
//! (the key its presentation carried, which the installer feeds every box as
//! the `front` peer), listing each fed box by its overlay /32;
//! * the twin's certification ([`crate::certify`]) once the verdict names a
//! primary and a twin with utility addresses and the console bearer is fed —
//! on a thread of its own, retried until it takes, once per pair.
//!
//! gunnar's front runs the same two duties (its `role.rs`); this is holger's,
//! on the shared crates.
use std::sync::Arc;
use std::time::Duration;
use nordisk_pairing::join::JoinReply;
use crate::FrontVm;
/// How often the fed reply is looked at again.
const POLL: Duration = Duration::from_secs(2);
/// **The front's role as a duty** (holger-live `duty`, jera jobs; A28 M18): listed while it
/// runs, a look that fails or panics said as an ALERT, stopped at the front's clean stop.
pub const FRONT_ROLE: &str = "front.role";
/// The twin's certification, one duty per pair the front certifies.
pub const CERTIFY_TWIN: &str = "front.certify-twin";
/// Whether the front's role runs now (a listed duty).
pub fn is_running() -> bool {
holger_live::duty::find(FRONT_ROLE).is_some()
}
/// How often the relay's counters are said (lane D12).
pub const COUNTERS_EVERY: Duration = Duration::from_secs(180);
/// **When the relay's counters are said** (lane D12, test3-holger-rs r6: the boxes never
/// completed a handshake and the relay, which drops a wrong-key initiation silently, said
/// nothing): the first look, then once per `every`.
pub struct CountersSay {
pub every: Duration,
last: Option<std::time::Instant>,
}
impl CountersSay {
pub fn new(every: Duration) -> CountersSay {
CountersSay { every, last: None }
}
/// The line to say now, if one is due.
pub fn due(
&mut self,
now: std::time::Instant,
line: impl FnOnce() -> String,
) -> Option<String> {
if self
.last
.is_some_and(|at| now.duration_since(at) < self.every)
{
return None;
}
self.last = Some(now);
Some(line())
}
}
/// The relay's bound port — a test binds 0.
pub struct RoleConfig {
pub relay_port: u16,
/// The courier's waits between attempts (`nordisk_twin::certify::waits()`).
pub waits: fn() -> Box<dyn Iterator<Item = Duration> + Send>,
}
impl Default for RoleConfig {
fn default() -> RoleConfig {
RoleConfig {
relay_port: nordisk_pairing::RELAY_PORT,
waits: || Box::new(nordisk_twin::certify::waits()),
}
}
}
/// What the role keeps from one look to the next.
struct Role {
relay: Option<estate_front::relay::Relay>,
counters: CountersSay,
last: Option<JoinReply>,
certifying: Option<nordisk_twin::certify::Target>,
// The bearers the running certify loop reads at every call, followed by every later
// feed (gate A12: the verdict came before the twin's console key, and a loop that
// held the first feed's bearers asked the twin with none for ever).
live: Option<crate::console_key::LiveBearers>,
said: Vec<String>,
}
/// Said once: a line the role would otherwise repeat at every look.
fn say_once(said: &mut Vec<String>, l: String) {
if !said.contains(&l) {
tracing::warn!("front: {l}");
said.push(l);
}
}
/// Where the front's duties say their failures.
fn alert() -> holger_live::duty::Say {
Arc::new(|l: &str| tracing::error!("front: {l}"))
}
/// **Be the front once fed, and keep taking what is fed.** A duty ([`FRONT_ROLE`]), one look
/// per [`POLL`]; returns at once. `handle` is the runtime the duties and the certification
/// courier's calls run on.
pub fn start(vm: Arc<FrontVm>, handle: tokio::runtime::Handle, cfg: RoleConfig) {
let _on = handle.enter();
// The pair every certify loop is told is wanted (A7 row 26): a loop for another pair stops.
let wanted: nordisk_twin::certify::Wanted = std::sync::Arc::default();
let role = std::sync::Mutex::new(Role {
relay: None,
counters: CountersSay::new(COUNTERS_EVERY),
last: None,
certifying: None,
live: None,
said: Vec::new(),
});
let h = handle.clone();
let every = |n: u64| if n == 1 { Duration::ZERO } else { POLL };
// No ledger: a look every two seconds for the front's life is no row worth keeping.
let started = holger_live::duty::start(FRONT_ROLE, None, every, alert(), move |_| {
let mut st = role.lock().unwrap_or_else(|p| p.into_inner());
look(&mut st, &vm, &h, &cfg, &wanted);
Ok(holger_live::duty::Look::Again)
});
if let Err(e) = started {
tracing::error!(error = %e, "front: the role could not start");
}
}
/// One look of the role at what is fed.
fn look(
st: &mut Role,
vm: &Arc<FrontVm>,
handle: &tokio::runtime::Handle,
cfg: &RoleConfig,
wanted: &nordisk_twin::certify::Wanted,
) {
if let Some(reply) = vm.latest() {
if st.relay.is_none() {
// An estate front's own overlay interface rides the relay (lane E1): the
// front reaches main and the twin over the tunnel itself.
match vm.product.relay_key().and_then(|k| {
estate_front::relay::Relay::start_with_hub(
cfg.relay_port,
k,
nordisk_pairing::OVERLAY_FRONT,
crate::wg_only::hub(),
)
}) {
Ok(r) => {
tracing::info!(relay = %r.public(), port = r.port(), "front: the relay is up on this box's own tunnel key");
st.relay = Some(r);
vm.relay_started();
}
Err(e) => say_once(
&mut st.said,
format!("the relay cannot start — asking again: {e}"),
),
}
}
if let Some(r) = st.relay.as_ref() {
if let Some(line) = st.counters.due(std::time::Instant::now(), || r.counters()) {
tracing::info!("front: relay counters: {line}");
}
}
if let Some(r) = st.relay.as_mut() {
if st.last.as_ref() != Some(&reply) {
for line in list_on_relay(r, &reply) {
tracing::info!("front: {line}");
}
// Every box may now be listed: PAIRED is the front's own reading.
vm.advance();
}
}
let target = crate::certify::target_of(&reply);
if want(wanted, target.as_ref()) {
// A re-feed moved the pair (or named none): the loop for the old pair stops
// (it dialled the dead twin for ever, Gunnar's estate-box3 R2), and the new
// pair gets a loop of its own below.
st.certifying = None;
}
if let (Some(l), Some(_)) = (st.live.as_ref(), st.certifying.as_ref()) {
if l.follow(crate::console_key::adopted(), &reply) {
tracing::info!(
"front: the certify courier takes the boxes' console keys of the latest feed"
);
}
}
if let Some(t) = target {
let done = crate::certify::Record::load(vm.root()).is_some_and(|rec| rec.covers(&t));
if !done && st.certifying.as_ref() != Some(&t) {
match crate::console_key::bearers(&reply) {
Ok(Some(bearers)) => {
tracing::info!("front: the election is in — certifying the twin at {} against the primary at {}", t.reader_ctl, t.source_ctl);
st.certifying = Some(t.clone());
let bearers = crate::console_key::LiveBearers::new(bearers);
st.live = Some(bearers.clone());
let (root, handle, waits, wanted) = (vm.root().to_path_buf(), handle.clone(), cfg.waits, Arc::clone(wanted));
// One duty per pair: its one look is the certify loop, run on the blocking
// pool; a panic in it is an ALERT (the raw thread ended without a word).
let job = std::sync::Mutex::new(Some((bearers, t)));
let started = holger_live::duty::start(CERTIFY_TWIN, None, |_| Duration::ZERO, alert(), move |_| {
let Some((bearers, t)) = job.lock().unwrap_or_else(|p| p.into_inner()).take() else { return Ok(holger_live::duty::Look::Done) };
let boxes = crate::certify::HolgerBoxes { handle: handle.clone(), bearers, timeout: Duration::from_secs(15), wg_only: crate::wg_only::wg_only() };
match crate::certify::run_while(&root, &boxes, &t, waits(), &wanted) {
Err(e) if nordisk_twin::certify::is_superseded(&e) => tracing::warn!("front: {e}"),
Err(e) => tracing::error!("front: the twin was not certified: {e}"),
Ok(_) => {}
}
Ok(holger_live::duty::Look::Done)
});
if let Err(e) = started {
// Asked again at the next look, as if never started.
tracing::error!("front: the twin's certification could not start: {e}");
st.certifying = None;
}
}
Ok(None) => say_once(&mut st.said, format!("the verdict is in, and no box has offered its console key (`{}`) yet — the front cannot certify the twin until they do", crate::console_key::OFFER_BOX_PUBKEY)),
Err(e) => say_once(&mut st.said, e),
}
}
}
st.last = Some(reply);
}
}
/// **Tell every certify loop which pair the front wants now** (A7 row 26; Gunnar b69c4f00,
/// `FrontRole::take`): `now` is the pair the latest feed names, `None` when it names none.
/// True when that changed — every loop started for another pair then stops at its next
/// look ([`nordisk_twin::certify::run_while`]). Gunnar kept the old pair wanted when a feed
/// named none, so a loop for a twin the feed dropped went on; here it stops too.
pub fn want(
wanted: &std::sync::Mutex<Option<nordisk_twin::certify::Target>>,
now: Option<&nordisk_twin::certify::Target>,
) -> bool {
let mut w = wanted.lock().unwrap_or_else(|p| p.into_inner());
if w.as_ref() == now {
return false;
}
*w = now.cloned();
true
}
/// **List every fed box on the relay, each with the pre-shared key the installer fed the
/// front for it** (lane G5, A7 row 13; Gunnar 356e8dd2): `wg_psk.<box>` read off the front's
/// feed (`nordisk_pairing::frontside::relay_list_with_psk`) and keyed into every session
/// (edda `estate_front::relay::Relay::take_with_psk`). Said, never the key.
pub fn list_on_relay(r: &mut estate_front::relay::Relay, reply: &JoinReply) -> Vec<String> {
r.take_with_psk(&nordisk_pairing::frontside::relay_list_with_psk(reply))
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::Instant;
/// ★ **The front's clean stop stops its duties** before the power goes (A28 M18).
#[test]
fn the_front_stops_its_duties_at_the_clean_stop() {
let main = include_str!("main.rs");
let stop = &main[main
.find("shutdown_request().await")
.expect("the clean stop")..];
assert!(
stop[..stop.find("power_off_cleanly").expect("the power-off")]
.contains("holger_live::duty::stop_all()")
);
}
/// ★ **The role and the twin's certification are duties, not raw threads** (A28 M18): a
/// panic in the raw `certify-twin` thread ended it silently and the pair was never asked
/// again; the role's own thread ended the front's duties the same way.
#[test]
fn the_role_and_the_certification_are_duties_not_raw_threads() {
let src = include_str!("role.rs");
let code = &src[..src.find("#[cfg(test)]").unwrap()];
assert!(
!code.contains("std::thread::"),
"a raw thread is left in the role"
);
assert!(
code.contains("duty::start(FRONT_ROLE") && code.contains("duty::start(CERTIFY_TWIN"),
"both start as duties"
);
}
/// ★ **Each box is listed on the relay with the PSK the installer fed the front for it**
/// (lane G5, A7 row 13; Gunnar 356e8dd2): `wg_psk.<box>`, said, never the key.
#[test]
fn each_box_is_listed_on_the_relay_with_the_psk_the_front_was_fed_for_it() {
let psk = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=";
let mut r = estate_front::relay::Relay::start(0, [6u8; 32], nordisk_pairing::OVERLAY_FRONT)
.unwrap();
let key = wg_appliance_core::encode_key(
&wg_appliance_core::keys::public_from_private_bytes(&[9u8; 32]),
);
let main = nordisk_pairing::join::Peer {
name: "vm2".into(),
role: nordisk_pairing::Role::Primary,
tunnel_pubkey: key,
overlay: nordisk_pairing::OVERLAY_BOX_A,
endpoint: None,
utility: None,
};
let mut reply = crate::tests::fed(&[], vec![main]);
reply.feed.insert(
nordisk_pairing::installer::feed_wg_psk_for("vm2"),
psk.into(),
);
let said = list_on_relay(&mut r, &reply);
assert_eq!(
said,
vec![format!(
"`vm2` listed on the relay as {} with its pre-shared key",
nordisk_pairing::OVERLAY_BOX_A
)]
);
assert_eq!(r.with_psk(), vec![nordisk_pairing::OVERLAY_BOX_A]);
}
/// ★ **The relay's counters are said at the first look, then once per window** (lane D12).
#[test]
fn the_relay_counters_are_said_first_and_then_once_per_window() {
let mut c = CountersSay::new(Duration::from_secs(180));
let t0 = Instant::now();
assert_eq!(
c.due(t0, || "10.44.0.2: handshakes 0 | refused 3".into())
.as_deref(),
Some("10.44.0.2: handshakes 0 | refused 3"),
"said at once"
);
assert_eq!(
c.due(t0 + Duration::from_secs(60), || unreachable!(
"not asked inside the window"
)),
None
);
assert!(
c.due(t0 + Duration::from_secs(181), || "again".into())
.is_some(),
"said again after the window"
);
}
/// ★ **The certify loop follows the re-feed** (A7 row 26; Gunnar b69c4f00). The role
/// started a new loop per pair and never stopped the old one, so the loop for a DEAD
/// twin kept dialling it. The pair the feed names is published to every loop; a move,
/// or a feed that names none, supersedes the loops before it.
#[test]
fn a_re_feed_supersedes_the_loop_for_the_old_pair() {
use nordisk_twin::certify::{is_superseded, run_while, Boxes, Issued, Target};
let at = |reader: &str| Target {
reader_ctl: reader.into(),
source_ctl: "10.6.0.14:50051".into(),
become_twin_of: Some("10.6.0.14:50051".into()),
principal: "holger-twin".into(),
purpose: "twin".into(),
};
let (old, new) = (at("10.6.0.13:50051"), at("10.6.0.15:50051"));
let wanted: nordisk_twin::certify::Wanted = std::sync::Arc::default();
assert!(want(&wanted, Some(&old)), "the first pair is a change");
assert!(!want(&wanted, Some(&old)), "the same feed again is not");
struct Dead<'a>(
&'a nordisk_twin::certify::Wanted,
Target,
std::cell::Cell<u32>,
);
impl Boxes for Dead<'_> {
fn become_twin(&self, _: &str, _: &str) -> Result<String, String> {
self.2.set(self.2.get() + 1);
// The installer re-feeds the front while this dial times out.
want(self.0, Some(&self.1));
Err("could not be reached (Unavailable)".into())
}
fn request(&self, _: &str, _: &str) -> Result<String, String> {
unreachable!()
}
fn issue(&self, _: &str, _: &str, _: &str) -> Result<Issued, String> {
unreachable!()
}
fn install(&self, _: &str, _: &str, _: &Issued) -> Result<i64, String> {
unreachable!()
}
}
let dead = Dead(&wanted, new.clone(), std::cell::Cell::new(0));
let e = run_while(
&dead,
&old,
std::iter::repeat_n(Duration::ZERO, 5),
&wanted,
&mut |_, _| panic!("a superseded attempt is not said"),
)
.unwrap_err();
assert!(is_superseded(&e), "{e}");
assert_eq!(
dead.2.get(),
1,
"one dial of the dead twin, then the loop stops"
);
assert!(
want(&wanted, None),
"a feed that names no pair is a change too"
);
assert_eq!(*wanted.lock().unwrap(), None, "and no loop is wanted");
}
}