vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

hive-agent: the swarm queue connection is required; retry, then exit

Browse source 
An agent's harness connects to the swarm queue with the agent's own
credential or not at all. The first connect retries after 5s, doubling
(7 attempts, about five minutes), with a warning per failed attempt;
when every attempt fails the harness logs an error and exits 1. Missing
coordinates (HIVE_AGENT_NATS_URL, or the secret store's BAO_ADDR /
HIVE_AGENT_NAME / certificate) fail the boot outright.

Previously the first connect's outcome was memoized for the life of the
process: a store 403 or queue refusal at boot left the agent on the
hive's shared client, or disconnected, until someone restarted it, and
an agent with no queue coordinates ran on without one. Both paths are
gone, with the hive-scoped subjects that only the shared client used.

The harness unit now restarts after 30s instead of 2s, so a swarm-wide
outage doesn't restart every agent every 2s, and has no start limit, so
it keeps coming back however long the outage lasts.

The hive's shared client credential is still imported into the harness
unit for swarm-logs; the two HIVE_AGENT_OIDC_*_FILE variables that
pointed the harness at it are dropped, as is the unset
HIVE_AGENT_OIDC_CA_FILE override.

With the fallback gone, the pieces #4944 added only for it go too:
swarm_input's skip-on-shared-client branch and swarm-queue-client's
connect_with_inbox_prefix. swarm-nats-auth's grant for the hive's shared
agent client is unchanged.

Closes #4909
This commit is contained in:
atlas 2026-10-04 19:05:48 +02:00 • committed by mara
commit 235bc481b3
16 changed files with 392 additions and 964 deletions
Download patch file Download diff file Expand all files Collapse all files

683
hive-agent/src/swarm_queue.rs
View file

