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(#2591): make NodeKind the queue payload — drop JobPayload, agent into variants

Browse source
This commit is contained in:
atlas 2026-07-20 23:00:20 +02:00 • committed by mara
commit be2dfa8cd3
8 changed files with 233 additions and 177 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -4,22 +4,22 @@
//! watcher, deferred-start follow-up, meta-update cascade) collapse into DAG
//! *shapes* over the shared node primitives ([`model::NodeKind`]).
//!
//! [`hive_jobq`] owns the graph, the two-class resource pool, and the settle
//! loop; this module maps hive-c0re's concepts onto it:
//! - `NodeKind` + agent → the crate payload [`resource::JobPayload`];
//! - the two resource classes → [`resource::Resource`]
//! ([`resource::Resource::BuildSlot`] node-held, [`resource::Resource::Agent`]
//! subtree-held), derived per node by [`resource::JobPayload::resource_deps`];
//! - a host **DAG id** groups a set of crate nodes; a node declares only its
//! `deps` (chain edges + resource deps), and the crate scheduler infers lease
//! re-entrancy from the [`Dep::Node`] graph — a node needing an agent lease a
//! node it depends on already holds re-enters it, with no parent annotation;
//! - per-DAG terminal work is a *focused* graph node per concern — `ResolveApproval`
//! (approval DAGs), `EmitRebuilt` (rebuild / perm-change), `RevertIntent`
//! (cancelled power-ops) — weak-depending on the DAG's tail nodes (extended onto
//! any runtime-appended subgraph's tail), so it runs once everything settles.
//! Hook-less DAGs (meta-update / boot / reconcile) get none. No drained event
//! stream, and no one node branching on DAG metadata.
//! [`hive_jobq`] owns the graph, the two-class resource pool, and the roll-up
//! settle loop; this module maps hive-c0re's concepts onto it:
//! - [`model::NodeKind`] **is** the crate payload `N` directly — each variant
//! carries the agent it targets ([`NodeKind::agent`]); the two resource
//! classes are [`resource::Resource`] (`BuildSlot` node-held, `Agent` lease
//! subtree-held), derived per node by [`NodeKind::resource_deps`];
//! - a **DAG is a single container node** ([`NodeKind::Dag`], `parent = None`)
//! carrying the group's metadata, with the template's nodes hung under it as
//! its subtree (the **parent axis** groups; `deps` order). So the container's
//! `NodeId` is the DAG id, its rolled-up state is the DAG state, and membership
//! is a graph walk — there are no host grouping side-tables. The lease is owned
//! by a subtree root and borrowed by its descendants (continuity);
//! - per-DAG terminal work runs **inline** ([`exec::run_terminal_hook`]) when the
//! container rolls up terminal — dispatched off its template
//! ([`terminal_hook`]): approval-resolve, `Rebuilt`-emit, or power-intent
//! revert. No terminal-hook node, no drained event stream.
//!
//! The queue is runtime-only (no persistence): an empty graph on boot; desired
//! state is re-derived by the reconcile sweep. A single scheduler task
@ -49,7 +49,7 @@ use crate::coordinator::TransientKind;
pub use model::{
DagSpec, DagView, DepWhen, NodeKind, NodeSpec, PermPayload, Source, State, Template,
};
use resource::{JobPayload, Resource};
use resource::Resource;
/// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) to retain per
/// template in the snapshot, matching the old per-kind history cap.
@ -130,7 +130,7 @@ struct DagMeta {
/// are graph queries ([`QueueInner::container`] / [`QueueInner::subtree`] /
/// [`QueueInner::dag_meta`]). One shared crate [`Graph`] holds every DAG.
struct QueueInner {
sched: Scheduler<JobPayload, Resource>,
sched: Scheduler<NodeKind, Resource>,
/// Per-node runtime metadata (build-log id, step, timestamps, error) —
/// mutable after insert, so it can't ride the immutable node payload.
node_rt: HashMap<NodeId, NodeRuntime>,
@ -232,10 +232,7 @@ fn insert_group(
) -> anyhow::Result<Vec<NodeId>> {
let mut ids: Vec<NodeId> = Vec::with_capacity(nodes.len());
for ns in nodes {
let payload = JobPayload {
kind: ns.kind.clone(),
agent: ns.agent.clone(),
};
let payload = ns.kind.clone();
let mut deps = payload.resource_deps();
for d in &ns.deps {
deps.push(Dep::Node {
@ -293,18 +290,15 @@ impl JobQueue {
let container = inner
.sched
.append(
JobPayload {
kind: NodeKind::Dag {
template: spec.template,
source: spec.source,
reason: spec.reason,
transient: spec.transient,
approval_id: spec.approval_id,
inputs: spec.inputs,
perm_payload: spec.perm_payload,
created_at: now_unix(),
},
agent: String::new(),
NodeKind::Dag {
template: spec.template,
source: spec.source,
reason: spec.reason,
transient: spec.transient,
approval_id: spec.approval_id,
inputs: spec.inputs,
perm_payload: spec.perm_payload,
created_at: now_unix(),
},
Vec::new(),
None,
@ -377,8 +371,8 @@ impl JobQueue {
let Some(node) = inner.sched.graph().node(id) else {
continue;
};
let kind = node.payload.kind.clone();
let agent = node.payload.agent.clone();
let kind = node.payload.clone();
let agent = node.payload.agent().to_owned();
let Some(container) = inner.dag_of(id) else {
continue;
};
@ -605,7 +599,7 @@ impl QueueInner {
self.sched.graph().nodes().find_map(|n| {
(n.parent.is_none()
&& n.id.get() == dag_id
&& matches!(n.payload.kind, NodeKind::Dag { .. }))
&& matches!(n.payload, NodeKind::Dag { .. }))
.then_some(n.id)
})
}
@ -646,7 +640,7 @@ impl QueueInner {
inputs,
perm_payload,
created_at,
} = &self.sched.graph().node(container)?.payload.kind
} = &self.sched.graph().node(container)?.payload
else {
return None;
};
@ -714,9 +708,9 @@ impl QueueInner {
let mut seen: Vec<String> = Vec::new();
for id in self.subtree(container) {
if let Some(n) = self.sched.graph().node(id) {
let agent = &n.payload.agent;
let agent = n.payload.agent();
if !agent.is_empty() && !seen.iter().any(|s| s == agent) {
seen.push(agent.clone());
seen.push(agent.to_owned());
}
}
}
@ -780,8 +774,8 @@ impl QueueInner {
}
nodes.push(NodeView {
id: id.get(),
agent: node.payload.agent.clone(),
kind: node.payload.kind.as_str().to_owned(),
agent: node.payload.agent().to_owned(),
kind: node.payload.as_str().to_owned(),
deps,
state: to_wire_state(node.state),
step: rt.and_then(|r| r.step.clone()),
@ -827,7 +821,7 @@ impl QueueInner {
self.sched
.graph()
.nodes()
.filter(|n| n.parent.is_none() && matches!(n.payload.kind, NodeKind::Dag { .. }))
.filter(|n| n.parent.is_none() && matches!(n.payload, NodeKind::Dag { .. }))
.map(|n| n.id)
.collect()
}