vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

wip(#3001): remove submit layer, rescue power ops into job_queue/power.rs

Browse source 
TREE IS RED ON PURPOSE — there is no compiling intermediate between
deleting submit and converting every caller. Checkpoint commit so the
work is durable; do not "fix" it by restoring submit.

Done:
- JobQueue::submit -> JobQueue::insert (no source/reason/container;
  returns the ids insert_job names).
- submit.rs deleted. Its 6 pure chain builders + 3 async *_many
  gatherers were NOT wrapper code and are rescued into
  job_queue/power.rs (templates.rs documents power ops as living
  outside it, because their shape needs a live is_running read).
- Converted: meta_inputs 1, topology 2, permissions 3, auto_update 2,
  actions 3, lifecycle_handlers 3.
- Dropped source/reason at every converted site: nothing ever read
  NodeKind::Dag's fields (only `{ .. }` matches exist), so they are
  write-only. Dead reason-only locals deleted; the boot sweep's summary
  became a tracing::info! rather than being lost.

Remaining: dashboard/lifecycle_ops 7, server.rs 7, and the test suite —
tests.rs has its own submit() helper whose u64 return is used as the
handle to navigate the inserted DAG, so those need a different way to
find nodes, not a mechanical port.
This commit is contained in:
atlas 2026-08-04 11:17:37 +02:00 • committed by mara
commit 7c0d9d2379
9 changed files with 198 additions and 301 deletions
Download patch file Download diff file Expand all files Collapse all files

52
hive-c0re/src/actions.rs
View file

@ -55,13 +55,12 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
// nothing). // nothing).
let inputs: Vec<String> = let inputs: Vec<String> =
serde_json::from_str(&approval.commit_ref).unwrap_or_default(); serde_json::from_str(&approval.commit_ref).unwrap_or_default();
let submitted = coord.job_queue.submit( let inserted = coord.job_queue.insert(|b| {
crate::job_queue::Source::Approval, crate::job_queue::templates::meta_update(b, inputs, Some(id));
format!("approval #{id} meta input update"), Vec::new()
|b| crate::job_queue::templates::meta_update(b, inputs, Some(id)), });
); if let Err(e) = inserted {
if let Err(e) = submitted { return Err(e.context("insert meta-update dag"));
return Err(e.context("submit meta-update dag"));
} }
coord.emit_rebuild_queue_snapshot(); coord.emit_rebuild_queue_snapshot();
Ok(()) Ok(())
@ -75,13 +74,12 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
{ {
tracing::warn!(agent = %approval.agent, error = ?e, "agent_power: seed on spawn failed"); tracing::warn!(agent = %approval.agent, error = ?e, "agent_power: seed on spawn failed");
} }
let submitted = coord.job_queue.submit( let inserted = coord.job_queue.insert(|b| {
crate::job_queue::Source::Approval, crate::job_queue::templates::spawn(b, approval.agent.as_str(), id);
format!("approval #{id} spawn"), Vec::new()
|b| crate::job_queue::templates::spawn(b, approval.agent.as_str(), id), });
); if let Err(e) = inserted {
if let Err(e) = submitted { return Err(e.context("insert spawn dag"));
return Err(e.context("submit spawn dag"));
} }
coord.emit_rebuild_queue_snapshot(); coord.emit_rebuild_queue_snapshot();
Ok(()) Ok(())
@ -106,12 +104,7 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
// `run_deploy_apply` (ff-merge, then grows the rebuild subgraph), // `run_deploy_apply` (ff-merge, then grows the rebuild subgraph),
// `run_finalize_deploy` (deploy tag + lock commit) and // `run_finalize_deploy` (deploy tag + lock commit) and
// `run_deploy_tail` (compensation + forge mirror). // `run_deploy_tail` (compensation + forge mirror).
enqueue_approval_rebuild( enqueue_approval_rebuild(&coord, approval.agent.as_str(), id);
&coord,
approval.agent.as_str(),
id,
format!("approval #{id} merge config pr"),
);
Ok(()) Ok(())
} }
} }
@ -121,19 +114,12 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
/// dispatch arm — the work ends in a container rebuild routed through the /// dispatch arm — the work ends in a container rebuild routed through the
/// queue. See [`crate::job_queue::templates::approval_deploy`] for the node /// queue. See [`crate::job_queue::templates::approval_deploy`] for the node
/// shape; the executor dispatches each node to the `run_deploy_*` bodies below. /// shape; the executor dispatches each node to the `run_deploy_*` bodies below.
fn enqueue_approval_rebuild( fn enqueue_approval_rebuild(coord: &Arc<Coordinator>, agent: &str, approval_id: i64) {
coord: &Arc<Coordinator>, if let Err(e) = coord.job_queue.insert(|b| {
agent: &str, crate::job_queue::templates::approval_deploy(b, agent, approval_id);
approval_id: i64, Vec::new()
reason: String, }) {
) { tracing::error!(%agent, approval_id, error = ?e, "insert approval deploy dag failed");
if let Err(e) = coord
.job_queue
.submit(crate::job_queue::Source::Approval, reason, |b| {
crate::job_queue::templates::approval_deploy(b, agent, approval_id);
})
{
tracing::error!(%agent, approval_id, error = ?e, "submit approval deploy dag failed");
} }
coord.emit_rebuild_queue_snapshot(); coord.emit_rebuild_queue_snapshot();
} }

