diff --git a/hive-bash-mcp/src/runner.rs b/hive-bash-mcp/src/runner.rs index e2e9e25f..ac1915b1 100644 --- a/hive-bash-mcp/src/runner.rs +++ b/hive-bash-mcp/src/runner.rs @@ -90,13 +90,7 @@ fn signal_group(pgid: Option, sig: i32) { // Helpers // --------------------------------------------------------------------------- -fn now_unix() -> i64 { - SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_secs() - .cast_signed() -} +use hive_sh4re::wire_time::now_unix; /// Generate a task ID: ``. #[must_use] diff --git a/hive-c0re/src/bin/hivectl.rs b/hive-c0re/src/bin/hivectl.rs index 515860a8..f2b20942 100644 --- a/hive-c0re/src/bin/hivectl.rs +++ b/hive-c0re/src/bin/hivectl.rs @@ -1482,12 +1482,15 @@ async fn wait_for_dags(socket: &Path, ids: Vec, no_wait: bool) -> Result<() println!("{line}"); last.insert(d.id, line); } - match d.state { - hive_sh4re::jobs::State::Failed => { - failed.push(format!("{} {}", d.kind.as_str(), d.agent)); - } - hive_sh4re::jobs::State::Done | hive_sh4re::jobs::State::Cancelled => {} - _ => all_terminal = false, + // Node-level terminality, not the roll-up: a DAG rolls + // up `failed` the moment one node fails while its + // after-any recovery node (rebuild's tail Reconcile) + // may still be running — keep watching so the operator + // sees whether the agent came back. + if !d.nodes.iter().all(|n| n.state.is_terminal()) { + all_terminal = false; + } else if d.state == hive_sh4re::jobs::State::Failed { + failed.push(format!("{} {}", d.kind.as_str(), d.agent)); } } if all_terminal { @@ -1651,16 +1654,21 @@ async fn stop( hive_c0re::client::request(socket, hive_sh4re::HostRequest::Stop { scope, graceful }) .await .with_context(|| format!("connect to daemon socket {}", socket.display()))?; - render_lifecycle(&resp, "stop queued")?; - wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), no_wait).await + // Render first, but even when an infra failure makes it bail, + // watch the already-queued agent DAGs before surfacing the error — + // they run regardless. + let rendered = render_lifecycle(&resp, "stop queued"); + wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), no_wait).await?; + rendered } async fn start(socket: &Path, scope: hive_sh4re::LifecycleScope, no_wait: bool) -> Result<()> { let resp = hive_c0re::client::request(socket, hive_sh4re::HostRequest::Start { scope }) .await .with_context(|| format!("connect to daemon socket {}", socket.display()))?; - render_lifecycle(&resp, "start queued")?; - wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), no_wait).await + let rendered = render_lifecycle(&resp, "start queued"); + wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), no_wait).await?; + rendered } /// Restart = `stop` then `start` over the same scope, composed client-side @@ -1723,6 +1731,12 @@ async fn subvol_upgrade(socket: &Path, name: &str, yes: bool) -> Result<()> { stop_resp.error.as_deref().unwrap_or("unknown error") ); } + // The stop is a queued DAG now — the migration below snapshots + + // swaps the state dir and MUST NOT run under a live bind mount, so + // wait for the stop to actually execute before touching anything. + wait_for_dags(socket, stop_resp.queued_dags.unwrap_or_default(), false) + .await + .with_context(|| format!("waiting for {name} to stop before the migration"))?; println!("migrating {name} state dir to a btrfs subvolume…"); let upgrade = hive_c0re::priv_client::upgrade_agent_subvolume(name).await; @@ -1762,6 +1776,14 @@ async fn subvol_upgrade(socket: &Path, name: &str, yes: bool) -> Result<()> { start_resp.error.as_deref().unwrap_or("unknown error") ); } + wait_for_dags(socket, start_resp.queued_dags.unwrap_or_default(), false) + .await + .with_context(|| { + format!( + "{name} migrated to a btrfs subvolume, but its restart job failed — run \ + `hivectl start --agent {name}` to retry" + ) + })?; println!("upgraded {name} to a btrfs subvolume and restarted it"); Ok(()) } diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index f15ebe2c..bfb118b2 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -39,6 +39,14 @@ pub use model::{ /// per template in the snapshot, matching the old per-kind history cap. const MAX_HISTORY_PER_TEMPLATE: usize = 5; +/// Terminal DAGs younger than this are exempt from the per-template +/// history cap. A broad `hivectl stop`/`start` submits many +/// same-template DAGs that can all settle within one poll interval — +/// without the grace, the cap would evict some before the ~1s +/// `QueueDag` poller ever observes their terminal state, silently +/// swallowing failures. +const HISTORY_GRACE_SECS: i64 = 300; + /// Cap on stored node error strings. const MAX_ERROR_LEN: usize = 2_000; @@ -399,7 +407,7 @@ impl JobQueue { for agent in freed { inner.leases.remove(&agent); } - Self::trim_history(inner); + Self::trim_history(inner, now_unix() - HISTORY_GRACE_SECS); } /// Cancel a DAG that hasn't started yet (roll-up `Queued`): every @@ -540,11 +548,12 @@ impl JobQueue { } /// Keep only the newest `MAX_HISTORY_PER_TEMPLATE` terminal DAGs - /// per template. Live DAGs are never evicted — and neither is a - /// terminal parent that still has live children (a fan-out parent - /// is terminal the moment its `MetaLock` completes; evicting it - /// while cascade rebuilds run would orphan their dashboard group). - fn trim_history(inner: &mut Inner) { + /// per template. Never evicted: live DAGs; terminal parents with + /// live children (a fan-out parent is terminal the moment its + /// `MetaLock` completes — evicting it while cascade rebuilds run + /// would orphan their dashboard group); and terminal DAGs that + /// finished after `grace_cutoff` (see [`HISTORY_GRACE_SECS`]). + fn trim_history(inner: &mut Inner, grace_cutoff: i64) { let live_parents: std::collections::HashSet = inner .dags .iter() @@ -560,6 +569,10 @@ impl JobQueue { if !d.is_terminal() || live_parents.contains(&d.id) { return true; } + let finished = d.nodes.iter().filter_map(|n| n.finished_at).max(); + if finished.is_none_or(|t| t > grace_cutoff) { + return true; + } let n = counts.entry(d.template).or_insert(0); *n += 1; *n <= MAX_HISTORY_PER_TEMPLATE @@ -568,4 +581,12 @@ impl JobQueue { .collect(); inner.dags = kept.into_iter().rev().collect(); } + + /// Test hook: trim with the grace window disabled, so eviction + /// behavior is assertable without aging real timestamps. + #[cfg(test)] + pub(crate) fn trim_ignoring_grace(&self) { + let mut inner = self.inner.lock().expect("job_queue mutex poisoned"); + Self::trim_history(&mut inner, i64::MAX); + } } diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index f7ec112e..321a6419 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -804,6 +804,16 @@ fn history_evicts_old_terminals_per_template() { let c = claim_one(&q); q.complete_node(id, c.node_id, Ok(())); } + // Fresh terminals are inside the grace window: nothing evicts yet, + // so a ~1s QueueDag poller can still observe every terminal state + // (a broad stop/start settles many same-template DAGs at once). + assert_eq!( + q.snapshot().len(), + 8, + "grace window protects fresh terminals" + ); + // Past the grace window the per-template cap applies. + q.trim_ignoring_grace(); assert_eq!(q.snapshot().len(), 5, "per-template history cap"); assert_eq!(q.live_count(), 0); } diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index b60705aa..63c5f885 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -291,19 +291,22 @@ async fn handle_restart_all(coord: &Arc) -> Result { tracing::info!("restart-all"); let agents = lifecycle::list().await?; let mut ok_agents: Vec = Vec::new(); + let mut queued: Vec = Vec::new(); for agent in &agents { let Some(logical) = agent.strip_prefix(lifecycle::AGENT_PREFIX) else { continue; }; - crate::job_queue::submit::restart( + queued.push(crate::job_queue::submit::restart( coord, logical, crate::job_queue::Source::Manual, "manual restart via hivectl restart-all".to_owned(), - ); + )); ok_agents.push(logical.to_owned()); } - Ok(HostResponse::list(ok_agents)) + let mut resp = HostResponse::list(ok_agents); + resp.queued_dags = Some(queued); + Ok(resp) } /// Stop the given `agents` (resolved logical names) then `infra` containers @@ -356,6 +359,14 @@ async fn handle_stop( ok_items.push(agent.clone()); } + // Agents go down before infra so they're not mid-request against a + // forge/matrix that's already gone. Hard stops are quick kills — + // await their DAGs (bounded) before touching infra. Graceful stops + // keep the immediate return (drains take minutes and the + // agents-then-infra race pre-existed there). + if !graceful && !infra.is_empty() { + await_dags(coord, &queued, std::time::Duration::from_mins(2)).await; + } for &container in infra { let name = container.unit_name(); match crate::priv_client::control_infra_container(container, InfraAction::Stop).await { @@ -372,6 +383,28 @@ async fn handle_stop( Ok(resp) } +/// Best-effort server-side wait for a set of DAGs to settle terminal, +/// bounded by `timeout` — used to preserve ordering invariants inside +/// one request (agent stops before infra stops) without trusting the +/// client to wait. +async fn await_dags(coord: &Arc, ids: &[u64], timeout: std::time::Duration) { + let deadline = std::time::Instant::now() + timeout; + loop { + let snap = coord.job_queue.snapshot(); + let pending = ids + .iter() + .any(|id| snap.iter().any(|d| d.id == *id && !d.state.is_terminal())); + if !pending { + return; + } + if std::time::Instant::now() >= deadline { + tracing::warn!(?ids, "await_dags: timed out; proceeding"); + return; + } + tokio::time::sleep(std::time::Duration::from_millis(250)).await; + } +} + /// Start the given `infra` containers then `agents` (`hivectl start`) — the /// inverse of [`handle_stop`]. Infra comes up before agents so the agents /// find forge/matrix/gateway ready. Per-target failures aggregated. Callers