crates/monetize-guardian/src/runner.rs
//! **The runner**: one pass over every estate per tick — diagnose, open a case, gather the
//! provider's word, wait for a button (or auto-fix), plan, and walk the plan one step at a
//! time: cloud steps through `Cloud`, orders queued SIGNED for the box to pull,
//! awaits read from reports.
use std::path::PathBuf;
use std::sync::Arc;
use monetize::sign::Keyring;
use monetize_cloud::lay::{same_resource, LaidMachine};
use monetize_cloud::{Cloud, CloudError, ResourceStatus};
use monetize_embed::fleet::{kind, SignedOrder};
use monetize_product::guardian::{
role as roles, BoxView, CloudStep, EstateGuardian, EstateView, Finding, Remedy, Step, StepState,
};
use monetize_product::EntitlementFact;
use crate::case::{step_state, Case, CaseState, CaseStep};
use crate::store::{Store, StoreError};
/// Who approved a case the estate's auto-fix opt-in approved.
pub const AUTO_PRINCIPAL: &str = "guardian (auto-fix)";
#[derive(Clone, Debug)]
pub struct GuardianConfig {
/// A box not heard for this long is silent: 3 report intervals of 30 s.
pub silent_after_ms: u64,
/// An order's `not_after`: past it, the box refuses to start it.
pub order_ttl_ms: u64,
/// How long an order step waits for its ack before the case fails.
pub order_within_ms: u64,
/// Where each finished case's receipt is written (`<id>.json`), if anywhere.
pub receipts_dir: Option<PathBuf>,
/// How long a box laid with no named key may pin its own on its first call (TOFU).
pub pin_window_ms: u64,
/// How long a fence waits for the provider to LIST the box it gave back as gone: UpCloud
/// keeps a deleted server in `maintenance` for up to five minutes (behaviour 7).
pub fence_within_ms: u64,
/// **How often the provider is asked about a silent main** (PROD-FIX 3, prod 2026-10-26:
/// UpCloud's storages were read several times a second). One look is a whole listing of
/// the account (UpCloud: `GET /storage` and one `GET /storage/<uuid>` per storage), so the
/// word is kept and asked again only this often ([`PROVIDER_LOOK_EVERY_MS`]).
pub provider_look_every_ms: u64,
/// **The tick asks the due looks itself** ([`Guardian::refresh_looks`]). A server that runs
/// them as a background job of its own ([`Guardian::looks_due`]) sets this `false`, and its
/// tick reads only what was looked at.
pub looks_in_tick: bool,
}
/// The provider is asked about a silent main at most once in five minutes ([`GuardianConfig::provider_look_every_ms`]).
pub const PROVIDER_LOOK_EVERY_MS: u64 = 5 * 60_000;
/// The word of a silent main the provider was not asked about yet: unknown, never a guess (the
/// guard reads it as "the provider could not say", and opens nothing on it).
pub const NOT_LOOKED_YET: &str =
"unknown (the provider was not asked yet: it is asked in the background)";
impl Default for GuardianConfig {
fn default() -> Self {
Self {
silent_after_ms: 90_000,
order_ttl_ms: 15 * 60_000,
order_within_ms: 6 * 3_600_000,
receipts_dir: None,
pin_window_ms: 10 * 60_000,
fence_within_ms: 15 * 60_000,
provider_look_every_ms: PROVIDER_LOOK_EVERY_MS,
looks_in_tick: true,
}
}
}
/// **What the provider last said of one box** ([`Guardian::view`] reads it, never the provider).
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ProviderLook {
/// Its word ([`provider_word`]); empty until the first look.
pub word: String,
/// When it was said (unix ms); 0: never.
pub looked_unix_ms: u64,
/// When it was last asked for (unix ms): a look is not asked again before
/// [`GuardianConfig::provider_look_every_ms`] past this, answered or not.
pub asked_unix_ms: u64,
}
/// One box laid again ([`World::lay_one`]).
#[derive(Clone, PartialEq, Eq, Debug, Default)]
pub struct LaidBox {
/// The provider's id for the new server.
pub server: String,
/// Its utility IPv4.
pub ipv4: String,
/// The new box's fleet key (base64), from the installer's verdict; empty when the
/// installer named none (then nothing it says is believed until a key is registered).
pub appliance_key: String,
pub words: String,
}
/// **The world the runner works in**: the products' guardians, the estates' clouds and
/// laid boxes, and the shop's ONE lay + installer path. Implemented by monetize-server.
pub trait World: Send + Sync {
fn guardian(&self, product: &str) -> Option<&dyn EstateGuardian>;
/// The cloud the estate's boxes stand in.
fn cloud(&self, instance: &str) -> Option<&dyn Cloud>;
/// **The box monetize laid in `role`, as its laid record holds it** — the estate's
/// reference (the label every resource of it carries) and the machine: its server, disks
/// and release clone. `None` when monetize laid none. What the fence gives back WHOLE
/// (`Change::Retire`) and the lay's fence invariant reads.
fn laid_machine(&self, instance: &str, role: &str) -> Option<(String, LaidMachine)>;
/// **What monetize's books hold for `product` at this estate** — every current verdict,
/// signed — re-stated to a recharged main as a signed `Snapshot` it pulls (`$entitlements`).
/// `Err` when the books cannot be read: the step fails and nothing is queued.
fn entitlements(&self, product: &str, instance: &str) -> Result<Vec<EntitlementFact>, String>;
/// Where monetize laid the box in `role` (its utility IPv4); empty when it laid none.
fn laid_ipv4(&self, instance: &str, role: &str) -> String;
/// **Lay ONE box again** for `role` and have the installer lay it — the same path a sold
/// estate is laid with. Called only once the role's old box is gone.
fn lay_one(&self, instance: &str, role: &str) -> Result<LaidBox, String>;
/// **Restore the box in `role` from its provider's newest backup** (S8: the front) — the
/// shop's own restore path, asked by `principal`. Answers what was done, in words. The
/// default restores nothing, by name.
fn restore_one(&self, instance: &str, role: &str, principal: &str) -> Result<String, String> {
let _ = (instance, principal);
Err(format!("this world restores no {role} from a backup"))
}
/// ★ SEC-S1: **why monetize holds the whole estate stopped ON PURPOSE**, when it does — its
/// term stopped it (unpaid past its grace, cancelled at the end of its paid period, being
/// deleted). The guardian then starts, fences and lays nothing of it: a stopped box of it
/// is monetize's own stop, never an outage. `None` (the default): nothing is held.
fn estate_held(&self, instance: &str) -> Option<String> {
let _ = instance;
None
}
/// ★ SEC-S1: **why monetize holds the box in `role` stopped on purpose**, when it does (the
/// term stopped it, and starts it itself). The start pass never starts it. `None` (the
/// default): a stopped box is an outage.
fn box_held(&self, instance: &str, role: &str) -> Option<String> {
let _ = (instance, role);
None
}
}
/// The proof a fence of an operator-laid estate carries: its first words.
pub const OPERATOR_FENCE: &str = "the operator's notice";
/// How long a case waits for an operator-laid estate's operator to lay the box again.
pub const OPERATOR_RELAY_WITHIN_MS: u64 = 24 * 3_600_000;
/// Why a button press is refused.
#[derive(Clone, PartialEq, Eq, Debug, thiserror::Error)]
pub enum PressRefusal {
/// No principal, or not a strong (fresh, user-verified) session.
#[error("{0}")]
Weak(String),
/// A case is already approved or running for the estate.
#[error("{0}")]
Busy(String),
#[error("{0}")]
Unknown(String),
#[error("{0}")]
Refused(String),
}
pub struct Guardian {
pub store: Arc<Store>,
/// The product keys (G8): an order for product `p` is signed with `p`'s.
pub(crate) keys: Arc<Keyring>,
pub cfg: GuardianConfig,
/// **What the provider last said of each silent main**, by `(instance, role)` (PROD-FIX 3).
looks: std::sync::Mutex<std::collections::BTreeMap<(String, String), ProviderLook>>,
}
/// What one step did this tick.
enum Moved {
Met(String),
Waiting,
Failed(String),
}
/// **Does laying `role` at step `i` promote a new box to take the writes?** A new main is
/// one; so is any box this plan later recharges from the twin. A promotion beside an old
/// box that still serves is split brain.
fn promotes(steps: &[CaseStep], i: usize, role: &str) -> bool {
role == roles::MAIN || steps[i..].iter().any(|s| matches!(&s.step, Step::Order { role: r, kind: k, .. } if r == role && k == kind::RECHARGE_FROM_TWIN))
}
/// **The confirmed fence of the old `role` box** before step `i`: the proof its fence step
/// holds once the provider's listing read the box gone. `None` when the plan fences no old
/// `role` box before `i`, or its fence confirmed nothing.
fn confirmed_fence(steps: &[CaseStep], i: usize, role: &str) -> Option<String> {
steps[..i]
.iter()
.rev()
.find(|s| matches!(&s.step, Step::Cloud(CloudStep::Fence { role: r }) if r == role))
.map(|s| s.fenced.clone())
.filter(|f| !f.is_empty())
}
/// **No promotion without a fence first** (stinky #18, owner 20:20Z): a plan that lays a
/// box it promotes with no fence of the old one before the lay is refused before any step.
fn unfenced_promotion(steps: &[CaseStep]) -> Option<String> {
steps.iter().enumerate().find_map(|(i, s)| match &s.step {
Step::Cloud(CloudStep::Lay { role }) if promotes(steps, i, role) && !steps[..i].iter().any(|s| matches!(&s.step, Step::Cloud(CloudStep::Fence { role: r }) if r == role)) => {
Some(format!("the plan promotes a new {role} with no fence of the old {role} before its lay: no promotion without a confirmed fence"))
}
_ => None,
})
}
/// **What the provider says of the box monetize laid in `role`** (`BoxView::at_provider`):
/// `gone` only when NOTHING of it stands (its server and every disk: the guard never calls a
/// box whose data disk stands "gone"), else its server's status word; empty when monetize laid
/// no box there or has no provider to ask (an operator's estate). Lane G11.
pub fn provider_word(world: &dyn World, instance: &str, role: &str) -> String {
let (Some(cloud), Some((reference, m))) =
(world.cloud(instance), world.laid_machine(instance, role))
else {
return String::new();
};
match monetize_cloud::ask::machine_standing(&cloud, &reference, &m) {
Ok(standing) if standing.is_empty() => "gone".into(),
Ok(standing) => match standing.iter().find(|i| same_resource(&i.id, &m.server)) {
Some(server) => status_word(&monetize_cloud::ask::status(&cloud, server)),
None => format!(
"unknown (the server {} is gone, {} of its laid resource(s) stand)",
m.server,
standing.len()
),
},
Err(e) => format!("unknown ({e})"),
}
}
/// **The provider's word on several boxes of one estate, from ONE listing of the account**
/// (BOX-START: every silent box is asked about, not only the main; PROD-FIX 3 bounds the
/// listings to one per estate per look). Each word as [`provider_word`] says it.
pub fn provider_words(
world: &dyn World,
instance: &str,
roles: &[String],
) -> std::collections::BTreeMap<String, String> {
let mut out = std::collections::BTreeMap::new();
let laid: Vec<(String, String, LaidMachine)> = roles
.iter()
.filter_map(|r| {
world
.laid_machine(instance, r)
.map(|(reference, m)| (r.clone(), reference, m))
})
.collect();
let Some(cloud) = world.cloud(instance) else {
return out;
};
if laid.is_empty() {
return out;
}
let listed = monetize_cloud::ask::account(&cloud);
for (role, reference, m) in laid {
let word = match &listed {
Err(e) => format!("unknown ({e})"),
Ok(_) if reference.trim().is_empty() => "unknown (a machine is looked up under its estate's reference; an empty one matches nothing)".into(),
Ok(inv) => {
let mut standing: Vec<monetize_cloud::ResourceId> = Vec::new();
for id in m.laid_ids() {
for r in inv.owned.iter().filter(|r| r.kind != monetize_cloud::OwnedKind::Dataset && r.reference.as_deref() == Some(reference.as_str()) && same_resource(&r.id.id, id)) {
if !standing.contains(&r.id) {
standing.push(r.id.clone());
}
}
}
match standing.iter().find(|i| same_resource(&i.id, &m.server)) {
_ if standing.is_empty() => "gone".into(),
Some(server) => status_word(&monetize_cloud::ask::status(&cloud, server)),
None => format!("unknown (the server {} is gone, {} of its laid resource(s) stand)", m.server, standing.len()),
}
}
};
out.insert(role, word);
}
out
}
fn status_word(r: &Result<ResourceStatus, CloudError>) -> String {
match r {
Ok(ResourceStatus::Active) => "active".into(),
Ok(ResourceStatus::Provisioning) => "provisioning".into(),
Ok(ResourceStatus::Stopped) => "stopped".into(),
Ok(ResourceStatus::Gone) | Err(CloudError::UnknownResource(_)) => "gone".into(),
Err(e) => format!("unknown ({e})"),
}
}
impl Guardian {
pub fn new(store: Arc<Store>, keys: Arc<Keyring>, cfg: GuardianConfig) -> Self {
Self {
store,
keys,
cfg,
looks: Default::default(),
}
}
/// **An order for one box** of a product's estate, valid `ttl_ms` from `now`; unsigned,
/// with no seq — [`Guardian::queue_order`] gives it both.
pub fn order(
product: &str,
instance: &str,
role: &str,
kind: &str,
body: String,
now_unix_ms: u64,
ttl_ms: u64,
) -> SignedOrder {
SignedOrder {
seq: 0,
product: product.into(),
instance: instance.into(),
role: role.into(),
kind: kind.into(),
body,
issued_unix_ms: now_unix_ms,
not_after_unix_ms: now_unix_ms + ttl_ms,
signature: Vec::new(),
}
}
/// **Queue `order` signed**; every unacked order `stale` names is dropped with it
/// (newest wins).
pub fn queue_order(
&self,
order: SignedOrder,
stale: impl Fn(&SignedOrder) -> bool,
) -> Result<SignedOrder, StoreError> {
self.store.queue_superseding(order, stale, |o| {
self.keys.product(&o.product).sign_order(o)
})
}
/// **Queue `order` signed, once per `mark`** ([`Store::queue_marked`]).
pub fn queue_order_marked(
&self,
order: SignedOrder,
mark: &str,
) -> Result<SignedOrder, StoreError> {
self.store
.queue_marked(order, mark, |o| self.keys.product(&o.product).sign_order(o))
}
/// **The estate as reported**: one [`BoxView`] per registered box, a silent main's
/// provider word as it was LAST looked at ([`Self::refresh_looks`]) — the provider is
/// never asked here (PROD-FIX 1/3: the console's Overview read every estate's view, and
/// each silent main cost a listing of the whole account at UpCloud, seconds each).
pub fn view(&self, world: &dyn World, instance: &str, now_unix_ms: u64) -> EstateView {
self.view_with(world, instance, now_unix_ms, false)
}
/// [`Self::view`] with every silent main's provider word asked NOW (and kept): for a
/// decision a person asked for, never for a page or a tick.
pub fn view_now(&self, world: &dyn World, instance: &str, now_unix_ms: u64) -> EstateView {
self.view_with(world, instance, now_unix_ms, true)
}
/// Is the box in `role` silent at `now`?
fn silent_at(&self, instance: &str, role: &str, now_unix_ms: u64) -> bool {
let rec = self.store.box_record(instance, role).unwrap_or_default();
rec.heard_unix_ms == 0
|| now_unix_ms.saturating_sub(rec.heard_unix_ms) > self.cfg.silent_after_ms
}
/// Does the provider have a word on the box in `role`: any box, silent, that monetize laid
/// in a cloud it links (BOX-START: a silent twin or front is asked too; a stopped one is
/// started, never fenced). Nothing is asked: what monetize holds of it.
pub(crate) fn has_provider_word(
&self,
world: &dyn World,
instance: &str,
role: &str,
now_unix_ms: u64,
) -> bool {
self.silent_at(instance, role, now_unix_ms)
&& world.cloud(instance).is_some()
&& world.laid_machine(instance, role).is_some()
}
/// **What the provider last said of the box in `role`**, kept ([`ProviderLook`]).
pub fn look_of(&self, instance: &str, role: &str) -> Option<ProviderLook> {
self.looks
.lock()
.unwrap_or_else(|e| e.into_inner())
.get(&(instance.to_string(), role.to_string()))
.cloned()
}
/// Keep `word` as the provider's word on the box in `role` (a start that found it gone).
pub(crate) fn keep_look(&self, instance: &str, role: &str, word: &str, now_unix_ms: u64) {
let mut looks = self.looks.lock().unwrap_or_else(|e| e.into_inner());
let l = looks
.entry((instance.to_string(), role.to_string()))
.or_default();
l.word = word.into();
l.looked_unix_ms = now_unix_ms.max(1);
}
/// **Ask the provider about the box in `role` now** and keep its word.
pub fn look_now(
&self,
world: &dyn World,
instance: &str,
role: &str,
now_unix_ms: u64,
) -> String {
self.look_now_all(world, instance, &[role.to_string()], now_unix_ms)
.remove(role)
.unwrap_or_default()
}
/// **Ask the provider about several boxes of one estate now**, in one listing
/// ([`provider_words`]), and keep each word.
pub fn look_now_all(
&self,
world: &dyn World,
instance: &str,
roles: &[String],
now_unix_ms: u64,
) -> std::collections::BTreeMap<String, String> {
let words = provider_words(world, instance, roles);
let mut looks = self.looks.lock().unwrap_or_else(|e| e.into_inner());
for (role, word) in &words {
let l = looks
.entry((instance.to_string(), role.clone()))
.or_default();
l.word = word.clone();
l.looked_unix_ms = now_unix_ms;
l.asked_unix_ms = l.asked_unix_ms.max(now_unix_ms);
}
words
}
/// **The looks due at `now`**: every silent main monetize laid whose provider word is
/// older than [`GuardianConfig::provider_look_every_ms`] (or was never asked). Nothing is
/// asked here.
pub fn looks_due(&self, world: &dyn World, now_unix_ms: u64) -> Vec<(String, String)> {
let every = self.cfg.provider_look_every_ms.max(1);
let looks = self.looks.lock().unwrap_or_else(|e| e.into_inner()).clone();
let mut due = Vec::new();
for (instance, _) in self.store.estates() {
for role in self.store.roles_of(&instance) {
// SEC-S1: an estate monetize holds stopped is not asked about (nothing is
// started, fenced or laid of it).
if !self.has_provider_word(world, &instance, &role, now_unix_ms)
|| world.estate_held(&instance).is_some()
{
continue;
}
let asked = looks
.get(&(instance.clone(), role.clone()))
.map(|l| l.asked_unix_ms)
.unwrap_or(0);
if asked == 0 || now_unix_ms >= asked.saturating_add(every) {
due.push((instance.clone(), role));
}
}
}
due
}
/// **Ask the provider for every look due** ([`Self::looks_due`]), each marked asked first
/// (a look that fails is not asked again before its interval either). Answers how many.
pub fn refresh_looks(&self, world: &dyn World, now_unix_ms: u64) -> usize {
let due = self.looks_due(world, now_unix_ms);
{
let mut looks = self.looks.lock().unwrap_or_else(|e| e.into_inner());
for key in &due {
looks.entry(key.clone()).or_default().asked_unix_ms = now_unix_ms;
}
}
let mut by_estate: std::collections::BTreeMap<&str, Vec<String>> =
std::collections::BTreeMap::new();
for (instance, role) in &due {
by_estate
.entry(instance.as_str())
.or_default()
.push(role.clone());
}
for (instance, roles) in by_estate {
self.look_now_all(world, instance, &roles, now_unix_ms);
}
due.len()
}
fn view_with(
&self,
world: &dyn World,
instance: &str,
now_unix_ms: u64,
now: bool,
) -> EstateView {
let mut v = EstateView {
product: self.store.product_of(instance).unwrap_or_default(),
instance: instance.into(),
now_unix_ms,
silent_after_ms: self.cfg.silent_after_ms,
..Default::default()
};
let roles = self.store.roles_of(instance);
let asked: Vec<String> = roles
.iter()
.filter(|r| self.has_provider_word(world, instance, r, now_unix_ms))
.cloned()
.collect();
let mut words_now = if now && !asked.is_empty() {
self.look_now_all(world, instance, &asked, now_unix_ms)
} else {
Default::default()
};
for role in roles {
let rec = self.store.box_record(instance, &role).unwrap_or_default();
// The guard's R1 (lane G11, Gunnar's `rebuild::judge`): a silent main is judged on
// what the PROVIDER says of it, never on its silence. Its word is the one last
// looked at (unknown before the first look), or asked now when a person asks.
let at_provider = if !self.has_provider_word(world, instance, &role, now_unix_ms) {
String::new()
} else if now {
words_now.remove(&role).unwrap_or_default()
} else {
self.look_of(instance, &role)
.filter(|l| l.looked_unix_ms > 0)
.map(|l| l.word)
.unwrap_or_else(|| NOT_LOOKED_YET.to_string())
};
v.boxes.insert(
role.clone(),
BoxView {
role: role.clone(),
heard_unix_ms: rec.heard_unix_ms,
build: rec.report.build.clone(),
health: rec.report.health.clone(),
twin_review: rec.report.twin_review.clone().unwrap_or_default(),
acks: rec.acks.clone(),
laid_ipv4: world.laid_ipv4(instance, &role),
at_provider,
},
);
}
v
}
/// Open a case for `finding` (detected), numbered `<instance>-<n>`.
pub(crate) fn open(
&self,
instance: &str,
product: &str,
finding: Finding,
now_unix_ms: u64,
) -> Result<Case, StoreError> {
let n = self.store.next(&format!("case/{instance}"))?;
let mut case = Case {
id: format!("{instance}-{n}"),
instance: instance.into(),
product: product.into(),
state: CaseState::Detected,
opened_unix_ms: now_unix_ms,
finding,
..Default::default()
};
case.say(now_unix_ms, format!("detected: {}", case.finding.words));
for ev in case.finding.evidence.clone() {
case.say(now_unix_ms, format!("evidence: {ev}"));
}
self.store.put_case(&case)?;
Ok(case)
}
/// **One pass over every estate.** Answers a line per case that moved.
pub fn tick(&self, world: &dyn World, now_unix_ms: u64) -> Vec<String> {
let mut out = Vec::new();
if self.cfg.looks_in_tick {
self.refresh_looks(world, now_unix_ms);
}
for (instance, product) in self.store.estates() {
let Some(guardian) = world.guardian(&product) else {
continue;
};
let before = self.store.open_case(&instance);
// ★ SEC-S1: an estate monetize stopped on purpose (its term) is left alone: nothing
// started, diagnosed, fenced or laid. A case nobody approved (or a start) is closed
// once, saying why; an approved one waits for the estate to be released.
if let Some(why) = world.estate_held(&instance) {
if let Some(mut c) = before.filter(|c| {
matches!(c.state, CaseState::Detected | CaseState::Investigating)
|| c.finding.code == crate::start::STOPPED_AT_PROVIDER
}) {
c.state = CaseState::Resolved;
c.outcome = format!("monetize holds the estate stopped on purpose ({why}): nothing is started or rebuilt");
c.say(now_unix_ms, c.outcome.clone());
if self.store.put_case(&c).is_ok() {
out.extend(self.write_receipt(&c).err());
out.push(format!("{} resolved: {}", c.id, c.outcome));
}
}
continue;
}
// ★ Start, don't rebuild (BOX-START): a box stopped at the provider is started
// before any product's remedy runs; a case only detected is superseded.
if let Some((case, lines)) =
self.start_pass(world, &instance, &product, before.as_ref(), now_unix_ms)
{
out.extend(lines);
let after = self.advance(world, guardian, case, now_unix_ms);
match self.store.put_case(&after) {
Ok(()) if !after.state.open() => out.extend(self.write_receipt(&after).err()),
Ok(()) => {}
Err(err) => out.push(format!("{} ALERT guardian.case-unsaved the case could not be saved ({err}): the next tick moves it again from what the store holds", after.id)),
}
out.push(format!(
"{} {}: {}",
after.id,
after.state.as_str(),
after.log.last().map(|l| l.words.as_str()).unwrap_or("")
));
out.extend(self.after_failed_start(world, &after, now_unix_ms));
continue;
}
let moved = match before.clone() {
None => {
let view = self.view(world, &instance, now_unix_ms);
match guardian.diagnose(&view).into_iter().next() {
Some(f) => self.open(&instance, &product, f, now_unix_ms).ok(),
None => None,
}
}
Some(case) => {
let after = self.advance(world, guardian, case, now_unix_ms);
match self.store.put_case(&after) {
// Closed, and kept: its receipt. Unsaved, it is not closed: the store
// still holds it open and the next tick moves it again.
Ok(()) if !after.state.open() => out.extend(self.write_receipt(&after).err()),
Ok(()) => {}
Err(err) => out.push(format!("{} ALERT guardian.case-unsaved the case could not be saved ({err}): the next tick moves it again from what the store holds", after.id)),
}
out.extend(self.after_failed_start(world, &after, now_unix_ms));
Some(after)
}
};
if let Some(c) = moved {
if before
.as_ref()
.is_none_or(|b| b.state != c.state || b.log.len() != c.log.len())
{
out.push(format!(
"{} {}: {}",
c.id,
c.state.as_str(),
c.log.last().map(|l| l.words.as_str()).unwrap_or("")
));
}
}
}
out
}
/// A closed case's receipt, kept in `receipts_dir` (none configured: nothing kept). `Err`
/// is the tick's ALERT line: the case stays closed in the store, its receipt file is missing.
pub(crate) fn write_receipt(&self, case: &Case) -> Result<(), String> {
let Some(dir) = &self.cfg.receipts_dir else {
return Ok(());
};
let at = dir.join(format!("{}.json", case.id));
std::fs::create_dir_all(dir)
.and_then(|()| std::fs::write(&at, case.receipt_json()))
.map_err(|e| format!("{} ALERT guardian.receipt-unwritten the closed case's receipt {} could not be written ({e}): the store holds the case", case.id, at.display()))
}
/// Move one open case as far as it goes this tick.
fn advance(
&self,
world: &dyn World,
guardian: &dyn EstateGuardian,
mut case: Case,
now: u64,
) -> Case {
let instance = case.instance.clone();
if case.state == CaseState::Detected {
let role = case.finding.role.clone();
let word = match (world.cloud(&instance), world.laid_machine(&instance, &role)) {
(Some(cloud), Some((reference, m))) => match monetize_cloud::ask::machine_standing(&cloud, &reference, &m) {
Ok(standing) if standing.is_empty() => format!("the provider says the {role} box {} is gone, and all {} of its laid resource(s)", m.server, m.laid_ids().len()),
Ok(standing) => {
let server = standing.iter().find(|i| same_resource(&i.id, &m.server)).map(|i| status_word(&monetize_cloud::ask::status(&cloud, i))).unwrap_or_else(|| "gone".into());
format!("the provider says the {role} box {} is {server}; {} of its {} laid resource(s) stand", m.server, standing.len(), m.laid_ids().len())
}
Err(e) => format!("the provider could not be asked about the {role} box {}: {e}", m.server),
},
(None, _) => format!("no provider to ask about the {role} box"),
(_, None) => format!("the provider has no {role} box laid by monetize"),
};
case.say(now, word);
case.state = CaseState::Investigating;
return case;
}
if case.state == CaseState::Investigating {
let view = self.view(world, &instance, now);
let found = guardian
.diagnose(&view)
.into_iter()
.find(|f| f.code == case.finding.code);
let still = found.is_some();
let pressed_or_asked =
case.finding.code == "pressed" || case.finding.code.starts_with("asked:");
// Auto-fix takes what the product says NOW (lane G11): a main judged standing when
// the case opened may be gone at the provider by now, and the other way round.
let auto = if pressed_or_asked {
case.finding.auto
} else {
found.and_then(|f| f.auto)
};
if !still && !pressed_or_asked {
case.state = CaseState::Resolved;
case.outcome = "the estate recovered on its own before anyone pressed".into();
case.say(now, case.outcome.clone());
return case;
}
match auto {
Some(remedy) if self.store.auto_fix(&instance) => {
self.approve(&mut case, remedy, AUTO_PRINCIPAL, true, now);
}
_ => return case,
}
}
if case.state == CaseState::Approved {
let Some(remedy) = case.remedy else {
return self.fail(case, now, "approved with no remedy");
};
// The plan is made ONCE per approved case: on what the provider says now.
let view = self.view_now(world, &instance, now);
match guardian.plan(remedy, &view) {
Ok(steps) => {
case.epoch = self.store.next(&format!("epoch/{instance}")).unwrap_or(0);
case.steps = steps
.into_iter()
.map(|step| CaseStep {
step,
state: step_state::PENDING.into(),
started_unix_ms: 0,
done_unix_ms: 0,
words: String::new(),
seq: 0,
asked: Vec::new(),
fenced: String::new(),
})
.collect();
if let Some(why) = unfenced_promotion(&case.steps) {
return self.fail(
case,
now,
format!(
"the product's plan for {} is refused: {why}",
remedy.as_str()
),
);
}
case.state = CaseState::Running;
let plan: Vec<String> = case.steps.iter().map(|s| s.step.words()).collect();
case.say(
now,
format!(
"running {} (epoch {}): {}",
remedy.as_str(),
case.epoch,
plan.join(" → ")
),
);
}
Err(why) => {
return self.fail(
case,
now,
format!("the product will not plan {}: {why}", remedy.as_str()),
)
}
}
}
if case.state == CaseState::Running {
while let Some(i) = case.steps.iter().position(|s| s.state != step_state::MET) {
if case.steps[i].state == step_state::PENDING {
case.steps[i].state = step_state::RUNNING.into();
case.steps[i].started_unix_ms = now;
}
match self.step(world, guardian, &mut case, i, now) {
Moved::Met(words) => {
case.steps[i].state = step_state::MET.into();
case.steps[i].done_unix_ms = now;
case.steps[i].words = words.clone();
case.say(
now,
format!("met: {} — {words}", case.steps[i].step.words()),
);
}
Moved::Waiting => return case,
Moved::Failed(why) => {
case.steps[i].state = step_state::FAILED.into();
case.steps[i].done_unix_ms = now;
case.steps[i].words = why.clone();
let what = format!("{}: {why}", case.steps[i].step.words());
return self.fail(case, now, what);
}
}
}
case.state = CaseState::Done;
case.outcome = match case.remedy {
Some(r) => format!("{} done", r.as_str()),
None if case.finding.code == crate::start::STOPPED_AT_PROVIDER => "start done: every stopped box started, and each but the front heard again; nothing rebuilt".into(),
None => "? done".into(),
};
case.say(now, case.outcome.clone());
}
case
}
fn fail(&self, mut case: Case, now: u64, why: impl Into<String>) -> Case {
case.state = CaseState::Failed;
case.outcome = why.into();
case.say(now, format!("failed: {}", case.outcome));
case
}
fn approve(&self, case: &mut Case, remedy: Remedy, principal: &str, auto: bool, now: u64) {
case.state = CaseState::Approved;
case.remedy = Some(remedy);
case.approved_by = principal.into();
case.approved_unix_ms = now;
case.auto = auto;
case.say(now, format!("approved: {} by {principal}", remedy.words()));
}
/// **The fence of a box monetize did not lay** (main rebuild, owner 2026-10-12): an
/// estate whose cloud is its operator's — gunnar.rs, test2. Monetize holds no laid record
/// and no token there, so it lays nothing and gives nothing back. Its operator rebuilds
/// the box (`cargo xtask estate rebuild main --apply`: gunnar's guard decides whether the
/// old disk may go) and the kit sends the owner's signed installed-notice naming the role
/// laid again (PIN-1). That notice, newer than this case (or at most `fence_within_ms`
/// before it opened), is this fence's proof; until it comes the case waits and says what
/// the operator runs, for up to [`OPERATOR_RELAY_WITHIN_MS`].
fn fence_by_operator(&self, case: &mut Case, i: usize, role: &str, now: u64) -> Moved {
let notice = self.store.notice_taken(&case.instance, role);
if notice > 0 && notice >= case.opened_unix_ms.saturating_sub(self.cfg.fence_within_ms) {
case.steps[i].fenced = format!("{OPERATOR_FENCE} at {notice}: the owner's signed installed-notice names a {role} laid again (case {})", case.id);
return Moved::Met(format!("the {role} was rebuilt by the estate's operator: the owner's signed notice at {notice} names it laid again"));
}
if case.steps[i].asked.is_empty() {
case.steps[i].asked = vec![OPERATOR_FENCE.to_string()];
case.say(
now,
format!(
"monetize laid no {role} box for {}: its cloud is its operator's. The operator rebuilds it — `cargo xtask estate rebuild {role} --apply` (the product's guardian decides whether the old disk may go) — and the case goes on once the owner's notice names the new {role}",
case.instance
),
);
}
if now.saturating_sub(case.steps[i].started_unix_ms) > OPERATOR_RELAY_WITHIN_MS {
return Moved::Failed(format!("the estate's operator did not lay a new {role} within {} h (no owner's notice): nothing was laid here", OPERATOR_RELAY_WITHIN_MS / 3_600_000));
}
Moved::Waiting
}
/// One step, once.
fn step(
&self,
world: &dyn World,
guardian: &dyn EstateGuardian,
case: &mut Case,
i: usize,
now: u64,
) -> Moved {
let instance = case.instance.clone();
let started = case.steps[i].started_unix_ms;
match case.steps[i].step.clone() {
Step::Cloud(CloudStep::Fence { role }) => {
// **The whole box given back** (gap-2): server, disks and clone, asked for ONCE;
// then **read back gone** from the provider's listing, tick by tick (gap-3: a
// provider may delete slowly), inside `fence_within_ms`.
let Some((reference, m)) = world.laid_machine(&instance, &role) else {
return self.fence_by_operator(case, i, &role, now);
};
let Some(cloud) = world.cloud(&instance) else {
return Moved::Failed("no provider for the estate".into());
};
// ★ BOX-START: a box only STOPPED at the provider keeps its disks: it is started,
// never given back.
if case.steps[i].asked.is_empty()
&& provider_word(world, &instance, &role) == "stopped"
{
return Moved::Failed(format!("the {role} box {} is only stopped at the provider (its disks stand): it is started, never fenced; nothing was given back", m.server));
}
if case.steps[i].asked.is_empty() {
match monetize_cloud::ask::retire_machine(&cloud, &reference, &m) {
Ok(gone) if gone.is_empty() => {
// H24: a box gone at the provider whose key stays pinned is no fence.
if let Err(err) = self.store.forget_box(&instance, &role) {
case.say(now, format!("the {role} box {} is gone at the provider, but monetize could not forget the old {role}'s key ({err}): the next tick tries again", m.server));
return Moved::Waiting;
}
case.steps[i].fenced = format!("the old {role} box {} read back gone at the provider at {now} (case {})", m.server, case.id);
return Moved::Met(format!("the {role} box {} is already gone, with all {} of its laid resource(s)", m.server, m.laid_ids().len()));
}
Ok(gone) => {
// Not marked asked until the key is forgotten: the next tick asks again.
if let Err(err) = self.store.forget_box(&instance, &role) {
case.say(now, format!("asked the provider to give the {role} box {} back, but monetize could not forget the old {role}'s key ({err}): the next tick asks again", m.server));
return Moved::Waiting;
}
case.steps[i].asked = gone.iter().map(|i| i.id.clone()).collect();
case.say(
now,
format!(
"asked the provider to give the {role} box {} back whole: {}",
m.server,
case.steps[i].asked.join(", ")
),
);
}
Err(e) => {
return Moved::Failed(format!(
"the fence of the {role} box {} is not met: {e}",
m.server
))
}
}
}
let asked = case.steps[i].asked.join(", ");
let left = match monetize_cloud::ask::machine_standing(&cloud, &reference, &m) {
Ok(left) if left.is_empty() => {
case.steps[i].fenced = format!("the old {role} box {} given back whole and read back gone at the provider at {now} (case {})", m.server, case.id);
return Moved::Met(format!("the {role} box {} given back whole at the provider: {asked} — read back as gone", m.server));
}
Ok(left) => left
.iter()
.map(|i| i.id.clone())
.collect::<Vec<_>>()
.join(", "),
Err(e) => format!("(the provider could not be read: {e})"),
};
if now.saturating_sub(started) > self.cfg.fence_within_ms {
Moved::Failed(format!("the fence of the {role} box {} is not met: not read back gone within {} s: {left} still stand(s)", m.server, self.cfg.fence_within_ms / 1000))
} else {
Moved::Waiting
}
}
Step::Cloud(CloudStep::Lay { role }) => {
if promotes(&case.steps, i, &role)
&& confirmed_fence(&case.steps, i, &role).is_none()
{
return Moved::Failed(format!(
"no new {role} is promoted: the old {role}'s fence is not confirmed (monetize holds no laid record of it to give back and read back gone) — a box nobody fenced may still take writes; a human stops it at the provider first"
));
}
// ★ An estate whose cloud is its OPERATOR's (gunnar.rs, test2): monetize lays
// nothing there. Its operator laid the box again — the fence before this step
// is the owner's signed notice ([`Guardian::fence_by_operator`]); the new box
// pins its key in the window that notice opened (PIN-1).
if world.laid_machine(&instance, &role).is_none() {
if let Some(f) = confirmed_fence(&case.steps, i, &role)
.filter(|f| f.starts_with(OPERATOR_FENCE))
{
return Moved::Met(format!("the {role} was laid again by the estate's operator ({f}); its new key pins in the window the owner's notice opened"));
}
}
if let Some((reference, m)) = world.laid_machine(&instance, &role) {
let standing = match world.cloud(&instance).map(|c| monetize_cloud::ask::machine_standing(&c, &reference, &m)) {
Some(Ok(s)) => s,
Some(Err(e)) => return Moved::Failed(format!("no lay while the old {role} box {} cannot be looked up ({e}): fence it first (the fence invariant)", m.server)),
None => return Moved::Failed("no provider for the estate".into()),
};
if !standing.is_empty() {
return Moved::Failed(format!("no lay while the old {role} box {} stands ({} of its resource(s)): fence it first (the fence invariant)", m.server, standing.len()));
}
}
let laid = match world.lay_one(&instance, &role) {
Ok(l) => l,
Err(e) => return Moved::Failed(format!("lay {role}: {e}")),
};
let keyed = if laid.appliance_key.is_empty() {
if let Err(err) = self.store.forget_box(&instance, &role) {
return Moved::Failed(format!(
"forget the old {role}'s key before the new one pins: {err}"
));
}
match self
.store
.open_pin(&instance, &role, now + self.cfg.pin_window_ms)
{
Ok(()) => "its key is pinned on its first call home".to_string(),
Err(e) => return Moved::Failed(format!("open the pin window: {e}")),
}
} else {
match self
.store
.register_key(&instance, &role, &laid.appliance_key)
{
Ok(()) => "its key from the installer's verdict".to_string(),
Err(e) => {
return Moved::Failed(format!("register the new {role}'s key: {e}"))
}
}
};
Moved::Met(format!(
"server {} at {} — {}; {keyed}",
laid.server, laid.ipv4, laid.words
))
}
Step::Cloud(CloudStep::Restore { role }) => {
match world.restore_one(&instance, &role, &case.approved_by) {
Ok(words) => Moved::Met(words),
Err(e) => Moved::Failed(format!("restore {role}: {e}")),
}
}
Step::Await { role, what, .. } if what == crate::start::START_WHAT => {
match self.start_one(world, &instance, &role, now) {
Ok(words) => Moved::Met(words),
Err(why) => Moved::Failed(why),
}
}
Step::Await {
role,
what,
within_ms,
} => {
if what == "heard" {
let heard = self
.store
.box_record(&instance, &role)
.map(|r| r.heard_unix_ms)
.unwrap_or(0);
if heard > started {
return Moved::Met(format!("the {role} called home"));
}
} else {
match guardian.observe(&role, &what, &self.view(world, &instance, now)) {
StepState::Met(w) => return Moved::Met(w),
StepState::Failed(w) => return Moved::Failed(w),
StepState::Waiting(_) => {}
}
}
if now.saturating_sub(started) > within_ms {
Moved::Failed(format!(
"the {role} did not show {what} within {} s",
within_ms / 1000
))
} else {
Moved::Waiting
}
}
Step::Order { role, kind, body } => {
// ★ Main rebuild: a re-laid main whose estate runs auto-fix fills ITSELF from its
// twin. A recharge order is then not queued (or not held against it): the box
// reporting its data whole since the case opened is the recharge met.
// Only into a box this case laid (or saw laid by its operator): a recharge
// pressed on a main that serves is refused by the box itself, as before.
let is_recharge = kind == monetize_embed::fleet::kind::RECHARGE_FROM_TWIN && case.steps[..i].iter().any(|s| matches!(&s.step, Step::Cloud(CloudStep::Lay { role: r }) if *r == role));
let opened = case.opened_unix_ms;
let filled_itself = || {
let heard = self
.store
.box_record(&instance, &role)
.map(|r| r.heard_unix_ms)
.unwrap_or(0);
heard > opened
&& matches!(
guardian.observe(&role, "recharged", &self.view(world, &instance, now)),
StepState::Met(_)
)
};
if is_recharge && case.steps[i].seq == 0 && filled_itself() {
return Moved::Met(format!("the {role} reports its data whole already — it filled itself from the twin (auto-fix): no recharge queued"));
}
if case.steps[i].seq == 0 {
let mut json = serde_json::Map::new();
for (k, v) in &body {
let value = match self.fill(world, case, v, now) {
Ok(v) => v,
Err(why) => return Moved::Failed(why),
};
json.insert(k.clone(), value);
}
json.insert("case".into(), serde_json::Value::String(case.id.clone()));
// A recharge into a box this case laid PROMOTES it: it carries the old
// box's confirmed fence, or it is not queued.
let laid_here = case.steps[..i].iter().any(|s| matches!(&s.step, Step::Cloud(CloudStep::Lay { role: r }) if *r == role));
if kind == monetize_embed::fleet::kind::RECHARGE_FROM_TWIN && laid_here {
match confirmed_fence(&case.steps, i, &role) {
Some(proof) => {
json.insert("fenced".into(), serde_json::Value::String(proof));
}
None => return Moved::Failed(format!("no promotion of the new {role}: the old {role}'s fence is not confirmed")),
}
}
let order = SignedOrder {
seq: 0,
product: case.product.clone(),
instance: instance.clone(),
role: role.clone(),
kind: kind.clone(),
body: serde_json::Value::Object(json).to_string(),
issued_unix_ms: now,
not_after_unix_ms: now + self.cfg.order_ttl_ms,
signature: Vec::new(),
};
match self
.store
.queue(order, |o| self.keys.product(&o.product).sign_order(o))
{
Ok(q) => {
case.steps[i].seq = q.seq;
case.say(
now,
format!(
"queued signed order #{} {kind} for the {role}; it pulls it",
q.seq
),
);
}
Err(e) => return Moved::Failed(format!("queue the order: {e}")),
}
}
let seq = case.steps[i].seq;
if let Some(ack) = self
.store
.box_record(&instance, &role)
.and_then(|r| r.acks.into_iter().rev().find(|a| a.seq == seq))
{
if ack.ok {
return Moved::Met(format!("the {role} acked #{seq}: {}", ack.detail));
}
// A recharge refused because the box is filling itself (or already did):
// met once it reports its data whole, waited for while it says it runs.
if is_recharge && filled_itself() {
return Moved::Met(format!("the {role} refused #{seq} ({}) and reports its data whole: it filled itself from the twin", ack.detail));
}
if !(is_recharge && ack.detail.starts_with("already_running")) {
return Moved::Failed(format!("the {role} refused #{seq}: {}", ack.detail));
}
}
if now.saturating_sub(started) > self.cfg.order_within_ms {
Moved::Failed(format!(
"the {role} did not ack #{seq} within {} s",
self.cfg.order_within_ms / 1000
))
} else {
Moved::Waiting
}
}
}
}
/// A body value: `$addr:<role>`, `$epoch`, `$principal`, `$entitlements` (a signed
/// `Snapshot` of [`World::entitlements`], issued `now`); `true`/`false` and digits typed.
fn fill(
&self,
world: &dyn World,
case: &Case,
v: &str,
now: u64,
) -> Result<serde_json::Value, String> {
use serde_json::Value;
if v == "$entitlements" {
let facts = world
.entitlements(&case.product, &case.instance)
.map_err(|e| format!("the books for {}: {e}", case.product))?;
let snapshot = self.keys.product(&case.product).sign_snapshot(now, facts);
return serde_json::to_value(&snapshot).map_err(|e| format!("the snapshot: {e}"));
}
if let Some(role) = v.strip_prefix("$addr:") {
let reported = self
.store
.box_record(&case.instance, role)
.and_then(|r| r.report.health.get("serves_at").cloned())
.unwrap_or_default();
let addr = if reported.is_empty() {
world.laid_ipv4(&case.instance, role)
} else {
reported
};
if addr.is_empty() {
return Err(format!("no address known for the {role}"));
}
return Ok(Value::String(addr));
}
Ok(match v {
"$epoch" => Value::from(case.epoch),
"$principal" => Value::String(case.approved_by.clone()),
"true" => Value::Bool(true),
"false" => Value::Bool(false),
n if !n.is_empty() && n.bytes().all(|b| b.is_ascii_digit()) => n
.parse::<u64>()
.map(Value::from)
.unwrap_or_else(|_| Value::String(n.into())),
s => Value::String(s.into()),
})
}
/// **The button.** `strong`: the caller proved a fresh, user-verified session for
/// `principal`; anything less is refused. Approves the estate's open case with `remedy`,
/// or opens one approved at once when nothing was detected.
#[allow(clippy::too_many_arguments)]
pub fn press(
&self,
world: &dyn World,
instance: &str,
remedy: Remedy,
case_id: &str,
principal: &str,
strong: bool,
now_unix_ms: u64,
) -> Result<Case, PressRefusal> {
if !strong || principal.trim().is_empty() {
return Err(PressRefusal::Weak(
"a button needs a named principal in a fresh, user-verified session".into(),
));
}
let Some(product) = self.store.product_of(instance) else {
return Err(PressRefusal::Unknown(format!("no estate {instance:?}")));
};
if world.guardian(&product).is_none() {
return Err(PressRefusal::Refused(format!(
"the product {product:?} has no guardian"
)));
}
if let Some(why) = world.estate_held(instance) {
return Err(PressRefusal::Refused(format!("monetize holds {instance} stopped on purpose ({why}): nothing is started or rebuilt until it is released")));
}
let mut case = match self.store.open_case(instance) {
Some(c) if matches!(c.state, CaseState::Approved | CaseState::Running) => {
return Err(PressRefusal::Busy(format!(
"case {} is already {}",
c.id,
c.state.as_str()
)));
}
Some(c) => {
if !case_id.is_empty() && case_id != c.id {
return Err(PressRefusal::Unknown(format!(
"case {case_id:?} is not the estate's open case ({})",
c.id
)));
}
c
}
None => {
if !case_id.is_empty() {
return Err(PressRefusal::Unknown(format!(
"case {case_id:?} is not open"
)));
}
let f = Finding {
code: "pressed".into(),
role: String::new(),
words: format!("{principal} pressed {}", remedy.words()),
evidence: Vec::new(),
remedies: vec![remedy],
auto: None,
};
self.open(instance, &product, f, now_unix_ms)
.map_err(|e| PressRefusal::Refused(e.to_string()))?
}
};
self.approve(&mut case, remedy, principal, false, now_unix_ms);
self.store
.put_case(&case)
.map_err(|e| PressRefusal::Refused(e.to_string()))?;
Ok(case)
}
/// **The estate's auto-fix opt-in** — off by default; turning it either way takes the same
/// strong session as a button.
pub fn set_auto_fix(
&self,
instance: &str,
on: bool,
principal: &str,
strong: bool,
) -> Result<bool, PressRefusal> {
if !strong || principal.trim().is_empty() {
return Err(PressRefusal::Weak(
"the auto-fix switch needs a named principal in a fresh, user-verified session"
.into(),
));
}
if self.store.product_of(instance).is_none() {
return Err(PressRefusal::Unknown(format!("no estate {instance:?}")));
}
self.store
.set_auto_fix(instance, on)
.map_err(|e| PressRefusal::Refused(e.to_string()))?;
Ok(on)
}
/// The estate's cases, newest first.
pub fn cases(&self, instance: &str) -> Vec<Case> {
self.store.cases_of(instance)
}
}