16
hive-c0re/src/dashboard/meta_inputs.rs
View file

@ -208,16 +208,18 @@ pub(super) async fn post_meta_update(
if inputs.is_empty() { if inputs.is_empty() {
return error_response("meta-update: no inputs selected"); return error_response("meta-update: no inputs selected");
} }
let inputs_label = inputs.join(", ");
// Cascade rebuild children fan out from the MetaLock node when the // Cascade rebuild children fan out from the MetaLock node when the
// lock bump lands — appended by the scheduler so they build against // lock bump lands — appended by the scheduler so they build against
// the post-bump lock, and a failed bump simply fans out nothing. // the post-bump lock, and a failed bump simply fans out nothing.
crate::job_queue::submit::meta_update( state
&state.coord, .coord
inputs, .job_queue
crate::job_queue::Source::Manual, .insert(|b| {
format!("meta-update via dashboard ({inputs_label})"), crate::job_queue::templates::meta_update(b, inputs, None);
); Vec::new()
})
.expect("template-declared shapes are acyclic");
state.coord.emit_rebuild_queue_snapshot();
(StatusCode::OK, "ok").into_response() (StatusCode::OK, "ok").into_response()
} }

68
hive-c0re/src/dashboard/permissions.rs
View file

@ -155,15 +155,21 @@ pub(super) async fn post_tool_groups(
// META_LOCK inside the WritePermFile node, so concurrent // META_LOCK inside the WritePermFile node, so concurrent
// batch-apply actions for different agents never race on the // batch-apply actions for different agents never race on the
// shared tool-groups.json. // shared tool-groups.json.
crate::job_queue::submit::perm_change( state
&state.coord, .coord
&logical, .job_queue
crate::job_queue::Source::Manual, .insert(|b| {
"tool-group change via permissions UI".to_owned(), crate::job_queue::templates::perm_change(
crate::job_queue::PermPayload::ToolGroups { b,
groups: body.groups.clone(), &logical,
}, crate::job_queue::PermPayload::ToolGroups {
); groups: body.groups.clone(),
},
);
Vec::new()
})
.expect("template-declared shapes are acyclic");
state.coord.emit_rebuild_queue_snapshot();
tracing::info!(agent = %logical, groups = ?body.groups, "operator: set tool-groups via dashboard"); tracing::info!(agent = %logical, groups = ?body.groups, "operator: set tool-groups via dashboard");
Ok((StatusCode::OK, "ok").into_response()) Ok((StatusCode::OK, "ok").into_response())
} }
@ -264,15 +270,21 @@ pub(super) async fn post_capabilities(
// META_LOCK inside the WritePermFile node, so concurrent // META_LOCK inside the WritePermFile node, so concurrent
// batch-apply actions for different agents never race on the // batch-apply actions for different agents never race on the
// shared capabilities.json. // shared capabilities.json.
crate::job_queue::submit::perm_change( state
&state.coord, .coord
&logical, .job_queue
crate::job_queue::Source::Manual, .insert(|b| {
"capability change via dashboard".to_owned(), crate::job_queue::templates::perm_change(
crate::job_queue::PermPayload::Capabilities { b,
caps: body.caps.clone(), &logical,
}, crate::job_queue::PermPayload::Capabilities {
); caps: body.caps.clone(),
},
);
Vec::new()
})
.expect("template-declared shapes are acyclic");
state.coord.emit_rebuild_queue_snapshot();
tracing::info!(agent = %logical, caps = ?body.caps, "operator: set capabilities via dashboard"); tracing::info!(agent = %logical, caps = ?body.caps, "operator: set capabilities via dashboard");
Ok((StatusCode::OK, "ok").into_response()) Ok((StatusCode::OK, "ok").into_response())
} }
@ -359,13 +371,19 @@ pub(super) async fn post_permissions(
} }
// Phase 2 — submit one combined PermChange DAG per affected agent. // Phase 2 — submit one combined PermChange DAG per affected agent.
for (logical, groups, caps) in staged { for (logical, groups, caps) in staged {
crate::job_queue::submit::perm_change( state
&state.coord, .coord
&logical, .job_queue
crate::job_queue::Source::Manual, .insert(|b| {
"batch permission change via permissions UI".to_owned(), crate::job_queue::templates::perm_change(
crate::job_queue::PermPayload::Combined { groups, caps }, b,
); &logical,
crate::job_queue::PermPayload::Combined { groups, caps },
);
Vec::new()
})
.expect("template-declared shapes are acyclic");
state.coord.emit_rebuild_queue_snapshot();
tracing::info!(agent = %logical, "operator: batch perm change via dashboard"); tracing::info!(agent = %logical, "operator: batch perm change via dashboard");
} }
Ok((StatusCode::OK, "ok").into_response()) Ok((StatusCode::OK, "ok").into_response())

