crates/monetize-server/src/grow.rs
//! **What the shop reads about a product's data set, and how an order past its room is queued.**
//!
//! A box grows through the v2 path only (owner 2026-11-02): a paid step is a signed
//! `grow_go_ahead` the box PULLS, and the box asks `Fleet.Ask{grow_set}` to size its own data
//! disks ([`crate::estate_order::sold::grow_own`], journaled by `monetize_cloud::grow`). The
//! appliance drive that once ran a growth from here (approve, flush, start, wait, resume, its
//! batcher and twin lane, the operator's grow console) is deleted.
//!
//! What stays here:
//!
//! * **the wallet reading** ([`affordable_bytes`], [`affordable_within`]): how much more data
//! set the cloud's spend ceiling can still buy for `[products.<id>.grow]`'s set, read off every
//! customer's path — [`crate::oversell`] judges the shop's promise against it;
//! * **the queue past the room** ([`queues_for_resize`], [`queue_past_room`],
//! [`can_absorb_queued`]): the customer never sees a cap (owner 2026-10-08) — an order past
//! the room of an estate that grows its own set when told is SOLD and QUEUED, and the estate
//! is told by a signed `grow_room` order;
//! * [`next_window_boundary`]: the daily boundary a twin's own plan change waits for.
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use monetize_cloud::grow::{DataSet, VolumeId};
use crate::config::GrowSetConfig;
use crate::state::AppState;
/// What a provider's price was asked for: the members' surveyed sizes
/// `(name, GiB)` and the target GiB.
type PriceKey = (Vec<(String, u64)>, u64);
/// **One configured product's data set** (`[products.<id>.grow]`) and its wallet reading.
pub struct SetWallet {
pub set: GrowSetConfig,
/// The last [`affordable_bytes`] reading, refreshed OFF the caller's path
/// once the state is armed — see [`affordable_within`].
wallet: WalletCell,
/// **The provider's prices the reading's bisect was given**, by (the
/// members' surveyed sizes, the target GiB) — [`PRICE_FRESH`].
prices: Mutex<BTreeMap<PriceKey, (std::time::Instant, u64)>>,
}
impl SetWallet {
fn new(set: GrowSetConfig) -> SetWallet {
SetWallet { set, wallet: WalletCell::default(), prices: Mutex::new(BTreeMap::new()) }
}
/// One price for the reading's bisect: the cached one while younger than
/// [`PRICE_FRESH`], else `ask` (and kept, when it priced — a failure is
/// never remembered, so the next reading asks again).
fn price_cached(&self, sizes: &[(String, u64)], target: u64, ask: impl FnOnce() -> Option<u64>) -> Option<u64> {
let key = (sizes.to_vec(), target);
if let Some((at, p)) = self.prices.lock().unwrap_or_else(|e| e.into_inner()).get(&key) {
if at.elapsed() < PRICE_FRESH {
return Some(*p);
}
}
let p = ask()?;
let mut cache = self.prices.lock().unwrap_or_else(|e| e.into_inner());
cache.retain(|_, (at, _)| at.elapsed() < PRICE_FRESH);
cache.insert(key, (std::time::Instant::now(), p));
Some(p)
}
pub fn data_set(&self, product: &str, target_gib: u64) -> DataSet {
DataSet {
product: product.to_owned(),
server: monetize_cloud::ServerId(self.set.server.clone()),
members: self.set.members.iter().map(|m| VolumeId(m.clone())).collect(),
target_gib,
}
}
}
/// **Every configured product's data set**, by product id.
pub struct GrowSets {
products: BTreeMap<String, Arc<SetWallet>>,
/// The state this lives in, once [`arm`] armed it: what a reading
/// refreshed on its own thread reads from. Unarmed (a bare test state),
/// a reading is made inline, as it always was.
armed: std::sync::OnceLock<std::sync::Weak<AppState>>,
/// **How old a wallet reading may get before the keeper refreshes it in
/// the background** ([`KEEP_WARM_EVERY`]). `None` = no keeper (a test
/// that counts every cloud op its own asks make).
pub keep_warm: Option<Duration>,
}
impl Default for GrowSets {
fn default() -> Self {
GrowSets { products: BTreeMap::new(), armed: std::sync::OnceLock::new(), keep_warm: Some(KEEP_WARM_EVERY) }
}
}
impl GrowSets {
/// From `[products.<id>.grow]`.
pub fn from_config(cfg: &crate::config::Config) -> anyhow::Result<GrowSets> {
let mut sets = GrowSets::default();
for (id, p) in &cfg.products {
let Some(set) = &p.grow else { continue };
sets.add(id, set.clone()).map_err(|e| anyhow::anyhow!("[products.{id}.grow] {e}"))?;
}
Ok(sets)
}
/// Register one product's set.
pub fn add(&mut self, product: &str, set: GrowSetConfig) -> Result<(), String> {
if set.cloud.is_empty() || set.server.is_empty() || set.members.is_empty() {
return Err("needs `cloud`, `server` and at least one member".into());
}
self.products.insert(product.to_owned(), Arc::new(SetWallet::new(set)));
Ok(())
}
pub fn get(&self, product: &str) -> Option<Arc<SetWallet>> {
self.products.get(product).cloned()
}
pub fn products(&self) -> impl Iterator<Item = (&String, &Arc<SetWallet>)> {
self.products.iter()
}
}
/// What the cloud's ceiling has left, the tighter of month and total — or `None` on a cloud
/// nothing wraps.
pub fn ceiling_remaining(state: &AppState, cloud: &crate::hot::HotCloud) -> Option<u64> {
let ceiling = cloud.spend_ceiling()?;
let month = cloud.spent_this_month_minor().unwrap_or(0);
let total = state.ledger.total_provisioned_minor_by(cloud.name(), ceiling.currency).unwrap_or(0);
let m = ceiling.monthly_minor.map(|c| c.saturating_sub(month));
let t = ceiling.total_minor.map(|c| c.saturating_sub(total));
match (m, t) {
(Some(a), Some(b)) => Some(a.min(b)),
(Some(a), None) | (None, Some(a)) => Some(a),
(None, None) => None,
}
}
/// **The LAST AFFORDABLE DISK: how many more bytes of data set this wallet can
/// still buy, at today's prices, for this product.**
///
/// `None` when the question cannot be answered rather than 0 — an unconfigured
/// product, a cloud nothing wraps (no ceiling: the wallet is not the limit), a
/// set that will not survey, a delta that will not price. Every one of those is
/// "we could not ask", and this estate's rule is that such an answer is never
/// rendered as "no".
///
/// # Why this function exists (STINK S2)
///
/// The shop's ceiling and the wallet's ceiling are **different units**:
/// `[signups.<product>] max_sold_percent_of_capacity` is a percentage OF THE IRON, so
/// every growth raises it, while `[clouds.<n>] cap` is a fixed number that only
/// goes down. Measured 2026-09-19 in `arrival_rate.rs`: the €25 stops buying
/// disk at customer ~12 and the shop goes on selling to customer 87, because
/// the allowance rose with each growth and the wallet did not. Nothing in the
/// code held the two numbers in one place, so nothing could say how far apart
/// they were.
///
/// This is that one place. `capacity + affordable_bytes` is **the largest data
/// set this estate can ever serve without a bigger budget** — the number the
/// shop's promise should be judged against, and the number
/// [`crate::oversell::past_the_wallet`] judges it against.
///
pub fn affordable_bytes(state: &AppState, product: &str) -> Option<u64> {
let pg = state.grow.get(product)?;
let cloud = state.registry.find_cloud(&pg.set.cloud)?;
// No ceiling at all = this cloud's wallet does not bound the estate; the
// question has no answer here rather than the answer "nothing".
let remaining = ceiling_remaining(state, cloud)?;
let survey = monetize_cloud::ask::survey(&cloud, &pg.data_set(product, 0)).ok()?;
let current = survey.current_gib;
let members = survey.volumes.len().max(1) as u64;
// **The provider's price, from its cache** ([`PRICE_FRESH`], owner
// 2026-10-05: "cache the provider price for at least 10 minutes"): keyed
// by the members' surveyed sizes, so a set that grew is priced afresh. This
// reading is for SHOWING offers; a growth prices live.
let sizes: Vec<(String, u64)> = survey.volumes.iter().map(|v| (v.id.0.clone(), v.gib)).collect();
let price = |t: u64| pg.price_cached(&sizes, t, || monetize_cloud::Cloud::price(&cloud, &monetize_cloud::cloud::Priced::Grow(&pg.data_set(product, t))).ok().map(|p| p.minor_per_month));
// One GiB per member is the smallest step there is. If that does not fit,
// nothing does — and if it is free, this cloud cannot cost a growth, which
// is not the same as an infinite budget.
let step = price(current + 1)?;
if step == 0 {
return None;
}
if step > remaining {
return Some(0);
}
// Priced per GiB-month in every provider this has met, so the linear guess
// is exact; the bisect below is what makes it correct anyway if it is not.
let mut hi = current.saturating_add(remaining / step).max(current + 1);
while price(hi).is_some_and(|p| p <= remaining) {
// The guess was low (a provider that charges less per GiB above a
// threshold). Doubling the reach, never the price.
let next = current + (hi - current) * 2;
if next == hi {
break;
}
hi = next;
}
let (mut lo, mut hi) = (current, hi);
while hi - lo > 1 {
let mid = lo + (hi - lo) / 2;
match price(mid) {
Some(p) if p <= remaining => lo = mid,
Some(_) => hi = mid,
// A size that will not price is not a size we may buy.
None => hi = mid,
}
}
// A member grows by `lo − current`; the SET grows by that times its members.
Some((lo - current).saturating_mul(members).saturating_mul(1 << 30))
}
/// **How long a wallet reading is fresh: 10 minutes** (owner 2026-10-05, "at
/// least 10 minutes"; it was 5). A reading is ~11 cloud-guest ops (~43 UpCloud
/// calls, each said on the log), and it moves only when money is spent or a
/// month turns over; the shop window asks the reach per row, so a short life
/// would re-read the provider for as long as anybody browses. Within it the
/// window answers from the reading at once.
pub const WALLET_FRESH: Duration = Duration::from_secs(10 * 60);
/// **How long one provider price (`price_grow` for one target, on one set of
/// member sizes) is reused by the reading's bisect: an hour** — the UpCloud
/// provider's own price-list life (`price_ttl`), and well past the owner's
/// 10-minute floor. So a refresh of an unchanged set is ONE survey op, not a
/// survey and a dozen priced ops. Keyed by the surveyed member sizes: a set
/// that grew is priced afresh. Showing offers only — a growth prices live.
pub const PRICE_FRESH: Duration = Duration::from_secs(60 * 60);
/// **How long a plan's shop window waits on a reading that is being made** before it
/// answers `reading` (`ListBundlesResponse.reading`) and lets the console ask
/// again (`crate::plans`). Every window asked meanwhile waits on the same one reading.
pub const WINDOW_WAIT: Duration = Duration::from_secs(2);
/// How old a wallet reading may be and still be used while a fresh one is
/// being made. The wallet moves only when money is spent (a growth, which is
/// slow) or a month turns over; a reading from within the hour is the honest
/// figure a growth in flight leaves the shop with.
pub const WALLET_STALE_MAX: Duration = Duration::from_secs(60 * 60);
/// **The keeper refreshes a reading this old**, before [`WALLET_FRESH`] runs
/// out, so a customer normally finds one made — owner 2026-10-05: "warm it at
/// monetize start and refresh it in the background before it expires, so a
/// customer normally never waits". Between keeps an unchanged set costs one
/// survey op (its prices are cached for [`PRICE_FRESH`]).
pub const KEEP_WARM_EVERY: Duration = Duration::from_secs(8 * 60);
/// **The last [`affordable_bytes`] reading for one product, and whether a
/// refresh is running.**
///
/// # Why this exists (test2, 2026-09-30)
///
/// `affordable_bytes` is a survey and a BISECT of `price_grow` — about a dozen
/// ops on the cloud. On test2 the cloud is the UpCloud GUEST, where each op is
/// a fresh guest (its price book lives one op, so every op re-reads
/// `GET /1.3/price`) plus `GET /1.3/server/{uuid}` and a `GET /1.3/storage/{uuid}`
/// per member, and every op queues behind the guest's one `serial` lock, which
/// a growth holds for its whole detach-resize-attach. `oversell::admit` asked
/// it INLINE on every `Signup.Start` for a reading that refuses nothing: the
/// owner's live signups took 25.5 s and 33 s, and one did not answer for more
/// than 100 s. Now the reading is made on its own thread, one at a time, and a
/// caller waits for it at most as long as it says.
#[derive(Default)]
pub struct WalletCell {
cell: Mutex<WalletState>,
done: std::sync::Condvar,
}
#[derive(Default)]
struct WalletState {
/// When the last reading was made, and what it said.
last: Option<(std::time::Instant, Option<u64>)>,
/// **Whether the cloud's grow gate was open at the last reading.** Asked
/// on the reading's own thread: on the UpCloud guest `grow_gate` is an OP
/// behind the guest's one lock, and test2's shop window queued 19.5 s
/// behind a reading asking it inline (2026-10-05).
gate: Option<bool>,
/// A refresh thread is running.
refreshing: bool,
/// Bumped by every refresh that ends, so a waiter knows one did.
generation: u64,
}
/// What [`affordable_within`] could say in the time it was given.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Wallet {
/// A reading: [`affordable_bytes`]'s own answer (`None` = the wallet
/// cannot be asked). Fresh, or the last one within [`WALLET_STALE_MAX`].
Read(Option<u64>),
/// No reading in time: none has been made yet, or the last is too old,
/// and the refresh did not finish within the wait.
NotInTime,
}
/// **[`affordable_bytes`], without making the caller wait for the cloud past
/// `wait`.** Fresh (younger than [`WALLET_FRESH`]): answered at once. Otherwise a
/// refresh is started on its own thread — at most one per product — and the
/// caller waits up to `wait` for it; late, the last reading answers if it is
/// younger than [`WALLET_STALE_MAX`], else [`Wallet::NotInTime`].
///
/// An UNARMED state (no [`arm`]: a bare test rig) has no thread to hand
/// the work to and reads inline, exactly as before.
pub fn affordable_within(state: &AppState, product: &str, wait: Duration) -> Wallet {
let Some(pg) = state.grow.get(product) else { return Wallet::Read(None) };
let Some(me) = state.grow.armed.get().and_then(std::sync::Weak::upgrade) else {
let (gate, v) = read_wallet(state, product);
let mut c = pg.wallet.cell.lock().unwrap_or_else(|e| e.into_inner());
c.gate = gate;
return Wallet::Read(v);
};
let w = &pg.wallet;
let mut c = w.cell.lock().unwrap_or_else(|e| e.into_inner());
if let Some((at, v)) = c.last {
if at.elapsed() < WALLET_FRESH {
return Wallet::Read(v);
}
}
// **ONE reading at a time** (single-flight): a caller that finds one
// running waits on it below, and never starts a second survey or bisect.
start_reading(&me, product, &pg, &mut c);
let seen = c.generation;
let until = std::time::Instant::now() + wait;
while c.generation == seen {
let left = until.saturating_duration_since(std::time::Instant::now());
if left.is_zero() {
break;
}
c = w.done.wait_timeout(c, left).unwrap_or_else(|e| e.into_inner()).0;
}
match c.last {
Some((_, v)) if c.generation != seen => Wallet::Read(v),
Some((at, v)) if at.elapsed() < WALLET_STALE_MAX => Wallet::Read(v),
_ => Wallet::NotInTime,
}
}
/// **Start a reading of `product`'s wallet as a jera job, unless one is
/// already running.** The caller holds the cell's lock, so two callers cannot
/// both find none running.
fn start_reading(me: &Arc<AppState>, product: &str, pg: &Arc<SetWallet>, c: &mut WalletState) {
if c.refreshing {
return;
}
c.refreshing = true;
let product_id = product.to_owned();
let pg2 = Arc::clone(pg);
let me = Arc::clone(me);
// A jera job on the product's own lane (A32-14): a reading that panics is an ALERT line and a
// FAILED row, and the cell is let go on every path (the raw thread left `refreshing` set for
// good, and no reading was made again).
let started = crate::jobs::once(WALLET_JOB, &crate::jobs::of(&me, product), move || {
let pg3 = Arc::clone(&pg2);
let _let_go = crate::jobs::Finally(move || {
let mut c = pg3.wallet.cell.lock().unwrap_or_else(|e| e.into_inner());
if c.refreshing {
c.refreshing = false;
pg3.wallet.done.notify_all();
}
});
let began = std::time::Instant::now();
let (gate, v) = read_wallet(&me, &product_id);
drop(me);
let took = began.elapsed();
if took > Duration::from_secs(2) {
eprintln!(
"monetize-server: {product_id}: the wallet reading (the set's survey and the price bisect at the cloud) took {} ms — off every customer's path",
took.as_millis()
);
}
let mut c = pg2.wallet.cell.lock().unwrap_or_else(|e| e.into_inner());
c.last = Some((std::time::Instant::now(), v));
c.gate = gate;
c.refreshing = false;
c.generation += 1;
pg2.wallet.done.notify_all();
Ok(())
});
match started {
Ok(true) => {}
// The product's last reading is still closing its lane: the next ask starts anew.
Ok(false) => c.refreshing = false,
Err(e) => {
c.refreshing = false;
eprintln!("monetize-server: {product}: the wallet reading could not start: {e}");
}
}
}
/// The jera job kind of one wallet reading (A32-14).
pub const WALLET_JOB: &str = "monetize_wallet_reading";
/// The jera duty that keeps the wallet readings warm (A32-14).
pub const WALLET_KEEPER: &str = "monetize_wallet_keeper";
/// **One reading**: the cloud's grow gate, then [`affordable_bytes`]. Both are
/// asked HERE, in the reading's own jera job, so nothing on a customer's path is
/// an op on the cloud guest.
fn read_wallet(state: &AppState, product: &str) -> (Option<bool>, Option<u64>) {
let gate = state.grow.get(product).and_then(|pg| state.registry.find_cloud(&pg.set.cloud)).map(|cloud| monetize_cloud::ask::grow_gate(&cloud).is_ok());
(gate, affordable_bytes(state, product))
}
/// The reading as it stands, never waiting: [`affordable_within`] with no wait.
pub fn wallet_reading(state: &AppState, product: &str) -> Wallet {
affordable_within(state, product, Duration::ZERO)
}
/// **The keeper**: refresh every product's reading once it is
/// [`GrowSets::keep_warm`] old, in the background, so it never runs out under
/// a customer. Held WEAKLY; it ends with the state.
fn keep_warm(state: &Arc<AppState>) {
let Some(every) = state.grow.keep_warm else { return };
let weak = Arc::downgrade(state);
let tick = (every / 4).clamp(Duration::from_millis(100), Duration::from_secs(15));
// A jera duty (A32-14): a look that panics is said and the next one runs; it ends with the state.
let mut first = true;
let started = crate::jobs::every(WALLET_KEEPER, tick, move || {
// The first look only waits, as the thread's first sleep did.
if std::mem::take(&mut first) {
return Ok(crate::jobs::Round::Again);
}
let Some(me) = weak.upgrade() else { return Ok(crate::jobs::Round::Done) };
for (id, pg) in me.grow.products.iter() {
let mut c = pg.wallet.cell.lock().unwrap_or_else(|e| e.into_inner());
let due = c.last.is_none_or(|(at, _)| at.elapsed() >= every);
if due {
start_reading(&me, id, pg, &mut c);
}
}
Ok(crate::jobs::Round::Again)
});
if let Err(e) = started {
eprintln!("monetize-server: ALERT jobs.unstarted {WALLET_KEEPER}: the wallet keeper could not start: {e}; readings are made when a customer asks");
}
}
/// **Make every reading `by` older** — the wallet reading and its prices — so a test can stand at any point of their lives without waiting.
pub fn age_readings_for_test(state: &AppState, by: Duration) {
for pg in state.grow.products.values() {
let mut c = pg.wallet.cell.lock().unwrap_or_else(|e| e.into_inner());
if let Some((at, v)) = c.last {
c.last = Some((at.checked_sub(by).unwrap_or(at), v));
}
drop(c);
for (at, _) in pg.prices.lock().unwrap_or_else(|e| e.into_inner()).values_mut() {
*at = at.checked_sub(by).unwrap_or(*at);
}
}
}
/// **Forget every reading**, as a monetize that has just started, or that sat
/// idle past [`WALLET_STALE_MAX`], has none. For tests and the bench.
pub fn forget_readings_for_test(state: &AppState) {
for pg in state.grow.products.values() {
{
let mut c = pg.wallet.cell.lock().unwrap_or_else(|e| e.into_inner());
c.last = None;
c.gate = None;
}
pg.prices.lock().unwrap_or_else(|e| e.into_inner()).clear();
}
}
/// Age every reading past [`WALLET_FRESH`], so the
/// next caller is the one who would have asked the cloud. For tests.
pub fn expire_readings_for_test(state: &AppState) {
for pg in state.grow.products.values() {
let mut c = pg.wallet.cell.lock().unwrap_or_else(|e| e.into_inner());
if let Some((at, v)) = c.last {
c.last = Some((at.checked_sub(WALLET_FRESH + Duration::from_secs(1)).unwrap_or(at), v));
}
}
}
/// **Arm the state**: wallet readings are made on their own job from now on (warmed now, and
/// kept warm), and the `server` cloud QUEUES a grant past its walls where the set is resized
/// later ([`queues_for_resize`]). Held WEAKLY — the registry lives inside the state.
pub fn arm(state: &Arc<AppState>) {
let _ = state.grow.armed.set(Arc::downgrade(state));
// Warm every product's wallet reading now, on its own thread, so the
// first customer finds one made — and keep it warm from here on.
for id in state.grow.products.keys() {
let _ = wallet_reading(state, id);
}
keep_warm(state);
let weak = Arc::downgrade(state);
state.registry.grow_reach.queue_with(Arc::new(move |id: &str| {
let Some(s) = weak.upgrade() else { return false };
s.registry.find_product(id).is_some_and(|product| queues_for_resize(&s, product))
}));
}
/// ★ **Is an order past the room QUEUED for this set's resize?** (round 5.3,
/// owner 2026-10-08: "the customer must not know about a cap! it must be
/// queued and resized later".)
///
/// `true` for a product whose OWN estate grows its set when told (UW-4, the pull channel,
/// [`queue_past_room`]): an order past the room on hand is SOLD, its estate is told by a signed
/// `grow_room` order, and the OPERATOR hears what waits, never the customer. A box that cannot
/// grow at all is no queue — its refusal stands, in words that name no room and no cap.
pub fn queues_for_resize(_state: &AppState, product: monetize_product::Served<'_>) -> bool {
product.queues_room()
}
/// ★ UW-4: **an order placed past the room of an estate that grows its own set when told**
/// ([`monetize_product::pulled::Pulled::queues_room`], the pull channel): the estate serving
/// `tenant` is told by [`monetize_product::pulled::Pulled::queue_room`] (a signed `grow_room`
/// order it pulls; the box then grows through the v2 path), and the operator by the returned
/// line. `None` when the order fits, or the product queues no room this way.
pub fn queue_past_room(_state: &AppState, product: monetize_product::Served<'_>, tenant: &monetize_product::TenantId, caps: &BTreeMap<String, u64>) -> Option<String> {
if !product.queues_room() {
return None;
}
let why = product.can_absorb(tenant, caps).err()?;
let demand = stored_demand(product, caps);
let line = match product.queue_room(tenant, demand, &format!("{}: {why}", tenant.0)) {
Ok(()) => format!(
"monetize-server: {}/{}: {} of stored bytes sold past the room ({why}) — queued; the estate is told to grow its set (grow_room)",
product.id(),
tenant.0,
monetize_product::bytes(demand)
),
Err(e) => format!("monetize-server: ALERT grow.not_told {}/{}: sold past the room ({why}) and the estate could NOT be told to grow: {e}", product.id(), tenant.0),
};
Some(line)
}
/// Pure: **the stored bytes `caps` asks of `product`'s set** — every meter the product backs
/// with disk (`Backing::DiskBytes`: Holger's `disk_bytes`, Gunnar's `pack_bytes` and
/// `lfs_bytes`), and Gunnar's two by name for a plugin that declares no meters. Each meter
/// once.
pub fn stored_demand(product: monetize_product::Served<'_>, caps: &BTreeMap<String, u64>) -> u64 {
let mut names: std::collections::BTreeSet<&str> = product.meters().iter().filter(|m| m.backing == Some(monetize_product::Backing::DiskBytes)).map(|m| m.name).collect();
names.extend(["pack_bytes", "lfs_bytes"]);
names.iter().filter_map(|n| caps.get(*n)).fold(0u64, |a, b| a.saturating_add(*b))
}
/// **The product's `can_absorb`, with the queue past the room**: a refusal of the box's
/// CURRENT room stands unless the order asks stored bytes of a set that is resized later
/// ([`queues_for_resize`]) — then it is SOLD and served from the box (`OnHand`) once the estate
/// has grown it; the operator, not the customer, hears what waits ([`queue_past_room`]).
pub fn can_absorb_queued(state: &AppState, product: monetize_product::Served<'_>, tenant: &monetize_product::TenantId, caps: &BTreeMap<String, u64>) -> Result<monetize_product::Absorb, String> {
let why = match product.can_absorb(tenant, caps) {
Ok(a) => return Ok(a),
Err(why) => why,
};
let demand = stored_demand(product, caps);
if demand > 0 && queues_for_resize(state, product) {
eprintln!(
"monetize-server: {}: {} of stored bytes is past the room on hand ({why}) — SOLD and QUEUED; the estate grows its set when told",
product.id(),
monetize_product::bytes(demand)
);
return Ok(monetize_product::Absorb::OnHand);
}
Err(why)
}
/// **The first growth-window boundary at or after `t`** — the daily batch
/// (`grow_batch_every_secs`, offset `grow_batch_at_secs` from the epoch's midnight, UTC); `t`
/// itself when the daily batch is off. A twin's own plan change waits for it.
pub fn next_window_boundary(cfg: &crate::config::Config, t: u64) -> u64 {
let every = cfg.grow_batch_every_secs * 1000;
if every == 0 {
return t;
}
let at = (cfg.grow_batch_at_secs * 1000) % every;
if t <= at {
return at;
}
at + (t - at).div_ceil(every) * every
}
#[cfg(test)]
mod tests {
use super::next_window_boundary;
/// The daily boundary: `t` with the batch off; else the first boundary at or after `t`.
#[test]
fn the_next_window_boundary_is_the_first_daily_boundary_at_or_after_now() {
let mut cfg = crate::config::Config::default();
assert_eq!(next_window_boundary(&cfg, 12_345), 12_345, "off: now");
cfg.grow_batch_every_secs = 86_400;
cfg.grow_batch_at_secs = 7_200;
assert_eq!(next_window_boundary(&cfg, 1), 7_200_000);
assert_eq!(next_window_boundary(&cfg, 7_200_000), 7_200_000, "on the boundary");
assert_eq!(next_window_boundary(&cfg, 7_200_001), 7_200_000 + 86_400_000);
}
}