src/release/pipeline.rs
//! Release pipeline orchestrator.
//!
//! Walks the workspace dep graph in build order and, for each repo,
//! runs the release gates (build → test → publish) while recording
//! provenance into the warehouse's `release_lineage` table.
//!
//! The recorded lineage — per repo: git SHA, dirty flag, gate status,
//! published versions, plus the dep-graph snapshot id that pinned the
//! plan — is enough to reconstruct *exactly which sources shipped
//! together*. That's the rollback story: pick a release_id from Urðr,
//! get back the per-repo SHAs, `git checkout` them, and you have the
//! tree as it was at release time.
//!
//! Cargo subprocess use here follows the same rationale documented in
//! the forge's `release_doctor::publish::run_cargo_publish` — cargo owns build/test/publish;
//! we don't reimplement it. Everything else (git inspection,
//! warehouse writes) stays pure Rust via `gix` and `iceberg-rust`.
use std::collections::{BTreeMap, VecDeque};
use std::io::{BufRead, BufReader};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::{Arc, Mutex};
use std::time::Instant;
use anyhow::{Context, Result};
use arrow::array::{
Array, BooleanArray, Int32Array, RecordBatch, StringArray, TimestampMicrosecondArray,
};
use chrono::Utc;
use iceberg::arrow::schema_to_arrow_schema;
use uuid::Uuid;
use crate::warehouse::Warehouse;
use crate::warehouse::dep_graph::WorkspaceGraph;
use crate::warehouse::iceberg::{self as wh_iceberg};
use crate::warehouse::release_events::{ReleaseEventEmitter, phase, status};
use crate::workspace::descriptor::WorkspaceDescriptor;
/// The upstream components `repo` waited on in the build order — the
/// dep-graph producers (`from == repo` edges point consumer→producer, so
/// `to` is the dependency). De-duplicated + sorted for a stable
/// `depends_on` list. Empty `Vec` for a root with no in-workspace deps.
fn depends_on_for(graph: &WorkspaceGraph, repo: &str) -> Vec<String> {
let mut deps: Vec<String> = graph
.dependencies_of(repo)
.iter()
.map(|e| e.to.clone())
.filter(|d| d != repo)
.collect();
deps.sort();
deps.dedup();
deps
}
/// Options controlling a single release run.
#[derive(Debug, Clone)]
pub struct PipelineOptions {
/// If true, skip the publish gate entirely (don't even call cargo
/// with `--dry-run`). Lineage row records `dry_run = true` and
/// `published_versions_csv` stays empty.
pub dry_run: bool,
/// Dep-graph snapshot this release is pinned to. Use
/// [`crate::warehouse::dep_graph::record_dep_graph`] first, then
/// pass the returned UUID here.
pub dep_graph_snapshot_id: Uuid,
/// If true (default), refuse to start the release when any repo
/// has uncommitted changes or untracked files. Rationale: nornir
/// only remembers git SHAs as the identity of shipped code; a
/// dirty tree means the SHA doesn't fully describe what would
/// ship, so the release can't be safely rolled back later. The
/// fix is always one of: `git commit` the work, or
/// `git reset --hard <sha>` to the last clean release SHA (look
/// it up via [`query_release_history`]).
pub refuse_if_dirty: bool,
/// Optional progress writer. When set, `run_pipeline` emits live
/// `ReleaseEvent`s into the underlying ndjson file so the egui
/// viz / `nornir-server` SSE can render the run live.
pub progress: Option<crate::release::progress::ProgressWriter>,
/// Optional gate configuration. When present, per-repo gates that
/// need config (`no_regression`, `guard_intact`) are evaluated as
/// release-blocking checks. Absent ⇒ those gates are skipped
/// (back-compat: existing callers that only smoke-test build/test).
pub gate_config: Option<PipelineGateConfig>,
}
/// Config for the release-blocking gates that need more than the
/// workspace descriptor: regression history keys off per-repo
/// [`Gates`], and `guard_intact` needs the workspace root to load the
/// recorded guard manifest.
#[derive(Debug, Clone)]
pub struct PipelineGateConfig {
/// Workspace root (parent of `workspace_holger`); where
/// `.nornir/guard-manifest.json` lives.
pub workspace_root: PathBuf,
/// Per-repo gate flags, keyed by repo name (from `nornir.toml`).
pub repo_gates: BTreeMap<String, crate::config::Gates>,
}
impl Default for PipelineOptions {
fn default() -> Self {
Self {
dry_run: true,
dep_graph_snapshot_id: Uuid::nil(),
refuse_if_dirty: true,
progress: None,
gate_config: None,
}
}
}
/// Whether a recorded per-repo `gate_status` represents success. Single
/// source of truth shared by the loop-abort check and the final-ok
/// rollup, so they can never diverge (the historical BUG#1 was exactly
/// such a divergence — only `failed_test` aborted/flipped, so
/// `failed_bench` shipped silently).
pub(crate) fn gate_succeeded(status: &str) -> bool {
matches!(status, "succeeded" | "succeeded_dry_run")
}
#[derive(Debug, Clone)]
pub struct RepoGitState {
pub sha: String,
pub branch: String,
pub dirty: bool,
}
#[derive(Debug, Clone)]
pub struct RepoReleaseRecord {
pub repo: String,
pub build_order_idx: usize,
pub git: RepoGitState,
pub gate_status: String,
pub tests_passed: u32,
pub tests_failed: u32,
pub published_versions: Vec<(String, String)>,
/// Urðr time-machine pins captured by the release run's capture phase
/// (`release_lineage` cols 15/16). `None` when the projection wasn't
/// captured (e.g. `--skip-snapshot`, or a build with no binary for dwarf).
pub tantivy_snapshot_id: Option<String>,
pub dwarf_snapshot_id: Option<String>,
}
#[derive(Debug, Clone)]
pub struct ReleaseReport {
pub release_id: Uuid,
pub workspace_name: String,
/// Dep-graph snapshot pinned at release time. `Uuid::nil()` for
/// pre-1.x rows that didn't record it.
pub dep_graph_snapshot_id: Uuid,
pub repos: Vec<RepoReleaseRecord>,
}
/// Run the pipeline. Returns a [`ReleaseReport`] AND persists the
/// same data into the warehouse before returning.
///
/// Stops at the first repo whose test gate fails — but every repo
/// processed up to (and including) the failure is recorded.
pub async fn run_pipeline(
wh: &dyn Warehouse,
desc: &WorkspaceDescriptor,
graph: &WorkspaceGraph,
opts: &PipelineOptions,
) -> Result<ReleaseReport> {
let release_id = Uuid::new_v4();
let workspace_name = desc.workspace.name.clone();
let order = graph.build_order().context("derive build order")?;
// Release-op DAG emitter. Keyed by the run's release_id so every event row
// groups under one run; `seq` is minted per-emit. Emits are best-effort +
// non-fatal (see `ReleaseEventEmitter`) so a logging-row failure never
// aborts the release. The whole run is bracketed by a `run/start`/`run/end`.
let events = ReleaseEventEmitter::new(release_id.to_string());
events
.emit_async(
wh,
&workspace_name,
"",
"run",
phase::START,
status::RUNNING,
&format!("order: {}", order.join(" → ")),
None,
None,
)
.await;
let sources: BTreeMap<String, PathBuf> =
crate::workspace::resolve::resolve_sources(desc).context("resolve workspace sources")?;
let mut records: Vec<RepoReleaseRecord> = Vec::new();
use crate::release::progress::{ReleaseEvent, now};
if let Some(p) = &opts.progress {
p.emit(&ReleaseEvent::RunStart {
ts: now(),
run_id: release_id.to_string(),
workspace: workspace_name.clone(),
});
}
// Pre-flight: collect git state for every repo. If any is dirty
// and `refuse_if_dirty` is set, abort before running any cargo —
// the SHA on a dirty tree doesn't fully describe what would
// ship, so the release would be unrollable.
let mut pre: BTreeMap<String, RepoGitState> = BTreeMap::new();
for repo in &order {
let root = sources
.get(repo)
.ok_or_else(|| anyhow::anyhow!("no source path for repo `{repo}` in descriptor"))?;
let git = read_git_state(root)
.with_context(|| format!("read git state for {}", root.display()))?;
pre.insert(repo.clone(), git);
}
if opts.refuse_if_dirty {
let dirty: Vec<&String> = order.iter().filter(|r| pre[*r].dirty).collect();
if !dirty.is_empty() {
let detail = dirty
.iter()
.map(|r| format!(" - {r} (HEAD={})", &pre[*r].sha[..12]))
.collect::<Vec<_>>()
.join("\n");
anyhow::bail!(
"release refused: {} repo(s) have uncommitted changes:\n{detail}\n\
commit the work, or `git reset --hard <last-clean-sha>` to roll back, \
then re-run. (Set refuse_if_dirty=false to override.)",
dirty.len()
);
}
}
for (idx, repo) in order.iter().enumerate() {
let root = sources.get(repo).expect("checked above");
let git = pre.remove(repo).expect("checked above");
println!(
"\n\u{2501}\u{2501}\u{2501} [{i}/{n}] {repo} \u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}\u{2501}",
i = idx + 1,
n = order.len(),
);
println!(
"📍 sha={} branch={} dirty={}",
&git.sha[..git.sha.len().min(12)],
git.branch,
if git.dirty { "yes" } else { "no" }
);
if let Some(p) = &opts.progress {
p.emit(&ReleaseEvent::RepoStart {
ts: now(),
repo: repo.clone(),
sha: git.sha.clone(),
});
p.emit(&ReleaseEvent::PhaseStart {
ts: now(),
repo: repo.clone(),
phase: "test".to_string(),
});
}
// Dep-graph producers this repo waited on (the DAG `depends_on` edges).
let deps = depends_on_for(graph, repo);
let phase_start = Instant::now();
events
.emit_async(
wh,
repo,
repo,
"test",
phase::START,
status::RUNNING,
"",
Some(deps.clone()),
None,
)
.await;
let (tests_passed, tests_failed, test_ok) =
run_cargo_test(root, Some(repo), opts.progress.as_ref())
.with_context(|| format!("cargo test for repo `{repo}`"))?;
println!(
" \u{2192} test totals: {tests_passed} passed, {tests_failed} failed ({:.1}s) {}",
phase_start.elapsed().as_secs_f32(),
if test_ok {
"\u{2713}"
} else {
"\u{2717} FAILED"
}
);
events
.emit_async(
wh,
repo,
repo,
"test",
phase::END,
if test_ok { status::OK } else { status::FAIL },
&format!("{tests_passed} passed, {tests_failed} failed"),
Some(deps.clone()),
Some(phase_start.elapsed().as_millis() as i64),
)
.await;
if let Some(p) = &opts.progress {
p.emit(&ReleaseEvent::PhaseEnd {
ts: now(),
repo: repo.clone(),
phase: "test".to_string(),
ok: test_ok,
duration_ms: phase_start.elapsed().as_millis() as u64,
});
}
let bench_start = Instant::now();
events
.emit_async(
wh,
repo,
repo,
"bench",
phase::START,
status::RUNNING,
"",
Some(deps.clone()),
None,
)
.await;
// `bench_status` is one of: "skipped" | "ok" (passed) or a fatal
// "failed_*" status. A fatal status both aborts the run and flips
// final_ok (via `gate_succeeded`). Note: a run is persisted as a
// future regression baseline ONLY if it passes every bench gate —
// we must not let a red/regressed/errored run poison the history.
let bench_status = match run_bench_example(root, None) {
Ok(None) => {
println!("\u{26A1} bench: skipped (no examples/nornir-bench.rs)");
"skipped".to_string()
}
Ok(Some(run)) => {
let n_res = run.results.len();
let n_failed = run.tests.iter().filter(|t| !t.passed).count();
println!(
"\u{26A1} bench ({:.1}s): {} result(s), {} test fail(s)",
bench_start.elapsed().as_secs_f32(),
n_res,
n_failed,
);
for r in &run.results {
let mut kv: Vec<String> = r
.metrics
.iter()
.filter_map(|(k, v)| v.as_f64().map(|f| format!("{k}={f:.2}")))
.collect();
kv.sort();
println!(" \u{2022} {:30} {}", r.name, kv.join(" "));
}
if n_failed > 0 {
for t in run.tests.iter().filter(|t| !t.passed) {
println!(" \u{2717} test {}", t.name);
}
// Bench unit tests failed → fatal. Do NOT persist (a
// red run must never become a regression baseline).
"failed_bench".to_string()
} else {
// Regression gate (if configured) runs BEFORE the
// append, so the candidate can't be compared against
// itself (which would make every drop 0%).
let regression = match &opts.gate_config {
Some(gc)
if gc
.repo_gates
.get(repo)
.map(|g| g.no_regression)
.unwrap_or(false) =>
{
let g = gc.repo_gates.get(repo).expect("checked");
let pct = if g.max_regression_pct > 0.0 {
g.max_regression_pct
} else {
10.0
};
let hist = wh
.query_bench_runs_async(&crate::warehouse::BenchFilter {
repo: Some(repo.clone()),
machine: Some(run.machine.clone()),
limit: None,
})
.await
.with_context(|| format!("read bench history for `{repo}`"))?;
crate::release::gate::no_regression_against(&run, &hist, pct).err()
}
_ => None,
};
// Core-saturation LAW gate (if configured): a parallel bench
// that under-used the cores is a bug, not Amdahl-fine. Checks
// the run's telemetry footprint IN MEMORY, before the append
// below, so a red run never becomes a baseline.
let saturation = match &opts.gate_config {
Some(gc)
if gc
.repo_gates
.get(repo)
.map(|g| g.core_saturation)
.unwrap_or(false) =>
{
let g = gc.repo_gates.get(repo).expect("checked");
crate::release::gate::core_saturation_run(&run, g.min_core_busy_frac)
.err()
}
_ => None,
};
if let Some(e) = regression {
println!(" \u{2717} regression gate: {e:#}");
"failed_regression".to_string()
} else if let Some(e) = saturation {
println!(" \u{2717} core-saturation gate: {e:#}");
"failed_core_saturation".to_string()
} else {
wh.append_bench_run_async(repo, &run)
.await
.with_context(|| format!("persist bench run for `{repo}`"))?;
println!(" \u{2713} persisted into bench_runs/bench_results");
"ok".to_string()
}
}
}
Err(e) => {
// Bench failed to even execute → fatal. The configured
// release evidence could not be produced; correctness-first
// means we must not ship.
println!("\u{26A1} bench: \u{2717} error: {e:#}");
"failed_bench_exec".to_string()
}
};
let bench_ev_status = match bench_status.as_str() {
"ok" => status::OK,
"skipped" => status::WARN,
_ => status::FAIL,
};
events
.emit_async(
wh,
repo,
repo,
"bench",
phase::END,
bench_ev_status,
&bench_status,
Some(deps.clone()),
Some(bench_start.elapsed().as_millis() as i64),
)
.await;
// guard_intact gate: the recorded-vs-current tamper check. Only
// evaluated for repos that opt in via `[gates].guard_intact` and
// when a gate_config (workspace root + manifest) is available.
let guard_failure: Option<String> = match &opts.gate_config {
Some(gc)
if gc
.repo_gates
.get(repo)
.map(|g| g.guard_intact)
.unwrap_or(false) =>
{
match crate::guard::read_manifest(&gc.workspace_root) {
Ok(manifest) => crate::guard::intact(&gc.workspace_root, &manifest)
.err()
.map(|e| format!("{e:#}")),
Err(e) => Some(format!("guard manifest unavailable: {e:#}")),
}
}
_ => None,
};
let bench_ok = matches!(bench_status.as_str(), "ok" | "skipped");
let (gate_status, published) = if !test_ok {
("failed_test".to_string(), Vec::new())
} else if !bench_ok {
// bench_status already holds the specific fatal reason.
(bench_status.clone(), Vec::new())
} else if let Some(reason) = &guard_failure {
println!(" \u{2717} guard_intact gate: {reason}");
("failed_guard_intact".to_string(), Vec::new())
} else if opts.dry_run {
("succeeded_dry_run".to_string(), Vec::new())
} else {
// Real publish path is intentionally NOT wired here: the
// per-repo publish_order TOML lives outside the workspace
// descriptor (see release/publish.rs) and we don't want
// to push to crates.io as a side-effect of a smoke-test.
// When that's wired, switch this arm to call publish_all
// and populate `published`.
("succeeded".to_string(), Vec::new())
};
let repo_ok = gate_succeeded(&gate_status);
// Gate outcome event: a single `gate/end` row carrying the per-repo
// `gate_status` (the same string `release_lineage` records), so the DAG
// shows exactly where a partial release stalled. `start`+`end` collapse
// to one boundary here — the gate rollup is instantaneous.
events
.emit_async(
wh,
repo,
repo,
"gate",
phase::END,
if repo_ok { status::OK } else { status::FAIL },
&gate_status,
Some(deps.clone()),
None,
)
.await;
// Historize the git-heat half of the knowledge map at the release SHA
// (best-effort, non-fatal). The symbol / call half is not a release
// step any more: the code room's consumer writes it per commit.
if repo_ok && !opts.dry_run {
let root_owned = root.clone();
let repo_owned = repo.clone();
match tokio::task::spawn_blocking(move || {
crate::knowledge::git_heat::scan_repo(
&root_owned,
&repo_owned,
uuid::Uuid::new_v4(),
chrono::Utc::now(),
)
})
.await
{
Ok(Ok(git)) => match wh.append_git_heat_scan_async(&git).await {
Ok(_) => println!(
" \u{2713} historized git-heat (symbols / calls are the code room's: its consumer writes them per commit, a release does not)"
),
Err(e) => eprintln!(" \u{26A0} git-heat capture failed for `{repo}`: {e:#}"),
},
Ok(Err(e)) => eprintln!(" \u{26A0} git-heat scan skipped for `{repo}`: {e:#}"),
Err(e) => eprintln!(" \u{26A0} git-heat scan task failed for `{repo}`: {e}"),
}
}
records.push(RepoReleaseRecord {
repo: repo.clone(),
build_order_idx: idx,
git,
gate_status: gate_status.clone(),
tests_passed,
tests_failed,
published_versions: published,
// The library orchestrator historizes the knowledge map but not the
// index/dwarf; the rich CLI path (`run_release_run`) pins those.
tantivy_snapshot_id: None,
dwarf_snapshot_id: None,
});
if let Some(p) = &opts.progress {
p.emit(&ReleaseEvent::RepoEnd {
ts: now(),
repo: repo.clone(),
ok: repo_ok,
});
}
// Any non-success status aborts the remaining build order: a
// dependency failed its gate, so dependents can't be validated
// against it. (Historically only "failed_test" broke here —
// BUG#1 — letting bench failures ship silently.)
if !repo_ok {
break;
}
}
persist_lineage(
wh,
release_id,
&workspace_name,
&opts.dep_graph_snapshot_id,
&records,
opts.dry_run,
)
.await?;
let final_ok = records.iter().all(|r| gate_succeeded(&r.gate_status));
events
.emit_async(
wh,
&workspace_name,
"",
"run",
phase::END,
if final_ok { status::OK } else { status::FAIL },
&format!("{} repo(s) processed", records.len()),
None,
None,
)
.await;
if let Some(p) = &opts.progress {
p.emit(&ReleaseEvent::RunEnd {
ts: now(),
run_id: release_id.to_string(),
ok: final_ok,
});
}
Ok(ReleaseReport {
release_id,
workspace_name,
dep_graph_snapshot_id: opts.dep_graph_snapshot_id,
repos: records,
})
}
pub fn read_git_state(root: &Path) -> Result<RepoGitState> {
let repo = gix::open(root).with_context(|| format!("gix::open {}", root.display()))?;
let head = repo.head().context("read HEAD")?;
let sha = head
.id()
.ok_or_else(|| anyhow::anyhow!("HEAD has no commit (unborn?)"))?
.to_string();
let branch = match repo.head_name()? {
Some(name) => name.shorten().to_string(),
None => "(detached)".to_string(),
};
// The ONE shared status walk (LAW 5) — never a second gix status iteration
// here. Errors still swallow to `false`: provenance must not fail on an
// exotic repo layout, which is what the local twin this replaced was for.
//
// The twin was also a PROCESS LEAK. It ran `repo.status(...)` on a plainly
// opened repository, and with `/etc/gitconfig` declaring
// `filter.lfs.process`, every such walk parks a `git-lfs filter-process`
// child in a `gix_filter::driver::State` that has no `Drop` — dead, never
// reaped, `<defunct>` for the life of the process. `worktree_freshness` goes
// through `gitio::open_for_status`, which disarms the long-running driver.
let dirty = crate::gitio::worktree_freshness(root)
.map(|f| f.dirty)
.unwrap_or(false);
Ok(RepoGitState { sha, branch, dirty })
}
/// Returns `(passed, failed, ok)` where `ok = failed == 0 && cargo
/// itself exited cleanly`. Parsing is line-oriented over stdout —
/// cargo's "test result: ok. N passed; M failed" summary.
///
/// `progress`, when set, receives structured events for every
/// `Running ...` binary and every per-test pass/fail. The live egui
/// pane and `nornir-server`'s `/release/progress` SSE both consume
/// this stream. Pass `None` for headless / unit-test calls.
pub fn run_cargo_test(
root: &Path,
repo_for_events: Option<&str>,
progress: Option<&crate::release::progress::ProgressWriter>,
) -> Result<(u32, u32, bool)> {
use crate::release::progress::{ReleaseEvent, now};
let repo_label = repo_for_events.unwrap_or("").to_string();
let emit = |ev: ReleaseEvent| {
if let Some(p) = progress {
p.emit(&ev);
}
};
// Heavy tests (`#[ignore]`'d: container integration, large corpora, model
// downloads, the workspace×function viz matrix) run only ON REQUEST now —
// default OFF everywhere (release, `nornir test`, viz). They burn minutes and
// spin the fans, and a flaky heavy/network arm turning a release red is the
// opposite of a gate; the always-run SYNTHETIC tier IS the gate. Opt IN with
// the global `--heavy` flag (`NORNIR_RUN_HEAVY=1`). Single source of truth:
// `test_matrix::heavy_enabled()` — see its doc for the full precedence.
let heavy = crate::test_matrix::heavy_enabled();
let verbose = std::env::var("NORNIR_VERBOSE")
.map(|v| !v.is_empty() && v != "0")
.unwrap_or(false);
// NOTE: deliberately *no* top-level `--quiet`. We want cargo's
// `Running tests/foo.rs` lines on stderr and libtest's
// `test result:` rollup on stdout. Warnings are filtered out
// below so the operator sees signal, not noise.
// Always serial across the workspace:
// `-j 1` → one test *binary* at a time (otherwise cargo
// runs different test binaries in parallel and
// e.g. two Sonatype Nexus testcontainers fight
// for RAM + ports).
// `--test-threads=1` → one test *function* at a time inside each
// binary (kills the classic libtest race on
// shared `/tmp` caches).
// Extra cargo features for the test gate, e.g. `NORNIR_TEST_FEATURES=viz,server`
// so nornir's OWN release compiles + runs the feature-gated viz matrix tests
// (the workspace×function sweep). Cargo args go BEFORE `--`; harness args after.
let feat = std::env::var("NORNIR_TEST_FEATURES")
.ok()
.filter(|s| !s.is_empty());
// This gate LINKS the whole workspace. A workspace that links the code
// room needs the system GraphBLAS: say so in one line instead of running
// cargo into a linker error per test binary (decision D5). A nornir built
// WITHOUT `code-room` (korp's dev-dependency, the thin atoms) has no
// preflight to ask — the linker speaks for itself there.
#[cfg(feature = "code-room")]
if let Err(missing) = crate::code_room::preflight::check(root) {
anyhow::bail!("cargo test preflight for {}: {missing}", root.display());
}
let mut args: Vec<&str> = vec!["test", "--workspace", "--no-fail-fast", "-j", "1"];
if let Some(f) = &feat {
args.push("--features");
args.push(f);
}
args.push("--");
args.push("--test-threads=1");
if heavy {
args.push("--include-ignored");
}
println!("🧪 cargo {}", args.join(" "));
let mut child = Command::new(nornir_release_core::cargo::cargo_bin())
.args(&args)
.current_dir(root)
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.context("spawn cargo test")?;
let stdout = child.stdout.take().expect("piped stdout");
let stderr = child.stderr.take().expect("piped stderr");
// Shared queue: stderr pushes binary names in cargo's emission
// order; stdout pops one each time libtest announces a new
// `running N tests` block. This is the only correct way to
// pair the two streams without parsing cargo JSON output.
let pending_bins: Arc<Mutex<VecDeque<String>>> = Arc::new(Mutex::new(VecDeque::new()));
let bins_for_stderr = Arc::clone(&pending_bins);
let progress_for_stderr = progress.cloned();
let repo_for_stderr = repo_label.clone();
let stderr_handle = std::thread::spawn(move || {
for line in BufReader::new(stderr).lines().map_while(Result::ok) {
let trimmed = line.trim_start();
if let Some(what) = trimmed.strip_prefix("Running ") {
let what = what.trim().to_string();
println!(" 📂 {what}");
if let Some(p) = &progress_for_stderr {
p.emit(&ReleaseEvent::BinaryStart {
ts: now(),
repo: repo_for_stderr.clone(),
binary: what.clone(),
});
}
bins_for_stderr.lock().unwrap().push_back(what);
} else if verbose && trimmed.starts_with("Compiling ") {
println!(" 🔨 {}", trimmed.trim_start_matches("Compiling "));
} else if trimmed.starts_with("error") {
// Compiler / cargo errors — always surface.
println!(" {line}");
}
// Warnings are intentionally dropped.
}
});
let mut passed = 0u32;
let mut failed = 0u32;
let mut failure_names: Vec<String> = Vec::new();
let mut current_bin: String = String::new();
// Counters scoped to the currently-running binary, so we can
// emit BinaryDone with the right rollup before the next block.
let mut bin_passed: u32 = 0;
let mut bin_failed: u32 = 0;
for line in BufReader::new(stdout).lines().map_while(Result::ok) {
let trimmed = line.trim_start();
if let Some(rest) = trimmed.strip_prefix("running ") {
// `running N tests` → start of a new binary block.
if let Some((n, _)) = rest.split_once(' ') {
if let Ok(n) = n.parse::<u32>() {
current_bin = pending_bins.lock().unwrap().pop_front().unwrap_or_default();
let bin_short = current_bin
.split_whitespace()
.next()
.unwrap_or(¤t_bin);
println!(" \u{2192} {n} test(s) {bin_short}");
bin_passed = 0;
bin_failed = 0;
}
}
} else if let Some(name) = trimmed.strip_prefix("test ") {
if let Some(nm) = name.strip_suffix(" ... FAILED") {
println!(" \u{2717} {nm}");
failure_names.push(format!(
"{}::{nm}",
current_bin.split_whitespace().next().unwrap_or("?")
));
bin_failed += 1;
emit(ReleaseEvent::TestFail {
ts: now(),
repo: repo_label.clone(),
binary: current_bin.clone(),
name: nm.to_string(),
});
} else if let Some(nm) = name.strip_suffix(" ... ok") {
if verbose {
println!(" \u{2713} {nm}");
}
bin_passed += 1;
emit(ReleaseEvent::TestPass {
ts: now(),
repo: repo_label.clone(),
binary: current_bin.clone(),
name: nm.to_string(),
});
}
} else if trimmed.starts_with("test result:") {
for chunk in trimmed.split(';') {
let chunk = chunk.trim();
if let Some(n) = chunk.strip_suffix(" passed") {
if let Some(n) = n.split_whitespace().last() {
passed += n.parse::<u32>().unwrap_or(0);
}
} else if let Some(n) = chunk.strip_suffix(" failed") {
if let Some(n) = n.split_whitespace().last() {
failed += n.parse::<u32>().unwrap_or(0);
}
}
}
println!(" \u{2261} {trimmed}");
emit(ReleaseEvent::BinaryDone {
ts: now(),
repo: repo_label.clone(),
binary: current_bin.clone(),
passed: bin_passed,
failed: bin_failed,
});
}
}
let status = child.wait().context("wait cargo test")?;
let _ = stderr_handle.join();
// If cargo aborted before printing the per-binary `test result:`
// rollup, the libtest counters above can read 0/0 while we still
// captured concrete failure names from `... FAILED` lines. Promote
// the failure-name list to the failed counter in that case so the
// operator-facing summary doesn't lie.
if failure_names.len() as u32 > failed {
failed = failure_names.len() as u32;
}
if !failure_names.is_empty() {
println!(" ✗ {} failure(s):", failure_names.len());
for fn_ in &failure_names {
println!(" • {fn_}");
}
}
Ok((passed, failed, status.success() && failed == 0))
}
/// Append one `release_lineage` row per record, pinning the Urðr snapshot ids
/// each record carries. Used by both the library orchestrator (`run_pipeline`)
/// and the rich CLI path (`bin/nornir.rs::run_release_run`) so the table is
/// written the same way regardless of which drives the release.
pub async fn persist_lineage(
wh: &dyn Warehouse,
release_id: Uuid,
workspace_name: &str,
dep_graph_snapshot_id: &Uuid,
records: &[RepoReleaseRecord],
dry_run: bool,
) -> Result<()> {
if records.is_empty() {
return Ok(());
}
// Build against the CANONICAL release_lineage schema (the table was created
// from it); `append_arrow_async` recasts to the stored field-id schema.
let arrow_schema = Arc::new(schema_to_arrow_schema(
&crate::warehouse::iceberg_schema::release_lineage()?,
)?);
let ts = Utc::now().timestamp_micros();
let rid = release_id.to_string();
let dsid = dep_graph_snapshot_id.to_string();
let n = records.len();
let mut release_ids = Vec::with_capacity(n);
let mut ws_names = Vec::with_capacity(n);
let mut ts_vals = Vec::with_capacity(n);
let mut dgsids = Vec::with_capacity(n);
let mut order_idx = Vec::with_capacity(n);
let mut repos = Vec::with_capacity(n);
let mut shas = Vec::with_capacity(n);
let mut branches = Vec::with_capacity(n);
let mut dirties = Vec::with_capacity(n);
let mut statuses = Vec::with_capacity(n);
let mut tpass = Vec::with_capacity(n);
let mut tfail = Vec::with_capacity(n);
let mut dryruns = Vec::with_capacity(n);
let mut pubs = Vec::with_capacity(n);
for r in records {
release_ids.push(rid.clone());
ws_names.push(workspace_name.to_string());
ts_vals.push(ts);
dgsids.push(dsid.clone());
order_idx.push(r.build_order_idx as i32);
repos.push(r.repo.clone());
shas.push(r.git.sha.clone());
branches.push(r.git.branch.clone());
dirties.push(r.git.dirty);
statuses.push(r.gate_status.clone());
tpass.push(r.tests_passed as i32);
tfail.push(r.tests_failed as i32);
dryruns.push(dry_run);
pubs.push(
r.published_versions
.iter()
.map(|(c, v)| format!("{c}@{v}"))
.collect::<Vec<_>>()
.join(","),
);
}
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(release_ids)),
Arc::new(StringArray::from(ws_names)),
Arc::new(TimestampMicrosecondArray::from(ts_vals).with_timezone("+00:00")),
Arc::new(StringArray::from(dgsids)),
Arc::new(Int32Array::from(order_idx)),
Arc::new(StringArray::from(repos)),
Arc::new(StringArray::from(shas)),
Arc::new(StringArray::from(branches)),
Arc::new(BooleanArray::from(dirties)),
Arc::new(StringArray::from(statuses)),
Arc::new(Int32Array::from(tpass)),
Arc::new(Int32Array::from(tfail)),
Arc::new(BooleanArray::from(dryruns)),
Arc::new(StringArray::from(pubs)),
// Time-machine pin columns (15-17). `tantivy_snapshot_id` +
// `dwarf_snapshot_id` come from each record's capture phase
// (`run_release_run`); the legacy `run_pipeline` leaves them None.
// `gimli_snapshot_id` (17) is not yet produced.
Arc::new(StringArray::from(
records
.iter()
.map(|r| r.tantivy_snapshot_id.clone())
.collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
records
.iter()
.map(|r| r.dwarf_snapshot_id.clone())
.collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(vec![None as Option<String>; n])),
// Cargo-pipeline pins (fields 18-21) — also null from the
// legacy orchestrator; populated by gates/wirings as they land.
Arc::new(StringArray::from(vec![None as Option<String>; n])), // git_tag
Arc::new(BooleanArray::from(vec![None as Option<bool>; n])), // git_tag_matches_published
Arc::new(StringArray::from(vec![None as Option<String>; n])), // rustdoc_json_snapshot_id
Arc::new(Int32Array::from(vec![None as Option<i32>; n])), // patch_lines_stripped
];
let batch = RecordBatch::try_new(arrow_schema, cols)?;
wh.append_arrow_async(wh_iceberg::TABLE_RELEASE_LINEAGE, batch)
.await?;
Ok(())
}
/// Read every recorded release for `workspace_name`, grouped by
/// release_id, sorted oldest-first. Optional `limit` keeps only the
/// most recent N. An **empty** `workspace_name` matches every workspace
/// (used by the cross-workspace `regression::trace`).
pub async fn query_release_history(
wh: &dyn Warehouse,
workspace_name: &str,
limit: Option<usize>,
) -> Result<Vec<ReleaseReport>> {
// Full-snapshot scan through the backend-neutral async reader (column order
// preserved, so the positional `down::<T>(batch, idx)` decode stays valid).
let batches: Vec<RecordBatch> = wh
.scan_arrow_async(wh_iceberg::TABLE_RELEASE_LINEAGE)
.await?;
// Key is (timestamp_micros, release_uuid) so iteration is CHRONOLOGICAL
// (oldest-first) — the contract the regression bisect and `limit` rely on.
// Keying by uuid first would order by a random `Uuid::new_v4()`, making the
// returned timeline non-chronological and the bisect's blame unreliable.
let mut by_release: BTreeMap<(i64, Uuid), (String, Uuid, Vec<RepoReleaseRecord>)> =
BTreeMap::new();
for batch in &batches {
let rid = down::<StringArray>(batch, 0)?;
let wss = down::<StringArray>(batch, 1)?;
let tss = down::<TimestampMicrosecondArray>(batch, 2)?;
let dgs = down::<StringArray>(batch, 3)?;
let oid = down::<Int32Array>(batch, 4)?;
let rep = down::<StringArray>(batch, 5)?;
let sha = down::<StringArray>(batch, 6)?;
let br = down::<StringArray>(batch, 7)?;
let dty = down::<BooleanArray>(batch, 8)?;
let st = down::<StringArray>(batch, 9)?;
let tp = down::<Int32Array>(batch, 10)?;
let tf = down::<Int32Array>(batch, 11)?;
let _dr = down::<BooleanArray>(batch, 12)?;
let pub_csv = down::<StringArray>(batch, 13)?;
// Time-machine pins (arrow cols 14/15 = field ids 15/16).
let tantivy = down::<StringArray>(batch, 14)?;
let dwarf = down::<StringArray>(batch, 15)?;
for i in 0..batch.num_rows() {
if !workspace_name.is_empty() && wss.value(i) != workspace_name {
continue;
}
let release_uuid = Uuid::parse_str(rid.value(i))?;
let snap_uuid = Uuid::parse_str(dgs.value(i)).unwrap_or(Uuid::nil());
let key = (tss.value(i), release_uuid);
let entry = by_release
.entry(key)
.or_insert_with(|| (wss.value(i).to_string(), snap_uuid, Vec::new()));
let published = if pub_csv.value(i).is_empty() {
Vec::new()
} else {
pub_csv
.value(i)
.split(',')
.filter_map(|s| {
let (c, v) = s.split_once('@')?;
Some((c.to_string(), v.to_string()))
})
.collect()
};
entry.2.push(RepoReleaseRecord {
repo: rep.value(i).to_string(),
build_order_idx: oid.value(i) as usize,
git: RepoGitState {
sha: sha.value(i).to_string(),
branch: br.value(i).to_string(),
dirty: dty.value(i),
},
gate_status: st.value(i).to_string(),
tests_passed: tp.value(i) as u32,
tests_failed: tf.value(i) as u32,
published_versions: published,
tantivy_snapshot_id: (!tantivy.is_null(i)).then(|| tantivy.value(i).to_string()),
dwarf_snapshot_id: (!dwarf.is_null(i)).then(|| dwarf.value(i).to_string()),
});
}
}
let mut out: Vec<ReleaseReport> = by_release
.into_iter()
.map(|((_ts, release_id), (ws, snap, mut repos))| {
repos.sort_by_key(|r| r.build_order_idx);
ReleaseReport {
release_id,
workspace_name: ws,
dep_graph_snapshot_id: snap,
repos,
}
})
.collect();
if let Some(n) = limit {
let drop_n = out.len().saturating_sub(n);
out.drain(..drop_n);
}
Ok(out)
}
fn down<T: 'static>(batch: &RecordBatch, idx: usize) -> Result<&T> {
batch
.column(idx)
.as_any()
.downcast_ref::<T>()
.ok_or_else(|| anyhow::anyhow!("column {idx} downcast failed"))
}
// ── the bench RUNNER — moved to `nornir_bench::runner` ──────────────────────
//
// `bench_pkg_has_feature`, `BenchExample`, `metadata_bench_example`,
// `find_bench_example`, `subject_version` and `run_bench_example` used to be
// defined here, ~300 lines of them, and that put the rules for FINDING and
// SPAWNING a repo's bench example inside the warehouse crate. The forge needs
// exactly those rules to run a bench (`dwarves bench run`) and could not reach
// them without linking nornir.
//
// They now live in `nornir_bench::runner` (edda, behind its `runner` feature) —
// the same crate that defines the inner half of the contract they implement.
// This module keeps its public names by RE-EXPORT, so `pipeline::BenchExample`
// and `pipeline::find_bench_example` resolve exactly as before, and the two
// in-tree callers of `run_bench_example` (this file's release gate, and
// `server_grpc`'s `Ops.RunBench`) are untouched.
pub use nornir_bench::runner::{BenchExample, find_bench_example};
/// Invoke the repo's `nornir-bench` example and parse the `BenchRun` it prints.
///
/// A thin adapter over [`nornir_bench::runner::run_bench_example`] that supplies
/// nornir's answers to the two policy questions the leaf refuses to guess at:
/// which `cargo` (the shared `nornir_release_core::cargo::cargo_bin` resolver)
/// and whether this is a heavy run (`test_matrix::heavy_enabled`). Behaviour is
/// unchanged for every existing caller — same signature, same defaults.
pub fn run_bench_example(
root: &Path,
features: Option<&str>,
) -> Result<Option<crate::bench::BenchRun>> {
let opts = nornir_bench::runner::RunOptions::new(nornir_release_core::cargo::cargo_bin())
.with_features(features)
.with_heavy(crate::test_matrix::heavy_enabled());
nornir_bench::runner::run_bench_example(root, &opts)
}
#[cfg(test)]
mod gate_status_tests {
use super::gate_succeeded;
#[test]
fn only_success_statuses_count_as_succeeded() {
assert!(gate_succeeded("succeeded"));
assert!(gate_succeeded("succeeded_dry_run"));
// Every fatal status must NOT count as success — this is the
// single predicate that drives both loop-abort and final_ok, so
// a regression here would resurrect BUG#1 (silent bench ship).
for bad in [
"failed_test",
"failed_bench",
"failed_bench_exec",
"failed_regression",
"failed_guard_intact",
"skipped",
"",
] {
assert!(!gate_succeeded(bad), "`{bad}` must not be a success status");
}
}
}