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

feat(jobq): add a job builder that names nodes instead of counting them

Browse source 
`JobBuilder::node(payload)` hands back a `NodeRef` handle obtainable no
other way, and edges are handle -> handle. `insert_into(graph,
root_parent)` consumes the builder, inserts in declaration order, and
returns the id each handle was minted as. There is no intermediate
node-description type: the builder inserts through `Graph::insert`
directly, so nothing has to stay in sync with that signature.

`root_parent` is the group's attachment point — a node that declared no
parent hangs there. That is what makes a job a self-contained sub-DAG: a
template is written without knowing which container node it will live
under, and the same builder serves a runtime-emitted subgraph hanging
off its emitting node.

Generic over the same `N`/`R` as `Graph`, so it belongs to the library
rather than to any one caller's node kind. `.needs(name)` /
`.needs_units(name, count)` declare resource deps at the construction
site, next to the node that needs them.

`Scheduler::insert_job` is the same over a scheduler, via the shared
`insert_with` sink, so a caller building a job never reaches past the
scheduler at the graph underneath.

Declaration order is enforced rather than papered over: a node
referencing one declared later is a named `BuildError::ForwardEdge` /
`ForwardParent`. Sorting for the caller would silently accept a shape
the graph cannot express, and would put ordering logic in a second
place.

Additive — `Graph` is untouched.
This commit is contained in:
atlas 2026-08-02 12:32:13 +02:00 • committed by mara
commit f161f8e40f
3 changed files with 547 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

521
hive-jobq/src/builder.rs Normal file
View file

