f3c0a87c0f
HIVE_AGENT_NATS_URL now reaches each container in its hive facts, from the host's deploy.hive-controller.queue.agentNatsUrl, instead of being rendered into the meta flake. The agent option services.hyperhive.agent.queue.natsUrl is removed; the harness unit imports the queue credential whenever the token endpoint is set. On deploy every agent rebuilds once (the rendered flake loses the address), and the harness reads the address from the facts file hive-priv writes before each start. Refs #4857
1774 lines
71 KiB
Rust
1774 lines
71 KiB
Rust
//! Single hive-c0re-owned flake at `/var/lib/hyperhive/meta/` that
|
|
//! exports one `nixosConfiguration` per agent and drives the system-wide
|
|
//! deploy audit trail. Flow (`sync_agents`, `lock_update_for_rebuild`,
|
|
//! `lock_update_hyperhive`):
|
|
//! `docs/agent-lifecycle/config-changes.md::Meta flake`.
|
|
|
|
use std::path::Path;
|
|
use std::sync::{Mutex as StdMutex, PoisonError};
|
|
use std::time::Duration;
|
|
|
|
use anyhow::{Context, Result, bail};
|
|
use tokio::process::Command;
|
|
use tokio::sync::{Mutex, MutexGuard};
|
|
use tokio::time::Instant;
|
|
|
|
pub use hive_meta_flake::{AgentPin, AgentSpec};
|
|
#[cfg(test)]
|
|
use hive_meta_flake::{FORWARDED_VAR_OPTIONS, push_forwarded_var_options};
|
|
use hive_meta_flake::{FORWARDED_VARS, canonical_inputs_in_lock, pinned_git_url};
|
|
|
|
use crate::coordinator::HiveEnv;
|
|
use crate::lifecycle;
|
|
|
|
const GIT_NAME: &str = "c0re";
|
|
const GIT_EMAIL: &str = "c0re@hyperhive.local";
|
|
|
|
/// Single-writer lock around every meta-repo operation. Git isn't
|
|
/// safe to drive from concurrent processes against the same `.git/`
|
|
/// — two simultaneous `git add` / `commit` invocations race on
|
|
/// `.git/index.lock`; if either dies before releasing, the lock
|
|
/// sticks and the next operation hits "another git process seems to
|
|
/// be running" until somebody `rm`s it manually. Holding this mutex
|
|
/// across each public function's git+nix calls makes parallel
|
|
/// rebuilds (`auto_update` + dashboard-triggered + apply-commit)
|
|
/// take turns instead of colliding.
|
|
static META_LOCK: Mutex<()> = Mutex::const_new(());
|
|
|
|
/// Name of the operation holding [`META_LOCK`], empty when free. Read only to
|
|
/// name the holder in a stalled-waiter warning.
|
|
static META_HOLDER: StdMutex<&'static str> = StdMutex::new("");
|
|
|
|
/// A wait on [`META_LOCK`] longer than this logs a warning, then again every
|
|
/// interval, naming the waiter and the holder.
|
|
const LOCK_WAIT_WARN: Duration = Duration::from_mins(1);
|
|
|
|
/// [`META_LOCK`] guard that records its holder in [`META_HOLDER`].
|
|
struct MetaGuard<'a> {
|
|
_lock: MutexGuard<'a, ()>,
|
|
holder: &'a StdMutex<&'static str>,
|
|
}
|
|
|
|
impl Drop for MetaGuard<'_> {
|
|
// Runs before the field drop releases the mutex, so the next holder's
|
|
// name is never overwritten by this one's clear.
|
|
fn drop(&mut self) {
|
|
*self.holder.lock().unwrap_or_else(PoisonError::into_inner) = "";
|
|
}
|
|
}
|
|
|
|
/// Take [`META_LOCK`] for `op`, warning while the wait drags on.
|
|
async fn lock_meta(op: &'static str) -> MetaGuard<'static> {
|
|
lock_notifying(
|
|
&META_LOCK,
|
|
&META_HOLDER,
|
|
op,
|
|
LOCK_WAIT_WARN,
|
|
|holder, waited| {
|
|
tracing::warn!(
|
|
op,
|
|
holder,
|
|
waited_secs = waited.as_secs(),
|
|
"meta_lock: still waiting for META_LOCK"
|
|
);
|
|
},
|
|
)
|
|
.await
|
|
}
|
|
|
|
/// Take `lock`, calling `on_stall(holder, waited)` each time `interval` passes
|
|
/// without getting it.
|
|
async fn lock_notifying<'a>(
|
|
lock: &'a Mutex<()>,
|
|
holder: &'a StdMutex<&'static str>,
|
|
op: &'static str,
|
|
interval: Duration,
|
|
mut on_stall: impl FnMut(&'static str, Duration),
|
|
) -> MetaGuard<'a> {
|
|
let start = Instant::now();
|
|
// One acquire future for the whole wait: dropping it loses the place in
|
|
// the mutex's FIFO queue, so the stall timer runs beside it, not around it.
|
|
let acquire = lock.lock();
|
|
tokio::pin!(acquire);
|
|
let guard = loop {
|
|
tokio::select! {
|
|
guard = &mut acquire => break guard,
|
|
() = tokio::time::sleep(interval) => on_stall(
|
|
*holder.lock().unwrap_or_else(PoisonError::into_inner),
|
|
start.elapsed(),
|
|
),
|
|
}
|
|
};
|
|
*holder.lock().unwrap_or_else(PoisonError::into_inner) = op;
|
|
MetaGuard {
|
|
_lock: guard,
|
|
holder,
|
|
}
|
|
}
|
|
|
|
// Exclusivity for meta-repo *windows* that span multiple `META_LOCK`
|
|
// acquisitions is **not** a mutex in this module. `META_LOCK` above
|
|
// serializes individual git ops but cannot keep another op out of a window
|
|
// spanning several of them; that window is owned by the job queue instead,
|
|
// as `Resource::MetaWindow`, declared by every meta-mutating node kind. A
|
|
// resource can be held by a subtree root across its children, which a `MutexGuard`
|
|
// (bounded by one executor fn) cannot.
|
|
|
|
/// Where the manager sees this directory inside its container (RO bind).
|
|
pub const CONTAINER_MANAGER_META_MOUNT: &str = "/meta";
|
|
|
|
/// Stage topology.json if it exists: it is regenerated by `reconcile`
|
|
/// whenever the agent set changed. Without staging, an
|
|
/// existing-but-untracked file shows up as untracked in the meta repo,
|
|
/// which can confuse nix's dirty-tree fetch. `git add` is a no-op when
|
|
/// content is unchanged.
|
|
async fn stage_generated_meta_files(dir: &std::path::Path) -> Result<()> {
|
|
if crate::topology::topology_path().exists() {
|
|
git(dir, &["add", "topology.json"]).await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Meta files `sync_agents` writes and locks; an uncommitted change to them
|
|
/// means an earlier sync stopped before its commit.
|
|
const SYNC_OWNED_PATHS: &[&str] = &["flake.nix", "flake.lock"];
|
|
|
|
/// Whether `sync_agents` has nothing to do: the repo exists, the rendered
|
|
/// flake equals the one on disk, and no sync-owned file is uncommitted.
|
|
fn can_skip_sync(initial: bool, flake_unchanged: bool, uncommitted: bool) -> bool {
|
|
!initial && flake_unchanged && !uncommitted
|
|
}
|
|
|
|
/// Idempotently reconcile the meta repo with the current agent set.
|
|
/// First call inits the git repo, runs `nix flake lock`, and lands a
|
|
/// seed commit. Subsequent calls only touch `flake.nix` when the
|
|
/// rendered contents differ from disk; an unchanged `flake.nix` is a
|
|
/// no-op unless `flake.nix` or `flake.lock` is uncommitted, in which case
|
|
/// the lock and commit run again.
|
|
pub async fn sync_agents(hive: &HiveEnv, agents: &[AgentSpec]) -> Result<()> {
|
|
let _guard = lock_meta("sync_agents").await;
|
|
// Before anything is written: a hive without a forge URL would deploy a
|
|
// whole fleet of agents that silently never log in, because the agent
|
|
// option treats "unset" as "no forge configured" rather than erroring.
|
|
// This is the layer that knows a forge is mandatory, so it says so here.
|
|
require_service_urls(&forwarded_env_vars())?;
|
|
let dir = crate::paths::meta_root();
|
|
std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
|
|
|
|
// One agent whose forge main cannot be read fails the whole
|
|
// sync, like a missing forge URL or hyperhive rev above: rendering the
|
|
// fleet without it would drop a running agent from the flake.
|
|
let pinned = pin_agents(agents).await?;
|
|
let agents = pinned.as_slice();
|
|
|
|
let new_flake = render_hive_flake(hive, agents)?;
|
|
let flake_path = dir.join("flake.nix");
|
|
let on_disk = std::fs::read_to_string(&flake_path).unwrap_or_default();
|
|
let initial = !dir.join(".git").exists();
|
|
|
|
let flake_unchanged = on_disk == new_flake;
|
|
let uncommitted = if !initial && flake_unchanged {
|
|
paths_dirty(&dir, SYNC_OWNED_PATHS).await?
|
|
} else {
|
|
false
|
|
};
|
|
if can_skip_sync(initial, flake_unchanged, uncommitted) {
|
|
return Ok(());
|
|
}
|
|
if uncommitted {
|
|
tracing::info!("sync_agents: resuming an uncommitted flake.nix/flake.lock change");
|
|
}
|
|
|
|
// Safety guard: refuse to write an empty agent list over a non-empty
|
|
// on-disk flake. An empty `agents` slice is never intentional — it
|
|
// means `nixos-container list` failed and the caller got an empty
|
|
// fallback. Overwriting here would drop every agent from the meta
|
|
// flake and trigger unnecessary (and potentially destructive) cascade
|
|
// rebuilds. Callers that genuinely need to clear the agent list
|
|
// (there are none today) must handle this case explicitly.
|
|
if agents.is_empty() && !initial && !on_disk.is_empty() {
|
|
tracing::warn!(
|
|
"sync_agents: refusing to overwrite non-empty meta flake with empty agent list \
|
|
(nixos-container list may have failed)"
|
|
);
|
|
return Ok(());
|
|
}
|
|
|
|
std::fs::write(&flake_path, &new_flake)
|
|
.with_context(|| format!("write {}", flake_path.display()))?;
|
|
|
|
// Agents fetch the swarm root from the store at boot, so no CA file sits
|
|
// next to the flake; drop the ones earlier renders left behind.
|
|
let ca_touched = remove_stale_ca_files(&dir);
|
|
|
|
// Reconcile topology.json against the live agent set — adds
|
|
// newly-spawned agents and drops removed ones. Idempotent; when
|
|
// nothing changed the file isn't touched.
|
|
let agent_names: Vec<String> = agents.iter().map(|a| a.name.clone()).collect();
|
|
let pending: Vec<String> = crate::coordinator::Coordinator::pending_init_names()
|
|
.into_iter()
|
|
.map(hive_types::Ident::into_string)
|
|
.collect();
|
|
crate::topology::reconcile(&agent_names, &pending)
|
|
.with_context(|| format!("reconcile {}", crate::topology::topology_path().display()))?;
|
|
|
|
// Refresh /var/lib/hyperhive/run/agent-sockets.json — drives the
|
|
// gateway's unix-socket upstreams. Every agent binds a unix socket
|
|
// (`HIVE_WEB_SOCKET`); the gateway routes via this map, falling back
|
|
// to a computed TCP loopback port (`lifecycle::agent_web_port`) only
|
|
// while an agent's socket marker is absent. Best-effort + non-fatal.
|
|
// See `docs/networking/gateway.md::Per-agent unix-socket upstream`.
|
|
if let Err(e) = crate::agent_sockets::write(&agent_names) {
|
|
tracing::warn!(error = ?e, "agent_sockets::write failed (non-fatal)");
|
|
}
|
|
|
|
// Refresh /var/lib/hive-gateway/conf/agents.conf — the nginx include
|
|
// file the gateway reads at runtime. c0re then triggers a reload (or
|
|
// start) of the host's nginx via hive-priv, since c0re is unprivileged.
|
|
// Same best-effort + non-fatal shape.
|
|
if let Err(e) = crate::gateway_nginx::write(&agent_names).await {
|
|
tracing::warn!(error = ?e, "gateway_nginx::write failed (non-fatal)");
|
|
}
|
|
|
|
if initial {
|
|
git(&dir, &["init", "--initial-branch=main"]).await?;
|
|
}
|
|
// Stage flake.nix *before* running nix flake lock. When meta is
|
|
// a git repo, nix treats it as a `git+file://` self-reference;
|
|
// its dirty-tree fetcher includes index entries (tracked +
|
|
// staged) but skips untracked files, so without the stage step
|
|
// an untracked flake.nix surfaces as "source tree does not
|
|
// contain '/flake.nix'". Lock then commit once with both
|
|
// flake.nix and flake.lock — single commit per change.
|
|
git(&dir, &["add", "flake.nix"]).await?;
|
|
// Stage every stale CA file removed above. `git add <path>` stages a deletion when the path is
|
|
// tracked and now gone; best-effort so the never-tracked-and-absent case
|
|
// (pathspec mismatch) is a harmless no-op.
|
|
for name in &ca_touched {
|
|
let _ = git(&dir, &["add", "--", name]).await;
|
|
}
|
|
stage_generated_meta_files(&dir).await?;
|
|
nix(&dir, &["flake", "lock"]).await?;
|
|
if std::path::Path::new(&dir).join("flake.lock").exists() {
|
|
git(&dir, &["add", "flake.lock"]).await?;
|
|
}
|
|
// Build the commit message from what's actually staged so it
|
|
// reflects reality — and skip the commit entirely when nothing
|
|
// changed (avoids "nothing to commit" errors on redundant syncs).
|
|
let staged = git_staged_names(&dir).await?;
|
|
if staged.is_empty() {
|
|
return Ok(());
|
|
}
|
|
let msg = if initial {
|
|
format!("seed meta from {} agent(s)", agents.len())
|
|
} else {
|
|
// Compose a message that names every file that actually changed,
|
|
// mapping the on-disk filename to a short human label.
|
|
let labels: Vec<&str> = staged
|
|
.iter()
|
|
.filter_map(|f| match f.as_str() {
|
|
"flake.nix" => Some("flake"),
|
|
"flake.lock" => Some("lock"),
|
|
// Only ever a removal; see `remove_stale_ca_files`.
|
|
f if is_stale_ca_name(f) => Some("stale-ca"),
|
|
"topology.json" => Some("topology"),
|
|
_ => None,
|
|
})
|
|
.collect();
|
|
if labels.is_empty() {
|
|
"meta: update".to_owned()
|
|
} else {
|
|
format!("meta: update {}", labels.join(", "))
|
|
}
|
|
};
|
|
git_commit(&dir, &msg).await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// One-shot used by the manual-rebuild path: relock just one
|
|
/// agent's input and commit the lock change if any. Single-phase
|
|
/// (no separate finalize) because rebuild has no failure-revert
|
|
/// semantics. The input follows the url `sync_agents` rendered just before.
|
|
pub async fn lock_update_for_rebuild(name: &str) -> Result<()> {
|
|
let _guard = lock_meta("lock_update_for_rebuild").await;
|
|
let dir = crate::paths::meta_root();
|
|
let input = format!("agent-{name}");
|
|
nix(&dir, &rebuild_relock_args(&input)).await?;
|
|
if !paths_dirty(&dir, &["flake.lock"]).await? {
|
|
return Ok(());
|
|
}
|
|
git(&dir, &["add", "flake.lock"]).await?;
|
|
git_commit_paths(
|
|
&dir,
|
|
&format!("rebuild {name}: lock update"),
|
|
&["flake.lock"],
|
|
)
|
|
.await
|
|
}
|
|
|
|
/// `nix` arguments re-locking one agent's config input. The input is updated
|
|
/// against the url `sync_agents` rendered, the forge url pinned by `?rev=`:
|
|
/// there is no `--override-input`, so the lock never records another source.
|
|
fn rebuild_relock_args(input: &str) -> [&str; 3] {
|
|
["flake", "update", input]
|
|
}
|
|
|
|
/// Update one or more named inputs in the meta flake and commit
|
|
/// the resulting lock change with a single combined message.
|
|
/// Used by the dashboard's "update meta inputs" form so the
|
|
/// operator can bulk-bump `hyperhive` + selected agents in one
|
|
/// shot. Each input name is passed verbatim to
|
|
/// Run `nix flake update [inputs...]` on the meta flake and commit the
|
|
/// resulting lock changes. When `inputs` is empty, updates ALL inputs
|
|
/// (bare `nix flake update`). The caller is responsible for picking
|
|
/// real input keys (e.g. via `inputs_view()` snapshotted from the lock
|
|
/// file) when targeting specific inputs.
|
|
pub async fn lock_update(inputs: &[String]) -> Result<()> {
|
|
let _guard = lock_meta("lock_update").await;
|
|
let dir = crate::paths::meta_root();
|
|
let mut args: Vec<&str> = vec!["flake", "update"];
|
|
for i in inputs {
|
|
args.push(i.as_str());
|
|
}
|
|
nix(&dir, &args).await?;
|
|
if !paths_dirty(&dir, &["flake.lock"]).await? {
|
|
return Ok(());
|
|
}
|
|
git(&dir, &["add", "flake.lock"]).await?;
|
|
let msg = if inputs.is_empty() {
|
|
"lock update: all inputs".to_string()
|
|
} else if inputs.len() == 1 {
|
|
format!("lock update: {}", inputs[0])
|
|
} else {
|
|
format!("lock update: {}", inputs.join(", "))
|
|
};
|
|
git_commit_paths(&dir, &msg, &["flake.lock"]).await
|
|
}
|
|
|
|
/// One-shot used by the auto-update path: pin the latest hyperhive
|
|
/// rev, commit if the lock changed. Cheaper than `sync_agents`
|
|
/// because the per-agent inputs aren't touched.
|
|
pub async fn lock_update_hyperhive() -> Result<()> {
|
|
let _guard = lock_meta("lock_update_hyperhive").await;
|
|
let dir = crate::paths::meta_root();
|
|
nix(&dir, &["flake", "update", "hyperhive"]).await?;
|
|
if !paths_dirty(&dir, &["flake.lock"]).await? {
|
|
return Ok(());
|
|
}
|
|
git(&dir, &["add", "flake.lock"]).await?;
|
|
git_commit_paths(&dir, "bump hyperhive", &["flake.lock"]).await
|
|
}
|
|
|
|
/// Delete the meta repo's permission files and commit the removal under
|
|
/// `META_LOCK`. Their grants live in the swarm's `agent-permissions` bucket
|
|
/// once `workers::permissions_import` has sent them.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Removing a file or a git stage/commit step fails.
|
|
pub async fn remove_permission_files() -> Result<()> {
|
|
let _guard = lock_meta("remove_permission_files").await;
|
|
let dir = crate::paths::meta_root();
|
|
let mut removed = Vec::new();
|
|
for file in crate::workers::permissions_import::META_FILES {
|
|
match std::fs::remove_file(dir.join(file)) {
|
|
Ok(()) => {}
|
|
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
|
|
Err(e) => return Err(e).with_context(|| format!("removing {file}")),
|
|
}
|
|
git(&dir, &["rm", "--cached", "--ignore-unmatch", "-q", file]).await?;
|
|
// Only a tracked file has a removal to commit; naming an untracked
|
|
// one makes `git commit -- <path>` fail.
|
|
if paths_dirty(&dir, &[file]).await? {
|
|
removed.push(*file);
|
|
}
|
|
}
|
|
if !removed.is_empty() {
|
|
git_commit_paths(
|
|
&dir,
|
|
"permissions moved to the swarm's agent-permissions",
|
|
&removed,
|
|
)
|
|
.await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Pinned forge URL of the hyperhive flake: `git+<forge>/hyperhive/hyperhive.git?rev=<rev>`.
|
|
/// Errors when the host has no rev, rather than rendering an unpinned URL.
|
|
fn hyperhive_input_url(hive: &HiveEnv) -> Result<String> {
|
|
pinned_git_url(
|
|
&crate::forge::forge_git_url("hyperhive/hyperhive"),
|
|
hive.hyperhive_rev.as_deref(),
|
|
)
|
|
}
|
|
|
|
/// [`render_flake`] with the hyperhive pin taken from `hive`, as
|
|
/// [`sync_agents`] renders it. Nothing else in `hive` reaches the flake: the
|
|
/// values that differ per host go to the container at start
|
|
/// (`lifecycle::hive_facts`).
|
|
fn render_hive_flake(hive: &HiveEnv, agents: &[AgentSpec]) -> Result<String> {
|
|
render_flake(&hyperhive_input_url(hive)?, agents)
|
|
}
|
|
|
|
/// [`hive_meta_flake::render_flake`] with the forwarded values and the GitHub
|
|
/// switch read from this process's environment.
|
|
fn render_flake(hyperhive_url: &str, agents: &[AgentSpec]) -> Result<String> {
|
|
hive_meta_flake::render_flake(
|
|
hyperhive_url,
|
|
agents,
|
|
&forwarded_env_vars(),
|
|
std::env::var_os("HYPERHIVE_GITHUB_DISABLED").is_some(),
|
|
)
|
|
}
|
|
|
|
/// Service URLs a running hive must always supply, checked before rendering.
|
|
///
|
|
/// The forge is not optional on a real hive: the hyperhive NixOS module sets
|
|
/// `HIVE_FORGE_URL` unconditionally, so its absence means this daemon was
|
|
/// started outside its unit. The agent option is nullable — `null`
|
|
/// legitimately means "no forge" when the modules are evaluated on their own —
|
|
/// which is exactly why the hive has to assert its own requirement here rather
|
|
/// than leaning on the module to reject the empty case.
|
|
const REQUIRED_SERVICE_URL_VARS: &[&str] = &["HIVE_FORGE_URL"];
|
|
|
|
/// Fails unless every [`REQUIRED_SERVICE_URL_VARS`] entry is present in
|
|
/// `vars`.
|
|
///
|
|
/// Pure over the already-collected pairs so it can be tested without touching
|
|
/// process env (the same reason [`hive_meta_flake::push_forwarded_var_options`] is split out).
|
|
///
|
|
/// Checked by `sync_agents` — the point at which the hive commits a rendered
|
|
/// flake to disk — rather than inside the renderer. Rendering is a pure string
|
|
/// operation that many tests exercise directly; making *it* env-dependent
|
|
/// would mean every one of those tests either sets a process-wide var (the
|
|
/// parallel-test race this module already avoids) or fails for reasons that
|
|
/// have nothing to do with what it asserts.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// When a required var is missing. The hyperhive NixOS module sets it
|
|
/// unconditionally, so this means the daemon is running outside its unit;
|
|
/// refusing to
|
|
/// write the flake beats writing one whose agents would silently all lack a
|
|
/// forge.
|
|
fn require_service_urls(vars: &[(&'static str, String)]) -> Result<()> {
|
|
for required in REQUIRED_SERVICE_URL_VARS {
|
|
if !vars.iter().any(|(name, _)| name == required) {
|
|
anyhow::bail!(
|
|
"{required} is unset — the hyperhive NixOS module sets it unconditionally on \
|
|
the hive-c0re unit, so this process was started outside that unit. Refusing \
|
|
to write a meta flake whose agents would every one of them have no forge \
|
|
configured."
|
|
);
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
fn forwarded_env_vars() -> Vec<(&'static str, String)> {
|
|
FORWARDED_VARS
|
|
.iter()
|
|
.filter_map(|&name| {
|
|
std::env::var(name)
|
|
.ok()
|
|
.filter(|v| !v.is_empty())
|
|
.map(|v| (name, v))
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// Remove the CA files earlier renders embedded next to `flake.nix`
|
|
/// (`hive-ca.pem`, and the per-peer `peer-ca-<N>.pem` before it), returning
|
|
/// every filename removed so the caller stages the deletion. Nothing writes
|
|
/// one now: agents fetch the swarm root from the store at boot, so a file left
|
|
/// here is referenced by nothing.
|
|
fn remove_stale_ca_files(dir: &Path) -> Vec<String> {
|
|
let mut removed = Vec::new();
|
|
if let Ok(entries) = std::fs::read_dir(dir) {
|
|
for e in entries.flatten() {
|
|
let fname = e.file_name();
|
|
let Some(name) = fname.to_str() else { continue };
|
|
if is_stale_ca_name(name) && std::fs::remove_file(dir.join(name)).is_ok() {
|
|
removed.push(name.to_owned());
|
|
}
|
|
}
|
|
}
|
|
removed
|
|
}
|
|
|
|
/// True for the filenames [`remove_stale_ca_files`] owns.
|
|
fn is_stale_ca_name(name: &str) -> bool {
|
|
name == "hive-ca.pem"
|
|
|| (name.starts_with("peer-ca-")
|
|
&& Path::new(name)
|
|
.extension()
|
|
.is_some_and(|ext| ext.eq_ignore_ascii_case("pem")))
|
|
}
|
|
|
|
/// Resolve every custom agent's [`AgentPin`] and return `agents` with it set.
|
|
async fn pin_agents(agents: &[AgentSpec]) -> Result<Vec<AgentSpec>> {
|
|
let mut out = Vec::with_capacity(agents.len());
|
|
for spec in agents {
|
|
let mut spec = spec.clone();
|
|
spec.pin = if spec.archetype.is_some() {
|
|
None
|
|
} else {
|
|
Some(
|
|
pin_agent(
|
|
&spec.name,
|
|
&crate::paths::applied_dir(&spec.name),
|
|
&crate::forge::forge_git_url(&format!(
|
|
"{}/{}",
|
|
crate::forge::CONFIG_ORG,
|
|
spec.name
|
|
)),
|
|
crate::forge::FORGE_MAIN_REF,
|
|
crate::forge::fetch_forge_main(&spec.name),
|
|
)
|
|
.await?,
|
|
)
|
|
};
|
|
out.push(spec);
|
|
}
|
|
Ok(out)
|
|
}
|
|
|
|
/// Pin the config input of agent `name` to the head of `agent-configs/<name>`
|
|
/// `main` on the forge. `fetch_forge` fetches that branch into `repo` as
|
|
/// `forge_ref` ([`crate::forge::fetch_forge_main`]); `repo` is only the object
|
|
/// store that fetch lands in, its own branches and files are never read. The
|
|
/// pin is a commit the forge has, so a flake rendered from it resolves off this
|
|
/// host. Errors, with no fallback to a local path, when the forge cannot be
|
|
/// reached or has no `main`.
|
|
async fn pin_agent<F>(
|
|
name: &str,
|
|
repo: &Path,
|
|
forge_git_url: &str,
|
|
forge_ref: &str,
|
|
fetch_forge: F,
|
|
) -> Result<AgentPin>
|
|
where
|
|
F: std::future::Future<Output = Result<std::path::PathBuf>>,
|
|
{
|
|
fetch_forge
|
|
.await
|
|
.with_context(|| format!("agent {name}: read agent-configs/{name} main from the forge"))?;
|
|
let rev = lifecycle::git_rev_parse(repo, forge_ref)
|
|
.await
|
|
.with_context(|| format!("agent {name}: agent-configs/{name} has no main on the forge"))?;
|
|
let lock = lifecycle::git_show_file(repo, &rev, "flake.lock").await?;
|
|
Ok(AgentPin {
|
|
url: pinned_git_url(forge_git_url, Some(&rev))?,
|
|
canonical_inputs: lock
|
|
.as_deref()
|
|
.map(canonical_inputs_in_lock)
|
|
.unwrap_or_default(),
|
|
})
|
|
}
|
|
|
|
/// Return the list of file names that are currently staged (index differs
|
|
/// from HEAD). On the initial commit (`HEAD` doesn't exist yet) falls back
|
|
/// to `git diff --cached --name-only HEAD` failing gracefully by using
|
|
/// `git status --porcelain` and collecting the `A ` / `M ` prefix lines.
|
|
async fn git_staged_names(dir: &Path) -> Result<Vec<String>> {
|
|
// `--diff-filter=ACM` skips deleted entries — we only care about
|
|
// additions and modifications for message-building purposes.
|
|
let out = git_output(
|
|
dir,
|
|
&["diff", "--cached", "--name-only", "--diff-filter=ACM"],
|
|
)
|
|
.await?;
|
|
if out.status.success() {
|
|
let names = String::from_utf8_lossy(&out.stdout)
|
|
.lines()
|
|
.map(str::trim)
|
|
.filter(|l| !l.is_empty())
|
|
.map(ToOwned::to_owned)
|
|
.collect();
|
|
return Ok(names);
|
|
}
|
|
// Fallback for initial commit (no HEAD yet): parse `git status --porcelain`.
|
|
let st = git_output(dir, &["status", "--porcelain"]).await?;
|
|
let names = String::from_utf8_lossy(&st.stdout)
|
|
.lines()
|
|
.filter(|l| l.starts_with("A ") || l.starts_with("M "))
|
|
.filter_map(|l| l.get(3..))
|
|
.map(ToOwned::to_owned)
|
|
.collect();
|
|
Ok(names)
|
|
}
|
|
|
|
/// Run `git <args>` in `dir` within [`lifecycle::GIT_LOCAL_DEADLINE`]; the caller
|
|
/// reads the exit status.
|
|
async fn git_output(dir: &Path, args: &[&str]) -> Result<std::process::Output> {
|
|
lifecycle::run_with_deadline(
|
|
lifecycle::git_command().current_dir(dir).args(args),
|
|
lifecycle::GIT_LOCAL_DEADLINE,
|
|
&format!("git {} in {}", args.join(" "), dir.display()),
|
|
)
|
|
.await
|
|
}
|
|
|
|
async fn git(dir: &Path, args: &[&str]) -> Result<()> {
|
|
let out = git_output(dir, args).await?;
|
|
if !out.status.success() {
|
|
bail!(
|
|
"git {} failed ({}): {}",
|
|
args.join(" "),
|
|
out.status,
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
);
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
async fn git_commit(dir: &Path, message: &str) -> Result<()> {
|
|
git(
|
|
dir,
|
|
&[
|
|
"-c",
|
|
&format!("user.name={GIT_NAME}"),
|
|
"-c",
|
|
&format!("user.email={GIT_EMAIL}"),
|
|
"commit",
|
|
"-m",
|
|
message,
|
|
],
|
|
)
|
|
.await?;
|
|
// Best-effort mirror to the bundled forge. No-op when the forge
|
|
// isn't seeded (no core token on disk); push failures log a warn
|
|
// but don't bubble up — a missing mirror shouldn't fail an
|
|
// otherwise successful deploy.
|
|
if let Err(e) = crate::forge::push_meta(dir).await {
|
|
tracing::warn!(error = ?e, "forge: meta push after commit failed (non-fatal)");
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Path-limited commit: commits ONLY the given paths, so unrelated
|
|
/// staged content can never be swept into someone else's commit. Every targeted
|
|
/// meta commit (limits, topology, lock bumps) goes through this;
|
|
/// only `sync_agents` uses the bare [`git_commit`], because its
|
|
/// staged set *is* its intentional commit set.
|
|
async fn git_commit_paths(dir: &Path, message: &str, paths: &[&str]) -> Result<()> {
|
|
let name = format!("user.name={GIT_NAME}");
|
|
let email = format!("user.email={GIT_EMAIL}");
|
|
let mut args = vec!["-c", &name, "-c", &email, "commit", "-m", message, "--"];
|
|
args.extend_from_slice(paths);
|
|
git(dir, &args).await?;
|
|
if let Err(e) = crate::forge::push_meta(dir).await {
|
|
tracing::warn!(error = ?e, "forge: meta push after commit failed (non-fatal)");
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// True when any of `paths` differs between HEAD and the index or
|
|
/// working tree — the path-scoped replacement for whole-tree
|
|
/// `git_is_clean` / `has_staged_changes` guards, which a concurrently
|
|
/// staged deploy lock would otherwise trip.
|
|
async fn paths_dirty(dir: &Path, paths: &[&str]) -> Result<bool> {
|
|
let mut args = vec!["diff", "--quiet", "HEAD", "--"];
|
|
args.extend_from_slice(paths);
|
|
let out = git_output(dir, &args).await?;
|
|
// Exit 0 = no differences; 1 = differences; anything else (e.g.
|
|
// no HEAD yet on a fresh repo) → treat as dirty so the commit
|
|
// path runs and surfaces real errors loudly.
|
|
Ok(!out.status.success())
|
|
}
|
|
|
|
/// Full argv for a `nix` invocation, prefixed with the flakes-enabling
|
|
/// experimental-features flag. `--extra-experimental-features` is
|
|
/// belt-and-suspenders for hosts that haven't set this in nix.conf: the
|
|
/// hyperhive module's deploy guide assumes flakes are already enabled,
|
|
/// but the cost of being defensive is one extra argv each call.
|
|
fn nix_argv<'a>(args: &[&'a str]) -> Vec<&'a str> {
|
|
let mut all = vec!["--extra-experimental-features", "nix-command flakes"];
|
|
all.extend_from_slice(args);
|
|
all
|
|
}
|
|
|
|
/// Run `nix <args>` in `dir` (flakes enabled), capturing combined output.
|
|
/// The core of [`nix`].
|
|
///
|
|
/// Every call site in this module is `flake lock` or `flake update`, which
|
|
/// fetch inputs from the forge, so the whole function runs under
|
|
/// [`lifecycle::FETCH_DEADLINE`]. A call that builds or evaluates a system
|
|
/// needs its own bound, not this one.
|
|
async fn nix_output(dir: &Path, args: &[&str]) -> Result<std::process::Output> {
|
|
nix_output_with("nix".as_ref(), dir, args, lifecycle::FETCH_DEADLINE).await
|
|
}
|
|
|
|
/// [`nix_output`] with the program and deadline injectable for tests.
|
|
async fn nix_output_with(
|
|
program: &std::ffi::OsStr,
|
|
dir: &Path,
|
|
args: &[&str],
|
|
deadline: Duration,
|
|
) -> Result<std::process::Output> {
|
|
lifecycle::run_with_deadline(
|
|
Command::new(program).current_dir(dir).args(nix_argv(args)),
|
|
deadline,
|
|
&format!("nix {} in {}", args.join(" "), dir.display()),
|
|
)
|
|
.await
|
|
}
|
|
|
|
/// Turn a finished nix [`Output`](std::process::Output) into `Result<()>`,
|
|
/// bailing with the trimmed stderr tail on non-zero exit.
|
|
fn nix_check(args: &[&str], out: &std::process::Output) -> Result<()> {
|
|
if !out.status.success() {
|
|
bail!(
|
|
"nix {} failed ({}): {}",
|
|
args.join(" "),
|
|
out.status,
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
);
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
async fn nix(dir: &Path, args: &[&str]) -> Result<()> {
|
|
let out = nix_output(dir, args).await?;
|
|
nix_check(args, &out)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn sync_skips_only_when_clean_unchanged_and_not_initial() {
|
|
assert!(can_skip_sync(false, true, false));
|
|
assert!(!can_skip_sync(false, true, true), "interrupted sync");
|
|
assert!(!can_skip_sync(false, false, false), "flake changed");
|
|
assert!(!can_skip_sync(true, true, false), "no repo yet");
|
|
}
|
|
|
|
/// An interrupted sync leaves the new `flake.nix` written with the old
|
|
/// lock; the dirty check over the sync-owned paths must see it, and stay
|
|
/// quiet on a clean tree and on files sync does not own.
|
|
#[tokio::test]
|
|
async fn interrupted_sync_leaves_sync_owned_paths_dirty() {
|
|
let tmp = tempfile::tempdir().expect("tempdir");
|
|
let dir = tmp.path();
|
|
git(dir, &["init", "--initial-branch=main"])
|
|
.await
|
|
.expect("git init");
|
|
std::fs::write(dir.join("flake.nix"), "old").expect("write");
|
|
std::fs::write(dir.join("flake.lock"), "lock-old").expect("write");
|
|
std::fs::write(dir.join("other.json"), "{}").expect("write");
|
|
git(dir, &["add", "-A"]).await.expect("add");
|
|
git_commit(dir, "seed").await.expect("seed commit");
|
|
assert!(!paths_dirty(dir, SYNC_OWNED_PATHS).await.expect("check"));
|
|
|
|
std::fs::write(dir.join("other.json"), r#"{"x":1}"#).expect("write");
|
|
assert!(
|
|
!paths_dirty(dir, SYNC_OWNED_PATHS).await.expect("check"),
|
|
"a file sync does not own is not an interrupted sync"
|
|
);
|
|
|
|
std::fs::write(dir.join("flake.nix"), "new").expect("write");
|
|
git(dir, &["add", "flake.nix"]).await.expect("stage");
|
|
assert!(
|
|
paths_dirty(dir, SYNC_OWNED_PATHS).await.expect("check"),
|
|
"new flake.nix written but not committed"
|
|
);
|
|
git(dir, &["add", "-A"]).await.expect("add");
|
|
git_commit(dir, "resume").await.expect("commit");
|
|
assert!(!paths_dirty(dir, SYNC_OWNED_PATHS).await.expect("check"));
|
|
}
|
|
|
|
/// A path-limited commit must leave an unrelated staged file untouched.
|
|
#[tokio::test]
|
|
async fn path_limited_commit_leaves_unrelated_staged_file_alone() {
|
|
let tmp = tempfile::tempdir().expect("tempdir");
|
|
let dir = tmp.path();
|
|
git(dir, &["init", "--initial-branch=main"])
|
|
.await
|
|
.expect("git init");
|
|
std::fs::write(dir.join("resource-limits.json"), "{}").expect("write");
|
|
std::fs::write(dir.join("flake.lock"), "v1").expect("write");
|
|
git(dir, &["add", "-A"]).await.expect("add");
|
|
git_commit(dir, "seed").await.expect("seed commit");
|
|
// Something stages a new lock (uncommitted)…
|
|
std::fs::write(dir.join("flake.lock"), "v2-staged-by-deploy").expect("write");
|
|
git(dir, &["add", "flake.lock"]).await.expect("stage lock");
|
|
// …and a limits change commits, path-limited.
|
|
std::fs::write(dir.join("resource-limits.json"), r#"{"alice":[]}"#).expect("write");
|
|
git(dir, &["add", "resource-limits.json"])
|
|
.await
|
|
.expect("add");
|
|
git_commit_paths(
|
|
dir,
|
|
"set resource limits for alice",
|
|
&["resource-limits.json"],
|
|
)
|
|
.await
|
|
.expect("path-limited commit");
|
|
// The limits file is committed; the staged lock is not.
|
|
assert!(
|
|
!paths_dirty(dir, &["resource-limits.json"])
|
|
.await
|
|
.expect("check"),
|
|
"limits file must be committed"
|
|
);
|
|
assert!(
|
|
paths_dirty(dir, &["flake.lock"]).await.expect("check"),
|
|
"staged deploy lock must survive the limits commit"
|
|
);
|
|
}
|
|
|
|
const SAMPLE_REV: &str = "1111111111111111111111111111111111111111";
|
|
|
|
fn sample_pin(name: &str, rev: &str, canonical_inputs: Vec<&'static str>) -> AgentPin {
|
|
AgentPin {
|
|
url: format!("git+https://forge.example/agent-configs/{name}.git?rev={rev}"),
|
|
canonical_inputs,
|
|
}
|
|
}
|
|
|
|
fn sample_spec(name: &str, port: u16) -> AgentSpec {
|
|
AgentSpec {
|
|
name: name.to_owned(),
|
|
port,
|
|
archetype: None,
|
|
pin: Some(sample_pin(name, SAMPLE_REV, Vec::new())),
|
|
}
|
|
}
|
|
|
|
fn render(hyperhive_url: &str, agents: &[AgentSpec]) -> String {
|
|
render_flake(hyperhive_url, agents).expect("render")
|
|
}
|
|
|
|
#[test]
|
|
fn rebuild_relock_updates_the_input_without_an_override() {
|
|
let args = rebuild_relock_args("agent-iris");
|
|
assert_eq!(args, ["flake", "update", "agent-iris"]);
|
|
assert!(
|
|
!args
|
|
.iter()
|
|
.any(|a| *a == "--override-input" || a.contains("git+file")),
|
|
"a rebuild must not lock a local path: {args:?}"
|
|
);
|
|
}
|
|
|
|
/// The url a rebuild relocks is the one `sync_agents` rendered: the forge,
|
|
/// pinned by rev.
|
|
#[test]
|
|
fn the_input_a_rebuild_relocks_is_a_pinned_forge_url() {
|
|
let out = render(
|
|
"git+https://forge.example/hyperhive/hyperhive.git?rev=0123abcd",
|
|
&[sample_spec("iris", 9001)],
|
|
);
|
|
let url = out
|
|
.lines()
|
|
.find_map(|l| l.trim().strip_prefix("agent-iris.url = \""))
|
|
.expect("agent-iris input is rendered");
|
|
assert!(
|
|
url.starts_with("git+https://forge.example/agent-configs/iris.git?rev="),
|
|
"{url}"
|
|
);
|
|
assert!(!url.contains("git+file"), "{url}");
|
|
}
|
|
|
|
#[test]
|
|
fn pinned_git_url_carries_the_rev_and_refuses_without_one() {
|
|
assert_eq!(
|
|
pinned_git_url(
|
|
"https://forge.example/hyperhive/hyperhive.git",
|
|
Some("abc123")
|
|
)
|
|
.unwrap(),
|
|
"git+https://forge.example/hyperhive/hyperhive.git?rev=abc123"
|
|
);
|
|
for missing in [None, Some("")] {
|
|
let err = pinned_git_url("https://forge.example/x.git", missing).unwrap_err();
|
|
assert!(err.to_string().contains("rev unknown"), "{err}");
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn render_flake_pins_hyperhive_by_url_and_follows_its_nixpkgs() {
|
|
let out = render(
|
|
"git+https://forge.example/hyperhive/hyperhive.git?rev=0123abcd",
|
|
&[sample_spec("alice", 9001)],
|
|
);
|
|
assert!(
|
|
out.contains(
|
|
"hyperhive.url = \"git+https://forge.example/hyperhive/hyperhive.git?rev=0123abcd\";"
|
|
),
|
|
"expected hyperhive pinned by forge URL:\n{out}"
|
|
);
|
|
assert!(
|
|
out.contains("nixpkgs.follows = \"hyperhive/nixpkgs\";"),
|
|
"expected nixpkgs to follow hyperhive:\n{out}"
|
|
);
|
|
assert!(
|
|
!out.contains("nixpkgs.url") && !out.contains("hyperhive.inputs.nixpkgs"),
|
|
"no separate nixpkgs input:\n{out}"
|
|
);
|
|
// docs come from the pinned hyperhive input: no separate docs input,
|
|
// no docs.source override.
|
|
assert!(
|
|
!out.contains("hyperhive-docs") && !out.contains("docs.source"),
|
|
"no docs input or override:\n{out}"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn render_flake_names_each_agent_container_after_the_agent() {
|
|
let out = render(
|
|
"git+https://forge.example/hyperhive/hyperhive.git?rev=0123abcd",
|
|
&[sample_spec("alice", 9001)],
|
|
);
|
|
assert!(
|
|
out.contains(r#"networking.hostName = "h-${name}";"#),
|
|
"an agent container with no hostname of its own is called `nixos`, \
|
|
and ships every log line under that:\n{out}"
|
|
);
|
|
// The machine name `nixos-container` and our own metrics already use,
|
|
// so a log line joins to a metric without a prefix transform.
|
|
assert!(
|
|
!out.contains("networking.hostName = name;"),
|
|
"must be the machine name, not the logical one:\n{out}"
|
|
);
|
|
// Inside `mkAgent`'s module block, so it applies to every agent and
|
|
// resolves to that agent's name — not emitted once at the top level.
|
|
let mk = out.find("mkAgent = {").expect("mkAgent block");
|
|
let host = out
|
|
.find(r#"networking.hostName = "h-${name}";"#)
|
|
.expect("hostname line");
|
|
assert!(
|
|
host > mk,
|
|
"hostname must be emitted inside the per-agent module block:\n{out}"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn render_flake_emits_follows_for_agents_declaring_nixpkgs() {
|
|
// `bitburner` + `dmatrix` declare `nixpkgs` at their root, while
|
|
// `argus` has no canonical inputs at all.
|
|
let with_nixpkgs = |name, port| AgentSpec {
|
|
pin: Some(sample_pin(name, SAMPLE_REV, vec!["nixpkgs"])),
|
|
..sample_spec(name, port)
|
|
};
|
|
let out = render(
|
|
"git+https://forge.example/hyperhive/hyperhive.git?rev=0123abcd",
|
|
&[
|
|
sample_spec("argus", 9001),
|
|
with_nixpkgs("bitburner", 9002),
|
|
with_nixpkgs("dmatrix", 9003),
|
|
],
|
|
);
|
|
// bitburner declares nixpkgs → follows emitted.
|
|
assert!(
|
|
out.contains("agent-bitburner.inputs.nixpkgs.follows = \"nixpkgs\""),
|
|
"missing bitburner nixpkgs follows:\n{out}"
|
|
);
|
|
// dmatrix declares nixpkgs → follows emitted.
|
|
assert!(out.contains("agent-dmatrix.inputs.nixpkgs.follows = \"nixpkgs\""));
|
|
// argus declares neither → no follows emitted for it. Asserting
|
|
// ABSENCE is the important bit: emitting a follows on a
|
|
// non-existent input errors at `nix flake lock` time.
|
|
assert!(
|
|
!out.contains("agent-argus.inputs.nixpkgs"),
|
|
"argus shouldn't have nixpkgs follows:\n{out}"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn render_flake_builds_every_agent_from_agent_base() {
|
|
let out = render(
|
|
"git+https://forge.example/hyperhive/hyperhive.git?rev=0123abcd",
|
|
&[sample_spec("alice", 9001), sample_spec("ruth", 9002)],
|
|
);
|
|
assert_eq!(
|
|
out.matches("hyperhive.nixosConfigurations.").count(),
|
|
1,
|
|
"one base for every agent:\n{out}"
|
|
);
|
|
assert!(
|
|
out.contains("base = hyperhive.nixosConfigurations.agent-base;"),
|
|
"{out}"
|
|
);
|
|
for line in [
|
|
"alice = mkAgent { name = \"alice\"; port = 9001; };",
|
|
"ruth = mkAgent { name = \"ruth\"; port = 9002; };",
|
|
] {
|
|
assert!(out.contains(line), "missing `{line}`:\n{out}");
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn render_flake_skips_canonical_follows_when_the_lock_declares_none() {
|
|
let out = render(
|
|
"git+https://forge.example/hyperhive/hyperhive.git?rev=0123abcd",
|
|
&[sample_spec("alice", 9001)],
|
|
);
|
|
// No agent-side follows when the lock declares nothing
|
|
// declared — protects agents whose flake.lock can't be read
|
|
// (missing / unparsable) from being broken by a follows on a
|
|
// non-existent input.
|
|
assert!(
|
|
!out.contains("agent-alice.inputs."),
|
|
"alice shouldn't have any inputs follows:\n{out}"
|
|
);
|
|
}
|
|
|
|
/// An archetype agent has no `agent-<name>` input, and `mkAgent` is
|
|
/// handed its archetype; a custom agent beside it keeps its input.
|
|
#[test]
|
|
fn render_flake_builds_an_archetype_agent_from_its_archetype() {
|
|
let android = AgentSpec {
|
|
archetype: Some(swarm_queue_client::agent_archetype::ArchetypeRef {
|
|
source: "hyperhive".to_owned(),
|
|
name: "android-dev".to_owned(),
|
|
}),
|
|
..sample_spec("droid", 9002)
|
|
};
|
|
let alice = AgentSpec {
|
|
pin: Some(sample_pin("alice", SAMPLE_REV, vec!["nixpkgs"])),
|
|
..sample_spec("alice", 9001)
|
|
};
|
|
let out = render(
|
|
"git+https://forge.example/hyperhive/hyperhive.git?rev=0123abcd",
|
|
&[alice, android],
|
|
);
|
|
assert!(
|
|
!out.contains("agent-droid"),
|
|
"an archetype agent has no config input:\n{out}"
|
|
);
|
|
assert!(
|
|
out.contains(
|
|
"droid = mkAgent { name = \"droid\"; port = 9002; \
|
|
archetype = { source = \"hyperhive\"; name = \"android-dev\"; }; };"
|
|
),
|
|
"{out}"
|
|
);
|
|
assert!(out.contains("agent-alice.url = \"git+https://"), "{out}");
|
|
assert!(out.contains("agent-alice.inputs.nixpkgs.follows"), "{out}");
|
|
assert!(
|
|
out.contains("alice = mkAgent { name = \"alice\"; port = 9001; };"),
|
|
"{out}"
|
|
);
|
|
assert!(
|
|
out.contains("hyperhiveArchetypes.${archetype.name}.module"),
|
|
"{out}"
|
|
);
|
|
assert!(
|
|
out.contains("environment.HIVE_ARCHETYPE = archetype.name"),
|
|
"{out}"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn render_flake_forwards_env_into_global_environment() {
|
|
// Regression for the isolation breakage where forwarded vars
|
|
// (HIVE_FORGE_URL etc.) landed only on the harness service env,
|
|
// so the bash-task runner / matrix daemon / shells defaulted to
|
|
// localhost and couldn't reach the in-cluster gateway. They must
|
|
// also appear in `systemd.globalEnvironment`, which every unit +
|
|
// shell in the container inherits.
|
|
//
|
|
// Serialised against every other env-mutating test in the crate:
|
|
// `render_flake_sets_service_url_options_from_forwarded_env` sets this
|
|
// same variable, and both land in the one test binary.
|
|
let _env = crate::test_env::lock();
|
|
// SAFETY: serialised by the guard above; restored before returning.
|
|
unsafe {
|
|
std::env::set_var("HIVE_FORGE_URL", "http://forge.example.test");
|
|
}
|
|
let out = render(
|
|
"git+https://forge.example/hyperhive/hyperhive.git?rev=0123abcd",
|
|
&[sample_spec("alice", 9001)],
|
|
);
|
|
unsafe {
|
|
std::env::remove_var("HIVE_FORGE_URL");
|
|
}
|
|
// The var must be emitted inside the globalEnvironment block, i.e.
|
|
// before the per-service harness env block that follows it.
|
|
let global_at = out
|
|
.find("systemd.globalEnvironment = {")
|
|
.expect("globalEnvironment block must exist");
|
|
let service_at = out
|
|
.find("systemd.services.${service}.environment")
|
|
.expect("harness service env block must exist");
|
|
let forge_at = out
|
|
.find("HIVE_FORGE_URL = \"http://forge.example.test\"")
|
|
.expect("HIVE_FORGE_URL must be forwarded into the flake");
|
|
assert!(
|
|
forge_at > global_at && forge_at < service_at,
|
|
"HIVE_FORGE_URL must land inside systemd.globalEnvironment, \
|
|
not only the harness service env:\n{out}"
|
|
);
|
|
}
|
|
|
|
/// The hive's name and domain reach an agent at runtime
|
|
/// (`/etc/hyperhive-hive-facts`), so the rendered flake does not change with
|
|
/// either.
|
|
#[test]
|
|
fn render_flake_is_identical_across_hive_names_and_domains() {
|
|
const VARS: [&str; 2] = ["HYPERHIVE_HIVE_NAME", "HYPERHIVE_HIVE_DOMAIN"];
|
|
let _env = crate::test_env::lock();
|
|
let saved = VARS.map(std::env::var_os);
|
|
let render_on = |name: &str, domain: &str| {
|
|
// SAFETY: serialised by the guard above; restored before returning.
|
|
unsafe {
|
|
std::env::set_var(VARS[0], name);
|
|
std::env::set_var(VARS[1], domain);
|
|
}
|
|
render(
|
|
"git+https://forge.example/hyperhive/hyperhive.git?rev=0123abcd",
|
|
&[sample_spec("alice", 9001)],
|
|
)
|
|
};
|
|
let alpha = render_on("alpha-hive", "alpha-hive.example.test");
|
|
let beta = render_on("beta-hive", "beta-hive.example.test");
|
|
for (var, value) in VARS.iter().zip(saved) {
|
|
// SAFETY: as above.
|
|
unsafe {
|
|
match value {
|
|
Some(v) => std::env::set_var(var, v),
|
|
None => std::env::remove_var(var),
|
|
}
|
|
}
|
|
}
|
|
for hive in ["alpha-hive", "beta-hive"] {
|
|
assert!(
|
|
!alpha.contains(hive) && !beta.contains(hive),
|
|
"the rendered flake must not name the hive ({hive}):\n{alpha}"
|
|
);
|
|
}
|
|
assert_eq!(alpha, beta, "the render must not depend on the hive");
|
|
}
|
|
|
|
#[test]
|
|
fn service_url_options_render_from_forwarded_pairs() {
|
|
// Pure over the pairs, deliberately: the render-level tests have to
|
|
// mutate process env, which makes them race each other under the
|
|
// parallel runner — the first version of this test did exactly that
|
|
// and failed for that reason rather than for a real defect.
|
|
let mut out = String::new();
|
|
push_forwarded_var_options(
|
|
&mut out,
|
|
&[
|
|
("HIVE_FORGE_URL", "http://forge.example.test".to_string()),
|
|
("HIVE_MATRIX_URL", "http://matrix.example.test".to_string()),
|
|
// Mapped too: `otel.nix` reads the option at eval time.
|
|
("HYPERHIVE_SWARM_NAME", "constellat1on".to_string()),
|
|
// Forwarded but genuinely unmapped — proves the map is a
|
|
// filter, not a pass-through.
|
|
(
|
|
"HIVE_FORGE_PUBLIC_URL",
|
|
"https://forge.example.test".to_string(),
|
|
),
|
|
],
|
|
);
|
|
assert_eq!(
|
|
out,
|
|
" services.hyperhive.agent.forge.url = \"http://forge.example.test\";\n\
|
|
\x20 services.hyperhive.agent.matrix.url = \"http://matrix.example.test\";\n\
|
|
\x20 services.hyperhive.agent.swarmName = \"constellat1on\";\n",
|
|
"expected exactly the mapped options, in input order, indented for the module block"
|
|
);
|
|
}
|
|
|
|
/// The queue address reaches the container in its hive facts, so the render
|
|
/// carries neither it nor any `services.hyperhive.agent.queue.*` option,
|
|
/// which no agent module declares.
|
|
#[test]
|
|
fn queue_address_is_not_rendered_into_the_flake() {
|
|
assert!(!FORWARDED_VARS.contains(&"HIVE_AGENT_NATS_URL"));
|
|
assert!(
|
|
!FORWARDED_VARS.contains(&"HIVE_AGENT_OIDC_TOKEN_ENDPOINT"),
|
|
"no agent reads a token endpoint"
|
|
);
|
|
for (var, option) in FORWARDED_VAR_OPTIONS {
|
|
assert!(
|
|
!option.starts_with("services.hyperhive.agent.queue."),
|
|
"{var} renders {option}, which no agent module declares"
|
|
);
|
|
}
|
|
}
|
|
|
|
/// `swarm-metrics` is installed only when the option is set, and reads
|
|
/// the env var at runtime, so the metrics store's origin needs both halves.
|
|
#[test]
|
|
fn the_metrics_url_is_forwarded_as_both_env_and_option() {
|
|
assert!(FORWARDED_VARS.contains(&"HIVE_AGENT_METRICS_URL"));
|
|
let mut out = String::new();
|
|
push_forwarded_var_options(
|
|
&mut out,
|
|
&[(
|
|
"HIVE_AGENT_METRICS_URL",
|
|
"https://metrics.t.local".to_string(),
|
|
)],
|
|
);
|
|
assert_eq!(
|
|
out,
|
|
" services.hyperhive.agent.metrics.url = \"https://metrics.t.local\";\n"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn require_service_urls_accepts_a_rendered_forge_url() {
|
|
require_service_urls(&[
|
|
("HIVE_FORGE_URL", "http://forge.example.test".to_string()),
|
|
("HYPERHIVE_SWARM_NAME", "constellat1on".to_string()),
|
|
])
|
|
.expect("a forwarded forge URL satisfies the requirement");
|
|
}
|
|
|
|
#[test]
|
|
fn require_service_urls_refuses_to_write_without_a_forge() {
|
|
// The agent option is nullable, so nothing downstream would complain:
|
|
// every agent would simply come up with no forge login and no way to
|
|
// tell that was unintended. The hive asserts its own requirement
|
|
// because it is the only layer that knows a forge is mandatory.
|
|
let err = require_service_urls(&[("HYPERHIVE_SWARM_NAME", "constellat1on".to_string())])
|
|
.expect_err("a missing forge URL must stop the write");
|
|
assert!(
|
|
err.to_string().contains("HIVE_FORGE_URL is unset"),
|
|
"the error must name the missing variable: {err}"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn service_url_options_emit_nothing_when_absent() {
|
|
// No guess when the host doesn't say — the agent option stays `null`
|
|
// ("not configured") and the units that would use it aren't generated,
|
|
// so absence is an absent integration rather than a misdirected one.
|
|
// Required services don't reach here: `require_service_urls` rejects
|
|
// them first.
|
|
// `HIVE_FORGE_PUBLIC_URL` is forwarded as env but carries no option —
|
|
// deliberately picked over a mapped var, since the whole point is a
|
|
// forwarded var the option map does not know about.
|
|
let mut out = String::new();
|
|
push_forwarded_var_options(
|
|
&mut out,
|
|
&[(
|
|
"HIVE_FORGE_PUBLIC_URL",
|
|
"https://forge.example.test".to_string(),
|
|
)],
|
|
);
|
|
assert!(
|
|
out.is_empty(),
|
|
"no option should be emitted for an unmapped forwarded var, got:\n{out}"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn service_url_options_escape_quotes() {
|
|
// The value lands inside a nix string literal; an unescaped quote
|
|
// would end the string and change the surrounding config rather than
|
|
// merely corrupting one value.
|
|
let mut out = String::new();
|
|
push_forwarded_var_options(
|
|
&mut out,
|
|
&[("HIVE_FORGE_URL", "http://x/\"; evil = \"".to_string())],
|
|
);
|
|
assert!(
|
|
out.contains("\\\"") && !out.contains("/\";"),
|
|
"quotes in the value must be escaped:\n{out}"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
#[ignore = "mutates process env; races the other render_flake env tests under the parallel runner"]
|
|
fn render_flake_sets_service_url_options_from_forwarded_env() {
|
|
// The forwarded env vars must ALSO become option assignments, because
|
|
// the two are consumed at different times: the option is baked into
|
|
// scripts at build time (forge-avatar-sync's FORGE_URL), the env var is read at
|
|
// runtime. Emitting only the env var leaves the option on its default,
|
|
// which is how a hive ends up with two disagreeing answers for the same
|
|
// URL.
|
|
//
|
|
// Asserts PLACEMENT, not just presence: an option line rendered outside
|
|
// the per-agent module block would appear in the file and change
|
|
// nothing. It has to land before the environment blocks that close the
|
|
// module out.
|
|
//
|
|
// Serialised against every other env-mutating test in the crate:
|
|
// `render_flake_forwards_env_into_global_environment` sets HIVE_FORGE_URL
|
|
// too, and both land in the one test binary.
|
|
let _env = crate::test_env::lock();
|
|
// SAFETY: serialised by the guard above; restored before returning.
|
|
unsafe {
|
|
std::env::set_var("HIVE_FORGE_URL", "http://forge.example.test");
|
|
std::env::set_var("HIVE_MATRIX_URL", "http://matrix.example.test");
|
|
}
|
|
let out = render(
|
|
"git+https://forge.example/hyperhive/hyperhive.git?rev=0123abcd",
|
|
&[sample_spec("alice", 9001)],
|
|
);
|
|
unsafe {
|
|
std::env::remove_var("HIVE_FORGE_URL");
|
|
std::env::remove_var("HIVE_MATRIX_URL");
|
|
}
|
|
let forge_opt_at = out
|
|
.find("services.hyperhive.agent.forge.url = \"http://forge.example.test\"")
|
|
.expect("services.hyperhive.agent.forge.url must be rendered from HIVE_FORGE_URL");
|
|
let matrix_opt_at = out
|
|
.find("services.hyperhive.agent.matrix.url = \"http://matrix.example.test\"")
|
|
.expect("services.hyperhive.agent.matrix.url must be rendered from HIVE_MATRIX_URL");
|
|
let env_block_at = out
|
|
.find("environment.variables = {")
|
|
.expect("per-agent environment.variables block must exist");
|
|
assert!(
|
|
forge_opt_at < env_block_at && matrix_opt_at < env_block_at,
|
|
"service URL options must land inside the per-agent module block:\n{out}"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn render_flake_pins_each_agent_input_to_its_own_forge_rev() {
|
|
let at = |name, port, rev: &str| AgentSpec {
|
|
pin: Some(sample_pin(name, rev, Vec::new())),
|
|
..sample_spec(name, port)
|
|
};
|
|
let out = render(
|
|
"git+https://forge.example/hyperhive/hyperhive.git?rev=0123abcd",
|
|
&[at("alice", 9001, "aaaa"), at("bob", 9002, "bbbb")],
|
|
);
|
|
assert!(
|
|
out.contains(
|
|
"agent-alice.url = \"git+https://forge.example/agent-configs/alice.git?rev=aaaa\";"
|
|
),
|
|
"{out}"
|
|
);
|
|
assert!(
|
|
out.contains(
|
|
"agent-bob.url = \"git+https://forge.example/agent-configs/bob.git?rev=bbbb\";"
|
|
),
|
|
"{out}"
|
|
);
|
|
assert!(
|
|
!out.contains("git+file://"),
|
|
"no host-local path may remain in the flake:\n{out}"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn render_flake_refuses_a_custom_agent_without_a_pin() {
|
|
let args = |agents: &[AgentSpec]| {
|
|
render_flake(
|
|
"git+https://forge.example/hyperhive/hyperhive.git?rev=0123abcd",
|
|
agents,
|
|
)
|
|
};
|
|
let unpinned = AgentSpec {
|
|
pin: None,
|
|
..sample_spec("alice", 9001)
|
|
};
|
|
let err = args(&[unpinned]).expect_err("an unpinned agent must not render");
|
|
assert!(
|
|
format!("{err:#}").contains("agent alice: no pinned rev"),
|
|
"{err:#}"
|
|
);
|
|
// Presence control: the same agent with a pin renders.
|
|
assert!(args(&[sample_spec("alice", 9001)]).is_ok());
|
|
// An archetype agent has no config input and needs no pin.
|
|
let android = AgentSpec {
|
|
archetype: Some(swarm_queue_client::agent_archetype::ArchetypeRef {
|
|
source: "hyperhive".to_owned(),
|
|
name: "android-dev".to_owned(),
|
|
}),
|
|
pin: None,
|
|
..sample_spec("droid", 9002)
|
|
};
|
|
assert!(args(&[android]).is_ok());
|
|
}
|
|
|
|
/// A repo whose commits carry a `flake.lock`; `lock` is its content, `None`
|
|
/// for a commit without one. Returns the dir and the commit.
|
|
async fn commit_lock(dir: &Path, lock: Option<&str>) -> String {
|
|
let _ = std::fs::remove_file(dir.join("flake.lock"));
|
|
if let Some(lock) = lock {
|
|
std::fs::write(dir.join("flake.lock"), lock).expect("write lock");
|
|
}
|
|
git(dir, &["add", "-A"]).await.expect("add");
|
|
git(
|
|
dir,
|
|
&[
|
|
"-c",
|
|
"user.name=t",
|
|
"-c",
|
|
"user.email=t@t",
|
|
"commit",
|
|
"-q",
|
|
"--allow-empty",
|
|
"-m",
|
|
"c",
|
|
],
|
|
)
|
|
.await
|
|
.expect("commit");
|
|
lifecycle::git_rev_parse(dir, "HEAD").await.expect("HEAD")
|
|
}
|
|
|
|
async fn repo() -> tempfile::TempDir {
|
|
let dir = tempfile::tempdir().expect("tempdir");
|
|
git(dir.path(), &["init", "-q", "--initial-branch=main"])
|
|
.await
|
|
.expect("init");
|
|
dir
|
|
}
|
|
|
|
const NIXPKGS_LOCK: &str = r#"{"nodes":{"root":{"inputs":{"nixpkgs":"nixpkgs"}},"nixpkgs":{}},"root":"root","version":7}"#;
|
|
const NO_INPUTS_LOCK: &str = r#"{"nodes":{"root":{"inputs":{}}},"root":"root","version":7}"#;
|
|
const TEST_FORGE_REF: &str = "refs/test/forge-main";
|
|
const TEST_URL: &str = "https://forge.example/agent-configs/alice.git";
|
|
|
|
/// `pin_agent` where the forge fetch leaves `forge_ref` at `forge_sha`.
|
|
async fn pin_with_forge_at(repo: &Path, forge_sha: Option<&str>) -> Result<AgentPin> {
|
|
pin_agent("alice", repo, TEST_URL, TEST_FORGE_REF, async {
|
|
if let Some(sha) = forge_sha {
|
|
lifecycle::git_update_ref(repo, TEST_FORGE_REF, sha).await?;
|
|
}
|
|
Ok(repo.to_path_buf())
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn pin_agent_pins_forge_main_not_the_local_main() {
|
|
let dir = repo().await;
|
|
let forge = commit_lock(dir.path(), Some(NIXPKGS_LOCK)).await;
|
|
// Local main moves on past the forge's: its commit and lock are never read.
|
|
let local = commit_lock(dir.path(), Some(NO_INPUTS_LOCK)).await;
|
|
std::fs::write(dir.path().join("flake.lock"), NO_INPUTS_LOCK).expect("edit");
|
|
assert_ne!(forge, local);
|
|
|
|
let pin = pin_with_forge_at(dir.path(), Some(&forge))
|
|
.await
|
|
.expect("pinned");
|
|
assert_eq!(pin.url, format!("git+{TEST_URL}?rev={forge}"));
|
|
// The lock at the forge commit declares nixpkgs; local main's does not.
|
|
assert_eq!(pin.canonical_inputs, vec!["nixpkgs"]);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn two_agents_at_different_forge_mains_get_distinct_pins() {
|
|
let a = repo().await;
|
|
let b = repo().await;
|
|
let rev_a = commit_lock(a.path(), None).await;
|
|
let rev_b = commit_lock(b.path(), Some(NIXPKGS_LOCK)).await;
|
|
assert_ne!(rev_a, rev_b);
|
|
let pin_a = pin_with_forge_at(a.path(), Some(&rev_a)).await.expect("a");
|
|
let pin_b = pin_with_forge_at(b.path(), Some(&rev_b)).await.expect("b");
|
|
assert_ne!(pin_a.url, pin_b.url);
|
|
assert!(pin_a.url.ends_with(&rev_a) && pin_b.url.ends_with(&rev_b));
|
|
// No lock at a's commit, so no follows; b's lock declares nixpkgs.
|
|
assert!(pin_a.canonical_inputs.is_empty());
|
|
assert_eq!(pin_b.canonical_inputs, vec!["nixpkgs"]);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn pin_agent_errors_when_the_forge_is_unreachable_or_has_no_main() {
|
|
let dir = repo().await;
|
|
let rev = commit_lock(dir.path(), None).await;
|
|
|
|
let unreachable = pin_agent("alice", dir.path(), TEST_URL, TEST_FORGE_REF, async {
|
|
anyhow::bail!("forge down")
|
|
})
|
|
.await
|
|
.expect_err("unreachable forge");
|
|
let msg = format!("{unreachable:#}");
|
|
assert!(
|
|
msg.contains("agent alice") && msg.contains("forge down"),
|
|
"{msg}"
|
|
);
|
|
|
|
// The fetch succeeds but leaves no forge main.
|
|
let no_main = pin_with_forge_at(dir.path(), None)
|
|
.await
|
|
.expect_err("no main on the forge");
|
|
let msg = format!("{no_main:#}");
|
|
assert!(
|
|
msg.contains("agent alice") && msg.contains("no main"),
|
|
"{msg}"
|
|
);
|
|
|
|
// Presence control: with a forge main, the same call pins it.
|
|
let pin = pin_with_forge_at(dir.path(), Some(&rev))
|
|
.await
|
|
.expect("pinned");
|
|
assert!(pin.url.ends_with(&rev));
|
|
}
|
|
|
|
#[test]
|
|
fn render_flake_embeds_no_ca() {
|
|
// Agents fetch the swarm root from the store at boot; the flake names
|
|
// no CA file whatever the host env says, including the variable
|
|
// hive-c0re used to embed from.
|
|
//
|
|
// Serialised against every other env-mutating test in the crate.
|
|
let _env = crate::test_env::lock();
|
|
let render = || {
|
|
render(
|
|
"git+https://forge.example/hyperhive/hyperhive.git?rev=0123abcd",
|
|
&[sample_spec("alice", 9001)],
|
|
)
|
|
};
|
|
// SAFETY: serialised by the guard at the top of this test; restored
|
|
// before returning.
|
|
unsafe {
|
|
std::env::set_var("HIVE_TLS_CA_PATH", "/nonexistent/trust-bundle.pem");
|
|
}
|
|
let flake = render();
|
|
unsafe {
|
|
std::env::remove_var("HIVE_TLS_CA_PATH");
|
|
}
|
|
// Presence control: the render does carry per-agent module config, so
|
|
// the absences below are not an empty render.
|
|
assert!(
|
|
flake.contains("services.hyperhive.agent."),
|
|
"per-agent module block missing:\n{flake}"
|
|
);
|
|
for needle in ["hive-ca.pem", "certificateFiles", "peer-ca-"] {
|
|
assert!(
|
|
!flake.contains(needle),
|
|
"the flake must not name {needle}:\n{flake}"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn stale_ca_files_are_removed_and_nothing_else() {
|
|
let dir = tempfile::tempdir().expect("tempdir");
|
|
for name in ["hive-ca.pem", "peer-ca-0.pem", "peer-ca-1.PEM", "flake.nix"] {
|
|
std::fs::write(dir.path().join(name), "x").expect("write");
|
|
}
|
|
// Not CA files this machinery owns, though they look similar.
|
|
for name in ["peer-ca-0.txt", "other-ca.pem"] {
|
|
std::fs::write(dir.path().join(name), "x").expect("write");
|
|
}
|
|
let mut removed = remove_stale_ca_files(dir.path());
|
|
removed.sort();
|
|
assert_eq!(removed, ["hive-ca.pem", "peer-ca-0.pem", "peer-ca-1.PEM"]);
|
|
for kept in ["flake.nix", "peer-ca-0.txt", "other-ca.pem"] {
|
|
assert!(dir.path().join(kept).exists(), "{kept} must survive");
|
|
}
|
|
assert!(remove_stale_ca_files(dir.path()).is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn services_pki_mount_is_forwarded_as_both_env_and_option() {
|
|
assert!(FORWARDED_VARS.contains(&"HIVE_AGENT_BAO_SERVICES_PKI_MOUNT"));
|
|
let mut out = String::new();
|
|
push_forwarded_var_options(
|
|
&mut out,
|
|
&[("HIVE_AGENT_BAO_SERVICES_PKI_MOUNT", "pki".to_string())],
|
|
);
|
|
assert_eq!(
|
|
out,
|
|
" services.hyperhive.agent.bao.servicesPkiMount = \"pki\";\n"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn render_flake_carries_no_otel_config_and_no_per_host_value() {
|
|
// The agent module defaults otel on and finds the collector by name, so
|
|
// the flake carries no enable, endpoint or protocol line however the
|
|
// host's OTEL env is set. The resource labels, dashboard port and
|
|
// pinned claude differ per host and reach the agent at container start
|
|
// (`lifecycle::hive_facts`), so none of them is rendered either.
|
|
//
|
|
// Serialised against every other env-mutating test in the crate.
|
|
// `stats::otel_metrics::tests` perturbs OTEL env too, and lands in this
|
|
// same test binary — "no other test asserts on these" was true of this
|
|
// module and false of the process.
|
|
let _env = crate::test_env::lock();
|
|
// SAFETY: serialised by the guard above; restored before returning.
|
|
let render = || {
|
|
render(
|
|
"git+https://forge.example/hyperhive/hyperhive.git?rev=0123abcd",
|
|
&[sample_spec("alice", 9001)],
|
|
)
|
|
};
|
|
unsafe {
|
|
std::env::remove_var("HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES");
|
|
std::env::set_var("HYPERHIVE_OTEL_ENDPOINT", "https://c.example/otel");
|
|
std::env::set_var("HYPERHIVE_OTEL_PROTOCOL", "grpc");
|
|
}
|
|
let plain = render();
|
|
unsafe {
|
|
std::env::set_var(
|
|
"HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES",
|
|
"deployment.environment=prod",
|
|
);
|
|
}
|
|
let labelled = render();
|
|
unsafe {
|
|
std::env::remove_var("HYPERHIVE_OTEL_ENDPOINT");
|
|
std::env::remove_var("HYPERHIVE_OTEL_PROTOCOL");
|
|
std::env::remove_var("HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES");
|
|
}
|
|
|
|
// Presence controls: the render is the real flake, carrying the pin and
|
|
// the harness env block the dashboard port used to sit in.
|
|
assert!(
|
|
plain.contains("rev=0123abcd") && plain.contains("HIVE_PORT = toString port;"),
|
|
"the hyperhive pin and the harness env must be rendered:\n{plain}"
|
|
);
|
|
for (name, flake) in [("plain", &plain), ("labelled", &labelled)] {
|
|
for needle in [
|
|
"agent.otel.enable",
|
|
"agent.otel.endpoint",
|
|
"agent.otel.protocol",
|
|
"c.example",
|
|
"extraResourceAttributes",
|
|
"deployment.environment",
|
|
"claudeCodePath",
|
|
"dashboardPort",
|
|
"HIVE_DASHBOARD_PORT",
|
|
] {
|
|
assert!(
|
|
!flake.contains(needle),
|
|
"{name}: `{needle}` must not be rendered:\n{flake}"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
#[test]
|
|
fn render_flake_injects_github_disable_only_when_signalled() {
|
|
// services.hyperhive.swarm.github.enable = false -> HYPERHIVE_GITHUB_DISABLED
|
|
// on hive-c0re's unit -> `services.hyperhive.agent.github.enable = false` injected into
|
|
// every agent. On by default, so nothing is emitted unless disabled.
|
|
//
|
|
// Serialised against every other env-mutating test in the crate — the
|
|
// variable is this module's alone, but the process it lives in is not.
|
|
let _env = crate::test_env::lock();
|
|
// SAFETY: serialised by the guard above; restored before returning.
|
|
let render = || {
|
|
render(
|
|
"git+https://forge.example/hyperhive/hyperhive.git?rev=0123abcd",
|
|
&[sample_spec("alice", 9001)],
|
|
)
|
|
};
|
|
unsafe {
|
|
std::env::remove_var("HYPERHIVE_GITHUB_DISABLED");
|
|
}
|
|
let on_default = render();
|
|
unsafe {
|
|
std::env::set_var("HYPERHIVE_GITHUB_DISABLED", "1");
|
|
}
|
|
let disabled = render();
|
|
unsafe {
|
|
std::env::remove_var("HYPERHIVE_GITHUB_DISABLED");
|
|
}
|
|
assert!(
|
|
!on_default.contains("services.hyperhive.agent.github.enable"),
|
|
"github.enable must not be emitted by default (agents keep the true default):\n{on_default}"
|
|
);
|
|
assert!(
|
|
disabled.contains("services.hyperhive.agent.github.enable = false;"),
|
|
"github.enable = false must be injected when the host disables it:\n{disabled}"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod lock_wait_tests {
|
|
use super::*;
|
|
|
|
/// A waiter behind a held lock is told, per interval, who holds it; once
|
|
/// the holder drops, it gets the lock and becomes the recorded holder.
|
|
#[tokio::test(start_paused = true)]
|
|
async fn stalled_waiter_reports_holder_then_acquires() {
|
|
let lock = Mutex::new(());
|
|
let holder = StdMutex::new("");
|
|
let first = lock_notifying(&lock, &holder, "first", Duration::from_mins(1), |_, _| {
|
|
panic!("an uncontended lock must not report a stall")
|
|
})
|
|
.await;
|
|
assert_eq!(*holder.lock().unwrap(), "first");
|
|
|
|
let stalls = std::cell::RefCell::new(Vec::new());
|
|
let waiter = lock_notifying(&lock, &holder, "second", Duration::from_mins(1), |h, w| {
|
|
stalls.borrow_mut().push((h, w.as_secs()));
|
|
});
|
|
tokio::pin!(waiter);
|
|
// 150s of virtual time: two intervals elapse while `first` is held.
|
|
tokio::select! {
|
|
_ = &mut waiter => panic!("lock is held"),
|
|
() = tokio::time::sleep(Duration::from_secs(150)) => {}
|
|
}
|
|
drop(first);
|
|
assert_eq!(*holder.lock().unwrap(), "");
|
|
let second = waiter.await;
|
|
assert_eq!(*holder.lock().unwrap(), "second");
|
|
drop(second);
|
|
assert_eq!(*stalls.borrow(), vec![("first", 60), ("first", 120)]);
|
|
}
|
|
|
|
/// Waiters get the lock in arrival order even when one of them stalls past
|
|
/// an interval: A queues at 0s, B at 30s, the holder releases at 70s. A
|
|
/// re-armed acquire would put A (stalled at 60s) behind B.
|
|
#[tokio::test(start_paused = true)]
|
|
async fn waiters_acquire_in_arrival_order_across_a_stall() {
|
|
use std::sync::Arc;
|
|
static LOCK: Mutex<()> = Mutex::const_new(());
|
|
static HOLDER: StdMutex<&'static str> = StdMutex::new("");
|
|
let order = Arc::new(StdMutex::new(Vec::new()));
|
|
let first = lock_notifying(&LOCK, &HOLDER, "first", Duration::from_mins(1), |_, _| {
|
|
panic!("an uncontended lock must not report a stall")
|
|
})
|
|
.await;
|
|
let spawn_waiter = |op: &'static str| {
|
|
let order = Arc::clone(&order);
|
|
tokio::spawn(async move {
|
|
let guard =
|
|
lock_notifying(&LOCK, &HOLDER, op, Duration::from_mins(1), |_, _| {}).await;
|
|
order.lock().unwrap().push(op);
|
|
drop(guard);
|
|
})
|
|
};
|
|
let a = spawn_waiter("A");
|
|
tokio::time::sleep(Duration::from_secs(30)).await;
|
|
let b = spawn_waiter("B");
|
|
tokio::time::sleep(Duration::from_secs(40)).await;
|
|
drop(first);
|
|
a.await.expect("A");
|
|
b.await.expect("B");
|
|
assert_eq!(*order.lock().unwrap(), vec!["A", "B"]);
|
|
}
|
|
|
|
/// A hung nix child comes back as a timeout error naming the command. The
|
|
/// deadline and program are injected; the production const is untouched.
|
|
#[tokio::test]
|
|
async fn nix_output_times_out_naming_the_command() {
|
|
let dir = tempfile::tempdir().expect("tempdir");
|
|
let fake = dir.path().join("nix");
|
|
std::fs::write(&fake, "#!/bin/sh\nexec sleep 30\n").expect("write");
|
|
std::fs::set_permissions(&fake, std::os::unix::fs::PermissionsExt::from_mode(0o755))
|
|
.expect("chmod");
|
|
// Parallel tests fork while the script's write fd is still open, so an
|
|
// early exec can hit ETXTBSY. That is a race in building the fake, not
|
|
// in `nix_output_with`: retry it, anything else fails the test.
|
|
let mut tries = 0;
|
|
let err = loop {
|
|
let res = nix_output_with(
|
|
fake.as_os_str(),
|
|
dir.path(),
|
|
&["flake", "update", "hyperhive"],
|
|
Duration::from_millis(300),
|
|
)
|
|
.await;
|
|
match res {
|
|
Err(e) if tries < 20 && format!("{e:#}").contains("Text file busy") => {
|
|
tries += 1;
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
}
|
|
other => break other.expect_err("hung nix must time out"),
|
|
}
|
|
};
|
|
let msg = format!("{err:#}");
|
|
assert!(
|
|
msg.contains("nix flake update hyperhive in") && msg.contains("timed out after 0.3s"),
|
|
"{msg}"
|
|
);
|
|
}
|
|
}
|