From f161f8e40f301928ea40c1ab97bf38c5b34516fa Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 12:32:13 +0200 Subject: [PATCH] feat(jobq): add a job builder that names nodes instead of counting them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `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. --- hive-jobq/src/builder.rs | 521 +++++++++++++++++++++++++++++++++++++ hive-jobq/src/lib.rs | 3 + hive-jobq/src/scheduler.rs | 23 ++ 3 files changed, 547 insertions(+) create mode 100644 hive-jobq/src/builder.rs diff --git a/hive-jobq/src/builder.rs b/hive-jobq/src/builder.rs new file mode 100644 index 00000000..884389da --- /dev/null +++ b/hive-jobq/src/builder.rs @@ -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 { + guid: NodeGuid, + payload: N, + deps: Vec<(NodeGuid, DepWhen)>, + resources: Vec<(R, u32)>, + parent: Option, +} + +/// 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 { + nodes: RefCell>>, + /// Source of handles. Monotonic and builder-local; the value is + /// deliberately meaningless outside this builder. + next_guid: Cell, +} + +// Hand-written rather than derived: `#[derive(Default)]` would demand +// `N: Default, R: Default`, which has nothing to do with an empty builder. +impl Default for JobBuilder { + fn default() -> Self { + Self { + nodes: RefCell::new(Vec::new()), + next_guid: Cell::new(0), + } + } +} + +impl JobBuilder { + /// 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, + root_parent: Option, + ) -> Result, 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, + mut insert: impl FnMut(N, Vec>, Option) -> Result, + ) -> Result, BuildError> { + let mut ids: HashMap = 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> = 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)) { + 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, + 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 Clone for NodeRef<'_, N, R> { + fn clone(&self) -> Self { + *self + } +} +impl Copy for NodeRef<'_, N, R> {} + +impl From> for NodeGuid { + fn from(n: NodeRef<'_, N, R>) -> Self { + n.guid + } +} + +impl From<&NodeRef<'_, N, R>> for NodeGuid { + fn from(n: &NodeRef<'_, N, R>) -> Self { + n.guid + } +} + +impl 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) -> 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) -> 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) -> 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) -> 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> { + 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); + } +} diff --git a/hive-jobq/src/lib.rs b/hive-jobq/src/lib.rs index d89a1763..e6f61e54 100644 --- a/hive-jobq/src/lib.rs +++ b/hive-jobq/src/lib.rs @@ -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. diff --git a/hive-jobq/src/scheduler.rs b/hive-jobq/src/scheduler.rs index e61dd64d..dbc10fb9 100644 --- a/hive-jobq/src/scheduler.rs +++ b/hive-jobq/src/scheduler.rs @@ -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 Scheduler { 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, + root_parent: Option, + ) -> Result, 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