src/build/pipeline.rs
//! **f5 — a build/pipeline is a root jera job whose body is a DAG of sub-jobs.**
//!
//! A build is not one job: it is a **root** [`crate::jobs`] job (kind
//! [`crate::jobs::kind::CI_BUILD`]) that **owns a [`PlanDag`] of sub-jobs**, and
//! **every pipeline step is a tracked sub-job** in its own right — its own
//! `job_id`, its own `queued → running → done|failed` lifecycle on the ledger, and
//! (via slice 2) its own build-info sub-stream filtered by `step_id`. Edges are
//! `needs` dependencies (a step runs only once every step it needs is `done`).
//!
//! ## The DAG is data — reuse, don't re-roll
//!
//! The graph vocabulary is **funnel-core's [`PlanDag`]** (`topo_order` /
//! `critical_path` / `depths` = the ready-wave layering). We do NOT invent a graph
//! type: a [`PipelineSpec`] projects onto a `PlanDag` ([`PipelineSpec::to_plan_dag`])
//! where **nodes = sub-jobs** and **edges = `needs` dependencies** ([`EdgeKind::Dep`]),
//! so `critical_path()` gives the bottleneck lineage and `depths()` the scheduling
//! wave for free.
//!
//! ## The ledger is the sub-job hierarchy — reuse, don't re-roll
//!
//! The lifecycle rides the **nornir-jobs** ledger ([`JobStore`]): the root is a
//! normal top-level job; each step is a **child** [`JobRecord`] (`parent_id` = the
//! root's `job_id`), admitted `queued` up front (the root literally *owns* the DAG
//! of sub-jobs), promoted `running` when its wave is ready, then `done|failed`. A
//! step whose `needs` did not all reach `done` is left blocked → `failed` without
//! running. The viz Jobs tree already folds children under their parent by
//! `parent_id`, so no viz change is needed to see the DAG per-step state.
//!
//! ## One engine for nornir AND dwarves
//!
//! This is the single `root job → pipeline DAG → sub-jobs` model the design (§f5)
//! calls for; dwarves' pipeline steps and nornir's build waves are the same shape,
//! not two schedulers.
//!
//! ## Additive
//!
//! The existing single-job build path is unchanged — it is simply a one-step
//! "DAG" (one node, zero edges, one sub-job), see
//! [`tests::single_step_is_a_one_node_dag_additive`].
use std::collections::BTreeMap;
use anyhow::{Result, bail};
use serde::{Deserialize, Serialize};
use crate::funnel::{DagStatus, EdgeKind, PlanDag, PlanDagEdge, PlanDagNode};
use crate::jobs::{JobRecord, JobStore, kind, status};
use nornir_build_thing::build_info::BuildInfoRow;
/// One declarative pipeline step = one sub-job to run + the step ids it `needs`.
///
/// `id` is the **logical, author-given** step id (stable across runs); it is what
/// [`PlanDag`] edges reference and what a caller passes to
/// [`PipelineOutcome::build_info_for`]. It is distinct from the ledger `job_id` the
/// run mints for the sub-job (a fresh uuid) — the build-info stream's `step_id`
/// column carries that `job_id` (see [`PipelineOutcome::job_id_for`]).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PipelineStep {
/// The stable logical step id (the [`PlanDag`] node id / edge endpoint).
pub id: String,
/// A human title for the sub-job (the ledger `target`).
pub name: String,
/// The step ids this step depends on — it runs only once all of them are `done`.
#[serde(default)]
pub needs: Vec<String>,
/// The ledger [`crate::jobs::kind`] for this sub-job; empty → [`kind::CI_BUILD`].
#[serde(default)]
pub kind: String,
}
impl PipelineStep {
/// A step `id` titled `name` that `needs` the given step ids (kind defaults to
/// [`kind::CI_BUILD`]).
pub fn new(id: &str, name: &str, needs: &[&str]) -> Self {
PipelineStep {
id: id.to_string(),
name: name.to_string(),
needs: needs.iter().map(|s| s.to_string()).collect(),
kind: String::new(),
}
}
/// The effective ledger kind for this sub-job.
fn job_kind(&self) -> &str {
if self.kind.is_empty() {
kind::CI_BUILD
} else {
&self.kind
}
}
}
/// A declarative build **pipeline**: a DAG of steps. The DAG *is data* — it
/// projects onto a funnel-core [`PlanDag`] ([`Self::to_plan_dag`]) for all graph
/// reasoning and runs onto the job ledger ([`Self::run_on_ledger`]) as a root job
/// owning one sub-job per step.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PipelineSpec {
/// The pipeline name (the root job's `target` + the DAG's `sel_plan` scope).
pub name: String,
/// The steps, in any order — dependencies come from `needs`, not list position.
pub steps: Vec<PipelineStep>,
}
impl PipelineSpec {
/// Project this spec onto a funnel-core [`PlanDag`] — the DAG vocabulary the
/// whole slice reuses. **Nodes = sub-jobs** (one per step), **edges = `needs`
/// dependencies** ([`EdgeKind::Dep`], `need → step`). All steps start
/// [`DagStatus::Idle`] (i.e. queued); `fanout` is the number of steps that need
/// this one. `critical_path()` / `depths()` / `topo_order()` then work verbatim.
#[must_use]
pub fn to_plan_dag(&self) -> PlanDag {
// fan-out = how many steps depend ON this step.
let mut needed_by: BTreeMap<&str, u64> = BTreeMap::new();
for s in &self.steps {
for n in &s.needs {
*needed_by.entry(n.as_str()).or_default() += 1;
}
}
let nodes: Vec<PlanDagNode> = self
.steps
.iter()
.map(|s| PlanDagNode {
id: s.id.clone(),
title: s.name.clone(),
group: Some(self.name.clone()),
status: DagStatus::Idle,
is_bug: false,
staleness_hours: 0.0,
weight_bytes: 0,
fanout: needed_by.get(s.id.as_str()).copied().unwrap_or(0),
})
.collect();
let mut edges: Vec<PlanDagEdge> = Vec::new();
for s in &self.steps {
for n in &s.needs {
edges.push(PlanDagEdge {
from: n.clone(),
to: s.id.clone(),
kind: EdgeKind::Dep,
});
}
}
edges.sort_by(|a, b| (&a.from, &a.to).cmp(&(&b.from, &b.to)));
PlanDag {
nodes,
edges,
sel_plan: Some(self.name.clone()),
}
}
/// Validate the spec: non-empty, unique step ids, every `needs` resolves, and
/// the graph is acyclic (a build pipeline must be a DAG).
pub fn validate(&self) -> Result<()> {
if self.steps.is_empty() {
bail!("pipeline `{}` has no steps", self.name);
}
let mut seen: BTreeMap<&str, ()> = BTreeMap::new();
for s in &self.steps {
if seen.insert(s.id.as_str(), ()).is_some() {
bail!(
"pipeline `{}` has a duplicate step id `{}`",
self.name,
s.id
);
}
}
for s in &self.steps {
for n in &s.needs {
if !seen.contains_key(n.as_str()) {
bail!("step `{}` needs unknown step `{}`", s.id, n);
}
}
}
self.to_plan_dag().topo_order().map_err(|c| {
anyhow::anyhow!(
"pipeline `{}` has a dependency cycle around {c:?}",
self.name
)
})?;
Ok(())
}
/// **Materialize + run** this pipeline on the ledger: a root [`kind::CI_BUILD`]
/// job that OWNS a [`PlanDag`] of sub-jobs. Each step becomes a child sub-job
/// with its own `job_id` and a `queued → running → done|failed` lifecycle;
/// steps run in dependency (ready-wave) order, and a step whose `needs` did not
/// all reach `done` is left blocked and marked `failed` without running.
///
/// `run_step` is the executor seam: `Ok(())` = the step succeeded (`done`),
/// `Err` = it failed (`failed`, error carried into the sub-job's `detail_json`).
/// The build itself is not run here — this is the DAG/ledger wiring; a real
/// caller passes a closure that drives `build_variant_with_info` (slice 2) per
/// step, tagging its build-info rows with the sub-job's `job_id` as `step_id`.
///
/// Returns the [`PipelineOutcome`]: the root id, the projected DAG, and every
/// sub-job's `(step_id, job_id, status)` — all robot-observable via
/// [`PipelineOutcome::state_json`].
pub fn run_on_ledger(
&self,
store: &JobStore,
workspace: &str,
mut run_step: impl FnMut(&PipelineStep) -> Result<()>,
) -> Result<PipelineOutcome> {
self.validate()?;
let dag = self.to_plan_dag();
// Acyclic guaranteed by validate(); topo order = dependency-before-dependent.
let order = dag
.topo_order()
.map_err(|c| anyhow::anyhow!("pipeline `{}` cycle around {c:?}", self.name))?;
// The root job OWNS the DAG.
let root = store.start(
kind::CI_BUILD,
&self.name,
workspace,
serde_json::json!({
"pipeline": self.name,
"steps": self.steps.len(),
"critical_path": dag.critical_path(),
"critical_path_len": dag.critical_path_len(),
}),
);
let root_id = root.job_id().to_string();
// Admit every step as a QUEUED sub-job up front (the root owns the whole DAG
// of sub-jobs; they light up wave by wave).
let mut job_id_of: BTreeMap<String, String> = BTreeMap::new();
let mut rec_of: BTreeMap<String, JobRecord> = BTreeMap::new();
for s in &self.steps {
let rec = new_child_record(&root_id, s, workspace, status::QUEUED);
store.submit(&rec)?;
job_id_of.insert(s.id.clone(), rec.job_id.clone());
rec_of.insert(s.id.clone(), rec);
}
// Ready-wave execution: topo order visits dependencies before dependents,
// so by the time we reach a step every `need` has a final status.
let step_by_id: BTreeMap<&str, &PipelineStep> =
self.steps.iter().map(|s| (s.id.as_str(), s)).collect();
let mut final_status: BTreeMap<String, String> = BTreeMap::new();
for sid in &order {
let Some(step) = step_by_id.get(sid.as_str()) else {
continue;
};
let rec = rec_of.get_mut(sid).expect("every step has a queued record");
// Blocked if any need did not reach `done`.
let blocked = step.needs.iter().any(|n| {
final_status
.get(n)
.map(|st| st != status::DONE)
.unwrap_or(true)
});
if blocked {
let blockers: Vec<&String> = step
.needs
.iter()
.filter(|n| {
final_status
.get(n.as_str())
.map(|st| st != status::DONE)
.unwrap_or(true)
})
.collect();
terminate(
rec,
status::FAILED,
serde_json::json!({ "step_id": step.id, "needs": step.needs, "blocked_by": blockers }),
);
store.submit(rec)?;
final_status.insert(sid.clone(), status::FAILED.to_string());
continue;
}
// queued → running.
promote_running(rec);
store.submit(rec)?;
let st = match run_step(step) {
Ok(()) => {
terminate(
rec,
status::DONE,
serde_json::json!({ "step_id": step.id, "needs": step.needs }),
);
status::DONE
}
Err(e) => {
terminate(
rec,
status::FAILED,
serde_json::json!({ "step_id": step.id, "needs": step.needs, "error": format!("{e:#}") }),
);
status::FAILED
}
};
store.submit(rec)?;
final_status.insert(sid.clone(), st.to_string());
}
let sub_jobs: Vec<SubJob> = self
.steps
.iter()
.map(|s| SubJob {
step_id: s.id.clone(),
job_id: job_id_of[&s.id].clone(),
status: final_status
.get(&s.id)
.cloned()
.unwrap_or_else(|| status::QUEUED.to_string()),
needs: s.needs.clone(),
})
.collect();
let ok = sub_jobs.iter().all(|s| s.status == status::DONE);
if ok {
root.finish(
serde_json::json!({ "steps": self.steps.len(), "outcome": "done" }),
"",
);
} else {
let failed: Vec<&String> = sub_jobs
.iter()
.filter(|s| s.status != status::DONE)
.map(|s| &s.step_id)
.collect();
root.fail_with_detail(serde_json::json!({ "failed_steps": failed }));
}
Ok(PipelineOutcome {
root_id,
dag,
sub_jobs,
ok,
})
}
}
/// One sub-job's recorded outcome — the tie between a logical `step_id`, its ledger
/// `job_id`, and its final lifecycle `status`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SubJob {
/// The logical, author-given step id ([`PipelineStep::id`]).
pub step_id: String,
/// The ledger `job_id` minted for this sub-job — the build-info stream tags its
/// rows' `step_id` column with THIS value (see [`PipelineOutcome::build_info_for`]).
pub job_id: String,
/// The final ledger [`crate::jobs::status`] (`done` / `failed`; `queued` only if
/// the run never reached it).
pub status: String,
/// The step ids this sub-job depended on (the `needs` edges).
pub needs: Vec<String>,
}
/// The result of a [`PipelineSpec::run_on_ledger`]: the root job id, the projected
/// [`PlanDag`], and every sub-job.
#[derive(Debug, Clone)]
pub struct PipelineOutcome {
/// The root job's `job_id` — the build-info stream's `build_id`.
pub root_id: String,
/// The DAG that was owned/run (funnel-core [`PlanDag`]).
pub dag: PlanDag,
/// One entry per step, in spec order.
pub sub_jobs: Vec<SubJob>,
/// Whether every step reached `done`.
pub ok: bool,
}
impl PipelineOutcome {
/// The ledger `job_id` a logical `step_id` was run as — i.e. the value the
/// build-info stream carries in its `step_id` column for that sub-job.
#[must_use]
pub fn job_id_for(&self, step_id: &str) -> Option<&str> {
self.sub_jobs
.iter()
.find(|s| s.step_id == step_id)
.map(|s| s.job_id.as_str())
}
/// **Filter a slice-2 build-info stream to ONE sub-job.** Returns the rows whose
/// `build_id` is this run's root AND whose `step_id` is the sub-job's ledger
/// `job_id` — the per-step feed the design (§f5) calls for ("each step's
/// log/metric stream is the f3 feed filtered by `step_id`"). Unknown `step_id`
/// → empty.
#[must_use]
pub fn build_info_for<'a>(
&self,
rows: &'a [BuildInfoRow],
step_id: &str,
) -> Vec<&'a BuildInfoRow> {
let Some(job_id) = self.job_id_for(step_id) else {
return Vec::new();
};
rows.iter()
.filter(|r| r.build_id == self.root_id && r.step_id == job_id)
.collect()
}
/// The robot-observable introspection blob (LAW 6): the root id, the DAG
/// vocabulary (critical path etc.), and every sub-job's `(step_id, job_id,
/// status, needs)`.
#[must_use]
pub fn state_json(&self) -> serde_json::Value {
serde_json::json!({
"root_id": self.root_id,
"ok": self.ok,
"sub_job_count": self.sub_jobs.len(),
"critical_path": self.dag.critical_path(),
"critical_path_len": self.dag.critical_path_len(),
"dep_edges": self.dag.dep_edges(),
"sub_jobs": self.sub_jobs.iter().map(|s| serde_json::json!({
"step_id": s.step_id,
"job_id": s.job_id,
"status": s.status,
"needs": s.needs,
})).collect::<Vec<_>>(),
})
}
}
fn now_micros() -> i64 {
chrono::Utc::now().timestamp_micros()
}
/// A fresh child [`JobRecord`] for `step` under `root_id`, in `initial_status`. The
/// `needs` edges are stashed into `detail_json` so the ledger row itself carries the
/// dependency set (the DAG is recoverable from the ledger, not only the projection).
fn new_child_record(
root_id: &str,
step: &PipelineStep,
ws: &str,
initial_status: &str,
) -> JobRecord {
JobRecord {
job_id: uuid::Uuid::new_v4().to_string(),
kind: step.job_kind().to_string(),
target: step.name.clone(),
workspace: ws.to_string(),
status: initial_status.to_string(),
ts_start_micros: now_micros(),
ts_end_micros: None,
elapsed_ms: None,
detail_json: serde_json::json!({ "step_id": step.id, "needs": step.needs }).to_string(),
result_ref: String::new(),
parent_id: Some(root_id.to_string()),
node: None,
phase: None,
actor: None,
actor_method: None,
}
}
/// Promote a queued record to `running`, re-stamping `ts_start` so `elapsed_ms`
/// measures actual work, not the queue wait (mirrors the ledger's own scheduler).
fn promote_running(rec: &mut JobRecord) {
rec.status = status::RUNNING.to_string();
rec.ts_start_micros = now_micros();
}
/// Write a terminal status onto a record (mirrors [`crate::jobs::JobHandle`]'s
/// private `terminate`, which we cannot reach for a caller-owned record).
fn terminate(rec: &mut JobRecord, st: &str, detail: serde_json::Value) {
let end = now_micros();
rec.status = st.to_string();
rec.ts_end_micros = Some(end);
rec.elapsed_ms = Some((end - rec.ts_start_micros) / 1000);
rec.detail_json = detail.to_string();
}
#[cfg(test)]
mod tests {
use super::*;
use crate::jobs::JobSelector;
use nornir_build_thing::build_info::{BuildInfoKind, BuildInfoRow};
fn tmpdir(tag: &str) -> std::path::PathBuf {
std::env::temp_dir().join(format!(
"nornir-pipeline-{tag}-{}-{}",
std::process::id(),
now_micros()
))
}
/// The canonical 3-node build DAG: fetch → compile → test.
fn three_node() -> PipelineSpec {
PipelineSpec {
name: "build".into(),
steps: vec![
PipelineStep::new("fetch", "fetch sources", &[]),
PipelineStep::new("compile", "cargo build", &["fetch"]),
PipelineStep::new("test", "cargo test", &["compile"]),
],
}
}
/// A build-info row tagged for a given `(build_id, step_id)`.
fn row(build_id: &str, step_id: &str, kind: BuildInfoKind, name: &str) -> BuildInfoRow {
let mut r = BuildInfoRow::new(build_id, 1, kind);
r.step_id = step_id.to_string();
r.name = name.to_string();
r
}
/// **RED-when-broken (the slice's headline):** a 3-node DAG creates a root +
/// three sub-jobs on the ledger, with the correct `needs` edges and per-step
/// status.
#[test]
fn three_node_dag_makes_root_plus_three_subjobs_with_needs_and_status() {
let dir = tmpdir("3node");
let store = JobStore::open(&dir).unwrap();
let spec = three_node();
// The DAG vocabulary is funnel-core's PlanDag (reuse, not re-rolled).
let dag = spec.to_plan_dag();
assert_eq!(dag.node_count(), 3, "one node per step");
assert_eq!(dag.dep_edges(), 2, "fetch→compile, compile→test");
assert_eq!(dag.critical_path(), vec!["fetch", "compile", "test"]);
assert_eq!(
dag.critical_path_len(),
2,
"two sequential waves define the schedule"
);
let out = spec
.run_on_ledger(&store, "nordisk", |_step| Ok(()))
.unwrap();
assert!(out.ok, "all steps ran green");
let all = store.list(&JobSelector::All).unwrap();
assert_eq!(all.len(), 4, "one root + three sub-jobs on the ledger");
// The root: top-level CI_BUILD, done.
let root = all
.iter()
.find(|r| r.job_id == out.root_id)
.expect("root on ledger");
assert_eq!(root.kind, kind::CI_BUILD);
assert_eq!(root.parent_id, None, "root is top-level");
assert_eq!(root.status, status::DONE);
// Exactly three children, all done, all folded under the root by parent_id.
let children: Vec<&JobRecord> = all
.iter()
.filter(|r| r.parent_id.as_deref() == Some(out.root_id.as_str()))
.collect();
assert_eq!(children.len(), 3, "three sub-jobs under the root");
assert!(
children.iter().all(|r| r.status == status::DONE),
"every sub-job done"
);
// The `needs` edges are recorded on each sub-job's ledger row.
let compile = children
.iter()
.find(|r| r.detail_json.contains("\"step_id\":\"compile\""))
.unwrap();
assert!(
compile.detail_json.contains("fetch"),
"compile's needs edge (fetch) is on the ledger row"
);
let fetch = children
.iter()
.find(|r| r.detail_json.contains("\"step_id\":\"fetch\""))
.unwrap();
assert!(
fetch.detail_json.contains("\"needs\":[]"),
"fetch has no needs"
);
// Per-step status via the outcome.
assert!(out.sub_jobs.iter().all(|s| s.status == status::DONE));
assert_eq!(out.sub_jobs.len(), 3);
nornir_testmatrix::functional_status(
"build-pipeline-subjobs",
"three_node_root_plus_subjobs",
all.len() == 4 && children.len() == 3 && dag.dep_edges() == 2,
"3-node DAG → 1 root + 3 sub-jobs with needs edges + per-step done",
);
std::fs::remove_dir_all(&dir).ok();
}
/// A failed step blocks its dependents and fails the root: compile fails → test
/// is blocked (never runs) → both non-done, root failed. RED-when-broken.
#[test]
fn failed_step_blocks_dependents_and_fails_root() {
let dir = tmpdir("fail");
let store = JobStore::open(&dir).unwrap();
let spec = three_node();
let out = spec
.run_on_ledger(&store, "ws", |s| {
if s.id == "compile" {
bail!("compile boom")
} else {
Ok(())
}
})
.unwrap();
assert!(!out.ok, "a failed step fails the pipeline");
let by = |id: &str| out.sub_jobs.iter().find(|s| s.step_id == id).unwrap();
assert_eq!(by("fetch").status, status::DONE, "fetch ran green");
assert_eq!(by("compile").status, status::FAILED, "compile failed");
assert_eq!(
by("test").status,
status::FAILED,
"test is blocked by the failed compile"
);
// The blocked step records WHY (blocked_by), and the failed step its error.
let all = store.list(&JobSelector::All).unwrap();
let test = all.iter().find(|r| r.job_id == by("test").job_id).unwrap();
assert!(
test.detail_json.contains("blocked_by"),
"test row says it was blocked"
);
assert!(test.detail_json.contains("compile"), "blocked by compile");
let compile = all
.iter()
.find(|r| r.job_id == by("compile").job_id)
.unwrap();
assert!(
compile.detail_json.contains("compile boom"),
"compile carries its error"
);
let root = all.iter().find(|r| r.job_id == out.root_id).unwrap();
assert_eq!(root.status, status::FAILED, "root fails when a step fails");
nornir_testmatrix::functional_status(
"build-pipeline-subjobs",
"failed_step_blocks_dependents",
!out.ok && by("test").status == status::FAILED && root.status == status::FAILED,
"compile fails → test blocked → root failed",
);
std::fs::remove_dir_all(&dir).ok();
}
/// Additive: the existing single-job build path is a one-step "DAG" — one node,
/// zero edges, root + one sub-job, both done.
#[test]
fn single_step_is_a_one_node_dag_additive() {
let dir = tmpdir("single");
let store = JobStore::open(&dir).unwrap();
let spec = PipelineSpec {
name: "single".into(),
steps: vec![PipelineStep::new("build", "cargo build", &[])],
};
let dag = spec.to_plan_dag();
assert_eq!(dag.node_count(), 1);
assert_eq!(dag.dep_edges(), 0, "a lone step has no needs edges");
assert_eq!(
dag.critical_path_len(),
0,
"one node = zero waves of dependency"
);
let out = spec.run_on_ledger(&store, "ws", |_| Ok(())).unwrap();
assert!(out.ok);
let all = store.list(&JobSelector::All).unwrap();
assert_eq!(all.len(), 2, "root + one sub-job");
assert!(all.iter().all(|r| r.status == status::DONE));
std::fs::remove_dir_all(&dir).ok();
}
/// The build-info stream (slice 2) filters per `step_id` to each sub-job: rows
/// tagged with a sub-job's ledger `job_id` come back only for that step.
#[test]
fn build_info_stream_filters_per_step_id_to_each_subjob() {
let dir = tmpdir("binfo");
let store = JobStore::open(&dir).unwrap();
let spec = three_node();
let out = spec.run_on_ledger(&store, "ws", |_| Ok(())).unwrap();
let fetch_job = out.job_id_for("fetch").unwrap().to_string();
let compile_job = out.job_id_for("compile").unwrap().to_string();
assert_ne!(fetch_job, compile_job, "each sub-job has its own id");
// A mixed stream for the whole build (build_id = root), tagged per sub-job.
let rows = vec![
row(&out.root_id, &fetch_job, BuildInfoKind::Log, "fetching"),
row(&out.root_id, &compile_job, BuildInfoKind::Log, "compiling"),
row(
&out.root_id,
&compile_job,
BuildInfoKind::Metric,
"warnings",
),
// A row for a different build entirely — must NOT leak in.
row("other-build", &compile_job, BuildInfoKind::Log, "elsewhere"),
];
let fetch_rows = out.build_info_for(&rows, "fetch");
assert_eq!(fetch_rows.len(), 1, "one row for fetch's sub-job");
assert_eq!(fetch_rows[0].name, "fetching");
let compile_rows = out.build_info_for(&rows, "compile");
assert_eq!(
compile_rows.len(),
2,
"compile's own two rows, not the other build's"
);
assert!(compile_rows.iter().all(|r| r.build_id == out.root_id));
// Unknown step → empty (not a panic).
assert!(out.build_info_for(&rows, "nope").is_empty());
nornir_testmatrix::functional_status(
"build-pipeline-subjobs",
"build_info_filtered_per_step",
fetch_rows.len() == 1 && compile_rows.len() == 2,
"slice-2 stream filtered by sub-job step_id",
);
std::fs::remove_dir_all(&dir).ok();
}
/// A cyclic spec is rejected (a build pipeline must be a DAG).
#[test]
fn cyclic_pipeline_is_rejected() {
let dir = tmpdir("cycle");
let store = JobStore::open(&dir).unwrap();
let spec = PipelineSpec {
name: "loop".into(),
steps: vec![
PipelineStep::new("a", "a", &["b"]),
PipelineStep::new("b", "b", &["a"]),
],
};
assert!(spec.validate().is_err(), "a cycle is invalid");
assert!(
spec.run_on_ledger(&store, "ws", |_| Ok(())).is_err(),
"run refuses a cyclic spec"
);
// And a needs pointing at a non-existent step is rejected too.
let bad = PipelineSpec {
name: "x".into(),
steps: vec![PipelineStep::new("a", "a", &["ghost"])],
};
assert!(bad.validate().is_err(), "dangling needs is invalid");
std::fs::remove_dir_all(&dir).ok();
}
/// The DAG projection reuses funnel-core's critical_path across a branchy DAG:
/// a diamond fetch → (build_x ‖ build_y) → link picks a 2-wave critical path.
#[test]
fn diamond_dag_reuses_funnel_core_critical_path() {
let spec = PipelineSpec {
name: "diamond".into(),
steps: vec![
PipelineStep::new("fetch", "fetch", &[]),
PipelineStep::new("build_x", "build x", &["fetch"]),
PipelineStep::new("build_y", "build y", &["fetch"]),
PipelineStep::new("link", "link", &["build_x", "build_y"]),
],
};
let dag = spec.to_plan_dag();
assert_eq!(dag.dep_edges(), 4);
assert_eq!(
dag.critical_path_len(),
2,
"fetch → build_* → link = 2 waves"
);
// fetch has fan-out 2 (both builds need it); link has fan-out 0.
let fetch = dag.nodes.iter().find(|n| n.id == "fetch").unwrap();
assert_eq!(fetch.fanout, 2, "two steps need fetch");
let s = out_state(&spec);
assert_eq!(s["dep_edges"], 4);
}
fn out_state(spec: &PipelineSpec) -> serde_json::Value {
let dir = tmpdir("state");
let store = JobStore::open(&dir).unwrap();
let out = spec.run_on_ledger(&store, "ws", |_| Ok(())).unwrap();
let s = out.state_json();
std::fs::remove_dir_all(&dir).ok();
s
}
}