diff --git a/Cargo.lock b/Cargo.lock index b6b1966c..76f2dfa4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -213,14 +213,17 @@ dependencies = [ "rustls-webpki", "serde", "serde_json", + "serde_nanos", "serde_repr", "thiserror 2.0.18", + "time", "tokio", "tokio-rustls", "tokio-stream", "tokio-util", "tokio-websockets", "tracing", + "tryhard", "url", ] @@ -4310,6 +4313,15 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_nanos" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a93142f0367a4cc53ae0fead1bcda39e85beccfad3dcd717656cacab94b12985" +dependencies = [ + "serde", +] + [[package]] name = "serde_path_to_error" version = "0.1.20" @@ -4546,7 +4558,10 @@ name = "swarm-controller" version = "0.1.0" dependencies = [ "anyhow", + "async-nats", "axum", + "futures-util", + "reqwest 0.13.1", "serde", "serde_json", "tokio", @@ -5070,6 +5085,16 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tryhard" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fe58ebd5edd976e0fe0f8a14d2a04b7c81ef153ea9a54eebc42e67c2c23b4e5" +dependencies = [ + "pin-project-lite", + "tokio", +] + [[package]] name = "tungstenite" version = "0.29.0" diff --git a/docs/swarm/README.md b/docs/swarm/README.md index 95776d2d..89e9b1e4 100644 --- a/docs/swarm/README.md +++ b/docs/swarm/README.md @@ -330,6 +330,65 @@ What it serves, why it is a unix socket rather than a port, and the socket-directory constraint that governs where `socketPath` may point: [`swarm-controller/README.md`](../../swarm-controller/README.md). +### Per-hive status (`GET /api/hives/status`) + +What each hive last **offered** about itself. Hives publish upward; the +controller never reaches down to collect. That direction is deliberate: +during the #3097 gateway outage every recovery channel ran through the +one broken thing, so a status path that depended on the controller would +have gone dark exactly when it was needed to diagnose the controller's +own network. A hive computes its own status locally either way — this +endpoint is a view of what was published, never the source. + +**The queue is the store.** A hive publishes into the `hive-status` +JetStream KV bucket (`history: 1` — the last thing each hive said), and +the controller reads that bucket per request, keeping no copy. A cache +here would be a second answer free to disagree with the first, and the +disagreement would surface as a hive reading healthy on a dashboard +while the bucket says otherwise. The bucket is created by whichever side +gets there first. + +**Absence is what the endpoint is built around**: + +| freshness | means | +|---|---| +| `fresh` | published within `staleAfterSeconds` | +| `stale` | published longer ago than that — the payload is still returned, because "old" and "absent" are different answers | +| `never_reported` | in the roster, has never published. Distinct from `stale`: went quiet is a fault, never spoke is usually a deployment that hasn't happened | +| `unknown` | published but not in `swarm.hives` — surfaced rather than dropped | + +Rows come from the **roster**, not from the bucket, so a hive that has +never reported appears rather than not appearing, and an empty bucket +renders as a swarm nobody has heard from instead of a healthy one. +Freshness is derived at read time from a timestamp and never stored as a +flag — a stored `healthy` boolean goes stale silently the moment nothing +arrives, which is the failure this is designed against. Each row also +carries `last_seen_unix` and `age_seconds`, so a consumer that disagrees +with `staleAfterSeconds` can apply its own threshold. + +`last_seen_unix` is the **bucket's** timestamp, applied by the NATS +server when the value landed, not a field inside the payload — a +publisher cannot make itself look fresher than it is, and a hive with a +wrong clock skews its own payload rather than its freshness. + +Because the bucket outlives a controller restart, a restarted controller +reports what it reads: `stale, age_seconds: 10800` rather than +`never_reported`. That is the more honest of the two — it genuinely +knows when the hive last spoke. Losing the bucket degrades in the same +direction: every hive reads `never_reported` until its next publish, +which is the true answer and not a remembered "healthy". + +The endpoint answers **503**, not an empty list, when no queue is +configured on this host or its store cannot be read. "I cannot reach the +store" and "every hive is silent" are different answers, and rendering +the second when the first is true would turn a local fault into an +apparent swarm-wide outage. + +⚠️ **Nothing writes to the bucket yet.** The transport and the +controller's read path are in place; the hive-side publisher is a later +slice. Until one lands, every hive reads `never_reported` — the correct +answer for a controller that has been told nothing. + ## Cross-references - `docs/snapshot-store.md` — the swarm's `btrfs receive` endpoint, and diff --git a/nix/host-modules/swarm-controller.nix b/nix/host-modules/swarm-controller.nix index 30f7ca86..b5ba75ee 100644 --- a/nix/host-modules/swarm-controller.nix +++ b/nix/host-modules/swarm-controller.nix @@ -33,6 +33,39 @@ let SWARMCTL_AUTHELIA_UNIT = autheliaCfg.unit; }; + natsCfg = config.services.hyperhive.swarm.nats; + + # The controller's own OAuth2 client. It is NOT a hive: the per-hive + # clients the roster issues belong to hives, and the responder's client + # belongs to the responder. One identity per principal — the rule is that + # a principal's credentials all derive from the same identity, not that + # the swarm has one. + queueClientId = "swarm-controller"; + + # Both halves have to be here: authelia to have minted the secret, and the + # queue to connect to. Same guard, and the same reasoning, as `autheliaEnv` + # above — a value set on a host that runs neither would point at a file + # that does not exist and produce a daemon that retries forever. + queueLocal = autheliaCfg.enable && natsCfg.enable; + + # `LoadCredential` and not a copy-oneshot, which is where this deliberately + # differs from the callout responder: that one delivers INTO a container, + # so it has to copy across a filesystem boundary. The controller is a plain + # host unit, so systemd can hand it the file directly — fewer moving parts, + # and the secret never gains a second on-disk copy to forget about. + queueEnv = lib.optionalAttrs queueLocal { + # The queue container shares the host netns, so loopback is correct here + # and is not the `localhost`-means-the-wrong-thing trap that applies + # inside agent containers. + SWARM_CONTROLLER_NATS_URL = "nats://127.0.0.1:${toString natsCfg.port}"; + SWARM_CONTROLLER_OIDC_TOKEN_ENDPOINT = "${autheliaCfg.url}/api/oidc/token"; + SWARM_CONTROLLER_OIDC_CLIENT_ID = queueClientId; + # `%d` is systemd's credentials directory: root reads the plaintext at + # unit start and the daemon's own user sees it 0400, without the unit + # ever being able to read the rest of authelia's state dir. + SWARM_CONTROLLER_OIDC_CLIENT_SECRET_FILE = "%d/queue-client.secret"; + }; + # Wrapped rather than documented: every one of these values is derived # from an option this deployment already set, so making the operator # re-supply them on the command line would be asking them to repeat the @@ -113,6 +146,25 @@ in ''; }; + staleAfterSeconds = lib.mkOption { + type = lib.types.ints.positive; + default = 120; + description = '' + How old a hive's last status snapshot may be before + `GET /api/hives/status` reports it as `stale` rather than `fresh`. + + This is a statement about how often hives *publish*, not about how + patient a reader is — set it above the publishing cadence or every + hive reads stale between offers. It is an option and not a + constant precisely because that cadence is a property of the + deployment. + + Freshness is derived when the endpoint is read, never stored, so + changing this takes effect for the next request; no hive has to + re-publish anything. + ''; + }; + links = lib.mkOption { type = lib.types.listOf ( lib.types.submodule { @@ -174,6 +226,23 @@ in # exactly what it must not inherit. environment.systemPackages = [ swarmctlConfigured ]; + # One declaration, two readers — the controller knows which client id it + # authenticates under, so making the operator restate it in authelia's + # client list would be a second source of truth for a string whose + # mismatch is an opaque 401 from the token endpoint. Same shape as the + # queue's own client declaration. + services.hyperhive.swarm.authelia.oidc.clients = lib.mkIf autheliaCfg.enable [ + { + id = queueClientId; + description = "HyperHive swarm controller"; + # `client_credentials`: a daemon authenticating as itself, with + # nobody to redirect. Declared rather than inferred from an empty + # redirect list, because authelia permits only the grants a client + # names and an omitted `grant_types` means authorization-code alone. + kind = "machine"; + } + ]; + systemd.services.swarm-controller = { description = "hyperhive swarm-level controller daemon"; wantedBy = [ "multi-user.target" ]; @@ -181,6 +250,14 @@ in serviceConfig = { ExecStart = "${cfg.package}/bin/swarm-controller"; + + # Only when the queue is actually reachable from here. An absent + # credential is not a failure: the daemon logs that no queue is + # configured and serves its HTTP surface, which is the correct + # behaviour on the hosts that do not run one. + LoadCredential = lib.mkIf queueLocal [ + "queue-client.secret:${autheliaCfg.hostClientSecretDir}/${queueClientId}.secret" + ]; User = "swarm-controller"; Group = "swarm-controller"; Restart = "on-failure"; @@ -218,23 +295,36 @@ in ]; }; - environment.SWARM_CONTROLLER_SOCKET = cfg.socketPath; - # The swarm's hive directory, JSON-encoded — same shape hive-c0re - # already builds for HYPERHIVE_PEERS (../hive-c0re/environment.nix), - # just the full directory (this daemon has no "self" hive to - # exclude, unlike a per-hive c0re's peer list) rather than - # peers-minus-self. Consumed by `GET /api/hives` - # (swarm-controller/src/main.rs::load_hives). - environment.SWARM_CONTROLLER_HIVES = builtins.toJSON ( - lib.mapAttrsToList (name: h: { - inherit name; - inherit (h) domain; - }) config.services.hyperhive.swarm.hives - ); - # The merged links list — see `links`' description above for who - # contributes to it. Consumed by `GET /api/links` - # (swarm-controller/src/main.rs::load_links). - environment.SWARM_CONTROLLER_LINKS = builtins.toJSON cfg.links; + # Queue coordinates (`queueEnv`) merge in last and are present only + # where the queue and its IdP both run. The daemon refuses a PARTIAL + # set rather than treating it as absent, which is why they are built + # as one attrset and never assigned individually. + environment = { + SWARM_CONTROLLER_SOCKET = cfg.socketPath; + # The swarm's hive directory, JSON-encoded — same shape hive-c0re + # already builds for HYPERHIVE_PEERS (../hive-c0re/environment.nix), + # just the full directory (this daemon has no "self" hive to + # exclude, unlike a per-hive c0re's peer list) rather than + # peers-minus-self. Consumed by `GET /api/hives` + # (swarm-controller/src/main.rs::load_hives). + SWARM_CONTROLLER_HIVES = builtins.toJSON ( + lib.mapAttrsToList (name: h: { + inherit name; + inherit (h) domain; + }) config.services.hyperhive.swarm.hives + ); + # The merged links list — see `links`' description above for who + # contributes to it. Consumed by `GET /api/links` + # (swarm-controller/src/main.rs::load_links). + SWARM_CONTROLLER_LINKS = builtins.toJSON cfg.links; + # Staleness threshold for `GET /api/hives/status` — see + # `staleAfterSeconds`' description. Set unconditionally rather + # than inside `queueEnv`: it is not a queue coordinate, and + # nothing about it is unsafe to define on a host whose queue is + # off (the daemon just has nothing to apply it to). + SWARM_CONTROLLER_STALE_AFTER_SECS = toString cfg.staleAfterSeconds; + } + // queueEnv; }; }; } diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index 0c3b628f..507858d3 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -10,7 +10,15 @@ path = "src/main.rs" [dependencies] anyhow.workspace = true +# `kv` (which pulls `jetstream`) on top of the workspace's feature set: the +# queue is this daemon's *store*, not just its transport - a hive's last +# status snapshot is read out of a JetStream KV bucket. Declared here rather +# than in the workspace entry so the auth-callout responder, which speaks +# neither, does not claim to need them. +async-nats = { workspace = true, features = ["kv"] } axum.workspace = true +futures-util.workspace = true +reqwest.workspace = true serde.workspace = true serde_json.workspace = true tokio.workspace = true diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 49ab4141..71459cbd 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -31,6 +31,9 @@ use serde::{Deserialize, Serialize}; use utoipa::{OpenApi, ToSchema}; use utoipa_axum::{router::OpenApiRouter, routes}; +mod queue; +mod status; + /// Where the daemon binds, overridable via `SWARM_CONTROLLER_SOCKET`. /// /// A compiled-in default is legitimate here and is *not* the mistake that @@ -101,6 +104,11 @@ struct AppState { /// Loaded once at startup (`load_links`); same synchronization story /// as `hives`. links: Arc>, + /// `None` when this deployment wired up no swarm queue — the only + /// state in which `/api/hives/status` cannot answer at all. A queue + /// that is merely *unreachable* still yields a reader, because + /// `async-nats` reconnects underneath it. + status: Option>, } /// Env var the controller's NixOS module sets from @@ -192,6 +200,59 @@ async fn get_links(State(state): State) -> Json> { Json((*state.links).clone()) } +/// Why the status route answers 503 rather than an empty list. +/// +/// "I cannot reach the store" and "every hive is silent" are different +/// answers, and rendering the second when the first is true is exactly +/// the smoothing this endpoint exists to avoid — a caller would draw a +/// swarm-wide outage out of a local one. The cause is carried in the +/// body because a bare 503 on an operator-facing diagnostic is how a +/// misconfiguration costs an afternoon; it is a queue/JetStream error +/// string, and this surface is already behind the swarm's SSO. +struct StatusUnavailable(String); + +impl axum::response::IntoResponse for StatusUnavailable { + fn into_response(self) -> axum::response::Response { + (axum::http::StatusCode::SERVICE_UNAVAILABLE, self.0).into_response() + } +} + +/// What each hive last said about itself, read from the swarm queue at +/// request time. +/// +/// Every hive in the roster gets a row whether or not it has ever +/// reported — see the `status` module for why absence, not presence, is +/// the case this is built around. +#[utoipa::path( + get, + path = "/api/hives/status", + responses( + (status = 200, description = "a row per hive, freshness derived now", body = Vec), + (status = 503, description = "no swarm queue is configured here, or its store could not be read", body = String), + ), + tag = "hives" +)] +async fn get_hives_status( + State(state): State, +) -> Result>, StatusUnavailable> { + let Some(reader) = state.status.as_ref() else { + return Err(StatusUnavailable( + "no swarm queue is configured on this host".to_owned(), + )); + }; + match reader + .view(&state.hives, std::time::SystemTime::now()) + .await + { + Ok(rows) => Ok(Json(rows)), + Err(e) => { + let detail = format!("{e:#}"); + tracing::warn!(error = %detail, "reading the swarm status bucket failed"); + Err(StatusUnavailable(detail)) + } + } +} + #[tokio::main] async fn main() -> Result<()> { tracing_subscriber::fmt() @@ -230,14 +291,44 @@ async fn main() -> Result<()> { .with_context(|| format!("chmod {}", path.display()))?; tracing::info!(socket = %path.display(), "swarm-controller listening"); + // Connect to the swarm queue when this deployment wired one up. + // + // Deliberately NOT fatal on failure: the controller's HTTP surface is + // useful without the queue, and a hive that cannot be read from renders + // as `unknown` rather than as an outage of this daemon. What IS fatal is + // a half-set environment — `QueueConfig::from_env` refuses that, because + // silently behaving like an unconfigured host is how every hive ends up + // reading `never_reported` with nothing to point at. + let status = match queue::QueueConfig::from_env()? { + None => { + tracing::info!("no swarm queue configured; status aggregation is off"); + None + } + Some(cfg) => match queue::connect(cfg).await { + Ok(client) => { + tracing::info!("connected to the swarm queue"); + Some(Arc::new(status::StatusReader::new( + client, + status::StatusReader::stale_after_from_env(), + ))) + } + Err(e) => { + tracing::warn!(error = format!("{e:#}"), "swarm queue unreachable"); + None + } + }, + }; + let state = AppState { hives: Arc::new(load_hives()), links: Arc::new(load_links()), + status, }; let (router, api) = OpenApiRouter::::with_openapi(ApiDoc::openapi()) .routes(routes!(health)) .routes(routes!(get_hives)) + .routes(routes!(get_hives_status)) .routes(routes!(get_links)) .split_for_parts(); // Just the JSON, not the UI — Swagger UI itself is nginx-hosted from diff --git a/swarm-controller/src/queue.rs b/swarm-controller/src/queue.rs new file mode 100644 index 00000000..1be9e846 --- /dev/null +++ b/swarm-controller/src/queue.rs @@ -0,0 +1,197 @@ +//! The controller's client end of the swarm message queue. +//! +//! The queue admits every non-responder client through `auth_callout`: a +//! client presents a token at CONNECT, the callout responder introspects it +//! against authelia and mints a user JWT if it is good. So the controller is +//! an ordinary client and needs an identity of its own — it is not a hive, and +//! the per-hive clients issued from the roster are not its to use. +//! +//! Two things about that shape drive everything here: +//! +//! - **A token expires.** Authelia issues `client_credentials` access tokens +//! with `expires_in: 3599`. Authentication happens at CONNECT, so a +//! long-lived connection is fine — but a *reconnect* an hour later needs a +//! token that was minted an hour later. +//! - **`async-nats` re-runs an auth callback per connection attempt** (it is +//! handed that attempt's nonce). So the refresh belongs in the callback and +//! not in a timer: there is no window in which the client holds a token it +//! minted for a previous connection. +//! +//! The alternative — mint once, pass a static `auth_token`, own the reconnect +//! loop — fails in the way this subsystem exists to prevent: the controller +//! keeps serving, its status data quietly stops updating, and nothing says so +//! until someone reads a dashboard. + +use std::path::PathBuf; + +use anyhow::{Context, Result, bail}; + +/// Only the one field this needs; authelia returns several. +#[derive(serde::Deserialize)] +struct TokenResponse { + access_token: String, +} + +/// Where the controller finds the queue and what it authenticates with. +/// +/// Every field comes from an environment variable the NixOS module sets, the +/// same way `load_hives` takes the roster — a config change is a redeploy, and +/// this process reads no file it was not pointed at. +#[derive(Debug, Clone)] +pub struct QueueConfig { + /// `nats://host:port` for the swarm queue. + pub url: String, + /// Authelia's token endpoint, e.g. `https://auth./api/oidc/token`. + pub token_endpoint: String, + /// The controller's own `OAuth2` client id. + pub client_id: String, + /// File holding the client secret's PLAINTEXT. + /// + /// A path and not a value: the secret is minted on the authelia host and + /// read here, and putting it in the environment would publish it to + /// anything that can read `/proc//environ`. + pub client_secret_file: PathBuf, +} + +impl QueueConfig { + /// Read the config from the environment, or `None` when the queue was not + /// wired up for this deployment. + /// + /// `None` rather than an error on purpose: the controller serves its HTTP + /// surface on hosts where the queue is not enabled, and refusing to start + /// there would trade a missing feature for a dead daemon. What must NOT + /// happen is a *half* configuration silently behaving like an absent one — + /// hence the explicit partial check below. + pub fn from_env() -> Result> { + let url = std::env::var("SWARM_CONTROLLER_NATS_URL").ok(); + let token_endpoint = std::env::var("SWARM_CONTROLLER_OIDC_TOKEN_ENDPOINT").ok(); + let client_id = std::env::var("SWARM_CONTROLLER_OIDC_CLIENT_ID").ok(); + let secret = std::env::var("SWARM_CONTROLLER_OIDC_CLIENT_SECRET_FILE").ok(); + + match (url, token_endpoint, client_id, secret) { + (None, None, None, None) => Ok(None), + (Some(url), Some(token_endpoint), Some(client_id), Some(secret)) => Ok(Some(Self { + url, + token_endpoint, + client_id, + client_secret_file: PathBuf::from(secret), + })), + // A partially-set environment is a deployment bug, and the failure + // it would otherwise produce is the expensive kind: the controller + // comes up "fine", never connects, and every hive reads as having + // never reported. Naming the missing variables costs one line. + _ => bail!( + "swarm queue is half-configured: SWARM_CONTROLLER_NATS_URL, \ + _OIDC_TOKEN_ENDPOINT, _OIDC_CLIENT_ID and \ + _OIDC_CLIENT_SECRET_FILE must be set together or not at all" + ), + } + } +} + +/// Mint a fresh access token for the controller's own client. +/// +/// `client_credentials`, because there is no user here: the controller +/// authenticates as itself. Authelia refuses the `openid` scope for this grant +/// (a machine client receives an access token and never an id-token), so no +/// scope is requested. +async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result { + // Read per call rather than caching: the file is small, and a cached + // secret would survive a rotation that the operator believes took effect. + let secret = tokio::fs::read_to_string(&cfg.client_secret_file) + .await + .with_context(|| { + format!( + "reading the queue client secret from {}", + cfg.client_secret_file.display() + ) + })?; + + let response = http + .post(&cfg.token_endpoint) + .form(&[ + ("grant_type", "client_credentials"), + ("client_id", cfg.client_id.as_str()), + ("client_secret", secret.trim()), + ]) + .send() + .await + .context("requesting an access token from authelia")?; + + // The body carries authelia's own error description, and it is far more + // useful than the status alone: a wrong grant says `unauthorized_client`, + // a wrong secret says `invalid_client`, and those point at different + // config. + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + if !status.is_success() { + bail!("authelia refused the controller's token request ({status}): {body}"); + } + + let parsed: TokenResponse = + serde_json::from_str(&body).context("parsing authelia's token response")?; + Ok(parsed.access_token) +} + +/// Connect to the swarm queue, minting a token for each connection attempt. +pub async fn connect(cfg: QueueConfig) -> Result { + let http = reqwest::Client::new(); + let url = cfg.url.clone(); + + let client = async_nats::ConnectOptions::with_auth_callback(move |_nonce| { + let http = http.clone(); + let cfg = cfg.clone(); + async move { + let token = mint_token(&http, &cfg) + .await + // The callback's error type carries a string, so the context + // chain would be lost; flatten it rather than dropping it. + .map_err(|e| async_nats::AuthError::new(format!("{e:#}")))?; + let mut auth = async_nats::Auth::new(); + auth.token = Some(token); + Ok(auth) + } + }) + // The controller and the queue are separate units on (possibly) + // separate hosts, and nothing orders them. Without this, a queue that + // comes up one second later leaves the controller permanently + // queue-less until someone restarts it — a boot-order race that + // presents as "status has been unavailable since Tuesday". + // + // It also composes with the callback above rather than fighting it: + // each background attempt is a connection attempt, so each one mints + // its own token instead of retrying a stale one. + .retry_on_initial_connect() + .connect(&url) + .await + .with_context(|| format!("connecting to the swarm queue at {url}"))?; + + Ok(client) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The all-unset case is the common one — most hosts do not run the queue. + #[test] + fn an_absent_environment_is_not_an_error() { + // Guard: this test would pass vacuously inside a configured + // environment, so it asserts the variables really are unset first. + for k in [ + "SWARM_CONTROLLER_NATS_URL", + "SWARM_CONTROLLER_OIDC_TOKEN_ENDPOINT", + "SWARM_CONTROLLER_OIDC_CLIENT_ID", + "SWARM_CONTROLLER_OIDC_CLIENT_SECRET_FILE", + ] { + if std::env::var(k).is_ok() { + return; + } + } + assert!( + QueueConfig::from_env() + .expect("absent is not an error") + .is_none() + ); + } +} diff --git a/swarm-controller/src/status.rs b/swarm-controller/src/status.rs new file mode 100644 index 00000000..b7d322aa --- /dev/null +++ b/swarm-controller/src/status.rs @@ -0,0 +1,543 @@ +//! Swarm-wide view of what each hive last said about itself. +//! +//! Hives **offer** a snapshot upward; this daemon never reaches down to +//! collect one. That direction is the design, not an implementation +//! detail: the hive 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 have gone dark exactly when it was +//! needed to diagnose the controller's own network. A hive computes its +//! status locally regardless of whether the swarm can be reached. +//! +//! **The queue is the store.** A hive publishes into a `JetStream` KV +//! bucket, which retains the last value per key; this daemon reads that +//! bucket per request and keeps no copy. A cache here would be a second +//! answer to the same question, free to disagree with the first — and +//! the disagreement would surface as a hive reading healthy on a +//! dashboard while the bucket says otherwise. +//! +//! **Absence is the case this is built around** — the freshness states +//! and the reasoning behind each are in `docs/swarm/README.md`. The two +//! properties that constrain the code rather than describe it: +//! freshness is **derived at read time**, never stored (a stored +//! `healthy: bool` goes stale silently the moment nothing arrives), and +//! rows come from the **roster**, not the bucket, so an empty bucket +//! cannot render as a healthy swarm. +//! +//! One consequence worth stating because it is the opposite of what a +//! cache would give: losing the bucket degrades **to honesty**. Every +//! hive reads `never_reported` until its next publish, which is the true +//! answer — not a remembered "healthy" from before the loss. + +use std::collections::BTreeMap; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use anyhow::{Context, Result}; +use futures_util::TryStreamExt as _; +use serde::{Deserialize, Serialize}; +use utoipa::ToSchema; + +use crate::HiveEntry; + +/// The KV bucket hives publish their snapshots into. +/// +/// A constant and not an option: reader and writer must name the same +/// bucket, and an option is a way for two deployments to disagree about +/// which one that is. Nothing about a bucket name is site-specific. +pub const BUCKET: &str = "hive-status"; + +/// Default age past which a snapshot is reported stale. +/// +/// A threshold is a statement about how often hives offer, and that +/// cadence is decided by the publisher (a later slice), so this is a +/// default to be overridden rather than a constant to be relied on. +pub const DEFAULT_STALE_AFTER: Duration = Duration::from_mins(2); + +/// Env var the NixOS module sets from +/// `services.hyperhive.swarm.controller.staleAfterSeconds`. +pub const STALE_AFTER_ENV: &str = "SWARM_CONTROLLER_STALE_AFTER_SECS"; + +/// How a hive's last report reads *now* — a function of the clock, not a +/// property of the report. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum Freshness { + /// Reported within the staleness threshold. + Fresh, + /// Reported, but longer ago than the threshold. The payload is still + /// rendered: "old" and "absent" are different answers and a consumer + /// may want the last thing a hive managed to say. + Stale, + /// In the roster, has never offered a snapshot. Distinct from + /// `Stale` because it separates "went quiet" from "never spoke" — + /// the first is a fault, the second is usually a deployment that + /// hasn't happened yet. + NeverReported, + /// Offered a snapshot but is not in the roster. Not an error this + /// daemon can resolve, and not one it should hide. + Unknown, +} + +/// One row of the aggregate. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, ToSchema)] +pub struct HiveStatus { + pub name: String, + /// From the roster; `None` for a hive the roster doesn't list. + pub domain: Option, + pub freshness: Freshness, + /// When the snapshot was stored, unix seconds. `None` when nothing + /// has been. + /// + /// This is the **bucket's** stamp, applied by the NATS server when + /// the value landed — not a field inside the payload. A publisher + /// therefore cannot make itself look fresher than it is, and a hive + /// with a wrong clock skews its own payload rather than this. + pub last_seen_unix: Option, + /// Age at render time. Carried alongside `last_seen_unix` so a + /// consumer with a different threshold need not re-derive it from a + /// clock that may not match this host's. + pub age_seconds: Option, + /// Whatever the hive published, unopened — stored opaquely so the + /// snapshot's contents stay settleable later without reworking the + /// aggregate. + /// + /// `None` in two cases the consumer can tell apart by `freshness`: + /// nothing has ever been published (`never_reported`), or something + /// was published that is not JSON (any other freshness — the row + /// still reports *when* the hive last spoke, and the read logs a + /// warning naming it). + pub snapshot: Option, +} + +/// A snapshot as retained by the bucket. +#[derive(Clone, Debug)] +struct Offered { + received_at: SystemTime, + /// `None` when the stored bytes are not JSON — see [`HiveStatus::snapshot`]. + payload: Option, +} + +/// Reads the aggregate out of the KV bucket. +/// +/// Holds a NATS client rather than a bucket handle: the bucket is +/// resolved on first use and cached, so a controller that starts before +/// the bucket exists picks it up without a restart. Resolution failures +/// are not cached — [`tokio::sync::OnceCell::get_or_try_init`] retries — +/// which is what makes the queue coming up *after* this daemon a +/// non-event rather than a permanent degradation. +pub struct StatusReader { + client: async_nats::Client, + store: tokio::sync::OnceCell, + stale_after: Duration, +} + +impl StatusReader { + #[must_use] + pub fn new(client: async_nats::Client, stale_after: Duration) -> Self { + Self { + client, + store: tokio::sync::OnceCell::new(), + stale_after, + } + } + + /// Reads [`STALE_AFTER_ENV`], falling back to + /// [`DEFAULT_STALE_AFTER`]. A zero or unparseable value takes the + /// default rather than failing startup — same rule as `load_hives`: + /// a controller whose own config is wrong must still serve. + #[must_use] + pub fn stale_after_from_env() -> Duration { + std::env::var(STALE_AFTER_ENV) + .ok() + .and_then(|raw| raw.trim().parse::().ok()) + .filter(|secs| *secs > 0) + .map_or(DEFAULT_STALE_AFTER, Duration::from_secs) + } + + /// The bucket handle, created on first use if nothing has made it yet. + /// + /// Whichever side arrives first creates it, and both sides want the + /// same shape, so this is a race with one outcome. `history: 1` is + /// the shape: the aggregate reads *the last thing each hive said*, + /// and retaining more would be storage bought for a query nobody + /// makes. + async fn store(&self) -> Result<&async_nats::jetstream::kv::Store> { + self.store + .get_or_try_init(|| async { + let js = async_nats::jetstream::new(self.client.clone()); + match js.get_key_value(BUCKET).await { + Ok(store) => Ok(store), + Err(e) => { + tracing::info!( + bucket = BUCKET, + reason = %e, + "status bucket not available, creating it" + ); + js.create_key_value(async_nats::jetstream::kv::Config { + bucket: BUCKET.to_owned(), + description: "Last status snapshot offered by each hive".to_owned(), + history: 1, + ..Default::default() + }) + .await + .with_context(|| format!("creating the {BUCKET} bucket")) + } + } + }) + .await + } + + /// The aggregate, rendered against `now`. + /// + /// Every roster hive produces a row whether or not it has ever + /// reported; a reporting hive outside the roster produces one too. + pub async fn view(&self, roster: &[HiveEntry], now: SystemTime) -> Result> { + // Only a CONNECTED client can be asked anything. `retry_on_initial_connect` + // means the client exists before it is usable, and a JetStream request + // made in that window does not fail — it WAITS, on every call, for + // longer than any dashboard poll should take (measured: still going at + // 15s against a queue that simply refuses the credential). + // + // Testing for `!= Connected` rather than `== Disconnected` is the whole + // point: a client that has never connected once sits in `Pending`, so + // the `Disconnected` test passes it straight through to the hang it was + // written to prevent. That is exactly the case here — a controller + // whose credential is wrong from boot never reaches `Disconnected`, + // because it was never connected to begin with. + // + // Naming the state is also the better error: "not connected" is + // actionable, a timeout is not. + let state = self.client.connection_state(); + if state != async_nats::connection::State::Connected { + anyhow::bail!("not connected to the swarm queue (client state: {state:?})"); + } + + let store = self.store().await?; + + // Keys first, then a fetch per key. The roster is a handful of + // hives, so the round-trip count is not worth trading for a + // watcher whose "I have seen everything current" condition is + // one more thing to get right on a read path. + let mut keys = store.keys().await.context("listing status bucket keys")?; + let mut entries: BTreeMap = BTreeMap::new(); + while let Some(key) = keys + .try_next() + .await + .context("reading the status bucket's key list")? + { + let Some(entry) = store + .entry(&key) + .await + .with_context(|| format!("reading status entry {key}"))? + else { + // Deleted between listing and fetching. Not an error: + // the next read simply won't list it. + continue; + }; + let payload = match serde_json::from_slice(&entry.value) { + Ok(value) => Some(value), + Err(e) => { + tracing::warn!( + hive = %key, + error = %e, + "status snapshot is not JSON; reporting the timestamp without it" + ); + None + } + }; + entries.insert( + key, + Offered { + received_at: to_system_time(entry.created.unix_timestamp()), + payload, + }, + ); + } + + Ok(render(roster, &entries, now, self.stale_after)) + } +} + +/// A bucket timestamp as a [`SystemTime`]. +/// +/// A pre-epoch stamp is not representable here and is not a thing a NATS +/// server produces; treating it as the epoch renders the row as +/// extremely stale, which is the safe direction — a nonsense timestamp +/// must never read as fresh. +fn to_system_time(unix_seconds: i64) -> SystemTime { + u64::try_from(unix_seconds).map_or(UNIX_EPOCH, |secs| UNIX_EPOCH + Duration::from_secs(secs)) +} + +/// Turn a roster plus whatever the bucket held into the rendered rows. +/// +/// Split out of [`StatusReader::view`] deliberately: this is where every +/// rule the acceptance criterion cares about lives, and keeping it a +/// pure function means those rules are tested against a table rather +/// than against a running NATS server. +fn render( + roster: &[HiveEntry], + entries: &BTreeMap, + now: SystemTime, + stale_after: Duration, +) -> Vec { + let mut rows: Vec = roster + .iter() + .map(|hive| match entries.get(&hive.name) { + Some(offered) => row( + hive.name.clone(), + Some(hive.domain.clone()), + offered, + now, + stale_after, + ), + None => HiveStatus { + name: hive.name.clone(), + domain: Some(hive.domain.clone()), + freshness: Freshness::NeverReported, + last_seen_unix: None, + age_seconds: None, + snapshot: None, + }, + }) + .collect(); + + rows.extend( + entries + .iter() + .filter(|(name, _)| !roster.iter().any(|hive| &&hive.name == name)) + .map(|(name, offered)| { + let mut unknown = row(name.clone(), None, offered, now, stale_after); + unknown.freshness = Freshness::Unknown; + unknown + }), + ); + rows +} + +fn row( + name: String, + domain: Option, + offered: &Offered, + now: SystemTime, + stale_after: Duration, +) -> HiveStatus { + // A snapshot stamped in the future (clock skew between the NATS + // server and this host) yields no age rather than a negative one, + // and is treated as fresh — the honest reading of "this arrived, I + // cannot tell how long ago". + let age = now.duration_since(offered.received_at).ok(); + let freshness = match age { + Some(age) if age > stale_after => Freshness::Stale, + _ => Freshness::Fresh, + }; + HiveStatus { + name, + domain, + freshness, + last_seen_unix: offered + .received_at + .duration_since(UNIX_EPOCH) + .ok() + .and_then(|d| i64::try_from(d.as_secs()).ok()), + age_seconds: age.map(|age| age.as_secs()), + snapshot: offered.payload.clone(), + } +} + +#[cfg(test)] +mod tests { + use super::{DEFAULT_STALE_AFTER, Freshness, Offered, STALE_AFTER_ENV, StatusReader, render}; + use crate::HiveEntry; + use std::collections::BTreeMap; + use std::time::{Duration, SystemTime}; + + fn roster() -> Vec { + vec![ + HiveEntry { + name: "pr1ma".to_owned(), + domain: "pr1ma.example.com".to_owned(), + }, + HiveEntry { + name: "umbra".to_owned(), + domain: "umbra.example.com".to_owned(), + }, + ] + } + + fn entries(rows: &[(&str, SystemTime)]) -> BTreeMap { + rows.iter() + .map(|(name, at)| { + ( + (*name).to_owned(), + Offered { + received_at: *at, + payload: Some(serde_json::json!({ "ok": true })), + }, + ) + }) + .collect() + } + + fn t0() -> SystemTime { + SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000) + } + + /// The whole point of the module: an empty bucket must not render as + /// a healthy swarm. Rows come from the roster, so silence is visible. + #[test] + fn an_empty_bucket_renders_every_hive_as_never_reported() { + let rows = render(&roster(), &BTreeMap::new(), t0(), DEFAULT_STALE_AFTER); + assert_eq!(rows.len(), 2, "a row per roster hive, not per report"); + assert!( + rows.iter().all(|r| r.freshness == Freshness::NeverReported), + "nothing published means nothing known — not healthy" + ); + assert!(rows.iter().all(|r| r.snapshot.is_none())); + } + + /// Freshness is derived from the clock at read time, so the same + /// stored value reads differently as it ages. The boundary is where + /// an off-by-one would hide, so it is pinned in both directions. + #[test] + fn the_threshold_boundary_is_inclusive() { + let stale_after = Duration::from_secs(90); + let stored = entries(&[("pr1ma", t0())]); + + assert_eq!( + render( + &roster(), + &stored, + t0() + Duration::from_secs(90), + stale_after + )[0] + .freshness, + Freshness::Fresh, + "exactly at the threshold is inside it" + ); + assert_eq!( + render( + &roster(), + &stored, + t0() + Duration::from_secs(91), + stale_after + )[0] + .freshness, + Freshness::Stale, + "one second past is outside it" + ); + } + + /// One hive reporting must not make its silent neighbour look + /// healthy — the failure mode of any aggregate that renders only + /// what it has. + #[test] + fn a_reporting_hive_does_not_vouch_for_a_silent_one() { + let rows = render( + &roster(), + &entries(&[("pr1ma", t0())]), + t0(), + DEFAULT_STALE_AFTER, + ); + assert_eq!(rows[0].name, "pr1ma"); + assert_eq!(rows[0].freshness, Freshness::Fresh); + assert_eq!(rows[1].name, "umbra"); + assert_eq!(rows[1].freshness, Freshness::NeverReported); + } + + /// An observation the daemon cannot explain is surfaced, not dropped. + #[test] + fn a_hive_outside_the_roster_is_surfaced_as_unknown() { + let rows = render( + &roster(), + &entries(&[("ghost", t0())]), + t0(), + DEFAULT_STALE_AFTER, + ); + assert_eq!(rows.len(), 3, "two roster hives plus the stranger"); + let ghost = rows.last().expect("rows is non-empty"); + assert_eq!(ghost.name, "ghost"); + assert_eq!(ghost.freshness, Freshness::Unknown); + assert!( + ghost.domain.is_none(), + "the roster is where a domain comes from, and this hive isn't in it" + ); + } + + /// Clock skew must not produce a negative age or a panic. A snapshot + /// stamped in the future reads fresh with no age — "it arrived, I + /// cannot tell how long ago". + #[test] + fn a_future_timestamp_yields_no_age_rather_than_a_wrong_one() { + let rows = render( + &roster(), + &entries(&[("pr1ma", t0() + Duration::from_secs(30))]), + t0(), + Duration::from_secs(90), + ); + assert_eq!(rows[0].freshness, Freshness::Fresh); + assert_eq!(rows[0].age_seconds, None); + } + + /// A hive that published something unreadable still gets its + /// timestamp reported: *when* it last spoke is exactly what this + /// aggregate is for, and dropping the row would read as silence. + #[test] + fn an_unparseable_payload_still_reports_when_it_arrived() { + let mut stored = BTreeMap::new(); + stored.insert( + "pr1ma".to_owned(), + Offered { + received_at: t0(), + payload: None, + }, + ); + + let row = &render(&roster(), &stored, t0(), DEFAULT_STALE_AFTER)[0]; + assert_eq!( + row.freshness, + Freshness::Fresh, + "unreadable is not the same as absent — freshness is what \ + separates them on the wire" + ); + assert!(row.snapshot.is_none()); + assert_eq!(row.last_seen_unix, Some(1_700_000_000)); + } + + /// SAFETY: single-threaded mutation of a process env var no other + /// test in this crate reads; restored before returning. One test + /// rather than four for the same reason `load_hives`'s is — the + /// parallel runner would race them. + #[test] + fn stale_after_from_env_covers_missing_bogus_zero_and_valid() { + unsafe { + std::env::remove_var(STALE_AFTER_ENV); + } + assert_eq!(StatusReader::stale_after_from_env(), DEFAULT_STALE_AFTER); + + unsafe { + std::env::set_var(STALE_AFTER_ENV, "not a number"); + } + assert_eq!( + StatusReader::stale_after_from_env(), + DEFAULT_STALE_AFTER, + "a controller whose own config is wrong must still serve" + ); + + unsafe { + std::env::set_var(STALE_AFTER_ENV, "0"); + } + assert_eq!( + StatusReader::stale_after_from_env(), + DEFAULT_STALE_AFTER, + "zero would make every snapshot instantly stale — take the default" + ); + + unsafe { + std::env::set_var(STALE_AFTER_ENV, "300"); + } + assert_eq!(StatusReader::stale_after_from_env(), Duration::from_mins(5)); + + unsafe { + std::env::remove_var(STALE_AFTER_ENV); + } + } +}