hyperhive/hive-c0re/src/agent_set.rs

//! The host-side files that follow the set of agents on this hive.
//!
//! [`refresh`] runs wherever the set changes (spawn, destroy, rebuild, the
//! startup migration): it brings `topology.json`, the gateway's agent-socket
//! map and its nginx include in line with the live agents.

use std::path::Path;

use anyhow::{Context, Result, bail};
use tokio::sync::Mutex;

/// Serialises [`refresh`]: `topology.json` is read, changed and written back,
/// so two concurrent refreshes would lose one of the changes.
static REFRESH: Mutex<()> = Mutex::const_new(());

/// Bring the host-side files in line with `agents`, the live agent set.
///
/// An empty `agents` is never intentional: it means `nixos-container list`
/// failed and the caller fell back to an empty list, and writing it would drop
/// every agent from the topology and the gateway. It is refused.
///
/// # Errors
/// `HIVE_FORGE_URL` is unset, or `topology.json` could not be reconciled. The
/// gateway files are best-effort: a failure there is logged, because the
/// 10-second poll in `workers::agent_sockets` writes them again.
pub async fn refresh(agents: &[hive_meta_flake::AgentSpec]) -> Result<()> {
    let _guard = REFRESH.lock().await;
    require_forge_url(std::env::var("HIVE_FORGE_URL").ok().as_deref())?;
    let dir = crate::paths::meta_root();
    std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
    let Some(agent_names) = live_names(agents) else {
        tracing::warn!(
            "agent_set: refusing to refresh with an empty agent list \
             (nixos-container list may have failed)"
        );
        return Ok(());
    };
    remove_stale_ca_files(&dir);

    let pending: Vec<String> = crate::coordinator::Coordinator::pending_init_names()
        .into_iter()
        .map(hive_types::Ident::into_string)
        .collect();
    crate::topology::reconcile(&agent_names, &pending)
        .with_context(|| format!("reconcile {}", crate::topology::topology_path().display()))?;

    // Every agent binds a unix socket (`HIVE_WEB_SOCKET`); the gateway routes
    // via this map, falling back to a computed TCP loopback port
    // (`lifecycle::agent_web_port`) only while an agent's socket marker is
    // absent. See `docs/networking/gateway.md::Per-agent unix-socket upstream`.
    if let Err(e) = crate::agent_sockets::write(&agent_names) {
        tracing::warn!(error = ?e, "agent_sockets::write failed (non-fatal)");
    }
    // c0re is unprivileged, so the nginx reload goes through hive-priv.
    if let Err(e) = crate::gateway_nginx::write(&agent_names).await {
        tracing::warn!(error = ?e, "gateway_nginx::write failed (non-fatal)");
    }
    Ok(())
}

/// The names of `agents`, or `None` for an empty list, which [`refresh`]
/// refuses.
fn live_names(agents: &[hive_meta_flake::AgentSpec]) -> Option<Vec<String>> {
    (!agents.is_empty()).then(|| agents.iter().map(|a| a.name.clone()).collect())
}

/// The forge is not optional on a real hive: the hyperhive NixOS module sets
/// `HIVE_FORGE_URL` unconditionally on the hive-c0re unit, so its absence
/// means this daemon was started outside it. The agent option is nullable
/// (`null` legitimately means "no forge" when the modules are evaluated on
/// their own), so nothing downstream would object: every agent would come up
/// with no forge login and no hint that was unintended.
fn require_forge_url(value: Option<&str>) -> Result<()> {
    if value.is_none_or(str::is_empty) {
        bail!(
            "HIVE_FORGE_URL is unset — the hyperhive NixOS module sets it unconditionally on \
             the hive-c0re unit, so this process was started outside that unit. Refusing to \
             register agents that would every one of them have no forge configured."
        );
    }
    Ok(())
}

/// Delete the CA files earlier renders left next to the meta flake
/// (`hive-ca.pem`, and the per-peer `peer-ca-<N>.pem` before it). Agents fetch
/// the swarm root from the store at boot, so nothing reads one.
fn remove_stale_ca_files(dir: &Path) {
    let Ok(entries) = std::fs::read_dir(dir) else {
        return;
    };
    for e in entries.flatten() {
        let fname = e.file_name();
        let Some(name) = fname.to_str() else { continue };
        if is_stale_ca_name(name) {
            let _ = std::fs::remove_file(dir.join(name));
        }
    }
}

/// True for the filenames [`remove_stale_ca_files`] owns.
fn is_stale_ca_name(name: &str) -> bool {
    name == "hive-ca.pem"
        || (name.starts_with("peer-ca-")
            && Path::new(name)
                .extension()
                .is_some_and(|ext| ext.eq_ignore_ascii_case("pem")))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn an_empty_agent_list_is_refused_and_a_live_one_is_named() {
        let spec = |name: &str| hive_meta_flake::AgentSpec {
            name: name.to_owned(),
            port: hive_meta_flake::agent_web_port(name),
            archetype: None,
            pin: None,
        };
        assert_eq!(live_names(&[]), None);
        assert_eq!(
            live_names(&[spec("alice"), spec("bob")]),
            Some(vec!["alice".to_owned(), "bob".to_owned()])
        );
    }

    #[test]
    fn a_forge_url_is_required() {
        require_forge_url(Some("http://forge.example.test"))
            .expect("a forge URL satisfies the requirement");
        for missing in [None, Some("")] {
            let err = require_forge_url(missing).expect_err("no forge URL must stop the refresh");
            assert!(
                err.to_string().contains("HIVE_FORGE_URL is unset"),
                "the error must name the missing variable: {err}"
            );
        }
    }

    #[test]
    fn stale_ca_files_are_removed_and_nothing_else() {
        let dir = tempfile::tempdir().expect("tempdir");
        for name in [
            "hive-ca.pem",
            "peer-ca-0.pem",
            "peer-ca-1.PEM",
            "topology.json",
        ] {
            std::fs::write(dir.path().join(name), "x").expect("write");
        }
        // Not CA files this machinery owns, though they look similar.
        for name in ["peer-ca-0.txt", "other-ca.pem"] {
            std::fs::write(dir.path().join(name), "x").expect("write");
        }
        remove_stale_ca_files(dir.path());
        for gone in ["hive-ca.pem", "peer-ca-0.pem", "peer-ca-1.PEM"] {
            assert!(!dir.path().join(gone).exists(), "{gone} must go");
        }
        for kept in ["topology.json", "peer-ca-0.txt", "other-ca.pem"] {
            assert!(dir.path().join(kept).exists(), "{kept} must survive");
        }
    }
}