fix: close second review round on the queue-routed CLI
- subvol upgrade waits for the queued stop DAG before migrating (was snapshotting + swapping state under a live bind mount) and for the restart job after - history trim gets a 5-min grace for fresh terminals so broad stop/start waits can't miss a failed DAG evicted by the per-template cap (cap still applies past the grace) - restart-all returns its DAG ids so hivectl actually waits - hard stops await their agent DAGs (bounded) before infra goes down, restoring the agents-before-infra invariant - hivectl wait uses node-level terminality so the after-any recovery reconcile is watched to completion; infra render errors no longer skip watching already-queued agent DAGs - fold hive-bash-mcp's last local now_unix into wire_time
This commit is contained in:
parent
0e4b5a1120
commit
2486251b32
5 changed files with 106 additions and 26 deletions
|
|
@ -90,13 +90,7 @@ fn signal_group(pgid: Option<i32>, 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: `<timestamp_hex><seq_hex>`.
|
||||
#[must_use]
|
||||
|
|
|
|||
|
|
@ -1482,12 +1482,15 @@ async fn wait_for_dags(socket: &Path, ids: Vec<u64>, 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(())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<u64> = 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);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -291,19 +291,22 @@ async fn handle_restart_all(coord: &Arc<Coordinator>) -> Result<HostResponse> {
|
|||
tracing::info!("restart-all");
|
||||
let agents = lifecycle::list().await?;
|
||||
let mut ok_agents: Vec<String> = Vec::new();
|
||||
let mut queued: Vec<u64> = 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<Coordinator>, 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
|
||||
|
|
|
|||
Loading…
Reference in a new issue