Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0aac20d863 | ||
|
|
eb557ee3c1 |
4 changed files with 79 additions and 6 deletions
|
|
@ -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"
|
||||
)]
|
||||
|
|
|
|||
|
|
@ -345,12 +345,17 @@ 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.sched.graph().resolve_id(id) else {
|
||||
return false;
|
||||
};
|
||||
if !inner.sched.cancel_node(container) {
|
||||
if !inner.sched.cancel_node(node) {
|
||||
return false;
|
||||
}
|
||||
drop(inner);
|
||||
|
|
|
|||
|
|
@ -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::<Vec<_>>()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cancel_refuses_running_dag() {
|
||||
let q = JobQueue::new(1);
|
||||
|
|
|
|||
|
|
@ -461,6 +461,21 @@ impl<N, R> Graph<N, R> {
|
|||
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<NodeId> {
|
||||
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<Item = &Node<N, R>> {
|
||||
|
|
@ -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());
|
||||
|
|
|
|||
Loading…
Reference in a new issue