refactor(#2772): enumflags2 for the edge set; drop re-export shims
Review follow-ups on #2785. `DepWhen` wraps `BitFlags<TerminalState>` instead of a hand-rolled `u8`, so the bit manipulation belongs to the library and `TerminalState` gains its flag value from `#[bitflags]` rather than a `bit()` match anyone could get wrong. `of`/`accepts`/`is_empty` become one-liners over it. Serialization is written out by hand rather than derived: clippy's `unsafe_derive_deserialize` fires on deriving over a type with unsafe internals, and the honest fix is to say what the wire form is. It is now the list of accepted outcomes — `["done","failed"]` — which reads better than a bitmask and survives the bits being renumbered. Also drops the `pub use` re-export of `DepWhen` / `TerminalState` from hive-c0re's `model`. It existed so that `use super::model::…` kept compiling, which is a shim for one consumer's convenience; the sites import from `hive_jobq` directly now. And removes comments narrating what the code used to be. Git holds that.
This commit is contained in:
parent
07078b76ef
commit
940c928fee
8 changed files with 74 additions and 50 deletions
22
Cargo.lock
generated
22
Cargo.lock
generated
|
|
@ -1116,6 +1116,27 @@ dependencies = [
|
||||||
"cfg-if",
|
"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]]
|
[[package]]
|
||||||
name = "equivalent"
|
name = "equivalent"
|
||||||
version = "1.0.2"
|
version = "1.0.2"
|
||||||
|
|
@ -1705,6 +1726,7 @@ name = "hive-jobq"
|
||||||
version = "0.1.0"
|
version = "0.1.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"chrono",
|
"chrono",
|
||||||
|
"enumflags2",
|
||||||
"serde",
|
"serde",
|
||||||
"serde_json",
|
"serde_json",
|
||||||
"thiserror 2.0.18",
|
"thiserror 2.0.18",
|
||||||
|
|
|
||||||
|
|
@ -11,7 +11,9 @@ use std::sync::Arc;
|
||||||
use anyhow::{Context as _, Result};
|
use anyhow::{Context as _, Result};
|
||||||
|
|
||||||
use super::Claim;
|
use super::Claim;
|
||||||
use super::model::{NodeKind, NodeSpec, TerminalState};
|
use hive_jobq::TerminalState;
|
||||||
|
|
||||||
|
use super::model::{NodeKind, NodeSpec};
|
||||||
use crate::coordinator::Coordinator;
|
use crate::coordinator::Coordinator;
|
||||||
use crate::power::{ReconcileAction, reconcile_action};
|
use crate::power::{ReconcileAction, reconcile_action};
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -18,10 +18,9 @@
|
||||||
//! by a subtree root and borrowed by its descendants (continuity);
|
//! by a subtree root and borrowed by its descendants (continuity);
|
||||||
//! - per-DAG terminal work is an ordinary **tail node**
|
//! - per-DAG terminal work is an ordinary **tail node**
|
||||||
//! ([`NodeKind::ResolveApproval`] / [`NodeKind::EmitRebuilt`]) that the builder
|
//! ([`NodeKind::ResolveApproval`] / [`NodeKind::EmitRebuilt`]) that the builder
|
||||||
//! appends in [`templates`], weak-edged (`AfterAny`) onto the DAG's other group
|
//! appends in [`templates`], edged onto the DAG's other group roots by the
|
||||||
//! roots so it runs on success, failure and cancel alike. It reads how the work
|
//! outcome it reports. Templates emit one tail per outcome and the graph runs
|
||||||
//! went off its own [`Claim::deps`] — no inline hook fired from outside the
|
//! exactly one, so nothing branches at runtime.
|
||||||
//! graph, no drained event stream.
|
|
||||||
//!
|
//!
|
||||||
//! The queue is runtime-only (no persistence): an empty graph on boot; desired
|
//! 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
|
//! 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 tokio::sync::Notify;
|
||||||
|
|
||||||
use crate::coordinator::TransientKind;
|
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;
|
use resource::Resource;
|
||||||
|
|
||||||
/// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) the snapshot
|
/// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) the snapshot
|
||||||
|
|
|
||||||
|
|
@ -17,11 +17,7 @@ use serde::Serialize;
|
||||||
|
|
||||||
use crate::coordinator::TransientKind;
|
use crate::coordinator::TransientKind;
|
||||||
|
|
||||||
/// When a dependency edge is satisfied — re-exported from [`hive_jobq`] rather
|
use hive_jobq::{DepWhen, TerminalState};
|
||||||
/// 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};
|
|
||||||
|
|
||||||
/// A dependency edge (intra-DAG only — cross-DAG ordering comes from
|
/// A dependency edge (intra-DAG only — cross-DAG ordering comes from
|
||||||
/// the per-agent lease + dedup, never from edges between DAGs).
|
/// 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
|
/// they `Failed`, and **nothing at all** when they `Cancelled` (a cancelled DAG
|
||||||
/// never ran, so there is no rebuild to report).
|
/// never ran, so there is no rebuild to report).
|
||||||
///
|
///
|
||||||
/// One node **per agent**, unlike the DAG-wide hook it replaces: a multi-agent
|
/// One node per **agent** — a multi-agent DAG reports each agent's own
|
||||||
/// DAG now reports each agent's own outcome instead of painting every agent with
|
/// outcome rather than painting all of them with the whole DAG's roll-up —
|
||||||
/// the whole DAG's roll-up. And one per *outcome* — `ok` isn't computed here,
|
/// and one per **outcome**: `ok` isn't computed here, it's which of the pair
|
||||||
/// it's which of the pair the graph let run.
|
/// the graph let run.
|
||||||
///
|
///
|
||||||
/// There is deliberately no cancel variant: a DAG dropped before it started
|
/// No cancel variant, deliberately: a DAG dropped before it started has no
|
||||||
/// has no rebuild to report, and neither tail's edge accepts `Cancelled`, so
|
/// rebuild to report, and neither tail's edge accepts `Cancelled`, so both
|
||||||
/// both are cancelled with the rest and nothing is emitted.
|
/// are cancelled with the rest and nothing is emitted.
|
||||||
EmitRebuilt { agent: String, ok: bool },
|
EmitRebuilt { agent: String, ok: bool },
|
||||||
/// Write the agent's durable power intent (`wanted = Up` when `up`, else
|
/// 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
|
/// `Offline`) as a first-class DAG node, at the head of a power-op
|
||||||
|
|
|
||||||
|
|
@ -30,7 +30,9 @@
|
||||||
|
|
||||||
use anyhow::{Result, bail};
|
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;
|
use crate::coordinator::TransientKind;
|
||||||
|
|
||||||
/// After-ok edge on the previous node — the common chain link. Shared with
|
/// After-ok edge on the previous node — the common chain link. Shared with
|
||||||
|
|
|
||||||
|
|
@ -6,7 +6,9 @@
|
||||||
//! scheduler's async loop is a thin claim/complete pump over the same
|
//! scheduler's async loop is a thin claim/complete pump over the same
|
||||||
//! methods exercised here.
|
//! methods exercised here.
|
||||||
|
|
||||||
use super::model::{Dep, DepWhen, NodeKind, NodeSpec};
|
use hive_jobq::DepWhen;
|
||||||
|
|
||||||
|
use super::model::{Dep, NodeKind, NodeSpec};
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|
||||||
fn submit(q: &JobQueue, spec: DagSpec) -> u64 {
|
fn submit(q: &JobQueue, spec: DagSpec) -> u64 {
|
||||||
|
|
|
||||||
|
|
@ -9,6 +9,7 @@ workspace = true
|
||||||
|
|
||||||
[dependencies]
|
[dependencies]
|
||||||
chrono = { workspace = true }
|
chrono = { workspace = true }
|
||||||
|
enumflags2 = { version = "0.7.12", features = ["serde"] }
|
||||||
serde = { workspace = true }
|
serde = { workspace = true }
|
||||||
thiserror = { workspace = true }
|
thiserror = { workspace = true }
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -59,6 +59,8 @@ impl NodeId {
|
||||||
/// How a node finished. The terminal subset of [`State`], as its own type so an
|
/// How a node finished. The terminal subset of [`State`], as its own type so an
|
||||||
/// edge condition cannot name `Pending` / `Running` / `Finishing` — those are
|
/// edge condition cannot name `Pending` / `Running` / `Finishing` — those are
|
||||||
/// meaningless in a dependency and are better unrepresentable than rejected.
|
/// meaningless in a dependency and are better unrepresentable than rejected.
|
||||||
|
#[enumflags2::bitflags]
|
||||||
|
#[repr(u8)]
|
||||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
|
||||||
#[serde(rename_all = "snake_case")]
|
#[serde(rename_all = "snake_case")]
|
||||||
pub enum TerminalState {
|
pub enum TerminalState {
|
||||||
|
|
@ -80,18 +82,6 @@ pub enum TerminalState {
|
||||||
Skipped,
|
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 {
|
impl State {
|
||||||
/// This state as a [`TerminalState`], or `None` while the node is still
|
/// This state as a [`TerminalState`], or `None` while the node is still
|
||||||
/// in flight.
|
/// in flight.
|
||||||
|
|
@ -122,15 +112,31 @@ impl State {
|
||||||
///
|
///
|
||||||
/// [`AFTER_OK`]: DepWhen::AFTER_OK
|
/// [`AFTER_OK`]: DepWhen::AFTER_OK
|
||||||
/// [`AFTER_ANY`]: DepWhen::AFTER_ANY
|
/// [`AFTER_ANY`]: DepWhen::AFTER_ANY
|
||||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||||
pub struct DepWhen(u8);
|
pub struct DepWhen(enumflags2::BitFlags<TerminalState>);
|
||||||
|
|
||||||
|
/// 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<S: serde::Serializer>(&self, ser: S) -> Result<S::Ok, S::Error> {
|
||||||
|
serde::Serialize::serialize(&self.0.iter().collect::<Vec<_>>(), ser)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<'de> serde::Deserialize<'de> for DepWhen {
|
||||||
|
fn deserialize<D: serde::Deserializer<'de>>(de: D) -> Result<Self, D::Error> {
|
||||||
|
let outcomes = Vec::<TerminalState>::deserialize(de)?;
|
||||||
|
Ok(Self(outcomes.into_iter().collect()))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
impl DepWhen {
|
impl DepWhen {
|
||||||
/// The dependency must reach [`TerminalState::Done`]. The default chain
|
/// The dependency must reach [`TerminalState::Done`]. The default chain
|
||||||
/// edge: if the dependency fails, the dependent must not run and is
|
/// edge: if the dependency fails, the dependent must not run and is
|
||||||
/// cancelled down the chain — e.g. a failed `Prebuild` must not let
|
/// cancelled down the chain — e.g. a failed `Prebuild` must not let
|
||||||
/// `StopForUpdate` stop a healthy container.
|
/// `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
|
/// Anything **except the work being dropped** — `Done`, `Failed` or
|
||||||
/// `Skipped`. For steps that must converge regardless of how the run went,
|
/// `Skipped`. For steps that must converge regardless of how the run went,
|
||||||
/// e.g. `Reconcile` bringing a container back up even when the preceding
|
/// 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
|
/// 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
|
/// step anyway would act on work that provably never happened. A node that
|
||||||
/// must report a cancellation names `Cancelled` explicitly.
|
/// must report a cancellation names `Cancelled` explicitly.
|
||||||
pub const AFTER_ANY: Self = Self(
|
pub const AFTER_ANY: Self =
|
||||||
TerminalState::Done.bit() | TerminalState::Failed.bit() | TerminalState::Skipped.bit(),
|
Self(enumflags2::make_bitflags!(TerminalState::{Done | Failed | Skipped}));
|
||||||
);
|
|
||||||
|
|
||||||
/// An edge satisfied by exactly the listed outcomes.
|
/// An edge satisfied by exactly the listed outcomes.
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub const fn of(outcomes: &[TerminalState]) -> Self {
|
pub fn of(outcomes: &[TerminalState]) -> Self {
|
||||||
let mut bits = 0u8;
|
Self(outcomes.iter().copied().collect())
|
||||||
let mut i = 0;
|
|
||||||
while i < outcomes.len() {
|
|
||||||
bits |= outcomes[i].bit();
|
|
||||||
i += 1;
|
|
||||||
}
|
|
||||||
Self(bits)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Whether `outcome` satisfies this edge.
|
/// Whether `outcome` satisfies this edge.
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub const fn accepts(self, outcome: TerminalState) -> bool {
|
pub fn accepts(self, outcome: TerminalState) -> bool {
|
||||||
self.0 & outcome.bit() != 0
|
self.0.contains(outcome)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// An edge no outcome can satisfy — rejected at [`Graph::validate`].
|
/// An edge no outcome can satisfy — rejected at [`Graph::validate`].
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub const fn is_empty(self) -> bool {
|
pub fn is_empty(self) -> bool {
|
||||||
self.0 == 0
|
self.0.is_empty()
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Whether a dependency in `dep_state` satisfies this edge. A non-terminal
|
/// Whether a dependency in `dep_state` satisfies this edge. A non-terminal
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue