From fe52037b0d004f96b89bc15debebb2c97c70c802 Mon Sep 17 00:00:00 2001 From: atlas Date: Tue, 4 Aug 2026 17:41:15 +0200 Subject: [PATCH] wip(#3001): rename insert -> insert_job per mara's 50056 --- hive-c0re/src/actions.rs | 6 ++--- hive-c0re/src/dashboard/lifecycle_ops.rs | 4 +-- hive-c0re/src/job_queue/mod.rs | 27 +++++++++---------- hive-c0re/src/job_queue/power.rs | 6 ++--- hive-c0re/src/job_queue/tests.rs | 2 +- .../src/socket_server/lifecycle_handlers.rs | 2 +- hive-c0re/src/workers/auto_update.rs | 4 +-- 7 files changed, 24 insertions(+), 27 deletions(-) diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index 8c531076..11c94c2f 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -55,7 +55,7 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { // nothing). let inputs: Vec = 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, 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, 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, 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() }) { diff --git a/hive-c0re/src/dashboard/lifecycle_ops.rs b/hive-c0re/src/dashboard/lifecycle_ops.rs index b4f399e0..0b864107 100644 --- a/hive-c0re/src/dashboard/lifecycle_ops.rs +++ b/hive-c0re/src/dashboard/lifecycle_ops.rs @@ -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) -> 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() }) { diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 4c5e74f2..51febed8 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -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, ) -> anyhow::Result> { diff --git a/hive-c0re/src/job_queue/power.rs b/hive-c0re/src/job_queue/power.rs index 67d2650e..3ed1d4e1 100644 --- a/hive-c0re/src/job_queue/power.rs +++ b/hive-c0re/src/job_queue/power.rs @@ -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, 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() })?; diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index b6e260fe..bd7cecb6 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -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() }) diff --git a/hive-c0re/src/socket_server/lifecycle_handlers.rs b/hive-c0re/src/socket_server/lifecycle_handlers.rs index 968d31a3..c0261405 100644 --- a/hive-c0re/src/socket_server/lifecycle_handlers.rs +++ b/hive-c0re/src/socket_server/lifecycle_handlers.rs @@ -145,7 +145,7 @@ pub(super) fn handle_update(coord: &Arc, 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() }) { diff --git a/hive-c0re/src/workers/auto_update.rs b/hive-c0re/src/workers/auto_update.rs index a629e73f..0084b102 100644 --- a/hive-c0re/src/workers/auto_update.rs +++ b/hive-c0re/src/workers/auto_update.rs @@ -109,7 +109,7 @@ pub async fn ensure_root_agent(coord: &Arc) -> 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() }) {