From ddc017f01b230a171268778553913765e9561e23 Mon Sep 17 00:00:00 2001 From: atlas Date: Tue, 4 Aug 2026 19:31:33 +0200 Subject: [PATCH] wip(#3001): convert tests off the container id; drop Source + insert_group 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. --- hive-c0re/src/dashboard/lifecycle_ops.rs | 21 +- hive-c0re/src/job_queue/mod.rs | 37 +- hive-c0re/src/job_queue/model.rs | 26 +- hive-c0re/src/job_queue/tests.rs | 410 +++++++++++++---------- hive-c0re/src/server.rs | 10 +- hive-host-sock/src/jobs.rs | 42 +-- 6 files changed, 272 insertions(+), 274 deletions(-) diff --git a/hive-c0re/src/dashboard/lifecycle_ops.rs b/hive-c0re/src/dashboard/lifecycle_ops.rs index e1c46463..e12723fa 100644 --- a/hive-c0re/src/dashboard/lifecycle_ops.rs +++ b/hive-c0re/src/dashboard/lifecycle_ops.rs @@ -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() diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 24f58559..04ed00af 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -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, -) -> 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] diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index fff79617..57cf0dca 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -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; diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index bd7cecb6..41e45018 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -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, +) -> Vec { + 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 { + 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 { 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 { 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 { - // 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 { + 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!( diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 9baaefdc..8e534ba1 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -840,16 +840,10 @@ async fn handle_stop( let mut errors: Vec = Vec::new(); let mut queued: Vec = 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)); diff --git a/hive-host-sock/src/jobs.rs b/hive-host-sock/src/jobs.rs index 7502277b..7930807e 100644 --- a/hive-host-sock/src/jobs.rs +++ b/hive-host-sock/src/jobs.rs @@ -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.