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

workers: skip a cycle instead of spawning/alerting on an unreadable container list

Browse source 
crash_watch's 10s poll and auto_update's ensure_root_agent both read
lifecycle::list().await.unwrap_or_default(), which turned a failed read
into 'zero containers'. In crash_watch that made every previously-running
agent look like it crashed simultaneously (prev.difference(current) over
an empty current), and left prev empty for the next cycle too, so a
second wave of false 'agent logged in' / 'agent needs login' events fired
against the next successful read. In ensure_root_agent it read as
'manager container missing' and called lifecycle::spawn on a manager
that might already exist.

Both sites now treat a list error as its own outcome: log it at warn and
skip the cycle's decision entirely. crash_watch leaves prev exactly as
the last good read produced it. ensure_root_agent attempts no spawn.

Factors each site's decision into a pure helper (plan_cycle /
plan_root_agent) matching the check_not_live / confirm_gone_after_failed_destroy
pattern, with unit tests for the error case, a control for the readable
case, and (for crash_watch) an invert-proof run locally against the old
unwrap_or_default logic before reverting.
This commit is contained in:
atlas 2026-09-24 12:50:07 +02:00 • committed by mara
commit f62ec349d5
2 changed files with 172 additions and 43 deletions
Download patch file Download diff file Expand all files Collapse all files

131
hive-c0re/src/workers/crash_watch.rs
View file

@ -28,46 +28,57 @@ pub fn spawn(coord: Arc<Coordinator>) {
let mut prev_sub_agents: HashSet<String> = HashSet::new();
let mut seeded = false;
loop {
let raw = lifecycle::list().await.unwrap_or_default();
let mut current_running = HashSet::new();
let mut current_logged_in = HashSet::new();
let mut sub_agents: Vec<String> = Vec::new();
for c in &raw {
let Some(logical) = c.strip_prefix(AGENT_PREFIX) else {
continue;
};
let logical = logical.to_owned();
sub_agents.push(logical.clone());
if lifecycle::is_running(&logical).await {
current_running.insert(logical.clone());
}
if hive_types::Ident::parse(&logical)
.is_ok_and(|id| claude_has_session(&Coordinator::agent_claude_dir(&id)))
{
current_logged_in.insert(logical.clone());
}
let list_result = lifecycle::list().await;
if let Err(e) = &list_result {
tracing::warn!(
error = ?e,
"crash watcher: container list unreadable — skipping this cycle"
);
}
if let CyclePlan::Process(raw) = plan_cycle(list_result) {
let mut current_running = HashSet::new();
let mut current_logged_in = HashSet::new();
let mut sub_agents: Vec<String> = Vec::new();
for c in &raw {
let Some(logical) = c.strip_prefix(AGENT_PREFIX) else {
continue;
};
let logical = logical.to_owned();
sub_agents.push(logical.clone());
if lifecycle::is_running(&logical).await {
current_running.insert(logical.clone());
}
if hive_types::Ident::parse(&logical)
.is_ok_and(|id| claude_has_session(&Coordinator::agent_claude_dir(&id)))
{
current_logged_in.insert(logical.clone());
}
}
if seeded {
emit_crash_transitions(&coord, &prev_running, &current_running);
emit_login_transitions(
&prev_logged_in,
&current_logged_in,
&sub_agents,
&prev_sub_agents,
)
.await;
if seeded {
emit_crash_transitions(&coord, &prev_running, &current_running);
emit_login_transitions(
&prev_logged_in,
&current_logged_in,
&sub_agents,
&prev_sub_agents,
)
.await;
}
// Periodic container rescan — catches state flips that
// happen outside our mutation surface (operator runs
// `nixos-container stop` over ssh, agent logs in via its
// own web UI, etc.) so the dashboard converges within one
// POLL_INTERVAL. Idempotent + cheap when nothing changed.
coord.rescan_containers_and_emit().await;
prev_running = current_running;
prev_logged_in = current_logged_in;
prev_sub_agents = sub_agents.into_iter().collect();
seeded = true;
}
// Periodic container rescan — catches state flips that
// happen outside our mutation surface (operator runs
// `nixos-container stop` over ssh, agent logs in via its
// own web UI, etc.) so the dashboard converges within one
// POLL_INTERVAL. Idempotent + cheap when nothing changed.
coord.rescan_containers_and_emit().await;
prev_running = current_running;
prev_logged_in = current_logged_in;
prev_sub_agents = sub_agents.into_iter().collect();
seeded = true;
// On `CyclePlan::Skip`, `prev_*` and `seeded` are left exactly as
// the last good cycle produced them, so the next successful read
// diffs against real prior state instead of an empty one.
tokio::select! {
() = tokio::time::sleep(POLL_INTERVAL) => {}
@ -80,6 +91,29 @@ pub fn spawn(coord: Arc<Coordinator>) {
});
}
/// What this poll cycle does with `lifecycle::list()`'s result.
enum CyclePlan {
/// The list read; diff `Vec` against `prev` as usual.
Process(Vec<String>),
/// The read failed: skip the cycle outright.
Skip,
}
/// Turn a `lifecycle::list()` result into this poll's plan. An unreadable
/// list is not "zero containers" — treating it as such made every
/// previously-running agent land in `prev.difference(current)`, i.e. look
/// like it crashed simultaneously with every other agent. So a read failure
/// plans `Skip`: no diff against `prev`, no crash or login events this
/// cycle, and (at the call site) `prev` stays exactly what the last good
/// read left it, so the next successful read compares against real state
/// rather than an empty one.
fn plan_cycle(list_result: anyhow::Result<Vec<String>>) -> CyclePlan {
match list_result {
Ok(raw) => CyclePlan::Process(raw),
Err(_) => CyclePlan::Skip,
}
}
fn emit_crash_transitions(coord: &Coordinator, prev: &HashSet<String>, current: &HashSet<String>) {
let transients = coord.transient_snapshot();
// Operator actions whose RAII guard already cleared but only just;
@ -177,6 +211,29 @@ async fn emit_login_transitions(
mod tests {
use super::*;
/// The regression this whole fix exists for: a read failure must not
/// reach `emit_crash_transitions` at all, so `prev` is left untouched
/// for the next cycle to diff against.
#[test]
fn list_error_skips_the_cycle() {
let result: anyhow::Result<Vec<String>> =
Err(anyhow::anyhow!("connect to hive-priv socket"));
assert!(matches!(plan_cycle(result), CyclePlan::Skip));
}
/// Control: a real drop from running to absent is a *readable*, empty
/// list — it must still be processed (and so still reach
/// `emit_crash_transitions`, which fires the crash event), unlike a
/// read failure.
#[test]
fn readable_empty_list_still_processes() {
let result: anyhow::Result<Vec<String>> = Ok(Vec::new());
match plan_cycle(result) {
CyclePlan::Process(raw) => assert!(raw.is_empty()),
CyclePlan::Skip => panic!("a readable empty list must not be treated as unreadable"),
}
}
#[test]
fn deliberate_when_either_source_says_so() {
// Active guard, and the race repro: a lifecycle action completes and