crates/monetize-server/src/estate_order/update.rs
//! **The ONE box updates the estates and sites it laid** (owner 2026-10-14: environments are
//! set up by the application, not by a kit — point 3, "upgrade + live update of sites driven by
//! the ONE box").
//!
//! Monetize never dials an estate. The operator OPENS an update here (`EstateOrder.UpdateEstate`,
//! the console's "Update to release X", `monetize estate update <estate>`); the estate's FRONT —
//! which holds the kept installer material and reaches its twin's and main's log doors over the
//! estate's own WireGuard — PULLS it (`SoldEstate.UpdateTake`, signed by its client key) and
//! does it, one box at a time, reporting each (`SoldEstate.UpdateReport`):
//!
//! ```text
//! operator ──UpdateEstate{estate, release}──▶ run.update (twin, main, front: waiting)
//! front ──UpdateTake──▶ {release, live?, next role, the live records + presigned archives}
//! twin, then main: GET /live/running on the box's log door
//! same BASE, live on ──▶ POST /live/stage (the release's live app) → confirmed
//! BASE moved / live off ──▶ SoldEstate.Upgrade (the keep medium) → re-admitted
//! ──UpdateReport{role, done | failed}──▶ the next role; a failure STOPS it (ALERT)
//! front: same BASE, live on ──▶ stages ITSELF (re-exec) → reports done once it runs it
//! else ──UpdateReport{front, needs-keep}──▶ THIS box keep-boots the front from the
//! release's `front-keep` medium over the cloud API (its /data kept) → done
//! ```
//!
//! **A front whose primary keeps no installer material** (F9, 2026-10-21: test3-holger-rs's r6
//! front never took it): while the custody still holds the estate's pairing material and every
//! box is PAIRED, the ONE box takes the main's and the twin's steps itself ([`custody_step`],
//! a jera job on the tick) — the box's live door over the installer link with the custody's
//! material, the live app staged and confirmed, or its keep medium — and hands the front only
//! its own step. A front that reports its primary keeps none hands that box over the same way.
//!
//! **Live off per estate** ([`EstateRun::live_update_off`], `EstateOrder.SetLiveUpdate`; a
//! military customer): the update is opened `live = false` and every box goes through its keep
//! medium; the box itself refuses a live payload too when its BASE says `korp.live_update=off`.
use std::collections::BTreeMap;
use std::sync::Arc;
use monetize::sign::EstateCall;
use super::sold::SoldRefusal;
use super::{phase, release_id, EstateOrderConfig, EstateRun, Transition};
use crate::state::AppState;
/// `SoldEstate.UpdateTake`'s signed call name.
pub const CALL_UPDATE_TAKE: &str = "update-take";
/// `SoldEstate.UpdateReport`'s signed call name; the outcome and words ride in `carries`.
pub const CALL_UPDATE_REPORT: &str = "update-report";
/// **The order the boxes are updated in**: the twin first (the primary keeps serving), then main,
/// the front last (it drives the others).
pub const ORDER: [&str; 3] = ["twin", "main", "front"];
/// The medium kind the FRONT is upgraded in place from (`install_then=poweroff`, the release's).
pub const FRONT_KEEP_MEDIUM: &str = "front-keep";
/// The media-object kind suffix of a release's live-app archive (`gunnar-live`, `front-live`):
/// the korp medium's `<medium>.live-app.znippy`, uploaded with the release.
pub const LIVE_SUFFIX: &str = "-live";
/// A step not reported for this long is told to the operator (once per step): a whole keep
/// boot ([`monetize_cloud::lay::KEEP_BOOT_WITHIN`], 2 h) and an hour more (audit 7).
pub const STALL_MS: u64 = monetize_cloud::lay::KEEP_BOOT_WITHIN.as_millis() as u64 + 3600 * 1000;
/// ★ A32-7: a stall still unreported this long after it was last told is told AGAIN (an
/// ALERT and its CODE RED once more), never only once.
pub const STALL_RETELL_MS: u64 = 6 * 3600 * 1000;
/// ★ A32-7: **an update is finished within a day** of its opening, or it is FAILED by itself
/// (`the boxes after it are NOT touched`, as any failure), so a front that never reports — or a
/// keep boot whose outcome was lost — never holds the estate's next update off for good.
pub const UPDATE_DEADLINE_MS: u64 = 24 * 3600 * 1000;
/// How many finished updates a run keeps.
pub const KEEP_FINISHED: usize = 20;
/// How often a front's look is written down ([`EstateRun::update_taken_at_ms`]).
pub const TAKEN_KEPT_EVERY_MS: u64 = 3600 * 1000;
/// The ALERT kind every failure and stall is told under.
pub const ALERT_KIND: &str = "estate-orders.update";
/// The update's own states.
pub mod state {
pub const RUNNING: &str = "running";
/// The front asked to be upgraded from its keep medium: this box does it on its tick.
pub const FRONT_KEEP: &str = "front-keep";
pub const DONE: &str = "done";
pub const FAILED: &str = "failed";
/// ★ A32-7: the operator cancelled it (`estate update-cancel`): the boxes done stay done.
pub const CANCELLED: &str = "cancelled";
}
/// One box's step's states.
pub mod step {
pub const WAITING: &str = "waiting";
pub const DONE: &str = "done";
pub const FAILED: &str = "failed";
}
/// **A release's live record for one medium kind** (`[estate_orders.live_apps.<kind>]`): korp's
/// `<medium>.live.toml` as the release's build wrote it — the BASE an installed box of the medium
/// runs, the app's pin (`app.blake3`), its length and its signature (hex, under the root the boxes
/// enrolled). The archive itself is the media object `<kind>-live`.
///
/// Lane B5: what the release DECLARES beside its app rides with it — the release declaration
/// korp composed beside the `.iso` (`<medium>.live-release.bundle` / `.sig`, hex here) and the
/// release's own `korp.env.*` tokens (the record's `env`). A record written before B5 carries
/// neither, and the box is told nothing.
#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LiveRecord {
pub base: String,
pub app: String,
pub app_len: u64,
pub sig: String,
/// The release declaration's bundle, hex (empty: none).
#[serde(default, skip_serializing_if = "String::is_empty")]
pub release: String,
/// Its signature over the app's root, hex (empty: unsigned).
#[serde(default, skip_serializing_if = "String::is_empty")]
pub release_sig: String,
/// The release's `korp.env.*` tokens (empty: none declared).
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub env: Vec<String>,
}
/// One box's step of an update.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct Step {
/// `twin` | `main` | `front`.
pub role: String,
/// [`step`].
pub state: String,
/// How it was done (`live`, `keep`, `nothing` — it ran the release already), once done.
#[serde(default)]
pub how: String,
#[serde(default)]
pub words: String,
#[serde(default)]
pub at_ms: u64,
/// **The ONE box takes this box's step itself, from the custody's copy of the installer
/// material** (F9): set when the front reported that its primary keeps no material; an
/// estate whose front never took it is driven so without the flag ([`custody_drives`]).
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub custody: bool,
}
/// **One update of an estate to a release**, as the run keeps it.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct EstateUpdate {
/// The release main and twin are updated to ([`release_id`] of `gunnar`).
pub release: String,
/// The release the front is updated to ([`release_id`] of `front`).
#[serde(default)]
pub front_release: String,
/// May a box be updated live (its app slot), or only from the keep medium?
pub live: bool,
/// Who opened it (the operator's principal or `operator`).
pub by: String,
pub opened_at_ms: u64,
pub steps: Vec<Step>,
/// [`state`].
pub state: String,
#[serde(default)]
pub words: String,
#[serde(default)]
pub closed_at_ms: u64,
/// When the stall of the step now open was last told.
#[serde(default)]
pub stall_told_at_ms: u64,
/// ★ A32-7: how often the stall of the step now open was told (again every
/// [`STALL_RETELL_MS`] while it lasts).
#[serde(default, skip_serializing_if = "is_zero")]
pub stalls_told: u32,
/// ★ A32-20: when the shop's keep boot of the front BEGAN (claimed under the run's pen before
/// the cloud is asked): a second job finds it claimed and boots nothing; an outcome that could
/// not be written leaves it set, and the stall, the deadline or a cancel ends the update.
#[serde(default, skip_serializing_if = "is_zero_ms")]
pub front_keep_began_at_ms: u64,
/// **The release's media, as it was when the update opened** (2026-10-17): its media ids and
/// objects (app, front, keep, front keep, live archives) and its live records. A newer
/// release fed while the update runs never leaks into it: what is handed out, keep-booted
/// and upgraded is THIS release's ([`as_opened`]). Empty: an update opened before pinning.
#[serde(default, skip_serializing_if = "UpdateMedia::is_empty")]
pub media: UpdateMedia,
}
/// The media an update was opened with ([`EstateUpdate::media`]).
#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(default)]
pub struct UpdateMedia {
/// Every media key the update reads (a kind, or `<kind>-live`).
pub kinds: Vec<String>,
pub media: BTreeMap<String, String>,
pub media_objects: BTreeMap<String, monetize_cloud::lay::MediaObject>,
pub live_apps: BTreeMap<String, LiveRecord>,
}
impl UpdateMedia {
pub fn is_empty(&self) -> bool {
self.kinds.is_empty()
}
}
/// Pure: **pin the media an update reads** from `cfg` (the release sold as it opens).
pub fn pin(cfg: &EstateOrderConfig, plan: &super::lifecycle::UpdatePlan) -> UpdateMedia {
let mut m = UpdateMedia::default();
for kind in [&plan.app_medium, &plan.front_medium, &plan.keep_medium, &plan.front_keep_medium] {
if kind.is_empty() {
continue;
}
for key in [kind.clone(), format!("{kind}{LIVE_SUFFIX}")] {
if !m.kinds.contains(&key) {
m.kinds.push(key);
}
}
if let Some(r) = cfg.live_apps.get(kind.as_str()) {
m.live_apps.insert(kind.clone(), r.clone());
}
}
for key in &m.kinds {
if let Some(v) = cfg.media.get(key) {
m.media.insert(key.clone(), v.clone());
}
if let Some(o) = cfg.media_objects.get(key) {
m.media_objects.insert(key.clone(), o.clone());
}
}
m
}
/// Pure: **`terms` as the open update `u` reads them** — its pinned media over whatever is fed
/// now; an update opened before pinning reads `terms` as they are.
pub fn as_opened(terms: &EstateOrderConfig, u: &EstateUpdate) -> EstateOrderConfig {
let mut c = terms.clone();
let m = &u.media;
if m.is_empty() {
return c;
}
for key in &m.kinds {
c.media.remove(key);
c.media_objects.remove(key);
c.live_apps.remove(key);
if let Some(v) = m.media.get(key) {
c.media.insert(key.clone(), v.clone());
}
if let Some(o) = m.media_objects.get(key) {
c.media_objects.insert(key.clone(), o.clone());
}
if let Some(r) = m.live_apps.get(key) {
c.live_apps.insert(key.clone(), r.clone());
}
}
c
}
impl EstateUpdate {
pub fn open(release: &str, front_release: &str, live: bool, by: &str, now: u64) -> EstateUpdate {
EstateUpdate::open_in(&ORDER.map(String::from), release, front_release, live, by, now)
}
/// [`EstateUpdate::open`] over the boxes `order` names (a product's
/// [`super::lifecycle::UpdatePlan::order`]).
pub fn open_in(order: &[String], release: &str, front_release: &str, live: bool, by: &str, now: u64) -> EstateUpdate {
EstateUpdate {
release: release.to_string(),
front_release: front_release.to_string(),
live,
by: by.to_string(),
opened_at_ms: now,
steps: order.iter().map(|r| Step { role: r.clone(), state: step::WAITING.into(), how: String::new(), words: String::new(), at_ms: now, custody: false }).collect(),
state: state::RUNNING.into(),
words: String::new(),
closed_at_ms: 0,
stall_told_at_ms: 0,
stalls_told: 0,
front_keep_began_at_ms: 0,
media: UpdateMedia::default(),
}
}
/// Still to be finished (running, or the front's keep boot).
pub fn is_open(&self) -> bool {
matches!(self.state.as_str(), state::RUNNING | state::FRONT_KEEP)
}
/// The box whose turn it is, while the update runs.
pub fn next(&self) -> Option<&Step> {
if !self.is_open() {
return None;
}
self.steps.iter().find(|s| s.state == step::WAITING)
}
/// When the step now open started waiting (the update's opening, or the step before it done).
fn waiting_since(&self) -> u64 {
self.steps.iter().filter(|s| s.state == step::DONE).map(|s| s.at_ms).max().unwrap_or(self.opened_at_ms).max(self.opened_at_ms)
}
/// One line for a console or the CLI.
pub fn line(&self) -> String {
let steps: Vec<String> = self.steps.iter().map(|s| format!("{} {}{}", s.role, s.state, if s.how.is_empty() { String::new() } else { format!(" ({})", s.how) })).collect();
format!("update to {} ({}): {} — {}{}", short(&self.release), if self.live { "live where it can" } else { "keep media only" }, self.state, steps.join(", "), if self.words.is_empty() { String::new() } else { format!(" — {}", self.words) })
}
}
/// What a front reports for its box.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Outcome {
/// The box runs the release now: `live`, `keep` or `nothing` (it did already).
Done(String),
/// The box could not be updated; the words say why. The update stops.
Failed,
/// The FRONT's BASE moved (or live is off): this box keep-boots it.
NeedsKeep,
}
impl Outcome {
/// The outcome word as signed (`done:live`, `done:keep`, `done:nothing`, `failed`, `needs-keep`).
pub fn parse(word: &str) -> Result<Outcome, String> {
match word.trim() {
"failed" => Ok(Outcome::Failed),
"needs-keep" => Ok(Outcome::NeedsKeep),
w => match w.strip_prefix("done:") {
Some(how @ ("live" | "keep" | "nothing")) => Ok(Outcome::Done(how.to_string())),
_ => Err(format!("{w:?} is not an update outcome (done:live | done:keep | done:nothing | failed | needs-keep)")),
},
}
}
pub fn word(&self) -> String {
match self {
Outcome::Done(how) => format!("done:{how}"),
Outcome::Failed => "failed".into(),
Outcome::NeedsKeep => "needs-keep".into(),
}
}
}
/// What a report did to the update.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Effect {
/// The next box's turn.
Advanced,
/// Every box runs the release.
Finished,
/// Stopped: the operator is told this.
Failed(String),
/// The front waits for this box to keep-boot it.
FrontKeep,
}
/// The `carries` of a signed report: the outcome word, a newline, the words.
pub fn report_carries(outcome: &Outcome, words: &str) -> String {
format!("{}\n{}", outcome.word(), words)
}
/// Pure: **open an update** of `run` to the release `cfg` sells now. `wanted`: the release the
/// operator's button named (empty: whatever is sold now) — a stale button is refused by name.
pub fn begin(run: &EstateRun, cfg: &EstateOrderConfig, wanted: &str, by: &str, now: u64) -> Result<EstateUpdate, String> {
let host = cfg.host_of(&run.company);
admits(run, &host)?;
let laid = run.laid.as_ref().ok_or_else(|| format!("{host} has nothing laid"))?;
let plan = super::lifecycle::of(cfg)?.update_plan();
for role in &plan.order {
if laid.machine(role).is_none() {
return Err(format!("{host} has no {role} laid: an update runs over {}", and_words(&plan.order)));
}
}
// Every laid member of each role (the Lake's `be`, `be-2`, …), the order's roles first.
let laid_roles: Vec<&str> = laid.machines.iter().map(|m| m.role.as_str()).collect();
let order = super::lifecycle::members_in(&plan.order, &laid_roles);
if let Some(u) = run.update.as_ref().filter(|u| u.is_open()) {
return Err(format!("{host} is already being updated: {}", u.line()));
}
let app = plan.app_medium.as_str();
let release = release_id(cfg, app);
if release.is_empty() {
return Err(format!("No release is published for {host} yet: there is nothing to update to."));
}
let wanted = wanted.trim();
if !wanted.is_empty() && !release.starts_with(wanted) {
return Err(format!("{host} is sold release {} now, not {wanted}: the page was stale — look again", short(&release)));
}
let front_release = release_id(cfg, &plan.front_medium);
let runs = |role: &str, rel: &str| !rel.is_empty() && run.release_of.get(role).map(String::as_str) == Some(rel);
let front = plan.front_role.as_str();
if order.iter().filter(|r| *r != front).all(|r| runs(r, &release)) && (front_release.is_empty() || runs(front, &front_release)) {
return Err(format!("{host} is up to date: it runs release {} already.", short(&release)));
}
// A box with no live slot (a guest that declares none) is updated from its keep medium only.
let live = plan.live_slot && !run.live_update_off;
let has_live = cfg.live_apps.contains_key(app) && cfg.media_objects.contains_key(&format!("{app}{LIVE_SUFFIX}"));
let keep = plan.keep_medium.as_str();
let has_keep = cfg.media.contains_key(keep) || cfg.media_objects.contains_key(keep);
if !has_keep && !(live && has_live) {
let _ = keep;
return Err(format!(
"No newer release with update media is published yet: release {} carries {} for {host}.",
short(&release),
if live { "neither an in-place upgrade nor a live update" } else { "no in-place upgrade, and live updates are off" }
));
}
let mut u = EstateUpdate::open_in(&order, &release, &front_release, live, by, now);
u.media = pin(cfg, &plan);
Ok(u)
}
/// Pure: **may `run` take an estate update in its phase?** (F7, 2026-10-20) A delivered estate
/// (`captured`, `grace`, `suspended`, `deleting`) does; so does a `paired` one — its installer
/// said PASS (every box PAIRED), or a re-check adopted it ([`super::ADOPTED_ON_RECHECK`]) — whose
/// main may run a release that can never report (test3-holger-rs: r6, its report-when-fed fix
/// in r10). The update only brings the boxes to the current release, their data kept: it is no
/// delivery and sends no ready mail; the installed path (the main's report, the sale check,
/// captured, ready) goes on after it. An install still running (`laying`, `installing`) and
/// every other phase are refused by name. Gunnar's and Holger's alike: one lifecycle.
pub fn admits(run: &EstateRun, host: &str) -> Result<(), String> {
if phase::is_term(&run.phase) || run.phase == phase::PAIRED {
return Ok(());
}
if matches!(run.phase.as_str(), phase::LAYING | phase::INSTALLING) {
return Err(format!("{host} is not delivered ({}): its install is still running, and nothing is updated before every box is PAIRED", run.phase));
}
Err(format!("{host} is not delivered ({}): nothing is updated", run.phase))
}
/// Pure: **may the shop keep-boot `run`'s `role` now?** Only while an update the operator
/// opened runs and it is that very box's turn — the rule [`take`] hands the front its box by
/// (F-U-02, A17: never on the front's own daily look), and only on an estate that takes an
/// update ([`admits`]). Refused by name otherwise.
pub fn upgrade_admits(run: &EstateRun, host: &str, role: &str) -> Result<(), String> {
admits(run, host)?;
let Some(u) = run.update.as_ref().filter(|u| u.state == state::RUNNING) else {
return Err(format!("no update is open for {host}'s {role}: a box is upgraded only through an update the operator opens (`estate update`)"));
};
match u.next() {
Some(s) if s.role == role => Ok(()),
Some(s) => Err(format!("no update is open for {host}'s {role} now: it is the {}'s turn", s.role)),
None => Err(format!("no update is open for {host}'s {role}: no box waits")),
}
}
/// `a, b and c`.
fn and_words(v: &[String]) -> String {
match v.split_last() {
Some((last, rest)) if !rest.is_empty() => format!("{} and {last}", rest.join(", ")),
Some((last, _)) => last.clone(),
None => String::new(),
}
}
/// Pure: **a front's report on its box**, applied to the update. Refused by name: no update
/// running, another box than the one whose turn it is, a live update reported where live is off,
/// `needs-keep` from another box than the front (main and twin go through `Upgrade`).
pub fn apply(u: &mut EstateUpdate, role: &str, outcome: &Outcome, words: &str, now: u64) -> Result<Effect, String> {
if u.state != state::RUNNING {
return Err(format!("no update is running ({}): nothing to report", u.state));
}
let next = u.next().map(|s| s.role.clone()).ok_or("no box waits")?;
if next != role {
return Err(format!("it is the {next}'s turn, not the {role}'s: the boxes are updated {}", u.steps.iter().map(|s| s.role.as_str()).collect::<Vec<_>>().join(", ")));
}
if matches!(outcome, Outcome::Done(how) if how == "live") && !u.live {
return Err(format!("live update is OFF for this estate: the {role} is updated from its keep medium only"));
}
// The box that drives the update is updated last (Gunnar's front).
if *outcome == Outcome::NeedsKeep && u.steps.last().map(|s| s.role.as_str()) != Some(role) {
return Err(format!("the {role} is upgraded from its keep medium through `Upgrade`, by the front"));
}
let s = u.steps.iter_mut().find(|s| s.role == role).ok_or("no such step")?;
s.at_ms = now;
s.words = words.to_string();
match outcome {
Outcome::Done(how) => {
s.state = step::DONE.into();
s.how = how.clone();
u.stall_told_at_ms = 0;
u.stalls_told = 0;
if u.next().is_none() {
u.state = state::DONE.into();
u.closed_at_ms = now;
u.words = format!("every box runs release {}", short(&u.release));
Ok(Effect::Finished)
} else {
Ok(Effect::Advanced)
}
}
Outcome::Failed => {
s.state = step::FAILED.into();
u.state = state::FAILED.into();
u.closed_at_ms = now;
u.words = format!("the {role} was not updated: {words} — the boxes after it are NOT touched");
Ok(Effect::Failed(u.words.clone()))
}
Outcome::NeedsKeep => {
u.state = state::FRONT_KEEP.into();
u.words = format!("the front is upgraded from its keep medium by the shop: {words}");
Ok(Effect::FrontKeep)
}
}
}
/// Pure: the update closed, kept with the run's finished ones (the newest [`KEEP_FINISHED`]).
fn close(run: &mut EstateRun) {
if let Some(u) = run.update.take() {
run.updates.push(u);
let over = run.updates.len().saturating_sub(KEEP_FINISHED);
run.updates.drain(..over);
}
}
/// A release identity, short.
pub fn short(release: &str) -> &str {
if release.is_empty() { "(unnamed)" } else { &release[..release.len().min(12)] }
}
fn alert(run: &EstateRun, words: &str) {
eprintln!("monetize-server: ALERT {ALERT_KIND} {}: {words}", run.reference);
}
/// The run of `estate` (its company, or a site's label) under its pen, changed by `f` and saved.
fn with_run<T>(state: &AppState, cfg: &EstateOrderConfig, estate: &str, f: impl FnOnce(&mut EstateRun, &EstateOrderConfig) -> Result<T, SoldRefusal>) -> Result<T, SoldRefusal> {
let company = state.estates.get(estate).map(|r| r.company).ok_or_else(|| SoldRefusal::Refused(format!("{estate} is no estate laid here")))?;
let Ok(_pen) = state.estates.pen_for_a_call(cfg, &company) else {
return Err(SoldRefusal::Unavailable(format!("{} is being worked on; ask again", cfg.host_of(&company))));
};
let mut run = state.estates.get(&company).ok_or_else(|| SoldRefusal::Refused(format!("{estate} is no estate laid here")))?;
let site_terms = super::terms_of(state, cfg, &run).unwrap_or_else(|_| cfg.clone());
let out = f(&mut run, &site_terms)?;
super::save(state, &run).map_err(|e| SoldRefusal::Unavailable(format!("the run could not be written: {e}")))?;
Ok(out)
}
// ── the operator's verbs ─────────────────────────────────────────────────────
/// **`EstateOrder.UpdateEstate`** (the operator's alone): open an update of `estate` (a sold
/// estate's company or an operator site's label) to the release sold now. Answers it.
/// **The laid estate's firewall, written again before an update's first step** (owner
/// 2026-10-18: self-healing — a front laid before the WireGuard-answers rule could never reach
/// its twin's log door). Every machine's rules as the cloud lays them now; a cloud that cannot
/// is refused by name and the update is not opened.
fn rewrite_firewall(state: &AppState, cfg: &EstateOrderConfig, estate: &str) -> Result<(), SoldRefusal> {
let run = state.estates.get(estate).ok_or_else(|| SoldRefusal::Refused(format!("{estate} is no estate laid here")))?;
let terms = super::terms_of(state, cfg, &run).unwrap_or_else(|_| cfg.clone());
let host = terms.host_of(&run.company);
let laid = run.laid.as_ref().ok_or_else(|| SoldRefusal::Refused(format!("{host} has nothing laid")))?;
let cloud = state.registry.find_cloud(&terms.cloud).ok_or_else(|| SoldRefusal::Unavailable(format!("the shop's cloud {:?} is not linked", terms.cloud)))?;
// DEV-NARRATE as the site says NOW (lane N2): turned on, the update opens udp/51822 from
// the ONE box; turned off, it closes it — whatever the laid record kept. The ports each role
// opens to anyone as the product declares them NOW (OX-WGDS5): a front laid before its
// product declared one opens it here, on the normal update path.
let laid = &monetize_cloud::lay::LaidEstate { narrate_from: terms.narrate_source().to_string(), ..super::lifecycle::shaped_now(&terms, laid) };
monetize_cloud::Cloud::firewall(&cloud, laid).map_err(|e| SoldRefusal::Unavailable(format!("{host}'s firewall could not be written again before the update ({e}): nothing was opened — update the plugins, then ask again")))?;
eprintln!("monetize-server/update: {}: the estate's firewall written again before the update", run.reference);
Ok(())
}
pub fn start(state: &AppState, estate: &str, wanted: &str, by: &str) -> Result<EstateUpdate, SoldRefusal> {
let cfg = state.estates.cfg.as_ref().ok_or_else(|| SoldRefusal::Refused("this monetize lays no estates".into()))?;
admits_now(state, cfg, estate)?;
// A front that never asked for an update carries no taker: it could never pull this one —
// the front goes first (the console's one button, owner 2026-10-17).
if let Some(run) = state.estates.get(estate).filter(|r| r.update_taken_at_ms == 0) {
let terms = super::terms_of(state, cfg, &run).unwrap_or_else(|_| cfg.clone());
let keep = front_keep_of(&terms);
if terms.media.contains_key(&keep) || terms.media_objects.contains_key(&keep) {
return start_front_first(state, estate, wanted, by);
}
}
rewrite_firewall(state, cfg, estate)?;
let now = state.now();
with_run(state, cfg, estate, |run, terms| {
let u = begin(run, terms, wanted, by, now).map_err(SoldRefusal::Refused)?;
let words = format!("{} opened an {}", by, u.line());
run.history.push(Transition { phase: String::new(), at_ms: now, note: words.clone() });
eprintln!("monetize-server/update: {}: {words}", run.reference);
run.update = Some(u.clone());
Ok(u)
})
}
/// [`admits`] on the run as it stands, before the firewall is written or the run's pen asked
/// (an install still running holds it): refused by name.
fn admits_now(state: &AppState, cfg: &EstateOrderConfig, estate: &str) -> Result<(), SoldRefusal> {
match state.estates.get(estate) {
Some(run) => {
let terms = super::terms_of(state, cfg, &run).unwrap_or_else(|_| cfg.clone());
admits(&run, &terms.host_of(&run.company)).map_err(SoldRefusal::Refused)
}
None => Ok(()),
}
}
/// The medium kind `terms`' product keep-boots its front from (its lifecycle's; Gunnar's and
/// Holger's: [`FRONT_KEEP_MEDIUM`]).
fn front_keep_of(terms: &EstateOrderConfig) -> String {
super::lifecycle::of(terms).map(|l| l.update_plan().front_keep_medium).unwrap_or_else(|_| FRONT_KEEP_MEDIUM.to_string())
}
/// The role that drives `terms`' product's update (its lifecycle's; Gunnar's and Holger's: `front`).
fn front_role_of(terms: &EstateOrderConfig) -> String {
super::lifecycle::of(terms).map(|l| l.update_plan().front_role).unwrap_or_else(|_| "front".to_string())
}
/// **`EstateOrder.UpdateEstate` with `front_first`** (owner 2026-10-17): the same update, but the
/// FRONT goes first — this box keep-boots it from the release's `front-keep` on its next tick (its
/// `/data` kept) — and the new front then takes twin and main in their turn. For an estate whose
/// front runs a release with no update taker yet: it could never pull an update itself.
pub fn start_front_first(state: &AppState, estate: &str, wanted: &str, by: &str) -> Result<EstateUpdate, SoldRefusal> {
let cfg = state.estates.cfg.as_ref().ok_or_else(|| SoldRefusal::Refused("this monetize lays no estates".into()))?;
admits_now(state, cfg, estate)?;
rewrite_firewall(state, cfg, estate)?;
let now = state.now();
with_run(state, cfg, estate, |run, terms| {
let mut u = begin(run, terms, wanted, by, now).map_err(SoldRefusal::Refused)?;
let keep = front_keep_of(terms);
let has_front_keep = terms.media.contains_key(&keep) || terms.media_objects.contains_key(&keep);
if !has_front_keep {
return Err(SoldRefusal::Refused(format!("release {} names no `{keep}` medium: the front cannot go first", short(&u.release))));
}
u.steps.rotate_right(1);
u.state = state::FRONT_KEEP.into();
u.words = "front first: the shop keep-boots the front, then the new front takes twin and main".into();
let words = format!("{} opened an {}", by, u.line());
run.history.push(Transition { phase: String::new(), at_ms: now, note: words.clone() });
eprintln!("monetize-server/update: {}: {words}", run.reference);
run.update = Some(u.clone());
Ok(u)
})
}
/// **`EstateOrder.SetLiveUpdate`** (the operator's alone): live updates on or OFF for `estate`
/// (a military customer: off — every box then goes through its keep medium). An update already
/// running keeps what it was opened with.
pub fn set_live(state: &AppState, estate: &str, on: bool, by: &str) -> Result<String, SoldRefusal> {
let cfg = state.estates.cfg.as_ref().ok_or_else(|| SoldRefusal::Refused("this monetize lays no estates".into()))?;
let now = state.now();
with_run(state, cfg, estate, |run, terms| {
run.live_update_off = !on;
let words = format!("{by}: live updates {} for {}", if on { "ON" } else { "OFF (keep media only)" }, terms.host_of(&run.company));
run.history.push(Transition { phase: String::new(), at_ms: now, note: words.clone() });
Ok(words)
})
}
/// What [`status`] answers.
#[derive(Clone, Debug)]
pub struct Status {
/// What the Sites page offers (2026-10-17): up to date, the button or why not, the release's
/// name, each box's line.
pub offer: Offer,
pub host: String,
/// The release sold now (what an update goes to).
pub release: String,
pub release_of: BTreeMap<String, String>,
pub live_on: bool,
pub open: Option<EstateUpdate>,
pub last: Option<EstateUpdate>,
}
/// **Where `estate`'s update stands** (the operator's alone).
pub fn status(state: &AppState, estate: &str) -> Result<Status, SoldRefusal> {
let cfg = state.estates.cfg.as_ref().ok_or_else(|| SoldRefusal::Refused("this monetize lays no estates".into()))?;
let run = state.estates.get(estate).ok_or_else(|| SoldRefusal::Refused(format!("{estate} is no estate laid here")))?;
let terms = super::terms_of(state, cfg, &run).unwrap_or_else(|_| cfg.clone());
Ok(Status {
offer: offer(&run, &terms),
host: terms.host_of(&run.company),
release: super::lifecycle::of(&terms).map(|l| release_id(&terms, &l.update_plan().app_medium)).unwrap_or_default(),
release_of: run.release_of.clone(),
live_on: !run.live_update_off && super::lifecycle::of(&terms).is_ok_and(|l| l.update_plan().live_slot),
open: run.update.clone(),
last: run.updates.last().cloned(),
})
}
/// **What a release IS, for a person** (`[estate_orders.release_notes."<id>"]`).
#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(default)]
pub struct ReleaseNote {
/// Its label (`r5.4`).
pub label: String,
/// When it was built (`2026-10-17`).
pub built: String,
/// repository → the commit it carries (`gunnar` → `aa14357b…`).
pub carries: BTreeMap<String, String>,
}
/// One box as the Sites page shows it.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct BoxLine {
pub role: String,
/// `current` | `behind` | `not-reported`
pub state: String,
pub words: String,
}
/// **What the Sites page offers for an estate's update** — every fact monetize knows before
/// the press: whether it is up to date, whether the button is offered (a NEWER release that
/// carries a way to update every box) and if not why, in plain words; the release's name; each
/// box's line.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Offer {
pub up_to_date: bool,
pub offered: bool,
pub why_not: String,
pub release_name: String,
pub boxes: Vec<BoxLine>,
}
/// Pure: **a release's name for a person** — its note's label, date and the commits it carries;
/// empty when the kit wrote none (the page then shows the short id).
pub fn release_name(cfg: &EstateOrderConfig, release: &str) -> String {
let Some(n) = cfg.release_notes.get(release) else { return String::new() };
let mut parts: Vec<String> = [n.label.trim(), n.built.trim()].into_iter().filter(|p| !p.is_empty()).map(str::to_string).collect();
parts.extend(n.carries.iter().map(|(repo, commit)| format!("{repo} {}", &commit[..commit.len().min(7)])));
parts.join(" · ")
}
/// Pure: [`Offer`] for `run` under `cfg` (its site's terms).
pub fn offer(run: &EstateRun, cfg: &EstateOrderConfig) -> Offer {
let Ok(life) = super::lifecycle::of(cfg) else { return Offer { why_not: "This box sells no estate of this kind: there is nothing to update.".into(), ..Offer::default() } };
let plan = life.update_plan();
let release = release_id(cfg, &plan.app_medium);
let boxes: Vec<BoxLine> = plan
.order
.iter()
.map(|role| {
let want = release_id(cfg, &life.medium_kind(role));
let runs = run.release_of.get(role).cloned().unwrap_or_default();
if runs.is_empty() {
let words = if *role == plan.front_role && run.update_taken_at_ms == 0 {
format!("not reported yet — the {role} has no update taker yet; the update starts with the {role}")
} else {
"not reported yet".to_string()
};
BoxLine { role: role.clone(), state: "not-reported".into(), words }
} else if !want.is_empty() && runs == want {
BoxLine { role: role.clone(), state: "current".into(), words: format!("current · release {}", short(&runs)) }
} else {
BoxLine { role: role.clone(), state: "behind".into(), words: format!("runs release {} · release {} is published", short(&runs), short(&want)) }
}
})
.collect();
let up_to_date = !release.is_empty() && boxes.iter().all(|b| b.state == "current");
let (offered, why_not) = if up_to_date {
(false, String::new())
} else {
match begin(run, cfg, "", "", 0) {
Ok(_) => (true, String::new()),
Err(w) => (false, w),
}
};
Offer { up_to_date, offered, why_not, release_name: release_name(cfg, &release), boxes }
}
// ── the front's calls ────────────────────────────────────────────────────────
/// One kind's live app, as a front is handed it: its record and the archive's (presigned) url.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct LiveApp {
pub record: LiveRecord,
pub url: String,
pub sha256: String,
}
/// [`take`]'s answer.
#[derive(Clone, Debug, Default)]
pub struct TakeAnswer {
/// An update runs and a box of this estate is to be updated now.
pub open: bool,
pub release: String,
pub live: bool,
/// The box whose turn it is.
pub role: String,
/// The live app the app boxes run (the lifecycle's `app_medium`: Gunnar's main and twin run
/// `gunnar`), when the release carries one and live is on.
pub app: Option<LiveApp>,
/// The front's (`front`).
pub front: Option<LiveApp>,
pub words: String,
}
/// The live app of medium `kind`, its archive presigned.
fn live_app(cfg: &EstateOrderConfig, objects: &BTreeMap<String, monetize_cloud::lay::MediaObject>, kind: &str) -> Option<LiveApp> {
let record = cfg.live_apps.get(kind)?.clone();
let o = objects.get(&format!("{kind}{LIVE_SUFFIX}")).filter(|o| !o.url.trim().is_empty())?;
Some(LiveApp { record, url: o.url.clone(), sha256: o.sha256.clone() })
}
/// **`SoldEstate.UpdateTake`**: the calling estate's open update — the box whose turn it is and,
/// while live is on, the release's live apps with their archives presigned. Nothing open: `open
/// = false`. Signed by the estate's client key.
pub fn take(state: &AppState, company: &str, at_ms: i64, signature: &[u8]) -> Result<TakeAnswer, SoldRefusal> {
take_as(state, company, super::sold::Caller::Signed { at_ms, signature })
}
/// [`take`], asked by `by` (the old door's signed call, or the front's `Fleet.Ask{update_take}`).
pub fn take_as(state: &AppState, company: &str, by: super::sold::Caller<'_>) -> Result<TakeAnswer, SoldRefusal> {
let cfg = state.estates.cfg.as_ref().ok_or(SoldRefusal::NotSigned)?;
let now = state.now();
let call = EstateCall { v: 1, call: CALL_UPDATE_TAKE.into(), estate: company.to_string(), members: vec![], gib: 0, at_unix_ms: 0, carries: String::new() };
let run = by.run(state.estates.get(company).as_ref(), &call, now)?;
if now.saturating_sub(run.update_taken_at_ms) >= TAKEN_KEPT_EVERY_MS {
let _ = with_run(state, cfg, company, |r, _| {
r.update_taken_at_ms = now.max(1);
Ok(())
});
}
let terms = super::terms_of(state, cfg, &run).unwrap_or_else(|_| cfg.clone());
let Some(u) = run.update.as_ref().filter(|u| u.state == state::RUNNING) else {
let words = match run.update.as_ref() {
Some(u) => u.line(),
None => "no update is open".into(),
};
return Ok(TakeAnswer { words, ..TakeAnswer::default() });
};
let role = u.next().map(|s| s.role.clone()).unwrap_or_default();
// The release the update was OPENED with, never one fed since.
let terms = as_opened(&terms, u);
// F9: the ONE box takes this box's step from the custody's copy: the front is handed no box.
if let Some(r) = custody_drives(&run, &shop_roles_of(&terms), custody_holds(state, company)) {
let words = format!("the ONE box updates the {r} itself, from the custody's copy of the installer material (the primary keeps none yet) — {}", u.line());
return Ok(TakeAnswer { open: true, release: u.release.clone(), live: u.live, role: String::new(), app: None, front: None, words });
}
let (app, front) = if u.live {
let objects = super::presigned_objects(&terms, state.estates.custody.as_ref(), state.estates.media_key_fed.as_deref(), now / 1_000).map_err(SoldRefusal::Unavailable)?;
let plan = super::lifecycle::of(&terms).map_err(SoldRefusal::Refused)?.update_plan();
(live_app(&terms, &objects, &plan.app_medium), live_app(&terms, &objects, &plan.front_medium))
} else {
(None, None)
};
Ok(TakeAnswer { open: true, release: u.release.clone(), live: u.live, role, app, front, words: u.line() })
}
/// **`SoldEstate.UpdateReport`**: the front reports `role`'s step (`outcome` and `words`, both
/// signed in the call's `carries`). A failure stops the update and the operator is told
/// ([`ALERT_KIND`]); `needs-keep` from the front hands the front's keep boot to this box's tick.
pub fn report(state: &AppState, company: &str, role: &str, carries: &str, at_ms: i64, signature: &[u8]) -> Result<String, SoldRefusal> {
report_as(state, company, role, carries, super::sold::Caller::Signed { at_ms, signature })
}
/// [`report`], asked by `by` (the old door's signed call, or `Fleet.Ask{update_report}`).
pub fn report_as(state: &AppState, company: &str, role: &str, carries: &str, by: super::sold::Caller<'_>) -> Result<String, SoldRefusal> {
let cfg = state.estates.cfg.as_ref().ok_or(SoldRefusal::NotSigned)?;
let now = state.now();
let call = EstateCall { v: 1, call: CALL_UPDATE_REPORT.into(), estate: company.to_string(), members: vec![role.to_string()], gib: 0, at_unix_ms: 0, carries: carries.to_string() };
by.run(state.estates.get(company).as_ref(), &call, now)?;
let (word, words) = carries.split_once('\n').unwrap_or((carries, ""));
let outcome = Outcome::parse(word).map_err(SoldRefusal::Refused)?;
// F9: the front's primary keeps no installer material — the ONE box takes the step from the
// custody's copy instead of stopping the update.
let held = custody_holds(state, company);
with_run(state, cfg, company, |run, terms| {
let host = terms.host_of(&run.company);
if outcome == Outcome::Failed && held && words.contains(NO_MATERIAL_WORDS) && custody_may(run) {
let shop_roles = super::lifecycle::of(terms).map(|l| l.update_plan().upgraded_by_shop).unwrap_or_default();
if let Some(s) = run.update.as_mut().filter(|u| u.state == state::RUNNING && u.next().is_some_and(|s| s.role == role) && shop_roles.iter().any(|r| r == role)).and_then(|u| u.steps.iter_mut().find(|s| s.role == role)) {
s.custody = true;
s.words = words.to_string();
let said = format!("{host}'s front reached no live door of its {role} ({words}): the ONE box takes the {role}'s step from the custody's copy of the installer material");
run.history.push(Transition { phase: String::new(), at_ms: now, note: said.clone() });
eprintln!("monetize-server/update: {}: {said}", run.reference);
return Ok(said);
}
}
recorded(state, run, terms, role, &outcome, words, now)
})
}
/// **One box's step recorded on the run**: applied to the update, the box's release written,
/// the history, a failure told (ALERT and CODE RED) and the update closed when it ends. Answers
/// the update's line. The front's report and the ONE box's own step ([`custody_step`]) alike.
fn recorded(state: &AppState, run: &mut EstateRun, terms: &EstateOrderConfig, role: &str, outcome: &Outcome, words: &str, now: u64) -> Result<String, SoldRefusal> {
let host = terms.host_of(&run.company);
let u = run.update.as_mut().ok_or_else(|| SoldRefusal::Refused(format!("{host} has no update open")))?;
let effect = apply(u, role, outcome, words, now).map_err(SoldRefusal::Refused)?;
let release = if role == front_role_of(terms) { u.front_release.clone() } else { u.release.clone() };
let line = u.line();
if let Outcome::Done(how) = outcome {
super::code_red::clear(state, &format!("update:{}:{role}", run.company));
if !release.is_empty() {
run.release_of.insert(role.to_string(), release.clone());
}
let said = format!("{host}'s {role} runs release {} ({how}): {words}", short(&release));
run.history.push(Transition { phase: String::new(), at_ms: now, note: said });
}
match &effect {
Effect::Failed(w) => {
run.history.push(Transition { phase: String::new(), at_ms: now, note: format!("update stopped: {w}") });
alert(run, &format!("{host}: {w}"));
super::code_red::raise(
state,
&format!("update:{}:{role}", run.company),
&host,
role,
&format!("the update stopped at the {role}; the boxes after it are not touched"),
words,
&format!("look at {host}'s {role} (is it started?); once fixed, press \"Update\" again — the boxes already done stay done"),
);
close(run);
}
Effect::Finished => {
run.history.push(Transition { phase: String::new(), at_ms: now, note: format!("update done: {line}") });
close(run);
}
Effect::FrontKeep => run.history.push(Transition { phase: String::new(), at_ms: now, note: line.clone() }),
Effect::Advanced => {}
}
eprintln!("monetize-server/update: {}: {role} {}: {line}", run.reference, outcome.word());
Ok(line)
}
// ── the tick: the front's keep boot, and a stalled step told ─────────────────
/// The jera job kind of a front's keep boot.
pub const FRONT_KEEP_JOB: &str = "monetize_front_keep";
/// **The front keep boots' lane** of `state`: one jera ChainQueue, started on the first ask on the
/// tick's runtime. `Err` when the tick runs with no runtime (the next tick asks again).
fn front_keep_lane(state: &Arc<AppState>) -> Result<Arc<jera::jobs::chain::ChainQueue>, String> {
if let Some(q) = state.estates.front_keep_lane.get() {
return Ok(q.clone());
}
let rt = tokio::runtime::Handle::try_current().map_err(|_| "no runtime to run the front keep 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("update: front keep");
let queue = jera::jobs::chain::ChainQueue::new(sink);
let q = state.estates.front_keep_lane.get_or_init(|| queue.clone()).clone();
if Arc::ptr_eq(&q, &queue) {
let weak = Arc::downgrade(state);
// A32-14, stays a task: the front-keep lane's jera worker.
rt.spawn(queue.clone().run(move |spec| {
let company = spec.payload.get("company").and_then(|v| v.as_str()).unwrap_or_default().to_string();
let Some(st) = weak.upgrade() else { anyhow::bail!("the server stopped") };
FRONT_KEEPING.lock().unwrap_or_else(|p| p.into_inner()).insert(company.clone());
front_keep(&st, &company);
FRONT_KEEPING.lock().unwrap_or_else(|p| p.into_inner()).remove(&company);
Ok(serde_json::json!({ "company": company }))
}));
}
Ok(q)
}
/// The fronts keep-booting now, by company: the sweep holds their clones (busy).
pub(crate) static FRONT_KEEPING: std::sync::Mutex<std::collections::BTreeSet<String>> = std::sync::Mutex::new(std::collections::BTreeSet::new());
fn is_zero(n: &u32) -> bool {
*n == 0
}
fn is_zero_ms(n: &u64) -> bool {
*n == 0
}
/// **On every tick**: an update past its [`UPDATE_DEADLINE_MS`] is FAILED by itself; each
/// update waiting for the front's keep boot gets it (a queued jera job); a step — the front's
/// keep boot too — not reported within [`STALL_MS`] is told to the operator, and told again
/// every [`STALL_RETELL_MS`] while it stalls (A32-7).
pub fn look(state: &Arc<AppState>) {
let now = state.now();
for run in state.estates.all() {
let Some(u) = run.update.as_ref().filter(|u| u.is_open()) else { continue };
if now.saturating_sub(u.opened_at_ms) >= UPDATE_DEADLINE_MS && !FRONT_KEEPING.lock().unwrap_or_else(|p| p.into_inner()).contains(&run.company) {
fail_at_deadline(state, &run.company, now);
continue;
}
if u.is_open() && now.saturating_sub(u.waiting_since()) >= STALL_MS && (u.stall_told_at_ms == 0 || now.saturating_sub(u.stall_told_at_ms) >= STALL_RETELL_MS) {
tell_stall(state, &run.company, now);
}
if u.state == state::FRONT_KEEP {
// A queued jera job per estate: the same key while it is queued or running is that job.
let lane = match front_keep_lane(state) {
Ok(l) => l,
Err(e) => {
eprintln!("monetize-server/update: the front's keep boot waits: {e}");
continue;
}
};
let spec = jera::jobs::chain::ChainSpec::new(FRONT_KEEP_JOB, format!("front-keep={}", run.company), format!("front-keep-{}", run.company), 0, "monetize", serde_json::json!({ "company": run.company }))
.with_target(format!("keep-boot the front of {}", run.company))
.with_priority(jera::jobs::chain::Priority::Normal);
let _ = lane.submit(spec);
} else if u.state == state::RUNNING && custody_due(state, &run) {
// F9: the ONE box's own step of a box the front cannot reach, on the ESTATE'S OWN lane
// (A32-14): a step waits up to 20 min between its polls, and on the shared front-keep
// lane that wait held every other estate's front keep boot. A second ask while the
// step runs is that step.
let (st, company) = (Arc::clone(state), run.company.clone());
if let Err(e) = crate::jobs::once(CUSTODY_STEP_JOB, &crate::jobs::of(state, &run.company), move || {
// A hand-off that failed is a FAILED row with its words (RA4-4); its words are the
// update's record either way.
custody_step_checked(&st, &company, &CustodyTweak::default()).map(|_| ())
}) {
eprintln!("monetize-server/update: the custody's step waits: {e}");
}
}
}
}
/// A32-7: the stall of `company`'s open update told (again): an ALERT and its CODE RED.
fn tell_stall(state: &AppState, company: &str, now: u64) {
let Some(cfg) = state.estates.cfg.as_ref() else { return };
let said = with_run(state, cfg, company, |r, terms| {
let Some(u) = r.update.as_mut().filter(|u| u.is_open()) else { return Ok(()) };
u.stall_told_at_ms = now;
u.stalls_told = u.stalls_told.saturating_add(1);
let role = u.next().map(|s| s.role.clone()).unwrap_or_default();
let hours = now.saturating_sub(u.waiting_since()) / 3_600_000;
let what = if u.state == state::FRONT_KEEP && u.front_keep_began_at_ms > 0 {
format!("the {role}'s keep boot began {} min ago and its outcome was never written: look at the {role} (did it come up from its disk?)", now.saturating_sub(u.front_keep_began_at_ms) / 60_000)
} else if u.state == state::FRONT_KEEP {
format!("the {role}'s keep boot has not finished for {hours} h")
} else { format!("the {role}'s update has not been reported for {hours} h") };
let w = format!("{}: {what} (told {} time(s); the update fails by itself {} h after it opened, or the operator cancels it: `estate update-cancel`) — is the front up? ({})", terms.host_of(&r.company), u.stalls_told, UPDATE_DEADLINE_MS / 3_600_000, u.line());
alert(r, &w);
super::code_red::raise(state, &format!("update:{}:{role}", r.company), &terms.host_of(&r.company), &role, &format!("the update has not moved for {hours} h"), &w, "look at the front (it drives the update): is it up and does it reach the ONE box? Or cancel the update (`estate update-cancel <estate> --apply`)");
Ok(())
});
if let Err(e) = said {
eprintln!("monetize-server/update: ALERT {ALERT_KIND} {company}: a stalled update could not be told: {e:?}");
}
}
/// A32-7: `company`'s open update is past its [`UPDATE_DEADLINE_MS`]: FAILED by itself, the boxes
/// done kept, the box waiting not touched; said (ALERT, CODE RED) and closed, so a new update may
/// open.
fn fail_at_deadline(state: &AppState, company: &str, now: u64) {
let Some(cfg) = state.estates.cfg.as_ref() else { return };
let said = with_run(state, cfg, company, |r, terms| {
let Some(u) = r.update.as_mut().filter(|u| u.is_open()) else { return Ok(()) };
let role = u.next().map(|s| s.role.clone()).unwrap_or_default();
let host = terms.host_of(&r.company);
if let Some(s) = u.steps.iter_mut().find(|s| s.role == role) {
s.state = step::FAILED.into();
s.at_ms = now;
}
u.state = state::FAILED.into();
u.closed_at_ms = now;
u.words = format!("the update was not finished within its deadline of {} h: the {role} never reported — the boxes after it are NOT touched", UPDATE_DEADLINE_MS / 3_600_000);
let words = u.words.clone();
let w = format!("{host}: {words}");
alert(r, &w);
super::code_red::raise(state, &format!("update:{}:{role}", r.company), &host, &role, &format!("the update passed its deadline of {} h", UPDATE_DEADLINE_MS / 3_600_000), &w, &format!("look at {host}'s {role} and its front; once fixed, press \"Update\" again"));
r.history.push(Transition { phase: String::new(), at_ms: now, note: format!("update stopped: {words}") });
close(r);
Ok(())
});
if let Err(e) = said {
eprintln!("monetize-server/update: ALERT {ALERT_KIND} {company}: an update past its deadline could not be closed now: {e:?}");
}
}
/// ★ **A32-7: `EstateOrder.CancelUpdate`** (the operator's alone): the open update of `estate`
/// closed as CANCELLED — the boxes done stay done (a mixed estate until "Update" again), the box
/// waiting is not touched, its stall's CODE RED goes. Refused by name: no update open, or the
/// front's keep boot running right now (cancel once it ends). A new update may open after it.
pub fn cancel(state: &AppState, estate: &str, by: &str) -> Result<String, SoldRefusal> {
let cfg = state.estates.cfg.as_ref().ok_or_else(|| SoldRefusal::Refused("this monetize lays no estates".into()))?;
let now = state.now();
let words = with_run(state, cfg, estate, |r, terms| {
let host = terms.host_of(&r.company);
let Some(u) = r.update.as_mut().filter(|u| u.is_open()) else {
return Err(SoldRefusal::Refused(format!("no update is open for {host}: nothing to cancel")));
};
if FRONT_KEEPING.lock().unwrap_or_else(|p| p.into_inner()).contains(&r.company) {
return Err(SoldRefusal::Unavailable(format!("{host}'s front is being keep-booted right now: cancel once the boot ends")));
}
let done: Vec<String> = u.steps.iter().filter(|s| s.state == step::DONE).map(|s| s.role.clone()).collect();
let waiting = u.next().map(|s| s.role.clone()).unwrap_or_default();
u.state = state::CANCELLED.into();
u.closed_at_ms = now;
u.words = format!(
"cancelled by {by}: {} — the {waiting} was not touched",
if done.is_empty() { "no box was updated".to_string() } else { format!("{} run(s) release {} and stay so", and_words(&done), short(&u.release)) }
);
let words = format!("{host}: {}", u.words);
r.history.push(Transition { phase: String::new(), at_ms: now, note: format!("update {}", u.words) });
close(r);
Ok(words)
})?;
super::code_red::clear(state, &format!("update:{}:", state.estates.get(estate).map(|r| r.company).unwrap_or_else(|| estate.to_string())));
eprintln!("monetize-server/update: {words}");
Ok(words)
}
/// **The front booted once from the release's `front-keep` medium** (its system rewritten, its
/// `/data` kept — its keys, its kept material) over the cloud API, and started again from its
/// disk. Public for the tests, which call it synchronously.
pub fn front_keep(state: &AppState, company: &str) {
let Some(cfg) = state.estates.cfg.as_ref() else { return };
let now = state.now();
let Some(run) = state.estates.get(company) else { return };
let terms = super::terms_of(state, cfg, &run).unwrap_or_else(|_| cfg.clone());
// The front is keep-booted from the release the update was OPENED with.
let terms = match run.update.as_ref() {
Some(u) => as_opened(&terms, u),
None => terms,
};
let host = terms.host_of(company);
let front_role = front_role_of(&terms);
// ★ A32-20: the boot is CLAIMED under the run's pen before the cloud is asked — only an
// update waiting for it now, and only once: a second job (or a tick that queued one before
// the first outcome landed) finds it claimed or the update moved on, and boots nothing.
let claimed = with_run(state, cfg, company, |r, _| {
let Some(u) = r.update.as_mut().filter(|u| u.state == state::FRONT_KEEP) else { return Ok(false) };
if u.front_keep_began_at_ms > 0 {
return Ok(false);
}
u.front_keep_began_at_ms = now;
Ok(true)
});
match claimed {
Ok(true) => {}
Ok(false) => {
eprintln!("monetize-server/update: {host}: no keep boot of the front is waiting (none open, or one already began): nothing booted");
return;
}
Err(e) => {
eprintln!("monetize-server/update: {host}: the front's keep boot waits: {e:?}");
return;
}
}
let outcome: Result<String, String> = (|| {
let laid = run.laid.as_ref().ok_or("nothing laid")?;
let plan = super::lifecycle::of(&terms)?.update_plan();
let front = laid.machine(&plan.front_role).cloned().ok_or(format!("no {} laid", plan.front_role))?;
let objects = super::presigned_objects(&terms, state.estates.custody.as_ref(), state.estates.media_key_fed.as_deref(), now / 1_000)?;
let keep = plan.front_keep_medium.as_str();
let medium = terms.media.get(keep).map(|u| u.trim().to_string()).unwrap_or_default();
let medium_object = if medium.is_empty() { objects.get(keep).filter(|o| !o.url.trim().is_empty()).cloned() } else { None };
if medium.is_empty() && medium_object.is_none() {
return Err(format!("the release names no `{keep}` medium: the front's BASE cannot be upgraded by the shop"));
}
let cloud = state.registry.find_cloud(&terms.cloud).ok_or_else(|| format!("the shop's cloud {:?} is not linked", terms.cloud))?;
let journal = state.estates.journal(company)?;
let keep = monetize_cloud::lay::KeepBoot { reference: laid.reference.clone(), zone: laid.zone.clone(), machine: front, medium, medium_object };
let m = monetize_cloud::ask::keep_boot(&cloud, &keep, &journal).map_err(|e| format!("the provider: {e}"))?;
Ok(format!("booted from the release's front keep medium, its data kept; started again at {} / {}", m.public_ipv4, m.utility_ipv4))
})();
let written = with_run(state, cfg, company, |r, _| {
let Some(u) = r.update.as_mut().filter(|u| u.state == state::FRONT_KEEP) else { return Ok(()) };
u.state = state::RUNNING.into();
u.front_keep_began_at_ms = 0;
let (o, words) = match &outcome {
Ok(w) => (Outcome::Done("keep".into()), w.clone()),
Err(w) => (Outcome::Failed, w.clone()),
};
let effect = apply(u, &front_role, &o, &words, now).map_err(SoldRefusal::Refused)?;
let line = u.line();
if matches!(o, Outcome::Done(_)) {
let rel = u.front_release.clone();
if !rel.is_empty() {
r.release_of.insert(front_role.clone(), rel);
}
}
match effect {
Effect::Failed(w) => {
alert(r, &format!("{host}: {w}"));
super::code_red::raise(
state,
&format!("update:{company}:front"),
&host,
"front",
"the front's keep boot failed; the update stopped before twin and main",
&w,
&format!("look at {host}'s front (is it started, from its disk?); once fixed, press \"Update\" again"),
);
r.history.push(Transition { phase: String::new(), at_ms: now, note: format!("update stopped: {w}") });
close(r);
}
// front first: twin and main still wait for the new front
Effect::Advanced => {
super::code_red::clear(state, &format!("update:{company}:front"));
r.history.push(Transition { phase: String::new(), at_ms: now, note: format!("the front was keep-booted first: {line}") })
}
_ => {
super::code_red::clear(state, &format!("update:{company}:front"));
r.history.push(Transition { phase: String::new(), at_ms: now, note: format!("update done: {line}") });
close(r);
}
}
Ok(())
});
// ★ A32-20: an outcome that could not be written is said, never dropped: the boot stays
// claimed (no second boot), and the stall, the deadline or a cancel ends the update.
if let Err(e) = written {
let w = format!("{host}: the front's keep boot ran ({}) but its outcome could not be written: {e:?}", match &outcome { Ok(w) | Err(w) => w });
eprintln!("monetize-server/update: ALERT {ALERT_KIND} {company}: {w}");
super::code_red::raise(state, &format!("update:{company}:front"), &host, "front", "the front's keep boot's outcome was not written", &w, "look at the front; then cancel the update (`estate update-cancel <estate> --apply`) and press \"Update\" again");
}
}
// ── F9: the ONE box's own step, from the custody's copy of the installer material ────────
/// The jera job kind of the ONE box's own step of a box ([`custody_step`]).
pub const CUSTODY_STEP_JOB: &str = "monetize_custody_step";
/// What a front says when its primary keeps no installer material (holger-estate-front, the
/// sealed slot; gunnar-front alike): `no installer material is kept in … yet`.
pub const NO_MATERIAL_WORDS: &str = "no installer material is kept";
/// How often a staged box is asked again (the front's [`POLL_EVERY`]).
pub const CUSTODY_POLL: std::time::Duration = std::time::Duration::from_secs(10);
/// How long a staged box has to confirm (the front's budget: 15 min).
pub const CUSTODY_CONFIRM: std::time::Duration = std::time::Duration::from_secs(15 * 60);
/// How long a keep-booted box is read back on its live door (a box without one is done all the same).
pub const CUSTODY_KEEP_READ: std::time::Duration = std::time::Duration::from_secs(5 * 60);
/// The old app back with no trial this long after a stage: rolled back.
pub const CUSTODY_ROLLED_BACK: std::time::Duration = std::time::Duration::from_secs(60);
/// A box's live slot as its log door says it (nordisk-pairing `live::Running`).
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct LiveRunning {
/// The BASE hash.
pub base: String,
/// The running app's `app.blake3`.
pub app: String,
/// `a` | `b`, and the trial state (`a`, `b trial`, `b booted`).
pub slot: String,
/// `korp.live_update` on the box.
pub live_update: bool,
}
/// A live app staged into a box's slot (nordisk-pairing `live::Stage` with its
/// `live::Declared`).
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct LiveStage {
/// `app.blake3`, hex.
pub pin_hex: String,
/// `app.sig`, raw.
pub signature: Vec<u8>,
pub base: String,
/// The app `.znippy`.
pub archive: Vec<u8>,
/// Lane B5: the release declaration `(bundle, signature)`, raw; `None`: none rides.
pub release: Option<(Vec<u8>, Vec<u8>)>,
/// Lane B5: the release's `korp.env.*` tokens; `None`: undeclared.
pub env: Option<Vec<String>>,
}
/// The refusals of a box's stage that a KEEP boot answers (nordisk-pairing
/// `live::KEEP_REFUSALS`, lane B5): the release moved what only a keep boot lays.
pub const KEEP_REFUSALS: &[&str] = &["[live-env-moved]", "[live-release-absent]", "[live-declared-unsupported]"];
/// What [`declared_of`] reads: the release declaration with its signature (`None`: the release
/// declares nothing), and the env the box is handed (`None`: none).
pub type Declared = (Option<(Vec<u8>, Vec<u8>)>, Option<Vec<String>>);
/// Pure: **what the release declares, off its live record** (lane B5) — the declaration and
/// the env the box is handed with the app. A bundle that is not hex is refused by name.
pub fn declared_of(record: &LiveRecord) -> Result<Declared, String> {
let release = match record.release.trim() {
"" => None,
b => Some((unhex(b).map_err(|e| format!("[update-release] the release declaration: {e}"))?, unhex(&record.release_sig).map_err(|e| format!("[update-release] its signature: {e}"))?)),
};
let env = (!record.env.is_empty()).then(|| record.env.clone());
Ok((release, env))
}
/// Whether the shop's custody still holds `company`'s pairing material — the installer CA and
/// its keys and the link keys minted before the lay, the bundle its pairing handed back beside
/// them — which the hand-off has not forgotten. The installer's link is reached with the minted
/// material and the plan of the order's laid file (F5's re-check does the same): an install
/// stopped before its late PASS handed the bundle back is reached all the same.
pub fn custody_holds(state: &AppState, company: &str) -> bool {
let Some(c) = state.estates.custody.as_ref() else { return false };
state.estates.get(company).is_some_and(|r| r.secrets_forgotten_at_ms == 0) && c.holds(company)
}
/// Pure: **may the ONE box take an update step of `run` from the custody?** Every box PAIRED
/// (`paired`, or delivered: `captured`, `grace`, `suspended`), never an estate being deleted or
/// gone.
pub fn custody_may(run: &EstateRun) -> bool {
admits(run, "").is_ok() && run.phase != phase::DELETING && !phase::is_gone(&run.phase)
}
/// The boxes the shop upgrades for `terms`' product (main and twin), whose step the ONE box may take.
fn shop_roles_of(terms: &EstateOrderConfig) -> Vec<String> {
super::lifecycle::of(terms).map(|l| l.update_plan().upgraded_by_shop).unwrap_or_default()
}
/// Pure: **the box whose step the ONE box takes itself** (F9) — the open update's next box is a
/// main or twin (`shop_roles`), the custody still holds the estate's installer material
/// (`held`), every box is PAIRED ([`custody_may`]), and the front's primary keeps no material:
/// its front never took it (test3-holger-rs: the r6 front never did), or it reported that its
/// primary keeps none ([`Step::custody`]). `None`: the front takes the box, as always.
pub fn custody_drives(run: &EstateRun, shop_roles: &[String], held: bool) -> Option<String> {
if !held || !custody_may(run) {
return None;
}
let s = run.update.as_ref().filter(|u| u.state == state::RUNNING)?.next()?;
(shop_roles.contains(&s.role) && (run.material_taken_at_ms == 0 || s.custody)).then(|| s.role.clone())
}
/// [`custody_drives`] for `run` as the tick sees it: its update's release, the custody asked.
fn custody_due(state: &AppState, run: &EstateRun) -> bool {
let Some(cfg) = state.estates.cfg.as_ref() else { return false };
let Some(u) = run.update.as_ref().filter(|u| u.state == state::RUNNING) else { return false };
let terms = as_opened(&super::terms_of(state, cfg, run).unwrap_or_else(|_| cfg.clone()), u);
// The cheap checks first: the custody is unsealed only for a box it could take.
custody_drives(run, &shop_roles_of(&terms), true).is_some() && custody_holds(state, &run.company)
}
/// What the ONE box can do to a box of the estate (prod: the installer's link with the custody's
/// material, the release bucket, the cloud's keep boot; a test fakes the fetch and the clock).
pub trait CustodyHands {
fn running(&self, role: &str) -> Result<LiveRunning, String>;
fn stage(&self, role: &str, stage: &LiveStage) -> Result<String, String>;
fn fetch(&self, url: &str) -> Result<Vec<u8>, String>;
/// `role` booted from the release's keep medium (its data kept).
fn keep(&self, role: &str) -> Result<String, String>;
fn wait(&self, d: std::time::Duration);
}
/// What one box gets ([`custody_decide`]): Holger's and Gunnar's fronts decide alike.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum CustodyPlan {
Nothing,
Live,
Keep(String),
Wait(String),
}
/// Pure: **what one box gets** — its running app against the release's live app (the fronts'
/// `decide`, word for word).
pub fn custody_decide(r: &LiveRunning, app: Option<&LiveApp>, live: bool) -> CustodyPlan {
let Some(app) = app else {
return CustodyPlan::Keep(if live { "the release carries no live app for it".into() } else { "live updates are OFF for this estate".into() });
};
if r.slot.split_whitespace().nth(1).is_some() {
return CustodyPlan::Wait(format!("[live-trial-open] slot {}: its last update is not confirmed yet", r.slot));
}
if r.app.trim() == app.record.app.trim() {
return CustodyPlan::Nothing;
}
if !live {
return CustodyPlan::Keep("live updates are OFF for this estate".into());
}
if !r.live_update {
return CustodyPlan::Keep("live update is OFF on the box (korp.live_update=off, its BASE)".into());
}
if r.base.trim() != app.record.base.trim() {
return CustodyPlan::Keep(format!("its BASE moved ({} → {}): a base update", short(&r.base), short(&app.record.base)));
}
CustodyPlan::Live
}
fn unhex(s: &str) -> Result<Vec<u8>, String> {
let s = s.trim();
if s.len() % 2 != 0 {
return Err("odd hex".into());
}
(0..s.len()).step_by(2).map(|i| u8::from_str_radix(&s[i..i + 2], 16).map_err(|e| e.to_string())).collect()
}
/// Wait until `role`'s live door answers again, at most `within` (a keep-booted box).
fn answers_again(h: &dyn CustodyHands, role: &str, within: std::time::Duration) -> Result<LiveRunning, String> {
let mut waited = std::time::Duration::ZERO;
let mut last = String::new();
while waited < within {
h.wait(CUSTODY_POLL);
waited += CUSTODY_POLL;
match h.running(role) {
Ok(r) => return Ok(r),
Err(e) => last = e,
}
}
Err(format!("the {role}'s live door did not answer within {} min ({last})", within.as_secs() / 60))
}
/// Pure over `h`: **one box's step, the ONE box's own** — the fronts' step for a main or twin.
/// `None`: nothing to record now (a trial still open on the box).
pub fn custody_step_over(h: &dyn CustodyHands, role: &str, app: Option<&LiveApp>, live: bool) -> Option<(Outcome, String)> {
let failed = |w: String| Some((Outcome::Failed, w));
// The keep boot is the cloud's (the delivered `Upgrade`'s own, sold::keep_boot_box): it
// needs no live door. A box with one is read back after it; a box without one (r6) is done
// once the cloud started it again from its disk (F9, c1 2026-10-21).
let keep = |why: String| match h.keep(role) {
Ok(k) => {
let after = match answers_again(h, role, CUSTODY_KEEP_READ) {
Ok(r) => format!("its live door answers again (app {}, slot {})", short(&r.app), r.slot),
Err(_) => "it has no live door to read back: started again from its disk".to_string(),
};
Some((Outcome::Done("keep".into()), format!("{why}: {k}; {after}")))
}
Err(e) => failed(format!("{why}; the keep upgrade of the {role}: {e}")),
};
// Live off: every box from its keep medium; its live door is never asked.
if !live {
return keep("live updates are OFF for this estate".into());
}
let running = match h.running(role) {
Ok(r) => r,
// A box with no live door (an r6 box, or one the installer link cannot ask live): its
// keep medium, as the front's `[live-unsupported]` fallback.
Err(e) => return keep(format!("its live door did not answer the ONE box ({e})")),
};
match custody_decide(&running, app, live) {
CustodyPlan::Wait(_) => None,
CustodyPlan::Nothing => Some((Outcome::Done("nothing".into()), format!("the {role} runs app {}", short(&running.app)))),
CustodyPlan::Keep(why) => keep(why),
CustodyPlan::Live => {
// custody_decide says Live only on an app; none is the keep.
let Some(app) = app else { return keep("the release carries no live app for it".into()) };
if app.record.sig.trim().is_empty() {
return failed("[update-unsigned] the release's live app carries no signature: never staged".into());
}
let archive = match h.fetch(&app.url) {
Ok(a) => a,
Err(e) => return failed(format!("[update-fetch] {e}")),
};
use sha2::Digest as _;
let sha: String = sha2::Sha256::digest(&archive).iter().map(|b| format!("{b:02x}")).collect();
if !app.sha256.trim().is_empty() && sha != app.sha256.trim() {
return failed(format!("[update-fetch] the archive's sha256 is {}, the release says {}", short(&sha), short(&app.sha256)));
}
if archive.len() as u64 != app.record.app_len {
return failed(format!("[update-fetch] the archive is {} bytes, the release says {}", archive.len(), app.record.app_len));
}
let signature = match unhex(&app.record.sig) {
Ok(s) => s,
Err(e) => return failed(format!("[update-unsigned] the signature: {e}")),
};
let (release, env) = match declared_of(&app.record) {
Ok(d) => d,
Err(e) => return failed(e),
};
let stage = LiveStage { pin_hex: app.record.app.trim().to_string(), signature, base: app.record.base.trim().to_string(), archive, release, env };
match h.stage(role, &stage) {
Ok(_) => {}
// Only a keep boot lays what the release moved (lane B5).
Err(e) if KEEP_REFUSALS.iter().any(|t| e.contains(t)) => return keep(format!("the {role} refused the live stage ({})", e.trim())),
Err(e) => return failed(format!("the stage was refused: {e}")),
}
let mut waited = std::time::Duration::ZERO;
let mut last = String::new();
while waited < CUSTODY_CONFIRM {
h.wait(CUSTODY_POLL);
waited += CUSTODY_POLL;
match h.running(role) {
Ok(r) if r.app.trim() == app.record.app.trim() && r.slot.split_whitespace().nth(1).is_none() => {
return Some((Outcome::Done("live".into()), format!("the ONE box staged it from the custody's installer material: the {role} runs app {} confirmed (slot {})", short(&r.app), r.slot)))
}
Ok(r) if r.app.trim() == running.app.trim() && r.slot.split_whitespace().nth(1).is_none() && waited >= CUSTODY_ROLLED_BACK => {
return failed(format!("[update-rolled-back] the {role} runs its old app {} again (slot {}): the new one never confirmed", short(&running.app), r.slot))
}
Ok(r) => last = format!("slot {}, app {}", r.slot, short(&r.app)),
Err(e) => last = e,
}
}
failed(format!("[update-not-confirmed] the {role} did not confirm within {} min ({last})", CONFIRM_MIN))
}
}
}
const CONFIRM_MIN: u64 = CUSTODY_CONFIRM.as_secs() / 60;
/// The ONE box's hands on an estate: the installer's link with the custody's material (handed
/// off for the step, taken back after it), the release bucket, the cloud's keep boot.
struct ShopHands<'a> {
state: &'a AppState,
terms: &'a EstateOrderConfig,
run: &'a EstateRun,
req: super::InstallRequest,
keep_medium: String,
tweak: &'a CustodyTweak,
}
/// A test's stand-in for the release bucket: the archive at a url.
pub type FetchFn = Box<dyn Fn(&str) -> Result<Vec<u8>, String> + Send + Sync>;
/// A test's tweaks of the ONE box's hands: the bucket's archive and the clock (prod: none).
#[derive(Default)]
pub struct CustodyTweak {
pub fetch: Option<FetchFn>,
/// No real waits (the test's boxes answer at once).
pub no_wait: bool,
}
impl CustodyHands for ShopHands<'_> {
fn running(&self, role: &str) -> Result<LiveRunning, String> {
self.state.estates.installer.live_running(&self.req, role)
}
fn stage(&self, role: &str, stage: &LiveStage) -> Result<String, String> {
self.state.estates.installer.live_stage(&self.req, role, stage)
}
fn fetch(&self, url: &str) -> Result<Vec<u8>, String> {
if let Some(f) = self.tweak.fetch.as_ref() {
return f(url);
}
let _ = rustls::crypto::ring::default_provider().install_default();
let client = reqwest::blocking::Client::builder()
.tls_certs_only(webpki_root_certs::TLS_SERVER_ROOT_CERTS.iter().filter_map(|c| reqwest::Certificate::from_der(c).ok()))
.connect_timeout(std::time::Duration::from_secs(30))
.timeout(std::time::Duration::from_secs(600))
.build()
.map_err(|e| format!("no HTTPS client: {e}"))?;
let resp = client.get(url).send().map_err(|e| e.without_url().to_string())?;
if !resp.status().is_success() {
return Err(format!("the bucket answered {}", resp.status()));
}
Ok(resp.bytes().map_err(|e| e.without_url().to_string())?.to_vec())
}
fn keep(&self, role: &str) -> Result<String, String> {
let laid = self.run.laid.as_ref().ok_or("nothing laid")?;
let words = |r: SoldRefusal| match r {
SoldRefusal::Refused(w) | SoldRefusal::Unavailable(w) => w,
SoldRefusal::NotSigned => "not signed".into(),
};
super::sold::keep_boot_box(self.state, self.terms, &self.run.company, role, laid, &self.keep_medium, self.state.now()).map(|(_, w)| w).map_err(words)
}
fn wait(&self, d: std::time::Duration) {
if !self.tweak.no_wait {
std::thread::sleep(d);
}
}
}
/// **The ONE box takes the open update's next box itself, from the custody's copy of the
/// installer material** (F9, 2026-10-21; [`custody_drives`]): test3-holger-rs's r6 front never
/// took its material, so its primary keeps none, the front reaches no live door of its twin, and
/// the front cannot be updated while the update itself needs that material. The box's step is
/// the fronts' own — its live door asked over the installer link, the release's live app staged
/// and confirmed, or the keep medium when its BASE moved — then recorded as a front's report is,
/// and the next box is handed out (the front, last, to the front itself). Answers what it did.
pub fn custody_step(state: &AppState, company: &str) -> String {
custody_step_with(state, company, &CustodyTweak::default())
}
/// [`custody_step`] with a test's tweaks.
pub fn custody_step_with(state: &AppState, company: &str, tweak: &CustodyTweak) -> String {
custody_step_checked(state, company, tweak).unwrap_or_else(|why| why)
}
/// [`custody_step_with`], a failed hand-off (the custody's refusal, an object not presigned) an
/// `Err` with its words: the job's FAILED row (RA4-4).
pub fn custody_step_checked(state: &AppState, company: &str, tweak: &CustodyTweak) -> Result<String, String> {
custody_step_words(state, company, tweak)
}
fn custody_step_words(state: &AppState, company: &str, tweak: &CustodyTweak) -> Result<String, String> {
let Some(cfg) = state.estates.cfg.as_ref() else { return Ok("this monetize lays no estates".into()) };
let Some(run) = state.estates.get(company) else { return Ok(format!("{company} is no estate laid here")) };
let Some(u) = run.update.clone().filter(|u| u.state == state::RUNNING) else { return Ok("no update runs".into()) };
let terms = as_opened(&super::terms_of(state, cfg, &run).unwrap_or_else(|_| cfg.clone()), &u);
let host = terms.host_of(company);
let Ok(plan) = super::lifecycle::of(&terms).map(|l| l.update_plan()) else { return Ok(format!("{host} sells no estate of this kind")) };
let Some(role) = custody_drives(&run, &plan.upgraded_by_shop, custody_holds(state, company)) else {
return Ok(format!("{host}'s front takes its next box itself"));
};
let eo = &state.estates;
let (Some(dir), Some(custody)) = (eo.run_dir(company), eo.custody.as_ref()) else { return Ok("this box keeps no estate-orders directory or no custody".into()) };
let laid_path = dir.join("laid.json");
let handoff = match custody.hand_off(company, &dir.join("custody-step")) {
Ok(h) => h,
Err(e) => return Err(format!("the custody would not hand the installer's material over: {e}")),
};
let app = if u.live {
match super::presigned_objects(&terms, eo.custody.as_ref(), eo.media_key_fed.as_deref(), state.now() / 1_000) {
Ok(objects) => live_app(&terms, &objects, &plan.app_medium),
Err(e) => {
let _sealed = super::seal_back(state, custody, company, &handoff, "the update's custody step");
return Err(format!("the release's objects could not be presigned: {e}"));
}
}
} else {
None
};
let facts = super::run_terms(state, &run).map(|c| super::pairing_facts(&c)).unwrap_or_default();
let req = super::InstallRequest { facts, company: company.to_string(), apex: cfg.apex.clone(), account: run.account.clone(), laid: laid_path, handoff: handoff.dir().to_path_buf(), run: dir, role: role.clone(), narrate: Default::default(), relay: BTreeMap::new(), site: None, offsite: BTreeMap::new() };
eprintln!("monetize-server/update: {}: the ONE box takes the {role}'s step from the custody's installer material (its front's primary keeps none)", run.reference);
let hands = ShopHands { state, terms: &terms, run: &run, req, keep_medium: plan.keep_medium.clone(), tweak };
let done = custody_step_over(&hands, &role, app.as_ref(), u.live);
drop(hands);
let _sealed = super::seal_back(state, custody, company, &handoff, &format!("the update's custody step on the {role}"));
drop(handoff);
let Some((outcome, words)) = done else { return Ok(format!("the {role}'s last update is still on trial: asked again on the next tick")) };
let now = state.now();
match with_run(state, cfg, company, |r, t| {
// The update this step was taken for, still at this box.
if r.update.as_ref().is_none_or(|x| x.state != state::RUNNING || x.opened_at_ms != u.opened_at_ms || x.next().is_none_or(|s| s.role != role)) {
return Err(SoldRefusal::Refused(format!("{host}'s update moved while the ONE box took its {role}: nothing recorded")));
}
recorded(state, r, t, &role, &outcome, &format!("the ONE box, from the custody's installer material: {words}"), now)
}) {
Ok(line) => Ok(line),
Err(SoldRefusal::Refused(w) | SoldRefusal::Unavailable(w)) => Ok(w),
Err(SoldRefusal::NotSigned) => Ok("not signed".into()),
}
}
#[cfg(test)]
mod tests {
use super::*;
fn u() -> EstateUpdate {
EstateUpdate::open("01-gunnar", "01-front", true, "operator", 1_000)
}
#[test]
fn the_boxes_go_twin_main_front_and_each_must_be_reported_in_its_turn() {
let mut u = u();
assert_eq!(u.next().unwrap().role, "twin");
let wrong = apply(&mut u, "main", &Outcome::Done("live".into()), "", 2_000).unwrap_err();
assert!(wrong.contains("twin's turn"), "{wrong}");
assert_eq!(apply(&mut u, "twin", &Outcome::Done("live".into()), "staged and confirmed", 2_000).unwrap(), Effect::Advanced);
assert_eq!(u.next().unwrap().role, "main");
assert_eq!(apply(&mut u, "main", &Outcome::Done("keep".into()), "", 3_000).unwrap(), Effect::Advanced);
assert_eq!(apply(&mut u, "front", &Outcome::Done("nothing".into()), "", 4_000).unwrap(), Effect::Finished);
assert_eq!(u.state, state::DONE);
assert!(u.next().is_none());
assert!(apply(&mut u, "front", &Outcome::Done("live".into()), "", 5_000).is_err(), "a finished update takes no report");
}
#[test]
fn a_failure_stops_the_update_and_no_later_box_is_turned() {
let mut u = u();
let e = apply(&mut u, "twin", &Outcome::Failed, "[update-rolled-back] the old app is back", 2_000).unwrap();
assert!(matches!(&e, Effect::Failed(w) if w.contains("twin") && w.contains("NOT touched")), "{e:?}");
assert_eq!(u.state, state::FAILED);
assert!(u.next().is_none(), "main is never updated after a twin that failed");
assert_eq!(u.steps.iter().filter(|s| s.state == step::WAITING).count(), 2);
}
#[test]
fn live_off_takes_no_live_report_and_needs_keep_is_the_fronts_alone() {
let mut u = EstateUpdate::open("01", "01f", false, "operator", 0);
assert!(apply(&mut u, "twin", &Outcome::Done("live".into()), "", 1).unwrap_err().contains("OFF"));
assert!(apply(&mut u, "twin", &Outcome::NeedsKeep, "", 1).unwrap_err().contains("Upgrade"));
apply(&mut u, "twin", &Outcome::Done("keep".into()), "", 1).unwrap();
apply(&mut u, "main", &Outcome::Done("keep".into()), "", 2).unwrap();
assert_eq!(apply(&mut u, "front", &Outcome::NeedsKeep, "BASE moved", 3).unwrap(), Effect::FrontKeep);
assert_eq!(u.state, state::FRONT_KEEP);
assert!(u.is_open() && u.next().is_some_and(|s| s.role == "front"));
assert!(apply(&mut u, "front", &Outcome::Done("keep".into()), "", 4).is_err(), "the shop finishes a front keep boot, not a report");
}
#[test]
fn outcomes_round_trip_and_strangers_are_refused_by_name() {
for o in [Outcome::Done("live".into()), Outcome::Done("keep".into()), Outcome::Done("nothing".into()), Outcome::Failed, Outcome::NeedsKeep] {
assert_eq!(Outcome::parse(&o.word()).unwrap(), o);
}
assert!(Outcome::parse("done:maybe").unwrap_err().contains("not an update outcome"));
assert_eq!(report_carries(&Outcome::Failed, "x y"), "failed\nx y");
}
#[test]
fn a_stalled_step_counts_from_the_step_before_it() {
let mut u = u();
assert_eq!(u.waiting_since(), 1_000);
apply(&mut u, "twin", &Outcome::Done("live".into()), "", 9_000).unwrap();
assert_eq!(u.waiting_since(), 9_000);
}
}