diff --git a/frontend/packages/dashboard/src/builds.html b/frontend/packages/dashboard/src/builds.html index 45ad8276..0c38371d 100644 --- a/frontend/packages/dashboard/src/builds.html +++ b/frontend/packages/dashboard/src/builds.html @@ -27,8 +27,10 @@
+ first-spawns. Rendered from GET /api/jobq/graph by + ; `rebuild_queue_changed` over + /api/dashboard/stream is the refresh trigger and carries no + payload of its own. Default tab. -->

pending + running rebuilds, meta-updates, and first-spawns. one runs at a time; meta-update cascades nest under their parent. dedup: re-enqueueing a still-queued op collapses into the existing entry.

diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 0c88b9ea..52dd4611 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -640,10 +640,8 @@ impl Coordinator { /// wrappers below) and the worker so every state transition /// surfaces on the dashboard without extra plumbing. pub fn emit_rebuild_queue_snapshot(self: &Arc) { - let queue = self.job_queue.snapshot(); self.emit_dashboard_event(DashboardEvent::RebuildQueueChanged { seq: self.next_seq(), - queue, }); } diff --git a/hive-c0re/src/dashboard/state_snapshot.rs b/hive-c0re/src/dashboard/state_snapshot.rs index ccf04b78..df3f486b 100644 --- a/hive-c0re/src/dashboard/state_snapshot.rs +++ b/hive-c0re/src/dashboard/state_snapshot.rs @@ -78,13 +78,6 @@ pub(super) struct StateSnapshot { /// disabled "updating…" state; live transitions arrive via the /// `MetaUpdateRunning` event. meta_update_running: bool, - /// Current state of the global job queue — pending + running DAGs - /// (rebuild / meta-update / spawn / power ops) with their per-node - /// breakdowns, plus the most recent few terminal DAGs the queue - /// retains for history. Live transitions arrive via the - /// `RebuildQueueChanged` event. See `job_queue/`. Field name kept - /// from the old flat queue for wire compatibility. - rebuild_queue: Vec, /// Whether the hive-forge container is up. When true the dashboard /// links each container's config + each approval's commit into the /// forge's `agent-configs` repos. @@ -424,7 +417,6 @@ pub(super) async fn api_state( question_history, tombstones, port_conflicts, - rebuild_queue: state.coord.job_queue.snapshot(), forge_present: crate::forge::is_present().await, matrix_gui_enabled: std::env::var_os("HIVE_MATRIX_GUI_ENABLED").is_some_and(|v| { // Accept any truthy string ("1", "true", "yes") since the diff --git a/hive-c0re/src/dashboard_events.rs b/hive-c0re/src/dashboard_events.rs index 97cd50da..bf3245e6 100644 --- a/hive-c0re/src/dashboard_events.rs +++ b/hive-c0re/src/dashboard_events.rs @@ -8,7 +8,6 @@ use serde::Serialize; use crate::container_view::ContainerView; use crate::dashboard::{MetaInputView, TombstoneView}; -use crate::job_queue::DagView; use chrono::{DateTime, Utc}; #[derive(Debug, Clone, Serialize)] @@ -217,15 +216,20 @@ pub enum DashboardEvent { /// when the active-run count crosses 0, so concurrent updates flip /// the flag exactly once. MetaUpdateRunning { seq: u64, running: bool }, - /// Full snapshot of the rebuild queue (`hive-c0re::rebuild_queue`) - /// — every entry, in enqueue order, including the few most-recent - /// terminal entries the queue retains for history. Same - /// snapshot-shape rationale as `TombstonesChanged` / - /// `MetaInputsChanged`: the list is small, snapshot semantics avoid - /// the add/remove races a per-row event would have, and the - /// dashboard renders each DAG's multi-agent shape from its `nodes` - /// (grouped by `NodeView::agent`) — no cross-DAG grouping needed. - RebuildQueueChanged { seq: u64, queue: Vec }, + /// The job queue changed — **a bare trigger, no payload.** + /// + /// It used to carry a full typed snapshot of the queue. Nothing reads + /// that any more: the dashboard renders the queue from + /// `GET /api/jobq/graph`, so the event's whole job is telling a client + /// *when* to re-fetch. Shipping the graph twice — once here, once on + /// the endpoint — is two projections to keep in agreement for a client + /// that only ever used one. + /// + /// Kept as an event rather than deleted in favour of polling because + /// push-on-change is what every other live surface here does + /// (transients, container state, approvals), and a poll loop would be + /// slower for the same information. + RebuildQueueChanged { seq: u64 }, /// Full snapshot of all scheduled prompts. Emitted after every /// operator mutation (new / edit / cancel / fire-now) and after the /// worker fires or rearms a row. Same snapshot-shape rationale as @@ -421,10 +425,7 @@ mod tests { seq: 1, running: false, }, - DashboardEvent::RebuildQueueChanged { - seq: 1, - queue: Vec::new(), - }, + DashboardEvent::RebuildQueueChanged { seq: 1 }, DashboardEvent::SchedulesChanged { seq: 1, schedules: Vec::new(), diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index c996582b..97ae16c2 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -39,15 +39,14 @@ mod tests; use std::sync::{Arc, Mutex}; use chrono::{DateTime, Utc}; -use hive_host_sock::jobs::NodeView; use hive_jobq::resources::ResourceTable; use hive_jobq::scheduler::{Outcome, Scheduler}; -use hive_jobq::{Dep, Graph, NodeId}; +use hive_jobq::{Graph, NodeId}; use hive_jobq_wire::{GraphNode, GraphWire}; use tokio::sync::Notify; pub use hive_jobq::TerminalState; -pub use model::{DagView, NodeKind, PermPayload, Source, State}; +pub use model::{NodeKind, PermPayload, Source, State}; use resource::Resource; /// A job under construction: `hive_jobq`'s builder over this queue's payload @@ -380,17 +379,6 @@ impl JobQueue { hive_jobq_wire::state_rollup(inner.graph(), visible_roots(&inner)) } - /// Snapshot every live + retained DAG for `/api/state` + `RebuildQueueChanged`. - #[must_use] - pub fn snapshot(&self) -> Vec { - let inner = self.lock(); - let mut ids = visible_dags(&inner); - ids.sort_unstable_by_key(|c| c.get()); - ids.into_iter() - .filter_map(|c| dag_view(&inner, c)) - .collect() - } - /// One or more nodes plus their live subtrees, as generic wire nodes — /// the `QueueNodes` polling surface behind `hivectl`'s wait/progress /// loop. Sibling of [`Self::snapshot`] (which serves the same graph @@ -410,8 +398,8 @@ impl JobQueue { /// the result rather than erroring the whole batch — some ids in a /// batch may already be evicted while others are still live. Today /// that only happens for a genuinely unknown id: nothing prunes the - /// graph yet (bounded-prune is a Stage-C follow-up, see - /// [`visible_dags`]), so a *completed* DAG's nodes keep riding here + /// graph yet (bounded-prune is a Stage-C follow-up; [`visible_roots`] + /// bounds the *view*, not the graph), so a completed group's nodes keep riding here /// with a terminal `state` rather than disappearing — callers /// watching for "done" should read the root's `state`, not absence. #[must_use] @@ -431,191 +419,13 @@ fn find_node(sched: &Sched, id: u64) -> Option { .find_map(|n| (n.id.get() == id).then_some(n.id)) } -/// Project a DAG into its wire [`DagView`]: a near-raw view of the -/// container's work nodes, with `Done` nodes excluded. Lifecycle -/// (`state` / `started_at` / `finished_at` / `error`) is read straight -/// off each `hive_jobq::Node`; the client derives the DAG label, roll-up -/// state, and DAG timestamps from the node set. Non-derivable per-node -/// payload (`approval_id`, meta `inputs`) rides the owning node. Returns -/// `None` when every work node is `Done` or `Skipped` — a fully-settled -/// DAG drops out of the snapshot entirely (a `Failed` one lingers until -/// aged out). -fn dag_view(sched: &Sched, container: NodeId) -> Option { - // Read straight off the container's payload: the domain fields have a - // single home there, so an intermediate owned copy of them was a second - // type describing the same data rather than a grouping side-table. - // - // `created_at` is NOT among them — it comes off the container node itself - // below, where the graph stamps it for every node. Keeping a payload copy - // would record one instant in two places. - let node = sched.graph().node(container)?; - let NodeKind::Dag { source, reason } = &node.payload else { - return None; - }; - let all: Vec<_> = sched.graph().descendants(container).collect(); - // DAG-level timestamps are taken over *all* subtree nodes (including the - // `Done` ones excluded from the wire) — the client can't derive them - // from a `Done`-filtered node set, so the host computes them here. - let mut started: Vec> = Vec::new(); - let mut finished: Vec> = Vec::new(); - for node in &all { - if let Some(s) = node.started_at { - started.push(s); - } - if let Some(f) = node.finished_at { - finished.push(f); - } - } - // Decide which nodes ride the wire *before* projecting any of them: a - // `NodeView` costs a `build_logs` lookup, so building one for a node - // that's about to be dropped would be a query per finished step. - let shown = shown_on_wire(&all.iter().map(|n| n.state).collect::>())?; - let mut nodes = Vec::new(); - for node in shown.into_iter().map(|i| all[i]) { - let id = node.id; - let deps: Vec = node - .deps - .iter() - .filter_map(|d| match d { - Dep::Node { id, .. } => Some(id.get()), - Dep::Resource { .. } => None, - }) - .collect(); - // Non-derivable per-node payload rides the node that owns it. Every - // deploy phase carries the approval id, but only the subtree root - // projects it onto the wire — hanging the approval link off all of - // them would render the same card once per phase. - let approval_id = match &node.payload { - NodeKind::DeployWindow { approval_id, .. } => Some(*approval_id), - _ => None, - }; - let inputs = match &node.payload { - NodeKind::MetaLock { inputs, .. } => inputs.clone(), - _ => Vec::new(), - }; - // Looked up from the log row itself (`build_logs.node_id`), not a - // host-side map. One indexed query per node in the snapshot; the - // node set is bounded by `MAX_HISTORY_DAGS` and the store is a - // local sqlite file, so this is cheaper than the lock contention - // a second shared map would reintroduce. - let build_log_id = crate::build_logs::global().and_then(|h| h.id_for_node(id.get())); - // `node.parent` is the structural jobq parent. Top-level nodes - // have `parent == Some(container)` (direct children of the Dag - // container); those become `parent: None` on the wire since the - // container itself is not part of the work-node payload. Sub-nodes - // carry the id of their containing parent work-node. - let parent = node - .parent - .filter(|&p| p != container) - .map(hive_jobq::NodeId::get); - nodes.push(NodeView { - id: id.get(), - agent: node.payload.agent().to_owned(), - kind: node.payload.as_str().to_owned(), - deps, - state: node.state, - started_at: node.started_at, - finished_at: node.finished_at, - error: node.error.clone(), - approval_id, - inputs, - build_log_id, - parent, - }); - } - let is_terminal = sched.graph().is_settled(container) == Some(true); - Some(DagView { - id: container.get(), - source: *source, - reason: reason.clone(), - created_at: node.created_at, - started_at: started.into_iter().min(), - finished_at: is_terminal.then(|| finished.into_iter().max()).flatten(), - nodes, - }) -} - -/// Which of a DAG's work nodes ride the wire, by index into `states` — or -/// `None` when the DAG has nothing left worth showing and drops out of the -/// snapshot entirely. -/// -/// Two separate decisions, and conflating them pins every completed deploy in -/// the queue view forever: -/// - **`Done` drops off the wire.** A finished step isn't interesting. -/// `Skipped` stays: which branch a run *didn't* take is the readable half of -/// an outcome-branched DAG. -/// - **`Skipped` alone doesn't hold a DAG in the snapshot.** So "the node list -/// is non-empty" and "there's still something here worth showing" are -/// different questions, and only the second one may drop the DAG. -/// -/// Takes states rather than projected nodes so the caller can skip the work of -/// projecting what it's about to discard, and so this is testable without a -/// graph — the states it keys on are ones only a run can produce. -fn shown_on_wire(states: &[State]) -> Option> { - let worth_showing = states - .iter() - .any(|s| !matches!(s, State::Done | State::Skipped)); - if !worth_showing { - return None; - } - Some( - states - .iter() - .enumerate() - .filter(|(_, s)| !matches!(s, State::Done)) - .map(|(i, _)| i) - .collect(), - ) -} - -/// When a DAG's work node finishes on `finished_at` — the max over its -/// subtree (read off the graph `Node`, as unix seconds), for the history -/// cap ordering. -fn dag_finished_at(sched: &Sched, container: NodeId) -> i64 { - sched - .graph() - .descendants(container) - .filter_map(|n| n.finished_at) - .map(|t| t.timestamp()) - .max() - .unwrap_or(0) -} - -/// Every DAG container node id in the graph. -fn containers(sched: &Sched) -> Vec { - sched - .graph() - .nodes() - .filter(|n| n.parent.is_none() && matches!(n.payload, NodeKind::Dag { .. })) - .map(|n| n.id) - .collect() -} - -/// The **visible** DAG set for the snapshot: every live (non-terminal) DAG, -/// plus the newest [`MAX_HISTORY_DAGS`] terminal ones. Crate nodes for -/// evicted DAGs linger in the graph (bounded-prune is a Stage-C follow-up); -/// this filter is what bounds what the dashboard sees. -fn visible_dags(sched: &Sched) -> Vec { - let mut live: Vec = Vec::new(); - let mut terminal: Vec<(NodeId, i64, u64)> = Vec::new(); - for c in containers(sched) { - if sched.graph().is_settled(c) == Some(true) { - terminal.push((c, dag_finished_at(sched, c), c.get())); - } else { - live.push(c); - } - } - retain_history(live, terminal, MAX_HISTORY_DAGS) -} - /// The visible **group** set for [`Queue::graph_snapshot`]: every live group /// root, plus the newest [`MAX_HISTORY_DAGS`] settled ones. /// -/// Same policy as [`visible_dags`], selected *structurally* — a root is a node -/// with no parent. The `DagView` path next door keys on `NodeKind::Dag` -/// instead, which is fine for a projection that already only means anything to -/// hive-c0re, but would make the generic endpoint depend on one node kind that -/// is itself slated for removal. +/// Selected *structurally* — a root is a node with no parent. The typed +/// projection this replaced keyed on `NodeKind::Dag` instead, which made the +/// visible set depend on one host node kind; nothing here knows what a node +/// means. /// /// **This bound is load-bearing, not tidiness.** Nothing ever removes a node /// from the graph (bounded pruning is a Stage-C follow-up), so serving @@ -636,10 +446,13 @@ fn visible_roots(sched: &Sched) -> Vec { } /// When a whole group last finished: the newest `finished_at` across the root -/// **and** its descendants. Unlike [`dag_finished_at`] the root itself counts, -/// because a generic group root can be an ordinary node with no children at -/// all — reading only descendants would date every such group to the epoch and -/// evict it first. +/// **and** its descendants. +/// +/// The root itself counts, because a group root can be an ordinary node with +/// no children at all — reading only descendants would date every such group +/// to the epoch and evict it first. (The typed path this replaced read +/// descendants only, and could get away with it: its roots were always DAG +/// containers, which always have children.) fn group_finished_at(sched: &Sched, root: NodeId) -> i64 { sched .graph() @@ -657,8 +470,8 @@ fn group_finished_at(sched: &Sched, root: NodeId) -> i64 { .unwrap_or(0) } -/// [`visible_dags`]'s policy, split from the graph it reads: keep every live -/// DAG, plus the newest `cap` terminal ones. +/// [`visible_roots`]'s policy, split from the graph it reads: keep every live +/// group, plus the newest `cap` terminal ones. /// /// `terminal` rows are `(handle, finished_at, tiebreak)`. The tiebreak orders /// DAGs that settled inside the same wall-clock second — which is *most* of diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 9f45c3b1..849d8b04 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -1,10 +1,10 @@ //! Data model for the generic job-DAG queue: node kinds (the primitive //! operations), dependency edges, and the runtime `Dag` / `Node` store. -//! The serialized *views* — `DagView` / `NodeView` plus the `Source` / -//! `State` / `PermPayload` wire enums — live in `hive_host_sock::jobs` -//! (they travel on the host admin socket and the dashboard channels -//! served off the same snapshot, and nowhere else) and are re-exported -//! here for the queue's internal use. +//! 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. //! //! Two levels: the **DAG** is the unit of cancel / approval-resolution //! and the dashboard group; the **node** is the unit of scheduling / @@ -12,7 +12,7 @@ //! DAG can span agents). See `docs/coordinator.md::Job queue` for the //! full design. -pub use hive_host_sock::jobs::{DagView, PermPayload, Source, State}; +pub use hive_host_sock::jobs::{PermPayload, Source, 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 e297726e..10759af5 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -284,14 +284,27 @@ fn declared_shape_for(q: &JobQueue, dag: u64, agent: &str) -> Vec { } fn state_of(q: &JobQueue, dag_id: u64) -> State { - // A DAG whose nodes have all settled `Done` or `Skipped` drops out of the - // snapshot — absence is the completion signal, so map it to `Done`. - // Otherwise derive the roll-up from the node set, exactly as every wire - // consumer does. - q.snapshot() + // A group root's own state *is* its subtree's roll-up — that is the + // scheduler's contract (`Finishing` until children settle, then the + // rolled-up outcome), so there is nothing to derive here any more. + // A root aged out of the retained history reads `Done`: it settled, or + // it would still be live. + q.graph_snapshot() .iter() - .find(|d| d.id == dag_id) - .map_or(State::Done, DagView::rollup_state) + .find(|n| n.id == dag_id) + .map_or(State::Done, |n| n.state) +} + +/// How many groups the queue is showing — roots, not nodes. +/// +/// `graph_snapshot` is flat (every node under every visible root), so a test +/// asking "how many DAGs" counts the parentless ones. Counting rows would +/// count steps, which is a different number: one rebuild is ~7 nodes. +fn dag_count(q: &JobQueue) -> usize { + q.graph_snapshot() + .iter() + .filter(|n| n.parent.is_none()) + .count() } // ---- submit (dedup removed — every submit is a fresh DAG) ---- @@ -302,7 +315,7 @@ fn submit_assigns_distinct_ids() { let first = submit(&q, "first", |builder| rebuild(builder, "agent-a")); let second = submit(&q, "second", |builder| rebuild(builder, "agent-b")); assert_ne!(first, second); - assert_eq!(q.snapshot().len(), 2); + assert_eq!(dag_count(&q), 2); } /// Submit-time dedup was removed with the agent-per-node refactor (a @@ -316,7 +329,7 @@ fn identical_resubmit_is_a_distinct_dag() { let first = submit(&q, "first", |builder| rebuild(builder, "agent-a")); let resubmit = submit(&q, "again", |builder| rebuild(builder, "agent-a")); assert_ne!(first, resubmit, "no dedup: identical resubmit is a new DAG"); - assert_eq!(q.snapshot().len(), 2); + assert_eq!(dag_count(&q), 2); } #[test] @@ -329,7 +342,7 @@ fn distinct_submits_never_collapse() { }); assert_ne!(rebuild_a, rebuild_b); assert_ne!(rebuild_a, restart_a); - assert_eq!(q.snapshot().len(), 3); + assert_eq!(dag_count(&q), 3); } #[test] @@ -349,7 +362,7 @@ fn resubmit_while_running_is_new_dag() { rebuild(builder, "agent-a"); }); assert_ne!(a, again); - assert_eq!(q.snapshot().len(), 2); + assert_eq!(dag_count(&q), 2); } // ---- malformed specs: no longer expressible ---- @@ -488,50 +501,6 @@ fn non_graceful_rebuild_has_no_signal_or_drain() { ); } -/// Which nodes ride the wire, and whether a DAG is still worth showing. -/// -/// This used to submit a rebuild, claim and complete all seven of its nodes, -/// then assert the DAG was absent from `snapshot()` — the scheduler, the -/// roll-up and the whole projection standing in for one predicate over a list -/// of states. `shown_on_wire` is that predicate, so the cases can be named -/// instead of arranged. -#[test] -fn skipped_nodes_ride_the_wire_but_do_not_hold_a_settled_dag_in_the_snapshot() { - // Nothing left worth showing → the DAG drops out of the snapshot, and its - // absence is what signals completion. - assert_eq!(shown_on_wire(&[State::Done, State::Done]), None); - // The case the old test was built around: a clean run whose not-taken - // failure branch is still in the graph as `Skipped`. "The node list is - // non-empty" and "there's something here worth showing" are different - // questions, and conflating them pins every completed deploy in the queue - // view forever. - assert_eq!( - shown_on_wire(&[State::Done, State::Skipped, State::Done]), - None, - "a skipped branch does not keep a finished DAG alive" - ); - // A `Failed` DAG lingers — and takes its skipped branches with it, which is - // how an operator sees which steps the run never reached. - assert_eq!( - shown_on_wire(&[State::Done, State::Failed, State::Skipped, State::Done]), - Some(vec![1, 2]), - "done nodes drop off, the failure and what it ruled out stay" - ); - // Live DAGs keep everything but their finished steps. - assert_eq!( - shown_on_wire(&[State::Done, State::Running, State::Pending]), - Some(vec![1, 2]) - ); - assert_eq!( - shown_on_wire(&[State::Cancelled, State::Skipped]), - Some(vec![0, 1]), - "a cancelled DAG is still worth showing to whoever cancelled it" - ); - // An empty DAG has nothing worth showing either — no special case needed, - // but it is the one input where "any" and "all" disagree, so it is pinned. - assert_eq!(shown_on_wire(&[]), None); -} - // ---- build slots ---- #[test] @@ -590,7 +559,7 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() { restart_online(builder, &["agent-a", "agent-b"], false); }); // A hive-wide restart is ONE DAG, not one-per-agent. - assert_eq!(q.snapshot().len(), 1); + assert_eq!(dag_count(&q), 1); // 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. @@ -626,7 +595,7 @@ fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() { stop_online(builder, &["agent-a", "agent-b"], false); }); // A hive-wide stop is ONE DAG, not one-per-agent. - assert_eq!(q.snapshot().len(), 1); + assert_eq!(dag_count(&q), 1); // 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 @@ -665,7 +634,7 @@ fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() { ); }); // One DAG spanning both agents. - assert_eq!(q.snapshot().len(), 1); + assert_eq!(dag_count(&q), 1); // 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 @@ -712,13 +681,12 @@ fn offline_agents_skip_mechanical_nodes_but_keep_reconcile() { submit::restart_nodes(builder, &[("down2".to_owned(), false)], true); }); let shape = |id: u64| -> Vec { - q.snapshot() + // The group's work nodes: its subtree minus the container itself, + // which the generic view carries as an ordinary node. + q.node_subtrees(&[id]) .iter() - .find(|d| d.id == id) - .expect("dag") - .nodes - .iter() - .map(|n| n.kind.clone()) + .filter(|n| n.id != id) + .map(|n| n.payload.label.clone()) .collect() }; assert_eq!( @@ -827,11 +795,13 @@ fn rebuild_chain_nodes_suppress_crash_watch() { // c0re's declaration of *which* edge is which is asserted in // `rebuild_chain_is_declared_serial` — the reconcile's accepted outcome // set is right there in the shape table. -// 2. the wire projection — `Done` off, `Skipped` on. That is -// `shown_on_wire`, tested directly above. -// 3. the roll-up — a DAG with a failed node reads `Failed`, and `Skipped` -// contributes nothing. That is `DagView::rollup_state`, which lives in -// hive-host-sock and is now tested there, next to the invariant. +// 2. the wire projection — which nodes ride and which are filtered out. +// **That decision no longer exists:** the generic view serves every node +// under a visible root, terminal ones included, so there is no predicate +// left to test. The `Done`-off/`Skipped`-on filter died with `DagView`. +// 3. the roll-up — a group with a failed node reads `Failed`. That is the +// root node's own `state`, stamped by `hive_jobq`'s settle loop and +// tested there; nothing re-derives it host-side any more. // // Reconstructing all three from one arranged run made none of them // individually legible, and the arrangement was the only reason this module @@ -1035,13 +1005,21 @@ fn cancel_drops_one_agents_branch_leaving_the_rest() { let id = submit(&q, "r", |builder| { restart_online(builder, &["agent-a", "agent-b"], false); }); - // 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 + // 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.agent == "agent-a" && n.parent.is_none()) + .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"); assert!(q.cancel(a_root.id), "an interior/group root cancels alone"); @@ -1154,11 +1132,18 @@ fn cancelled_dag_still_runs_its_approval_tail() { ), "only the cancelled-outcome tail is spared, got {spared:?}" ); - assert_eq!(state_of(&q, id), State::Cancelled); - // An unrelated DAG landing in the same graph doesn't disturb this one's - // roll-up — the snapshot is per-DAG, not a global state machine. + // ⚠️ `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); + // 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 = submit(&q, "r", |builder| rebuild(builder, "agent-b")); - assert_eq!(state_of(&q, id), State::Cancelled); + assert_eq!(state_of(&q, id), State::Finishing); } // ---- approval deploy subtree ---- diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 5427a7f6..26799a9a 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -150,16 +150,6 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { HostRequest::Rebuild { name } => { submit_single(&coord, name.as_str(), Verb::Rebuild).await } - HostRequest::QueueDag { id } => { - // A multi-step op is one DAG now (no fan-out children to gather). - let dags = coord - .job_queue - .snapshot() - .into_iter() - .filter(|d| d.id == *id) - .collect(); - HostResponse::dags(dags) - } HostRequest::QueueNodes { ids } => { HostResponse::nodes(coord.job_queue.node_subtrees(ids)) } @@ -901,15 +891,15 @@ async fn handle_stop( async fn await_dags(coord: &Arc, ids: &[u64], timeout: std::time::Duration) { let deadline = std::time::Instant::now() + timeout; loop { - let snap = coord.job_queue.snapshot(); - // A DAG has settled when it's either gone from the snapshot (fully - // `Done` DAGs drop out) or still present but with every node terminal - // (a `Failed`/`Cancelled` DAG lingers). It's pending only while it has - // a non-terminal node. - let pending = ids.iter().any(|id| { - snap.iter() - .any(|d| d.id == *id && d.nodes.iter().any(|n| !n.state.is_terminal())) - }); + // Every node under the named roots, terminal ones included — unlike the + // old typed snapshot, a settled group does not drop out of this view. + // So "pending" is simply "some node hasn't finished", with no second + // rule for the disappeared case. + let pending = coord + .job_queue + .node_subtrees(ids) + .iter() + .any(|n| !n.state.is_terminal()); if !pending { return; } diff --git a/hive-host-sock/src/jobs.rs b/hive-host-sock/src/jobs.rs index 2f567bfd..7502277b 100644 --- a/hive-host-sock/src/jobs.rs +++ b/hive-host-sock/src/jobs.rs @@ -1,12 +1,15 @@ -//! Wire shapes of hive-c0re's job-DAG queue: what a queued job looks -//! like on the dashboard SSE channel (`rebuild_queue_changed`), the -//! `/api/state.rebuild_queue` snapshot, and the host admin socket's -//! `QueueDag` polling surface (`hivectl`'s wait/progress loop). The -//! queue *internals* — node kinds, dependency edges, scheduling state — -//! live in `hive-c0re::job_queue`; these are the serialized views it -//! produces. Semantics: `docs/coordinator.md::Job queue`. +//! 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`]. +//! +//! **The typed `DagView`/`NodeView` projection that used to live here is +//! gone.** One graph is served one way now — `hive_jobq_wire`'s generic +//! `GraphNode`, over the `QueueNodes` socket request and +//! `GET /api/jobq/graph` — so there is no second shape to keep in +//! agreement with the first. Queue internals (node kinds, edges, +//! scheduling) live in `hive-c0re::job_queue`. +//! Semantics: `docs/coordinator.md::Job queue`. -use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; /// Where the submit request originated — drives the "why" chip on the @@ -64,248 +67,3 @@ pub enum PermPayload { /// just within one. Consumers treat it opaquely (grouping + dep matching), /// so the widening from the old dag-local `u32` is transparent. pub type NodeId = u64; - -/// One node of a queued DAG, serialized near-raw from the scheduler -/// graph. Lifecycle (`state` / `started_at` / `finished_at` / `error`) -/// comes straight off the `hive_jobq::Node`. The client derives DAG-level -/// roll-ups (label, state, timestamps) from the node set — nothing is -/// rolled up host-side. Build logs are fetched on demand by node id -/// (`GET /api/build-log/`), not carried inline. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct NodeView { - pub id: NodeId, - /// The agent whose container (or meta repo, for `hyperhive` meta-level - /// nodes) this node operates on. Agent is per-node — a single DAG can - /// span multiple agents (e.g. a hive-wide restart), so there is no - /// DAG-level agent field; consumers group by this. - pub agent: String, - /// Node primitive tag: `"prebuild"`, `"stop_for_update"`, - /// `"swap"`, `"create"`, `"meta_lock"`, `"reconcile"`, `"signal"`, - /// `"drain"`, `"write_dropin"`, `"write_perm_file"`, - /// `"approval_deploy"`. - pub kind: String, - /// Ids of the nodes this one waits for. May reference an already-`Done` - /// node that's been filtered out of the wire — the client treats a dep - /// on an absent node as satisfied. - #[serde(default)] - pub deps: Vec, - pub state: State, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub started_at: Option>, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub finished_at: Option>, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub error: Option, - /// Approval-queue row id — present only on the `approval_deploy` node. - /// The client links a DAG to its pending approval through this (it is - /// not derivable from the graph, so it rides the node that owns it). - #[serde(default, skip_serializing_if = "Option::is_none")] - pub approval_id: Option, - /// Meta-flake inputs being bumped — present only on the `meta_lock` - /// node. Display-only payload, not derivable from the graph. - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub inputs: Vec, - /// The build-log history row id, when this node has a captured build - /// log fetchable at `GET /api/build-log/` and deep-linkable to - /// `/builds.html?id=#buildlogs`. Only the nix-heavy nodes that - /// stream build output set one; the client gates its log link on - /// `.is_some()` so lock / noop / store-only nodes don't render a link - /// that 404s. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub build_log_id: Option, - /// Structural parent in the jobq tree — `None` for top-level nodes - /// (direct children of the DAG container). Sub-nodes carry the id of - /// their containing parent node. The client uses this to render the - /// recursive tree rather than inferring structure from `deps` alone. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub parent: Option, -} - -/// A queued / running / failed DAG — a thin projection of one container -/// node plus its (non-`Done`) subtree from the scheduler graph. Only -/// non-derivable facts live here: `id`, `source`, `reason`, `created_at`, -/// and the node set. The client derives the card label, roll-up state, and -/// DAG timestamps from `nodes` (per-node `kind` + lifecycle) — nothing is -/// rolled up host-side. There is no DAG-level `agent`: a DAG can span -/// agents, so agent is per-[`NodeView`]; consumers group by `NodeView::agent`. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct DagView { - pub id: u64, - pub source: Source, - pub reason: String, - /// When the DAG was enqueued. - pub created_at: DateTime, - /// When the DAG's first node started (min over *all* its nodes) — computed - /// host-side, **not** derived on the client: `Done` nodes are excluded from - /// `nodes` below, so the earliest-started node is usually absent from the - /// wire and the client can't take the min itself. `None` until a node runs. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub started_at: Option>, - /// When the DAG finished (max `finished_at` over all its nodes), set only - /// once the DAG has settled terminal. Host-computed for the same reason as - /// `started_at`. `None` while the DAG is still live. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub finished_at: Option>, - /// Nodes of this DAG with `Done` ones excluded. `Skipped` nodes are - /// carried so the dashboard can show which branches weren't taken, but - /// they don't keep a DAG alive: one whose nodes are all `Done` or - /// `Skipped` is omitted from the snapshot entirely, and its absence is - /// what signals completion. A `Failed` DAG lingers until aged out by the - /// history cap. - pub nodes: Vec, -} - -impl DagView { - /// Roll-up state derived from the node set — the shared derivation every - /// Rust consumer (hivectl, the wait loops, tests) uses so the dashboard's - /// JS render and the host agree: `Failed` if any node failed, else - /// **`Cancelled` if any cancelled**, else `Running` if any running, else - /// `Pending` if any pending, else `Done`. `Done` nodes are excluded - /// from the wire, so a DAG that is *entirely* done isn't sent at all — - /// its absence from the snapshot is what signals completion. - /// - /// `Cancelled` outranks both `Running` and `Pending` because a cancelled DAG - /// still has its weak-edged tail node to run (it reports the cancellation), - /// so `Pending`-then-`Running` would flicker back at the operator who just - /// cancelled it and read as "the cancel didn't take". Outside that window - /// the states barely co-occur: a cancel *cascade* originates at a `Failed` - /// node, which returns early above. - /// - /// `Finishing` counts as running: the node's own work is done but its - /// sub-nodes are still going, so the DAG is still in flight. - /// - /// `Skipped` contributes nothing: a not-taken branch is an expected part of - /// a healthy run, so counting it would make every successful DAG roll up - /// non-`Done`. - /// - /// This ordering matches `frontend/packages/dashboard/src/builds.js`'s - /// `rollupState`. The two implementations must be edited together — they - /// have silently disagreed before. - #[must_use] - pub fn rollup_state(&self) -> State { - let mut any_running = false; - let mut any_pending = false; - let mut any_cancelled = false; - for n in &self.nodes { - match n.state { - State::Failed => return State::Failed, - State::Running | State::Finishing => any_running = true, - State::Pending => any_pending = true, - State::Cancelled => any_cancelled = true, - State::Done | State::Skipped => {} - } - } - if any_cancelled { - State::Cancelled - } else if any_running { - State::Running - } else if any_pending { - State::Pending - } else { - State::Done - } - } -} - -#[cfg(test)] -mod tests { - use chrono::Utc; - - use super::{DagView, NodeView, Source, State}; - - /// A node set carrying nothing but the states — the only input - /// `rollup_state` reads. - fn dag(states: &[State]) -> DagView { - DagView { - id: 1, - source: Source::Manual, - reason: "test".to_owned(), - created_at: Utc::now(), - started_at: None, - finished_at: None, - nodes: states - .iter() - .enumerate() - .map(|(i, &state)| NodeView { - id: i as u64, - agent: "a".to_owned(), - kind: "reconcile".to_owned(), - deps: Vec::new(), - state, - started_at: None, - finished_at: None, - error: None, - approval_id: None, - inputs: Vec::new(), - build_log_id: None, - parent: None, - }) - .collect(), - } - } - - #[test] - fn a_failure_outranks_everything_and_skipped_counts_for_nothing() { - // The case this replaces used to be arranged in hive-c0re by running a - // rebuild until its Prebuild failed. Only the states ever mattered. - assert_eq!( - dag(&[State::Done, State::Failed, State::Skipped]).rollup_state(), - State::Failed - ); - // A failure wins even against a node still going — the DAG's verdict - // is already decided. - assert_eq!( - dag(&[State::Running, State::Failed]).rollup_state(), - State::Failed - ); - // Skipped is an expected part of a healthy run: an outcome-branched - // DAG always leaves one branch untaken, so counting it would make - // every successful DAG roll up non-Done. - assert_eq!( - dag(&[State::Done, State::Skipped]).rollup_state(), - State::Done - ); - } - - #[test] - fn cancelled_outranks_running_and_pending() { - // A cancelled DAG still has its weak-edged tail node to run, so - // Pending-then-Running would flicker back at the operator who just - // cancelled it and read as "the cancel didn't take". - assert_eq!( - dag(&[State::Cancelled, State::Pending]).rollup_state(), - State::Cancelled - ); - assert_eq!( - dag(&[State::Cancelled, State::Running]).rollup_state(), - State::Cancelled - ); - } - - #[test] - fn finishing_still_counts_as_running() { - // The node's own work is done but its sub-nodes are still going, so - // the DAG is in flight. A parent parked in Finishing is the normal - // shape of a subtree mid-run, not an edge case. - assert_eq!( - dag(&[State::Finishing, State::Pending]).rollup_state(), - State::Running - ); - assert_eq!( - dag(&[State::Running, State::Pending]).rollup_state(), - State::Running - ); - assert_eq!( - dag(&[State::Done, State::Pending]).rollup_state(), - State::Pending - ); - } - - #[test] - fn an_empty_node_set_reads_done() { - // Every node Done means every node is filtered off the wire, so this - // is what a finished DAG actually looks like to a consumer that has - // one in hand at all. - assert_eq!(dag(&[]).rollup_state(), State::Done); - } -} diff --git a/hive-host-sock/src/lib.rs b/hive-host-sock/src/lib.rs index caf7e3d9..5073c72f 100644 --- a/hive-host-sock/src/lib.rs +++ b/hive-host-sock/src/lib.rs @@ -210,15 +210,9 @@ pub enum HostRequest { /// matrix GUI disabled). Backs `hivectl open` + the federation /// peer-config block (which reads the bare `domain`). Urls, - /// Fetch one job-queue DAG by id — the polling surface behind - /// `hivectl`'s wait/progress loop. A multi-step op is a single DAG - /// (its whole graph in `nodes`). Result: [`HostResponse::dags`]. - QueueDag { id: u64 }, /// Fetch one or more job-queue nodes plus their live subtrees, as - /// generic `hive-jobq-wire` nodes — `hivectl`'s wait/progress loop. - /// Sibling of [`Self::QueueDag`]: same graph, through the generic - /// projection instead of the typed `DagView`/`NodeView` (kept for - /// `QueueDag`'s other consumer, `/api/state.rebuild_queue`). No + /// generic `hive-jobq-wire` nodes — `hivectl`'s wait/progress loop, + /// and the only way the queue is served. No /// assumption that an id names a DAG container or root — whatever /// node has that id, the backend hands back its subtree as-is. A /// batch op that submits several independent roots (e.g. one per @@ -542,23 +536,16 @@ pub struct HostResponse { /// different question". #[serde(default, skip_serializing_if = "Option::is_none")] pub agent_exists: Option, - /// Ids of the job-queue DAGs this request submitted (rebuild / + /// Ids of the job-queue roots this request submitted (rebuild / /// restart / power ops). Clients poll them via - /// [`HostRequest::QueueDag`]; `None` for non-submitting requests. + /// [`HostRequest::QueueNodes`]; `None` for non-submitting requests. #[serde(default, skip_serializing_if = "Option::is_none")] pub queued_dags: Option>, - /// `QueueDag` result — the requested DAG followed by its live - /// fan-out children ([`jobs::DagView`]). Empty when the DAG has - /// been evicted from the queue's history tail. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub dags: Option>, /// `QueueNodes` result — the requested nodes plus their live subtrees, /// as generic `hive-jobq-wire` nodes, all roots' subtrees combined in /// one flat list. `None` for every other request kind. An id with no /// live node in the graph is silently dropped rather than erroring - /// the whole batch — see `JobQueue::node_subtrees`'s doc comment for - /// why a *completed* DAG's nodes don't vanish the same way - /// `QueueDag`'s do; callers should read each root node's `state` for + /// the whole batch — callers should read each root node's `state` for /// terminality, not absence. #[serde(default, skip_serializing_if = "Option::is_none")] pub nodes: Option>, @@ -652,16 +639,6 @@ impl HostResponse { } } - /// `QueueDag` result — the polled DAG + its live children. - #[must_use] - pub fn dags(dags: Vec) -> Self { - Self { - ok: true, - dags: Some(dags), - ..Self::default() - } - } - /// `QueueNodes` result — the polled node + its subtree, as generic /// wire nodes. #[must_use]