crates/monetize-product/src/pulled.rs
//! **The product as monetize serves it** (5.4, G1 pull; lane MONETIZE-FIX step 3c): every
//! answer the server loops ask of a product — its tenants and who they are, their usage and
//! activity, what the box can serve, the paid door, the room a raise finds, a grant of users,
//! a purge — read from the newest report of the estate's main ([`ProductReport`]), and every
//! change queued as a signed order the main pulls. Monetize never calls into an estate.
//!
//! [`Pulled`] is concrete: there is no per-product host adapter. A product's own words are its
//! facts ([`ProductFacts`], from its guest); the box's are its report. [`PullLane`] is the
//! guardian's store as a pulled product reads and queues on it (`monetize-guardian`'s `Lane`
//! in production).
use std::collections::BTreeMap;
use std::sync::Arc;
use crate::pull::{ProductReport, PurgeAnswer};
use crate::{
Absorb, Activity, EntitlementFact, MeteredDoor, ProductError, ProductFacts, Purged, Servable,
TenantId, TenantOwner, TenantUsage, Usage, UsagePage, UsagePageReq, UsersAnswer, UsersAsk,
};
/// 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);
/// Where a product's facts come from, asked each time (a guest replaced live answers anew).
pub type FactsSource = Arc<dyn Fn() -> ProductFacts + Send + Sync>;
/// **One product's estate over the pull channel**, as [`Pulled`] reads and orders it: the
/// estate's main's reports, the fleet's other estates of the product, and the signed orders
/// queued for them (marked, so a repeat answers from the first one's ack).
pub trait PullLane: Send + Sync {
/// The product's own estate.
fn instance(&self) -> &str;
/// The estate whose main serves `tenant` and the tenant's name there.
fn serving(&self, tenant: &TenantId) -> Result<(String, String), String>;
/// `instance`'s main's newest report and when monetize heard it.
fn report_of(&self, instance: &str) -> Result<(ProductReport, u64), String>;
/// The same, refused when the main has been silent past the guardian's bound.
fn fresh_of(&self, instance: &str) -> Result<ProductReport, String>;
/// The product's other estates on the fleet.
fn sold_estates(&self) -> Vec<String>;
/// Queue the fact for the main that serves its tenant (newest per tenant wins).
fn queue_entitlement(&self, fact: &EntitlementFact) -> Result<(), String>;
/// Whether the grant under `reference` is queued and not yet acked.
fn users_queued(&self, tenant: &TenantId, reference: &str) -> bool;
/// Queue a grant of users once per `reference`; `Some` once the main acked it.
fn add_users(
&self,
tenant: &TenantId,
ask: &UsersAsk,
reference: &str,
) -> Result<Option<Result<UsersAnswer, String>>, String>;
/// Queue ONE `grow_room` order for the main that serves `tenant`.
fn queue_room(&self, tenant: &TenantId, need_bytes: u64, why: &str) -> Result<(), String>;
/// Queue the purge once; `Some` once the main says how it went.
fn purge(
&self,
tenant: &TenantId,
reason: &str,
) -> Result<Option<Result<PurgeAnswer, String>>, String>;
/// The own estate's newest report.
fn report(&self) -> Result<(ProductReport, u64), String> {
self.report_of(self.instance())
}
/// The own estate's report, fresh.
fn fresh(&self) -> Result<ProductReport, String> {
self.fresh_of(self.instance())
}
}
/// **A product, served** (see the module): its facts and the lane of its estate.
#[derive(Clone)]
pub struct Pulled {
facts: FactsSource,
lane: Arc<dyn PullLane>,
/// How long `add_users` waits for its ack ([`USERS_ACK_WAIT`]).
ack_wait: std::time::Duration,
}
impl std::fmt::Debug for Pulled {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Pulled")
.field("instance", &self.lane.instance())
.finish()
}
}
impl Pulled {
/// The product whose facts `facts` answers, served from `lane`.
pub fn new(facts: FactsSource, lane: Arc<dyn PullLane>) -> Self {
Self {
facts,
lane,
ack_wait: USERS_ACK_WAIT,
}
}
/// [`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
}
/// The estate whose main serves the product's own tenants.
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.lane.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)
}
pub fn facts(&self) -> crate::ProductFacts {
(self.facts)()
}
pub fn list_tenants(&self) -> Result<Vec<TenantId>, ProductError> {
Ok(self
.every_report()?
.into_iter()
.flat_map(|p| p.tenants)
.map(|o| o.tenant)
.collect())
}
pub 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
))
})
}
pub fn read_usage_page(&self, req: &UsagePageReq) -> Result<Option<UsagePage>, ProductError> {
let _ = req;
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()
}))
}
pub 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).
pub 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)
}
pub 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.
pub fn can_absorb(
&self,
tenant: &TenantId,
caps: &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("; "))),
}
}
pub 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.
pub 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),
}
}
pub 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.
pub 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.
pub 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.
pub fn queues_room(&self) -> bool {
matches!(self.fresh().ok().and_then(|p| p.servable), Some(Servable::Measured { growth, .. }) if !matches!(growth, crate::Growth::Sealed(_)))
}
pub 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.
pub 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 }),
}
}
}
/// **A product placed on no estate**: `[products.<id>]` names no `instance`, so no main's report
/// answers for it; every read and order is refused by name.
pub struct Unplaced(pub &'static str);
impl Unplaced {
fn no(&self) -> String {
format!("[products.{}] names no instance: no estate's main reports for it, and monetize never dials one", self.0)
}
}
impl PullLane for Unplaced {
fn instance(&self) -> &str {
""
}
fn serving(&self, _tenant: &TenantId) -> Result<(String, String), String> {
Err(self.no())
}
fn report_of(&self, _instance: &str) -> Result<(ProductReport, u64), String> {
Err(self.no())
}
fn fresh_of(&self, _instance: &str) -> Result<ProductReport, String> {
Err(self.no())
}
fn sold_estates(&self) -> Vec<String> {
Vec::new()
}
fn queue_entitlement(&self, _fact: &EntitlementFact) -> Result<(), String> {
Err(self.no())
}
fn users_queued(&self, _tenant: &TenantId, _reference: &str) -> bool {
false
}
fn add_users(
&self,
_tenant: &TenantId,
_ask: &UsersAsk,
_reference: &str,
) -> Result<Option<Result<UsersAnswer, String>>, String> {
Err(self.no())
}
fn queue_room(&self, _tenant: &TenantId, _need_bytes: u64, _why: &str) -> Result<(), String> {
Err(self.no())
}
fn purge(
&self,
_tenant: &TenantId,
_reason: &str,
) -> Result<Option<Result<PurgeAnswer, String>>, String> {
Err(self.no())
}
}
/// **One estate's main held in memory** (tests, and a host that keeps no guardian store): its
/// newest report, the orders queued for it, and the main's answers to the ones that answer.
/// Every report is fresh. What [`Pulled`] reads and queues on, as on a real main.
#[derive(Default)]
pub struct HeldMain {
instance: String,
held: std::sync::Mutex<Held>,
}
/// What a [`HeldMain`] holds.
#[derive(Default, Clone)]
pub struct Held {
pub report: Option<ProductReport>,
/// The entitlement facts queued, oldest first.
pub entitlements: Vec<EntitlementFact>,
/// The grants of users queued, oldest first.
pub users: Vec<crate::pull::AddUsers>,
/// The newest `grow_room` order, unacked.
pub room: Option<crate::pull::GrowRoom>,
/// The purges queued.
pub purges: Vec<crate::pull::PurgeTenant>,
/// The main's answer to a grant of users, once it ran one.
pub users_answer: Option<Result<UsersAnswer, String>>,
/// The main's answer to a purge, once it ran one.
pub purge_answer: Option<Result<PurgeAnswer, String>>,
}
impl HeldMain {
/// The main of `instance`, which has not reported yet.
pub fn new(instance: impl Into<String>) -> Arc<HeldMain> {
Arc::new(HeldMain {
instance: instance.into(),
..Default::default()
})
}
/// **The main reports**: `edit` changes its newest report (an empty one at first).
pub fn report(&self, edit: impl FnOnce(&mut ProductReport)) {
edit(
self.lock()
.report
.get_or_insert_with(ProductReport::default),
);
}
/// **The main changes what it holds or answers** (its answers to grants and purges).
pub fn hold(&self, edit: impl FnOnce(&mut Held)) {
edit(&mut self.lock());
}
/// What it holds now.
pub fn held(&self) -> Held {
self.lock().clone()
}
/// The entitlement facts queued, oldest first.
pub fn entitlements(&self) -> Vec<EntitlementFact> {
self.lock().entitlements.clone()
}
fn lock(&self) -> std::sync::MutexGuard<'_, Held> {
self.held.lock().unwrap_or_else(|e| e.into_inner())
}
/// A product whose facts `facts` answers, served from this main (no ack wait).
pub fn serve(self: &Arc<Self>, facts: ProductFacts) -> Pulled {
Pulled::new(
Arc::new(move || facts.clone()),
Arc::clone(self) as Arc<dyn PullLane>,
)
.with_ack_wait(std::time::Duration::ZERO)
}
}
impl PullLane for HeldMain {
fn instance(&self) -> &str {
&self.instance
}
fn serving(&self, tenant: &TenantId) -> Result<(String, String), String> {
match tenant.estate() {
Some((estate, ns)) if estate != self.instance => {
Err(format!("{estate}: no estate of this product holds {ns}"))
}
Some((_, ns)) => Ok((self.instance.clone(), ns.to_owned())),
None => Ok((self.instance.clone(), tenant.0.clone())),
}
}
fn report_of(&self, instance: &str) -> Result<(ProductReport, u64), String> {
if instance != self.instance {
return Err(format!("{instance}: not this estate"));
}
self.lock().report.clone().map(|r| (r, 1)).ok_or_else(|| {
format!("{instance}: its main has sent no report yet; monetize never dials it")
})
}
fn fresh_of(&self, instance: &str) -> Result<ProductReport, String> {
self.report_of(instance).map(|(r, _)| r)
}
fn sold_estates(&self) -> Vec<String> {
Vec::new()
}
fn queue_entitlement(&self, fact: &EntitlementFact) -> Result<(), String> {
self.lock().entitlements.push(fact.clone());
Ok(())
}
fn users_queued(&self, _tenant: &TenantId, reference: &str) -> bool {
let h = self.lock();
h.users_answer.is_none() && h.users.iter().any(|u| u.order == reference)
}
fn add_users(
&self,
tenant: &TenantId,
ask: &UsersAsk,
reference: &str,
) -> Result<Option<Result<UsersAnswer, String>>, String> {
let ns = self.serving(tenant)?.1;
let mut h = self.lock();
if !h.users.iter().any(|u| u.order == reference) {
h.users.push(crate::pull::AddUsers {
tenant: ns,
ask: ask.clone(),
order: reference.to_owned(),
});
}
Ok(h.users_answer.clone())
}
fn queue_room(&self, _tenant: &TenantId, need_bytes: u64, why: &str) -> Result<(), String> {
self.lock().room = Some(crate::pull::GrowRoom {
need_bytes,
reason: why.to_owned(),
});
Ok(())
}
fn purge(
&self,
tenant: &TenantId,
reason: &str,
) -> Result<Option<Result<PurgeAnswer, String>>, String> {
let ns = self.serving(tenant)?.1;
let mut h = self.lock();
if !h.purges.iter().any(|p| p.tenant == ns) {
h.purges.push(crate::pull::PurgeTenant {
tenant: ns,
reason: reason.to_owned(),
..Default::default()
});
}
Ok(h.purge_answer.clone())
}
}