vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Compare commits

base: vinzenz:b8e19a31b287714f3f8cea1830a05fe6e5dd80b0
Branches Tags
vinzenz:main
..
compare: vinzenz:6712cdb796e2948d9bdc0ff65f3406c34e92cbc3
Branches Tags
vinzenz:main
16 changed files with 90 additions and 639 deletions
Download patch file Download diff file Expand all files Collapse all files

2
Cargo.lock generated
View file

@ -1664,7 +1664,6 @@ name = "hive-c0re"
version = "0.1.0"
dependencies = [
"anyhow",
"async-nats",
"axum",
"base64",
"bcrypt",
@ -1696,7 +1695,6 @@ dependencies = [
"serde",
"serde_json",
"sha2 0.11.0",
"swarm-queue-client",
"tempfile",
"tokio",
"tokio-stream",

39
docs/swarm/README.md
View file

@ -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.
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.
⚠️ **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`.
| freshness | what to do about it |
|---|---|
@ -311,37 +311,8 @@ 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. 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-<hiveName>.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-<hiveName>`
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.
between reports. It takes effect on the next request; nothing has to
re-publish.
The endpoint answers **503** when this host has no swarm queue
configured, or has one and cannot read it — deliberately not an empty

7
hive-c0re/Cargo.toml
View file

@ -9,9 +9,6 @@ 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
@ -55,10 +52,6 @@ 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

37
hive-c0re/src/dashboard/health.rs
View file

@ -26,6 +26,8 @@ use axum::{
use serde::Serialize;
use utoipa::ToSchema;
use crate::host_stats::ServerWarning;
#[derive(Serialize, ToSchema)]
struct LiveBody {
status: &'static str,
@ -42,37 +44,40 @@ 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<ServerWarning>,
}
/// Readiness.
///
/// `200` with `{"status":"ok", "warnings": [...]}` unless a `crit`-level
/// 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.
/// 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.
#[utoipa::path(
get,
path = "/health/ready",
responses(
(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),
(status = 200, description = "no crit-level warning set", body = ReadyBody),
(status = 503, description = "at least one crit-level warning set", body = ReadyBody),
),
tag = "health"
)]
pub(super) async fn get_health_ready() -> Response {
let body = crate::warnings::readiness();
let code = if body.is_degraded() {
let warnings = crate::warnings::snapshot();
let degraded = warnings.iter().any(|w| w.level == "crit");
let code = if degraded {
StatusCode::SERVICE_UNAVAILABLE
} else {
StatusCode::OK
};
let body = ReadyBody {
status: if degraded { "degraded" } else { "ok" },
warnings,
};
(code, axum::Json(body)).into_response()
}

6
hive-c0re/src/main.rs
View file

@ -35,7 +35,6 @@ mod snapshot_push;
mod socket_server;
mod stats;
mod stores;
mod swarm_status;
mod webhook_secret;
mod workers;
@ -477,11 +476,6 @@ 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

52
hive-c0re/src/stats/warnings.rs
View file

@ -30,9 +30,6 @@ 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
@ -92,55 +89,6 @@ 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<ServerWarning>,
}
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]

177
hive-c0re/src/swarm_status.rs
View file

@ -1,177 +0,0 @@
//! 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<bool>) {
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 <url>"
// 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(())
}

17
nix/host-modules/hive-c0re/default.nix
View file

@ -251,20 +251,9 @@ 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}"
# 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}";
LoadCredential = lib.optional (
config.services.hyperhive.otel.enable && config.services.hyperhive.otel.headersCredential != null
) "otel-headers:${config.services.hyperhive.otel.headersCredential}";
# 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.

19
nix/host-modules/hive-c0re/environment.nix
View file

@ -209,22 +209,3 @@ 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";
}

134
nix/host-modules/swarm.nix
View file

@ -53,23 +53,6 @@ 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 {
@ -315,43 +298,6 @@ 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.
'';
}
];
};
@ -397,84 +343,4 @@ 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-<hiveName>` — 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-<hiveName>` 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/<pid>/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.
'';
};
};
}

6
swarm-controller/Cargo.toml
View file

@ -24,11 +24,7 @@ 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.
#
# `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"] }
swarm-queue-client.workspace = true
tokio.workspace = true
tracing.workspace = true
tracing-subscriber.workspace = true

76
swarm-controller/src/status.rs
View file

@ -38,6 +38,13 @@ 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
@ -148,25 +155,34 @@ impl StatusReader {
/// The bucket handle, created on first use if nothing has made it yet.
///
/// 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> {
/// 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(|| swarm_queue_client::status::open_or_create(&self.client))
.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
}
@ -175,11 +191,25 @@ 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<Vec<HiveStatus>> {
// 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)?;
// 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?;

20
swarm-queue-client/Cargo.toml
View file

@ -4,23 +4,11 @@ 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]
# Bare (no `kv`/`jetstream`) unless a consumer opts into the `kv` feature
# above - the connect itself needs none of them.
# 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.
async-nats.workspace = true
reqwest.workspace = true
serde.workspace = true

23
swarm-queue-client/README.md
View file

@ -50,23 +50,6 @@ effect actually did.
## What this crate does not do
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.
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.

47
swarm-queue-client/src/lib.rs
View file

@ -84,21 +84,6 @@ 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.
@ -127,12 +112,6 @@ 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 {
@ -239,32 +218,6 @@ async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result<String,
Ok(parsed.access_token)
}
/// Fail fast unless the client is actually connected.
///
/// **Call this before every `JetStream` request.** `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*, for longer than any caller
/// should (measured: still going at 15s against a queue that simply refuses
/// the credential). On a request path that hangs a poll; on a periodic task it
/// hangs the task, including the shutdown branch it never reaches.
///
/// 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 that matters — a process 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.
pub fn ensure_connected(client: &async_nats::Client) -> 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<async_nats::Client, Error> {
// A timeout, because this client runs INSIDE the auth callback: a token

67
swarm-queue-client/src/status.rs
View file

@ -1,67 +0,0 @@
//! 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<async_nats::jetstream::kv::Store, Error> {
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,
})
}
}
}