diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index cf110437..b90827b4 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -317,13 +317,22 @@ pub(crate) fn deploy_rebuild_nodes(builder: &JobBuilder, agent: &str, approval_i /// children. /// /// Closed by an [`NodeKind::EmitRebuilt`] tail edged onto all three group-roots -/// (`MetaSync`, `Prebuild`, `Reconcile`) — `Prebuild`'s roll-up carries the -/// whole `StopForUpdate`→`Swap`→`RebuildBookkeeping` subtree, so those three cover every -/// node. Edging `Reconcile` alone would not do: it is `AfterAny` `Prebuild`, so -/// it reaches `Done` even after a failed swap and the tail would report success. -pub fn rebuild(builder: &JobBuilder, agent: &str, relock: bool) { +/// (`MetaSync`, the `AgentWindow` brace, `Reconcile`) — the brace's roll-up +/// carries the whole `StopForUpdate`→`Swap`→`RebuildBookkeeping` subtree, so +/// those three cover every node. Edging `Reconcile` alone would not do: it is +/// `AfterAny` the brace, so it reaches `Done` even after a failed swap and the +/// tail would report success. +/// +/// Returns those three roots, so a caller that needs to wait on the rebuild can +/// name them. +pub fn rebuild(builder: &JobBuilder, agent: &str, relock: bool) -> Vec { let roots = rebuild_nodes(builder, agent, relock, None); emit_rebuilt_tails(builder, agent, &roots.all()); + vec![ + roots.meta_sync.guid(), + roots.agent_window.guid(), + roots.reconcile.guid(), + ] } /// Approval-driven deploy (`MergeConfigPr`) as a phase subtree rather than the @@ -483,10 +492,16 @@ pub fn meta_update(builder: &JobBuilder, inputs: Vec, approval_id: Optio /// checks), so a parent move needs no container rebuild to take effect. /// No transient pill either — the node is agentless (no lease to hang one /// off of) and near-instant. No tail node: the write is the whole effect. -pub fn reparent(builder: &JobBuilder, moves: Vec<(hive_types::Ident, Option)>) { - let _reparent = builder +/// +/// Returns the single node's guid so a caller can wait on it. +pub fn reparent( + builder: &JobBuilder, + moves: Vec<(hive_types::Ident, Option)>, +) -> hive_jobq::NodeGuid { + builder .node(NodeKind::Reparent { moves }) - .needs(Resource::MetaWindow); + .needs(Resource::MetaWindow) + .guid() } // The boot is assembled inline in `workers/auto_update.rs::submit_boot_tree` diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 26799a9a..9baaefdc 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -180,13 +180,19 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { // submit returns a DAG id immediately, the caller polls // `QueueDag` (`hivectl`'s wait/progress loop) for the // outcome instead of blocking here on the commit. - let id = crate::job_queue::submit::reparent( - &coord, - vec![(child.clone(), new_parent.clone())], - crate::job_queue::Source::Manual, - "manual set-parent via hivectl".to_owned(), - ); - HostResponse::queued(vec![id]) + let inserted = coord.job_queue.insert_job(|b| { + vec![crate::job_queue::templates::reparent( + b, + vec![(child.clone(), new_parent.clone())], + )] + }); + match inserted { + Ok(ids) => { + coord.emit_rebuild_queue_snapshot(); + HostResponse::queued(ids.into_iter().map(hive_jobq::NodeId::get).collect()) + } + Err(e) => HostResponse::error(format!("queue reparent: {e}")), + } } HostRequest::SetResourceLimits { name, @@ -777,39 +783,36 @@ enum Verb { } async fn submit_single(coord: &Arc, name: &str, verb: Verb) -> HostResponse { - use crate::job_queue::{Source, submit}; - let id = match verb { + use crate::job_queue::power; + // A single-target op is the N-target one with N = 1 — the shapes are + // identical, so there is no separate builder to keep in sync. + let targets = [name.to_owned()]; + let ids = match verb { Verb::Kill => { tracing::info!(%name, "kill"); - submit::stop( - coord, - name, - Source::Manual, - "manual kill via hivectl".to_owned(), - ) - .await + power::stop_many(coord, &targets, false).await } Verb::Restart => { tracing::info!(%name, "restart"); - submit::restart( - coord, - name, - Source::Manual, - "manual restart via hivectl".to_owned(), - ) - .await + power::restart_many(coord, &targets, false).await } Verb::Rebuild => { tracing::info!(%name, "rebuild"); - submit::rebuild( - coord, - name, - Source::Manual, - "manual rebuild via hivectl".to_owned(), - ) + // Not a power op: a rebuild's shape doesn't depend on live state, + // so it is a plain template insert rather than a `*_many` gather. + let inserted = coord + .job_queue + .insert_job(|b| crate::job_queue::templates::rebuild(b, name, true)); + if inserted.is_ok() { + coord.emit_rebuild_queue_snapshot(); + } + inserted } }; - HostResponse::queued(vec![id]) + match ids { + Ok(ids) => HostResponse::queued(ids.into_iter().map(hive_jobq::NodeId::get).collect()), + Err(e) => HostResponse::error(format!("queue insert failed: {e}")), + } } /// Stop the given `agents` (resolved logical names) then `infra` containers @@ -847,17 +850,13 @@ async fn handle_stop( } else { "manual via hivectl stop" }; - queued.push( - crate::job_queue::submit::stop_many( - coord, - agents, - graceful, - crate::job_queue::Source::Manual, - reason.to_owned(), - ) - .await, - ); - ok_items.extend(agents.iter().cloned()); + match crate::job_queue::power::stop_many(coord, agents, graceful).await { + Ok(ids) => { + queued.extend(ids.into_iter().map(hive_jobq::NodeId::get)); + ok_items.extend(agents.iter().cloned()); + } + Err(e) => errors.push(format!("queue stop: {e}")), + } } // Agents go down before infra so they're not mid-request against a @@ -944,16 +943,13 @@ async fn handle_start( // hivectl's wait loop. let mut queued: Vec = Vec::new(); if !agents.is_empty() { - queued.push( - crate::job_queue::submit::start_many( - coord, - agents, - crate::job_queue::Source::Manual, - "manual via hivectl start".to_owned(), - ) - .await, - ); - ok_items.extend(agents.iter().cloned()); + match crate::job_queue::power::start_many(coord, agents).await { + Ok(ids) => { + queued.extend(ids.into_iter().map(hive_jobq::NodeId::get)); + ok_items.extend(agents.iter().cloned()); + } + Err(e) => errors.push(format!("queue start: {e}")), + } } let mut resp = finish_lifecycle(ok_items, &errors); @@ -982,28 +978,20 @@ async fn handle_restart_scoped( let mut errors: Vec = Vec::new(); let mut queued: Vec = Vec::new(); - // One DAG for all targeted agents — a per-agent restart subgraph each - // (`SetWanted → [Signal → Drain →] StopForUpdate → Reconcile`), - // independent roots that run concurrently on their own leases. A - // hive-wide `hivectl restart` is now a single DAG, not N. No - // client-side stop-then-start composition — the whole restart survives - // a dropped connection because the DAG owns it. + // One insert for all targeted agents — a per-agent restart subgraph each + // (`[Signal → Drain →] StopForUpdate → Reconcile`; no `SetWanted`, since a + // restart converges to the agent's *existing* intent rather than rewriting + // it), independent roots running concurrently on their own leases. No + // client-side stop-then-start composition — the whole restart survives a + // dropped connection because the graph owns it. if !agents.is_empty() { - queued.push( - crate::job_queue::submit::restart_many( - coord, - &agents, - graceful, - crate::job_queue::Source::Manual, - if graceful { - "manual via hivectl restart --graceful".to_owned() - } else { - "manual restart via hivectl restart".to_owned() - }, - ) - .await, - ); - ok_items.extend(agents.iter().cloned()); + match crate::job_queue::power::restart_many(coord, &agents, graceful).await { + Ok(ids) => { + queued.extend(ids.into_iter().map(hive_jobq::NodeId::get)); + ok_items.extend(agents.iter().cloned()); + } + Err(e) => errors.push(format!("queue restart: {e}")), + } } for &container in &infra {