crates/monetize-server/src/state.rs
//! **`AppState`** — everything a handler needs, behind one `Arc`: the registry,
//! the ledger, the signer, the deployment's terms (the price lists, the policy,
//! the hosts), the usage store, the clock, and the state-change log that
//! becomes `functional_status` rows.
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, RwLock};
use std::time::{SystemTime, UNIX_EPOCH};
use monetize::ledger::LedgerError;
use monetize::{Ledger, Policy, PriceList, Signer, Terms, Transaction};
use monetize_pay::Market;
use monetize_product::{EntitlementFact, State, TenantId, Usage};
use redb::{Database, TableDefinition, ReadableTable};
use crate::bus::{ledger_event, EventBus};
use crate::health::HealthStore;
use crate::registry::Registry;
use crate::store::StoreError;
pub type Clock = Box<dyn Fn() -> u64 + Send + Sync>;
pub fn system_clock() -> Clock {
Box::new(|| SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or_default())
}
/// One entitlement state change, as it is reported. Kept in memory for the tests
/// and the console; emitted to the testmatrix through skidbladnir.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StateChange {
pub product: String,
pub tenant: String,
pub from: Option<State>,
pub to: State,
pub source: String,
pub at_unix_ms: u64,
/// The push to the product succeeded.
pub pushed: bool,
}
const USAGE: TableDefinition<(&str, &str), &[u8]> = TableDefinition::new("usage");
const HISTORY: TableDefinition<(&str, &str, u64), &[u8]> = TableDefinition::new("usage_history");
/// The poller's readings, in redb: the LATEST per (product, tenant) in `usage`,
/// and the kept SERIES per (product, tenant, measured_at) in `usage_history`.
///
/// # Why both tables
///
/// `usage` answers "how much now" in one lookup and is what the console's tenant
/// list joins against; folding a series on every page load to get the same answer
/// would be strictly worse. `usage_history` answers the questions a single slot
/// cannot: how fast is this tenant growing, when did it cross the cap, what did
/// last month look like now that last month is over. Before it existed every
/// poll tick overwrote the previous reading, so monetize measured its customers
/// continuously and remembered nothing.
///
/// # Retention — 400 days, pruned by the writer
///
/// **Readings are NOT kept forever.** [`UsageStore::RETENTION_MS`] is 400 days
/// and [`UsageStore::put`] deletes everything older than that FOR THE TENANT IT
/// IS WRITING, inside the same redb transaction as the insert. There is no timer,
/// no sweeper task and no operator verb: the only way a row is added is the same
/// call that removes the stale ones, so the table cannot grow while nothing is
/// watching it, and a tenant that stops being polled stops being pruned only
/// because it has also stopped growing.
///
/// The number is chosen, not arbitrary. 400 days is a full year plus a margin, so
/// "the same month last year" is always answerable while an annual plan is being
/// renewed or argued about, and a bill can be reconstructed after the fact for
/// any period a plan covers. Beyond that the reading is history, not evidence.
///
/// The bound matters because monetize runs on an APPLIANCE. At the default
/// fifteen-minute tick that is ~96 readings per tenant per day, ~38 400 in the
/// window; at a few hundred bytes of JSON each that is a few tens of megabytes
/// per tenant at the ceiling and it stops there. An unbounded table written on a
/// schedule is not a feature that grows, it is a disk-fill bug with a date on it.
///
/// The history is deliberately absent from [`crate::backup`]: a backup carries
/// the ledger (money, irreplaceable) and the latest reading (cheap, and needed to
/// start), while the series is a measurement the poller reconstructs by simply
/// continuing to run. Putting 400 days of telemetry into a git-replicated
/// repository would drown the rows that actually matter.
pub struct UsageStore {
db: Database,
}
impl UsageStore {
/// How long a reading is kept. See the type's doc for why 400 days.
pub const RETENTION_MS: u64 = 400 * 24 * 60 * 60 * 1000;
/// The most readings one [`UsageStore::history`] call will return when the
/// caller names no limit — a request for "everything" from a console must not
/// be able to ask the server to build a 38 000-element response.
pub const HISTORY_CEILING: usize = 5_000;
pub fn open(path: &Path) -> Result<Self, StoreError> {
Self::init(Database::create(path)?)
}
pub fn in_memory() -> Result<Self, StoreError> {
Self::init(Database::builder().create_with_backend(redb::backends::InMemoryBackend::new())?)
}
fn init(db: Database) -> Result<Self, StoreError> {
let txn = db.begin_write()?;
txn.open_table(USAGE)?;
txn.open_table(HISTORY)?;
txn.commit()?;
Ok(Self { db })
}
/// Store one reading: the latest slot, the history row, and the prune, in ONE
/// redb transaction — so a reading is never kept without its slot being
/// current, and the table is never left holding rows past the window.
///
/// The reading's own `measured_at_unix_ms` is the history key AND the instant
/// the prune measures from. No clock is injected: the store then has no
/// opinion about the time separate from the data, and a test that stands the
/// poller at any date gets the retention behaviour of that date.
/// **Many readings in ONE transaction** (2026-10-06) — a poll tick's whole
/// harvest, so a tick costs one commit rather than one per tenant. Each
/// reading is stored exactly as [`UsageStore::put`] stores one: the latest
/// slot, the history row, the prune.
pub fn put_many(&self, product: &str, readings: &[(String, Usage)]) -> Result<(), StoreError> {
if readings.is_empty() {
return Ok(());
}
let txn = self.db.begin_write()?;
{
let mut latest = txn.open_table(USAGE)?;
let mut hist = txn.open_table(HISTORY)?;
for (tenant, usage) in readings {
let bytes = StoreError::encode("a usage reading", usage)?;
latest.insert((product, tenant.as_str()), bytes.as_slice())?;
hist.insert((product, tenant.as_str(), usage.measured_at_unix_ms), bytes.as_slice())?;
let cutoff = usage.measured_at_unix_ms.saturating_sub(Self::RETENTION_MS);
if cutoff > 0 {
hist.retain_in((product, tenant.as_str(), 0)..(product, tenant.as_str(), cutoff), |_, _| false)?;
}
}
}
txn.commit()?;
Ok(())
}
pub fn put(&self, product: &str, tenant: &str, usage: &Usage) -> Result<(), StoreError> {
let bytes = StoreError::encode("a usage reading", usage)?;
let txn = self.db.begin_write()?;
{
txn.open_table(USAGE)?.insert((product, tenant), bytes.as_slice())?;
let mut hist = txn.open_table(HISTORY)?;
hist.insert((product, tenant, usage.measured_at_unix_ms), bytes.as_slice())?;
// The prune: everything for THIS tenant measured before the window
// opens. One range delete, not a read-then-remove loop.
let cutoff = usage.measured_at_unix_ms.saturating_sub(Self::RETENTION_MS);
if cutoff > 0 {
hist.retain_in((product, tenant, 0)..(product, tenant, cutoff), |_, _| false)?;
}
}
txn.commit()?;
Ok(())
}
/// **The kept series for one tenant, oldest first**, from `since` (0 = the
/// whole window), at most `limit` readings taken from the NEWEST end (0 = the
/// server's [`UsageStore::HISTORY_CEILING`]).
///
/// `limit` cuts from the newest end and the result is still ordered oldest
/// first, because "the last hundred readings" is the question a chart asks and
/// reversing it in the caller is how a series ends up drawn backwards.
pub fn history(&self, product: &str, tenant: &str, since_unix_ms: u64, limit: usize) -> Result<Vec<Usage>, StoreError> {
let limit = if limit == 0 { Self::HISTORY_CEILING } else { limit.min(Self::HISTORY_CEILING) };
let txn = self.db.begin_read()?;
let t = txn.open_table(HISTORY)?;
let mut out: std::collections::VecDeque<Usage> = std::collections::VecDeque::new();
// Walked BACKWARDS so `limit` takes the newest without reading the rest.
for entry in t.range((product, tenant, since_unix_ms)..=(product, tenant, u64::MAX))?.rev() {
if out.len() >= limit {
break;
}
let (_, v) = entry?;
if let Ok(u) = serde_json::from_slice(v.value()) {
out.push_front(u);
}
}
Ok(out.into())
}
/// How many readings are kept for one tenant, whatever a `history` call's
/// `since` and `limit` cut away.
pub fn kept(&self, product: &str, tenant: &str) -> Result<u64, StoreError> {
let txn = self.db.begin_read()?;
let t = txn.open_table(HISTORY)?;
let mut n = 0;
for entry in t.range((product, tenant, 0)..=(product, tenant, u64::MAX))? {
entry?;
n += 1;
}
Ok(n)
}
pub fn get(&self, product: &str, tenant: &str) -> Result<Option<Usage>, StoreError> {
let txn = self.db.begin_read()?;
let t = txn.open_table(USAGE)?;
Ok(t.get((product, tenant))?.and_then(|v| serde_json::from_slice(v.value()).ok()))
}
/// **Every reading, across every product, in key order** — what a backup
/// exports. Ordered and total so two exports of the same state are
/// byte-identical.
pub fn export_all(&self) -> Result<Vec<(String, String, Usage)>, StoreError> {
let txn = self.db.begin_read()?;
let t = txn.open_table(USAGE)?;
let mut out = Vec::new();
for entry in t.iter()? {
let (k, v) = entry?;
let (product, tenant) = k.value();
if let Ok(u) = serde_json::from_slice(v.value()) {
out.push((product.to_owned(), tenant.to_owned(), u));
}
}
Ok(out)
}
pub fn all(&self, product: &str) -> Result<Vec<(TenantId, Usage)>, StoreError> {
let txn = self.db.begin_read()?;
let t = txn.open_table(USAGE)?;
let mut out = Vec::new();
for entry in t.range((product, "")..)? {
let (k, v) = entry?;
if k.value().0 != product {
break;
}
if let Ok(u) = serde_json::from_slice(v.value()) {
out.push((TenantId(k.value().1.to_string()), u));
}
}
Ok(out)
}
}
/// **The deployment's terms**: what a unit costs the customer per product per
/// market, the one policy, and where each product's disks attach. Read from
/// `monetize.toml`; changed by editing it and restarting — there is no RPC
/// that writes a price, deliberately.
#[derive(Clone, Debug)]
pub struct TermsStore {
pub prices: BTreeMap<String, BTreeMap<Market, PriceList>>,
pub policy: Policy,
/// `[nonpay]`: the warnings, the deletion, the renewal reminder.
pub nonpay: crate::nonpay::NonpayPolicy,
/// product id → host a bought disk attaches to.
pub hosts: BTreeMap<String, String>,
pub default_market: Market,
}
impl Default for TermsStore {
/// No list — free everywhere — the default policy, SE.
fn default() -> Self {
TermsStore { prices: BTreeMap::new(), policy: Policy::default(), nonpay: Default::default(), hosts: BTreeMap::new(), default_market: Market::Se }
}
}
impl TermsStore {
/// The list for `product` in `market`, or `None` — which is free.
pub fn list(&self, product: &str, market: Market) -> Option<&PriceList> {
self.prices.get(product)?.get(&market)
}
/// The host a product's disks attach to: configured, or the product id.
pub fn host(&self, product: &str) -> String {
self.hosts.get(product).cloned().unwrap_or_else(|| product.to_owned())
}
/// The terms an order for `product` in `market` runs under. The host is
/// passed in so the borrow is the caller's.
pub fn terms<'a>(&'a self, product: &str, market: Market, host: &'a str) -> Terms<'a> {
Terms { list: self.list(product, market), market, policy: self.policy, host }
}
/// A fixture: 22.00 SEK per GiB-month of disk for the fixture's product
/// ([`monetize::testing::PRODUCT`]) in SE, disks on [`monetize::testing::HOST`], the
/// fixture policy — the in-memory server's terms.
pub fn fixture() -> TermsStore {
TermsStore {
prices: [(monetize::testing::PRODUCT.to_string(), [(Market::Se, monetize::testing::price_list())].into())].into(),
policy: monetize::testing::POLICY,
nonpay: Default::default(),
hosts: [(monetize::testing::PRODUCT.to_string(), monetize::testing::HOST.to_string())].into(),
default_market: Market::Se,
}
}
}
/// **What a ladder move came to** — [`AppState::apply_ladder_fact`].
#[derive(Debug)]
pub struct LadderApplied {
/// **The fact that STANDS afterwards**: the ladder's own, or the newer one
/// that replaced it while the push was in flight. The poller's reclamation
/// queue reads its state, so a tenant who paid mid-tick is never entered on
/// a teardown list on the strength of a verdict that no longer holds.
pub stands: EntitlementFact,
/// Whether the PRODUCT acknowledged the fact the ladder pushed.
pub pushed: Result<(), String>,
/// A newer fact replaced the ladder's between the store and the push
/// landing. The product is holding the older one; `stands` is left pending
/// so the retry corrects it.
pub superseded_in_flight: bool,
}
pub struct AppState {
pub registry: Registry,
pub ledger: Ledger,
/// **The product keys** (G8): what a fact, snapshot, go-ahead, estate
/// growth or fleet order for product `p` is signed with —
/// [`AppState::signer_for`]. The root signs nothing a product sees.
pub keys: Arc<monetize::Keyring>,
/// The price lists, the policy, the hosts. See [`TermsStore`].
pub terms: RwLock<TermsStore>,
/// Where `Ledger.Export` writes. `None` in tests that never back up.
pub backup_dir: Option<PathBuf>,
pub usage: UsageStore,
/// **Where each product's last batched usage read left off** — the cursor
/// the next tick asks "what changed since" with (2026-10-06). In memory: a
/// restart's first tick reads everything, which is the honest first tick.
pub usage_cursors: crate::usage_poll::Cursors,
/// **What each product last said about itself** — its capacity answer and
/// whether it answers at all, refreshed by the poller and never asked from
/// inside a page load. See [`crate::health`] for why it is a cache.
pub health: HealthStore,
pub clock: Clock,
/// `vendor -> checkout URL template` (`{handle}`, `{reference}`).
/// Live: a `[vendors.<name>] checkout_url` edited or added while the
/// server runs is followed by the next purchase ([`crate::guests::apply`]).
pub checkout_urls: RwLock<BTreeMap<String, String>>,
pub changes: Mutex<Vec<StateChange>>,
/// **The live feed.** Bounded, lossy-with-a-signal, and never in the way of
/// a write — see [`crate::bus`]. `Arc` because the ledger observer and every
/// `Events.Watch` stream hold it independently of `AppState`.
pub bus: Arc<EventBus>,
/// **The refused vendor callbacks.** A bounded in-memory ring an
/// UNAUTHENTICATED caller can put rows in, read by the console over
/// `CallbackAdmin.ListRefusals` and by nothing else — no decision in this
/// server branches on it. See [`crate::refusals`] for the bound, for why it
/// is not persisted, and for which refusal arms belong in it.
pub refusals: crate::refusals::RefusalLog,
/// **The paid-signup store**: `(vendor, session) → what became of it`. See
/// [`crate::signup`] for why it is not the ledger.
pub signups: crate::signup::SignupStore,
/// **The vendor's deliveries, kept before they are acted on**: `(vendor,
/// event id) → what became of it`. See [`crate::inbox`].
pub callbacks: crate::inbox::CallbackInbox,
/// **One resume per reference at a time** — the webhook's grant, a
/// redelivery's, the poller's and `Purchase.Status` share it, so a retried
/// event cannot grant twice. See [`crate::inbox::Flights`].
pub resuming: crate::inbox::Flights,
/// **What the vendor says it took and monetize did not fulfil**: the
/// reconciler's case rows, `(vendor, session) → what was decided`, and
/// each vendor's last pass. Written by the poller's reconcile pass alone,
/// from the vendor's answer to OUR authenticated read — never from a POST.
/// See [`crate::reconcile`] for the table it is classified by.
pub reconcile: crate::reconcile::ReconcileStore,
/// **Who stopped paying, what they were told, and what was deleted** —
/// written by the poller alone. See [`crate::nonpay`].
pub nonpay: crate::nonpay::NonpayJournal,
/// **What turns a paid signup into a customer.** The open build's is
/// [`crate::signup::LoudSink`], which creates nothing and says so; the vetra
/// build registers the real one through `Linked::signup`.
pub signup_sink: crate::signup::Sink,
/// `[signups.<product>.plans]`, verbatim. Not in [`TermsStore`] with the price lists,
/// because a signup plan is not a price list: it is one bundle sold whole
/// from a page this process never rendered, and the only arithmetic done
/// with it is an equality check against what the vendor took.
pub signup_plans: crate::signup::Plans,
/// **The `[signups.<product>]` block taken in live** — the guest watcher hands it every
/// config it applies ([`crate::guests::apply_signup`]). `None`: the sink is
/// the start's for the life of the process (the open default, a test).
pub signup_reload: Option<Arc<crate::signup::SignupReload>>,
/// **THE SHOP WINDOW, per product**: `[products.<id>.bundles]`, verbatim —
/// product id → bundle id → the declared cap delta.
///
/// Kept as DECLARED and priced per request, never priced once at start.
/// A bundle's price depends on the market the caller names
/// (`[prices.<product>.SE]` and `.FI` are different numbers), and a shop
/// window priced at boot would show a Finnish customer Swedish öre.
///
/// Behind a lock and re-read with every guest reload ([`crate::guests::apply`]):
/// a product that registers ITS OWN plugin after this server started (holger's
/// primary, gunnar's) brings its `[products.<id>.bundles]` in the same block,
/// and a window read only at start stayed empty until the front restarted
/// (MEASURED on holger's mock 2026-09-28: `Purchase.Bundles` answered `[]`).
pub bundles: std::sync::RwLock<BTreeMap<String, BTreeMap<String, monetize::BundleSpec>>>,
/// **How much of each product's iron this deployment may sell**,
/// `[signups.<product>] max_sold_percent_of_capacity`, by product (absent:
/// [`crate::oversell::DEFAULT_MAX_SOLD_PERCENT`]; see
/// [`AppState::max_sold_percent_of`]). Read by the signup gate
/// (`crate::oversell`) and by nothing else; it is a percentage of what the
/// product says it can hold, and there is no value that means "no limit".
pub signup_max_sold_percents: std::collections::BTreeMap<String, u32>,
/// **Slots held by checkouts that have been minted and not yet settled** —
/// what stops two concurrent `Signup.Start` calls being told yes for the
/// same last slot. In memory, bounded by its own TTL; see
/// [`crate::oversell::Reservations`] for exactly what it does not guard.
pub signup_reservations: crate::oversell::Reservations,
/// **What turns the ledger into a tax return.** A started server's is
/// `Linked::build_tax`'s: [`monetize::PerCountry`] with no `[tax]` declared
/// (each payment's country, rate and tax the vendor's); the vetra build
/// registers a jurisdiction through `Linked::tax`. [`AppState::in_memory`]
/// starts on [`monetize::LoudTax`] until a test sets one.
pub tax: Arc<dyn monetize::TaxReport>,
/// **Where this deployment is, and where each of its suppliers is** — the
/// operator's declaration (`default_market`, and `country` on each
/// `[vendors.<name>]` / `[clouds.<name>]`). What a `TaxSale`'s countries are
/// resolved from. See [`monetize::Books`].
/// Live, like the vendors and clouds whose `country` it carries.
pub books: RwLock<monetize::Books>,
/// **The data sets**, one per product that declares `[products.<id>.grow]`,
/// and their wallet readings. See [`crate::grow`].
pub grow: crate::grow::GrowSets,
/// **The private-estate orders** (`[estate_orders]`): the lay-estate jobs,
/// their installer, their DNS writer and the custody of their secrets.
/// See [`crate::estate_order`]. Default: this server sells none.
pub estates: crate::estate_order::EstateOrders,
/// **The names a paid "Add users" order waits to add** (G13): beside the
/// ledger, which names no member ([`crate::named_users`]).
pub named_users: crate::named_users::NamedUsers,
/// **The deployment's own declaration, as this process loaded it.**
///
/// Kept whole for the one thing that is DRAWN rather than decided: the
/// estate picture ([`crate::estate_svg::Estate::from_config`]), whose
/// `[machines]` and `[clouds]` blocks no other field here carries.
///
/// **It is not a second source of truth.** Every decision in this server
/// took its input out of this file once, at boot, and holds it in its own
/// field above; nothing consults this one to decide anything, and
/// `[machines]` decides nothing by construction — see
/// [`crate::config::MachineConfig`]. It is kept rather than re-read so the
/// picture can never show an estate the running process has not loaded.
///
/// Behind a lock since vendors and clouds apply live: the watcher swaps
/// in the file it just applied, so the picture never shows an estate the
/// running process has not taken in — nor misses one it has.
pub config: RwLock<crate::config::Config>,
/// **The plugin updates** (2026-10-17): the jera lane the operator's `Plugins.Update`
/// queues on, and what each job came to (`crate::plugins`).
pub plugin_jobs: crate::plugins::PluginJobs,
/// **How many webhook deliveries may be inside this process at once**,
/// counting the settle or grant each one leaves running behind its answer.
/// See [`crate::callback::Slots`].
pub callback_slots: crate::callback::Slots,
/// The admin principals whose per-call SSHSIG opens the control API from
/// outside the estate (`crate::control`); re-read with the config.
pub control: Arc<crate::control::ControlKeys>,
/// **Room sold as plans, by the price rule** — `[pricing.<product>]` and
/// what was last read of the provider's price. See [`crate::plans`].
pub plans: crate::plans::PlanBook,
/// **The pull channel and the estate guardian** (5.4): what the estates' boxes report,
/// the signed orders queued for them, the cases. See [`crate::fleet`].
pub fleet: crate::fleet::FleetState,
/// **The operator's TEST BUTTON's runs** (`TestPurchase`).
pub tests: crate::test_purchase::TestRuns,
}
impl AppState {
/// `product`'s sold-capacity allowance: its `[signups.<product>]`
/// `max_sold_percent_of_capacity`, else the finite default.
pub fn max_sold_percent_of(&self, product: &str) -> u32 {
self.signup_max_sold_percents.get(product).copied().unwrap_or(crate::oversell::DEFAULT_MAX_SOLD_PERCENT)
}
/// In-memory everything, under [`TermsStore::fixture`]'s terms: tests and
/// `--dry` bring-up.
pub fn in_memory(registry: Registry) -> Self {
let keys = Arc::new(monetize::Keyring::new(Signer::generate()));
let fleet = crate::fleet::FleetState::sharing(registry.pull_slot());
fleet.install(crate::fleet::in_memory_guardian(Arc::clone(&keys)));
let state = AppState {
registry,
ledger: Ledger::in_memory().expect("in-memory ledger"),
keys,
fleet,
terms: RwLock::new(TermsStore::fixture()),
backup_dir: None,
usage: UsageStore::in_memory().expect("in-memory usage store"),
usage_cursors: Default::default(),
health: HealthStore::default(),
clock: system_clock(),
checkout_urls: RwLock::new(BTreeMap::new()),
changes: Mutex::new(Vec::new()),
bus: Arc::new(EventBus::default()),
refusals: crate::refusals::RefusalLog::default(),
signups: crate::signup::SignupStore::in_memory().expect("in-memory signup store"),
callbacks: crate::inbox::CallbackInbox::in_memory().expect("in-memory callback inbox"),
resuming: crate::inbox::Flights::default(),
reconcile: crate::reconcile::ReconcileStore::in_memory().expect("in-memory reconcile store"),
nonpay: crate::nonpay::NonpayJournal::in_memory(),
// The open default in the open constructor. A test that wants to
// watch a sink refuse replaces the field; the vetra build replaces
// it through the registration seam.
signup_sink: Arc::new(crate::signup::LoudSink),
signup_plans: crate::signup::Plans::default(),
signup_reload: None,
// No shop window: `Purchase.Bundles` answers an empty list and the
// typed order path is untouched. A test that wants one sets it.
bundles: std::sync::RwLock::new(BTreeMap::new()),
signup_max_sold_percents: Default::default(),
signup_reservations: Default::default(),
// Same rule, same constructor: the OPEN default, replaced by a test
// that wants a jurisdiction or by the vetra build's registration.
tax: Arc::new(monetize::LoudTax),
books: RwLock::new(monetize::Books { home_country: TermsStore::fixture().default_market.country().to_owned(), supplier_countries: BTreeMap::new() }),
// No data set configured: no wallet is read. A test that wants one
// replaces the field.
grow: crate::grow::GrowSets::default(),
// No `[estate_orders]`: `EstateOrder.Start` refuses by name. A test
// that sells private estates replaces the field.
estates: crate::estate_order::EstateOrders::default(),
named_users: crate::named_users::NamedUsers::in_memory(),
// No `[machines]`: the estate picture draws an empty board, which
// is the honest drawing of a deployment that has not written one
// down. A test that wants chips on it replaces the field.
config: RwLock::new(crate::config::Config::default()),
plugin_jobs: Default::default(),
callback_slots: Default::default(),
control: Default::default(),
// No `[pricing]`: no product sells plans. A test that wants them sets it.
plans: Default::default(),
tests: crate::test_purchase::TestRuns::in_memory(),
};
state.wire_events();
state
}
/// **Point the ledger at the bus.** Every row `Ledger::append` commits
/// becomes a `LedgerAppended` event from here on.
///
/// It is a separate call rather than something `AppState`'s constructor
/// hides because `main` builds its `AppState` with a struct literal, and a
/// wiring step that only ran on one of the two construction paths is exactly
/// the bug that would leave the live console silent on the real box while
/// every test stayed green. Both paths call it, and
/// [`tests::in_memory_points_the_ledger_at_its_own_bus`] is the guard on the
/// one a test can reach.
pub fn wire_events(&self) {
let bus = Arc::clone(&self.bus);
self.ledger.observe(Arc::new(move |row, open| {
bus.publish(ledger_event(row, open));
}));
}
pub fn now(&self) -> u64 {
(self.clock)()
}
/// The key `product`'s facts are signed with (G8).
pub fn signer_for(&self, product: &str) -> &Signer {
self.keys.product(product)
}
/// **The product a sold estate is** (OM-4, one monetize 2026-10-10):
/// `[estate_orders] product`, named or derived at the start
/// ([`crate::estate_order::EstateOrderConfig::product`]) — never a
/// hard-coded "gunnar". No `[estate_orders]`: nothing is sold here, and
/// the asker is told so by name.
pub fn sold_product(&self) -> Result<&str, String> {
let Some(c) = self.estates.cfg.as_ref() else {
return Err("no [estate_orders]: this monetize sells no private estates".into());
};
c.sold_product()
}
pub fn transaction<'a>(&'a self, product: monetize_product::Served<'a>, pay: monetize_pay::Vendor<'a>) -> Transaction<'a> {
Transaction {
pay,
cloud: self.registry.cloud_for_transactions(),
product,
ledger: &self.ledger,
signer: self.signer_for(product.id()),
now_unix_ms: self.now(),
}
}
/// Store a signed fact, push it to the product, and REPORT the state change —
/// one writer for the operator's `Set` and the poller's ladder moves. The fact
/// is stored even when the push fails, and then a [`monetize::ledger::PendingPush`]
/// is stored beside it so the poller's [`crate::poller::retry_pushes`] pushes it
/// again. Until 2026-09-14 this comment said the poller retried and nothing did.
/// **The LADDER's apply: it replaces only the fact it decided from.**
///
/// [`AppState::apply_fact`] pushes and then overwrites, which is right for
/// an operator and for a purchase — both are somebody saying what IS true,
/// and the last word should win. The ladder is not saying that. It is
/// saying what FOLLOWS from a fact it read before a gRPC call that can take
/// thirty seconds, and in that window a card clears: the purchase stores
/// Paid with next month's `paid_until`, this then overwrites it with
/// Suspended carrying LAST month's, and every later tick derives Suspended
/// from that stale date and agrees. Money captured, service withdrawn, and
/// nothing that ever disagrees again.
///
/// So: a compare-and-swap against `expect`, with the pending-push row in the
/// same transaction (the fact is stored BEFORE it is pushed here, and a
/// crash in between must leave a to-do and not a fact the product is never
/// told). `Ok(None)` means somebody else moved the tenant while this was
/// being decided — their fact is newer, they pushed it, and the ladder's
/// verdict is simply void.
pub fn apply_ladder_fact(
&self,
product: monetize_product::Served<'_>,
expect: Option<&EntitlementFact>,
fact: &EntitlementFact,
) -> Result<Option<LadderApplied>, LedgerError> {
let before = expect.map(|f| f.state);
let pending = monetize::ledger::PendingPush {
fact: fact.clone(),
since_unix_ms: self.now(),
attempts: 0,
last_error: "stored, not yet pushed".into(),
};
if !self.ledger.replace_entitlement_pending(product.id(), expect, &pending)? {
return Ok(None);
}
let answer = product.push_entitlement(fact);
// **WHAT STANDS AFTER THE CALL.** The push is a gRPC round trip, so a
// payment lands inside it as readily as inside the window the CAS above
// closes — and a payment's write is a BLIND one (`Step::EntitlementPushed`
// inserts the fact and clears the pending row in its own transaction),
// so it wins the ledger and cannot be stopped from here. What it cannot
// do is un-send the bytes already on their way to the appliance.
//
// MEASURED (`monetize-server/tests/renewal.rs`): a renewal that lands
// mid-push left monetize holding Paid and the APPLIANCE holding
// Suspended, with nothing pending — a customer who had just paid,
// locked out on the box, and no row anywhere that would ever fix it.
let stands = self.ledger.entitlement(product.id(), &fact.tenant.0)?.unwrap_or_else(|| fact.clone());
let superseded_in_flight = &stands != fact;
match &answer {
// Acknowledged — but only of OUR fact. When a newer one replaced it
// while the call was in flight, the newer one is what the product
// must end up holding, so it is left PENDING and `retry_pushes`
// delivers it on the next tick.
Ok(()) if !superseded_in_flight => self.ledger.push_acknowledged(product.id(), &fact.tenant.0)?,
Ok(()) => self.ledger.push_pending(
product.id(),
&monetize::ledger::PendingPush {
fact: stands.clone(),
since_unix_ms: self.now(),
attempts: 0,
last_error: "the ladder's push landed after a newer fact replaced it; the product is holding the older one".into(),
},
)?,
Err(e) => self.ledger.push_pending(
product.id(),
&monetize::ledger::PendingPush {
fact: stands.clone(),
attempts: 1,
last_error: e.to_string(),
..pending
},
)?,
}
let pushed = answer.map_err(|e| e.to_string());
self.report_change(StateChange {
product: product.id().into(),
tenant: fact.tenant.0.clone(),
from: before,
to: stands.state,
source: stands.source.clone(),
at_unix_ms: self.now(),
pushed: pushed.is_ok() && !superseded_in_flight,
});
Ok(Some(LadderApplied { stands, pushed, superseded_in_flight }))
}
pub fn apply_fact(&self, product: monetize_product::Served<'_>, fact: &EntitlementFact) -> Result<Result<(), String>, LedgerError> {
let before = self.ledger.entitlement(product.id(), &fact.tenant.0)?.map(|f| f.state);
let answer = product.push_entitlement(fact);
self.ledger.set_entitlement(product.id(), fact)?;
match &answer {
Ok(()) => self.ledger.push_acknowledged(product.id(), &fact.tenant.0)?,
Err(e) => self.ledger.push_pending(
product.id(),
&monetize::ledger::PendingPush {
fact: fact.clone(),
since_unix_ms: self.now(),
attempts: 1,
last_error: e.to_string(),
},
)?,
}
let pushed = answer.map_err(|e| e.to_string());
self.report_change(StateChange {
product: product.id().into(),
tenant: fact.tenant.0.clone(),
from: before,
to: fact.state,
source: fact.source.clone(),
at_unix_ms: self.now(),
pushed: pushed.is_ok(),
});
Ok(pushed)
}
/// Every state change → a `functional_status` row (skidbladnir → the nornir
/// testmatrix when built with `testmatrix`), plus the in-memory log.
pub fn report_change(&self, change: StateChange) {
let detail = format!(
"{}/{}: {} -> {:?} source={} pushed={}",
change.product,
change.tenant,
change.from.map_or("none".to_string(), |s| format!("{s:?}")),
change.to,
change.source,
change.pushed
);
skidbladnir::functional_status("monetize-server/entitlement", "state-change", change.pushed, &detail);
eprintln!("monetize-server: entitlement {detail}");
// The live feed, from the SAME one writer the testmatrix row comes from.
// Publishing at the call sites instead would have been three places
// (operator `Set`, the ladder, a paid purchase) to forget one.
self.bus.publish(monetize_proto::event::Payload::EntitlementChanged(monetize_proto::EntitlementChanged {
product: change.product.clone(),
tenant: change.tenant.clone(),
order: self
.ledger
.entitlement(&change.product, &change.tenant)
.ok()
.flatten()
.map(|f| f.plan)
.unwrap_or_default(),
from_state: change.from.map(crate::convert::state_to_pb).unwrap_or(monetize_proto::State::Unspecified) as i32,
had_previous: change.from.is_some(),
to_state: crate::convert::state_to_pb(change.to) as i32,
source: change.source.clone(),
pushed: change.pushed,
}));
self.changes.lock().unwrap_or_else(|e| e.into_inner()).push(change);
}
pub fn changes(&self) -> Vec<StateChange> {
self.changes.lock().unwrap_or_else(|e| e.into_inner()).clone()
}
}
#[cfg(test)]
mod tests {
use super::*;
/// (r) `in_memory` really does point the ledger at its bus.
///
/// Every event test in this crate stands on it: without the `wire_events`
/// call in the constructor they would all be watching a bus that nothing
/// writes to, and a bus nobody writes to answers every "did you get it?" the
/// same way a broken one does. Asserted on the bus's own counter after a
/// REAL append, both before and after, so a bus that counted a phantom event
/// at construction would not satisfy it either.
#[test]
fn in_memory_points_the_ledger_at_its_own_bus() {
let state = AppState::in_memory(Registry::new());
assert_eq!(state.bus.stats().published, 0, "a fresh state has published nothing");
state
.ledger
.append(1, monetize::Step::Declined { reference: "gunnar/team/sub/2026-09-04".into(), reason: "no".into() })
.unwrap();
assert_eq!(state.bus.stats().published, 1, "the committed row reached the bus");
assert_eq!(state.bus.last_seq(), 1);
}
fn reading(at: u64, bytes: u64) -> Usage {
Usage { meters: [("pack_bytes".to_string(), bytes)].into(), measured_at_unix_ms: at }
}
/// (r) TWO readings for the SAME tenant at DIFFERENT times both survive, in
/// order, and the latest slot still answers with the newer of them.
///
/// This is the one the single-slot store could not pass: `put` used to write
/// `(product, tenant) -> Usage` and nothing else, so the second reading
/// REPLACED the first and any history call could only ever return one row.
/// Asserting the METER VALUES and the timestamps, not the count: a series of
/// two rows that both hold the newest reading is the same bug wearing a
/// different shape.
#[test]
fn two_readings_for_one_tenant_are_both_kept_oldest_first() {
let s = UsageStore::in_memory().unwrap();
s.put("gunnar", "team/sub", &reading(1_000, 5)).unwrap();
s.put("gunnar", "team/sub", &reading(2_000, 9)).unwrap();
let h = s.history("gunnar", "team/sub", 0, 0).unwrap();
assert_eq!(
h.iter().map(|u| (u.measured_at_unix_ms, u.meters["pack_bytes"])).collect::<Vec<_>>(),
vec![(1_000, 5), (2_000, 9)],
"both readings, oldest first"
);
assert_eq!(s.kept("gunnar", "team/sub").unwrap(), 2);
// The latest slot is untouched by the history table.
assert_eq!(s.get("gunnar", "team/sub").unwrap().unwrap().meters["pack_bytes"], 9);
assert_eq!(s.export_all().unwrap().len(), 1, "a backup still exports one row per tenant, not the series");
}
/// (r) `since` and `limit` cut the right ends, and a tenant's series never
/// bleeds into a tenant whose name it is a PREFIX of — the failure mode of a
/// range scan over a tuple key with a variable-length middle.
#[test]
fn since_cuts_the_old_end_limit_cuts_the_new_and_a_prefix_tenant_does_not_bleed() {
let s = UsageStore::in_memory().unwrap();
for (i, at) in [10u64, 20, 30, 40].iter().enumerate() {
s.put("gunnar", "team", &reading(*at, i as u64)).unwrap();
s.put("gunnar", "team/sub", &reading(*at, 100 + i as u64)).unwrap();
s.put("holger", "team", &reading(*at, 900 + i as u64)).unwrap();
}
let at = |h: &[Usage]| h.iter().map(|u| u.measured_at_unix_ms).collect::<Vec<_>>();
assert_eq!(at(&s.history("gunnar", "team", 0, 0).unwrap()), vec![10, 20, 30, 40]);
// `since` is inclusive.
assert_eq!(at(&s.history("gunnar", "team", 20, 0).unwrap()), vec![20, 30, 40]);
assert_eq!(at(&s.history("gunnar", "team", 41, 0).unwrap()), Vec::<u64>::new());
// `limit` keeps the NEWEST, still ordered oldest first.
assert_eq!(at(&s.history("gunnar", "team", 0, 2).unwrap()), vec![30, 40]);
assert_eq!(at(&s.history("gunnar", "team", 20, 2).unwrap()), vec![30, 40]);
// Neither the prefix tenant nor the other product is in this tenant's series.
let sub = s.history("gunnar", "team/sub", 0, 0).unwrap();
assert_eq!(sub.len(), 4);
assert!(sub.iter().all(|u| u.meters["pack_bytes"] >= 100 && u.meters["pack_bytes"] < 900), "{sub:?}");
assert!(s.history("gunnar", "team", 0, 0).unwrap().iter().all(|u| u.meters["pack_bytes"] < 100));
assert_eq!(s.kept("gunnar", "team").unwrap(), 4);
assert_eq!(s.history("gunnar", "nobody", 0, 0).unwrap(), Vec::<Usage>::new());
}
/// (r) The retention window is real: a reading past `RETENTION_MS` is GONE
/// after the next write, and one inside it stays. Without the prune in `put`
/// this table grows forever on an appliance.
#[test]
fn a_reading_past_the_retention_window_is_pruned_by_the_next_write() {
let s = UsageStore::in_memory().unwrap();
let now = 2_000_000_000_000u64;
let ancient = now - UsageStore::RETENTION_MS - 1;
let inside = now - UsageStore::RETENTION_MS + 1;
s.put("gunnar", "team", &reading(ancient, 1)).unwrap();
s.put("gunnar", "team", &reading(inside, 2)).unwrap();
assert_eq!(s.kept("gunnar", "team").unwrap(), 2, "nothing is pruned before the window has moved past it");
s.put("gunnar", "team", &reading(now, 3)).unwrap();
assert_eq!(
s.history("gunnar", "team", 0, 0).unwrap().iter().map(|u| u.measured_at_unix_ms).collect::<Vec<_>>(),
vec![inside, now],
"the reading older than the window is gone, the one inside it is not"
);
}
}
/// R5a: every state change a push makes is recorded here; a panic that once
/// held the list must not make every later push panic.
#[cfg(test)]
mod poisoned_changes_tests {
use super::*;
#[test]
fn poisoned_changes_still_answer() {
let state = AppState::in_memory(Registry::new());
let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let _held = state.changes.lock();
panic!("a panic while the changes are held");
}));
assert!(state.changes.is_poisoned());
assert!(state.changes().is_empty());
}
}