@ -0,0 +1,521 @@
//! Build a job's sub-DAG by **naming** nodes instead of counting them.
//!
//! [`JobBuilder::node`] hands back a handle, obtainable no other way; edges are
//! handle → handle. So a caller names the node it wants to depend on rather
//! than computing where that node landed, and there is no positional index to
//! get wrong.
//!
//! **An insertion API, not a spec factory.** [`JobBuilder::insert_into`]
//! consumes the builder and puts the nodes straight into a [`Graph`], returning
//! the ids the graph minted. Nothing job-shaped comes back out — there is no
//! intermediate node-description type to keep in sync with [`Graph::insert`]'s
//! signature.
//!
//! **Payload-agnostic.** Generic over the same `N` and `R` as [`Graph`]: the
//! builder knows nothing about what a node *does*, only how nodes relate.
//!
//! # Declaration order
//!
//! Nodes are inserted in declaration order, so a node must be declared *after*
//! everything it references. That is not a limitation the builder invents: node
//! ids are minted by the graph at insert time, so a forward edge has nothing to
//! point at. Declaring one is a [`BuildError::ForwardEdge`] rather than a
//! silent reorder — a builder that sorted for you would quietly accept a shape
//! the graph itself cannot express.
use std::cell::{Cell, RefCell};
use std::collections::HashMap;
use crate::{Dep, DepWhen, Graph, GraphError, NodeId, TerminalState};
/// An opaque identity for a node **within the job being built**.
///
/// Obtainable only from [`JobBuilder::node`], so it can only ever name a node
/// that exists. Deliberately not a [`NodeId`]: those are minted by the graph at
/// insert time and are meaningful system-wide, while this is a builder-local
/// label that stops existing once the job is inserted.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct NodeGuid(u64);
/// Why a job could not be inserted.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum BuildError {
/// A node depends on one declared later. See the module docs on
/// declaration order.
#[error(
"node {node:?} depends on {dep:?}, which is declared later — a dependency \
must be declared before the node that names it"
)]
ForwardEdge {
/// The node carrying the edge.
node: NodeGuid,
/// The not-yet-declared target.
dep: NodeGuid,
},
/// A node is grouped under a parent declared later. Same cause as
/// [`BuildError::ForwardEdge`].
#[error(
"node {node:?} is grouped under {parent:?}, which is declared later — a \
parent must be declared before its children"
)]
ForwardParent {
/// The child node.
node: NodeGuid,
/// The not-yet-declared parent.
parent: NodeGuid,
},
/// The graph rejected an otherwise well-formed node — an out-of-group edge,
/// an unsatisfiable [`DepWhen`], and so on.
#[error(transparent)]
Graph(#[from] GraphError),
}
/// One node as the builder holds it: edges and parent still name *handles*, so
/// nothing here depends on ids the graph has not minted yet.
#[derive(Debug)]
struct Pending<N, R> {
guid: NodeGuid,
payload: N,
deps: Vec<(NodeGuid, DepWhen)>,
resources: Vec<(R, u32)>,
parent: Option<NodeGuid>,
}
/// Accumulates a job's nodes, handing back a handle for each.
///
/// Sub-builders take `&JobBuilder` and **return their handles**, which is what
/// removes the base-offset parameter a positional API needs: a caller names the
/// node it wants rather than computing where that node landed.
#[derive(Debug)]
pub struct JobBuilder<N, R> {
nodes: RefCell<Vec<Pending<N, R>>>,
/// Source of handles. Monotonic and builder-local; the value is
/// deliberately meaningless outside this builder.
next_guid: Cell<u64>,
}
// Hand-written rather than derived: `#[derive(Default)]` would demand
// `N: Default, R: Default`, which has nothing to do with an empty builder.
impl<N, R> Default for JobBuilder<N, R> {
fn default() -> Self {
Self {
nodes: RefCell::new(Vec::new()),
next_guid: Cell::new(0),
}
}
}
impl<N, R> JobBuilder<N, R> {
/// A fresh, empty builder.
#[must_use]
pub fn new() -> Self {
Self::default()
}
/// Add a node carrying `payload`, with no edges, resources, or parent yet.
///
/// The returned handle is where those are declared; it is [`Copy`], so it
/// can be named as a dependency as many times as needed.
pub fn node(&self, payload: N) -> NodeRef<'_, N, R> {
let guid = NodeGuid(self.next_guid.get());
self.next_guid.set(guid.0 + 1);
self.nodes.borrow_mut().push(Pending {
guid,
payload,
deps: Vec::new(),
resources: Vec::new(),
parent: None,
});
NodeRef {
builder: self,
guid,
}
}
/// Insert every declared node into `graph`, in declaration order, and
/// return the id each handle's node was minted as.
///
/// `root_parent` is the group's attachment point: a node that declared no
/// [`NodeRef::part_of`] hangs there rather than at the top level. That is
/// what makes a job a *self-contained sub-DAG* — the whole thing goes under
/// one container node, or under the runtime node that emitted it, and the
/// job's own declarations stay relative.
///
/// The returned map is how a caller finds out what its handles became: hold
/// the handles it cares about, then look them up here. Handles that were
/// never looked up cost nothing.
///
/// # Errors
///
/// [`BuildError::ForwardEdge`] / [`BuildError::ForwardParent`] if a node
/// references one declared after it, or [`BuildError::Graph`] if the graph
/// rejects a node (see [`Graph::insert`]).
pub fn insert_into(
self,
graph: &mut Graph<N, R>,
root_parent: Option<NodeId>,
) -> Result<HashMap<NodeGuid, NodeId>, BuildError> {
self.insert_with(root_parent, |payload, deps, parent| {
graph.insert(payload, deps, parent)
})
}
/// [`JobBuilder::insert_into`] against an arbitrary sink — the same
/// resolution, for a caller that inserts through something wrapping the
/// graph (e.g. [`crate::scheduler::Scheduler::insert_job`], which has
/// bookkeeping of its own to do per node).
///
/// # Errors
///
/// As [`JobBuilder::insert_into`].
///
/// # Partial insertion
///
/// An error leaves the nodes inserted *before* it in the sink. Callers that
/// need all-or-nothing should insert into a scratch graph, or treat a
/// failure as fatal — every variant is a programming error in the job's own
/// shape, not a runtime condition to recover from.
pub fn insert_with(
self,
root_parent: Option<NodeId>,
mut insert: impl FnMut(N, Vec<Dep<R>>, Option<NodeId>) -> Result<NodeId, GraphError>,
) -> Result<HashMap<NodeGuid, NodeId>, BuildError> {
let mut ids: HashMap<NodeGuid, NodeId> = HashMap::new();
for pending in self.nodes.into_inner() {
let parent = match pending.parent {
None => root_parent,
Some(p) => Some(*ids.get(&p).ok_or(BuildError::ForwardParent {
node: pending.guid,
parent: p,
})?),
};
let mut deps: Vec<Dep<R>> = Vec::with_capacity(pending.deps.len());
for (on, when) in pending.deps {
let id = *ids.get(&on).ok_or(BuildError::ForwardEdge {
node: pending.guid,
dep: on,
})?;
deps.push(Dep::Node { id, when });
}
deps.extend(
pending
.resources
.into_iter()
.map(|(name, count)| Dep::Resource { name, count }),
);
let id = insert(pending.payload, deps, parent)?;
ids.insert(pending.guid, id);
}
Ok(ids)
}
/// Apply `f` to the node named by `guid`.
fn with_node(&self, guid: NodeGuid, f: impl FnOnce(&mut Pending<N, R>)) {
let mut nodes = self.nodes.borrow_mut();
if let Some(n) = nodes.iter_mut().find(|n| n.guid == guid) {
f(n);
}
}
}
/// A handle to one declared node, and the surface its edges are declared on.
///
/// [`Copy`] because it is only a builder borrow plus a handle: naming a node as
/// a dependency must not consume the ability to name it again.
#[derive(Debug)]
pub struct NodeRef<'a, N, R> {
builder: &'a JobBuilder<N, R>,
guid: NodeGuid,
}
// Hand-written for the same reason as `Default` above: deriving would demand
// `N: Copy, R: Copy`, but a handle copies a borrow and an integer.
impl<N, R> Clone for NodeRef<'_, N, R> {
fn clone(&self) -> Self {
*self
}
}
impl<N, R> Copy for NodeRef<'_, N, R> {}
impl<N, R> From<NodeRef<'_, N, R>> for NodeGuid {
fn from(n: NodeRef<'_, N, R>) -> Self {
n.guid
}
}
impl<N, R> From<&NodeRef<'_, N, R>> for NodeGuid {
fn from(n: &NodeRef<'_, N, R>) -> Self {
n.guid
}
}
impl<N, R> NodeRef<'_, N, R> {
/// This node's handle, for callers that want to hold the identity without
/// the builder borrow (e.g. to look the id up after
/// [`JobBuilder::insert_into`]).
#[must_use]
pub fn guid(self) -> NodeGuid {
self.guid
}
/// Run only after `on` succeeds — the common chain link.
///
/// Chained against several nodes it means *all* of them succeeded, and this
/// node is ruled out ([`TerminalState::Skipped`]) the moment one doesn't —
/// which is the signal [`NodeRef::on_elimination_of`] waits for.
#[must_use]
pub fn after_ok(self, on: impl Into<NodeGuid>) -> Self {
self.edge(on.into(), DepWhen::AFTER_OK)
}
/// Run once `on` is terminal, however it ended. The cleanup edge: a node
/// that must run whether or not the work it follows succeeded.
#[must_use]
pub fn after_any(self, on: impl Into<NodeGuid>) -> Self {
self.edge(on.into(), DepWhen::AFTER_ANY)
}
/// Run only if `on` was **ruled out** — i.e. its own `after_ok` edges did
/// not hold.
///
/// This is how *"any one of several nodes failed"* gets expressed at all.
/// Dependency edges are conjunctive, so that condition cannot be written
/// directly; the composition that expresses it is a pair — point the
/// success branch at every root with [`NodeRef::after_ok`], then hang the
/// failure branch off *that* node's elimination. Exactly one of the pair
/// ever runs.
///
/// It accepts [`TerminalState::Skipped`] and **not**
/// [`TerminalState::Cancelled`] on purpose: if the whole job was dropped
/// before it started, the success branch is `Cancelled` directly and this
/// branch is ruled out too — a job nobody ran reports nothing.
#[must_use]
pub fn on_elimination_of(self, on: impl Into<NodeGuid>) -> Self {
self.edge(on.into(), DepWhen::of(&[TerminalState::Skipped]))
}
/// Hold one unit of the named resource while this node and its subtree run.
///
/// Declared on every node that needs it, even when an ancestor already
/// holds the same resource — the ancestor's grant is re-used (a re-entrant
/// borrow), so declaring it costs nothing and makes the node's needs
/// readable where the node is written.
#[must_use]
pub fn needs(self, name: R) -> Self {
self.needs_units(name, 1)
}
/// [`NodeRef::needs`] with an explicit unit count, for a resource whose
/// capacity is a budget rather than a mutex.
#[must_use]
pub fn needs_units(self, name: R, count: u32) -> Self {
self.builder
.with_node(self.guid, |n| n.resources.push((name, count)));
self
}
/// Group this node under `parent`.
///
/// The child runs once its parent reaches [`crate::State::Finishing`] (the
/// parent gate), so it must not also depend on that parent — dep-scope
/// validation rejects that, and it would deadlock. A group root owns
/// whatever resource it declares for its whole subtree.
#[must_use]
pub fn part_of(self, parent: impl Into<NodeGuid>) -> Self {
let parent = parent.into();
self.builder
.with_node(self.guid, |n| n.parent = Some(parent));
self
}
/// Record one dependency edge and hand the handle back for chaining.
fn edge(self, on: NodeGuid, when: DepWhen) -> Self {
self.builder
.with_node(self.guid, |n| n.deps.push((on, when)));
self
}
}
#[cfg(test)]
mod tests {
use super::{BuildError, JobBuilder, NodeGuid};
use crate::{Dep, DepWhen, Graph, NodeId};
/// A graph whose payload is a name and whose resources are strings.
fn graph() -> Graph<&'static str, &'static str> {
Graph::new()
}
fn deps_of(g: &Graph<&'static str, &'static str>, id: NodeId) -> Vec<Dep<&'static str>> {
g.node(id).expect("node present").deps.clone()
}
/// The point of the handle layer: an edge declared against a *handle* comes
/// out addressing the id that node was actually minted as.
#[test]
fn edges_resolve_to_minted_ids() {
let mut g = graph();
let b = JobBuilder::new();
let first = b.node("a");
let second = b.node("b").after_ok(first);
let (first, second) = (first.guid(), second.guid());
let ids = b.insert_into(&mut g, None).expect("insert");
assert_eq!(
deps_of(&g, ids[&second]),
vec![Dep::Node {
id: ids[&first],
when: DepWhen::AFTER_OK
}]
);
assert!(deps_of(&g, ids[&first]).is_empty());
}
#[test]
fn parent_resolves_to_a_minted_id() {
let mut g = graph();
let b = JobBuilder::new();
let root = b.node("a");
let child = b.node("b").part_of(root).guid();
let root = root.guid();
let ids = b.insert_into(&mut g, None).expect("insert");
assert_eq!(g.node(ids[&root]).expect("root").parent, None);
assert_eq!(g.node(ids[&child]).expect("child").parent, Some(ids[&root]));
}
/// A handle is `Copy`, so naming the same node as a dependency twice must
/// not consume it — the fan-out every composite job needs.
#[test]
fn one_handle_can_be_depended_on_twice() {
let mut g = graph();
let b = JobBuilder::new();
let shared = b.node("a");
let ok = b.node("b").after_ok(shared).guid();
let any = b.node("c").after_any(shared).guid();
let shared = shared.guid();
let ids = b.insert_into(&mut g, None).expect("insert");
assert_eq!(
deps_of(&g, ids[&ok]),
vec![Dep::Node {
id: ids[&shared],
when: DepWhen::AFTER_OK
}]
);
assert_eq!(
deps_of(&g, ids[&any]),
vec![Dep::Node {
id: ids[&shared],
when: DepWhen::AFTER_ANY
}]
);
}
/// Resource deps ride along with the node deps, in one insert.
#[test]
fn resources_become_resource_deps() {
let mut g = graph();
let b = JobBuilder::new();
let only = b.node("a").needs("agent/atlas").needs_units("build", 2);
let only = only.guid();
let ids = b.insert_into(&mut g, None).expect("insert");
assert_eq!(
deps_of(&g, ids[&only]),
vec![
Dep::Resource {
name: "agent/atlas",
count: 1
},
Dep::Resource {
name: "build",
count: 2
},
]
);
}
/// A handle stays live across later `node()` calls, so it is possible to
/// aim an edge *backwards* in declaration order. The graph mints ids as it
/// inserts, so there is nothing for that edge to point at — say so instead
/// of quietly reordering.
#[test]
fn a_forward_edge_is_rejected_by_name() {
let mut g = graph();
let b = JobBuilder::new();
let first = b.node("a");
let second = b.node("b");
let _ = first.after_any(second);
let err = b.insert_into(&mut g, None).expect_err("forward edge");
assert_eq!(
err,
BuildError::ForwardEdge {
node: NodeGuid(0),
dep: NodeGuid(1)
}
);
}
#[test]
fn a_forward_parent_is_rejected_by_name() {
let mut g = graph();
let b = JobBuilder::new();
let child = b.node("a");
let parent = b.node("b");
let _ = child.part_of(parent);
let err = b.insert_into(&mut g, None).expect_err("forward parent");
assert_eq!(
err,
BuildError::ForwardParent {
node: NodeGuid(0),
parent: NodeGuid(1)
}
);
}
/// The graph's own validation still applies — the builder does not
/// pre-empt it.
#[test]
fn graph_rejection_surfaces_as_is() {
let mut g = graph();
let b = JobBuilder::new();
let root = b.node("root");
// A child may not depend on its own parent: the parent gate already
// orders them, and the edge would deadlock.
let _ = b.node("child").part_of(root).after_ok(root);
let err = b.insert_into(&mut g, None).expect_err("out-of-group dep");
assert!(matches!(err, BuildError::Graph(_)), "{err:?}");
}
/// A job's own roots hang under the group's attachment point, while a node
/// that named a parent inside the job keeps it. This is what lets a
/// template be written without knowing the container it will live under.
#[test]
fn root_parent_adopts_only_the_jobs_own_roots() {
let mut g = graph();
let container = g.insert("container", Vec::new(), None).expect("container");
let b = JobBuilder::new();
let root = b.node("root");
let child = b.node("child").part_of(root).guid();
let root = root.guid();
let ids = b.insert_into(&mut g, Some(container)).expect("insert");
assert_eq!(g.node(ids[&root]).expect("root").parent, Some(container));
assert_eq!(g.node(ids[&child]).expect("child").parent, Some(ids[&root]));
}
#[test]
fn an_empty_builder_inserts_nothing() {
let mut g = graph();
let ids = JobBuilder::<&str, &str>::new()
.insert_into(&mut g, None)
.expect("insert");
assert!(ids.is_empty());
assert_eq!(g.nodes().count(), 0);
}
}

