crates/monetize-cloud/src/pool.rs
//! **The pool ledger: what of the provisioned disk is already sold, and which
//! `(reference, resource)` pairs this box has acted on.**
//!
//! One JSON file at a configured path, rewritten atomically (`.tmp` + rename) on every
//! change. It is small — one entry per sold resource — and an operator can read it.
//!
//! Shared by every provider that keeps a pool: `upcloud` owns a [`Ledger`] file of
//! its own, and the `sandbox` cloud embeds a [`LedgerData`] inside its single state
//! file. The sold/grant arithmetic lives on [`LedgerData`] so both route through
//! the one writer of it, and [`write_atomically`] is the one way either file
//! reaches disk.
//!
//! # The pool policy, in three lines
//!
//! 1. **Provisioned** GiB on a server = the sizes of its attached storages as the
//! provider reports them right now, minus the storages listed in `pool_exclude`
//! (the boot disk). The provider is the truth for what exists.
//! 2. **Sold** GiB per server = this ledger's count of GiB granted to references,
//! whether the grant was `OnHand` or came with a bought disk. **Free = provisioned −
//! sold − reserve.**
//! 3. A `Disk` request that fits in *free* is `OnHand` and costs nothing; one that does
//! not buys a new storage of `max(requested, disk_step)` rounded up to `disk_step`,
//! attaches it, and the surplus over the request becomes pool for the next tenant.
//!
//! Growing the filesystem on the box (gunnar's `/data` is `md` RAID0 over the members)
//! is the product's job, not this crate's: monetize buys and attaches; the appliance
//! adopts.
//!
//! # Why a write-ahead intent
//!
//! `ensure` may die between "the provider created it" and "we wrote it down". The label
//! on the provider-side resource covers disks and servers: a retry finds it by
//! `monetize_ref`+`monetize_res` and adopts it. A **plan change** has no resource of
//! its own to label, so its intent — *from* which plan — is written here **before** the
//! `PUT`. A retry then knows the delta to charge even when the server already reports
//! the new plan.
use crate::{CloudError, Price};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum Grant {
/// Served from the pool; `gib` was added to `sold[server]`.
OnHand { server: String, gib: u64, free_after: u64 },
/// Bought. `id` is the `ResourceId::id` (`storage/<uuid>`, `server/<uuid>`,
/// `compute/<uuid>/<from_plan>`). For a disk, `gib` was added to `sold[server]`.
Provisioned {
id: String,
cost: Price,
#[serde(default)]
server: Option<String>,
#[serde(default)]
gib: u64,
},
/// A plan change that was started and not yet confirmed.
ComputeIntent { server: String, from_plan: String, to_plan: String },
}
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct LedgerData {
#[serde(default)]
pub version: u32,
/// server uuid → GiB sold out of that server's attached storage.
#[serde(default)]
pub sold: BTreeMap<String, u64>,
/// `"<reference>|<resource fingerprint>"` → what was done.
#[serde(default)]
pub grants: BTreeMap<String, Grant>,
}
impl LedgerData {
pub fn grant(&self, key: &str) -> Option<&Grant> {
self.grants.get(key)
}
pub fn sold(&self, server: &str) -> u64 {
self.sold.get(server).copied().unwrap_or(0)
}
/// Record a grant; a disk grant adds its GiB to `sold[server]`.
pub fn record(&mut self, key: &str, grant: Grant) {
match &grant {
Grant::OnHand { server, gib, .. } => *self.sold.entry(server.clone()).or_default() += gib,
Grant::Provisioned { server: Some(server), gib, .. } if *gib > 0 => {
*self.sold.entry(server.clone()).or_default() += gib
}
_ => {}
}
self.grants.insert(key.to_string(), grant);
}
/// Replace an intent with its outcome without touching `sold`.
pub fn replace(&mut self, key: &str, grant: Grant) {
self.grants.insert(key.to_string(), grant);
}
/// Drop a grant; a disk grant gives its GiB back to the pool count. Answers
/// whether there was one to drop.
pub fn forget(&mut self, key: &str) -> bool {
let Some(g) = self.grants.remove(key) else { return false };
let (server, gib) = match g {
Grant::OnHand { server, gib, .. } => (Some(server), gib),
Grant::Provisioned { server, gib, .. } => (server, gib),
Grant::ComputeIntent { .. } => (None, 0),
};
if let (Some(s), true) = (server, gib > 0) {
if let Some(v) = self.sold.get_mut(&s) {
*v = v.saturating_sub(gib);
}
}
true
}
/// The grant keys whose `Provisioned.id` is `id` — used by `destroy`.
pub fn keys_for_id(&self, id: &str) -> Vec<String> {
self.grants
.iter()
.filter(|(_, g)| matches!(g, Grant::Provisioned { id: gid, .. } if gid == id))
.map(|(k, _)| k.clone())
.collect()
}
}
/// `bytes` to `path` via `<path>.tmp` + rename, creating the directory: a reader
/// never sees a half-written ledger. `what` names the file in the error.
pub fn write_atomically(path: &Path, bytes: &[u8], what: &str) -> Result<(), CloudError> {
let tmp = path.with_extension("json.tmp");
if let Some(dir) = path.parent().filter(|p| !p.as_os_str().is_empty()) {
std::fs::create_dir_all(dir)
.map_err(|e| CloudError::Unavailable(format!("{what} dir {}: {e}", dir.display())))?;
}
std::fs::write(&tmp, bytes)
.and_then(|_| std::fs::rename(&tmp, path))
.map_err(|e| CloudError::Unavailable(format!("{what} write {}: {e}", path.display())))
}
/// **Where a [`Ledger`] keeps its bytes.** A file beside the server's ledger
/// for a linked provider ([`FileStore`]); the host's durable memory for a
/// provider running as a guest, which has no filesystem and is a fresh
/// instance per call. The ledger's arithmetic and its write-on-every-change
/// rule are the same whichever it is.
pub trait LedgerStore: Send {
/// The bytes last saved, or `None` when nothing was ever saved.
fn load(&self) -> Result<Option<Vec<u8>>, CloudError>;
fn save(&self, bytes: &[u8]) -> Result<(), CloudError>;
/// For a refusal: where the ledger lives.
fn describe(&self) -> String;
}
/// The ledger as one JSON file, rewritten atomically.
pub struct FileStore(pub PathBuf);
impl LedgerStore for FileStore {
fn load(&self) -> Result<Option<Vec<u8>>, CloudError> {
match std::fs::read(&self.0) {
Ok(bytes) => Ok(Some(bytes)),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(e) => Err(CloudError::Unavailable(format!("pool ledger {}: {e}", self.0.display()))),
}
}
fn save(&self, bytes: &[u8]) -> Result<(), CloudError> {
write_atomically(&self.0, bytes, "pool ledger")
}
fn describe(&self) -> String {
self.0.display().to_string()
}
}
pub struct Ledger {
store: Box<dyn LedgerStore>,
data: LedgerData,
}
impl Ledger {
/// Load, or start empty if the file does not exist yet. A file that exists but does
/// not parse is `Unavailable` — guessing over a corrupt ledger could sell the same
/// gigabyte twice.
pub fn open(path: &Path) -> Result<Ledger, CloudError> {
Ledger::with_store(Box::new(FileStore(path.to_path_buf())))
}
/// [`Ledger::open`] over any [`LedgerStore`]. Stored bytes that do not
/// parse are `Unavailable`, exactly as a corrupt file is.
pub fn with_store(store: Box<dyn LedgerStore>) -> Result<Ledger, CloudError> {
let data = match store.load()? {
Some(bytes) => serde_json::from_slice::<LedgerData>(&bytes)
.map_err(|e| CloudError::Unavailable(format!("pool ledger {} unreadable: {e}", store.describe())))?,
None => LedgerData { version: 1, ..Default::default() },
};
Ok(Ledger { store, data })
}
pub fn key(reference: &str, fingerprint: &str) -> String {
format!("{reference}|{fingerprint}")
}
pub fn data(&self) -> &LedgerData {
&self.data
}
pub fn grant(&self, key: &str) -> Option<&Grant> {
self.data.grant(key)
}
pub fn sold(&self, server: &str) -> u64 {
self.data.sold(server)
}
pub fn record(&mut self, key: &str, grant: Grant) -> Result<(), CloudError> {
self.data.record(key, grant);
self.flush()
}
/// Replace an intent with its outcome without touching `sold`.
pub fn replace(&mut self, key: &str, grant: Grant) -> Result<(), CloudError> {
self.data.replace(key, grant);
self.flush()
}
/// Drop a grant; a disk grant gives its GiB back to the pool count.
pub fn forget(&mut self, key: &str) -> Result<(), CloudError> {
if self.data.forget(key) {
self.flush()?;
}
Ok(())
}
/// The grant keys whose `Provisioned.id` is `id` — used by `destroy`.
pub fn keys_for_id(&self, id: &str) -> Vec<String> {
self.data.keys_for_id(id)
}
fn flush(&self) -> Result<(), CloudError> {
let bytes = serde_json::to_vec_pretty(&self.data)
.map_err(|e| CloudError::Unavailable(format!("pool ledger encode: {e}")))?;
self.store.save(&bytes)
}
}
#[cfg(test)]
mod tests {
use super::*;
use monetize_pay::Currency;
fn tmp(name: &str) -> PathBuf {
let dir = std::env::temp_dir().join(format!("monetize-cloud-ledger-{}-{name}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
dir.join("pool.json")
}
#[test]
fn counts_sold_and_survives_reopen() {
let p = tmp("reopen");
let mut l = Ledger::open(&p).unwrap();
assert_eq!(l.sold("s1"), 0);
l.record("r1|disk", Grant::OnHand { server: "s1".into(), gib: 10, free_after: 5 }).unwrap();
l.record(
"r2|disk",
Grant::Provisioned {
id: "storage/u".into(),
cost: Price { minor_per_month: 1, currency: Currency::Eur },
server: Some("s1".into()),
gib: 20,
},
)
.unwrap();
assert_eq!(l.sold("s1"), 30);
let l2 = Ledger::open(&p).unwrap();
assert_eq!(l2.sold("s1"), 30);
assert!(matches!(l2.grant("r1|disk"), Some(Grant::OnHand { gib: 10, .. })));
assert_eq!(l2.keys_for_id("storage/u"), vec!["r2|disk".to_string()]);
}
#[test]
fn forget_returns_gib_and_corrupt_file_is_refused_not_guessed() {
let p = tmp("forget");
let mut l = Ledger::open(&p).unwrap();
l.record("r|d", Grant::OnHand { server: "s".into(), gib: 7, free_after: 0 }).unwrap();
l.forget("r|d").unwrap();
assert_eq!(l.sold("s"), 0);
assert!(l.grant("r|d").is_none());
std::fs::write(&p, b"{ not json").unwrap();
assert!(matches!(Ledger::open(&p), Err(CloudError::Unavailable(_))));
}
}