hyperhive/hive-c0re/src/workers/wanted.rs

//! Converging this hive onto the agent set the swarm controller declares.
//!
//! The controller writes one key per hive into the `hive-wanted` bucket (see
//! [`swarm_queue_client::wanted`]); this reads its own and acts on it. It is
//! the **repair** path, not the fast one: a deploy event
//! ([`crate::swarm_status`]) is core NATS, so a hive that was down never hears
//! it, and until this existed that left the hive on its old shape with nothing
//! looking wrong.
//!
//! # Two things a declaration does not mean
//!
//! **Absence is not a deletion order.** An agent missing from the value — or
//! no value, or no bucket — is one the controller has said nothing about.
//! Swarm-side lifecycle does not yet cover agents that predate it, so
//! converging "to exactly this set" would tear down every agent the swarm has
//! not adopted.
//!
//! **An unknown state is not a partial instruction.** [`AgentState`] is closed,
//! so a value this build cannot read fails the whole decode: the hive converges
//! nothing rather than obeying the agents it happened to understand.
//!
//! # What "diverged" is measured against
//!
//! The hive's **stored power intent**, not the container's observed running
//! state. An agent down while its intent says `Up` is already the boot
//! reconcile's work; a loop reading `is_running` would insert a start DAG
//! behind that reconcile's back on every boot.
//!
//! One consequence, deliberate: a declaration outranks a local stop — an agent
//! declared `Up` returns at the next boot, because the controller owns that.

use std::collections::{BTreeMap, BTreeSet};
use std::sync::Arc;

use anyhow::{Context, Result};
use swarm_queue_client::wanted::{AgentState, HiveWanted};

use crate::coordinator::Coordinator;
use crate::power::Wanted;

/// Read this hive's declaration and converge the agents it names.
///
/// `Ok(())` covers three quiet cases that are **not** an empty declaration: no
/// hive name, no bucket, no key. Each logs its own line, because a loop that
/// silently does nothing and one that is working look identical otherwise.
/// Anything else — an unreachable queue, an undecodable value, an agent list
/// this hive cannot enumerate — is an error, so the node fails visibly on the
/// dashboard instead of reading as "nothing was declared".
///
/// # Errors
/// Propagates a failed queue connect, KV read, decode, or agent enumeration.
pub async fn pull(coord: &Arc<Coordinator>) -> Result<()> {
    let Some(hive) = crate::container_view::hive_swarm_names().0 else {
        tracing::debug!("wanted state: HYPERHIVE_HIVE_NAME unset; no key to read");
        return Ok(());
    };
    let Some(client) = crate::swarm_queue::client().await else {
        // Absent or failed — either way already reported by the shared
        // connector (an `info` log or a `swarm_queue_config` banner).
        return Ok(());
    };
    // An unconnected client does not fail a JetStream request, it hangs on it.
    // A hung node is worse than a failed one: the dashboard shows it running.
    swarm_queue_client::ensure_connected(&client)?;

    let Some(store) = swarm_queue_client::wanted::open_read_only(&client, &hive).await else {
        tracing::info!(%hive, "wanted state: no bucket yet; nothing has been declared");
        return Ok(());
    };
    let Some(raw) = store
        .get(&hive)
        .await
        .context("reading this hive's wanted state")?
    else {
        tracing::info!(%hive, "wanted state: no key for this hive; nothing has been declared");
        return Ok(());
    };
    let declared: HiveWanted =
        serde_json::from_slice(&raw).context("decoding this hive's wanted state")?;

    converge(coord, &declared).await
}

