vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

job_queue: drop the NodeKind::Dag/root filter from the QueueNodes lookup

Browse source
This commit is contained in:
damocles 2026-08-03 20:03:48 +02:00 • committed by mara
commit 309c1c91c6
4 changed files with 48 additions and 43 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -289,8 +289,8 @@ impl JobQueue {
#[must_use] #[must_use]
pub fn first_error(&self, dag_id: u64) -> Option<String> { pub fn first_error(&self, dag_id: u64) -> Option<String> {
let inner = self.lock(); let inner = self.lock();
let container = container(&inner, dag_id)?; let node = find_node(&inner, dag_id)?;
inner.graph().first_error(container).map(ToOwned::to_owned) inner.graph().first_error(node).map(ToOwned::to_owned)
} }
/// `(agent, label, takes_container_down)` for the live transient-pill set, /// `(agent, label, takes_container_down)` for the live transient-pill set,
@ -379,38 +379,40 @@ impl JobQueue {
.collect() .collect()
} }
/// One DAG's container node plus its live subtree, as generic wire /// A node plus its live subtree, as generic wire nodes — the
/// nodes — the `QueueNodes` polling surface behind `hivectl`'s /// `QueueNodes` polling surface behind `hivectl`'s wait/progress loop.
/// wait/progress loop. Sibling of [`Self::snapshot`] /// Sibling of [`Self::snapshot`] (which serves the same graph through
/// (which serves the same graph through the typed `DagView`/`NodeView` /// the typed `DagView`/`NodeView` projection for the dashboard's
/// projection for the dashboard's `/api/state.rebuild_queue`), this one /// `/api/state.rebuild_queue`), this one goes through
/// goes through [`GraphWire::wire_snapshot`] instead — no `Done`-node /// [`GraphWire::wire_snapshot`] instead — no `Done`-node filtering, no
/// filtering, no roll-up field (the root's own `state` answers that, /// roll-up field (a node's own `state` answers that, see
/// see `hive_jobq_wire`'s doc comment). /// `hive_jobq_wire`'s doc comment). Looks the id up by identity alone —
/// no assumption that it names a DAG container or a root; "just show
/// whatever the backend sends" for whatever id the caller asks about.
/// ///
/// Empty when `dag_id` names no DAG container in the graph. Today that /// Empty when `id` names no node in the graph. Today that only happens
/// only happens for a genuinely unknown id: nothing prunes the graph /// for a genuinely unknown id: nothing prunes the graph yet
/// yet (bounded-prune is a Stage-C follow-up, see [`visible_dags`]), so /// (bounded-prune is a Stage-C follow-up, see [`visible_dags`]), so a
/// a *completed* DAG's nodes keep riding here with a terminal `state` /// *completed* DAG's nodes keep riding here with a terminal `state`
/// rather than disappearing — callers watching for "done" should read /// rather than disappearing — callers watching for "done" should read
/// the root's `state`, not emptiness. /// the root's `state`, not emptiness.
#[must_use] #[must_use]
pub fn dag_nodes(&self, dag_id: u64) -> Vec<GraphNode> { pub fn node_subtree(&self, id: u64) -> Vec<GraphNode> {
let inner = self.lock(); let inner = self.lock();
let Some(root) = container(&inner, dag_id) else { let Some(node) = find_node(&inner, id) else {
return Vec::new(); return Vec::new();
}; };
inner.graph().wire_snapshot([root]) inner.graph().wire_snapshot([node])
} }
} }
/// The container node of `dag_id` — the `NodeKind::Dag` root whose id equals /// The graph node whose id equals `id`, whatever its kind or depth.
/// `dag_id`. `NodeId` is un-fabricable from a raw `u64`, so this is a search. /// `NodeId` is un-fabricable from a raw `u64`, so this is a search.
fn container(sched: &Sched, dag_id: u64) -> Option<NodeId> { fn find_node(sched: &Sched, id: u64) -> Option<NodeId> {
sched.graph().nodes().find_map(|n| { sched
(n.parent.is_none() && n.id.get() == dag_id && matches!(n.payload, NodeKind::Dag { .. })) .graph()
.then_some(n.id) .nodes()
}) .find_map(|n| (n.id.get() == id).then_some(n.id))
} }
/// Project a DAG into its wire [`DagView`]: a near-raw view of the /// Project a DAG into its wire [`DagView`]: a near-raw view of the

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

@ -160,7 +160,9 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
.collect(); .collect();
HostResponse::dags(dags) HostResponse::dags(dags)
} }
HostRequest::QueueNodes { id } => HostResponse::nodes(coord.job_queue.dag_nodes(*id)), HostRequest::QueueNodes { id } => {
HostResponse::nodes(coord.job_queue.node_subtree(*id))
}
HostRequest::List => HostResponse::list(lifecycle::list().await?), HostRequest::List => HostResponse::list(lifecycle::list().await?),
// The agents root is ours and not world-traversable, so this // The agents root is ours and not world-traversable, so this
// question is only answerable on this side of the socket — // question is only answerable on this side of the socket —

