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

wip(#3001): rename insert -> insert_job per mara's 50056

Browse source
This commit is contained in:
atlas 2026-08-04 17:41:15 +02:00 • committed by mara
commit fe52037b0d
7 changed files with 24 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -55,7 +55,7 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
// nothing).
let inputs: Vec<String> =
serde_json::from_str(&approval.commit_ref).unwrap_or_default();
let inserted = coord.job_queue.insert(|b| {
let inserted = coord.job_queue.insert_job(|b| {
crate::job_queue::templates::meta_update(b, inputs, Some(id));
Vec::new()
});
@ -74,7 +74,7 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
{
tracing::warn!(agent = %approval.agent, error = ?e, "agent_power: seed on spawn failed");
}
let inserted = coord.job_queue.insert(|b| {
let inserted = coord.job_queue.insert_job(|b| {
crate::job_queue::templates::spawn(b, approval.agent.as_str(), id);
Vec::new()
});
@ -115,7 +115,7 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
/// queue. See [`crate::job_queue::templates::approval_deploy`] for the node
/// shape; the executor dispatches each node to the `run_deploy_*` bodies below.
fn enqueue_approval_rebuild(coord: &Arc<Coordinator>, agent: &str, approval_id: i64) {
if let Err(e) = coord.job_queue.insert(|b| {
if let Err(e) = coord.job_queue.insert_job(|b| {
crate::job_queue::templates::approval_deploy(b, agent, approval_id);
Vec::new()
}) {

4
hive-c0re/src/dashboard/lifecycle_ops.rs
View file

@ -50,7 +50,7 @@ pub(super) async fn post_rebuild(
if let Some(reject) = guard_agent_name(&state, &logical).await {
return reject;
}
if let Err(e) = state.coord.job_queue.insert(|b| {
if let Err(e) = state.coord.job_queue.insert_job(|b| {
crate::job_queue::templates::rebuild(b, &logical, true);
Vec::new()
}) {
@ -399,7 +399,7 @@ pub(super) async fn post_update_all(State(state): State<AppState>) -> Response {
else {
continue;
};
if let Err(e) = state.coord.job_queue.insert(|b| {
if let Err(e) = state.coord.job_queue.insert_job(|b| {
crate::job_queue::templates::rebuild(b, &logical, true);
Vec::new()
}) {

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

@ -188,27 +188,24 @@ impl JobQueue {
self.sched.lock().expect("job_queue mutex poisoned")
}
/// Submit a DAG: insert a [`NodeKind::Dag`] **container node** carrying the
/// group's metadata, then insert the template's nodes as its subtree (their
/// roots re-parented to the container). Returns the container's id as the
/// DAG id — its rolled-up state is the DAG state.
/// Insert a job's nodes into the shared graph, then wake the run loop.
///
/// The container is an ordinary node: it declares no resources, so the
/// scheduler claims it on the next pass, runs its (empty) logic and parks
/// it in `Finishing`, at which point its children become runnable. Nothing
/// here completes it by hand — a node with no work of its own still goes
/// the way every other node goes.
/// Deliberately named for the [`hive_jobq`] primitive it wraps, because
/// that is nearly all it is. **The wrapper earns its place on the wake**:
/// the crate is sync and runtime-free — it holds no `Notify` at all — so
/// the channel the run loop parks on belongs to the host, and something has
/// to ping it. Left to call sites, an insert whose ping was forgotten would
/// leave a correct DAG sitting unscheduled until an unrelated event
/// happened along; nothing would fail, and no test in isolation would see
/// it.
///
/// `source` and `reason` are the container node's own payload — they are
/// arguments here rather than fields of a spec struct because that is all
/// they ever were. `declare` is the recipe, taken by generic and run
/// against a builder `hive_jobq` owns: it goes from the template straight
/// into this call, so there is nothing to allocate for.
/// Returns exactly what the primitive returns: the ids of the nodes the
/// template named, in the order it named them.
///
/// # Errors
/// Propagates a graph-insert error (dependencies that aren't
/// dependency-topological).
pub fn insert(
pub fn insert_job(
&self,
declare: impl FnOnce(&JobBuilder) -> Vec<hive_jobq::NodeGuid>,
) -> anyhow::Result<Vec<NodeId>> {

6
hive-c0re/src/job_queue/power.rs
View file

@ -216,7 +216,7 @@ pub async fn restart_many(
for agent in agents {
targets.push((agent.clone(), lifecycle::is_running(agent).await));
}
let ids = coord.job_queue.insert(|b| {
let ids = coord.job_queue.insert_job(|b| {
restart_nodes(b, &targets, graceful);
Vec::new()
})?;
@ -245,7 +245,7 @@ pub async fn start_many(coord: &Arc<Coordinator>, agents: &[String]) -> anyhow::
}
targets.push((agent.clone(), running, stale));
}
let ids = coord.job_queue.insert(|b| {
let ids = coord.job_queue.insert_job(|b| {
start_nodes(b, &targets);
Vec::new()
})?;
@ -269,7 +269,7 @@ pub async fn stop_many(
for agent in agents {
targets.push((agent.clone(), lifecycle::is_running(agent).await));
}
let ids = coord.job_queue.insert(|b| {
let ids = coord.job_queue.insert_job(|b| {
stop_nodes(b, &targets, graceful);
Vec::new()
})?;

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

@ -21,7 +21,7 @@ use super::*;
/// graph. A test that needs a handle calls `q.insert` directly and names the
/// node it cares about.
fn insert(q: &JobQueue, declare: impl FnOnce(&JobBuilder)) {
q.insert(|b| {
q.insert_job(|b| {
declare(b);
Vec::new()
})

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

@ -145,7 +145,7 @@ pub(super) fn handle_update(coord: &Arc<Coordinator>, agent: &str, name: &str) -
return err;
}
tracing::info!(%agent, %name, "submit rebuild");
if let Err(e) = coord.job_queue.insert(|b| {
if let Err(e) = coord.job_queue.insert_job(|b| {
crate::job_queue::templates::rebuild(b, name, true);
Vec::new()
}) {

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

@ -109,7 +109,7 @@ pub async fn ensure_root_agent(coord: &Arc<Coordinator>) -> Result<()> {
tracing::warn!(
"manager container exists but no applied flake — forcing rebuild to migrate"
);
if let Err(e) = coord.job_queue.insert(|b| {
if let Err(e) = coord.job_queue.insert_job(|b| {
crate::job_queue::templates::rebuild(b, MANAGER_NAME, true);
Vec::new()
}) {
@ -397,7 +397,7 @@ fn submit_boot_tree(
// The subgraphs also carry their own per-agent crash-watch suppression
// during their `Swap` (applied at claim time); a reconcile-only boot needs
// no transient.
if let Err(e) = coord.job_queue.insert(|b| {
if let Err(e) = coord.job_queue.insert_job(|b| {
boot_nodes(b, any_stale, fanout, drifted);
Vec::new()
}) {