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): convert tests off the container id; drop Source + insert_group

Browse source 
The last of the DAG-container removal. `tests.rs` navigated by the id
`submit` returned, so removing the container removed the tests' way of
finding what they inserted; they name the roots they assert on now, which
is the same handle production uses.

Three findings the port surfaced, each a behaviour change rather than a
test fix:

- Cancelling a rebuild's head no longer drops the job. `Reconcile`'s edge
  accepts a skipped brace, and a cancel-cascade skips rather than cancels,
  so the tail stays claimable. Dropping a job means cancelling every id the
  insert returned.
- A directly-cancelled group root reads terminal while a spared tail still
  runs; the cancel used to land on a node above it, which rolled up
  Finishing instead.
- "One DAG per hive-wide op" is not expressible without a container. The
  three tests asserting it now assert that every named root is top-level,
  which is what makes the per-agent subgraphs concurrent.

Deletes two tests: one asserted only that two containers get distinct ids,
the other re-ran an existing case under a second name.

`Source`, `insert_group` and the stop path's `reason` string went dead with
the container and are removed with it.
This commit is contained in:
atlas 2026-08-04 19:31:33 +02:00 • committed by mara
commit ddc017f01b
6 changed files with 273 additions and 275 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -94,7 +94,8 @@ pub(super) async fn post_kill(
// lease keeps it from racing an in-flight rebuild for the same
// agent, and per-node progress surfaces on the queue snapshot.
if let Err(e) =
crate::job_queue::power::stop_many(&state.coord, &[logical.clone()], true).await
crate::job_queue::power::stop_many(&state.coord, std::slice::from_ref(&logical), true)
.await
{
tracing::error!(agent = %logical, error = ?e, "graceful stop: insert failed");
}
@ -111,7 +112,8 @@ pub(super) async fn post_kill(
// manager calling Kill on its own container is self-suicide
// mid-call, not a legitimate operator action.
if let Err(e) =
crate::job_queue::power::stop_many(&state.coord, &[logical.clone()], false).await
crate::job_queue::power::stop_many(&state.coord, std::slice::from_ref(&logical), false)
.await
{
tracing::error!(agent = %logical, error = ?e, "stop: insert failed");
}
@ -146,15 +148,20 @@ pub(super) async fn post_restart(
return reject;
}
if params.graceful {
if let Err(e) =
crate::job_queue::power::restart_many(&state.coord, &[logical.clone()], true).await
if let Err(e) = crate::job_queue::power::restart_many(
&state.coord,
std::slice::from_ref(&logical),
true,
)
.await
{
tracing::error!(agent = %logical, error = ?e, "graceful restart: insert failed");
}
return (StatusCode::OK, "ok").into_response();
}
if let Err(e) =
crate::job_queue::power::restart_many(&state.coord, &[logical.clone()], false).await
crate::job_queue::power::restart_many(&state.coord, std::slice::from_ref(&logical), false)
.await
{
tracing::error!(agent = %logical, error = ?e, "restart: insert failed");
}
@ -219,7 +226,9 @@ pub(super) async fn post_start(
return (StatusCode::OK, "ok").into_response();
}
}
if let Err(e) = crate::job_queue::power::start_many(&state.coord, &[logical.clone()]).await {
if let Err(e) =
crate::job_queue::power::start_many(&state.coord, std::slice::from_ref(&logical)).await
{
tracing::error!(agent = %logical, error = ?e, "start: insert failed");
}
(StatusCode::OK, "ok").into_response()

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

@ -45,7 +45,7 @@ use hive_jobq_wire::{GraphNode, GraphWire};
use tokio::sync::Notify;
pub use hive_jobq::TerminalState;
pub use model::{NodeKind, PermPayload, Source, State};
pub use model::{NodeKind, PermPayload, State};
use resource::Resource;
/// A job under construction: `hive_jobq`'s builder over this queue's payload
@ -137,35 +137,12 @@ fn outcome_of(result: Result<(), String>) -> Outcome {
}
}
/// Insert a declared `job` into the shared graph, returning the inserted ids.
///
/// A node that declared no parent hangs under `group_parent` — the DAG
/// container for a template, the emitting node for a runtime-appended
/// subgraph. Templates declare the parent axis + sibling ordering directly, so
/// there is no dep-on-root to drop and no lease to hoist: each node declares
/// its own resources, and the crate's borrow model keeps a resource continuous
/// across a subtree (a root owns it, descendants borrow it). Independent group
/// roots carry no cross-links, so a multi-agent DAG's per-agent subgraphs run
/// concurrently, each on its own lease.
///
/// # Errors
/// Propagates a crate graph-insert error (malformed dep/parent / dep-scope).
fn insert_group(
inner: &mut Sched,
declare: impl FnOnce(&JobBuilder),
group_parent: Option<NodeId>,
) -> anyhow::Result<()> {
inner
.insert_job(group_parent, |b| {
declare(b);
// A runtime-appended subgraph is addressed by the node that emitted
// it (`group_parent`), so this path names nothing. Callers that DO
// want a handle use `JobQueue::insert` and name the node there.
Vec::new()
})
.map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?;
Ok(())
}
// `insert_group` lived here: a `group_parent`-taking insert whose only
// remaining caller was the DAG container, everything under it. Runtime growth
// never went through it — an executor declares into the builder `hive_jobq`
// hands it, which parents the new work under the emitting node by
// construction. With no container to be the other kind of parent, the
// distinction it existed to express is gone.
impl JobQueue {
#[must_use]

26
hive-c0re/src/job_queue/model.rs
View file

@ -1,18 +1,18 @@
//! Data model for the generic job-DAG queue: node kinds (the primitive
//! operations), dependency edges, and the runtime `Dag` / `Node` store.
//! The `Source` / `State` / `PermPayload` wire enums live in
//! `hive_host_sock::jobs` (they travel on the host admin socket) and are
//! re-exported here for the queue's internal use. The graph itself is
//! served through `hive_jobq_wire`'s generic projection — there is no
//! second, typed view of it any more.
//! Data model for the generic job-DAG queue: the node kinds — the primitive
//! operations — and what each one carries. The `State` / `PermPayload` wire
//! enums live in `hive_host_sock::jobs` (they travel on the host admin
//! socket) and are re-exported here for the queue's internal use. The graph
//! itself is served through `hive_jobq_wire`'s generic projection — there is
//! no second, typed view of it any more.
//!
//! Two levels: the **DAG** is the unit of cancel / approval-resolution
//! and the dashboard group; the **node** is the unit of scheduling /
//! execution / build-log, and carries its own `agent` (a
//! DAG can span agents). See `docs/coordinator.md::Job queue` for the
//! full design.
//! **One level, not two.** The node is the unit of everything: scheduling,
//! execution, build-log, cancel, and the dashboard group (a group root's
//! subtree *is* the group). A DAG used to be a second level above it, with
//! its own store and its own id; there is no container node any more, so a
//! job is exactly the nodes it declared. See `docs/coordinator.md::Job queue`
//! for the full design.
pub use hive_host_sock::jobs::{PermPayload, Source, State};
pub use hive_host_sock::jobs::{PermPayload, State};
use serde::Serialize;
use hive_jobq::TerminalState;

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

@ -28,28 +28,53 @@ fn insert(q: &JobQueue, declare: impl FnOnce(&JobBuilder)) {
.expect("valid shape");
}
/// Insert a declared job and hand back the ids of the nodes it **named**, in
/// the order it named them.
///
/// This is the handle that replaced the DAG id: there is no container to point
/// at any more, so a test that needs to cancel a job or read its state names
/// the roots it cares about — exactly what production does with the ids
/// [`JobQueue::insert_job`] returns.
/// Handed back as raw `u64`, the same form production passes to `cancel` and
/// `node_subtrees` — a `NodeId` cannot be fabricated, so the read surface takes
/// raw ids and searches for them.
fn insert_named(
q: &JobQueue,
declare: impl FnOnce(&JobBuilder) -> Vec<hive_jobq::NodeGuid>,
) -> Vec<u64> {
q.insert_job(declare)
.expect("valid shape")
.into_iter()
.map(NodeId::get)
.collect()
}
fn ident(s: &str) -> hive_types::Ident {
hive_types::Ident::parse(s).expect("valid test ident")
}
fn rebuild(builder: &JobBuilder, agent: &str) {
templates::rebuild(builder, agent, true);
fn rebuild(builder: &JobBuilder, agent: &str) -> Vec<hive_jobq::NodeGuid> {
templates::rebuild(builder, agent, true)
}
/// Restart shape with every agent treated as **running** — the online
/// shape (`[Signal→Drain→] StopForUpdate → Reconcile`, no `SetWanted` head)
/// most queue-mechanics tests assume. Mirrors the pre-dynamic
/// `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,
) -> Vec<hive_jobq::NodeGuid> {
let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect();
power::restart_nodes(builder, &targets, graceful);
power::restart_nodes(builder, &targets, graceful)
}
/// Stop shape with every agent treated as **running** — the online shape
/// (`SetWanted → [Signal→Drain→](graceful) Reconcile`).
fn stop_online(builder: &JobBuilder, agents: &[&str], graceful: bool) {
fn stop_online(builder: &JobBuilder, agents: &[&str], graceful: bool) -> Vec<hive_jobq::NodeGuid> {
let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect();
power::stop_nodes(builder, &targets, graceful);
power::stop_nodes(builder, &targets, graceful)
}
// `Claimed` / `ClaimReady` / `CompleteNode` lived here: a claim snapshot type
@ -299,63 +324,68 @@ fn dag_count(q: &JobQueue) -> usize {
.count()
}
// ---- submit (dedup removed — every submit is a fresh DAG) ----
// ---- insert (dedup removed — every insert is a fresh job) ----
//
// `submit_assigns_distinct_ids` lived here and is gone. Its whole body was
// "two inserts get different container ids" — an assertion about the id
// allocation of a node type this issue deleted, and in any case `hive_jobq`'s
// property rather than c0re's. What the tests below keep is the part that was
// about c0re: **no dedup**, now read off the group count instead of off an id.
#[test]
fn submit_assigns_distinct_ids() {
let q = JobQueue::new(1);
let first = insert(&q, |builder| rebuild(builder, "agent-a"));
let second = insert(&q, |builder| rebuild(builder, "agent-b"));
assert_ne!(first, second);
assert_eq!(dag_count(&q), 2);
}
/// Submit-time dedup was removed with the agent-per-node refactor (a
/// multi-agent DAG has no single agent to key a dedup on), so an identical
/// resubmit — same template + agent, still queued — now enqueues a distinct
/// DAG instead of collapsing into the pending one. Whether any dedup needs
/// Insert-time dedup was removed with the agent-per-node refactor (a
/// multi-agent job has no single agent to key a dedup on), so an identical
/// re-insert — same template + agent, still queued — now enqueues a distinct
/// group instead of collapsing into the pending one. Whether any dedup needs
/// reintroducing is tracked as a follow-up.
#[test]
fn identical_resubmit_is_a_distinct_dag() {
let q = JobQueue::new(1);
let first = insert(&q, |builder| rebuild(builder, "agent-a"));
let resubmit = insert(&q, |builder| rebuild(builder, "agent-a"));
assert_ne!(first, resubmit, "no dedup: identical resubmit is a new DAG");
assert_eq!(dag_count(&q), 2);
let first = insert_named(&q, |builder| rebuild(builder, "agent-a"));
let after_first = dag_count(&q);
let resubmit = insert_named(&q, |builder| rebuild(builder, "agent-a"));
assert_ne!(
first, resubmit,
"no dedup: an identical re-insert declares its own nodes"
);
// Counted as a doubling rather than against a literal: one rebuild is
// several group roots now, and pinning the number here would make this
// test fail on any shape change while saying nothing about dedup.
assert_eq!(
dag_count(&q),
after_first * 2,
"the re-insert added its own roots instead of collapsing into the pending ones"
);
}
#[test]
fn distinct_submits_never_collapse() {
let q = JobQueue::new(1);
let rebuild_a = insert(&q, |builder| rebuild(builder, "agent-a"));
let rebuild_b = insert(&q, |builder| rebuild(builder, "agent-b"));
let restart_a = insert(&q, |builder| {
restart_online(builder, &["agent-a"], false);
});
let rebuild_a = insert_named(&q, |builder| rebuild(builder, "agent-a"));
let one_rebuild = dag_count(&q);
let rebuild_b = insert_named(&q, |builder| rebuild(builder, "agent-b"));
let two_rebuilds = dag_count(&q);
let restart_a = insert_named(&q, |builder| restart_online(builder, &["agent-a"], false));
assert_ne!(rebuild_a, rebuild_b);
assert_ne!(rebuild_a, restart_a);
assert_eq!(dag_count(&q), 3);
assert_eq!(
two_rebuilds,
one_rebuild * 2,
"two rebuilds, nothing merged"
);
assert_eq!(
dag_count(&q),
two_rebuilds + restart_a.len(),
"a restart of an agent that already has a queued rebuild is still its own group"
);
}
#[test]
fn resubmit_while_running_is_new_dag() {
// The "while running" is not load-bearing and used to be staged by claiming
// a node first. `submit` appends a container and inserts the declared
// group; it never consults the state of any existing node, so whether an
// earlier DAG is running cannot change the outcome. What is actually being
// asserted — no dedup, ever — is `identical_resubmit_is_a_distinct_dag`.
//
// Kept as the *named* case because "a config bump mid-build must not be
// swallowed" is the scenario people worry about, and a reader looking for
// it should find it.
let q = JobQueue::new(1);
let a = insert(&q, |builder| rebuild(builder, "agent-a"));
let again = insert(&q, |builder| {
rebuild(builder, "agent-a");
});
assert_ne!(a, again);
assert_eq!(dag_count(&q), 2);
}
// `resubmit_while_running_is_new_dag` lived here: the same two inserts as
// above, kept under a second name so a reader looking for "a config bump
// mid-build must not be swallowed" would find it. It asserted nothing the
// test above doesn't — `insert_job` never consults the state of an existing
// node, so "while running" could not change the outcome and was never staged.
// The scenario is named in that test's doc instead; a duplicate test is a
// second place for the same fact to rot.
// ---- malformed specs: no longer expressible ----
//
@ -393,7 +423,9 @@ fn rebuild_chain_is_declared_serial() {
// The non-graceful shape asserted here has no quiesce chain, so nothing here
// is concurrent — but the name would mislead about the graceful one.
let q = JobQueue::new(1);
let id = insert(&q, |builder| rebuild(builder, "agent-a"));
insert(&q, |builder| {
rebuild(builder, "agent-a");
});
assert_eq!(
declared_shape(&q),
vec![
@ -460,15 +492,9 @@ fn rebuild_chain_is_declared_serial() {
#[test]
fn graceful_rebuild_chain_drains_before_stopping() {
let q = JobQueue::new(1);
let id = q
.submit(
Source::AutoUpdate,
"sweep".to_owned(),
|builder: &JobBuilder| {
templates::graceful_rebuild_nodes(builder, "agent-a", true, None);
},
)
.expect("valid shape");
insert(&q, |builder| {
templates::graceful_rebuild_nodes(builder, "agent-a", true, None);
});
// Asserted as full rows, not just kinds: the kind list is identical whether
// the quiesce chain runs beside the build or nested under it, so a
// kind-only assertion cannot see the bug this shape exists to fix.
@ -522,7 +548,7 @@ fn non_graceful_rebuild_has_no_signal_or_drain() {
// job keeps its nodes to itself and inserts them, so what it built is
// observable where it matters — in what the scheduler runs.
let q = JobQueue::new(1);
let id = insert(&q, |builder| {
insert(&q, |builder| {
templates::rebuild_nodes(builder, "agent-a", true, None);
});
assert_eq!(
@ -559,7 +585,9 @@ fn rebuild_chain_declares_its_resources_on_the_brace() {
// `a_contended_resource_goes_to_the_oldest_waiter`. What is c0re's is
// *which* nodes contend in the first place — a declaration, asserted here.)
let q = JobQueue::new(1);
let id = insert(&q, |builder| rebuild(builder, "agent-a"));
insert(&q, |builder| {
rebuild(builder, "agent-a");
});
let res = |kind: &str| declared_resources(&q, node_of(&q, kind));
let agent = || Resource::Agent("agent-a".to_owned());
@ -602,13 +630,18 @@ fn rebuild_chain_declares_its_resources_on_the_brace() {
// ---- per-agent lease ----
#[test]
fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
fn multi_agent_restart_declares_concurrent_per_agent_subgraphs() {
let q = JobQueue::new(4);
let id = insert(&q, |builder| {
restart_online(builder, &["agent-a", "agent-b"], false);
let roots = insert_named(&q, |builder| {
restart_online(builder, &["agent-a", "agent-b"], false)
});
// A hive-wide restart is ONE DAG, not one-per-agent.
assert_eq!(dag_count(&q), 1);
// Was "a hive-wide restart is ONE DAG": one container over both agents.
// With the container gone it is one *insert* over N independent groups —
// which is the same claim about the operator's action and a better one
// about the graph, since independence is what lets them run at once.
// Asserted against the named count rather than a literal: the point is
// that every root the job named is top-level, with nothing above it.
assert_eq!(dag_count(&q), roots.len(), "every named root is top-level");
// Each agent's subgraph head (StopForUpdate, since both are running) is a
// group root with no deps, so nothing orders them against each other; and
// each declares only its OWN agent's lease, so nothing makes them contend.
@ -638,13 +671,13 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
}
#[test]
fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() {
fn multi_agent_stop_declares_concurrent_per_agent_subgraphs() {
let q = JobQueue::new(4);
let id = insert(&q, |builder| {
stop_online(builder, &["agent-a", "agent-b"], false);
let roots = insert_named(&q, |builder| {
stop_online(builder, &["agent-a", "agent-b"], false)
});
// A hive-wide stop is ONE DAG, not one-per-agent.
assert_eq!(dag_count(&q), 1);
// See the restart case above for why this is a named-root count now.
assert_eq!(dag_count(&q), roots.len(), "every named root is top-level");
// Same declared story as the restart case above: each agent's subgraph head
// is a group root with no node-deps, holding only its own agent's lease.
// Independent roots on disjoint resources is what "concurrently" means at
@ -669,21 +702,24 @@ fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() {
}
#[test]
fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() {
fn multi_agent_start_folds_per_agent_stale_rebuild() {
let q = JobQueue::new(4);
// fresh: offline + not stale → SetWanted → Reconcile.
// stale: offline + stale → SetWanted → «rebuild subgraph».
let id = insert(&q, |builder| {
let roots = insert_named(&q, |builder| {
power::start_nodes(
builder,
&[
("fresh".to_owned(), false, false),
("stale".to_owned(), false, true),
],
);
)
});
// One DAG spanning both agents.
assert_eq!(dag_count(&q), 1);
// One insert spanning both agents — and an *uneven* number of roots, which
// is the shape this test is about: the fresh agent names one, the stale one
// names four (its rebuild chains behind `SetWanted` rather than nesting
// under it, so the head alone would report the start done mid-rebuild).
assert_eq!(dag_count(&q), roots.len(), "every named root is top-level");
// The fold is a *declared* difference, readable the moment submit returns:
// both agents get a `SetWanted(Up)` group root, but the fresh agent's
// subgraph ends at the Reconcile behind it while the stale agent's carries
@ -733,31 +769,32 @@ fn offline_agents_skip_mechanical_nodes_but_keep_reconcile() {
// read and node exec.
let q = JobQueue::new(4);
// Offline graceful stop → SetWanted(Off) → Reconcile (no Signal/Drain).
let stop = insert(&q, |builder| {
power::stop_nodes(builder, &[("down".to_owned(), false)], true);
let stop = insert_named(&q, |builder| {
power::stop_nodes(builder, &[("down".to_owned(), false)], true)
});
// Offline restart → a lone Reconcile (no SetWanted, no StopForUpdate):
// nothing to bounce, and restart never rewrites intent, so the tail
// Reconcile converges the down agent to its existing `wanted`.
let restart = insert(&q, |builder| {
power::restart_nodes(builder, &[("down2".to_owned(), false)], true);
let restart = insert_named(&q, |builder| {
power::restart_nodes(builder, &[("down2".to_owned(), false)], true)
});
let shape = |id: u64| -> Vec<String> {
// The group's work nodes: its subtree minus the container itself,
// which the generic view carries as an ordinary node.
q.node_subtrees(&[id])
// The whole group, **root included** — the root is a work node now
// (`SetWanted` for the stop, the lone `Reconcile` for the restart), not a
// container to be filtered out. One id per agent, which is what the chain
// named.
let shape = |roots: &[u64]| -> Vec<String> {
q.node_subtrees(roots)
.iter()
.filter(|n| n.id != id)
.map(|n| n.payload.label.clone())
.collect()
};
assert_eq!(
shape(stop),
shape(&stop),
vec!["set_wanted".to_owned(), "reconcile".to_owned()],
"offline graceful stop skips the signal/drain quiesce, keeps Reconcile"
);
assert_eq!(
shape(restart),
shape(&restart),
vec!["reconcile".to_owned()],
"offline restart is a lone Reconcile (no SetWanted head, nothing to stop)"
);
@ -774,20 +811,14 @@ fn boot_sweep_nodes_declare_their_own_resources() {
// meta commit inside another node's staged deploy window. Nothing failed to
// compile; only an exhaustive caller list would have caught it.
let q = JobQueue::new(4);
let id = q
.submit(
Source::AutoUpdate,
"boot".to_owned(),
|builder: &JobBuilder| {
crate::workers::auto_update::boot_nodes(
builder,
true,
vec!["stale-agent".to_owned()],
vec!["drifted-agent".to_owned()],
);
},
)
.expect("valid shape");
insert(&q, |builder| {
crate::workers::auto_update::boot_nodes(
builder,
true,
vec!["stale-agent".to_owned()],
vec!["drifted-agent".to_owned()],
);
});
let mut lock = declared_resources(&q, node_of(&q, "meta_lock"));
lock.sort_by_key(|r| format!("{r:?}"));
@ -893,7 +924,9 @@ fn rebuild_reconcile_waits_for_the_whole_build_subtree() {
// **did** flatten this chain, and the guarantee survives only because the
// roll-up point moved with it.
let q = JobQueue::new(1);
let id = insert(&q, |builder| rebuild(builder, "agent-a"));
insert(&q, |builder| {
rebuild(builder, "agent-a");
});
let shape = declared_shape(&q);
let parent_of = |kind: &str| {
shape
@ -965,7 +998,7 @@ fn rebuild_reconcile_waits_for_the_whole_build_subtree() {
#[test]
fn a_fanned_out_mechanical_node_declares_its_agent_lease() {
let q = JobQueue::new(4);
let id = insert(&q, |builder| {
insert(&q, |builder| {
templates::fanned_out_mechanical(
builder,
NodeKind::Start {
@ -1000,15 +1033,9 @@ fn a_fanned_out_mechanical_node_declares_its_agent_lease() {
fn a_meta_lock_grows_one_rebuild_subgraph_per_agent() {
let q = JobQueue::new(4);
let agents = vec!["alice".to_owned(), "bob".to_owned()];
let id = q
.submit(
Source::AutoUpdate,
"sweep".to_owned(),
|builder: &JobBuilder| {
templates::grown_graceful_rebuilds(builder, &agents, true);
},
)
.expect("valid shape");
insert(&q, |builder| {
templates::grown_graceful_rebuilds(builder, &agents, true);
});
// One chain per agent, each an independent group root — so the two rebuild
// concurrently, each on its own lease.
@ -1038,20 +1065,39 @@ fn a_meta_lock_grows_one_rebuild_subgraph_per_agent() {
#[test]
fn cancel_clears_queued_dag() {
let q = JobQueue::new(1);
let id = insert(&q, |builder| rebuild(builder, "agent-a"));
assert!(q.cancel(id), "fully-queued dag cancels");
// The operator sees `Cancelled` the moment the cancel returns — the spared
// tail is still `Pending`, and a DAG must not read `Queued` back to the
// operator who just cancelled it (the dashboard renders this roll-up from
// the snapshot `post_rebuild_queue_cancel` emits synchronously).
assert_eq!(state_of(&q, id), State::Cancelled, "no stale Queued gap");
// Neither `EmitRebuilt` tail accepts a *dropped* dependency — the ok one is
// `AFTER_OK`, the failure one keys on elimination — so both are cancelled
// with the work and **nothing is left that could still run**: no node is
// spared, so a rebuild that never ran emits nothing.
let roots = insert_named(&q, |builder| rebuild(builder, "agent-a"));
let [head, brace, tail] = roots.as_slice() else {
panic!("a rebuild names three roots, got {roots:?}")
};
assert!(q.cancel(*head), "fully-queued dag cancels");
// The operator sees `Cancelled` the moment the cancel returns — a group must
// not read `Queued` back to the operator who just cancelled it (the
// dashboard renders this roll-up from the snapshot
// `post_rebuild_queue_cancel` emits synchronously).
assert_eq!(state_of(&q, *head), State::Cancelled, "no stale Queued gap");
// Both `EmitRebuilt` tails go with it — the ok one is `AFTER_OK`, the
// failure one keys on elimination — so a rebuild that never ran emits
// nothing.
//
// 🎯 **But `Reconcile` survives, and that is the finding.** Its edge onto
// the brace accepts `done|failed|skipped`, and a cancel-cascade *skips* the
// brace rather than cancelling it — so the edge is satisfied and the tail
// stays claimable. With a container above them all, one cancel took the
// whole group; a job is its roots now, and dropping it means dropping every
// id the insert returned. That is what the ids are for.
assert_eq!(
pending_kinds(&q),
vec!["reconcile"],
"the convergence tail outlives its head's cancel"
);
assert!(
!q.cancel(*brace),
"the brace was eliminated with the head — there is nothing left to cancel"
);
assert!(q.cancel(*tail), "the surviving tail cancels on its own id");
assert!(
pending_kinds(&q).is_empty(),
"a dropped rebuild leaves nothing alive, got {:?}",
"cancelling every named root leaves nothing alive, got {:?}",
pending_kinds(&q)
);
}
@ -1067,27 +1113,17 @@ fn cancel_clears_queued_dag() {
#[test]
fn cancel_drops_one_agents_branch_leaving_the_rest() {
let q = JobQueue::new(2);
let id = insert(&q, |builder| {
restart_online(builder, &["agent-a", "agent-b"], false);
let roots = insert_named(&q, |builder| {
restart_online(builder, &["agent-a", "agent-b"], false)
});
// Per-agent subgraphs hang directly off the container, one per agent.
// ⚠️ `parent` is the *graph* parent here, not a DAG-relative one: what
// the typed view called a parentless group root is a direct child of the
// container node in the generic view.
let snap = q.node_subtrees(&[id]);
let a_root = snap
.iter()
.find(|n| {
n.parent == Some(id)
&& n.payload
.data
.get("agent")
.and_then(serde_json::Value::as_str)
== Some("agent-a")
})
.expect("agent-a has a group root");
// One root per agent, **in the order the chain named them** — that ordering
// is `insert_job`'s contract, and it is what replaced digging the right
// subgraph out of a snapshot by matching on its payload's agent field.
let [a_root, _b_root] = roots.as_slice() else {
panic!("a two-agent restart names one root per agent, got {roots:?}")
};
assert!(q.cancel(a_root.id), "an interior/group root cancels alone");
assert!(q.cancel(*a_root), "an interior/group root cancels alone");
// agent-a's subgraph is gone; agent-b's is untouched and still alive.
assert!(
@ -1116,23 +1152,26 @@ fn cancel_drops_one_agents_branch_leaving_the_rest() {
/// deliberately-stopped as far as reconcile and crash-watch are concerned.
#[test]
fn cancelled_power_op_runs_no_compensating_node() {
/// Submit-cancel-assert for one power op. Taking the already-submitted DAG
/// id is what removes the need to put three differently-typed recipes in
/// one array: each caller submits its own spec, so no closure type has to
/// be erased to a boxed one.
fn assert_cancels_clean(q: &JobQueue, id: u64, writes_intent: bool, case: &str) {
// Read the intent head off the submitted DAG rather than out of the
// spec: a declared job holds its own nodes and inserts them.
let has_intent = declared_shape(q, id).iter().any(|d| d.kind == "set_wanted");
/// Insert-cancel-assert for one power op. Taking the roots the job already
/// named is what removes the need to put three differently-typed recipes in
/// one array: each caller inserts its own, so no closure type has to be
/// erased to a boxed one.
fn assert_cancels_clean(q: &JobQueue, roots: &[u64], writes_intent: bool, case: &str) {
// Read the intent head off the inserted nodes rather than out of a
// spec: a declared job holds its own nodes and inserts them. The queue
// is fresh per case, so the whole graph is this one op.
let has_intent = declared_shape(q).iter().any(|d| d.kind == "set_wanted");
assert_eq!(has_intent, writes_intent, "{case}: intent head");
assert!(q.cancel(id), "{case}: cancelled while queued");
assert_eq!(state_of(q, id), State::Cancelled);
for root in roots {
assert!(q.cancel(*root), "{case}: cancelled while queued");
assert_eq!(state_of(q, *root), State::Cancelled);
}
// Nothing is left that *could* run. Asserting on the pending set rather
// than on "what is ready this instant" also covers a node that is alive
// but blocked — which is exactly what a leftover compensating node
// would look like.
assert_eq!(
pending_kinds(q, id),
pending_kinds(q),
Vec::<&str>::new(),
"{case}: a power op emits no tail node, so a cancelled one leaves nothing"
);
@ -1144,22 +1183,20 @@ fn cancelled_power_op_runs_no_compensating_node() {
let case = format!("graceful={graceful} running={running}");
let q = JobQueue::new(1);
let id = insert(&q, |builder| {
power::restart_nodes(builder, &targets, graceful);
let roots = insert_named(&q, |builder| {
power::restart_nodes(builder, &targets, graceful)
});
assert_cancels_clean(&q, id, false, &format!("restart {case}"));
assert_cancels_clean(&q, &roots, false, &format!("restart {case}"));
let q = JobQueue::new(1);
let id = insert(&q, |builder| {
power::stop_nodes(builder, &targets, graceful);
});
assert_cancels_clean(&q, id, true, &format!("stop {case}"));
let roots = insert_named(&q, |builder| power::stop_nodes(builder, &targets, graceful));
assert_cancels_clean(&q, &roots, true, &format!("stop {case}"));
let q = JobQueue::new(1);
let id = insert(&q, |builder| {
power::start_nodes(builder, &[("agent-a".to_owned(), running, false)]);
let roots = insert_named(&q, |builder| {
power::start_nodes(builder, &[("agent-a".to_owned(), running, false)])
});
assert_cancels_clean(&q, id, true, &format!("start {case}"));
assert_cancels_clean(&q, &roots, true, &format!("start {case}"));
}
}
}
@ -1177,10 +1214,15 @@ fn cancelled_power_op_runs_no_compensating_node() {
#[test]
fn cancelled_dag_still_runs_its_approval_tail() {
let q = JobQueue::new(1);
let id = insert(&q, |builder| {
insert(&q, |builder| {
templates::approval_deploy(builder, "agent-a", 7);
});
assert!(q.cancel(id), "fully-queued dag cancels");
// Found by kind, not returned: `approval_deploy` deliberately names
// nothing, because nothing polls it — the approval row is how an operator
// follows a deploy, so the template is fire-and-forget in production and a
// test must not make it return an id it would otherwise have no use for.
let window = node_of(&q, "deploy_window").get();
assert!(q.cancel(window), "fully-queued dag cancels");
// The `Cancelled` tail is the only node whose edge accepts a dropped
// dependency, so it is the only one `cancel` spares — and *which* tail
// survives is the whole assertion: the template emits one per outcome and
@ -1197,18 +1239,20 @@ fn cancelled_dag_still_runs_its_approval_tail() {
),
"only the cancelled-outcome tail is spared, got {spared:?}"
);
// ⚠️ `Finishing`, not `Cancelled` — and the change is a **fix**, not a
// regression. This used to read the host-side `DagView::rollup_state`,
// which flattened the spared tail away and reported the group settled
// while a node of it was still pending. The root's own state is the
// scheduler's answer: `Finishing` means "own logic done, children still
// running", and the tail this test exists to protect *is* such a child.
// A group that still has work to do does not read terminal.
assert_eq!(state_of(&q, id), State::Finishing);
// ⚠️ `Cancelled`, and it reads terminal **while the spared tail is still
// pending** — the one place this differs from the container era, where the
// cancel landed on a node *above* the window and the window rolled up
// `Finishing`. Here the operator cancels the window itself, so its own
// state is `Cancelled` however its subtree is doing. Deliberately asserted
// rather than routed around: the group's card goes terminal while a
// bookkeeping node runs on. That is acceptable for the tail this test
// protects (it resolves the approval row and nothing waits on it), and it
// would not be for work an operator expects to still be watching.
assert_eq!(state_of(&q, window), State::Cancelled);
// An unrelated group landing in the same graph doesn't disturb this one's
// state — a root rolls up its own subtree, not the graph.
let _other = insert(&q, |builder| rebuild(builder, "agent-b"));
assert_eq!(state_of(&q, id), State::Finishing);
let _other = insert_named(&q, |builder| rebuild(builder, "agent-b"));
assert_eq!(state_of(&q, window), State::Cancelled);
}
// ---- approval deploy subtree ----
@ -1221,7 +1265,7 @@ fn cancelled_dag_still_runs_its_approval_tail() {
#[test]
fn deploy_dag_runs_phases_in_order_and_tails_a_failed_apply() {
let q = JobQueue::new(1);
let id = insert(&q, |builder| {
insert(&q, |builder| {
templates::approval_deploy(builder, "agent-a", 7);
});
@ -1279,7 +1323,7 @@ fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() {
// children run (`a_completing_node_grows_the_work_it_declared`,
// `parent_parks_in_finishing_until_children_roll_up`).
let q = JobQueue::new(1);
let id = insert(&q, |builder| {
insert(&q, |builder| {
templates::deploy_rebuild_nodes(builder, "agent-a", 11);
});
@ -1459,7 +1503,7 @@ fn error_truncation_cuts_on_a_char_boundary() {
#[test]
fn graceful_stop_shape_signal_drain_reconcile() {
let q = JobQueue::new(1);
let id = insert(&q, |builder| {
insert(&q, |builder| {
stop_online(builder, &["agent-a"], true);
});
assert_eq!(
@ -1497,7 +1541,7 @@ fn graceful_stop_shape_signal_drain_reconcile() {
#[test]
fn spawn_shape_provision_create_dropin_reconcile() {
let q = JobQueue::new(1);
let id = insert(&q, |builder| {
insert(&q, |builder| {
templates::spawn(builder, "newbie", 7);
});
assert_eq!(
@ -1522,7 +1566,7 @@ fn spawn_shape_provision_create_dropin_reconcile() {
#[test]
fn perm_change_shape_prefixes_rebuild_chain() {
let q = JobQueue::new(1);
let id = insert(&q, |builder| {
insert(&q, |builder| {
templates::perm_change(
builder,
"agent-a",
@ -1561,7 +1605,7 @@ fn reparent_shape_is_a_lone_agentless_meta_window_node() {
// `MetaLock`, and it must declare the meta window — a topology commit
// must not land inside another node's staged deploy window.
let q = JobQueue::new(1);
let id = insert(&q, |builder| {
insert(&q, |builder| {
templates::reparent(builder, vec![(ident("alice"), Some(ident("bob")))]);
});
assert_eq!(
@ -1585,7 +1629,7 @@ fn reparent_bulk_shape_carries_every_move_on_one_node() {
// request is the reason a single node was chosen in the first place.
let moves = vec![(ident("alice"), Some(ident("bob"))), (ident("carol"), None)];
let q = JobQueue::new(1);
let id = insert(&q, |builder| {
insert(&q, |builder| {
templates::reparent(builder, moves.clone());
});
assert_eq!(

10
hive-c0re/src/server.rs
View file

@ -840,16 +840,10 @@ async fn handle_stop(
let mut errors: Vec<String> = Vec::new();
let mut queued: Vec<u64> = Vec::new();
// One DAG for all targeted agents — a per-agent stop subgraph each
// One insert for all targeted agents — a per-agent stop subgraph each
// (`SetWanted(Offline) → [Signal → Drain →] Reconcile`), independent
// roots that run concurrently on their own leases. A hive-wide
// `hivectl stop` is now a single DAG, not N.
// roots that run concurrently on their own leases.
if !agents.is_empty() {
let reason = if graceful {
"manual via hivectl graceful stop"
} else {
"manual via hivectl stop"
};
match crate::job_queue::power::stop_many(coord, agents, graceful).await {
Ok(ids) => {
queued.extend(ids.into_iter().map(hive_jobq::NodeId::get));

42
hive-host-sock/src/jobs.rs
View file

@ -1,6 +1,11 @@
//! Vocabulary hive-c0re's job queue shares with its clients: where a job
//! came from ([`Source`]), what a permission change carries
//! ([`PermPayload`]), and the scheduler's lifecycle [`State`].
//! Vocabulary hive-c0re's job queue shares with its clients: what a
//! permission change carries ([`PermPayload`]) and the scheduler's
//! lifecycle [`State`].
//!
//! A `Source` enum lived here too — where a job came from, rendered as the
//! "why" chip. It was a field on the DAG container, and it went with it: a
//! job is its nodes now, and a node says what it does rather than who asked
//! for it.
//!
//! **The typed `DagView`/`NodeView` projection that used to live here is
//! gone.** One graph is served one way now — `hive_jobq_wire`'s generic
@ -12,37 +17,6 @@
use serde::{Deserialize, Serialize};
/// Where the submit request originated — drives the "why" chip on the
/// dashboard.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Source {
/// Operator action (dashboard button, CLI, manager tool).
Manual,
/// Meta-update cascade rebuild (grown into the meta-update DAG).
MetaUpdate,
/// Boot-time submission (the boot sweep DAG + boot reconciles).
AutoUpdate,
/// Crash recovery path (future use).
CrashRecover,
/// Operator approved a pending `Approval` row; `approval_id` on
/// the DAG points back at the source row.
Approval,
}
impl Source {
#[must_use]
pub fn as_str(self) -> &'static str {
match self {
Source::Manual => "manual",
Source::MetaUpdate => "meta_update",
Source::AutoUpdate => "auto_update",
Source::CrashRecover => "crash_recover",
Source::Approval => "approval",
}
}
}
pub use hive_jobq::State;
/// Kind-specific payload for `Template::PermChange` DAGs.