hyperhive/hive-c0re/src/swarm_agent_status.rs

//! Offering each locally-hosted agent's status upward to the swarm.
//!
//! Sibling task to [`crate::swarm_status`], not a field folded into it — see
//! `swarm_wire::agent_status`'s module doc for why hive status and
//! agent status live in separate buckets. Same direction, though: a hive
//! **offers** what it already knows about its own agents (via
//! [`crate::container_view::read_agent_status_bounded`], the bounded form of
//! the read `GetAgentMeta` serves to the dashboard), and the swarm
//! controller never reaches down to collect it.
//!
//! **One publish per agent, not one document for the hive.** Each agent
//! gets its own key (`swarm_wire::agent_status::key`), so one
//! agent's status changing does not touch any other agent's key, and a
//! consumer watching one agent reads exactly its own revision history.

use std::time::Duration;

use anyhow::{Context, Result};

use crate::stats::sweep_health::{self, SweepHealth};

/// How often this hive offers a snapshot of every agent it hosts.
///
/// Same value as [`crate::swarm_status::PUBLISH_INTERVAL`] on purpose: both
/// feed the same controller-side `staleAfterSeconds` freshness window, and
/// there is no reason for an agent's status to go stale on a different
/// cadence than the hive's own.
pub const PUBLISH_INTERVAL: Duration = crate::swarm_status::PUBLISH_INTERVAL;

/// Consecutive failed sweeps before the sweep raises a hive warning. Mirrors
/// [`crate::swarm_status`]'s constant of the same name and purpose.
const FAILURES_BEFORE_WARNING: u32 = 3;

/// Start the publish loop, if this deployment wired up a swarm queue.
///
/// Shares its connect gate with [`crate::swarm_status::spawn`] — both wait on
/// `crate::swarm_queue::wait_for_client()`, one connection per process — so a
/// hive with no queue coordinates pays for this decision once, not twice.
///
/// Takes the `Coordinator` only for the set of agents an update has down
/// (`Coordinator::is_updating`), which it publishes with the rest and wakes on.
/// Everything else it reads is agent state that already exists on disk / in the
/// container runtime (`lifecycle::agents_for_meta_listing`,
/// `container_view::read_agent_status_bounded`); it never touches the job queue
/// or broker the way `swarm_status::spawn`'s deploy-event listener does.
pub fn spawn(coord: std::sync::Arc<crate::coordinator::Coordinator>) {
    let mut shutdown = coord.shutdown_rx();
    let Some(hive) = crate::container_view::hive_swarm_names().0 else {
        // Already warned once by `swarm_status::spawn`, which runs the
        // identical check — a second warning for the same missing env var
        // would just be the same fact told twice.
        return;
    };

    tokio::spawn(async move {
        let Some(client) = crate::swarm_queue::wait_for_client().await else {
            return;
        };

        let mut health = SweepHealth::new(
            "swarm_agent_status_publish",
            "warn",
            FAILURES_BEFORE_WARNING,
        );
        loop {
            match publish_all(&client, &hive, &coord).await {
                Ok(published) => {
                    health.record_ok();
                    tracing::debug!(published, "agent status: sweep complete");
                }
                Err(e) => {
                    tracing::warn!(error = ?e, "agent status: sweep failed");
                    let err = format!("{e:#}");
                    health.record_err(|ctx| {
                        let age = ctx.since_last_ok.map_or_else(
                            || "no success this session".to_owned(),
                            |d| format!("last ok {} ago", sweep_health::fmt_age(d)),
                        );
                        format!(
                            "agent status publishing is failing ({} consecutive, {age}) \
                             — the swarm sees this hive's agents as stale, the agents \
                             themselves are unaffected: {err}",
                            ctx.consecutive
                        )
                    });
                }
            }
            // Publish first, then wait — same reasoning as
            // `swarm_status::spawn`: a hive that just came up is exactly the
            // one whose agents someone is looking at.
            tokio::select! {
                () = tokio::time::sleep(PUBLISH_INTERVAL) => {}
                // An agent entering or leaving an update is the one change worth
                // sending ahead of the sweep: the window can be shorter than it.
                () = coord.updating_changed() => {}
                _ = shutdown.changed() => {
                    tracing::info!("agent status: shutdown signal received");
                    break;
                }
            }
        }
    });
}

/// Offer one snapshot of every agent this hive currently hosts.
///
/// Returns the count published, purely so the caller can log it — nothing
/// downstream reads the number. Enumeration failure (cannot list agents at
/// all) is the one thing that fails the whole sweep; a single agent's
/// status read or publish failing is logged and skipped; skipping one
/// agent should not report every *other* agent on this hive as stale too.
async fn publish_all(
    client: &async_nats::Client,
    hive: &str,
    coord: &crate::coordinator::Coordinator,
) -> Result<usize> {
    // Same hang risk `swarm_status::publish` guards against: an unconnected
    // client does not fail a JetStream request, it hangs on it.
    swarm_queue_client::ensure_connected(client)?;

    let agents = crate::lifecycle::agents_for_meta_listing()
        .await
        .context("enumerating this hive's agents")?;

    let store = swarm_queue_client::agent_status::open_or_create(client).await?;
    let mut published = 0usize;
    for spec in &agents {
        let Ok(ident) = hive_types::Ident::parse(&spec.name) else {
            // An agent name that fails `Ident::parse` here would mean
            // something already on disk violates the naming charset every
            // other path enforces at creation — worth a warning, not a
            // reason to abandon everyone else's publish.
            tracing::warn!(agent = %spec.name, "agent status: name failed Ident::parse, skipping");
            continue;
        };
        let Some((status_text, status_set_at, running)) =
            crate::container_view::read_agent_status_bounded(&ident).await
        else {
            tracing::warn!(agent = %spec.name, "agent status: unit state unreadable, skipping");
            continue;
        };
        let payload = swarm_wire::agent_status::AgentStatus {
            status_text,
            status_set_at,
            running,
            updating: coord.is_updating(&spec.name),
        };
        let bytes = match serde_json::to_vec(&payload) {
            Ok(b) => b,
            Err(e) => {
                tracing::warn!(agent = %spec.name, error = %e, "agent status: serialising failed, skipping");
                continue;
            }
        };
        let key = swarm_wire::agent_status::key(hive, &spec.name);
        if let Err(e) = store.put(key, bytes.into()).await {
            tracing::warn!(agent = %spec.name, error = %e, "agent status: publish failed, skipping");
            continue;
        }
        published += 1;
    }
    Ok(published)
}