crates/monetize-server/src/fleet.rs
//! **The pull channel and the estate guardian, served** (5.4).
//!
//! An estate's boxes call [`monetize_proto::fleet_server::Fleet`]: they report, they pull the
//! SIGNED orders queued for them, they ask. Monetize never calls into an estate; the guardian
//! ([`monetize_guardian::Guardian`]) only ever QUEUES an order, and does the cloud part (fence,
//! lay one box again through the shop's ONE lay + installer path, [`crate::estate_order::relay`])
//! itself. The operator's console is [`monetize_proto::guardian_server::Guardian`]: the cases per
//! estate, the four buttons and the auto-fix opt-in.
use std::sync::{Arc, RwLock};
use monetize_cloud::lay::LaidMachine;
use monetize_cloud::Cloud;
use monetize_guardian::fleet::{AskAnswer, FleetRefusal};
use monetize_guardian::{Guardian, LaidBox, World, UNTIL_PINNED};
use monetize_product::guardian::EstateGuardian;
use monetize_product::EntitlementFact;
use crate::state::AppState;
/// **The fleet's half of the state**: the guardian (absent: the doors answer UNAVAILABLE) and
/// the product guardians this binary links beside the product plugins (a WASM guest product
/// cannot carry one).
#[derive(Default)]
pub struct FleetState {
guardian: monetize_guardian::pulled::GuardianSlot,
linked: RwLock<Vec<(&'static str, &'static dyn EstateGuardian)>>,
joined: std::sync::OnceLock<std::path::PathBuf>,
/// **Each joined estate's ONE bearer** (estate → its digest and life): what names the
/// SITE an estate-facing call comes from ([`crate::auth::CallerEstate`]). One per estate:
/// a re-enrolment replaces it, a revocation takes it away (ST-13).
bearers: RwLock<std::collections::BTreeMap<String, Bearer>>,
/// The day (unix ms / day) each estate's bearer was last told about ([`bearer_alerts`]):
/// one line a day, never a storm.
bearer_told: std::sync::Mutex<std::collections::BTreeMap<String, u64>>,
/// **The joined records that wait for their product** (JOINED-REFUSE, prod 2026-10-26):
/// file name → the product it waits for. Said once per wait ([`take_joined`]), never a
/// storm on every tick.
joined_waiting: std::sync::Mutex<std::collections::BTreeMap<String, String>>,
/// The `.refused` records put back once by this process ([`take_joined`]): a record
/// refused again is never put back again, so a refusal can never turn into a loop.
joined_put_back: std::sync::Mutex<std::collections::BTreeSet<String>>,
/// **The joined records a transient error keeps from being taken** (A32-1): file name →
/// (the error, when it was last told). Told once by an `ALERT joined.transient` line, again
/// when the error changes or every [`JOINED_TRANSIENT_RETELL_MS`] while it lasts; said taken
/// once the store answers again.
joined_transient: std::sync::Mutex<std::collections::BTreeMap<String, (String, u64)>>,
/// **The guardian's provider looks' lane** ([`looks_behind`], PROD-FIX 3): one jera
/// ChainQueue, started on the first tick that has a runtime.
look_lane: std::sync::OnceLock<Arc<jera::jobs::chain::ChainQueue>>,
}
/// One estate's bearer as held: its digest, until when it opens, when it was revoked.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
struct Bearer {
digest: String,
until_unix_ms: u64,
revoked_unix_ms: u64,
}
/// **An estate's bearer as the console shows it** (ST-13): `until_unix_ms` when it stops
/// opening (0: revoked), `revoked_unix_ms` when the operator revoked it (0: never).
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct BearerView {
pub until_unix_ms: u64,
pub revoked_unix_ms: u64,
}
impl std::fmt::Debug for FleetState {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "FleetState(guardian: {})", self.guardian.get().is_some())
}
}
impl FleetState {
pub fn with(guardian: Guardian) -> Self {
let s = Self::default();
let _ = s.guardian.set(Arc::new(guardian));
s
}
/// A fleet state whose guardian is set in `slot` — the registry's, so its pulled
/// products read the reports and queue on the guardian installed here.
pub fn sharing(slot: monetize_guardian::pulled::GuardianSlot) -> Self {
Self {
guardian: slot,
..Default::default()
}
}
/// Set the guardian once; `false` when one was set already.
pub fn install(&self, guardian: Guardian) -> bool {
self.guardian.set(Arc::new(guardian)).is_ok()
}
pub fn guardian(&self) -> Option<&Arc<Guardian>> {
self.guardian.get()
}
/// **Where the estates that joined are recorded** (G7): [`tick`] takes them from `dir` from
/// now on. Set once; `false` when it was set already.
pub fn joined_at(&self, dir: std::path::PathBuf) -> bool {
self.joined.set(dir).is_ok()
}
/// Know `estate`'s bearer by its digest ([`bearer_digest`]), with no expiry (tests).
pub fn know_bearer(&self, digest: &str, estate: &str) {
self.know_bearer_until(digest, estate, u64::MAX);
}
/// **`estate`'s bearer is `digest` until `until_unix_ms`** — REPLACING the one it held
/// (ST-13: a re-enrolment's new bearer kills the old one at once, no restart).
pub fn know_bearer_until(&self, digest: &str, estate: &str, until_unix_ms: u64) {
self.bearers
.write()
.unwrap_or_else(|e| e.into_inner())
.insert(
estate.to_owned(),
Bearer {
digest: digest.to_owned(),
until_unix_ms,
revoked_unix_ms: 0,
},
);
}
/// **`estate`'s bearer is revoked** (at `at_unix_ms`): it opens nothing from now on.
pub fn revoke_bearer(&self, estate: &str, at_unix_ms: u64) {
self.bearers
.write()
.unwrap_or_else(|e| e.into_inner())
.insert(
estate.to_owned(),
Bearer {
digest: String::new(),
until_unix_ms: 0,
revoked_unix_ms: at_unix_ms,
},
);
}
/// `estate`'s bearer state, when it ever held one here.
pub fn bearer_of(&self, estate: &str) -> Option<BearerView> {
self.bearers
.read()
.unwrap_or_else(|e| e.into_inner())
.get(estate)
.map(|b| BearerView {
until_unix_ms: b.until_unix_ms,
revoked_unix_ms: b.revoked_unix_ms,
})
}
/// **The estate whose bearer `bearer` is**, if any joined with it (the wall clock).
pub fn estate_of_bearer(&self, bearer: &str) -> Option<String> {
self.estate_of_bearer_at(bearer, crate::bus::now_unix_ms())
}
/// **The estate whose LIVE bearer `bearer` is at `now`**: its current one (never one a
/// re-enrolment replaced), not revoked, not past its expiry.
pub fn estate_of_bearer_at(&self, bearer: &str, now: u64) -> Option<String> {
if bearer.is_empty() {
return None;
}
let d = bearer_digest(bearer);
self.bearers
.read()
.unwrap_or_else(|e| e.into_inner())
.iter()
.find(|(_, b)| {
!b.digest.is_empty()
&& b.digest == d
&& b.revoked_unix_ms == 0
&& now < b.until_unix_ms
})
.map(|(e, _)| e.clone())
}
/// Link a product's guardian by product id; a second link of the same id replaces the first.
pub fn link(&self, product: &'static str, g: &'static dyn EstateGuardian) {
let mut l = self.linked.write().unwrap_or_else(|e| e.into_inner());
l.retain(|(p, _)| *p != product);
l.push((product, g));
}
fn linked(&self, product: &str) -> Option<&'static dyn EstateGuardian> {
self.linked
.read()
.unwrap_or_else(|e| e.into_inner())
.iter()
.find(|(p, _)| *p == product)
.map(|(_, g)| *g)
}
}
/// **The world the guardian works in, over the shop's state**: the estate's cloud is the
/// shop's `[estate_orders] cloud`, its boxes are the run's laid machines, and laying one again
/// is [`crate::estate_order::relay::relay_one`].
pub struct ShopWorld<'a> {
pub state: &'a AppState,
}
impl ShopWorld<'_> {
fn cloud_name(&self) -> Option<&str> {
self.state.estates.cfg.as_ref().map(|c| c.cloud.as_str())
}
}
impl World for ShopWorld<'_> {
fn guardian(&self, product: &str) -> Option<&dyn EstateGuardian> {
if let Some(g) = self.state.fleet.linked(product) {
return Some(g);
}
self.state.registry.guardian_of(product)
}
fn cloud(&self, _instance: &str) -> Option<&dyn Cloud> {
self.state
.registry
.find_cloud(self.cloud_name()?)
.map(|c| c as &dyn Cloud)
}
fn laid_machine(&self, instance: &str, role: &str) -> Option<(String, LaidMachine)> {
// An ADOPTED site's machines are its kit's (no label of this shop): the guardian waits
// for its operator, as before the adoption — it fences and lays nothing of it.
// An estate being terminated (or gone) is its give-back's alone (F4): the guardian fences
// and lays nothing of it.
let laid = self
.state
.estates
.get(instance)
.filter(|r| !crate::estate_order::adopt::is_adopted(r))
.filter(|r| {
r.phase != crate::estate_order::phase::TERMINATING
&& !crate::estate_order::phase::is_gone(&r.phase)
})?
.laid?;
let m = laid.machine(role)?.clone();
if m.laid_ids().is_empty() {
return None;
}
Some((laid.reference, m))
}
/// **ONE monetize** (G16, M-D28 reversed): no estate runs a monetize. A SOLD estate's
/// tenants are the product's accounts on the ONE ledger with the estate in the path
/// (`@<estate>/<ns>`); its restatement is exactly those, each as its gunnar names it
/// ([`monetize::sign::Signer::on_estate`]). The product's own estate gets the product's
/// verdicts as monetize signs a snapshot of them, signup-granted tenants folded in, and
/// never a sold estate's tenant.
fn entitlements(&self, product: &str, instance: &str) -> Result<Vec<EntitlementFact>, String> {
let facts = self
.state
.ledger
.entitlements(product)
.map_err(|e| format!("the ledger: {e}"))?;
// An ADOPTED site keeps the restatement it had as the kit-installed site it is.
if self
.state
.estates
.get(instance)
.is_some_and(|r| !crate::estate_order::adopt::is_adopted(&r))
{
return Ok(facts
.iter()
.filter_map(|f| self.state.signer_for(product).on_estate(f, instance))
.collect());
}
let facts = crate::signup::with_signup_facts(self.state, product, facts)
.map_err(|e| format!("the signup store: {e}"))?;
Ok(facts
.into_iter()
.filter(|f| f.tenant.estate().is_none())
.collect())
}
fn laid_ipv4(&self, instance: &str, role: &str) -> String {
self.state
.estates
.get(instance)
.and_then(|r| {
r.laid
.as_ref()
.and_then(|l| l.machine(role).map(|m| m.utility_ipv4.clone()))
})
.unwrap_or_default()
}
fn lay_one(&self, instance: &str, role: &str) -> Result<LaidBox, String> {
crate::estate_order::relay::relay_one(self.state, instance, role)
}
/// ★ SEC-S1: **an estate its term stopped is monetize's own stop**, never an outage: unpaid
/// past its grace or cancelled at the end of its paid period (SUSPENDED), or being deleted.
/// The guardian starts, fences and lays nothing of it; a renewal starts it (the term).
fn estate_held(&self, instance: &str) -> Option<String> {
use crate::estate_order::phase;
let r = self.state.estates.get(instance)?;
if r.site_order || !matches!(r.phase.as_str(), phase::SUSPENDED | phase::DELETING) {
return None;
}
let why = if r.reversed_at_ms > 0 {
"its money went back at the vendor"
} else if r.cancelled_at_ms > 0 {
"cancelled by its buyer, stopped at the end of its paid period"
} else {
"unpaid past its grace"
};
Some(format!("the term stopped it: {why}; {}", r.phase))
}
/// ★ SEC-S1: a box the term stopped (the run's `stopped` list: a renewal's start it still
/// owes) is the term's to start, never the guardian's.
fn box_held(&self, instance: &str, role: &str) -> Option<String> {
let r = self.state.estates.get(instance)?;
if r.stopped.is_empty() {
return None;
}
let m = r.laid.as_ref()?.machine(role)?.clone();
r.stopped
.iter()
.any(|id| {
monetize_cloud::lay::same_resource(id, &m.server)
|| monetize_cloud::lay::same_resource(&m.server, id)
})
.then(|| {
format!(
"the term stopped the {role} box {} and starts it itself",
m.server
)
})
}
/// S8: the front restored from its daily backup in the operator's name — the same path the
/// buyer's `EstateOrder.RestoreFront` takes.
fn restore_one(&self, instance: &str, role: &str, principal: &str) -> Result<String, String> {
if role != monetize_product::guardian::role::FRONT {
return Err(format!(
"only the front is restored from a backup; the {role} is recharged or laid again"
));
}
crate::estate_order::sold::restore_front(
self.state,
instance,
crate::estate_order::sold::RestoreBy::Operator(principal.into()),
)
.map(|a| a.words)
.map_err(|e| match e {
crate::estate_order::sold::SoldRefusal::NotSigned => {
crate::estate_order::sold::NOT_SIGNED.to_string()
}
crate::estate_order::sold::SoldRefusal::Refused(w)
| crate::estate_order::sold::SoldRefusal::Unavailable(w) => w,
})
}
}
/// **One guardian tick over the shop**: every estate's reports read, cases moved. What it did,
/// one line each (also on stderr). The provider looks due are asked here, inline (a caller
/// with no runtime: a test, `--dry`); the server's loop asks them behind the tick
/// ([`tick_on`]).
pub fn tick(state: &AppState) -> Vec<String> {
if let Some(g) = state.fleet.guardian() {
g.refresh_looks(&ShopWorld { state }, state.now());
}
tick_reading_looks(state)
}
/// **The server's guardian tick** ([`run`]): [`tick`], the provider looks due queued as ONE
/// slow jera job behind it ([`looks_behind`]) instead of asked in the tick (PROD-FIX 3).
pub fn tick_on(state: &Arc<AppState>) -> Vec<String> {
looks_behind(state);
tick_reading_looks(state)
}
/// The jera job kind of the guardian's provider looks ([`looks_behind`]).
pub const LOOK_JOB: &str = "monetize_guardian_provider_looks";
/// **The provider looks due, as a queued jera job** (PROD-FIX 3, prod 2026-10-26: UpCloud's
/// storages were read several times a second — every 5 s tick listed the whole account once
/// per silent main, and so did every console read of the fleet). A look is due at most once
/// per [`monetize_guardian::GuardianConfig::provider_look_every_ms`] per main; the job is ONE
/// keyed job on its own lane at low priority, so a tick never waits on the provider and a
/// queued job is not asked twice. No runtime: asked inline, as [`tick`] does.
pub fn looks_behind(state: &Arc<AppState>) -> usize {
let Some(g) = state.fleet.guardian() else {
return 0;
};
let due = g.looks_due(&ShopWorld { state }, state.now()).len();
if due == 0 {
return 0;
}
match look_lane(state) {
Ok(lane) => {
if let Some(line) = look_submitted(&lane.submit(look_spec(state.now(), due))) {
eprintln!("monetize-server: guardian: {line}");
}
}
Err(_) => {
g.refresh_looks(&ShopWorld { state }, state.now());
}
}
due
}
/// The ONE key of the provider looks' job (RA3-2): the same while a look is queued, so a tick
/// that finds looks due while one is queued asks nothing new, and one asked while it runs runs
/// once more after it (jera's re-arm).
pub const LOOK_KEY: &str = "guardian-looks";
/// **The provider looks' job** asked at `now` for `due` looks: one key, one group, one
/// generation (RA3-2).
pub fn look_spec(now: u64, due: usize) -> jera::jobs::chain::ChainSpec {
jera::jobs::chain::ChainSpec::new(
LOOK_JOB,
LOOK_KEY,
"guardian-looks",
0,
"monetize",
serde_json::json!({ "at": now, "due": due }),
)
.with_target("ask the provider about the silent mains monetize laid")
.with_priority(jera::jobs::chain::Priority::Low)
}
/// What to SAY of what jera answered the looks' submit (RA3-2): nothing for a job queued or
/// one already queued; a refusal by name.
pub fn look_submitted(s: &jera::jobs::chain::Submitted) -> Option<String> {
match s {
jera::jobs::chain::Submitted::Accepted(_) | jera::jobs::chain::Submitted::Duplicate(_) => None,
jera::jobs::chain::Submitted::Stale(g) => Some(format!("ALERT guardian.looks-refused the provider looks' job was refused by jera as stale (generation {g} holds the group): no look is asked this tick")),
}
}
/// **The looks' lane** of `state`: one jera ChainQueue on the tick's runtime; `Err` with none.
fn look_lane(state: &Arc<AppState>) -> Result<Arc<jera::jobs::chain::ChainQueue>, String> {
if let Some(q) = state.fleet.look_lane.get() {
return Ok(q.clone());
}
let rt = tokio::runtime::Handle::try_current()
.map_err(|_| "no runtime to run the guardian's looks 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("guardian: the provider looks");
let queue = jera::jobs::chain::ChainQueue::new(sink);
let q = state.fleet.look_lane.get_or_init(|| queue.clone()).clone();
if Arc::ptr_eq(&q, &queue) {
let weak = Arc::downgrade(state);
// A32-14, stays a task: the guardian looks lane's jera worker.
rt.spawn(queue.clone().run(move |_spec| {
let Some(st) = weak.upgrade() else {
anyhow::bail!("the server stopped")
};
let Some(g) = st.fleet.guardian() else {
anyhow::bail!("this monetize runs no guardian")
};
let asked = g.refresh_looks(&ShopWorld { state: &st }, st.now());
Ok(serde_json::json!({ "asked": asked }))
}));
}
Ok(q)
}
/// The tick itself, on the provider words last looked at (none asked here).
fn tick_reading_looks(state: &AppState) -> Vec<String> {
let Some(g) = state.fleet.guardian() else {
return Vec::new();
};
let mut lines = state
.fleet
.joined
.get()
.map(|d| take_joined(state, d))
.unwrap_or_default();
lines.extend(g.tick(&ShopWorld { state }, state.now()));
lines.extend(bearer_alerts(state));
lines.extend(operator_rebuild_alerts(g));
for l in &lines {
eprintln!("monetize-server: guardian: {l}");
}
lines
}
/// The alert's kind ([`operator_rebuild_alerts`]): `estate watch` and the functional status read it.
pub const OPERATOR_REBUILD_ALERT: &str = "guardian.operator_rebuild";
/// **The operator is told to rebuild** (main rebuild, owner 2026-10-12: the ONE box lays no
/// main for an estate whose cloud is its operator's — it detects, then WAITS for the
/// operator's `estate rebuild main --apply`). Once per case: an ALERT line naming the
/// command, and a functional-status row beside it (as the non-payment alerts).
pub fn operator_rebuild_alerts(g: &monetize_guardian::Guardian) -> Vec<String> {
static SAID: std::sync::Mutex<std::collections::BTreeSet<String>> =
std::sync::Mutex::new(std::collections::BTreeSet::new());
let mut out = Vec::new();
for (instance, _) in g.store.estates() {
let Some(case) = g.store.open_case(&instance) else {
continue;
};
let waits = case.steps.iter().find(|s| {
s.asked
.iter()
.any(|a| a == monetize_guardian::runner::OPERATOR_FENCE)
&& s.fenced.is_empty()
&& s.state == "running"
});
let Some(step) = waits else { continue };
if !SAID
.lock()
.unwrap_or_else(|p| p.into_inner())
.insert(case.id.clone())
{
continue;
}
let role = match &step.step {
monetize_product::guardian::Step::Cloud(
monetize_product::guardian::CloudStep::Fence { role },
) => role.clone(),
_ => "main".to_string(),
};
let detail = format!(
"{instance}: case {} waits for its OPERATOR — this ONE box lays nothing in the estate's own cloud. Rebuild its {role} with the product's own ops kit (`cargo xtask --instance <{instance}'s instance> estate rebuild {role} --apply`; the product's guardian decides whether the old disk may go); the case goes on by itself once the owner's notice names the new {role}",
case.id
);
skidbladnir::functional_status(
"monetize-server/guardian",
OPERATOR_REBUILD_ALERT,
false,
&detail,
);
let line = format!("ALERT {OPERATOR_REBUILD_ALERT} {detail}");
eprintln!("monetize-server: {line}");
out.push(line);
}
out
}
/// Where, under the data root, the estates that joined (`POST /attach/enrol`) are recorded:
/// `<data>/joined/<estate>.json`, one record each.
pub const JOINED_REL: &str = "joined";
/// The clause a joined record's refusal names when its ONLY fault was that its product was
/// not served yet ([`take_joined`]): written first in `<estate>.json.refused.why`, the one
/// recorded reason a put-aside record is put back for.
pub const JOINED_NOT_SERVED: &str = "product-not-served";
/// **How often a transient error that keeps a joined record from being taken is told again**
/// while it lasts (A32-1): an hour — the guardian ticks every few seconds, the operator is
/// mailed once an hour, never a storm.
pub const JOINED_TRANSIENT_RETELL_MS: u64 = 3600 * 1000;
/// **Take the estates that joined** (G7, pull): every `<estate>.json` in `dir` whose product
/// this monetize sells is declared on the guardian, and the box that enrolled may pin its key
/// on its first call inside the pin window counted from its enrolment. Monetize never dials it.
///
/// ★ **JOINED-REFUSE** (prod 2026-10-26): a record whose product is not served HERE YET (the
/// gap of an update: the new app is up, its product guests load only with the new feed) is
/// never put aside. It stays in place, is said ONCE as a `WARN` (no ALERT: nothing is mailed),
/// and is taken on the first tick once the products change (a feed taken, a guest loaded).
/// A record refused for a REAL reason (not a join record, an estate known under another
/// product, not an estate name) is put aside as `<estate>.json.refused`, said by name.
///
/// ★ **A32-1**: a TRANSIENT error — the record cannot be read this tick, the guardian's store
/// refuses a write (a full disk, a volume gone read-only) — is no defect of the record: it
/// stays as `<estate>.json`, the operator is told by an `ALERT joined.transient` line (mailed),
/// and every tick tries it again, so the first tick once the store writes takes it.
pub fn take_joined(state: &AppState, dir: &std::path::Path) -> Vec<String> {
let Some(g) = state.fleet.guardian() else {
return Vec::new();
};
let mut lines = put_back_refused(state, g, dir);
let Ok(rd) = std::fs::read_dir(dir) else {
return lines;
};
let mut files: Vec<std::path::PathBuf> = rd
.filter_map(|e| e.ok().map(|e| e.path()))
.filter(|p| p.extension().is_some_and(|x| x == "json"))
.collect();
files.sort();
let mut waiting = state
.fleet
.joined_waiting
.lock()
.unwrap_or_else(|e| e.into_inner());
let mut transient = state
.fleet
.joined_transient
.lock()
.unwrap_or_else(|e| e.into_inner());
let now = state.now();
for path in files {
let name = path
.file_name()
.map(|n| n.to_string_lossy().to_string())
.unwrap_or_default();
let outcome = take_one(state, g, &path);
if !matches!(outcome, Err(NotTaken::Transient(_))) {
if let Some((why, _)) = transient.remove(&name) {
if outcome.is_ok() {
lines.push(format!(
"joined: {name} is taken now (the error that kept it, {why}, is gone)"
));
}
}
}
match outcome {
Ok(said) => {
if let Some(product) = waiting.remove(&name) {
lines.push(format!("joined: {name} is taken now that {product:?} is served (it waited for its product)"));
}
lines.extend(said);
}
Err(NotTaken::Transient(why)) => {
let tell = match transient.get(&name) {
Some((was, at)) => {
*was != why || now.saturating_sub(*at) >= JOINED_TRANSIENT_RETELL_MS
}
None => true,
};
if tell {
lines.push(format!("ALERT joined.transient {name}: {why} — the record stays and every tick tries it again; it is taken once the store answers"));
transient.insert(name, (why, now));
}
}
Err(NotTaken::Waits { product, why }) => {
if waiting.get(&name) != Some(&product) {
lines.push(format!("WARN [joined] {name}: {why}; the record stays and is taken once {product:?} is served (a feed taken, its guest loaded)"));
waiting.insert(name, product);
}
}
Err(NotTaken::Refused(why)) => {
waiting.remove(&name);
lines.extend(put_aside(state, &path, &name, &why));
}
}
}
// A record that is gone (taken away, re-enrolled under another name) waits no more.
waiting.retain(|name, _| dir.join(name).exists());
transient.retain(|name, _| dir.join(name).exists());
lines
}
/// The field THIS app stamps into every record it puts aside (RA1-6): the refusal's reason, in
/// the record itself. A put-aside record that carries it is this app's, whatever is (or is not)
/// beside it, so a lost `.why` is never read as an older app's record.
pub const JOINED_REFUSED_STAMP: &str = "refused_by_monetize";
/// **Put a refused joined record aside, its reason recorded twice** (RA1-6): in the record
/// itself ([`JOINED_REFUSED_STAMP`], written as `<estate>.json.refused` and the original then
/// removed) and beside it (`<estate>.json.refused.why`). A reason that cannot be written beside
/// it is an ALERT by name, and the record still goes aside (stamped; if even the stamped copy
/// cannot be written it is moved as it is, and this process never puts it back). A record that
/// is no JSON object is moved as it is: it is never a join record, so it is never put back.
fn put_aside(state: &AppState, path: &std::path::Path, name: &str, why: &str) -> Vec<String> {
let mut aside = path.as_os_str().to_owned();
aside.push(".refused");
let aside = std::path::PathBuf::from(aside);
let mut beside = aside.as_os_str().to_owned();
beside.push(".why");
let mut lines = Vec::new();
// The reason beside it first: a record put aside for a REAL reason is never put back.
if let Err(e) = std::fs::write(&beside, format!("refused: {why}\n")) {
lines.push(format!("ALERT joined.why-lost {name}: the reason it is refused could not be written beside it ({e}); it is kept aside, its reason in the record itself, and never put back"));
}
let stamped = std::fs::read(path)
.ok()
.and_then(|b| serde_json::from_slice::<serde_json::Value>(&b).ok())
.and_then(|mut v| {
v.as_object_mut()?.insert(
JOINED_REFUSED_STAMP.to_string(),
serde_json::Value::String(format!("refused: {why}")),
);
serde_json::to_vec(&v).ok()
});
let moved = match stamped {
Some(bytes) => {
let mut tmp = aside.as_os_str().to_owned();
tmp.push(".tmp");
let tmp = std::path::PathBuf::from(tmp);
match std::fs::write(&tmp, &bytes)
.and_then(|()| std::fs::rename(&tmp, &aside))
.and_then(|()| std::fs::remove_file(path))
{
Ok(()) => Ok(()),
Err(e) => {
let _ = std::fs::remove_file(&tmp);
lines.push(format!("ALERT joined.unstamped {name}: it could not be marked as refused in itself ({e}); it is moved aside as it is and this process never puts it back"));
state
.fleet
.joined_put_back
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert(format!("{name}.refused"));
// The original as it is over any half-made copy; if it is gone the stamped copy is in place.
if path.exists() {
std::fs::rename(path, &aside)
} else {
Ok(())
}
}
}
}
None => std::fs::rename(path, &aside),
};
let moved = moved
.map(|_| String::new())
.unwrap_or_else(|e| format!(" (and it cannot be put aside: {e})"));
lines.insert(0, format!("REFUSED [joined] {name}: {why}{moved}"));
lines
}
/// ★ **JOINED-REFUSE, the way back**: every `<estate>.json.refused` put aside only because its
/// product was not served yet (prod 2026-10-26: `gunnar-rs.json.refused`, the adopted
/// gunnar.rs's only standing, put aside by an OLDER app) is put back as `<estate>.json` once
/// that product IS served, said once by name; the take that follows takes it. Which records:
/// * its recorded reason (`<estate>.json.refused.why`, written by this app on every real
/// refusal) names [`JOINED_NOT_SERVED`]: put back; any other recorded reason: NEVER;
/// * no reason beside it (an older app's): re-checked — it is a join record, its product is
/// served now, its estate is not known here under another product. An older app put a
/// record aside for exactly three things: not a join record, an estate of another product,
/// a product not served; the first two are re-checked as false, so only the third is left;
/// * a record THIS app put aside carries its reason in itself ([`JOINED_REFUSED_STAMP`],
/// RA1-6): only a recorded "product not served" is put back, whatever is beside it.
///
/// A newer `<estate>.json` (a re-enrolment) wins: the old one stays aside. Each is put back at
/// most once by a process: one refused again stays aside.
fn put_back_refused(state: &AppState, g: &Guardian, dir: &std::path::Path) -> Vec<String> {
let Ok(rd) = std::fs::read_dir(dir) else {
return Vec::new();
};
let mut aside: Vec<std::path::PathBuf> = rd
.filter_map(|e| e.ok().map(|e| e.path()))
.filter(|p| p.extension().is_some_and(|x| x == "refused"))
.collect();
aside.sort();
let mut lines = Vec::new();
for path in aside {
let back = path.with_extension("");
if back.extension().is_none_or(|x| x != "json") || back.exists() {
continue;
}
let mut why_path = path.clone().into_os_string();
why_path.push(".why");
let why_path = std::path::PathBuf::from(why_path);
let recorded = std::fs::read_to_string(&why_path).ok();
if recorded
.as_deref()
.is_some_and(|w| !w.trim_start().starts_with(JOINED_NOT_SERVED))
{
continue;
}
let Ok(text) = std::fs::read(&path) else {
continue;
};
// RA1-6: a record THIS app put aside carries its reason in itself; a missing `.why`
// beside it is a lost write, never an older app's file.
let stamp = serde_json::from_slice::<serde_json::Value>(&text)
.ok()
.and_then(|v| {
v.get(JOINED_REFUSED_STAMP)
.and_then(|s| s.as_str())
.map(str::to_owned)
});
if stamp.is_some_and(|w| {
!w.trim_start_matches("refused: ")
.starts_with(JOINED_NOT_SERVED)
}) {
continue;
}
let Ok(j) = serde_json::from_slice::<Joined>(&text) else {
continue;
};
if state.registry.find_product(&j.product).is_none()
|| g.store
.product_of(&j.estate)
.is_some_and(|p| p != j.product)
{
continue;
}
let name = path
.file_name()
.map(|n| n.to_string_lossy().to_string())
.unwrap_or_default();
if !state
.fleet
.joined_put_back
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert(name.clone())
{
continue;
}
match std::fs::rename(&path, &back) {
Ok(()) => {
if recorded.is_some() {
let _ = std::fs::remove_file(&why_path);
}
lines.push(format!("joined: {name} is put back: it was put aside only because its product {:?} was not served yet, and it is served now", j.product));
}
Err(e) => lines.push(format!("WARN [joined] {name} cannot be put back: {e}")),
}
}
lines
}
/// Why [`take_one`] did not take a joined record.
#[derive(Debug, Clone, PartialEq, Eq)]
enum NotTaken {
/// Its product is not served here YET (JOINED-REFUSE): it stays and is taken later.
Waits { product: String, why: String },
/// A real refusal (a defect of the record): the record is put aside.
Refused(String),
/// ★ A32-1: a transient error (the record unreadable this tick, the store refusing a
/// write): the record stays and is tried again on the next tick.
Transient(String),
}
/// A read error in [`take_one`] is transient (A32-1); a store error says itself which it is
/// (RA3-7). A defect of the record itself is named [`NotTaken::Refused`] where it is found.
impl From<String> for NotTaken {
fn from(why: String) -> Self {
NotTaken::Transient(why)
}
}
/// ★ RA3-7: the store says what its error is — a refusal of the record (a name that is no
/// estate, a key that is no key) is a defect, put aside; a failure of the store is transient.
impl From<monetize_guardian::StoreError> for NotTaken {
fn from(e: monetize_guardian::StoreError) -> Self {
if e.is_refusal() {
NotTaken::Refused(e.to_string())
} else {
NotTaken::Transient(e.to_string())
}
}
}
/// **How long an ATTACHED estate's boxes may pin their keys after its owner's install
/// notice** (i54a run16): the kit sends the notice right BEFORE the installer feeds the boxes,
/// and a box's first report comes when it is fed and serving — minutes later on a slow host.
/// The window covers a whole install, as gunnar-server's `repair::NOT_YET_KNOWN_GRACE` does
/// on the box's side. Still only the enrolled owner opens it, and a box pins once.
pub const ATTACHED_INSTALL_WINDOW_MS: u64 = 45 * 60_000;
/// **How long an estate's bearer opens the estate-facing doors** (ST-13) when its record
/// names no expiry of its own: a year from its enrolment. Rotation is a re-enrolment (a new
/// one-shot token); the operator revokes one at the door (`POST /attach/revoke`).
pub const BEARER_LIFE_MS: u64 = 365 * 24 * 3600 * 1000;
/// **How long before an estate's bearer stops opening the operator is told** (audit round 2,
/// BEARER-1): 30 days, once a day ([`bearer_alerts`]).
pub const BEARER_WARN_MS: u64 = 30 * 24 * 3600 * 1000;
/// **The operator is told before an estate's bearer stops opening** (audit round 2, BEARER-1):
/// one `ALERT attach.bearer_expiring <estate>` line a day inside [`BEARER_WARN_MS`] of its end,
/// one `ALERT attach.bearer_expired <estate>` line a day after it — the ONE box mails what it
/// marks ALERT (monetize-admin `alerts`). A revoked bearer is the operator's own doing: no line.
pub fn bearer_alerts(state: &AppState) -> Vec<String> {
use monetize_embed::civil::{iso_date, DAY_MS};
let now = state.now();
let today = now / DAY_MS;
let bearers: Vec<(String, Bearer)> = state
.fleet
.bearers
.read()
.unwrap_or_else(|e| e.into_inner())
.iter()
.map(|(e, b)| (e.clone(), b.clone()))
.collect();
let mut told = state
.fleet
.bearer_told
.lock()
.unwrap_or_else(|e| e.into_inner());
let mut lines = Vec::new();
for (estate, b) in bearers {
// Revoked (the operator's own doing), never ending (a test's `know_bearer`), far off.
if b.revoked_unix_ms > 0
|| b.digest.is_empty()
|| b.until_unix_ms == u64::MAX
|| b.until_unix_ms > now.saturating_add(BEARER_WARN_MS)
{
continue;
}
if told.get(&estate) == Some(&today) {
continue;
}
told.insert(estate.clone(), today);
let on = iso_date(b.until_unix_ms);
lines.push(if now >= b.until_unix_ms {
format!(
"ALERT attach.bearer_expired {estate}: its ONE bearer stopped opening Signup, Purchase and EstateOrder on {on} — its signups and purchases are refused until it is enrolled again (an attach token, then the kit's attach)"
)
} else {
let days = (b.until_unix_ms - now).div_ceil(DAY_MS);
format!(
"ALERT attach.bearer_expiring {estate}: its ONE bearer stops opening Signup, Purchase and EstateOrder in {days} days, on {on} — enrol the estate again before then (an attach token, then the kit's attach)"
)
});
}
lines
}
/// **The digest an estate's bearer is kept by** (hex SHA-256): the central never keeps the
/// bearer itself (monetize-front `attach` records only this).
pub fn bearer_digest(bearer: &str) -> String {
use sha2::Digest as _;
sha2::Sha256::digest(bearer.as_bytes())
.iter()
.map(|b| format!("{b:02x}"))
.collect()
}
/// What one estate that joined records; the rest of the front's record is informational.
#[derive(serde::Deserialize)]
struct Joined {
estate: String,
product: String,
box_name: String,
enrolled_unix_ms: u64,
/// The boxes the enrolled owner named after the install (monetize-front
/// `attach::installed_answer`), each pinned from [`Joined::installed_unix_ms`].
#[serde(default)]
roles: Vec<String>,
#[serde(default)]
installed_unix_ms: u64,
/// [`bearer_digest`] of the bearer minted at enrolment; empty from an older front.
#[serde(default)]
bearer_sha256: String,
/// Until when the bearer opens; 0: [`BEARER_LIFE_MS`] from the enrolment (an older front).
#[serde(default)]
bearer_until_unix_ms: u64,
/// When the operator revoked it; 0: never.
#[serde(default)]
bearer_revoked_unix_ms: u64,
/// The site's ticket key its owner's notice delivered (ST-3); empty: none yet.
#[serde(default)]
appliance_key: String,
/// Whether the owner's newest notice carried the ticket key (monetize-front
/// `Enrolled::installed_with_ticket_key`): the kit's second notice of one install.
#[serde(default)]
installed_with_ticket_key: bool,
/// Each box's own fleet key as the owner's newest notice named it (ST-14): role → 64 hex.
#[serde(default)]
fleet_keys: std::collections::BTreeMap<String, String>,
}
fn take_one(
state: &AppState,
g: &Guardian,
path: &std::path::Path,
) -> Result<Option<String>, NotTaken> {
let text = std::fs::read(path).map_err(|e| format!("the record cannot be read: {e}"))?;
let j: Joined = serde_json::from_slice(&text)
.map_err(|e| NotTaken::Refused(format!("not a join record: {e}")))?;
// A record that names no product is no join record (the store's own refusals — a name that
// is no estate — come back typed, RA3-7, below).
if j.product.trim().is_empty() {
return Err(NotTaken::Refused(
"not a join record: it names no product".into(),
));
}
if state.registry.find_product(&j.product).is_none() {
// Checked FIRST, before anything is known from the record: a waiting record opens
// nothing (no bearer, no declaration, no pin window) until its product is served.
return Err(NotTaken::Waits {
why: format!(
"{} joined as a {:?} estate, a product not served yet by this monetize",
j.estate, j.product
),
product: j.product,
});
}
// The estate's own bearer names the site its estate-facing calls come from — its ONE
// bearer: the record's replaces whatever it held (ST-13), a revoked record takes it away.
// An operator site the ONE box laid ITSELF after this enrolment holds the bearer the shop
// minted for it at that lay (`estate_order::site`): the older enrolment's stays replaced.
if crate::estate_order::site::bearer_newer_than(
state,
&j.estate,
j.enrolled_unix_ms.max(j.bearer_revoked_unix_ms),
) {
crate::estate_order::site::reassert(state);
} else if j.bearer_revoked_unix_ms > 0 {
state
.fleet
.revoke_bearer(&j.estate, j.bearer_revoked_unix_ms);
} else if !j.bearer_sha256.is_empty() {
let until = if j.bearer_until_unix_ms > 0 {
j.bearer_until_unix_ms
} else {
j.enrolled_unix_ms.saturating_add(BEARER_LIFE_MS)
};
state
.fleet
.know_bearer_until(&j.bearer_sha256, &j.estate, until);
}
// **A site the ONE box laid ITSELF since this record's newest word** (its enrolment, its
// owner's install notice) is the shop's to key (`estate_order::site`): the record's ticket
// key, its boxes' fleet keys and its pin windows are the OLD boxes' — restated here on
// every tick they shut the new boxes out and put the old main's ticket key back (test2,
// kit-installed then shop-laid, 2026-10-14). A NEWER word of the estate's own counts again.
if crate::estate_order::site::laid_since(
state,
&j.estate,
j.enrolled_unix_ms.max(j.installed_unix_ms),
) {
return Ok(None);
}
let mut did = Vec::new();
// ST-3: the site's ticket key, delivered by the estate itself. A refused key never undoes
// the estate's join: only the key is not taken, said once.
let mut keyed = None;
if !j.appliance_key.trim().is_empty() {
keyed = match state
.registry
.deliver_appliance_key(&j.product, &j.estate, &j.appliance_key)
{
Ok(line) => line,
Err(refused) => Some(refused),
};
// The product's OWN estate (its `instance`): the host checks the product's own tickets
// under it too (A32-13) — no `[products.<id>] appliance_key` copied by hand.
if state.registry.home_of(&j.product).as_deref() == Some(j.estate.as_str()) {
if let Ok(line) = state
.registry
.deliver_product_key(&j.product, &j.appliance_key)
{
keyed = Some(keyed.map_or(line.clone(), |k| format!("{k}; {line}")));
}
}
}
match g.store.product_of(&j.estate) {
Some(p) if p != j.product => {
return Err(NotTaken::Refused(format!(
"{} is known here as a {p:?} estate, not {:?}",
j.estate, j.product
)))
}
Some(_) => {}
None => {
g.store.declare(&j.estate, &j.product)?;
did.push(format!("{} is a {} estate", j.estate, j.product));
}
}
// The enrolled box from its enrolment; every box the owner named after the install from
// then (an attached estate's install PASS: its boxes are laid long after it enrolled).
let mut windows = vec![(
j.box_name.clone(),
j.enrolled_unix_ms,
"enrolled",
g.cfg.pin_window_ms,
)];
if j.installed_unix_ms > 0 {
windows.extend(j.roles.iter().map(|r| {
(
r.clone(),
j.installed_unix_ms,
"installed",
ATTACHED_INSTALL_WINDOW_MS,
)
}));
}
// ★ ST-14: a box the owner named WITH its key is pinned to exactly that key — no window,
// so no stranger first inside one takes the role; a different key pinned before (an
// impostor, or the box laid again) is replaced by the owner's word.
for (role, hex) in &j.fleet_keys {
let raw: Option<[u8; 32]> = (hex.len() == 64)
.then(|| {
(0..32)
.map(|i| u8::from_str_radix(&hex[2 * i..2 * i + 2], 16).ok())
.collect::<Option<Vec<u8>>>()
})
.flatten()
.and_then(|v| v.try_into().ok());
let Some(raw) = raw.filter(|r| monetize_embed::VerifyingKey::from_bytes(r).is_ok()) else {
did.push(format!(
"REFUSED the {role} key the owner named: not an ed25519 key (64 hex)"
));
continue;
};
if g.store.key_of(&j.estate, role).map(|k| k.to_bytes()) != Some(raw) {
g.store.register_key(
&j.estate,
role,
&monetize_embed::signing::base64_encode(&raw),
)?;
did.push(format!("its {role}'s key pinned as its owner named it"));
}
g.store.open_pin(&j.estate, role, 0)?;
if j.installed_unix_ms > 0 {
g.store.take_notice(&j.estate, role, j.installed_unix_ms)?;
}
}
for (role, from, when, width) in windows {
if j.fleet_keys.contains_key(&role) {
continue;
}
let until = from.saturating_add(width);
if g.store.key_of(&j.estate, &role).is_some() {
// ★ PIN-1: a NEWER owner's notice — the box in `role` was laid again with a new
// fleet key: its window opens once more, and inside it the new key replaces the
// pinned one ([`monetize_guardian::Guardian::first_call`]). The notice already
// taken, or one past its window, opens nothing; nor does the notice that carried
// the ticket key — the same install telling again after its pairing (the kit's
// second notice), never a box laid again: the boxes that pinned stay theirs.
// ★ PIN-2 (audit round 2): told apart by the key, never by time — an install laid
// again inside the last one's window (a failed install laid at once) is known.
let taken = g.store.notice_taken(&j.estate, &role);
if when == "installed"
&& until > state.now()
&& taken < from
&& !j.installed_with_ticket_key
{
g.store.open_pin(&j.estate, &role, until)?;
g.store.take_notice(&j.estate, &role, from)?;
did.push(format!("its {role} was laid again: its new key may replace the pinned one on its first call (installed)"));
}
continue;
}
if until <= state.now() {
if g.store.pin_window(&j.estate, &role) == 0 && !did.is_empty() {
did.push(format!("its {role} {when} too long ago to pin its key: the pin window is closed, the operator names the key"));
}
} else if g.store.pin_window(&j.estate, &role) < until {
g.store.open_pin(&j.estate, &role, until)?;
if when == "installed" {
g.store.take_notice(&j.estate, &role, from)?;
}
did.push(format!(
"its {role} may pin its key on its first call ({when})"
));
}
}
let mut said: Vec<String> = keyed.into_iter().collect();
if !did.is_empty() {
said.insert(0, format!("joined: {}", did.join("; ")));
}
Ok((!said.is_empty()).then(|| said.join("; ")))
}
/// What an estate's own console reads and presses about its MAIN ([`rebuild_main`]).
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct RebuildMain {
pub finding: String,
pub words: String,
pub remedy: String,
pub case_id: String,
pub case_state: String,
pub steps: Vec<String>,
pub outcome: String,
pub auto_fix: bool,
}
/// Why [`rebuild_main`] refused.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RebuildRefusal {
/// Not an estate's own call (no site bearer), or an estate the guardian does not know.
NotAnEstate(String),
/// The guardian's own refusal of the press.
Press(monetize_guardian::PressRefusal),
/// The guardian finds nothing about the main to press for.
Nothing(String),
}
/// **An estate's own console about its MAIN** (main rebuild, owner 2026-10-12): `estate` is
/// the caller's own (its site bearer named it). A look answers the guardian's finding about
/// the main and the open case; a press presses that finding's first remedy — `new_server`
/// for a silent main (the twin is never promoted: a new main is laid and filled from it),
/// `recharge_from_twin` for one that says its data is lost — in `principal`'s name, which
/// the guardian refuses unless `strong`. One path with the ONE box's own buttons.
pub fn rebuild_main(
state: &AppState,
estate: Option<&str>,
press: bool,
principal: &str,
strong: bool,
) -> Result<RebuildMain, RebuildRefusal> {
let Some(estate) = estate.map(str::trim).filter(|e| !e.is_empty()) else {
return Err(RebuildRefusal::NotAnEstate(
"only an estate's own console rebuilds its main: the call carries no estate's bearer"
.into(),
));
};
let g = state
.fleet
.guardian()
.ok_or_else(|| RebuildRefusal::NotAnEstate("this monetize runs no guardian".into()))?;
let product = g.store.product_of(estate).ok_or_else(|| {
RebuildRefusal::NotAnEstate(format!("the guardian knows no estate {estate:?}"))
})?;
let world = ShopWorld { state };
let now = state.now();
let finding = world
.guardian(&product)
.map(|eg| eg.diagnose(&g.view_now(&world, estate, now)))
.unwrap_or_default()
.into_iter()
.find(|f| f.role == monetize_product::guardian::role::MAIN);
let mut out = RebuildMain {
auto_fix: g.store.auto_fix(estate),
..RebuildMain::default()
};
if let Some(f) = &finding {
out.finding = f.code.clone();
out.words = f.words.clone();
out.remedy = f
.remedies
.first()
.map(|r| r.as_str().to_string())
.unwrap_or_default();
} else {
out.words = "the guardian finds nothing wrong with the main: nothing to rebuild".into();
}
let case = if press {
let remedy = finding
.as_ref()
.and_then(|f| f.remedies.first().copied())
.ok_or_else(|| RebuildRefusal::Nothing(out.words.clone()))?;
let c = g
.press(&world, estate, remedy, "", principal, strong, now)
.map_err(RebuildRefusal::Press)?;
eprintln!("monetize-server: guardian: {principal} (the estate's own console) pressed {} on {estate} (case {})", remedy.as_str(), c.id);
Some(c)
} else {
g.store
.open_case(estate)
.or_else(|| g.cases(estate).into_iter().next())
};
if let Some(c) = case {
out.case_id = c.id.clone();
out.case_state = c.state.as_str().to_string();
out.steps = c
.steps
.iter()
.map(|s| {
format!(
"{} — {}{}",
s.state,
s.step.words(),
if s.words.is_empty() {
String::new()
} else {
format!(": {}", s.words)
}
)
})
.collect();
out.outcome = c.outcome.clone();
}
Ok(out)
}
/// The jera duty kind of the guardian's tick ([`start`]).
pub const TICK_JOB: &str = "monetize_guardian_tick";
/// **The guardian's tick as a jera duty** (A32-14, P4b): [`tick_on`] at once and every `every`
/// after, on the blocking pool. A tick that PANICS is an `ALERT jobs.failed` line and a FAILED
/// row, and the next runs as due (the raw loop it replaces dropped the JoinError). The duty
/// holds the state weakly and ends with it, or at the server's clean stop (`jobs::stop_all`).
pub fn start(
state: &Arc<AppState>,
every: std::time::Duration,
) -> Result<Arc<crate::jobs::Job>, String> {
let weak = Arc::downgrade(state);
crate::jobs::every(TICK_JOB, every, move || {
let Some(s) = weak.upgrade() else {
return Ok(crate::jobs::Round::Done);
};
tick_on(&s);
Ok(crate::jobs::Round::Again)
})
}
/// What `Fleet.Ask` answers: a case opened (its id), the words, a page to open (empty today),
/// and a front's ask's answer (the old `SoldEstate` reply, encoded; empty for every other ask).
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Asked {
pub case_id: String,
pub words: String,
pub url: String,
pub answer: Vec<u8>,
}
impl From<AskAnswer> for Asked {
fn from(a: AskAnswer) -> Asked {
Asked {
case_id: a.case_id,
words: a.words,
..Asked::default()
}
}
}
/// **A front's ask, proven** ([`front_ask`]): the run's company, the kind and its body.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct FrontAsked {
pub company: String,
pub kind: String,
pub body: monetize_embed::fleet::FrontAsk,
}
/// **The run a fleet instance names**: a sold estate's fleet name is its company; an operator
/// site's is its label, its run kept under [`crate::estate_order::site::company_of`] the label.
pub fn company_of_instance(state: &AppState, instance: &str) -> String {
let named = |company: &str| {
state
.estates
.get(company)
.is_some_and(|r| r.fleet_name() == instance)
};
if named(instance) {
return instance.to_string();
}
let site = crate::estate_order::site::company_of(instance);
if !site.is_empty() && named(&site) {
return site;
}
instance.to_string()
}
/// The refusal code of a front's ask by a company that names more than one estate.
pub const FRONT_COMPANY_AMBIGUOUS: &str = "[front-company-ambiguous]";
/// What a front's `instance` names, when it is not a fleet name ([`front_by_company`]).
#[derive(Debug)]
pub enum ByCompany {
/// Exactly one run: the call is proven by its keys only.
One(Box<crate::estate_order::EstateRun>),
/// More than one run could be meant: refused by name, but only to a call one of them
/// signed (RA5-44); any other caller is UNAUTHENTICATED.
Ambiguous(Vec<crate::estate_order::EstateRun>),
}
/// **The estate a front fed before the fold asks as its COMPANY** (R6-FIX-4): a front fed
/// before P3-FLEET 6/8 (3dcd6154, 2026-10-25: no `product` in its `shop_door.toml`) is moved to
/// Fleet by its product's `or_over_fleet`, which signs as the run's company (`test2holgerrs`),
/// while the fleet knows an operator site by its label (`test2-holger-rs`). A ONE box serves
/// the fronts one release behind (fronts update after the box), so such an `instance` names
/// that run, but only when it is unambiguous: `None` when the fleet knows `instance` itself (the
/// fleet's own name: the path as before) or no run has that company and another fleet name of
/// `product`; the one run when exactly one does; [`ByCompany::Ambiguous`] when more than one run
/// could be meant (a company that is also another site's label), which [`front_ask`] says by
/// name only after the call verified under one of them. Logged once per company, by name.
/// The call is then proven by THAT run's own keys only ([`front_ask`]).
pub fn front_by_company(
state: &AppState,
g: &Guardian,
instance: &str,
product: &str,
) -> Option<ByCompany> {
if g.store.product_of(instance).is_some() {
return None;
}
let named: Vec<_> = state
.estates
.all()
.into_iter()
.filter(|r| r.company == instance || (r.site_order && r.estate == instance))
.collect();
let ours: Vec<_> = named
.iter()
.filter(|r| {
r.fleet_name() != instance
&& g.store.product_of(r.fleet_name()).as_deref() == Some(product)
})
.cloned()
.collect();
if named.len() > 1 && !ours.is_empty() {
return Some(ByCompany::Ambiguous(named));
}
let run = ours.into_iter().next()?;
static SAID: std::sync::Mutex<std::collections::BTreeSet<String>> =
std::sync::Mutex::new(std::collections::BTreeSet::new());
if SAID
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert(instance.to_string())
{
eprintln!(
"monetize-server: fleet: front ask by company {instance} resolved to {}",
run.fleet_name()
);
}
Some(ByCompany::One(Box::new(run)))
}
/// Is `c` signed by `run`'s box in `c.role`: the fleet key pinned for the run's fleet name, or
/// (its front) the estate's client key.
fn signed_by_run(
state: &AppState,
g: &Guardian,
run: &crate::estate_order::EstateRun,
c: &monetize_embed::fleet::FleetCall,
signature: &[u8],
now: u64,
) -> bool {
let by_fleet_key = g
.store
.key_of(run.fleet_name(), &c.role)
.is_some_and(|k| monetize_embed::fleet::verify_fleet_call(c, signature, &k, now).is_ok());
by_fleet_key
|| (c.role == crate::estate_order::sold::front_role_of(state, run)
&& crate::estate_order::sold::client_key_of(run).is_some_and(|k| {
monetize_embed::fleet::verify_fleet_call(c, signature, &k, now).is_ok()
}))
}
/// **The front's ask proven** (plan §8: the `SoldEstate` door folded into `Fleet.Ask`): the
/// call is an `ask` of a known estate's product, signed by the estate's FRONT — under the fleet
/// key pinned for `(instance, front)`, or under the estate's client key the custody minted it
/// (the key its `SoldEstate` calls were signed with: an estate installed before the fold has no
/// fleet key for its front, and needs none). Any other signer is UNAUTHENTICATED, said the same
/// way; a proven box that is not the front is refused by name. Nothing is done here. A front
/// asking as its run's company ([`front_by_company`], R6-FIX-4) is proven by that run's own
/// client key or the fleet key pinned for that run's fleet name, and nothing else.
pub fn front_ask(
state: &AppState,
g: &Guardian,
c: &monetize_embed::fleet::FleetCall,
signature: &[u8],
ask: monetize_embed::fleet::Ask,
) -> Result<FrontAsked, FleetRefusal> {
let now = state.now();
if c.call != monetize_embed::fleet::call::ASK {
return Err(FleetRefusal::Malformed(format!(
"a {:?} call on the ask door",
c.call
)));
}
let (run, by_fleet_key) = match front_by_company(state, g, &c.instance, &c.product) {
Some(ByCompany::Ambiguous(named)) => {
// RA5-44: proven under one of the candidates' keys first; said by name only then.
if !named
.iter()
.any(|run| signed_by_run(state, g, run, c, signature, now))
{
return Err(FleetRefusal::Unauthenticated);
}
return Err(FleetRefusal::Refused(format!(
"{FRONT_COMPANY_AMBIGUOUS} a front asks as {:?}, which names {} estates here: it must ask as its estate's fleet name",
c.instance,
named.len()
)));
}
Some(ByCompany::One(run)) => {
let pinned = g.store.key_of(run.fleet_name(), &c.role);
let by_fleet_key = pinned.is_some_and(|k| {
monetize_embed::fleet::verify_fleet_call(c, signature, &k, now).is_ok()
});
(*run, by_fleet_key)
}
None => {
if g.store.product_of(&c.instance).as_deref() != Some(c.product.as_str()) {
return Err(FleetRefusal::Unauthenticated);
}
let company = company_of_instance(state, &c.instance);
let run = state
.estates
.get(&company)
.ok_or(FleetRefusal::Unauthenticated)?;
(
run,
g.authenticate(c, signature, monetize_embed::fleet::call::ASK, now)
.is_ok(),
)
}
};
let company = run.company.clone();
let front = crate::estate_order::sold::front_role_of(state, &run);
let by_client_key = || {
c.role == front
&& crate::estate_order::sold::client_key_of(&run).is_some_and(|k| {
monetize_embed::fleet::verify_fleet_call(c, signature, &k, now).is_ok()
})
};
if !(by_fleet_key || by_client_key()) {
return Err(FleetRefusal::Unauthenticated);
}
if c.role != front {
return Err(FleetRefusal::Refused(format!(
"{:?} is the estate's {front}'s ask: the {} asks only for its own disks and remedies",
ask.kind, c.role
)));
}
let body = serde_json::from_value(ask.body)
.map_err(|e| FleetRefusal::Malformed(format!("not a {} ask: {e}", ask.kind)))?;
Ok(FrontAsked {
company,
kind: ask.kind,
body,
})
}
/// **`Fleet.Ask`**: a remedy opens a case ([`Guardian::ask`]); a sold estate's box asking to
/// size its own data disks (`grow_set`, G16-d) is answered at once, by its fleet key and role
/// ([`crate::estate_order::sold::grow_own`]), with no case. (The `renew` ask went in v2.0 step
/// 1: a buyer renews on the ONE box's own page, never through the estate's box.)
pub fn ask(
state: &AppState,
g: &Guardian,
c: &monetize_embed::fleet::FleetCall,
signature: &[u8],
) -> Result<Asked, FleetRefusal> {
let now = state.now();
g.authenticate(c, signature, monetize_embed::fleet::call::ASK, now)?;
let ask: monetize_embed::fleet::Ask = serde_json::from_str(&c.body)
.map_err(|e| FleetRefusal::Malformed(format!("not an ask: {e}")))?;
if monetize_embed::fleet::ask::is_front(&ask.kind) {
return Err(FleetRefusal::Refused(format!(
"{:?} is a front's ask: answered by the Fleet door itself",
ask.kind
)));
}
if ask.kind == monetize_embed::fleet::ask::FETCH {
// A product order's file, pulled by the box its order is queued for (MONETIZE-FIX).
let b: monetize_embed::fleet::FetchAsk = serde_json::from_value(ask.body)
.map_err(|e| FleetRefusal::Malformed(format!("not a fetch ask: {e}")))?;
let bytes = crate::estate_order::product_order::fetch(state, &c.instance, &c.role, &b.id)
.map_err(FleetRefusal::Refused)?;
return Ok(Asked {
words: format!("the file {} ({} bytes)", b.id, bytes.len()),
answer: bytes,
..Asked::default()
});
}
if ask.kind != monetize_embed::fleet::ask::GROW_SET {
return g.ask(c, signature, now).map(Asked::from);
}
if !grow_set_is_ours(state, &c.instance, &c.product) {
return Err(FleetRefusal::Refused(format!(
"{}: grow_set is a sold estate's ask; {} grows through its own lane",
c.instance, c.product
)));
}
let body: monetize_embed::fleet::GrowSetAsk = serde_json::from_value(ask.body)
.map_err(|e| FleetRefusal::Malformed(format!("not a grow_set ask: {e}")))?;
use crate::estate_order::sold::{grow_own, SoldRefusal};
let a =
grow_own(state, &c.instance, &c.role, &body.members, body.gib).map_err(|e| match e {
SoldRefusal::NotSigned => FleetRefusal::Unauthenticated,
SoldRefusal::Refused(w) => FleetRefusal::Refused(w),
SoldRefusal::Unavailable(w) => FleetRefusal::Unavailable(w),
})?;
Ok(Asked {
words: a.words,
..Asked::default()
})
}
/// **The product an estate laid here is sold as**: its site's (`[estates.<e>] product`) when that
/// is another kind than the shop's own sold product (a Holger site on a box that sells Gunnar),
/// else the shop's ([`AppState::sold_product`], as before).
pub fn own_product(state: &AppState, instance: &str) -> Result<String, String> {
// A sold estate's run names its site; an operator site's instance is the site itself.
let site = state
.estates
.get(instance)
.map(|r| r.estate)
.filter(|e| !e.trim().is_empty())
.unwrap_or_else(|| instance.to_string());
product_of_site(state, &site)
}
/// **The product a sold estate's run is** ([`own_product`] for a run in hand): its site's
/// product, or the shop's. What every order and fact signed for that one estate is signed as.
pub fn run_product(
state: &AppState,
run: &crate::estate_order::EstateRun,
) -> Result<String, String> {
let site = if run.estate.trim().is_empty() {
run.company.as_str()
} else {
run.estate.as_str()
};
product_of_site(state, site)
}
fn product_of_site(state: &AppState, site: &str) -> Result<String, String> {
let sold = state.sold_product()?.to_string();
let kind = crate::estate_order::lifecycle::family_of;
match state.registry.product_of_estate(site) {
Some(p) if kind(&p) != kind(&sold) => Ok(p),
_ => Ok(sold),
}
}
/// **Is `product`'s `grow_set` ask from `instance` a sold estate's of that product?**
pub fn grow_set_is_ours(state: &AppState, instance: &str, product: &str) -> bool {
own_product(state, instance).ok().as_deref() == Some(product)
}
/// **A sold estate's install PASSed**: the estate is known to the fleet, each box's key the
/// installer's verdict named is registered, and a box it named none for may pin its own on its
/// first call (TOFU, G-D5).
///
/// ★ F6 (2026-10-20): **an adopted estate's windows stay open until each box pins**
/// ([`UNTIL_PINNED`], `adopted`: F5's re-check found every box PAIRED long after its install
/// began) — test3-holger-rs's main first called home 36 minutes after its adopt, was refused
/// by the 10-minute window and sat at "installing (paired)". The window is bounded by the
/// estate's life: it closes with the pin, or at the estate's terminate
/// ([`monetize_guardian::Store::forget_box`]). A role whose key is already pinned keeps it
/// on an adopt (the same boxes): no window. A fresh install's PASS keeps the guardian's own
/// window (its boxes paired that moment and call home at once; the operator's `site
/// open-pin` is there for one that does not). The pin is the same either way: exactly the
/// first key that calls for a named role. Gunnar and Holger estates take this one path.
pub fn estate_installed(
state: &AppState,
company: &str,
roles: &[String],
appliance_keys: &std::collections::BTreeMap<String, String>,
adopted: bool,
) -> Result<String, String> {
let Some(g) = state.fleet.guardian() else {
return Ok(String::new());
};
g.store
.declare(company, &own_product(state, company)?)
.map_err(|e| e.to_string())?;
let until = if adopted {
UNTIL_PINNED
} else {
state.now() + g.cfg.pin_window_ms
};
let (mut named, mut open, mut kept) = (0, 0, Vec::new());
for role in roles {
match appliance_keys.get(role).filter(|k| !k.is_empty()) {
Some(k) => {
g.store
.register_key(company, role, k)
.map_err(|e| e.to_string())?;
named += 1;
}
None if adopted && g.store.key_of(company, role).is_some() => kept.push(role.as_str()),
None => {
g.store
.open_pin(company, role, until)
.map_err(|e| e.to_string())?;
open += 1;
}
}
}
let kept = if kept.is_empty() {
String::new()
} else {
format!(", {} already pinned ({})", kept.len(), kept.join(", "))
};
let how = if adopted { " (open until pinned)" } else { "" };
Ok(format!("the guardian knows {company}: {named} box key(s) named by the installer, {open} to pin on first call{how}{kept}"))
}
/// **`site open-pin <site>`** (F6, the operator's): every box in `roles` whose key is not pinned
/// may pin its own on its first call again, until it does ([`UNTIL_PINNED`]). A box whose key
/// is pinned is never given a window here: its key is never replaced by this verb. Answers
/// one line naming both.
pub fn open_pins(state: &AppState, company: &str, roles: &[String]) -> Result<String, String> {
let g = state
.fleet
.guardian()
.ok_or("this monetize runs no guardian: no pin window to open")?;
if g.store.product_of(company).is_none() {
return Err(format!(
"the guardian does not know {company} yet: its install has not passed, nothing to pin"
));
}
let (mut opened, mut pinned) = (Vec::new(), Vec::new());
for role in roles {
if g.store.key_of(company, role).is_some() {
pinned.push(role.as_str());
} else {
g.store
.open_pin(company, role, UNTIL_PINNED)
.map_err(|e| e.to_string())?;
opened.push(role.as_str());
}
}
let opened = if opened.is_empty() {
"no pin window opened".to_string()
} else {
format!("pin window opened until pinned for {}", opened.join(", "))
};
let pinned = if pinned.is_empty() {
String::new()
} else {
format!("; already pinned, kept: {}", pinned.join(", "))
};
Ok(format!("{company}: {opened}{pinned}"))
}
/// **Has the estate called home since `since_ms`?** (G1 #14: the pull model's "does the estate
/// answer"). Monetize never dials an estate: it answers when one of its `roles` has sent a
/// Fleet report at or after `since_ms`. Answers the roles heard.
pub fn heard_since(
state: &AppState,
company: &str,
roles: &[String],
since_ms: u64,
) -> Result<Vec<String>, String> {
let g = state
.fleet
.guardian()
.ok_or("the fleet store is not open: monetize cannot hear the estate call home")?;
let heard: Vec<String> = roles
.iter()
.filter(|r| {
g.store
.box_record(company, r)
.is_some_and(|b| b.heard_unix_ms > 0 && b.heard_unix_ms >= since_ms)
})
.cloned()
.collect();
if heard.is_empty() {
return Err(format!("{company} has not called home yet: none of its boxes ({}) has reported on the pull channel since; monetize never dials an estate", roles.join(", ")));
}
Ok(heard)
}
/// **Does the estate SERVE, by its own word?** (S4, the sale check in the pull model). The
/// newest report of its `primary` (its lifecycle's primary role: `main` for Gunnar and Holger)
/// at or after `since_ms` says `serving` = `yes` (not `no`, not `fenced`) and `data` = `ok` —
/// gunnar-server `repair::health`: a main with no identity and no repository (no welcome
/// repository yet) says `fresh`. Answers one line; refused naming the word that is missing. An
/// estate with no `primary` among `roles` has no such word to give, and [`heard_since`] alone
/// answers for it.
pub fn serves_since(
state: &AppState,
company: &str,
roles: &[String],
primary: &str,
since_ms: u64,
) -> Result<String, String> {
if !roles.iter().any(|r| r == primary) {
return Ok(String::new());
}
let g = state
.fleet
.guardian()
.ok_or("the fleet store is not open: monetize cannot hear the estate say it serves")?;
let rec = g
.store
.box_record(company, primary)
.filter(|b| b.heard_unix_ms > 0 && b.heard_unix_ms >= since_ms);
let Some(rec) = rec else {
return Err(format!("{company}'s {primary} has not reported since the install began: the sale waits for its word that it serves"));
};
let word = |k: &str| {
rec.report
.health
.get(k)
.map(|v| v.trim().to_owned())
.unwrap_or_default()
};
let (serving, data) = (word("serving"), word("data"));
let shown = |v: &str| {
if v.is_empty() {
"nothing".to_owned()
} else {
v.to_owned()
}
};
if serving != "yes" {
return Err(format!(
"{company}'s {primary} reports serving = {}, not yes: the sale waits until it serves",
shown(&serving)
));
}
if data != "ok" {
return Err(format!("{company}'s {primary} reports data = {}, not ok (no identity or welcome repository yet): the sale waits", shown(&data)));
}
Ok(format!("the {primary} serves its data"))
}
/// **Declare every pulled product's estate** on the guardian's store (`[products.<id>]
/// instance`, `fleet_keys`): its product and its boxes' keys, so their first report is heard.
pub fn declare_pulled(state: &AppState, cfg: &crate::config::Config) -> anyhow::Result<()> {
let g = state.fleet.guardian().ok_or_else(|| {
anyhow::anyhow!("the fleet store is not open: no pulled product can be declared")
})?;
for (id, p) in &cfg.products {
let Some(instance) = p.instance.as_deref() else {
continue;
};
declare_one(g, id, instance, &p.fleet_keys).map_err(|e| anyhow::anyhow!(e))?;
}
Ok(())
}
/// **The pulled products of a config fed LIVE** (the guest watcher's look, [`crate::guests::look`]):
/// the ONE box's feed arrives after monetize serves, and there are no restarts. Every
/// `[products.<id>] instance` not yet declared as `<id>`, or whose `fleet_keys` differ from the
/// store's, is declared now. Answers one line per estate that changed; a refusal is a line too, and
/// the rest still applies. A block that goes is not undeclared: its estate's ledger rows stay.
pub fn apply_pulled(state: &AppState, cfg: &crate::config::Config) -> Vec<String> {
let Some(g) = state.fleet.guardian() else {
return Vec::new();
};
let mut lines = Vec::new();
for (id, p) in &cfg.products {
let Some(instance) = p.instance.as_deref() else {
continue;
};
let declared = g.store.product_of(instance).as_deref() == Some(id.as_str());
let keys_held = p.fleet_keys.iter().all(|(role, key)| {
g.store
.key_of(instance, role)
.is_some_and(|k| monetize_embed::signing::base64_encode(k.as_bytes()) == key.trim())
});
if declared && keys_held {
continue;
}
lines.push(match declare_one(g, id, instance, &p.fleet_keys) {
Ok(()) => format!("[products.{id}] instance = {instance:?} DECLARED live: its boxes' first reports are heard"),
Err(e) => format!("[products.{id}] REFUSED live: {e}"),
});
}
lines
}
fn declare_one(
g: &Guardian,
id: &str,
instance: &str,
fleet_keys: &std::collections::BTreeMap<String, String>,
) -> Result<(), String> {
g.store
.declare(instance, id)
.map_err(|e| format!("[products.{id}] instance = {instance:?}: {e}"))?;
for (role, key) in fleet_keys {
g.store
.register_key(instance, role, key)
.map_err(|e| format!("[products.{id}] fleet_keys.{role}: {e}"))?;
}
Ok(())
}
/// The guardian over a fresh in-memory store, signing with `keys` — tests and `--dry`.
pub fn in_memory_guardian(keys: Arc<monetize::Keyring>) -> Guardian {
let store = monetize_guardian::Store::in_memory().expect("in-memory fleet store");
// The server asks the provider looks itself ([`tick`], [`tick_on`]), never inside the tick.
Guardian::new(
Arc::new(store),
keys,
monetize_guardian::GuardianConfig {
looks_in_tick: false,
..Default::default()
},
)
}
#[cfg(test)]
mod tests;
/// **What a box's report that acked orders sets moving at once** (i54a A2, 2026-10-12): the
/// pending signups are settled again, so an account the main just made gets its plan's
/// entitlement queued for the main's open long-poll now, not at the poller's next tick.
/// Answers how many signups it decided.
pub fn after_report(state: &AppState, acked: &[u64]) -> usize {
if acked.is_empty()
|| state
.signups
.pending()
.map(|p| p.is_empty())
.unwrap_or(true)
{
return 0;
}
crate::signup::settle_pending(state)
}