crates/lake-product/src/bench.rs
//! **The Lake bench** (`cargo xtask bench lake`; `bench echo` is its alias for one release): the
//! whole lifecycle of one Lake estate on the ONE box's REAL code, in one process, at container
//! level.
//!
//! What is real: monetize-server's registry (the signed Lake guest loaded and gated by its
//! grants, which are none), the PULLED shape (`[products.lake] instance`, no endpoint: the
//! product is answered from its boxes' reports and its orders are queued for them to pull), its
//! estate orders (the quote from the guest's plan, the hold, the lay, the install, the sale
//! check, the capture, the renewal, the update, the terminate), its custody, its guardian and
//! its gRPC router (every Lake box reports to `Fleet` over the wire), and the shared buy relay
//! (`monetize-site-buy`) a product site runs. What is a double: the vendor
//! (`testing::CallbackPay`: a hold authorised when the bench says so), the cloud
//! (`monetize::testing::MockCloud`), the DNS (printed), the installer (the bench's, on loopback;
//! see [`crate::installer`]), and the registry's linked set (`Registry::from_config`, the open
//! set the ONE box's `monetize_admin::linked()` also starts from; the real program in
//! containers is the fast test's, `crates/monetize-fast`).
//!
//! On the signed axes (`--cloud upcloud|scaleway`, `--vendor stripe|klarna|swish`,
//! [`crate::signed`]) the cloud and the vendor are monetize-impl's signed guests instead,
//! loaded from their blocks by the same registry, against their fakes (the cloud's mock in
//! process, the vendor's fake as a process whose control door is the buyer).
//!
//! Each step is one receipt row: `PASS`, or `FAIL` with the words the ONE box answered. What
//! the Lake cannot prove before monetize v2.0 P3 is said in the receipt, not judged.
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use monetize_cloud::dns::{DnsRecord, DnsWriter};
use monetize_server::estate_order::{self, phase, EstateOrderConfig, EstateSizes};
use monetize_server::state::AppState;
use crate::installer::Boxes;
use crate::lake_box::{role, METER};
use crate::say;
/// **The bench's axes**: what stands for the cloud and what for the vendor.
/// `mock` is the in-process test double (`MockCloud`, `CallbackPay`: the bench authorises the
/// hold itself); `sandbox` is the open DUMMY the money-free default build ships (the sandbox
/// cloud with declared prices, the sandbox vendor that pays on the spot), built from the ONE
/// box's config as a deployment builds it. The SIGNED guests (`upcloud`, `scaleway`; `stripe`,
/// `klarna`, `swish`: [`crate::signed`]) are monetize-impl's, loaded from the blocks the bench
/// writes, against their fakes; `signed` is the directory the xtask built them into.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Axes {
pub cloud: String,
pub vendor: String,
pub signed: Option<PathBuf>,
}
impl Default for Axes {
fn default() -> Self {
Axes {
cloud: "mock".into(),
vendor: "mock".into(),
signed: None,
}
}
}
impl Axes {
/// The axes, refused by name: an unknown cloud or vendor, or a signed axis whose built
/// guest or fake is not in [`Axes::signed`].
pub fn check(&self) -> Result<String, String> {
if !DUMMIES.contains(&self.cloud.as_str())
&& !crate::signed::CLOUDS.contains(&self.cloud.as_str())
{
return Err(format!(
"no cloud {:?}: the bench knows {DUMMIES:?} and the signed {:?}",
self.cloud,
crate::signed::CLOUDS
));
}
if !DUMMIES.contains(&self.vendor.as_str())
&& !crate::signed::VENDORS.contains(&self.vendor.as_str())
{
return Err(format!(
"no vendor {:?}: the bench knows {DUMMIES:?} and the signed {:?}",
self.vendor,
crate::signed::VENDORS
));
}
let signed: Vec<&str> = [self.cloud.as_str(), self.vendor.as_str()]
.into_iter()
.filter(|a| crate::signed::is_signed(a))
.collect();
if signed.is_empty() {
return Ok(format!(
"cloud {}, vendor {} (the open dummies)",
self.cloud, self.vendor
));
}
let dir = self.signed.clone().ok_or_else(|| format!("{signed:?}: a signed guest needs its build (`--signed <dir>`); `cargo xtask bench lake` builds it from the sibling ../monetize-impl"))?;
crate::signed::Signed(dir.clone()).check(&self.cloud, &self.vendor)?;
Ok(format!(
"cloud {}, vendor {}; the signed {signed:?} from {}",
self.cloud,
self.vendor,
dir.display()
))
}
/// The cloud's name in the registry.
pub fn cloud_name(&self) -> &str {
match self.cloud.as_str() {
"mock" => "mockcloud",
other => other,
}
}
/// The vendor's name in the registry.
pub fn vendor_name(&self) -> String {
match self.vendor.as_str() {
"mock" => "mockpay".into(),
"sandbox" => SANDBOX_VENDOR.into(),
other => crate::signed::vendor_name(other),
}
}
/// The zone the estate is laid in.
pub fn zone(&self) -> &'static str {
match self.cloud.as_str() {
"sandbox" => "se-sto1",
"upcloud" => crate::signed::UPCLOUD_ZONE,
"scaleway" => crate::signed::SCALEWAY_ZONE,
_ => "fi-hel2",
}
}
/// The ingestors' plan the bench's own orders name: the cloud's name for 2 CPUs, 4 GB.
pub fn ingestor_plan(&self) -> &'static str {
if self.cloud == "scaleway" {
"DEV1-M"
} else {
"STARTER-2xCPU-4GB"
}
}
}
/// The dummies the bench knows.
pub const DUMMIES: [&str; 2] = ["mock", "sandbox"];
/// The sandbox vendor's instance in the bench's config.
pub const SANDBOX_VENDOR: &str = "sandbox/play";
/// The service bearer of the bench's ONE box (a site's link presents it).
pub const BEARER: &str = "lake-bench-service-bearer";
/// The company the bench buys.
pub const COMPANY: &str = "acme";
/// The buyer's account and address (never a real one).
pub const ACCOUNT: &str = "acme";
pub const EMAIL: &str = "buyer@mock.invalid";
/// **The DNS writer of the bench**: every record printed, none written anywhere.
pub struct LakeDns;
impl DnsWriter for LakeDns {
/// It prints, it writes into no zone (RA1-5).
fn zoneless(&self) -> bool {
true
}
fn write(&self, records: &[DnsRecord]) -> Result<String, String> {
for r in records {
say(
"dns",
"write",
"monetize-server",
&format!("{} {} {}", r.name, r.kind, r.value),
);
}
Ok(format!("{} record(s) printed", records.len()))
}
fn remove(&self, names: &[String]) -> Result<String, String> {
say("dns", "remove", "monetize-server", &names.join(" "));
Ok(format!("{} name(s) printed", names.len()))
}
fn listed(&self) -> Result<Vec<DnsRecord>, String> {
Ok(Vec::new())
}
}
/// One receipt row.
#[derive(Clone, Debug, serde::Serialize)]
pub struct Row {
pub step: String,
pub pass: bool,
pub words: String,
}
#[derive(Default)]
pub struct Receipt {
pub rows: Mutex<Vec<Row>>,
/// What waits for P3, said and not judged.
pub said: Mutex<Vec<String>>,
}
impl Receipt {
pub fn step(&self, step: &str, r: Result<String, String>) -> bool {
let (pass, words) = match r {
Ok(w) => (true, w),
Err(w) => (false, w),
};
println!(
"bench: {} {step}: {words}",
if pass { "PASS" } else { "FAIL" }
);
self.rows.lock().unwrap().push(Row {
step: step.into(),
pass,
words,
});
pass
}
pub fn say(&self, line: &str) {
println!("bench: SAID {line}");
self.said.lock().unwrap().push(line.to_string());
}
pub fn passed(&self) -> bool {
self.rows.lock().unwrap().iter().all(|r| r.pass)
}
/// `receipt.txt` and `receipt.json` in `out`.
pub fn write(&self, out: &Path, commit: &str, axes: &Axes) -> std::io::Result<PathBuf> {
std::fs::create_dir_all(out)?;
let rows = self.rows.lock().unwrap().clone();
let said = self.said.lock().unwrap().clone();
let pass = rows.iter().filter(|r| r.pass).count();
let mut t = format!("Lake bench receipt\nmonetize {commit}\ncloud {} · vendor {}\nresult: {} ({pass} of {} steps passed)\n\n", axes.cloud, axes.vendor, if pass == rows.len() { "PASS" } else { "FAIL" }, rows.len());
for r in &rows {
t.push_str(&format!(
"{} {}: {}\n",
if r.pass { "PASS" } else { "FAIL" },
r.step,
r.words
));
}
t.push_str("\nwhat waits for monetize v2.0 P3 (said, not judged):\n");
for s in &said {
t.push_str(&format!("- {s}\n"));
}
t.push_str("\nthe calls, as the Lake printed them:\n");
for l in crate::lines() {
t.push_str(&l);
t.push('\n');
}
let p = out.join("receipt.txt");
std::fs::write(&p, t)?;
std::fs::write(
out.join("receipt.json"),
serde_json::to_vec_pretty(
&serde_json::json!({ "commit": commit, "cloud": axes.cloud, "vendor": axes.vendor, "rows": rows, "p3_waits": said, "calls": crate::lines() }),
)?,
)?;
Ok(p)
}
}
/// Sign the Lake guest with a bench CA and write the ONE box's `monetize.toml` naming it: the
/// pulled shape (`instance`, no endpoint, no grants), as the ONE box's feed writes a product.
fn config(dir: &Path, wasm: &Path, axes: &Axes, signed_blocks: &str) -> Result<PathBuf, String> {
use guest_host::verify::sign::{self, dev};
let module = dir.join("lake.wasm");
std::fs::copy(wasm, &module).map_err(|e| format!("{}: {e}", wasm.display()))?;
let (ca_key, ca_der) = dev::mint_ca("lake-bench-plugin-root").map_err(|e| e.to_string())?;
let root = dir.join("bench-root.der");
std::fs::write(&root, &ca_der).map_err(|e| e.to_string())?;
sign::sign_module_file(
&module,
&dev::new_ed25519_signer(&ca_key, &ca_der, "lake-bench-signing")
.map_err(|e| e.to_string())?,
)?;
let mut toml = format!(
"data_dir = {:?}\nsecrets_dir = {:?}\n\n[products.lake]\ninstance = {COMPANY:?}\nguest = {:?}\nguest_trust_roots = [{:?}]\ngrants = []\n\n[signups.lake.plans.\"lake/account\"]\namount_minor = 0\ncurrency = \"Eur\"\ncap_bytes = 0\nperiod_days = 30\n",
dir.join("data").display().to_string(),
dir.join("secrets").display().to_string(),
module.display().to_string(),
root.display().to_string()
);
if axes.vendor == "sandbox" {
toml.push_str("\n[vendors.sandbox.play]\nenabled = true\n");
}
if axes.cloud == "sandbox" {
// The Lake's plans, declared; the ingestors' plan of the quotes below among them.
// The Lake's plans, the quotes' ingestor plan, and the shop's Gunnar shape's three (the
// quote resolves front, main and the twin's plan for every product until P3's order).
let plans = [
monetize_guest_lake::plans::FRONT,
monetize_guest_lake::plans::BE,
monetize_guest_lake::plans::FE,
monetize_guest_lake::plans::GRAPH,
monetize_guest_lake::plans::INGESTOR,
axes.ingestor_plan(),
"STARTER-4xCPU-8GB",
"STARTER-1xCPU-4GB",
]
.iter()
.collect::<std::collections::BTreeSet<_>>()
.into_iter()
.enumerate()
.map(|(i, p)| format!("{p:?} = {}", 1000 + 500 * i))
.collect::<Vec<_>>()
.join(", ");
toml.push_str(&format!("\n[clouds.sandbox]\nenabled = true\ncap = {{ total_minor = 100000000, currency = {currency:?} }}\nzone = \"se-sto1\"\npool_gib = 0\ndisk_step_gib = 10\nprices = {{ disk_minor_per_gib_month = 23, plan_minor_per_month = {{ {plans} }}, currency = {currency:?} }}\n", currency = if axes.vendor == "swish" { "Sek" } else { "Eur" }));
}
toml.push_str(signed_blocks);
let p = dir.join("monetize.toml");
std::fs::write(&p, toml).map_err(|e| e.to_string())?;
Ok(p)
}
/// The media of every release the bench feeds (`lake-r1`, `lake-keep-r1`, `lake-r2`, …).
pub const MEDIA: [&str; 4] = ["lake-r1", "lake-keep-r1", "lake-r2", "lake-keep-r2"];
/// `media`: a medium's id at a signed cloud's mock, by its name (absent: the name itself).
fn estate_cfg(
release: &str,
axes: &Axes,
media: &std::collections::BTreeMap<String, String>,
) -> EstateOrderConfig {
let at = |name: String| media.get(&name).cloned().unwrap_or(name);
let mut c = EstateOrderConfig {
product: "lake".into(),
vendor: axes.vendor_name(),
cloud: axes.cloud_name().into(),
apex: "lake.mock.invalid".into(),
currency: crate::signed::vendor_currency(&axes.vendor).into(),
resize_notice_secs: 0,
shop_door: "https://one.mock.invalid".into(),
zone: axes.zone().into(),
// The release the shop sells now: the Lake's one medium and its keep medium.
media: [
("lake".to_string(), at(format!("lake-{release}"))),
("lake-keep".to_string(), at(format!("lake-keep-{release}"))),
]
.into_iter()
.collect(),
..EstateOrderConfig::default()
};
// The quote resolves the shop's front, main and twin plans for every product until P3's
// order: at Scaleway they are Scaleway's names (the twin's at least 4 GB, its floor).
if axes.cloud == "scaleway" {
c.shape.front.plan = "DEV1-S".into();
c.shape.main.plan = "DEV1-L".into();
c.shape.twin.plan = "DEV1-M".into();
// The ladder the shop prices for Sells, in Scaleway's names too.
c.ladder = vec![monetize_server::estate_order::LadderStep::new(
"Mini", "DEV1-S", "DEV1-L", 10,
)];
}
c
}
fn phase_of(state: &AppState) -> String {
state
.estates
.get(COMPANY)
.map(|r| r.phase.clone())
.unwrap_or_else(|| "none".into())
}
/// Every box of the estate reports, as running boxes do.
async fn report_all(boxes: &Boxes, fleet: &str, now: u64) {
for b in boxes.all() {
if !b.held.lock().unwrap().instance.is_empty() {
let _ = b.report(fleet, now).await;
}
}
}
/// Drive the order until `want` (or a phase that ends it), the boxes reporting between ticks,
/// for at most `secs` (a signed cloud's mock lays at its own pace).
async fn drive_until(
state: &Arc<AppState>,
pay: Option<&monetize_server::testing::CallbackPay>,
boxes: &Boxes,
fleet: &str,
want: &[&str],
secs: u64,
) -> Result<String, String> {
let mut last = String::new();
let mut caps = pay.map(|p| p.captures()).unwrap_or(0);
let until = std::time::Instant::now() + std::time::Duration::from_secs(secs);
while std::time::Instant::now() < until {
report_all(boxes, fleet, state.now()).await;
let s = state.clone();
tokio::task::spawn_blocking(move || estate_order::drive(&s, COMPANY))
.await
.map_err(|e| e.to_string())?;
let p = phase_of(state);
if let Some(pay) = pay {
if pay.captures() != caps {
caps = pay.captures();
say(
"vendor",
"capture",
"monetize-server",
&format!("captures={caps} at phase {p}"),
);
}
}
if p != last {
say("bench", "phase", "monetize-server", &p);
last = p.clone();
}
if want.contains(&p.as_str()) {
return Ok(p);
}
if p == phase::FAILED || p == phase::RELEASED || p == phase::OWED {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}
let history = state
.estates
.get(COMPANY)
.map(|r| {
r.history
.iter()
.map(|t| format!("{}: {}", t.phase, t.note))
.collect::<Vec<_>>()
.join(" | ")
})
.unwrap_or_default();
let why = state
.estates
.get(COMPANY)
.map(|r| r.why.clone())
.unwrap_or_default();
Err(format!(
"the order stayed at {last}, not {want:?}; why: {why}; notes: {history}"
))
}
/// The sizes of a Lake order of `numbers` (the Lake's numbers ride `EstateSizes.order`).
fn order(numbers: serde_json::Value) -> EstateSizes {
EstateSizes {
order: numbers,
..EstateSizes::default()
}
}
/// **The parametric proof**: the core prices exactly the plan the guest answers for the buyer's
/// numbers. Blocking (the guest host runs its own wait).
fn quotes(
cfg: &EstateOrderConfig,
cloud: &dyn monetize_cloud::Cloud,
ingestor_plan: &str,
) -> Vec<(String, Result<String, String>)> {
let q = |o: serde_json::Value| estate_order::price::quote(cfg, cloud, COMPANY, &order(o));
let roles = |q: &estate_order::Quote| {
q.machines
.iter()
.map(|m| m.role.clone())
.collect::<Vec<_>>()
};
let mut v = Vec::new();
let small = q(serde_json::Value::Null);
v.push((
"quote: no numbers is the smallest Lake (front, ingestor, be, fe)".to_string(),
match &small {
Ok(q) if roles(q) == ["front", "ingestor", "be", "fe"] => {
Ok(format!("{} a month", q.monthly.in_words()))
}
Ok(q) => Err(format!("{:?}", roles(q))),
Err(e) => Err(e.clone()),
},
));
let starter = serde_json::json!({"store_gib": 500, "ingestors": 2, "ingestor_plan": ingestor_plan, "be": 3, "fe": 1, "graph": false});
let s = q(starter.clone());
v.push((
format!("quote: the buyer's numbers are the shape (500 GiB, 2 ingestors on {ingestor_plan}, 3 BE, 1 FE: 7 machines, no main, no twin)"),
match &s {
Ok(q) if q.machines.len() == 7 && q.machines.iter().filter(|m| m.role.starts_with("ingestor")).all(|m| m.plan == ingestor_plan) && !roles(q).iter().any(|r| r == "main" || r == "twin") => {
Ok(format!("{:?}, {} a month", roles(q), q.monthly.in_words()))
}
Ok(q) => Err(format!("{:?}", q.machines)),
Err(e) => Err(e.clone()),
},
));
let mut graph = starter.clone();
graph["graph"] = serde_json::json!(true);
let g = q(graph);
v.push((
"quote: a graph database is one machine more".into(),
match (&s, &g) {
(Ok(a), Ok(b))
if b.machines.len() == a.machines.len() + 1
&& roles(b).last().map(String::as_str) == Some(role::GRAPH) =>
{
Ok(format!(
"{} machines, {} a month",
b.machines.len(),
b.monthly.in_words()
))
}
(a, b) => Err(format!("{a:?} / {b:?}")),
},
));
let mut be4 = starter.clone();
be4["be"] = serde_json::json!(4);
let grown = q(be4);
v.push((
"quote: grow = changing the numbers (be 3 -> 4): one backend more, a higher price".into(),
match (&s, &grown) {
(Ok(a), Ok(b))
if b.machines.len() == a.machines.len() + 1
&& b.monthly.minor > a.monthly.minor =>
{
Ok(format!(
"{} -> {} a month, + {}",
a.monthly.in_words(),
b.monthly.in_words(),
b.machines
.iter()
.filter(|m| m.role.starts_with("be"))
.count()
))
}
(a, b) => Err(format!("{a:?} / {b:?}")),
},
));
let bad = q(serde_json::json!({"fe": 2}));
v.push((
"quote: an order out of range is refused by name (2 frontends)".into(),
match bad {
Err(e) if e.contains("1, 3 or 5 frontends") => Ok(e),
other => Err(format!("{other:?}")),
},
));
v
}
/// The vendor's own word on a hold: the mock's state, or the sandbox's through the registry.
fn vendor_status(
state: &AppState,
pay: Option<&monetize_server::testing::CallbackPay>,
vendor: &str,
handle: &monetize_pay::ChargeHandle,
) -> Option<monetize_pay::ChargeStatus> {
match pay {
Some(p) => p.vendor_state(&handle.id),
None => state
.registry
.find_vendor(vendor)
.and_then(|v| v.charge(handle).ok()),
}
}
/// **The buyer pays the page** of `handle` (`checkout`, its URL): the mock vendor is told so,
/// the sandbox pays on the spot, a signed vendor's page is paid at its fake and the vendor's
/// signed callback delivered to the ONE box.
fn buyer_pays(
pay: Option<&monetize_server::testing::CallbackPay>,
fake: Option<&crate::signed::VendorFake>,
one: std::net::SocketAddr,
handle: &monetize_pay::ChargeHandle,
checkout: &str,
) -> Result<String, String> {
match (pay, fake) {
(Some(p), _) => {
p.authorize(&handle.id);
Ok(format!(
"authorised {} at the mock vendor; captures so far {}",
handle.id,
p.captures()
))
}
(None, Some(f)) => f.pay(one, handle, checkout),
(None, None) => Ok(format!(
"{}: the sandbox vendor pays on the spot; nobody authorises",
handle.id
)),
}
}
/// What the cloud still holds under the estate's reference.
fn live_under(
state: &AppState,
mock: Option<&monetize::testing::MockCloud>,
cloud: &str,
reference: &str,
) -> usize {
match mock {
Some(c) => c.live_under(reference),
None => state
.registry
.find_cloud(cloud)
.and_then(|c| monetize_cloud::ask::account(&c).ok())
.map(|i| {
i.owned
.iter()
.filter(|o| o.reference.as_deref() == Some(reference))
.count()
})
.unwrap_or(usize::MAX),
}
}
/// **GROW THE STORE** (A4-FACTS: new numbers that grow the disks of machines that stand): the
/// Lake's input store spread over its standing backends, `store_gib` 100 → 200, grows each
/// backend's data disk whose share is bigger now — at the cloud, by the shop, once paid — and
/// the store never shrinks. A cloud that grows no disk (the sandbox's dummy) refuses the change
/// by name, nothing held. The bench's MockCloud is given a growth path at the backends' size.
#[allow(clippy::too_many_arguments)]
async fn grow_the_store(
r: &Receipt,
state: &Arc<AppState>,
pay: Option<&monetize_server::testing::CallbackPay>,
mock: Option<&monetize::testing::MockCloud>,
fake: Option<&crate::signed::VendorFake>,
one: std::net::SocketAddr,
boxes: &Boxes,
fleet: &str,
axes: &Axes,
vendor_name: &str,
) -> Result<(), String> {
let data_gib = |role: &str| {
state
.estates
.get(COMPANY)
.and_then(|run| run.laid)
.and_then(|l| l.machine(role).and_then(|m| m.disks.get(1)).map(|d| d.gib))
.unwrap_or(0)
};
let before: Vec<(String, u64)> = [role::BE, "be-2"]
.iter()
.map(|b| (b.to_string(), data_gib(b)))
.collect();
if let Some(m) = mock {
let smallest = before
.iter()
.map(|(_, g)| *g)
.filter(|g| *g > 0)
.min()
.unwrap_or(50);
m.grows(monetize::testing::GrowScript {
current_gib: smallest,
..monetize::testing::GrowScript::default()
});
}
let (st, cn) = (state.clone(), axes.cloud_name().to_string());
let grows_disks = tokio::task::spawn_blocking(move || {
st.registry
.find_cloud(&cn)
.map(|c| monetize_cloud::ask::grow_gate(&c).is_ok())
.unwrap_or(false)
})
.await
.map_err(|e| e.to_string())?;
let monthly_before = state
.estates
.get(COMPANY)
.map(|run| run.quote.monthly)
.unwrap_or(monetize_pay::Money {
minor: 0,
currency: monetize_pay::Currency::Eur,
});
let st = state.clone();
let store = tokio::task::spawn_blocking(move || {
estate_order::change::change(&st, COMPANY, &order(serde_json::json!({"store_gib": 200})))
})
.await
.map_err(|e| e.to_string())?;
if !grows_disks {
r.step(&format!("grow the store: store_gib 100 → 200 on a cloud that grows no disk ({}) is refused by name, nothing held", axes.cloud), match &store {
Err(e) if format!("{e:?}").contains("grows no disk") => Ok(format!("{e:?}")),
other => Err(format!("{other:?}")),
});
return Ok(());
}
let opened = r.step(
"grow the store: store_gib 100 → 200 on a standing Lake grows the backends' data disks whose share is bigger (no machine laid), its difference held",
match &store {
// The mock axis prices every resource flat (a bigger disk costs the same): nothing to
// hold there; a cloud that prices by size holds the difference.
Ok(c) if !c.grows.is_empty() && c.adds.is_empty() && c.removes.is_empty() && c.quote.monthly.minor >= monthly_before.minor => Ok(format!("{}: {} ({} held)", c.reference, c.words, c.amount.in_words())),
other => Err(format!("{other:?}")),
},
);
let Ok(c) = store else { return Ok(()) };
if !opened {
return Ok(());
}
if let Some(h) = &c.handle {
let (ch, co) = (h.clone(), c.checkout_url.clone());
let paid = tokio::task::block_in_place(|| buyer_pays(pay, fake, one, &ch, &co));
if fake.is_some() {
r.step(
&format!(
"grow the store: the buyer pays the change at the fake {}",
axes.vendor
),
paid,
);
}
}
let until = std::time::Instant::now() + std::time::Duration::from_secs(240);
while std::time::Instant::now() < until
&& state
.estates
.get(COMPANY)
.is_some_and(|run| run.change.is_some())
{
report_all(boxes, fleet, state.now()).await;
let s = state.clone();
tokio::task::spawn_blocking(move || estate_order::drive(&s, COMPANY))
.await
.map_err(|e| e.to_string())?;
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}
let run = state.estates.get(COMPANY);
let done = run.as_ref().and_then(|r| r.changes.last().cloned());
let now: Vec<(String, u64)> = c
.grows
.iter()
.map(|g| (g.role.clone(), data_gib(&g.role)))
.collect();
let (st, vn, pm, h) = (
state.clone(),
vendor_name.to_string(),
pay.cloned(),
c.handle.clone(),
);
let captured = match h {
Some(h) => tokio::task::spawn_blocking(move || vendor_status(&st, pm.as_ref(), &vn, &h))
.await
.map_err(|e| e.to_string())?,
None => None,
};
let monthly_now = run
.as_ref()
.map(|r| r.quote.monthly)
.unwrap_or(monthly_before);
r.step(
"grow the store: each standing backend's data disk is grown at the cloud and read back at its new size, the hold captured, the Lake billed for it",
match (&done, &captured) {
(Some(d), cap) if (c.handle.is_none() || matches!(cap, Some(monetize_pay::ChargeStatus::Captured { .. }))) && d.status == estate_order::change::status::DONE && c.grows.iter().all(|g| now.iter().any(|(r, gib)| *r == g.role && *gib == g.to_gib)) && monthly_now == c.quote.monthly => {
Ok(format!("{} done: {}; {} a month (was {})", d.reference, now.iter().map(|(r, g)| format!("{r} {g} GiB")).collect::<Vec<_>>().join(", "), monthly_now.in_words(), monthly_before.in_words()))
}
other => Err(format!("{other:?}; before {before:?}, now {now:?}")),
},
);
let st = state.clone();
let smaller = tokio::task::spawn_blocking(move || {
estate_order::change::change(&st, COMPANY, &order(serde_json::json!({"store_gib": 100})))
})
.await
.map_err(|e| e.to_string())?;
r.step("grow the store: the store never shrinks (facts.grow never_shrinks store_gib): 200 → 100 refused by name, nothing held", match smaller {
Err(e) if format!("{e:?}").contains("never shrinks") => Ok(format!("{e:?}")),
other => Err(format!("{other:?}")),
});
Ok(())
}
/// **The whole lifecycle.** `wasm`: the built Lake guest. `out`: where the receipt goes.
pub async fn run(wasm: &Path, out: &Path, commit: &str, axes: &Axes) -> Result<PathBuf, String> {
let r = Receipt::default();
let known = axes.check();
let known_ok = r.step("the bench's axes", known.clone());
let dir = out.join("one-box");
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).map_err(|e| e.to_string())?;
if !known_ok {
let p = r.write(out, commit, axes).map_err(|e| e.to_string())?;
return Err(format!(
"the bench FAILED: {}; see {}",
known.unwrap_err(),
p.display()
));
}
// ── the signed guests' fakes: the cloud's mock in process, the vendor's as a process ─
let signed = axes.signed.clone().map(crate::signed::Signed);
let (cloud_fake, media) = if crate::signed::CLOUDS.contains(&axes.cloud.as_str()) {
let (f, m) = crate::signed::CloudFake::start(&axes.cloud, &MEDIA.map(String::from)).await?;
(Some(f), m)
} else {
(None, std::collections::BTreeMap::new())
};
let vendor_fake = match (
crate::signed::VENDORS.contains(&axes.vendor.as_str()),
&signed,
) {
(true, Some(s)) => Some(crate::signed::VendorFake::start(&axes.vendor, s)?),
_ => None,
};
let signed_blocks = match &signed {
Some(s) if cloud_fake.is_some() || vendor_fake.is_some() => crate::signed::blocks(
&dir,
s,
cloud_fake.as_ref(),
vendor_fake.as_ref(),
&dir.join("secrets"),
)?,
_ => String::new(),
};
let vendor_name = axes.vendor_name();
// ── the ONE box, with the Lake pulled ──────────────────────────────────────────────
let boxes = Boxes::default();
let cfg_path = config(&dir, wasm, axes, &signed_blocks)?;
let cfg = monetize_server::config::Config::load(&cfg_path)
.map_err(|e| format!("{}: {e}", cfg_path.display()))?;
let lake_block = cfg
.products
.get("lake")
.map(|p| (p.instance.clone(), p.endpoint.clone()));
r.step(
"the product block is pulled: an instance, no endpoint, no grants",
match lake_block {
Some((Some(i), e)) if e.is_empty() => Ok(format!("[products.lake] instance = {i:?}")),
other => Err(format!("{other:?}")),
},
);
// The mock axes are the bench's own doubles; the sandbox axes are built from the config as a
// deployment builds them (the open default's `Linked::open()`), and the bench holds no handle.
let mock_pay = (axes.vendor == "mock").then(|| {
monetize_server::testing::CallbackPay::with_checkout(
"lake-bench-webhook-secret",
"https://pay.mock.invalid/c/",
)
});
let mock_cloud = (axes.cloud == "mock").then(|| Arc::new(monetize::testing::MockCloud::new()));
// A signed guest's host runs its own wait: built off the runtime's threads.
let mut registry =
tokio::task::block_in_place(|| monetize_server::registry::Registry::from_config(&cfg))
.map_err(|e| e.to_string())?;
if let Some(p) = &mock_pay {
registry = registry.sells("lake", p.clone());
}
if let Some(c) = &mock_cloud {
registry = registry.cloud(c.clone());
}
let pay = mock_pay.as_ref();
let cloud_name = axes.cloud_name().to_string();
if signed.is_some() {
let refused: Vec<String> = registry
.vendor_start_refusals()
.iter()
.map(|c| format!("{c:?}"))
.collect();
let found = (
registry.find_cloud(&cloud_name).is_some(),
registry.find_vendor(&vendor_name).is_some(),
);
r.step(
"the ONE box loads the signed guests from their blocks (signed under the run's CA, grants http/hmac/state, the host holding the keys)",
if refused.is_empty() && found == (true, true) { Ok(format!("cloud {cloud_name} and vendor {vendor_name} serve")) } else { Err(format!("cloud {cloud_name} found {}, vendor {vendor_name} found {}; refused {refused:?}", found.0, found.1)) },
);
}
let mut state = AppState::in_memory(registry);
state.signup_plans =
monetize_server::signup::Plans::of(cfg.signup_plans(), cfg.signup_cap_meters());
let custody = monetize_server::custody::Custody::open(dir.join("custody"))?;
state.estates = crate::installer::estate_orders(
estate_cfg("r1", axes, &media),
dir.join("estate-orders"),
boxes.clone(),
custody,
);
// The bench's clock: the real one, moved on when a step needs time to pass.
let offset = Arc::new(std::sync::atomic::AtomicU64::new(0));
let o = offset.clone();
state.clock = Box::new(move || {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0)
+ o.load(std::sync::atomic::Ordering::SeqCst)
});
let state = Arc::new(state);
let lines = estate_order::guest_lifecycle::follow(&state.registry);
let life = estate_order::lifecycle::find("lake");
r.step(
"the ONE box loads the signed Lake guest and takes its lifecycle (its own roles, no twin)",
match &life {
Some(l) if !l.roles().iter().any(|x| x == "main" || x == "twin") => Ok(format!(
"{}; roles {:?}, primary {}",
lines.join("; "),
l.roles(),
l.primary_role()
)),
Some(l) => Err(format!("roles {:?}", l.roles())),
None => Err(format!("no Lake lifecycle: {lines:?}")),
},
);
let server = monetize_server::testing::TestServer::spawn(state.clone(), BEARER).await;
let fleet = server.endpoint();
say("bench", "one-box", "bench", &format!("gRPC at {fleet}"));
// ── the parametric proof: the quote is the guest's plan for the buyer's numbers ────
let (st, c1, cn, ip) = (
state.clone(),
estate_cfg("r1", axes, &media),
cloud_name.clone(),
axes.ingestor_plan(),
);
for (step, res) in tokio::task::spawn_blocking(move || match st.registry.find_cloud(&cn) {
Some(c) => quotes(&c1, c, ip),
None => vec![(
"quote: the cloud is linked".to_string(),
Err(format!("no cloud {cn}")),
)],
})
.await
.map_err(|e| e.to_string())?
{
r.step(&step, res);
}
// ── BUY: the site's relay → EstateOrder.Start ──────────────────────────────────────
// The shop prices its ladder once before the first visitor (EstateOrder.Sells answers a
// kept answer and waits two seconds at most: a signed cloud's first price lists may take
// longer, and the relay would say "not sold" until the next ask).
let st = state.clone();
let sells = tokio::task::spawn_blocking(move || estate_order::sells_refresh(&st))
.await
.map_err(|e| e.to_string())?;
say(
"bench",
"sells",
"monetize-server",
&format!("the ladder priced: sells={sells}"),
);
let link = monetize_site_buy::ShopLink::connect_lazy(
&monetize_site_buy::Door {
url: fleet.clone(),
at: None,
ca_pem: None,
},
BEARER,
"lake",
)?;
let relay = monetize_site_buy::Relay::new(
crate::site::LakeSite {
boxes: boxes.clone(),
origin: "https://lake.mock.invalid".into(),
},
link,
);
let ask = monetize_site_buy::relay::Ask::new(axum::http::HeaderMap::new(), "127.0.0.1");
let sizes: monetize_site_buy::relay::Sizes = serde_json::from_value(
serde_json::json!({ "company": COMPANY, "account": ACCOUNT, "email": EMAIL }),
)
.map_err(|e| e.to_string())?;
let a = monetize_site_buy::relay::private_start(&relay, &ask, &sizes).await;
let started = a.status.is_success();
r.step(
"buy: a visitor starts the order through the site's relay (EstateOrder.Start over gRPC)",
if started {
Ok(a.body.to_string())
} else {
Err(format!("{} {}", a.status, a.body))
},
);
r.say(monetize_guest_lake::P3_WAITS[0]);
// ★ A33-9: a buy that does not start through the site's relay stops the bench: there is no
// second way in (the in-process `start_for` fallback that skipped the RPC is deleted).
if !started {
let p = r.write(out, commit, axes).map_err(|e| e.to_string())?;
return Err(format!("the bench FAILED: the visitor's order did not start through the site's relay ({} {}); see {}", a.status, a.body, p.display()));
}
// ── PAY ────────────────────────────────────────────────────────────────────────────
let charge = state
.estates
.get(COMPANY)
.and_then(|run| run.handle.clone())
.unwrap_or_else(|| monetize_pay::ChargeHandle {
vendor: String::new(),
id: String::new(),
});
let checkout = state
.estates
.get(COMPANY)
.map(|run| run.checkout_url.clone())
.unwrap_or_default();
let handle = charge.id.clone();
let one = server.addr;
let (vf, ch, co) = (vendor_fake.as_ref(), charge.clone(), checkout.clone());
let paid = tokio::task::block_in_place(|| buyer_pays(pay, vf, one, &ch, &co));
match &paid {
Ok(w) => say("bench", "vendor", "buyer", w),
Err(e) => say(
"bench",
"vendor",
"buyer",
&format!("the payment did not go through: {e}"),
),
}
if vendor_fake.is_some() {
r.step(
&format!(
"pay: the buyer pays at the fake {} and its signed callback reaches the ONE box",
axes.vendor
),
paid,
);
}
// ── LAY, INSTALL, SALE CHECK, CAPTURE ──────────────────────────────────────────────
let got = drive_until(&state, pay, &boxes, &fleet, &[phase::CAPTURED], 240).await;
r.step("lay → install → sale check → capture", got.clone());
let laid = state.estates.get(COMPANY).and_then(|run| run.laid.clone());
let laid_roles: Vec<String> = laid
.as_ref()
.map(|l| l.machines.iter().map(|m| m.role.clone()).collect())
.unwrap_or_default();
r.step(
"instantiate: the cloud laid exactly the guest's plan (front, ingestor, be, fe)",
if laid_roles == ["front", "ingestor", "be", "fe"] {
Ok(laid_roles.join(", "))
} else {
Err(format!("{laid_roles:?}"))
},
);
let dns: Vec<String> = crate::lines()
.into_iter()
.filter(|l| l.starts_with("lake: dns.write"))
.collect();
r.step(
"dns: the estate's names are the ones the Lake serves (s3., no git.)",
if dns.iter().any(|l| l.contains(" s3.")) && dns.iter().all(|l| !l.contains(" git.")) {
Ok(format!("{} record(s)", dns.len()))
} else {
Err(format!("{dns:?}"))
},
);
let (st, vn, ch) = (state.clone(), vendor_name.clone(), charge.clone());
let pm = mock_pay.clone();
let status = tokio::task::spawn_blocking(move || vendor_status(&st, pm.as_ref(), &vn, &ch))
.await
.map_err(|e| e.to_string())?;
let calls = vendor_fake
.as_ref()
.and_then(|f| tokio::task::block_in_place(|| f.calls_word(&checkout)));
r.step(
"the vendor holds the money captured",
match status {
Some(monetize_pay::ChargeStatus::Captured { .. }) => Ok(format!(
"{handle} captured at the {} vendor{}{}",
axes.vendor,
pay.map(|p| format!(" ({} capture call(s))", p.captures()))
.unwrap_or_default(),
calls.map(|c| format!("; {c}")).unwrap_or_default()
)),
other => Err(format!("{handle}: {other:?}")),
},
);
// ── THE BUYER, SIGNED IN ───────────────────────────────────────────────────────────
let mut headers = axum::http::HeaderMap::new();
headers.insert(crate::site::USER_HEADER, ACCOUNT.parse().unwrap());
let signed_in = monetize_site_buy::relay::Ask::new(headers, "127.0.0.1");
let a = monetize_site_buy::relay::private_status(&relay, &signed_in, COMPANY).await;
r.step("buy: the signed-in buyer reads the order's state through the relay (no ticket: the Lake mints none, so the answer is the state anyone may see)", if a.status.is_success() && a.body.get("records").is_none_or(|r| r.as_array().is_none_or(|v| v.is_empty())) { Ok(a.body.to_string()) } else { Err(format!("{} {}", a.status, a.body)) });
// ── READY: every machine reports; the pulled product answers from the primary's report ─
report_all(&boxes, &fleet, state.now()).await;
let instance = boxes
.get(role::FRONT)
.map(|b| b.held.lock().unwrap().instance.clone())
.unwrap_or_default();
let unheard: Vec<String> = laid_roles
.iter()
.filter(|role| {
!state
.fleet
.guardian()
.and_then(|g| g.store.box_record(&instance, role))
.is_some_and(|b| b.heard_unix_ms > 0)
})
.cloned()
.collect();
r.step(
"ready: the first report of every role arrived",
if unheard.is_empty() && !laid_roles.is_empty() {
Ok(format!("{instance}: {}", laid_roles.join(", ")))
} else {
Err(format!("no report of {unheard:?}"))
},
);
let st = state.clone();
let pulled = tokio::task::spawn_blocking(move || {
let p = st.registry.find_product("lake").ok_or("no lake product")?;
let tenants = p.list_tenants().map_err(|e| e.to_string())?;
let usage = p
.read_usage(&monetize_product::TenantId(ACCOUNT.into()))
.map_err(|e| e.to_string())?;
Ok::<_, String>((tenants, usage))
})
.await
.map_err(|e| e.to_string())?;
r.step("ready: the pulled product's tenants and usage come from its primary's report (no call into the estate)", match pulled {
Ok((t, u)) if u.meters.get(METER) == Some(&42) => Ok(format!("tenants {t:?}, {METER}=42 from the {} report", life.as_ref().map(|l| l.primary_role()).unwrap_or_default())),
other => Err(format!("{other:?}")),
});
r.say(monetize_guest_lake::P3_WAITS[1]);
// An entitlement is an ORDER the primary pulls, never a push.
let st = state.clone();
let queued = tokio::task::spawn_blocking(move || {
let p = st.registry.find_product("lake").ok_or("no lake product")?;
let fact = monetize_product::EntitlementFact {
tenant: monetize_product::TenantId(ACCOUNT.into()),
plan: "lake/mini".into(),
caps: [(METER.to_string(), 100u64 << 30)].into_iter().collect(),
source: "bench".into(),
state: monetize_product::State::Paid,
paid_until_unix_ms: None,
signature: Vec::new(),
issued_unix_ms: None,
issued_signature: Vec::new(),
};
p.push_entitlement(&fact).map_err(|e| e.to_string())
})
.await
.map_err(|e| e.to_string())?;
let primary = life.as_ref().map(|l| l.primary_role()).unwrap_or_default();
let pulled_kinds = match (&queued, boxes.get(&primary)) {
(Ok(()), Some(b)) => b.orders(&fleet, state.now(), 100).await,
(Err(e), _) => Err(e.clone()),
(_, None) => Err(format!("no {primary} box")),
};
r.step(
"fleet: an entitlement is a signed order the primary pulls (Fleet.Orders)",
match pulled_kinds {
Ok(k)
if k.iter()
.any(|k| k == monetize_product::pull::order::ENTITLEMENT) =>
{
Ok(format!("the {primary} pulled {k:?}"))
}
other => Err(format!("{other:?}")),
},
);
// ── GROW: new numbers on a standing Lake lay the machines they add (P3-FACTS) ────────
let monthly_before = state
.estates
.get(COMPANY)
.map(|run| run.quote.monthly)
.unwrap_or(monetize_pay::Money {
minor: 0,
currency: monetize_pay::Currency::Eur,
});
let st = state.clone();
let grow = tokio::task::spawn_blocking(move || {
estate_order::change::change(&st, COMPANY, &order(serde_json::json!({"be": 2})))
})
.await
.map_err(|e| e.to_string())?;
let opened = r.step(
"grow: be 1 → 2 on a standing Lake opens a plan change that lays be-2 (the product's plan for the new numbers; facts.grow may_add be), its difference held",
match &grow {
Ok(c) if c.adds.iter().map(|a| a.role.as_str()).eq(["be-2"]) && c.quote.monthly.minor > monthly_before.minor => Ok(format!("{}: {} ({} held, {} a month after)", c.reference, c.words, c.amount.in_words(), c.quote.monthly.in_words())),
other => Err(format!("{other:?}")),
},
);
r.say(monetize_guest_lake::P3_WAITS[4]);
if let (true, Ok(c)) = (opened, &grow) {
let handle = c.handle.clone();
if let Some(h) = &handle {
let (vf, ch, co) = (vendor_fake.as_ref(), h.clone(), c.checkout_url.clone());
let paid = tokio::task::block_in_place(|| buyer_pays(pay, vf, one, &ch, &co));
if vendor_fake.is_some() {
r.step(
&format!(
"grow: the buyer pays the change at the fake {}",
axes.vendor
),
paid,
);
}
}
let until = std::time::Instant::now() + std::time::Duration::from_secs(240);
while std::time::Instant::now() < until
&& state
.estates
.get(COMPANY)
.is_some_and(|run| run.change.is_some())
{
report_all(&boxes, &fleet, state.now()).await;
let s = state.clone();
tokio::task::spawn_blocking(move || estate_order::drive(&s, COMPANY))
.await
.map_err(|e| e.to_string())?;
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}
let run = state.estates.get(COMPANY);
let done = run.as_ref().and_then(|r| r.changes.last().cloned());
let roles_now: Vec<String> = run
.as_ref()
.and_then(|r| r.laid.as_ref())
.map(|l| l.machines.iter().map(|m| m.role.clone()).collect())
.unwrap_or_default();
let paired = boxes
.get("be-2")
.map(|b| !b.held.lock().unwrap().instance.is_empty())
.unwrap_or(false);
let heard = state
.fleet
.guardian()
.and_then(|g| g.store.box_record(&instance, "be-2"))
.is_some_and(|b| b.heard_unix_ms > 0);
r.step(
"grow: be-2 is laid at the cloud beside the estate, paired by the estate's installer and heard on the pull channel",
if roles_now.iter().any(|x| x == "be-2") && paired && heard { Ok(format!("laid {}", roles_now.join(", "))) } else { Err(format!("laid {roles_now:?}, paired {paired}, heard {heard}; the change: {:?}", run.as_ref().and_then(|r| r.change.clone()).or(done.clone()))) },
);
let st = state.clone();
let (vn, pm) = (vendor_name.clone(), mock_pay.clone());
let captured = match handle {
Some(h) => {
tokio::task::spawn_blocking(move || vendor_status(&st, pm.as_ref(), &vn, &h))
.await
.map_err(|e| e.to_string())?
}
None => None,
};
let monthly_now = run
.as_ref()
.map(|r| r.quote.monthly)
.unwrap_or(monthly_before);
r.step(
"grow: the change is done, its hold captured, and the Lake is billed for its new machine from now on",
match (&done, &captured) {
(Some(d), Some(monetize_pay::ChargeStatus::Captured { .. })) if d.status == estate_order::change::status::DONE && monthly_now == c.quote.monthly => Ok(format!("{} done; {} a month (was {})", d.reference, monthly_now.in_words(), monthly_before.in_words())),
other => Err(format!("{other:?}; {} a month", monthly_now.in_words())),
},
);
}
let st = state.clone();
let smaller = tokio::task::spawn_blocking(move || {
estate_order::change::change(&st, COMPANY, &order(serde_json::json!({"be": 1})))
})
.await
.map_err(|e| e.to_string())?;
r.step("grow: a backend is never taken away by a buy (facts.grow may_remove names none): refused by name, nothing held", match smaller {
Err(e) if format!("{e:?}").contains("takes no be-2 away") => Ok(format!("{e:?}")),
other => Err(format!("{other:?}")),
});
grow_the_store(
&r,
&state,
pay,
mock_cloud.as_deref(),
vendor_fake.as_ref(),
one,
&boxes,
&fleet,
axes,
&vendor_name,
)
.await?;
let laid_roles: Vec<String> = state
.estates
.get(COMPANY)
.and_then(|r| {
r.laid
.map(|l| l.machines.into_iter().map(|m| m.role).collect())
})
.unwrap_or(laid_roles);
// ── RENEW ──────────────────────────────────────────────────────────────────────────
if let Some(run) = state.estates.get(COMPANY) {
let at = run.paid_until_ms.saturating_sub(20 * 86_400_000);
offset.store(
at.saturating_sub(state.now()),
std::sync::atomic::Ordering::SeqCst,
);
}
let st = state.clone();
let renew = tokio::task::spawn_blocking(move || {
estate_order::term::renew(&st, COMPANY).map_err(|e| format!("{e:?}"))
})
.await
.map_err(|e| e.to_string())?;
match renew {
Ok(n) => {
r.step(
"renew: EstateOrder.Renew opens the renewal's page",
Ok(format!("{} {}", n.reference, n.checkout_url)),
);
let before = state
.estates
.get(COMPANY)
.map(|r| r.paid_until_ms)
.unwrap_or(0);
let (vf, nh, nc) = (
vendor_fake.as_ref(),
n.handle.clone(),
n.checkout_url.clone(),
);
let paid = tokio::task::block_in_place(|| buyer_pays(pay, vf, one, &nh, &nc));
if vendor_fake.is_some() {
r.step(
&format!(
"renew: the buyer pays the renewal at the fake {}",
axes.vendor
),
paid,
);
}
let mut moved = false;
for _ in 0..40 {
report_all(&boxes, &fleet, state.now()).await;
let s = state.clone();
tokio::task::spawn_blocking(move || estate_order::drive(&s, COMPANY))
.await
.map_err(|e| e.to_string())?;
if state
.estates
.get(COMPANY)
.map(|r| r.paid_until_ms)
.unwrap_or(0)
> before
{
moved = true;
break;
}
}
let after = state
.estates
.get(COMPANY)
.map(|r| r.paid_until_ms)
.unwrap_or(0);
r.step(
"renew: the paid renewal moves paid_until on",
if moved {
Ok(format!("paid until {before} -> {after}"))
} else {
Err(format!("paid until stayed {before}"))
},
);
}
Err(e) => {
r.step("renew: EstateOrder.Renew opens the renewal's page", Err(e));
}
}
// ── UPDATE: r2 fed; every member of every role takes its step ──────────────────────
server.stop().await;
let mut st = Arc::try_unwrap(state)
.map_err(|_| "the bench still holds the ONE box's state".to_string())?;
let custody = monetize_server::custody::Custody::open(dir.join("custody"))?;
st.estates = crate::installer::estate_orders(
estate_cfg("r2", axes, &media),
dir.join("estate-orders"),
boxes.clone(),
custody,
);
let state = Arc::new(st);
say(
"bench",
"release",
"operator",
"release r2 fed; the ONE box restarted on it",
);
let server = monetize_server::testing::TestServer::spawn(state.clone(), BEARER).await;
let fleet = server.endpoint();
let st = state.clone();
let upd = tokio::task::spawn_blocking(move || {
estate_order::update::start(&st, COMPANY, "", "bench")
.map(|u| u.line())
.map_err(|e| format!("{e:?}"))
})
.await
.map_err(|e| e.to_string())?;
let opened = r.step("update: UpdateEstate opens the update to r2", upd);
// A machine runs r2 when its release is r2's medium: its name, or its id at a signed cloud.
let r2: Vec<String> = ["lake-r2", "lake-keep-r2"]
.iter()
.map(|m| media.get(*m).cloned().unwrap_or_else(|| m.to_string()))
.collect();
let on_r2 = |x: &String| r2.iter().any(|m| x.contains(m.as_str()));
let keep_run = Arc::new(std::sync::atomic::AtomicBool::new(true));
let keep_player = cloud_fake
.as_ref()
.and_then(|c| c.play_keep_media(keep_run.clone()));
if opened {
let front = boxes.get(role::FRONT).ok_or("no front box")?;
let mut stepped = Vec::new();
let until = std::time::Instant::now() + std::time::Duration::from_secs(240);
while std::time::Instant::now() < until {
report_all(&boxes, &fleet, state.now()).await;
if let Ok(t) = front.update_take(&fleet, state.now()).await {
if t.open && !t.role.is_empty() {
let outcome = if t.live { "done:live" } else { "done:keep" };
if !stepped.contains(&t.role) {
stepped.push(t.role.clone());
}
let _ = front
.update_report(
&fleet,
&t.role,
outcome,
"the Lake box runs the release",
state.now(),
)
.await;
}
}
let s = state.clone();
tokio::task::spawn_blocking(move || estate_order::tick(&s))
.await
.map_err(|e| e.to_string())?;
// The front's keep boot is the server loop's queued jera job (proven by
// estate_order/tests/front_keep_job.rs); the bench plays the loop and takes the step
// itself when the update waits for it.
if state
.estates
.get(COMPANY)
.and_then(|r| {
r.update
.as_ref()
.map(|u| u.state == estate_order::update::state::FRONT_KEEP)
})
.unwrap_or(false)
{
let s = state.clone();
tokio::task::spawn_blocking(move || estate_order::update::front_keep(&s, COMPANY))
.await
.map_err(|e| e.to_string())?;
say("bench", "front-keep", "bench", "the shop's keep boot of the front, taken by the bench as the server loop's job would");
}
let run = state.estates.get(COMPANY);
if run.as_ref().is_some_and(|r| {
laid_roles
.iter()
.all(|role| r.release_of.get(role).is_some_and(on_r2))
}) {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}
let run = state.estates.get(COMPANY);
let line = run
.as_ref()
.and_then(|r| r.update.clone().map(|u| u.line()))
.unwrap_or_default();
let behind: Vec<String> = laid_roles
.iter()
.filter(|role| {
!run.as_ref()
.is_some_and(|r| r.release_of.get(*role).is_some_and(on_r2))
})
.map(|role| {
format!(
"{role} at {:?}",
run.as_ref().and_then(|r| r.release_of.get(role).cloned())
)
})
.collect();
// The front never asked before this update, so the shop keep-boots it first (the step the
// bench took); then the new front takes every other member in the facts' order.
let others: Vec<String> = laid_roles
.iter()
.filter(|r| *r != role::FRONT)
.cloned()
.collect();
r.step(
"update: the shop keep-boots the front first, then the front takes every other member; every machine runs r2",
if behind.is_empty() && stepped.iter().all(|s| others.contains(s)) && stepped.len() == others.len() { Ok(format!("the front keep-booted by the shop, then it took {stepped:?}; {line}")) } else { Err(format!("behind {behind:?}, stepped {stepped:?}: {line}")) },
);
}
keep_run.store(false, std::sync::atomic::Ordering::SeqCst);
if let Some(p) = keep_player {
let _ = p.join();
}
// ── TERMINATE ──────────────────────────────────────────────────────────────────────
let st = state.clone();
let term = tokio::task::spawn_blocking(move || {
estate_order::terminate::terminate(&st, None, COMPANY, COMPANY)
.map(|t| format!("{t:?}"))
.map_err(|e| format!("{e:?}"))
})
.await
.map_err(|e| e.to_string())?;
r.step("terminate: no backup, the estate ends", term);
// A signed cloud gives back at its own pace (a server stops before it is deleted): up to
// three minutes, ticked as the server loop ticks.
let until = std::time::Instant::now() + std::time::Duration::from_secs(180);
while std::time::Instant::now() < until {
if state
.estates
.get(COMPANY)
.map(|r| r.phase == phase::TERMINATED)
.unwrap_or(true)
&& tokio::task::block_in_place(|| {
live_under(
&state,
mock_cloud.as_deref(),
&cloud_name,
&format!("estate/{COMPANY}"),
)
}) == 0
{
break;
}
let s = state.clone();
tokio::task::spawn_blocking(move || estate_order::tick(&s))
.await
.map_err(|e| e.to_string())?;
tokio::time::sleep(std::time::Duration::from_millis(250)).await;
}
say("bench", "phase", "monetize-server", &phase_of(&state));
r.step("terminate: the cloud holds nothing of the estate", {
let left = tokio::task::block_in_place(|| {
live_under(
&state,
mock_cloud.as_deref(),
&cloud_name,
&format!("estate/{COMPANY}"),
)
});
let servers = cloud_fake.as_ref().map(|c| c.servers_left()).unwrap_or(0);
if left == 0 && servers == 0 {
Ok(format!(
"0 resources left{}",
cloud_fake
.as_ref()
.map(|c| format!("; the {} mock holds 0 servers", c.name()))
.unwrap_or_default()
))
} else {
Err(format!(
"{left} resources left under estate/{COMPANY}, {servers} server(s) at the mock"
))
}
});
for w in &monetize_guest_lake::P3_WAITS[2..] {
if *w != monetize_guest_lake::P3_WAITS[4] {
r.say(w);
}
}
server.stop().await;
// A signed guest's host is dropped off the runtime's threads too.
tokio::task::block_in_place(move || drop(state));
let p = r.write(out, commit, axes).map_err(|e| e.to_string())?;
if r.passed() {
Ok(p)
} else {
Err(format!("the bench FAILED; see {}", p.display()))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_axes_take_the_dummies_and_the_signed_guests_and_refuse_the_rest_by_name() {
let a = |cloud: &str, vendor: &str| Axes {
cloud: cloud.into(),
vendor: vendor.into(),
signed: None,
};
assert!(a("mock", "mock").check().is_ok());
assert!(a("sandbox", "sandbox").check().is_ok());
let e = a("aws", "mock").check().unwrap_err();
assert!(
e.contains("no cloud \"aws\"") && e.contains("scaleway"),
"{e}"
);
let e = a("mock", "paypal").check().unwrap_err();
assert!(
e.contains("no vendor \"paypal\"") && e.contains("klarna"),
"{e}"
);
let e = a("upcloud", "stripe").check().unwrap_err();
assert!(
e.contains("--signed") && e.contains("../monetize-impl"),
"a signed axis without its build is refused by name: {e}"
);
let s = Axes {
signed: Some(std::env::temp_dir().join("lake-bench-no-such-dir")),
..a("scaleway", "mock")
};
assert!(s.check().unwrap_err().contains("scaleway.wasm"));
assert_eq!(
(
a("upcloud", "stripe").cloud_name(),
a("upcloud", "stripe").vendor_name().as_str()
),
("upcloud", "stripe/lake")
);
assert_eq!(
(
a("mock", "sandbox").cloud_name(),
a("mock", "sandbox").vendor_name().as_str()
),
("mockcloud", SANDBOX_VENDOR)
);
assert_eq!(
(
a("scaleway", "mock").zone(),
a("scaleway", "mock").ingestor_plan()
),
("fr-par-1", "DEV1-M")
);
}
#[test]
fn a_medium_at_a_signed_cloud_is_its_id_there_and_the_vendor_names_the_currency() {
let media: std::collections::BTreeMap<String, String> =
[("lake-r1".to_string(), "01-uuid".to_string())]
.into_iter()
.collect();
let c = estate_cfg(
"r1",
&Axes {
cloud: "upcloud".into(),
vendor: "swish".into(),
signed: None,
},
&media,
);
assert_eq!(c.media.get("lake").map(String::as_str), Some("01-uuid"));
assert_eq!(
c.media.get("lake-keep").map(String::as_str),
Some("lake-keep-r1"),
"a name the mock does not hold stays a name"
);
assert_eq!(
(
c.vendor.as_str(),
c.cloud.as_str(),
c.currency.as_str(),
c.zone.as_str()
),
("swish/lake", "upcloud", "SEK", "se-sto1")
);
}
}
#[cfg(test)]
mod one_way_in {
/// ★ A33-9: the bench buys only through the site's relay — its code starts no order in
/// process (the old fallback that skipped `EstateOrder.Start` is deleted).
#[test]
fn the_bench_starts_no_order_in_process() {
let src = include_str!("bench.rs");
let call = ["estate_order::", "start_for("].concat();
assert!(
!src.contains(&call),
"the bench must not start an order in process"
);
}
}