From eb557ee3c1a5cb39fb120aadd03b0485f00a3923 Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 1 Aug 2026 15:09:05 +0200 Subject: [PATCH 1/2] refactor(#2908): cancel a node, not a DAG MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `JobQueue::cancel(dag_id)` resolved the id to a `NodeKind::Dag` container and cancelled that. But the container lookup was the only DAG-specific part — everything that makes cancel work already lives in the scheduler: `cancel_node` marks the node `Cancelled` and cascades to its pending descendants, sparing any node whose edge accepts `Cancelled` (which is what keeps a dropped approval DAG from dangling its row). So `cancel` now takes any node id. A group root cancels the whole group, which is what the dashboard's button does today and why nothing about its behaviour changes: a DAG id *is* its root node's id. An interior node cancels just that branch — a capability the DAG-scoped version could not express, covered by the new test (a hive-wide restart drops one agent's subgraph while the other keeps running). `QueueInner::node_by_id` replaces `container()` here: same search, same cost, without asserting the node is a DAG container. `container()` stays for `first_error` and the append-subgraph guard, which are genuinely DAG-scoped. No wire change. The route is `POST /api/rebuild-queue/{id}/cancel` with a `u64` path param — same type, same route, and the client keeps sending the same number. Only the param's documented meaning moves from "DAG id" to "node id". Checked with clippy (`--all-targets -D warnings`), `cargo test -p hive-c0re` (322 passed) and `nix fmt`. No option surface touched, so no nix-eval gate. --- hive-c0re/src/dashboard/schedules.rs | 13 ++++++++--- hive-c0re/src/job_queue/mod.rs | 21 +++++++++++++++--- hive-c0re/src/job_queue/tests.rs | 32 ++++++++++++++++++++++++++++ 3 files changed, 60 insertions(+), 6 deletions(-) diff --git a/hive-c0re/src/dashboard/schedules.rs b/hive-c0re/src/dashboard/schedules.rs index 0b0148ff..f5ea431f 100644 --- a/hive-c0re/src/dashboard/schedules.rs +++ b/hive-c0re/src/dashboard/schedules.rs @@ -153,8 +153,15 @@ pub(super) async fn post_schedule_fire_now( } } -/// `POST /api/rebuild-queue/{id}/cancel` — drop a still-fully-queued -/// DAG from the job queue. Refuses `Running` / terminal DAGs: an +/// `POST /api/rebuild-queue/{id}/cancel` — drop still-queued work from the job +/// queue. +/// +/// `id` is a **node** id. A DAG's root cancels the whole group (the scheduler +/// cascades to pending descendants), which is what the dashboard's cancel +/// button sends today — a DAG id *is* its root node's id. An interior node +/// cancels just that branch. +/// +/// Refuses `Running` / terminal nodes: an /// in-flight node owns the agent's nix store + nixos-container update /// lock and can't be safely interrupted from the queue side. Always /// returns 200; the body is `{"cancelled": true}` on a successful @@ -164,7 +171,7 @@ pub(super) async fn post_schedule_fire_now( #[utoipa::path( post, path = "/api/rebuild-queue/{id}/cancel", - params(("id" = u64, Path, description = "job-queue DAG id")), + params(("id" = u64, Path, description = "job-queue node id (a DAG's root cancels the group)")), responses((status = 200, description = "whether the DAG was cancelled", body = serde_json::Value)), tag = "schedules" )] diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index f0c27484..aa4be634 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -345,12 +345,16 @@ impl JobQueue { /// say) is cancelled along with everything else — there is nothing to converge /// when no node ever ran. Only a node that named `Cancelled` survives, and it /// survives because it asked to. - pub fn cancel(&self, dag_id: u64) -> bool { + /// `id` names **any node**, not specifically a DAG. Cancelling a group root + /// drops that whole group (the cascade is the scheduler's), which is what + /// the dashboard's whole-DAG cancel does; cancelling an interior node drops + /// just that branch. Nothing here knows about DAGs. + pub fn cancel(&self, id: u64) -> bool { let mut inner = self.lock(); - let Some(container) = inner.container(dag_id) else { + let Some(node) = inner.node_by_id(id) else { return false; }; - if !inner.sched.cancel_node(container) { + if !inner.sched.cancel_node(node) { return false; } drop(inner); @@ -460,6 +464,17 @@ impl QueueInner { /// The container node of `dag_id` — the `NodeKind::Dag` root whose id equals /// `dag_id`. `NodeId` is un-fabricable from a raw `u64`, so this is a search. + /// Resolve a raw wire `u64` to a graph [`NodeId`], whatever kind of node it + /// names. `NodeId` is un-fabricable from a `u64`, so this is a search — + /// same cost as [`QueueInner::container`], without asserting the node is a + /// DAG container. + fn node_by_id(&self, id: u64) -> Option { + self.sched + .graph() + .nodes() + .find_map(|n| (n.id.get() == id).then_some(n.id)) + } + fn container(&self, dag_id: u64) -> Option { self.sched.graph().nodes().find_map(|n| { (n.parent.is_none() diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 3f5681d1..5fd97c49 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -1010,6 +1010,38 @@ fn cancel_clears_queued_dag() { assert_eq!(state_of(&q, id), State::Cancelled); } +/// `cancel` takes a **node** id, not a DAG id — so an interior node can be +/// dropped without touching the rest of the group. +/// +/// This is the capability the DAG-scoped version couldn't express, and the +/// reason it reads naturally: a DAG id *is* its root node's id, so the +/// whole-group cancel every other test does is just this called on a root. +/// Here a hive-wide restart drops **one agent's** subgraph and the other agent +/// still runs. +#[test] +fn cancel_drops_one_agents_branch_leaving_the_rest() { + let q = JobQueue::new(2); + let id = submit(&q, restart_online(&["agent-a", "agent-b"], false, "r")); + // Per-agent subgraphs are independent roots; find agent-a's. + let snap = q.snapshot(); + let dag = snap.iter().find(|d| d.id == id).expect("dag in snapshot"); + let a_root = dag + .nodes + .iter() + .find(|n| n.agent == "agent-a" && n.parent.is_none()) + .expect("agent-a has a group root"); + + assert!(q.cancel(a_root.id), "an interior/group root cancels alone"); + + // agent-b's work is untouched and still claimable; agent-a's is not. + let claims = q.claim_ready(); + assert!( + !claims.is_empty() && claims.iter().all(|c| c.agent == "agent-b"), + "only agent-b remains runnable, got {:?}", + claims.iter().map(|c| c.agent.as_str()).collect::>() + ); +} + #[test] fn cancel_refuses_running_dag() { let q = JobQueue::new(1); From 0aac20d863276316a01c0da21e8d5e286caa2900 Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 1 Aug 2026 15:34:39 +0200 Subject: [PATCH 2/2] refactor(#2908): resolve a wire id in the graph, not the c0re layer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit mara on !2909: "shouldnt node_by_id be part of jobq?" — yes. Resolving a raw value to a `NodeId` is the exact inverse of `NodeId::get`, which already lives in hive-jobq, and it is only a search because the graph owns the counter that makes ids unfabricable. Both halves of that round-trip belong on the same side of the crate boundary. Placing it in `QueueInner` also put it in a layer slated for removal, so the c0re-side helper would have had to move later anyway — and it was private there, leaving any other caller needing the same resolution to write the same `nodes().find_map(…)` by hand. `Graph::resolve_id` replaces it, with a unit test covering the round-trip and the rejection of a value that was never an id. While re-reading the diff for that question: the doc comment added in the previous commit landed *between* `container`'s doc comment and its signature, silently reattaching "The container node of `dag_id`" to the new helper and leaving `container` undocumented. Restored. No behaviour change and no wire change — same search, same call site. Checked with clippy (`--all-targets -D warnings`), `cargo test -p hive-jobq -p hive-c0re` (41 + 322 passed) and `nix fmt`. --- hive-c0re/src/job_queue/mod.rs | 14 ++------------ hive-jobq/src/lib.rs | 29 +++++++++++++++++++++++++++++ 2 files changed, 31 insertions(+), 12 deletions(-) diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index aa4be634..9300b526 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -345,13 +345,14 @@ impl JobQueue { /// say) is cancelled along with everything else — there is nothing to converge /// when no node ever ran. Only a node that named `Cancelled` survives, and it /// survives because it asked to. + /// /// `id` names **any node**, not specifically a DAG. Cancelling a group root /// drops that whole group (the cascade is the scheduler's), which is what /// the dashboard's whole-DAG cancel does; cancelling an interior node drops /// just that branch. Nothing here knows about DAGs. pub fn cancel(&self, id: u64) -> bool { let mut inner = self.lock(); - let Some(node) = inner.node_by_id(id) else { + let Some(node) = inner.sched.graph().resolve_id(id) else { return false; }; if !inner.sched.cancel_node(node) { @@ -464,17 +465,6 @@ impl QueueInner { /// The container node of `dag_id` — the `NodeKind::Dag` root whose id equals /// `dag_id`. `NodeId` is un-fabricable from a raw `u64`, so this is a search. - /// Resolve a raw wire `u64` to a graph [`NodeId`], whatever kind of node it - /// names. `NodeId` is un-fabricable from a `u64`, so this is a search — - /// same cost as [`QueueInner::container`], without asserting the node is a - /// DAG container. - fn node_by_id(&self, id: u64) -> Option { - self.sched - .graph() - .nodes() - .find_map(|n| (n.id.get() == id).then_some(n.id)) - } - fn container(&self, dag_id: u64) -> Option { self.sched.graph().nodes().find_map(|n| { (n.parent.is_none() diff --git a/hive-jobq/src/lib.rs b/hive-jobq/src/lib.rs index 4646428f..d89a1763 100644 --- a/hive-jobq/src/lib.rs +++ b/hive-jobq/src/lib.rs @@ -461,6 +461,21 @@ impl Graph { self.nodes.iter().find(|n| n.id == id) } + /// Resolve a raw value back to the opaque [`NodeId`] it names — the inverse + /// of [`NodeId::get`], and the only way to perform that direction. A caller + /// holding a value that crossed a wire cannot fabricate an id from it (that + /// impossibility is the point of the type), so it has to be matched against + /// the graph, which is what makes this a search rather than a cast. + /// + /// `None` when no node carries that value, which covers both a value that + /// was never an id and one whose node has since been reaped. + #[must_use] + pub fn resolve_id(&self, raw: u64) -> Option { + self.nodes + .iter() + .find_map(|n| (n.id.0 == raw).then_some(n.id)) + } + /// Every node in the graph, in insertion order. The scheduler iterates /// this to find runnable pending nodes. pub fn nodes(&self) -> impl Iterator> { @@ -729,6 +744,20 @@ mod tests { ); } + #[test] + fn resolve_id_inverts_get_and_rejects_a_value_that_was_never_an_id() { + let mut g: Graph<&str, String> = Graph::new(); + let a = g.insert("a", vec![], None).unwrap(); + let b = g.insert("b", vec![], None).unwrap(); + // Round-trips every id the graph handed out: this is the only way back + // from a raw value, since NodeId can't be constructed from one. + assert_eq!(g.resolve_id(a.get()), Some(a)); + assert_eq!(g.resolve_id(b.get()), Some(b)); + // A value that was never an id resolves to nothing, so a caller can't + // reach a node by guessing a number off the wire. + assert_eq!(g.resolve_id(u64::MAX), None); + } + #[test] fn state_terminality() { assert!(State::Done.is_terminal());