diff --git a/crates/mind-body/src/cli.rs b/crates/mind-body/src/cli.rs index b1debbc..de79120 100644 --- a/crates/mind-body/src/cli.rs +++ b/crates/mind-body/src/cli.rs @@ -8,8 +8,15 @@ //! mind-control --instance brake release //! mind-control --instance dial set --heat H --base-cooldown-s S --base-run-usd U //! mind-control [--root ] --instance check +//! mind-control [--root ] swarm show +//! mind-control swarm brake hold | release +//! mind-control swarm dial set --heat H --base-rate R --base-run U --k K +//! mind-control swarm share set //! ``` //! +//! The `swarm` verbs write `/swarm.cc` and stop no unit: the swarm's +//! brake and heat 0 are read by the organs that act on the swarm's behalf. +//! //! `check` exits 0 only when the brake is released and the dial in bounds; a //! systemd unit runs it as its `ExecCondition`. It writes nothing. //! @@ -27,11 +34,13 @@ use chrono::{DateTime, Utc}; use eureka_pipeline::{RunTurn, Slug}; use rust_decimal::Decimal; -use crate::control::{BurnRate, ControlWriter, Effective, FileSource, control_path, derive, load_state}; +use crate::control::{ + BurnRate, ControlWriter, Effective, FileSource, Share, SwarmDial, SwarmWriter, control_path, derive, derive_swarm, load_state, load_swarm_state, swarm_path, +}; use crate::launch::unit; const USAGE: &str = - "usage: mind-control [--root ] --instance show | check | brake hold | brake release | dial set --heat H --base-cooldown-s S --base-run-usd U"; + "usage: mind-control [--root ] --instance show | check | brake hold | brake release | dial set --heat H --base-cooldown-s S --base-run-usd U | mind-control [--root ] swarm show | brake hold | brake release | dial set --heat H --base-rate R --base-run U --k K | share set "; /// The arguments every organ binary shares: `--instance ` and an optional /// `--endpoint rudp://:`, which `env_endpoint` (the binary's reading of @@ -86,6 +95,11 @@ pub fn run( [flag, dir, rest @ ..] if flag == "--root" => (Path::new(dir), rest), _ => (root, args), }; + if let [word, rest @ ..] = args + && word == "swarm" + { + return swarm(root, rest, set_by, now, out); + } let (instance, command) = match args { [flag, instance, command @ ..] if flag == "--instance" => (Slug(instance.clone()), command), _ => bail!("{USAGE}"), @@ -117,6 +131,87 @@ pub fn run( } } +/// The `swarm` verbs over `/swarm.cc`. Every refusal happens before the +/// writer is opened, so a refused command creates nothing. +fn swarm(root: &Path, args: &[String], set_by: &str, now: DateTime, out: &mut impl Write) -> Result<()> { + let path = swarm_path(root); + let words: Vec<&str> = args.iter().map(String::as_str).collect(); + match words.as_slice() { + ["show"] => show_swarm(&path, out), + ["brake", "hold"] => SwarmWriter::open(&path)?.set_brake(false, now, set_by), + ["brake", "release"] => SwarmWriter::open(&path)?.set_brake(true, now, set_by), + ["dial", "set", flags @ ..] => { + let dial = parse_swarm_dial(flags, now, set_by)?; + SwarmWriter::open(&path)?.set_dial(dial) + } + ["share", "set", instance, weight] => { + let share = Share { instance: (*instance).to_string(), weight: Decimal::from_str(weight).context("the weight is not a decimal")? }; + SwarmWriter::open(&path)?.set_share(share) + } + _ => bail!("{USAGE}"), + } +} + +fn parse_swarm_dial(flags: &[&str], now: DateTime, set_by: &str) -> Result { + let (mut heat, mut rate, mut run_usd, mut k) = (None, None, None, None); + let mut rest = flags.iter(); + while let Some(flag) = rest.next() { + let value = rest.next().ok_or_else(|| anyhow!("{flag} needs a value"))?; + let slot = match *flag { + "--heat" => &mut heat, + "--base-rate" => &mut rate, + "--base-run" => &mut run_usd, + "--k" => &mut k, + other => bail!("unknown flag {other:?}"), + }; + if slot.replace(*value).is_some() { + bail!("{flag} given twice"); + } + } + Ok(SwarmDial { + heat: Decimal::from_str(need(heat, "--heat")?).context("--heat is not a decimal")?, + base_rate_usd_per_h: Decimal::from_str(need(rate, "--base-rate")?).context("--base-rate is not a decimal")?, + base_run_usd: Decimal::from_str(need(run_usd, "--base-run")?).context("--base-run is not a decimal")?, + k: need(k, "--k")?.parse().context("--k is not a whole number")?, + set_at: now, + set_by: set_by.to_string(), + }) +} + +fn show_swarm(path: &Path, out: &mut impl Write) -> Result<()> { + let source = FileSource::new(path); + writeln!(out, "swarm: {}", path.display())?; + match load_swarm_state(&source) { + Ok(state) => { + match &state.brake { + Some(brake) => writeln!(out, "brake: {} (set_at {}, set_by {})", if brake.released { "released" } else { "held" }, brake.set_at.to_rfc3339(), brake.set_by)?, + None => writeln!(out, "brake: absent")?, + } + match &state.dial { + Some(dial) => writeln!( + out, + "dial: heat {} base_rate_usd_per_h {} base_run_usd {} k {} (set_at {}, set_by {})", + dial.heat, + dial.base_rate_usd_per_h, + dial.base_run_usd, + dial.k, + dial.set_at.to_rfc3339(), + dial.set_by + )?, + None => writeln!(out, "dial: absent")?, + } + let mut shares: Vec<_> = state.shares.iter().collect(); + shares.sort_by(|a, b| a.instance.cmp(&b.instance)); + for share in shares { + writeln!(out, "share: {} {}", share.instance, share.weight)?; + } + } + Err(error) => writeln!(out, "store: undecodable ({error:#})")?, + } + writeln!(out, "effective: {:?}", derive_swarm(&source))?; + Ok(()) +} + fn parse_dial(flags: &[&str], now: DateTime, set_by: &str) -> Result { let (mut heat, mut cooldown, mut run_usd) = (None, None, None); let mut rest = flags.iter(); diff --git a/crates/mind-body/src/control.rs b/crates/mind-body/src/control.rs index 9d0cac0..3082933 100644 --- a/crates/mind-body/src/control.rs +++ b/crates/mind-body/src/control.rs @@ -26,6 +26,13 @@ //! the filesystem is the authority. The writer only refuses to write through an //! existing directory, lock or store that is not root's or that group or world //! can write (`UnsafeStore`); it checks no caller identity. +//! +//! The swarm has one more store of the same kind, `CONTROL_ROOT/swarm.cc`: a +//! file beside the instance directories, so no instance name reaches it. It +//! holds the swarm's brake, the swarm dial (heat, the spend rate, the per-run +//! cap and the concurrency) and each instance's share. `read_envelope` is its +//! one reader, fail closed like `read_effective`; `Envelope::grant_for` is the +//! only way to narrow the envelope into a `Grant`, and it never widens one. use std::path::{Path, PathBuf}; use std::time::Duration; @@ -45,6 +52,14 @@ pub const CONTROL_ROOT: &str = "/etc/gamecult/minds"; pub const BRAKE_KEY: &str = "brake"; pub const DIAL_KEY: &str = "dial"; +/// The swarm's store, a file directly under `CONTROL_ROOT`. +pub const SWARM_FILE: &str = "swarm.cc"; + +/// `/swarm.cc`. +pub fn swarm_path(root: &Path) -> PathBuf { + root.join(SWARM_FILE) +} + /// `//control.cc`, after the leaf's slug grammar has passed, so /// an instance name cannot walk out of the root. pub fn control_path(root: &Path, instance: &Slug) -> Result { @@ -116,6 +131,10 @@ pub enum HeldReason { BrakeHeld, DialAbsent, DialOutOfBounds, + /// The swarm dial's heat is 0: the operator stopped everything. + HeatZero, + /// A share in the swarm store is out of bounds or not keyed by its instance. + ShareOutOfBounds, Undecodable, } @@ -258,6 +277,14 @@ fn registered_cache() -> Result { Ok(cache) } +fn swarm_cache() -> Result { + let mut cache = CultCache::new(); + cache.register_entry_type::()?; + cache.register_entry_type::()?; + cache.register_entry_type::()?; + Ok(cache) +} + /// Decodes the store into its two documents. Any envelope the cache does not /// know, or any payload that does not decode, is an error: a store the reader /// cannot fully read is not one it may act on. @@ -408,6 +435,44 @@ fn set_mode(_path: &Path, _mode: u32) -> Result<()> { Ok(()) } +/// Before the first write: the control directory exists at 0755 and the +/// lock file at 0644, set explicitly because creation takes the umask. +/// A directory, lock or store that already exists must be root's and not +/// group- or world-writable, else the write is refused (`UnsafeStore`): the +/// operator fixes the install. Called after a setter's own refusals, so a +/// refused write creates nothing. +fn prepare_store(path: &Path) -> Result<()> { + let directory = path.parent().context("control path has no directory")?; + let mut lock_name = path.file_name().context("control path has no file name")?.to_os_string(); + lock_name.push(".lock"); + let lock = path.with_file_name(lock_name); + for existing in [directory, lock.as_path(), path] { + if existing.exists() { + require_root_only_writable(existing)?; + } + } + // Every directory this write creates is 0755, ancestors included: a + // restrictive process umask would otherwise leave /etc/gamecult/minds + // untraversable by the units that only read. + let missing: Vec<&Path> = directory.ancestors().take_while(|ancestor| !ancestor.exists()).collect(); + if !missing.is_empty() { + std::fs::create_dir_all(directory).with_context(|| format!("failed to create {}", directory.display()))?; + for created in missing { + set_mode(created, DIR_MODE)?; + } + } + if !lock.exists() { + std::fs::OpenOptions::new() + .create(true) + .write(true) + .truncate(false) + .open(&lock) + .with_context(|| format!("failed to open {}", lock.display()))?; + set_mode(&lock, FILE_MODE)?; + } + Ok(()) +} + impl ControlWriter { pub fn open(path: impl Into) -> Result { let path = path.into(); @@ -417,42 +482,8 @@ impl ControlWriter { Ok(Self { cache, path }) } - /// Before the first write: the control directory exists at 0755 and the - /// lock file at 0644, set explicitly because creation takes the umask. - /// A directory, lock or store that already exists must be root's and not - /// group- or world-writable, else the write is refused (`UnsafeStore`): the - /// operator fixes the install. Called after a setter's own refusals, so a - /// refused write creates nothing. fn prepare(&self) -> Result<()> { - let directory = self.path.parent().context("control path has no directory")?; - let mut lock_name = self.path.file_name().context("control path has no file name")?.to_os_string(); - lock_name.push(".lock"); - let lock = self.path.with_file_name(lock_name); - for existing in [directory, lock.as_path(), self.path.as_path()] { - if existing.exists() { - require_root_only_writable(existing)?; - } - } - // Every directory this write creates is 0755, ancestors included: a - // restrictive process umask would otherwise leave /etc/gamecult/minds - // untraversable by the units that only read. - let missing: Vec<&Path> = directory.ancestors().take_while(|ancestor| !ancestor.exists()).collect(); - if !missing.is_empty() { - std::fs::create_dir_all(directory).with_context(|| format!("failed to create {}", directory.display()))?; - for created in missing { - set_mode(created, DIR_MODE)?; - } - } - if !lock.exists() { - std::fs::OpenOptions::new() - .create(true) - .write(true) - .truncate(false) - .open(&lock) - .with_context(|| format!("failed to open {}", lock.display()))?; - set_mode(&lock, FILE_MODE)?; - } - Ok(()) + prepare_store(&self.path) } /// After a write: the store file the atomic rename put in place takes the @@ -480,3 +511,234 @@ impl ControlWriter { self.settle() } } + +// --------------------------------------------------------------------------- +// The swarm store: `CONTROL_ROOT/swarm.cc`. +// --------------------------------------------------------------------------- + +/// The swarm dial: `heat` scales the whole swarm's activity, 0 stops it. The +/// base values say what heat 1.0 means: the total spend rate, the most one run +/// may spend, and `k`, how many turns may run at once. The envelope derives +/// rate and run cap from it, never stored. +#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] +#[cultcache(type = "eureka.control.swarm_dial.v1", schema = "ControlSwarmDial")] +pub struct SwarmDial { + #[cultcache(key = 0)] + pub heat: Decimal, + #[cultcache(key = 1)] + pub base_rate_usd_per_h: Decimal, + #[cultcache(key = 2)] + pub base_run_usd: Decimal, + #[cultcache(key = 3)] + pub k: u32, + #[cultcache(key = 4)] + pub set_at: DateTime, + #[cultcache(key = 5)] + pub set_by: String, +} + +impl SwarmDial { + /// The one bounds rule for the swarm dial: the reader holds on it and the + /// CLI refuses on it. Heat 0 is in bounds (it is the stop); the reader + /// answers it as held. + pub fn in_bounds(&self) -> bool { + self.heat >= Decimal::ZERO && self.heat <= heat_max() && self.base_rate_usd_per_h > Decimal::ZERO && self.base_run_usd > Decimal::ZERO && self.k >= 1 + } +} + +/// One instance's share of the swarm, keyed in the store by the instance name. +/// An instance with no share has weight 1. +#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] +#[cultcache(type = "eureka.control.share.v1", schema = "ControlShare")] +pub struct Share { + #[cultcache(key = 0)] + pub instance: String, + #[cultcache(key = 1)] + pub weight: Decimal, +} + +impl Share { + /// A share is a slug-named instance with a positive weight. + pub fn in_bounds(&self) -> bool { + Slug(self.instance.clone()).validate_slug().is_ok() && self.weight > Decimal::ZERO + } +} + +/// The swarm store's contents, decoded and not yet judged. +#[derive(Clone, Debug, PartialEq, Eq, Default)] +pub struct SwarmState { + pub brake: Option, + pub dial: Option, + pub shares: Vec, +} + +/// What the swarm store allows right now. +#[derive(Debug, PartialEq, Eq)] +pub enum SwarmEffective { + Held(HeldReason), + Released(Envelope), +} + +/// The swarm's budget while the brake is released and heat is above 0. Its +/// fields are private and only `read_envelope` makes one, so a run cap derived +/// from it exists only as the reading of the root-owned swarm store. +#[derive(Debug, PartialEq, Eq)] +pub struct Envelope { + rate_usd_per_h: Decimal, + run_cap_usd: Decimal, + k: u32, + shares: std::collections::BTreeMap, +} + +impl Envelope { + /// The swarm's total spend rate: `base_rate_usd_per_h x heat`. + pub fn rate_usd_per_h(&self) -> Decimal { + self.rate_usd_per_h + } + + /// The most any one run may spend: `base_run_usd x heat`. + pub fn run_cap_usd(&self) -> Decimal { + self.run_cap_usd + } + + /// How many turns may run at once. + pub fn k(&self) -> u32 { + self.k + } + + /// The instance's weight; 1 when the store holds no share for it. + pub fn share(&self, instance: &Slug) -> Decimal { + self.shares.get(&instance.0).copied().unwrap_or(Decimal::ONE) + } + + /// Narrows this envelope and the instance's own reading into the Grant for + /// one run of `budget_usd`. It is the only new way to build a Grant and it + /// never widens: `None` unless the instance's own reading is released for + /// this instance and `0 < budget_usd` is within both the swarm's run cap and + /// the instance's own. The Grant's cap is the budget. + pub fn grant_for(&self, instance: &Slug, instance_effective: &Effective, budget_usd: Decimal) -> Option { + let Effective::Released(own) = instance_effective else { + return None; + }; + if own.instance != *instance || budget_usd <= Decimal::ZERO || budget_usd > self.run_cap_usd || budget_usd > own.run_cap_usd { + return None; + } + Some(Grant { instance: instance.clone(), cadence: own.cadence, run_cap_usd: budget_usd }) + } +} + +impl SwarmState { + /// The one derivation. Order: brake, dial bounds, heat, shares. + pub(crate) fn effective(&self) -> SwarmEffective { + let held = SwarmEffective::Held; + let Some(brake) = &self.brake else { + return held(HeldReason::BrakeAbsent); + }; + if !brake.released { + return held(HeldReason::BrakeHeld); + } + let Some(dial) = &self.dial else { + return held(HeldReason::DialAbsent); + }; + if !dial.in_bounds() { + return held(HeldReason::DialOutOfBounds); + } + if dial.heat.is_zero() { + return held(HeldReason::HeatZero); + } + if !self.shares.iter().all(Share::in_bounds) { + return held(HeldReason::ShareOutOfBounds); + } + match (dial.base_rate_usd_per_h.checked_mul(dial.heat), dial.base_run_usd.checked_mul(dial.heat)) { + (Some(rate_usd_per_h), Some(run_cap_usd)) => SwarmEffective::Released(Envelope { + rate_usd_per_h, + run_cap_usd, + k: dial.k, + shares: self.shares.iter().map(|share| (share.instance.clone(), share.weight)).collect(), + }), + _ => held(HeldReason::DialOutOfBounds), + } + } +} + +/// Decodes the swarm store. An envelope the cache does not know, a payload that +/// does not decode or a share not keyed by its own instance is an error. +pub fn load_swarm_state(source: &impl ControlSource) -> Result { + let mut cache = swarm_cache()?; + for envelope in source.snapshot().context("swarm store cannot be read")? { + cache.put_raw_envelope(envelope)?; + } + let mut shares = Vec::new(); + for (key, share) in cache.get_all_with_keys::()? { + anyhow::ensure!(key == share.instance, "a share is not keyed by its instance"); + shares.push(share); + } + Ok(SwarmState { brake: cache.get::(BRAKE_KEY)?, dial: cache.get::(DIAL_KEY)?, shares }) +} + +/// The one reader of the swarm store, at `CONTROL_ROOT/swarm.cc`; the caller +/// supplies neither a root, a path nor a source. Anything but a decodable store +/// holding a released brake, an in-bounds dial with heat above 0 and in-bounds +/// shares is `Held`, with the reason. +pub fn read_envelope() -> SwarmEffective { + derive_swarm(&FileSource::new(swarm_path(&control_root()))) +} + +/// The derivation over any source; the CLI's `show` is its other caller. +pub(crate) fn derive_swarm(source: &impl ControlSource) -> SwarmEffective { + match load_swarm_state(source) { + Ok(state) => state.effective(), + Err(_) => SwarmEffective::Held(HeldReason::Undecodable), + } +} + +/// The operator's writer over the swarm file, with the same root-only refusals +/// as `ControlWriter`. Each setter replaces its document whole. +pub struct SwarmWriter { + cache: CultCache, + path: PathBuf, +} + +impl SwarmWriter { + pub fn open(path: impl Into) -> Result { + let path = path.into(); + let mut cache = swarm_cache()?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path))?; + cache.pull_all_backing_stores()?; + Ok(Self { cache, path }) + } + + fn settle(&self) -> Result<()> { + set_mode(&self.path, FILE_MODE) + } + + pub fn set_brake(&mut self, released: bool, set_at: DateTime, set_by: &str) -> Result<()> { + let brake = Brake { released, set_at, set_by: set_by.to_string() }; + prepare_store(&self.path)?; + self.cache.put(BRAKE_KEY, &brake)?; + self.settle() + } + + /// Refuses an out-of-bounds dial before anything is written. + pub fn set_dial(&mut self, dial: SwarmDial) -> Result<()> { + if !dial.in_bounds() { + anyhow::bail!( + "swarm dial refused: heat must be within 0..={}, base_rate_usd_per_h and base_run_usd above 0, k at least 1", + heat_max() + ); + } + prepare_store(&self.path)?; + self.cache.put(DIAL_KEY, &dial)?; + self.settle() + } + + /// Refuses a share for a non-slug instance or a weight at or below 0. + pub fn set_share(&mut self, share: Share) -> Result<()> { + if !share.in_bounds() { + anyhow::bail!("share refused: the instance must be a slug and the weight above 0"); + } + prepare_store(&self.path)?; + self.cache.put(share.instance.clone(), &share)?; + self.settle() + } +} diff --git a/crates/mind-body/src/control/tests.rs b/crates/mind-body/src/control/tests.rs index 0106f5c..d55795b 100644 --- a/crates/mind-body/src/control/tests.rs +++ b/crates/mind-body/src/control/tests.rs @@ -425,3 +425,221 @@ mod grant { assert!(format!("{error:#}").contains("asked the mind"), "{error:#}"); } } + +/// The swarm store: `/swarm.cc`, read by `read_envelope`. +mod swarm { + use super::*; + + fn swarm_file(dir: &Path) -> PathBuf { + swarm_path(dir) + } + + fn swarm_dial(heat: &str, rate: &str, run_usd: &str, k: u32) -> SwarmDial { + SwarmDial { heat: dec(heat), base_rate_usd_per_h: dec(rate), base_run_usd: dec(run_usd), k, set_at: at(0), set_by: "op".into() } + } + + fn share(instance: &str, weight: &str) -> Share { + Share { instance: instance.into(), weight: dec(weight) } + } + + /// A released swarm store with the given dial, through the real writer. + fn released_swarm(dir: &Path, dial: SwarmDial) -> PathBuf { + let path = swarm_file(dir); + let mut writer = SwarmWriter::open(&path).unwrap(); + writer.set_brake(true, at(0), "op").unwrap(); + writer.set_dial(dial).unwrap(); + path + } + + fn envelope_at(dir: &Path) -> SwarmEffective { + with_root(dir, read_envelope) + } + + fn released(dir: &Path) -> Envelope { + match envelope_at(dir) { + SwarmEffective::Released(envelope) => envelope, + other => panic!("expected a released envelope, got {other:?}"), + } + } + + /// A swarm dial written around the writer's bounds check. + fn write_unchecked(path: &Path, dial: &SwarmDial) { + let mut cache = swarm_cache().unwrap(); + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(path)).unwrap(); + cache.pull_all_backing_stores().unwrap(); + cache.put(DIAL_KEY, dial).unwrap(); + } + + fn write_unchecked_share(path: &Path, key: &str, share: &Share) { + let mut cache = swarm_cache().unwrap(); + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(path)).unwrap(); + cache.pull_all_backing_stores().unwrap(); + cache.put(key, share).unwrap(); + } + + fn slug(name: &str) -> Slug { + Slug(name.into()) + } + + #[test] + fn missing_or_corrupt_swarm_store_is_held() { + let dir = tempfile::tempdir().unwrap(); + // No file at all. + assert_eq!(envelope_at(dir.path()), SwarmEffective::Held(HeldReason::BrakeAbsent)); + // Bytes that are not a store. + std::fs::write(swarm_file(dir.path()), b"not a cultcache store").unwrap(); + assert_eq!(envelope_at(dir.path()), SwarmEffective::Held(HeldReason::Undecodable)); + std::fs::remove_file(swarm_file(dir.path())).unwrap(); + // A brake and no dial. + let path = swarm_file(dir.path()); + SwarmWriter::open(&path).unwrap().set_brake(true, at(0), "op").unwrap(); + assert_eq!(envelope_at(dir.path()), SwarmEffective::Held(HeldReason::DialAbsent)); + // Every out-of-bounds dial, written around the writer. + for (what, bad) in [ + ("negative heat", swarm_dial("-0.1", "5", "2", 2)), + ("heat above the maximum", swarm_dial("2.1", "5", "2", 2)), + ("zero rate", swarm_dial("1", "0", "2", 2)), + ("zero run cap", swarm_dial("1", "5", "0", 2)), + ("zero k", swarm_dial("1", "5", "2", 0)), + ] { + write_unchecked(&path, &bad); + assert_eq!(envelope_at(dir.path()), SwarmEffective::Held(HeldReason::DialOutOfBounds), "{what}"); + } + // Heat 0 is in bounds and stops everything. + write_unchecked(&path, &swarm_dial("0", "5", "2", 2)); + assert_eq!(envelope_at(dir.path()), SwarmEffective::Held(HeldReason::HeatZero)); + } + + #[test] + fn a_held_swarm_brake_stops_everything_whatever_the_dial_says() { + let dir = tempfile::tempdir().unwrap(); + let path = released_swarm(dir.path(), swarm_dial("1", "5", "2", 2)); + let _ = released(dir.path()); + SwarmWriter::open(&path).unwrap().set_brake(false, at(60), "alice").unwrap(); + assert_eq!(envelope_at(dir.path()), SwarmEffective::Held(HeldReason::BrakeHeld)); + SwarmWriter::open(&path).unwrap().set_brake(true, at(120), "bob").unwrap(); + let _ = released(dir.path()); + // Heat 0 stops it again, with the brake released. + SwarmWriter::open(&path).unwrap().set_dial(swarm_dial("0", "5", "2", 2)).unwrap(); + assert_eq!(envelope_at(dir.path()), SwarmEffective::Held(HeldReason::HeatZero)); + // The dial survived the brake writes, provenance included. + let state = load_swarm_state(&FileSource::new(&path)).unwrap(); + assert_eq!(state.brake.unwrap().set_by, "bob"); + assert_eq!(state.dial.unwrap().heat, dec("0")); + } + + #[test] + fn a_share_that_is_out_of_bounds_or_not_keyed_by_its_instance_holds_the_swarm() { + let dir = tempfile::tempdir().unwrap(); + let path = released_swarm(dir.path(), swarm_dial("1", "5", "2", 2)); + write_unchecked_share(&path, "weksa", &share("weksa", "0")); + assert_eq!(envelope_at(dir.path()), SwarmEffective::Held(HeldReason::ShareOutOfBounds)); + write_unchecked_share(&path, "weksa", &share("weksa", "-1")); + assert_eq!(envelope_at(dir.path()), SwarmEffective::Held(HeldReason::ShareOutOfBounds)); + write_unchecked_share(&path, "weksa", &share("weksa", "1")); + let _ = released(dir.path()); + write_unchecked_share(&path, "mallory", &share("weksa", "9")); + assert_eq!(envelope_at(dir.path()), SwarmEffective::Held(HeldReason::Undecodable)); + } + + #[test] + fn envelope_scales_by_heat() { + for (heat, rate, run_cap) in [("0.5", "2.5", "1.25"), ("2", "10", "5")] { + let dir = tempfile::tempdir().unwrap(); + released_swarm(dir.path(), swarm_dial(heat, "5", "2.5", 3)); + let envelope = released(dir.path()); + assert_eq!(envelope.rate_usd_per_h(), dec(rate), "rate at heat {heat}"); + assert_eq!(envelope.run_cap_usd(), dec(run_cap), "run cap at heat {heat}"); + assert_eq!(envelope.k(), 3); + } + } + + #[test] + fn a_missing_share_is_weight_one_and_a_set_share_is_read_back() { + let dir = tempfile::tempdir().unwrap(); + let path = released_swarm(dir.path(), swarm_dial("1", "5", "2", 2)); + SwarmWriter::open(&path).unwrap().set_share(share("weksa", "3")).unwrap(); + let envelope = released(dir.path()); + assert_eq!(envelope.share(&slug("weksa")), dec("3")); + assert_eq!(envelope.share(&slug("eureka")), Decimal::ONE); + } + + /// An instance's own released reading, with a run cap of `run_usd`. + fn instance_reading(instance: &str, run_usd: &str) -> Effective { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join(instance).join("control.cc"); + let mut writer = ControlWriter::open(&path).unwrap(); + writer.set_brake(true, at(0), "op").unwrap(); + writer.set_dial(dial("1", 600, run_usd)).unwrap(); + with_root(dir.path(), || read_effective(&slug(instance))) + } + + #[test] + fn grant_for_never_exceeds_cap() { + let dir = tempfile::tempdir().unwrap(); + released_swarm(dir.path(), swarm_dial("1", "5", "2", 2)); + let envelope = released(dir.path()); + let eureka = slug("eureka"); + let own = instance_reading("eureka", "10"); + for (budget, expected) in [("2.01", None), ("100", None), ("0", None), ("-1", None), ("2", Some("2")), ("0.5", Some("0.5")), ("0.01", Some("0.01"))] { + let grant = envelope.grant_for(&eureka, &own, dec(budget)); + assert_eq!(grant.map(|g| g.run_cap_usd()), expected.map(dec), "budget {budget}"); + } + let grant = envelope.grant_for(&eureka, &own, dec("1")).unwrap(); + assert_eq!(grant.instance(), &eureka); + assert_eq!(grant.cadence(), Duration::from_secs(600), "the instance cadence is carried, not invented"); + } + + #[test] + fn grant_for_does_not_widen_the_instance_own_cap() { + let dir = tempfile::tempdir().unwrap(); + released_swarm(dir.path(), swarm_dial("1", "5", "2", 2)); + let envelope = released(dir.path()); + let eureka = slug("eureka"); + let own = instance_reading("eureka", "0.5"); + assert!(envelope.grant_for(&eureka, &own, dec("0.51")).is_none()); + assert_eq!(envelope.grant_for(&eureka, &own, dec("0.5")).unwrap().run_cap_usd(), dec("0.5")); + } + + #[test] + fn grant_for_refuses_held_instance() { + let dir = tempfile::tempdir().unwrap(); + released_swarm(dir.path(), swarm_dial("1", "5", "2", 2)); + let envelope = released(dir.path()); + let eureka = slug("eureka"); + assert!(envelope.grant_for(&eureka, &Effective::Held { reason: HeldReason::BrakeHeld }, dec("1")).is_none()); + // Another instance released reading is no proof for this one. + let weksa_reading = instance_reading("weksa", "10"); + assert!(envelope.grant_for(&eureka, &weksa_reading, dec("1")).is_none()); + assert!(envelope.grant_for(&slug("weksa"), &weksa_reading, dec("1")).is_some()); + } + + #[cfg(unix)] + #[test] + fn the_swarm_store_is_written_by_root_only() { + use std::os::unix::fs::PermissionsExt; + let dir = tempfile::tempdir().unwrap(); + let path = released_swarm(dir.path(), swarm_dial("1", "5", "2", 2)); + let before = std::fs::read(&path).unwrap(); + for mode in [0o666, 0o664] { + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(mode)).unwrap(); + let error = SwarmWriter::open(&path).unwrap().set_dial(swarm_dial("0", "5", "2", 2)).unwrap_err(); + assert!(error.downcast_ref::().is_some(), "mode {mode:o}: {error:#}"); + } + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap(); + assert_eq!(std::fs::read(&path).unwrap(), before, "a refused write changed the store"); + } + + #[test] + fn the_swarm_writer_refuses_out_of_bounds_values_and_writes_nothing() { + let dir = tempfile::tempdir().unwrap(); + let path = swarm_file(dir.path()); + let mut writer = SwarmWriter::open(&path).unwrap(); + assert!(writer.set_dial(swarm_dial("2.5", "5", "2", 2)).is_err()); + assert!(writer.set_dial(swarm_dial("1", "5", "2", 0)).is_err()); + assert!(writer.set_share(share("weksa", "0")).is_err()); + assert!(writer.set_share(share("Not A Slug", "1")).is_err()); + assert!(!path.exists(), "a refused write created the store"); + assert!(std::fs::read_dir(dir.path()).unwrap().next().is_none(), "a refused write created a sibling"); + } +} diff --git a/crates/mind-body/tests/cli.rs b/crates/mind-body/tests/cli.rs index 8616bd6..d82f8e8 100644 --- a/crates/mind-body/tests/cli.rs +++ b/crates/mind-body/tests/cli.rs @@ -454,3 +454,65 @@ mod shared_arguments { } } } + +fn swarm_show(root: &std::path::Path) -> String { + let (result, out) = go(root, &["swarm", "show"], "op", 0); + result.unwrap(); + out +} + +#[test] +fn swarm_verbs_round_trip() { + let root = tempfile::tempdir().unwrap(); + let shown = swarm_show(root.path()); + assert!(shown.contains("brake: absent") && shown.contains("dial: absent") && shown.contains("BrakeAbsent"), "{shown}"); + + for args in [ + &["swarm", "brake", "release"][..], + &["swarm", "dial", "set", "--heat", "1.5", "--base-rate", "4", "--base-run", "2", "--k", "3"], + &["swarm", "share", "set", "weksa", "2.5"], + ] { + go(root.path(), args, "alice", 5).0.unwrap(); + } + let shown = swarm_show(root.path()); + assert!(shown.contains("brake: released") && shown.contains("set_by alice"), "{shown}"); + assert!(shown.contains("dial: heat 1.5 base_rate_usd_per_h 4 base_run_usd 2 k 3"), "{shown}"); + assert!(shown.contains("share: weksa 2.5"), "{shown}"); + assert!(shown.contains("Released") && shown.contains("rate_usd_per_h: 6.0") && shown.contains("run_cap_usd: 3.0"), "{shown}"); + + // Heat 0 stops the swarm and is one command away; the hold stops it too. + go(root.path(), &["swarm", "dial", "set", "--heat", "0", "--base-rate", "4", "--base-run", "2", "--k", "3"], "alice", 6).0.unwrap(); + assert!(swarm_show(root.path()).contains("HeatZero")); + go(root.path(), &["swarm", "brake", "hold"], "alice", 7).0.unwrap(); + assert!(swarm_show(root.path()).contains("BrakeHeld")); +} + +#[test] +fn swarm_verbs_write_only_the_swarm_file_and_never_call_the_stopper() { + let root = tempfile::tempdir().unwrap(); + let (result, _, calls) = go_stopping(root.path(), &["swarm", "brake", "hold"], "op", 0, |_| Ok(())); + result.unwrap(); + assert!(calls.is_empty(), "the swarm hold has no instance to stop: {calls:?}"); + let names = listing(root.path()); + assert!(names.iter().all(|name| name.starts_with("swarm.cc")), "{names:?}"); +} + +#[test] +fn swarm_commands_refuse_bad_input_and_write_nothing() { + let root = tempfile::tempdir().unwrap(); + for args in [ + &["swarm"][..], + &["swarm", "dial", "set", "--heat", "1"], + &["swarm", "dial", "set", "--heat", "3", "--base-rate", "4", "--base-run", "2", "--k", "3"], + &["swarm", "dial", "set", "--heat", "1", "--base-rate", "4", "--base-run", "2", "--k", "0"], + &["swarm", "dial", "set", "--heat", "1", "--heat", "1", "--base-rate", "4", "--base-run", "2", "--k", "3"], + &["swarm", "share", "set", "weksa", "0"], + &["swarm", "share", "set", "weksa", "heavy"], + &["swarm", "share", "set", "Not A Slug", "1"], + &["swarm", "share", "set", "weksa"], + &["swarm", "brake", "release", "now"], + ] { + assert!(go(root.path(), args, "op", 0).0.is_err(), "{args:?} was accepted"); + } + assert!(listing(root.path()).is_empty(), "a refused command wrote something"); +} diff --git a/crates/mind-body/tests/control.rs b/crates/mind-body/tests/control.rs index 73ac41a..3dcd2b0 100644 --- a/crates/mind-body/tests/control.rs +++ b/crates/mind-body/tests/control.rs @@ -19,7 +19,7 @@ use std::time::Duration; use chrono::{DateTime, TimeZone, Utc}; use eureka_pipeline::Slug; -use mind_body::control::{BurnRate, ControlWriter, FileSource, control_path, load_state}; +use mind_body::control::{BurnRate, ControlWriter, FileSource, SwarmWriter, control_path, load_state, swarm_path}; use rust_decimal::Decimal; use std::str::FromStr; @@ -112,6 +112,9 @@ mod modes { "brake" => { ControlWriter::open(&path).unwrap().set_brake(true, at(0), "op").unwrap(); } + "swarm_brake" => { + SwarmWriter::open(swarm_path(Path::new(&std::env::var(ROOT).unwrap()))).unwrap().set_brake(true, at(0), "op").unwrap(); + } "attacker" => { // Spins on an open-for-write of the live store; any success is a hit. let stop = std::env::var("MIND_BODY_STOP").unwrap(); @@ -182,6 +185,14 @@ mod modes { } } + #[test] + fn the_swarm_store_is_created_0644_whatever_the_operators_umask() { + for umask in ["000", "022", "077"] { + let (dir, _) = written_under("swarm_brake", umask); + assert_eq!(mode(&swarm_path(dir.path())), 0o644, "swarm store under umask {umask}"); + } + } + fn running_as_root() -> bool { Command::new("id").arg("-u").output().map(|out| String::from_utf8_lossy(&out.stdout).trim() == "0").unwrap_or(false) } diff --git a/crates/mind-body/tests/public_surface.txt b/crates/mind-body/tests/public_surface.txt index 354f7ee..0d888bd 100644 --- a/crates/mind-body/tests/public_surface.txt +++ b/crates/mind-body/tests/public_surface.txt @@ -7,16 +7,21 @@ mind_body: mod wake mind_body::cli: fn instance_and_endpoint (args : & [String] , env_endpoint : Option < String > , usage : & str) -> Result < (Slug , SocketAddr) > mind_body::cli: fn operator_name (sudo_user : Option < String > , user : Option < String >) -> String mind_body::cli: fn run (args : & [String] , root : & Path , set_by : & str , now : DateTime < Utc > , out : & mut impl Write , stop_units : & mut dyn FnMut (& Slug) -> Result < () > ,) -> Result < () > -mind_body::control: # [derive (Clone , Copy , Debug , PartialEq , Eq)] pub enum HeldReason { BrakeAbsent , BrakeHeld , DialAbsent , DialOutOfBounds , Undecodable , } +mind_body::control: # [derive (Clone , Copy , Debug , PartialEq , Eq)] pub enum HeldReason { BrakeAbsent , BrakeHeld , DialAbsent , DialOutOfBounds , HeatZero , ShareOutOfBounds , Undecodable , } mind_body::control: # [derive (Debug , PartialEq , Eq)] pub enum Effective { Held { reason : HeldReason } , Released (Grant) , } +mind_body::control: # [derive (Debug , PartialEq , Eq)] pub enum SwarmEffective { Held (HeldReason) , Released (Envelope) , } mind_body::control: const BRAKE_KEY : & str mind_body::control: const CONTROL_ROOT : & str mind_body::control: const DIAL_KEY : & str +mind_body::control: const SWARM_FILE : & str mind_body::control: fn control_path (root : & Path , instance : & Slug) -> Result < PathBuf > mind_body::control: fn heat_max () -> Decimal mind_body::control: fn heat_min () -> Decimal mind_body::control: fn load_state (source : & impl ControlSource) -> Result < ControlState > +mind_body::control: fn load_swarm_state (source : & impl ControlSource) -> Result < SwarmState > mind_body::control: fn read_effective (instance : & Slug) -> Effective +mind_body::control: fn read_envelope () -> SwarmEffective +mind_body::control: fn swarm_path (root : & Path) -> PathBuf mind_body::control: impl BurnRate mind_body::control: impl BurnRate: fn in_bounds (& self) -> bool mind_body::control: impl ControlSource for FileSource @@ -25,12 +30,28 @@ mind_body::control: impl ControlWriter mind_body::control: impl ControlWriter: fn open (path : impl Into < PathBuf >) -> Result < Self > mind_body::control: impl ControlWriter: fn set_brake (& mut self , released : bool , set_at : DateTime < Utc > , set_by : & str) -> Result < () > mind_body::control: impl ControlWriter: fn set_dial (& mut self , dial : BurnRate) -> Result < () > +mind_body::control: impl Envelope +mind_body::control: impl Envelope: fn grant_for (& self , instance : & Slug , instance_effective : & Effective , budget_usd : Decimal) -> Option < Grant > +mind_body::control: impl Envelope: fn k (& self) -> u32 +mind_body::control: impl Envelope: fn rate_usd_per_h (& self) -> Decimal +mind_body::control: impl Envelope: fn run_cap_usd (& self) -> Decimal +mind_body::control: impl Envelope: fn share (& self , instance : & Slug) -> Decimal mind_body::control: impl FileSource mind_body::control: impl FileSource: fn new (path : impl Into < PathBuf >) -> Self mind_body::control: impl Grant mind_body::control: impl Grant: fn cadence (& self) -> Duration mind_body::control: impl Grant: fn instance (& self) -> & Slug mind_body::control: impl Grant: fn run_cap_usd (& self) -> Decimal +mind_body::control: impl Share +mind_body::control: impl Share: fn in_bounds (& self) -> bool +mind_body::control: impl SwarmDial +mind_body::control: impl SwarmDial: fn in_bounds (& self) -> bool +mind_body::control: impl SwarmState +mind_body::control: impl SwarmWriter +mind_body::control: impl SwarmWriter: fn open (path : impl Into < PathBuf >) -> Result < Self > +mind_body::control: impl SwarmWriter: fn set_brake (& mut self , released : bool , set_at : DateTime < Utc > , set_by : & str) -> Result < () > +mind_body::control: impl SwarmWriter: fn set_dial (& mut self , dial : SwarmDial) -> Result < () > +mind_body::control: impl SwarmWriter: fn set_share (& mut self , share : Share) -> Result < () > mind_body::control: impl std :: error :: Error for UnsafeStore mind_body::control: impl std :: fmt :: Display for UnsafeStore mind_body::control: pub trait ControlSource { fn snapshot (& self) -> Result < Vec < CultCacheEnvelope > > ; } @@ -38,8 +59,13 @@ mind_body::control: struct Brake # [derive (Clone , Debug , PartialEq , Eq , Dat mind_body::control: struct BurnRate # [derive (Clone , Debug , PartialEq , Eq , DatabaseEntry)] pub-fields[heat: Decimal, base_cooldown_s: u32, base_run_usd: Decimal, set_at: DateTime < Utc >, set_by: String] mind_body::control: struct ControlState # [derive (Clone , Debug , PartialEq , Eq , Default)] pub-fields[brake: Option < Brake >, dial: Option < BurnRate >] mind_body::control: struct ControlWriter pub-fields[] +mind_body::control: struct Envelope # [derive (Debug , PartialEq , Eq)] pub-fields[] mind_body::control: struct FileSource pub-fields[] mind_body::control: struct Grant # [derive (Debug , PartialEq , Eq)] pub-fields[] +mind_body::control: struct Share # [derive (Clone , Debug , PartialEq , Eq , DatabaseEntry)] pub-fields[instance: String, weight: Decimal] +mind_body::control: struct SwarmDial # [derive (Clone , Debug , PartialEq , Eq , DatabaseEntry)] pub-fields[heat: Decimal, base_rate_usd_per_h: Decimal, base_run_usd: Decimal, k: u32, set_at: DateTime < Utc >, set_by: String] +mind_body::control: struct SwarmState # [derive (Clone , Debug , PartialEq , Eq , Default)] pub-fields[brake: Option < Brake >, dial: Option < SwarmDial >, shares: Vec < Share >] +mind_body::control: struct SwarmWriter pub-fields[] mind_body::control: struct UnsafeStore # [derive (Debug)] pub-fields[path: PathBuf, mode: u32, uid: u32] mind_body::launch: # [derive (Clone , Debug , PartialEq , Eq)] pub enum Declined { Admission (MindRefusal) , Conflict , AlreadyAdmitted , StartFailed , GrantForOtherInstance , PersonaRunTakesNoClaims , } mind_body::launch: # [derive (Clone , Debug , PartialEq , Eq)] pub enum LaunchOutcome { Launched { run : PipelineRef , } , NotQueued { item : PipelineRef , } , Tripped { item : PipelineRef , question : PipelineRef , } , Busy { run : PipelineRef , } , Refused (Declined) , }