crates/monetize-ui-server/src/estates.rs
//! **The estates this console shows are the FLEET's** (ONE monetize, owner
//! 2026-10-10). No estate runs a monetize of its own, so there is nothing to
//! dial: the console is given ONE address (the ONE monetize) and reads every
//! estate from it. An estate is one the ONE monetize has declared
//! (`Guardian.Cases`: a sold estate at its install, a product's own estate
//! from `[products.<id>] instance`). It is SERVING when one of its boxes has
//! called home within the guardian's silence bound. That is the rule
//! `fleet::heard_since` applies to the same box records.
//!
//! Its customers are ACCOUNTS on the ONE ledger (`Accounts.List`), named by
//! their path ([`account_path`]):
//! * `gunnar/<t>` is a tenant on the product's own estate;
//! * `gunnar/@<company>/<ns>` is a tenant on the sold estate `<company>`;
//! * `holger/<t>` is a tenant on holger's estate.
//!
//! LEGACY: the `--estates` file (`[[estate]] control = "https://monetize.<apex>"`),
//! with one pinned, signed client and one router per estate, belongs to the
//! per-estate monetize. The binary refuses it by name ([`LEGACY_ESTATES_FILE`]).
use monetize_proto as pb;
use serde::Serialize;
use crate::grpc::{Fail, Monetize};
/// The refusal for the old `--estates` file.
pub const LEGACY_ESTATES_FILE: &str = "REFUSED [estates-legacy] --estates is LEGACY: no estate runs a monetize of its own any more, and this console dials none. Give it the ONE monetize (--control) and it reads every estate from the fleet and every customer from the ONE ledger";
/// One estate as the fleet knows it.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
pub struct FleetEstate {
/// The instance the ONE monetize declared it as (a sold estate's company).
pub name: String,
/// The product it serves.
pub product: String,
/// Every box the fleet knows of it, by role.
pub roles: Vec<String>,
/// The boxes that called home within the guardian's bound ([`heard`]).
pub heard: Vec<String>,
/// **The machines**: every box that has reported in at least once
/// ([`reported`]), live from the fleet's own records — a box appears the
/// moment its first report lands, never from a declared or fixed list.
pub reported: Vec<String>,
/// Some machine of it ever reported in. False: the estate is declared and
/// nothing of it is installed yet.
pub installed: bool,
/// Each reported machine's own `serves_at` (its host), by role.
pub hosts: std::collections::BTreeMap<String, String>,
/// Its newest call home; `None` when no box has ever called.
pub last_heard_unix_ms: Option<u64>,
/// Some box called home within the bound.
pub serving: bool,
/// Why it is not serving, in the pull model's words.
pub error: Option<String>,
/// The state of its open guardian case, if one is open.
pub open_case: Option<String>,
}
/// **The roles heard**: `fleet::heard_since(now - silent_after)`, as the
/// guardian words it. A box that never reported is not heard.
pub fn heard(boxes: &[pb::GuardianBox]) -> Vec<String> {
boxes.iter().filter(|b| b.heard_unix_ms > 0 && !b.silent).map(|b| b.role.clone()).collect()
}
/// **The machines**, in role order: a box whose first fleet report has landed
/// (`heard_unix_ms > 0`), silent now or not, or one this box laid.
pub fn reported(boxes: &[pb::GuardianBox]) -> Vec<String> {
// ★ PROD-5 (owner 2026-10-26: "Machines 2" for front, main and twin): a box THIS box laid
// (it holds its laid address) is a machine whether it reports or not — a front does not
// call home. A key only declared is still not one.
let mut r: Vec<String> = boxes.iter().filter(|b| b.heard_unix_ms > 0 || !b.laid_ipv4.trim().is_empty()).map(|b| b.role.clone()).collect();
r.sort_by_key(|role| (ROLE_ORDER.iter().position(|o| o == role).unwrap_or(ROLE_ORDER.len()), role.clone()));
r
}
/// The order an estate's machines are named in.
const ROLE_ORDER: [&str; 3] = ["front", "main", "twin"];
/// The words for an estate with no machine installed yet.
pub const NOT_INSTALLED: &str = "not installed";
/// A case still waiting on something (not done, failed or resolved).
fn open(state: &str) -> bool {
!matches!(state, "done" | "failed" | "resolved")
}
/// **The fleet, pure**: every estate of a `Guardian.Cases` answer, by name.
pub fn fleet(answer: pb::GuardianCasesResponse) -> Vec<FleetEstate> {
let mut out: Vec<FleetEstate> = answer
.estates
.into_iter()
.map(|e| {
let roles: Vec<String> = e.boxes.iter().map(|b| b.role.clone()).collect();
let heard = heard(&e.boxes);
let reported = reported(&e.boxes);
// Installed = some machine of it has reported in, ever. A key declared
// for a box (`fleet_keys`) is not an install (gunnar.rs on the new
// account, 2026-10-12: declared, nothing installed).
let installed = e.boxes.iter().any(|b| b.heard_unix_ms > 0);
let hosts = e.boxes.iter().filter(|b| b.heard_unix_ms > 0).filter_map(|b| b.health.get("serves_at").map(|h| (b.role.clone(), h.clone()))).collect();
let last = e.boxes.iter().map(|b| b.heard_unix_ms).max().filter(|ms| *ms > 0);
let error = heard.is_empty().then(|| {
if !installed {
format!("{} is not installed yet: no machine of it has reported in", e.instance)
} else {
format!("{} is not answering: none of its machines ({}) has reported in lately", e.instance, roles.join(", "))
}
});
FleetEstate {
serving: !heard.is_empty(),
open_case: e.cases.iter().find(|c| open(&c.state)).map(|c| c.state.clone()),
name: e.instance,
product: e.product,
roles,
heard,
reported,
installed,
hosts,
last_heard_unix_ms: last,
error,
}
})
.collect();
out.sort_by(|a, b| a.name.cmp(&b.name));
out
}
/// **Every estate the ONE monetize's fleet knows** (one call to the ONE).
pub async fn read(m: &Monetize) -> Result<Vec<FleetEstate>, Fail> {
Ok(fleet(cases(m, "").await?.answer))
}
/// [`read`], with how old the state it answers is ([`FleetRead::stale_secs`]).
pub async fn read_aged(m: &Monetize) -> Result<(Vec<FleetEstate>, Option<u64>), Fail> {
let r = cases(m, "").await?;
Ok((fleet(r.answer), r.stale_secs))
}
/// **How long a page waits for the fleet** (PROD-4): [`crate::bounded::PAGE_WITHIN`].
pub const FLEET_WITHIN: std::time::Duration = crate::bounded::PAGE_WITHIN;
/// One `Guardian.Cases` read as a page gets it.
#[derive(Debug, Clone, Default)]
pub struct FleetRead {
pub answer: pb::GuardianCasesResponse,
/// `None`: read now. `Some(s)`: the fleet did not answer within [`FLEET_WITHIN`]; this is
/// the state it answered `s` seconds ago.
pub stale_secs: Option<u64>,
}
/// **`Guardian.Cases` for `instance` (empty: every estate), within the page's bound**
/// ([`crate::bounded::within`], PROD-4): one whole read is kept, and past the bound a page is
/// answered from it with its age (filtered to `instance`); a refusal (an estate the guardian
/// does not know) is the refusal.
///
/// RA3-1: the page asks nothing itself. The [`FleetKeeper`]'s jera job reads the fleet (one
/// read at a time, every [`FLEET_EVERY`] and whenever a page asks), and the page answers what
/// it brings within the bound, else what it kept. With no runtime to keep it on, the page's
/// own single flight reads it.
pub async fn cases(m: &Monetize, instance: &str) -> Result<FleetRead, Fail> {
let (mut answer, stale_secs) = match m.fleet_keeper.lane(m) {
Ok(lane) => crate::bounded::within_kept(&m.fleet_seen, "", "the fleet", FLEET_WITHIN, || {
FleetKeeper::ask(&lane);
})
.await?,
Err(_) => crate::bounded::within(&m.fleet_seen, "", "the fleet", fleet_call(m)).await?,
};
if !instance.is_empty() {
answer.estates.retain(|e| e.instance == instance);
if answer.estates.is_empty() {
return Err(Fail { code: "notfound".into(), message: format!("the guardian knows no estate {instance:?}") });
}
}
Ok(FleetRead { answer, stale_secs })
}
/// **`Estates.Overview` — the estates that exist, as the ONE monetize classifies them**
/// (OVW-CLEAN), within the page's bound and kept the way the fleet is ([`cases`]): the
/// [`FleetKeeper`]'s job reads it beside the fleet, and the page answers what it brings within
/// the bound, else what it kept, with its age.
pub async fn overview(m: &Monetize) -> Result<(pb::EstatesOverviewResponse, Option<u64>), Fail> {
match m.fleet_keeper.lane(m) {
Ok(lane) => crate::bounded::within_kept(&m.estates_seen, "", "the estates", FLEET_WITHIN, || {
FleetKeeper::ask(&lane);
})
.await,
Err(_) => crate::bounded::within(&m.estates_seen, "", "the estates", overview_call(m)).await,
}
}
/// One whole `Estates.Overview` read.
fn overview_call(m: &Monetize) -> impl std::future::Future<Output = Result<pb::EstatesOverviewResponse, tonic::Status>> + Send + 'static {
let mut e = m.estates();
async move { e.overview(pb::EstatesOverviewRequest {}).await.map(|a| a.into_inner()) }
}
/// One whole `Guardian.Cases` read (every estate).
fn fleet_call(m: &Monetize) -> impl std::future::Future<Output = Result<pb::GuardianCasesResponse, tonic::Status>> + Send + 'static {
let mut g = m.guardian();
async move { g.cases(pb::GuardianCasesRequest { instance: String::new() }).await.map(|a| a.into_inner()) }
}
/// The jera job kind of the console's fleet read (RA3-1).
pub const FLEET_JOB: &str = "monetize_console_fleet_read";
/// The fleet read's ONE jera key: a second ask while it is queued is the same job.
pub const FLEET_KEY: &str = "console-fleet";
/// How often the keeper reads the fleet when no page asks (RA3-1: the kept fleet is the
/// keeper's, not the pages').
pub const FLEET_EVERY: std::time::Duration = std::time::Duration::from_secs(30);
/// **The jera job that keeps the console's fleet** (RA3-1, owner: "from the jera cache, a jera
/// job refreshes"). One ChainQueue lane per console, started on the first use within a
/// runtime: its job asks the ONE `Guardian.Cases` read under the single flight of
/// [`crate::bounded::Seen::fly`] (a job while the read is out starts nothing), and the lane
/// asks it again every [`FLEET_EVERY`]. Pages read what it kept ([`cases`]).
#[derive(Clone, Default)]
pub struct FleetKeeper(std::sync::Arc<std::sync::OnceLock<std::sync::Arc<jera::jobs::chain::ChainQueue>>>);
impl FleetKeeper {
/// The ONE fleet job, by its stable key.
pub fn spec() -> jera::jobs::chain::ChainSpec {
jera::jobs::chain::ChainSpec::new(FLEET_JOB, FLEET_KEY, "console-fleet", 0, "monetize-console", serde_json::json!({})).with_target("read the ONE monetize's fleet (Guardian.Cases) for the console's pages").with_priority(jera::jobs::chain::Priority::Normal)
}
/// **Ask the keeper for a fresh fleet.** A `Stale` answer cannot come (one key, one
/// generation); it is said if it does.
pub fn ask(lane: &std::sync::Arc<jera::jobs::chain::ChainQueue>) -> jera::jobs::chain::Submitted {
let r = lane.submit(Self::spec());
if let jera::jobs::chain::Submitted::Stale(g) = &r {
tracing::warn!(generation = g, "the console's fleet job was refused as stale");
}
r
}
/// **The keeper's lane**, started on first use: `Err` with no runtime to run it on.
pub fn lane(&self, m: &Monetize) -> Result<std::sync::Arc<jera::jobs::chain::ChainQueue>, String> {
if let Some(q) = self.0.get() {
return Ok(q.clone());
}
let rt = tokio::runtime::Handle::try_current().map_err(|_| "no runtime to keep the console's fleet on".to_string())?;
let sink = jera::jobs::JobSink::remote(|rec: &jera::jobs::JobRecord| {
if rec.status == "failed" {
tracing::error!("ALERT console.fleet-job-failed {}: {}", rec.job_id, rec.detail_json);
}
});
// RA3-8: jera's own re-ask; since jera 0.2.1 a look at the exclusive gate never takes it.
let queue = jera::jobs::chain::ChainQueue::new(sink);
let q = self.0.get_or_init(|| queue.clone()).clone();
if std::sync::Arc::ptr_eq(&q, &queue) {
let (seen, estates_seen, call_of, every) = (m.fleet_seen.clone(), m.estates_seen.clone(), m.clone(), queue.clone());
// A32-14, stays a task: the console fleet lane's jera worker and its clock, ONE task.
rt.spawn(async move {
let clock = async move {
loop {
tokio::time::sleep(FLEET_EVERY).await;
Self::ask(&every);
}
};
let worker = queue.run(move |_spec| {
let started = seen.fly("", fleet_call(&call_of));
// OVW-CLEAN: the Overview's classified estates, kept beside the fleet.
let estates = estates_seen.fly("", overview_call(&call_of));
Ok(serde_json::json!({ "started": started, "estates": estates }))
});
tokio::join!(worker, clock);
});
Self::ask(&q);
}
Ok(q)
}
}
/// **An account's path on the ONE ledger**: `<product>/<tenant>`, e.g.
/// `gunnar/alice`, `gunnar/@acme/ns`, `holger/bob`.
pub fn account_path(product: &str, tenant: &str) -> String {
format!("{product}/{tenant}")
}
/// The sold estate a tenant lives on: `@acme/ns` → `acme`. `None` for a
/// tenant of the product's own estate.
pub fn sold_estate_of(tenant: &str) -> Option<&str> {
let (estate, ns) = tenant.strip_prefix('@')?.split_once('/')?;
(!estate.is_empty() && !ns.is_empty()).then_some(estate)
}
/// Does the tenant's account belong to `estate`? A sold estate holds its
/// `@<estate>/` tenants. A product's own estate holds the product's tenants
/// with no `@`.
pub fn tenant_on(estate: &FleetEstate, sold: bool, tenant: &str) -> bool {
match sold_estate_of(tenant) {
Some(e) => sold && e == estate.name,
None => !sold,
}
}
#[cfg(test)]
mod tests {
use super::*;
fn b(role: &str, heard: u64, silent: bool) -> pb::GuardianBox {
pb::GuardianBox { role: role.into(), heard_unix_ms: heard, silent, ..Default::default() }
}
#[test]
fn an_estate_serves_when_a_box_called_home_within_the_bound() {
let answer = pb::GuardianCasesResponse {
estates: vec![
pb::GuardianEstate { instance: "zed".into(), product: "holger".into(), boxes: vec![b("main", 5, true)], ..Default::default() },
pb::GuardianEstate { instance: "acme".into(), product: "gunnar".into(), boxes: vec![b("main", 9, false), b("twin", 0, false)], cases: vec![pb::GuardianCase { state: "done".into(), ..Default::default() }, pb::GuardianCase { state: "detected".into(), ..Default::default() }], ..Default::default() },
pb::GuardianEstate { instance: "new".into(), product: "gunnar".into(), ..Default::default() },
],
};
let f = fleet(answer);
assert_eq!(f.iter().map(|e| e.name.as_str()).collect::<Vec<_>>(), ["acme", "new", "zed"]);
assert_eq!((f[0].serving, f[0].heard.clone(), f[0].last_heard_unix_ms, f[0].open_case.as_deref()), (true, vec!["main".to_string()], Some(9), Some("detected")));
assert!(f[1].error.as_deref().unwrap().contains("not installed yet"));
assert!(!f[1].installed);
assert_eq!((f[2].serving, f[2].last_heard_unix_ms), (false, Some(5)));
assert!(f[2].error.as_deref().unwrap().contains("(main)"));
}
/// ★ Owner 2026-10-12: test2 has front, main and twin, the console said
/// "Machines 2". The count is the machines that REPORTED IN, live: each
/// appears with its first report, none from a fixed list.
#[test]
fn the_machines_are_the_boxes_that_reported_in_and_appear_as_they_pair() {
let one = |boxes: Vec<pb::GuardianBox>| fleet(pb::GuardianCasesResponse { estates: vec![pb::GuardianEstate { instance: "test2-gunnar-rs".into(), product: "gunnar".into(), boxes, ..Default::default() }] }).remove(0);
// Paired (key pinned) but not reported yet: not a machine yet.
let e = one(vec![b("main", 9, false), b("twin", 0, false)]);
assert_eq!(e.reported, ["main"]);
// The twin's first report lands: it appears.
let e = one(vec![b("main", 9, false), b("twin", 10, false)]);
assert_eq!(e.reported, ["main", "twin"]);
// The front reports too: three machines, named in estate order.
let e = one(vec![b("twin", 10, false), b("main", 9, false), b("front", 11, false)]);
assert_eq!(e.reported, ["front", "main", "twin"]);
// A silent box that once reported is still a machine of the estate.
let e = one(vec![b("main", 9, true), b("twin", 10, false)]);
assert_eq!(e.reported.len(), 2);
}
#[test]
fn a_tenant_path_names_its_estate() {
assert_eq!(account_path("gunnar", "@acme/ns"), "gunnar/@acme/ns");
assert_eq!(sold_estate_of("@acme/ns"), Some("acme"));
assert_eq!(sold_estate_of("alice"), None);
assert_eq!(sold_estate_of("@acme"), None);
let fleet = [
FleetEstate { name: "acme".into(), product: "gunnar".into(), ..Default::default() },
FleetEstate { name: "gunnar-rs".into(), product: "gunnar".into(), ..Default::default() },
];
assert!(tenant_on(&fleet[0], true, "@acme/x") && !tenant_on(&fleet[0], true, "alice"));
assert!(tenant_on(&fleet[1], false, "alice") && !tenant_on(&fleet[1], false, "@acme/x"));
}
}