From 02e916feeed8cf7db20d0d48dd1f6fb05b5be7be Mon Sep 17 00:00:00 2001 From: atlas Date: Tue, 4 Aug 2026 17:54:28 +0200 Subject: [PATCH] wip(#3001): power chains name their group roots MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `*_many` entry points returned `insert_job`'s result while their closures ended in `Vec::new()` — naming nothing, so the returned id list was always empty. `queued_dags` would have shipped `Some([])` and hivectl's wait loop would have had nothing to poll. Silent: it compiles, the op still runs, and no test in isolation looks. Each `*_chain` now returns its group root's guid and the `*_nodes` collectors gather them, so the ids a caller gets back are the roots it can actually wait on. `start_chain` returns *four* in the stale branch, not one: `rebuild_nodes` chains its roots behind `SetWanted` with `after_ok` rather than nesting them under it, so `SetWanted` rolls up only itself. Naming it alone would have reported the start complete while the rebuild was still running — the same under-reporting bug one level down. --- hive-c0re/src/dashboard/lifecycle_ops.rs | 3 +- hive-c0re/src/job_queue/power.rs | 123 ++++++++++++++++------- 2 files changed, 89 insertions(+), 37 deletions(-) diff --git a/hive-c0re/src/dashboard/lifecycle_ops.rs b/hive-c0re/src/dashboard/lifecycle_ops.rs index 0b864107..e1c46463 100644 --- a/hive-c0re/src/dashboard/lifecycle_ops.rs +++ b/hive-c0re/src/dashboard/lifecycle_ops.rs @@ -110,7 +110,8 @@ pub(super) async fn post_kill( // `socket_server.rs::Request::Kill` stays in place: a // manager calling Kill on its own container is self-suicide // mid-call, not a legitimate operator action. - if let Err(e) = crate::job_queue::power::stop_many(&state.coord, &[logical.clone()], false).await + if let Err(e) = + crate::job_queue::power::stop_many(&state.coord, &[logical.clone()], false).await { tracing::error!(agent = %logical, error = ?e, "stop: insert failed"); } diff --git a/hive-c0re/src/job_queue/power.rs b/hive-c0re/src/job_queue/power.rs index 3ed1d4e1..b1606a57 100644 --- a/hive-c0re/src/job_queue/power.rs +++ b/hive-c0re/src/job_queue/power.rs @@ -35,7 +35,16 @@ use crate::lifecycle; /// actually running (nothing to drain on a down container). The `Reconcile` /// stays even for a down agent so a race-up between the state read and exec /// is still stopped in-DAG. -fn stop_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: bool) { +/// Returns the group root's guid, which is what the caller names so +/// `insert_job` hands its id back — that id is how `hivectl` polls this agent's +/// progress. A chain that returned nothing would insert correctly and leave the +/// caller with nothing to wait on. +fn stop_chain( + builder: &JobBuilder, + agent: &str, + graceful: bool, + running: bool, +) -> hive_jobq::NodeGuid { // `SetWanted` is the group root and owns the agent lease; the mechanical // steps are its children (borrow the lease, run once it reaches `Finishing`, // dep-ordered among themselves). @@ -64,13 +73,26 @@ fn stop_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: bool) .needs(Resource::Agent(a())) .part_of(wanted); } + wanted.guid() } /// One agent's **start** subgraph. `SetWanted(Up)` head; a down + stale-rev /// agent gets the rebuild subgraph (its tail `Reconcile` starts it on /// current derivations), otherwise a plain `Reconcile` (which starts a down /// agent and noops an already-running one). -fn start_chain(builder: &JobBuilder, agent: &str, running: bool, stale: bool) { +/// +/// Returns **every** group root — see [`stop_chain`] for why they are named. +/// +/// ⚠️ More than one in the stale branch: `rebuild_nodes` chains its roots +/// *behind* `SetWanted` with `after_ok`, it does **not** nest them under it. So +/// `SetWanted` rolls up only itself, and naming it alone would report the whole +/// start finished while the rebuild was still running. +fn start_chain( + builder: &JobBuilder, + agent: &str, + running: bool, + stale: bool, +) -> Vec { let wanted = builder .node(NodeKind::SetWanted { agent: agent.to_owned(), @@ -78,10 +100,16 @@ fn start_chain(builder: &JobBuilder, agent: &str, running: bool, stale: bool) { }) .needs(Resource::Agent(agent.to_owned())); if !running && stale { - // Rebuild subtree chained behind the `SetWanted` head. `MetaSync`, - // `Prebuild` + `Reconcile` are their own group roots (top-level, per - // `rebuild_nodes`). - rebuild_nodes(builder, agent, true, Some(wanted)); + // Rebuild subtree chained behind the `SetWanted` head. `MetaSync`, the + // `AgentWindow` brace and `Reconcile` are their own group roots + // (top-level, per `rebuild_nodes`) — hence all four names. + let roots = rebuild_nodes(builder, agent, true, Some(wanted)); + vec![ + wanted.guid(), + roots.meta_sync.guid(), + roots.agent_window.guid(), + roots.reconcile.guid(), + ] } else { let _ = builder .node(NodeKind::Reconcile { @@ -89,6 +117,7 @@ fn start_chain(builder: &JobBuilder, agent: &str, running: bool, stale: bool) { }) .needs(Resource::Agent(agent.to_owned())) .part_of(wanted); + vec![wanted.guid()] } } @@ -101,14 +130,22 @@ fn start_chain(builder: &JobBuilder, agent: &str, running: bool, stale: bool) { /// before `Reconcile`; a down agent gets just `Reconcile`, which /// converges to intent — a stopped (`wanted = Off`) agent stays stopped, /// a crashed (`wanted = Up`) agent comes back up. -fn restart_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: bool) { +/// +/// Returns the group root's guid — see [`stop_chain`]. +fn restart_chain( + builder: &JobBuilder, + agent: &str, + graceful: bool, + running: bool, +) -> hive_jobq::NodeGuid { let a = || agent.to_owned(); if !running { - // Nothing to bounce — a lone Reconcile converges to intent. - let _ = builder + // Nothing to bounce — a lone Reconcile converges to intent, and is + // itself the root. + return builder .node(NodeKind::Reconcile { agent: a() }) - .needs(Resource::Agent(a())); - return; + .needs(Resource::Agent(a())) + .guid(); } // Running: mechanical stop then Reconcile. The first stop node is the group // ROOT (no SetWanted head) and owns the agent lease; the rest are its @@ -144,6 +181,7 @@ fn restart_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: boo .needs(Resource::Agent(a())) .part_of(signal) .after_ok(stop); + signal.guid() } else { let stop = builder .node(NodeKind::StopForUpdate { agent: a() }) @@ -152,6 +190,7 @@ fn restart_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: boo .node(NodeKind::Reconcile { agent: a() }) .needs(Resource::Agent(a())) .part_of(stop); + stop.guid() } } @@ -169,10 +208,15 @@ fn restart_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: boo // subgraph's indices onto another's used to be a function. /// Declare the stop DAG from explicit `(agent, running)` targets. -pub(crate) fn stop_nodes(builder: &JobBuilder, targets: &[(String, bool)], graceful: bool) { - for (agent, running) in targets { - stop_chain(builder, agent, graceful, *running); - } +pub(crate) fn stop_nodes( + builder: &JobBuilder, + targets: &[(String, bool)], + graceful: bool, +) -> Vec { + targets + .iter() + .map(|(agent, running)| stop_chain(builder, agent, graceful, *running)) + .collect() } /// Assemble the start DAG from explicit `(agent, running, stale)` targets. @@ -181,17 +225,26 @@ pub(crate) fn stop_nodes(builder: &JobBuilder, targets: &[(String, bool)], grace /// running under its lease, so a down+stale agent that grew a rebuild subgraph /// reports `rebuilding` during its swap and `starting` at its reconcile, /// without the DAG having to guess one label covering every target. -pub(crate) fn start_nodes(builder: &JobBuilder, targets: &[(String, bool, bool)]) { - for (agent, running, stale) in targets { - start_chain(builder, agent, *running, *stale); - } +pub(crate) fn start_nodes( + builder: &JobBuilder, + targets: &[(String, bool, bool)], +) -> Vec { + targets + .iter() + .flat_map(|(agent, running, stale)| start_chain(builder, agent, *running, *stale)) + .collect() } /// Declare the restart DAG from explicit `(agent, running)` targets. -pub(crate) fn restart_nodes(builder: &JobBuilder, targets: &[(String, bool)], graceful: bool) { - for (agent, running) in targets { - restart_chain(builder, agent, graceful, *running); - } +pub(crate) fn restart_nodes( + builder: &JobBuilder, + targets: &[(String, bool)], + graceful: bool, +) -> Vec { + targets + .iter() + .map(|(agent, running)| restart_chain(builder, agent, graceful, *running)) + .collect() } // ---- entry points --------------------------------------------------------- @@ -216,10 +269,9 @@ pub async fn restart_many( for agent in agents { targets.push((agent.clone(), lifecycle::is_running(agent).await)); } - let ids = coord.job_queue.insert_job(|b| { - restart_nodes(b, &targets, graceful); - Vec::new() - })?; + let ids = coord + .job_queue + .insert_job(|b| restart_nodes(b, &targets, graceful))?; coord.emit_rebuild_queue_snapshot(); Ok(ids) } @@ -231,7 +283,10 @@ pub async fn restart_many( /// /// # Errors /// Propagates a graph-insert error. -pub async fn start_many(coord: &Arc, agents: &[String]) -> anyhow::Result> { +pub async fn start_many( + coord: &Arc, + agents: &[String], +) -> anyhow::Result> { let current = crate::auto_update::current_flake_rev(&coord.hyperhive_flake); let mut targets = Vec::with_capacity(agents.len()); for agent in agents { @@ -245,10 +300,7 @@ pub async fn start_many(coord: &Arc, agents: &[String]) -> anyhow:: } targets.push((agent.clone(), running, stale)); } - let ids = coord.job_queue.insert_job(|b| { - start_nodes(b, &targets); - Vec::new() - })?; + let ids = coord.job_queue.insert_job(|b| start_nodes(b, &targets))?; coord.emit_rebuild_queue_snapshot(); Ok(ids) } @@ -269,10 +321,9 @@ pub async fn stop_many( for agent in agents { targets.push((agent.clone(), lifecycle::is_running(agent).await)); } - let ids = coord.job_queue.insert_job(|b| { - stop_nodes(b, &targets, graceful); - Vec::new() - })?; + let ids = coord + .job_queue + .insert_job(|b| stop_nodes(b, &targets, graceful))?; coord.emit_rebuild_queue_snapshot(); Ok(ids) }