crates/monetize-server/src/estate_order/change.rs
//! **Plan changes on a delivered estate** (PRIVATE-GUNNAR.md §10): more or
//! fewer cores and memory for main and the front, chosen by the buyer from
//! the provider's plans; the twin follows main (an eighth of it, the rule
//! the order used) and grows by itself when it stays behind
//! ([`super::twin`]).
//!
//! ```text
//! ChangePlan(sizes) ─▶ priced at the provider's price NOW + 20 %
//! (refused over the estate's spend cap)
//! dearer ─▶ waiting_for_payment: the difference for the rest of the
//! paid period is HELD (a month is 30 days)
//! cheaper ─▶ queued: nothing now (the period is prepaid; nothing is
//! refunded), the lower price from the next renewal
//! queued ─▶ the price read again at the buy (under 10 %: nothing is
//! changed, the hold released) ─▶ resizing
//! resizing ─▶ each server: stopped, its plan changed, started — the front,
//! the twin, main LAST (its downtime is the forge's, said in
//! the order's words); journaled; resumed after a restart
//! the estate answers again ─▶ the hold CAPTURED ─▶ done (mail)
//! refused / no answer ─▶ the hold RELEASED ─▶ failed (mail)
//! ```
//!
//! **A data step** (one shop, 2026-10-07: a step of the ladder that holds more
//! data) is priced here the same way — main's four data disks and the twin's
//! at the new size, the difference for the days left held — but no disk is
//! resized here: a disk grows while the estate's Gunnar has it unmounted, so
//! once the money is captured this shop SIGNS a growth go-ahead
//! ([`super::sold::grant`]) and the estate grows by it (G16: no monetize of its own), its
//! cloud step done at this shop (`SoldEstate.GrowSet`). Data never shrinks.
use monetize_cloud::lay::PlanSpec;
use monetize_cloud::{Cloud, CloudError, Price, ResourceId, Zone};
use monetize_pay::{ChargeHandle, ChargeRequest, ChargeStatus, Money};
use super::term::DAY_MS;
use super::twin::{decide, TwinContext, TwinVerdict};
use super::{
iso_date, phase, price, EstateOrderConfig, EstateRun, EstateSizes, Quote, StartRefusal,
Transition,
};
use crate::state::AppState;
/// The change's status words (the customer's and the operator's).
pub mod status {
pub const WAITING_FOR_PAYMENT: &str = "waiting_for_payment";
pub const QUEUED: &str = "queued";
pub const RESIZING: &str = "resizing";
pub const DONE: &str = "done";
pub const FAILED: &str = "failed";
}
/// A change that resized and does not call home within this long fails (the
/// machines stay on their new plans; an operator looks).
pub const ANSWER_WITHIN_MS: u64 = 30 * 60 * 1000;
/// One server's plan change within a [`PlanChange`].
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct PlanStep {
/// `front` | `twin` | `main`.
pub role: String,
/// The provider's server id.
pub server: String,
pub from: String,
pub to: String,
/// What the new plan adds a month at the provider (0 for a smaller one).
pub delta_minor: u64,
pub done: bool,
}
/// **A plan change**, as the run keeps it.
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct PlanChange {
/// `estate/<company>/change-<n>`: the vendor's reference.
pub reference: String,
/// `buyer` (ChangePlan) or `twin` (the twin grew by itself).
pub kind: String,
pub status: String,
pub from: EstateSizes,
pub from_twin: String,
pub to: EstateSizes,
pub to_twin: String,
/// The estate's price after the change (today's provider price + the margin).
pub quote: Quote,
/// Held now: the monthly difference for the days left of the paid period.
pub amount: Money,
pub handle: Option<ChargeHandle>,
pub checkout_url: String,
pub steps: Vec<PlanStep>,
pub opened_at_ms: u64,
pub resized_at_ms: u64,
/// What the change does, as the buyer is told (the downtime included).
pub words: String,
/// Why it failed, or what it waits for.
pub why: String,
/// **The day's batch** (the room rule, 2026-10-05): a change the twin made
/// for itself is resized no sooner than this — the next daily boundary
/// when `grow_batch_every_secs` is on, else at once (0).
#[serde(default)]
pub not_before_ms: u64,
/// When it was first queued to resize: a buyer's change that stops
/// machines waits [`super::EstateOrderConfig::resize_notice_secs`] from
/// here, announced to the estate ([`notice`]).
#[serde(default)]
pub queued_at_ms: u64,
/// **The machines the change lays** (P3-FACTS: a plan change that ADDS machines, the Lake's
/// backends 1 → 2 lay `be-2`), as the product's plan names them; laid once the money is held
/// and the machines that stand are resized, installed into the estate by its installer, and
/// heard on the pull channel before the hold is captured. Which roles a change may add is
/// the product's grow rule ([`crate::facts::GrowRule::may_add`]).
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub adds: Vec<AddStep>,
/// **The machines the change takes away** (A4-FACTS: the herd's nodes 2 → 1 takes `node-2`
/// away), as laid: each emptied first by the product's drain (a signed order of its own kind
/// the box pulls and acks), then given back whole at the cloud, read gone, forgotten by the
/// guardian and named no more by the estate's DNS, peers and firewall. Which roles may go is
/// the product's grow rule ([`crate::facts::GrowRule::may_remove`], its `drain`, `moves_data`).
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub removes: Vec<RemoveStep>,
/// **The disks of standing machines the change grows** (A4-FACTS: new numbers that grow a
/// standing machine's data disk, the Lake's store over its backends), each grown at the cloud
/// once the money is held (detached, resized, put back at its address), its size read back.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub grows: Vec<DiskStep>,
}
/// One machine a [`PlanChange`] takes away from the estate that stands.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct RemoveStep {
/// The machine's role as laid (`node-2`).
pub role: String,
/// The provider's server id, as laid.
pub server: String,
/// The seq of the drain order queued for it (0: not queued yet).
#[serde(default)]
pub drain_seq: u64,
/// When its drain was queued, and when it acked that it was emptied (0: not yet).
#[serde(default)]
pub drain_at_ms: u64,
#[serde(default)]
pub drained_at_ms: u64,
/// When the cloud was asked to give it back, and when the listing read it gone (0: not yet).
#[serde(default)]
pub retired_at_ms: u64,
#[serde(default)]
pub gone_at_ms: u64,
}
/// The data disks of one standing machine a [`PlanChange`] grows to one size.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct DiskStep {
/// The machine's role as laid (`node`, `be-2`).
pub role: String,
/// The provider's server id.
pub server: String,
/// The provider's ids of the disks that grow.
pub members: Vec<String>,
pub from_gib: u64,
pub to_gib: u64,
#[serde(default)]
pub done: bool,
}
/// One machine a [`PlanChange`] lays beside the estate that stands.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct AddStep {
/// The machine's role as laid (`be-2`).
pub role: String,
pub plan: String,
/// The provider's server id, once laid.
#[serde(default)]
pub server: String,
/// When it was laid and installed (0: not yet); its first report must come after.
#[serde(default)]
pub laid_at_ms: u64,
}
impl PlanChange {
fn open(&self) -> bool {
!matches!(self.status.as_str(), status::DONE | status::FAILED)
}
/// GB a disk main's data grows to with this change; 0 when it holds no data step.
pub fn data_to_gib(&self) -> u64 {
if self.to.main_data_gib > self.from.main_data_gib {
self.to.main_data_gib
} else {
0
}
}
}
/// Pure: **the numbers a change asks, over the numbers laid**: each value `asked` names in its
/// place, the rest as laid. An order that is not an object over one that is, or none laid, is
/// `asked` as it is (the form then says what is wrong with it).
pub fn over(laid: &serde_json::Value, asked: &serde_json::Value) -> serde_json::Value {
match (laid, asked) {
(serde_json::Value::Object(l), serde_json::Value::Object(a)) => {
let mut out = l.clone();
out.extend(a.iter().map(|(k, v)| (k.clone(), v.clone())));
serde_json::Value::Object(out)
}
_ => asked.clone(),
}
}
/// **The difference for the rest of the paid period**: the monthly price's
/// rise × the days left (rounded up) / 30. A cheaper plan holds nothing.
pub fn prorated(old_monthly: Money, new_monthly: Money, now_ms: u64, paid_until_ms: u64) -> Money {
let rise = new_monthly.minor.saturating_sub(old_monthly.minor);
let days = paid_until_ms.saturating_sub(now_ms).div_ceil(DAY_MS);
Money {
minor: (rise.saturating_mul(days)).div_ceil(30),
currency: new_monthly.currency,
}
}
/// **The roles a change speaks of by their part** (P3-FACTS): the product's primary and its
/// twin, from its facts — Gunnar's and Holger's `main` and `twin`; the Lake's `fe` and none.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Parts {
pub primary: String,
pub twin: Option<String>,
}
impl Parts {
/// The parts `cfg`'s product's lifecycle names; none for a product with no lifecycle.
pub fn of(cfg: &EstateOrderConfig) -> Parts {
super::lifecycle::of(cfg)
.map(|l| Parts {
primary: l.primary_role(),
twin: l.twin().map(|t| t.role),
})
.unwrap_or_default()
}
}
/// Pure: **the server steps from what is laid to what is asked** — every machine that stands
/// whose plan in `quote` is another, in the order they are laid with the primary LAST (its
/// downtime is the estate's): Gunnar's front, twin, main. A machine the quote does not name
/// (one a change takes away) is not resized; one the estate does not have yet is laid by the
/// change ([`adds_for`]), never resized.
pub fn steps_for(
run: &EstateRun,
catalog: &[PlanSpec],
quote: &Quote,
parts: &Parts,
) -> Result<Vec<PlanStep>, String> {
let laid = run.laid.as_ref().ok_or("the estate has nothing laid")?;
let price_of = |plan: &str| {
catalog
.iter()
.find(|p| p.name == plan)
.map(|p| p.price.minor_per_month)
.unwrap_or(0)
};
let mut out = Vec::new();
let mut last = Vec::new();
for m in &laid.machines {
let Some(to) = quote
.machines
.iter()
.find(|q| q.role == m.role)
.map(|q| q.plan.as_str())
else {
continue;
};
if m.plan != to {
let step = PlanStep {
role: m.role.clone(),
server: m.server.clone(),
from: m.plan.clone(),
to: to.to_string(),
delta_minor: price_of(to).saturating_sub(price_of(&m.plan)),
done: false,
};
if m.role == parts.primary {
last.push(step)
} else {
out.push(step)
}
}
}
out.extend(last);
Ok(out)
}
/// Pure: **the machines a change adds and takes away**: the roles of `quote` that are not laid,
/// and the laid roles `quote` does not name, each in its own order.
pub fn adds_for(run: &EstateRun, quote: &Quote) -> (Vec<AddStep>, Vec<String>) {
let laid: Vec<&str> = run
.laid
.as_ref()
.map(|l| l.machines.iter().map(|m| m.role.as_str()).collect())
.unwrap_or_default();
let adds = quote
.machines
.iter()
.filter(|q| !laid.contains(&q.role.as_str()))
.map(|q| AddStep {
role: q.role.clone(),
plan: q.plan.clone(),
server: String::new(),
laid_at_ms: 0,
})
.collect();
let gone = laid
.iter()
.filter(|r| !quote.machines.iter().any(|q| q.role == **r))
.map(|r| r.to_string())
.collect();
(adds, gone)
}
/// Pure: **the data disks of standing machines the new plan grows** (A4-FACTS): for each machine
/// that stands and that `quote` names, its data disks (the system disk is the first) the quote
/// sizes bigger, one [`DiskStep`] per machine and size. `Err` names a system disk that would grow:
/// a system disk is the release's, never grown by a plan change.
pub fn grows_for(run: &EstateRun, quote: &Quote) -> Result<Vec<DiskStep>, String> {
let Some(laid) = run.laid.as_ref() else {
return Ok(Vec::new());
};
let mut out: Vec<DiskStep> = Vec::new();
for m in &laid.machines {
let Some(line) = quote.machines.iter().find(|q| q.role == m.role) else {
continue;
};
for (i, (want, had)) in line.disks_gib.iter().zip(m.disks.iter()).enumerate() {
if *want <= had.gib {
continue;
}
if i == 0 {
return Err(format!("{}'s system disk would grow from {} to {want} GiB: a plan change grows data disks only", m.role, had.gib));
}
match out
.iter_mut()
.find(|s| s.role == m.role && s.to_gib == *want)
{
Some(s) => {
s.members.push(had.id.clone());
s.from_gib = s.from_gib.min(had.gib);
}
None => out.push(DiskStep {
role: m.role.clone(),
server: m.server.clone(),
members: vec![had.id.clone()],
from_gib: had.gib,
to_gib: *want,
done: false,
}),
}
}
}
Ok(out)
}
/// Pure: **the laid machines a change takes away**, as [`RemoveStep`]s, in the order laid.
pub fn removes_for(run: &EstateRun, gone: &[String]) -> Vec<RemoveStep> {
let Some(laid) = run.laid.as_ref() else {
return Vec::new();
};
laid.machines
.iter()
.filter(|m| gone.contains(&m.role))
.map(|m| RemoveStep {
role: m.role.clone(),
server: m.server.clone(),
drain_seq: 0,
drain_at_ms: 0,
drained_at_ms: 0,
retired_at_ms: 0,
gone_at_ms: 0,
})
.collect()
}
/// Pure: **does the laid machine `role` hold data** (a disk beyond its system disk)?
pub fn holds_data(run: &EstateRun, role: &str) -> bool {
run.laid
.as_ref()
.and_then(|l| l.machine(role))
.is_some_and(|m| m.disks.len() > 1)
}
fn describe(steps: &[PlanStep], parts: &Parts) -> String {
steps
.iter()
.map(|s| {
if s.role == parts.primary {
format!("{p} {} → {} ({p} is stopped for a few minutes while its plan changes: the estate does not serve meanwhile)", s.from, s.to, p = s.role)
} else if parts.twin.as_deref() == Some(s.role.as_str()) {
format!("the {} follows: {} → {}", s.role, s.from, s.to)
} else {
format!("{} {} → {} (stopped for a few minutes)", s.role, s.from, s.to)
}
})
.collect::<Vec<_>>()
.join("; ")
}
/// The machines a change lays, as words.
fn describe_adds(adds: &[AddStep]) -> Option<String> {
(!adds.is_empty()).then(|| {
format!(
"{} laid beside the machines that stand (nothing that stands is stopped for it)",
adds.iter()
.map(|a| format!("a new {} ({})", a.role, a.plan))
.collect::<Vec<_>>()
.join(", ")
)
})
}
/// **What a sold estate is told while its machines are being resized**: the
/// change's words and until when — while a buyer's change that stops servers
/// is queued or resizing; nothing otherwise.
pub fn notice(cfg: &EstateOrderConfig, run: &EstateRun, now: u64) -> Option<(String, u64)> {
let c = run.change.as_ref().filter(|c| {
c.kind == "buyer" && matches!(c.status.as_str(), status::QUEUED | status::RESIZING)
})?;
let left: Vec<PlanStep> = c.steps.iter().filter(|s| !s.done).cloned().collect();
if left.is_empty() {
return None;
}
let from = if c.queued_at_ms > 0 {
c.queued_at_ms
} else {
now
};
let until = from
.saturating_add(cfg.resize_notice_secs.saturating_mul(1_000))
.saturating_add(ANSWER_WITHIN_MS)
.max(now + 60_000);
Some((
format!(
"{} is being resized at {} ({}): {}",
cfg.host_of(&run.company),
cfg.apex,
c.reference,
describe(&left, &Parts::of(cfg))
),
until,
))
}
/// The data step as words: main's four disks, and the twin's that holds them.
fn describe_data(from_gib: u64, to_gib: u64, twin_gib: u64) -> String {
format!(
"main's data disks grow from {from_gib} to {to_gib} GB each and the twin's to {twin_gib} GB, once paid (the data set pauses for a few minutes while each disk grows; nothing is stopped)"
)
}
/// The disks a change grows, as words.
fn describe_grows(grows: &[DiskStep]) -> Option<String> {
(!grows.is_empty()).then(|| {
let each = grows.iter().map(|g| format!("{}'s data disk grows from {} to {} GiB", g.role, g.from_gib, g.to_gib)).collect::<Vec<_>>().join(", ");
format!("{each}, once paid (each disk is detached for a few minutes while it grows and put back at its place; its machine is not stopped)")
})
}
/// The machines a change takes away, as words.
fn describe_removes(removes: &[RemoveStep]) -> Option<String> {
(!removes.is_empty()).then(|| {
let names = removes.iter().map(|r| r.role.clone()).collect::<Vec<_>>();
format!("{} {} emptied and taken away (the estate moves what it holds to the machines that stay first; nothing that stays is stopped for it)", super::lifecycle::listed(&names, "and"), if names.len() == 1 { "is" } else { "are" })
})
}
/// What a change does, as the buyer is told: the machines resized, the disks grown, the machines
/// laid, the machines taken away, then the data.
fn describe_change(
steps: &[PlanStep],
grows: &[DiskStep],
adds: &[AddStep],
removes: &[RemoveStep],
data: Option<String>,
roles: &Parts,
) -> String {
let mut parts = Vec::new();
if !steps.is_empty() {
parts.push(describe(steps, roles));
}
parts.extend(describe_grows(grows));
parts.extend(describe_adds(adds));
parts.extend(describe_removes(removes));
parts.extend(data);
parts.join("; ")
}
/// **`EstateOrder.ChangePlan`**: change main's and/or the front's plan on a
/// delivered estate. Empty sizes keep what is laid. Refused by name before
/// any money moves.
pub fn change(
state: &AppState,
company: &str,
asked: &EstateSizes,
) -> Result<PlanChange, StartRefusal> {
change_back(state, company, asked, None)
}
/// [`change`], the payer sent back to `back` after the vendor's page (the ONE shop's own status
/// page, lane G12); `None`: [`change_back_to`].
pub fn change_back(
state: &AppState,
company: &str,
asked: &EstateSizes,
back: Option<&str>,
) -> Result<PlanChange, StartRefusal> {
change_paid_by(state, company, asked, back, None)
}
/// [`change_back`], a dearer change held from `payer` instead of the run's address: the site's
/// vendor's TEST card for the operator's test change of his own site (MONETIZE-FIX,
/// [`super::site::site_test_card`]), nobody at a page.
pub fn change_paid_by(
state: &AppState,
company: &str,
asked: &EstateSizes,
back: Option<&str>,
payer: Option<&str>,
) -> Result<PlanChange, StartRefusal> {
let eo = &state.estates;
let base = eo.cfg.as_ref().ok_or_else(|| {
StartRefusal::NotSold(
"this monetize sells no private estates ([estate_orders] is absent)".into(),
)
})?;
// ★ ST-5: a plan change is held through the account of the site the estate was bought
// on, and named on that site — never a fixed `[estate_orders] vendor`.
let site_terms = match eo.get(company) {
Some(run) => super::terms_of(state, base, &run).map_err(StartRefusal::NotSold)?,
None => base.clone(),
};
let cfg = &site_terms;
// Refused at once when there is nothing to change: a run being laid or
// installed holds its pen for as long as that takes.
changeable(cfg, company, eo.get(company))?;
// **The run's pen, from the read to the write** (`super::Pens`): the
// driver's step writes nothing between them, and this writes over nothing.
let _pen = eo.pen_for_a_call(cfg, company)?;
let mut run = changeable(cfg, company, eo.get(company))?;
let now_sizes = run.quote.sizes.clone();
// ★ The product's own order (v2.0 P3) is checked against its form and filled in, as at the
// start, so the numbers asked are compared with the numbers laid.
// A change names the numbers it changes: the ones it does not name stay as laid (never the
// form's defaults, which would shrink what the buyer has).
let over_laid = EstateSizes {
order: over(&now_sizes.order, &asked.order),
..asked.clone()
};
let filled = match super::lifecycle::of(cfg) {
Ok(l) if !asked.order.is_null() => {
price::order_of(&l.facts(), &over_laid).map_err(StartRefusal::BadName)?
}
_ => asked.clone(),
};
let asked = &filled;
let life = super::lifecycle::of(cfg).ok();
let rule = life
.as_ref()
.map(|l| l.grow())
.unwrap_or_else(crate::facts::GrowRule::sizes);
let to = EstateSizes {
front_plan: if asked.front_plan.trim().is_empty() {
now_sizes.front_plan.clone()
} else {
asked.front_plan.trim().to_string()
},
main_plan: if asked.main_plan.trim().is_empty() {
now_sizes.main_plan.clone()
} else {
asked.main_plan.trim().to_string()
},
main_data_gib: if asked.main_data_gib == 0 {
now_sizes.main_data_gib
} else {
asked.main_data_gib
},
order: if asked.order.is_null() {
now_sizes.order.clone()
} else {
asked.order.clone()
},
};
// **What never shrinks does not** (P3-FACTS: the product's grow rule; one shop, 2026-10-07:
// main's data disks): a change that holds less is refused; more is a data step.
let json = |s: &EstateSizes| serde_json::to_value(s).unwrap_or_default();
if let Some((field, now, was)) = rule.shrinks(&json(&now_sizes), &json(&to)) {
return Err(StartRefusal::BadName(if field == "main_data_gib" {
format!("data never shrinks: main's data disks are {now} GB each now, and {was} GB was asked")
} else {
format!("data never shrinks: {field} is {now} now, and {was} was asked")
}));
}
if to == now_sizes {
return Err(StartRefusal::NotNow("nothing to change: front and main are already on those plans and main's data at that size".into()));
}
let cloud = state.registry.find_cloud(&cfg.cloud).ok_or_else(|| {
StartRefusal::NotSold(format!("the cloud {:?} is not linked here", cfg.cloud))
})?;
let q = price::quote_with_twin(cfg, cloud, company, &to, &run.twin_floor)
.map_err(StartRefusal::Price)?;
// The estate's spend cap is never the buyer's to meet (A53): a change
// over it is taken, and waits in `queued` until the operator raises it.
let catalog = monetize_cloud::ask::plan_catalog(&cloud, &Zone(cfg.zone.clone()))
.map_err(|e| StartRefusal::Price(format!("the provider's plan list: {e}")))?;
let parts = Parts::of(cfg);
let steps = steps_for(&run, &catalog, &q, &parts).map_err(StartRefusal::NotNow)?;
// **Machines added or taken away** (P3-FACTS): what the product's new plan lays that is not
// laid, and what is laid that it no longer names, each allowed by its grow rule — refused
// by name, nothing held. A change lays new machines; none is taken away yet.
let (adds, gone) = adds_for(&run, &q);
let role_facts = life.as_ref().map(|l| l.role_facts()).unwrap_or_default();
let names = |a: &[AddStep]| a.iter().map(|x| x.role.clone()).collect::<Vec<_>>();
if let Some(why) = rule.machines_refusal(&role_facts, &names(&adds), &gone) {
return Err(StartRefusal::NotNow(format!("{why}; nothing was held")));
}
// **A machine taken away is emptied first** (A4-FACTS): one that holds data goes only when
// the product's estate moves that data off it (its grow rule's `moves_data`), else refused
// by name before anything is held.
if let Some(why) = rule.data_refusal(&role_facts, &gone, &|m: &str| holds_data(&run, m)) {
return Err(StartRefusal::NotNow(format!("{why}; nothing was held")));
}
let removes = removes_for(&run, &gone);
// **New numbers that grow a STANDING machine's disk** (A4-FACTS: the Lake's store over the
// backends that stand, the herd's node_gib): grown at the cloud by this change once paid. A
// data step of the sizes (`main_data_gib`) stays the estate's own growth (the go-ahead).
let grows = if to.order != now_sizes.order {
grows_for(&run, &q)
.map_err(|why| StartRefusal::NotNow(format!("{why}; nothing was held")))?
} else {
Vec::new()
};
if !grows.is_empty() {
if let Err(e) = monetize_cloud::ask::grow_gate(&cloud) {
let names: Vec<String> = grows.iter().map(|g| g.role.clone()).collect();
return Err(StartRefusal::NotNow(format!("the new numbers grow the data disks of {}, and this cloud grows no disk ({e}); nothing was held", super::lifecycle::listed(&names, "and"))));
}
}
// ★ A change of a site the OPERATOR owns meets the cap before it is paid (owner
// 2026-10-14): refused plainly, in EUR, nothing held. A customer's is taken and waits.
if run.site_order {
let currency = run.quote.cost.currency;
let laid_more = if adds.is_empty() && grows.is_empty() {
0
} else {
q.cost_in_sale
.minor
.saturating_sub(run.quote.cost_in_sale.minor)
.saturating_sub(steps.iter().map(|s| s.delta_minor).sum())
};
let more = Money {
minor: steps.iter().map(|s| s.delta_minor).sum::<u64>() + laid_more,
currency,
};
super::operator_cap_first(
cloud,
q.cost_in_sale,
super::effective_cap(cfg, &run),
more,
&plan_keys(&steps),
)?;
}
let now = state.now();
let amount = prorated(run.quote.monthly, q.monthly, now, run.paid_until_ms);
let reference = format!("{}/change-{}", run.reference, run.changes.len() + 1);
let host = if run.domain.is_empty() {
cfg.host_of(company)
} else {
run.domain.clone()
};
let days = run.paid_until_ms.saturating_sub(now).div_ceil(DAY_MS);
let money_words = if amount.minor > 0 {
format!("The price becomes {} a month; for the {days} day(s) left of the period paid until {}, {} is held now and charged when the change is done.", q.monthly.in_words(), iso_date(run.paid_until_ms), amount.in_words())
} else {
format!("The period is paid until {}; nothing more is charged now, and from the renewal the price is {} a month.", iso_date(run.paid_until_ms), q.monthly.in_words())
};
let data = (q.sizes.main_data_gib > now_sizes.main_data_gib).then(|| {
let twin = parts
.twin
.as_ref()
.and_then(|t| q.machines.iter().find(|m| &m.role == t))
.and_then(|m| m.disks_gib.get(1))
.copied()
.unwrap_or(0);
describe_data(now_sizes.main_data_gib, q.sizes.main_data_gib, twin)
});
let words = format!(
"{}. {money_words}",
describe_change(&steps, &grows, &adds, &removes, data, &parts)
);
let (status, handle, checkout_url) = if amount.minor > 0 {
let vendor = state.registry.find_vendor(&cfg.vendor).ok_or_else(|| {
StartRefusal::Vendor(format!("the vendor {:?} is not linked here", cfg.vendor))
})?;
let req = ChargeRequest {
reference: reference.clone(),
amount,
market: amount.currency.market(),
description: format!("{} — {host}: a plan change for the rest of the paid period (charged when it is done)", super::lifecycle::name_of(cfg)),
payer: Some(payer.map(str::to_owned).unwrap_or_else(|| run.email.clone())),
};
// An OPERATOR SITE's change comes back to the ONE box's Sites page: the site's own
// front is the machine being resized (live 2026-10-14: back to
// `https://test2.gunnar.rs/paid/estate/…/change-1`, down during the resize).
let reserved = match back
.map(str::to_owned)
.or_else(|| change_back_to(cfg, &run))
{
Some(back) => vendor.reserve(&req, &monetize_pay::payment::Return::to(&back, "")),
// A16: back to the site it was bought on, whichever instance takes the money.
None => vendor.reserve(
&req,
&crate::returns::back_for(
state.registry.site_of_estate(&run.estate).as_deref(),
&req.reference,
),
),
};
let h = reserved
.map_err(|e| StartRefusal::Vendor(format!("the vendor refused the hold: {e}")))?;
let url = vendor.page_url(&h).unwrap_or_default();
(status::WAITING_FOR_PAYMENT, Some(h), url)
} else {
(status::QUEUED, None, String::new())
};
let c = PlanChange {
reference,
kind: "buyer".into(),
status: status.into(),
from: now_sizes,
from_twin: run.quote.twin_plan.clone(),
to: q.sizes.clone(),
to_twin: q.twin_plan.clone(),
quote: q,
amount,
handle,
checkout_url,
steps,
opened_at_ms: now,
resized_at_ms: 0,
words,
why: String::new(),
not_before_ms: 0,
queued_at_ms: 0,
adds,
removes,
grows,
};
run.change = Some(c.clone());
run.history.push(Transition {
phase: String::new(),
at_ms: now,
note: format!("plan change {} opened: {}", c.reference, c.words),
});
super::save(state, &run).map_err(StartRefusal::NotSold)?;
Ok(c)
}
/// Pure: **where the payment page sends the payer back after a change** — `None`: the
/// vendor's own return (the selling site's page, a customer's; it stays up while their
/// machines restart). An operator site's change returns to the ONE box's Sites page
/// (`<shop_door>/sites`), never to the site being resized.
pub fn change_back_to(cfg: &EstateOrderConfig, run: &EstateRun) -> Option<String> {
if !run.site_order {
return None;
}
cfg.shop_door().ok().map(|door| format!("{door}/sites"))
}
/// The run, if it may change plans now: delivered, paid, no change open.
fn changeable(
cfg: &EstateOrderConfig,
company: &str,
run: Option<EstateRun>,
) -> Result<EstateRun, StartRefusal> {
let run = run.ok_or_else(|| {
StartRefusal::NotNow(format!("there is no estate {}", cfg.host_of(company)))
})?;
if run.phase != phase::CAPTURED {
return Err(StartRefusal::NotNow(format!(
"{} is {}: only a delivered, paid estate changes plans",
cfg.host_of(company),
run.phase
)));
}
if let Some(c) = run.change.as_ref().filter(|c| c.open()) {
return Err(StartRefusal::NotNow(format!(
"a plan change is already {} ({})",
c.status, c.reference
)));
}
Ok(run)
}
/// **Drive the change in flight** — or, with none, ask whether the twin
/// should grow by itself. Mutates the run (the caller saves it); answers the
/// change's state in words, if there is one.
pub(super) fn advance(
state: &AppState,
cfg: &EstateOrderConfig,
run: &mut EstateRun,
now: u64,
) -> Option<String> {
if run.change.is_none() {
// An adopted site's twin is its kit's: it does not grow by itself here.
if run.phase == phase::CAPTURED && !super::adopt::is_adopted(run) {
autoscale(state, cfg, run, now);
}
run.change.as_ref()?;
}
let cloud = state.registry.find_cloud(&cfg.cloud)?;
for _ in 0..8 {
let mut c = run.change.clone()?;
let before = c.status.clone();
match c.status.as_str() {
status::WAITING_FOR_PAYMENT => waiting(state, cfg, run, &mut c, now),
status::QUEUED => queued(cfg, cloud, run, &mut c, now),
status::RESIZING => resizing(state, cfg, cloud, run, &mut c, now),
status::DONE | status::FAILED => {
finish(state, cfg, run, c, now);
return None;
}
other => c.why = format!("nothing drives the status {other:?}"),
}
let moved = c.status != before;
if moved {
run.history.push(Transition {
phase: String::new(),
at_ms: now,
note: format!(
"plan change {}: {before} → {}{}",
c.reference,
c.status,
if c.why.is_empty() {
String::new()
} else {
format!(" ({})", c.why)
}
),
});
}
let why = c.why.clone();
run.change = Some(c);
if !moved {
return Some(why);
}
// Every finished step is written down before the next is asked for.
if !super::kept(
state,
run,
&format!(
"plan change {} moved",
run.change
.as_ref()
.map(|c| c.reference.as_str())
.unwrap_or("")
),
) {
return Some(format!("the plan change's step is not written down (CODE RED save:{}:run); it goes on once the run is written", run.company));
}
}
None
}
fn waiting(
state: &AppState,
cfg: &EstateOrderConfig,
run: &mut EstateRun,
c: &mut PlanChange,
now: u64,
) {
let Some(vendor) = state.registry.find_vendor(&cfg.vendor) else {
c.why = format!("the vendor {:?} is not linked", cfg.vendor);
return;
};
let Some(handle) = c.handle.clone() else {
c.status = status::QUEUED.into();
return;
};
if run.phase != phase::CAPTURED {
let closed = super::pages::close(vendor, &handle);
super::pages::settle(
run,
&c.reference,
&handle,
closed,
"the estate's paid period ended",
now,
);
fail(c, format!("the estate's paid period ended ({}); the change is called off and nothing is charged", run.phase));
return;
}
match vendor.charge(&handle) {
Ok(ChargeStatus::Reserved) => {
c.status = status::QUEUED.into();
c.why = String::new();
}
Ok(ChargeStatus::AwaitingCustomer) if now > c.opened_at_ms.saturating_add(cfg.order_ttl_secs.saturating_mul(1_000)) => match super::pages::close(vendor, &handle) {
super::pages::Closed::Closed => fail(c, "the payment page expired unpaid; nothing was changed".into()),
// ★ F-S-02: paid in its last second — the change goes on (captured once resized).
super::pages::Closed::Held => {
c.status = status::QUEUED.into();
c.why = String::new();
}
super::pages::Closed::Paid(_) => fail(c, "the vendor CAPTURED the money before the change; it must be refunded".into()),
super::pages::Closed::Failed(e) => c.why = format!("the payment page expired, and closing it did not go through yet: {e}; asking again"),
},
Ok(ChargeStatus::AwaitingCustomer) => c.why = "waiting for the payment".into(),
Ok(ChargeStatus::Released) | Ok(ChargeStatus::Declined { .. }) => fail(c, "the payment was not authorised; nothing was changed".into()),
Ok(ChargeStatus::Captured { .. }) => fail(c, "the vendor CAPTURED the money before the change; it must be refunded".into()),
Err(e) => c.why = format!("the vendor did not answer: {e}"),
}
}
/// A change the cap holds back: the operator's ALERT (once, again when its
/// words change, and an hourly reminder — [`super::wait_alert_due`]) and the
/// run's `why`. The buyer's `change_words` stay what the change does.
fn cap_waits(reference: &str, words: &str, now: u64) -> String {
super::cap_alert(reference, words, false, now);
words.to_string()
}
/// The spend cap's reservation keys of the change's machines (`plan:<server>`):
/// a resumed change's own, never counted against it.
fn plan_keys(steps: &[PlanStep]) -> Vec<String> {
steps.iter().map(|s| format!("plan:{}", s.server)).collect()
}
fn fail(c: &mut PlanChange, why: String) {
c.status = status::FAILED.into();
c.why = why;
}
/// **At the buy**: the provider's price read again for exactly what is about
/// to be bought — under the floor (a buyer's change) or over the cap,
/// nothing is changed.
fn queued(
cfg: &EstateOrderConfig,
cloud: &crate::hot::HotCloud,
run: &EstateRun,
c: &mut PlanChange,
now: u64,
) {
if now < c.not_before_ms {
c.why = format!(
"waiting for the next growth window at {}",
iso_time(c.not_before_ms)
);
return;
}
if c.kind == "buyer" && run.phase != phase::CAPTURED {
fail(c, format!("the estate's paid period ended ({}); the change is called off and nothing is charged", run.phase));
return;
}
// **The estate is told before its machines stop** (one shop 4): a
// buyer's change that stops servers waits the notice, announced.
if c.kind == "buyer" && !c.steps.is_empty() && cfg.resize_notice_secs > 0 {
if c.queued_at_ms == 0 {
c.queued_at_ms = now;
}
let at = c
.queued_at_ms
.saturating_add(cfg.resize_notice_secs.saturating_mul(1_000));
if now < at {
c.why = format!(
"{} is told first: its machines stop at {}",
cfg.host_of(&run.company),
iso_time(at)
);
return;
}
}
let plan = match super::lifecycle::lay_plan_of(cfg, &run.company, &c.to, &c.to_twin) {
Ok(p) => p,
Err(e) => return fail(c, e),
};
let live =
match monetize_cloud::Cloud::price(&cloud, &monetize_cloud::cloud::Priced::Plan(&plan)) {
Ok(p) => p,
Err(CloudError::Unavailable(w)) => {
c.why = format!("the provider's price could not be read; asking again: {w}");
return;
}
Err(e) => return fail(c, format!("the provider's price: {e}; nothing was changed")),
};
let live = Money {
minor: live.minor_per_month,
currency: live.currency,
};
// **Over the estate's spend cap: the change WAITS** (A53, the owner
// 2026-10-08): the operator is told; the buyer reads what the change
// does, never the cap.
let cap = super::effective_cap(cfg, run);
if cap.minor > 0 && live.minor > cap.minor {
c.why = cap_waits(
&run.reference,
&format!(
"{}: {}",
super::ORDERS_WAIT,
super::term::over_cap(live, cap)
),
now,
);
return;
}
// **The cloud's spend cap, for the WHOLE change, before any machine is
// touched** (live 2026-10-14: a paid grow resized the front, and only the
// next machine's resize met the cap — a half-done grow). Over it, nothing
// is changed and the change waits, the operator told.
let left: Vec<PlanStep> = c.steps.iter().filter(|s| !s.done).cloned().collect();
// The machines the change lays are its too (P3-FACTS): what the new plan costs over the old
// beyond the resizes, reserved under the change's own key when laid.
let resized: u64 = c.steps.iter().map(|s| s.delta_minor).sum();
// The disks it grows likewise (A4-FACTS): the bigger disks are in the new plan's cost.
let pending_adds = c.adds.iter().any(|a| a.laid_at_ms == 0);
let laid_more = if pending_adds || c.grows.iter().any(|g| !g.done) {
c.quote
.cost
.minor_per_month
.saturating_sub(run.quote.cost.minor_per_month)
.saturating_sub(resized)
} else {
0
};
let mut keys = plan_keys(&left);
if laid_more > 0 && pending_adds {
keys.push(format!(
"lay:{}",
monetize_cloud::lay::relay_key(
&run.reference,
c.reference.rsplit('/').next().unwrap_or("change")
)
));
}
let adds = Price {
minor_per_month: left.iter().map(|s| s.delta_minor).sum::<u64>() + laid_more,
currency: c.quote.cost.currency,
};
if adds.minor_per_month > 0 {
match cloud.cap_short(adds, &keys) {
Ok(None) => {}
Ok(Some(short)) => {
c.why = cap_waits(
&run.reference,
&format!("{}: {}", super::ORDERS_WAIT, short.numbers()),
now,
);
return;
}
Err(e) => {
return fail(
c,
format!("the cloud's spend cap: {e}; nothing was changed"),
)
}
}
}
// The twin's growth is ours to pay until the renewal: no margin to keep.
if c.kind == "buyer" {
if let Err(sentence) = price::recheck(cfg, cloud, &plan, c.quote.monthly) {
if sentence.starts_with("unavailable:") {
c.why = format!("the provider's price could not be read; asking again: {sentence}");
return;
}
return fail(
c,
format!("at the buy: {sentence} For this plan change: nothing was changed."),
);
}
}
c.status = status::RESIZING.into();
c.resized_at_ms = now;
c.why = String::new();
}
fn resizing(
state: &AppState,
cfg: &EstateOrderConfig,
cloud: &dyn Cloud,
run: &mut EstateRun,
c: &mut PlanChange,
now: u64,
) {
let journal = match state.estates.journal(&run.company) {
Ok(j) => j,
Err(e) => {
c.why = format!("the journal: {e}");
return;
}
};
let currency = c.quote.cost.currency;
for i in 0..c.steps.len() {
if c.steps[i].done {
continue;
}
let s = c.steps[i].clone();
let id = ResourceId {
provider: cloud.manifest().provider,
id: s.server.clone(),
};
match monetize_cloud::ask::change_plan(
&cloud,
&id,
&s.to,
Price {
minor_per_month: s.delta_minor,
currency,
},
&journal,
) {
Ok(()) => {
c.steps[i].done = true;
if let Some(m) = run
.laid
.as_mut()
.and_then(|l| l.machines.iter_mut().find(|m| m.role == s.role))
{
m.plan = s.to.clone();
}
if !super::kept(
state,
&EstateRun {
change: Some(c.clone()),
..run.clone()
},
&format!("{} was resized", s.role),
) {
c.why = format!("{} was resized but the run is not written down (CODE RED save:{}:run); the next step waits for it", s.role, run.company);
return;
}
}
Err(CloudError::Unavailable(w)) => {
c.why = format!(
"the cloud is unavailable; resizing {} again next tick: {w}",
s.role
);
return;
}
// The shop's cloud cap: the change WAITS for the operator to raise
// it (A53), asked again next tick; never a failure the buyer reads.
Err(CloudError::OverCap(w)) => {
c.why = cap_waits(&run.reference, &format!("{}: {w}", super::ORDERS_WAIT), now);
return;
}
Err(e) => {
let done: Vec<String> = c
.steps
.iter()
.filter(|x| x.done)
.map(|x| format!("{} is on {}", x.role, x.to))
.collect();
let left = if done.is_empty() {
"nothing was changed".to_string()
} else {
done.join(", ")
};
return fail(
c,
format!("{} could not change to {}: {e}; {left}", s.role, s.to),
);
}
}
}
// **The disks of standing machines the change grows** (A4-FACTS), once every standing one is
// on its plan: each grown at the cloud, the provider's sizes read back into the run.
if c.grows.iter().any(|g| !g.done) {
match grow_disks(state, cfg, cloud, run, c) {
Ok(true) => {}
Ok(false) => return,
Err(e) => {
let stands: Vec<String> = c
.steps
.iter()
.filter(|x| x.done)
.map(|x| format!("{} is on {}", x.role, x.to))
.chain(
c.grows
.iter()
.filter(|g| g.done)
.map(|g| format!("{}'s data disk is {} GiB", g.role, g.to_gib)),
)
.collect();
return fail(
c,
format!(
"{e}; {}",
if stands.is_empty() {
"nothing was changed".to_string()
} else {
stands.join(", ")
}
),
);
}
}
}
// **The machines the change adds are laid** (P3-FACTS), once every standing one is on its
// plan: the product's plan for the new numbers, kept to the new machines, laid beside the
// estate and installed into it — the path the guardian's relay lays a box by.
if c.adds.iter().any(|a| a.laid_at_ms == 0) {
if let Err(e) = lay_adds(state, cfg, cloud, run, c, now) {
if now > c.resized_at_ms.saturating_add(ANSWER_WITHIN_MS) {
let stands: Vec<String> = c
.steps
.iter()
.filter(|x| x.done)
.map(|x| format!("{} is on {}", x.role, x.to))
.collect();
return fail(
c,
format!(
"the new machines were not laid within {} minutes ({e}); {}",
ANSWER_WITHIN_MS / 60_000,
if stands.is_empty() {
"nothing was changed".to_string()
} else {
stands.join(", ")
}
),
);
}
c.why = format!("laying the new machines again next tick: {e}");
return;
}
}
// **The machines the change takes away** (A4-FACTS), once the new ones stand: each emptied by
// the product's drain, given back at the cloud, read gone, and forgotten.
if c.removes.iter().any(|x| x.gone_at_ms == 0) {
match take_away(state, cfg, cloud, run, c, now) {
Ok(true) => {}
Ok(false) => return,
Err(e) => return fail(c, e),
}
}
// Every machine is on its plan: the estate must call home again (G1 #14, never a probe).
let Some(laid) = run.laid.clone() else {
return fail(c, "the estate has nothing laid".into());
};
let roles: Vec<String> = laid.machines.iter().map(|m| m.role.clone()).collect();
// Each new machine is heard by itself: a new box that never called home is not delivered.
let unheard: Vec<String> = c
.adds
.iter()
.filter(|a| {
crate::fleet::heard_since(
state,
run.fleet_name(),
std::slice::from_ref(&a.role),
a.laid_at_ms,
)
.is_err()
})
.map(|a| a.role.clone())
.collect();
if !unheard.is_empty() {
let since = c
.adds
.iter()
.map(|a| a.laid_at_ms)
.max()
.unwrap_or(c.resized_at_ms);
if now > since.saturating_add(ANSWER_WITHIN_MS) {
return fail(c, format!("laid, and the new {} did not call home within {} minutes; the machines stand, an operator looks", super::lifecycle::listed(&unheard, "and"), ANSWER_WITHIN_MS / 60_000));
}
c.why = format!(
"laid; waiting for the new {} to call home",
super::lifecycle::listed(&unheard, "and")
);
return;
}
if let Err(e) = crate::fleet::heard_since(state, run.fleet_name(), &roles, c.resized_at_ms) {
if now > c.resized_at_ms.saturating_add(ANSWER_WITHIN_MS) {
return fail(c, format!("resized, and the estate did not call home within {} minutes ({e}); the machines are on their new plans, an operator looks", ANSWER_WITHIN_MS / 60_000));
}
c.why = format!("resized; waiting for the estate to call home: {e}");
return;
}
if let Some(handle) = c.handle.clone() {
let Some(vendor) = state.registry.find_vendor(&cfg.vendor) else {
c.why = format!("the vendor {:?} is not linked", cfg.vendor);
return;
};
match vendor.capture(&handle) {
Ok(ChargeStatus::Captured { .. }) => {}
Ok(other) => {
c.why = format!("the capture answered {other:?}; asking again");
return;
}
Err(e) => {
c.why = format!("the capture did not go through yet: {e}");
return;
}
}
}
// **A data step**: paid and captured, so the estate may grow — this shop
// signs the go-ahead its own monetize fetches (`sold`).
let gib = c.data_to_gib();
if gib > 0
&& run
.growth
.as_ref()
.map(|g| g.reference != c.reference)
.unwrap_or(true)
{
super::sold::grant(state, run, gib, &c.reference, now);
}
c.status = status::DONE.into();
c.why = String::new();
}
/// **Grow the data disks of the standing machines the change grows** (A4-FACTS): each machine's
/// set at the cloud on the provider's estate lane (`grow_estate_set`: detached, resized, put
/// back at the same address, journaled under the change), the provider's own sizes read back
/// into the run's laid record (only ever up). `Ok(true)`: every disk is at its size; `Ok(false)`:
/// it waits (the cloud did not answer, the cap holds it, a growth stopped half-way), `c.why`
/// says so and the next tick asks again; `Err`: the provider refused by name.
fn grow_disks(
state: &AppState,
cfg: &EstateOrderConfig,
cloud: &dyn Cloud,
run: &mut EstateRun,
c: &mut PlanChange,
) -> Result<bool, String> {
let mark = c
.reference
.rsplit('/')
.next()
.unwrap_or("change")
.to_string();
let dir = state
.estates
.run_dir(&run.company)
.ok_or("no estate-orders directory")?
.join(&mark);
std::fs::create_dir_all(&dir).map_err(|e| format!("{}: {e}", dir.display()))?;
let suffixes: Vec<(String, String)> = super::lifecycle::of(cfg)
.map(|l| {
l.data_roles()
.into_iter()
.map(|d| (d.role, d.set_suffix))
.collect()
})
.unwrap_or_default();
for i in 0..c.grows.len() {
if c.grows[i].done {
continue;
}
let g = c.grows[i].clone();
let journal =
monetize_cloud::grow::Journal::open(dir.join(format!("grow-{}.journal", g.role)))
.map_err(|e| format!("the growth journal of {}: {e}", g.role))?;
let suffix = suffixes
.iter()
.find(|(r, _)| *r == g.role)
.map(|(_, s)| s.clone())
.unwrap_or_default();
let set = super::sold::set_of_as(&cfg.product, &suffix, &g.server, &g.members, g.to_gib);
let outcome = match monetize_cloud::ask::grow(&cloud, &set, true, &journal) {
Ok(o) => o,
Err(CloudError::Unavailable(w)) => {
c.why = format!(
"the cloud is unavailable; growing {}'s data disk again next tick: {w}",
g.role
);
return Ok(false);
}
Err(CloudError::OverCap(w)) => {
c.why = cap_waits(
&run.reference,
&format!("{}: {w}", super::ORDERS_WAIT),
state.now(),
);
return Ok(false);
}
Err(e) => {
return Err(format!(
"{}'s data disk could not grow to {} GiB: {e}",
g.role, g.to_gib
))
}
};
if let monetize_cloud::grow::GrowOutcome::Partial { why, .. }
| monetize_cloud::grow::GrowOutcome::Untouched { why } = &outcome
{
c.why = format!(
"{}'s data disk is not at {} GiB yet ({why}); asking again next tick",
g.role, g.to_gib
);
return Ok(false);
}
// What the box sees now, as the provider answers it: read back, only ever up.
let survey = monetize_cloud::ask::survey(
&cloud,
&monetize_cloud::grow::DataSet {
target_gib: 0,
..set.clone()
},
)
.ok();
let at = survey
.as_ref()
.map(|s| s.current_gib)
.filter(|g| *g > 0)
.unwrap_or(g.to_gib);
if at < g.to_gib {
c.why = format!(
"{}'s data disk reads {at} GiB at the cloud, not {} yet; asking again next tick",
g.role, g.to_gib
);
return Ok(false);
}
if let Some(m) = run
.laid
.as_mut()
.and_then(|l| l.machines.iter_mut().find(|m| m.role == g.role))
{
for d in m
.disks
.iter_mut()
.skip(1)
.filter(|d| g.members.contains(&d.id))
{
let read = survey
.as_ref()
.and_then(|s| s.volumes.iter().find(|v| v.id.0 == d.id))
.map(|v| v.gib)
.unwrap_or(at);
d.gib = d.gib.max(read);
}
}
c.grows[i].done = true;
run.history.push(Transition {
phase: String::new(),
at_ms: state.now(),
note: format!(
"plan change {}: {}'s data disk grown {} → {at} GiB at the cloud ({})",
c.reference,
g.role,
g.from_gib,
outcome.word()
),
});
if !super::kept(
state,
&EstateRun {
change: Some(c.clone()),
..run.clone()
},
&format!("{}'s data disk grew", g.role),
) {
c.why = format!("{}'s data disk grew but the run is not written down (CODE RED save:{}:run); the next step waits for it", g.role, run.company);
return Ok(false);
}
}
Ok(true)
}
/// **Take away the machines the change removes** (A4-FACTS), one step a tick, each recorded on
/// its [`RemoveStep`]:
///
/// 1. its drain is queued: the product's own order kind ([`crate::facts::GrowRule::drain`]),
/// signed, for that box alone, once (by its mark);
/// 2. the box acks it: emptied (its data moved to the boxes that stay, by the product's estate);
/// acked NOT ok, or no ack within [`ANSWER_WITHIN_MS`], fails the change with nothing taken
/// away;
/// 3. it is given back WHOLE at the cloud (`retire_machine`: its server, disks and clone);
/// 4. it is read gone from the provider's listing; then the guardian forgets its key, it leaves
/// the run's laid record, its WireGuard peer and every DNS record at its address go, and the
/// estate's firewall is written again without it.
///
/// `Ok(true)`: every one is gone; `Ok(false)`: it waits (`c.why`); `Err`: the change fails.
fn take_away(
state: &AppState,
cfg: &EstateOrderConfig,
cloud: &dyn Cloud,
run: &mut EstateRun,
c: &mut PlanChange,
now: u64,
) -> Result<bool, String> {
let life = super::lifecycle::of(cfg)?;
let drain = life.grow().drain;
let g = state
.fleet
.guardian()
.ok_or("the fleet store is not open: no box can be sent its drain")?
.clone();
let product = crate::fleet::run_product(state, run)?;
let fleet = run.fleet_name().to_string();
let stays: Vec<String> = run
.laid
.as_ref()
.map(|l| {
l.machines
.iter()
.map(|m| m.role.clone())
.filter(|r| !c.removes.iter().any(|x| &x.role == r))
.collect()
})
.unwrap_or_default();
let mut gone_now = false;
for i in 0..c.removes.len() {
if c.removes[i].gone_at_ms > 0 {
continue;
}
let x = c.removes[i].clone();
let role = x.role.as_str();
// 1. The drain, queued once.
if x.drain_seq == 0 {
let body = serde_json::json!({ "change": c.reference, "role": role, "stays": stays })
.to_string();
let order = monetize_guardian::Guardian::order(
&product,
&fleet,
role,
&drain,
body,
now,
ANSWER_WITHIN_MS,
);
let mark = format!(
"drain/{fleet}/{}/{role}",
c.reference.rsplit('/').next().unwrap_or("change")
);
let seq = match g.store.mark(&mark) {
Some(seq) => seq,
None => {
g.queue_order_marked(order, &mark)
.map_err(|e| format!("the drain of {role} could not be queued: {e}"))?
.seq
}
};
c.removes[i].drain_seq = seq;
c.removes[i].drain_at_ms = now;
run.history.push(Transition { phase: String::new(), at_ms: now, note: format!("plan change {}: {role} is sent its drain ({drain}, seq {seq}); it is emptied before it goes", c.reference) });
c.why = format!("waiting for {role} to be emptied (its drain is queued)");
return Ok(false);
}
// 2. Its ack.
if x.drained_at_ms == 0 {
let ack = g
.store
.box_record(&fleet, role)
.and_then(|b| b.acks.into_iter().find(|a| a.seq == x.drain_seq));
match ack {
Some(a) if a.ok => {
c.removes[i].drained_at_ms = now;
run.history.push(Transition {
phase: String::new(),
at_ms: now,
note: format!(
"plan change {}: {role} was emptied: {}",
c.reference, a.detail
),
});
}
Some(a) => {
return Err(format!(
"{role} was not emptied: {}; nothing was taken away",
a.detail
))
}
None if now > x.drain_at_ms.saturating_add(ANSWER_WITHIN_MS) => {
return Err(format!(
"{role} did not ack its drain within {} minutes; nothing was taken away",
ANSWER_WITHIN_MS / 60_000
))
}
None => {
c.why = format!("waiting for {role} to be emptied (its drain is queued)");
return Ok(false);
}
}
}
let Some(machine) = run.laid.as_ref().and_then(|l| l.machine(role)).cloned() else {
c.removes[i].gone_at_ms = now;
continue;
};
// 3. Given back whole.
if c.removes[i].retired_at_ms == 0 {
match monetize_cloud::ask::retire_machine(&cloud, &run.reference, &machine) {
Ok(ids) => {
c.removes[i].retired_at_ms = now;
run.history.push(Transition { phase: String::new(), at_ms: now, note: format!("plan change {}: {role} given back at the cloud ({})", c.reference, ids.iter().map(|i| i.id.as_str()).collect::<Vec<_>>().join(", ")) });
}
Err(e) if now > c.removes[i].drained_at_ms.saturating_add(ANSWER_WITHIN_MS) => return Err(format!("{role} was emptied, and the cloud did not take it back within {} minutes: {e}; an operator looks", ANSWER_WITHIN_MS / 60_000)),
Err(e) => {
c.why = format!("{role} is emptied; giving it back at the cloud again next tick: {e}");
return Ok(false);
}
}
}
// 4. Read gone, then forgotten.
match monetize_cloud::ask::machine_standing(&cloud, &run.reference, &machine) {
Ok(left) if left.is_empty() => {}
Ok(left) => {
if now > c.removes[i].retired_at_ms.saturating_add(ANSWER_WITHIN_MS) {
return Err(format!("{role} was given back, and the cloud still lists {} after {} minutes; an operator looks", left.iter().map(|i| i.id.as_str()).collect::<Vec<_>>().join(", "), ANSWER_WITHIN_MS / 60_000));
}
// A provider may delete slowly (UpCloud lists a deleted server in maintenance):
// asked again until it is gone.
c.why = match monetize_cloud::ask::retire_machine(&cloud, &run.reference, &machine) {
Ok(_) => format!("{role} is being given back at the cloud ({} still listed)", left.len()),
Err(e) => format!("{role} is being given back at the cloud ({} still listed; asked again, the cloud answered: {e})", left.len()),
};
return Ok(false);
}
Err(e) => {
c.why = format!(
"the cloud's listing could not be read to see {role} gone; asking again: {e}"
);
return Ok(false);
}
}
if let Err(e) = g.store.forget_box(&fleet, role) {
c.why = format!("{role} is gone at the cloud; the guardian did not forget it yet: {e}");
return Ok(false);
}
let mut words = vec![format!(
"{role} is gone at the cloud and forgotten by the guardian"
)];
let addresses: Vec<&str> = [machine.public_ipv4.as_str(), machine.utility_ipv4.as_str()]
.into_iter()
.filter(|a| !a.trim().is_empty())
.collect();
let named: Vec<String> = run
.dns
.iter()
.filter(|r| addresses.contains(&r.value.as_str()))
.map(|r| r.name.clone())
.collect();
if !named.is_empty() {
let (writer, _) = state.estates.dns_of_run(run);
match writer.remove(&named) {
Ok(_) => {
run.dns.retain(|r| !named.contains(&r.name));
words.push(format!("its DNS records {} removed", named.join(", ")));
}
Err(e) => {
c.why = format!(
"{role} is gone; its DNS records {} were not removed yet: {e}",
named.join(", ")
);
return Ok(false);
}
}
}
run.peers.retain(|p| p.name != role);
if let Some(l) = run.laid.as_mut() {
l.machines.retain(|m| m.role != role);
}
c.removes[i].gone_at_ms = now;
gone_now = true;
run.history.push(Transition {
phase: String::new(),
at_ms: now,
note: format!("plan change {}: {}", c.reference, words.join("; ")),
});
if !super::kept(
state,
&EstateRun {
change: Some(c.clone()),
..run.clone()
},
&format!("{role} taken away"),
) {
c.why = format!("{role} was taken away but the run is not written down (CODE RED save:{}:run); the next step waits for it", run.company);
return Ok(false);
}
}
if gone_now {
// The firewall names every machine of the estate: written again without the ones gone.
if let Some(laid) = run.laid.as_ref() {
if let Err(e) =
monetize_cloud::Cloud::firewall(&cloud, &super::lifecycle::shaped_now(cfg, laid))
{
run.history.push(Transition {
phase: String::new(),
at_ms: now,
note: format!(
"plan change {}: the firewall was not written again yet: {e}",
c.reference
),
});
}
}
}
Ok(true)
}
/// **Lay the machines the change adds** (P3-FACTS, a plan change that adds machines): the
/// product's plan for the change's numbers, kept to the machines not laid yet, each marked
/// [`monetize_cloud::lay::ADDED_LABEL`] with the change's name (its own spend), laid beside the
/// estate and installed into it ([`super::relay::lay_members`]), its box key named to the
/// guardian (or its pin window opened), the estate's firewall written again for it. A lay that
/// is refused or does not answer is asked again: the cloud adopts what it laid by its title.
fn lay_adds(
state: &AppState,
cfg: &EstateOrderConfig,
cloud: &dyn Cloud,
run: &mut EstateRun,
c: &mut PlanChange,
now: u64,
) -> Result<(), String> {
let life = super::lifecycle::of(cfg)?;
let roles: Vec<String> = c
.adds
.iter()
.filter(|a| a.laid_at_ms == 0)
.map(|a| a.role.clone())
.collect();
let mut plan = super::lifecycle::plan_of(life.as_ref(), cfg, &run.company, &c.to, &c.to_twin)?;
plan.machines.retain(|m| roles.contains(&m.role));
if plan.machines.len() != roles.len() {
let planned: Vec<String> = plan.machines.iter().map(|m| m.role.clone()).collect();
return Err(format!(
"the product's plan for the new numbers lays {}, not {}",
super::lifecycle::listed(&planned, "and"),
super::lifecycle::listed(&roles, "and")
));
}
let mark = c
.reference
.rsplit('/')
.next()
.unwrap_or("change")
.to_string();
for m in &mut plan.machines {
m.labels
.insert(monetize_cloud::lay::ADDED_LABEL.into(), mark.clone());
}
let dir = state
.estates
.run_dir(&run.company)
.ok_or("no estate-orders directory")?
.join(&mark);
std::fs::create_dir_all(&dir).map_err(|e| format!("{}: {e}", dir.display()))?;
let journal =
monetize_cloud::grow::Journal::open(dir.join("journal")).map_err(|e| e.to_string())?;
let listed = super::lifecycle::listed(&roles, "and");
let (during, narrate_as) = (
format!("the plan change {} laying {listed}", c.reference),
format!("change {mark}"),
);
let how = super::relay::Members {
roles: &roles,
dir: &dir,
journal: &journal,
during: &during,
narrate_as: &narrate_as,
};
let laid = super::relay::lay_members(state, cfg, cloud, run, life.as_ref(), plan, &how)?;
let mut keys = std::collections::BTreeMap::new();
for v in &laid.verdicts {
for p in &v.peers {
run.peers.retain(|q| q.name != p.name);
run.peers.push(p.clone());
}
keys.extend(
v.appliance_keys
.iter()
.filter(|(r, _)| roles.contains(r))
.map(|(r, k)| (r.clone(), k.clone())),
);
}
let fleet = crate::fleet::estate_installed(state, run.fleet_name(), &roles, &keys, false)?;
// The firewall names every machine of the estate: written again with the new ones in it.
let firewall = match monetize_cloud::Cloud::firewall(
&cloud,
&super::lifecycle::shaped_now(cfg, &laid.estate),
) {
Ok(()) => String::new(),
Err(e) => format!("; its firewall was not written again yet: {e}"),
};
for m in &laid.machines {
if let Some(a) = c.adds.iter_mut().find(|a| a.role == m.role) {
a.server = m.server.clone();
a.laid_at_ms = now;
}
}
run.laid = Some(laid.estate);
let words = format!(
"plan change {}: {} laid ({}) and installed, the installer's verdict PASS; {fleet}; {} release clone(s) given back{firewall}",
c.reference,
listed,
laid.machines.iter().map(|m| format!("{} on {}", m.role, m.server)).collect::<Vec<_>>().join(", "),
laid.clones
);
run.history.push(Transition {
phase: String::new(),
at_ms: now,
note: words,
});
if !super::kept(
state,
&EstateRun {
change: Some(c.clone()),
..run.clone()
},
&format!("{listed} laid"),
) {
return Err(format!(
"{listed} laid but the run is not written down (CODE RED save:{}:run)",
run.company
));
}
Ok(())
}
/// The change is over: the run takes the new plans (done) or keeps its own
/// (failed), the hold is released on a failure, the buyer is told.
fn finish(state: &AppState, cfg: &EstateOrderConfig, run: &mut EstateRun, c: PlanChange, now: u64) {
let host = if run.domain.is_empty() {
cfg.host_of(&run.company)
} else {
run.domain.clone()
};
if c.status == status::DONE {
if c.kind == "twin" {
// Ours to pay until the renewal: the buyer's price stands; the
// record says what is laid and what it costs us.
run.twin_floor = c.to_twin.clone();
run.quote.twin_plan = c.to_twin.clone();
run.quote.cost = c.quote.cost;
run.quote.cost_in_sale = c.quote.cost_in_sale;
let twin = Parts::of(cfg).twin.unwrap_or_default();
if let Some(line) = run.quote.machines.iter_mut().find(|m| m.role == twin) {
if let Some(new) = c.quote.machines.iter().find(|m| m.role == twin) {
*line = new.clone();
}
}
} else {
run.quote = c.quote.clone();
}
if c.from_twin != c.to_twin {
run.twin_changed_at_ms = now;
}
if c.kind == "buyer" {
mail(
state,
run,
"estate.changed",
format!("Your {} {host} has its new plans", super::lifecycle::name_of(cfg)),
format!(
"{host} is changed: {}.\n\nIts price is now {} a month. {}\n",
c.steps
.iter()
.map(|s| format!("{} is on {}", s.role, s.to))
.chain(c.grows.iter().map(|g| format!("{}'s data disk is {} GiB", g.role, g.to_gib)))
.chain(c.adds.iter().map(|a| format!("a new {} runs on {}", a.role, a.plan)))
.chain(c.removes.iter().map(|x| format!("{} is taken away", x.role)))
.chain((c.data_to_gib() > 0).then(|| format!("main's data disks grow to {} GB each at {host}'s next growth, without stopping it", c.data_to_gib())))
.collect::<Vec<_>>()
.join(", "),
run.quote.monthly.in_words(),
if c.amount.minor > 0 {
format!("{} was charged for the rest of the period paid until {}.", c.amount.in_words(), iso_date(run.paid_until_ms))
} else {
format!("The period is paid until {}: nothing more is charged now.", iso_date(run.paid_until_ms))
}
),
);
}
} else {
if let (Some(h), Some(vendor)) =
(c.handle.as_ref(), state.registry.find_vendor(&cfg.vendor))
{
let closed = super::pages::close(vendor, h);
super::pages::settle(
run,
&c.reference,
h,
closed,
"the plan change did not go through",
now,
);
}
if c.kind == "buyer" {
mail(
state,
run,
"estate.change_failed",
format!("Your {} {host}'s plan change did not go through — nothing was charged", super::lifecycle::name_of(cfg)),
format!("The plan change of {host} did not go through.\n\nNothing was charged for it; the amount held, if any, is released. Your {} runs on as before.\n", super::lifecycle::name_of(cfg)),
);
}
}
run.history.push(Transition {
phase: String::new(),
at_ms: now,
note: format!("plan change {} {}", c.reference, c.status),
});
run.changes.push(c);
run.change = None;
}
/// **The twin grows by itself** (the damped rule, [`super::twin::decide`]):
/// its samples from the estate's source (the installer's native read of main's
/// heartbeat, kept in the estate's window), the rule's hysteresis and
/// spacing, and — a resize being a provider operation — the day's batch.
fn autoscale(state: &AppState, cfg: &EstateOrderConfig, run: &mut EstateRun, now: u64) {
// **A product with no twin has none to grow** (P3-FACTS: its facts' `twin`; the Lake).
let parts = Parts::of(cfg);
let Some(twin) = super::lifecycle::of(cfg).ok().and_then(|l| l.twin()) else {
return;
};
let twin_role = twin.role.clone();
let Some(run_dir) = state.estates.run_dir(&run.company) else {
return;
};
let at = super::twin::SampleAt {
company: &run.company,
apex: &cfg.apex,
now_ms: now,
run_dir: &run_dir,
reports: state.fleet.guardian().map(|g| &*g.store),
every_ms: cfg.twin.sample_every_secs.max(1).saturating_mul(1_000),
twin: &twin,
};
let samples = match state.estates.twin_health.samples(&at) {
Ok(s) => s,
Err(e) => {
// Said once per distinct reason, not on every tick.
let line = format!("the twin's health was not read: {e}");
if run.history.last().map(|t| t.note != line).unwrap_or(true)
&& !run.history.iter().rev().take(4).any(|t| t.note == line)
{
run.history.push(Transition {
phase: String::new(),
at_ms: now,
note: line,
});
}
return;
}
};
if samples.is_empty() {
return;
}
let Some(cloud) = state.registry.find_cloud(&cfg.cloud) else {
return;
};
let Ok(catalog) = monetize_cloud::ask::plan_catalog(&cloud, &Zone(cfg.zone.clone())) else {
return;
};
let spec = |name: &str| catalog.iter().find(|p| p.name == name).cloned();
let (Some(twin), Some(main)) = (spec(&run.quote.twin_plan), spec(&run.quote.sizes.main_plan))
else {
return;
};
let cx = TwinContext {
twin: &twin,
main: &main,
catalog: &catalog,
last_change_ms: run.twin_changed_at_ms,
now_ms: now,
cost_minor: run.quote.cost_in_sale.minor,
cap_minor: run.spend_cap.minor,
};
let TwinVerdict::Grow { to, why } = decide(&cfg.twin, &samples, &cx) else {
return;
};
let Ok(q) = price::quote_with_twin(cfg, cloud, &run.company, &run.quote.sizes, &to.name) else {
return;
};
let Ok(steps) = steps_for(run, &catalog, &q, &parts) else {
return;
};
// The twin's own growth resizes the twin alone.
let steps: Vec<PlanStep> = steps.into_iter().filter(|s| s.role == twin_role).collect();
if steps.is_empty() {
return;
}
let not_before = crate::grow::next_window_boundary(
&state.config.read().unwrap_or_else(|p| p.into_inner()),
now,
);
let c = PlanChange {
reference: format!("{}/change-{}", run.reference, run.changes.len() + 1),
kind: "twin".into(),
status: status::QUEUED.into(),
from: run.quote.sizes.clone(),
from_twin: run.quote.twin_plan.clone(),
to: q.sizes.clone(),
to_twin: q.twin_plan.clone(),
amount: Money {
minor: 0,
currency: q.monthly.currency,
},
quote: q,
handle: None,
checkout_url: String::new(),
words: format!(
"{} — {why}; ours to pay until the renewal",
describe(&steps, &parts)
),
steps,
opened_at_ms: now,
resized_at_ms: 0,
why: String::new(),
not_before_ms: if not_before > now { not_before } else { 0 },
queued_at_ms: 0,
adds: Vec::new(),
removes: Vec::new(),
grows: Vec::new(),
};
let when = if c.not_before_ms > 0 {
format!(
" (resized at the next growth window, {})",
iso_time(c.not_before_ms)
)
} else {
String::new()
};
run.history.push(Transition {
phase: String::new(),
at_ms: now,
note: format!("the twin grows by itself: {}{when}", c.words),
});
run.change = Some(c);
}
/// `YYYY-MM-DD HH:MM UTC`.
fn iso_time(ms: u64) -> String {
let secs = (ms / 1000) % 86_400;
format!(
"{} {:02}:{:02} UTC",
iso_date(ms),
secs / 3600,
(secs % 3600) / 60
)
}
/// Pure: **the product an instance's mail goes out as** (A16): the product of the site it was
/// bought on, else the shop's.
pub(crate) fn notice_product(of_site: Option<String>, sold: Result<&str, String>) -> String {
of_site.unwrap_or_else(|| sold.map(str::to_owned).unwrap_or_default())
}
fn mail(state: &AppState, run: &mut EstateRun, kind: &str, subject: String, body: String) {
let notice = monetize_pay::CustomerNotice {
product: notice_product(
state.registry.product_of_estate(&run.estate),
state.sold_product(),
),
tenant: run.company.clone(),
to: run.email.clone(),
kind: kind.into(),
subject,
body,
};
let line = match state.signup_sink.notify(¬ice) {
Ok(l) => format!("{kind} mail sent: {l}"),
Err(e) => format!("{kind} mail NOT sent: {e:?}"),
};
let at = state.now();
run.history.push(Transition {
phase: String::new(),
at_ms: at,
note: line,
});
}
#[cfg(test)]
mod tests {
use super::*;
use monetize_pay::Currency;
/// ★ A16 (owner 2026-10-23): **an instance's mail is its product's** — a Holger instance
/// sold on a Holger site mails as Holger (its sender, its words), never as the shop's product.
#[test]
fn an_instances_mail_is_its_products() {
assert_eq!(
super::notice_product(Some("holger".into()), Ok("gunnar")),
"holger"
);
assert_eq!(super::notice_product(None, Ok("gunnar")), "gunnar");
assert_eq!(super::notice_product(None, Err("none".into())), "");
}
/// ★ **A change's numbers lie over the laid ones** (P3-FACTS): what it names in their place,
/// the rest as laid; anything else is the change's own (for the form to refuse).
#[test]
fn a_changes_numbers_lie_over_the_laid_ones() {
let laid = serde_json::json!({"be": 1, "store_gib": 500});
assert_eq!(
over(&laid, &serde_json::json!({"be": 2})),
serde_json::json!({"be": 2, "store_gib": 500})
);
assert_eq!(
over(&serde_json::Value::Null, &serde_json::json!({"be": 2})),
serde_json::json!({"be": 2})
);
assert_eq!(over(&laid, &serde_json::json!([1])), serde_json::json!([1]));
assert_eq!(
over(&laid, &serde_json::Value::Null),
serde_json::Value::Null
);
}
#[test]
fn the_difference_is_the_monthly_rise_for_the_days_left() {
let eur = |minor| Money {
minor,
currency: Currency::Eur,
};
assert_eq!(
prorated(eur(13_200), eur(18_000), 0, 90 * DAY_MS),
eur(14_400)
);
assert_eq!(
prorated(eur(13_200), eur(18_000), 1, 90 * DAY_MS),
eur(14_400),
"a part day counts whole"
);
assert_eq!(
prorated(eur(18_000), eur(13_200), 0, 90 * DAY_MS),
eur(0),
"cheaper: nothing now"
);
assert_eq!(
prorated(eur(13_200), eur(18_000), 90 * DAY_MS, 90 * DAY_MS),
eur(0),
"nothing left of the period"
);
}
}