holger-estate-front/src/sold.rs
//! **The sold front's duties as jera jobs** (feature `metered`): keep the installer
//! material in the primary's sealed slot, do the update the ONE box opened, and recharge a
//! lost main or twin ([`crate::recharge`]). A ticker submits each under ONE idempotency key; the work runs on the
//! queue's blocking thread.
use std::sync::Arc;
use std::time::Duration;
use jera::jobs::{ChainQueue, ChainSpec, JobSink, Priority};
use crate::shop_door::{ShopDoor, UPGRADE_WITHIN};
use crate::FrontVm;
pub const KIND_MATERIAL: &str = "front.material";
pub const KIND_UPDATE: &str = "front.update";
/// A lost main or twin, recharged by the front ([`crate::recharge`]).
pub const KIND_RECHARGE: &str = "front.recharge";
/// How often the front looks.
pub const LOOK_EVERY: Duration = Duration::from_secs(120);
/// Where the front's jobs are recorded.
pub const JOBS_REL: &str = "jobs";
/// **The front's hands on a live estate**: the boxes' live doors over the kept installer
/// material (in the primary's sealed slot), its own live slot, an https GET of a release object,
/// the shop's keep upgrade and the box answering again on its control port after it.
pub struct EstateHands {
pub door: ShopDoor,
pub vm: Arc<FrontVm>,
pub handle: tokio::runtime::Handle,
/// The kept bundle, read once per look: the primary is down while it re-execs itself.
pub kept: std::sync::Mutex<Option<Vec<u8>>>,
}
impl EstateHands {
pub fn new(door: ShopDoor, vm: Arc<FrontVm>, handle: tokio::runtime::Handle) -> EstateHands {
EstateHands {
door,
vm,
handle,
kept: std::sync::Mutex::new(None),
}
}
fn bundle(&self) -> Result<Vec<u8>, String> {
let mut held = self.kept.lock().map_err(|e| e.to_string())?;
if let Some(b) = held.as_ref() {
return Ok(b.clone());
}
let reply = self.vm.latest().ok_or("not fed yet")?;
let slot = crate::sealed::slot_of(&reply, &self.handle)
.ok_or("no primary elected or no console bearer fed yet")?;
// The primary's slot first; else what this front took and holds (F9).
let b = match slot.get(crate::sealed::MATERIAL_OBJECT) {
Ok(Some(b)) if !b.is_empty() => b,
got => match crate::shop_door::held() {
Some(b) => b,
None => {
return Err(match got {
Err(e) => e,
Ok(_) => {
format!("no installer material is kept in {} yet", slot.describe())
}
})
}
},
};
*held = Some(b.clone());
Ok(b)
}
fn control_of(&self, role: &str) -> Option<String> {
let reply = self.vm.latest()?;
if role == "twin" {
return crate::twin_addr(&reply).map(|ip| format!("{ip}:{}", crate::CONTROL_PORT));
}
crate::primary_control(&reply)
}
/// The installer's client to `role`'s log door, from the kept material.
fn client(&self, role: &str) -> Result<nordisk_pairing::installer::Client, String> {
let kept = nordisk_pairing::kept::Kept::parse(&self.bundle()?)?;
let m = kept.material()?;
let link = m
.link
.clone()
.ok_or("the kept material holds no link keys")?;
let (_, path) = nordisk_pairing::kept::paths_for(&kept.plan, &link)?
.into_iter()
.find(|(vm, _)| vm.name == role)
.ok_or_else(|| format!("the kept plan names no {role}"))?;
nordisk_pairing::installer::Client::new(
path,
&m.ca.cert_pem,
&m.me,
Duration::from_secs(120),
)
}
}
impl crate::update::Hands for EstateHands {
fn running(&self, role: &str) -> Result<nordisk_pairing::live::Running, String> {
if role == "front" {
return nordisk_pairing::frontside::FrontProduct::live_running(&self.vm.product);
}
self.client(role)?.live_running()
}
fn stage(
&self,
role: &str,
stage: &nordisk_pairing::live::Stage,
declared: &nordisk_pairing::live::Declared,
) -> Result<String, String> {
if role == "front" {
return nordisk_pairing::frontside::FrontProduct::live_stage_declared(
&self.vm.product,
stage,
declared,
);
}
self.client(role)?.live_stage_declared(stage, declared)
}
fn fetch(&self, url: &str) -> Result<Vec<u8>, String> {
let agent = ureq::Agent::config_builder()
.http_status_as_error(false)
.timeout_global(Some(Duration::from_secs(600)))
.build()
.new_agent();
let mut r = agent
.get(&crate::update::reachable(url))
.call()
.map_err(|e| format!("[update-fetch] {e}"))?;
if !r.status().is_success() {
return Err(format!("[update-fetch] the bucket answered {}", r.status()));
}
r.body_mut()
.with_config()
.limit(nordisk_pairing::live::MAX_STAGE_BYTES as u64)
.read_to_vec()
.map_err(|e| format!("[update-fetch] the body: {e}"))
}
fn keep(&self, role: &str) -> Result<String, String> {
let r = self.door.upgrade(role, false)?;
if !r.upgraded {
return Ok(format!(
"the {role} runs this release already ({})",
r.words
));
}
let until = std::time::Instant::now() + UPGRADE_WITHIN;
loop {
if let Some(at) = self.control_of(role) {
if at
.parse::<std::net::SocketAddr>()
.ok()
.and_then(|a| {
std::net::TcpStream::connect_timeout(&a, Duration::from_secs(3)).ok()
})
.is_some()
{
break;
}
}
if std::time::Instant::now() >= until {
return Err(format!(
"the {role} did not answer within {} min of its keep boot",
UPGRADE_WITHIN.as_secs() / 60
));
}
std::thread::sleep(Duration::from_secs(10));
}
let done = self.door.upgrade(role, true)?;
Ok(format!(
"the {role} was booted from the keep medium and answers again ({})",
done.words
))
}
fn wait(&self, d: Duration) {
std::thread::sleep(d);
}
fn staged_self(&self, release: &str) -> bool {
let p = self.vm.root().join(crate::update::STAGED_SELF_REL);
let was = std::fs::read_to_string(&p).is_ok_and(|r| r.trim() == release.trim());
let _ = std::fs::remove_file(&p);
was
}
fn mark_staged_self(&self, release: &str) -> Result<(), String> {
nordisk_pairing::boxside::write_atomic(
self.vm.root(),
crate::update::STAGED_SELF_REL,
release.as_bytes(),
false,
)
.map_err(|e| e.to_string())
}
fn front_keep(&self) -> String {
crate::update::FRONT_KEEP.into()
}
/// **Boxes left at their open install listener, re-admitted together** (T16: the shared
/// update step asks it first, gunnar.rs 2026-10-25): from the kept bundle, over the
/// installer's link ([`nordisk_pairing::kept::readmit_together`]); the new ledger kept in the
/// primary's sealed slot and held here.
fn readmit(&self, roles: &[&str]) -> Result<String, String> {
let bundle = self.bundle()?;
let reply = self.vm.latest();
let slot = reply
.as_ref()
.and_then(|r| crate::sealed::slot_of(r, &self.handle));
let store = |b: &[u8]| match &slot {
Some(s) => s.put(crate::sealed::MATERIAL_OBJECT, b),
None => Err("no primary elected or no console bearer fed yet".into()),
};
let (after, words) = readmit_kept(
&bundle,
roles,
&nordisk_pairing::kept::readmit_together,
&store,
)?;
*self.kept.lock().map_err(|e| e.to_string())? = Some(after);
Ok(words)
}
}
/// What re-admits the kept boxes (`nordisk_pairing::kept::readmit_together`; a test's fake).
pub type Together<'a> = &'a dyn Fn(
&nordisk_pairing::kept::Kept,
&[&str],
) -> Result<(nordisk_pairing::kept::Kept, String), String>;
/// **The kept `bundle`'s boxes `roles` re-admitted together** (`together`), the bundle with the
/// ledger as it moved rendered and kept (`store`, the primary's sealed slot). A slot that does
/// not take it yet is said in the words: the bundle is held, the re-admission stands.
pub fn readmit_kept(
bundle: &[u8],
roles: &[&str],
together: Together<'_>,
store: &dyn Fn(&[u8]) -> Result<(), String>,
) -> Result<(Vec<u8>, String), String> {
let kept = nordisk_pairing::kept::Kept::parse(bundle)?;
let (after, words) = together(&kept, roles)?;
let rendered = after.render()?;
match store(&rendered) {
Ok(()) => Ok((rendered, words)),
Err(e) => {
tracing::warn!(error = %e, "update: the re-admission's new ledger is held here, not yet kept in the primary's slot");
Ok((
rendered,
format!(
"{words}; the new ledger is held here, not yet kept in the primary's slot: {e}"
),
))
}
}
}
fn spec(kind: &str, generation: i64) -> ChainSpec {
ChainSpec::new(
kind,
kind,
kind,
generation,
"holger-front",
serde_json::json!({}),
)
.with_priority(Priority::Normal)
}
/// One job.
fn run(
kind: &str,
door: &ShopDoor,
vm: &Arc<FrontVm>,
handle: &tokio::runtime::Handle,
watch: &crate::recharge::Watch,
) -> anyhow::Result<serde_json::Value> {
match kind {
KIND_RECHARGE => {
let reply = vm.latest().ok_or_else(|| anyhow::anyhow!("not fed yet"))?;
match watch.look(door, &reply, handle) {
None => Ok(serde_json::json!({ "due": false })),
Some((role, Ok(words))) => {
tracing::warn!(role = %role, "recharge: {words}");
Ok(serde_json::json!({ "role": role, "words": words }))
}
Some((role, Err(e))) => Err(anyhow::anyhow!("the recharge of the {role}: {e}")),
}
}
KIND_MATERIAL => {
let reply = vm.latest().ok_or_else(|| anyhow::anyhow!("not fed yet"))?;
let slot = crate::sealed::slot_of(&reply, handle).ok_or_else(|| {
tracing::warn!(
"material: no primary elected or no console bearer fed yet: not asked"
);
anyhow::anyhow!("no primary elected or no console bearer fed yet")
})?;
// Said at WARN, its refusal too (F9: a failed job says nothing on an estate box).
let said = crate::shop_door::keep_material(door, &slot).map_err(|e| {
tracing::warn!(error = %e, "material: not kept yet; asked again in {} s", LOOK_EVERY.as_secs());
anyhow::anyhow!(e)
})?;
Ok(serde_json::json!({ "words": said }))
}
KIND_UPDATE => {
let hands = EstateHands::new(door.clone(), Arc::clone(vm), handle.clone());
// A failed update pull is said with its reason (Gunnar 7c3feb61: a pull that failed
// every 2 min left nothing to read).
let looked = crate::update::look(door, &hands).inspect_err(|e| tracing::info!(error = %e, "update: the shop did not answer the update pull; asked again at the next look"));
match looked.map_err(|e| anyhow::anyhow!(e))? {
None => Ok(serde_json::json!({ "open": false })),
Some((role, outcome, shop)) => {
Ok(serde_json::json!({ "role": role, "outcome": outcome, "shop": shop }))
}
}
}
other => Err(anyhow::anyhow!("no front job {other:?}")),
}
}
/// The product this front signs its `Fleet.Ask` as when its feed names none (fed before
/// P3-FLEET); the ONE box knows a Holger estate as this product.
pub const PRODUCT: &str = "holger";
/// **The fed shop door, on `Fleet.Ask`** (P3-FLEET, plan §8: the `SoldEstate` door folded into
/// Fleet): the feed's door, signed as the fed product's front, or as Holger's when the feed is
/// from before the fold. `None`: no door fed.
pub fn door_of(
feed: &std::collections::BTreeMap<String, String>,
) -> Option<Result<ShopDoor, String>> {
ShopDoor::from_feed(feed).map(|d| d.map(|d| d.or_over_fleet(PRODUCT)))
}
/// **Start the sold front's duties** once it is fed a shop door (a task waits for the feed).
/// Must run inside a runtime.
pub fn start(vm: Arc<FrontVm>, root: std::path::PathBuf) {
let handle = tokio::runtime::Handle::current();
// A32-14, stays a task: waits for the fed shop door once, then starts the sold front's jera lanes.
handle.clone().spawn(async move {
let door = loop {
match vm.latest().map(|r| r.feed).as_ref().and_then(door_of) {
Some(Ok(d)) => break d,
Some(Err(e)) => tracing::error!(error = %e, "front: the fed shop door is not usable"),
None => {}
}
tokio::time::sleep(Duration::from_secs(10)).await;
};
let sink: JobSink = match jera::jobs::JobStore::open(&root.join(JOBS_REL)) {
Ok(s) => s.sink(),
Err(e) => return tracing::error!(error = %e, "front: the job ledger cannot open: no material kept, no update pulled"),
};
let q = ChainQueue::new(sink);
let (d, v, h) = (door.clone(), Arc::clone(&vm), handle.clone());
let watch = crate::recharge::Watch::default();
// A32-14, stays a task: the sold lane's jera worker.
handle.spawn(Arc::clone(&q).run(move |s| run(&s.kind, &d, &v, &h, &watch)));
tracing::info!(door = %door.door, company = %door.company, product = ?door.fleet_product(), "front: the ONE monetize's door is fed (Fleet.Ask): material kept, updates pulled, a lost box recharged");
let mut generation = 0i64;
loop {
generation += 1;
let _ = q.submit(spec(KIND_MATERIAL, generation));
let _ = q.submit(spec(KIND_UPDATE, generation));
let _ = q.submit(spec(KIND_RECHARGE, generation));
tokio::time::sleep(LOOK_EVERY).await;
}
});
}
#[cfg(test)]
mod tests {
use super::*;
use std::cell::RefCell;
fn bundle() -> Vec<u8> {
let ca = nordisk_pairing::mtls::InstallerCa::mint(
"acme.holger.example",
nordisk_pairing::kept::MATERIAL_VALID,
)
.unwrap();
let plan = serde_json::json!({"estate": "acme.holger.example", "tunnel": "wireguard", "vms": [], "budget_s": 60});
let files = serde_json::json!({"ca.cert.pem": ca.cert_pem, "ca.key.pem": ca.key_pem});
serde_json::to_vec(&serde_json::json!({"v": nordisk_pairing::kept::BUNDLE_V, "files": files, "ledger": "before", "plan": plan})).unwrap()
}
/// ★ T16: **Holger's front re-admits its boxes together from the kept bundle**: the boxes
/// named, the ledger as the re-admission moved it kept in the primary's slot; a slot that
/// does not take it yet is said, and a re-admission that fails is said by name.
#[test]
fn the_kept_boxes_are_readmitted_together_and_the_new_ledger_kept_in_the_primarys_slot() {
let asked = RefCell::new(Vec::<String>::new());
let together = |k: &nordisk_pairing::kept::Kept,
roles: &[&str]|
-> Result<(nordisk_pairing::kept::Kept, String), String> {
asked.borrow_mut().push(roles.join("+"));
let mut after = k.clone();
after.ledger = "moved".into();
Ok((after, format!("{} re-admitted by the front after its upgrade (its data kept); 3 box(es) closed", roles.join(" and "))))
};
let stored = RefCell::new(Vec::<Vec<u8>>::new());
let store = |b: &[u8]| -> Result<(), String> {
stored.borrow_mut().push(b.to_vec());
Ok(())
};
let (after, words) =
readmit_kept(&bundle(), &["twin", "main"], &together, &store).expect("re-admitted");
assert_eq!(
*asked.borrow(),
vec!["twin+main".to_string()],
"ONE run, both boxes"
);
assert_eq!(
nordisk_pairing::kept::Kept::parse(&after).unwrap().ledger,
"moved"
);
assert_eq!(
*stored.borrow(),
vec![after.clone()],
"the new ledger is kept in the primary's slot"
);
assert!(words.contains("twin and main re-admitted"), "{words}");
let refused = |_: &[u8]| -> Result<(), String> { Err("the primary is down".into()) };
let (_, words) = readmit_kept(&bundle(), &["main"], &together, &refused)
.expect("the re-admission stands");
assert!(words.contains("the primary is down"), "{words}");
let fails = |_: &nordisk_pairing::kept::Kept,
_: &[&str]|
-> Result<(nordisk_pairing::kept::Kept, String), String> {
Err("REFUSED [installer-incomplete] main: never answered".into())
};
assert!(readmit_kept(&bundle(), &["main"], &fails, &store)
.unwrap_err()
.contains("[installer-incomplete]"));
assert!(readmit_kept(b"not a bundle", &["main"], &together, &store).is_err());
}
}
#[cfg(test)]
mod fleet_tests {
use super::*;
use crate::shop_door::{FEED_FLEET_INSTANCE, FEED_SHOP_DOOR};
/// ★ P3-FLEET (plan §8): the front's shop door rides `Fleet.Ask` — as the fed product when
/// the shop feeds one, as Holger when the feed is from before the fold — signed as the fed
/// `fleet_instance`'s front, or the company's.
#[test]
fn the_fed_door_rides_fleet_as_holger_or_as_the_fed_product() {
let mut feed = std::collections::BTreeMap::from([
(
FEED_SHOP_DOOR.to_string(),
"door = \"https://monetize.ignalina.dk\"\ncompany = \"acme\"\n".to_string(),
),
("secret.shop_client_key".to_string(), "22".repeat(32)),
]);
assert!(
door_of(&std::collections::BTreeMap::new()).is_none(),
"no door fed"
);
let d = door_of(&feed).unwrap().unwrap();
assert_eq!(
(
d.on_fleet(),
d.fleet_product().as_deref(),
d.fleet_instance().as_deref()
),
(true, Some(PRODUCT), Some("acme"))
);
feed.insert(FEED_SHOP_DOOR.to_string(), "door = \"https://monetize.ignalina.dk\"\ncompany = \"holgerrs\"\nproduct = \"holger-site\"\n".to_string());
feed.insert(FEED_FLEET_INSTANCE.to_string(), "holger-rs".to_string());
let d = door_of(&feed).unwrap().unwrap();
assert_eq!(
(d.fleet_product().as_deref(), d.fleet_instance().as_deref()),
(Some("holger-site"), Some("holger-rs"))
);
}
}