3
hive-jobq/src/lib.rs
View file

@ -26,9 +26,12 @@
//! re-entrant borrow, one branch at a time). Single-threaded — the scheduler
//! owns the resource table and mutates it directly. See [`scheduler`].
pub mod builder;
pub mod resources;
pub mod scheduler;
pub use builder::{BuildError, JobBuilder, NodeGuid, NodeRef};
use chrono::{DateTime, Utc};
/// Opaque, stable, monotonic node identifier.

23
hive-jobq/src/scheduler.rs
View file

@ -31,6 +31,7 @@
use std::collections::HashMap;
use std::hash::Hash;
use crate::builder::{BuildError, JobBuilder, NodeGuid};
use crate::resources::ResourceTable;
use crate::{Dep, Graph, GraphError, NodeId, State, TerminalState};
@ -99,6 +100,28 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
self.graph.insert(payload, deps, parent)
}
/// Insert a whole job — every node a [`JobBuilder`] declared — under
/// `root_parent`, returning the id each handle's node was minted as.
///
/// The scheduler-side counterpart of [`JobBuilder::insert_into`]: same
/// resolution, but each node goes through [`Scheduler::append`], so a
/// caller building a job never has to reach past the scheduler at the graph
/// underneath. Call [`Scheduler::settle`] afterwards to start whatever
/// became runnable.
///
/// # Errors
/// Propagates [`BuildError`] — a forward reference in the job's own
/// declarations, or a graph rejection.
pub fn insert_job(
&mut self,
job: JobBuilder<N, R>,
root_parent: Option<NodeId>,
) -> Result<HashMap<NodeGuid, NodeId>, BuildError> {
job.insert_with(root_parent, |payload, deps, parent| {
self.append(payload, deps, parent)
})
}
/// Claim every currently-runnable pending node and start it: node-deps
/// satisfied and all resource-deps acquired atomically (all-or-nothing).
/// Each claimed node is marked `Running`, its acquired units recorded, and