diff --git a/Cargo.lock b/Cargo.lock index 58117b75..f0956b19 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1664,6 +1664,7 @@ name = "hive-c0re" version = "0.1.0" dependencies = [ "anyhow", + "async-nats", "axum", "base64", "bcrypt", @@ -1695,6 +1696,7 @@ dependencies = [ "serde", "serde_json", "sha2 0.11.0", + "swarm-queue-client", "tempfile", "tokio", "tokio-stream", diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index 9476a73d..8f23a07a 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -9,6 +9,9 @@ workspace = true [dependencies] anyhow.workspace = true +# Named directly only for the client type the swarm status publisher passes +# around; the connect itself lives in `swarm-queue-client` below. +async-nats.workspace = true axum.workspace = true chrono.workspace = true base64.workspace = true @@ -52,6 +55,10 @@ sha2.workspace = true rusqlite.workspace = true serde.workspace = true serde_json.workspace = true +# Offering this hive's status to the swarm (`swarm_status`). The same crate +# the swarm controller reads it with, and `kv` for the same reason: the +# bucket's name and creation config belong to neither end of it alone. +swarm-queue-client = { workspace = true, features = ["kv"] } tokio.workspace = true tokio-stream.workspace = true tracing.workspace = true diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 85c757f7..9515f47c 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -35,6 +35,7 @@ mod snapshot_push; mod socket_server; mod stats; mod stores; +mod swarm_status; mod webhook_secret; mod workers; @@ -476,6 +477,11 @@ async fn cmd_serve( } } }); + // Offer this hive's readiness to the swarm, if one is configured. + // A no-op on a standalone hive (no queue env, logged once) — see + // swarm_status, which owns the whole task including its own decision + // not to start. + swarm_status::spawn(coord.shutdown_rx()); // Per-agent events.sqlite + bash-tasks file cleanup now runs // agent-side in the harness (`hive_agent::vacuum`): the files are // agent-owned, so host-side deletes hit PermissionDenied / readonly-db diff --git a/hive-c0re/src/swarm_status.rs b/hive-c0re/src/swarm_status.rs new file mode 100644 index 00000000..a32b31fb --- /dev/null +++ b/hive-c0re/src/swarm_status.rs @@ -0,0 +1,163 @@ +//! Offering this hive's readiness upward to the swarm. +//! +//! A hive **offers**; the swarm controller never reaches down to collect. +//! That direction is the design and not an implementation detail: the +//! gateway has gone down in a way where every recovery channel ran +//! through the one broken thing, so a status path that depended on the +//! controller would go dark exactly when it is needed. A hive computes +//! its own status locally whether or not the swarm can be reached, and +//! this task is only the part that carries it. +//! +//! **It publishes what the hive already says about itself.** +//! [`crate::warnings::readiness`] is the same value `/health/ready` +//! serves — not a swarm-specific recomputation. Two producers of "is this +//! hive OK" would be free to disagree, and the disagreement would surface +//! as the dashboard and the swarm view contradicting each other about the +//! same host, each internally consistent. +//! +//! **The key is this hive's `hiveName`.** Nothing here has to arrange for +//! that to match the controller's roster: `swarm.nix` asserts that +//! `services.hyperhive.hiveName` is a key of `services.hyperhive.swarm.hives`, +//! so a hive that evaluates at all publishes under a name the roster +//! knows. The alternative — a key the controller has never heard of — +//! renders as `unknown` rather than being silently dropped, but the +//! assertion means it should not arise. +//! +//! **Nothing here stamps a time.** Freshness is derived by the reader +//! from when the value landed in the bucket, so a hive cannot make itself +//! look fresher than it is, and a hive with a wrong clock skews only its +//! own payload. + +use std::time::Duration; + +use anyhow::{Context, Result}; + +use crate::stats::sweep_health::{self, SweepHealth}; + +/// How often this hive offers a snapshot. +/// +/// **One decision with the controller's `staleAfterSeconds` (default +/// 120s), not two.** The ratio is what either option means: at 2, a +/// single lost publish still reads `fresh` and two consecutive misses +/// read `stale`. Set them equal and any one hiccup alarms; open the gap +/// wide and `stale` stops meaning anything. Changing one without the +/// other silently re-tunes the swarm's definition of "quiet". +pub const PUBLISH_INTERVAL: Duration = Duration::from_mins(1); + +/// Env var prefix for this daemon's swarm-queue credentials — see +/// [`swarm_queue_client::QueueConfig::from_env`]. All four or none. +const ENV_PREFIX: &str = "HIVE_C0RE"; + +/// Consecutive failed publishes before the dashboard banners. +/// +/// At [`PUBLISH_INTERVAL`] this is ~3 minutes of genuine failure, so a +/// blip does not flap a banner an operator learns to ignore. +const FAILURES_BEFORE_BANNER: u32 = 3; + +/// Start the publish loop, if this deployment wired up a swarm queue. +/// +/// Absent queue config is the ordinary case — most hives are not in a +/// swarm — so it is an `info` and not a warning. A *half*-set environment +/// is a different thing entirely and [`swarm_queue_client::QueueConfig::from_env`] +/// makes it a hard error; it is bannered here rather than swallowed, +/// because the failure it otherwise produces is a hive that looks fine +/// and silently never reports. +pub fn spawn(mut shutdown: tokio::sync::watch::Receiver) { + let cfg = match swarm_queue_client::QueueConfig::from_env(ENV_PREFIX) { + Ok(Some(cfg)) => cfg, + Ok(None) => { + tracing::info!("no swarm queue configured; this hive offers no status upward"); + return; + } + Err(e) => { + // A one-shot startup step with no later retry to clear it — + // exactly what `set_boot_warning` is for. The fix is a + // redeploy, which restarts this process anyway. + crate::warnings::set_boot_warning( + "swarm_status_config", + "warn", + format!("swarm status publishing is off: {e:#}"), + ); + return; + } + }; + + let Some(hive) = crate::container_view::hive_swarm_names().0 else { + crate::warnings::set_boot_warning( + "swarm_status_config", + "warn", + "swarm status publishing is off: HYPERHIVE_HIVE_NAME is unset, so this \ + hive has no key to publish under", + ); + return; + }; + + tokio::spawn(async move { + // `retry_on_initial_connect` inside, so this returns a client + // that may not be connected yet rather than failing on a queue + // that comes up second. The publish below is what discovers that, + // and it is already the thing that reports it. + let client = match swarm_queue_client::connect(cfg).await { + Ok(client) => client, + Err(e) => { + crate::warnings::set_boot_warning( + "swarm_status_config", + "warn", + format!("swarm status publishing is off: {e:#}"), + ); + return; + } + }; + + let mut health = SweepHealth::new("swarm_status_publish", "warn", FAILURES_BEFORE_BANNER); + loop { + match publish(&client, &hive).await { + Ok(()) => health.record_ok(), + Err(e) => { + tracing::warn!(error = ?e, "swarm status: publish 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!( + "swarm status publishing is failing ({} consecutive, {age}) \ + — the swarm sees this hive as stale, the hive itself is \ + unaffected: {err}", + ctx.consecutive + ) + }); + } + } + // Publish first, then wait: a hive that has just come up is + // exactly the one whose status someone is looking at, and + // sleeping first would make every restart read `stale` for a + // full interval. The shutdown arm means a stop is not spent + // waiting out that interval either. + tokio::select! { + () = tokio::time::sleep(PUBLISH_INTERVAL) => {} + _ = shutdown.changed() => { + tracing::info!("swarm status: shutdown signal received"); + break; + } + } + } + }); +} + +/// Offer one snapshot: this hive's current readiness, under its own key. +async fn publish(client: &async_nats::Client, hive: &str) -> Result<()> { + // An unconnected client does not fail a JetStream request, it hangs + // on it — which here would hang the loop, shutdown arm included. + swarm_queue_client::ensure_connected(client)?; + + let store = swarm_queue_client::status::open_or_create(client).await?; + let payload = + serde_json::to_vec(&crate::warnings::readiness()).context("serialising the readiness")?; + store + .put(hive, payload.into()) + .await + .with_context(|| format!("publishing the status snapshot for {hive}"))?; + Ok(()) +}