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

refactor(#2802): drop the wrappers that now only forward to the graph

Browse source 
`dag_of` and `dag_first_error` had shrunk to a single delegating call once
the walks moved into `hive-jobq`; their callers say what they mean without
the hop.

`subtree` was worse than redundant. It collected the descendant ids into a
`Vec` and both callers then looked each node up again by id — `dag_view`
needed a `let … else { continue }` for a lookup that could not fail.
Iterating `descendants()` hands back the node directly, so the round-trip
and the re-lookup both go.
This commit is contained in:
atlas 2026-07-27 21:10:24 +02:00
commit 6f551334de
1 changed files with 17 additions and 36 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -104,8 +104,8 @@ struct DagMeta {
/// container node** ([`NodeKind::Dag`], `parent = None`) whose subtree is the
/// DAG's work — so the container's `NodeId` is the DAG id, its rolled-up state
/// is the DAG state, and there are no grouping side-tables: membership + meta
/// are graph queries ([`QueueInner::container`] / [`QueueInner::subtree`] /
/// [`QueueInner::dag_meta`]). One shared crate [`Graph`] holds every DAG.
/// are graph queries ([`QueueInner::container`] / [`QueueInner::dag_meta`] +
/// the `hive_jobq::Graph` accessors). One shared crate [`Graph`] holds every DAG.
struct QueueInner {
sched: Scheduler<NodeKind, Resource>,
/// Per-node runtime metadata (the build-log id) — mutable after
@ -309,7 +309,7 @@ impl JobQueue {
};
let kind = node.payload.clone();
let agent = node.payload.agent().to_owned();
let Some(container) = inner.dag_of(id) else {
let Some(container) = inner.sched.graph().root_of(id) else {
continue;
};
let Some(meta) = inner.dag_meta(container) else {
@ -382,7 +382,9 @@ impl JobQueue {
/// Link a `build_logs` row to a specific `Running` node.
pub fn set_build_log_id(&self, dag_id: u64, node_id: NodeId, log_id: i64) -> bool {
let mut inner = self.lock();
if inner.dag_of(node_id).map(NodeId::get) != Some(dag_id) || !inner.node_running(node_id) {
if inner.sched.graph().root_of(node_id).map(NodeId::get) != Some(dag_id)
|| !inner.node_running(node_id)
{
return false;
}
inner.node_rt.entry(node_id).or_default().build_log_id = Some(log_id);
@ -417,7 +419,11 @@ impl JobQueue {
pub fn first_error(&self, dag_id: u64) -> Option<String> {
let inner = self.lock();
let container = inner.container(dag_id)?;
inner.dag_first_error(container)
inner
.sched
.graph()
.first_error(container)
.map(ToOwned::to_owned)
}
/// The `(dag_id, agent, kind)` triples for every per-agent lease currently
@ -435,7 +441,7 @@ impl JobQueue {
let Resource::Agent(agent) = res else {
return None;
};
let container = inner.dag_of(holder)?;
let container = inner.sched.graph().root_of(holder)?;
let kind = inner.dag_meta(container)?.transient?;
Some((container.get(), agent, kind))
})
@ -484,21 +490,6 @@ impl QueueInner {
})
}
/// The DAG container a node belongs to. A container is exactly a group root,
/// so this is the graph's own parent-chain walk.
fn dag_of(&self, id: NodeId) -> Option<NodeId> {
self.sched.graph().root_of(id)
}
/// The DAG's work nodes — its `container`'s subtree, excluding the container.
fn subtree(&self, container: NodeId) -> Vec<NodeId> {
self.sched
.graph()
.descendants(container)
.map(|n| n.id)
.collect()
}
/// The container's carried domain metadata as an owned read-view. The data
/// lives solely in the [`NodeKind::Dag`] payload — this is a derived read,
/// not a stored side-table.
@ -524,14 +515,6 @@ impl QueueInner {
})
}
/// First failed work node's error, for the dashboard's DAG-level error line.
fn dag_first_error(&self, container: NodeId) -> Option<String> {
self.sched
.graph()
.first_error(container)
.map(ToOwned::to_owned)
}
/// 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
@ -554,10 +537,8 @@ impl QueueInner {
// from a `Done`-filtered node set, so the host computes them here.
let mut started: Vec<DateTime<Utc>> = Vec::new();
let mut finished: Vec<DateTime<Utc>> = Vec::new();
for id in self.subtree(container) {
let Some(node) = self.sched.graph().node(id) else {
continue;
};
for node in self.sched.graph().descendants(container) {
let id = node.id;
if let Some(s) = node.started_at {
started.push(s);
}
@ -635,9 +616,9 @@ impl QueueInner {
/// subtree (read off the graph `Node`, as unix seconds), for the history
/// cap ordering.
fn dag_finished_at(&self, container: NodeId) -> i64 {
self.subtree(container)
.iter()
.filter_map(|id| self.sched.graph().node(*id))
self.sched
.graph()
.descendants(container)
.filter_map(|n| n.finished_at)
.map(|t| t.timestamp())
.max()