From f035b63b9ad898faa8edba0aea3964317252a523 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 14:39:37 +0200 Subject: [PATCH] jobq: one insertion entry point, and make it atomic Three findings from the operator's review, all correct. 1. Two insert_job's. Graph::insert_job had no caller outside hive-jobq's own tests -- production only ever went through Scheduler::insert_job. It existed because the graph-level one got written first. Deleted; the tests moved onto a Scheduler, which is where insertion belongs anyway. 2. insert_job was not atomic, and the previous commit made that worse: a forward edge or forward parent surfaced mid-loop, leaving the nodes before it in the graph, and resolve_wanted ran after every insert, so an unknown handle failed once the whole job was already committed. The module documented this under "Partial insertion" instead of fixing it -- prose describing a hole is not a design. All three are decidable from what the builder holds, so check_declaration_order now runs before the first insert and the loop indexes ids directly. A malformed job leaves the graph untouched. What remains mid-insert is the graph's own rejection (out-of-group dep, empty DepWhen); closing that needs a dry-run validate on Graph, which is a separate change. 3. DagSpec no longer boxes its recipe: it is generic over the closure, which travels from the template that built it straight into submit. The box bought type inference, and paying for it costs annotations -- `|b: &Job|` at each declaration site (the field needs an HRTB, and an unannotated closure binds one lifetime) and `+ use<>` on each returning signature (or the opaque type captures the caller's borrows). Erasure is still needed where several recipe shapes share one type: the boxed Declare stays for the executor's append_subgraph, and a test table uses an erase() helper. --- hive-c0re/src/job_queue/mod.rs | 20 +-- hive-c0re/src/job_queue/model.rs | 13 +- hive-c0re/src/job_queue/submit.rs | 18 +-- hive-c0re/src/job_queue/templates.rs | 48 +++++--- hive-c0re/src/job_queue/tests.rs | 60 +++++++-- hive-jobq/src/builder.rs | 178 +++++++++++++-------------- hive-jobq/src/lib.rs | 29 ----- hive-jobq/src/scheduler.rs | 21 ++-- 8 files changed, 211 insertions(+), 176 deletions(-) diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index c31e7be4..9b9db781 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -179,7 +179,7 @@ impl Default for JobQueue { /// Propagates a crate graph-insert error (malformed dep/parent / dep-scope). fn insert_group( inner: &mut QueueInner, - declare: Declare, + declare: impl FnOnce(&Job), group_parent: Option, ) -> anyhow::Result<()> { inner @@ -216,15 +216,19 @@ impl JobQueue { self.inner.lock().expect("job_queue mutex poisoned") } - /// Submit a DAG. Validates the spec, inserts a [`NodeKind::Dag`] **container - /// node** carrying the group's metadata, then inserts the template's nodes as - /// its subtree (their roots re-parented to the container). Returns the - /// container's id as the DAG id — its rolled-up state is the DAG state. + /// Submit a DAG: insert a [`NodeKind::Dag`] **container node** carrying the + /// group's metadata, then insert the template's nodes as its subtree (their + /// roots re-parented to the container). Returns the container's id as the + /// DAG id — its rolled-up state is the DAG state. + /// + /// Takes the spec's recipe by generic, not as a boxed [`Declare`]: a spec + /// travels from the template that built it directly into this call, so + /// there is nothing to allocate for. /// /// # Errors - /// Propagates the spec-validation error (empty / cyclic / bad parent) or a - /// graph-insert error (dependencies that aren't dependency-topological). - pub fn submit(&self, spec: DagSpec) -> anyhow::Result { + /// Propagates a graph-insert error (dependencies that aren't + /// dependency-topological). + pub fn submit(&self, spec: DagSpec) -> anyhow::Result { let mut inner = self.lock(); let container = inner .sched diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 2463097a..19556271 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -475,16 +475,23 @@ impl NodeKind { /// moment it inserts. A shape that has been declared is therefore always /// insertable — a dangling edge or a cycle cannot be expressed, so there is /// nothing left for a submit-time validation pass to reject. -pub struct DagSpec { +/// +/// Generic over the recipe rather than boxing it: a spec goes from the template +/// that returns it straight to the `submit` that consumes it, so the closure's +/// concrete type is known the whole way and needs neither an allocation nor a +/// `Send` bound. (The executor's `append_subgraph` is the case that *does* need +/// a boxed [`super::Declare`] — its recipes are collected into a `Vec` and +/// applied later, across a task boundary.) +pub struct DagSpec { pub source: Source, /// Free-form "why". pub reason: String, /// Declares the DAG's nodes — their edges, grouping and resources — onto /// the builder the queue hands it. - pub declare: super::Declare, + pub declare: F, } -impl std::fmt::Debug for DagSpec { +impl std::fmt::Debug for DagSpec { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { // The recipe is a closure; there is nothing to show of it, and its // nodes do not exist until the queue runs it. diff --git a/hive-c0re/src/job_queue/submit.rs b/hive-c0re/src/job_queue/submit.rs index edc3444f..7aaf1878 100644 --- a/hive-c0re/src/job_queue/submit.rs +++ b/hive-c0re/src/job_queue/submit.rs @@ -26,11 +26,11 @@ use std::sync::Arc; use super::model::{DagSpec, NodeKind}; use super::templates::{RebuildOpts, node, rebuild_nodes}; -use super::{Declare, Job, Source, templates}; +use super::{Job, Source, templates}; use crate::coordinator::Coordinator; use crate::lifecycle; -fn submit_and_emit(coord: &Arc, spec: super::DagSpec) -> u64 { +fn submit_and_emit(coord: &Arc, spec: super::DagSpec) -> u64 { let id = coord .job_queue .submit(spec) @@ -169,7 +169,7 @@ fn restart_chain(b: &Job, agent: &str, graceful: bool, running: bool) { /// and each keeps its own root, so the per-agent subgraphs are independent and /// run concurrently, each on its own lease. Rebasing one subgraph's indices /// onto another's used to be a function. -fn power_dag(source: Source, reason: String, declare: Declare) -> DagSpec { +fn power_dag(source: Source, reason: String, declare: F) -> DagSpec { DagSpec { source, reason, @@ -188,12 +188,12 @@ pub(crate) fn stop_spec( graceful: bool, source: Source, reason: String, -) -> DagSpec { +) -> DagSpec> { let targets = targets.to_vec(); power_dag( source, reason, - Box::new(move |b| { + Box::new(move |b: &Job| { for (agent, running) in targets { stop_chain(b, &agent, graceful, running); } @@ -211,12 +211,12 @@ pub(crate) fn start_spec( targets: &[(String, bool, bool)], source: Source, reason: String, -) -> DagSpec { +) -> DagSpec> { let targets = targets.to_vec(); power_dag( source, reason, - Box::new(move |b| { + Box::new(move |b: &Job| { for (agent, running, stale) in targets { start_chain(b, &agent, running, stale); } @@ -230,12 +230,12 @@ pub(crate) fn restart_spec( graceful: bool, source: Source, reason: String, -) -> DagSpec { +) -> DagSpec> { let targets = targets.to_vec(); power_dag( source, reason, - Box::new(move |b| { + Box::new(move |b: &Job| { for (agent, running) in targets { restart_chain(b, &agent, graceful, running); } diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index 71aedc05..7d29b946 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -241,7 +241,7 @@ pub(crate) fn rebuild_nodes<'a>( /// already holding it rather than deadlocking against it. pub(crate) fn deploy_rebuild_nodes(agent: &str, approval_id: i64) -> Declare { let agent = agent.to_owned(); - Box::new(move |b| { + Box::new(move |b: &Job| { let roots = rebuild_nodes( b, &agent, @@ -274,12 +274,17 @@ pub(crate) fn deploy_rebuild_nodes(agent: &str, approval_id: i64) -> Declare { /// whole `StopForUpdate`→`Swap`→`PostSwap` subtree, so those three cover every /// node. Edging `Reconcile` alone would not do: it is `AfterAny` `Prebuild`, so /// it reaches `Done` even after a failed swap and the tail would report success. -pub fn rebuild(agent: &str, source: Source, reason: String, relock: bool) -> DagSpec { +pub fn rebuild( + agent: &str, + source: Source, + reason: String, + relock: bool, +) -> DagSpec> { let agent = agent.to_owned(); DagSpec { source, reason, - declare: Box::new(move |b| { + declare: Box::new(move |b: &Job| { let roots = rebuild_nodes( b, &agent, @@ -319,12 +324,16 @@ pub fn rebuild(agent: &str, source: Source, reason: String, relock: bool) -> Dag /// /// The window still spans the container build, as it must: `prepare_deploy` /// leaves `flake.lock` staged-uncommitted for the build's whole duration. -pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec { +pub fn approval_deploy( + agent: &str, + approval_id: i64, + reason: String, +) -> DagSpec> { let agent = agent.to_owned(); DagSpec { source: Source::Approval, reason, - declare: Box::new(move |b| { + declare: Box::new(move |b: &Job| { let a = || agent.clone(); let window = node( b, @@ -371,12 +380,16 @@ pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec /// single-node lifecycle DAGs that exercise per-agent lease serialization /// in the queue tests); production paths no longer emit a bare reconcile. #[cfg(test)] -pub fn reconcile_only(agent: &str, source: Source, reason: String) -> DagSpec { +pub fn reconcile_only( + agent: &str, + source: Source, + reason: String, +) -> DagSpec> { let agent = agent.to_owned(); DagSpec { source, reason, - declare: Box::new(move |b| { + declare: Box::new(move |b: &Job| { let _reconcile = node(b, NodeKind::Reconcile { agent }); }), } @@ -393,12 +406,12 @@ pub fn reconcile_only(agent: &str, source: Source, reason: String) -> DagSpec { /// container was never created). Closed by a `ResolveApproval` tail root edged /// `AfterAny` onto `Provision` — the DAG's only other group-root, so its roll-up /// already carries the whole cascade. -pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec { +pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec> { let agent = agent.to_owned(); DagSpec { source: Source::Approval, reason, - declare: Box::new(move |b| { + declare: Box::new(move |b: &Job| { let a = || agent.clone(); let provision = node(b, NodeKind::Provision { agent: a() }); let create = node(b, NodeKind::Create { agent: a() }).part_of(provision); @@ -417,12 +430,17 @@ pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec { /// effect in the container. Group-roots are `WritePermFile` plus the rebuild /// subgraph's `MetaSync` / `Prebuild` / `Reconcile`, so the `EmitRebuilt` tail /// edges all four. -pub fn perm_change(agent: &str, source: Source, reason: String, payload: PermPayload) -> DagSpec { +pub fn perm_change( + agent: &str, + source: Source, + reason: String, + payload: PermPayload, +) -> DagSpec> { let agent = agent.to_owned(); DagSpec { source, reason, - declare: Box::new(move |b| { + declare: Box::new(move |b: &Job| { let write = node( b, NodeKind::WritePermFile { @@ -463,11 +481,11 @@ pub fn meta_update( source: Source, reason: String, approval_id: Option, -) -> DagSpec { +) -> DagSpec> { DagSpec { source, reason, - declare: Box::new(move |b| { + declare: Box::new(move |b: &Job| { let lock = node( b, NodeKind::MetaLock { @@ -501,11 +519,11 @@ pub fn reparent( moves: Vec<(hive_types::Ident, Option)>, source: Source, reason: String, -) -> DagSpec { +) -> DagSpec> { DagSpec { source, reason, - declare: Box::new(move |b| { + declare: Box::new(move |b: &Job| { let _reparent = node(b, NodeKind::Reparent { moves }); }), } diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 1584c281..8b026b01 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -9,15 +9,29 @@ use super::model::NodeKind; use super::*; -fn submit(q: &JobQueue, spec: DagSpec) -> u64 { +fn submit(q: &JobQueue, spec: DagSpec) -> u64 { q.submit(spec).expect("valid spec") } +/// Erase a spec's recipe to the boxed [`Declare`] so specs of *different* +/// shapes can share one type — e.g. a table of `(name, spec)` cases. +/// +/// Production never needs this: each submit path builds one spec and hands it +/// straight to `submit`, so the concrete closure type is known end to end. A +/// test table is the case where several shapes must be one type. +fn erase(spec: DagSpec) -> DagSpec { + DagSpec { + source: spec.source, + reason: spec.reason, + declare: Box::new(spec.declare), + } +} + fn ident(s: &str) -> hive_types::Ident { hive_types::Ident::parse(s).expect("valid test ident") } -fn rebuild(agent: &str, reason: &str) -> DagSpec { +fn rebuild(agent: &str, reason: &str) -> DagSpec> { templates::rebuild(agent, Source::Manual, reason.to_owned(), true) } @@ -25,14 +39,22 @@ fn rebuild(agent: &str, reason: &str) -> DagSpec { /// shape (`[Signal→Drain→] StopForUpdate → Reconcile`, no `SetWanted` head) /// most queue-mechanics tests assume. Mirrors the pre-dynamic /// `templates::restart` (which is now the state-aware `submit::restart_spec`). -fn restart_online(agents: &[&str], graceful: bool, reason: &str) -> DagSpec { +fn restart_online( + agents: &[&str], + graceful: bool, + reason: &str, +) -> DagSpec> { let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect(); submit::restart_spec(&targets, graceful, Source::Manual, reason.to_owned()) } /// Stop DAG spec with every agent treated as **running** — the online shape /// (`SetWanted → [Signal→Drain→](graceful) Reconcile`). -fn stop_online(agents: &[&str], graceful: bool, reason: &str) -> DagSpec { +fn stop_online( + agents: &[&str], + graceful: bool, + reason: &str, +) -> DagSpec> { let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect(); submit::stop_spec(&targets, graceful, Source::Manual, reason.to_owned()) } @@ -197,7 +219,7 @@ fn graceful_rebuild_chain_drains_before_stopping() { DagSpec { source: Source::AutoUpdate, reason: "sweep".to_owned(), - declare: Box::new(|b| { + declare: Box::new(|b: &Job| { templates::rebuild_nodes( b, "agent-a", @@ -246,7 +268,7 @@ fn non_graceful_rebuild_has_no_signal_or_drain() { DagSpec { source: Source::Manual, reason: "manual".to_owned(), - declare: Box::new(|b| { + declare: Box::new(|b: &Job| { templates::rebuild_nodes( b, "agent-a", @@ -695,7 +717,7 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() { let spec = DagSpec { source: Source::AutoUpdate, reason: "sweep".to_owned(), - declare: Box::new(|b| { + declare: Box::new(|b: &Job| { let _lock = templates::node( b, NodeKind::MetaLock { @@ -715,7 +737,7 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() { // match the sweep arm of `run_meta_lock` or this stops tracking production. let subgraph = |agent: &str| -> Declare { let agent = agent.to_owned(); - Box::new(move |b| { + Box::new(move |b: &Job| { templates::rebuild_nodes( b, &agent, @@ -821,7 +843,7 @@ fn meta_update_grows_cascade_in_dag() { // Simulate the executor growing the cascade in-DAG (`relock = false` — a // cascade child must not re-lock and revert the parent's bump). for agent in ["alice", "bob"] { - let declare: Declare = Box::new(move |b| { + let declare: Declare = Box::new(move |b: &Job| { templates::rebuild_nodes( b, agent, @@ -1073,25 +1095,37 @@ fn cancelled_power_op_runs_no_compensating_node() { for graceful in [false, true] { for running in [false, true] { let targets = vec![("agent-a".to_owned(), running)]; + // Erased to `DagSpec`: three different recipe types have to + // sit in one array. let cases = [ ( "restart", false, - submit::restart_spec(&targets, graceful, Source::Manual, "bounce".to_owned()), + erase(submit::restart_spec( + &targets, + graceful, + Source::Manual, + "bounce".to_owned(), + )), ), ( "stop", true, - submit::stop_spec(&targets, graceful, Source::Manual, "stop".to_owned()), + erase(submit::stop_spec( + &targets, + graceful, + Source::Manual, + "stop".to_owned(), + )), ), ( "start", true, - submit::start_spec( + erase(submit::start_spec( &[("agent-a".to_owned(), running, false)], Source::Manual, "start".to_owned(), - ), + )), ), ]; for (name, writes_intent, spec) in cases { diff --git a/hive-jobq/src/builder.rs b/hive-jobq/src/builder.rs index 19d34b4f..2a6a07d4 100644 --- a/hive-jobq/src/builder.rs +++ b/hive-jobq/src/builder.rs @@ -6,15 +6,15 @@ //! get wrong. //! //! **An insertion API, not a spec factory.** A builder is only ever handed to a -//! closure by an insertion entry point ([`Graph::insert_job`], -//! [`crate::scheduler::Scheduler::insert_job`]), which inserts the declared -//! nodes and returns the ids the graph minted. It cannot be constructed, held +//! closure by the single insertion entry point +//! ([`crate::scheduler::Scheduler::insert_job`]), which inserts the declared +//! nodes and returns the ids the job asked for. It cannot be constructed, held //! or inserted from outside this crate, and there is no intermediate -//! node-description type to keep in sync with [`Graph::insert`]'s signature — +//! node-description type to keep in sync with [`crate::Graph::insert`]'s signature — //! so a job has no representation that can be passed around instead of being //! inserted. //! -//! **Payload-agnostic.** Generic over the same `N` and `R` as [`Graph`]: the +//! **Payload-agnostic.** Generic over the same `N` and `R` as [`crate::Graph`]: the //! builder knows nothing about what a node *does*, only how nodes relate. //! //! # Declaration order @@ -29,7 +29,7 @@ use std::cell::RefCell; use std::collections::HashMap; -use crate::{Dep, DepWhen, Graph, GraphError, NodeId, TerminalState}; +use crate::{Dep, DepWhen, GraphError, NodeId, TerminalState}; /// An opaque identity for a node **within the job being built**. /// @@ -88,24 +88,48 @@ pub enum BuildError { Graph(#[from] GraphError), } -/// Look each handle a job asked for up in what the insert actually minted, -/// preserving the order it asked in — the last step of both insertion entry -/// points. +/// Reject a job whose own declarations don't hold up — **before anything is +/// inserted**, so these three failures cannot leave a partial job behind. +/// +/// Each is decidable from what the builder already holds: a node may only name +/// handles declared before it, and a job may only ask for handles it declared. +/// Running this first is what lets the insert loop index `ids` directly instead +/// of discovering a bad reference halfway through mutating the graph. /// /// # Errors -/// [`BuildError::UnknownNode`] for a handle this job never issued. -pub(crate) fn resolve_wanted( +/// [`BuildError::ForwardEdge`] / [`BuildError::ForwardParent`] for a reference +/// to a later node, [`BuildError::UnknownNode`] for a requested handle this job +/// never declared. +fn check_declaration_order( + pending: &[Pending], wanted: &[NodeGuid], - ids: &HashMap, -) -> Result, BuildError> { - wanted - .iter() - .map(|g| { - ids.get(g) - .copied() - .ok_or(BuildError::UnknownNode { node: *g }) - }) - .collect() +) -> Result<(), BuildError> { + let mut declared: std::collections::HashSet = std::collections::HashSet::new(); + for node in pending { + for (dep, _) in &node.deps { + if !declared.contains(dep) { + return Err(BuildError::ForwardEdge { + node: node.guid, + dep: *dep, + }); + } + } + if let Some(parent) = node.parent + && !declared.contains(&parent) + { + return Err(BuildError::ForwardParent { + node: node.guid, + parent, + }); + } + declared.insert(node.guid); + } + for guid in wanted { + if !declared.contains(guid) { + return Err(BuildError::UnknownNode { node: *guid }); + } + } + Ok(()) } /// One node as the builder holds it: edges and parent still name *handles*, so @@ -143,8 +167,8 @@ impl JobBuilder { /// A fresh, empty builder. /// /// **Crate-private, and that is the API.** A builder is only ever handed to - /// a closure by an insertion entry point ([`Graph::insert_job`], - /// [`crate::scheduler::Scheduler::insert_job`]), which inserts the declared + /// a closure by the single insertion entry point + /// ([`crate::scheduler::Scheduler::insert_job`]), which inserts the declared /// nodes and returns the ids. Nothing job-shaped is constructible or /// carryable outside this crate — otherwise it is a spec factory again, /// just with a builder's name on it. @@ -195,64 +219,35 @@ impl JobBuilder { /// /// [`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(crate) 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. + /// rejects a node (see [`crate::Graph::insert`]). pub(crate) fn insert_with( self, root_parent: Option, + wanted: &[NodeGuid], mut insert: impl FnMut(N, Vec>, Option) -> Result, - ) -> Result, BuildError> { + ) -> Result, BuildError> { + let pending = self.nodes.into_inner(); + check_declaration_order(&pending, wanted)?; + let mut ids: HashMap = HashMap::new(); - for pending in self.nodes.into_inner() { - let parent = match pending.parent { + for node in pending { + let parent = match node.parent { None => root_parent, - Some(p) => Some(*ids.get(&p).ok_or(BuildError::ForwardParent { - node: pending.guid, - parent: p, - })?), + Some(p) => Some(ids[&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 }); + let mut deps: Vec> = Vec::with_capacity(node.deps.len()); + for (on, when) in node.deps { + deps.push(Dep::Node { id: ids[&on], when }); } deps.extend( - pending - .resources + node.resources .into_iter() .map(|(name, count)| Dep::Resource { name, count }), ); - let id = insert(pending.payload, deps, parent)?; - ids.insert(pending.guid, id); + let id = insert(node.payload, deps, parent)?; + ids.insert(node.guid, id); } - Ok(ids) + Ok(wanted.iter().map(|g| ids[g]).collect()) } /// Apply `f` to the node named by `guid`. @@ -393,22 +388,25 @@ impl NodeRef<'_, N, R> { #[cfg(test)] mod tests { use super::BuildError; + use crate::resources::ResourceTable; + use crate::scheduler::Scheduler; 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() + /// A scheduler over a graph whose payload is a name and whose resources are + /// strings — the only way in, since insertion is a scheduler operation. + fn sched() -> Scheduler<&'static str, &'static str> { + Scheduler::new(Graph::new(), ResourceTable::new()) } - fn deps_of(g: &Graph<&'static str, &'static str>, id: NodeId) -> Vec> { - g.node(id).expect("node present").deps.clone() + fn deps_of(g: &Scheduler<&'static str, &'static str>, id: NodeId) -> Vec> { + g.graph().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 mut g = sched(); let ids = g .insert_job(None, |b| { let first = b.node("a"); @@ -431,7 +429,7 @@ mod tests { #[test] fn parent_resolves_to_a_minted_id() { - let mut g = graph(); + let mut g = sched(); let ids = g .insert_job(None, |b| { let root = b.node("a"); @@ -442,15 +440,15 @@ mod tests { let [root, child] = ids[..] else { panic!("two ids back") }; - assert_eq!(g.node(root).expect("root").parent, None); - assert_eq!(g.node(child).expect("child").parent, Some(root)); + assert_eq!(g.graph().node(root).expect("root").parent, None); + assert_eq!(g.graph().node(child).expect("child").parent, Some(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 mut g = sched(); let ids = g .insert_job(None, |b| { let shared = b.node("a"); @@ -481,7 +479,7 @@ mod tests { /// Resource deps ride along with the node deps, in one insert. #[test] fn resources_become_resource_deps() { - let mut g = graph(); + let mut g = sched(); let ids = g .insert_job(None, |b| { vec![ @@ -516,7 +514,7 @@ mod tests { /// of quietly reordering. #[test] fn a_forward_edge_is_rejected_by_name() { - let mut g = graph(); + let mut g = sched(); let mut named = None; let err = g .insert_job(None, |b| { @@ -541,7 +539,7 @@ mod tests { #[test] fn a_forward_parent_is_rejected_by_name() { - let mut g = graph(); + let mut g = sched(); let mut named = None; let err = g .insert_job(None, |b| { @@ -571,7 +569,7 @@ mod tests { /// would resolve, wrongly, to the second job's own node. #[test] fn a_handle_from_another_job_is_not_silently_resolved() { - let mut g = graph(); + let mut g = sched(); let mut foreign = None; g.insert_job(None, |b| { foreign = Some(b.node("first job").guid()); @@ -596,7 +594,7 @@ mod tests { /// pre-empt it. #[test] fn graph_rejection_surfaces_as_is() { - let mut g = graph(); + let mut g = sched(); let err = g .insert_job(None, |b| { let root = b.node("root"); @@ -614,8 +612,8 @@ mod tests { /// 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 mut g = sched(); + let container = g.append("container", Vec::new(), None).expect("container"); let ids = g .insert_job(Some(container), |b| { @@ -627,8 +625,8 @@ mod tests { let [root, child] = ids[..] else { panic!("two ids back") }; - assert_eq!(g.node(root).expect("root").parent, Some(container)); - assert_eq!(g.node(child).expect("child").parent, Some(root)); + assert_eq!(g.graph().node(root).expect("root").parent, Some(container)); + assert_eq!(g.graph().node(child).expect("child").parent, Some(root)); } /// A handle that names no node in *this* job is refused rather than @@ -636,7 +634,7 @@ mod tests { /// short vector would misalign every id after it. #[test] fn asking_for_a_foreign_handle_is_an_error() { - let mut g = graph(); + let mut g = sched(); let mut foreign = None; g.insert_job(None, |b| { foreign = Some(b.node("first job").guid()); @@ -656,9 +654,9 @@ mod tests { #[test] fn an_empty_builder_inserts_nothing() { - let mut g = graph(); + let mut g = sched(); let ids = g.insert_job(None, |_| Vec::new()).expect("insert"); assert!(ids.is_empty()); - assert_eq!(g.nodes().count(), 0); + assert_eq!(g.graph().nodes().count(), 0); } } diff --git a/hive-jobq/src/lib.rs b/hive-jobq/src/lib.rs index 2fcf0d84..e6f61e54 100644 --- a/hive-jobq/src/lib.rs +++ b/hive-jobq/src/lib.rs @@ -458,35 +458,6 @@ impl Graph { Ok(id) } - /// Insert a whole job under `root_parent`, returning the ids of the nodes - /// `declare` **asked for**, in the order it named them. - /// - /// `declare` receives a fresh [`JobBuilder`], names the job's nodes on it, - /// and returns the handles whose ids it wants back. The builder never - /// leaves this call, so a job cannot be built in one place and inserted in - /// another. The graph-only counterpart of - /// [`crate::scheduler::Scheduler::insert_job`] — prefer that one when a - /// scheduler owns the graph, since it also records what it started. - /// - /// Asking is how a caller addresses a node it created: the alternative — a - /// map of everything inserted, or a positional vector — either hands back a - /// lookup nobody performs or reintroduces the counting this API exists to - /// remove. - /// - /// # Errors - /// Propagates [`BuildError`] — a forward reference in the job's own - /// declarations, a handle from a different job, or a graph rejection. - pub fn insert_job( - &mut self, - root_parent: Option, - declare: impl FnOnce(&JobBuilder) -> Vec, - ) -> Result, BuildError> { - let job = JobBuilder::new(); - let wanted = declare(&job); - let ids = job.insert_into(self, root_parent)?; - crate::builder::resolve_wanted(&wanted, &ids) - } - /// Borrow a node by id. #[must_use] pub fn node(&self, id: NodeId) -> Option<&Node> { diff --git a/hive-jobq/src/scheduler.rs b/hive-jobq/src/scheduler.rs index d326f382..2727f0fa 100644 --- a/hive-jobq/src/scheduler.rs +++ b/hive-jobq/src/scheduler.rs @@ -110,15 +110,19 @@ impl Scheduler { /// insert one itself, so there is no way to end up with a job-shaped value /// being passed around as a spec. /// - /// The scheduler-side counterpart of [`Graph::insert_job`]: same - /// resolution, but each node goes through [`Scheduler::append`], so a - /// caller never has to reach past the scheduler at the graph underneath. - /// Call [`Scheduler::settle`] afterwards to start whatever became - /// runnable. + /// The one insertion entry point: every node goes through + /// [`Scheduler::append`], so a caller never has to reach past the scheduler + /// at the graph underneath. Call [`Scheduler::settle`] afterwards to start + /// whatever became runnable. + /// + /// **Atomic in the job's own shape.** A forward edge, a forward parent, or + /// a request for a handle this job never declared is rejected *before* the + /// first node is inserted, so a malformed job leaves the graph untouched + /// rather than half-built. /// /// # Errors /// Propagates [`BuildError`] — a forward reference in the job's own - /// declarations, or a graph rejection. + /// declarations, a handle from a different job, or a graph rejection. pub fn insert_job( &mut self, root_parent: Option, @@ -126,10 +130,9 @@ impl Scheduler { ) -> Result, BuildError> { let job = JobBuilder::new(); let wanted = declare(&job); - let ids = job.insert_with(root_parent, |payload, deps, parent| { + job.insert_with(root_parent, &wanted, |payload, deps, parent| { self.append(payload, deps, parent) - })?; - crate::builder::resolve_wanted(&wanted, &ids) + }) } /// Claim every currently-runnable pending node and start it: node-deps