hyperhive/swarm-queue-client/src/wanted.rs

//! The hive-wanted KV bucket: the agent set the controller declares for each
//! hive, keyed 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 KV bucket the controller publishes per-hive wanted state into, one key
/// per hive keyed by `hiveName`.
///
/// A constant and not an option, 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: &str = "hive-wanted";

/// One hive's whole declaration: the agents the controller names, and what it
/// wants of each. The value under `BUCKET`/`<hiveName>`.
///
/// 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<String, AgentWanted>,
}

/// 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, 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,
}

/// 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 — that
/// question is answered by the controller's own records, not by replaying a
/// bucket.
#[cfg(feature = "kv")]
pub async fn open_or_create(
    client: &async_nats::Client,
) -> Result<async_nats::jetstream::kv::Store, Error> {
    let js = async_nats::jetstream::new(client.clone());
    match js.get_key_value(BUCKET).await {
        Ok(store) => Ok(store),
        Err(e) => {
            tracing::info!(
                bucket = BUCKET,
                reason = %e,
                "wanted-state bucket not available, creating it"
            );
            js.create_key_value(async_nats::jetstream::kv::Config {
                bucket: BUCKET.to_owned(),
                description: "Agent set the swarm controller declares for each hive".to_owned(),
                history: 1,
                ..Default::default()
            })
            .await
            .map_err(|source| Error::CreateBucket {
                bucket: 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,
) -> Option<async_nats::jetstream::kv::Store> {
    let js = async_nats::jetstream::new(client.clone());
    js.get_key_value(BUCKET).await.ok()
}

#[cfg(test)]
mod tests {
    use super::{AgentState, HiveWanted};

    #[test]
    fn the_known_states_decode_and_round_trip() {
        let doc = r#"{"agents":{"a":{"state":"up"},"b":{"state":"offline"}}}"#;
        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!(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() {
        let doc = r#"{"agents":{"a":{"state":"up"},"c":{"state":"paused"}}}"#;
        assert!(serde_json::from_str::<HiveWanted>(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::<HiveWanted>(r#"{"agents":{"a":{}}}"#).is_err());
    }
}