crates/monetize-server/src/poller.rs
//! **The usage poller.** Every N minutes, for every tenant of every product:
//! `read_usage` → the usage store; then the ladder is re-derived from the stored
//! entitlement's `paid_until` and the deployment's policy, and a changed state is
//! signed and pushed (Grace expiring → Suspended, Suspended → Retention). Open
//! ledger attempts are resumed on the same tick (a callback that never came, a
//! push that failed), and so are PENDING paid signups (a customer who closed the
//! return page during an outage). Every state change is a `functional_status` row.
//!
//! The poller never blocks the product: it reads, it pushes, and a product that
//! is down costs one `false` row and a retry next tick.
use std::sync::Arc;
use std::time::Duration;
use monetize::derive_state;
use monetize_product::{EntitlementFact, MeteredDoor};
use crate::health::Reachability;
use crate::services::purchase::resume_reference;
use crate::state::AppState;
/// Source stamped on a fact the ladder moved (as opposed to `operator` or
/// `payment:<vendor>:<ref>`): the product's attestation log then says the clock
/// did it, not a person and not a payment.
pub const LADDER_SOURCE: &str = "ladder";
#[derive(Debug, Default, PartialEq, Eq)]
pub struct Tick {
pub tenants_read: usize,
pub read_failures: usize,
pub state_changes: usize,
pub push_failures: usize,
pub resumed: usize,
/// PENDING paid signups this tick DECIDED (`crate::signup::settle_pending`).
pub signups_settled: usize,
/// Unacknowledged entitlement pushes this tick got the product to
/// ACKNOWLEDGE ([`retry_pushes`]).
pub pushes_retried: usize,
/// Payments the vendors answered this tick — every fact examined
/// ([`crate::reconcile::pass`]).
pub reconciled: usize,
/// Of those, the ones monetize had NO ROW FOR: new cases this tick,
/// handed to the signup engine or left for an operator. Zero is the
/// quiet tick; anything else is a payment that would have been lost.
pub unfulfilled: usize,
/// **Ladder moves this tick THREW AWAY** because a payment (or an
/// operator) replaced the tenant's fact while the move was in flight. Zero
/// on a quiet tick; anything else is a tenant who was one overwrite away
/// from being suspended after paying.
pub ladder_superseded: usize,
/// **Paid-signup tenants this tick ADOPTED into the ledger** so the ladder
/// can move them — see [`poll_product`]. Zero on every tick after the
/// first one that saw them; anything else is a signup the ladder had never
/// walked.
pub signups_adopted: usize,
/// **Payments taken for accounts that were never created** — the whole
/// standing queue, not this tick's new ones: the number an operator owes
/// somebody back by hand (`crate::signup::SignupStore::owed_back`). It is a
/// STANDING count on purpose, because the failure this exists to break is a
/// refusal that happened once, months ago, and has been invisible ever
/// since.
///
/// `None` is the store refusing to answer. It is not 0: a store that will
/// not open is not a deployment that owes nobody anything, and the health
/// line prints `?` for it.
pub owed_back: Option<usize>,
/// **Products whose PAID DOOR is not proven to be there** this tick —
/// `Absent` (the box answered `UNIMPLEMENTED`: a community build) plus
/// `Unproven` (nobody could ask). Both count, because a deployment that
/// sells may not call itself healthy on either. See [`probe_metered_door`].
pub metered_unproven: usize,
/// **Person accounts this tick GIVEN the shop's account fact** (Free, no room) because
/// they held none — see [`owed_account_facts`]. Zero on a quiet tick; anything else is an
/// account an invite, an operator or a failed sign-up made, which was UNCAPPED until now.
pub accounts_given: usize,
/// **What the nonpay engine did this tick** — warnings, deletions,
/// cancellations, reminders ([`crate::nonpay`]). Not on the poll line (a
/// deployment check greps its tail); [`nonpay_line`] says it when any is
/// non-zero.
pub nonpay: crate::nonpay::Did,
/// **Unacknowledged pushes that have failed [`PUSH_ALERT_AFTER`] times or
/// more and are still asked every tick** (R5.1-F). Each one is ALERTed on the
/// front's log as it crosses every multiple of the threshold; the push is
/// never given up. Not on the poll line (a deployment check greps its tail).
pub pushes_stuck: usize,
}
/// A pending push is ALERTed after this many failed attempts (6 ticks of 5
/// minutes = half an hour of a paid fact not reaching the product) and again at
/// every multiple of it — and is still pushed every tick: no give-up.
pub const PUSH_ALERT_AFTER: u32 = 6;
/// One pass over every product. Blocking; the loop runs it on `spawn_blocking`.
pub fn tick(state: &AppState) -> Tick {
let mut t = Tick::default();
for product in state.registry.products() {
poll_product(state, product, &mut t);
}
match state.ledger.unfinished() {
Ok(refs) => {
for r in refs {
match resume_reference(state, &r) {
Ok(Some(v)) if v.state != monetize_proto::PurchaseState::Pending => {
t.resumed += 1
}
Ok(_) => {}
Err(e) => eprintln!("monetize-server: resume {r}: {}", e.message()),
}
}
}
Err(e) => eprintln!("monetize-server: ledger unfinished: {e}"),
}
// **Paid signups nobody is watching.** Same tick, same blocking thread, so
// a retry never overlaps the previous one. See `settle_pending` for the
// interval and for why it is this loop and not a thread of its own.
t.signups_settled = crate::signup::settle_pending(state);
// **And the ones nobody CAN watch any more**: money taken, account refused.
// Counted after the settle, so a signup this very tick decided is in the
// number. See `owed_back`.
t.owed_back = count_owed_back(state);
// **Facts the product never acknowledged.** After the ladder, so a move
// this tick that failed is not retried in the same breath against the
// same unreachable box; the next tick asks.
t.pushes_retried = retry_pushes(state);
// ★ ALICA-4: a paid fact a site's main does not hold (its order refused or lost) is
// queued again; the line says which.
resend_unheld_facts(state);
t.pushes_stuck = stuck_pushes(state);
// **What the vendor took that nobody told us about.** Last, so a signup
// it hands to the engine and a sink that says "not now" meet on the next
// tick's `settle_pending` rather than being retried in the same breath.
let r = crate::reconcile::pass(state);
t.reconciled = r.examined;
t.unfulfilled = r.unfulfilled;
t
}
/// **Ask the box one metered verb, and say so loudly when it is not there.**
///
/// The poller's other calls — `list_tenants`, `read_usage` — are OPEN-build
/// services. gunnar's community ISO answers both perfectly, so
/// `poll tenants=11 failures=0` proves connectivity and NOTHING about whether
/// the appliance has `Entitlement` compiled in. The first thing to find out
/// otherwise is a paying customer: the principal is created, `Entitlement.Set`
/// answers `UNIMPLEMENTED`, and the signup is refused with the money taken
/// (`crate::signup::Owed`). The last proof this estate's appliance is a metered
/// build was a journal line from 2026-09-15.
///
/// **It is not fatal and it is not quiet.** Fatal would be wrong twice over: a
/// box that is merely down answers `Unproven`, and a deployment that sells
/// nothing (the open build, a free estate) is not misconfigured for running
/// against a community appliance. So the unhealthy row is written when this
/// deployment SELLS — `[signups.<product>.plans]` is not empty — and the count goes on the
/// poll line either way.
fn probe_metered_door(
state: &AppState,
product: monetize_product::Served<'_>,
now: u64,
t: &mut Tick,
) {
// A box that would not list its tenants will not answer a metered verb
// either, and asking anyway costs a second timeout per tick against a box
// that is down — the reason the capacity question is skipped there too.
let door = match state.health.get(product.id()).reachability {
Reachability::Down(why) => MeteredDoor::Unproven(format!(
"{} did not answer at all this tick: {why}",
product.id()
)),
_ => product.metered_door(),
};
state
.health
.set_metered_door(product.id(), door.clone(), now);
if door.proven() {
return;
}
t.metered_unproven += 1;
let sells = !state.signup_plans.is_empty();
let detail = match (&door, sells) {
(MeteredDoor::Absent(why), true) => format!(
"{}: THIS DEPLOYMENT SELLS AND THE BOX CANNOT ENFORCE WHAT IT SELLS — a metered verb answered UNIMPLEMENTED, \
which is a community build: {why}. Every paid signup against it will take the money and fail to grant the \
entitlement. Either re-image the appliance with the metered build or empty [signups.<product>.plans]",
product.id()
),
(MeteredDoor::Absent(why), false) => {
format!("{}: a metered verb answered UNIMPLEMENTED ({why}); this deployment sells no signups, so nothing depends on it today", product.id())
}
(_, true) => format!("{}: the paid door is UNPROVEN and this deployment sells: {}", product.id(), door.detail()),
(_, false) => format!("{}: the paid door is unproven: {}", product.id(), door.detail()),
};
// A deployment that sells against a box that cannot enforce entitlements
// must not read as healthy — that is the whole of this probe. One that
// sells nothing says the same thing without the red.
skidbladnir::functional_status(
"monetize-server/entitlement",
"metered-door",
!sells,
&detail,
);
eprintln!("monetize-server: {detail}");
}
/// **The owed-back queue, counted and REPORTED**, once per tick.
///
/// A row on a console page is only seen by somebody who opens it. This is the
/// other half: every tick with a non-empty queue writes a FAILING
/// `functional_status` row naming the count, so a deployment that has taken
/// money for accounts it never made cannot read as healthy in the one place an
/// operator actually watches. A queue of zero writes nothing — a green line per
/// tick would be the noise that hides the red one.
///
/// `None` when the store would not answer, and the row then says so: the one
/// thing this must never do is print 0 because it could not count.
fn count_owed_back(state: &AppState) -> Option<usize> {
match state.signups.owed_back() {
Ok(rows) => {
if !rows.is_empty() {
let oldest = rows.first().map(|r| r.recorded_at_unix_ms).unwrap_or(0);
let detail = format!(
"{} payment(s) taken for accounts that were never created and are owed back; oldest recorded at {oldest}; \
SignupAdmin.ListOwedBack has the sessions, the buyers and what each one needs, and SignupAdmin.Refund \
(operator bearer; `monetize-server --refund <vendor>:<session> --reason …`) gives it back once",
rows.len()
);
skidbladnir::functional_status(
"monetize-server/signup",
"owed-back",
false,
&detail,
);
eprintln!("monetize-server: signup {detail}");
}
Some(rows.len())
}
Err(e) => {
skidbladnir::functional_status(
"monetize-server/signup",
"owed-back",
false,
&format!("the signup store could not be counted: {e}"),
);
eprintln!("monetize-server: signup owed-back could not be counted: {e}");
None
}
}
}
/// ★ ALICA-4 (live 2026-10-26): **every paid fact a pulled site's main does not hold is
/// queued for it again** ([`monetize_guardian::pulled::Lane::resend_unheld`]): the ledger's
/// facts and the signup-granted ones not adopted yet, per product. A refused or lost
/// `entitlement` order is otherwise never sent again (a site's gunnar said Free for a paid
/// plan). Answers how many were queued.
pub fn resend_unheld_facts(state: &AppState) -> usize {
let now = state.now();
let mut queued = 0;
for product in state.registry.product_ids() {
let facts = match state.ledger.entitlements(product) {
Ok(f) => f,
Err(e) => {
eprintln!("monetize-server: {product}: unheld facts: the ledger: {e}");
continue;
}
};
let facts = match crate::signup::with_signup_facts(state, product, facts) {
Ok(f) => f,
Err(e) => {
eprintln!("monetize-server: {product}: unheld facts: the signup store: {e}");
continue;
}
};
let home = state.registry.home_of(product).unwrap_or_default();
let lane = monetize_guardian::pulled::Lane::new(
product,
home,
state.registry.pull_slot(),
std::sync::Arc::new(move || now),
);
for line in lane.resend_unheld(&facts) {
queued += 1;
eprintln!("monetize-server: ALERT entitlement.site_not_holding {product}/{line}");
}
}
queued
}
/// **Push every stored fact the product has not acknowledged, once.** Returns
/// how many the product acknowledged this time.
///
/// * A pending fact that is no longer the tenant's current fact is dropped: a
/// newer one replaced it (and carries its own pending row if it too failed).
/// * A failed push (the fact could not be queued for the main) counts an attempt
/// and stays pending for the next tick. There is no give-up: a product that is
/// down for a day must still get the lapse when it comes back. A served
/// product queues the fact and never refuses one (v2.0 step 1).
/// * A pending row whose product is no longer linked is left alone.
pub fn retry_pushes(state: &AppState) -> usize {
let pending = match state.ledger.pending_pushes() {
Ok(v) => v,
Err(e) => {
eprintln!("monetize-server: pending pushes: {e}");
return 0;
}
};
let mut acknowledged = 0;
for (product_id, mut p) in pending {
let Some(product) = state.registry.products().find(|x| x.id() == product_id) else {
continue;
};
let tenant = p.fact.tenant.0.clone();
let who = format!("{product_id}/{tenant}");
match state.ledger.entitlement(&product_id, &tenant) {
Ok(Some(current)) if current == p.fact => {}
Ok(_) => {
if let Err(e) = state.ledger.push_acknowledged(&product_id, &tenant) {
eprintln!("monetize-server: pending push {who}: {e}");
}
continue;
}
Err(e) => {
eprintln!("monetize-server: pending push {who}: {e}");
continue;
}
}
let outcome = match product.push_entitlement(&p.fact) {
Ok(()) => {
acknowledged += 1;
skidbladnir::functional_status(
"monetize-server/entitlement",
"push-retry",
true,
&format!(
"{who}: {:?} acknowledged after {} failed push(es)",
p.fact.state, p.attempts
),
);
state.ledger.push_acknowledged(&product_id, &tenant)
}
Err(e) => {
p.attempts = p.attempts.saturating_add(1);
p.last_error = e.to_string();
if p.attempts % PUSH_ALERT_AFTER == 0 {
// R5.1-F: said at every multiple of the threshold, never given up.
let detail = format!(
"{who}: {:?} has not reached the product after {} pushes since {} — still pushed every tick; last: {e}",
p.fact.state, p.attempts, p.since_unix_ms
);
skidbladnir::functional_status(
"monetize-server/entitlement",
"push-stuck",
false,
&detail,
);
eprintln!("monetize-server: ALERT entitlement.push_stuck {detail}");
}
state.ledger.push_pending(&product_id, &p)
}
};
if let Err(e) = outcome {
eprintln!("monetize-server: pending push {who}: {e}");
}
}
acknowledged
}
/// **Pending pushes that have failed [`PUSH_ALERT_AFTER`] times or more** and
/// (every one is still asked). R5.1-F.
pub fn stuck_pushes(state: &AppState) -> usize {
match state.ledger.pending_pushes() {
Ok(v) => v
.iter()
.filter(|(_, p)| p.attempts >= PUSH_ALERT_AFTER)
.count(),
Err(e) => {
eprintln!("monetize-server: pending pushes: {e}");
0
}
}
}
fn poll_product(state: &AppState, product: monetize_product::Served<'_>, t: &mut Tick) {
let now = state.now();
let tenants = match product.list_tenants() {
Ok(mut v) => {
// One pass per tenant however the product lists them: the ladder,
// the queue and the nonpay engine each act once per tick.
let mut seen = std::collections::BTreeSet::new();
v.retain(|t| seen.insert(t.0.clone()));
// **THE REACHABILITY DOT** (`UI.md` screen 1). Recorded here and not
// measured from a page: this is the one place that already talks to
// every product, already runs on the blocking pool, and is already
// bounded by the plugin's own per-call deadline. A page that asked
// the question itself would hang on the slowest box in the estate,
// which is worse than a page that says "unknown".
state
.health
.set_reachability(product.id(), Reachability::Up, now);
v
}
Err(e) => {
skidbladnir::functional_status(
"monetize-server/poller",
product.id(),
false,
&format!("list_tenants: {e}"),
);
eprintln!("monetize-server: {} list_tenants: {e}", product.id());
state
.health
.set_reachability(product.id(), Reachability::Down(e.to_string()), now);
t.read_failures += 1;
// **And the paid door goes back to UNPROVEN**, rather than keeping
// whatever the last tick saw. A cached `enforcing` about a box
// nobody can reach is exactly the stale-answer-rendered-as-current
// that `crate::health` exists to avoid — and if this box were
// replaced by a community ISO, the old yes is the answer that would
// survive the swap.
probe_metered_door(state, product, now, t);
// A product that will not list its tenants will not answer a
// capacity question either, and asking anyway would cost a second
// timeout on every tick against a box that is down.
return;
}
};
// **THE PAID DOOR.** One read-only metered verb per product per tick, for
// the reason `probe_metered_door` gives: nothing else in this loop can tell
// a metered appliance from a community ISO.
probe_metered_door(state, product, now, t);
// **THE FLEET'S DENOMINATOR.** One call per product per tick, cached, so
// `Usage.Fleet` never blocks a page load on the product. It is asked BEFORE
// the per-tenant loop so a product with many tenants still refreshes it.
state
.health
.set_servable(product.id(), product.servable(), now);
// **THE SHOP'S OWN CUSTOMERS.** A paid signup writes no ledger row
// (`crate::signup`'s header says why), and until this fold existed the loop
// below `continue`d past every one of them on every tick forever: the
// ladder never moved them, the lapse was never pushed, and gunnar — which
// does no clock arithmetic of its own — served one month's money without
// end. One fold per product per tick; a signup whose plan the config no
// longer names contributes nothing, exactly as `signup_facts` has it.
let granted: std::collections::BTreeMap<String, EntitlementFact> =
match crate::signup::granted_facts(state, product.id()) {
Ok(v) => v.into_iter().map(|f| (f.tenant.0.clone(), f)).collect(),
Err(e) => {
eprintln!("monetize-server: {} signup facts: {e}", product.id());
Default::default()
}
};
let listed: std::collections::BTreeSet<String> = tenants.iter().map(|t| t.0.clone()).collect();
give_account_facts(state, product, &granted, now, t);
// **THE USAGE, for every listed tenant, before the per-tenant loop** —
// batched, and after the first tick only what changed (2026-10-06): see
// `crate::usage_poll`. One transaction stores the whole harvest.
let harvest = crate::usage_poll::collect(state, product, &tenants);
if let Err(e) = state.usage.put_many(product.id(), &harvest.readings) {
eprintln!("monetize-server: usage store: {e}");
}
for (tenant, reading) in &harvest.readings {
// The live feed. Published HERE and not from `UsageStore::put`,
// deliberately: `put` is also how a backup RESTORE lands its
// readings, and a restore replays measurements, it does not take
// them. An event that fired there would tell a console that
// three hundred tenants were just measured by a poller that
// never ran.
state
.bus
.publish(monetize_proto::event::Payload::UsageMeasured(
monetize_proto::UsageMeasured {
product: product.id().into(),
tenant: tenant.clone(),
meters: reading
.meters
.iter()
.map(|(name, value)| monetize_proto::Meter {
name: name.clone(),
value: *value,
})
.collect(),
measured_at_unix_ms: reading.measured_at_unix_ms,
},
));
}
t.tenants_read += harvest.readings.len();
t.read_failures += harvest.failed.len();
for (tenant, e) in &harvest.failed {
eprintln!(
"monetize-server: {}/{} read_usage: {e}",
product.id(),
tenant
);
}
eprintln!("monetize-server: {}", harvest.line(product.id()));
for tenant in tenants {
// The ladder.
let fact = match state.ledger.entitlement(product.id(), &tenant.0) {
Ok(Some(fact)) => fact,
// **ADOPTION.** No ledger row, but a paid signup granted this
// tenant an entitlement and the appliance is holding it. The
// GRANTED fact is seeded — the Paid one the sink pushed, not a
// state re-derived at `now`, or the ladder below would find
// nothing to move and the lapse would never be signed. Seeded with
// `set_entitlement` and NOT `apply_fact`: the appliance already
// has this exact fact, and pushing it again would be a second
// attestation line for something that never changed.
Ok(None) => match granted.get(&tenant.0) {
Some(f) => {
if let Err(e) = state.ledger.set_entitlement(product.id(), f) {
eprintln!("monetize-server: adopt {}/{}: {e}", product.id(), tenant.0);
continue;
}
t.signups_adopted += 1;
skidbladnir::functional_status(
"monetize-server/poller",
"signup-adopted",
true,
&format!("{}/{}: the paid signup's entitlement is on the ledger; the ladder walks it from now on", product.id(), tenant.0),
);
f.clone()
}
None => continue,
},
Err(e) => {
eprintln!(
"monetize-server: {}/{} entitlement: {e}",
product.id(),
tenant.0
);
continue;
}
};
if let Some(next) = ladder_move(state, &fact, now) {
let signed = state.signer_for(product.id()).sign_fact(next);
// **COMPARE-AND-SWAP against the fact this move was derived from.**
// A payment that lands while the push is in flight must win; see
// `AppState::apply_ladder_fact`.
match state.apply_ladder_fact(product, Some(&fact), &signed) {
// The move was written. `stands` is what monetize believes
// afterwards — the ladder's fact, or a newer one that replaced
// it while the push was in flight. The fact is stored whether or
// not the product acknowledged it; the pending-push retry chases
// the product.
Ok(Some(applied)) => {
t.state_changes += 1;
if applied.pushed.is_err() {
t.push_failures += 1;
}
if applied.superseded_in_flight {
t.ladder_superseded += 1;
eprintln!(
"monetize-server: ladder {}/{}: {:?} was replaced by {:?} while the push was in flight; the newer fact is pending for the product",
product.id(),
tenant.0,
signed.state,
applied.stands.state
);
}
}
// **Somebody moved the tenant while this tick was deciding** —
// a payment, or an operator. Their fact is newer and they
// pushed it; this verdict is void (nonpay below re-reads what
// is ACTUALLY stored now).
Ok(None) => {
t.ladder_superseded += 1;
let who = format!("{}/{}", product.id(), tenant.0);
eprintln!("monetize-server: ladder {who}: {:?} was superseded while the move was in flight; the newer fact stands", signed.state);
}
// The ledger write itself failed: nothing moved.
Err(e) => eprintln!("monetize-server: ledger: {e}"),
}
}
// **Nonpay** (owner ruling g-nonpay): the warnings, the deletion at
// `delete_after_days`, the cancellation when they pay — and the
// renewal reminder and the declined-payment mail for those who do.
// After the ladder, so it reads the fact that stands.
t.nonpay += crate::nonpay::observe(state, product, &tenant, now);
}
// **A deletion in flight is followed whether or not the product still lists
// the tenant**: the purge takes the namespace, and the product's listing with it.
for tenant in state.nonpay.deleting(product.id()) {
if !listed.contains(&tenant) {
t.nonpay +=
crate::nonpay::observe(state, product, &monetize_product::TenantId(tenant), now);
}
}
}
/// **The nonpay engine's line**, when it did anything this tick.
pub fn nonpay_line(t: &Tick) -> Option<String> {
let n = &t.nonpay;
(*n != crate::nonpay::Did::default()).then(|| {
format!(
"nonpay warned={} deleted={} cancelled={} reminded={} failures={}",
n.warned, n.deleted, n.cancelled, n.reminded, n.failures
)
})
}
/// The fact the ladder says should replace `fact` at `now`, or `None` if it
/// holds. The day counts are the deployment's ONE policy; a fact with no
/// `paid_until` (a hand-set Free) is never moved.
pub fn ladder_move(state: &AppState, fact: &EntitlementFact, now: u64) -> Option<EntitlementFact> {
fact.paid_until_unix_ms?;
let policy = state.terms.read().unwrap_or_else(|e| e.into_inner()).policy;
let next = derive_state(&policy, fact.paid_until_unix_ms, now);
if next == fact.state {
return None;
}
Some(EntitlementFact {
state: next,
source: LADDER_SOURCE.into(),
signature: Vec::new(),
..fact.clone()
})
}
/// **Does this shop SELL `product`'s room?** Not served free (`[products.<id>] free`, a sold
/// estate: its customer never sees a cap) and priced (`[prices.<id>]` holds a market). Only
/// such a shop gives every person's account a fact ([`give_account_facts`]).
pub fn sells_room(state: &AppState, product: &str) -> bool {
let free = state
.config
.read()
.map(|c| c.products.get(product).is_some_and(|p| p.free))
.unwrap_or(true);
let priced = state
.terms
.read()
.map(|t| t.prices.get(product).is_some_and(|m| !m.is_empty()))
.unwrap_or(false);
!free && priced
}
/// **Does this shop sell `product`'s room to `tenant`?** (HENRIK-1, prod 2026-10-26.) The shop
/// sells room ([`sells_room`]) AND the tenant lives where it is sold: the home (a plain
/// tenant, or one on the product's own estate) or one of the operator's own declared sites
/// (`[estates.<e>] operator_site`). A sold estate's tenant (`@<estate>/<ns>` on a customer's
/// private instance) is served free: its customer never sees a cap, so it is never given one.
pub fn sells_room_to(state: &AppState, product: &str, tenant: &str) -> bool {
if !sells_room(state, product) {
return false;
}
let id = monetize_product::TenantId(tenant.to_owned());
match id.estate() {
None => true,
Some((estate, _)) => {
state.registry.home_of(product).as_deref() == Some(estate)
|| state.registry.operator_site(estate)
}
}
}
/// Pure: **the accounts owed the shop's account fact** (2026-10-08, the free-room leak): every
/// PERSON's account that holds no fact — an invite's, an operator's, a failed sign-up's — and
/// is not the SELLER's own. Without one gunnar reads the account as UNCAPPED; with the
/// sign-up's account fact (Free, no end, no room) it holds what every account at the shop
/// holds until it buys room. `has_fact` answers true on a ledger it cannot read: nothing is
/// issued on a guess.
pub fn owed_account_facts(
owners: &[monetize_product::TenantOwner],
has_fact: impl Fn(&str) -> bool,
) -> Vec<monetize_product::TenantId> {
owners
.iter()
.filter(|o| o.person && !o.seller && !has_fact(&o.tenant.0))
.map(|o| o.tenant.clone())
.collect()
}
/// Pure: **the seller's own namespaces**, by name — what the poller leaves out of
/// [`owed_account_facts`] and says so once per start ([`give_account_facts`]).
pub fn sellers_own(owners: &[monetize_product::TenantOwner]) -> Vec<String> {
owners
.iter()
.filter(|o| o.seller)
.map(|o| o.tenant.0.clone())
.collect()
}
/// What each product's seller-own line last said in this process: the line is written once
/// per start, and again only when the list changes.
static SELLERS_OWN_SAID: std::sync::Mutex<std::collections::BTreeMap<String, Vec<String>>> =
std::sync::Mutex::new(std::collections::BTreeMap::new());
/// **Every person's account at a selling shop holds a fact** (2026-10-08). The product says
/// who each tenant is ([`monetize_product::Product::tenant_owners`]); each person's account
/// with no fact on the ledger and none granted by a sign-up is given the sign-up's account
/// fact, signed and pushed (a failed push is retried like any other). The seller's own
/// namespaces are named in the log, once per start. A product that cannot say, or a shop
/// that sells nothing (a sold estate), issues nothing; nor does the ONE box to a sold estate's
/// tenant it serves ([`sells_room_to`], HENRIK-1).
fn give_account_facts(
state: &AppState,
product: monetize_product::Served<'_>,
granted: &std::collections::BTreeMap<String, EntitlementFact>,
now: u64,
t: &mut Tick,
) {
if !sells_room(state, product.id()) {
return;
}
let owners = match product.tenant_owners() {
Ok(Some(owners)) => owners,
Ok(None) => return,
Err(e) => {
eprintln!("monetize-server: {} tenant_owners: {e}", product.id());
return;
}
};
let own = sellers_own(&owners);
{
let mut said = SELLERS_OWN_SAID.lock().unwrap_or_else(|p| p.into_inner());
if said.get(product.id()) != Some(&own) {
eprintln!(
"monetize-server: {}: the seller's own namespaces (their owner holds a server grant; no account fact is given them): {}",
product.id(),
if own.is_empty() { "none".to_string() } else { own.join(", ") }
);
said.insert(product.id().to_string(), own);
}
}
let has_fact = |tenant: &str| {
granted.contains_key(tenant)
|| !matches!(state.ledger.entitlement(product.id(), tenant), Ok(None))
};
for tenant in owed_account_facts(&owners, has_fact) {
if !sells_room_to(state, product.id(), &tenant.0) {
continue;
}
match give_account_fact(state, product, &tenant, now) {
Ok(pushed) => {
t.accounts_given += 1;
if pushed.is_err() {
t.push_failures += 1;
}
}
Err(e) => eprintln!("monetize-server: {e}"),
}
}
}
/// **The shop's account fact, given to `tenant`** — signed, stored and pushed (a failed push
/// is stored and retried like any other: `Ok(Err(why))`).
fn give_account_fact(
state: &AppState,
product: monetize_product::Served<'_>,
tenant: &monetize_product::TenantId,
now: u64,
) -> Result<Result<(), String>, String> {
let fact = crate::signup::account_fact(
state.signer_for(product.id()),
product.id(),
tenant.clone(),
now,
state.signup_plans.cap_meter_of(product.id()),
crate::signup::GIVEN_VENDOR,
);
let pushed = state.apply_fact(product, &fact).map_err(|e| {
format!(
"{}/{}: the account fact was not stored: {e}",
product.id(),
tenant.0
)
})?;
if let Err(e) = &pushed {
eprintln!(
"monetize-server: {}/{}: the account fact is stored and its push failed (retried): {e}",
product.id(),
tenant.0
);
}
skidbladnir::functional_status(
"monetize-server/poller",
"account-fact-given",
true,
&format!("{}/{}: a person's account held no fact (uncapped); it now holds the shop's account fact (Free, no room)", product.id(), tenant.0),
);
Ok(pushed)
}
/// **A signup-granted tenant on the ledger NOW** — the tick's ADOPTION for one tenant, at once
/// (the buy trials on `linked()`, 2026-10-24): a raise is resolved against the ledger
/// (`services::purchase::caps_of_bundle`), and an account made since the last tick held its
/// fact in the signup store only, while the site's product already held it — the site's
/// ticket (held + delta) and the box's caps differed, and the raise was refused. Seeded with
/// `set_entitlement`, as the tick does: the product already holds this exact fact. `Ok(true)`:
/// adopted now; a tenant on the ledger already, or with no granted signup, is left as it is.
pub fn adopt_signup(state: &AppState, product: &str, tenant: &str) -> Result<bool, String> {
if !matches!(state.ledger.entitlement(product, tenant), Ok(None)) {
return Ok(false);
}
let granted = crate::signup::granted_facts(state, product)
.map_err(|e| format!("{product}: signup facts: {e}"))?;
let Some(fact) = granted.into_iter().find(|f| f.tenant.0 == tenant) else {
return Ok(false);
};
state
.ledger
.set_entitlement(product, &fact)
.map_err(|e| format!("adopt {product}/{tenant}: {e}"))?;
skidbladnir::functional_status(
"monetize-server/poller",
"signup-adopted",
true,
&format!("{product}/{tenant}: the signup's entitlement is on the ledger before a raise; the ladder walks it from now on"),
);
Ok(true)
}
/// **Before a sale asks the product about an account, the account holds a fact** — the
/// poller's rule for ONE tenant, at once ("Add users" asks gunnar whether the users fit, and an
/// account with no fact fits everything). `Ok(true)`: it was given one now. A shop that sells
/// nothing, a product that cannot say, the seller's own and a tenant that already holds a
/// fact are left as they are.
pub fn account_holds_a_fact(
state: &AppState,
product: monetize_product::Served<'_>,
tenant: &str,
) -> Result<bool, String> {
if !sells_room_to(state, product.id(), tenant)
|| !matches!(state.ledger.entitlement(product.id(), tenant), Ok(None))
{
return Ok(false);
}
let granted = crate::signup::granted_facts(state, product.id())
.map_err(|e| format!("{}: signup facts: {e}", product.id()))?;
if granted.iter().any(|f| f.tenant.0 == tenant) {
return Ok(false);
}
let Some(owners) = product
.tenant_owners()
.map_err(|e| format!("{} tenant_owners: {e}", product.id()))?
else {
return Ok(false);
};
let one: Vec<monetize_product::TenantOwner> = owners
.into_iter()
.filter(|o| o.tenant.0 == tenant)
.collect();
match owed_account_facts(&one, |_| false).first() {
Some(owed) => give_account_fact(state, product, owed, state.now()).map(|_| true),
None => Ok(false),
}
}
/// **The tick's one log line.** Counts are APPENDED, never reordered: a
/// deployment check greps `poll tenants=N failures=0`, and a count inserted in
/// the middle would break it.
pub fn poll_line(t: &Tick) -> String {
format!(
"poll tenants={} failures={} changes={} push_failures={} resumed={} signups_settled={} pushes_retried={} reconciled={} unfulfilled={} signups_adopted={} ladder_superseded={} owed_back={} metered_unproven={} accounts_given={}",
t.tenants_read,
t.read_failures,
t.state_changes,
t.push_failures,
t.resumed,
t.signups_settled,
t.pushes_retried,
t.reconciled,
t.unfulfilled,
t.signups_adopted,
t.ladder_superseded,
// `?`, never 0: a count that could not be taken is not a count of none.
t.owed_back.map_or("?".to_string(), |n| n.to_string()),
t.metered_unproven,
t.accounts_given
)
}
/// The loop: a tick every `every`, until the runtime stops.
pub async fn run(state: Arc<AppState>, every: Duration) {
let mut interval = tokio::time::interval(every);
loop {
interval.tick().await;
let s = state.clone();
match tokio::task::spawn_blocking(move || tick(&s)).await {
Ok(t) => {
eprintln!("monetize-server: {}", poll_line(&t));
if let Some(line) = nonpay_line(&t) {
eprintln!("monetize-server: {line}");
}
}
Err(e) => eprintln!("monetize-server: poller panicked: {e}"),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use monetize::testing::MockProduct;
use monetize_product::Servable;
use std::sync::Arc;
use crate::registry::Registry;
use crate::state::AppState;
const NOW: u64 = 1_788_523_200_000;
fn state(reg: Registry) -> AppState {
let mut s = AppState::in_memory(reg);
s.clock = Box::new(|| NOW);
s
}
/// The signed fact an operator hands the product: no `paid_until`, so the
/// ladder never moves it and nothing but the retry touches it.
fn operator_fact(state: &AppState, tenant: &str, cap: u64) -> EntitlementFact {
state.signer_for("gunnar").sign_fact(EntitlementFact {
tenant: monetize_product::TenantId(tenant.into()),
plan: "trial".into(),
state: monetize_product::State::Paid,
paid_until_unix_ms: None,
caps: [("pack_bytes".to_string(), cap)].into(),
source: "operator".into(),
signature: Vec::new(),
issued_unix_ms: None,
issued_signature: Vec::new(),
})
}
/// A state whose product double the test keeps a handle to, so it can be
/// brought back up between ticks.
fn state_with(product: Arc<MockProduct>) -> AppState {
state(Registry::new().shared_product(product))
}
/// **A push that failed is retried until the product acknowledges it.**
///
/// RED before 2026-09-14: the fact was stored, the push failed, and nothing
/// ever pushed it again — the ladder only moves a fact whose derived state
/// differs from the stored one, and the stored one had already moved. gunnar
/// could miss a cap raise for good while every monetize screen said it had it.
#[test]
fn a_failed_push_stays_pending_and_the_next_tick_pushes_it_until_it_lands() {
let product = Arc::new(MockProduct::gunnar());
*product.fail_push.lock().unwrap() = true;
let s = state_with(product.clone());
let fact = operator_fact(&s, "alice", 100 << 30);
let pushed = s
.apply_fact(s.registry.products().next().unwrap(), &fact)
.unwrap();
assert!(pushed.is_err(), "the product was down");
// The FACT is stored — monetize's own answer has moved — and so is the
// to-do, which is the half that was missing.
assert_eq!(
s.ledger.entitlement("gunnar", "alice").unwrap().as_ref(),
Some(&fact)
);
let pending = s
.ledger
.pending_push("gunnar", "alice")
.unwrap()
.expect("a failed push leaves a pending push");
assert_eq!(pending.attempts, 1);
assert!(
pending.last_error.contains("mock product down"),
"{}",
pending.last_error
);
assert_eq!(pending.since_unix_ms, NOW);
// A tick against a product that is STILL down counts an attempt and
// keeps it pending — no give-up: a lapse must land when the box is back.
let t = tick(&s);
assert_eq!(t.pushes_retried, 0);
assert_eq!(
s.ledger
.pending_push("gunnar", "alice")
.unwrap()
.unwrap()
.attempts,
2
);
assert!(
product.pushed().is_empty(),
"nothing landed while it was down"
);
// …and the tick after it comes back pushes the fact and clears the row.
*product.fail_push.lock().unwrap() = false;
let t = tick(&s);
assert_eq!(t.pushes_retried, 1, "the fact must reach the product");
assert_eq!(
product.pushed(),
vec![fact],
"the SAME signed fact, unchanged"
);
assert!(
s.ledger.pending_push("gunnar", "alice").unwrap().is_none(),
"acknowledged: nothing left to do"
);
// Idempotent: the next tick pushes nothing again.
let before = product.push_attempts().len();
let t = tick(&s);
assert_eq!(
(t.pushes_retried, product.push_attempts().len()),
(0, before)
);
}
/// **R5.1-F: a push that keeps failing is ALERTed and never given up.** A
/// paid fact that has not reached the product for [`PUSH_ALERT_AFTER`]
/// ticks is a customer without the room they paid for: the tick counts it
/// (and the front's log says ALERT), it is STILL pushed every tick, and it
/// lands when the product is back.
#[test]
fn a_push_failing_for_many_ticks_is_alerted_and_still_pushed_until_it_lands() {
let product = Arc::new(MockProduct::gunnar());
*product.fail_push.lock().unwrap() = true;
let s = state_with(product.clone());
let fact = operator_fact(&s, "alice", 100 << 30);
assert!(s
.apply_fact(s.registry.products().next().unwrap(), &fact)
.unwrap()
.is_err());
// attempts 1 after the first push; every tick adds one.
for attempts in 2..PUSH_ALERT_AFTER {
let t = tick(&s);
assert_eq!(
s.ledger
.pending_push("gunnar", "alice")
.unwrap()
.unwrap()
.attempts,
attempts
);
assert_eq!(
t.pushes_stuck, 0,
"below the threshold ({attempts} attempts) nothing is stuck yet"
);
}
for attempts in PUSH_ALERT_AFTER..PUSH_ALERT_AFTER * 2 + 1 {
let t = tick(&s);
let p = s.ledger.pending_push("gunnar", "alice").unwrap().unwrap();
assert_eq!(p.attempts, attempts, "still asked every tick — no give-up");
assert_eq!(
t.pushes_stuck, 1,
"{attempts} failed pushes: the tick says it is stuck"
);
}
*product.fail_push.lock().unwrap() = false;
let t = tick(&s);
assert_eq!(
(t.pushes_retried, t.pushes_stuck),
(1, 0),
"back up: it lands and nothing is stuck"
);
assert_eq!(product.pushed(), vec![fact]);
}
/// The log line's counts are appended, never reordered: a deployment check
/// greps the head of it. The reconcile counts are the newest and sit at
/// the END, behind `pushes_retried=`, which keeps its place.
#[test]
fn the_poll_line_keeps_its_old_head_and_gains_the_reconcile_counts_at_the_end() {
let line = poll_line(&Tick {
tenants_read: 3,
pushes_retried: 2,
reconciled: 7,
unfulfilled: 1,
..Tick::default()
});
assert!(line.starts_with("poll tenants=3 failures=0"), "{line}");
assert!(
line.contains(" pushes_retried=2 "),
"the previous tail keeps its place: {line}"
);
assert!(line.contains(" reconciled=7 unfulfilled=1 "), "{line}");
// The growth's count went with the appliance grow drive (2026-11-02),
// as the reclamation queue's two went with the queue (owner 2026-10-11).
assert!(line.contains(" unfulfilled=1 signups_adopted=0 "), "{line}");
assert!(
line.contains(" signups_adopted=0 "),
"the previous tail keeps its place: {line}"
);
assert!(
line.contains(" ladder_superseded=0 "),
"the previous tail keeps its place: {line}"
);
// The two newest counts sit at the very end in the order they were
// added — appended, never inserted, because a deployment check greps
// this line.
assert!(
line.contains(" owed_back=? "),
"a tick nobody counted says so: {line}"
);
assert!(line.contains(" metered_unproven=0 "), "{line}");
assert!(
line.ends_with(" accounts_given=0"),
"the newest count sits at the very end: {line}"
);
// A quiet tick reads as one: nothing examined, nothing unfulfilled.
assert!(poll_line(&Tick { owed_back: Some(0), ..Tick::default() }).ends_with(
"reconciled=0 unfulfilled=0 signups_adopted=0 ladder_superseded=0 owed_back=0 metered_unproven=0 accounts_given=0"
));
}
/// (r) **The reachability dot is written by the poller, and a product that
/// does not answer goes DOWN with its own reason — while a product nobody
/// has polled stays UNKNOWN.**
///
/// Three states, and the third is the point: on a freshly started server the
/// dot must not claim either way. `UI.md` screen 1 renders `○ holger — not
/// connected` for a product that refused, and that has to be a different
/// pixel from a product nothing has asked about yet.
#[test]
fn a_tick_records_up_for_a_product_that_answers_and_down_for_one_that_does_not() {
// The holger plugin pointed at a port nothing listens on: its
// `list_tenants` is a real `Admin.ListRepositories` that cannot connect,
// which is `Unavailable` and exactly the "not connected" row. (It was a
// stub answering `Unavailable` to everything until 2026-09-18; the row
// this test asserts on is the same either way, which is the point of
// asserting on the row.)
let reg = Registry::new()
.product(MockProduct::gunnar())
.product(crate::testing::DownProduct("holger", "127.0.0.1:1"));
let s = state(reg);
// BEFORE the tick: nothing has been asked, and that is its own state.
assert_eq!(s.health.get("gunnar").reachability.name(), "unknown");
assert_eq!(s.health.get("holger").reachability.name(), "unknown");
assert_eq!(s.health.get("gunnar").checked_at_unix_ms, 0);
let t = tick(&s);
assert_eq!(
t.read_failures, 1,
"exactly one product failed to list its tenants"
);
let gunnar = s.health.get("gunnar");
assert_eq!(gunnar.reachability, crate::health::Reachability::Up);
assert_eq!(
(gunnar.checked_at_unix_ms, gunnar.last_up_unix_ms),
(NOW, NOW)
);
let holger = s.health.get("holger");
assert_eq!(holger.reachability.name(), "down");
// The dot carries the PLUGIN's own sentence, whatever it happens to be
// — here a real transport failure against a port nothing listens on. It
// was the stub's `not wired` until 2026-09-18; what this asserts is that
// the reason reaches the dot, not which reason it was.
assert!(
holger.reachability.detail().contains("127.0.0.1:1"),
"the dot carries the plugin's own reason, naming the box: {}",
holger.reachability.detail()
);
assert_eq!(holger.checked_at_unix_ms, NOW, "it WAS asked");
assert_eq!(holger.last_up_unix_ms, 0, "and it has never answered");
assert_ne!(gunnar.reachability, holger.reachability);
// A product that is down is NOT then asked for its capacity: that would
// cost a second deadline every tick against a box that is not there.
assert!(
s.health.get("holger").servable.is_none(),
"a down product is not interrogated further"
);
assert_eq!(s.health.get("holger").servable_at_unix_ms, 0);
// The one that answered was asked, and its answer is cached for the
// fleet total — Unmeasured here, because the double measures nothing.
assert!(
matches!(gunnar.servable, Some(Servable::Unmeasured(_))),
"{:?}",
gunnar.servable
);
assert_eq!(gunnar.servable_at_unix_ms, NOW);
}
}
/// **The paid-signup half of the ladder.**
///
/// A paid signup writes NO ledger row — `crate::signup`'s header says why, and
/// the entitlement it granted is folded out of the signup store at read time
/// (`signup_facts`). That fold is what the console draws.
///
/// It was not what the POLLER walked. `poll_product` asked
/// `ledger.entitlement(product, tenant)` and `continue`d on `None`, so a tenant
/// whose only entitlement came from a paid signup was skipped on every tick,
/// forever:
///
/// * the ladder never moved them off Paid;
/// * no lapse was ever signed or pushed, and gunnar does no clock arithmetic of
/// its own (`gunnar-server/src/entitlement.rs`: *"the ladder moves when
/// monetize pushes the next fact"*), so the appliance kept serving the 30 days
/// the customer paid for as though they were forever;
/// * and the reclamation queue never saw them, so the one record of who is
/// standing on unpaid iron had a hole in it exactly the shape of the shop's
/// own customers.
#[cfg(test)]
mod signup_ladder_tests {
use super::*;
use monetize::testing::MockProduct;
use monetize_pay::signup::PaidSignup;
use std::sync::Arc;
use crate::registry::Registry;
use crate::signup::tests::{a_signup, plans};
use crate::signup::SignupState;
use crate::state::AppState;
const DAY: u64 = 24 * 60 * 60 * 1000;
/// The state a shop deployment is in the moment a paid signup has landed:
/// bob exists on the appliance, holds the plan's cap, and the LEDGER HAS NO
/// ROW FOR HIM — which is the deployment, not a shortcut for the test.
fn a_shop_with_one_paid_signup() -> (AppState, Arc<MockProduct>, u64) {
let product = Arc::new(MockProduct::gunnar());
let mut s = AppState::in_memory(Registry::new().shared_product(product.clone()));
s.signup_plans = plans().into();
let signup: PaidSignup = a_signup("cs_bob", "bob");
let paid_at = signup.paid_at_unix_ms as u64;
s.clock = Box::new(move || paid_at);
s.signups
.record_pending(&signup, signup.paid_at_unix_ms)
.unwrap();
s.signups
.set(
"mockpay",
"cs_bob",
SignupState::Created {
namespace: "bob".into(),
username: "bob".into(),
at_unix_ms: signup.paid_at_unix_ms,
},
)
.unwrap();
// The sink pushed the signed fact to the appliance and wrote nothing to
// the ledger — `monetize_signup_impl::GunnarSignup::paid` step (c).
let fact = crate::signup::signup_facts(&s, "gunnar")
.unwrap()
.pop()
.expect("the signup granted an entitlement");
assert_eq!(fact.state, monetize_product::State::Paid);
product.push_entitlement(&fact).unwrap();
assert!(
s.ledger.entitlement("gunnar", "bob").unwrap().is_none(),
"a signup writes no ledger row: that is the deployment"
);
let paid_until = fact.paid_until_unix_ms.unwrap();
(s, product, paid_until)
}
/// **A signup customer whose 30 days ran out is suspended and pushed —
/// not served forever** (nonpay purges them after the grace).
///
/// RED before this lane: `tick` walked past bob on every pass. gunnar held
/// his Paid fact and nothing ever replaced it, so one month's money bought
/// unbounded service; the reclamation queue, whose whole job is to say who
/// stands on unpaid iron, never heard of him.
#[test]
fn a_paid_signup_that_lapses_is_moved_down_the_ladder_and_pushed() {
let (mut s, product, paid_until) = a_shop_with_one_paid_signup();
// A day into grace: still served — grace is the period whose whole
// purpose is that nothing happens.
let in_grace = paid_until + DAY;
s.clock = Box::new(move || in_grace);
tick(&s);
assert_eq!(
product.pushed().last().unwrap().state,
monetize_product::State::Grace,
"the lapse into grace must reach the appliance"
);
// Grace is over (14 days, the deployed policy). The customer paid for
// one month and has had six weeks.
let suspended = paid_until + 15 * DAY;
s.clock = Box::new(move || suspended);
let t = tick(&s);
assert_eq!(t.state_changes, 1, "the ladder must move a signup tenant");
assert_eq!(t.push_failures, 0);
let last = product
.pushed()
.last()
.cloned()
.expect("the appliance must be told");
assert_eq!(
(last.tenant.0.as_str(), last.state),
("bob", monetize_product::State::Suspended),
"gunnar does no clock arithmetic: the lapse only exists if monetize pushes it"
);
assert_eq!(
last.source, LADDER_SOURCE,
"the clock did it, not a person and not a payment"
);
// Idempotent: the tick after changes nothing and pushes nothing again.
let pushes = product.push_attempts().len();
let t = tick(&s);
assert_eq!(
(t.state_changes, product.push_attempts().len()),
(0, pushes)
);
}
}
/// **The tick and a payment, in the same second.**
#[cfg(test)]
mod ladder_race_tests {
use super::*;
use monetize::testing::MockProduct;
use monetize_product::{State, TenantId};
use std::sync::Arc;
use crate::registry::Registry;
use crate::state::AppState;
const DAY: u64 = 24 * 60 * 60 * 1000;
const NOW: u64 = 1_788_523_200_000;
/// ★ **A TENANT WHO PAID IS NOT SUSPENDED BY THE TICK THAT WAS ALREADY
/// DECIDING ABOUT THEM.**
///
/// `poll_product` read the stored fact, called `ladder_move` on it, and
/// then `apply_fact` — which PUSHES first and STORES second, and
/// `Ledger::set_entitlement` is a blind overwrite. A push is a gRPC call to
/// an appliance with a 30 s deadline, so the gap between the read and the
/// write is seconds wide, not microseconds, and it is wide open:
///
/// the tick reads bob at Grace, paid_until = yesterday ·
/// bob's card clears and the purchase stores Paid, paid_until = next
/// month, and pushes it ·
/// the tick finishes and overwrites BOTH with Suspended carrying
/// YESTERDAY's paid_until.
///
/// And it does not heal: the next tick derives Suspended from that stale
/// paid_until, agrees with itself, and moves nothing, forever. The money
/// was captured, the ledger's purchase row is there, and the customer is
/// locked out with no row anywhere saying why. This is the worst shape a
/// bug in this repository has: a tenant suspended who paid.
///
/// The window is made deterministic by `MockProduct::during_push` — the
/// payment lands while the tick's push is in flight, which is exactly where
/// a real one lands.
#[test]
fn a_payment_that_lands_during_a_tick_is_not_overwritten_by_the_ladder() {
let product = Arc::new(MockProduct::gunnar());
let mut state = AppState::in_memory(Registry::new().shared_product(product.clone()));
state.clock = Box::new(|| NOW);
let state = Arc::new(state);
// bob is one day into grace: the tick WILL want to move him, later.
let lapsed = NOW - 15 * DAY;
let grace = state
.signer_for("gunnar")
.sign_fact(monetize_product::EntitlementFact {
tenant: TenantId("bob".into()),
plan: "gunnar/bob/2026-08-01".into(),
state: State::Grace,
paid_until_unix_ms: Some(lapsed),
caps: [("pack_bytes".to_string(), 10u64 << 30)].into(),
source: LADDER_SOURCE.into(),
signature: Vec::new(),
issued_unix_ms: None,
issued_signature: Vec::new(),
});
state
.apply_fact(state.registry.products().next().unwrap(), &grace)
.unwrap()
.unwrap();
// THE PAYMENT, armed to land while the tick's push is in flight.
let renewed_until = NOW + 30 * DAY;
let weak = Arc::downgrade(&state);
*product.during_push.lock().unwrap() = Some(Box::new(move || {
let s = weak.upgrade().expect("the state outlives the tick");
let paid = s
.signer_for("gunnar")
.sign_fact(monetize_product::EntitlementFact {
state: State::Paid,
paid_until_unix_ms: Some(renewed_until),
source: "payment:stripe:cs_bob".into(),
signature: Vec::new(),
..s.ledger.entitlement("gunnar", "bob").unwrap().unwrap()
});
s.ledger
.set_entitlement("gunnar", &paid)
.expect("the purchase stores its fact");
}));
tick(&state);
// THE ASSERTION THAT IS ABOUT MONEY.
let held = state
.ledger
.entitlement("gunnar", "bob")
.unwrap()
.expect("bob has a fact");
assert_eq!(
held.state,
State::Paid,
"the tick suspended a tenant who had just paid"
);
assert_eq!(
held.paid_until_unix_ms,
Some(renewed_until),
"and threw away the period they paid for"
);
assert!(
held.source.starts_with("payment:"),
"the payment's fact must win, not the ladder's: {}",
held.source
);
// And it is stable: the next tick agrees and moves nothing.
let t = tick(&state);
assert_eq!(t.state_changes, 0, "a paid tenant must not be moved");
assert_eq!(
state
.ledger
.entitlement("gunnar", "bob")
.unwrap()
.unwrap()
.state,
State::Paid
);
}
}
/// **The health line's owed-back count.** The queue is a page somebody has to
/// open; this is the half that reaches the journal a deployment check greps.
#[cfg(test)]
mod owed_back_tests {
use super::*;
use monetize::testing::MockProduct;
use std::sync::Arc;
use crate::registry::Registry;
use crate::signup::tests::{a_signup, plans};
use crate::signup::{settle, SignupState};
use crate::state::AppState;
/// (r) **A payment taken for an account that was never created is COUNTED
/// ON THE POLL LINE**, so it is not only a row on a console page.
///
/// The count is the STANDING queue and not this tick's new refusals: the
/// failure being broken is a refusal from months ago that nobody has seen
/// since, and a per-tick delta would print 0 for exactly that case. The
/// suffix is asserted verbatim because a deployment check greps this line.
#[test]
fn the_poll_line_counts_the_payments_that_are_owed_back() {
let product = Arc::new(MockProduct::gunnar());
let mut s = AppState::in_memory(Registry::new().shared_product(product));
s.signup_plans = plans().into();
s.clock = Box::new(|| 1_788_436_800_000);
// A quiet deployment says so — and says 0, not `?`.
let quiet = tick(&s);
assert_eq!(quiet.owed_back, Some(0));
assert!(
poll_line(&quiet).contains(" owed_back=0 "),
"{}",
poll_line(&quiet)
);
// One refusal. The money moved; no account exists.
let mut refused = a_signup("cs_refused", "bob");
refused.plan = "gunnar/1tb-forever".into();
assert!(matches!(settle(&s, &refused), SignupState::Refused { .. }));
let after = tick(&s);
assert_eq!(after.owed_back, Some(1));
assert!(
poll_line(&after).contains(" owed_back=1 "),
"{}",
poll_line(&after)
);
// A tick LATER — nothing new refused — still says 1: the queue stands
// until a human clears it, and a count that fell back to 0 would be the
// silence this whole lane exists to break.
assert_eq!(tick(&s).owed_back, Some(1));
// A CREATED signup does not join it.
settle(&s, &a_signup("cs_made", "ada"));
assert!(matches!(
s.signups.get("mockpay", "cs_made").unwrap().unwrap().state,
SignupState::Created { .. }
));
assert_eq!(
tick(&s).owed_back,
Some(1),
"an account that WAS created is not money owed back"
);
// A count that could not be taken prints `?` and never 0.
let unknown = Tick {
owed_back: None,
..Default::default()
};
assert!(
poll_line(&unknown).contains(" owed_back=? "),
"{}",
poll_line(&unknown)
);
}
}
/// **The paid door, probed.** Both arms: a box that answers a metered verb and
/// one that answers `UNIMPLEMENTED`.
#[cfg(test)]
mod metered_door_tests {
use super::*;
use monetize::testing::MockProduct;
use monetize_product::MeteredDoor;
use std::sync::Arc;
use crate::registry::Registry;
use crate::signup::tests::plans;
use crate::state::AppState;
fn shop(door: Option<MeteredDoor>, sells: bool) -> (AppState, Arc<MockProduct>) {
let product = Arc::new(MockProduct::gunnar());
*product.metered_door.lock().unwrap() = door;
let mut s = AppState::in_memory(Registry::new().shared_product(product.clone()));
if sells {
s.signup_plans = plans().into();
}
s.clock = Box::new(|| 1_788_436_800_000);
(s, product)
}
/// (r) **A COMMUNITY APPLIANCE IS COUNTED AND VISIBLE, AND A METERED ONE IS
/// NOT.**
///
/// `poll tenants=11 failures=0` was the whole proof this estate had, and
/// every call behind it — `Repos.ListNamespaces`, `Usage.Read` — is an
/// OPEN-build service that a community ISO answers perfectly. So a box that
/// cannot enforce a single entitlement polled exactly like one that can,
/// and the first thing to find out otherwise was a paying customer
/// (`Entitlement.Set` → UNIMPLEMENTED → a refusal with the money taken).
///
/// Asserted on both arms and on the health cache, because a probe that
/// counted everything — or nothing — would satisfy either half alone.
#[test]
fn a_community_appliance_is_counted_on_the_poll_line_and_a_metered_one_is_not() {
// THE PAID BUILD: nothing to report, nothing counted.
let (paid, _p) = shop(
Some(MeteredDoor::Enforcing(
"Entitlement.List answered (2 facts)".into(),
)),
true,
);
let t = tick(&paid);
assert_eq!(t.metered_unproven, 0);
assert!(
poll_line(&t).contains(" metered_unproven=0 "),
"{}",
poll_line(&t)
);
assert_eq!(
paid.health.get("gunnar").metered_door_or_unasked().name(),
"enforcing"
);
assert!(paid.health.get("gunnar").metered_door_or_unasked().proven());
assert_eq!(
paid.health.get("gunnar").metered_door_at_unix_ms,
1_788_436_800_000
);
// THE COMMUNITY ISO: counted, and the cache carries the box's own words
// so an operator reads what answered rather than a paraphrase.
let (open, _p) = shop(
Some(MeteredDoor::Absent(
"gunnar Entitlement.Set -> Unimplemented".into(),
)),
true,
);
let t = tick(&open);
assert_eq!(t.metered_unproven, 1);
assert!(
poll_line(&t).contains(" metered_unproven=1 "),
"{}",
poll_line(&t)
);
let door = open.health.get("gunnar").metered_door_or_unasked();
assert_eq!(door.name(), "absent");
assert!(
!door.proven(),
"a community build must never read as proven"
);
assert_eq!(door.detail(), "gunnar Entitlement.Set -> Unimplemented");
// **A BOX NOBODY COULD ASK IS NOT A YES EITHER.** The plugin with no
// probe, and the box that is down, both land here — and both are
// counted, because a deployment that sells may not call itself healthy
// on the strength of a question nobody answered.
let (silent, _p) = shop(None, true);
assert_eq!(tick(&silent).metered_unproven, 1);
let door = silent.health.get("gunnar").metered_door_or_unasked();
assert_eq!(door.name(), "unproven");
assert!(!door.proven());
// **A box nobody can reach proves nothing** — and is not asked twice.
// holger's plugin pointed at a dead port: its catalog call cannot
// connect, which is the down product this loop already skips the
// capacity question for.
let mut down = AppState::in_memory(
Registry::new().product(crate::testing::DownProduct("holger", "127.0.0.1:1")),
);
down.signup_plans = plans().into();
let t = tick(&down);
assert_eq!(
(t.read_failures, t.metered_unproven),
(1, 1),
"an unreachable box is a read failure AND an unproven door"
);
let door = down.health.get("holger").metered_door_or_unasked();
assert_eq!(door.name(), "unproven");
assert!(
door.detail().contains("did not answer at all"),
"{}",
door.detail()
);
// And a server that has not polled at all says so rather than claiming
// either answer — the state every fresh process is in.
let (fresh, _p) = shop(Some(MeteredDoor::Enforcing("x".into())), true);
let never = fresh.health.get("gunnar");
assert!(never.metered_door.is_none() && never.metered_door_at_unix_ms == 0);
assert_eq!(never.metered_door_or_unasked().name(), "unproven");
assert!(never
.metered_door_or_unasked()
.detail()
.contains("has not yet asked"));
}
/// (r) **A deployment that SELLS against a box that cannot enforce is
/// UNHEALTHY; one that sells nothing is merely told.**
///
/// The distinction is the whole reason this is not fatal at start: the open
/// build against a community appliance is a correct deployment, and taking
/// it down — or painting it red for ever — would teach an operator to
/// ignore the row that matters. `[signups.<product>.plans]` non-empty is what "sells"
/// means here, because that is what turns an `Unimplemented` into a charged
/// card.
///
/// Asserted on the count and on the deployment's own state, since
/// `functional_status` is a sink this crate cannot read back.
#[test]
fn selling_against_a_community_appliance_is_the_unhealthy_case_and_not_selling_is_not() {
let absent = || MeteredDoor::Absent("gunnar Entitlement.Set -> Unimplemented".into());
let (selling, _p) = shop(Some(absent()), true);
assert!(
!selling.signup_plans.is_empty(),
"this deployment takes money for accounts"
);
assert_eq!(tick(&selling).metered_unproven, 1);
// The same box under a deployment that sells nothing: still counted —
// the fact is the fact — and not the deployment's emergency.
let (free, _p) = shop(Some(absent()), false);
assert!(free.signup_plans.is_empty());
assert_eq!(tick(&free).metered_unproven, 1);
assert_eq!(
free.health.get("gunnar").metered_door_or_unasked().name(),
"absent"
);
}
}
/// **Every person's account at a selling shop holds a fact** (2026-10-08, the free-room leak:
/// an account an invite, an operator or a failed sign-up made held none, and gunnar read it
/// as UNCAPPED).
#[cfg(test)]
mod account_facts {
use super::*;
use monetize::testing::MockProduct;
use monetize_product::{State, TenantId, TenantOwner};
use std::sync::Arc;
use crate::registry::Registry;
use crate::state::AppState;
fn owner(tenant: &str, person: bool, seller: bool) -> TenantOwner {
TenantOwner {
tenant: TenantId(tenant.into()),
person,
seller,
}
}
fn a_shop(owners: Vec<TenantOwner>) -> (AppState, Arc<MockProduct>) {
let product = Arc::new(MockProduct::gunnar());
*product.owners.lock().unwrap() = Some(owners);
let s = AppState::in_memory(Registry::new().shared_product(product.clone()));
(s, product)
}
/// ★ **Only a person's account with no fact is owed one; the seller's own never is**, and a
/// normal person is never the seller's: only the product's `seller` word (its owner holds a
/// server grant) leaves a person out — no fact, no name, nothing else does.
#[test]
fn a_person_with_no_fact_is_owed_the_account_fact_and_only_the_sellers_own_are_left_out() {
let owners = [
owner("invited", true, false),
owner("paid", true, false),
owner("nordisk", true, true),
owner("team", false, false),
owner("ops", false, true),
];
let owed = owed_account_facts(&owners, |t| t == "paid");
assert_eq!(
owed,
vec![TenantId("invited".into())],
"the person with no fact, and only them"
);
assert_eq!(
sellers_own(&owners),
vec!["nordisk".to_string(), "ops".to_string()]
);
// A normal person — whatever their name, whatever they hold — is owed one while they
// hold no fact: the exemption is the product's word, never a name the shop knows.
for name in [
"nordisk-customer",
"operator",
"admin",
"gunnar",
"monetize",
] {
assert_eq!(
owed_account_facts(&[owner(name, true, false)], |_| false),
vec![TenantId(name.into())],
"{name} is a person like any other"
);
}
assert!(
owed_account_facts(&owners, |_| true).is_empty(),
"a ledger that cannot be read issues nothing"
);
}
/// ★ **The tick gives the account fact — Free, no end, no room — and pushes it**; the
/// seller's own and a group get nothing; the next tick gives nothing twice.
#[test]
fn the_tick_gives_a_person_with_no_fact_the_shops_account_fact_once() {
let (s, product) = a_shop(vec![
owner("invited", true, false),
owner("nordisk", true, true),
owner("team", false, false),
]);
assert!(
sells_room(&s, "gunnar"),
"the fixture shop is priced and not free"
);
let t = tick(&s);
assert_eq!(t.accounts_given, 1, "{}", poll_line(&t));
let fact = s
.ledger
.entitlement("gunnar", "invited")
.unwrap()
.expect("the person's account holds a fact");
assert_eq!(
(
fact.state,
fact.paid_until_unix_ms,
fact.caps.get("pack_bytes").copied()
),
(State::Free, None, Some(crate::signup::NO_ROOM_CAP)),
"the sign-up's account fact: Free, no end, no room"
);
assert!(
fact.source.starts_with("account:given:"),
"its attestation says it was given: {}",
fact.source
);
assert!(
product.pushed().iter().any(|f| f.tenant.0 == "invited"
&& f.caps.get("pack_bytes") == Some(&crate::signup::NO_ROOM_CAP)),
"gunnar is told"
);
for none in ["nordisk", "team"] {
assert!(
s.ledger.entitlement("gunnar", none).unwrap().is_none(),
"{none} is given nothing"
);
}
assert_eq!(tick(&s).accounts_given, 0, "a fact is given once");
}
/// ★ **"Add users" never asks about an account with no fact**: the sale gives it the
/// account fact first (so gunnar answers with a cap, and a shortfall opens a checkout);
/// the seller's own and an account that holds a fact are left as they are.
#[test]
fn a_sale_gives_the_account_its_fact_before_it_asks() {
let (s, product) = a_shop(vec![
owner("invited", true, false),
owner("nordisk", true, true),
]);
assert!(
account_holds_a_fact(&s, product.served(), "invited").unwrap(),
"given now"
);
assert!(s.ledger.entitlement("gunnar", "invited").unwrap().is_some());
assert!(
!account_holds_a_fact(&s, product.served(), "invited").unwrap(),
"once"
);
assert!(
!account_holds_a_fact(&s, product.served(), "nordisk").unwrap(),
"the seller's own"
);
assert!(s.ledger.entitlement("gunnar", "nordisk").unwrap().is_none());
}
/// ★ **A shop that sells nothing gives nothing**: a sold estate's own monetize is served
/// free, and its customer never sees a cap.
#[test]
fn a_free_shop_gives_no_account_fact() {
let (s, _) = a_shop(vec![owner("customer", true, false)]);
s.config
.write()
.unwrap()
.products
.entry("gunnar".into())
.or_default()
.free = true;
assert!(!sells_room(&s, "gunnar"));
assert_eq!(tick(&s).accounts_given, 0);
assert!(s
.ledger
.entitlement("gunnar", "customer")
.unwrap()
.is_none());
}
/// The fixture shop with the operator's own second site `test2-gunnar-rs` declared
/// (`operator_site = true`) and a declared customer estate `kund` (not the operator's).
fn a_shop_with_sites(owners: Vec<TenantOwner>) -> (AppState, Arc<MockProduct>) {
let product = Arc::new(MockProduct::gunnar());
*product.owners.lock().unwrap() = Some(owners);
let site = crate::config::EstateConfig {
product: "gunnar".into(),
owner: "alica".into(),
operator_site: true,
..Default::default()
};
let kund = crate::config::EstateConfig {
product: "gunnar".into(),
owner: "kund".into(),
..Default::default()
};
let s = AppState::in_memory(
Registry::new()
.shared_product(product.clone())
.estate("test2-gunnar-rs", site)
.estate("kund", kund),
);
(s, product)
}
/// ★ HENRIK-1 (prod 2026-10-26 19:52:59, `entitlement gunnar/@rickardtest/henrik: none ->
/// Free … pack_bytes=1`): **a SOLD estate's tenant is given no account fact.** The ONE box
/// sells gunnar room on its own sites, and a customer's private instance (`@rickardtest/…`,
/// laid by an estate order, or a declared estate that is not the operator's) is served
/// free: its customer never sees a cap. The operator's own sites and the home keep it.
#[test]
fn a_sold_estates_tenant_is_given_no_account_fact_and_the_operators_sites_still_are() {
let (s, product) = a_shop_with_sites(vec![
owner("@rickardtest/henrik", true, false),
owner("@kund/eve", true, false),
owner("@test2-gunnar-rs/bob", true, false),
owner("invited", true, false),
]);
let t = tick(&s);
assert_eq!(
t.accounts_given,
2,
"the home's and the operator site's person only: {}",
poll_line(&t)
);
for sold in ["@rickardtest/henrik", "@kund/eve"] {
assert!(
s.ledger.entitlement("gunnar", sold).unwrap().is_none(),
"{sold} is on a sold estate: given nothing"
);
assert!(
!product.pushed().iter().any(|f| f.tenant.0 == sold),
"{sold}: nothing is pushed to a private instance"
);
}
for kept in ["@test2-gunnar-rs/bob", "invited"] {
let fact = s
.ledger
.entitlement("gunnar", kept)
.unwrap()
.expect("the operator's own sites keep the account fact");
assert_eq!(
fact.caps.get("pack_bytes").copied(),
Some(crate::signup::NO_ROOM_CAP),
"{kept}"
);
}
}
/// ★ HENRIK-1: **a sale never gives a sold estate's tenant the account fact either**
/// ([`account_holds_a_fact`]); an operator's site's tenant is given it as before.
#[test]
fn a_sale_gives_no_account_fact_on_a_sold_estate() {
let (s, product) = a_shop_with_sites(vec![
owner("@rickardtest/henrik", true, false),
owner("@test2-gunnar-rs/bob", true, false),
]);
assert!(
!account_holds_a_fact(&s, product.served(), "@rickardtest/henrik").unwrap(),
"a sold estate: nothing given"
);
assert!(s
.ledger
.entitlement("gunnar", "@rickardtest/henrik")
.unwrap()
.is_none());
assert!(
account_holds_a_fact(&s, product.served(), "@test2-gunnar-rs/bob").unwrap(),
"the operator's site: given now"
);
}
/// ★ HENRIK-1, pure: **which estates sell room** — the home (a plain tenant) and the
/// operator's own declared sites; never an undeclared (sold) estate or a customer's.
#[test]
fn room_is_sold_on_the_home_and_the_operators_own_sites_only() {
let (s, _) = a_shop_with_sites(Vec::new());
assert!(sells_room_to(&s, "gunnar", "invited"));
assert!(sells_room_to(&s, "gunnar", "@test2-gunnar-rs/bob"));
assert!(!sells_room_to(&s, "gunnar", "@rickardtest/henrik"));
assert!(!sells_room_to(&s, "gunnar", "@kund/eve"));
s.config
.write()
.unwrap()
.products
.entry("gunnar".into())
.or_default()
.free = true;
assert!(
!sells_room_to(&s, "gunnar", "invited"),
"a free shop sells room nowhere"
);
}
}
#[cfg(test)]
mod unheld_fact_tests {
//! ★ ALICA-4 across monetize-server and monetize-guardian: the poller's tick sends a paid
//! ledger fact again to a site's main that does not hold it.
use super::*;
use monetize_embed::fleet::Report;
use monetize_product::pull::{order, ProductReport};
use monetize_product::{EntitlementFact, State, TenantId};
const GIB: u64 = 1 << 30;
fn test2_report(alica_gib: u64) -> Report {
let p = ProductReport {
promised: [
(
"bob".to_string(),
[("pack_bytes".to_string(), 3 * GIB)].into(),
),
(
"alica".to_string(),
[("pack_bytes".to_string(), alica_gib * GIB)].into(),
),
]
.into(),
names: vec!["alica".into(), "bob".into()],
..Default::default()
};
Report {
product: Some(serde_json::to_string(&p).unwrap()),
..Default::default()
}
}
/// Live 2026-10-26: monetize's ledger said alica paid 10 GB on test2.gunnar.rs, and
/// test2's gunnar said Free: its entitlement order was gone and nothing sent it again.
#[test]
fn a_paid_ledger_fact_test2s_main_does_not_hold_is_queued_for_it_again() {
let reg = crate::registry::Registry::new()
.product(monetize::testing::MockProduct::gunnar())
.home("gunnar", "gunnar-rs");
let s = crate::state::AppState::in_memory(reg);
// AppState::in_memory may already hold a guardian in the registry's slot; else one is installed.
if s.fleet.guardian().is_none() {
let keys = monetize::sign::Keyring::new(monetize::sign::Signer::from_bytes(&[7u8; 32]));
let store = std::sync::Arc::new(monetize_guardian::store::Store::in_memory().unwrap());
assert!(s.fleet.install(monetize_guardian::runner::Guardian::new(
store,
std::sync::Arc::new(keys),
Default::default()
)));
}
let g = s.fleet.guardian().unwrap().clone();
g.store.declare("test2-gunnar-rs", "gunnar").unwrap();
let now = s.now();
g.store
.record_report("test2-gunnar-rs", "main", now, &test2_report(0))
.unwrap();
let fact = EntitlementFact {
tenant: TenantId("@test2-gunnar-rs/alica".into()),
plan: "gunnar/10gb-1m-fi".into(),
state: State::Paid,
paid_until_unix_ms: Some(now + 30 * 86_400_000),
caps: [("pack_bytes".to_string(), 10 * GIB)].into(),
source: "payment:stripe/gunnar-test2:x".into(),
signature: vec![],
issued_unix_ms: None,
issued_signature: vec![],
};
s.ledger
.set_entitlement("gunnar", &s.signer_for("gunnar").sign_fact(fact))
.unwrap();
assert_eq!(
resend_unheld_facts(&s),
1,
"test2's main does not hold it: queued again"
);
let queued: Vec<_> = g
.store
.queued("test2-gunnar-rs")
.into_iter()
.filter(|o| o.kind == order::ENTITLEMENT)
.collect();
assert_eq!(queued.len(), 1);
let body: EntitlementFact = serde_json::from_str(&queued[0].body).unwrap();
assert_eq!(
(body.tenant.0.as_str(), body.caps.get("pack_bytes").copied()),
("alica", Some(10 * GIB)),
"as test2's gunnar names her, with her room"
);
assert_eq!(resend_unheld_facts(&s), 0, "one waits: none more");
g.store
.record_report(
"test2-gunnar-rs",
"main",
now,
&Report {
acked: vec![monetize_embed::fleet::OrderAck {
seq: queued[0].seq,
ok: true,
..Default::default()
}],
..test2_report(10)
},
)
.unwrap();
assert_eq!(resend_unheld_facts(&s), 0, "held: nothing more is sent");
}
}