diff --git a/Cargo.lock b/Cargo.lock index a1ab379c..7a6c2e2b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1116,6 +1116,27 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "enumflags2" +version = "0.7.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1027f7680c853e056ebcec683615fb6fbbc07dbaa13b4d5d9442b146ded4ecef" +dependencies = [ + "enumflags2_derive", + "serde", +] + +[[package]] +name = "enumflags2_derive" +version = "0.7.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67c78a4d8fdf9953a5c9d458f9efe940fd97a0cab0941c075a813ac594733827" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "equivalent" version = "1.0.2" @@ -1705,6 +1726,7 @@ name = "hive-jobq" version = "0.1.0" dependencies = [ "chrono", + "enumflags2", "serde", "serde_json", "thiserror 2.0.18", diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 51de0607..b69e29a2 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -11,7 +11,9 @@ use std::sync::Arc; use anyhow::{Context as _, Result}; use super::Claim; -use super::model::{NodeKind, NodeSpec, TerminalState}; +use hive_jobq::TerminalState; + +use super::model::{NodeKind, NodeSpec}; use crate::coordinator::Coordinator; use crate::power::{ReconcileAction, reconcile_action}; diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 9bb9335a..f03ffe45 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -18,10 +18,9 @@ //! by a subtree root and borrowed by its descendants (continuity); //! - per-DAG terminal work is an ordinary **tail node** //! ([`NodeKind::ResolveApproval`] / [`NodeKind::EmitRebuilt`]) that the builder -//! appends in [`templates`], weak-edged (`AfterAny`) onto the DAG's other group -//! roots so it runs on success, failure and cancel alike. It reads how the work -//! went off its own [`Claim::deps`] — no inline hook fired from outside the -//! graph, no drained event stream. +//! appends in [`templates`], edged onto the DAG's other group roots by the +//! outcome it reports. Templates emit one tail per outcome and the graph runs +//! exactly one, so nothing branches at runtime. //! //! 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 +48,8 @@ use hive_sh4re::wire_time::now_unix; use tokio::sync::Notify; use crate::coordinator::TransientKind; -pub use model::{DagSpec, DagView, NodeKind, NodeSpec, PermPayload, Source, State, TerminalState}; +pub use hive_jobq::TerminalState; +pub use model::{DagSpec, DagView, NodeKind, NodeSpec, PermPayload, Source, State}; use resource::Resource; /// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) the snapshot diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index d6f75b3c..dcddd2f4 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -17,11 +17,7 @@ use serde::Serialize; use crate::coordinator::TransientKind; -/// When a dependency edge is satisfied — re-exported from [`hive_jobq`] rather -/// than mirrored here. It used to be a duplicate enum with a `to_crate_when` -/// translation beside it; the copy bought nothing and had to be widened in -/// lockstep every time the crate's edge model grew. -pub use hive_jobq::{DepWhen, TerminalState}; +use hive_jobq::{DepWhen, TerminalState}; /// A dependency edge (intra-DAG only — cross-DAG ordering comes from /// the per-agent lease + dedup, never from edges between DAGs). @@ -248,14 +244,14 @@ pub enum NodeKind { /// they `Failed`, and **nothing at all** when they `Cancelled` (a cancelled DAG /// never ran, so there is no rebuild to report). /// - /// One node **per agent**, unlike the DAG-wide hook it replaces: a multi-agent - /// DAG now reports each agent's own outcome instead of painting every agent with - /// the whole DAG's roll-up. And one per *outcome* — `ok` isn't computed here, - /// it's which of the pair the graph let run. + /// One node per **agent** — a multi-agent DAG reports each agent's own + /// outcome rather than painting all of them with the whole DAG's roll-up — + /// and one per **outcome**: `ok` isn't computed here, it's which of the pair + /// the graph let run. /// - /// There is deliberately no cancel variant: a DAG dropped before it started - /// has no rebuild to report, and neither tail's edge accepts `Cancelled`, so - /// both are cancelled with the rest and nothing is emitted. + /// No cancel variant, deliberately: a DAG dropped before it started has no + /// rebuild to report, and neither tail's edge accepts `Cancelled`, so both + /// are cancelled with the rest and nothing is emitted. EmitRebuilt { agent: String, ok: bool }, /// Write the agent's durable power intent (`wanted = Up` when `up`, else /// `Offline`) as a first-class DAG node, at the head of a power-op diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index f893e9a5..08608491 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -30,7 +30,9 @@ use anyhow::{Result, bail}; -use super::model::{DagSpec, Dep, DepWhen, NodeKind, NodeSpec, PermPayload, Source, TerminalState}; +use hive_jobq::{DepWhen, TerminalState}; + +use super::model::{DagSpec, Dep, NodeKind, NodeSpec, PermPayload, Source}; use crate::coordinator::TransientKind; /// After-ok edge on the previous node — the common chain link. Shared with diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index a7c45229..0088e837 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -6,7 +6,9 @@ //! scheduler's async loop is a thin claim/complete pump over the same //! methods exercised here. -use super::model::{Dep, DepWhen, NodeKind, NodeSpec}; +use hive_jobq::DepWhen; + +use super::model::{Dep, NodeKind, NodeSpec}; use super::*; fn submit(q: &JobQueue, spec: DagSpec) -> u64 { diff --git a/hive-jobq/Cargo.toml b/hive-jobq/Cargo.toml index aa5370f7..e791f702 100644 --- a/hive-jobq/Cargo.toml +++ b/hive-jobq/Cargo.toml @@ -9,6 +9,7 @@ workspace = true [dependencies] chrono = { workspace = true } +enumflags2 = { version = "0.7.12", features = ["serde"] } serde = { workspace = true } thiserror = { workspace = true } diff --git a/hive-jobq/src/lib.rs b/hive-jobq/src/lib.rs index 83a746da..60e5b6cc 100644 --- a/hive-jobq/src/lib.rs +++ b/hive-jobq/src/lib.rs @@ -59,6 +59,8 @@ impl NodeId { /// How a node finished. The terminal subset of [`State`], as its own type so an /// edge condition cannot name `Pending` / `Running` / `Finishing` — those are /// meaningless in a dependency and are better unrepresentable than rejected. +#[enumflags2::bitflags] +#[repr(u8)] #[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "snake_case")] pub enum TerminalState { @@ -80,18 +82,6 @@ pub enum TerminalState { Skipped, } -impl TerminalState { - /// Bit for this outcome in a [`DepWhen`] set. - const fn bit(self) -> u8 { - match self { - TerminalState::Done => 1, - TerminalState::Failed => 1 << 1, - TerminalState::Cancelled => 1 << 2, - TerminalState::Skipped => 1 << 3, - } - } -} - impl State { /// This state as a [`TerminalState`], or `None` while the node is still /// in flight. @@ -122,15 +112,31 @@ impl State { /// /// [`AFTER_OK`]: DepWhen::AFTER_OK /// [`AFTER_ANY`]: DepWhen::AFTER_ANY -#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -pub struct DepWhen(u8); +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct DepWhen(enumflags2::BitFlags); + +/// Serialised as the list of outcomes it accepts (`["done","failed"]`) rather +/// than the underlying bitmask, so the wire form stays readable and survives the +/// bits being renumbered. +impl serde::Serialize for DepWhen { + fn serialize(&self, ser: S) -> Result { + serde::Serialize::serialize(&self.0.iter().collect::>(), ser) + } +} + +impl<'de> serde::Deserialize<'de> for DepWhen { + fn deserialize>(de: D) -> Result { + let outcomes = Vec::::deserialize(de)?; + Ok(Self(outcomes.into_iter().collect())) + } +} impl DepWhen { /// The dependency must reach [`TerminalState::Done`]. The default chain /// edge: if the dependency fails, the dependent must not run and is /// cancelled down the chain — e.g. a failed `Prebuild` must not let /// `StopForUpdate` stop a healthy container. - pub const AFTER_OK: Self = Self(TerminalState::Done.bit()); + pub const AFTER_OK: Self = Self(enumflags2::make_bitflags!(TerminalState::{Done})); /// Anything **except the work being dropped** — `Done`, `Failed` or /// `Skipped`. For steps that must converge regardless of how the run went, /// e.g. `Reconcile` bringing a container back up even when the preceding @@ -140,32 +146,25 @@ impl DepWhen { /// started at all there is nothing to converge, and running the recovery /// step anyway would act on work that provably never happened. A node that /// must report a cancellation names `Cancelled` explicitly. - pub const AFTER_ANY: Self = Self( - TerminalState::Done.bit() | TerminalState::Failed.bit() | TerminalState::Skipped.bit(), - ); + pub const AFTER_ANY: Self = + Self(enumflags2::make_bitflags!(TerminalState::{Done | Failed | Skipped})); /// An edge satisfied by exactly the listed outcomes. #[must_use] - pub const fn of(outcomes: &[TerminalState]) -> Self { - let mut bits = 0u8; - let mut i = 0; - while i < outcomes.len() { - bits |= outcomes[i].bit(); - i += 1; - } - Self(bits) + pub fn of(outcomes: &[TerminalState]) -> Self { + Self(outcomes.iter().copied().collect()) } /// Whether `outcome` satisfies this edge. #[must_use] - pub const fn accepts(self, outcome: TerminalState) -> bool { - self.0 & outcome.bit() != 0 + pub fn accepts(self, outcome: TerminalState) -> bool { + self.0.contains(outcome) } /// An edge no outcome can satisfy — rejected at [`Graph::validate`]. #[must_use] - pub const fn is_empty(self) -> bool { - self.0 == 0 + pub fn is_empty(self) -> bool { + self.0.is_empty() } /// Whether a dependency in `dep_state` satisfies this edge. A non-terminal