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/Cargo.toml b/Cargo.toml index 9cd6ab8c..12ea54ef 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -51,6 +51,7 @@ 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 d00d9227..4dadeace 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -507,20 +507,18 @@ 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. /// -/// `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. +/// `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. /// /// [`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, - state: crate::job_queue::State, + outcome: crate::job_queue::TerminalState, error: Option<&str>, ) { - use crate::job_queue::State; + use crate::job_queue::TerminalState; let approval = match coord.approvals.get(approval_id) { Ok(Some(a)) => a, Ok(None) => { @@ -532,10 +530,14 @@ pub(crate) async fn resolve_approval_dag( return; } }; - 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 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 mut terminal_tag = None; match approval.kind { @@ -551,7 +553,7 @@ pub(crate) async fn resolve_approval_dag( } } ApprovalKind::MergeConfigPr => { - terminal_tag = deploy_terminal_tag(approval.agent.as_str(), approval_id, state).await; + terminal_tag = deploy_terminal_tag(approval.agent.as_str(), approval_id, outcome).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 @@ -577,13 +579,14 @@ pub(crate) async fn resolve_approval_dag( async fn deploy_terminal_tag( agent: &str, approval_id: i64, - state: crate::job_queue::State, + outcome: crate::job_queue::TerminalState, ) -> Option { - use crate::job_queue::State; - let candidate = match state { - State::Done => format!("deployed/{approval_id}"), - State::Cancelled => return None, - _ => format!("failed/{approval_id}"), + 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}"), }; 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 21ea0728..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, State}; +use hive_jobq::TerminalState; + +use super::model::{NodeKind, NodeSpec}; use crate::coordinator::Coordinator; use crate::power::{ReconcileAction, reconcile_action}; @@ -95,10 +97,11 @@ 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, .. } => { - run_resolve_approval(coord, claim, *approval_id).await - } - NodeKind::EmitRebuilt { .. } => Ok(run_emit_rebuilt(coord, claim)), + 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::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 @@ -107,58 +110,39 @@ pub(super) async fn run_node(coord: &Arc, claim: &Claim) -> Result< } } -/// 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. +/// 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. async fn run_resolve_approval( coord: &Arc, claim: &Claim, approval_id: i64, + outcome: TerminalState, ) -> Result { - let reason = failure_reason(coord, claim); - crate::actions::resolve_approval_dag(coord, approval_id, claim.deps_state(), reason.as_deref()) - .await; + 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; Ok(NodeOutput::default()) } -/// 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, - }), - _ => {} - } +/// 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, + }); NodeOutput::default() } diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 2a3b4e7a..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 @@ -44,12 +43,13 @@ 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, DepWhen as JobDepWhen, Graph, NodeId, State as JobState}; +use hive_jobq::{Dep, Graph, NodeId, State as JobState}; use hive_sh4re::wire_time::now_unix; use tokio::sync::Notify; use crate::coordinator::TransientKind; -pub use model::{DagSpec, DagView, DepWhen, NodeKind, NodeSpec, PermPayload, Source, State}; +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 @@ -60,31 +60,6 @@ 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)] @@ -101,38 +76,6 @@ 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 @@ -190,23 +133,21 @@ 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 => State::Cancelled, + JobState::Cancelled | JobState::Skipped => State::Cancelled, } } @@ -241,7 +182,7 @@ fn insert_group( for d in &ns.deps { deps.push(Dep::Node { id: ids[dep_index(d.on)], - when: to_crate_when(d.when), + when: d.when, }); } let parent = match ns.parent { @@ -373,24 +314,6 @@ 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; }; @@ -405,7 +328,6 @@ 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. @@ -438,17 +360,17 @@ impl JobQueue { /// so each is cancelled. `false` once any work node is running or terminal — /// an in-flight nix build isn't interruptible. /// - /// **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. + /// **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. /// - /// 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. + /// 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. pub fn cancel(&self, dag_id: u64) -> bool { let mut inner = self.lock(); let Some(container) = inner.container(dag_id) else { @@ -466,12 +388,7 @@ impl JobQueue { return false; } for id in work { - if inner - .sched - .graph() - .node(id) - .is_some_and(|n| n.payload.is_tail()) - { + if inner.observes_cancellation(id) { continue; } inner.sched.cancel_node(id); @@ -650,6 +567,18 @@ 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 { @@ -690,7 +619,11 @@ impl QueueInner { if let Some(f) = node.finished_at { finished.push(f); } - if node.state == JobState::Done { + // `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) { continue; } let deps: Vec = node diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 0d321a36..dcddd2f4 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -17,18 +17,7 @@ use serde::Serialize; use crate::coordinator::TransientKind; -/// 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, -} +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). @@ -230,7 +219,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::AfterAny`) like [`NodeKind::DeployTail`], so it runs on + /// Weak-edged (`DepWhen::AFTER_ANY`) 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 @@ -240,18 +229,30 @@ 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. - /// - /// [`Claim::deps`]: super::Claim::deps - ResolveApproval { approval_id: i64 }, + 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, + }, /// 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**, 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 }, + /// 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 }, /// 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 @@ -340,7 +341,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 { .. } @@ -349,20 +350,6 @@ 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 d47be49d..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}; +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 @@ -39,27 +41,124 @@ use crate::coordinator::TransientKind; pub(crate) fn after_ok(on: u64) -> Vec { vec![Dep { on, - when: DepWhen::AfterOk, + when: DepWhen::AFTER_OK, }] } -/// Weak edges onto every one of a DAG's other **group-roots** — how a tail node -/// (`ResolveApproval` / `EmitRebuilt`) sees the whole DAG's outcome. +/// `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. /// /// 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 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`]. +/// 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. pub(crate) fn after_any_all(ons: &[u64]) -> Vec { ons.iter() .map(|&on| Dep { on, - when: DepWhen::AfterAny, + when: DepWhen::AFTER_ANY, }) .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 @@ -135,7 +234,7 @@ pub(crate) fn rebuild_nodes(agent: &str, relock: bool, base: u64) -> Vec Vec { vec![ Dep { on: 1, - when: DepWhen::AfterOk, + when: DepWhen::AFTER_OK, }, Dep { on: 5, - when: DepWhen::AfterOk, + when: DepWhen::AFTER_OK, }, ], )); @@ -198,12 +297,7 @@ 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.push(node( - NodeKind::EmitRebuilt { - agent: agent.to_owned(), - }, - after_any_all(&[0, 1, 5]), - )); + nodes.extend(emit_rebuilt_tails(agent, &[0, 1, 5], 6)); DagSpec { source, reason, @@ -256,14 +350,13 @@ pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec NodeKind::DeployTail { agent: a() }, vec![Dep { on: 2, - when: DepWhen::AfterAny, + when: DepWhen::AFTER_ANY, }], ), - node( - NodeKind::ResolveApproval { approval_id }, - after_any_all(&[0]), - ), - ], + ] + .into_iter() + .chain(resolve_approval_tails(approval_id, 0)) + .collect(), } } @@ -319,11 +412,10 @@ 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() }, } } @@ -342,12 +434,7 @@ pub fn perm_change(agent: &str, source: Source, reason: String, payload: PermPay Vec::new(), )]; nodes.extend(rebuild_nodes(agent, true, 1)); - nodes.push(node( - NodeKind::EmitRebuilt { - agent: agent.to_owned(), - }, - after_any_all(&[0, 1, 2, 6]), - )); + nodes.extend(emit_rebuilt_tails(agent, &[0, 1, 2, 6], 7)); DagSpec { source, reason, @@ -382,14 +469,11 @@ 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 a - // tail 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 the + // per-outcome tails 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.push(node( - NodeKind::ResolveApproval { approval_id }, - after_any_all(&[0]), - )); + nodes.extend(resolve_approval_tails(approval_id, 0)); } DagSpec { source, diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 40251fd7..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 { @@ -48,35 +50,34 @@ fn claim_one(q: &JobQueue) -> Claim { claims.pop().expect("one claim") } -/// 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) { +/// 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) { let tail = claim_one(q); assert!( - matches!(tail.kind, NodeKind::ResolveApproval { approval_id: got } if got == approval_id), - "expected the ResolveApproval tail for #{approval_id}, got {:?}", + matches!( + tail.kind, + NodeKind::ResolveApproval { approval_id: got, outcome } + if got == approval_id && outcome == expect + ), + "expected the {expect:?} 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 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) { +/// 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) { let tail = claim_one(q); - 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`" + 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 ); q.complete_node(dag_id, tail.node_id, Ok(())); } @@ -154,7 +155,7 @@ fn cyclic_dag_is_rejected_at_submit() { }, deps: vec![Dep { on: 1, - when: DepWhen::AfterOk, + when: DepWhen::AFTER_OK, }], parent: None, }, @@ -164,7 +165,7 @@ fn cyclic_dag_is_rejected_at_submit() { }, deps: vec![Dep { on: 0, - when: DepWhen::AfterOk, + when: DepWhen::AFTER_OK, }], parent: None, }, @@ -183,7 +184,7 @@ fn unknown_dep_is_rejected_at_submit() { }, deps: vec![Dep { on: 9, - when: DepWhen::AfterOk, + when: DepWhen::AFTER_OK, }], parent: None, }]; @@ -229,7 +230,7 @@ fn rebuild_chain_claims_in_dep_order() { ); q.complete_node(id, c.node_id, Ok(())); } - settle_rebuild_tail(&q, id, "agent-a", State::Done); + settle_rebuild_tail(&q, id, "agent-a", true); assert_eq!(state_of(&q, id), State::Done); } @@ -736,73 +737,6 @@ 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] @@ -831,9 +765,19 @@ fn failed_node_cancels_downstream_but_afterany_reconcile_runs() { .state }; assert_eq!(by_kind("prebuild"), State::Failed); - assert_eq!(by_kind("stop_for_update"), State::Cancelled); - assert_eq!(by_kind("swap"), State::Cancelled); - assert_eq!(by_kind("post_swap"), State::Cancelled); + // `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" + ); + } // 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!( @@ -872,14 +816,18 @@ 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 == "post_swap") - .expect("post_swap node") + .find(|n| n.kind == "swap") + .expect("swap node") .state, - State::Cancelled, - "PostSwap must cancel-cascade when Swap fails" + State::Failed, + "and the failure that ruled it out is still on the wire" ); assert_eq!(state_of(&q, id), State::Failed); } @@ -911,7 +859,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", State::Done); + settle_rebuild_tail(&q, id, "agent-a", true); assert_eq!(state_of(&q, id), State::Done); } @@ -939,18 +887,14 @@ 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"); - // 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`" + // 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" ); - q.complete_node(id, tail.node_id, Ok(())); assert_eq!(state_of(&q, id), State::Cancelled); - assert!(q.claim_ready().is_empty()); } #[test] @@ -1065,21 +1009,10 @@ 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"); - 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(())); + // 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); 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")); @@ -1130,7 +1063,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, State::Failed); + settle_approval_tail(&q, id, 7, TerminalState::Failed); assert_eq!( state_of(&q, id), State::Failed, @@ -1203,7 +1136,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, State::Done); + settle_approval_tail(&q, id, 11, TerminalState::Done); assert_eq!(state_of(&q, id), State::Done); } @@ -1254,7 +1187,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, State::Failed); + settle_approval_tail(&q, id, 13, TerminalState::Failed); assert_eq!(state_of(&q, id), State::Failed); assert_eq!( q.first_error(id).as_deref(), @@ -1292,7 +1225,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, State::Failed); + settle_approval_tail(&q, id, 9, TerminalState::Failed); assert_eq!(state_of(&q, id), State::Failed); } @@ -1432,7 +1365,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, State::Done); + settle_approval_tail(&q, id, 7, TerminalState::Done); assert_eq!(state_of(&q, id), State::Done); } @@ -1464,7 +1397,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", State::Done); + settle_rebuild_tail(&q, id, "agent-a", true); assert_eq!(state_of(&q, id), State::Done); } diff --git a/hive-jobq/Cargo.toml b/hive-jobq/Cargo.toml index aa5370f7..4fc70e0a 100644 --- a/hive-jobq/Cargo.toml +++ b/hive-jobq/Cargo.toml @@ -9,6 +9,7 @@ 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 5963f042..60e5b6cc 100644 --- a/hive-jobq/src/lib.rs +++ b/hive-jobq/src/lib.rs @@ -56,29 +56,122 @@ impl NodeId { } } -/// When a [`Dep::Node`] edge is satisfied — the strong/weak distinction the -/// current queue carries as `DepWhen`, load-bearing for failure safety. +/// 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)] -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, +#[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())) + } } impl DepWhen { - /// Whether a dependency in `dep_state` satisfies this edge. + /// 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. #[must_use] pub fn satisfied_by(self, dep_state: State) -> bool { - match self { - DepWhen::AfterOk => dep_state == State::Done, - DepWhen::AfterAny => dep_state.is_terminal(), - } + dep_state.terminal().is_some_and(|t| self.accepts(t)) } } @@ -128,18 +221,24 @@ pub enum State { Done, /// Completed unsuccessfully — own logic failed, or a sub-node did. Failed, - /// Never ran: an `AfterOk` dependency failed, so this node (and the rest of - /// its strong-dependent chain) is cancelled rather than run. + /// Never ran: the work was dropped while still queued. See + /// [`TerminalState::Cancelled`]. 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, - /// or cancelled — which is when its resources are released and dependents - /// are re-evaluated. + /// dropped, or ruled out — 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) + matches!( + self, + State::Done | State::Failed | State::Cancelled | State::Skipped + ) } } @@ -207,6 +306,16 @@ 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 @@ -427,10 +536,16 @@ impl Graph { return Err(GraphError::UnknownParent(p)); } for dep in &node.deps { - if let Dep::Node { id, .. } = dep { + if let Dep::Node { id, when } = 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, @@ -465,7 +580,7 @@ mod tests { "update", vec![Dep::Node { id: a, - when: DepWhen::AfterOk, + when: DepWhen::AFTER_OK, }], None, ) @@ -494,19 +609,25 @@ 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::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)); + 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)); // Finishing satisfies neither — a dependent waits until the node rolls // up to a terminal state (all its sub-nodes done). - assert!(!DepWhen::AfterOk.satisfied_by(State::Finishing)); - assert!(!DepWhen::AfterAny.satisfied_by(State::Finishing)); + assert!(!DepWhen::AFTER_OK.satisfied_by(State::Finishing)); + assert!(!DepWhen::AFTER_ANY.satisfied_by(State::Finishing)); } #[test] @@ -515,7 +636,7 @@ mod tests { let bogus = NodeId(42); let deps = vec![Dep::Node { id: bogus, - when: DepWhen::AfterOk, + when: DepWhen::AFTER_OK, }]; assert_eq!( g.insert("x", deps, None).unwrap_err(), @@ -547,7 +668,7 @@ mod tests { "b".to_owned(), vec![Dep::Node { id: a, - when: DepWhen::AfterAny, + when: DepWhen::AFTER_ANY, }], None, ) @@ -568,7 +689,7 @@ mod tests { // roll-up model — the parent stays `Finishing` awaiting its children). let on_parent = vec![Dep::Node { id: root, - when: DepWhen::AfterOk, + when: DepWhen::AFTER_OK, }]; assert_eq!( g.insert("child", on_parent, Some(root)).unwrap_err(), @@ -598,7 +719,7 @@ mod tests { fn after_ok_dep(on: NodeId) -> Dep { Dep::Node { id: on, - when: DepWhen::AfterOk, + when: DepWhen::AFTER_OK, } } @@ -613,7 +734,7 @@ mod tests { payload: "x".to_owned(), deps: vec![Dep::Node { id: NodeId(99), - when: DepWhen::AfterOk, + when: DepWhen::AFTER_OK, }], state: State::Pending, started_at: None, diff --git a/hive-jobq/src/scheduler.rs b/hive-jobq/src/scheduler.rs index 5cd80975..f18437d5 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, DepWhen, Graph, GraphError, NodeId, State}; +use crate::{Dep, Graph, GraphError, NodeId, State}; /// The result of a node's own execution, reported to [`Scheduler::complete`]. /// @@ -235,6 +235,9 @@ 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() @@ -244,6 +247,12 @@ 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 @@ -253,7 +262,7 @@ impl Scheduler { State::Done }; self.graph.set_state(id, state); - if state == State::Failed { + if state.is_terminal() { self.cascade_cancel(id); } } @@ -276,14 +285,13 @@ impl Scheduler { State::Done }; self.graph.set_state(a, state); - if state == State::Failed { - self.cascade_cancel(a); - } + // Any terminal outcome can rule a dependent out — see `settle_terminal`. + self.cascade_cancel(a); cur = self.graph.node(a).and_then(|n| n.parent); } } - /// Cancel a still-*pending* node (and cascade to its `AfterOk` dependents): + /// Cancel a still-*pending* node (and cascade to the dependents it rules out): /// 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 @@ -316,16 +324,33 @@ impl Scheduler { .collect() } - /// 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. + /// 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. 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() @@ -333,13 +358,13 @@ impl Scheduler { n.state == State::Pending && (n.parent == Some(cur) || n.deps.iter().any(|d| { - matches!(d, Dep::Node { id, when: DepWhen::AfterOk } if *id == cur) + matches!(d, Dep::Node { id, when } if *id == cur && !when.accepts(outcome)) })) }) .map(|n| n.id) .collect(); for d in doomed { - self.graph.set_state(d, State::Cancelled); + self.graph.set_state(d, State::Skipped); stack.push(d); } } @@ -413,6 +438,7 @@ impl Scheduler { #[cfg(test)] mod tests { use super::*; + use crate::{DepWhen, TerminalState}; fn res(name: &str) -> String { name.to_owned() @@ -436,7 +462,7 @@ mod tests { fn after_ok(on: NodeId) -> Dep { Dep::Node { id: on, - when: DepWhen::AfterOk, + when: DepWhen::AFTER_OK, } } @@ -489,10 +515,10 @@ mod tests { assert_eq!(n.error.as_deref(), Some("boom")); assert!(n.finished_at.is_some()); - // The `AfterOk`-cancelled node: cascade-cancelled, so finished_at is set, + // The `AFTER_OK` dependent: ruled out by its edge, 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::Cancelled); + assert_eq!(n.state, State::Skipped); assert!(n.started_at.is_none()); assert!(n.finished_at.is_some()); assert_eq!(n.error, None); @@ -727,18 +753,148 @@ mod tests { "weak", vec![Dep::Node { id: root, - when: DepWhen::AfterAny, + when: DepWhen::AFTER_ANY, }], 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::Cancelled); - assert_eq!(s.graph().node(strong2).unwrap().state, State::Cancelled); + assert_eq!(s.graph().node(strong1).unwrap().state, State::Skipped); + assert_eq!(s.graph().node(strong2).unwrap().state, State::Skipped); 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 @@ -749,8 +905,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::Cancelled); - assert_eq!(s.graph().node(grandchild).unwrap().state, State::Cancelled); + assert_eq!(s.graph().node(child).unwrap().state, State::Skipped); + assert_eq!(s.graph().node(grandchild).unwrap().state, State::Skipped); } #[test] @@ -759,8 +915,9 @@ 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::Cancelled); + assert_eq!(s.graph().node(b).unwrap().state, State::Skipped); let c = s.append("c", vec![], None).expect("c"); assert_eq!(s.settle(), vec![c]); assert!(!s.cancel_node(c));