crates/monetize-server/src/estate_order/narrate.rs
//! **DEV-NARRATE on the ONE box** (lane N2, 2026-10-20; the owner: debug logs from test estates
//! exactly like Gunnar's dev builds).
//!
//! A DEV build of Gunnar or Holger narrates its boot log over udp/51822, sealed to the
//! installer's X25519 key (Skidbladnir `nordisk-narrate`: knock, WHO, authenticated HELLO,
//! then the log from line 0). For an estate the kit installs, the kit listens
//! (private-gunnar-ops `xtask narrate`). For an estate the ONE box lays, the installer IS the
//! ONE box: its key is the custody's `link.toml` ([`crate::custody::LINK_KEYS_FILE`]), and
//! this module listens.
//!
//! * **Which estates**: the site order of a site that says `[estates.<label>] narrate = true`
//! ([`crate::registry::Registry::narrates`]), laid and not gone, on a box that says where it
//! is heard from (`[estate_orders] narrate_from`, the box's public IPv4 — what the estate's
//! firewall admits udp/51822 from, and nothing else). Never an estate sold on a site.
//! * **How**: one jera job on its own lane ([`look`], from every tick; never a raw thread)
//! listens to every such estate's boxes for [`WINDOW`], then ends; the next tick queues the
//! next window. A window is queued while ANY site says narrate, hearable or not, and runs
//! its whole length (a site laid during it is picked up at the next [`RELOOK`]).
//! * **What it says**: ONE line per site that says narrate per window, at its end —
//! `narrate <label>: heard N lines` or `narrate <label>: not heard: <reason>` ([`Heard::said`];
//! 2026-10-20: after a re-exec the box's receiver said nothing at all, and nobody could tell
//! whether it ran). The installer key is read IN PLACE from the custody for each window
//! ([`crate::custody::Custody::installer_link`]) and held in memory only — never written,
//! never logged. The streams' state (which boot, which line is next) outlives a window, so a
//! line is written once.
//! * **Where**: `<data>/estate-orders/<company>/narrate/<vm>.narrate.log`, one per machine
//! (its role), capped at [`LOG_CAP`]: past it the file becomes `<vm>.narrate.log.1` (the one
//! before is dropped) and a new one starts ([`RotatingLog`]).
//! * **Who reads it**: the operator alone — `EstateOrder.NarrateLog` ([`tail`]), the console's
//! admin-only "Logs" page and its text route.
//!
//! The receiver half (the crypto) is built with this crate's `narrate` feature (the ONE box's
//! monetize-admin enables it); without it the window listens to nothing and each site's line
//! says so.
use std::io::Write as _;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use crate::state::AppState;
/// nordisk-narrate's port on every VM (`nordisk_narrate::PORT`, the same literal).
pub const PORT: u16 = 51822;
/// The jera job kind of one listening window.
pub const NARRATE_JOB: &str = "monetize_narrate_listen";
/// How long one job listens before it ends (the next tick queues the next).
pub const WINDOW: Duration = Duration::from_secs(10 * 60);
/// How often a window looks again at which estates are narrated (a site turned off stops here).
pub const RELOOK: Duration = Duration::from_secs(30);
/// One VM's log is cut at this size: the older half moves to `.1`.
pub const LOG_CAP: u64 = 4 * 1024 * 1024;
/// The narration's subdirectory of a run's directory.
pub const DIR: &str = "narrate";
/// What the console's tail reads at most, per VM, unless asked for less.
pub const TAIL_BYTES: usize = 64 * 1024;
/// **One machine to listen to**: its role (the log's name) and `<public IPv4>:51822`.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Vm {
pub role: String,
pub addr: SocketAddr,
}
/// **One narrated estate**: its run's company, its site's label, its log directory, its boxes.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Narrated {
pub company: String,
pub label: String,
pub dir: PathBuf,
pub vms: Vec<Vm>,
}
/// `<run dir>/narrate/<vm>.narrate.log`.
pub fn log_path(run_dir: &Path, vm: &str) -> PathBuf {
run_dir.join(DIR).join(format!("{vm}.narrate.log"))
}
/// **Every site that says narrate, and whether it can be heard now**, in label order: `Ok` the
/// estate to listen to; `Err` one sentence, why not (said in the window's line and the tail's
/// words). A site that does not say `narrate = true` is not here.
pub fn sites(state: &AppState) -> Vec<(String, Result<Narrated, String>)> {
let mut labels: Vec<String> = state
.registry
.declared_estates()
.into_iter()
.map(|(l, _)| l)
.filter(|l| state.registry.narrates(l))
.collect();
labels.sort();
labels
.into_iter()
.map(|label| {
let v = site(state, &label);
(label, v)
})
.collect()
}
fn site(state: &AppState, label: &str) -> Result<Narrated, String> {
let cfg = state
.estates
.cfg
.as_ref()
.ok_or("this box sells no estates ([estate_orders] is absent)")?;
let from = cfg.narrate_from.trim();
if from.is_empty() {
return Err("this box names no `[estate_orders] narrate_from` (its public IPv4, the one address a narrated estate admits udp/51822 from)".into());
}
let r = state
.estates
.all()
.into_iter()
.filter(|r| r.site_order && r.estate == label && !super::phase::is_gone(&r.phase))
.max_by_key(|r| r.ordered_at_ms)
.ok_or("no site order of it stands here (never laid from this box, or gone)")?;
let laid = r
.laid
.as_ref()
.ok_or_else(|| format!("its site order is {}: nothing of it is laid yet", r.phase))?;
let dir = state
.estates
.run_dir(&r.company)
.ok_or("this box keeps no estate-orders directory")?;
let vms = laid
.machines
.iter()
.filter_map(|m| {
m.public_ipv4
.trim()
.parse::<std::net::Ipv4Addr>()
.ok()
.map(|ip| Vm {
role: m.role.clone(),
addr: SocketAddr::from((ip, PORT)),
})
})
.collect::<Vec<_>>();
if vms.is_empty() {
return Err("no laid box of it has a public IPv4 to knock on".into());
}
Ok(Narrated {
company: r.company.clone(),
label: r.estate.clone(),
dir,
vms,
})
}
/// **The estates narrated now**: each laid site order, not gone, whose site says `narrate`,
/// with the boxes that have a public address — none when the box says no `narrate_from`.
pub fn narrated(state: &AppState) -> Vec<Narrated> {
sites(state)
.into_iter()
.filter_map(|(_, n)| n.ok())
.collect()
}
/// One target of a window: a site's label, and its estate with the installer's key — or why
/// it is not heard.
pub type Target = (String, Result<(Narrated, [u8; 32]), String>);
/// **[`sites`] with each hearable estate's installer key** read IN PLACE from the custody
/// ([`crate::custody::Custody::installer_link`]): held in memory only, never written, never logged.
pub fn keyed(state: &AppState) -> Vec<Target> {
sites(state)
.into_iter()
.map(|(label, n)| {
let t = n.and_then(|n| {
let custody = state.estates.custody.as_ref().ok_or("this box keeps no custody, so no installer key")?;
match custody.installer_link(&n.company)? {
Some(link) => installer_secret_of(&link).map(|key| (n, key)),
None => Err("the custody holds no installer link file for it (laid before narrate was on, or adopted from a kit)".to_string()),
}
});
(label, t)
})
.collect()
}
/// **What one window heard**: lines written per site, and each site's last look (its boxes,
/// or why it is not heard).
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Heard {
pub lines: std::collections::BTreeMap<String, u64>,
pub last: std::collections::BTreeMap<String, Result<Vec<Vm>, String>>,
}
impl Heard {
/// Note a look at the targets.
pub fn saw(&mut self, targets: &[Target]) {
for (label, t) in targets {
self.last.insert(
label.clone(),
t.as_ref().map(|(n, _)| n.vms.clone()).map_err(Clone::clone),
);
}
}
/// Every line heard this window.
pub fn total(&self) -> u64 {
self.lines.values().sum()
}
/// Pure: **ONE line per site**, in label order — `narrate <label>: heard N lines`, or
/// `narrate <label>: not heard: <reason>`; `from` is the box's `narrate_from`.
pub fn said(&self, from: &str) -> Vec<String> {
self.last
.iter()
.map(|(label, last)| match (self.lines.get(label).copied().unwrap_or(0), last) {
(n, _) if n > 0 => format!("narrate {label}: heard {n} line{}", if n == 1 { "" } else { "s" }),
(_, Err(why)) => format!("narrate {label}: not heard: {why}"),
(_, Ok(vms)) => format!(
"narrate {label}: not heard: none of its boxes ({}) sent a line over udp/{PORT} this window (not a DEV build, not running, or its firewall does not admit {})",
vms.iter().map(|v| format!("{} {}", v.role, v.addr.ip())).collect::<Vec<_>>().join(", "),
if from.is_empty() { "this box" } else { from }
),
})
.collect()
}
}
/// **A VM's log, capped**: appended line by line; past [`LOG_CAP`] (or the cap given) the file
/// is renamed `<name>.1` — the one before it dropped — and a new one begun.
pub struct RotatingLog {
path: PathBuf,
cap: u64,
}
impl RotatingLog {
pub fn new(path: PathBuf, cap: u64) -> RotatingLog {
RotatingLog {
path,
cap: cap.max(1),
}
}
/// The rotated file beside it.
pub fn older(path: &Path) -> PathBuf {
let mut s = path.as_os_str().to_owned();
s.push(".1");
PathBuf::from(s)
}
/// Append `lines`, rotating first when they would carry the file past its cap.
pub fn append(&self, lines: &[String]) -> std::io::Result<()> {
if lines.is_empty() {
return Ok(());
}
if let Some(d) = self.path.parent() {
std::fs::create_dir_all(d)?;
}
let adding: u64 = lines.iter().map(|l| l.len() as u64 + 1).sum();
let now = std::fs::metadata(&self.path).map(|m| m.len()).unwrap_or(0);
if now > 0 && now + adding > self.cap {
std::fs::rename(&self.path, Self::older(&self.path))?;
}
let mut f = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&self.path)?;
let mut text = String::with_capacity(adding as usize);
for l in lines {
text.push_str(l);
text.push('\n');
}
f.write_all(text.as_bytes())?;
f.flush()
}
}
/// **One VM's log as the console reads it.**
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct VmLog {
pub vm: String,
/// The tail: whole lines, oldest first, at most the bytes asked (the rotated file's end
/// first when the current one is shorter).
pub text: String,
/// The bytes held for this VM (current and rotated).
pub bytes: u64,
/// The last write, Unix ms; 0: none yet.
pub updated_unix_ms: u64,
}
/// **What the operator reads**: an estate's narration.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct EstateLogs {
pub estate: String,
/// Whether the estate is narrated NOW (its site says so, and the box says where from).
pub narrated: bool,
/// One sentence: what is heard, or why nothing is.
pub words: String,
pub vms: Vec<VmLog>,
}
fn tail_of(path: &Path, max: usize) -> (String, u64, u64) {
let read = |p: &Path| std::fs::read(p).unwrap_or_default();
let older = RotatingLog::older(path);
let (cur, old) = (read(path), read(&older));
let bytes = (cur.len() + old.len()) as u64;
let mtime = |p: &Path| {
std::fs::metadata(p)
.and_then(|m| m.modified())
.ok()
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
.map(|d| d.as_millis() as u64)
.unwrap_or(0)
};
let updated = mtime(path).max(mtime(&older));
let mut all = old;
all.extend_from_slice(&cur);
let from = all.len().saturating_sub(max);
let mut slice = &all[from..];
// Whole lines only: a cut in the middle of one drops its head.
if from > 0 {
if let Some(nl) = slice.iter().position(|b| *b == b'\n') {
slice = &slice[nl + 1..];
}
}
(String::from_utf8_lossy(slice).into_owned(), bytes, updated)
}
/// **`EstateOrder.NarrateLog`**: the narration of `estate` (a site's label, or its run's
/// company) — every VM's tail, or the one VM named. The operator's alone (the service checks).
pub fn tail(
state: &AppState,
estate: &str,
vm: &str,
max_bytes: usize,
) -> Result<EstateLogs, String> {
let run = state
.estates
.get(estate.trim())
.ok_or_else(|| format!("{estate} is no estate laid here"))?;
let dir = state
.estates
.run_dir(&run.company)
.ok_or("this box keeps no estate-orders directory")?;
let max = if max_bytes == 0 {
TAIL_BYTES
} else {
max_bytes.min(LOG_CAP as usize)
};
let narrated = narrated(state).iter().any(|n| n.company == run.company);
let mut roles: Vec<String> = run
.laid
.as_ref()
.map(|l| l.machines.iter().map(|m| m.role.clone()).collect())
.unwrap_or_default();
// A log left from a machine no longer laid is still read.
if let Ok(rd) = std::fs::read_dir(dir.join(DIR)) {
for e in rd.flatten() {
if let Some(role) = e
.file_name()
.to_str()
.and_then(|n| n.strip_suffix(".narrate.log"))
{
if !roles.iter().any(|r| r == role) {
roles.push(role.to_string());
}
}
}
}
let vm = vm.trim();
if !vm.is_empty() {
if !roles.iter().any(|r| r == vm) {
return Err(format!(
"{estate} has no machine {vm:?} (it has {})",
roles.join(", ")
));
}
roles.retain(|r| r == vm);
}
let vms: Vec<VmLog> = roles
.into_iter()
.map(|role| {
let (text, bytes, updated_unix_ms) = tail_of(&log_path(&dir, &role), max);
VmLog {
vm: role,
text,
bytes,
updated_unix_ms,
}
})
.collect();
let label = if run.estate.is_empty() {
run.company.clone()
} else {
run.estate.clone()
};
let from = state
.estates
.cfg
.as_ref()
.map(|c| c.narrate_from.trim().to_string())
.unwrap_or_default();
let words = if narrated {
format!("{label} is narrated: its boxes' boot logs are heard here (udp/{PORT} open from {from} only)")
} else if !run.site_order || !state.registry.narrates(&run.estate) {
format!("{label} is not narrated: its site does not say `narrate = true` (only an operator's own test site may)")
} else {
let why = site(state, &run.estate)
.err()
.unwrap_or_else(|| "it is not the site order this box hears".into());
format!("{label} says narrate and is not heard: {why}")
};
Ok(EstateLogs {
estate: label,
narrated,
words,
vms,
})
}
/// **Queue the next listening window** when an estate is narrated and none is queued: from
/// every tick. Owner rule: background work is a jera job, never a raw thread.
pub fn look(state: &Arc<AppState>) {
// Any site that SAYS narrate: a window runs, and says why a site is not heard.
if sites(state).is_empty() {
return;
}
let lane = match lane(state) {
Ok(l) => l,
Err(e) => {
eprintln!("monetize-server: narrate: not listening this tick: {e}");
return;
}
};
let spec = jera::jobs::chain::ChainSpec::new(
NARRATE_JOB,
"narrate",
"narrate",
0,
"monetize",
serde_json::json!({ "window_s": WINDOW.as_secs() }),
)
.with_target("listen to the narrated estates' boot logs (DEV-NARRATE)")
.with_priority(jera::jobs::chain::Priority::Low);
let _ = lane.submit(spec);
}
/// The narrate lane of `state`: one jera ChainQueue, started on the first ask on the tick's
/// runtime. `Err` when the tick runs with no runtime.
fn lane(state: &Arc<AppState>) -> Result<Arc<jera::jobs::chain::ChainQueue>, String> {
if let Some(q) = state.estates.narrate_lane.get() {
return Ok(q.clone());
}
let rt = tokio::runtime::Handle::try_current()
.map_err(|_| "no runtime to run the narrate lane on".to_string())?;
// The ledger, sampled (RA4-6): a FAILED row always, other rows at most once per 30 s.
let sink = crate::jobs::lane_sink("narrate: the listening");
let queue = jera::jobs::chain::ChainQueue::new(sink);
let q = state
.estates
.narrate_lane
.get_or_init(|| queue.clone())
.clone();
if Arc::ptr_eq(&q, &queue) {
let weak = Arc::downgrade(state);
// A32-14, stays a task: the narrate lane's jera worker.
rt.spawn(queue.clone().run(move |_spec| {
let Some(st) = weak.upgrade() else {
anyhow::bail!("the server stopped")
};
drop(st);
let heard = listen(&weak, WINDOW).map_err(|e| anyhow::anyhow!(e))?;
Ok(serde_json::json!({ "lines": heard }))
}));
}
Ok(q)
}
/// The installer's X25519 private key out of a custody `link.toml` (`installer_private`, base64).
pub fn installer_secret_of(link_toml: &str) -> Result<[u8; 32], String> {
let v: toml::Value = toml::from_str(link_toml)
.map_err(|e| format!("the link file is not TOML: {}", e.message()))?;
let k = v
.get("installer_private")
.and_then(|k| k.as_str())
.ok_or("the link file names no installer_private")?;
b64_32(k)
}
/// A standard-base64 32-byte key (44 characters, one `=`).
fn b64_32(s: &str) -> Result<[u8; 32], String> {
let val = |c: u8| -> Option<u32> {
Some(match c {
b'A'..=b'Z' => c - b'A',
b'a'..=b'z' => c - b'a' + 26,
b'0'..=b'9' => c - b'0' + 52,
b'+' => 62,
b'/' => 63,
_ => return None,
} as u32)
};
let t = s.trim().trim_end_matches('=').as_bytes();
let mut out = Vec::with_capacity(33);
let (mut acc, mut bits) = (0u32, 0u32);
for &c in t {
acc = (acc << 6) | val(c).ok_or("the installer key is not base64")?;
bits += 6;
if bits >= 8 {
bits -= 8;
out.push((acc >> bits) as u8);
acc &= (1 << bits) - 1;
}
}
out.try_into()
.map_err(|v: Vec<u8>| format!("the installer key is {} bytes, not 32", v.len()))
}
/// **One listening window** over every narrated estate's boxes, for `window` (or until the
/// server stops); then ONE line per site that says narrate ([`Heard::said`]). Answers how many
/// lines were written. The streams outlive the window.
pub fn listen(state: &std::sync::Weak<AppState>, window_len: Duration) -> Result<u64, String> {
let from = state
.upgrade()
.and_then(|s| {
s.estates
.cfg
.as_ref()
.map(|c| c.narrate_from.trim().to_string())
})
.unwrap_or_default();
let heard = window(
&|| {
let st = state.upgrade()?;
Some(keyed(&st))
},
window_len,
RELOOK,
)?;
for line in heard.said(&from) {
eprintln!("monetize-server: {line}");
}
Ok(heard.total())
}
/// **One window** over what `targets` names (asked again every `relook`; `None`: the server
/// stopped, the window ends). It runs its whole length even when nothing is hearable.
#[cfg(feature = "narrate")]
pub fn window(
targets: &dyn Fn() -> Option<Vec<Target>>,
window: Duration,
relook: Duration,
) -> Result<Heard, String> {
receiver::listen(targets, window, relook)
}
/// Without the `narrate` feature: nothing is heard, and each site's line says so.
#[cfg(not(feature = "narrate"))]
pub fn window(
targets: &dyn Fn() -> Option<Vec<Target>>,
window: Duration,
relook: Duration,
) -> Result<Heard, String> {
const NO_RECEIVER: &str = "ALERT narrate.no-receiver this build links no DEV-NARRATE receiver (monetize-server feature `narrate`)";
let t0 = std::time::Instant::now();
let mut heard = Heard::default();
while t0.elapsed() < window {
let Some(t) = targets() else { break };
let t: Vec<Target> = t
.into_iter()
.map(|(l, t)| (l, t.and_then(|_| Err(NO_RECEIVER.to_string()))))
.collect();
heard.saw(&t);
let next = std::time::Instant::now() + relook;
while std::time::Instant::now() < next && t0.elapsed() < window {
std::thread::sleep(Duration::from_millis(50).min(relook));
}
}
Ok(heard)
}
/// **The receiver over nordisk-narrate's pure state machine** (`receiver::Stream`): one UDP
/// socket per window, every narrated VM knocked, hello'd and read, each line written once.
#[cfg(feature = "narrate")]
pub mod receiver {
use super::*;
use nordisk_narrate::receiver::Stream;
use std::collections::BTreeMap;
use std::sync::Mutex;
use std::time::Instant;
/// Each VM's stream, by `<company>/<role>`, across windows (a line is written once).
static STREAMS: Mutex<BTreeMap<String, Stream>> = Mutex::new(BTreeMap::new());
/// The streams' state, for a test: `(key, lines written)`.
pub fn seen() -> Vec<(String, u64)> {
STREAMS
.lock()
.unwrap_or_else(|p| p.into_inner())
.iter()
.map(|(k, s)| (k.clone(), s.seen.lines))
.collect()
}
/// **Listen for `window`** to what `targets` names (asked again every `relook`; `None`: the
/// server stopped). Each estate's key is used in memory only. A window with nothing to hear
/// runs its length all the same.
pub fn listen(
targets: &dyn Fn() -> Option<Vec<Target>>,
window: Duration,
relook: Duration,
) -> Result<Heard, String> {
let sock = std::net::UdpSocket::bind("0.0.0.0:0")
.map_err(|e| format!("narrate: no UDP socket: {e}"))?;
sock.set_read_timeout(Some(Duration::from_millis(50)))
.map_err(|e| e.to_string())?;
let t0 = Instant::now();
let mut heard = Heard::default();
let mut buf = vec![0u8; 4096];
let mut looked: Option<Instant> = None;
let mut now_targets: Vec<(Narrated, crypto_box::SecretKey)> = Vec::new();
while t0.elapsed() < window {
if looked.is_none_or(|t| t.elapsed() >= relook) {
looked = Some(Instant::now());
let Some(t) = targets() else { break };
heard.saw(&t);
now_targets = t
.into_iter()
.filter_map(|(_, t)| t.ok())
.map(|(n, k)| (n, crypto_box::SecretKey::from(k)))
.collect();
}
if now_targets.is_empty() {
std::thread::sleep(Duration::from_millis(50));
continue;
}
let mut streams = STREAMS.lock().unwrap_or_else(|p| p.into_inner());
let now = Instant::now();
for (n, key) in &now_targets {
for vm in &n.vms {
let k = format!("{}/{}", n.company, vm.role);
let s = streams
.entry(k.clone())
.or_insert_with(|| Stream::new(&vm.role, vm.addr));
if s.addr != vm.addr {
// The machine was laid again at a new address: a new stream.
*s = Stream::new(&vm.role, vm.addr);
}
for d in s.due(key, now) {
let _ = sock.send_to(&d, vm.addr);
}
}
}
let drain = Instant::now();
while drain.elapsed() < Duration::from_millis(200) {
let Ok((len, from)) = sock.recv_from(&mut buf) else {
break;
};
let Some((n, key, vm)) = now_targets.iter().find_map(|(n, key)| {
n.vms.iter().find(|v| v.addr == from).map(|v| (n, key, v))
}) else {
continue;
};
let Some(s) = streams.get_mut(&format!("{}/{}", n.company, vm.role)) else {
continue;
};
let lines = s.on_datagram(key, &buf[..len], Instant::now());
if lines.is_empty() {
continue;
}
let log = RotatingLog::new(log_path(&n.dir, &vm.role), LOG_CAP);
match log.append(&lines) {
Ok(()) => {
*heard.lines.entry(n.label.clone()).or_default() += lines.len() as u64
}
Err(e) => eprintln!(
"monetize-server: narrate {} {}: the log could not be written: {e}",
n.label, vm.role
),
}
}
}
Ok(heard)
}
}
#[cfg(test)]
mod tests {
use super::*;
fn tmp(name: &str) -> PathBuf {
let d =
std::env::temp_dir().join(format!("monetize-narrate-{}-{name}", std::process::id()));
let _ = std::fs::remove_dir_all(&d);
d
}
/// ★ **A VM's log is capped**: past the cap the file moves to `.1` (the one before is
/// dropped) and a new one begins; the tail reads across both, whole lines only.
#[test]
fn the_log_rotates_at_its_cap_and_the_tail_reads_whole_lines() {
let dir = tmp("rotate");
let path = log_path(&dir, "main");
let log = RotatingLog::new(path.clone(), 100);
log.append(
&(0..5)
.map(|i| format!("line {i:02} ........"))
.collect::<Vec<_>>(),
)
.unwrap();
assert!(
!RotatingLog::older(&path).exists(),
"under the cap: one file"
);
log.append(&["line 05 ........".to_string()]).unwrap();
assert!(
RotatingLog::older(&path).exists(),
"past the cap: the old file moved to .1"
);
assert_eq!(
std::fs::read_to_string(&path).unwrap(),
"line 05 ........\n"
);
log.append(
&(6..12)
.map(|i| format!("line {i:02} ........"))
.collect::<Vec<_>>(),
)
.unwrap();
let older = std::fs::read_to_string(RotatingLog::older(&path)).unwrap();
assert!(
older.starts_with("line 05") && !older.contains("line 00"),
"only one older file is kept: {older}"
);
let (text, bytes, at) = tail_of(&path, 40);
assert!(bytes > 0 && at > 0);
assert!(
text.ends_with("line 11 ........\n") && text.starts_with("line "),
"whole lines: {text:?}"
);
let _ = std::fs::remove_dir_all(&dir);
}
/// ★ **One line per narrated site per window** (2026-10-20: the receiver logged nothing
/// after a re-exec): "heard N lines", or "not heard: <reason>" — a box that never answered
/// names its boxes and the one address its firewall admits.
#[test]
fn a_window_says_one_line_per_site_heard_or_why_not() {
let vm = Vm {
role: "main".into(),
addr: "198.51.100.9:51822".parse().unwrap(),
};
let mut h = Heard::default();
h.last.insert(
"test2-gunnar-rs".into(),
Err("the custody holds no installer link file for it".into()),
);
h.last
.insert("test3-holger-rs".into(), Ok(vec![vm.clone()]));
h.last.insert("test4-gunnar-rs".into(), Ok(vec![vm]));
h.lines.insert("test4-gunnar-rs".into(), 12);
let said = h.said("203.0.113.7");
assert_eq!(said.len(), 3, "{said:#?}");
assert_eq!(
said[0],
"narrate test2-gunnar-rs: not heard: the custody holds no installer link file for it"
);
assert!(
said[1].starts_with("narrate test3-holger-rs: not heard: ")
&& said[1].contains("main 198.51.100.9")
&& said[1].contains("203.0.113.7"),
"{}",
said[1]
);
assert_eq!(said[2], "narrate test4-gunnar-rs: heard 12 lines");
assert_eq!(h.total(), 12);
}
/// ★ **A window with nothing to hear still runs its length** (a site laid during it is
/// picked up at the next relook) and says why each site is not heard.
#[test]
fn a_window_with_nothing_to_hear_runs_its_length_and_says_why() {
let targets = || {
Some(vec![(
"test2-gunnar-rs".to_string(),
Err::<(Narrated, [u8; 32]), String>("no link".to_string()),
)])
};
let t0 = std::time::Instant::now();
let h = window(
&targets,
Duration::from_millis(400),
Duration::from_millis(100),
)
.unwrap();
assert!(
t0.elapsed() >= Duration::from_millis(400),
"the window was cut short: {:?}",
t0.elapsed()
);
assert_eq!(
h.said(""),
vec!["narrate test2-gunnar-rs: not heard: no link".to_string()]
);
let gone = || None;
let t0 = std::time::Instant::now();
assert_eq!(
window(&gone, Duration::from_secs(30), Duration::from_millis(100))
.unwrap()
.said(""),
Vec::<String>::new()
);
assert!(
t0.elapsed() < Duration::from_secs(5),
"a stopped server ends the window"
);
}
#[test]
fn the_installer_key_is_read_out_of_the_link_file() {
let key = [7u8; 32];
let b64 = "BwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwc=";
assert_eq!(
installer_secret_of(&format!(
"installer_private = \"{b64}\"\nvm_public = \"x\"\n"
))
.unwrap(),
key
);
assert!(installer_secret_of("vm_public = \"x\"")
.unwrap_err()
.contains("installer_private"));
assert!(installer_secret_of("installer_private = \"AAAA\"")
.unwrap_err()
.contains("32"));
}
/// ★ **The receiver, given a local sender sealed to a test installer key, writes the
/// log** (lane N2): the box's buffered lines from line 0, then a live one, each once, into
/// `<run>/narrate/<vm>.narrate.log`; a second window does not write them again.
#[cfg(feature = "narrate")]
#[test]
fn the_receiver_writes_a_local_senders_log_once_across_windows() {
use nordisk_narrate::sender::{self, Config, Source};
struct Log(std::sync::Mutex<Vec<String>>);
impl Source for Log {
fn since(&self, from: u64, max: usize) -> (u64, Vec<String>, u64) {
let l = self.0.lock().unwrap();
let from = (from as usize).min(l.len());
let lines: Vec<String> = l[from..].iter().take(max).cloned().collect();
let next = (from + lines.len()) as u64;
(0, lines, next)
}
}
let installer = nordisk_narrate::wire::generate();
let log: &'static Log = Box::leak(Box::new(Log(std::sync::Mutex::new(vec![
"korp-start: STAGE entered".into(),
"holger: STAGE listen :8443".into(),
]))));
let at = sender::start(
Config {
installer_pub: *installer.public_key().as_bytes(),
bind: "127.0.0.1:0".parse().unwrap(),
product: "holger".into(),
mac: "52:54:00:00:00:09".into(),
},
log,
)
.unwrap();
let dir = tmp("receiver");
let n = Narrated {
company: "test3holgerrs".into(),
label: "test3-holger-rs".into(),
dir: dir.clone(),
vms: vec![Vm {
role: "main".into(),
addr: at,
}],
};
let key = installer.to_bytes();
let targets = || Some(vec![(n.label.clone(), Ok((n.clone(), key)))]);
let path = log_path(&dir, "main");
let t0 = std::time::Instant::now();
while !std::fs::read_to_string(&path)
.unwrap_or_default()
.contains("STAGE listen")
{
assert!(
t0.elapsed() < Duration::from_secs(20),
"the box's lines never arrived"
);
receiver::listen(&targets, Duration::from_millis(600), RELOOK).unwrap();
}
log.0.lock().unwrap().push("holger: STAGE ready".into());
while !std::fs::read_to_string(&path)
.unwrap_or_default()
.contains("STAGE ready")
{
assert!(
t0.elapsed() < Duration::from_secs(30),
"the live line never arrived"
);
receiver::listen(&targets, Duration::from_millis(600), RELOOK).unwrap();
}
let text = std::fs::read_to_string(&path).unwrap();
let lines: Vec<&str> = text
.lines()
.filter(|l| !l.starts_with("── narrate"))
.collect();
assert_eq!(
lines,
vec![
"korp-start: STAGE entered",
"holger: STAGE listen :8443",
"holger: STAGE ready"
],
"each line once, in order:\n{text}"
);
assert_eq!(
text.lines().filter(|l| l.starts_with("── narrate")).count(),
1,
"one boot, one opening line across windows:\n{text}"
);
assert!(
text.lines().next().unwrap().contains("holger") && text.contains("52:54:00:00:00:09")
);
let _ = std::fs::remove_dir_all(&dir);
}
}