//! The hive-wanted KV buckets: the agent set the controller declares for each //! hive, **one bucket per hive** (see [`crate::wanted::bucket`] for why the //! split is a grant boundary rather than a data-modelling choice), keyed inside //! it by `hiveName`. //! //! Sibling of [`crate::status`] and deliberately **not** a mirror of it: that //! one is **observed** — each hive republishes what it is, so its store losing //! everything "degrades to honesty" (`swarm-nats.nix`). This one is //! **declared**, and nothing regenerates it; lose it and a reconcile loop has //! nothing to converge to, which is the failure such a loop exists to remove. //! //! # Absence is not a deletion order //! //! A hive that finds no key for itself — or no bucket at all — has learned //! nothing about the agents it is running, not that it should have none. //! Swarm-side lifecycle does not yet cover agents that predate it, so callers //! converge the agents a value NAMES and leave the rest alone. //! //! # Only the controller creates it //! //! Unlike [`crate::status::open_or_create`], where either end may legitimately //! arrive first, the writer here is single and known. A hive opens read-only //! and treats absence as the case above, so "not published yet" stays quiet //! rather than looking like an error a hive could fix. #[cfg(feature = "kv")] use crate::Error; /// The prefix every hive's wanted-state bucket name starts with. /// /// Derived from, and not configurable, for the reason [`crate::status::BUCKET`] /// gives: writer and reader must name the same bucket, and an option is a way /// for two deployments to disagree about which one that is. pub const BUCKET_PREFIX: &str = "hive-wanted-"; /// The KV bucket one hive's declaration lives in — **one bucket per hive**, not /// one bucket keyed by hive. /// /// The split is a grant-scoping decision, not a data-modelling one. A KV read /// scopes per key, because `DIRECT.GET` carries the key in the subject; a /// *watch* does not, because a consumer's filter travels in the request payload /// and `$JS.API.CONSUMER.CREATE.` therefore grants the whole stream. So /// with every hive in one bucket, letting a hive watch its own declaration /// means letting it read every other hive's. One stream per hive makes the /// grant a hive can hold exactly as wide as what it is allowed to see. /// /// A bucket name may contain `[a-zA-Z0-9_-]`, and a hive name is an `Ident` /// (`[a-z0-9-]`), so this composition is always a legal bucket name — plain /// code span rather than an intra-doc link because this crate does not depend /// on `hive_types`. #[must_use] pub fn bucket(hive: &str) -> String { format!("{BUCKET_PREFIX}{hive}") } /// One hive's whole declaration: the agents the controller names, and what it /// wants of each. The value under `BUCKET`/``. /// /// The agents it does **not** name are not part of the declaration at all — /// see the module docs. A hive converges the ones listed here and leaves the /// rest alone. #[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct HiveWanted { /// Agent name → what the controller wants of it. #[serde(default)] pub agents: std::collections::BTreeMap, } /// One agent's entry in a [`HiveWanted`]. /// /// A struct around one field rather than the bare state, because this is where /// a later declaration grows — a config revision, a schedule — and a hive /// built before that field existed keeps deserialising the whole document. #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct AgentWanted { /// The state to converge this agent to. pub state: AgentState, } /// What the controller wants an agent to be. /// /// **Closed on purpose.** A value this build does not know fails the whole /// document's decode, so a hive running older code converges **nothing** rather /// than part of a declaration it only half understands. Adding a state means /// adding a variant here and shipping it to both ends — which is the intended /// workflow, not an obstacle to route around with a catch-all variant. #[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "snake_case")] pub enum AgentState { /// Exists on the hive and is running. Up, /// Exists on the hive and is not running. Offline, /// Exists on the hive, is running, and its turn loop is parked — the /// swarm-declared counterpart of the hive-local `hivectl agent pause` /// marker. Orthogonal to the power axis `Up`/`Offline` express: a /// paused agent still has a running container (the web UI and MCP /// daemons stay reachable), it just drives no turns. See /// `hive-c0re::workers::wanted::decide_pause` for the convergence side /// of that split. Paused, /// Torn down entirely — the hive runs the destroy template once, then /// treats absence as agreement rather than re-running it. The key /// stays in the declared set with this state forever, on purpose: it's /// what lets a redeclare of an already-destroyed agent be rejected as /// a terminal-state transition instead of silently reviving it. /// Irreversible: there is no state that brings a destroyed agent back /// short of a fresh deploy. Destroyed, } impl AgentState { /// The wire spelling, for a reader that renders rather than decodes. /// /// Kept beside the enum so it cannot drift from the `rename_all` above; /// a test pins the two together. #[must_use] pub fn as_str(self) -> &'static str { match self { AgentState::Up => "up", AgentState::Offline => "offline", AgentState::Paused => "paused", AgentState::Destroyed => "destroyed", } } } /// Open the wanted-state bucket for writing, creating it if nothing has yet. /// /// **Controller-side only.** `history: 1` because a hive converges to the /// current declaration and never asks what the previous one was. The /// controller keeps no second copy either: the value published here is the /// record, and it reads it back from this bucket when it needs it. #[cfg(feature = "kv")] pub async fn open_or_create( client: &async_nats::Client, hive: &str, ) -> Result { let js = async_nats::jetstream::new(client.clone()); let bucket = bucket(hive); match js.get_key_value(&bucket).await { Ok(store) => Ok(store), Err(e) => { tracing::info!( %bucket, reason = %e, "wanted-state bucket not available, creating it" ); js.create_key_value(async_nats::jetstream::kv::Config { bucket: bucket.clone(), description: format!("Agent set the swarm controller declares for hive {hive}"), history: 1, ..Default::default() }) .await .map_err(|source| Error::CreateBucket { bucket, source }) } } } /// Open the wanted-state bucket for reading, or `None` when it does not exist. /// /// **Hive-side.** `None` is the ordinary pre-publication state, not a failure: /// a hive holds no grant to create this bucket and must not treat its absence /// as a reason to converge to an empty agent set — see the module docs. #[cfg(feature = "kv")] pub async fn open_read_only( client: &async_nats::Client, hive: &str, ) -> Option { let js = async_nats::jetstream::new(client.clone()); js.get_key_value(bucket(hive)).await.ok() } /// Watch this hive's own declaration for changes. /// /// **Hive-side.** `None` on the same terms as [`open_read_only`] — no bucket /// yet — plus one more: a watch is a `JetStream` *consumer*, so it needs a grant /// a plain `get` does not. A hive missing `CONSUMER.CREATE` on its own stream /// gets `None` here while `get` keeps working, which is why the caller must /// treat this as "not watching yet" and retry rather than as a dead end. /// /// Updates only, deliberately: the boot-time read already has the current /// value, and a watch that replayed history would re-converge the whole /// declaration on every reconnect for nothing. #[cfg(feature = "kv")] pub async fn watch( client: &async_nats::Client, hive: &str, ) -> Option { // The bucket holds exactly one key, named for the hive — same key // `open_read_only`'s caller reads, so both paths address one declaration. open_read_only(client, hive).await?.watch(hive).await.ok() } #[cfg(test)] mod tests { use super::{AgentState, BUCKET_PREFIX, HiveWanted, bucket}; #[test] fn each_hive_gets_its_own_bucket_name() { assert_eq!(bucket("alpha"), "hive-wanted-alpha"); assert_ne!(bucket("alpha"), bucket("beta")); } #[test] fn a_bucket_name_is_legal_for_every_legal_hive_name() { // The client rejects a bucket name outside `[a-zA-Z0-9_-]`, and a hive // name is an `Ident` — lowercase, digits, hyphen. Pinned here because // the grant scoping in `swarm-nats-auth` names this string, so a bucket // the client refuses to open would surface as a permissions problem // rather than as the naming problem it is. let legal = |s: &str| { s.chars() .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-') }; assert!(legal(&bucket("alpha"))); assert!(legal(&bucket("a-hive-with-hyphens"))); assert!(legal(&bucket("h9"))); // Control: the predicate can fail, so the assertions above are not // vacuously true of any string. assert!(!legal("hive.wanted")); } #[test] fn the_prefix_is_what_every_bucket_starts_with() { assert!(bucket("alpha").starts_with(BUCKET_PREFIX)); } #[test] fn the_known_states_decode_and_round_trip() { let doc = r#"{"agents":{"a":{"state":"up"},"b":{"state":"offline"},"c":{"state":"destroyed"},"d":{"state":"paused"}}}"#; let decoded: HiveWanted = serde_json::from_str(doc).expect("decodes"); assert_eq!(decoded.agents["a"].state, AgentState::Up); assert_eq!(decoded.agents["b"].state, AgentState::Offline); assert_eq!(decoded.agents["c"].state, AgentState::Destroyed); assert_eq!(decoded.agents["d"].state, AgentState::Paused); assert_eq!(serde_json::to_string(&decoded).expect("serialises"), doc); } /// The enum is closed, so a state this build does not know takes the /// **whole document** down rather than one agent. That is the point: a hive /// converges nothing rather than part of a declaration it half understands. /// The `up` beside it is the control — a valid entry in the same document /// does not rescue it, and this test would pass on a broken decoder without /// the round-trip test above proving the happy path still works. #[test] fn an_unknown_state_fails_the_whole_declaration() { // `"paused"` used to be this test's unknown example; it is a real // variant now, so `"sleeping"` takes its place as one this build // still does not know. let doc = r#"{"agents":{"a":{"state":"up"},"c":{"state":"sleeping"}}}"#; assert!(serde_json::from_str::(doc).is_err()); } /// Closed *values*, tolerant *fields*: a field a newer controller adds must /// not cost an older hive the document, since adding one is not a semantic /// a hive has to understand to obey the rest. #[test] fn unknown_fields_are_ignored() { let doc = r#"{"agents":{"a":{"state":"up","config_rev":"deadbeef"}},"epoch":3}"#; let decoded: HiveWanted = serde_json::from_str(doc).expect("decodes"); assert_eq!(decoded.agents["a"].state, AgentState::Up); } /// A declaration missing the one required field is an error, so a malformed /// document cannot read as "no agents". #[test] fn a_missing_state_is_an_error() { assert!(serde_json::from_str::(r#"{"agents":{"a":{}}}"#).is_err()); } /// `as_str` and the `rename_all` are two spellings of one fact, and a /// renderer that disagrees with the wire is worse than one that does not /// exist. The `match` is what makes this exhaustive: a new variant fails /// to compile here rather than quietly going untested. #[test] fn as_str_matches_the_serde_spelling() { let every = [ AgentState::Up, AgentState::Offline, AgentState::Paused, AgentState::Destroyed, ]; for state in every { match state { AgentState::Up | AgentState::Offline | AgentState::Paused | AgentState::Destroyed => {} } assert_eq!( serde_json::to_string(&state).expect("serialises"), format!("\"{}\"", state.as_str()) ); } } }