/// Queue whatever the declaration asks for and this hive is not already doing.
async fn converge(coord: &Arc<Coordinator>, declared: &HiveWanted) -> Result<()> {
    // Fail closed, for the reason the deploy event's own arm gives: without
    // the list, "this hive does not have that agent" cannot be told from
    // "this hive cannot see its agents", and acting on the guess creates a
    // container over one that already exists.
    let present: BTreeSet<String> = crate::lifecycle::agents_for_meta_listing()
        .await
        .context("enumerating this hive's agents")?
        .into_iter()
        .map(|spec| spec.name)
        .collect();

    let mut intents = BTreeMap::new();
    for agent in declared.agents.keys() {
        match coord.power.get(agent) {
            Ok(intent) => {
                intents.insert(agent.clone(), intent);
            }
            // Left out of the map, which is what makes `plan` skip it.
            Err(e) => {
                tracing::warn!(%agent, error = ?e, "wanted state: cannot read power intent; skipping");
            }
        }
    }

    let plan = plan(declared, &present, &intents);
    tracing::info!(
        declared = declared.agents.len(),
        deploy = plan.deploy.len(),
        start = plan.start.len(),
        stop = plan.stop.len(),
        "wanted state: converging"
    );

    let mut queued = false;
    for agent in &plan.deploy {
        match crate::swarm_status::queue_first_deploy(coord, agent) {
            Ok(_) => queued = true,
            // Per agent rather than the whole sweep: one agent that cannot be
            // queued is not a reason to leave the others undeclared.
            Err(e) => tracing::warn!(%agent, error = %e, "wanted state: queueing a deploy failed"),
        }
    }
    if !plan.start.is_empty() {
        crate::job_queue::power::start_many(coord, &plan.start)
            .await
            .context("queueing the declared starts")?;
    }
    if !plan.stop.is_empty() {
        crate::job_queue::power::stop_many(coord, &plan.stop, true)
            .await
            .context("queueing the declared stops")?;
    }
    if queued {
        // The power ops emit their own; the first deploys do not.
        coord.emit_rebuild_queue_snapshot();
    }
    Ok(())
}

/// Everything one declaration asks this hive to queue.
#[derive(Debug, Default, PartialEq, Eq)]
struct Plan {
    deploy: Vec<String>,
    start: Vec<String>,
    stop: Vec<String>,
}

/// Turn one declaration into that plan — pure, so what the loop does with a
/// given declaration is testable without a queue, a container or a store.
///
/// `intents` is the hive's stored power intent per agent, and an agent
/// **missing from the map** is one whose intent could not be read: it is
/// skipped. That is not `Some(None)` — "no row yet" is a reading and gets
/// converged, "the read failed" is not and would be a guess.
///
/// An agent the declaration does not name reaches none of the three lists.
/// That is the whole of "absence is not a deletion order": the agents this
/// hive runs that the swarm has not adopted are never even inspected.
fn plan(
    declared: &HiveWanted,
    present: &BTreeSet<String>,
    intents: &BTreeMap<String, Option<Wanted>>,
) -> Plan {
    let mut plan = Plan::default();
    for (agent, decl) in &declared.agents {
        let Some(intent) = intents.get(agent) else {
            continue;
        };
        match decide(decl.state, present.contains(agent), *intent) {
            Converge::Deploy => plan.deploy.push(agent.clone()),
            Converge::Start => plan.start.push(agent.clone()),
            Converge::Stop => plan.stop.push(agent.clone()),
            Converge::Nothing => {}
        }
    }
    plan
}

/// What the loop will do about one declared agent.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Converge {
    /// Declared, and this hive has no container for it — the missed-deploy
    /// repair.
    Deploy,
    Start,
    Stop,
    /// The hive already agrees with the declaration.
    Nothing,
}

