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/docs/swarm/README.md b/docs/swarm/README.md index 748478b7..14499b37 100644 --- a/docs/swarm/README.md +++ b/docs/swarm/README.md @@ -294,9 +294,9 @@ it said. Hives publish upward through the swarm queue; the controller never reaches down to collect, so a hive that cannot reach the swarm still knows its own state — you just cannot see it from here. -⚠️ **Nothing publishes yet.** The read path is in place; the hive-side -publisher lands in a later change. Until it does, every hive reads -`never_reported`. +A hive publishes only once it has been given the three +`swarm.statusPublish` coordinates below. A hive that has not reads +`never_reported` — it is not broken, it just has nothing to say upward. | freshness | what to do about it | |---|---| @@ -311,8 +311,37 @@ arrival, not one the hive put in its own payload. Set `services.hyperhive.swarm.controller.staleAfterSeconds` (default `120`) **above the rate hives publish at**, or everything reads `stale` -between reports. It takes effect on the next request; nothing has to -re-publish. +between reports. Hives publish once a minute, so the default tolerates +one missed report and flags two. It takes effect on the next request; +nothing has to re-publish. + +### Making a hive report (`swarm.statusPublish`) + +Three options, on the **hive**, set together or not at all — a +half-configured hive is an eval error rather than one that quietly never +reports: + +| option | what to set it to | +|---|---| +| `natsUrl` | where the swarm queue listens, as this hive reaches it | +| `tokenEndpoint` | the swarm IdP's `/api/oidc/token` | +| `clientSecretFile` | path to this hive's client secret, plaintext | + +On a host that runs the queue and the IdP itself, all three default to +the local ones and there is nothing to set. Any other hive needs them +spelled out, and needs the secret to physically be there: the swarm does +not distribute it. Copy `hive-.secret` out of the swarm host's +`swarm.authelia.hostClientSecretDir` with whatever secret management the +deployment already uses. + +The identity is not a choice — a hive authenticates as `hive-` +and publishes under `hiveName`, the same name that keys `swarm.hives`. + +If a hive stops reporting, its own dashboard is the place to look: a +failure to publish raises a warning banner there after three consecutive +misses. It stays `warn` rather than `crit` on purpose — a hive that +cannot reach the queue is not itself unhealthy, so it does not start +calling itself degraded for being unable to say it is fine. The endpoint answers **503** when this host has no swarm queue configured, or has one and cannot read it — deliberately not an empty 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/dashboard/health.rs b/hive-c0re/src/dashboard/health.rs index 6f5837fc..32a669e5 100644 --- a/hive-c0re/src/dashboard/health.rs +++ b/hive-c0re/src/dashboard/health.rs @@ -26,8 +26,6 @@ use axum::{ use serde::Serialize; use utoipa::ToSchema; -use crate::host_stats::ServerWarning; - #[derive(Serialize, ToSchema)] struct LiveBody { status: &'static str, @@ -44,40 +42,37 @@ pub(super) async fn get_health_live() -> Response { (StatusCode::OK, axum::Json(LiveBody { status: "ok" })).into_response() } -#[derive(Serialize, ToSchema)] -struct ReadyBody { - status: &'static str, - warnings: Vec, -} - /// Readiness. /// /// `200` with `{"status":"ok", "warnings": [...]}` unless a `crit`-level -/// warning is currently set in [`crate::warnings::snapshot`], in which -/// case `503` with `{"status":"degraded", ...}`. `warnings` always -/// carries the full current list (including `warn`-level entries not -/// affecting the status) so a poller gets detail either way. +/// warning is currently set, in which case `503` with +/// `{"status":"degraded", ...}`. `warnings` always carries the full +/// current list (including `warn`-level entries not affecting the status) +/// so a poller gets detail either way. +/// +/// The body is [`crate::warnings::Readiness`] rather than a type of this +/// module's own, and the verdict comes from +/// [`crate::warnings::readiness`] rather than being computed here. The +/// swarm status publisher offers that same document upward, and this +/// endpoint deciding "unhealthy" for itself is precisely how the two +/// would drift apart. What stays here is the only part that *is* this +/// endpoint's: the mapping onto an HTTP status code. #[utoipa::path( get, path = "/health/ready", responses( - (status = 200, description = "no crit-level warning set", body = ReadyBody), - (status = 503, description = "at least one crit-level warning set", body = ReadyBody), + (status = 200, description = "no crit-level warning set", body = crate::warnings::Readiness), + (status = 503, description = "at least one crit-level warning set", body = crate::warnings::Readiness), ), tag = "health" )] pub(super) async fn get_health_ready() -> Response { - let warnings = crate::warnings::snapshot(); - let degraded = warnings.iter().any(|w| w.level == "crit"); - let code = if degraded { + let body = crate::warnings::readiness(); + let code = if body.is_degraded() { StatusCode::SERVICE_UNAVAILABLE } else { StatusCode::OK }; - let body = ReadyBody { - status: if degraded { "degraded" } else { "ok" }, - warnings, - }; (code, axum::Json(body)).into_response() } 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/stats/warnings.rs b/hive-c0re/src/stats/warnings.rs index 9a322176..05ea79e3 100644 --- a/hive-c0re/src/stats/warnings.rs +++ b/hive-c0re/src/stats/warnings.rs @@ -30,6 +30,9 @@ use std::collections::BTreeMap; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Mutex, OnceLock}; +use serde::Serialize; +use utoipa::ToSchema; + use crate::host_stats::ServerWarning; /// Monotonic guard-id source. Each [`set_warning`] mints a fresh id so a @@ -89,6 +92,55 @@ pub fn set_warning( WarningGuard { kind, id } } +/// `status` value for a hive with no `crit`-level warning set. +pub const STATUS_OK: &str = "ok"; +/// `status` value for a hive with at least one `crit`-level warning set. +pub const STATUS_DEGRADED: &str = "degraded"; + +/// What this hive currently says about its own health. +/// +/// One type with one producer ([`readiness`]) because there is more than +/// one consumer: `/health/ready` answers a poller with it, and the swarm +/// status publisher offers the same document upward. **Two consumers each +/// deciding for themselves what counts as unhealthy is how they end up +/// disagreeing** — and the disagreement would be invisible, since each +/// would look internally consistent. The day a second degraded condition +/// is added, it is added here and both consumers get it. +#[derive(Debug, Clone, Serialize, ToSchema)] +pub struct Readiness { + /// [`STATUS_OK`] or [`STATUS_DEGRADED`]. + pub status: &'static str, + /// The full current warning list, `warn`-level entries included even + /// though they do not affect `status` — a consumer gets the detail + /// either way rather than having to ask twice. + pub warnings: Vec, +} + +impl Readiness { + /// Whether anything `crit`-level is set. The predicate lives next to + /// the constants that encode it so a caller never spells the string. + #[must_use] + pub fn is_degraded(&self) -> bool { + self.status == STATUS_DEGRADED + } +} + +/// Derive the readiness verdict from the current registry contents. +/// +/// The rule — degraded iff any warning is `crit` — is stated exactly +/// once, here. `warn` is deliberately not degrading: it is the level for +/// "an operator should look", not "stop sending me work". +#[must_use] +pub fn readiness() -> Readiness { + let warnings = snapshot(); + let status = if warnings.iter().any(|w| w.level == "crit") { + STATUS_DEGRADED + } else { + STATUS_OK + }; + Readiness { status, warnings } +} + /// Snapshot of the currently-active warnings, kind-sorted. Cheap read /// behind the registry mutex — safe to call on every `/api/state`. #[must_use] diff --git a/hive-c0re/src/swarm_status.rs b/hive-c0re/src/swarm_status.rs new file mode 100644 index 00000000..822b8564 --- /dev/null +++ b/hive-c0re/src/swarm_status.rs @@ -0,0 +1,177 @@ +//! 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. + // + // `chain`, not `{:#}`: this is the queue client's own error + // type, whose Display ignores the alternate flag, so `{:#}` + // would show only "swarm queue is half-configured" and drop + // which variables are missing. + crate::warnings::set_boot_warning( + "swarm_status_config", + "warn", + format!( + "swarm status publishing is off: {}", + swarm_queue_client::chain(&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) => { + // `chain` for the same reason as above: without it this + // banner reads "connecting to the swarm queue at " + // and drops the nats error that says why. + crate::warnings::set_boot_warning( + "swarm_status_config", + "warn", + format!( + "swarm status publishing is off: {}", + swarm_queue_client::chain(&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(()) +} diff --git a/nix/host-modules/hive-c0re/default.nix b/nix/host-modules/hive-c0re/default.nix index fc3086b9..01362c69 100644 --- a/nix/host-modules/hive-c0re/default.nix +++ b/nix/host-modules/hive-c0re/default.nix @@ -251,9 +251,20 @@ in # path. Same secret the agent containers get (forwarded there via # nspawn --load-credential); this just also hands it to c0re itself. # Empty list (no credential) when otel is off or no header is set. - LoadCredential = lib.optional ( - config.services.hyperhive.otel.enable && config.services.hyperhive.otel.headersCredential != null - ) "otel-headers:${config.services.hyperhive.otel.headersCredential}"; + LoadCredential = + lib.optional ( + config.services.hyperhive.otel.enable && config.services.hyperhive.otel.headersCredential != null + ) "otel-headers:${config.services.hyperhive.otel.headersCredential}" + # The swarm-queue client secret this hive authenticates with to + # publish its own status. `LoadCredential` and not a copy: root + # reads the plaintext at unit start and hive-core sees it 0400 + # under `%d`, so the secret never gains a second on-disk copy + # and the daemon never needs read access to wherever it lives. + # (The callout responder copies instead only because it + # delivers into a container, across a filesystem boundary.) + ++ lib.optional ( + config.services.hyperhive.swarm.statusPublish.clientSecretFile != null + ) "swarm-status-client.secret:${config.services.hyperhive.swarm.statusPublish.clientSecretFile}"; # Sandboxing. hive-c0re is unprivileged (runs as hive-core, never # setuid), makes HTTP requests to forge/matrix/Anthropic (keeps INET), # and delegates all privileged ops to hive-priv via a Unix socket. diff --git a/nix/host-modules/hive-c0re/environment.nix b/nix/host-modules/hive-c0re/environment.nix index 41003e78..a88572fb 100644 --- a/nix/host-modules/hive-c0re/environment.nix +++ b/nix/host-modules/hive-c0re/environment.nix @@ -209,3 +209,22 @@ in in "${s.address}:${toString s.port}"; } +// + # Swarm-queue coordinates for offering this hive's status upward + # (hive-c0re::swarm_status). All four together or none: a half-set + # environment is a deployment bug the daemon refuses to treat as + # "no queue configured", because the failure it would otherwise + # produce is a hive that comes up fine and silently never reports. + # The three-option version of that same rule is asserted at eval in + # ./../swarm.nix, so this can only ever emit a complete set. + lib.optionalAttrs (config.services.hyperhive.swarm.statusPublish.natsUrl != null) { + HIVE_C0RE_NATS_URL = config.services.hyperhive.swarm.statusPublish.natsUrl; + HIVE_C0RE_OIDC_TOKEN_ENDPOINT = config.services.hyperhive.swarm.statusPublish.tokenEndpoint; + # The identity swarm-authelia.nix already declares for every entry in + # `swarm.hives` — the hive does not choose its own name here, it uses + # the one the roster gave it. + HIVE_C0RE_OIDC_CLIENT_ID = "hive-${config.services.hyperhive.hiveName}"; + # `%d` is systemd's credentials directory — see the LoadCredential in + # ./default.nix. The daemon reads a path, never a value. + HIVE_C0RE_OIDC_CLIENT_SECRET_FILE = "%d/swarm-status-client.secret"; + } diff --git a/nix/host-modules/swarm.nix b/nix/host-modules/swarm.nix index e2756ca2..4d8c9d37 100644 --- a/nix/host-modules/swarm.nix +++ b/nix/host-modules/swarm.nix @@ -53,6 +53,23 @@ let pinnedHives = lib.attrNames ( lib.filterAttrs (_: hive: hive.certFingerprint != null) swarmCfg.hives ); + + # Whether this host can derive its own status-publishing coordinates — + # ONE condition for all three of them, deliberately. + # + # 🩸 They were three independent conditions first, and that was wrong in + # a way only an eval gate finds: `enableRequiredServices` turns on + # matrix and authelia but NOT nats (nats has no mode that enables it — + # see the auto-deploy question on the swarm-queue issue), so an ordinary + # all-local hive resolved authelia's two coordinates and not the queue + # URL. Two of three set is exactly what the assertion below rejects, so + # every `enableAllLocalDefaults` hive would have stopped evaluating. + # + # Deriving all three from one predicate makes the partial state + # unrepresentable rather than merely detected: a default set is all or + # nothing, and the assertion is then only ever about what an operator + # typed. + queueLocal = swarmCfg.nats.enable && swarmCfg.authelia.enable && cfg.hiveName != null; in { options.services.hyperhive.swarm.hives = lib.mkOption { @@ -298,6 +315,43 @@ in designed for it. ''; } + { + # Deliberately an assertion and not a silent "then publish + # nothing": a half-set trio is a config an operator believes is + # working, and its runtime failure mode is the expensive one — + # the daemon comes up fine, never connects, and the hive reads + # `never_reported` on a dashboard nobody is watching yet. + # + # Safe to add to an existing deployment: every `statusPublish` + # default is either all-local or all-null, so no config that + # evaluates today can be caught by this. It also encodes a + # property of the code rather than an intended shape — hive-c0re + # genuinely cannot publish with two of three coordinates — which + # is the distinction the `serviceDomains'` comment at the top of + # this file was written about. + assertion = + let + set = lib.filter (v: v != null) [ + swarmCfg.statusPublish.natsUrl + swarmCfg.statusPublish.tokenEndpoint + swarmCfg.statusPublish.clientSecretFile + ]; + in + builtins.length set == 0 || builtins.length set == 3; + message = '' + services.hyperhive.swarm.statusPublish needs natsUrl, + tokenEndpoint and clientSecretFile set together or not at all + — this hive has only some of them. + + Currently: + natsUrl = ${toString swarmCfg.statusPublish.natsUrl} + tokenEndpoint = ${toString swarmCfg.statusPublish.tokenEndpoint} + clientSecretFile = ${toString swarmCfg.statusPublish.clientSecretFile} + + Set the missing ones to publish this hive's status to the + swarm, or set all three to null to turn publishing off. + ''; + } ]; }; @@ -343,4 +397,84 @@ in }; }; + # How this hive reaches the swarm queue to offer its own status + # (hive-c0re's `swarm_status`). Three coordinates, defaulted from the + # local swarm services when this host runs them, and set by hand + # otherwise — one code path for both deployments. + # + # The alternative was the shape swarm-controller uses: emit the + # coordinates only when authelia and NATS are local, and nothing + # otherwise. That is right for the controller, which *is* a swarm-host + # service — but a hive is the one thing in a swarm that routinely is + # not on the swarm host, so the same rule would mean status publishing + # works on exactly the deployment that needs it least. + # + # There is no `enable`: three coordinates that are all set is the + # enable. An extra flag would let a hive be configured-but-off, which + # is one more state to explain and one more way to be silently quiet. + options.services.hyperhive.swarm.statusPublish = { + natsUrl = lib.mkOption { + type = lib.types.nullOr lib.types.str; + default = if queueLocal then "nats://127.0.0.1:${toString swarmCfg.nats.port}" else null; + defaultText = lib.literalExpression ''"nats://127.0.0.1:''${swarm.nats.port}" when this host runs the queue and the IdP, else null''; + example = "nats://10.100.0.1:4222"; + description = '' + Where the swarm queue listens, as seen from *this* hive. + + Defaults to loopback when this host runs the queue container + itself (it shares the host netns, so loopback is correct there + and is not the "localhost means the wrong thing" trap that + applies inside agent containers). A hive that is not the swarm + host has to name the swarm's mesh address. + + Null disables status publishing: this hive computes its own + readiness as always, and simply offers it to nobody. The swarm + controller then reports it `never_reported`, which is the honest + reading. + ''; + }; + + tokenEndpoint = lib.mkOption { + type = lib.types.nullOr lib.types.str; + default = + if queueLocal && swarmCfg.authelia.url != null then + "${swarmCfg.authelia.url}/api/oidc/token" + else + null; + defaultText = lib.literalExpression ''"''${swarm.authelia.url}/api/oidc/token" when this host runs both the queue and the IdP, else null''; + example = "https://auth.example.com/api/oidc/token"; + description = '' + The swarm IdP's OAuth2 token endpoint. This hive mints a + `client_credentials` access token there and presents it when + connecting to the queue, which authenticates it as + `hive-` — the client + {file}`nix/host-modules/swarm-authelia.nix` already declares for + every entry in {option}`services.hyperhive.swarm.hives`. + ''; + }; + + clientSecretFile = lib.mkOption { + type = lib.types.nullOr lib.types.str; + default = + if queueLocal then "${swarmCfg.authelia.hostClientSecretDir}/hive-${cfg.hiveName}.secret" else null; + defaultText = lib.literalExpression ''"''${swarm.authelia.hostClientSecretDir}/hive-''${hiveName}.secret" when this host runs both the queue and the IdP, else null''; + example = "/var/lib/secrets/swarm-queue-client.secret"; + description = '' + Path to a file holding the plaintext client secret for this + hive's `hive-` identity. + + A path and not a value: a secret in the Nix store is world + readable, and one in the environment is readable by anything + that can open {file}`/proc//environ`. + + Defaults to authelia's own minted secret when the IdP runs on + this host. On any other hive the secret has to get here somehow, + and the swarm does not distribute it — copy it out of the swarm + host's + {option}`services.hyperhive.swarm.authelia.hostClientSecretDir` + with whatever secret management this deployment already uses. + ''; + }; + }; + } diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index c5012025..8448176c 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -24,7 +24,11 @@ serde_json.workspace = true # every other participant - a hive publishing its own status runs the same # code with a different client id. Two copies of credential handling is one # token-refresh fix that has to be found twice. -swarm-queue-client.workspace = true +# +# `kv` for the same reason one level in: the status bucket's name and +# creation config are shared with the hive that writes it, so this end does +# not get to declare them privately. +swarm-queue-client = { workspace = true, features = ["kv"] } tokio.workspace = true tracing.workspace = true tracing-subscriber.workspace = true diff --git a/swarm-controller/src/status.rs b/swarm-controller/src/status.rs index b7d322aa..eaff6038 100644 --- a/swarm-controller/src/status.rs +++ b/swarm-controller/src/status.rs @@ -38,13 +38,6 @@ 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 @@ -155,34 +148,25 @@ impl StatusReader { /// 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> { + /// The name and the creation config come from + /// [`swarm_queue_client::status`] rather than from here: the hive that + /// writes this bucket opens it with the same call, and a bucket both + /// ends may create is one both ends have to describe identically. + /// + /// Whichever side arrives first creates it, and both sides ask for the + /// same shape, so this is a race with one outcome. + /// + /// Returns the queue client's own error rather than an `anyhow::Error`: + /// `OnceCell::get_or_try_init` takes its error type from the closure, + /// so widening here would mean converting *inside* the closure for no + /// gain. `view` below `?`s it and anyhow converts there — which is the + /// whole point of the library keeping a typed error while the binary + /// keeps anyhow. + async fn store( + &self, + ) -> std::result::Result<&async_nats::jetstream::kv::Store, swarm_queue_client::Error> { 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")) - } - } - }) + .get_or_try_init(|| swarm_queue_client::status::open_or_create(&self.client)) .await } @@ -191,25 +175,11 @@ impl StatusReader { /// 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:?})"); - } + // Before anything JetStream: an unconnected client does not fail a + // request, it hangs on it. The rule and the measurement behind it live + // in `swarm_queue_client::ensure_connected` — every consumer of the + // queue needs it, so it is not this daemon's to keep. + swarm_queue_client::ensure_connected(&self.client)?; let store = self.store().await?; diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index 32a2d0e5..0c25927d 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -4,11 +4,23 @@ version.workspace = true readme = "README.md" edition.workspace = true +[features] +# OFF by default, and that default is the point: the auth-callout responder +# consumes this crate for the connect alone and speaks neither `jetstream` +# nor `kv`. A consumer that needs the status bucket says so in its own +# Cargo.toml, so the requirement stays visible where it is incurred. +# +# What is behind the flag is deliberately narrow - the *name and shape* of +# one bucket two crates open from opposite ends (`src/status.rs`), not a +# general "KV support" surface. The crate's job still ends at a connected +# client; the exception exists because an agreement between two crates has +# to live in one of them, and neither end of that bucket is senior to the +# other. +kv = ["async-nats/kv"] + [dependencies] -# No `kv`/`jetstream` feature here on purpose: this crate's job ends at a -# connected client. What a consumer does with it - KV for the controller and -# the hive, plain messaging for anything later - is the consumer's business, -# and its Cargo.toml is where that requirement should be visible. +# Bare (no `kv`/`jetstream`) unless a consumer opts into the `kv` feature +# above - the connect itself needs none of them. async-nats.workspace = true reqwest.workspace = true serde.workspace = true diff --git a/swarm-queue-client/README.md b/swarm-queue-client/README.md index 79230794..07a75129 100644 --- a/swarm-queue-client/README.md +++ b/swarm-queue-client/README.md @@ -50,6 +50,23 @@ effect actually did. ## What this crate does not do -It ends at a connected client. No `jetstream`/`kv` feature is enabled here — -what a consumer does with the connection is its own business, and its -`Cargo.toml` is where that requirement should be visible. +It ends at a connected client. `jetstream`/`kv` are **off by default** — what a +consumer does with the connection is its own business, and its `Cargo.toml` is +where that requirement should be visible. The auth-callout responder speaks the +connect and nothing else, and pays for nothing else. + +## The one exception: the `kv` feature + +`kv` adds `status`, which holds the name and the creation config of the +`hive-status` bucket — nothing more. + +It is here because that bucket has **two ends in two crates**: a hive writes its +own key, the controller reads every key. The name being a repeated literal is +the mild half of the problem; the sharp half is that either end may arrive first +on a fresh swarm, so both create the bucket if it is missing. Two `Config`s that +drift means whichever end created it wins and the other opens a bucket it did +not ask for — no error, no log, just a retention policy nobody chose. + +An agreement between two crates has to live in one of them, and neither end of +this bucket is senior to the other. Behind a default-off feature, the consumer +that needs none of it still pays nothing. diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 94a4af0d..d6d0ce62 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -84,6 +84,21 @@ pub enum Error { #[source] source: async_nats::ConnectError, }, + + /// Distinct from [`Error::Connect`] on purpose: that one is a connect + /// that was attempted and refused, this one is a request made before + /// any connection exists. The second is a caller-ordering problem and + /// the first is a deployment one. + #[error("not connected to the swarm queue (client state: {0:?})")] + NotConnected(async_nats::connection::State), + + #[cfg(feature = "kv")] + #[error("creating the {bucket} bucket")] + CreateBucket { + bucket: &'static str, + #[source] + source: async_nats::jetstream::context::CreateKeyValueError, + }, } /// Render an error and its source chain on one line. @@ -112,6 +127,12 @@ pub fn chain(error: &dyn std::error::Error) -> String { rendered } +/// The hive-status KV bucket, shared by the hive that writes it and the +/// controller that reads it. Behind the `kv` feature — see the module doc +/// for why a bucket name and its config belong to neither end alone. +#[cfg(feature = "kv")] +pub mod status; + /// Only the one field this needs; authelia returns several. #[derive(serde::Deserialize)] struct TokenResponse { @@ -218,6 +239,32 @@ async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result Result<(), Error> { + let state = client.connection_state(); + if state != async_nats::connection::State::Connected { + return Err(Error::NotConnected(state)); + } + Ok(()) +} + /// Connect to the swarm queue, minting a token for each connection attempt. pub async fn connect(cfg: QueueConfig) -> Result { // A timeout, because this client runs INSIDE the auth callback: a token diff --git a/swarm-queue-client/src/status.rs b/swarm-queue-client/src/status.rs new file mode 100644 index 00000000..1c8689b2 --- /dev/null +++ b/swarm-queue-client/src/status.rs @@ -0,0 +1,67 @@ +//! The hive-status KV bucket: its name, and the shape it is created with. +//! +//! Two processes touch this bucket from opposite ends — a hive writes its +//! own key, the swarm controller reads every key — and they live in +//! different crates. That is the whole reason this module exists rather +//! than a `const` on each side: **the two ends must agree, and a literal +//! repeated across crates is an agreement nothing checks.** +//! +//! The name is the obvious half. The sharper half is the *config*: both +//! ends open the bucket with [`crate::status::open_or_create`], because either may +//! arrive first on a fresh swarm and neither can assume the other has +//! run. If the two ends passed different `Config`s, whichever created it +//! would win and the other's `get_key_value` would succeed against a +//! bucket it did not ask for — no error, no log, just a retention policy +//! nobody chose. Sharing the constructor makes the race have one outcome +//! instead of two. +//! +//! Feature-gated (`kv`) so the crate's other consumer, the auth-callout +//! responder, still pulls neither `jetstream` nor `kv`: it speaks the +//! connect and nothing else. + +use crate::Error; + +/// The KV bucket hives publish their status snapshots into, one key per +/// hive keyed by `hiveName`. +/// +/// 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"; + +/// Open the status bucket, creating it if nothing has yet. +/// +/// `history: 1` is the shape: every consumer reads *the last thing each +/// hive said*, and retaining more would be storage bought for a query +/// nobody makes. +/// +/// Creating rather than requiring a provisioning step is deliberate — the +/// controller and the hives come up in no particular order, and a bucket +/// that must pre-exist turns "the swarm was deployed in the wrong order" +/// into a permanent, silent absence of data. +pub async fn open_or_create( + client: &async_nats::Client, +) -> Result { + let js = async_nats::jetstream::new(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 + .map_err(|source| Error::CreateBucket { + bucket: BUCKET, + source, + }) + } + } +}