30
hive-c0re/src/dashboard/topology.rs
View file

@ -94,12 +94,15 @@ pub(super) async fn post_set_parent(
new_parent = ?new_parent, new_parent = ?new_parent,
"operator: set-parent via dashboard" "operator: set-parent via dashboard"
); );
submit::reparent( state
&state.coord, .coord
vec![(child, new_parent)], .job_queue
Source::Manual, .insert(|b| {
"manual set-parent via dashboard".to_owned(), crate::job_queue::templates::reparent(b, vec![(child, new_parent)]);
); Vec::new()
})
.expect("template-declared shapes are acyclic");
state.coord.emit_rebuild_queue_snapshot();
Ok((StatusCode::OK, "ok").into_response()) Ok((StatusCode::OK, "ok").into_response())
} }
@ -150,11 +153,14 @@ pub(super) async fn post_set_parent_bulk(
.map_err(|e| error_problem(&e))?; .map_err(|e| error_problem(&e))?;
let names: Vec<&str> = body.iter().map(|e| e.child.as_str()).collect(); let names: Vec<&str> = body.iter().map(|e| e.child.as_str()).collect();
tracing::info!(agents = ?names, "operator: set-parent-bulk via dashboard"); tracing::info!(agents = ?names, "operator: set-parent-bulk via dashboard");
submit::reparent( state
&state.coord, .coord
moves, .job_queue
Source::Manual, .insert(|b| {
"manual set-parent-bulk via dashboard".to_owned(), crate::job_queue::templates::reparent(b, moves);
); Vec::new()
})
.expect("template-declared shapes are acyclic");
state.coord.emit_rebuild_queue_snapshot();
Ok((StatusCode::OK, "ok").into_response()) Ok((StatusCode::OK, "ok").into_response())
} }

19
hive-c0re/src/job_queue/mod.rs
View file

