//! `hive-jobq` — a persistent job-DAG scheduler, extracted from hive-c0re's //! in-tree `job_queue` as a domain-agnostic library. //! //! # Model (v2) //! //! One **persistent graph** for the whole system, not a DAG per job. Enqueuing //! inserts a self-contained **node group** and returns its id; the scheduler //! runs a continuous loop, starting every node whose [`Dep`]s are satisfied: //! //! - **Resource** deps are named counting semaphores ([`ResourceName`]): //! `build-slot` (capacity N), `agent/` (capacity 1), or any name //! (capacity 1, created on use). A node acquires *all* its resource deps //! atomically at start (all-or-nothing) — no hold-and-wait, so no deadlock //! and no cycle detection needed. //! - **Node** deps wait on a node/group per [`DepWhen`]: `AfterOk` needs //! success (a failed dep cancels the dependent), `AfterAny` only terminal. //! //! A **node group** is a self-contained sub-graph; things depend on it as a //! whole (done = every inner node terminal), never on an inner node. Groups //! nest; a running node may grow its own group but not reach outside it. //! //! A [`NodeId`] is opaque, stable, and monotonic — persisted, so it survives //! restarts. Group membership is a parent edge ([`Node::parent`]), *not* in the //! id; the `1/1/2` hierarchy is a derived UI label. The node payload is generic //! (`N`) so the library stays container-agnostic — the caller supplies its own //! node kind. Resources are held by the acquiring node and released on //! completion via guard objects, recursive within a group. //! //! The resource-acquisition machinery, the guards, and the scheduler loop are //! follow-ups; this is the data model they build on. /// Opaque, stable, monotonic node identifier. /// /// Assigned by the [`Graph`] on insert and persisted, so it is stable across /// restarts. Group membership is a separate parent edge ([`Node::parent`]) — it /// is deliberately *not* encoded in the id, so the id never changes as the tree /// grows or collapses. The hierarchical `1/1/2` path used in the UI is derived /// from the parent tree at render time. #[derive( Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize, )] pub struct NodeId(pub u64); /// A named counting semaphore. /// /// Examples: `build-slot` (capacity configured to the number of build slots), /// `agent/` (capacity 1 — the per-agent lifecycle lock), or any other /// name, which is assumed to have capacity 1 and is created on first use. #[derive( Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize, )] pub struct ResourceName(pub String); /// When a [`Dep::Node`] edge is satisfied — the strong/weak distinction the /// current queue carries as `DepWhen`, load-bearing for failure safety. #[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub enum DepWhen { /// The dependency must reach [`State::Done`]. This is the default chain /// edge: if the dependency *fails*, the dependent must not run and is /// cancelled ([`State::Cancelled`]) down the chain — e.g. a failed /// `Prebuild` must not let `StopForUpdate` stop a healthy container. AfterOk, /// The dependency need only be terminal — success or failure both satisfy /// it. For steps that must converge regardless, e.g. `Reconcile` running /// even when the preceding `Swap` failed. AfterAny, } impl DepWhen { /// Whether a dependency in `dep_state` satisfies this edge. #[must_use] pub fn satisfied_by(self, dep_state: State) -> bool { match self { DepWhen::AfterOk => dep_state == State::Done, DepWhen::AfterAny => dep_state.is_terminal(), } } } /// One dependency of a node. A node becomes runnable once every [`Dep::Node`] /// edge it names is satisfied (per its [`DepWhen`]) *and* every [`Dep::Resource`] /// it names can be acquired (all of them, atomically). #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub enum Dep { /// Depend on another node (or a group, by its group node's id). Whether a /// *failed* dependency satisfies the edge is decided by `when`: `AfterOk` /// requires success (and cancels this node if the dep fails), `AfterAny` /// only requires the dep to be terminal. Node { /// The node (or group) depended on. id: NodeId, /// Strong (`AfterOk`) vs weak (`AfterAny`). when: DepWhen, }, /// Hold `count` units of a named resource for the duration of this node's /// run. Acquired atomically with the node's other resource deps at start, /// released when the node completes. Resource { /// The resource to acquire. name: ResourceName, /// How many units to hold (usually 1). count: u32, }, } /// A node's lifecycle state. #[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub enum State { /// Waiting on dependencies (node or resource). Pending, /// Dependencies satisfied, resources held, currently executing. Running, /// Completed successfully. Done, /// Completed unsuccessfully. Failed, /// Never ran: an `AfterOk` dependency failed, so this node (and the rest of /// its strong-dependent chain) is cancelled rather than run. Cancelled, } impl State { /// A node is *terminal* once it has finished — successfully, unsuccessfully, /// or cancelled — which is when its resources are released and dependents /// are re-evaluated. #[must_use] pub fn is_terminal(self) -> bool { matches!(self, State::Done | State::Failed | State::Cancelled) } } /// A node in the graph, carrying a caller-defined payload `N`. /// /// The library schedules over `Node`s and resources without interpreting the /// payload; the caller supplies `N` (its own node kind) and a runner to execute /// a claimed node. #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct Node { /// Stable identity, assigned on insert. pub id: NodeId, /// The group this node belongs to, if any. `None` for a top-level group /// node. Group membership lives here, not in the id. pub parent: Option, /// Caller-defined payload (the node's kind / work description). pub payload: N, /// What must hold before this node runs (other nodes + resources). pub deps: Vec, /// Lifecycle state. pub state: State, } /// The single persistent graph of all nodes. /// /// New jobs are inserted as node groups; the scheduler (added in a follow-up) /// walks this graph filling open slots. Completed groups are retained (no /// pruning in v1). #[derive(Debug, serde::Serialize, serde::Deserialize)] pub struct Graph { nodes: Vec>, next_id: u64, } // A `derive(Default)` would wrongly require `N: Default` (an empty graph holds // no payload); an empty `Vec>` needs no such bound, so impl it directly. impl Default for Graph { fn default() -> Self { Self::new() } } impl Graph { /// An empty graph. #[must_use] pub fn new() -> Self { Self { nodes: Vec::new(), next_id: 0, } } /// Mint the next stable node id. fn mint_id(&mut self) -> NodeId { let id = NodeId(self.next_id); self.next_id += 1; id } /// Insert a node with the given payload, deps, and parent group, returning /// its freshly-minted id. The node starts [`State::Pending`]. pub fn insert(&mut self, payload: N, deps: Vec, parent: Option) -> NodeId { let id = self.mint_id(); self.nodes.push(Node { id, parent, payload, deps, state: State::Pending, }); id } /// Borrow a node by id. #[must_use] pub fn node(&self, id: NodeId) -> Option<&Node> { self.nodes.iter().find(|n| n.id == id) } /// The direct children of a group node (nodes whose `parent` is `id`). pub fn children(&self, id: NodeId) -> impl Iterator> { self.nodes.iter().filter(move |n| n.parent == Some(id)) } /// A group is terminal once the group node itself is terminal *and* every /// node inside it (recursively) is terminal. The node's own state matters: /// a group node still `Pending`/`Running` is not terminal even with no /// children yet, since a running node may still append some. (Deciding when /// to *settle* a group node to terminal once its children are all done is a /// separate concern the scheduler owns.) An unknown id is not terminal. #[must_use] pub fn group_terminal(&self, id: NodeId) -> bool { let Some(node) = self.node(id) else { return false; }; node.state.is_terminal() && self.children(id).all(|child| self.group_terminal(child.id)) } } #[cfg(test)] mod tests { use super::*; #[test] fn insert_mints_stable_monotonic_ids() { let mut g: Graph<&str> = Graph::new(); let a = g.insert("sweep", vec![], None); let b = g.insert( "update", vec![Dep::Node { id: a, when: DepWhen::AfterOk, }], Some(a), ); assert_eq!(a, NodeId(0)); assert_eq!(b, NodeId(1)); // Membership is the parent edge, not the id. assert_eq!(g.node(b).unwrap().parent, Some(a)); assert_eq!(g.node(a).unwrap().parent, None); } fn set_state(g: &mut Graph, id: NodeId, state: State) { let idx = g.nodes.iter().position(|n| n.id == id).unwrap(); g.nodes[idx].state = state; } #[test] fn group_terminal_requires_the_group_node_and_all_children_terminal() { let mut g: Graph<&str> = Graph::new(); let group = g.insert("group", vec![], None); let child = g.insert("child", vec![], Some(group)); // Both pending → not terminal. assert!(!g.group_terminal(group)); // Child done, but the group node itself is still pending → NOT terminal: // the group node's own state is load-bearing, not just its children. set_state(&mut g, child, State::Done); assert!(!g.group_terminal(group)); // Group node terminal too → the whole group is terminal. set_state(&mut g, group, State::Done); assert!(g.group_terminal(group)); } #[test] fn empty_running_group_is_not_terminal() { // A running node with no children yet may still append some, so it must // not read as terminal just because its child set is currently empty. let mut g: Graph<&str> = Graph::new(); let group = g.insert("group", vec![], None); set_state(&mut g, group, State::Running); assert!(!g.group_terminal(group)); // Once it finishes (having grown no children), it is terminal. set_state(&mut g, group, State::Done); assert!(g.group_terminal(group)); } #[test] fn state_terminality() { assert!(State::Done.is_terminal()); assert!(State::Failed.is_terminal()); assert!(State::Cancelled.is_terminal()); assert!(!State::Pending.is_terminal()); assert!(!State::Running.is_terminal()); } #[test] fn after_ok_needs_success_after_any_needs_terminal() { // AfterOk: only Done satisfies; a Failed/Cancelled dep does NOT (the // dependent must be cancelled, not run). assert!(DepWhen::AfterOk.satisfied_by(State::Done)); assert!(!DepWhen::AfterOk.satisfied_by(State::Failed)); assert!(!DepWhen::AfterOk.satisfied_by(State::Cancelled)); assert!(!DepWhen::AfterOk.satisfied_by(State::Running)); // AfterAny: any terminal state satisfies. assert!(DepWhen::AfterAny.satisfied_by(State::Done)); assert!(DepWhen::AfterAny.satisfied_by(State::Failed)); assert!(DepWhen::AfterAny.satisfied_by(State::Cancelled)); assert!(!DepWhen::AfterAny.satisfied_by(State::Pending)); } }