crates/nornir-build-thing/src/build_info.rs
//! **Build-info** — the second half of a build (design §f2/§f3).
//!
//! A build produces two things at once: the *thing* (artifact bytes, hashed by
//! [`crate::provenance::BuildProvenance`] and stored in holger's CAS) and the
//! *build info* — the columnar record ABOUT the build. This module is the
//! transport-agnostic build-info binding: a flat, dep-light [`BuildInfoRow`]
//! shaped exactly like the design's §f3 Arrow schema, a pluggable
//! [`BuildInfoSink`] seam with an in-process default, and — behind the optional
//! `build-info-arrow` feature — the Arrow [`build_info_schema`] +
//! [`build_info_batch`] emitter.
//!
//! # One schema, three record kinds
//!
//! Build info is inherently record-shaped, so it IS columnar. One row shape,
//! discriminated by [`BuildInfoRow::kind`], carries the whole feed — the
//! Artifactory-parity **manifest rows** (`provenance` / `dep` / `artifact`,
//! derived from [`BuildProvenance`]) AND the **live feed** (`log` / `metric` /
//! `step`). The columns are, in order:
//!
//! | column | type | notes |
//! |---|---|---|
//! | `build_id` | Utf8 | the jera job id (root) — here the provenance record id |
//! | `step_id` | Utf8 | the sub-job id (pipeline node); empty for the root |
//! | `ts_micros` | Int64 | wall-clock micros, stamped by the emitter |
//! | `kind` | Utf8 | `provenance` \| `dep` \| `artifact` \| `log` \| `metric` \| `step` |
//! | `level` | Utf8 | log level / step state |
//! | `name` | Utf8 | free key for provenance & metrics |
//! | `value` | Utf8 | free value |
//! | `content_id` | Utf8 | for `artifact` rows — the holger CAS id (sha256) |
//! | `blob` | Binary (nullable) | small inline payload (e.g. a log chunk) |
//!
//! # Dep-light by construction
//!
//! The rows + the sink are pure `std` + `serde` (no arrow): a leaf repo's xtask
//! gets the emitter for free. The Arrow RecordBatch binding is gated behind the
//! `build-info-arrow` feature so the default crate keeps its no-heavy-deps
//! promise — the arrow crates (and znippy's Arrow-IPC/Flight seal at rest) are
//! only linked when a caller actually wants columnar batches.
//!
//! # Reuse note (design §9)
//!
//! znippy's `build_metadata_batch` builds znippy's *own* blob-index schema (path
//! / chunk_seq / checksum …), which is a different shape from §f3 — so we reuse
//! the **arrow builder plumbing** (`arrow-array` / `arrow-schema`, the same
//! crates + version znippy pins) rather than re-rolling arrow, and shape the §f3
//! columns here. Pulling `znippy-common` wholesale (wasmtime / zoomies / sysinfo)
//! would break this crate's dep-light law.
use std::sync::{Arc, Mutex};
use anyhow::Result;
use crate::image_build::TargetProvenance;
use crate::provenance::BuildProvenance;
/// Schema tag stamped into the Arrow metadata + every logical batch so a reader
/// can reject an unknown shape (mirrors [`crate::provenance::PROVENANCE_SCHEMA`]).
pub const BUILD_INFO_SCHEMA: &str = "nornir.build-info/v1";
/// The kind discriminator for a [`BuildInfoRow`] (the §f3 `kind` column). The
/// first three are the **manifest** (Artifactory-parity) rows; the last three are
/// the **live feed**.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BuildInfoKind {
/// A scalar provenance fact (repo, vcs rev, toolchain, host …).
Provenance,
/// A consumed input / dependency (here: an enabled feature).
Dep,
/// A produced artifact — `content_id` carries its CAS id (sha256).
Artifact,
/// A live log line.
Log,
/// A live metric sample.
Metric,
/// A pipeline step lifecycle event; `level` carries the step state.
Step,
}
impl BuildInfoKind {
/// The exact string written to the `kind` column.
pub fn as_str(self) -> &'static str {
match self {
BuildInfoKind::Provenance => "provenance",
BuildInfoKind::Dep => "dep",
BuildInfoKind::Artifact => "artifact",
BuildInfoKind::Log => "log",
BuildInfoKind::Metric => "metric",
BuildInfoKind::Step => "step",
}
}
}
/// One build-info record — a flat row shaped exactly like the §f3 Arrow schema.
///
/// Dep-light (`std` + `serde` only): the Arrow conversion lives behind the
/// `build-info-arrow` feature ([`build_info_batch`]). Free-form columns default
/// to the empty string, and `blob` is the only nullable column.
#[derive(Debug, Clone, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)]
pub struct BuildInfoRow {
/// The root build id (the provenance record id).
pub build_id: String,
/// The sub-job / pipeline-step id; empty for the root.
pub step_id: String,
/// Wall-clock micros, stamped by the emitter.
pub ts_micros: i64,
/// The record kind (`provenance` / `dep` / `artifact` / `log` / `metric` / `step`).
pub kind: String,
/// Log level / step state.
pub level: String,
/// Free key.
pub name: String,
/// Free value.
pub value: String,
/// For `artifact` rows: the holger CAS id (sha256).
pub content_id: String,
/// Small inline payload (e.g. a log chunk); the only nullable column.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub blob: Option<Vec<u8>>,
}
impl BuildInfoRow {
/// A row of `kind`, stamped `ts_micros`, for the root build (`step_id` empty).
pub fn new(build_id: &str, ts_micros: i64, kind: BuildInfoKind) -> Self {
BuildInfoRow {
build_id: build_id.to_string(),
ts_micros,
kind: kind.as_str().to_string(),
..Default::default()
}
}
/// Fluent: set the pipeline step id.
pub fn step(mut self, step_id: &str) -> Self {
self.step_id = step_id.to_string();
self
}
/// Fluent: set `level`.
pub fn level(mut self, level: &str) -> Self {
self.level = level.to_string();
self
}
/// Fluent: set the `name`/`value` k/v pair.
pub fn kv(mut self, name: &str, value: impl Into<String>) -> Self {
self.name = name.to_string();
self.value = value.into();
self
}
/// Fluent: set the artifact CAS id.
pub fn content_id(mut self, content_id: &str) -> Self {
self.content_id = content_id.to_string();
self
}
/// Fluent: attach an inline blob payload.
pub fn blob(mut self, blob: Vec<u8>) -> Self {
self.blob = Some(blob);
self
}
}
/// Current wall-clock in micros — the emitter's timestamp source (§f3
/// "stamped by the emitter").
pub fn now_micros() -> i64 {
chrono::Utc::now().timestamp_micros()
}
/// Derive the **manifest rows** (`provenance` / `dep` / `artifact`) for a captured
/// [`BuildProvenance`] — the Artifactory-parity record (design §7). Every row is
/// stamped with `ts_micros` and tagged with the record's id as `build_id`.
///
/// - **provenance** rows: one k/v per scalar fact (repo, variant, resolved
/// inputs, toolchain, host, outcome, builder …).
/// - **dep** rows: one per enabled feature (the consumed inputs present on the
/// record; a resolved-crate-graph dep row is a later slice — see problems file).
/// - **artifact** rows: one per produced artifact, `content_id` = its sha256.
pub fn manifest_rows(prov: &BuildProvenance, ts_micros: i64) -> Vec<BuildInfoRow> {
let id = prov.id.as_str();
let p = |name: &str, value: String| {
BuildInfoRow::new(id, ts_micros, BuildInfoKind::Provenance).kv(name, value)
};
let mut rows = vec![
p("repo", prov.repo.clone()),
p("variant", prov.variant.name.clone()),
p("schema", prov.schema.clone()),
p("profile", prov.inputs.profile.clone()),
p("target", prov.inputs.target.clone().unwrap_or_default()),
p("cargo_args", prov.inputs.cargo_args.join(" ")),
p("vcs_rev", prov.inputs.vcs_rev.clone()),
p("vcs_branch", prov.inputs.vcs_branch.clone()),
p("vcs_dirty", prov.inputs.vcs_dirty.to_string()),
p("worktree_digest", prov.inputs.worktree_digest.clone()),
p("rustc", prov.toolchain.rustc.clone()),
p("cargo", prov.toolchain.cargo.clone()),
p("host_triple", prov.toolchain.host.clone()),
p("cores", prov.host.cores.to_string()),
p("memory_bytes", prov.host.memory_bytes.to_string()),
p("outcome", format!("{:?}", prov.outcome).to_lowercase()),
p("built_at", prov.built_at.clone()),
p("builder", prov.builder.clone()),
];
// dep rows — the consumed inputs present on the record (enabled features).
for f in &prov.inputs.features {
rows.push(
BuildInfoRow::new(id, ts_micros, BuildInfoKind::Dep).kv("feature", f.clone()),
);
}
// artifact rows — content-addressed by sha256 (the holger CAS id).
for a in &prov.artifacts {
rows.push(
BuildInfoRow::new(id, ts_micros, BuildInfoKind::Artifact)
.kv(&a.path, a.bytes.to_string())
.content_id(&a.sha256),
);
}
rows
}
/// Derive the **manifest rows** (`provenance` / `dep` / `artifact`) for a captured
/// [`TargetProvenance`] — the image/VM analogue of [`manifest_rows`], so a
/// container/appliance build lowers into the SAME §f3 lineage triple as a cargo
/// build. The unified `assemble()` seam ([`crate::assemble`]) uses this so a
/// container/VM receipt graphs by `content_id` exactly like an executable's.
///
/// - **provenance** rows: one k/v per scalar fact (repo, target, kind, engine,
/// descriptor fingerprint, host, VCS, outcome …).
/// - **dep** rows: the descriptor's own `descriptor_id` (the SPEC hash consumed).
/// - **artifact** rows: one per produced artifact (image digest / kernel+rootfs),
/// `content_id` = its sha256 — the graph key across builds.
pub fn target_manifest_rows(prov: &TargetProvenance, ts_micros: i64) -> Vec<BuildInfoRow> {
let id = prov.id.as_str();
let p = |name: &str, value: String| {
BuildInfoRow::new(id, ts_micros, BuildInfoKind::Provenance).kv(name, value)
};
let mut rows = vec![
p("repo", prov.repo.clone()),
p("target", prov.target.clone()),
p("kind", prov.kind.as_str().to_string()),
p("schema", prov.schema.clone()),
p("descriptor_id", prov.descriptor_id.clone()),
p("engine", prov.engine.clone()),
p("cores", prov.host.cores.to_string()),
p("memory_bytes", prov.host.memory_bytes.to_string()),
p("vcs_rev", prov.vcs_rev.clone()),
p("vcs_branch", prov.vcs_branch.clone()),
p("vcs_dirty", prov.vcs_dirty.to_string()),
p("outcome", format!("{:?}", prov.outcome).to_lowercase()),
p("built_at", prov.built_at.clone()),
p("builder", prov.builder.clone()),
];
// dep row — the descriptor spec fingerprint (what was asked for), the consumed
// input distinct from the produced-artifact hashes below.
rows.push(
BuildInfoRow::new(id, ts_micros, BuildInfoKind::Dep)
.kv("descriptor_id", prov.descriptor_id.clone()),
);
// artifact rows — content-addressed by sha256 (image digest / kernel+rootfs).
for a in &prov.artifacts {
rows.push(
BuildInfoRow::new(id, ts_micros, BuildInfoKind::Artifact)
.kv(&a.path, a.bytes.to_string())
.content_id(&a.sha256),
);
}
rows
}
/// Derive the **live-feed rows** (`step` / `log` / `metric`) for a finished
/// `cargo build` step of a variant. Complements [`manifest_rows`]: together they
/// are the whole §f3 stream for one build.
pub fn build_feed_rows(
build_id: &str,
ts_micros: i64,
variant: &str,
success: bool,
artifact_count: usize,
cores: u32,
elapsed_micros: i64,
) -> Vec<BuildInfoRow> {
let step = "cargo-build";
let state = if success { "done" } else { "failed" };
vec![
BuildInfoRow::new(build_id, ts_micros, BuildInfoKind::Step)
.step(step)
.level(state)
.kv("cargo build", variant.to_string()),
BuildInfoRow::new(build_id, ts_micros, BuildInfoKind::Log)
.step(step)
.level(if success { "info" } else { "error" })
.kv("cargo", format!("{artifact_count} artifact(s), {state}")),
BuildInfoRow::new(build_id, ts_micros, BuildInfoKind::Metric)
.step(step)
.kv("artifacts", artifact_count.to_string()),
BuildInfoRow::new(build_id, ts_micros, BuildInfoKind::Metric)
.step(step)
.kv("cores", cores.to_string()),
BuildInfoRow::new(build_id, ts_micros, BuildInfoKind::Metric)
.step(step)
.kv("elapsed_micros", elapsed_micros.to_string()),
]
}
/// The **build-info sink** seam (mirrors [`crate::sink::BuildProvenanceSink`]).
///
/// A sink receives batches of [`BuildInfoRow`]s and routes them somewhere:
/// the in-process [`InMemBuildInfoSink`] (the default; single-box localhost —
/// zero transport tax), a `NullBuildInfoSink` no-op, or — a LATER slice — the
/// `nornir-flight` seam for cross-host live streaming. The emitter only deals in
/// rows; the transport is picked by deployment. Kept `Send + Sync` so it can be
/// shared across the build's threads.
pub trait BuildInfoSink: Send + Sync {
/// Emit a batch of rows. Best-effort transports may buffer; the in-mem
/// default appends synchronously.
fn emit(&self, rows: &[BuildInfoRow]) -> Result<()>;
}
/// Discards rows — the no-op default so existing [`crate::build_variant`] callers
/// pay nothing.
#[derive(Debug, Default, Clone, Copy)]
pub struct NullBuildInfoSink;
impl BuildInfoSink for NullBuildInfoSink {
fn emit(&self, _rows: &[BuildInfoRow]) -> Result<()> {
Ok(())
}
}
/// The **in-process default** sink: accumulates rows in a shared `Vec` (a
/// zero-transport localhost seam). Cloneable — every clone shares one buffer —
/// so a caller can hand a clone to the build and read the rows back afterwards.
#[derive(Debug, Default, Clone)]
pub struct InMemBuildInfoSink {
rows: Arc<Mutex<Vec<BuildInfoRow>>>,
}
impl InMemBuildInfoSink {
/// A fresh, empty in-process sink.
pub fn new() -> Self {
Self::default()
}
/// Snapshot of all rows emitted so far.
pub fn rows(&self) -> Vec<BuildInfoRow> {
self.rows.lock().expect("build-info sink mutex").clone()
}
/// Take all rows, leaving the buffer empty.
pub fn drain(&self) -> Vec<BuildInfoRow> {
std::mem::take(&mut *self.rows.lock().expect("build-info sink mutex"))
}
/// Count of rows accumulated so far.
pub fn len(&self) -> usize {
self.rows.lock().expect("build-info sink mutex").len()
}
/// Whether no rows have been emitted.
pub fn is_empty(&self) -> bool {
self.len() == 0
}
}
impl BuildInfoSink for InMemBuildInfoSink {
fn emit(&self, rows: &[BuildInfoRow]) -> Result<()> {
self.rows
.lock()
.expect("build-info sink mutex")
.extend_from_slice(rows);
Ok(())
}
}
// ─── Arrow RecordBatch binding (feature `build-info-arrow`) ───────────────────
//
// Reuses the arrow builder plumbing (arrow-array / arrow-schema, the crates +
// version znippy pins) to shape the §f3 columns. Only linked when a caller wants
// columnar batches — the default crate stays dep-light.
#[cfg(feature = "build-info-arrow")]
mod arrow_batch {
use std::sync::Arc;
use arrow_array::builder::{BinaryBuilder, Int64Builder, StringBuilder};
use arrow_array::{ArrayRef, RecordBatch};
use arrow_schema::{ArrowError, DataType, Field, Schema};
use super::{BuildInfoRow, BUILD_INFO_SCHEMA};
/// The §f3 Arrow schema, columns in declared order. `blob` is the only
/// nullable column; the schema carries a `schema` metadata key set to
/// [`BUILD_INFO_SCHEMA`].
pub fn build_info_schema() -> Schema {
let fields = vec![
Field::new("build_id", DataType::Utf8, false),
Field::new("step_id", DataType::Utf8, false),
Field::new("ts_micros", DataType::Int64, false),
Field::new("kind", DataType::Utf8, false),
Field::new("level", DataType::Utf8, false),
Field::new("name", DataType::Utf8, false),
Field::new("value", DataType::Utf8, false),
Field::new("content_id", DataType::Utf8, false),
Field::new("blob", DataType::Binary, true),
];
let mut meta = std::collections::HashMap::new();
meta.insert("schema".to_string(), BUILD_INFO_SCHEMA.to_string());
Schema::new(fields).with_metadata(meta)
}
/// Build one Arrow [`RecordBatch`] over `rows`, matching [`build_info_schema`].
pub fn build_info_batch(rows: &[BuildInfoRow]) -> Result<RecordBatch, ArrowError> {
let n = rows.len();
let mut build_id = StringBuilder::with_capacity(n, n * 40);
let mut step_id = StringBuilder::with_capacity(n, n * 16);
let mut ts = Int64Builder::with_capacity(n);
let mut kind = StringBuilder::with_capacity(n, n * 10);
let mut level = StringBuilder::with_capacity(n, n * 8);
let mut name = StringBuilder::with_capacity(n, n * 16);
let mut value = StringBuilder::with_capacity(n, n * 32);
let mut content_id = StringBuilder::with_capacity(n, n * 64);
let mut blob = BinaryBuilder::with_capacity(n, n * 32);
for r in rows {
build_id.append_value(&r.build_id);
step_id.append_value(&r.step_id);
ts.append_value(r.ts_micros);
kind.append_value(&r.kind);
level.append_value(&r.level);
name.append_value(&r.name);
value.append_value(&r.value);
content_id.append_value(&r.content_id);
match &r.blob {
Some(b) => blob.append_value(b),
None => blob.append_null(),
}
}
let columns: Vec<ArrayRef> = vec![
Arc::new(build_id.finish()),
Arc::new(step_id.finish()),
Arc::new(ts.finish()),
Arc::new(kind.finish()),
Arc::new(level.finish()),
Arc::new(name.finish()),
Arc::new(value.finish()),
Arc::new(content_id.finish()),
Arc::new(blob.finish()),
];
RecordBatch::try_new(Arc::new(build_info_schema()), columns)
}
}
#[cfg(feature = "build-info-arrow")]
pub use arrow_batch::{build_info_batch, build_info_schema};
#[cfg(test)]
mod tests {
use super::*;
use crate::provenance::{BuildProvenance, HostInfo, Outcome, Toolchain};
use crate::variant::{BuildVariant, Profile};
/// A captured record with one artifact and two features (no cargo run).
fn sample() -> (tempfile::TempDir, BuildProvenance) {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
let target = root.join("target/release");
std::fs::create_dir_all(&target).unwrap();
let art = target.join("holger-full");
std::fs::write(&art, b"ELF holger-full bytes").unwrap();
let variant = BuildVariant {
name: "full".into(),
bin: Some("holger-full".into()),
features: vec!["full".into(), "gui".into()],
profile: Profile::Release,
..Default::default()
};
let rec = BuildProvenance::capture(
&variant,
"holger",
root,
Toolchain { rustc: "rustc 1.85.0".into(), host: "x86_64-unknown-linux-gnu".into(), cargo: "cargo 1.85.0".into(), ..Default::default() },
HostInfo { cores: 12, memory_bytes: 64 * 1024 * 1024 * 1024 },
Outcome::Success,
&[art],
)
.unwrap();
(dir, rec)
}
#[test]
fn manifest_rows_cover_provenance_dep_and_artifact() {
let (_d, rec) = sample();
let rows = manifest_rows(&rec, 42);
// Every row is stamped + tagged with the build id (§f3 build_id column).
assert!(rows.iter().all(|r| r.ts_micros == 42 && r.build_id == rec.id));
// provenance: the repo fact and the content-addressed host facts landed.
let repo = rows.iter().find(|r| r.kind == "provenance" && r.name == "repo").unwrap();
assert_eq!(repo.value, "holger");
assert!(rows.iter().any(|r| r.name == "cores" && r.value == "12"));
assert!(rows.iter().any(|r| r.name == "vcs_rev"));
// dep: one row per enabled feature.
let deps: Vec<&str> = rows
.iter()
.filter(|r| r.kind == "dep")
.map(|r| r.value.as_str())
.collect();
assert_eq!(deps, vec!["full", "gui"]);
// artifact: content_id is the sha256 CAS id (RED-when-broken).
let artifact = rows.iter().find(|r| r.kind == "artifact").unwrap();
assert_eq!(artifact.content_id, rec.artifacts[0].sha256);
assert_eq!(artifact.name, "target/release/holger-full");
assert_eq!(artifact.blob, None);
}
#[test]
fn feed_rows_carry_step_log_and_metrics() {
let rows = build_feed_rows("bid", 7, "full", true, 3, 12, 999);
assert_eq!(rows[0].kind, "step");
assert_eq!(rows[0].step_id, "cargo-build");
assert_eq!(rows[0].level, "done");
assert!(rows.iter().any(|r| r.kind == "log"));
assert!(rows.iter().any(|r| r.kind == "metric" && r.name == "artifacts" && r.value == "3"));
assert!(rows.iter().any(|r| r.kind == "metric" && r.name == "elapsed_micros" && r.value == "999"));
}
#[test]
fn in_mem_sink_round_trips_rows() {
let (_d, rec) = sample();
let sink = InMemBuildInfoSink::new();
assert!(sink.is_empty());
let rows = manifest_rows(&rec, 1);
sink.emit(&rows).unwrap();
sink.emit(&build_feed_rows(&rec.id, 2, "full", true, 1, 12, 5)).unwrap();
// Round-trip: what went in comes back out, in order, byte-identical.
let back = sink.rows();
assert_eq!(back.len(), rows.len() + 5);
assert_eq!(&back[..rows.len()], rows.as_slice());
// drain empties the shared buffer; a clone sees the same buffer.
let clone = sink.clone();
let drained = sink.drain();
assert_eq!(drained.len(), back.len());
assert!(clone.is_empty());
}
#[test]
fn null_sink_is_a_noop() {
let (_d, rec) = sample();
assert!(NullBuildInfoSink.emit(&manifest_rows(&rec, 0)).is_ok());
}
#[cfg(feature = "build-info-arrow")]
#[test]
fn arrow_schema_has_f3_columns_in_order() {
use arrow_schema::DataType;
let schema = build_info_schema();
let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
// RED-when-broken: the exact §f3 column order.
assert_eq!(
names,
vec![
"build_id", "step_id", "ts_micros", "kind", "level", "name", "value",
"content_id", "blob"
]
);
assert_eq!(schema.field(2).data_type(), &DataType::Int64);
assert_eq!(schema.field(8).data_type(), &DataType::Binary);
// blob is the ONLY nullable column.
assert!(schema.field(8).is_nullable());
assert!(schema.fields().iter().take(8).all(|f| !f.is_nullable()));
assert_eq!(schema.metadata().get("schema").map(String::as_str), Some(BUILD_INFO_SCHEMA));
}
#[cfg(feature = "build-info-arrow")]
#[test]
fn arrow_batch_matches_schema_and_values() {
use arrow_array::{cast::AsArray, types::Int64Type, Array};
let (_d, rec) = sample();
let mut rows = manifest_rows(&rec, 55);
rows.push(BuildInfoRow::new(&rec.id, 56, BuildInfoKind::Log).blob(b"hello".to_vec()));
let batch = build_info_batch(&rows).unwrap();
assert_eq!(batch.num_columns(), 9);
assert_eq!(batch.num_rows(), rows.len());
// Column 0 (build_id) is the record id on every row.
let build_ids = batch.column(0).as_string::<i32>();
assert_eq!(build_ids.value(0), rec.id);
// ts_micros round-trips as Int64.
let ts = batch.column(2).as_primitive::<Int64Type>();
assert_eq!(ts.value(0), 55);
// blob column: null on manifest rows, present on the last log row.
let blob = batch.column(8).as_binary::<i32>();
assert!(blob.is_null(0));
assert_eq!(blob.value(rows.len() - 1), b"hello");
}
// ─── functional-status wiring (feature `testmatrix`) ─────────────────────
#[cfg(feature = "testmatrix")]
#[test]
fn emits_functional_rows_for_build_info() {
let (_d, rec) = sample();
let sink = InMemBuildInfoSink::new();
let rows = manifest_rows(&rec, 1);
sink.emit(&rows).unwrap();
let has_prov = rows.iter().any(|r| r.kind == "provenance");
let has_dep = rows.iter().any(|r| r.kind == "dep");
let has_art = rows.iter().filter(|r| r.kind == "artifact").count() == rec.artifacts.len();
crate::bt_assert_emit!(
"nornir-build-thing",
"build_info_manifest_rows_emitted",
has_prov && has_dep && has_art,
"prov={has_prov} dep={has_dep} art={has_art}"
);
let round_trip = sink.rows() == rows;
crate::bt_assert_emit!(
"nornir-build-thing",
"build_info_sink_round_trips",
round_trip,
"rows={}",
sink.len()
);
let drained = nornir_testmatrix::drain_functional_rows();
for want in ["build_info_manifest_rows_emitted", "build_info_sink_round_trips"] {
assert!(
drained.iter().any(|r| r.suite == "nornir-build-thing"
&& r.test_name == want
&& r.status == "pass"),
"missing green row {want}: {drained:?}"
);
}
}
}