/// The whole decision, pure: no queue, no store, no container. The three
/// inputs are exactly what [`converge`] reads per agent, so the table below is
/// the behaviour rather than a model of it.
fn decide(state: AgentState, present: bool, intent: Option<Wanted>) -> Converge {
    match state {
        AgentState::Up if !present => Converge::Deploy,
        AgentState::Up => {
            if intent == Some(Wanted::Up) {
                Converge::Nothing
            } else {
                Converge::Start
            }
        }
        // Declared offline and not here: creating a container in order to
        // leave it stopped is not what the declaration asks for.
        AgentState::Offline if !present => Converge::Nothing,
        AgentState::Offline => {
            if intent == Some(Wanted::Offline) {
                Converge::Nothing
            } else {
                Converge::Stop
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use std::collections::{BTreeMap, BTreeSet};

    use super::{Converge, Plan, decide, plan};
    use crate::power::Wanted;
    use swarm_queue_client::wanted::{AgentState, AgentWanted, HiveWanted};

    fn declaration(agents: &[(&str, AgentState)]) -> HiveWanted {
        HiveWanted {
            agents: agents
                .iter()
                .map(|(name, state)| ((*name).to_owned(), AgentWanted { state: *state }))
                .collect(),
        }
    }

    fn set(names: &[&str]) -> BTreeSet<String> {
        names.iter().map(|n| (*n).to_owned()).collect()
    }

    fn intents(entries: &[(&str, Option<Wanted>)]) -> BTreeMap<String, Option<Wanted>> {
        entries
            .iter()
            .map(|(name, intent)| ((*name).to_owned(), *intent))
            .collect()
    }

    /// The property the whole migration rides on: this hive's own agents are
    /// not converged, they are not looked at. `adopted` is the control — the
    /// same call must still act on what the declaration *does* name, or this
    /// passes on a loop that plans nothing at all.
    #[test]
    fn an_agent_the_declaration_does_not_name_is_never_touched() {
        let declared = declaration(&[("adopted", AgentState::Up)]);
        let planned = plan(
            &declared,
            &set(&["adopted", "legacy"]),
            &intents(&[("adopted", Some(Wanted::Offline)), ("legacy", None)]),
        );
        assert_eq!(
            planned,
            Plan {
                deploy: vec![],
                start: vec!["adopted".to_owned()],
                stop: vec![],
            }
        );
    }

    /// An intent that could not be read is skipped rather than treated as "no
    /// row". `readable` is the control for the same call.
    #[test]
    fn an_unreadable_intent_skips_only_that_agent() {
        let declared = declaration(&[("unreadable", AgentState::Up), ("readable", AgentState::Up)]);
        let planned = plan(
            &declared,
            &set(&["unreadable", "readable"]),
            &intents(&[("readable", None)]),
        );
        assert_eq!(planned.start, vec!["readable".to_owned()]);
        assert!(planned.deploy.is_empty() && planned.stop.is_empty());
    }

    #[test]
    fn a_declared_agent_this_hive_does_not_have_is_deployed() {
        assert_eq!(decide(AgentState::Up, false, None), Converge::Deploy);
        // Even holding an intent: a row can outlive its container.
        assert_eq!(
            decide(AgentState::Up, false, Some(Wanted::Up)),
            Converge::Deploy
        );
    }

    #[test]
    fn a_hive_that_already_agrees_queues_nothing() {
        assert_eq!(
            decide(AgentState::Up, true, Some(Wanted::Up)),
            Converge::Nothing
        );
        assert_eq!(
            decide(AgentState::Offline, true, Some(Wanted::Offline)),
            Converge::Nothing
        );
    }

    #[test]
    fn a_disagreeing_intent_is_converged_both_ways() {
        assert_eq!(
            decide(AgentState::Up, true, Some(Wanted::Offline)),
            Converge::Start
        );
        assert_eq!(
            decide(AgentState::Offline, true, Some(Wanted::Up)),
            Converge::Stop
        );
    }

    /// No row is not agreement. An agent that predates the power store is
    /// converged to what it was declared as, not left on whatever it happens
    /// to be doing.
    #[test]
    fn an_agent_with_no_intent_row_is_converged() {
        assert_eq!(decide(AgentState::Up, true, None), Converge::Start);
        assert_eq!(decide(AgentState::Offline, true, None), Converge::Stop);
    }

    /// Declared offline, no container: the one absent case that must not
    /// deploy. Its control is the `Up` case above, which must.
    #[test]
    fn an_absent_agent_declared_offline_is_left_absent() {
        assert_eq!(decide(AgentState::Offline, false, None), Converge::Nothing);
    }

    /// Version skew is handled one layer up, at the decode: `AgentState` is
    /// closed, so an unknown value never reaches [`decide`] — it fails the
    /// whole declaration in `swarm-queue-client`, which owns that test
    /// (`an_unknown_state_fails_the_whole_declaration`). There is deliberately
    /// no case for it here: a test asserting something unrepresentable would
    /// pass forever without measuring anything.
    #[test]
    fn every_state_this_build_knows_is_covered_above() {
        for state in [AgentState::Up, AgentState::Offline] {
            let seen = [true, false].iter().any(|present| {
                decide(state, *present, None) != Converge::Nothing
                    || decide(state, *present, Some(Wanted::Up)) != Converge::Nothing
            });
            assert!(seen, "{state:?} produces no action in any combination");
        }
    }
}