33
hive-host-sock/src/lib.rs
View file

@ -214,13 +214,14 @@ pub enum HostRequest {
/// `hivectl`'s wait/progress loop. A multi-step op is a single DAG /// `hivectl`'s wait/progress loop. A multi-step op is a single DAG
/// (its whole graph in `nodes`). Result: [`HostResponse::dags`]. /// (its whole graph in `nodes`). Result: [`HostResponse::dags`].
QueueDag { id: u64 }, QueueDag { id: u64 },
/// Fetch one job-queue DAG's container node plus its live subtree, as /// Fetch one job-queue node plus its live subtree, as generic
/// generic `hive-jobq-wire` nodes — `hivectl`'s wait/progress loop. /// `hive-jobq-wire` nodes — `hivectl`'s wait/progress loop. Sibling of
/// Sibling of [`Self::QueueDag`]: same DAG, same /// [`Self::QueueDag`]: same graph, same `id`, through the generic
/// `id` (the container's own node id, what `queued_dags` already /// projection instead of the typed `DagView`/`NodeView` (kept for
/// carries), through the generic projection instead of the typed /// `QueueDag`'s other consumer, `/api/state.rebuild_queue`). No
/// `DagView`/`NodeView` (kept for `QueueDag`'s other consumer, /// assumption that `id` names a DAG container or root — whatever node
/// `/api/state.rebuild_queue`). Result: [`HostResponse::nodes`]. /// has that id, the backend hands back its subtree as-is. Result:
/// [`HostResponse::nodes`].
QueueNodes { id: u64 }, QueueNodes { id: u64 },
/// List pending approval requests. /// List pending approval requests.
Pending, Pending,
@ -547,13 +548,13 @@ pub struct HostResponse {
/// been evicted from the queue's history tail. /// been evicted from the queue's history tail.
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub dags: Option<Vec<jobs::DagView>>, pub dags: Option<Vec<jobs::DagView>>,
/// `QueueNodes` result — the requested DAG's container node plus its /// `QueueNodes` result — the requested node plus its live subtree, as
/// live subtree, as generic `hive-jobq-wire` nodes. `None` for every /// generic `hive-jobq-wire` nodes. `None` for every other request kind.
/// other request kind. An empty `Vec` means `id` names no live DAG in /// An empty `Vec` means `id` names no live node in the graph (today: an
/// the graph (today: an unknown id — see `JobQueue::dag_nodes`'s doc /// unknown id — see `JobQueue::node_subtree`'s doc comment for why a
/// comment for why a *completed* DAG's nodes don't vanish the same way /// *completed* DAG's nodes don't vanish the same way `QueueDag`'s do);
/// `QueueDag`'s do); callers should read the root node's `state` for /// callers should read the root node's `state` for terminality, not
/// terminality, not emptiness. /// emptiness.
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub nodes: Option<Vec<hive_jobq_wire::GraphNode>>, pub nodes: Option<Vec<hive_jobq_wire::GraphNode>>,
/// Free-form operator-facing output lines the client prints verbatim /// Free-form operator-facing output lines the client prints verbatim
@ -656,8 +657,8 @@ impl HostResponse {
} }
} }
/// `QueueNodes` result — the polled DAG's container + subtree, as /// `QueueNodes` result — the polled node + its subtree, as generic
/// generic wire nodes. /// wire nodes.
#[must_use] #[must_use]
pub fn nodes(nodes: Vec<hive_jobq_wire::GraphNode>) -> Self { pub fn nodes(nodes: Vec<hive_jobq_wire::GraphNode>) -> Self {
Self { Self {

4
hivectl/src/dag_progress.rs
View file

@ -66,8 +66,8 @@ async fn wait_for_dags_plain(socket: &Path, ids: Vec<u64>) -> Result<()> {
let nodes = resp.nodes.unwrap_or_default(); let nodes = resp.nodes.unwrap_or_default();
let Some(root) = find_root(&nodes) else { let Some(root) = find_root(&nodes) else {
// Unknown id — nothing prunes the graph yet (see // Unknown id — nothing prunes the graph yet (see
// `JobQueue::dag_nodes`'s doc comment), so an id that // `JobQueue::node_subtree`'s doc comment), so an id that
// resolves to no container never named a real DAG. A // resolves to no node never named a real DAG. A
// *completed* DAG's nodes keep riding here instead, with a // *completed* DAG's nodes keep riding here instead, with a
// terminal root `state`, which is what the check below // terminal root `state`, which is what the check below
// watches for. // watches for.