@ -29,9 +29,9 @@
pub mod exec; pub mod exec;
pub mod model; pub mod model;
pub mod power;
pub mod resource; pub mod resource;
pub mod scheduler; pub mod scheduler;
pub mod submit;
pub mod templates; pub mod templates;
#[cfg(test)] #[cfg(test)]
mod tests; mod tests;
@ -208,20 +208,17 @@ impl JobQueue {
/// # Errors /// # Errors
/// Propagates a graph-insert error (dependencies that aren't /// Propagates a graph-insert error (dependencies that aren't
/// dependency-topological). /// dependency-topological).
pub fn submit( pub fn insert(
&self, &self,
source: Source, declare: impl FnOnce(&JobBuilder) -> Vec<hive_jobq::NodeGuid>,
reason: String, ) -> anyhow::Result<Vec<NodeId>> {
declare: impl FnOnce(&JobBuilder),
) -> anyhow::Result<u64> {
let mut inner = self.lock(); let mut inner = self.lock();
let container = inner let named = inner
.append(NodeKind::Dag { source, reason }, Vec::new(), None) .insert_job(None, declare)
.map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?; .map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?;
insert_group(&mut inner, declare, Some(container))?;
drop(inner); drop(inner);
self.notify.notify_one(); self.notify.notify_one();
Ok(container.get()) Ok(named)
} }
/// The scheduler itself, for `hive_jobq`'s run-loop seam /// The scheduler itself, for `hive_jobq`'s run-loop seam

228
hive-c0re/src/job_queue/submit.rs → hive-c0re/src/job_queue/power.rs
View file

@ -1,59 +1,26 @@
//! Request-level submit API — the surface the dashboard POST handlers, //! Power ops (`stop` / `start` / `restart`) — the DAG shapes whose per-agent
//! the MCP socket handlers, and `hivectl` paths call. //! form depends on **live** container state, so they cannot be static
//! [`super::templates`] entries.
//! //!
//! The **power ops** (`stop` / `start` / `restart`) are built here, not in //! The split is the purity line, not the subject matter: the `*_chain` /
//! `templates.rs`: each agent's subgraph shape depends on its *live* running //! `*_nodes` builders below are pure (they take `running` / `stale` as
//! state, which needs an async `lifecycle::is_running` read that a pure/sync //! parameters, which is what keeps them unit-testable without a container),
//! template can't do. So these fns are async — they read each agent's state, //! and only the `*_many` entry points do the async `lifecycle::is_running`
//! assemble a per-agent subgraph out of the shared pure primitives //! read that produces those parameters.
//! (`JobBuilder::node` + `templates::rebuild_nodes`), all declaring into ONE job
//! (independent per-agent roots, concurrent on their own leases).
//! //!
//! Dynamic shape rule: `stop`/`start` carry a head `SetWanted(w)` (durable //! Each entry point declares its group and inserts it. There is no metadata
//! intent write) — `restart` does NOT (it bounces the container but leaves //! parameter and no container node: attribution is not something every caller
//! `wanted` untouched, so a deliberately-stopped agent isn't forced up). The //! has to invent, and a DAG is addressed by the nodes a template names.
//! tail `Reconcile` (the convergence guarantee — cheap, noops when already
//! converged) is ALWAYS present; only the *mechanical* nodes
//! (`Signal`/`Drain`/`StopForUpdate`) are state-conditional (skipped for a
//! down agent — nothing to quiesce/stop). Keeping `Reconcile` in every shape
//! closes the TOCTOU window: if an agent flips state between the `is_running`
//! read and node execution, the tail `Reconcile` still converges it in-DAG,
//! with `StopForUpdate`-noop as the backstop — no reliance on an external
//! reconcile sweep. Every helper emits a fresh queue snapshot so the
//! dashboard shows the new DAG immediately.
use std::sync::Arc; use std::sync::Arc;
use super::model::NodeKind; use hive_jobq::NodeId;
use super::resource::Resource;
use super::templates::rebuild_nodes; use super::JobBuilder;
use super::{JobBuilder, Source, templates}; use super::templates;
use crate::coordinator::Coordinator; use crate::coordinator::Coordinator;
use crate::lifecycle; use crate::lifecycle;
fn submit_and_emit(
coord: &Arc<Coordinator>,
source: Source,
reason: String,
declare: impl FnOnce(&JobBuilder),
) -> u64 {
let id = coord
.job_queue
.submit(source, reason, declare)
.expect("template-declared shapes are acyclic");
coord.emit_rebuild_queue_snapshot();
id
}
/// Manual/approval-independent rebuild (always relocks the agent's
/// meta input — the meta-update cascade grows its own rebuild subgraphs
/// in-DAG instead of going through this surface).
pub fn rebuild(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 {
submit_and_emit(coord, source, reason, |builder| {
templates::rebuild(builder, agent, true);
})
}
// ---- dynamic power-op DAG assembly ---------------------------------------- // ---- dynamic power-op DAG assembly ----------------------------------------
// //
// The pure per-agent chain builders below take `running` (and `stale`) // The pure per-agent chain builders below take `running` (and `stale`)
@ -226,63 +193,44 @@ pub(crate) fn restart_nodes(builder: &JobBuilder, targets: &[(String, bool)], gr
} }
} }
/// Restart a single agent. Thin wrapper over [`restart_many`]. // ---- entry points ---------------------------------------------------------
pub async fn restart(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 { //
restart_many(coord, &[agent.to_owned()], false, source, reason).await // Each reads the live state its shape depends on, then declares + inserts.
}
/// Graceful restart of a single agent (signal → drain → stop → reconcile, /// Restart `agents` in a **single** DAG — one per-agent subgraph each, built
/// when running). Thin wrapper over [`restart_many`] with `graceful = true`. /// from live running state and run concurrently on their own leases. A running
pub async fn graceful_restart( /// agent gets the stop→reconcile chain (`graceful` prepends signal→drain); a
coord: &Arc<Coordinator>, /// down agent gets a lone `Reconcile`. Restart never writes `wanted`, so the
agent: &str,
source: Source,
reason: String,
) -> u64 {
restart_many(coord, &[agent.to_owned()], true, source, reason).await
}
/// Restart `agents` (one or many) in a **single** DAG — one per-agent
/// subgraph each, built dynamically from live running state and run
/// concurrently on their own leases. A running agent gets the stop→reconcile
/// chain (`graceful` prepends signal→drain); a down agent gets just a lone
/// `Reconcile` (nothing to stop). Restart never writes `wanted`, so the
/// tail `Reconcile` converges each agent to its EXISTING intent — a /// tail `Reconcile` converges each agent to its EXISTING intent — a
/// deliberately-stopped agent stays down. The whole hive-wide /// deliberately-stopped agent stays down.
/// `hivectl restart` is one DAG. ///
/// # Errors
/// Propagates a graph-insert error.
pub async fn restart_many( pub async fn restart_many(
coord: &Arc<Coordinator>, coord: &Arc<Coordinator>,
agents: &[String], agents: &[String],
graceful: bool, graceful: bool,
source: Source, ) -> anyhow::Result<Vec<NodeId>> {
reason: String,
) -> u64 {
let mut targets = Vec::with_capacity(agents.len()); let mut targets = Vec::with_capacity(agents.len());
for agent in agents { for agent in agents {
targets.push((agent.clone(), lifecycle::is_running(agent).await)); targets.push((agent.clone(), lifecycle::is_running(agent).await));
} }
submit_and_emit(coord, source, reason, |builder| { let ids = coord.job_queue.insert(|b| {
restart_nodes(builder, &targets, graceful); restart_nodes(b, &targets, graceful);
}) Vec::new()
})?;
coord.emit_rebuild_queue_snapshot();
Ok(ids)
} }
/// Start a single agent. Thin wrapper over [`start_many`]. /// Start `agents` in a **single** DAG. A down agent gets
pub async fn start(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 { /// `SetWanted(Up) → Reconcile` (or, rev stale, a rebuild-then-start so it comes
start_many(coord, &[agent.to_owned()], source, reason).await /// up on current derivations); an already-running agent gets the same shape
} /// with the reconcile noop'ing.
///
/// Start `agents` (one or many) in a **single** DAG — one per-agent subgraph /// # Errors
/// each, built dynamically from live state and run concurrently on their own /// Propagates a graph-insert error.
/// leases. A down agent gets `SetWanted(Up) → Reconcile` (or, rev stale, a pub async fn start_many(coord: &Arc<Coordinator>, agents: &[String]) -> anyhow::Result<Vec<NodeId>> {
/// rebuild-then-start so it comes up on current derivations); an already-
/// running agent gets `SetWanted(Up) → Reconcile` (the reconcile noops). The
/// whole hive-wide `hivectl start` is one DAG.
pub async fn start_many(
coord: &Arc<Coordinator>,
agents: &[String],
source: Source,
reason: String,
) -> u64 {
let current = crate::auto_update::current_flake_rev(&coord.hyperhive_flake); let current = crate::auto_update::current_flake_rev(&coord.hyperhive_flake);
let mut targets = Vec::with_capacity(agents.len()); let mut targets = Vec::with_capacity(agents.len());
for agent in agents { for agent in agents {
@ -296,88 +244,34 @@ pub async fn start_many(
} }
targets.push((agent.clone(), running, stale)); targets.push((agent.clone(), running, stale));
} }
submit_and_emit(coord, source, reason, |builder| { let ids = coord.job_queue.insert(|b| {
start_nodes(builder, &targets); start_nodes(b, &targets);
}) Vec::new()
})?;
coord.emit_rebuild_queue_snapshot();
Ok(ids)
} }
/// Hard stop a single agent. Thin wrapper over [`stop_many`]. /// Stop `agents` in a **single** DAG. A running agent gets
pub async fn stop(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 { /// `SetWanted(Off) → [Signal → Drain →](graceful) Reconcile`; a down agent
stop_many(coord, &[agent.to_owned()], false, source, reason).await /// skips the pointless quiesce but keeps the `Reconcile` as the race-up
} /// backstop.
///
/// Graceful stop of a single agent (signal → drain → reconcile, when /// # Errors
/// running). Thin wrapper over [`stop_many`] with `graceful = true`. /// Propagates a graph-insert error.
pub async fn graceful_stop(
coord: &Arc<Coordinator>,
agent: &str,
source: Source,
reason: String,
) -> u64 {
stop_many(coord, &[agent.to_owned()], true, source, reason).await
}
/// Stop `agents` (one or many) in a **single** DAG — one per-agent subgraph
/// each, built dynamically from live state and run concurrently on their own
/// leases. A running agent gets `SetWanted(Off) → [Signal → Drain →](graceful)
/// Reconcile`; a down agent gets just `SetWanted(Off) → Reconcile` (skips the
/// pointless quiesce, keeps the Reconcile as the race-up backstop). The whole
/// hive-wide `hivectl stop` is one DAG.
pub async fn stop_many( pub async fn stop_many(
coord: &Arc<Coordinator>, coord: &Arc<Coordinator>,
agents: &[String], agents: &[String],
graceful: bool, graceful: bool,
source: Source, ) -> anyhow::Result<Vec<NodeId>> {
reason: String,
) -> u64 {
let mut targets = Vec::with_capacity(agents.len()); let mut targets = Vec::with_capacity(agents.len());
for agent in agents { for agent in agents {
targets.push((agent.clone(), lifecycle::is_running(agent).await)); targets.push((agent.clone(), lifecycle::is_running(agent).await));
} }
submit_and_emit(coord, source, reason, |builder| { let ids = coord.job_queue.insert(|b| {
stop_nodes(builder, &targets, graceful); stop_nodes(b, &targets, graceful);
}) Vec::new()
} })?;
coord.emit_rebuild_queue_snapshot();
/// Perm change: commit the JSON file(s) then rebuild. Ok(ids)
pub fn perm_change(
coord: &Arc<Coordinator>,
agent: &str,
source: Source,
reason: String,
payload: super::PermPayload,
) -> u64 {
submit_and_emit(coord, source, reason, |builder| {
templates::perm_change(builder, agent, payload);
})
}
/// Meta-input lock bump; cascade rebuilds fan out on completion.
pub fn meta_update(
coord: &Arc<Coordinator>,
inputs: Vec<String>,
source: Source,
reason: String,
) -> u64 {
submit_and_emit(coord, source, reason, |builder| {
templates::meta_update(builder, inputs, None);
})
}
/// Topology move(s) as a queue DAG. `moves` is `(child, new_parent)` pairs —
/// one entry for `set-parent`, N for `set-parent-bulk`. Fire-and-forget like
/// everything else in this module: submits and returns a DAG id
/// immediately, the caller learns the outcome async (dashboard job view /
/// `hivectl`'s `QueueDag` poll). Wired from `server.rs`'s `HostRequest::
/// SetParent` (hivectl) and `dashboard/topology.rs`'s `set-parent`/
/// `set-parent-bulk` handlers.
pub fn reparent(
coord: &Arc<Coordinator>,
moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>,
source: Source,
reason: String,
) -> u64 {
submit_and_emit(coord, source, reason, |builder| {
templates::reparent(builder, moves);
})
} }

18
hive-c0re/src/job_queue/tests.rs
View file

@ -36,17 +36,17 @@ fn rebuild(builder: &JobBuilder, agent: &str) {
/// Restart shape with every agent treated as **running** — the online /// Restart shape with every agent treated as **running** — the online
/// shape (`[Signal→Drain→] StopForUpdate → Reconcile`, no `SetWanted` head) /// shape (`[Signal→Drain→] StopForUpdate → Reconcile`, no `SetWanted` head)
/// most queue-mechanics tests assume. Mirrors the pre-dynamic /// most queue-mechanics tests assume. Mirrors the pre-dynamic
/// `templates::restart` (which is now the state-aware `submit::restart_nodes`). /// `templates::restart` (which is now the state-aware `power::restart_nodes`).
fn restart_online(builder: &JobBuilder, agents: &[&str], graceful: bool) { fn restart_online(builder: &JobBuilder, agents: &[&str], graceful: bool) {
let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect(); let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect();
submit::restart_nodes(builder, &targets, graceful); power::restart_nodes(builder, &targets, graceful);
} }
/// Stop shape with every agent treated as **running** — the online shape /// Stop shape with every agent treated as **running** — the online shape
/// (`SetWanted → [Signal→Drain→](graceful) Reconcile`). /// (`SetWanted → [Signal→Drain→](graceful) Reconcile`).
fn stop_online(builder: &JobBuilder, agents: &[&str], graceful: bool) { fn stop_online(builder: &JobBuilder, agents: &[&str], graceful: bool) {
let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect(); let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect();
submit::stop_nodes(builder, &targets, graceful); power::stop_nodes(builder, &targets, graceful);
} }
// `Claimed` / `ClaimReady` / `CompleteNode` lived here: a claim snapshot type // `Claimed` / `ClaimReady` / `CompleteNode` lived here: a claim snapshot type
@ -682,7 +682,7 @@ fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() {
// fresh: offline + not stale → SetWanted → Reconcile. // fresh: offline + not stale → SetWanted → Reconcile.
// stale: offline + stale → SetWanted → «rebuild subgraph». // stale: offline + stale → SetWanted → «rebuild subgraph».
let id = submit(&q, "hive-wide start", |builder| { let id = submit(&q, "hive-wide start", |builder| {
submit::start_nodes( power::start_nodes(
builder, builder,
&[ &[
("fresh".to_owned(), false, false), ("fresh".to_owned(), false, false),
@ -742,13 +742,13 @@ fn offline_agents_skip_mechanical_nodes_but_keep_reconcile() {
let q = JobQueue::new(4); let q = JobQueue::new(4);
// Offline graceful stop → SetWanted(Off) → Reconcile (no Signal/Drain). // Offline graceful stop → SetWanted(Off) → Reconcile (no Signal/Drain).
let stop = submit(&q, "stop down", |builder| { let stop = submit(&q, "stop down", |builder| {
submit::stop_nodes(builder, &[("down".to_owned(), false)], true); power::stop_nodes(builder, &[("down".to_owned(), false)], true);
}); });
// Offline restart → a lone Reconcile (no SetWanted, no StopForUpdate): // Offline restart → a lone Reconcile (no SetWanted, no StopForUpdate):
// nothing to bounce, and restart never rewrites intent, so the tail // nothing to bounce, and restart never rewrites intent, so the tail
// Reconcile converges the down agent to its existing `wanted`. // Reconcile converges the down agent to its existing `wanted`.
let restart = submit(&q, "restart down", |builder| { let restart = submit(&q, "restart down", |builder| {
submit::restart_nodes(builder, &[("down2".to_owned(), false)], true); power::restart_nodes(builder, &[("down2".to_owned(), false)], true);
}); });
let shape = |id: u64| -> Vec<String> { let shape = |id: u64| -> Vec<String> {
// The group's work nodes: its subtree minus the container itself, // The group's work nodes: its subtree minus the container itself,
@ -1153,19 +1153,19 @@ fn cancelled_power_op_runs_no_compensating_node() {
let q = JobQueue::new(1); let q = JobQueue::new(1);
let id = submit(&q, "bounce", |builder| { let id = submit(&q, "bounce", |builder| {
submit::restart_nodes(builder, &targets, graceful); power::restart_nodes(builder, &targets, graceful);
}); });
assert_cancels_clean(&q, id, false, &format!("restart {case}")); assert_cancels_clean(&q, id, false, &format!("restart {case}"));
let q = JobQueue::new(1); let q = JobQueue::new(1);
let id = submit(&q, "stop", |builder| { let id = submit(&q, "stop", |builder| {
submit::stop_nodes(builder, &targets, graceful); power::stop_nodes(builder, &targets, graceful);
}); });
assert_cancels_clean(&q, id, true, &format!("stop {case}")); assert_cancels_clean(&q, id, true, &format!("stop {case}"));
let q = JobQueue::new(1); let q = JobQueue::new(1);
let id = submit(&q, "start", |builder| { let id = submit(&q, "start", |builder| {
submit::start_nodes(builder, &[("agent-a".to_owned(), running, false)]); power::start_nodes(builder, &[("agent-a".to_owned(), running, false)]);
}); });
assert_cancels_clean(&q, id, true, &format!("start {case}")); assert_cancels_clean(&q, id, true, &format!("start {case}"));
} }

33
hive-c0re/src/socket_server/lifecycle_handlers.rs
View file

@ -20,13 +20,9 @@ pub(super) async fn handle_start(coord: &Arc<Coordinator>, agent: &str, name: &s
// Persist `wanted = Up` and submit the Start DAG; the submit layer // Persist `wanted = Up` and submit the Start DAG; the submit layer
// upgrades a stale-rev start to a full rebuild so the container // upgrades a stale-rev start to a full rebuild so the container
// runs current nix derivations before it starts. // runs current nix derivations before it starts.
crate::job_queue::submit::start( if let Err(e) = crate::job_queue::power::start_many(coord, &[name.to_owned()]).await {
coord, tracing::error!(%agent, %name, error = ?e, "start: insert failed");
name, }
crate::job_queue::Source::Manual,
format!("agent `{agent}` start tool"),
)
.await;
Response::Ok Response::Ok
} }
@ -48,13 +44,9 @@ pub(super) async fn handle_restart(coord: &Arc<Coordinator>, agent: &str, name:
return err; return err;
} }
tracing::info!(%agent, %name, "submit restart"); tracing::info!(%agent, %name, "submit restart");
crate::job_queue::submit::restart( if let Err(e) = crate::job_queue::power::restart_many(coord, &[name.to_owned()], false).await {
coord, tracing::error!(%agent, %name, error = ?e, "restart: insert failed");
name, }
crate::job_queue::Source::Manual,
format!("agent `{agent}` restart tool"),
)
.await;
Response::Ok Response::Ok
} }
@ -153,12 +145,13 @@ pub(super) fn handle_update(coord: &Arc<Coordinator>, agent: &str, name: &str) -
return err; return err;
} }
tracing::info!(%agent, %name, "submit rebuild"); tracing::info!(%agent, %name, "submit rebuild");
crate::job_queue::submit::rebuild( if let Err(e) = coord.job_queue.insert(|b| {
coord, crate::job_queue::templates::rebuild(b, name, true);
name, Vec::new()
crate::job_queue::Source::Manual, }) {
format!("agent `{agent}` update tool"), tracing::error!(%agent, %name, error = ?e, "update: insert failed");
); }
coord.emit_rebuild_queue_snapshot();
Response::Ok Response::Ok
} }

35
hive-c0re/src/workers/auto_update.rs
View file

@ -109,12 +109,11 @@ pub async fn ensure_root_agent(coord: &Arc<Coordinator>) -> Result<()> {
tracing::warn!( tracing::warn!(
"manager container exists but no applied flake — forcing rebuild to migrate" "manager container exists but no applied flake — forcing rebuild to migrate"
); );
if let Err(e) = coord.job_queue.submit( if let Err(e) = coord.job_queue.insert(|b| {
crate::job_queue::Source::AutoUpdate, crate::job_queue::templates::rebuild(b, MANAGER_NAME, true);
"manager migration: no applied flake".to_owned(), Vec::new()
|b| crate::job_queue::templates::rebuild(b, MANAGER_NAME, true), }) {
) { tracing::warn!(error = ?e, "manager migration rebuild insert failed");
tracing::warn!(error = ?e, "manager migration rebuild submit failed");
} }
} else { } else {
tracing::debug!("manager container already present"); tracing::debug!("manager container already present");
@ -378,18 +377,19 @@ fn submit_boot_tree(
n_deferred: usize, n_deferred: usize,
n_skipped: usize, n_skipped: usize,
) { ) {
use crate::job_queue::Source; // Fully-quiet boot (nothing stale, nothing drifted) inserts nothing.
// Fully-quiet boot (nothing stale, nothing drifted) submits nothing.
if !any_stale && drifted.is_empty() { if !any_stale && drifted.is_empty() {
return; return;
} }
let reason = format!( // The summary the sweep used to hand the container as its `reason` is a log
"boot: {} rebuild(s), {} reconcile(s), {} deferred (offline), {} up-to-date", // line now: it was only ever stored on a node nobody read, and the counts
fanout.len(), // are worth having where they can actually be seen.
drifted.len(), tracing::info!(
n_deferred, rebuilds = fanout.len(),
n_skipped, reconciles = drifted.len(),
deferred = n_deferred,
up_to_date = n_skipped,
"boot: sweep"
); );
// The sweep's own rebuild subgraphs emit their `Rebuilt` events as they // The sweep's own rebuild subgraphs emit their `Rebuilt` events as they
@ -397,10 +397,11 @@ fn submit_boot_tree(
// The subgraphs also carry their own per-agent crash-watch suppression // The subgraphs also carry their own per-agent crash-watch suppression
// during their `Swap` (applied at claim time); a reconcile-only boot needs // during their `Swap` (applied at claim time); a reconcile-only boot needs
// no transient. // no transient.
if let Err(e) = coord.job_queue.submit(Source::AutoUpdate, reason, |b| { if let Err(e) = coord.job_queue.insert(|b| {
boot_nodes(b, any_stale, fanout, drifted); boot_nodes(b, any_stale, fanout, drifted);
Vec::new()
}) { }) {
tracing::warn!(error = ?e, "boot: sweep DAG submit failed"); tracing::warn!(error = ?e, "boot: sweep DAG insert failed");
} }
coord.emit_rebuild_queue_snapshot(); coord.emit_rebuild_queue_snapshot();
} }