@ -1,54 +1,29 @@
//! This agent's swarm-queue credentials, resolved once at boot.
//! This agent's swarm-queue connection, made with the agent's own credential.
//!
//! Three of the four coordinates arrive as environment variables the meta
//! flake renders into the harness unit. The fourth — the OIDC client id —
//! arrives as a *file*, delivered beside the secret as one systemd
//! credential pair (`nix/agent-modules/queue.nix`). That is the whole reason
//! this module exists rather than a bare [`QueueConfig::from_env`] call: the
//! id rides with the secret so a reader never has to spell `hive-<name>-agent`
//! a second time, and `from_env` wants it as a value.
//! The credential is minted per agent at swarm level. This process reads it
//! from the swarm secret store under the agent's own certificate and holds it
//! in memory only ([`AgentSecret`]), and publishes on the agent's own
//! hive-free subjects.
//!
//! Reading the file and assigning the variable would put the same rule back
//! in `from_env`'s hands, but `std::env::set_var` is unsound in a process that
//! has already spawned threads, and this one has. So the rule is restated
//! here over the inputs this consumer actually has, and [`decide`] is the
//! single place it lives.
//!
//! Beside all four sits a fifth coordinate, resolved by [`store_source`] and
//! not part of their group. The four are the *hive's* — one OIDC client
//! shared by every container on it — so at the queue's auth callout they say
//! which hive is connecting and never which agent. The fifth is this agent's
//! own, minted per agent at swarm level. This process reads it from the swarm
//! secret store under the agent's own certificate and holds it in memory only
//! ([`AgentSecret`]).
//!
//! When the store holds one, [`client`] connects with it first, and the agent
//! publishes on its own hive-free subjects. When it holds none, or the queue
//! refuses it, the first connect falls back to the hive's shared client and
//! the hive-scoped subjects. [`Connection::presented`] says which, so a
//! publisher builds the subject that credential is granted.
//! **The connection is required.** [`init`] fails when the queue's address or
//! the store's coordinates are missing, and [`client`] retries the first
//! connect on [`CONNECT_BACKOFF`] and then exits the process, so the harness
//! unit fails and systemd restarts it.
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;
use anyhow::{Context as _, anyhow};
use tokio::sync::OnceCell;
use swarm_queue_client::{ClientSecret, QueueConfig};
use swarm_secret_client::{
SecretStore,
client::{DEFAULT_CERT_MOUNT, ENV_ADDR, ENV_CACERT, Settings},
policy, queue,
};
/// Variable prefix for this agent's coordinates. Distinct from `HIVE_C0RE`'s
/// on purpose: an agent authenticates as its own client, not as its hive.
const ENV_PREFIX: &str = "HIVE_AGENT";
/// Resolved once at boot, so the publisher that connects to the queue reads
/// one answer rather than re-deriving it per call.
static CONFIG: OnceLock<Option<QueueConfig>> = OnceLock::new();
/// Where the swarm queue listens, as this container reaches it.
const ENV_NATS_URL: &str = "HIVE_AGENT_NATS_URL";
/// Names the agent whose queue secret this process reads. The cert-auth role
/// and the store path are both built from it by `swarm_secret_client`.
@ -59,19 +34,29 @@ const ENV_AGENT_NAME: &str = "HIVE_AGENT_NAME";
/// every reconnect behind it.
const STORE_READ_TIMEOUT: Duration = Duration::from_secs(10);
/// Where to present this agent's own credential, resolved once at boot.
static AGENT: OnceLock<Option<AgentPath>> = OnceLock::new();
/// The waits between first-connect attempts: 5s doubling, about five minutes
/// in all. After the last attempt fails the process exits.
const CONNECT_BACKOFF: [Duration; 6] = [
Duration::from_secs(5),
Duration::from_secs(10),
Duration::from_secs(20),
Duration::from_secs(40),
Duration::from_secs(80),
Duration::from_secs(160),
];
/// Where to connect, resolved once at boot by [`init`].
static AGENT: OnceLock<AgentPath> = OnceLock::new();
/// The one connection every publisher in this process shares. Separate from
/// [`CONFIG`] because resolving the coordinates is synchronous boot work and
/// [`AGENT`] because resolving the coordinates is synchronous boot work and
/// connecting is not — see [`client`].
static CLIENT: OnceCell<Option<Connection>> = OnceCell::const_new();
static CLIENT: OnceCell<async_nats::Client> = OnceCell::const_new();
/// What connecting with this agent's own credential needs.
#[derive(Debug, PartialEq, Eq)]
struct AgentPath {
url: String,
ca_file: Option<PathBuf>,
store: Arc<StoreSource>,
}
@ -150,8 +135,8 @@ impl<S: SecretSource> AgentSecret<S> {
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
/// Read the secret before the first connection attempt. An `Err` here is
/// what sends [`connect_once`] to the hive's shared client.
/// Read the secret before the first connection attempt. An `Err` here
/// fails that attempt.
async fn prime(&self) -> anyhow::Result<()> {
let value = self
.read()
@ -216,109 +201,31 @@ impl<S: SecretSource> AgentSecret<S> {
}
}
/// Which credential the shared connection presented.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Presented {
/// This agent's own, granted `<prefix>.<agent>`.
Agent,
/// The hive's shared client, granted `<prefix>.<hive>.>` for the hive the
/// client id names.
Hive { client_id: String },
}
/// The shared queue connection and the credential it was made with.
#[derive(Clone)]
pub struct Connection {
pub client: async_nats::Client,
pub presented: Presented,
}
/// The four variables the harness unit sets, before the client-id file is
/// read. Collected into a struct so [`decide`] is pure over them and the
/// process env is touched in exactly one place.
struct QueueEnv {
nats_url: Option<String>,
token_endpoint: Option<String>,
client_id_file: Option<String>,
client_secret_file: Option<String>,
/// Independent of the all-or-none group below, exactly as in
/// `QueueConfig::from_env`: a swarm behind a publicly-trusted
/// certificate needs no extra anchor. Nothing in this tree sets it for
/// an agent — a container already trusts the swarm root, which
/// `hive_c0re::meta` embeds at build time — so it is here for a
/// deployment that needs a different one, not for ours.
ca_file: Option<String>,
}
impl QueueEnv {
fn from_env() -> Self {
let var = |suffix: &str| std::env::var(format!("{ENV_PREFIX}_{suffix}")).ok();
Self {
nats_url: var("NATS_URL"),
token_endpoint: var("OIDC_TOKEN_ENDPOINT"),
client_id_file: var("OIDC_CLIENT_ID_FILE"),
client_secret_file: var("OIDC_CLIENT_SECRET_FILE"),
ca_file: var("OIDC_CA_FILE"),
}
}
}
/// What the environment plus the client-id file add up to.
enum Resolution {
/// Everything is here; the agent can reach the queue.
Configured(Box<QueueConfig>),
/// No queue coordinates for this agent, and that is a legal state — carries why.
Absent(&'static str),
/// Some of the environment, not all of it. A deployment bug rather than
/// an absent integration, so it is reported and then survived.
Partial,
}
/// Read the client id out of the file the credential landed at.
///
/// `None` for a missing or empty file, which is the ordinary state of a hive
/// whose secret store has nothing published yet — the credential simply is
/// not there, and nspawn forwards nothing. Trailing newline stripped: the
/// reader unit writes one and an id with a newline in it authenticates as
/// nobody.
fn read_client_id(path: &Path) -> Option<String> {
match std::fs::read_to_string(path) {
Ok(raw) => {
let id = raw.trim();
(!id.is_empty()).then(|| id.to_owned())
}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
Err(e) => {
tracing::warn!(path = %path.display(), error = %e, "reading the queue client id failed");
None
}
}
}
/// Where this agent reads its own queue secret from, or `None` when this
/// container was given no store.
/// Where this agent reads its own queue secret from.
///
/// Taken through a lookup so it can be asserted without mutating the process
/// environment. The shape follows `hive_matrix_mcp::credential`'s
/// `store_settings`, which reads the same variables for the same identity.
///
/// # Errors
/// A store address without the agent name or the identity beside it, which
/// is a half-delivered container rather than one without a store.
fn store_source(get: impl Fn(&str) -> Option<String>) -> anyhow::Result<Option<StoreSource>> {
/// No store address, or one without the agent name or the identity beside it.
fn store_source(get: impl Fn(&str) -> Option<String>) -> anyhow::Result<StoreSource> {
if get(ENV_ADDR).is_none_or(|v| v.is_empty()) {
return Ok(None);
anyhow::bail!(
"{ENV_ADDR} is unset or empty, so there is no secret store to read this agent's \
queue credential from"
);
}
let agent = get(ENV_AGENT_NAME)
.filter(|v| !v.is_empty())
.with_context(|| format!("{ENV_AGENT_NAME} is unset or empty"))?;
let settings = Settings::from_lookup(|k| get(k).filter(|v| k != ENV_CACERT || ca_is_usable(v)))
.context("reading the swarm secret store's coordinates from the environment")?;
Ok(Some(StoreSource {
Ok(StoreSource {
settings,
role: policy::agent_object_name(&agent)?,
path: queue::agent_queue_path(&agent)?,
}))
})
}
/// Whether the CA bundle at `path` is a file with bytes in it. `BAO_CACERT`
@ -328,196 +235,105 @@ fn ca_is_usable(path: &str) -> bool {
std::fs::metadata(path).is_ok_and(|m| m.len() > 0)
}
/// Decide whether this agent can connect with its own credential: it needs the
/// queue's address and a store to read its secret from, and nothing of the
/// hive's client.
fn decide_agent_path(env: &QueueEnv, store: Option<StoreSource>) -> Option<AgentPath> {
Some(AgentPath {
url: env.nats_url.clone()?,
ca_file: env.ca_file.as_ref().map(Into::into),
store: Arc::new(store?),
/// What connecting needs: the queue's address and a store to read this agent's
/// secret from.
///
/// # Errors
/// Either one missing.
fn agent_path(get: impl Fn(&str) -> Option<String>) -> anyhow::Result<AgentPath> {
let url = get(ENV_NATS_URL)
.filter(|v| !v.is_empty())
.with_context(|| format!("{ENV_NATS_URL} is unset or empty"))?;
Ok(AgentPath {
url,
store: Arc::new(store_source(&get)?),
})
}
/// Decide what this agent's queue configuration is, given the environment and
/// whatever the client-id file held.
/// Resolve this agent's queue coordinates and record them for [`client`].
///
/// All four variables or none, for the reason `QueueConfig::from_env` gives:
/// a half-set environment produces a process that comes up fine and never
/// connects. The client id is graded separately from the four because its
/// file has a legal absence the variables do not — the unit names the path a
/// credential *would* arrive at whether or not one has been published yet.
fn decide(env: &QueueEnv, client_id: Option<String>) -> Resolution {
match (
env.nats_url.as_ref(),
env.token_endpoint.as_ref(),
env.client_id_file.as_ref(),
env.client_secret_file.as_ref(),
) {
(None, None, None, None) => Resolution::Absent("this hive has no swarm queue configured"),
(Some(url), Some(token_endpoint), Some(_), Some(secret)) => match client_id {
Some(client_id) => Resolution::Configured(Box::new(QueueConfig {
url: url.clone(),
token_endpoint: token_endpoint.clone(),
client_id,
client_secret: ClientSecret::File(secret.into()),
ca_file: env.ca_file.as_ref().map(Into::into),
})),
None => {
Resolution::Absent("the queue credential has not been published to this hive yet")
}
},
_ => Resolution::Partial,
}
}
/// Resolve this agent's queue configuration and record it for later use.
///
/// Never fails: a harness that cannot reach the queue still serves its
/// operator, its web UI and its turn loop, so every outcome here is a log
/// line and not an exit.
pub fn init() {
let env = QueueEnv::from_env();
let client_id = env
.client_id_file
.as_deref()
.map(Path::new)
.and_then(read_client_id);
let resolved = match decide(&env, client_id) {
Resolution::Configured(cfg) => {
// url + client id only. The secret is a path in this struct and
// stays one: neither it nor its contents belong in a log.
tracing::info!(url = %cfg.url, client_id = %cfg.client_id, "swarm queue configured");
Some(*cfg)
}
Resolution::Absent(why) => {
tracing::info!(why, "no swarm queue coordinates for this agent");
None
}
Resolution::Partial => {
tracing::error!(
prefix = ENV_PREFIX,
"swarm queue half-configured: {ENV_PREFIX}_NATS_URL, \
{ENV_PREFIX}_OIDC_TOKEN_ENDPOINT, {ENV_PREFIX}_OIDC_CLIENT_ID_FILE and \
{ENV_PREFIX}_OIDC_CLIENT_SECRET_FILE are set together or not at all — \
this agent will not connect"
);
None
}
};
let _ = CONFIG.set(resolved);
// Independent of everything above: this agent may hold its own secret on
// a hive whose shared client is not published yet, or the shared client
// and no secret of its own.
let store = store_source(|k| std::env::var(k).ok()).unwrap_or_else(|e| {
tracing::warn!(
error = format!("{e:#}"),
"this agent's secret store coordinates are incomplete, so it cannot read \
its own swarm queue credential"
);
None
});
if let Some(store) = &store {
tracing::info!(
path = %store.path,
"this agent reads its own swarm queue credential from the secret store"
);
} else {
tracing::info!(
"no secret store for this agent; it is known to the queue by its hive's \
shared client"
);
}
let _ = AGENT.set(decide_agent_path(&env, store));
}
/// Whether this agent has any credential to reach the queue with. `false`
/// before [`init`] has run.
pub fn configured() -> bool {
config().is_some() || AGENT.get().is_some_and(Option::is_some)
}
/// What [`init`] resolved, or `None` when this agent has no queue.
///
/// Borrowed from the `OnceLock` rather than cloned: a caller wants the client
/// id to derive its subject from, and handing out an owned copy of a struct
/// holding a credential path invites it being stored somewhere with a
/// different lifetime than the one place that owns it.
///
/// `None` before [`init`] has run, which is the same answer as "no queue" and
/// deliberately not a panic — the ordering is a boot detail, and a harness
/// that reordered its boot should lose the queue, not die.
pub fn config() -> Option<&'static QueueConfig> {
CONFIG.get()?.as_ref()
/// # Errors
/// A coordinate is missing. The harness exits on it: an agent without a queue
/// connection is invisible to the operator at swarm level.
pub fn init() -> anyhow::Result<()> {
let agent = agent_path(|k| std::env::var(k).ok())
.context("resolving this agent's swarm queue coordinates")?;
tracing::info!(
url = %agent.url,
path = %agent.store.path,
"this agent connects to the swarm queue with its own credential from the secret store"
);
let _ = AGENT.set(agent);
Ok(())
}
/// The shared queue connection, made on first call and memoized for the rest
/// of the process.
///
/// One connection per process, not per publisher: the agent authenticates as
/// one client, so a second `connect` would be a second token mint and a second
/// live connection for the same identity rather than a second credential.
/// one client, so a second `connect` would be a second live connection for the
/// same identity rather than a second credential.
///
/// `None` covers both "no queue coordinates" and "configured but the connect
/// failed" — a caller does nothing differently between them, since either way
/// there is nothing to publish onto. Connecting is lazy so that an agent on a
/// hive with no queue pays nothing at boot.
pub async fn client() -> Option<Connection> {
CLIENT.get_or_init(connect_once).await.clone()
/// The first call retries the connect on [`CONNECT_BACKOFF`], and every other
/// caller waits on it. When no attempt succeeds the process exits with status
/// 1, so this never returns without a client.
pub async fn client() -> async_nats::Client {
CLIENT.get_or_init(connect_or_exit).await.clone()
}
/// This agent's own credential first, then the hive's. Each path taken is
/// logged once, here.
async fn connect_once() -> Option<Connection> {
if let Some(agent) = AGENT.get().and_then(Option::as_ref) {
let secret = AgentSecret::new(Arc::clone(&agent.store));
let name = crate::identity::label();
match connect_as_agent(&agent.url, agent.ca_file.as_deref(), name, secret).await {
Ok(client) => {
tracing::info!(
url = %agent.url,
"connected to the swarm queue with this agent's own credential"
);
return Some(Connection {
client,
presented: Presented::Agent,
});
}
Err(e) => tracing::warn!(
error = format!("{e:#}"),
"connecting with this agent's own credential failed; falling back to \
its hive's shared client"
),
}
}
let cfg = config()?;
let inbox = swarm_queue_client::client_inbox_prefix(&cfg.client_id);
match swarm_queue_client::connect_with_inbox_prefix(cfg.clone(), inbox).await {
Ok(client) => {
tracing::info!(
url = %cfg.url,
client_id = %cfg.client_id,
"connecting to the swarm queue with the hive's shared client; the \
client retries in the background until the queue accepts it"
);
Some(Connection {
client,
presented: Presented::Hive {
client_id: cfg.client_id.clone(),
},
})
}
Err(e) => {
// `chain`, not `{:#}`: this is `swarm_queue_client::Error`, whose
// `Display` ignores the alternate flag, so `{:#}` renders the
// headline and drops the cause that says which half failed.
tracing::warn!(
error = %swarm_queue_client::chain(&e),
"swarm queue connect failed; this agent publishes nothing upward"
);
None
}
async fn connect_or_exit() -> async_nats::Client {
let connected = match AGENT.get() {
Some(agent) => retry_first_connect(|| connect_once(agent), tokio::time::sleep).await,
None => Err(anyhow!("swarm_queue::init has not run")),
};
connected.unwrap_or_else(|e| {
tracing::error!(
error = format!("{e:#}"),
"no swarm queue connection; exiting, since this agent cannot run without one"
);
std::process::exit(1)
})
}
/// One connection attempt with this agent's own credential.
async fn connect_once(agent: &AgentPath) -> anyhow::Result<async_nats::Client> {
let secret = AgentSecret::new(Arc::clone(&agent.store));
let name = crate::identity::label();
let client = connect_as_agent(&agent.url, name, secret).await?;
tracing::info!(
url = %agent.url,
"connected to the swarm queue with this agent's own credential"
);
Ok(client)
}
/// Run `connect` until it succeeds, sleeping each wait in [`CONNECT_BACKOFF`]
/// between attempts, and return the last error once every attempt has failed.
async fn retry_first_connect<T, C, CFut, S, SFut>(mut connect: C, mut sleep: S) -> anyhow::Result<T>
where
C: FnMut() -> CFut,
CFut: Future<Output = anyhow::Result<T>>,
S: FnMut(Duration) -> SFut,
SFut: Future<Output = ()>,
{
let mut waits = CONNECT_BACKOFF.iter();
let mut attempt = 1;
loop {
let e = match connect().await {
Ok(client) => return Ok(client),
Err(e) => e,
};
let Some(&wait) = waits.next() else {
return Err(e.context(format!("{attempt} connection attempts failed")));
};
tracing::warn!(
error = format!("{e:#}"),
attempt,
retry_in = ?wait,
"connecting to the swarm queue failed"
);
sleep(wait).await;
attempt += 1;
}
}
@ -525,18 +341,19 @@ async fn connect_once() -> Option<Connection> {
///
/// The secret is read before the first attempt, and any failure of that read
/// or of the first attempt, a refusal included, is returned for
/// [`connect_once`] to fall back on. Once connected, each reconnect attempt
/// [`retry_first_connect`] to retry. Once connected, each reconnect attempt
/// takes its secret from [`AgentSecret::for_attempt`].
async fn connect_as_agent<S: SecretSource>(
url: &str,
ca_file: Option<&Path>,
name: String,
secret: AgentSecret<S>,
) -> anyhow::Result<async_nats::Client> {
secret.prime().await?;
let secret = Arc::new(secret);
let agent = name.clone();
swarm_queue_client::connect_with_token(url, ca_file, &agent, move || {
// No extra trust anchor: the container trusts the swarm root, which
// `hive_c0re::meta` embeds at build time.
swarm_queue_client::connect_with_token(url, None, &agent, move || {
let secret = Arc::clone(&secret);
let name = name.clone();
// Spawned so the future handed back is only a `JoinHandle`, which is
@ -551,6 +368,7 @@ async fn connect_as_agent<S: SecretSource>(
.await
.map_err(|e| anyhow!(swarm_queue_client::chain(&e)))
}
#[cfg(test)]
mod tests {
use std::collections::VecDeque;
@ -561,127 +379,25 @@ mod tests {
use anyhow::anyhow;
use super::{
AgentPath, AgentSecret, ClientSecret, QueueEnv, Resolution, SecretSource, StoreSource,
connect_as_agent, decide, decide_agent_path, read_client_id, store_source,
AgentPath, AgentSecret, CONNECT_BACKOFF, SecretSource, StoreSource, agent_path,
connect_as_agent, retry_first_connect, store_source,
};
fn env(parts: [Option<&str>; 4]) -> QueueEnv {
let [nats_url, token_endpoint, client_id_file, client_secret_file] = parts;
QueueEnv {
nats_url: nats_url.map(str::to_owned),
token_endpoint: token_endpoint.map(str::to_owned),
client_id_file: client_id_file.map(str::to_owned),
client_secret_file: client_secret_file.map(str::to_owned),
ca_file: None,
}
}
fn full() -> QueueEnv {
env([
Some("nats://10.42.0.1:4222"),
Some("https://auth.t.local/api/oidc/token"),
Some("/run/credentials/hive-agent.service/hive-queue-agent-client-id"),
Some("/run/credentials/hive-agent.service/hive-queue-agent-secret"),
])
}
/// The reader unit writes the id with a trailing newline; an id carrying
/// one is a client authelia has never heard of.
#[test]
fn a_client_id_file_is_read_without_its_newline() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("client_id");
std::fs::write(&path, "hive-h1-agent\n").expect("write");
assert_eq!(read_client_id(&path).as_deref(), Some("hive-h1-agent"));
}
/// Both shapes of "no credential here": never delivered, or delivered
/// empty. Neither is an id, and treating an empty string as one would
/// authenticate as the anonymous client rather than failing.
#[test]
fn a_missing_or_empty_client_id_file_reads_as_no_id() {
let dir = tempfile::tempdir().expect("tempdir");
let missing = dir.path().join("client_id");
assert_eq!(read_client_id(&missing), None);
std::fs::write(&missing, "\n").expect("write");
assert_eq!(read_client_id(&missing), None);
}
#[test]
fn a_complete_environment_with_a_published_id_configures_the_queue() {
let e = full();
let Resolution::Configured(cfg) = decide(&e, Some("hive-h1-agent".to_owned())) else {
panic!("expected a configured queue");
};
assert_eq!(cfg.url, "nats://10.42.0.1:4222");
assert_eq!(cfg.client_id, "hive-h1-agent");
let ClientSecret::File(path) = &cfg.client_secret else {
panic!("the secret stays a path: {:?}", cfg.client_secret);
};
assert!(
path.ends_with("hive-queue-agent-secret"),
"the secret stays a path: {}",
path.display()
);
}
/// A hive that has not been given its queue's coordinates at all. Silence
/// here is correct, and it has to be distinguishable from the half-set
/// case below — that distinction is the only thing that makes the error
/// branch worth logging.
#[test]
fn an_empty_environment_is_no_queue_rather_than_an_error() {
let e = env([None, None, None, None]);
assert!(matches!(decide(&e, None), Resolution::Absent(_)));
}
/// The credential's own absence. The unit names the path unconditionally
/// once the hive has a queue, so this is the state of every agent on a
/// swarm whose publisher has not run — legal, and not the error branch.
#[test]
fn a_complete_environment_with_no_published_id_is_no_queue() {
let e = full();
assert!(matches!(decide(&e, None), Resolution::Absent(_)));
}
/// Each single-variable omission, because the failure a partial set
/// produces is a harness that looks healthy and publishes nothing.
#[test]
fn any_missing_variable_is_a_partial_configuration() {
for drop in 0..4 {
let mut parts = [
Some("nats://10.42.0.1:4222"),
Some("https://auth.t.local/api/oidc/token"),
Some("/run/credentials/hive-agent.service/hive-queue-agent-client-id"),
Some("/run/credentials/hive-agent.service/hive-queue-agent-secret"),
];
parts[drop] = None;
let e = env(parts);
assert!(
matches!(
decide(&e, Some("hive-h1-agent".to_owned())),
Resolution::Partial
),
"dropping variable {drop} must report a partial configuration"
);
}
}
/// A lookup standing in for a container that was handed a store.
/// A lookup standing in for a container that was handed a store and the
/// queue's address.
fn with_store(k: &str) -> Option<String> {
match k {
"BAO_ADDR" => Some("https://bao.t.local:8200".to_owned()),
"BAO_CLIENT_CERT" => Some("/run/credentials/x/cert".to_owned()),
"BAO_CLIENT_KEY" => Some("/run/credentials/x/key".to_owned()),
"HIVE_AGENT_NAME" => Some("a1".to_owned()),
"HIVE_AGENT_NATS_URL" => Some("nats://10.42.0.1:4222".to_owned()),
_ => None,
}
}
fn source() -> StoreSource {
store_source(with_store)
.expect("a complete store environment")
.expect("a store address was given")
store_source(with_store).expect("a complete store environment")
}
#[test]
@ -691,20 +407,19 @@ mod tests {
assert_eq!(s.role, "hive-agent-a1");
}
/// No address: this container was given no store, so there is nothing to
/// read from and nothing wrong.
#[test]
fn no_store_address_is_no_store() {
let none = store_source(|k| if k == "BAO_ADDR" { None } else { with_store(k) });
assert!(none.expect("absent is not an error").is_none());
let empty = store_source(|k| {
if k == "BAO_ADDR" {
Some(String::new())
} else {
with_store(k)
}
});
assert!(empty.expect("empty is absent").is_none());
fn a_missing_store_address_is_an_error() {
for addr in [None, Some(String::new())] {
let e = store_source(|k| {
if k == "BAO_ADDR" {
addr.clone()
} else {
with_store(k)
}
})
.expect_err("no store to read the credential from");
assert!(format!("{e:#}").contains("BAO_ADDR"), "{e:#}");
}
}
#[test]
@ -727,37 +442,94 @@ mod tests {
with_store(k)
}
})
.expect("complete")
.expect("given a store");
.expect("complete");
assert_eq!(with_missing_ca, source());
}
/// The per-agent secret is resolved by a separate switch from the hive's
/// client: an agent whose hive has no queue client can still read its own
/// swarm-minted secret, and needs only the queue's address for it.
#[test]
fn an_agent_with_a_store_and_the_queue_address_connects_as_itself() {
let url_only = env([Some("nats://10.42.0.1:4222"), None, None, None]);
assert!(!matches!(
decide(&url_only, None),
Resolution::Configured(_)
));
fn a_store_and_the_queue_address_are_all_an_agent_connects_with() {
assert_eq!(
decide_agent_path(&url_only, Some(source())),
Some(AgentPath {
agent_path(with_store).expect("complete"),
AgentPath {
url: "nats://10.42.0.1:4222".to_owned(),
ca_file: None,
store: Arc::new(source()),
})
}
);
}
#[test]
fn without_a_store_or_the_queue_address_an_agent_does_not_connect_as_itself() {
assert_eq!(decide_agent_path(&full(), None), None);
fn without_a_store_or_the_queue_address_an_agent_has_nothing_to_connect_with() {
for var in ["HIVE_AGENT_NATS_URL", "BAO_ADDR"] {
let e = agent_path(|k| if k == var { None } else { with_store(k) })
.expect_err("a coordinate is missing");
assert!(format!("{e:#}").contains(var), "dropping {var}: {e:#}");
}
}
/// Runs `retry_first_connect` against a connect that fails `failures`
/// times and then succeeds, recording each wait instead of sleeping it.
async fn retry_with(failures: usize) -> (anyhow::Result<usize>, usize, Vec<Duration>) {
let mut calls = 0;
let mut waits = Vec::new();
let result = retry_first_connect(
|| {
calls += 1;
let n = calls;
async move {
if n > failures {
Ok(n)
} else {
Err(anyhow!("attempt {n} refused"))
}
}
},
|wait| {
waits.push(wait);
async {}
},
)
.await;
(result, calls, waits)
}
#[tokio::test]
async fn a_first_connect_that_succeeds_waits_for_nothing() {
let (result, calls, waits) = retry_with(0).await;
assert_eq!(result.expect("connected"), 1);
assert_eq!(calls, 1);
assert!(waits.is_empty(), "{waits:?}");
}
/// 5s doubling, one wait per failed attempt, and the attempt after the
/// last wait is the one that connects.
#[tokio::test]
async fn a_failed_connect_is_retried_on_a_doubling_backoff() {
let (result, calls, waits) = retry_with(3).await;
assert_eq!(result.expect("connected on the fourth attempt"), 4);
assert_eq!(calls, 4);
assert_eq!(
decide_agent_path(&env([None, None, None, None]), Some(source())),
None
waits,
[5, 10, 20].map(Duration::from_secs),
"each wait doubles the one before"
);
}
/// About five minutes of waits, then the last attempt's error rather than
/// another wait.
#[tokio::test]
async fn the_retry_gives_up_with_the_last_error_after_about_five_minutes() {
let (result, calls, waits) = retry_with(usize::MAX).await;
let e = result.expect_err("every attempt failed");
assert_eq!(calls, CONNECT_BACKOFF.len() + 1);
assert_eq!(waits, CONNECT_BACKOFF);
let total: Duration = waits.iter().sum();
assert!(
(Duration::from_mins(4)..=Duration::from_mins(6)).contains(&total),
"{total:?}"
);
assert!(
format!("{e:#}").contains(&format!("attempt {calls} refused")),
"{e:#}"
);
}
@ -875,18 +647,17 @@ mod tests {
.expect_err("the dropped value does not come back");
}
/// (c) Nothing to connect with before the first attempt is an error, which
/// is what sends `connect_once` to the hive's shared client. The queue is
/// never dialled: nothing listens at the address given.
/// (c) Nothing to connect with before the first attempt fails that attempt.
/// The queue is never dialled: nothing listens at the address given.
#[tokio::test]
async fn no_secret_before_the_first_connect_falls_back() {
async fn no_secret_before_the_first_connect_fails_the_attempt() {
for (answer, says) in [
(Answer::Fails, "unreachable"),
(Answer::Absent, "nothing is stored"),
(Answer::Value(" \n"), "nothing is stored"),
] {
let secret = AgentSecret::new(scripted([answer]));
let e = connect_as_agent("nats://127.0.0.1:9", None, "a1".to_owned(), secret)
let e = connect_as_agent("nats://127.0.0.1:9", "a1".to_owned(), secret)
.await
.expect_err("no secret to present");
assert!(format!("{e:#}").contains(says), "{e:#}");
@ -957,7 +728,7 @@ mod tests {
]);
let secret = AgentSecret::new(Arc::clone(&store));
let url = format!("nats://127.0.0.1:{port}");
let _client = connect_as_agent(&url, None, "a1".to_owned(), secret)
let _client = connect_as_agent(&url, "a1".to_owned(), secret)
.await
.expect("the first connect is accepted");