crates/monetize-guardian/src/pulled.rs
//! **A product over the pull channel** (5.4, G1): the [`Product`] the server loops talk to
//! when the product's estate CALLS monetize instead of being dialled.
//!
//! Monetize never calls into an estate. Every verb that used to dial the box — the tenant
//! list, usage, the paid door, servable, `can_absorb`, the entitlement push — is answered
//! here from the newest `Fleet.Report` of the estate's main ([`ProductReport`]) or queued as
//! a SIGNED order the main pulls. Verbs that need no box (meters, the guardian lane) are the
//! plugin's own; a ticket is the host's (`monetize_server::site_ticket`). Verbs with no order kind yet refuse by name; none dials.
use std::sync::{Arc, OnceLock};
use monetize_product::guardian::role;
use monetize_embed::fleet::SignedOrder;
use monetize_product::pull::{order, AddPrincipal, AddUsers, GrowRoom, ProductReport, PurgeAnswer, PurgeTenant};
use monetize_product::{
Absorb, Activity, EntitlementFact, MeteredDoor, Product, ProductError, Purged, Servable, TenantId, TenantOwner, TenantUsage, Usage, UsagePage, UsagePageReq, UsersAnswer, UsersAsk,
};
use crate::runner::Guardian;
/// The guardian a [`Pulled`] product reads reports from and queues orders on. Set once the
/// server has opened its store; built before it, so the registry can hand it out early.
pub type GuardianSlot = Arc<OnceLock<Arc<Guardian>>>;
/// Wall clock in Unix ms, injectable for tests.
pub type Clock = Arc<dyn Fn() -> u64 + Send + Sync>;
/// **The role whose box serves the product's tenants** (monetize v2.0 P1): asked each time,
/// because a product guest's lifecycle (its first data role) is taken while the server runs.
/// The default is `main` (Gunnar's and Holger's); the Lake's is its own (`be` today, `fe` once
/// P3's `facts.primary_role` says so).
pub type Primary = Arc<dyn Fn() -> String + Send + Sync>;
/// `main`, the primary of a product that names none.
pub fn main_primary() -> Primary {
Arc::new(|| role::MAIN.to_string())
}
/// How long an entitlement order waits for its box: an estate offline for a week still gets
/// the newest fact when it calls again.
pub const ENTITLEMENT_TTL_MS: u64 = 7 * 24 * 3_600_000;
/// The wall clock, in Unix ms.
pub fn system_clock() -> Clock {
Arc::new(|| std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0))
}
/// How long [`Pulled::add_users`] waits for the main's ack of a grant it just queued: the
/// main long-polls its orders, runs one at once and reports straight after, so a live estate
/// answers within a few seconds; past this the grant stays queued and is answered on the
/// next ask (the paid path asks again on every read).
pub const USERS_ACK_WAIT: std::time::Duration = std::time::Duration::from_secs(20);
/// How long a `grow_room` order waits for its box.
pub const GROW_ROOM_TTL_MS: u64 = 30 * 24 * 3_600_000;
/// What became of a signup's `add_principal` order ([`Lane::add_principal`]).
#[derive(Clone, PartialEq, Eq, Debug)]
pub enum Principal {
/// Queued (or queued again); the main has not acked it yet.
Queued,
/// The main made it; the namespace it made.
Created(String),
/// The name is somebody else's; the main's words.
Taken(String),
/// Refused, final; the main's words.
Refused(String),
}
/// **One product's estate over the pull channel**: its main's reports read, its orders
/// queued. What [`Pulled`] answers the server loops from, and what the server's pulled signup
/// sink creates accounts with.
#[derive(Clone)]
pub struct Lane {
product: String,
instance: String,
slot: GuardianSlot,
clock: Clock,
primary: Primary,
}
impl Lane {
pub fn new(product: impl Into<String>, instance: impl Into<String>, slot: GuardianSlot, clock: Clock) -> Self {
Self { product: product.into(), instance: instance.into(), slot, clock, primary: main_primary() }
}
/// This lane reading and ordering the box of the role `primary` answers ([`Primary`]).
pub fn with_primary(mut self, primary: Primary) -> Self {
self.primary = primary;
self
}
/// The role whose box serves the tenants now (`main` unless the product says otherwise).
pub fn primary(&self) -> String {
(self.primary)()
}
pub fn instance(&self) -> &str {
&self.instance
}
/// **This lane on `estate`**: another estate of the same product (ONE monetize, several
/// sites of one product) — its main's reports read, its orders queued. Unchecked: ask
/// [`Self::knows`] first.
pub fn on(&self, estate: &str) -> Lane {
Lane { instance: estate.to_owned(), ..self.clone() }
}
/// Whether the fleet knows `estate` as an estate of this lane's product; `Err` while the
/// guardian's store is not open (nothing can be said yet).
pub fn knows(&self, estate: &str) -> Result<bool, String> {
Ok(self.guardian()?.store.product_of(estate).as_deref() == Some(self.product.as_str()))
}
/// The lane's clock, Unix ms.
pub fn now(&self) -> u64 {
(self.clock)()
}
fn guardian(&self) -> Result<&Arc<Guardian>, String> {
self.slot.get().ok_or_else(|| format!("{}: the guardian store is not open yet, so no report of the main is read", self.instance))
}
/// The main's newest product report and when monetize heard it.
pub fn report(&self) -> Result<(ProductReport, u64), String> {
self.report_of(&self.instance)
}
/// The newest product report of `instance`'s main (the product's own estate, or one of its
/// [`Self::sold_estates`]) and when monetize heard it.
pub fn report_of(&self, instance: &str) -> Result<(ProductReport, u64), String> {
let g = self.guardian()?;
let primary = self.primary();
let rec = g.store.box_record(instance, &primary).filter(|r| r.heard_unix_ms > 0).ok_or_else(|| format!("{instance}: its {primary} has sent no report yet; monetize never dials it"))?;
let text = rec.report.product.as_deref().ok_or_else(|| format!("{instance}: its {primary}'s report carries no product part"))?;
let p = serde_json::from_str(text).map_err(|err| format!("{instance}: its {primary}'s product report does not parse: {err}"))?;
Ok((p, rec.heard_unix_ms))
}
/// The report, refused when the main has been silent past the guardian's bound: a
/// judgement that buys, serves or frees a name is never made on an old word.
pub fn fresh(&self) -> Result<ProductReport, String> {
self.fresh_of(&self.instance)
}
/// [`Self::fresh`] for `instance`'s main.
pub fn fresh_of(&self, instance: &str) -> Result<ProductReport, String> {
let (p, heard) = self.report_of(instance)?;
let silent_after = self.guardian()?.cfg.silent_after_ms;
let age = self.now().saturating_sub(heard);
if age > silent_after {
return Err(format!("{instance}: its {} has been silent {} s (bound {} s); nothing is judged on an old report", self.primary(), age / 1000, silent_after / 1000));
}
Ok(p)
}
/// **The product's other estates** (G1 multi-estate, ONE monetize): every estate the fleet
/// declares for this product beside its own — the sold ones (M-D32), whose tenants monetize
/// keys `@<estate>/<ns>`. Empty before the guardian's store is open.
pub fn sold_estates(&self) -> Vec<String> {
let Ok(g) = self.guardian() else { return Vec::new() };
g.store.estates().into_iter().filter(|(instance, product)| *product == self.product && *instance != self.instance).map(|(instance, _)| instance).collect()
}
/// The estate whose main serves `tenant`, and the tenant's name there: the product's own
/// estate for a plain tenant; for `@<estate>/<ns>`, that estate — refused by name when the
/// fleet does not know it for this product.
pub fn serving(&self, tenant: &TenantId) -> Result<(String, String), String> {
match tenant.estate() {
None => Ok((self.instance.clone(), tenant.0.clone())),
Some((estate, ns)) => {
let g = self.guardian()?;
if g.store.product_of(estate).as_deref() != Some(self.product.as_str()) {
return Err(format!("{estate}: no estate of {} on the fleet, so {} has no main that serves it", self.product, tenant.0));
}
Ok((estate.to_owned(), ns.to_owned()))
}
}
}
/// Queue `fact` for the main that serves its tenant; an unacked older fact for the same
/// tenant is dropped. A sold estate's tenant (`@<estate>/<ns>`, G16 ONE monetize) is that
/// ESTATE's main's to pull, as its gunnar names it ([`monetize::sign::Signer::on_estate`]);
/// an estate the fleet does not know for this product is refused by name.
pub fn queue_entitlement(&self, fact: &EntitlementFact) -> Result<(), String> {
self.queue_entitlement_marked(fact, None)
}
/// [`Self::queue_entitlement`], remembered under `mark` when one is given (RA5-16: the
/// resend of one version of a fact, [`Self::resend_unheld`]).
fn queue_entitlement_marked(&self, fact: &EntitlementFact, mark: Option<&str>) -> Result<(), String> {
let g = self.guardian()?;
let primary = self.primary();
let (instance, fact) = match fact.tenant.estate() {
None => (self.instance.as_str(), fact.clone()),
Some((estate, _)) => {
if g.store.product_of(estate).as_deref() != Some(self.product.as_str()) {
return Err(format!("{estate}: no estate of {} on the fleet, so {}'s entitlement has no main to pull it", self.product, fact.tenant.0));
}
(estate, g.keys.product(&self.product).on_estate(fact, estate).ok_or_else(|| format!("{}: not a tenant of {estate}", fact.tenant.0))?)
}
};
let body = serde_json::to_string(&fact).map_err(|err| format!("the entitlement fact does not serialise: {err}"))?;
let tenant = fact.tenant.clone();
let stale = |o: &SignedOrder| o.kind == order::ENTITLEMENT && o.role == primary && serde_json::from_str::<EntitlementFact>(&o.body).is_ok_and(|f| f.tenant == tenant);
let order = Guardian::order(&self.product, instance, &primary, order::ENTITLEMENT, body, self.now(), ENTITLEMENT_TTL_MS);
match mark {
None => g.queue_order(order, stale).map(|_| ()),
Some(mark) => g.queue_order_marked(order, mark).map(|_| ()),
}
.map_err(|err| err.to_string())
}
/// ★ ALICA-4 (live 2026-10-26): **a paid fact the main does not hold is queued again.** An
/// `entitlement` order is fire and forget: a refused ack removes it from the queue
/// ([`crate::store::Store::record_report`]) and nothing ever sends it again, and the
/// guardian restates a box only inside a recharge case. Live: alica paid 10 GB on
/// test2.gunnar.rs, monetize's ledger held the paid fact, and test2's gunnar said Free.
///
/// For each fact that is `Paid` or `Grace` (money that buys room now): the main that serves
/// its tenant ([`Self::serving`]) is read on a FRESH report ([`Self::fresh_of`]; nothing is
/// judged on an old word, and a silent estate is left for later). It holds the fact when
/// every cap of the fact is what the report says it promises that tenant (a cap of 0
/// matches a tenant it promises nothing). A main that reports no promises at all says
/// nothing, and nothing is queued for it. A fact not held and with no entitlement order
/// for that tenant still queued is queued again ([`Self::queue_entitlement`]), ONCE per
/// version of the fact ([`resend_mark`], RA5-16). Answers one line per fact queued.
pub fn resend_unheld(&self, facts: &[EntitlementFact]) -> Vec<String> {
let Ok(g) = self.guardian() else { return Vec::new() };
let primary = self.primary();
let mut said = Vec::new();
for fact in facts.iter().filter(|f| matches!(f.state, monetize_product::State::Paid | monetize_product::State::Grace)) {
let Ok((instance, ns)) = self.serving(&fact.tenant) else { continue };
let Ok(report) = self.fresh_of(&instance) else { continue };
if report.promised.is_empty() {
continue;
}
let promised = report.promised.get(&ns);
let held = fact.caps.iter().all(|(meter, cap)| promised.and_then(|p| p.get(meter)).copied().unwrap_or(0) == *cap);
if held {
continue;
}
let waiting = g.store.queued(&instance).iter().any(|o| {
o.kind == order::ENTITLEMENT && o.role == primary && serde_json::from_str::<EntitlementFact>(&o.body).is_ok_and(|f| f.tenant.0 == ns)
});
if waiting {
continue;
}
// RA5-16: ONE resend per version of the fact. A main that refuses it again (a tenant
// deleted on the site, a promise that differs from the cap) is not sent it on every
// tick with an ALERT each time; a new version (a renewal, a grow) is sent once more.
let mark = resend_mark(&instance, &ns, fact);
if g.store.mark(&mark).is_some() {
continue;
}
match self.queue_entitlement_marked(fact, Some(&mark)) {
Ok(()) => said.push(format!("{}: {instance}'s {primary} does not hold its paid fact ({}); queued again", fact.tenant.0, fact.plan)),
Err(why) => said.push(format!("{}: its paid fact is not held by {instance}'s {primary} and could not be queued again: {why}", fact.tenant.0)),
}
}
said
}
/// **The signup's account, made by the main**: queued once per signup (the order's seq is
/// remembered under it), then answered from the main's ack ([`order::ADD_PRINCIPAL`]).
/// An order neither queued nor acked any more (its ack scrolled out of the kept window)
/// is queued again; the main answers a signup it already served with the same ack.
pub fn add_principal(&self, ap: &AddPrincipal) -> Result<Principal, String> {
let g = self.guardian()?;
let primary = self.primary();
let mark = format!("principal/{}/{}", self.instance, ap.signup);
if let Some(seq) = g.store.mark(&mark) {
if g.store.queued(&self.instance).iter().any(|o| o.seq == seq) {
return Ok(Principal::Queued);
}
if let Some(a) = g.store.box_record(&self.instance, &primary).and_then(|r| r.acks.into_iter().rev().find(|a| a.seq == seq)) {
let detail = a.detail.trim().to_owned();
return Ok(match (a.ok, detail.starts_with(order::TAKEN)) {
(true, _) if detail.is_empty() => Principal::Created(ap.username.clone()),
(true, _) => Principal::Created(detail),
(false, true) => Principal::Taken(detail),
(false, false) => Principal::Refused(detail),
});
}
}
let body = serde_json::to_string(ap).map_err(|err| format!("the account order does not serialise: {err}"))?;
g.queue_order_marked(Guardian::order(&self.product, &self.instance, &primary, order::ADD_PRINCIPAL, body, self.now(), ENTITLEMENT_TTL_MS), &mark).map_err(|err| err.to_string())?;
Ok(Principal::Queued)
}
/// ★ UW-5: **"Add users", granted, as an order the serving main pulls** — queued once per
/// `order` reference (remembered under it), then answered from the main's ack: `Some(Ok)`
/// with its `UsersAnswer`, `Some(Err)` refused with its words, `None` while it waits. An
/// order neither queued nor acked any more is queued again (the box answers a repeat of
/// the same `order` with its first grant). No address leaves monetize.
pub fn add_users(&self, tenant: &TenantId, ask: &monetize_product::UsersAsk, reference: &str) -> Result<Option<Result<UsersAnswer, String>>, String> {
let (instance, ns) = self.serving(tenant)?;
let g = self.guardian()?;
let primary = self.primary();
let mark = format!("users/{instance}/{reference}");
if let Some(seq) = g.store.mark(&mark) {
if g.store.queued(&instance).iter().any(|o| o.seq == seq) {
return Ok(None);
}
if let Some(a) = g.store.box_record(&instance, &primary).and_then(|r| r.acks.into_iter().rev().find(|a| a.seq == seq)) {
if !a.ok {
return Ok(Some(Err(a.detail.trim().to_owned())));
}
return serde_json::from_str::<UsersAnswer>(&a.detail).map(|u| Some(Ok(u))).map_err(|e| format!("{instance}: the main granted {reference} and its answer does not parse: {e}"));
}
}
let ask = monetize_product::UsersAsk { email: String::new(), invite_origin: String::new(), ..ask.clone() };
let body = serde_json::to_string(&AddUsers { tenant: ns, ask, order: reference.to_owned() }).map_err(|e| format!("the users order does not serialise: {e}"))?;
g.queue_order_marked(Guardian::order(&self.product, &instance, &primary, order::ADD_USERS, body, self.now(), ENTITLEMENT_TTL_MS), &mark).map_err(|err| err.to_string())?;
Ok(None)
}
/// Whether the grant under `reference` is queued and not yet acked (asked before).
pub fn users_queued(&self, tenant: &TenantId, reference: &str) -> bool {
let (Ok((instance, _)), Ok(g)) = (self.serving(tenant), self.guardian()) else { return false };
g.store.mark(&format!("users/{instance}/{reference}")).is_some_and(|seq| g.store.queued(&instance).iter().any(|o| o.seq == seq))
}
/// ★ UW-4: **room sold past the set, queued for the serving estate to grow** — ONE
/// `grow_room` order to its main, the newest want replacing an unacked older one.
pub fn queue_room(&self, tenant: &TenantId, need_bytes: u64, why: &str) -> Result<(), String> {
let (instance, _) = self.serving(tenant)?;
let g = self.guardian()?;
let primary = self.primary();
let body = serde_json::to_string(&GrowRoom { need_bytes, reason: why.to_owned() }).map_err(|e| format!("the grow_room order does not serialise: {e}"))?;
g.queue_order(Guardian::order(&self.product, &instance, &primary, order::GROW_ROOM, body, self.now(), GROW_ROOM_TTL_MS), |o| o.kind == order::GROW_ROOM && o.role == primary).map_err(|err| err.to_string())?;
Ok(())
}
/// ★ UW-11: **the nonpay purge of `tenant`, as orders the serving main pulls** — a start,
/// then, while its job runs, ONE status order naming the job per ask. Answered from the
/// newest ack: `Ok(None)` while an order waits (the account stays locked), `Ok(Some)` the
/// main's word. A refusal, or a FAILED job, is answered once and a fresh start queued, so
/// the next ask starts again. Remembered per reason (one lapse is one purge).
pub fn purge(&self, tenant: &TenantId, reason: &str) -> Result<Option<Result<PurgeAnswer, String>>, String> {
let (instance, ns) = self.serving(tenant)?;
let g = self.guardian()?;
let primary = self.primary();
let mark = format!("purge/{instance}/{ns}/{:016x}", fnv1a(reason));
let queue = |job: &str| -> Result<(), String> {
let body = serde_json::to_string(&PurgeTenant { tenant: ns.clone(), reason: reason.to_owned(), job: job.to_owned() }).map_err(|e| format!("the purge order does not serialise: {e}"))?;
g.queue_order_marked(Guardian::order(&self.product, &instance, &primary, order::PURGE_TENANT, body, self.now(), ENTITLEMENT_TTL_MS), &mark).map(|_| ()).map_err(|err| err.to_string())
};
let Some(seq) = g.store.mark(&mark) else {
queue("")?;
return Ok(None);
};
if g.store.queued(&instance).iter().any(|o| o.seq == seq) {
return Ok(None);
}
let Some(a) = g.store.box_record(&instance, &primary).and_then(|r| r.acks.into_iter().rev().find(|a| a.seq == seq)) else {
// Neither queued nor acked any more: asked again (the main adopts a running job).
queue("")?;
return Ok(None);
};
if !a.ok {
queue("")?;
return Ok(Some(Err(a.detail.trim().to_owned())));
}
let answer: PurgeAnswer = serde_json::from_str(&a.detail).map_err(|e| format!("{instance}: the main's purge answer for {} does not parse: {e}", tenant.0))?;
if answer.running {
queue(&answer.job)?;
} else if answer.failed {
queue("")?;
}
Ok(Some(Ok(answer)))
}
/// Whether `name` is held on the main (its fresh report's `names`) or asked for by a
/// signup's queued `add_principal`. Never asks the box; an old or missing report is `Err`.
pub fn name_taken(&self, name: &str) -> Result<bool, String> {
let p = self.fresh()?;
if p.names.iter().any(|n| n == name) {
return Ok(true);
}
let g = self.guardian()?;
Ok(g.store.queued(&self.instance).iter().any(|o| o.kind == order::ADD_PRINCIPAL && serde_json::from_str::<AddPrincipal>(&o.body).is_ok_and(|a| a.username == name)))
}
}
/// FNV-1a, 64 bits: a stable short name for a purge's reason in its mark.
fn fnv1a(text: &str) -> u64 {
text.bytes().fold(0xcbf2_9ce4_8422_2325, |h, b| (h ^ u64::from(b)).wrapping_mul(0x0100_0000_01b3))
}
pub struct Pulled {
inner: Arc<dyn Product>,
lane: Lane,
/// How long `add_users` waits for its ack ([`USERS_ACK_WAIT`]).
ack_wait: std::time::Duration,
}
impl Pulled {
/// `inner` is the product's plugin; `instance` the estate whose main serves its tenants.
pub fn new(inner: Arc<dyn Product>, instance: impl Into<String>, slot: GuardianSlot, clock: Clock) -> Self {
let lane = Lane::new(inner.facts().id, instance, slot, clock);
Self { inner, lane, ack_wait: USERS_ACK_WAIT }
}
/// [`Self::new`] reading the box of the role `primary` answers ([`Primary`]; default `main`).
pub fn with_primary(mut self, primary: Primary) -> Self {
self.lane = self.lane.with_primary(primary);
self
}
/// [`Self::new`] waiting `wait` for a grant's ack (tests: none).
pub fn with_ack_wait(mut self, wait: std::time::Duration) -> Self {
self.ack_wait = wait;
self
}
pub fn instance(&self) -> &str {
self.lane.instance()
}
fn report(&self) -> Result<(ProductReport, u64), String> {
self.lane.report()
}
fn fresh(&self) -> Result<ProductReport, String> {
self.lane.fresh()
}
/// The own estate's report (required: the product is unavailable without it), then each
/// sold estate's that has reported, its tenants keyed `@<estate>/<ns>`. A sold estate that
/// has not called home is left out; monetize never dials it.
fn every_report(&self) -> Result<Vec<ProductReport>, ProductError> {
let (own, _) = self.report().map_err(ProductError::Unavailable)?;
let mut all = vec![own];
for estate in self.lane.sold_estates() {
let Ok((mut p, _)) = self.lane.report_of(&estate) else { continue };
for o in &mut p.tenants {
o.tenant = TenantId::on_estate(&estate, &o.tenant.0);
}
p.usage = std::mem::take(&mut p.usage).into_iter().map(|(t, u)| (TenantId::on_estate(&estate, &t).0, u)).collect();
all.push(p);
}
Ok(all)
}
}
impl Product for Pulled {
fn facts(&self) -> monetize_product::ProductFacts {
self.inner.facts()
}
fn list_tenants(&self) -> Result<Vec<TenantId>, ProductError> {
Ok(self.every_report()?.into_iter().flat_map(|p| p.tenants).map(|o| o.tenant).collect())
}
fn read_usage(&self, tenant: &TenantId) -> Result<Usage, ProductError> {
let (instance, ns) = self.lane.serving(tenant).map_err(ProductError::Refused)?;
let (mut p, _) = self.lane.report_of(&instance).map_err(ProductError::Unavailable)?;
p.usage.remove(&ns).ok_or_else(|| ProductError::Refused(format!("{instance}: the main's report has no usage for tenant {}", tenant.0)))
}
fn read_usage_page(&self, _req: &UsagePageReq) -> Result<Option<UsagePage>, ProductError> {
let all = self.every_report()?;
let as_of_unix_ms = all.iter().map(|p| p.measured_unix_ms).min().unwrap_or_default();
let rows = all.into_iter().flat_map(|p| p.usage).map(|(t, usage)| TenantUsage { tenant: TenantId(t), usage }).collect();
Ok(Some(UsagePage { rows, as_of_unix_ms, ..Default::default() }))
}
fn tenant_owners(&self) -> Result<Option<Vec<TenantOwner>>, ProductError> {
Ok(Some(self.every_report()?.into_iter().flat_map(|p| p.tenants).collect()))
}
/// ★ UW-10: the days of the asked window from the serving main's report, with how far
/// back the box remembers. A report that carries none for the tenant is unavailable,
/// never an empty series (that would read as a silent tenant).
fn read_activity(&self, tenant: &TenantId, since_unix_ms: u64, until_unix_ms: u64) -> Result<Activity, ProductError> {
let (instance, ns) = self.lane.serving(tenant).map_err(ProductError::Refused)?;
let (mut p, _) = self.lane.report_of(&instance).map_err(ProductError::Unavailable)?;
let mut a = p.activity.remove(&ns).ok_or_else(|| ProductError::Unavailable(format!("{instance}: the main's report carries no activity for tenant {}", tenant.0)))?;
let until = if until_unix_ms == 0 { u64::MAX } else { until_unix_ms };
a.days.retain(|d| d.day_unix_ms >= since_unix_ms && d.day_unix_ms <= until);
Ok(a)
}
fn push_entitlement(&self, fact: &EntitlementFact) -> Result<(), ProductError> {
self.lane.queue_entitlement(fact).map_err(ProductError::Unavailable)
}
/// A raise to `caps` needs `caps − promised` of the reported room of the main that serves
/// the tenant (a sold estate's own for `@<estate>/<ns>`), per meter the report names room
/// for. Short: bought iron helps only when the report says how.
fn can_absorb(&self, tenant: &TenantId, caps: &std::collections::BTreeMap<String, u64>) -> Result<Absorb, String> {
let (instance, ns) = self.lane.serving(tenant)?;
let p = self.lane.fresh_of(&instance)?;
let held = p.promised.get(&ns);
let short: Vec<String> = caps
.iter()
.filter_map(|(meter, target)| {
let room = *p.room.get(meter)?;
let need = target.saturating_sub(held.and_then(|h| h.get(meter)).copied().unwrap_or(0));
(need > room).then(|| format!("{meter} needs {need} beyond the {room} the main has room for"))
})
.collect();
if short.is_empty() {
return Ok(Absorb::OnHand);
}
match p.with_iron {
Some(note) => Ok(Absorb::WithIron { note }),
None => Err(format!("{instance}: {}", short.join("; "))),
}
}
fn servable(&self) -> Servable {
match self.fresh() {
Ok(p) => p.servable.unwrap_or_else(|| Servable::Unmeasured(format!("{}: its main's report does not say what it can serve", self.instance()))),
Err(w) => Servable::Unmeasured(w),
}
}
/// ST-11: another estate of the product answers from its OWN main's fresh report.
fn servable_on(&self, estate: &str) -> Servable {
if estate == self.instance() {
return self.servable();
}
match self.lane.fresh_of(estate) {
Ok(p) => p.servable.unwrap_or_else(|| Servable::Unmeasured(format!("{estate}: its main's report does not say what it can serve"))),
Err(w) => Servable::Unmeasured(w),
}
}
fn metered_door(&self) -> MeteredDoor {
match self.report() {
Ok((p, _)) => p.metered.unwrap_or_else(|| MeteredDoor::Unproven(format!("{}: its main's report does not say", self.instance()))),
Err(w) => MeteredDoor::Unproven(w),
}
}
/// ★ UW-5: from the serving main's fresh report — the account's room, its seats and the
/// room one user holds — and its names (a held name is refused before money moves). The
/// grant re-checks in the box; a refusal there is refunded.
fn users_fit(&self, tenant: &TenantId, ask: &UsersAsk) -> Result<UsersAnswer, ProductError> {
let (instance, ns) = self.lane.serving(tenant).map_err(ProductError::Refused)?;
let p = self.lane.fresh_of(&instance).map_err(ProductError::Unavailable)?;
let name = ask.name.trim();
if ask.count.max(1) == 1 && !name.is_empty() && p.names.iter().any(|n| n == name) {
return Err(ProductError::Refused(format!("{name} is already a name on {instance}: pick another")));
}
let room = p.accounts.get(&ns).ok_or_else(|| ProductError::Unavailable(format!("{instance}: the main's report says nothing of {}'s room yet", tenant.0)))?;
let room_bytes = if ask.room_bytes > 0 { ask.room_bytes } else { room.room_bytes };
let need_bytes = room_bytes.saturating_mul(u64::from(ask.count.max(1)));
let shortfall_bytes = room.unused_bytes.map(|u| need_bytes.saturating_sub(u)).unwrap_or(0);
Ok(UsersAnswer { principal: String::new(), room_bytes, need_bytes, unused_bytes: room.unused_bytes, shortfall_bytes, seats: room.seats, seats_open: room.seats_open, unchanged: true, invite_secret: String::new() })
}
/// ★ UW-5: ONE signed `add_users` order per `order`, answered from the main's ack — waited
/// for a short while (a live main runs it at once); still waiting, `Unavailable` (queued),
/// and the next ask reads the ack.
fn add_users(&self, tenant: &TenantId, ask: &UsersAsk, order: &str) -> Result<UsersAnswer, ProductError> {
// Waited for only by the call that queues it: a read of a grant queued before (the
// poller, a status page) answers at once and never holds a thread.
let waits = !self.lane.users_queued(tenant, order);
let until = std::time::Instant::now() + self.ack_wait;
loop {
match self.lane.add_users(tenant, ask, order).map_err(ProductError::Unavailable)? {
Some(Ok(answer)) => return Ok(answer),
Some(Err(words)) => return Err(ProductError::Refused(words)),
None if !waits || std::time::Instant::now() >= until => {
return Err(ProductError::Unavailable(format!("{order}: queued for {}'s main to grant on its next call", self.instance())))
}
None => std::thread::sleep(std::time::Duration::from_millis(250)),
}
}
}
/// ★ UW-4: the serving main says its set grows at runtime.
/// ★ The customer never sees a cap (owner): past the room of a main whose set is not
/// SEALED — it grows at runtime or at a restart, an operator grows it, or it cannot tell —
/// the order is queued and the main told (`grow_room`); a sealed set refuses, as before.
fn queues_room(&self) -> bool {
matches!(self.fresh().ok().and_then(|p| p.servable), Some(Servable::Measured { growth, .. }) if !matches!(growth, monetize_product::Growth::Sealed(_)))
}
fn queue_room(&self, tenant: &TenantId, need_bytes: u64, why: &str) -> Result<(), ProductError> {
self.lane.queue_room(tenant, need_bytes, why).map_err(ProductError::Unavailable)
}
/// ★ UW-11: ONE signed `purge_tenant` order at a time, asked again (naming the job) until
/// the main says it settled. Waiting or running is `still_locked` (nonpay asks again next
/// tick); a FAILED job is refused, so nothing of the tenant's is released.
fn purge_tenant(&self, tenant: &TenantId, reason: &str) -> Result<Purged, ProductError> {
let locked = |detail: String| Purged { tenant: tenant.clone(), detail, bytes_reclaimed: None, still_locked: true, failures: Vec::new() };
match self.lane.purge(tenant, reason).map_err(ProductError::Unavailable)? {
None => Ok(locked(format!("queued for {}'s main to purge on its next call; the account stays locked", self.instance()))),
Some(Err(words)) => Err(ProductError::Refused(words)),
Some(Ok(a)) if a.failed => Err(ProductError::Refused(format!(
"{}'s purge of {} did not finish ({}): {}{}",
self.instance(),
tenant.0,
a.detail,
if a.error.is_empty() { "no reason given" } else { &a.error },
if a.failures.is_empty() { String::new() } else { format!("; still on the disk: {}", a.failures.join("; ")) }
))),
Some(Ok(a)) if a.running => Ok(locked(format!("{}; the main's job {} is still running — the account stays LOCKED until it settles", a.detail, a.job))),
Some(Ok(a)) => Ok(Purged { tenant: tenant.clone(), detail: a.detail, bytes_reclaimed: a.bytes_reclaimed, still_locked: a.still_locked, failures: a.failures }),
}
}
}
/// **The mark of one resend of one version of a fact** (RA5-16): the estate, the tenant as its
/// main names it, and what makes the fact a new one (its order, its end, when it was issued).
pub fn resend_mark(instance: &str, ns: &str, fact: &EntitlementFact) -> String {
format!("resend/{instance}/{ns}/{}/{}/{}", fact.plan, fact.paid_until_unix_ms.unwrap_or(0), fact.issued_unix_ms.unwrap_or(0))
}
#[cfg(test)]
mod tests;