crates/installer/nordisk-narrate/src/receiver.rs
//! **The installer's side**: knock on every VM's narration port, say an
//! authenticated hello to whoever answers with a WHO that opens under the
//! installer's key, and write every line that arrives — in order, once — to
//! that VM's `<vm>.narrate.log`.
//!
//! It sends nothing but knocks and hellos, and reads nothing it cannot open:
//! a datagram that does not open under the installer's private key is counted
//! (`forged`) and dropped; one that opens but belongs to no stream it knows
//! (another boot, another ephemeral key) is `stray`; a line it already has is
//! a `duplicate` (a replay, or a resend after a loss).
//!
//! The state machine ([`Stream`]) is pure; [`start`] is the thread.
use crate::wire::{self, BoxId, Hello, Up};
use crypto_box::{PublicKey, SecretKey};
use std::collections::BTreeMap;
use std::io::Write as _;
use std::net::{SocketAddr, UdpSocket};
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
/// Knock this often until a VM answers (its NIC is not up yet, or the port
/// is closed).
pub const KNOCK_EVERY: Duration = Duration::from_secs(1);
/// Say hello this often once it answered — it is also the acknowledgement:
/// each hello carries the next line needed.
pub const HELLO_EVERY: Duration = Duration::from_secs(2);
/// Nothing from the box this long: knock again (it may have restarted, with a
/// new boot and a new ephemeral key).
pub const REKNOCK_AFTER: Duration = Duration::from_secs(6);
/// The marker line stored where lines were dropped — the log door's words
/// (nordisk-pairing `bootlog::dropped_marker`), so the two logs read alike.
pub fn dropped_marker(n: u64) -> String {
format!("… {n} lines dropped (the VM's log holds its install-time head and its newest lines) …")
}
/// The line that opens a boot's stream in `<vm>.narrate.log`.
pub fn boot_line(id: &BoxId, again: bool) -> String {
format!(
"── narrate: {} {} boot {:016x}{} — its log from line 0, sealed to this installer ──",
id.product,
id.mac,
id.boot,
if again { " (a NEW boot of this VM: it restarted)" } else { "" }
)
}
/// One VM to listen to.
#[derive(Debug, Clone)]
pub struct Target {
pub name: String,
/// `<public address>:`[`crate::PORT`].
pub addr: SocketAddr,
/// `<run>/logs/<vm>.narrate.log`.
pub out: PathBuf,
}
/// What was heard from one VM.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Seen {
pub who: Option<BoxId>,
pub boots: u32,
/// Lines written (markers included).
pub lines: u64,
pub duplicates: u64,
pub dropped: u64,
/// Datagrams that did not open under the installer's key.
pub forged: u64,
/// Datagrams that opened but belong to no stream (another boot's).
pub stray: u64,
/// The last line written.
pub last: Option<String>,
}
/// **One VM's stream**, pure.
pub struct Stream {
pub name: String,
pub addr: SocketAddr,
eph: Option<[u8; 32]>,
next: u64,
pending: BTreeMap<u64, String>,
/// `start → end` (exclusive): lines the box said it dropped.
gaps: BTreeMap<u64, u64>,
counter: u64,
last_knock: Option<Instant>,
last_hello: Option<Instant>,
last_heard: Option<Instant>,
pub seen: Seen,
}
impl Stream {
pub fn new(name: &str, addr: SocketAddr) -> Stream {
Stream {
name: name.into(),
addr,
eph: None,
next: 0,
pending: BTreeMap::new(),
gaps: BTreeMap::new(),
counter: 0,
last_knock: None,
last_hello: None,
last_heard: None,
seen: Seen::default(),
}
}
/// The next line this installer needs.
pub fn next(&self) -> u64 {
self.next
}
/// A hello counter that only grows — across receivers of one run too (a
/// box refuses a counter it has seen): the wall clock in µs, or one more.
fn tick(&mut self) -> u64 {
let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_micros() as u64).unwrap_or(0);
self.counter = now.max(self.counter + 1);
self.counter
}
/// **What to send now**, if anything: knocks until a WHO opened (and again
/// when the box went quiet), then a hello every [`HELLO_EVERY`].
pub fn due(&mut self, installer: &SecretKey, now: Instant) -> Vec<Vec<u8>> {
let mut out = Vec::new();
let quiet = self.last_heard.is_none_or(|t| now.duration_since(t) >= REKNOCK_AFTER);
if (self.eph.is_none() || quiet) && self.last_knock.is_none_or(|t| now.duration_since(t) >= KNOCK_EVERY) {
self.last_knock = Some(now);
out.push(wire::knock(wire::random_16()));
}
if let (Some(eph), Some(id)) = (self.eph, self.seen.who.as_ref()) {
if self.last_hello.is_none_or(|t| now.duration_since(t) >= HELLO_EVERY) {
self.last_hello = Some(now);
let h = Hello { boot: id.boot, counter: self.tick(), from: self.next };
out.push(wire::seal_hello(installer, &PublicKey::from(eph), &h));
}
}
out
}
/// **One datagram from this VM's address.** Returns the lines now ready,
/// in order (a boot's opening line and dropped markers included).
pub fn on_datagram(&mut self, installer: &SecretKey, d: &[u8], now: Instant) -> Vec<String> {
let (eph, up) = match wire::open_up(installer, d) {
Ok(x) => x,
Err(wire::WireError::NotOurs) => return Vec::new(),
Err(_) => {
self.seen.forged += 1;
return Vec::new();
}
};
let mut out = Vec::new();
match up {
Up::Who { id, .. } => {
self.last_heard = Some(now);
if self.seen.who.as_ref().map(|w| w.boot) != Some(id.boot) {
let again = self.seen.boots > 0;
self.seen.boots += 1;
self.eph = Some(eph);
self.next = 0;
self.pending.clear();
self.gaps.clear();
out.push(boot_line(&id, again));
self.seen.who = Some(id);
// Say hello at once rather than in two seconds.
self.last_hello = None;
}
}
Up::Data { id, first, dropped, lines } => {
if Some(eph) != self.eph || self.seen.who.as_ref().map(|w| w.boot) != Some(id.boot) {
self.seen.stray += 1;
return Vec::new();
}
self.last_heard = Some(now);
if dropped > 0 && first > self.next {
self.gaps.insert(first.saturating_sub(dropped).max(self.next), first);
}
for (i, l) in lines.into_iter().enumerate() {
let seq = first + i as u64;
if seq < self.next || self.pending.contains_key(&seq) {
self.seen.duplicates += 1;
} else {
self.pending.insert(seq, l);
}
}
loop {
if let Some(l) = self.pending.remove(&self.next) {
out.push(l);
self.next += 1;
} else if let Some(end) = self.gaps.remove(&self.next) {
out.push(dropped_marker(end - self.next));
self.seen.dropped += end - self.next;
self.next = end;
} else {
break;
}
}
}
}
self.seen.lines += out.len() as u64;
if let Some(l) = out.last() {
self.seen.last = Some(l.clone());
}
out
}
}
/// Called for every line written: `(vm, line)`.
pub type OnLine = Box<dyn Fn(&str, &str) + Send>;
/// **The running receiver.** [`Receiver::stop`] ends it and says what was heard.
pub struct Receiver {
stop: Arc<AtomicBool>,
seen: Arc<Mutex<BTreeMap<String, Seen>>>,
thread: Option<std::thread::JoinHandle<()>>,
}
impl Receiver {
/// What was heard so far, per VM.
pub fn seen(&self) -> BTreeMap<String, Seen> {
self.seen.lock().unwrap_or_else(|p| p.into_inner()).clone()
}
pub fn stop(mut self) -> BTreeMap<String, Seen> {
self.stop.store(true, Ordering::SeqCst);
if let Some(t) = self.thread.take() {
let _ = t.join();
}
self.seen()
}
}
impl Drop for Receiver {
fn drop(&mut self) {
self.stop.store(true, Ordering::SeqCst);
}
}
/// **Listen to `targets`** with the installer's private key, on one socket of
/// this machine's (any port: the boxes answer to where the hello came from).
pub fn start(installer_secret: [u8; 32], targets: Vec<Target>, on_line: OnLine) -> std::io::Result<Receiver> {
let sock = UdpSocket::bind("0.0.0.0:0")?;
sock.set_read_timeout(Some(Duration::from_millis(50)))?;
let key = SecretKey::from(installer_secret);
let stop = Arc::new(AtomicBool::new(false));
let seen = Arc::new(Mutex::new(targets.iter().map(|t| (t.name.clone(), Seen::default())).collect::<BTreeMap<_, _>>()));
let mut files = Vec::new();
for t in &targets {
if let Some(d) = t.out.parent() {
std::fs::create_dir_all(d)?;
}
files.push(std::fs::OpenOptions::new().create(true).append(true).open(&t.out)?);
}
let mut streams: Vec<Stream> = targets.iter().map(|t| Stream::new(&t.name, t.addr)).collect();
let (st, se) = (stop.clone(), seen.clone());
let thread = std::thread::Builder::new().name("narrate-receiver".into()).spawn(move || {
let mut buf = vec![0u8; 4096];
while !st.load(Ordering::SeqCst) {
let now = Instant::now();
for s in streams.iter_mut() {
for d in s.due(&key, now) {
let _ = sock.send_to(&d, s.addr);
}
}
// Drain what arrived, then go round again.
let t0 = Instant::now();
while t0.elapsed() < Duration::from_millis(200) {
let Ok((n, from)) = sock.recv_from(&mut buf) else { break };
let Some(i) = streams.iter().position(|s| s.addr == from) else { continue };
let lines = streams[i].on_datagram(&key, &buf[..n], Instant::now());
if !lines.is_empty() {
let f = &mut files[i];
for l in &lines {
let _ = writeln!(f, "{l}");
on_line(&streams[i].name, l);
}
let _ = f.flush();
}
se.lock().unwrap_or_else(|p| p.into_inner()).insert(streams[i].name.clone(), streams[i].seen.clone());
}
}
})?;
Ok(Receiver { stop, seen, thread: Some(thread) })
}
#[cfg(test)]
mod tests {
use super::*;
fn id(boot: u64) -> BoxId {
BoxId { product: "gunnar-server".into(), mac: "-".into(), boot }
}
/// **In order, once, with the gaps said.** Datagrams out of order, a
/// replayed one, and a dropped range: the log reads 0.. in order, the
/// replay is a duplicate, the drop is one marker line.
#[test]
fn lines_are_written_in_order_once_and_a_drop_is_one_marker() {
let installer = wire::generate();
let eph = wire::generate();
let pk = installer.public_key();
let mut s = Stream::new("main", SocketAddr::from(([127, 0, 0, 1], 1)));
let t = Instant::now();
let who = s.on_datagram(&installer, &wire::seal_up(&eph, &pk, &Up::Who { id: id(7), echo: [0; 16] }), t);
assert!(who[0].starts_with("── narrate: gunnar-server - boot 0000000000000007"), "{who:?}");
let d = |first: u64, dropped: u64, l: &[&str]| wire::seal_up(&eph, &pk, &Up::Data { id: id(7), first, dropped, lines: l.iter().map(|s| s.to_string()).collect() });
assert!(s.on_datagram(&installer, &d(2, 0, &["c", "d"]), t).is_empty(), "held until 0 and 1 arrive");
assert_eq!(s.on_datagram(&installer, &d(0, 0, &["a", "b"]), t), vec!["a", "b", "c", "d"]);
assert!(s.on_datagram(&installer, &d(0, 0, &["a", "b"]), t).is_empty(), "a replay writes nothing");
assert_eq!(s.seen.duplicates, 2);
assert_eq!(s.on_datagram(&installer, &d(10, 6, &["k"]), t), vec![dropped_marker(6), "k".to_string()]);
assert_eq!(s.next(), 11);
// Another boot's data before its WHO is stray, not a line.
let other = wire::generate();
assert!(s.on_datagram(&installer, &wire::seal_up(&other, &pk, &Up::Data { id: id(8), first: 0, dropped: 0, lines: vec!["x".into()] }), t).is_empty());
assert_eq!(s.seen.stray, 1);
// A garbled datagram is forged.
let mut bad = d(11, 0, &["l"]);
let n = bad.len();
bad[n - 1] ^= 1;
assert!(s.on_datagram(&installer, &bad, t).is_empty());
assert_eq!(s.seen.forged, 1);
// The restart: a new WHO, a new boot, from line 0 again.
let again = s.on_datagram(&installer, &wire::seal_up(&other, &pk, &Up::Who { id: id(8), echo: [0; 16] }), t);
assert!(again[0].contains("a NEW boot"), "{again:?}");
assert_eq!(s.next(), 0);
}
/// **Knock, then hello** — and the hello asks for the next line needed,
/// with a counter that only grows.
#[test]
fn it_knocks_until_a_who_then_hellos_from_the_next_line() {
let installer = wire::generate();
let eph = wire::generate();
let mut s = Stream::new("twin", SocketAddr::from(([127, 0, 0, 1], 1)));
let t = Instant::now();
let first = s.due(&installer, t);
assert_eq!((first.len(), &first[0][..4]), (1, &wire::TAG_KNOCK[..]));
assert!(s.due(&installer, t).is_empty(), "not again within the interval");
s.on_datagram(&installer, &wire::seal_up(&eph, &installer.public_key(), &Up::Who { id: id(3), echo: [0; 16] }), t);
let d = s.due(&installer, t);
let h1 = d.iter().find(|d| &d[..4] == wire::TAG_HELLO).expect("a hello at once");
let h1 = wire::open_hello(&eph, &installer.public_key(), h1).unwrap();
assert_eq!((h1.boot, h1.from), (3, 0));
let h2 = s.due(&installer, t + HELLO_EVERY).into_iter().find(|d| &d[..4] == wire::TAG_HELLO).unwrap();
assert!(wire::open_hello(&eph, &installer.public_key(), &h2).unwrap().counter > h1.counter);
}
}