crates/monetize-server/src/hot.rs
//! **Vendors and clouds that change while the server runs.**
//!
//! monetize runs as a thread inside the front's PID 1, and the owner's rule
//! there is: no restarts. So a `[vendors.<name>]` or `[clouds.<name>]` block
//! that is added, edited, disabled or removed is applied LIVE, by the same
//! watcher that takes product guests in ([`crate::guests`]).
//!
//! ```text
//! Registry ── Slots<HotCloud> ── HotCloud ── RwLock<Option<(Arc<dyn Cloud>, Option<Arc<Cap>>)>>
//! │ (the capped provider serving NOW)
//! └── gate: RwLock<()> — every spending call holds it
//! for reading; a swap takes it for writing
//! ```
//!
//! * **A slot is never emptied or moved**, so `Registry::clouds()` and
//! `Registry::vendors()` still hand out plain borrows: a slot, once filled,
//! stays; only what it points at changes.
//! * **A replacement swaps the WHOLE provider.** Calls in flight finish on the
//! instance they started on (they hold its `Arc`); the next call gets the
//! new one. A retired slot answers every call with a named refusal and is
//! not listed.
//! * **The spend ceiling is reconciled BEFORE a cloud is published.** The
//! registry builds the replacement off the serving path, then takes the
//! slot's gate for writing — which waits for any `ensure` in flight — wraps
//! the new provider in its `SpendCap` (reading the cap's file only now, so it
//! sees every purchase the old one committed), corrects it from the ledger
//! and only then swaps. No purchase runs between the reconcile and the swap.
//!
//! The borrowed tables the traits answer (`zones()`, `markets()`) are interned
//! once per distinct value ([`intern`]), so a hot reload leaks nothing new.
use std::sync::{Arc, OnceLock, RwLock, RwLockWriteGuard};
use monetize_cloud::cap::Cap;
use monetize_cloud::cloud::{Boot, Booted, Catalog, Change, Changed, CloudManifest, Give, Listed, Listing, Priced, RoomLeft};
use monetize_cloud::lay::{EstatePlan, LaidEstate};
use monetize_cloud::{grow, Cloud, CloudError, Price, Reclaimed, Zone};
use monetize_pay::payment::{Charge, Of, Page, PayManifest, Return, Standing};
use monetize_pay::{CallbackEvent, ChargeHandle, ChargeStatus, Market, Money, PayError, Payment, RawCallback, Settlement, Vendor};
/// How many vendors, and how many clouds, one process can hold — at start
/// and added live together. A slot is never freed (a retired one is reused
/// if its name comes back), so a deployment that meets the bound is refused
/// by name and restarts.
pub const PROVIDER_SLOTS: usize = 64;
/// **Append-only slots**: filled once, never emptied, so a borrow of one
/// outlives any number of swaps inside it.
pub struct Slots<T>(Box<[OnceLock<Arc<T>>]>);
impl<T> Default for Slots<T> {
fn default() -> Self {
Slots((0..PROVIDER_SLOTS).map(|_| OnceLock::new()).collect())
}
}
impl<T> Slots<T> {
pub fn iter(&self) -> impl Iterator<Item = &Arc<T>> {
self.0.iter().map_while(|s| s.get())
}
/// Fill the first empty slot. One writer at a time (the registry's
/// builder, or `apply_providers` under its lock).
pub fn push(&self, t: Arc<T>) -> Result<(), Arc<T>> {
match self.0.iter().find(|s| s.get().is_none()) {
Some(slot) => slot.set(t),
None => Err(t),
}
}
}
/// **A borrowed table, interned**: the same zones or markets always answer
/// the same `&'static` slice, so swapping a provider a thousand times leaks
/// one copy per distinct value, not one per swap.
pub fn intern<T: Clone + PartialEq + Send + Sync + 'static>(pool: &'static std::sync::Mutex<Vec<&'static [T]>>, v: &[T]) -> &'static [T] {
let mut p = pool.lock().unwrap_or_else(|e| e.into_inner());
if let Some(found) = p.iter().find(|s| **s == v) {
return found;
}
let leaked: &'static [T] = Box::leak(v.to_vec().into_boxed_slice());
p.push(leaked);
leaked
}
static ZONES: std::sync::Mutex<Vec<&'static [Zone]>> = std::sync::Mutex::new(Vec::new());
static MARKETS: std::sync::Mutex<Vec<&'static [Market]>> = std::sync::Mutex::new(Vec::new());
// ── clouds ──────────────────────────────────────────────────────────────────
/// The serving cloud and the cap's shared ledger beside it.
type Served = (Arc<dyn Cloud>, Option<Arc<Cap>>);
/// **One cloud, served by whichever cloud was last published into it**, with the spend
/// cap its guard shares ([`monetize_cloud::cap::Cap`], `None` for a cloud nothing caps).
/// Its NAME and ZONES are data the slot holds (read once at publish). A spend (a lay, a
/// plan change, a growth, an ensure) runs under the read half of `gate`; a live swap takes
/// the write half, so it waits for spends in flight and no spend starts on a half-swapped
/// cloud.
pub struct HotCloud {
name: &'static str,
cell: RwLock<Option<Served>>,
zones: RwLock<&'static [Zone]>,
/// The serving cloud's manifest, read once at publish: its name, zones, setting names and
/// grow switch are its block's, so a changed block publishes again.
manifest: RwLock<CloudManifest>,
gate: RwLock<()>,
source: std::sync::Mutex<String>,
account: std::sync::Mutex<(Option<String>, bool)>,
}
const ACTS: bool = true;
const READS: bool = false;
impl HotCloud {
/// The ACCOUNT's standing as the last call left it (CODE RED audit 5): an `Exhausted`
/// answer is remembered; the next call that ACTS and goes through clears it.
fn seen<T>(&self, acts: bool, r: Result<T, CloudError>) -> Result<T, CloudError> {
let mut a = self.account.lock().unwrap_or_else(|p| p.into_inner());
match &r {
Err(CloudError::Exhausted(w)) if a.0.as_deref() != Some(w.as_str()) => *a = (Some(w.clone()), true),
Ok(_) if acts && a.0.is_some() => *a = (None, true),
_ => {}
}
r
}
/// What changed about the account since the last ask: `Some(Some(words))` exhausted,
/// `Some(None)` working again, `None` nothing new.
pub fn account_change(&self) -> Option<Option<String>> {
let mut a = self.account.lock().unwrap_or_else(|p| p.into_inner());
if !a.1 {
return None;
}
a.1 = false;
Some(a.0.clone())
}
/// A slot serving `p`, capped by `cap` (its guard's shared ledger) when one wraps it.
pub fn serving(p: Box<dyn Cloud>, cap: Option<Arc<Cap>>, source: String) -> HotCloud {
let m = p.manifest();
HotCloud {
name: intern_name(&m.provider),
cell: RwLock::new(Some((Arc::from(p), cap))),
zones: RwLock::new(intern(&ZONES, &m.zones)),
manifest: RwLock::new(m.clone()),
gate: RwLock::new(()),
source: std::sync::Mutex::new(source),
account: std::sync::Mutex::new((None, false)),
}
}
pub fn name(&self) -> &'static str {
self.name
}
/// The zones it lays in, as its manifest said at publish.
pub fn zones(&self) -> &'static [Zone] {
*self.zones.read().unwrap_or_else(|p| p.into_inner())
}
pub fn current(&self) -> Option<Arc<dyn Cloud>> {
self.cell.read().unwrap_or_else(|p| p.into_inner()).as_ref().map(|(c, _)| Arc::clone(c))
}
/// The spend cap's ledger the serving cloud's guard shares; `None`: nothing caps it.
pub fn cap(&self) -> Option<Arc<Cap>> {
self.cell.read().unwrap_or_else(|p| p.into_inner()).as_ref().and_then(|(_, cap)| cap.clone())
}
pub fn is_serving(&self) -> bool {
self.cell.read().unwrap_or_else(|p| p.into_inner()).is_some()
}
pub fn source(&self) -> String {
self.source.lock().unwrap_or_else(|p| p.into_inner()).clone()
}
/// The write half of the spend gate: held across a live swap.
pub fn quiesce(&self) -> RwLockWriteGuard<'_, ()> {
self.gate.write().unwrap_or_else(|p| p.into_inner())
}
pub fn publish(&self, next: Arc<dyn Cloud>, cap: Option<Arc<Cap>>, source: String) {
let m = next.manifest();
*self.zones.write().unwrap_or_else(|p| p.into_inner()) = intern(&ZONES, &m.zones);
*self.manifest.write().unwrap_or_else(|p| p.into_inner()) = m;
*self.cell.write().unwrap_or_else(|p| p.into_inner()) = Some((next, cap));
*self.source.lock().unwrap_or_else(|p| p.into_inner()) = source;
}
pub fn retire(&self, source: String) {
*self.cell.write().unwrap_or_else(|p| p.into_inner()) = None;
*self.source.lock().unwrap_or_else(|p| p.into_inner()) = source;
}
fn cur(&self) -> Result<Arc<dyn Cloud>, CloudError> {
self.current().ok_or_else(|| {
CloudError::Refused(format!("[clouds.{}] was retired while the server runs (disabled, or its block is gone); nothing goes through it", self.name))
})
}
fn spend_gate(&self) -> std::sync::RwLockReadGuard<'_, ()> {
self.gate.read().unwrap_or_else(|p| p.into_inner())
}
// ── the spend cap, as the host holds it ─────────────────────────────
/// The ceiling its cap holds; `None`: nothing caps it.
pub fn spend_ceiling(&self) -> Option<monetize_cloud::cap::Ceiling> {
self.cap().map(|c| c.ceiling())
}
/// What the cap's month window has committed; `None`: no window (not zero).
pub fn spent_this_month_minor(&self) -> Option<u64> {
self.cap().map(|c| c.spent_this_month())
}
/// The cap's cached total corrected upward from the ledger; nothing to correct uncapped.
pub fn reconcile_spend(&self, authoritative_total_minor: u64) -> Result<(), CloudError> {
match self.cap() {
Some(c) => self.seen(READS, c.reconcile_total(authoritative_total_minor)),
None => Ok(()),
}
}
/// **Would laying `plan` fit under the cap now?** PURE; an order that does not WAITS.
pub fn can_lay(&self, plan: &EstatePlan) -> Result<(), CloudError> {
monetize_cloud::ask::can_lay(self, plan)
}
/// What `asking` more a month falls short of the cap by; `None`: it fits, or nothing caps it.
pub fn cap_short(&self, asking: Price, mine: &[String]) -> Result<Option<monetize_cloud::cap::Short>, CloudError> {
match self.cap() {
Some(c) => self.seen(READS, c.cap_short(asking, mine)),
None => Ok(None),
}
}
}
impl Cloud for HotCloud {
/// The name, zones and setting names as read at publish; the grow switch as the serving
/// cloud says it now (a guest cloud answers it from what it asked at load: no guest call).
fn manifest(&self) -> CloudManifest {
match self.current() {
Some(c) => CloudManifest { grow_refused: c.manifest().grow_refused, ..self.manifest.read().unwrap_or_else(|p| p.into_inner()).clone() },
None => CloudManifest { provider: self.name.to_string(), zones: self.zones().to_vec(), settings: Vec::new(), grow_refused: Some(format!("[clouds.{}] was retired while the server runs", self.name)) },
}
}
fn catalog(&self, zone: &Zone) -> Result<Catalog, CloudError> {
self.seen(READS, self.cur()?.catalog(zone))
}
fn price(&self, what: &Priced<'_>) -> Result<Price, CloudError> {
self.seen(READS, self.cur()?.price(what))
}
fn lay(&self, plan: &EstatePlan, journal: &grow::Journal) -> Result<LaidEstate, CloudError> {
let _g = self.spend_gate();
self.seen(ACTS, self.cur()?.lay(plan, journal))
}
fn change(&self, change: &Change, journal: &grow::Journal) -> Result<Changed, CloudError> {
match change {
Change::LabelAdopted { .. } => self.seen(ACTS, self.cur()?.change(change, journal)),
Change::Plan { .. } | Change::Grow { .. } | Change::Ensure { .. } => {
let _g = self.spend_gate();
self.seen(ACTS, self.cur()?.change(change, journal))
}
Change::Settle { .. } => self.seen(ACTS, self.cur()?.change(change, journal)),
Change::ForgetMedia { .. } | Change::Retire { .. } => self.seen(READS, self.cur()?.change(change, journal)),
}
}
fn boot(&self, boot: &Boot, journal: &grow::Journal) -> Result<Booted, CloudError> {
self.seen(ACTS, self.cur()?.boot(boot, journal))
}
fn firewall(&self, laid: &LaidEstate) -> Result<(), CloudError> {
self.seen(READS, self.cur()?.firewall(laid))
}
fn destroy(&self, what: &Give<'_>) -> Result<Reclaimed, CloudError> {
self.seen(READS, self.cur()?.destroy(what))
}
fn inventory(&self, what: &Listing<'_>) -> Result<Listed, CloudError> {
match (what, self.current()) {
// A retired cloud has no wall to serve within: nothing goes through it.
(Listing::Room { .. }, None) => Ok(Listed::Room(RoomLeft { serve: Err(format!("[clouds.{}] was retired while the server runs", self.name)), left_bytes: None, ceilings: Default::default() })),
_ => self.seen(READS, self.cur()?.inventory(what)),
}
}
}
// ── vendors ─────────────────────────────────────────────────────────────────
/// **One vendor, served by whichever vendor was last published into it.**
/// A payment in flight finishes on the instance it started on; a callback
/// that arrives after a swap is parsed by the new one, which reads the same
/// block (and the same state files) the old one did. Its NAME and MARKETS are
/// data the slot holds (read once at publish), never asked of the vendor per call.
pub struct HotVendor {
name: &'static str,
cell: RwLock<Option<Arc<dyn Payment>>>,
markets: RwLock<&'static [Market]>,
source: std::sync::Mutex<String>,
}
impl HotVendor {
/// A linked vendor (its manifest is cheap): its name and markets read from it once.
pub fn serving(p: Box<dyn Payment>, source: String) -> HotVendor {
let m = p.manifest();
HotVendor::serving_arc(Arc::from(p), intern_name(&m.vendor), &m.markets, source)
}
/// A slot serving a vendor something else also holds — a vendor guest's
/// own `HotVendor`, which the registry replaces in place — under the name and
/// markets its builder read.
pub fn serving_arc(p: Arc<dyn Payment>, name: &'static str, markets: &[Market], source: String) -> HotVendor {
HotVendor { name, cell: RwLock::new(Some(p)), markets: RwLock::new(intern(&MARKETS, markets)), source: std::sync::Mutex::new(source) }
}
/// A vendor guest's slot: the guest's name and markets, as it answered them at build.
pub fn serving_guest(g: Arc<monetize_product_wasm::HotVendor>, source: String) -> HotVendor {
let (name, markets) = (g.name(), g.markets());
HotVendor::serving_arc(g, name, markets, source)
}
pub fn name(&self) -> &'static str {
self.name
}
/// The markets it takes money in; none once retired.
pub fn markets(&self) -> &'static [Market] {
if !self.is_serving() {
return &[];
}
*self.markets.read().unwrap_or_else(|p| p.into_inner())
}
/// **The vendor as the host holds it**: this slot's name and markets over its ten ops.
pub fn vendor(&self) -> Vendor<'_> {
Vendor::new(self.name, self.markets(), self)
}
pub fn current(&self) -> Option<Arc<dyn Payment>> {
self.cell.read().unwrap_or_else(|p| p.into_inner()).clone()
}
pub fn is_serving(&self) -> bool {
self.cell.read().unwrap_or_else(|p| p.into_inner()).is_some()
}
pub fn source(&self) -> String {
self.source.lock().unwrap_or_else(|p| p.into_inner()).clone()
}
pub fn publish(&self, next: Arc<dyn Payment>, markets: &[Market], source: String) {
*self.markets.write().unwrap_or_else(|p| p.into_inner()) = intern(&MARKETS, markets);
*self.cell.write().unwrap_or_else(|p| p.into_inner()) = Some(next);
*self.source.lock().unwrap_or_else(|p| p.into_inner()) = source;
}
pub fn retire(&self, source: String) {
*self.cell.write().unwrap_or_else(|p| p.into_inner()) = None;
*self.source.lock().unwrap_or_else(|p| p.into_inner()) = source;
}
fn cur(&self) -> Result<Arc<dyn Payment>, PayError> {
self.current().ok_or_else(|| PayError::Unavailable(format!("{} was retired while the server runs (disabled, or its block is gone)", monetize_pay::instance::block_of(self.name))))
}
}
impl Payment for HotVendor {
fn manifest(&self) -> PayManifest {
match self.current() {
Some(v) => PayManifest { vendor: self.name.to_string(), markets: self.markets().to_vec(), ..v.manifest() },
None => PayManifest { vendor: self.name.to_string(), markets: Vec::new(), settings: Vec::new(), holds: true, test_payer: None },
}
}
fn page(&self, charge: &Charge<'_>, back: &Return) -> Result<Page, PayError> {
self.cur()?.page(charge, back)
}
fn status(&self, of: &Of<'_>) -> Result<Standing, PayError> {
match (self.current(), of) {
// A page's URL never fails: a retired vendor has none to give.
(None, Of::Page(_)) => Ok(Standing::Page(None)),
_ => self.cur()?.status(of),
}
}
fn capture(&self, handle: &ChargeHandle) -> Result<ChargeStatus, PayError> {
self.cur()?.capture(handle)
}
fn release(&self, handle: &ChargeHandle) -> Result<ChargeStatus, PayError> {
self.cur()?.release(handle)
}
fn refund(&self, handle: &ChargeHandle, amount: Money) -> Result<ChargeStatus, PayError> {
self.cur()?.refund(handle, amount)
}
fn callback(&self, raw: RawCallback<'_>) -> Result<CallbackEvent, PayError> {
self.cur()?.callback(raw)
}
fn settlement(&self, handle: &ChargeHandle) -> Option<Settlement> {
self.current().and_then(|v| v.settlement(handle))
}
fn paid_since(&self, since_unix_ms: i64) -> Result<Vec<monetize_pay::reconcile::PaidAtVendor>, PayError> {
self.cur()?.paid_since(since_unix_ms)
}
fn livemode(&self) -> Option<bool> {
self.current().and_then(|v| v.livemode())
}
}
static NAMES: std::sync::Mutex<Vec<&'static str>> = std::sync::Mutex::new(Vec::new());
/// A configured name as the `&'static str` a factory's name is — once per
/// distinct name, however often the config is re-read.
pub fn intern_name(name: &str) -> &'static str {
let mut p = NAMES.lock().unwrap_or_else(|e| e.into_inner());
if let Some(found) = p.iter().find(|s| **s == name) {
return found;
}
let leaked: &'static str = Box::leak(name.to_string().into_boxed_str());
p.push(leaked);
leaked
}
#[cfg(test)]
mod gate_tests {
use super::*;
use std::sync::mpsc;
use monetize_cloud::{Ensured, Inventory, Resource, ResourceId, ResourceStatus};
/// A provider whose lay waits until the test lets it go.
struct SlowLay {
entered: std::sync::Mutex<Option<mpsc::Sender<()>>>,
go: std::sync::Mutex<mpsc::Receiver<()>>,
}
impl SlowLay {
pub fn provider(&self) -> &'static str {
"slow"
}
pub fn zones(&self) -> &[Zone] {
&[]
}
pub fn price_resource(&self, _: &Resource) -> Result<Price, CloudError> {
Err(CloudError::Refused("no".into()))
}
pub fn ensure(&self, _: &str, _: &Resource) -> Result<Ensured, CloudError> {
Err(CloudError::Refused("no".into()))
}
pub fn destroy_id(&self, _: &ResourceId) -> Result<(), CloudError> {
Ok(())
}
pub fn status(&self, _: &ResourceId) -> Result<ResourceStatus, CloudError> {
Err(CloudError::Refused("no".into()))
}
pub fn list_owned(&self) -> Result<Inventory, CloudError> {
Err(CloudError::Refused("no".into()))
}
pub fn lay_estate(&self, _: &monetize_cloud::lay::EstatePlan, _: &grow::Journal) -> Result<monetize_cloud::lay::LaidEstate, CloudError> {
if let Some(tx) = self.entered.lock().unwrap().take() {
let _ = tx.send(());
}
let _ = self.go.lock().unwrap().recv();
Err(CloudError::Unavailable("let go".into()))
}
}
impl monetize_cloud::cloud::Cloud for SlowLay {
fn manifest(&self) -> monetize_cloud::cloud::CloudManifest {
monetize_cloud::cloud::CloudManifest { provider: self.provider().to_string(), zones: self.zones().to_vec(), settings: Vec::new(), grow_refused: Some(monetize_cloud::grow::unsupported(self.provider()).to_string()) }
}
fn catalog(&self, _zone: &monetize_cloud::Zone) -> Result<monetize_cloud::cloud::Catalog, monetize_cloud::CloudError> {
Err(monetize_cloud::lay::unsupported(self.provider()))
}
fn price(&self, what: &monetize_cloud::cloud::Priced<'_>) -> Result<monetize_cloud::Price, monetize_cloud::CloudError> {
match what {
monetize_cloud::cloud::Priced::Resource(r) => self.price_resource(r),
monetize_cloud::cloud::Priced::Plan(p) => monetize_cloud::price_estate_by_rows(p, |r| self.price_resource(r), None),
monetize_cloud::cloud::Priced::Grow(_set) => Err(monetize_cloud::grow::unsupported(self.provider())),
}
}
fn lay(&self, plan: &monetize_cloud::lay::EstatePlan, journal: &monetize_cloud::grow::Journal) -> Result<monetize_cloud::lay::LaidEstate, monetize_cloud::CloudError> {
self.lay_estate(plan, journal)
}
fn change(&self, change: &monetize_cloud::cloud::Change, _journal: &monetize_cloud::grow::Journal) -> Result<monetize_cloud::cloud::Changed, monetize_cloud::CloudError> {
Ok(match change {
monetize_cloud::cloud::Change::Plan { server: _, plan: _, delta: _ } => return Err(monetize_cloud::lay::unsupported_verb(self.provider(), "change_plan")),
monetize_cloud::cloud::Change::Grow { set: _, estate: _ } => return Err(monetize_cloud::grow::unsupported(self.provider())),
monetize_cloud::cloud::Change::Settle { set: _ } => return Err(monetize_cloud::grow::unsupported(self.provider())),
monetize_cloud::cloud::Change::ForgetMedia { laid: _ } => monetize_cloud::cloud::Changed::Forgotten(Vec::new()),
monetize_cloud::cloud::Change::Ensure { reference, resource, room: _ } => monetize_cloud::cloud::Changed::Ensured(self.ensure(reference, resource)?),
monetize_cloud::cloud::Change::Retire { reference, machine } => monetize_cloud::cloud::Changed::Retired(monetize_cloud::cloud::retire_by_rows(self, reference, machine)?),
monetize_cloud::cloud::Change::LabelAdopted { .. } => return Err(monetize_cloud::lay::unsupported_verb(self.provider(), "label_adopted")),
})
}
fn boot(&self, boot: &monetize_cloud::cloud::Boot, _journal: &monetize_cloud::grow::Journal) -> Result<monetize_cloud::cloud::Booted, monetize_cloud::CloudError> {
match boot {
monetize_cloud::cloud::Boot::Keep(_k) => Err(monetize_cloud::lay::unsupported_verb(self.provider(), "keep_boot")),
monetize_cloud::cloud::Boot::Power { server: _, on: _ } => Err(monetize_cloud::lay::unsupported_verb(self.provider(), "set_power")),
monetize_cloud::cloud::Boot::Restore(_r) => Err(monetize_cloud::lay::unsupported_verb(self.provider(), "restore_from_backup")),
}
}
fn firewall(&self, _laid: &monetize_cloud::lay::LaidEstate) -> Result<(), monetize_cloud::CloudError> {
Err(monetize_cloud::lay::unsupported_verb(self.provider(), "set_estate_firewall"))
}
fn destroy(&self, what: &monetize_cloud::cloud::Give<'_>) -> Result<monetize_cloud::Reclaimed, monetize_cloud::CloudError> {
match what {
monetize_cloud::cloud::Give::References(refs) => Ok(monetize_cloud::release_for_by_rows(self.list_owned()?, refs, |id| self.destroy_id(id))),
monetize_cloud::cloud::Give::Resource(id) => {
self.destroy_id(id)?;
Ok(monetize_cloud::Reclaimed { destroyed: vec![(*id).clone()], ..monetize_cloud::Reclaimed::default() })
}
}
}
fn inventory(&self, what: &monetize_cloud::cloud::Listing<'_>) -> Result<monetize_cloud::cloud::Listed, monetize_cloud::CloudError> {
Ok(match what {
monetize_cloud::cloud::Listing::Account => monetize_cloud::cloud::Listed::Account(self.list_owned()?),
monetize_cloud::cloud::Listing::Set(_set) => return Err(monetize_cloud::grow::unsupported(self.provider())),
monetize_cloud::cloud::Listing::Status(id) => monetize_cloud::cloud::Listed::Status(self.status(id)?),
monetize_cloud::cloud::Listing::Under(refs) => monetize_cloud::cloud::Listed::Under(monetize_cloud::cloud::under(self.list_owned()?, refs)),
monetize_cloud::cloud::Listing::Room { delta: _, room: _ } => monetize_cloud::cloud::Listed::Room(monetize_cloud::cloud::RoomLeft { serve: Ok(()), left_bytes: None, ceilings: Default::default() }),
monetize_cloud::cloud::Listing::Admits(_plan) => monetize_cloud::cloud::Listed::Admitted,
})
}
}
/// ★ **A26: a lay in flight holds the spend gate** — the operator raising
/// the cap (a swap, `quiesce`) waits for it, as it waits for an `ensure`;
/// it no longer loads the lay's reservation into a new cap the old one
/// then settles behind its back.
#[test]
fn a_swap_waits_for_a_lay_in_flight() {
let (entered_tx, entered_rx) = mpsc::channel();
let (go_tx, go_rx) = mpsc::channel();
let slot = Arc::new(HotCloud::serving(Box::new(SlowLay { entered: std::sync::Mutex::new(Some(entered_tx)), go: std::sync::Mutex::new(go_rx) }), None, String::new()));
let dir = std::env::temp_dir().join(format!("monetize-hot-gate-{}", std::process::id()));
let journal = grow::Journal::open(dir.join("journal")).unwrap();
let lay = {
let slot = Arc::clone(&slot);
std::thread::spawn(move || {
let plan = monetize_cloud::lay::EstatePlan { reference: "estate/acme".into(), company: "acme".into(), zone: Zone("se-sto1".into()), machines: Vec::new(), narrate_from: String::new(), shape: None };
monetize_cloud::Cloud::lay(&slot, &plan, &journal)
})
};
entered_rx.recv().unwrap();
assert!(slot.gate.try_write().is_err(), "a swap cannot start while the lay spends");
go_tx.send(()).unwrap();
let _ = lay.join().unwrap();
assert!(slot.gate.try_write().is_ok(), "and starts once it is done");
let _ = std::fs::remove_dir_all(&dir);
}
}