From 861a1f8f26839349bee360e41f5639ab0cbc6dca Mon Sep 17 00:00:00 2001 From: iris Date: Wed, 12 Aug 2026 20:54:53 +0200 Subject: [PATCH] job_queue: filter jobq graph snapshot by per-node state graph_snapshot previously filtered which whole roots got projected based on the root node's own state, so a group root that was still Running but had already-Done internal steps couldn't be filtered down to just its live nodes, and a filtered-out root hid its entire subtree even when a descendant still matched. Apply the states filter after GraphWire::wire_snapshot instead, over every node in the flattened tree, not just roots. The jobq-graph client already handles an orphaned node (parent filtered out) by promoting it to a rendered root, so this is safe on the client side with no changes needed there. Fixes hyperhive#3210 --- hive-c0re/src/job_queue/mod.rs | 66 +++++++++++++++--------------- hive-c0re/src/job_queue/tests.rs | 69 ++++++++++++++++++++++++++++++-- 2 files changed, 98 insertions(+), 37 deletions(-) diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index c83acf3f..e6011105 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -333,25 +333,32 @@ impl JobQueue { /// visible rather than vanishing from the payload, which is what makes a /// fast rebuild render as a single node). /// - /// `states`, when given, keeps only the **root** groups whose own state - /// is named — a root's state is already its subtree's rolled-up answer - /// (see `hive_jobq_wire`'s doc), so filtering the root filters the whole - /// group without needing to also filter descendants. `None` (or the - /// full state set) is the unfiltered call, matching prior behaviour. - /// A slice rather than a set: `State` (`hive_jobq`'s) derives `Eq` but - /// not `Hash`, and the whole vocabulary is 7 variants — a linear - /// membership check per root costs nothing at that size, so there is no - /// reason to widen `hive_jobq::State`'s derive list just for this. + /// `states`, when given, keeps every **individual node** (root or + /// descendant) whose own state is named — not just whole root groups. + /// A still-live group (root not yet terminal) can otherwise hold any + /// number of already-finished steps inside it; filtering only at the + /// root leaves every one of those visible regardless of the ask, which + /// is exactly the clutter a state filter exists to remove. `None` (or + /// the full state set) is the unfiltered call, matching prior + /// behaviour. A slice rather than a set: `State` derives `Eq` but not + /// `Hash`, and the vocabulary is 7 variants — a linear check per node + /// costs nothing at that size. + /// + /// A node whose *parent* got filtered out still rides with its original + /// `parent` id — `` (the one consumer) already treats + /// an unresolvable parent as a new root (`buildTree`'s fallback), so a + /// filtered-out ancestor surfaces a still-matching descendant one level + /// higher rather than hiding or orphaning it. /// /// The projection itself is [`hive_jobq_wire`]'s; all this layer supplies - /// is *which* groups to show — see [`visible_roots`] for why the graph - /// can't decide that for itself. + /// is *which* nodes to show — see [`visible_roots`] for why the graph + /// can't decide the root-visibility half of that for itself. #[must_use] pub fn graph_snapshot(&self, states: Option<&[State]>) -> Vec { let inner = self.lock(); let roots = visible_roots(&inner); - let roots = filter_roots_by_state(&inner, roots, states); - inner.graph().wire_snapshot(roots) + let nodes = inner.graph().wire_snapshot(roots); + filter_nodes_by_state(nodes, states) } /// Per-state counts over the **same** groups [`Queue::graph_snapshot`] @@ -406,30 +413,23 @@ fn find_node(sched: &Sched, id: u64) -> Option { .find_map(|n| (n.id.get() == id).then_some(n.id)) } -/// [`Queue::graph_snapshot`]'s `states` ask, applied to an already-bounded -/// root set: keeps only the roots whose own `state` is named. +/// [`Queue::graph_snapshot`]'s `states` ask, applied to the already- +/// projected node list: keeps every node — root or descendant — whose own +/// `state` is named. /// -/// Applied *after* [`visible_roots`] rather than folded into it — the -/// history cap bounds how much settled work is retained at all, which is a -/// different question from which of the retained (and live) groups the -/// caller wants shown right now. `None` (or an unrecognised/absent query) -/// is the identity filter. -fn filter_roots_by_state( - sched: &Sched, - roots: Vec, - states: Option<&[State]>, -) -> Vec { +/// Applied *after* [`GraphWire::wire_snapshot`] rather than as a root +/// pre-filter — narrowing which roots are visible at all is +/// [`visible_roots`]'s job (a different question: how much settled work is +/// retained, full stop); this is "of what's retained and live, which +/// individual nodes does the caller want shown right now." `None` (or an +/// unrecognised/absent query) is the identity filter. +fn filter_nodes_by_state(nodes: Vec, states: Option<&[State]>) -> Vec { let Some(states) = states else { - return roots; + return nodes; }; - roots + nodes .into_iter() - .filter(|&id| { - sched - .graph() - .node(id) - .is_some_and(|n| states.contains(&n.state)) - }) + .filter(|n| states.contains(&n.state)) .collect() } diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index b1ff098f..4ac0a3f8 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -1471,11 +1471,18 @@ fn history_retains_live_dags_and_the_newest_terminals() { ); } -/// `graph_snapshot`'s `states` ask filters by **root** state — a whole group -/// is kept or dropped together, never split mid-subtree. `None` is the -/// identity filter (every visible root, current default behaviour). +/// `graph_snapshot`'s `states` ask, exercised through the real scheduler: +/// a queued (never-run) DAG cancels as one unit, so every node in it shares +/// one state and this only proves root-level inclusion/exclusion — the +/// per-node case (a live group holding a mix of finished and unfinished +/// steps) isn't expressible without driving the scheduler, which this test +/// module deliberately can't do (see the module doc comment). See +/// `filter_nodes_by_state_keeps_matching_nodes_from_a_mixed_state_tree` +/// below for that half, exercised directly against hand-built `GraphNode`s. +/// `None` is the identity filter (every visible node, current default +/// behaviour). #[test] -fn graph_snapshot_states_filters_whole_groups_by_root_state() { +fn graph_snapshot_states_filters_by_node_state() { let q = JobQueue::new(2); let roots_a = insert_named(&q, |builder| restart_online(builder, &["agent-a"], false)); let roots_b = insert_named(&q, |builder| restart_online(builder, &["agent-b"], false)); @@ -1520,6 +1527,60 @@ fn graph_snapshot_states_filters_whole_groups_by_root_state() { ); } +/// One hand-built node, bypassing the scheduler entirely — `filter_nodes_by_state` +/// is a pure `Vec -> Vec` transform, so this exercises it +/// directly rather than trying (and failing, per this module's own rule) to +/// drive a live group into a genuinely mixed state through the real queue. +fn node(id: u64, parent: Option, state: State) -> GraphNode { + GraphNode { + id, + parent, + state, + deps: Vec::new(), + created_at: Utc::now(), + started_at: None, + finished_at: None, + error: None, + payload: hive_jobq_wire::NodePayload { + label: id.to_string(), + data: serde_json::Value::Null, + }, + } +} + +/// The case `graph_snapshot_states_filters_by_node_state` above can't reach: +/// a still-live group (root `Running`) holding a mix of already-`Done` and +/// still-`Pending` steps. Filtering out `Done` must drop exactly the `Done` +/// node and nothing else — the root and the `Pending` sibling both stay, +/// even though the root itself isn't in the requested state set. +#[test] +fn filter_nodes_by_state_keeps_matching_nodes_from_a_mixed_state_tree() { + let nodes = vec![ + node(1, None, State::Running), // root: whole group still live + node(2, Some(1), State::Done), // finished step, should be hidden + node(3, Some(1), State::Pending), // not-yet-run step, should stay + ]; + + let mut kept: Vec = + filter_nodes_by_state(nodes.clone(), Some(&[State::Running, State::Pending])) + .into_iter() + .map(|n| n.id) + .collect(); + kept.sort_unstable(); + assert_eq!( + kept, + vec![1, 3], + "Done is filtered out even though its still-live parent (root) isn't" + ); + + let mut unfiltered: Vec = filter_nodes_by_state(nodes, None) + .into_iter() + .map(|n| n.id) + .collect(); + unfiltered.sort_unstable(); + assert_eq!(unfiltered, vec![1, 2, 3], "None is the identity filter"); +} + #[test] fn error_truncation_cuts_on_a_char_boundary() { // `truncate_error` is a pure `&str -> String`. This used to submit a DAG,