crates/monetize-estate-side/src/sold_door.rs
//! **`ShopDoor`: a sold estate's signed shop-door client** (feature `sold-door`; P4a, A22 W9; over `Fleet.Ask` since P3-FLEET).
//!
//! A sold estate runs NO monetize of its own (G16): its boxes pull from the ONE monetize
//! ([`crate::pump`]), and its front calls the ONE monetize (`Fleet.Ask`, H16) for what only the
//! shop can do for it — every call signed by the estate's client key (the fed
//! `secret.shop_client_key`), nothing listening on the front:
//!
//! | call | what |
//! |---|---|
//! | `TakeMaterial` | the installer material the shop's pairing left, then `done` |
//! | `Recharge` | one LOST box laid again with a fresh seed, then `done` |
//! | `Upgrade` | one standing box booted once from the release's keep medium (data kept), then `done` |
//! | `UpdateTake` / `UpdateReport` | the update the operator opened, pulled box by box (the shared update step's [`Shop`]) |
//!
//! Moved from Holger's `holger-estate-front/src/shop_door.rs` (Holger first); Gunnar's
//! `gunnar-front/src/shop_door.rs` carried the same `unhex32`, `from_status` and `blocking` and
//! moves in T16 ([`ShopDoor::from_settings`] and [`ShopDoor::with_seam`] are its constructors).
//! What a front does with the answers (keeping the material in its primary's slot, re-admitting a
//! box) stays the product's.
//!
//! **Thin on purpose** (TASKS-4-v2 T16, A28 H16): the SoldEstate → Fleet merge replaces the
//! transport ([`Wire`]) and nothing else.
//!
//! **P3-FLEET (plan §8, decided 2026-10-24): the five calls ride `Fleet.Ask`.** [`FleetWire`] asks
//! each as a `Fleet.Ask` kind (`upgrade`, `recharge`, `material`, `update_take`,
//! `update_report`; monetize-embed `fleet::ask::FRONT`), a `FleetCall` of `role = front` signed
//! by the estate's client key, the old door's answer decoded from `FleetAskAnswer.answer`. A door
//! fed its `product` ([`FEED_SHOP_DOOR`]) or told it ([`ShopDoor::or_over_fleet`]) takes it.
//! **H16: the old `SoldEstate` door and its overlap are gone** (the ONE box served both doors
//! for one release, R3+R4): there is no fallback, and a door that knows no product to sign as
//! refuses every call by name ([`NoProduct`]) — every live front says its product.
use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Duration;
use ed25519_dalek::Signer as _;
use monetize_embed::fleet::{
ask as fleet_ask, call as fleet_call, fleet_call_message, Ask, FleetCall, FrontAsk,
};
use monetize_embed::signing::{estate_call_message, EstateCall};
use monetize_proto as pb;
use nordisk_pairing::frontside::update::{LiveApp, Shop, Take};
pub use crate::unhex32;
/// The fed settings (`monetize-shop-pairing` `FEED_SHOP_DOOR`): TOML `door`, `company`,
/// `shop_pubkey`, `client_key_secret`.
pub const FEED_SHOP_DOOR: &str = "shop_door.toml";
/// The secret the client key is fed as when the settings name none (64 hex, its seed).
pub const CLIENT_KEY_SECRET: &str = "shop_client_key";
pub const CALL_UPGRADE: &str = "upgrade";
pub const CALL_RECHARGE: &str = "recharge";
pub const CALL_TAKE_MATERIAL: &str = "take-material";
pub const CALL_UPDATE_TAKE: &str = "update-take";
pub const CALL_UPDATE_REPORT: &str = "update-report";
/// An in-place upgrade waits on the shop's whole keep boot (import, clone, stop, the medium's
/// 45 min, give-back, start): two hours, the ONE constant the shop holds its Upgrade call by
/// (monetize `KEEP_BOOT_WITHIN`; A7 row 9, Gunnar b74ca801). A front that gave up first counted a
/// slow keep boot as lost and asked for a Recharge while the shop still worked.
pub const UPGRADE_WITHIN: Duration = monetize_embed::fleet::KEEP_BOOT_WITHIN;
/// A Recharge waits up to an hour (the shop lays a lost box within 30 min). Holger's bound;
/// Gunnar's front waits 30 min today (T16 takes this one).
pub const RECHARGE_WITHIN: Duration = Duration::from_secs(60 * 60);
/// Every other call (a take, a report, the material).
pub const READ_WITHIN: Duration = Duration::from_secs(30);
/// **How an Upgrade is asked: started, then polled** (REL-T2-1, 2026-10-29). One ask waits at
/// most `ask` for its answer; an ask that was cut (no answer from the shop: the connection
/// dropped, or `ask` passed) or answered "already running" is asked again every `every`, until
/// the shop answers or `within` has passed. The shop's keep boot runs on whether or not the
/// front is still on the line.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Polling {
pub ask: Duration,
pub every: Duration,
pub within: Duration,
}
/// **The tag of a call that was cut**: no answer came from the shop (the connection dropped
/// mid-call, the ask's own bound passed, or the door could not be reached), so the shop may still
/// be working on it. Said as `unavailable: [shop-cut] <why>`.
pub const SHOP_CUT: &str = "[shop-cut]";
/// How often every call's connection pings the ONE box (HTTP/2 PING), and how long a ping may
/// go unanswered: the edge closes a connection silent for 120 s (edda
/// `estate-front::web::IDLE_TIMEOUT`), and a call waiting on the shop sends nothing else.
pub const KEEPALIVE_EVERY: Duration = Duration::from_secs(20);
pub const KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(10);
impl Polling {
/// The front's own: each ask well under the ONE box edge's 120 s idle cut (edda
/// `estate-front::web::IDLE_TIMEOUT`), polled every 15 s, for the shop's whole keep boot.
pub const UPGRADE: Polling = Polling {
ask: Duration::from_secs(60),
every: Duration::from_secs(15),
within: UPGRADE_WITHIN,
};
}
/// The fed keys of the ONE box's pull channel the seam is read from (the boxes' own).
pub const FEED_FLEET: &str = "fleet";
pub const FEED_FLEET_AT: &str = nordisk_pairing::fleet::FEED_FLEET_AT;
pub const FEED_FLEET_CA: &str = nordisk_pairing::fleet::FEED_FLEET_CA;
/// The estate's name on the pull channel, fed beside the door (monetize-shop-pairing's
/// `FEED_FLEET_INSTANCE`); the door's `company` when none is fed.
pub const FEED_FLEET_INSTANCE: &str = "fleet_instance";
/// The role a front asks as on `Fleet`.
pub const FRONT_ROLE: &str = "front";
/// **Where the shop is, and as whom this estate calls it.**
#[derive(Clone)]
pub struct ShopDoor {
/// `https://monetize.<apex>`.
pub door: String,
pub company: String,
client: [u8; 32],
clock: Option<Arc<dyn Fn() -> i64 + Send + Sync>>,
/// **The seam** (row 16, Gunnar 7c3feb61): where the door is DIALLED (`fleet_at`,
/// `host:port`) and the CA its certificate is verified under (`fleet_ca`), fed for the ONE
/// box's pull channel. The TLS name stays the door's host. None: the door's host, under the
/// public roots.
pub seam: Option<Seam>,
wire: Arc<dyn Wire>,
/// Whether the calls ride `Fleet.Ask` ([`FleetWire`]).
on_fleet: bool,
/// How an Upgrade is asked ([`Polling::UPGRADE`]; a test's shop polls faster).
polling: Polling,
}
/// The dialled address and the CA of the shop door ([`ShopDoor::seam`]).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Seam {
pub at: String,
pub ca_pem: Option<String>,
}
/// Where one call goes: the door, and its seam when fed.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Dial {
pub door: String,
pub seam: Option<Seam>,
}
/// **The transport the calls ride** — what the SoldEstate → Fleet merge swaps. Each call is the
/// signed request as monetize-proto shapes it, answered within `within`; an error is the shop's
/// refusal by name (`refused: …`) or worth asking again (`unavailable: …`).
pub trait Wire: Send + Sync {
fn upgrade(
&self,
d: &Dial,
r: pb::EstateUpgradeRequest,
within: Duration,
) -> Result<pb::EstateUpgradeResponse, String>;
fn recharge(
&self,
d: &Dial,
r: pb::EstateRechargeRequest,
within: Duration,
) -> Result<pb::EstateRechargeResponse, String>;
fn take_material(
&self,
d: &Dial,
r: pb::EstateTakeMaterialRequest,
within: Duration,
) -> Result<pb::EstateTakeMaterialResponse, String>;
fn update_take(
&self,
d: &Dial,
r: pb::EstateUpdateTakeRequest,
within: Duration,
) -> Result<pb::EstateUpdateTakeResponse, String>;
fn update_report(
&self,
d: &Dial,
r: pb::EstateUpdateReportRequest,
within: Duration,
) -> Result<pb::EstateUpdateReportResponse, String>;
/// The product and the estate the calls are signed as on `Fleet` ([`FleetWire`]); none for
/// any other wire.
fn fleet_as(&self) -> Option<(String, String)> {
None
}
}
/// Pure: the host of an `http(s)://host[:port]/…` URL.
fn host_of(url: &str) -> &str {
let rest = url.split_once("://").map(|(_, r)| r).unwrap_or(url);
let authority = rest.split('/').next().unwrap_or("");
authority
.rsplit_once(':')
.map(|(h, _)| h)
.unwrap_or(authority)
}
/// Pure: **the seam of the shop door `door`** from the feed: `fleet_at` (and `fleet_ca`) when
/// the fed `fleet` is a door on the same host — the ONE box the boxes already dial there.
pub fn seam_of(door: &str, feed: &BTreeMap<String, String>) -> Option<Seam> {
let get = |k: &str| feed.get(k).map(|v| v.trim()).filter(|v| !v.is_empty());
let at = get(FEED_FLEET_AT)?;
let fleet = get(FEED_FLEET)?;
(host_of(fleet) == host_of(door) && !host_of(door).is_empty()).then(|| Seam {
at: at.to_string(),
ca_pem: get(FEED_FLEET_CA).map(str::to_string),
})
}
impl std::fmt::Debug for ShopDoor {
/// Never the client key.
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ShopDoor")
.field("door", &self.door)
.field("company", &self.company)
.field("at", &self.seam.as_ref().map(|s| s.at.as_str()))
.finish_non_exhaustive()
}
}
impl ShopDoor {
/// **From the fed settings** (`shop_door.toml`'s `door`, `company`, `shop_pubkey`,
/// `client_key_secret`) and the fed client key itself (64 hex). The shop's key, when named,
/// must be a key. No seam: [`ShopDoor::with_seam`].
pub fn from_settings(settings: &toml::Table, client_key_hex: &str) -> Result<ShopDoor, String> {
ShopDoor::settings(settings, client_key_hex, true)
}
/// [`ShopDoor::from_settings`]; `shop_key` says whether a named `shop_pubkey` must be a key
/// (Gunnar's front checks it; Holger's front verifies nothing under it and never did).
fn settings(
settings: &toml::Table,
client_key_hex: &str,
shop_key: bool,
) -> Result<ShopDoor, String> {
let get = |k: &str| {
settings
.get(k)
.and_then(|v| v.as_str())
.map(str::trim)
.filter(|v| !v.is_empty())
.map(str::to_string)
};
let door = get("door")
.ok_or("`door` is not set: the shop's monetize door (`https://monetize.<apex>`)")?;
if !(door.starts_with("https://") || door.starts_with("http://")) {
return Err(format!("`door` {door:?} is not an http(s) URL"));
}
let company =
get("company").ok_or("`company` is not set: this estate's name at the shop")?;
if let Some(k) = get("shop_pubkey").filter(|_| shop_key) {
monetize_embed::VerifyingKey::from_bytes(&unhex32(&k, "`shop_pubkey`")?)
.map_err(|e| format!("`shop_pubkey` is not a key: {e}"))?;
}
let secret = get("client_key_secret").unwrap_or_else(|| CLIENT_KEY_SECRET.to_string());
let client = unhex32(
client_key_hex,
&format!("the client key (the fed secret `{secret}`)"),
)?;
let mut d = ShopDoor {
door: door.trim_end_matches('/').to_string(),
company,
client,
clock: None,
seam: None,
wire: Arc::new(NoProduct),
on_fleet: false,
polling: Polling::UPGRADE,
};
if let Some(product) = get("product") {
let instance = d.company.clone();
d = d.over_fleet(&product, &instance);
}
Ok(d)
}
/// **From the front's feed**; `None` when it names no shop door (an open or a site front). The
/// seam is the ONE box's pull channel's, taken when its `fleet` is the same host
/// ([`seam_of`]) and only as a bench feeds it (nordisk-pairing `fleet::seam_refusal`).
pub fn from_feed(feed: &BTreeMap<String, String>) -> Option<Result<ShopDoor, String>> {
let text = feed.get(FEED_SHOP_DOOR)?;
Some((|| {
let t: toml::Table =
toml::from_str(text).map_err(|e| format!("{FEED_SHOP_DOOR}: {e}"))?;
let secret = t
.get("client_key_secret")
.and_then(|v| v.as_str())
.map(str::trim)
.filter(|v| !v.is_empty())
.unwrap_or(CLIENT_KEY_SECRET)
.to_string();
let door = t
.get("door")
.and_then(|v| v.as_str())
.map(str::trim)
.unwrap_or_default()
.to_string();
if door.is_empty() {
return Err("`door` is not set: the shop's monetize door".into());
}
let key = feed
.get(&format!("secret.{secret}"))
.ok_or_else(|| format!("the client key (the fed secret `{secret}`) is not fed"))?;
let mut d = ShopDoor::settings(&t, key, false)?;
// A7 rows 16 and 23 (Gunnar 7c3feb61, 8134ec6f): the fed seam only as a bench's.
let fed = |k: &str| feed.get(k).map(|v| v.trim()).filter(|v| !v.is_empty());
if let Some(why) = nordisk_pairing::fleet::seam_refusal(
fed(FEED_FLEET_AT),
fed(FEED_FLEET_CA).is_some(),
) {
return Err(why);
}
d.seam = seam_of(&d.door, feed);
// The pull channel's name for this estate, when fed (an operator site's label).
if let Some(instance) = fed(FEED_FLEET_INSTANCE).filter(|_| d.on_fleet) {
let product = d.fleet_product().unwrap_or_default();
d = d.over_fleet(&product, instance);
}
Ok(d)
})())
}
/// **The bench's seam, named directly** (`fleet_at`, `fleet_ca`), honoured only as
/// nordisk-pairing `fleet::seam_refusal` allows — a bench `ip:port`, its CA beside it;
/// anything else refused by name.
pub fn with_seam(mut self, at: Option<&str>, ca_pem: Option<&str>) -> Result<ShopDoor, String> {
let at = at.map(str::trim).filter(|a| !a.is_empty());
let ca = ca_pem.map(str::trim).filter(|c| !c.is_empty());
if let Some(why) = nordisk_pairing::fleet::seam_refusal(at, ca.is_some()) {
return Err(why);
}
self.seam = at.map(|a| Seam {
at: a.to_string(),
ca_pem: ca.map(str::to_string),
});
Ok(self)
}
/// Sign calls at `clock`'s time (a test's shop runs on its own clock).
pub fn with_clock(mut self, clock: Arc<dyn Fn() -> i64 + Send + Sync>) -> ShopDoor {
self.clock = Some(clock);
self
}
/// Ask an Upgrade as `polling` says (a test's shop polls faster than [`Polling::UPGRADE`]).
pub fn with_polling(mut self, polling: Polling) -> ShopDoor {
self.polling = polling;
self
}
/// The calls ride `wire` instead of `Fleet.Ask` (a test's shop).
pub fn with_wire(mut self, wire: Arc<dyn Wire>) -> ShopDoor {
self.wire = wire;
self.on_fleet = false;
self
}
/// **The calls ride `Fleet.Ask`** (plan §8), signed as `product`'s estate `instance`'s
/// front, by the estate's client key.
pub fn over_fleet(mut self, product: &str, instance: &str) -> ShopDoor {
self.wire = Arc::new(FleetWire {
product: product.trim().to_string(),
instance: instance.trim().to_string(),
client: self.client,
});
self.on_fleet = true;
self
}
/// **Over `Fleet.Ask` as `product`, unless the feed already named one** — what a product's
/// front calls with its own product name, so a front fed before the fold (no `product` in
/// its `shop_door.toml`) moves to Fleet too.
pub fn or_over_fleet(self, product: &str) -> ShopDoor {
if self.on_fleet {
return self;
}
let instance = self.company.clone();
self.over_fleet(product, &instance)
}
/// Whether the calls ride `Fleet.Ask`.
pub fn on_fleet(&self) -> bool {
self.on_fleet
}
/// The product the calls are signed as on `Fleet`, when they ride it.
pub fn fleet_product(&self) -> Option<String> {
self.on_fleet
.then(|| self.wire.fleet_as().map(|(p, _)| p))
.flatten()
}
/// The estate's name on `Fleet`, when the calls ride it.
pub fn fleet_instance(&self) -> Option<String> {
self.on_fleet
.then(|| self.wire.fleet_as().map(|(_, i)| i))
.flatten()
}
/// The shop's name as a person reads it: `ignalina.dk` of `https://monetize.ignalina.dk`.
pub fn shop_host(&self) -> String {
let host = host_of(&self.door);
host.strip_prefix("monetize.").unwrap_or(host).to_string()
}
/// Where every call goes.
pub fn dial(&self) -> Dial {
Dial {
door: self.door.clone(),
seam: self.seam.clone(),
}
}
/// The signed call, as monetize-embed's `EstateCall` canonical JSON.
pub fn signed(
&self,
call: &str,
members: &[String],
gib: u64,
carries: &str,
) -> pb::EstateCallSignature {
let at = match &self.clock {
Some(c) => c(),
None => std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |d| d.as_millis() as i64),
};
let c = EstateCall {
v: 1,
call: call.into(),
estate: self.company.clone(),
members: members.to_vec(),
gib,
at_unix_ms: at,
carries: carries.to_string(),
};
let signature = ed25519_dalek::SigningKey::from_bytes(&self.client)
.sign(&estate_call_message(&c))
.to_bytes()
.to_vec();
pb::EstateCallSignature {
at_unix_ms: at,
signature,
}
}
/// **`SoldEstate.Upgrade`**: boot this estate's standing `role` once from the release's keep
/// medium, or, with `done`, record that it runs the release.
///
/// **Started, then polled** (REL-T2-1, [`Polling`]): each ask waits at most `polling.ask`;
/// one that was cut ([`SHOP_CUT`]: the ONE box's edge closed the silent connection, prod
/// 2026-10-29) or answered "already running" ([`monetize_embed::fleet::UPGRADE_RUNNING`]) is
/// asked again, freshly signed, until the shop answers. The shop's keep boot runs on without
/// the front, and it records the release on the box, so its "already runs" after a cut is the
/// keep boot this front started: answered `upgraded`, for the front to re-admit the box. A
/// refusal by name and the shop's own failure come back at once, never asked again.
pub fn upgrade(&self, role: &str, done: bool) -> Result<pb::EstateUpgradeResponse, String> {
let p = self.polling;
let started = std::time::Instant::now();
let mut unanswered: Option<String> = None;
loop {
let req = pb::EstateUpgradeRequest {
company: self.company.clone(),
role: role.into(),
done,
signed: Some(self.signed(CALL_UPGRADE, &[role.to_string()], u64::from(done), "")),
};
match self.wire.upgrade(&self.dial(), req, p.ask) {
Ok(mut a) => {
if let Some(why) = unanswered.as_ref().filter(|_| !done && !a.upgraded) {
a.upgraded = true;
a.words = format!(
"the shop keep-booted the {role} while the front polled it ({why}); {}",
a.words
);
}
return Ok(a);
}
Err(e) if still_working(&e) => {
if started.elapsed() + p.every >= p.within {
return Err(format!(
"{e} — no answer from the shop within {} min",
p.within.as_secs().div_ceil(60)
));
}
unanswered = Some(e.trim_start_matches("unavailable: ").to_string());
std::thread::sleep(p.every);
}
Err(e) => return Err(e),
}
}
}
/// **`SoldEstate.Recharge`**: lay this estate's LOST `role` again with `seed` as its
/// user_data, or, with `done`, give its release clone back (the box is re-admitted).
pub fn recharge_box(
&self,
role: &str,
seed: &str,
done: bool,
) -> Result<pb::EstateRechargeResponse, String> {
let carries = if done { "" } else { seed };
let req = pb::EstateRechargeRequest {
company: self.company.clone(),
role: role.into(),
seed: carries.into(),
done,
signed: Some(self.signed(CALL_RECHARGE, &[role.to_string()], u64::from(done), carries)),
};
self.wire.recharge(&self.dial(), req, RECHARGE_WITHIN)
}
/// **`SoldEstate.TakeMaterial`**: the installer material the pairing left, or, with `done`,
/// that it is kept (the shop forgets it at the hand-off).
pub fn take_material(&self, done: bool) -> Result<pb::EstateTakeMaterialResponse, String> {
let req = pb::EstateTakeMaterialRequest {
company: self.company.clone(),
done,
signed: Some(self.signed(CALL_TAKE_MATERIAL, &[], u64::from(done), "")),
};
self.wire.take_material(&self.dial(), req, READ_WITHIN)
}
/// **`SoldEstate.UpdateTake`**: the update the operator opened, as the shop hands it out.
pub fn update_take(&self) -> Result<pb::EstateUpdateTakeResponse, String> {
let req = pb::EstateUpdateTakeRequest {
company: self.company.clone(),
signed: Some(self.signed(CALL_UPDATE_TAKE, &[], 0, "")),
};
self.wire.update_take(&self.dial(), req, READ_WITHIN)
}
/// **`SoldEstate.UpdateReport`**: `role`'s step, `outcome` and `words` signed together
/// (`carries` = `<outcome>\n<words>`).
pub fn update_report(
&self,
role: &str,
outcome: &str,
words: &str,
) -> Result<pb::EstateUpdateReportResponse, String> {
let carries = format!("{outcome}\n{words}");
let req = pb::EstateUpdateReportRequest {
company: self.company.clone(),
role: role.into(),
signed: Some(self.signed(CALL_UPDATE_REPORT, &[role.to_string()], 0, &carries)),
carries,
};
self.wire.update_report(&self.dial(), req, READ_WITHIN)
}
}
/// Pure: **an Upgrade ask the shop is still working on**: cut before its answer ([`SHOP_CUT`]),
/// or answered "already running" by the shop.
pub fn still_working(e: &str) -> bool {
e.starts_with("unavailable: ")
&& (e.contains(SHOP_CUT) || e.contains(monetize_embed::fleet::UPGRADE_RUNNING))
}
/// Pure: **the shop's live app as the update step takes it**.
pub fn live_app_of(l: pb::EstateLiveApp) -> LiveApp {
LiveApp {
base: l.base,
app: l.app,
app_len: l.app_len,
sig: l.sig,
url: l.url,
sha256: l.sha256,
release: l.release,
release_sig: l.release_sig,
env: l.env,
}
}
/// Pure: **the shop's take as the update step takes it**.
pub fn take_of(r: pb::EstateUpdateTakeResponse) -> Take {
Take {
open: r.open,
release: r.release,
live: r.live,
role: r.role,
app: r.app.map(live_app_of),
front: r.front.map(live_app_of),
words: r.words,
}
}
/// **The ONE box's update door** (`SoldEstate.UpdateTake` / `UpdateReport`) as the shared update
/// step (nordisk-pairing `frontside::update`) pulls it.
impl Shop for ShopDoor {
fn take(&self) -> Result<Take, String> {
self.update_take().map(take_of)
}
fn report(&self, role: &str, outcome: &str, words: &str) -> Result<String, String> {
self.update_report(role, outcome, words).map(|r| r.words)
}
}
/// Pure: the address a call to `door` is dialled at — `scheme://<at>` over the seam, the door
/// itself otherwise.
pub fn dialled(door: &str, seam: Option<&Seam>) -> String {
match seam {
Some(s) => format!(
"{}://{}",
door.split_once("://").map(|(sc, _)| sc).unwrap_or("https"),
s.at.trim()
),
None => door.to_string(),
}
}
/// ★ **H16: the wire of a door that knows no product** — the old `SoldEstate` door is gone, so
/// a front that does not say which product it is (its `shop_door.toml` `product`, or
/// [`ShopDoor::or_over_fleet`]) calls nothing: every call is refused by name.
#[derive(Clone, Copy, Debug, Default)]
pub struct NoProduct;
/// The refusal [`NoProduct`] answers every call with.
pub const NO_PRODUCT: &str = "refused: this front's shop door names no product, and the ONE box is called over Fleet only (the old SoldEstate door is gone): feed `product` in shop_door.toml, or call or_over_fleet";
impl Wire for NoProduct {
fn upgrade(
&self,
_: &Dial,
_: pb::EstateUpgradeRequest,
_: Duration,
) -> Result<pb::EstateUpgradeResponse, String> {
Err(NO_PRODUCT.into())
}
fn recharge(
&self,
_: &Dial,
_: pb::EstateRechargeRequest,
_: Duration,
) -> Result<pb::EstateRechargeResponse, String> {
Err(NO_PRODUCT.into())
}
fn take_material(
&self,
_: &Dial,
_: pb::EstateTakeMaterialRequest,
_: Duration,
) -> Result<pb::EstateTakeMaterialResponse, String> {
Err(NO_PRODUCT.into())
}
fn update_take(
&self,
_: &Dial,
_: pb::EstateUpdateTakeRequest,
_: Duration,
) -> Result<pb::EstateUpdateTakeResponse, String> {
Err(NO_PRODUCT.into())
}
fn update_report(
&self,
_: &Dial,
_: pb::EstateUpdateReportRequest,
_: Duration,
) -> Result<pb::EstateUpdateReportResponse, String> {
Err(NO_PRODUCT.into())
}
}
/// The ONE box's door, dialled as [`Dial`] says: TLS named as the door, under the seam's CA or
/// the public roots; plain http only for a loopback test's door.
async fn channel(d: Dial, within: Duration) -> Result<tonic::transport::Channel, String> {
let door = d.door.as_str();
let dial = dialled(door, d.seam.as_ref());
let origin: tonic::codegen::http::Uri = door
.parse()
.map_err(|e| format!("the shop's door {door:?}: {e}"))?;
let mut ep = tonic::transport::Endpoint::from_shared(dial.clone())
.map_err(|e| format!("the shop's door {door:?} at {dial}: {e}"))?
.origin(origin)
.connect_timeout(Duration::from_secs(10))
.timeout(within)
// REL-T2-1: a call waiting on the shop sends nothing else; the ONE box's edge closes a silent connection.
.http2_keep_alive_interval(KEEPALIVE_EVERY)
.keep_alive_timeout(KEEPALIVE_TIMEOUT)
.keep_alive_while_idle(true)
.tcp_keepalive(Some(KEEPALIVE_EVERY));
if door.starts_with("https://") {
let tls = match d.seam.as_ref().and_then(|s| s.ca_pem.as_deref()) {
Some(pem) => tonic::transport::ClientTlsConfig::new()
.ca_certificate(tonic::transport::Certificate::from_pem(pem)),
None => tonic::transport::ClientTlsConfig::new().with_webpki_roots(),
};
ep = ep
.tls_config(tls.domain_name(host_of(door).to_string()))
.map_err(|e| format!("the shop's door {door:?}: {e}"))?;
}
ep.connect().await.map_err(|e| {
format!("unavailable: {SHOP_CUT} the shop's door {door} (at {dial}) does not answer: {e}")
})
}
/// The shop's refusal by name; anything else is worth asking again.
pub fn from_status(s: tonic::Status) -> String {
use tonic::Code;
match s.code() {
Code::Unauthenticated
| Code::FailedPrecondition
| Code::InvalidArgument
| Code::PermissionDenied
| Code::NotFound => format!("refused: {}", s.message()),
_ => format!("unavailable: {}", s.message()),
}
}
/// A call to the shop on its own thread and runtime: safe from any caller (a jera job's
/// blocking thread, a runtime's worker).
pub fn blocking<T: Send + 'static>(
f: impl std::future::Future<Output = Result<T, String>> + Send + 'static,
) -> Result<T, String> {
// A32-14, stays a task: one shop call on a runtime of its own, joined at once (safe from a jera job's blocking thread).
std::thread::spawn(move || {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|e| format!("a runtime for the shop's door: {e}"))?
.block_on(f)
})
.join()
.unwrap_or_else(|_| Err("the call to the shop panicked".into()))
}
/// **The Fleet door** (plan §8, P3-FLEET): each call a `Fleet.Ask` of the front's kind
/// (monetize-embed `fleet::ask::FRONT`), body [`FrontAsk`], a `FleetCall` of `role = front`
/// signed by the estate's client key at the call's own signed time; the answer is the call's
/// message, decoded from `FleetAskAnswer.answer`. (H16: no fallback to the old door.)
#[derive(Clone)]
pub struct FleetWire {
pub product: String,
pub instance: String,
client: [u8; 32],
}
impl std::fmt::Debug for FleetWire {
/// Never the client key.
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("FleetWire")
.field("product", &self.product)
.field("instance", &self.instance)
.finish_non_exhaustive()
}
}
/// Why a Fleet ask did not come back: the door's refusal (its gRPC status), or no door.
enum Asked {
Status(tonic::Status),
Other(String),
}
impl FleetWire {
/// Pure: **the signed `Fleet.Ask`** of `kind` with `body`, at `at_ms`.
pub fn signed(&self, kind: &str, body: &FrontAsk, at_ms: u64) -> pb::FleetSigned {
let body = serde_json::to_string(&Ask {
kind: kind.into(),
body: serde_json::to_value(body).unwrap_or_default(),
})
.unwrap_or_default();
let c = FleetCall {
product: self.product.clone(),
instance: self.instance.clone(),
role: FRONT_ROLE.into(),
call: fleet_call::ASK.into(),
body,
at_unix_ms: at_ms,
};
let signature = ed25519_dalek::SigningKey::from_bytes(&self.client)
.sign(&fleet_call_message(&c))
.to_bytes()
.to_vec();
pb::FleetSigned {
product: c.product,
instance: c.instance,
role: c.role,
call: c.call,
body: c.body,
at_unix_ms: c.at_unix_ms,
signature,
box_key: Vec::new(),
}
}
fn ask<M: prost::Message + Default>(
&self,
d: &Dial,
kind: &str,
body: FrontAsk,
signed: Option<&pb::EstateCallSignature>,
within: Duration,
) -> Result<M, Asked> {
let at = signed.map_or(0, |s| s.at_unix_ms.max(0) as u64);
let req = self.signed(kind, &body, at);
let d = d.clone();
let answer = blocking(async move {
let mut c = pb::fleet_client::FleetClient::new(channel(d, within).await?);
Ok(c.ask(req).await.map(|r| r.into_inner().answer))
})
.map_err(Asked::Other)?
.map_err(Asked::Status)?;
M::decode(answer.as_slice()).map_err(|e| {
Asked::Other(format!(
"unavailable: the ONE box's answer to {kind} is not its message: {e}"
))
})
}
/// One call over Fleet.
fn call<M: prost::Message + Default>(
&self,
d: &Dial,
kind: &str,
body: FrontAsk,
signed: Option<&pb::EstateCallSignature>,
within: Duration,
) -> Result<M, String> {
match self.ask(d, kind, body, signed, within) {
Ok(m) => Ok(m),
// A status the shop never sent (the connection dropped, the ask's bound passed): cut.
Err(Asked::Status(s)) if std::error::Error::source(&s).is_some() => {
Err(format!("unavailable: {SHOP_CUT} {}", s.message()))
}
Err(Asked::Status(s)) => Err(from_status(s)),
Err(Asked::Other(e)) => Err(e),
}
}
}
impl Wire for FleetWire {
fn upgrade(
&self,
d: &Dial,
r: pb::EstateUpgradeRequest,
within: Duration,
) -> Result<pb::EstateUpgradeResponse, String> {
let body = FrontAsk {
role: r.role.clone(),
done: r.done,
..Default::default()
};
self.call(d, fleet_ask::UPGRADE, body, r.signed.as_ref(), within)
}
fn recharge(
&self,
d: &Dial,
r: pb::EstateRechargeRequest,
within: Duration,
) -> Result<pb::EstateRechargeResponse, String> {
let body = FrontAsk {
role: r.role.clone(),
done: r.done,
seed: r.seed.clone(),
..Default::default()
};
self.call(d, fleet_ask::RECHARGE, body, r.signed.as_ref(), within)
}
fn take_material(
&self,
d: &Dial,
r: pb::EstateTakeMaterialRequest,
within: Duration,
) -> Result<pb::EstateTakeMaterialResponse, String> {
let body = FrontAsk {
done: r.done,
..Default::default()
};
self.call(d, fleet_ask::MATERIAL, body, r.signed.as_ref(), within)
}
fn update_take(
&self,
d: &Dial,
r: pb::EstateUpdateTakeRequest,
within: Duration,
) -> Result<pb::EstateUpdateTakeResponse, String> {
self.call(
d,
fleet_ask::UPDATE_TAKE,
FrontAsk::default(),
r.signed.as_ref(),
within,
)
}
fn update_report(
&self,
d: &Dial,
r: pb::EstateUpdateReportRequest,
within: Duration,
) -> Result<pb::EstateUpdateReportResponse, String> {
let body = FrontAsk {
role: r.role.clone(),
carries: r.carries.clone(),
..Default::default()
};
self.call(d, fleet_ask::UPDATE_REPORT, body, r.signed.as_ref(), within)
}
fn fleet_as(&self) -> Option<(String, String)> {
Some((self.product.clone(), self.instance.clone()))
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Mutex;
fn feed(door: &str) -> BTreeMap<String, String> {
BTreeMap::from([
(FEED_SHOP_DOOR.to_string(), format!("door = \"{door}\"\ncompany = \"acme\"\nshop_pubkey = \"{}\"\nclient_key_secret = \"shop_client_key\"\n", hex_of(&ed25519_dalek::SigningKey::from_bytes(&[0x11; 32]).verifying_key().to_bytes()))),
("secret.shop_client_key".to_string(), "22".repeat(32)),
])
}
fn hex_of(b: &[u8]) -> String {
crate::hex(b)
}
/// ★ Row 16 (Gunnar 7c3feb61): the door is dialled over the fed seam when the fed `fleet`
/// is the same ONE box — at `fleet_at`, verified under `fleet_ca`, named as the door — and
/// at its own host otherwise.
#[test]
fn the_shop_door_dials_the_fed_seam_of_the_same_one_box() {
let mut f = feed("https://monetize-test.ignalina.dk");
assert!(
ShopDoor::from_feed(&f).unwrap().unwrap().seam.is_none(),
"no seam fed: the door's host"
);
f.insert(
FEED_FLEET.into(),
"https://monetize-test.ignalina.dk".into(),
);
f.insert(FEED_FLEET_AT.into(), "10.13.12.5:443".into());
f.insert(
FEED_FLEET_CA.into(),
"-----BEGIN CERTIFICATE-----\nX\n-----END CERTIFICATE-----\n".into(),
);
let d = ShopDoor::from_feed(&f).unwrap().unwrap();
let seam = d.seam.clone().expect("the seam of the same box");
assert_eq!(seam.at, "10.13.12.5:443");
assert!(seam.ca_pem.is_some());
assert_eq!(dialled(&d.door, d.seam.as_ref()), "https://10.13.12.5:443");
assert_eq!(dialled(&d.door, None), "https://monetize-test.ignalina.dk");
f.insert(FEED_FLEET.into(), "https://monetize.other.example".into());
assert!(
ShopDoor::from_feed(&f).unwrap().unwrap().seam.is_none(),
"another box's seam is never this door's"
);
}
/// ★ A7 rows 16 and 23 (Gunnar 7c3feb61, 8134ec6f): the fed seam is honoured only as a bench
/// feeds it (nordisk-pairing `fleet::seam_refusal`): a public `fleet_at`, a name, or a
/// `fleet_ca` with no `fleet_at` refuses the door by name; a bench `ip:port` is dialled. The
/// same rule holds for a seam named directly (`with_seam`, Gunnar's constructor).
#[test]
fn a_seam_that_is_not_a_benchs_refuses_the_door_by_name() {
let mut f = feed("https://monetize-test.ignalina.dk");
f.insert(
FEED_FLEET.into(),
"https://monetize-test.ignalina.dk".into(),
);
for at in ["81.27.107.63:443", "monetize-test.ignalina.dk:443"] {
f.insert(FEED_FLEET_AT.into(), at.into());
let e = ShopDoor::from_feed(&f).unwrap().unwrap_err();
assert!(e.contains("[fleet-seam-bench-only]"), "{at}: {e}");
}
f.remove(FEED_FLEET_AT);
f.insert(
FEED_FLEET_CA.into(),
"-----BEGIN CERTIFICATE-----\nX\n-----END CERTIFICATE-----\n".into(),
);
assert!(
ShopDoor::from_feed(&f)
.unwrap()
.unwrap_err()
.contains("[fleet-seam-bench-only]"),
"a CA with no bench address"
);
f.insert(FEED_FLEET_AT.into(), "10.0.2.2:35219".into());
let d = ShopDoor::from_feed(&f).unwrap().unwrap();
assert_eq!(
d.seam.clone().map(|s| s.at),
Some("10.0.2.2:35219".to_string())
);
assert!(d
.clone()
.with_seam(Some("81.27.107.63:443"), None)
.unwrap_err()
.contains("[fleet-seam-bench-only]"));
assert_eq!(
d.with_seam(Some(" 10.0.2.2:1 "), Some("PEM")).unwrap().seam,
Some(Seam {
at: "10.0.2.2:1".into(),
ca_pem: Some("PEM".into())
})
);
}
/// ★ A7 row 9 (Gunnar b74ca801): an Upgrade waits on the shop's whole keep boot, the ONE
/// constant the shop holds its call by (monetize `KEEP_BOOT_WITHIN`, 2 h).
#[test]
fn an_upgrade_waits_on_the_shops_whole_keep_boot() {
assert_eq!(UPGRADE_WITHIN, Duration::from_secs(2 * 3600));
assert_eq!(
(RECHARGE_WITHIN, READ_WITHIN),
(Duration::from_secs(3600), Duration::from_secs(30))
);
}
#[test]
fn the_fed_door_signs_its_calls_the_way_the_shop_verifies_them() {
let d = ShopDoor::from_feed(&feed("https://monetize.ignalina.dk/"))
.unwrap()
.unwrap()
.with_clock(Arc::new(|| 1_000));
assert_eq!(
(d.door.as_str(), d.company.as_str(), d.shop_host().as_str()),
("https://monetize.ignalina.dk", "acme", "ignalina.dk")
);
let s = d.signed(CALL_TAKE_MATERIAL, &[], 0, "");
let c = EstateCall {
v: 1,
call: CALL_TAKE_MATERIAL.into(),
estate: "acme".into(),
members: vec![],
gib: 0,
at_unix_ms: 1_000,
carries: String::new(),
};
let vk = ed25519_dalek::SigningKey::from_bytes(&[0x22; 32]).verifying_key();
monetize_embed::signing::verify_estate_call(&c, &s.signature, &vk, 1_000)
.expect("the shop verifies the call under the estate's client key");
assert!(
ShopDoor::from_feed(&BTreeMap::new()).is_none(),
"no shop door fed: none"
);
let mut no_key = feed("https://x");
no_key.remove("secret.shop_client_key");
assert!(ShopDoor::from_feed(&no_key)
.unwrap()
.unwrap_err()
.contains("not fed"));
assert!(
!format!("{d:?}").contains("22"),
"the client key is never printed"
);
}
/// ★ The front's feed is taken as the shop feeds it (monetize-shop-pairing `front_feed`): the
/// `shop_pubkey` beside the door is the boxes' monetize key, and the front verifies nothing
/// under it, so a value that is no curve point does not refuse the door (Holger's
/// `shop_pairing_contract` feeds `ab`×32). Gunnar's `from_settings` still checks it.
#[test]
fn the_fed_door_does_not_refuse_a_shop_key_it_never_verifies_under() {
let mut f = feed("https://monetize.ignalina.dk");
f.insert(FEED_SHOP_DOOR.into(), format!("door = \"https://monetize.ignalina.dk\"\ncompany = \"acme\"\nshop_pubkey = \"{}\"\n", "ab".repeat(32)));
assert_eq!(ShopDoor::from_feed(&f).unwrap().unwrap().company, "acme");
let t: toml::Table = toml::from_str(&f[FEED_SHOP_DOOR]).unwrap();
assert!(ShopDoor::from_settings(&t, &"22".repeat(32))
.unwrap_err()
.contains("`shop_pubkey` is not a key"));
}
/// ★ Gunnar's constructor (`from_settings`): a door that is no URL, no company, a shop key
/// that is no key and a client key that is not 64 hex are each refused by name.
#[test]
fn the_settings_are_checked_by_name() {
let t = |s: &str| -> toml::Table { toml::from_str(s).unwrap() };
let key = "22".repeat(32);
assert!(ShopDoor::from_settings(&t("company = \"acme\""), &key)
.unwrap_err()
.contains("`door` is not set"));
assert!(
ShopDoor::from_settings(&t("door = \"monetize.x\"\ncompany = \"acme\""), &key)
.unwrap_err()
.contains("is not an http(s) URL")
);
assert!(ShopDoor::from_settings(&t("door = \"https://m.x\""), &key)
.unwrap_err()
.contains("`company` is not set"));
assert!(ShopDoor::from_settings(
&t("door = \"https://m.x\"\ncompany = \"acme\"\nshop_pubkey = \"12\""),
&key
)
.unwrap_err()
.contains("`shop_pubkey` is not 64 hex"));
assert!(ShopDoor::from_settings(
&t("door = \"https://m.x\"\ncompany = \"acme\"\nclient_key_secret = \"k\""),
"12"
)
.unwrap_err()
.contains("the client key (the fed secret `k`) is not 64 hex"));
assert_eq!(
ShopDoor::from_settings(&t("door = \"https://m.x/\"\ncompany = \" acme \""), &key)
.unwrap()
.dial(),
Dial {
door: "https://m.x".into(),
seam: None
}
);
}
/// A shop on a wire of its own: every request it was handed, and its answers.
#[derive(Default)]
struct Heard(Mutex<Vec<(String, Duration, String)>>);
impl Wire for Heard {
fn upgrade(
&self,
d: &Dial,
r: pb::EstateUpgradeRequest,
within: Duration,
) -> Result<pb::EstateUpgradeResponse, String> {
self.0.lock().unwrap().push((
format!("upgrade {} done={}", r.role, r.done),
within,
d.door.clone(),
));
Ok(pb::EstateUpgradeResponse {
words: "booted".into(),
..Default::default()
})
}
fn recharge(
&self,
_: &Dial,
r: pb::EstateRechargeRequest,
within: Duration,
) -> Result<pb::EstateRechargeResponse, String> {
self.0.lock().unwrap().push((
format!("recharge {} seed={}", r.role, r.seed),
within,
String::new(),
));
Err("refused: acme is not delivered".into())
}
fn take_material(
&self,
_: &Dial,
r: pb::EstateTakeMaterialRequest,
within: Duration,
) -> Result<pb::EstateTakeMaterialResponse, String> {
self.0.lock().unwrap().push((
format!("take-material done={}", r.done),
within,
String::new(),
));
Ok(pb::EstateTakeMaterialResponse {
bundle: b"kept".to_vec(),
words: "handed".into(),
})
}
fn update_take(
&self,
_: &Dial,
_: pb::EstateUpdateTakeRequest,
within: Duration,
) -> Result<pb::EstateUpdateTakeResponse, String> {
self.0
.lock()
.unwrap()
.push(("update-take".into(), within, String::new()));
Ok(pb::EstateUpdateTakeResponse {
open: true,
release: "03".into(),
live: true,
role: "twin".into(),
app: Some(pb::EstateLiveApp {
base: "b".into(),
app: "a".into(),
app_len: 3,
env: vec!["korp.env.X=1".into()],
..Default::default()
}),
front: None,
words: "your turn".into(),
})
}
fn update_report(
&self,
_: &Dial,
r: pb::EstateUpdateReportRequest,
within: Duration,
) -> Result<pb::EstateUpdateReportResponse, String> {
self.0.lock().unwrap().push((
format!("update-report {} {:?}", r.role, r.carries),
within,
String::new(),
));
Ok(pb::EstateUpdateReportResponse {
words: "recorded".into(),
})
}
}
/// ★ **The transport is swappable** (the SoldEstate → Fleet merge, A28 H16): every call rides
/// the door's [`Wire`] with its own bound and the shop's words come back as they are; the
/// update step's [`Shop`] is the same two calls.
#[test]
fn every_call_rides_the_doors_wire_with_its_own_bound() {
let heard = Arc::new(Heard::default());
let d = ShopDoor::from_feed(&feed("https://monetize.ignalina.dk"))
.unwrap()
.unwrap()
.with_wire(heard.clone());
assert_eq!(d.upgrade("twin", false).unwrap().words, "booted");
assert_eq!(
d.recharge_box("main", "SEED", false).unwrap_err(),
"refused: acme is not delivered"
);
assert_eq!(d.take_material(true).unwrap().bundle, b"kept");
let t = Shop::take(&d).unwrap();
assert_eq!(
(t.open, t.role.as_str(), t.words.as_str()),
(true, "twin", "your turn")
);
assert_eq!(t.app.unwrap().env, vec!["korp.env.X=1".to_string()]);
assert_eq!(
Shop::report(&d, "twin", "done:live", "the twin runs app a").unwrap(),
"recorded"
);
let h = heard.0.lock().unwrap();
assert_eq!(
h.iter()
.map(|(c, w, _)| (c.as_str(), *w))
.collect::<Vec<_>>(),
vec![
("upgrade twin done=false", Polling::UPGRADE.ask),
("recharge main seed=SEED", RECHARGE_WITHIN),
("take-material done=true", READ_WITHIN),
("update-take", READ_WITHIN),
(
"update-report twin \"done:live\\nthe twin runs app a\"",
READ_WITHIN
),
]
);
assert_eq!(h[0].2, "https://monetize.ignalina.dk");
}
/// ★ P3-FLEET (plan §8): a door fed its `product` rides `Fleet.Ask` as the fed
/// `fleet_instance`'s front; a door fed before the fold (no `product`) calls nothing until its
/// product's front says which product it is ([`ShopDoor::or_over_fleet`]) — H16: the old door
/// is gone, so it is refused by name; a test's wire is never Fleet.
#[test]
fn a_door_fed_its_product_rides_fleet_and_an_older_feed_is_moved_by_its_product() {
let mut f = feed("https://monetize.ignalina.dk");
let old = ShopDoor::from_feed(&f).unwrap().unwrap();
assert!(
!old.on_fleet() && old.fleet_product().is_none(),
"no product fed"
);
assert_eq!(
old.update_take().unwrap_err(),
NO_PRODUCT,
"H16: no old door to fall back to"
);
let moved = old.clone().or_over_fleet("holger");
assert_eq!(
(
moved.on_fleet(),
moved.fleet_product().as_deref(),
moved.fleet_instance().as_deref()
),
(true, Some("holger"), Some("acme"))
);
f.insert(
FEED_SHOP_DOOR.into(),
format!("{}product = \"gunnar\"\n", f[FEED_SHOP_DOOR]),
);
let fed = ShopDoor::from_feed(&f).unwrap().unwrap();
assert_eq!(
(
fed.fleet_product().as_deref(),
fed.fleet_instance().as_deref()
),
(Some("gunnar"), Some("acme"))
);
assert_eq!(
fed.clone()
.or_over_fleet("holger")
.fleet_product()
.as_deref(),
Some("gunnar"),
"the fed product wins"
);
f.insert(FEED_FLEET_INSTANCE.into(), "gunnar-rs".into());
assert_eq!(
ShopDoor::from_feed(&f)
.unwrap()
.unwrap()
.fleet_instance()
.as_deref(),
Some("gunnar-rs"),
"an operator site's label"
);
assert!(!fed.with_wire(Arc::new(Heard::default())).on_fleet());
let t: toml::Table =
toml::from_str("door = \"https://m.x\"\ncompany = \"acme\"\nproduct = \"gunnar\"")
.unwrap();
assert!(
ShopDoor::from_settings(&t, &"22".repeat(32))
.unwrap()
.on_fleet(),
"Gunnar's constructor reads it too"
);
}
/// ★ The front's ask is a `FleetCall` of `role = front` the ONE box verifies under the
/// estate's client key (the same key as the old door's `EstateCall`), the body the old
/// request's fields; the key is never printed.
#[test]
fn the_fleet_ask_is_signed_as_the_front_by_the_estates_client_key() {
let d = ShopDoor::from_feed(&feed("https://monetize.ignalina.dk"))
.unwrap()
.unwrap();
let w = FleetWire {
product: "gunnar".into(),
instance: "acme".into(),
client: [0x22; 32],
};
let s = w.signed(
fleet_ask::RECHARGE,
&FrontAsk {
role: "main".into(),
seed: "SEED".into(),
..Default::default()
},
1_000,
);
assert_eq!(
(
s.product.as_str(),
s.instance.as_str(),
s.role.as_str(),
s.call.as_str(),
s.at_unix_ms
),
("gunnar", "acme", "front", "ask", 1_000)
);
assert!(s.box_key.is_empty(), "a front pins nothing");
let c = FleetCall {
product: s.product.clone(),
instance: s.instance.clone(),
role: s.role.clone(),
call: s.call.clone(),
body: s.body.clone(),
at_unix_ms: s.at_unix_ms,
};
let vk = ed25519_dalek::SigningKey::from_bytes(&[0x22; 32]).verifying_key();
monetize_embed::fleet::verify_fleet_call(&c, &s.signature, &vk, 1_000)
.expect("the ONE box verifies the ask under the estate's client key");
let a: Ask = serde_json::from_str(&s.body).unwrap();
assert_eq!(a.kind, "recharge");
assert_eq!(
serde_json::from_value::<FrontAsk>(a.body).unwrap(),
FrontAsk {
role: "main".into(),
seed: "SEED".into(),
..Default::default()
}
);
assert!(
!format!("{w:?}").contains("22") && !format!("{d:?}").contains("2222"),
"the client key is never printed"
);
}
/// A shop whose Upgrade answers are scripted, one per ask, in order.
struct Scripted(
Mutex<Vec<Result<pb::EstateUpgradeResponse, String>>>,
Mutex<Vec<Duration>>,
);
impl Scripted {
fn new(answers: Vec<Result<pb::EstateUpgradeResponse, String>>) -> Arc<Scripted> {
Arc::new(Scripted(Mutex::new(answers), Mutex::default()))
}
fn asks(&self) -> Vec<Duration> {
self.1.lock().unwrap().clone()
}
}
impl Wire for Scripted {
fn upgrade(
&self,
_: &Dial,
_: pb::EstateUpgradeRequest,
within: Duration,
) -> Result<pb::EstateUpgradeResponse, String> {
self.1.lock().unwrap().push(within);
let mut a = self.0.lock().unwrap();
assert!(!a.is_empty(), "asked once too often");
a.remove(0)
}
fn recharge(
&self,
_: &Dial,
_: pb::EstateRechargeRequest,
_: Duration,
) -> Result<pb::EstateRechargeResponse, String> {
unreachable!()
}
fn take_material(
&self,
_: &Dial,
_: pb::EstateTakeMaterialRequest,
_: Duration,
) -> Result<pb::EstateTakeMaterialResponse, String> {
unreachable!()
}
fn update_take(
&self,
_: &Dial,
_: pb::EstateUpdateTakeRequest,
_: Duration,
) -> Result<pb::EstateUpdateTakeResponse, String> {
unreachable!()
}
fn update_report(
&self,
_: &Dial,
_: pb::EstateUpdateReportRequest,
_: Duration,
) -> Result<pb::EstateUpdateReportResponse, String> {
unreachable!()
}
}
const FAST: Polling = Polling {
ask: Duration::from_secs(60),
every: Duration::from_millis(1),
within: Duration::from_secs(30),
};
/// ★ **REL-T2-1 (prod 2026-10-29, test2.gunnar.rs and holger.rs)**: the ONE box's edge cut the
/// front's Upgrade after ~4 min ("unavailable: transport error") while the shop's keep boot ran
/// on to the end, and the update was recorded failed. The Upgrade is started, then polled: a
/// cut ask and an "already running" answer are asked again, and the shop's "already runs"
/// after a cut is the keep boot this front started — `upgraded`, so the front re-admits it.
#[test]
fn an_upgrade_cut_mid_call_is_polled_to_the_shops_answer_never_recorded_failed() {
let running = |host: &str| {
Err(format!(
"unavailable: an upgrade of {host} {}",
monetize_embed::fleet::UPGRADE_RUNNING
))
};
let shop = Scripted::new(vec![
Err(format!("unavailable: {SHOP_CUT} transport error")),
running("acme.example"),
Err(format!(
"unavailable: {SHOP_CUT} the shop's door https://m.x does not answer"
)),
running("acme.example"),
Ok(pb::EstateUpgradeResponse {
upgraded: false,
words: "acme.example's twin already runs release 03: nothing to upgrade".into(),
..Default::default()
}),
]);
let d = ShopDoor::from_feed(&feed("https://monetize.ignalina.dk"))
.unwrap()
.unwrap()
.with_wire(shop.clone())
.with_polling(FAST);
let up = d
.upgrade("twin", false)
.expect("a cut ask is never the update's failure");
assert!(
up.upgraded,
"the keep boot the cut ask started is re-admitted by the front: {up:?}"
);
assert!(up.words.contains("already runs release 03"), "{}", up.words);
assert_eq!(
shop.asks(),
vec![FAST.ask; 5],
"every ask bounded by the poll's own bound, never the whole keep boot"
);
// Asked at once, not cut: the shop's own "already runs" is nothing to re-admit.
let shop = Scripted::new(vec![Ok(pb::EstateUpgradeResponse {
upgraded: false,
words: "already runs".into(),
..Default::default()
})]);
let d = d.with_wire(shop.clone());
assert!(!d.upgrade("twin", false).unwrap().upgraded);
// A refusal by name, and the shop's own failure (its keep boot failed: CODE RED on its
// side), are answered at once: never asked again, no second keep boot.
for said in [
"refused: no update is open on the twin",
"unavailable: the provider: the medium left in",
] {
let shop = Scripted::new(vec![
Err(format!("unavailable: {SHOP_CUT} transport error")),
Err(said.to_string()),
]);
assert_eq!(
d.clone()
.with_wire(shop.clone())
.upgrade("twin", false)
.unwrap_err(),
said
);
assert_eq!(shop.asks().len(), 2);
}
// A shop that never answers within the poll's bound: the cut is said, by name.
let shop = Scripted::new(
(0..1000)
.map(|_| Err(format!("unavailable: {SHOP_CUT} transport error")))
.collect(),
);
let e = d
.with_wire(shop)
.with_polling(Polling {
within: Duration::from_millis(30),
..FAST
})
.upgrade("twin", false)
.unwrap_err();
assert!(
e.starts_with("unavailable: ")
&& e.contains("transport error")
&& e.contains("no answer from the shop within"),
"{e}"
);
}
/// ★ REL-T2-1: the front's own poll keeps each ask under the ONE box edge's 120 s idle cut
/// (edda `estate-front::web::IDLE_TIMEOUT`), and every call's connection pings the edge well
/// inside it (a Recharge waits up to an hour on one call).
#[test]
fn every_ask_stays_under_the_edges_idle_cut() {
let edge_idle = Duration::from_secs(120);
assert!(Polling::UPGRADE.ask < edge_idle && Polling::UPGRADE.within == UPGRADE_WITHIN);
assert!(KEEPALIVE_EVERY < edge_idle / 2 && KEEPALIVE_TIMEOUT < KEEPALIVE_EVERY);
}
#[test]
fn a_refusal_is_said_by_name_and_anything_else_is_worth_asking_again() {
assert_eq!(
from_status(tonic::Status::failed_precondition("[update-out-of-turn] x")),
"refused: [update-out-of-turn] x"
);
assert_eq!(
from_status(tonic::Status::unauthenticated("bad sig")),
"refused: bad sig"
);
assert_eq!(
from_status(tonic::Status::unavailable("down")),
"unavailable: down"
);
assert_eq!(blocking(async { Ok::<_, String>(7) }), Ok(7));
}
}