1021 lines
41 KiB
Rust
1021 lines
41 KiB
Rust
//! `nixos-container` lifecycle + per-agent config flake generation.
|
|
|
|
mod git;
|
|
mod host_config;
|
|
mod setup;
|
|
#[cfg(test)]
|
|
mod tests;
|
|
|
|
pub use git::{
|
|
git, git_command, git_fetch_to_tag, git_read_tree_reset, git_rev_parse, git_tag,
|
|
git_tag_annotated, git_update_ref,
|
|
};
|
|
pub use host_config::{
|
|
CONTAINER_MANAGER_AGENTS_MOUNT, CONTAINER_MANAGER_APPLIED_MOUNT, write_dropins,
|
|
};
|
|
pub use setup::{
|
|
ensure_agent_state_subvolume, ensure_claude_dir, ensure_state_dir, initial_flake_nix,
|
|
setup_applied, setup_proposed,
|
|
};
|
|
|
|
use std::path::Path;
|
|
|
|
use anyhow::{Context, Result, bail};
|
|
use tokio::process::Command;
|
|
|
|
use crate::coordinator::{AgentPaths, HiveEnv};
|
|
|
|
/// Sub-agent container prefix. `nixos-container` caps the total container name
|
|
/// at 11 chars (it gets encoded into network interface names), so the agent
|
|
/// name itself can be at most `MAX_AGENT_NAME` chars.
|
|
pub const AGENT_PREFIX: &str = "h-";
|
|
pub const MAX_AGENT_NAME: usize = 9;
|
|
/// Logical name of the manager agent (broker recipient, state-dir key,
|
|
/// meta flake attribute). All persistent state lives under `ruth/`.
|
|
pub const MANAGER_NAME: &str = "ruth";
|
|
/// Container name of the manager. Uses the same `h-` prefix as sub-agents
|
|
/// so `nixos-container list` output is uniform and the list filter is
|
|
/// a single `starts_with(AGENT_PREFIX)` check. Logical name → container
|
|
/// name: `ruth` → `h-ruth`.
|
|
pub const MANAGER_CONTAINER: &str = "h-ruth";
|
|
|
|
/// Mount point of the per-agent runtime directory inside the container.
|
|
pub const CONTAINER_RUNTIME_MOUNT: &str = "/run/hive";
|
|
|
|
/// Where the per-agent Claude credentials dir mounts inside the
|
|
/// container. The harness service runs as a non-root unix user
|
|
/// whose home is `/home/<agent>/`, so the mount path varies per
|
|
/// agent — `container_claude_mount(name)` returns
|
|
/// `/home/<name>/.claude` for every agent including the manager.
|
|
/// `claude` inside the container reads
|
|
/// `$HOME/.claude` and the service environment sets `HOME` to the
|
|
/// same path, so the OAuth session survives container restarts.
|
|
#[must_use]
|
|
pub fn container_claude_mount(name: &str) -> String {
|
|
format!("/home/{name}/.claude")
|
|
}
|
|
|
|
/// Mount point of the shared directory accessible to all agents.
|
|
/// All agents can read/write here; agents should only put things they're
|
|
/// willing to lose (other agents may delete them).
|
|
pub const CONTAINER_SHARED_MOUNT: &str = "/shared";
|
|
|
|
/// Sub-agent web UI port range. Deterministic from the agent's name (FNV-1a
|
|
/// hash mod range size), so the dashboard can compute the same port without
|
|
/// asking hive-c0re.
|
|
const WEB_PORT_BASE: u16 = 8100;
|
|
const WEB_PORT_RANGE: u16 = 900;
|
|
|
|
/// FNV-1a hash of a string — shared by `agent_web_port` and
|
|
/// `agent_network_ip` so the derivation rule is identical.
|
|
fn fnv1a(s: &str) -> u32 {
|
|
let mut hash: u32 = 2_166_136_261;
|
|
for b in s.bytes() {
|
|
hash ^= u32::from(b);
|
|
hash = hash.wrapping_mul(16_777_619);
|
|
}
|
|
hash
|
|
}
|
|
|
|
/// Per-agent web UI port — `WEB_PORT_BASE + FNV-1a(name) %
|
|
/// WEB_PORT_RANGE` for every agent including the manager. The port
|
|
/// allocation rule reads the same for every name; collisions are
|
|
/// possible (birthday paradox at ~30 agents) and the operator
|
|
/// resolves them by renaming an agent (different hash → different
|
|
/// port). Stable across hosts, restarts, and dashboard renders —
|
|
/// no state-file dance.
|
|
#[must_use]
|
|
pub fn agent_web_port(name: &str) -> u16 {
|
|
// Modulo of a u32 by a u16's value is guaranteed < u16::MAX, so try_from never fails.
|
|
WEB_PORT_BASE + u16::try_from(fnv1a(name) % u32::from(WEB_PORT_RANGE)).unwrap_or(0)
|
|
}
|
|
|
|
/// Deterministic IPv4 address for an agent inside an isolated subnet.
|
|
///
|
|
/// Parses `subnet_cidr` as `<network_ip>/<prefix_len>` (e.g.
|
|
/// `"10.42.0.0/24"`), then computes:
|
|
///
|
|
/// ```text
|
|
/// host_count = 2^(32 - prefix_len)
|
|
/// usable = host_count - 3 // skip .0 (network), .1 (gateway), .255 (broadcast)
|
|
/// offset = FNV-1a(name) % usable + 2 // .2 is the first agent slot
|
|
/// agent_ip = network_base_u32 + offset
|
|
/// ```
|
|
///
|
|
/// Returns `None` when `subnet_cidr` can't be parsed (invalid format,
|
|
/// prefix out of range, etc.) so callers can fall back gracefully.
|
|
/// Collisions are possible (birthday paradox) and the operator resolves
|
|
/// them by renaming an agent, same as for port collisions.
|
|
#[must_use]
|
|
pub fn agent_network_ip(name: &str, subnet_cidr: &str) -> Option<String> {
|
|
let (ip_str, prefix_str) = subnet_cidr.split_once('/')?;
|
|
let prefix_len: u32 = prefix_str.parse().ok()?;
|
|
if prefix_len > 30 {
|
|
// /31 and /32 have no room for agents; /30 has 1 usable slot.
|
|
// /0 (the other extreme) is handled further down: host_count
|
|
// overflows checked_shl(32) → 0 → usable = 0 → None.
|
|
return None;
|
|
}
|
|
// Parse dotted-decimal IPv4.
|
|
let octets: Vec<u8> = ip_str
|
|
.split('.')
|
|
.map(|o| o.parse::<u8>().ok())
|
|
.collect::<Option<Vec<_>>>()?;
|
|
if octets.len() != 4 {
|
|
return None;
|
|
}
|
|
let base_u32 = u32::from_be_bytes([octets[0], octets[1], octets[2], octets[3]]);
|
|
// Mask off host bits to get the true network address.
|
|
let mask = if prefix_len == 0 {
|
|
0u32
|
|
} else {
|
|
!0u32 << (32 - prefix_len)
|
|
};
|
|
let network_base = base_u32 & mask;
|
|
let host_count: u32 = 1u32.checked_shl(32 - prefix_len).unwrap_or(0);
|
|
// `.0` = network, `.1` = bridge gateway, last = broadcast → 3 reserved.
|
|
let usable = host_count.saturating_sub(3);
|
|
if usable == 0 {
|
|
return None;
|
|
}
|
|
let offset = fnv1a(name) % usable + 2; // +2: skip .0 and .1
|
|
let ip_u32 = network_base + offset;
|
|
let [a, b, c, d] = ip_u32.to_be_bytes();
|
|
Some(format!("{a}.{b}.{c}.{d}"))
|
|
}
|
|
|
|
/// Extract the bridge gateway IP from `HIVE_NETWORK_SUBNET`.
|
|
///
|
|
/// `HIVE_NETWORK_SUBNET` carries the host-side bridge address verbatim
|
|
/// (e.g. `10.42.0.1/24`), **not** the canonical network address — see
|
|
/// the note in `set_nspawn_flags` + `docs/network.md`. The IP part is
|
|
/// therefore the bridge IP itself: the host end of the bridge, the
|
|
/// default-route target for isolated containers, and the address the
|
|
/// hive dnsmasq resolver binds. Returns the dotted-decimal IP with the
|
|
/// `/<prefix>` stripped, or `None` if the input isn't a valid
|
|
/// `<ipv4>/<prefix>` pair.
|
|
///
|
|
/// Deliberately returns the operator-configured address verbatim rather
|
|
/// than deriving `network + 1`: an operator who sets `bridgeIp` to a
|
|
/// non-`.1` host address (e.g. `10.42.0.254`) runs the bridge + resolver
|
|
/// there, so that — not `.1` — is the real gateway.
|
|
#[must_use]
|
|
pub fn bridge_gateway_ip(subnet_cidr: &str) -> Option<String> {
|
|
let (ip_str, prefix_str) = subnet_cidr.split_once('/')?;
|
|
// Validate the prefix is a sane IPv4 CIDR length and the address is
|
|
// dotted-decimal IPv4 — same shape `agent_network_ip` accepts — so a
|
|
// malformed `HIVE_NETWORK_SUBNET` can't smuggle a bogus HOST_ADDRESS
|
|
// into the nspawn conf.
|
|
let prefix_len: u32 = prefix_str.parse().ok()?;
|
|
if prefix_len > 32 {
|
|
return None;
|
|
}
|
|
let octets: Vec<u8> = ip_str
|
|
.split('.')
|
|
.map(|o| o.parse::<u8>().ok())
|
|
.collect::<Option<Vec<_>>>()?;
|
|
if octets.len() != 4 {
|
|
return None;
|
|
}
|
|
Some(ip_str.to_owned())
|
|
}
|
|
|
|
#[must_use]
|
|
pub fn container_name(name: &str) -> String {
|
|
format!("{AGENT_PREFIX}{name}")
|
|
}
|
|
|
|
/// Read the agent user's `(uid, gid)` from the container's nixos-managed
|
|
/// `/etc/passwd`. Returns `None` when the container hasn't been built
|
|
/// yet, the passwd file is unparseable, or the agent user is missing
|
|
/// (e.g. legacy container that still runs as root).
|
|
///
|
|
/// Used by `forge` + `matrix` after writing per-agent state files so
|
|
/// the bind-mounted host file ends up readable by the agent user
|
|
/// without waiting for the next container activation to run the chown
|
|
/// fixup.
|
|
///
|
|
/// Notes:
|
|
/// - Reads the *container-local* passwd at
|
|
/// `/var/lib/nixos-containers/<container>/etc/passwd`, not the host's.
|
|
/// The container's user-namespace shares uids with the host (no
|
|
/// `PrivateUsers`), so the uid is directly usable in host-side
|
|
/// `chown(2)`.
|
|
/// - Best-effort: caller treats `None` as "skip the chown".
|
|
#[must_use]
|
|
pub fn agent_uid_gid(agent_name: &str) -> Option<(u32, u32)> {
|
|
let container = container_name(agent_name);
|
|
let passwd_path = format!("/var/lib/nixos-containers/{container}/etc/passwd");
|
|
let content = std::fs::read_to_string(&passwd_path).ok()?;
|
|
for line in content.lines() {
|
|
let mut parts = line.split(':');
|
|
let user = parts.next()?;
|
|
if user != agent_name {
|
|
continue;
|
|
}
|
|
let _ = parts.next()?; // x (password placeholder)
|
|
let uid: u32 = parts.next()?.parse().ok()?;
|
|
let gid: u32 = parts.next()?.parse().ok()?;
|
|
return Some((uid, gid));
|
|
}
|
|
None
|
|
}
|
|
|
|
/// Best-effort `chown(path, agent_uid, agent_gid)`. Resolves the agent's
|
|
/// uid/gid via [`agent_uid_gid`] and shells out to `std::os::unix::fs::chown`.
|
|
/// Silently no-ops when the container isn't built yet (`None` from
|
|
/// [`agent_uid_gid`]) and logs at debug on chown syscall failure — the
|
|
/// activation script in `harness-base.nix` is the steady-state safety
|
|
/// net. Used by per-agent state writers in `forge` + `matrix` so the
|
|
/// agent can read the file without waiting for the next container
|
|
/// rebuild.
|
|
pub fn chown_to_agent(name: &str, path: &Path, subsystem: &str) {
|
|
let Some((uid, gid)) = agent_uid_gid(name) else {
|
|
return;
|
|
};
|
|
if let Err(e) = std::os::unix::fs::chown(path, Some(uid), Some(gid)) {
|
|
tracing::debug!(%name, %subsystem, path = %path.display(), error = %e, "chown to agent failed");
|
|
}
|
|
}
|
|
|
|
fn validate(name: &str) -> Result<()> {
|
|
if name.is_empty() {
|
|
bail!("agent name must not be empty");
|
|
}
|
|
if name.len() > MAX_AGENT_NAME {
|
|
bail!(
|
|
"agent name '{name}' is too long ({} chars); max {MAX_AGENT_NAME}",
|
|
name.len()
|
|
);
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// First name (≠ `self_name`) currently running whose hashed port
|
|
/// matches this agent's. The harness inside the colliding container
|
|
/// would otherwise loop on `AddrInUse` forever; we surface the
|
|
/// conflict here so spawn / rebuild fails loudly with an actionable
|
|
/// message instead.
|
|
async fn port_collision(self_name: &str) -> Option<String> {
|
|
let port = agent_web_port(self_name);
|
|
let raw = list().await.unwrap_or_default();
|
|
for c in raw {
|
|
let Some(other) = c.strip_prefix(AGENT_PREFIX) else {
|
|
continue;
|
|
};
|
|
if other == self_name {
|
|
continue;
|
|
}
|
|
if agent_web_port(other) == port && is_running(other).await {
|
|
return Some(other.to_owned());
|
|
}
|
|
}
|
|
None
|
|
}
|
|
|
|
pub async fn spawn(name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()> {
|
|
create_container(name, hive, paths).await?;
|
|
// Runtime dir must exist before nixos-container start (nspawn bind-mount
|
|
// source). Create it here so callers don't need a separate preamble step.
|
|
ensure_agent_runtime_dir(name)?;
|
|
write_dropins(name, hive, paths).await?;
|
|
priv_run("start", name).await
|
|
}
|
|
|
|
/// First-spawn provisioning + `nixos-container create`, without the
|
|
/// drop-in write or the start — the job queue's `Create` node.
|
|
/// `spawn` composes this with `write_dropins` + start for direct
|
|
/// callers (root-agent bootstrap).
|
|
pub async fn create_container(name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()> {
|
|
validate(name)?;
|
|
if let Some(other) = port_collision(name).await {
|
|
bail!(
|
|
"port {} is already taken by '{other}' — rename one of them and retry",
|
|
agent_web_port(name)
|
|
);
|
|
}
|
|
setup_proposed(&paths.proposed_dir, name).await?;
|
|
setup_applied(&paths.applied_dir, Some(&paths.proposed_dir), name).await?;
|
|
ensure_agent_state_subvolume(name).await?;
|
|
ensure_claude_dir(&paths.claude_dir)?;
|
|
ensure_state_dir(&paths.notes_dir)?;
|
|
// Meta flake gets the new agent's input + nixosConfiguration
|
|
// before `nixos-container create` so the `--flake meta#<name>`
|
|
// ref resolves.
|
|
let agents = agents_after_spawn(name).await?;
|
|
crate::meta::sync_agents(hive, &agents).await?;
|
|
priv_run("create", name).await
|
|
}
|
|
|
|
/// Rebuild-path preamble shared by the job queue's `Prebuild` node and
|
|
/// `rebuild_no_meta`: fail fast on a port collision, then make sure
|
|
/// the applied repo + state dirs exist. Container untouched.
|
|
pub async fn prepare_rebuild_dirs(name: &str, paths: &AgentPaths) -> Result<()> {
|
|
validate(name)?;
|
|
if let Some(other) = port_collision(name).await {
|
|
bail!(
|
|
"port {} is already taken by '{other}' — rename one of them and retry",
|
|
agent_web_port(name)
|
|
);
|
|
}
|
|
setup_applied(&paths.applied_dir, None, name).await?;
|
|
ensure_agent_state_subvolume(name).await?;
|
|
ensure_claude_dir(&paths.claude_dir)?;
|
|
ensure_state_dir(&paths.notes_dir)?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Profile-swap for an existing, stopped container: re-apply the
|
|
/// drop-ins, then `nixos-container update`. The job queue's `Swap`
|
|
/// node. Requires the container stopped (the queue's `StopForUpdate`
|
|
/// upstream); does NOT start it — the DAG's tail `Reconcile` owns
|
|
/// bringing the agent back to its wanted power state.
|
|
pub async fn swap_update(
|
|
name: &str,
|
|
hive: &HiveEnv,
|
|
paths: &AgentPaths,
|
|
on_step: &(dyn Fn(&str) + Send + Sync),
|
|
on_build_log_id: &(dyn Fn(i64) + Send + Sync),
|
|
) -> Result<()> {
|
|
write_dropins(name, hive, paths).await?;
|
|
on_step("nixos-container update");
|
|
priv_run_inner("update", name, Some(on_build_log_id)).await
|
|
}
|
|
|
|
/// Build the `AgentSpec` list for the meta flake from `nixos-container
|
|
/// list` + a hypothetical extra name not yet in the list (for spawn
|
|
/// where the new agent's container doesn't exist yet). Pass empty
|
|
/// `name_to_add` from rebuild paths where the agent is already in the
|
|
/// container list.
|
|
///
|
|
/// Propagates errors from `list()` rather than swallowing them.
|
|
/// Using `.unwrap_or_default()` here would silently produce an empty
|
|
/// agent list when `nixos-container list` fails (priv helper down, race),
|
|
/// which `sync_agents` would then commit to meta — dropping every agent
|
|
/// from `flake.nix`. Callers that can tolerate failures (e.g. migration)
|
|
/// handle the `Err` themselves with `.unwrap_or_default()`.
|
|
async fn agents_for_meta(name_to_add: Option<&str>) -> Result<Vec<crate::meta::AgentSpec>> {
|
|
let containers = list().await?;
|
|
let mut out: Vec<crate::meta::AgentSpec> = containers
|
|
.into_iter()
|
|
.filter_map(|c| {
|
|
let name = c.strip_prefix(AGENT_PREFIX)?.to_owned();
|
|
Some(crate::meta::AgentSpec {
|
|
is_manager: name == MANAGER_NAME,
|
|
port: agent_web_port(&name),
|
|
name,
|
|
})
|
|
})
|
|
.collect();
|
|
if let Some(extra) = name_to_add
|
|
&& !out.iter().any(|a| a.name == extra)
|
|
{
|
|
out.push(crate::meta::AgentSpec {
|
|
is_manager: extra == MANAGER_NAME,
|
|
port: agent_web_port(extra),
|
|
name: extra.to_owned(),
|
|
});
|
|
}
|
|
out.sort_by(|a, b| a.name.cmp(&b.name));
|
|
Ok(out)
|
|
}
|
|
|
|
async fn agents_after_spawn(name: &str) -> Result<Vec<crate::meta::AgentSpec>> {
|
|
agents_for_meta(Some(name)).await
|
|
}
|
|
|
|
/// Like `agents_for_meta_listing` but with an extra agent added (for a
|
|
/// container that doesn't exist yet). Used by the first-spawn path in
|
|
/// `actions::run_apply_commit` to register the new agent in meta before
|
|
/// `prepare_deploy` tries to update its input lock.
|
|
pub async fn agents_for_meta_listing_with(extra: &str) -> Result<Vec<crate::meta::AgentSpec>> {
|
|
agents_for_meta(Some(extra)).await
|
|
}
|
|
|
|
/// Public enumeration of currently-existing agents (whatever
|
|
/// `nixos-container list` says), sorted, no extras. For callers
|
|
/// outside this module that need to reseed meta after lifecycle
|
|
/// changes — destroy, startup reconciliation, etc.
|
|
pub async fn agents_for_meta_listing() -> Result<Vec<crate::meta::AgentSpec>> {
|
|
agents_for_meta(None).await
|
|
}
|
|
|
|
/// True when the named container already exists (appears in
|
|
/// `nixos-container list`). Used by the apply-commit path to decide
|
|
/// between first-spawn (`nixos-container create`) and normal rebuild
|
|
/// (`nixos-container update`).
|
|
pub async fn container_exists(name: &str) -> bool {
|
|
let container = container_name(name);
|
|
list()
|
|
.await
|
|
.unwrap_or_default()
|
|
.iter()
|
|
.any(|c| c == &container)
|
|
}
|
|
|
|
pub async fn kill(name: &str) -> Result<()> {
|
|
validate(name)?;
|
|
priv_run("stop", name).await
|
|
}
|
|
|
|
/// Start a container. Success is defined as the container's unit reaching
|
|
/// `active`, **not** the `nixos-container start` exit code — a start-job
|
|
/// timeout on a slow boot (e.g. DHCP taking ~10s) exits non-zero while
|
|
/// `container@<c>.service` keeps retrying and the container comes up seconds
|
|
/// later. So a zero exit returns immediately (already active), and a non-zero
|
|
/// exit polls the unit state for [`START_SETTLE_TIMEOUT`] before concluding
|
|
/// the start actually failed. Every caller (dashboard restart, reconcile
|
|
/// start, the cold-start fallback) gets this truth for free.
|
|
pub async fn start(name: &str) -> Result<()> {
|
|
validate(name)?;
|
|
if priv_run("start", name).await.is_ok() {
|
|
return Ok(());
|
|
}
|
|
let container = container_name(name);
|
|
if wait_until_running(name, START_SETTLE_TIMEOUT).await {
|
|
tracing::info!(
|
|
container = %container,
|
|
"start exited non-zero but the container reached active on systemd's retry; treating as success"
|
|
);
|
|
return Ok(());
|
|
}
|
|
Err(anyhow::anyhow!(
|
|
"container {container} did not reach active within {}s of start",
|
|
START_SETTLE_TIMEOUT.as_secs()
|
|
))
|
|
}
|
|
|
|
/// Opaque token produced by [`converge_start_preamble`].
|
|
/// [`start_with_fallback`] requires this as proof that the pre-start
|
|
/// preamble (runtime dir + drop-ins) ran. Dropping the token without
|
|
/// calling `start_with_fallback` is a no-op.
|
|
#[must_use = "call lifecycle::start_with_fallback(token) to start the container"]
|
|
pub struct StartableAgent {
|
|
name: String,
|
|
}
|
|
|
|
/// Run the per-agent start preamble: ensure the runtime dir exists and write
|
|
/// the nspawn / resource-limits drop-ins. Returns a [`StartableAgent`] token
|
|
/// as typed proof that the preamble ran; pass it to [`start_with_fallback`].
|
|
/// Callers that omit this step cannot call `start_with_fallback` — the type
|
|
/// system makes forgetting the preamble a compile error.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Returns an error if `ensure_agent_runtime_dir` or `write_dropins` fails.
|
|
pub async fn converge_start_preamble(
|
|
name: &str,
|
|
hive: &HiveEnv,
|
|
paths: &AgentPaths,
|
|
) -> Result<StartableAgent> {
|
|
ensure_agent_runtime_dir(name)?;
|
|
write_dropins(name, hive, paths).await?;
|
|
Ok(StartableAgent {
|
|
name: name.to_owned(),
|
|
})
|
|
}
|
|
|
|
/// Start with the cold-start fallback: when a plain start fails (the
|
|
/// activation-error shape), retry once via stop + kill + start before
|
|
/// giving up. Used by the queue's fast-lane `Start` handler and the
|
|
/// inline start-after-rebuild path.
|
|
/// See `docs/coordinator.md::Cold-start fallback`.
|
|
///
|
|
/// Requires a [`StartableAgent`] token from [`converge_start_preamble`]
|
|
/// to prove the preamble ran. For internal use within this module (where
|
|
/// the preamble is already enforced structurally) call
|
|
/// `start_with_fallback_inner` directly.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates the retry's start error (annotated with the original
|
|
/// failure) when the fallback also fails.
|
|
pub async fn start_with_fallback(token: StartableAgent) -> Result<()> {
|
|
start_with_fallback_inner(&token.name).await
|
|
}
|
|
|
|
/// How long to wait for a container's unit to reach `active` after a
|
|
/// `nixos-container start` that returned a non-zero exit. A start-job timeout
|
|
/// on a slow boot (e.g. DHCP taking ~10s) returns an error while
|
|
/// `container@<c>.service` keeps retrying and the container comes up seconds
|
|
/// later — so the exit code is NOT authoritative, the unit state is. 60s
|
|
/// comfortably covers an observed slow boot (the reported incident settled
|
|
/// ~16s after the "failure") without pinning a queue node on a genuine
|
|
/// never-boots failure for too long.
|
|
const START_SETTLE_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(1);
|
|
|
|
/// Poll cadence while waiting for a container to reach `active`.
|
|
const START_SETTLE_POLL: std::time::Duration = std::time::Duration::from_secs(2);
|
|
|
|
/// Poll [`is_running`] until the container's unit is active or `timeout`
|
|
/// elapses; returns true as soon as it's active.
|
|
async fn wait_until_running(name: &str, timeout: std::time::Duration) -> bool {
|
|
let deadline = tokio::time::Instant::now() + timeout;
|
|
loop {
|
|
if is_running(name).await {
|
|
return true;
|
|
}
|
|
if tokio::time::Instant::now() >= deadline {
|
|
return false;
|
|
}
|
|
tokio::time::sleep(START_SETTLE_POLL).await;
|
|
}
|
|
}
|
|
|
|
/// Internal implementation of the cold-start fallback. Used by
|
|
/// [`start_with_fallback`] (public, token-gated) and by
|
|
/// [`rebuild_no_meta`] where the preamble is already enforced structurally.
|
|
///
|
|
/// [`start`] already treats unit-active (not the exit code) as success and
|
|
/// waits out a slow boot, so this only layers the activation-error recovery on
|
|
/// top: if the container is still down after `start`, tear it down and try
|
|
/// `start` once more. The teardown prefers a graceful `stop` and only
|
|
/// escalates to SIGKILL when that `stop` itself fails.
|
|
async fn start_with_fallback_inner(name: &str) -> Result<()> {
|
|
validate(name)?;
|
|
if start(name).await.is_ok() {
|
|
return Ok(());
|
|
}
|
|
tracing::warn!(
|
|
%name,
|
|
"start did not bring the container up; hard-resetting (stop, kill only if that fails) and retrying once"
|
|
);
|
|
// Graceful `stop` first; only escalate to SIGKILL if the stop itself
|
|
// fails (e.g. a wedged container whose `machinectl poweroff` never
|
|
// completes). A clean stop is enough on its own — SIGKILL is the
|
|
// last-resort teardown for when graceful shutdown can't finish.
|
|
if let Err(stop_err) = priv_run("stop", name).await {
|
|
tracing::warn!(
|
|
%name,
|
|
error = %stop_err,
|
|
"graceful stop failed; escalating to SIGKILL"
|
|
);
|
|
priv_run("kill", name).await.unwrap_or_else(|e| {
|
|
tracing::warn!(
|
|
%name,
|
|
error = %e,
|
|
"kill after failed stop also failed (ignored)"
|
|
);
|
|
});
|
|
}
|
|
start(name)
|
|
.await
|
|
.with_context(|| format!("cold-start fallback also failed for {name}"))
|
|
}
|
|
|
|
/// Stop + start without regenerating any config. For "kick the container"
|
|
/// without touching the flake or nspawn flags.
|
|
pub async fn restart(name: &str) -> Result<()> {
|
|
kill(name).await?;
|
|
start(name).await
|
|
}
|
|
|
|
/// True when the container's systemd unit is active. Used by the dashboard
|
|
/// to gate stop/restart buttons.
|
|
pub async fn is_running(name: &str) -> bool {
|
|
let container = container_name(name);
|
|
let unit = format!("container@{container}.service");
|
|
Command::new("systemctl")
|
|
.args(["is-active", "--quiet", &unit])
|
|
.status()
|
|
.await
|
|
.is_ok_and(|s| s.success())
|
|
}
|
|
|
|
/// Fully tear down a sub-agent's container: stop + remove via `nixos-container
|
|
/// destroy`, then clean our own systemd drop-in. Leaves it to the caller to
|
|
/// wipe `/var/lib/hyperhive/...` state and the per-agent runtime dir.
|
|
pub async fn destroy(name: &str) -> Result<()> {
|
|
validate(name)?;
|
|
let container = container_name(name);
|
|
// nixos-container destroy handles stop + removal of /var/lib/nixos-containers/<C>
|
|
// and /etc/nixos-containers/<C>.conf. Tolerate "no such container".
|
|
if let Err(e) = priv_run("destroy", name).await {
|
|
tracing::warn!(error = ?e, "nixos-container destroy returned an error; continuing cleanup");
|
|
}
|
|
// Remove the systemd resource-limits drop-in via hive-priv.
|
|
if let Err(e) = crate::priv_client::remove_service_dropin(&container).await {
|
|
tracing::warn!(error = ?e, "remove service drop-in failed (non-fatal)");
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Container-level rebuild without touching the meta repo. The one
|
|
/// remaining fused stop/update/start pipeline: the approval deploy
|
|
/// (`actions::deploy_applied_target`) drives meta through the
|
|
/// two-phase prepare/finalize/abort flow itself and needs the inline
|
|
/// start to verify the agent comes back up before finalizing. Every
|
|
/// other rebuild is a job-queue DAG (`Prebuild → StopForUpdate → Swap
|
|
/// → Reconcile`) whose `Prebuild` executor owns the meta sync +
|
|
/// relock this path's deleted `rebuild` wrapper used to do.
|
|
///
|
|
/// `on_step` is called at each phase boundary with a short human-readable
|
|
/// label so callers can surface progress (e.g. update the rebuild-queue
|
|
/// step shown in the dashboard). Pass `&|_| ()` when progress reporting
|
|
/// is not needed.
|
|
///
|
|
/// `on_build_log_id` is called with the build-log row id immediately after
|
|
/// the `nixos-container update` log row opens, before the actual update
|
|
/// command starts. Callers can use this to link the queue entry to the log
|
|
/// for live streaming. Pass `&|_| ()` when not needed.
|
|
///
|
|
/// `defer_start` skips the start-after-update for a previously-running
|
|
/// container and returns `true` instead, so a queue-side caller can hand
|
|
/// the (potentially slow) container boot to the fast lane rather than
|
|
/// holding the serialized build lane through it. With `defer_start =
|
|
/// false` the start (with cold-start fallback) runs inline as before and
|
|
/// the return value is always `false`. The spawn path always starts
|
|
/// inline — a freshly-created container boots as part of provisioning.
|
|
pub async fn rebuild_no_meta(
|
|
name: &str,
|
|
hive: &HiveEnv,
|
|
paths: &AgentPaths,
|
|
defer_start: bool,
|
|
on_step: &(dyn Fn(&str) + Send + Sync),
|
|
on_build_log_id: &(dyn Fn(i64) + Send + Sync),
|
|
) -> Result<bool> {
|
|
prepare_rebuild_dirs(name, paths).await?;
|
|
let flake_ref = format!("{}#{name}", crate::meta::meta_dir().display());
|
|
if container_exists(name).await {
|
|
// Rebuild strategy: stop-before-update + pre-build.
|
|
// See `docs/coordinator.md::Container lifecycle`.
|
|
let was_running = is_running(name).await;
|
|
write_dropins(name, hive, paths).await?;
|
|
if was_running {
|
|
on_step("nix build");
|
|
prebuild_toplevel(name, &flake_ref, &|_| ()).await?;
|
|
on_step("nixos-container stop");
|
|
priv_run("stop", name).await?;
|
|
}
|
|
on_step("nixos-container update");
|
|
let update_result = priv_run_inner("update", name, Some(on_build_log_id)).await;
|
|
if let Err(ref update_err) = update_result {
|
|
// The update failed (e.g. nix build error). If the agent was
|
|
// running before we stopped it, try to bring it back up on the
|
|
// previous successful configuration so it doesn't stay dead.
|
|
// The start failure is logged but not promoted to an error —
|
|
// we always propagate the original update error (below).
|
|
if was_running {
|
|
tracing::warn!(
|
|
%name,
|
|
error = %update_err,
|
|
"nixos-container update failed; attempting restart on old config"
|
|
);
|
|
on_step("nixos-container start (recovery)");
|
|
if let Err(e) = priv_run("start", name).await {
|
|
tracing::warn!(%name, error = %e, "recovery start after failed update also failed");
|
|
}
|
|
}
|
|
}
|
|
update_result?;
|
|
if was_running {
|
|
if defer_start {
|
|
// The caller re-queues the start on the fast lane so the
|
|
// build lane is freed for the next entry instead of
|
|
// waiting out the container boot here.
|
|
return Ok(true);
|
|
}
|
|
on_step("nixos-container start");
|
|
// write_dropins was called above; use the inner fn directly
|
|
// since the preamble is enforced structurally in this path.
|
|
start_with_fallback_inner(name).await?;
|
|
}
|
|
Ok(false)
|
|
} else {
|
|
// Spawn path: create is atomic, no prebuild needed.
|
|
// See `docs/coordinator.md::Spawn path`.
|
|
on_step("nixos-container create");
|
|
priv_run("create", name).await?;
|
|
// Runtime dir must exist before nixos-container start.
|
|
ensure_agent_runtime_dir(name)?;
|
|
write_dropins(name, hive, paths).await?;
|
|
on_step("nixos-container start");
|
|
priv_run("start", name).await?;
|
|
Ok(false)
|
|
}
|
|
}
|
|
|
|
/// Pre-build `system.build.toplevel` against `meta#<name>` so the
|
|
/// subsequent `nixos-container update` finds the result cached and
|
|
/// skips straight to the profile-swap. Store-warming only — container
|
|
/// is untouched. See `docs/coordinator.md::Rebuild path` for why
|
|
/// the prebuild happens before stop, and `docs/coordinator.md::Prebuild
|
|
/// attr path` for why the explicit nixosConfigurations attr is required.
|
|
///
|
|
/// `on_build_log_id` fires with the `build_logs` row id as soon as the
|
|
/// row opens, so queue-side callers can link their node to the live
|
|
/// stream. Pass `&|_| ()` when not needed.
|
|
pub async fn prebuild_toplevel(
|
|
name: &str,
|
|
flake_ref: &str,
|
|
on_build_log_id: &(dyn Fn(i64) + Send + Sync),
|
|
) -> Result<()> {
|
|
use tokio::io::{AsyncBufReadExt, BufReader};
|
|
// Split `<root>#<name>` so we can re-emit with the explicit
|
|
// `nixosConfigurations.<name>` segment. The flake_ref shape is
|
|
// constructed by `rebuild_no_meta` and always contains exactly one
|
|
// `#`; `split_once` returning None here would be a programmer
|
|
// error we'd want to surface loudly rather than paper over.
|
|
let (flake_root, fragment) = flake_ref
|
|
.split_once('#')
|
|
.with_context(|| format!("flake_ref {flake_ref:?} missing '#<name>' fragment"))?;
|
|
// Sanity-check the fragment matches the agent name we were
|
|
// passed — guards against future calls that pass a divergent
|
|
// pair (no current callsite does, but the pair is redundant
|
|
// and worth checking once).
|
|
if fragment != name {
|
|
anyhow::bail!("prebuild_toplevel: flake_ref fragment '{fragment}' ≠ agent name '{name}'");
|
|
}
|
|
let attr = format!("{flake_root}#nixosConfigurations.{name}.config.system.build.toplevel");
|
|
let args = vec![
|
|
"--extra-experimental-features",
|
|
"nix-command flakes",
|
|
"build",
|
|
"--no-link",
|
|
"--print-out-paths",
|
|
&attr,
|
|
];
|
|
let cmdline = format!("nix {}", args.join(" "));
|
|
tracing::info!(%name, %cmdline, "prebuild: warming system toplevel");
|
|
|
|
// Open a build_logs row for this attempt (best-effort — None when
|
|
// the global handle hasn't been installed, e.g. early startup
|
|
// or standalone tests). Lines pumped from stdout/stderr append
|
|
// into the row; `finish` lands the terminal status before we bail.
|
|
let logs = crate::build_logs::global();
|
|
let log_id = logs.as_ref().and_then(|h| {
|
|
h.start(name, "prebuild", &cmdline)
|
|
.map_err(|e| {
|
|
tracing::warn!(error = ?e, "build_logs: start failed (prebuild log dropped)");
|
|
})
|
|
.ok()
|
|
});
|
|
if let Some(id) = log_id {
|
|
on_build_log_id(id);
|
|
}
|
|
|
|
let mut child = Command::new("nix")
|
|
.args(&args)
|
|
.stdout(std::process::Stdio::piped())
|
|
.stderr(std::process::Stdio::piped())
|
|
.spawn()
|
|
.with_context(|| format!("spawn {cmdline}"))?;
|
|
|
|
let stdout = child.stdout.take().expect("piped stdout");
|
|
let stderr = child.stderr.take().expect("piped stderr");
|
|
|
|
let stdout_cmdline = cmdline.clone();
|
|
let stdout_logs = logs.clone();
|
|
let pump_stdout = tokio::spawn(async move {
|
|
let mut lines = BufReader::new(stdout).lines();
|
|
while let Ok(Some(line)) = lines.next_line().await {
|
|
tracing::info!(target: "nix-prebuild", cmdline = %stdout_cmdline, "{line}");
|
|
if let (Some(h), Some(id)) = (&stdout_logs, log_id) {
|
|
h.append_stdout(id, &line);
|
|
}
|
|
}
|
|
});
|
|
|
|
let stderr_cmdline = cmdline.clone();
|
|
let stderr_logs = logs.clone();
|
|
let pump_stderr = tokio::spawn(async move {
|
|
let mut lines = BufReader::new(stderr).lines();
|
|
while let Ok(Some(line)) = lines.next_line().await {
|
|
tracing::warn!(target: "nix-prebuild", cmdline = %stderr_cmdline, "{line}");
|
|
if let (Some(h), Some(id)) = (&stderr_logs, log_id) {
|
|
h.append_stderr(id, &line);
|
|
}
|
|
}
|
|
});
|
|
|
|
let status = child
|
|
.wait()
|
|
.await
|
|
.with_context(|| format!("wait {cmdline}"))?;
|
|
let _ = pump_stdout.await;
|
|
let _ = pump_stderr.await;
|
|
|
|
let ok = status.success();
|
|
if let (Some(h), Some(id)) = (&logs, log_id) {
|
|
h.finish(
|
|
id,
|
|
if ok {
|
|
crate::build_logs::BuildStatus::Ok
|
|
} else {
|
|
crate::build_logs::BuildStatus::Fail
|
|
},
|
|
);
|
|
}
|
|
if !ok {
|
|
match log_id {
|
|
Some(id) => bail!("prebuild {cmdline} failed ({status}); see build log #{id}"),
|
|
None => bail!("prebuild {cmdline} failed ({status})"),
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn list() -> Result<Vec<String>> {
|
|
let stdout = crate::priv_client::list_containers().await?;
|
|
Ok(stdout
|
|
.lines()
|
|
.map(str::trim)
|
|
.filter(|line| line.starts_with(AGENT_PREFIX))
|
|
.map(str::to_owned)
|
|
.collect())
|
|
}
|
|
|
|
/// Sync `/etc/tmpfiles.d/hyperhive-agents.conf` with the currently-known
|
|
/// agent set (from `nixos-container list`). Strips the `h-` prefix to get
|
|
/// logical names. Best-effort: errors are logged but never propagated — a
|
|
/// failed tmpfiles write shouldn't block a spawn or destroy.
|
|
///
|
|
/// Called at hive-c0re startup and after each spawn / destroy so the file
|
|
/// always reflects the live agent set. `systemd-tmpfiles-setup.service`
|
|
/// reads the file at boot (before any container units start), pre-creating
|
|
/// bind-mount source dirs so container@h-* units don't race hive-c0re.
|
|
pub async fn sync_tmpfiles() {
|
|
let agents = match list().await {
|
|
Ok(containers) => containers
|
|
.into_iter()
|
|
.filter_map(|c| c.strip_prefix(AGENT_PREFIX).map(str::to_owned))
|
|
.collect::<Vec<_>>(),
|
|
Err(e) => {
|
|
tracing::warn!(error = ?e, "sync_tmpfiles: list failed; skipping");
|
|
return;
|
|
}
|
|
};
|
|
if let Err(e) = crate::priv_client::sync_agent_tmpfiles(&agents).await {
|
|
tracing::warn!(error = ?e, "sync_tmpfiles: priv call failed");
|
|
} else {
|
|
tracing::debug!(count = agents.len(), "sync_tmpfiles: ok");
|
|
}
|
|
}
|
|
|
|
/// Ensure the per-agent runtime directory `/run/hyperhive/agents/<name>`
|
|
/// exists. The directory is also written by `SyncAgentTmpfiles` (run at
|
|
/// boot + spawn/destroy), but explicit creation in start/spawn paths guards
|
|
/// against races where hive-c0re starts a container before tmpfiles.d has
|
|
/// applied the new entry.
|
|
///
|
|
/// Pure filesystem op — no `Coordinator` dependency — so callers that only
|
|
/// need the dir do not have to hold an `Arc<Coordinator>`.
|
|
///
|
|
/// # Errors
|
|
/// Returns an error if `create_dir_all` fails.
|
|
pub fn ensure_agent_runtime_dir(name: &str) -> Result<()> {
|
|
let dir = std::path::PathBuf::from(format!("/run/hyperhive/agents/{name}"));
|
|
std::fs::create_dir_all(&dir)
|
|
.with_context(|| format!("create agent runtime dir {}", dir.display()))
|
|
}
|
|
|
|
/// Build the per-line callback for `create_container_streaming` /
|
|
/// `update_container_streaming`. Both ops share identical dispatch logic
|
|
/// (stdout → info + `append_stdout`, stderr → warn + `append_stderr`); this
|
|
/// helper avoids duplicating that match body across the two call sites.
|
|
fn make_log_callback(
|
|
logs: Option<std::sync::Arc<crate::build_logs::BuildLogs>>,
|
|
log_id: Option<i64>,
|
|
cmdline: String,
|
|
) -> impl FnMut(hive_sh4re::priv_proto::PrivStream, &str) {
|
|
use hive_sh4re::priv_proto::PrivStream;
|
|
move |stream, line| match stream {
|
|
PrivStream::Stdout => {
|
|
tracing::info!(target: "nixos-container", cmdline = %cmdline, "{line}");
|
|
if let (Some(h), Some(id)) = (&logs, log_id) {
|
|
h.append_stdout(id, line);
|
|
}
|
|
}
|
|
PrivStream::Stderr => {
|
|
tracing::warn!(target: "nixos-container", cmdline = %cmdline, "{line}");
|
|
if let (Some(h), Some(id)) = (&logs, log_id) {
|
|
h.append_stderr(id, line);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Execute a container operation via hive-priv and integrate with
|
|
/// `build_logs.sqlite`. hive-priv runs as root and forwards output lines
|
|
/// to hive-c0re in real time via the streaming priv protocol. Each line
|
|
/// is appended to the build-log row as it arrives, so the dashboard
|
|
/// shows live progress during long `nixos-container create` / `update` runs.
|
|
async fn priv_run(kind: &str, name: &str) -> Result<()> {
|
|
priv_run_inner(kind, name, None).await
|
|
}
|
|
|
|
/// Like `priv_run` but calls `on_log_id(log_id)` immediately after the
|
|
/// build-log row is opened — before the actual container op starts.
|
|
/// This lets callers surface the row id for live streaming (e.g. the
|
|
/// rebuild-queue worker sets `build_log_id` on the queue entry so the
|
|
/// dashboard can link to `/api/build-logs/id/{id}/stream`).
|
|
///
|
|
/// The callback fires only when a build-log row is successfully opened
|
|
/// (i.e. the global `BuildLogs` handle is installed AND `h.start()`
|
|
/// succeeds). No-op when `on_log_id` is `None` — that's the path for
|
|
/// all callers that don't need the id.
|
|
async fn priv_run_inner(
|
|
kind: &str,
|
|
name: &str,
|
|
on_log_id: Option<&(dyn Fn(i64) + Send + Sync)>,
|
|
) -> Result<()> {
|
|
let container = container_name(name);
|
|
let cmdline = format!("nixos-container {kind} {container}");
|
|
|
|
let logs = crate::build_logs::global();
|
|
let log_id = logs.as_ref().and_then(|h| {
|
|
h.start(name, kind, &cmdline)
|
|
.map_err(|e| {
|
|
tracing::warn!(error = ?e, "build_logs: start failed (priv_run log dropped)");
|
|
})
|
|
.ok()
|
|
});
|
|
// Notify the caller as soon as the log row exists so it can surface
|
|
// the id for live streaming before the container op even starts.
|
|
if let (Some(id), Some(cb)) = (log_id, on_log_id) {
|
|
cb(id);
|
|
}
|
|
|
|
// For long-running ops use the streaming protocol so build_logs
|
|
// receives lines in real time rather than as a batch at completion.
|
|
let result: Result<()> = match kind {
|
|
"create" => {
|
|
crate::priv_client::create_container_streaming(
|
|
name,
|
|
make_log_callback(logs.clone(), log_id, cmdline.clone()),
|
|
)
|
|
.await
|
|
}
|
|
"update" => {
|
|
crate::priv_client::update_container_streaming(
|
|
name,
|
|
make_log_callback(logs.clone(), log_id, cmdline.clone()),
|
|
)
|
|
.await
|
|
}
|
|
"start" => crate::priv_client::start_container(name).await,
|
|
"stop" => crate::priv_client::stop_container(name).await,
|
|
"kill" => crate::priv_client::kill_container(name).await,
|
|
"destroy" => crate::priv_client::destroy_container(name).await,
|
|
other => Err(anyhow::anyhow!("unknown container op: {other}")),
|
|
};
|
|
|
|
let succeeded = result.is_ok();
|
|
if let (Some(h), Some(id)) = (&logs, log_id) {
|
|
h.finish(
|
|
id,
|
|
if succeeded {
|
|
crate::build_logs::BuildStatus::Ok
|
|
} else {
|
|
crate::build_logs::BuildStatus::Fail
|
|
},
|
|
);
|
|
}
|
|
|
|
match result {
|
|
Ok(()) => Ok(()),
|
|
Err(e) => {
|
|
let journal = if kind == "update" {
|
|
container_journal_tail(&container).await
|
|
} else {
|
|
String::new()
|
|
};
|
|
match log_id {
|
|
Some(id) => bail!("{e:#}; see build log #{id}{journal}"),
|
|
None => bail!("{e:#}{journal}"),
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// On a failed `nixos-container update`, the stderr nixos-container
|
|
/// itself prints is often terse ("failed to reload container") — the
|
|
/// real reason (which unit failed `switch-to-configuration` during
|
|
/// the reload phase) lands in the *container's* own journal, not on
|
|
/// the host. Fetch the tail of it so a failed rebuild self-documents
|
|
/// the failing unit in the error string, no second round-trip.
|
|
///
|
|
/// Scoped to `update`: that's the reload-phase case, and the
|
|
/// container is still up (running the old generation) so
|
|
/// `journalctl -M` works. Best-effort — returns "" for other verbs
|
|
/// or when the journal can't be read (machine gone, journalctl
|
|
/// missing); it never produces an error of its own.
|
|
async fn container_journal_tail(container: &str) -> String {
|
|
// `-M` enters the container namespace and needs root, so the read
|
|
// is delegated to hive-priv (hive-c0re itself runs unprivileged).
|
|
let res = crate::priv_client::read_container_journal(
|
|
container,
|
|
hive_sh4re::priv_proto::JournalQuery {
|
|
lines: 40,
|
|
..Default::default()
|
|
},
|
|
)
|
|
.await;
|
|
match res {
|
|
Ok((stdout, _)) if !stdout.is_empty() => format!(
|
|
"\n--- last 40 journal lines from container '{container}' ---\n{}",
|
|
stdout.trim_end()
|
|
),
|
|
_ => String::new(),
|
|
}
|
|
}
|