diff --git a/Cargo.lock b/Cargo.lock index 7a6c2e2b..a1ab379c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1116,27 +1116,6 @@ 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" @@ -1726,7 +1705,6 @@ name = "hive-jobq" version = "0.1.0" dependencies = [ "chrono", - "enumflags2", "serde", "serde_json", "thiserror 2.0.18", diff --git a/Cargo.toml b/Cargo.toml index 12ea54ef..9cd6ab8c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -51,7 +51,6 @@ chrono = { version = "0.4", default-features = false, features = [ ] } clap = { version = "4", features = ["derive"] } clap_complete = "4" -enumflags2 = { version = "0.7.12", features = ["serde"] } indicatif = "0.18" hive-sh4re = { path = "hive-sh4re" } hive-agent-sock = { path = "hive-agent-sock" } diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index 4dadeace..d00d9227 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -507,18 +507,20 @@ async fn run_approval_schedule_prompt( /// so the work's terminal state is the authoritative outcome and there's no /// in-node resolution to skip around. /// -/// `outcome` is the one the calling node was built to report — a template emits -/// one tail per outcome, so success, failure and cancel each arrive here from -/// their own node rather than from one node branching. +/// `state` / `error` come from the tail node's own dependency roll-up +/// ([`Claim::deps_state`] / [`Claim::deps_error`]), so this runs on the success, +/// failure **and cancel** paths alike. /// /// [`NodeKind::ResolveApproval`]: crate::job_queue::NodeKind::ResolveApproval +/// [`Claim::deps_state`]: crate::job_queue::Claim::deps_state +/// [`Claim::deps_error`]: crate::job_queue::Claim::deps_error pub(crate) async fn resolve_approval_dag( coord: &Arc, approval_id: i64, - outcome: crate::job_queue::TerminalState, + state: crate::job_queue::State, error: Option<&str>, ) { - use crate::job_queue::TerminalState; + use crate::job_queue::State; let approval = match coord.approvals.get(approval_id) { Ok(Some(a)) => a, Ok(None) => { @@ -530,14 +532,10 @@ pub(crate) async fn resolve_approval_dag( return; } }; - let result: Result<()> = match outcome { - TerminalState::Done => Ok(()), - // `Skipped` never reaches here — no template emits a tail for it — but it - // reads the same to an operator either way: the work did not happen. - TerminalState::Cancelled | TerminalState::Skipped => { - Err(anyhow::anyhow!("cancelled before completion")) - } - TerminalState::Failed => Err(anyhow::anyhow!("{}", error.unwrap_or("job dag failed"))), + let result: Result<()> = match state { + State::Done => Ok(()), + State::Cancelled => Err(anyhow::anyhow!("cancelled before completion")), + _ => Err(anyhow::anyhow!("{}", error.unwrap_or("job dag failed"))), }; let mut terminal_tag = None; match approval.kind { @@ -553,7 +551,7 @@ pub(crate) async fn resolve_approval_dag( } } ApprovalKind::MergeConfigPr => { - terminal_tag = deploy_terminal_tag(approval.agent.as_str(), approval_id, outcome).await; + terminal_tag = deploy_terminal_tag(approval.agent.as_str(), approval_id, state).await; // On a failed deploy, surface the failing build log back onto the // PR so the manager sees why it was rejected without leaving the // forge. Posted here rather than inside a node because this is the @@ -579,14 +577,13 @@ pub(crate) async fn resolve_approval_dag( async fn deploy_terminal_tag( agent: &str, approval_id: i64, - outcome: crate::job_queue::TerminalState, + state: crate::job_queue::State, ) -> Option { - use crate::job_queue::TerminalState; - let candidate = match outcome { - TerminalState::Done => format!("deployed/{approval_id}"), - // Nothing ran, so nothing was planted. - TerminalState::Cancelled | TerminalState::Skipped => return None, - TerminalState::Failed => format!("failed/{approval_id}"), + use crate::job_queue::State; + let candidate = match state { + State::Done => format!("deployed/{approval_id}"), + State::Cancelled => return None, + _ => format!("failed/{approval_id}"), }; lifecycle::git_rev_parse(&crate::paths::applied_dir(agent), &candidate) .await diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index b69e29a2..21ea0728 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -11,9 +11,7 @@ use std::sync::Arc; use anyhow::{Context as _, Result}; use super::Claim; -use hive_jobq::TerminalState; - -use super::model::{NodeKind, NodeSpec}; +use super::model::{NodeKind, NodeSpec, State}; use crate::coordinator::Coordinator; use crate::power::{ReconcileAction, reconcile_action}; @@ -97,11 +95,10 @@ pub(super) async fn run_node(coord: &Arc, claim: &Claim) -> Result< NodeKind::DeployApply { .. } => run_deploy_apply(coord, claim).await, NodeKind::FinalizeDeploy { .. } => run_finalize_deploy(coord, claim).await, NodeKind::DeployTail { .. } => run_deploy_tail(coord, claim).await, - NodeKind::ResolveApproval { - approval_id, - outcome, - } => run_resolve_approval(coord, claim, *approval_id, *outcome).await, - NodeKind::EmitRebuilt { ok, .. } => Ok(run_emit_rebuilt(coord, claim, *ok)), + NodeKind::ResolveApproval { approval_id, .. } => { + run_resolve_approval(coord, claim, *approval_id).await + } + NodeKind::EmitRebuilt { .. } => Ok(run_emit_rebuilt(coord, claim)), NodeKind::SetWanted { up, .. } => run_set_wanted(coord, claim, *up), // Pure grouping container — no work; completing it lets it reach // `Finishing` so its child work nodes start. The DAG's terminal side @@ -110,39 +107,58 @@ pub(super) async fn run_node(coord: &Arc, claim: &Claim) -> Result< } } -/// Resolve the DAG's approval row the way this node's own `outcome` says. -/// -/// Nothing is inspected: a template emits one of these per outcome, each edged to -/// accept only that one, so *which* node the scheduler let run already is the -/// answer. Best-effort — a resolution failure is logged inside -/// [`crate::actions::resolve_approval_dag`], never surfaced as a node failure, -/// since the work already happened and failing the tail would only misreport it. +/// Resolve the DAG's approval row from how the work it follows ended. The +/// outcome comes off this node's own dependency roll-up, not from re-reading +/// the world. Best-effort: a resolution failure is logged inside +/// [`crate::actions::resolve_approval_dag`], never surfaced as a node failure — +/// the work already happened, and failing the tail would only misreport it. async fn run_resolve_approval( coord: &Arc, claim: &Claim, approval_id: i64, - outcome: TerminalState, ) -> Result { - let reason = (outcome == TerminalState::Failed) - .then(|| coord.job_queue.first_error(claim.dag_id)) - .flatten(); - crate::actions::resolve_approval_dag(coord, approval_id, outcome, reason.as_deref()).await; + let reason = failure_reason(coord, claim); + crate::actions::resolve_approval_dag(coord, approval_id, claim.deps_state(), reason.as_deref()) + .await; Ok(NodeOutput::default()) } -/// Emit this agent's `Rebuilt` manager event. `ok` is not computed — it is which -/// of the tail pair the graph let run. The failure note comes from the DAG's -/// first failing node, since the branch knows *that* it failed but not *why*. -fn run_emit_rebuilt(coord: &Arc, claim: &Claim, ok: bool) -> NodeOutput { - coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { - agent: claim.agent.clone(), - ok, - note: (!ok) - .then(|| coord.job_queue.first_error(claim.dag_id)) - .flatten(), - sha: None, - tag: None, - }); +/// Why the work a tail node follows failed, as a human-readable string. +/// +/// Prefers the tail's own dependency error, but a dep that is a **group root** +/// rolled up `Failed` from a child carries no error of its own (the reason lives +/// on the leaf that actually failed) — and a grafted subgraph's nodes can't be +/// edged statically anyway. So fall back to the DAG's first failing node. Still +/// the queue's own graph, not the outside world. +fn failure_reason(coord: &Arc, claim: &Claim) -> Option { + claim + .deps_error() + .map(str::to_owned) + .or_else(|| coord.job_queue.first_error(claim.dag_id)) +} + +/// Emit this agent's `Rebuilt` manager event — `ok` when the work it follows is +/// `Done`, `!ok` with the failure note when it `Failed`, and nothing at all when +/// it `Cancelled` (nothing ran, so there is no rebuild to report). +fn run_emit_rebuilt(coord: &Arc, claim: &Claim) -> NodeOutput { + let agent = claim.agent.clone(); + match claim.deps_state() { + State::Done => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { + agent, + ok: true, + note: None, + sha: None, + tag: None, + }), + State::Failed => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { + agent, + ok: false, + note: failure_reason(coord, claim), + sha: None, + tag: None, + }), + _ => {} + } NodeOutput::default() } diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index f03ffe45..2a3b4e7a 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -18,9 +18,10 @@ //! 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`], 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. +//! 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. //! //! 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 @@ -43,13 +44,12 @@ use chrono::{DateTime, Utc}; use hive_host_sock::jobs::NodeView; use hive_jobq::resources::ResourceTable; use hive_jobq::scheduler::{Outcome, Scheduler}; -use hive_jobq::{Dep, Graph, NodeId, State as JobState}; +use hive_jobq::{Dep, DepWhen as JobDepWhen, Graph, NodeId, State as JobState}; use hive_sh4re::wire_time::now_unix; use tokio::sync::Notify; use crate::coordinator::TransientKind; -pub use hive_jobq::TerminalState; -pub use model::{DagSpec, DagView, NodeKind, NodeSpec, PermPayload, Source, State}; +pub use model::{DagSpec, DagView, DepWhen, NodeKind, NodeSpec, PermPayload, Source, State}; use resource::Resource; /// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) the snapshot @@ -60,6 +60,31 @@ const MAX_HISTORY_DAGS: usize = 50; /// Cap on stored node error strings. const MAX_ERROR_LEN: usize = 2_000; +/// How one of a claimed node's dependencies finished, snapshotted at claim time. +/// +/// A node only starts once its edges are satisfied, so every dep named here is +/// already terminal: an `AfterOk` edge means [`State::Done`], an `AfterAny` edge +/// means any of `Done` / `Failed` / `Cancelled`. +/// +/// This is what lets a tail node be an ordinary node. A tail that must report +/// how the work below it went — "emit `Rebuilt { ok }`", "resolve this approval +/// with the failure note" — reads its deps' outcomes off its own claim, rather +/// than re-deriving them from the world after the fact. The alternative in this +/// codebase is `DeployTail`, which infers success by re-reading *git state*; that +/// works only because a deploy happens to write its result somewhere durable, and +/// it is not a pattern to copy. +/// +/// Carries no node id: a tail acts on *how* its dependencies ended, never on +/// which one it was, so an id here would be a field with no reader. +#[derive(Debug, Clone)] +pub struct DepOutcome { + /// Its terminal state, in wire terms. + pub state: State, + /// Its failure reason, when it failed with one of its own. A node that + /// rolled up `Failed` from a child, or was cancelled, carries no error. + pub error: Option, +} + /// A node claimed for execution — everything the executor needs, snapshotted at /// claim time. #[derive(Debug, Clone)] @@ -76,6 +101,38 @@ pub struct Claim { /// pill is currently shown is derived from live lease ownership /// ([`JobQueue::held_transients`]), not a per-claim edge. pub transient: Option, + /// How each node this one depends on finished. Empty for a head node. + /// See [`DepOutcome`] — this is how a tail node learns the outcome of the + /// work it follows without going back to the world to ask. + pub deps: Vec, +} + +impl Claim { + /// Roll this claim's dependencies up into one outcome — what a weak-edged + /// tail node acts on. `Done` only when every dep succeeded; `Failed` when any + /// failed; otherwise `Cancelled` (all terminal, none failed, so the work was + /// dropped before it ran). + /// + /// A head node has no deps and rolls up `Done` — vacuously true, and never + /// reached in practice since only tail kinds consult this. + pub fn deps_state(&self) -> State { + if self.deps.iter().all(|d| d.state == State::Done) { + State::Done + } else if self.deps.iter().any(|d| d.state == State::Failed) { + State::Failed + } else { + State::Cancelled + } + } + + /// The first failed dependency's error, for reporting *why* the work ended + /// badly. `None` when nothing failed. + pub fn deps_error(&self) -> Option<&str> { + self.deps + .iter() + .find(|d| d.state == State::Failed) + .and_then(|d| d.error.as_deref()) + } } /// Per-node runtime metadata the crate graph doesn't carry. Lifecycle @@ -133,21 +190,23 @@ impl Default for JobQueue { } } +/// Map a spec dependency edge kind onto the crate's. +fn to_crate_when(when: DepWhen) -> JobDepWhen { + match when { + DepWhen::AfterOk => JobDepWhen::AfterOk, + DepWhen::AfterAny => JobDepWhen::AfterAny, + } +} + /// Map a crate node state onto the wire state (`Pending` ↔ `Queued`; /// `Finishing` — own logic done, sub-nodes still running — reads as `Running`). -/// -/// `Skipped` has no wire counterpart and folds into `Cancelled`: to a reader -/// both mean "this never ran". The distinction is a *scheduling* one — it -/// decides whether a parent's roll-up counts the node — and the wire carries no -/// roll-up input, only display state. In practice a client never sees it either: -/// `dag_view` drops skipped nodes from the snapshot along with `Done` ones. fn to_wire_state(state: JobState) -> State { match state { JobState::Pending => State::Queued, JobState::Running | JobState::Finishing => State::Running, JobState::Done => State::Done, JobState::Failed => State::Failed, - JobState::Cancelled | JobState::Skipped => State::Cancelled, + JobState::Cancelled => State::Cancelled, } } @@ -182,7 +241,7 @@ fn insert_group( for d in &ns.deps { deps.push(Dep::Node { id: ids[dep_index(d.on)], - when: d.when, + when: to_crate_when(d.when), }); } let parent = match ns.parent { @@ -314,6 +373,24 @@ impl JobQueue { }; let kind = node.payload.clone(); let agent = node.payload.agent().to_owned(); + let dep_ids: Vec = node + .deps + .iter() + .filter_map(|d| match d { + Dep::Node { id, .. } => Some(*id), + Dep::Resource { .. } => None, + }) + .collect(); + let deps: Vec = dep_ids + .into_iter() + .filter_map(|dep| { + let n = inner.sched.graph().node(dep)?; + Some(DepOutcome { + state: to_wire_state(n.state), + error: n.error.clone(), + }) + }) + .collect(); let Some(container) = inner.dag_of(id) else { continue; }; @@ -328,6 +405,7 @@ impl JobQueue { approval_id: meta.approval_id, inputs: meta.inputs, transient: meta.transient, + deps, }); // `started_at` is stamped on the graph `Node` by the scheduler's // transition to `Running` — no host-side copy needed. @@ -360,17 +438,17 @@ impl JobQueue { /// so each is cancelled. `false` once any work node is running or terminal — /// an in-flight nix build isn't interruptible. /// - /// **Nodes that explicitly observe cancellation are spared** — a node whose - /// edge names [`hive_jobq::TerminalState::Cancelled`] is asking to run when - /// the work it follows was dropped, which is exactly what an approval tail - /// needs: cancel the work, and the tail still fires to resolve the approval - /// row rather than leaving it dangling forever. + /// **Tail nodes are spared** ([`NodeKind::is_tail`]). They are weak-edged + /// (`AfterAny`), and a `Cancelled` dep satisfies a weak edge, so sparing one + /// leaves it *ready* rather than stranded: the scheduler claims it on the next + /// pass, its [`Claim::deps_state`] reads `Cancelled`, and it resolves the + /// approval as "cancelled before completion". That is what stops a queued + /// approval DAG the operator cancelled from dangling its approval forever — + /// the job the inline hook used to do from outside the graph. /// - /// Nothing is special-cased by node kind. `AFTER_ANY` deliberately does *not* - /// accept `Cancelled`, so an ordinary weak-edged step (rebuild's `Reconcile`, - /// say) is cancelled along with everything else — there is nothing to converge - /// when no node ever ran. Only a node that named `Cancelled` survives, and it - /// survives because it asked to. + /// Cancelling the tail too would be the bug: `cancel_node` only cascades along + /// `AfterOk` edges and parent links, so nothing else would reach it, and the + /// approval row would simply never be touched. pub fn cancel(&self, dag_id: u64) -> bool { let mut inner = self.lock(); let Some(container) = inner.container(dag_id) else { @@ -388,7 +466,12 @@ impl JobQueue { return false; } for id in work { - if inner.observes_cancellation(id) { + if inner + .sched + .graph() + .node(id) + .is_some_and(|n| n.payload.is_tail()) + { continue; } inner.sched.cancel_node(id); @@ -567,18 +650,6 @@ impl QueueInner { .is_some_and(|n| n.state.is_terminal()) } - /// Whether `id` has an edge that accepts a **dropped** dependency — i.e. the - /// node exists to report on work that may never run. Used by - /// [`JobQueue::cancel`] to decide what to spare, so the decision comes from - /// the node's own declared edges rather than a hardcoded list of kinds. - fn observes_cancellation(&self, id: NodeId) -> bool { - self.sched.graph().node(id).is_some_and(|n| { - n.deps.iter().any(|d| { - matches!(d, Dep::Node { when, .. } if when.accepts(hive_jobq::TerminalState::Cancelled)) - }) - }) - } - /// First failed work node's error (read off the graph `Node`), for the /// dashboard's DAG-level error line. fn dag_first_error(&self, container: NodeId) -> Option { @@ -619,11 +690,7 @@ impl QueueInner { if let Some(f) = node.finished_at { finished.push(f); } - // `Done` nodes drop off the wire (a finished step isn't interesting), - // and so do `Skipped` ones: a branch that was never taken is noise on - // the dashboard, and surfacing it would also drag the client-side - // roll-up toward `Cancelled` for a run that went fine. - if matches!(node.state, JobState::Done | JobState::Skipped) { + if node.state == JobState::Done { continue; } let deps: Vec = node diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index dcddd2f4..0d321a36 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -17,7 +17,18 @@ use serde::Serialize; use crate::coordinator::TransientKind; -use hive_jobq::{DepWhen, TerminalState}; +/// When a dependency edge is considered satisfied. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum DepWhen { + /// Dep must reach `Done`. A `Failed` / `Cancelled` dep cancels this + /// node (cancel-downstream). + AfterOk, + /// Dep must merely reach a terminal state (ok *or* fail). Used only + /// by `rebuild`'s tail `Reconcile` so the recovery-start runs even + /// when `Swap` failed. + AfterAny, +} /// A dependency edge (intra-DAG only — cross-DAG ordering comes from /// the per-agent lease + dedup, never from edges between DAGs). @@ -219,7 +230,7 @@ pub enum NodeKind { /// Tail node of an approval-carrying DAG (spawn / opaque deploy / config-PR /// merge): resolve the approval row from how the work actually ended. /// - /// Weak-edged (`DepWhen::AFTER_ANY`) like [`NodeKind::DeployTail`], so it runs on + /// Weak-edged (`DepWhen::AfterAny`) like [`NodeKind::DeployTail`], so it runs on /// success, failure **and cancel** alike and decides internally. It reads its /// dependencies' terminal states off its own [`Claim::deps`] rather than /// re-deriving them from the world the way `DeployTail` reads git: a node is @@ -229,30 +240,18 @@ pub enum NodeKind { /// carrying one here would be a second copy free to drift. Like /// [`NodeKind::MetaLock`] it reports `""` from [`NodeKind::agent`] and takes /// no lease — which is also what lets one close a multi-agent DAG. - ResolveApproval { - approval_id: i64, - /// Which outcome this node reports. A template emits **one per outcome**, - /// each edged to accept only that one, so exactly one is ever runnable - /// and the executor has nothing to decide — it resolves the row the way - /// its own variant says. The `Cancelled` one is also the node that - /// [`super::JobQueue::cancel`] spares, since its edge is the only one - /// that accepts a dropped dependency. - outcome: TerminalState, - }, + /// + /// [`Claim::deps`]: super::Claim::deps + ResolveApproval { approval_id: i64 }, /// Tail node of a rebuild / perm-change: emit this agent's `Rebuilt` manager /// event — `ok` when its deps are `Done`, `!ok` carrying the failure note when /// 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** — 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. - /// - /// 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 }, + /// 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. + EmitRebuilt { agent: String }, /// 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 /// template so the downstream `Reconcile` reads it. Replaces the old @@ -341,7 +340,7 @@ impl NodeKind { | NodeKind::DeployApply { agent } | NodeKind::FinalizeDeploy { agent } | NodeKind::DeployTail { agent } - | NodeKind::EmitRebuilt { agent, .. } + | NodeKind::EmitRebuilt { agent } | NodeKind::SetWanted { agent, .. } => agent, NodeKind::MetaLock { .. } | NodeKind::Reparent { .. } @@ -350,6 +349,20 @@ impl NodeKind { } } + /// Whether this is a DAG's **tail** — a node that reports how the rest of the + /// DAG ended rather than doing work of its own. + /// + /// The one place this matters is [`super::JobQueue::cancel`], which spares + /// tails so they still run (and report `Cancelled`) on a cancelled DAG. Note + /// [`NodeKind::DeployTail`] is *not* one: despite the name it does real + /// compensating work, and on a cancelled DAG there is nothing to compensate. + pub fn is_tail(&self) -> bool { + matches!( + self, + NodeKind::ResolveApproval { .. } | NodeKind::EmitRebuilt { .. } + ) + } + /// Nix-heavy kinds hold one of the `buildSlots` semaphore permits /// for the node's duration. pub fn needs_build_slot(&self) -> bool { diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index 08608491..d47be49d 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -30,9 +30,7 @@ use anyhow::{Result, bail}; -use hive_jobq::{DepWhen, TerminalState}; - -use super::model::{DagSpec, Dep, NodeKind, NodeSpec, PermPayload, Source}; +use super::model::{DagSpec, Dep, DepWhen, NodeKind, NodeSpec, PermPayload, Source}; use crate::coordinator::TransientKind; /// After-ok edge on the previous node — the common chain link. Shared with @@ -41,124 +39,27 @@ use crate::coordinator::TransientKind; pub(crate) fn after_ok(on: u64) -> Vec { vec![Dep { on, - when: DepWhen::AFTER_OK, + when: DepWhen::AfterOk, }] } -/// `AfterOk` edges onto every one of a DAG's **group-roots** — the success -/// branch of a per-outcome tail pair, and the aggregator the failure branch -/// keys off. +/// Weak edges onto every one of a DAG's other **group-roots** — how a tail node +/// (`ResolveApproval` / `EmitRebuilt`) sees the whole DAG's outcome. /// /// Group-roots are the right granularity, not "every node": a root's state *is* /// its subtree's roll-up, so edging the roots covers every descendant while -/// keeping the dep list small and stable as subtrees grow. Because every edge is -/// `AFTER_OK`, this node runs only if *all* of them succeeded — and is ruled out -/// ([`TerminalState::Skipped`]) the moment one doesn't, which is precisely the -/// signal [`on_elimination_of`] waits for. -pub(crate) fn after_ok_all(ons: &[u64]) -> Vec { - ons.iter() - .map(|&on| Dep { - on, - when: DepWhen::AFTER_OK, - }) - .collect() -} - -/// `AFTER_ANY` edges onto every group-root — "wait for all of these to finish, -/// however they went". Ordering only; it accepts any outcome except the DAG -/// being dropped. +/// keeping the tail's dep list small and stable as subtrees grow. `AfterAny` +/// throughout, so the tail runs on success, failure and cancel alike and decides +/// from [`super::Claim::deps_state`]. pub(crate) fn after_any_all(ons: &[u64]) -> Vec { ons.iter() .map(|&on| Dep { on, - when: DepWhen::AFTER_ANY, + when: DepWhen::AfterAny, }) .collect() } -/// A single edge satisfied only when `on` was **ruled out** by its own edges. -/// -/// Dependency edges are conjunctive, so "any one of these several nodes failed" -/// cannot be written directly. This is the composition that expresses it: point -/// the success branch at every root with [`after_ok_all`], then hang the failure -/// branch off *that* node's elimination. Exactly one of the pair ever runs. -/// -/// Note it accepts `Skipped` and **not** `Cancelled`: if the whole DAG was -/// dropped before it started, the success branch is marked `Cancelled` directly -/// and this branch is ruled out too — a job nobody ran reports nothing. -pub(crate) fn on_elimination_of(on: u64) -> Vec { - vec![Dep { - on, - when: DepWhen::of(&[TerminalState::Skipped]), - }] -} - -/// A single edge satisfied only by the listed outcomes of `on` — for the -/// one-tail-per-outcome shape an approval DAG uses. -pub(crate) fn on_outcome(on: u64, outcomes: &[TerminalState]) -> Vec { - vec![Dep { - on, - when: DepWhen::of(outcomes), - }] -} - -/// The `Rebuilt`-reporting tail pair for a rebuild-shaped DAG: the success node -/// gated on every group-root in `roots`, and the failure node gated on *its* -/// elimination. `base` is the spec index the pair starts at. -/// -/// Exactly one runs on a DAG that executed, and neither runs on one the operator -/// dropped — see [`on_elimination_of`]. -fn emit_rebuilt_tails(agent: &str, roots: &[u64], base: u64) -> Vec { - // The failure branch needs *both*: the ok branch being ruled out (that is the - // "something went wrong" signal) **and** every root actually finished. The - // second half is easy to forget and gets the ordering wrong without it — a - // failed `Prebuild` eliminates the ok branch immediately, while the recovery - // `Reconcile` is still bringing the container back up, so reporting straight - // off the elimination would announce the failure mid-recovery. - let mut on_fail = after_any_all(roots); - on_fail.extend(on_elimination_of(base)); - vec![ - node( - NodeKind::EmitRebuilt { - agent: agent.to_owned(), - ok: true, - }, - after_ok_all(roots), - ), - node( - NodeKind::EmitRebuilt { - agent: agent.to_owned(), - ok: false, - }, - on_fail, - ), - ] -} - -/// The approval-resolving tails for an approval-carrying DAG: one per outcome of -/// the DAG's single group-root `root`, each accepting only its own. -/// -/// The `Cancelled` node is what keeps a dropped approval DAG from dangling its -/// row forever — its edge is the only one [`super::JobQueue::cancel`] spares. -fn resolve_approval_tails(approval_id: i64, root: u64) -> Vec { - [ - TerminalState::Done, - TerminalState::Failed, - TerminalState::Cancelled, - ] - .into_iter() - .map(|outcome| { - node( - NodeKind::ResolveApproval { - approval_id, - outcome, - }, - on_outcome(root, &[outcome]), - ) - }) - .collect() -} - /// Build one **top-level (group-root)** node — `parent = None`. `kind` carries /// the agent it targets ([`NodeKind`] is the payload directly). Shared with /// `submit.rs`'s dynamic power-op builders. A root owns whatever resource it @@ -234,7 +135,7 @@ pub(crate) fn rebuild_nodes(agent: &str, relock: bool, base: u64) -> Vec Vec { vec![ Dep { on: 1, - when: DepWhen::AFTER_OK, + when: DepWhen::AfterOk, }, Dep { on: 5, - when: DepWhen::AFTER_OK, + when: DepWhen::AfterOk, }, ], )); @@ -297,7 +198,12 @@ pub(crate) fn deploy_rebuild_nodes(agent: &str) -> Vec { /// 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 { let mut nodes = rebuild_nodes(agent, relock, 0); - nodes.extend(emit_rebuilt_tails(agent, &[0, 1, 5], 6)); + nodes.push(node( + NodeKind::EmitRebuilt { + agent: agent.to_owned(), + }, + after_any_all(&[0, 1, 5]), + )); DagSpec { source, reason, @@ -350,13 +256,14 @@ pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec NodeKind::DeployTail { agent: a() }, vec![Dep { on: 2, - when: DepWhen::AFTER_ANY, + when: DepWhen::AfterAny, }], ), - ] - .into_iter() - .chain(resolve_approval_tails(approval_id, 0)) - .collect(), + node( + NodeKind::ResolveApproval { approval_id }, + after_any_all(&[0]), + ), + ], } } @@ -412,10 +319,11 @@ pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec { child(0, NodeKind::Create { agent: a() }, Vec::new()), child(1, NodeKind::WriteDropin { agent: a() }, Vec::new()), child(1, NodeKind::Reconcile { agent: a() }, after_ok(2)), + node( + NodeKind::ResolveApproval { approval_id }, + after_any_all(&[0]), + ), ] - .into_iter() - .chain(resolve_approval_tails(approval_id, 0)) - .collect() }, } } @@ -434,7 +342,12 @@ pub fn perm_change(agent: &str, source: Source, reason: String, payload: PermPay Vec::new(), )]; nodes.extend(rebuild_nodes(agent, true, 1)); - nodes.extend(emit_rebuilt_tails(agent, &[0, 1, 2, 6], 7)); + nodes.push(node( + NodeKind::EmitRebuilt { + agent: agent.to_owned(), + }, + after_any_all(&[0, 1, 2, 6]), + )); DagSpec { source, reason, @@ -469,11 +382,14 @@ pub fn meta_update( Vec::new(), )]; // The bump itself has no side effect, so an operator-driven one ends at the - // `MetaLock`; an approval-driven one still has its row to resolve and gets the - // per-outcome tails edged onto that single group-root — whose roll-up covers - // the rebuild subgraphs `MetaLock` grows into itself. + // `MetaLock`; an approval-driven one still has its row to resolve and gets a + // tail edged onto that single group-root — whose roll-up covers the rebuild + // subgraphs `MetaLock` grows into itself. if let Some(approval_id) = approval_id { - nodes.extend(resolve_approval_tails(approval_id, 0)); + nodes.push(node( + NodeKind::ResolveApproval { approval_id }, + after_any_all(&[0]), + )); } DagSpec { source, diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 0088e837..40251fd7 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -6,9 +6,7 @@ //! scheduler's async loop is a thin claim/complete pump over the same //! methods exercised here. -use hive_jobq::DepWhen; - -use super::model::{Dep, NodeKind, NodeSpec}; +use super::model::{Dep, DepWhen, NodeKind, NodeSpec}; use super::*; fn submit(q: &JobQueue, spec: DagSpec) -> u64 { @@ -50,34 +48,35 @@ fn claim_one(q: &JobQueue) -> Claim { claims.pop().expect("one claim") } -/// Claim an approval DAG's `ResolveApproval` tail and complete it, asserting it -/// is the one built for `expect`. -/// -/// A template emits one tail per outcome and the graph runs exactly one, so the -/// assertion is on *which node was claimed* — that alone says what the approval -/// row is about to be resolved as. Nothing computes it. -fn settle_approval_tail(q: &JobQueue, dag_id: u64, approval_id: i64, expect: TerminalState) { +/// Claim an approval DAG's `ResolveApproval` tail, assert which approval it +/// carries and what it will report to that row, then complete it. Replaces the +/// old `terminal_summary()` assertions: the outcome is no longer a struct handed +/// to a hook, it's what this node reads off its own deps. +fn settle_approval_tail(q: &JobQueue, dag_id: u64, approval_id: i64, expect: State) { let tail = claim_one(q); assert!( - matches!( - tail.kind, - NodeKind::ResolveApproval { approval_id: got, outcome } - if got == approval_id && outcome == expect - ), - "expected the {expect:?} ResolveApproval tail for #{approval_id}, got {:?}", + matches!(tail.kind, NodeKind::ResolveApproval { approval_id: got } if got == approval_id), + "expected the ResolveApproval tail for #{approval_id}, got {:?}", tail.kind ); + assert_eq!( + tail.deps_state(), + expect, + "outcome the tail reports to approval #{approval_id}" + ); q.complete_node(dag_id, tail.node_id, Ok(())); } -/// The `EmitRebuilt` counterpart of [`settle_approval_tail`] — claim the tail the -/// graph let run and assert it's the `ok` one expected. -fn settle_rebuild_tail(q: &JobQueue, dag_id: u64, agent: &str, expect_ok: bool) { +/// The `EmitRebuilt` counterpart of [`settle_approval_tail`] — claim a rebuild / +/// perm-change DAG's tail, assert the `Rebuilt` event it will emit, complete it. +fn settle_rebuild_tail(q: &JobQueue, dag_id: u64, agent: &str, expect: State) { let tail = claim_one(q); - assert!( - matches!(&tail.kind, NodeKind::EmitRebuilt { agent: a, ok } if a == agent && *ok == expect_ok), - "expected the ok={expect_ok} EmitRebuilt tail for {agent}, got {:?}", - tail.kind + assert_eq!(tail.kind.as_str(), "emit_rebuilt"); + assert_eq!(tail.agent, agent, "`Rebuilt` is emitted per agent"); + assert_eq!( + tail.deps_state(), + expect, + "ok-ness of the emitted `Rebuilt`" ); q.complete_node(dag_id, tail.node_id, Ok(())); } @@ -155,7 +154,7 @@ fn cyclic_dag_is_rejected_at_submit() { }, deps: vec![Dep { on: 1, - when: DepWhen::AFTER_OK, + when: DepWhen::AfterOk, }], parent: None, }, @@ -165,7 +164,7 @@ fn cyclic_dag_is_rejected_at_submit() { }, deps: vec![Dep { on: 0, - when: DepWhen::AFTER_OK, + when: DepWhen::AfterOk, }], parent: None, }, @@ -184,7 +183,7 @@ fn unknown_dep_is_rejected_at_submit() { }, deps: vec![Dep { on: 9, - when: DepWhen::AFTER_OK, + when: DepWhen::AfterOk, }], parent: None, }]; @@ -230,7 +229,7 @@ fn rebuild_chain_claims_in_dep_order() { ); q.complete_node(id, c.node_id, Ok(())); } - settle_rebuild_tail(&q, id, "agent-a", true); + settle_rebuild_tail(&q, id, "agent-a", State::Done); assert_eq!(state_of(&q, id), State::Done); } @@ -737,6 +736,73 @@ fn meta_update_carries_rebuilding_transient_and_grows_cascade_in_dag() { ); } +// ---- dep outcomes on the claim ---- + +/// Every claimed node reports how each node it depends on finished, and those +/// deps are always already terminal — that is what a node's edges being +/// satisfied *means*. A tail node reads its `deps` instead of going back to the +/// world to find out how the work below it went. +#[test] +fn claim_carries_terminal_dep_outcomes() { + let q = JobQueue::new(1); + let id = submit(&q, rebuild("agent-a", "r")); + let mut saw_a_dep = false; + loop { + let mut claims = q.claim_ready(); + let Some(claim) = claims.pop() else { break }; + assert!(claims.is_empty(), "one build slot ⇒ one claim at a time"); + for dep in &claim.deps { + saw_a_dep = true; + assert!( + dep.state.is_terminal(), + "{} was claimed with a non-terminal dep ({:?}) — a node's edges \ + being satisfied is exactly the claim that its deps have finished", + claim.kind.as_str(), + dep.state + ); + assert_eq!( + dep.state, + State::Done, + "on the happy path every dep of {} finished Done", + claim.kind.as_str() + ); + assert_eq!(dep.error, None, "a Done dep carries no error"); + } + q.complete_node(id, claim.node_id, Ok(())); + } + assert!(saw_a_dep, "the rebuild DAG has at least one dependent node"); + assert_eq!(state_of(&q, id), State::Done); +} + +/// The failure direction, which is the whole point of carrying outcomes at all: +/// `Reconcile` hangs off `Prebuild` with `AfterAny`, so a failed prebuild +/// cancel-cascades `StopForUpdate`/`Swap`/`PostSwap` and `Reconcile` still runs +/// — and its claim hands it the failure, including the reason, rather than +/// leaving the executor to go and re-derive it from the world. +#[test] +fn claim_dep_outcome_reports_a_failed_dep_with_its_error() { + let q = JobQueue::new(1); + let id = submit(&q, rebuild("agent-a", "r")); + let meta_sync = claim_one(&q); + q.complete_node(id, meta_sync.node_id, Ok(())); + let prebuild = claim_one(&q); + assert_eq!(prebuild.kind.as_str(), "prebuild"); + q.complete_node(id, prebuild.node_id, Err("nix build exploded".to_owned())); + let reconcile = claim_one(&q); + assert_eq!(reconcile.kind.as_str(), "reconcile"); + assert_eq!( + reconcile + .deps + .iter() + .map(|d| (d.state, d.error.as_deref())) + .collect::>(), + vec![(State::Failed, Some("nix build exploded"))], + "reconcile's claim carries the failed prebuild and its reason" + ); + assert_eq!(reconcile.deps_state(), State::Failed); + assert_eq!(reconcile.deps_error(), Some("nix build exploded")); +} + // ---- failure: cancel-downstream + AfterAny ---- #[test] @@ -765,19 +831,9 @@ fn failed_node_cancels_downstream_but_afterany_reconcile_runs() { .state }; assert_eq!(by_kind("prebuild"), State::Failed); - // `StopForUpdate` / `Swap` / `PostSwap` were *ruled out* by the failed - // `Prebuild` — `Skipped`, and skipped nodes are filtered off the wire along - // with `Done` ones. The failure itself is still visible (the `prebuild` row - // above, and the roll-up), which is the part an operator acts on. - // Restoring that detail wants a real `Skipped` wire state the client renders - // as "not run" — surfacing them as `Cancelled` instead would make a - // *successful* DAG with a not-taken branch read as cancelled. - for ruled_out in ["stop_for_update", "swap", "post_swap"] { - assert!( - dag.nodes.iter().all(|n| n.kind != ruled_out), - "{ruled_out} was ruled out, so it is off the wire" - ); - } + assert_eq!(by_kind("stop_for_update"), State::Cancelled); + assert_eq!(by_kind("swap"), State::Cancelled); + assert_eq!(by_kind("post_swap"), State::Cancelled); // The AfterAny reconcile ran (claimed + completed Ok above) → it's `Done`, // and `Done` nodes are excluded from the wire, so it's absent here. assert!( @@ -816,18 +872,14 @@ fn swap_failure_still_runs_reconcile() { q.complete_node(id, reconcile.node_id, Ok(())); let all_dags = q.snapshot(); let dag = all_dags.iter().find(|d| d.id == id).expect("dag"); - assert!( - dag.nodes.iter().all(|n| n.kind != "post_swap"), - "PostSwap is ruled out by the failed Swap (`Skipped`, so off the wire)" - ); assert_eq!( dag.nodes .iter() - .find(|n| n.kind == "swap") - .expect("swap node") + .find(|n| n.kind == "post_swap") + .expect("post_swap node") .state, - State::Failed, - "and the failure that ruled it out is still on the wire" + State::Cancelled, + "PostSwap must cancel-cascade when Swap fails" ); assert_eq!(state_of(&q, id), State::Failed); } @@ -859,7 +911,7 @@ fn swap_ok_runs_post_swap_before_reconcile() { let reconcile = claim_one(&q); assert_eq!(reconcile.kind.as_str(), "reconcile"); q.complete_node(id, reconcile.node_id, Ok(())); - settle_rebuild_tail(&q, id, "agent-a", true); + settle_rebuild_tail(&q, id, "agent-a", State::Done); assert_eq!(state_of(&q, id), State::Done); } @@ -887,14 +939,18 @@ fn cancel_clears_queued_dag() { // operator who just cancelled it (the dashboard renders this roll-up from // the snapshot `post_rebuild_queue_cancel` emits synchronously). assert_eq!(state_of(&q, id), State::Cancelled, "no stale Queued gap"); - // Neither `EmitRebuilt` tail accepts a *dropped* dependency — the ok one is - // `AFTER_OK`, the failure one keys on elimination — so both are cancelled - // with the work and **nothing is emitted** for a rebuild that never ran. - assert!( - q.claim_ready().is_empty(), - "a dropped rebuild reports nothing" + // Every work node is `Cancelled`, but the tail is spared so it can still + // report the cancellation — so it is the one thing left to claim. + let tail = claim_one(&q); + assert_eq!(tail.kind.as_str(), "emit_rebuilt"); + assert_eq!( + tail.deps_state(), + State::Cancelled, + "the spared tail sees its deps cancelled, so it emits no `Rebuilt`" ); + q.complete_node(id, tail.node_id, Ok(())); assert_eq!(state_of(&q, id), State::Cancelled); + assert!(q.claim_ready().is_empty()); } #[test] @@ -1009,10 +1065,21 @@ fn cancelled_dag_still_runs_its_approval_tail() { templates::approval_deploy("agent-a", 7, "approval #7".to_owned()), ); assert!(q.cancel(id), "fully-queued dag cancels"); - // The `Cancelled` tail is the only node whose edge accepts a dropped - // dependency, so it is the only one `cancel` spares — and claiming it *is* - // the assertion that the approval gets resolved as cancelled. - settle_approval_tail(&q, id, 7, TerminalState::Cancelled); + let tail = claim_one(&q); + assert_eq!(tail.dag_id, id); + assert_eq!(tail.kind.as_str(), "resolve_approval"); + assert!( + matches!(tail.kind, NodeKind::ResolveApproval { approval_id: 7 }), + "the tail carries the approval to resolve, got {:?}", + tail.kind + ); + assert_eq!( + tail.deps_state(), + State::Cancelled, + "so the executor resolves the approval as cancelled, not as a failure" + ); + assert_eq!(tail.deps_error(), None, "a cancelled dep carries no error"); + q.complete_node(id, tail.node_id, Ok(())); assert_eq!(state_of(&q, id), State::Cancelled); // Unrelated later activity doesn't disturb the settled DAG. let other = submit(&q, rebuild("agent-b", "r")); @@ -1063,7 +1130,7 @@ fn deploy_dag_runs_phases_in_order_and_tails_a_failed_apply() { ); q.complete_node(id, tail.node_id, Ok(())); - settle_approval_tail(&q, id, 7, TerminalState::Failed); + settle_approval_tail(&q, id, 7, State::Failed); assert_eq!( state_of(&q, id), State::Failed, @@ -1136,7 +1203,7 @@ fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() { assert!(matches!(tail.kind, NodeKind::DeployTail { .. })); q.complete_node(id, tail.node_id, Ok(())); - settle_approval_tail(&q, id, 11, TerminalState::Done); + settle_approval_tail(&q, id, 11, State::Done); assert_eq!(state_of(&q, id), State::Done); } @@ -1187,7 +1254,7 @@ fn deploy_dag_skips_finalize_but_still_tails_a_failed_graft() { ); q.complete_node(id, tail.node_id, Ok(())); - settle_approval_tail(&q, id, 13, TerminalState::Failed); + settle_approval_tail(&q, id, 13, State::Failed); assert_eq!(state_of(&q, id), State::Failed); assert_eq!( q.first_error(id).as_deref(), @@ -1225,7 +1292,7 @@ fn deploy_dag_skips_apply_but_still_runs_tail_when_verify_fails() { ); q.complete_node(id, tail.node_id, Ok(())); - settle_approval_tail(&q, id, 9, TerminalState::Failed); + settle_approval_tail(&q, id, 9, State::Failed); assert_eq!(state_of(&q, id), State::Failed); } @@ -1365,7 +1432,7 @@ fn spawn_shape_provision_create_dropin_reconcile() { assert_eq!(c.approval_id, Some(7)); q.complete_node(id, c.node_id, Ok(())); } - settle_approval_tail(&q, id, 7, TerminalState::Done); + settle_approval_tail(&q, id, 7, State::Done); assert_eq!(state_of(&q, id), State::Done); } @@ -1397,7 +1464,7 @@ fn perm_change_shape_prefixes_rebuild_chain() { assert_eq!(c.kind.as_str(), expected); q.complete_node(id, c.node_id, Ok(())); } - settle_rebuild_tail(&q, id, "agent-a", true); + settle_rebuild_tail(&q, id, "agent-a", State::Done); assert_eq!(state_of(&q, id), State::Done); } diff --git a/hive-jobq/Cargo.toml b/hive-jobq/Cargo.toml index 4fc70e0a..aa5370f7 100644 --- a/hive-jobq/Cargo.toml +++ b/hive-jobq/Cargo.toml @@ -9,7 +9,6 @@ workspace = true [dependencies] chrono = { workspace = true } -enumflags2.workspace = true serde = { workspace = true } thiserror = { workspace = true } diff --git a/hive-jobq/src/lib.rs b/hive-jobq/src/lib.rs index 60e5b6cc..5963f042 100644 --- a/hive-jobq/src/lib.rs +++ b/hive-jobq/src/lib.rs @@ -56,122 +56,29 @@ 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)] +/// 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)] -#[serde(rename_all = "snake_case")] -pub enum TerminalState { - /// Own logic succeeded and every sub-node did too. - Done, - /// Own logic failed, or a sub-node did. - Failed, - /// Never ran because the work was **dropped** before it could start — the - /// caller cancelled the whole group while it was still queued. Counts as - /// not-success when a parent rolls up. - Cancelled, - /// Never ran because its own **edges ruled it out**: a dependency settled on - /// an outcome the edge doesn't accept. Expected, not a problem — the failure - /// branch of a run that succeeded is `Skipped`. - /// - /// A parent's roll-up **ignores** `Skipped` children entirely. Without that, - /// branching on outcome would be self-defeating: exactly one branch is always - /// ruled out, so every group containing one would roll up failed. - Skipped, -} - -impl State { - /// This state as a [`TerminalState`], or `None` while the node is still - /// in flight. - #[must_use] - pub fn terminal(self) -> Option { - match self { - State::Done => Some(TerminalState::Done), - State::Failed => Some(TerminalState::Failed), - State::Cancelled => Some(TerminalState::Cancelled), - State::Skipped => Some(TerminalState::Skipped), - State::Pending | State::Running | State::Finishing => None, - } - } -} - -/// Which outcomes of a dependency satisfy a [`Dep::Node`] edge — **a set**, not -/// a fixed set of named cases. -/// -/// Naming the cases (`AfterOk` / `AfterFail` / …) means a new variant every time -/// a combination is wanted. A set is closed under combination: "run regardless" -/// (systemd's `After=`) is all three; "anything that isn't a failure" is -/// `{Done, Cancelled}`; a compensating branch is `{Failed}`. [`AFTER_OK`] and -/// [`AFTER_ANY`] stay as named constants because they're the two the templates -/// overwhelmingly use. -/// -/// The empty set satisfies nothing, so a node carrying one could never run; -/// [`Graph::validate`] rejects it. -/// -/// [`AFTER_OK`]: DepWhen::AFTER_OK -/// [`AFTER_ANY`]: DepWhen::AFTER_ANY -#[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())) - } +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 { - /// 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(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 - /// `Swap` failed *or* was itself ruled out by a failed `MetaSync`. - /// - /// Deliberately excludes [`TerminalState::Cancelled`]: if the group never - /// 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(enumflags2::make_bitflags!(TerminalState::{Done | Failed | Skipped})); - - /// An edge satisfied by exactly the listed outcomes. - #[must_use] - pub fn of(outcomes: &[TerminalState]) -> Self { - Self(outcomes.iter().copied().collect()) - } - - /// Whether `outcome` satisfies this edge. - #[must_use] - pub fn accepts(self, outcome: TerminalState) -> bool { - self.0.contains(outcome) - } - - /// An edge no outcome can satisfy — rejected at [`Graph::validate`]. - #[must_use] - pub fn is_empty(self) -> bool { - self.0.is_empty() - } - - /// Whether a dependency in `dep_state` satisfies this edge. A non-terminal - /// dependency never does. + /// Whether a dependency in `dep_state` satisfies this edge. #[must_use] pub fn satisfied_by(self, dep_state: State) -> bool { - dep_state.terminal().is_some_and(|t| self.accepts(t)) + match self { + DepWhen::AfterOk => dep_state == State::Done, + DepWhen::AfterAny => dep_state.is_terminal(), + } } } @@ -221,24 +128,18 @@ pub enum State { Done, /// Completed unsuccessfully — own logic failed, or a sub-node did. Failed, - /// Never ran: the work was dropped while still queued. See - /// [`TerminalState::Cancelled`]. + /// Never ran: an `AfterOk` dependency failed, so this node (and the rest of + /// its strong-dependent chain) is cancelled rather than run. Cancelled, - /// Never ran: its own edges ruled it out. See [`TerminalState::Skipped`] — - /// notably, a parent's roll-up ignores these. - Skipped, } impl State { /// A node is *terminal* once it has finished — successfully, unsuccessfully, - /// dropped, or ruled out — which is when its resources are released and - /// dependents are re-evaluated. + /// 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 | State::Skipped - ) + matches!(self, State::Done | State::Failed | State::Cancelled) } } @@ -306,16 +207,6 @@ pub enum GraphError { /// A node's `parent` named an id not present in the graph. #[error("parent references unknown node {0:?}")] UnknownParent(NodeId), - /// A [`Dep::Node`] edge carries an empty [`DepWhen`] set, which no outcome - /// can satisfy — the node could never run. Rejected at insert rather than - /// left to wedge its subtree non-terminal at runtime. - #[error("node {node:?} has an unsatisfiable dependency on {dep:?}: empty outcome set")] - UnsatisfiableDep { - /// The node that could never run. - node: NodeId, - /// The dependency whose edge accepts nothing. - dep: NodeId, - }, /// A node's [`Dep::Node`] edge points outside its own parent group — the /// target must be a proper descendant of the depender's `parent` (a sibling /// or a sibling's sub-node), never the parent itself or a node in another @@ -536,16 +427,10 @@ impl Graph { return Err(GraphError::UnknownParent(p)); } for dep in &node.deps { - if let Dep::Node { id, when } = dep { + if let Dep::Node { id, .. } = dep { if self.node(*id).is_none() { return Err(GraphError::UnknownDep(*id)); } - if when.is_empty() { - return Err(GraphError::UnsatisfiableDep { - node: node.id, - dep: *id, - }); - } if !self.dep_target_in_group(node.parent, *id) { return Err(GraphError::DepOutsideParent { dep: *id, @@ -580,7 +465,7 @@ mod tests { "update", vec![Dep::Node { id: a, - when: DepWhen::AFTER_OK, + when: DepWhen::AfterOk, }], None, ) @@ -609,25 +494,19 @@ mod tests { 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::AFTER_OK.satisfied_by(State::Done)); - assert!(!DepWhen::AFTER_OK.satisfied_by(State::Failed)); - assert!(!DepWhen::AFTER_OK.satisfied_by(State::Cancelled)); - assert!(!DepWhen::AFTER_OK.satisfied_by(State::Running)); - assert!(!DepWhen::AFTER_OK.satisfied_by(State::Skipped)); - // AfterAny: the dep reached a terminal state *some other way than being - // dropped* — success, failure, or ruled out by its own edges. - assert!(DepWhen::AFTER_ANY.satisfied_by(State::Done)); - assert!(DepWhen::AFTER_ANY.satisfied_by(State::Failed)); - assert!(DepWhen::AFTER_ANY.satisfied_by(State::Skipped)); - assert!( - !DepWhen::AFTER_ANY.satisfied_by(State::Cancelled), - "a dropped dep does not converge a weak dependent — nothing ever ran" - ); - assert!(!DepWhen::AFTER_ANY.satisfied_by(State::Pending)); + 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)); // Finishing satisfies neither — a dependent waits until the node rolls // up to a terminal state (all its sub-nodes done). - assert!(!DepWhen::AFTER_OK.satisfied_by(State::Finishing)); - assert!(!DepWhen::AFTER_ANY.satisfied_by(State::Finishing)); + assert!(!DepWhen::AfterOk.satisfied_by(State::Finishing)); + assert!(!DepWhen::AfterAny.satisfied_by(State::Finishing)); } #[test] @@ -636,7 +515,7 @@ mod tests { let bogus = NodeId(42); let deps = vec![Dep::Node { id: bogus, - when: DepWhen::AFTER_OK, + when: DepWhen::AfterOk, }]; assert_eq!( g.insert("x", deps, None).unwrap_err(), @@ -668,7 +547,7 @@ mod tests { "b".to_owned(), vec![Dep::Node { id: a, - when: DepWhen::AFTER_ANY, + when: DepWhen::AfterAny, }], None, ) @@ -689,7 +568,7 @@ mod tests { // roll-up model — the parent stays `Finishing` awaiting its children). let on_parent = vec![Dep::Node { id: root, - when: DepWhen::AFTER_OK, + when: DepWhen::AfterOk, }]; assert_eq!( g.insert("child", on_parent, Some(root)).unwrap_err(), @@ -719,7 +598,7 @@ mod tests { fn after_ok_dep(on: NodeId) -> Dep { Dep::Node { id: on, - when: DepWhen::AFTER_OK, + when: DepWhen::AfterOk, } } @@ -734,7 +613,7 @@ mod tests { payload: "x".to_owned(), deps: vec![Dep::Node { id: NodeId(99), - when: DepWhen::AFTER_OK, + when: DepWhen::AfterOk, }], state: State::Pending, started_at: None, diff --git a/hive-jobq/src/scheduler.rs b/hive-jobq/src/scheduler.rs index f18437d5..5cd80975 100644 --- a/hive-jobq/src/scheduler.rs +++ b/hive-jobq/src/scheduler.rs @@ -32,7 +32,7 @@ use std::collections::HashMap; use std::hash::Hash; use crate::resources::ResourceTable; -use crate::{Dep, Graph, GraphError, NodeId, State}; +use crate::{Dep, DepWhen, Graph, GraphError, NodeId, State}; /// The result of a node's own execution, reported to [`Scheduler::complete`]. /// @@ -235,9 +235,6 @@ impl Scheduler { /// Whether any direct child of `id` ended `Failed`/`Cancelled` — the roll-up /// failure condition for the parent. - /// `Skipped` children are **not** counted: being ruled out by an edge is the - /// expected fate of every branch not taken, so counting it would make any - /// group that branches on outcome roll up failed no matter how the run went. fn any_child_failed(&self, id: NodeId) -> bool { self.graph .nodes() @@ -247,12 +244,6 @@ impl Scheduler { /// Transition a node whose own logic just *succeeded* to its resulting state: /// [`State::Finishing`] while any child is still non-terminal, else `Failed` /// if a child failed, else `Done`. A node with no children skips `Finishing`. - /// - /// Cascades on **any** terminal outcome, `Done` included. Since an edge names - /// the set of outcomes it accepts, success can rule a dependent out just as - /// failure can — a `{Failed}` compensation branch is unsatisfiable the moment - /// its dependency succeeds, and leaving it `Pending` would wedge the subtree - /// non-terminal forever. fn settle_terminal(&mut self, id: NodeId) { let state = if !self.all_children_terminal(id) { State::Finishing @@ -262,7 +253,7 @@ impl Scheduler { State::Done }; self.graph.set_state(id, state); - if state.is_terminal() { + if state == State::Failed { self.cascade_cancel(id); } } @@ -285,13 +276,14 @@ impl Scheduler { State::Done }; self.graph.set_state(a, state); - // Any terminal outcome can rule a dependent out — see `settle_terminal`. - self.cascade_cancel(a); + if state == State::Failed { + self.cascade_cancel(a); + } cur = self.graph.node(a).and_then(|n| n.parent); } } - /// Cancel a still-*pending* node (and cascade to the dependents it rules out): + /// Cancel a still-*pending* node (and cascade to its `AfterOk` dependents): /// mark it [`State::Cancelled`] and report whether it was cancellable. A /// node that has already started (`Running`) or finished is left untouched — /// an in-flight node's work is not interruptible. A pending node holds no @@ -324,33 +316,16 @@ impl Scheduler { .collect() } - /// Propagate elimination out from a just-terminal `origin`: every - /// still-`Pending` node that can no longer run gets marked - /// [`State::Skipped`], transitively. Skipped nodes were `Pending`, so they - /// hold no resources. - /// - /// `Skipped`, not `Cancelled`: these nodes were *ruled out by their edges*, - /// which is a normal outcome, not a dropped job. `Cancelled` is reserved for - /// work the caller abandoned before it started ([`Scheduler::cancel_node`]), - /// and the two are distinguished precisely so a parent's roll-up can ignore - /// the former while still treating the latter as not-success. - /// - /// One rule decides it: **a node is doomed once any edge it names can never - /// be satisfied** — the dep settled on an outcome that edge does not accept. - /// `AFTER_OK` on a `Failed` dep dooms (a strong dependency failed); - /// `AFTER_ANY` never dooms, which is what lets a tail node survive the - /// cancellation of the work it reports on. That falls out of the edge's own - /// set rather than being a special case for particular node kinds. - /// - /// Plus the structural edge: being a *child* of a doomed node. Its parent - /// will never reach `Finishing`, so it was gated from ever starting, and - /// leaving it pending would wedge the subtree non-terminal. + /// Propagate cancellation out from a just-failed/cancelled `origin`: every + /// still-`Pending` node that can no longer run gets marked `Cancelled`, + /// transitively. Two edges carry it: (a) an `AfterOk` dep on a cancelled node + /// (a strong dependency failed), and (b) being a *child* of one (its parent + /// will never reach `Finishing`, so it was gated from ever starting — and + /// leaving it pending would wedge the subtree non-terminal). Cancelled nodes + /// were `Pending`, so they hold no resources. fn cascade_cancel(&mut self, origin: NodeId) { let mut stack = vec![origin]; while let Some(cur) = stack.pop() { - let Some(outcome) = self.graph.node(cur).and_then(|n| n.state.terminal()) else { - continue; - }; let doomed: Vec = self .graph .nodes() @@ -358,13 +333,13 @@ impl Scheduler { n.state == State::Pending && (n.parent == Some(cur) || n.deps.iter().any(|d| { - matches!(d, Dep::Node { id, when } if *id == cur && !when.accepts(outcome)) + matches!(d, Dep::Node { id, when: DepWhen::AfterOk } if *id == cur) })) }) .map(|n| n.id) .collect(); for d in doomed { - self.graph.set_state(d, State::Skipped); + self.graph.set_state(d, State::Cancelled); stack.push(d); } } @@ -438,7 +413,6 @@ impl Scheduler { #[cfg(test)] mod tests { use super::*; - use crate::{DepWhen, TerminalState}; fn res(name: &str) -> String { name.to_owned() @@ -462,7 +436,7 @@ mod tests { fn after_ok(on: NodeId) -> Dep { Dep::Node { id: on, - when: DepWhen::AFTER_OK, + when: DepWhen::AfterOk, } } @@ -515,10 +489,10 @@ mod tests { assert_eq!(n.error.as_deref(), Some("boom")); assert!(n.finished_at.is_some()); - // The `AFTER_OK` dependent: ruled out by its edge, so finished_at is set, + // The `AfterOk`-cancelled node: cascade-cancelled, so finished_at is set, // but it never ran (no started_at) and carries no error of its own. let n = s.graph().node(downstream).unwrap(); - assert_eq!(n.state, State::Skipped); + assert_eq!(n.state, State::Cancelled); assert!(n.started_at.is_none()); assert!(n.finished_at.is_some()); assert_eq!(n.error, None); @@ -753,148 +727,18 @@ mod tests { "weak", vec![Dep::Node { id: root, - when: DepWhen::AFTER_ANY, + when: DepWhen::AfterAny, }], None, ) .expect("weak"); assert_eq!(s.settle(), vec![root]); s.complete(root, Outcome::Failed(String::new())); - assert_eq!(s.graph().node(strong1).unwrap().state, State::Skipped); - assert_eq!(s.graph().node(strong2).unwrap().state, State::Skipped); + assert_eq!(s.graph().node(strong1).unwrap().state, State::Cancelled); + assert_eq!(s.graph().node(strong2).unwrap().state, State::Cancelled); assert_eq!(s.settle(), vec![weak]); } - /// The direction only a *set* edge can express: a branch that runs solely on - /// failure. Success has to rule it out, which means the cascade must fire on - /// `Done` too — otherwise it sits `Pending` forever and wedges the graph. - #[test] - fn success_cancels_a_failure_only_branch() { - let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new()); - let root = s.append("root", vec![], None).expect("root"); - let on_fail = s - .append( - "compensate", - vec![Dep::Node { - id: root, - when: DepWhen::of(&[TerminalState::Failed]), - }], - None, - ) - .expect("compensate"); - assert_eq!(s.settle(), vec![root]); - s.complete(root, Outcome::Done); - assert_eq!( - s.graph().node(on_fail).unwrap().state, - State::Skipped, - "a Failed-only branch is unsatisfiable once its dep succeeds — and it is \ - `Skipped`, not `Cancelled`, so the parent roll-up ignores it" - ); - assert!(s.settle().is_empty(), "and nothing is left runnable"); - } - - /// The mirror: the same branch is exactly what *does* run on failure, while - /// an `AFTER_OK` sibling is cancelled. One edge set, both directions. - #[test] - fn failure_runs_the_failure_only_branch_and_cancels_the_ok_one() { - let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new()); - let root = s.append("root", vec![], None).expect("root"); - let on_ok = s - .append("on_ok", vec![after_ok(root)], None) - .expect("on_ok"); - let on_fail = s - .append( - "on_fail", - vec![Dep::Node { - id: root, - when: DepWhen::of(&[TerminalState::Failed]), - }], - None, - ) - .expect("on_fail"); - assert_eq!(s.settle(), vec![root]); - s.complete(root, Outcome::Failed("boom".to_owned())); - assert_eq!(s.graph().node(on_ok).unwrap().state, State::Skipped); - assert_eq!(s.settle(), vec![on_fail]); - } - - /// A weak edge accepts a dependency that was *ruled out*, so a tail still - /// runs when the work it reports on never happened — held by the edge itself - /// rather than by any node-kind special case. - #[test] - fn eliminated_dep_still_satisfies_a_weak_edge() { - let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new()); - let root = s.append("root", vec![], None).expect("root"); - let mid = s.append("mid", vec![after_ok(root)], None).expect("mid"); - let tail = s - .append( - "tail", - vec![Dep::Node { - id: mid, - when: DepWhen::AFTER_ANY, - }], - None, - ) - .expect("tail"); - assert_eq!(s.settle(), vec![root]); - s.complete(root, Outcome::Failed("boom".to_owned())); - assert_eq!(s.graph().node(mid).unwrap().state, State::Skipped); - assert_eq!( - s.settle(), - vec![tail], - "the tail runs off a cancelled dependency" - ); - } - - /// Edges are **conjunctive**, so "any of these several nodes failed" is not - /// directly expressible — a `{Failed}` edge on each would mean *all* failed. - /// The composition that does work: the success branch depends `AFTER_OK` on - /// every node (so it runs only if all succeeded, and is ruled out the moment - /// one doesn't), and the failure branch hangs off *it* with `{Skipped}` — - /// "run when the success branch was ruled out". The success branch is the - /// aggregator, and exactly one of the two runs. - #[test] - fn ok_branch_aggregates_and_failure_branch_hangs_off_its_elimination() { - let build = || { - let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new()); - let a = s.append("a", vec![], None).expect("a"); - let b = s.append("b", vec![], None).expect("b"); - let on_ok = s - .append("on_ok", vec![after_ok(a), after_ok(b)], None) - .expect("on_ok"); - let on_fail = s - .append( - "on_fail", - vec![Dep::Node { - id: on_ok, - when: DepWhen::of(&[TerminalState::Skipped]), - }], - None, - ) - .expect("on_fail"); - (s, a, b, on_ok, on_fail) - }; - - // Everything succeeds: the ok branch runs, the failure branch is ruled out. - let (mut s, a, b, on_ok, on_fail) = build(); - assert_eq!(s.settle(), vec![a, b], "both roots start; neither tail can"); - s.complete(a, Outcome::Done); - s.complete(b, Outcome::Done); - assert_eq!(s.settle(), vec![on_ok]); - s.complete(on_ok, Outcome::Done); - assert_eq!(s.graph().node(on_fail).unwrap().state, State::Skipped); - assert!(s.settle().is_empty()); - - // One of them fails: the ok branch is ruled out, which is precisely the - // signal the failure branch waits on. - let (mut s, a, b, on_ok, on_fail) = build(); - assert_eq!(s.settle(), vec![a, b]); - s.complete(a, Outcome::Failed("boom".to_owned())); - s.complete(b, Outcome::Done); - assert_eq!(s.graph().node(on_ok).unwrap().state, State::Skipped); - assert_eq!(s.settle(), vec![on_fail]); - } - #[test] fn failed_parent_cancels_its_pending_children() { // A failed group node cancels its sub-nodes (they were gated from ever @@ -905,8 +749,8 @@ mod tests { let grandchild = s.append("gc", vec![], Some(child)).expect("gc"); assert_eq!(s.settle(), vec![root]); s.complete(root, Outcome::Failed(String::new())); - assert_eq!(s.graph().node(child).unwrap().state, State::Skipped); - assert_eq!(s.graph().node(grandchild).unwrap().state, State::Skipped); + assert_eq!(s.graph().node(child).unwrap().state, State::Cancelled); + assert_eq!(s.graph().node(grandchild).unwrap().state, State::Cancelled); } #[test] @@ -915,9 +759,8 @@ mod tests { let a = s.append("a", vec![], None).expect("a"); let b = s.append("b", vec![after_ok(a)], None).expect("b"); assert!(s.cancel_node(a)); - // `a` was dropped by the caller; `b` was merely ruled out by its edge. assert_eq!(s.graph().node(a).unwrap().state, State::Cancelled); - assert_eq!(s.graph().node(b).unwrap().state, State::Skipped); + assert_eq!(s.graph().node(b).unwrap().state, State::Cancelled); let c = s.append("c", vec![], None).expect("c"); assert_eq!(s.settle(), vec![c]); assert!(!s.cancel_node(c));