src/warehouse/build_provenance.rs
//! Iceberg writer + reader + [`BuildProvenanceSink`] for **build-provenance**
//! (`build_provenance`) — task #36.
//!
//! When `nornir build --variant <v> --attest` captures a
//! [`BuildProvenance`] record, it is written to THIS ledger **first** (the
//! warehouse), then dumped as a znippy archive beside the artifact
//! (`nornir::build::seal`). This module is the warehouse half: one Iceberg
//! snapshot per recorded build, partitioned by `repo` so per-repo history reads
//! prune. It mirrors `warehouse::release_changes` / `warehouse::test_results`.
//!
//! Grain: one row per recorded build (`id` = the record UUID). The full record
//! travels in `record_json`, so a reader reconstructs the complete provenance
//! ([`BuildProvenanceRow::record`]) while the flat columns (variant, digest,
//! outcome, host cores/memory, artifact_count) stay queryable.
use std::sync::Arc;
use anyhow::{Result, anyhow};
use arrow::array::{
Array, Int32Array, Int64Array, RecordBatch, StringArray, TimestampMicrosecondArray,
};
use iceberg::Catalog;
use iceberg::arrow::schema_to_arrow_schema;
use nornir_build_thing::provenance::Outcome;
use nornir_build_thing::{BuildProvenance, BuildProvenanceSink};
use super::iceberg::{IcebergWarehouse, TABLE_BUILD_PROVENANCE, append_batch};
// Column indices match `iceberg_schema::build_provenance` field order.
const COL_ID: usize = 0;
const COL_REPO: usize = 1;
const COL_VARIANT: usize = 2;
const COL_OUTCOME: usize = 3;
const COL_PROFILE: usize = 4;
const COL_TARGET: usize = 5;
const COL_VCS_REV: usize = 6;
const COL_VCS_BRANCH: usize = 7;
const COL_RUSTC: usize = 8;
const COL_HOST_CORES: usize = 9;
const COL_HOST_MEMORY: usize = 10;
const COL_ARTIFACT_COUNT: usize = 11;
const COL_DIGEST: usize = 12;
const COL_BUILT_AT: usize = 13;
const COL_TS_MICROS: usize = 14;
const COL_RECORD_JSON: usize = 15;
/// A flat build-provenance row read back from the warehouse. The complete record
/// is in `record_json` (reconstruct via [`Self::record`]).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BuildProvenanceRow {
pub id: String,
pub repo: String,
pub variant: String,
pub outcome: String,
pub profile: String,
pub target: String,
pub vcs_rev: String,
pub vcs_branch: String,
pub rustc: String,
pub host_cores: i32,
pub host_memory_bytes: i64,
pub artifact_count: i64,
pub digest: String,
pub built_at: String,
pub ts_micros: i64,
pub record_json: String,
}
impl BuildProvenanceRow {
/// Reconstruct the full [`BuildProvenance`] record from `record_json`.
pub fn record(&self) -> Result<BuildProvenance> {
BuildProvenance::from_json(&self.record_json)
}
/// Project a [`BuildProvenance`] into a warehouse row (`digest` = the
/// attestation's SHA-256 over the canonical record).
fn from_record(prov: &BuildProvenance) -> Result<Self> {
let digest = prov.clone().attest()?.digest_hex();
Ok(BuildProvenanceRow {
id: prov.id.clone(),
repo: prov.repo.clone(),
variant: prov.variant.name.clone(),
outcome: match prov.outcome {
Outcome::Success => "success",
Outcome::Failed => "failed",
}
.to_string(),
profile: prov.inputs.profile.clone(),
target: prov.inputs.target.clone().unwrap_or_default(),
vcs_rev: prov.inputs.vcs_rev.clone(),
vcs_branch: prov.inputs.vcs_branch.clone(),
rustc: prov.toolchain.rustc.clone(),
host_cores: prov.host.cores as i32,
host_memory_bytes: prov.host.memory_bytes as i64,
artifact_count: prov.artifacts.len() as i64,
digest,
built_at: prov.built_at.clone(),
ts_micros: chrono::Utc::now().timestamp_micros(),
record_json: prov.to_json()?,
})
}
}
/// Append one build-provenance record (one Iceberg snapshot).
pub async fn append_build_provenance(wh: &IcebergWarehouse, prov: &BuildProvenance) -> Result<()> {
let row = BuildProvenanceRow::from_record(prov)?;
let table = wh
.catalog()
.load_table(&wh.table_ident(TABLE_BUILD_PROVENANCE))
.await?;
let arrow_schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(vec![row.id.clone()])),
Arc::new(StringArray::from(vec![row.repo.clone()])),
Arc::new(StringArray::from(vec![row.variant.clone()])),
Arc::new(StringArray::from(vec![row.outcome.clone()])),
Arc::new(StringArray::from(vec![row.profile.clone()])),
Arc::new(StringArray::from(vec![row.target.clone()])),
Arc::new(StringArray::from(vec![row.vcs_rev.clone()])),
Arc::new(StringArray::from(vec![row.vcs_branch.clone()])),
Arc::new(StringArray::from(vec![row.rustc.clone()])),
Arc::new(Int32Array::from(vec![row.host_cores])),
Arc::new(Int64Array::from(vec![row.host_memory_bytes])),
Arc::new(Int64Array::from(vec![row.artifact_count])),
Arc::new(StringArray::from(vec![row.digest.clone()])),
Arc::new(StringArray::from(vec![row.built_at.clone()])),
Arc::new(TimestampMicrosecondArray::from(vec![row.ts_micros]).with_timezone("+00:00")),
Arc::new(StringArray::from(vec![row.record_json.clone()])),
];
let batch = RecordBatch::try_new(arrow_schema, cols)?;
append_batch(wh.catalog(), table, batch).await?;
Ok(())
}
/// Read build-provenance rows, optionally scoped to `repo`, newest-first
/// (`ts_micros` desc).
pub async fn query_build_provenance(
wh: &IcebergWarehouse,
repo: Option<&str>,
) -> Result<Vec<BuildProvenanceRow>> {
let batches: Vec<RecordBatch> =
super::iceberg::load_and_read_all(wh, TABLE_BUILD_PROVENANCE).await?;
let mut out: Vec<BuildProvenanceRow> = Vec::new();
for b in &batches {
let id = col_str(b, COL_ID)?;
let repo_c = col_str(b, COL_REPO)?;
let variant = col_str(b, COL_VARIANT)?;
let outcome = col_str(b, COL_OUTCOME)?;
let profile = col_str(b, COL_PROFILE)?;
let target = col_str(b, COL_TARGET)?;
let vcs_rev = col_str(b, COL_VCS_REV)?;
let vcs_branch = col_str(b, COL_VCS_BRANCH)?;
let rustc = col_str(b, COL_RUSTC)?;
let cores = col_i32(b, COL_HOST_CORES)?;
let mem = col_i64(b, COL_HOST_MEMORY)?;
let artifacts = col_i64(b, COL_ARTIFACT_COUNT)?;
let digest = col_str(b, COL_DIGEST)?;
let built_at = col_str(b, COL_BUILT_AT)?;
let ts = col_ts(b, COL_TS_MICROS)?;
let record_json = col_str(b, COL_RECORD_JSON)?;
for i in 0..b.num_rows() {
let row = BuildProvenanceRow {
id: id.value(i).to_string(),
repo: repo_c.value(i).to_string(),
variant: variant.value(i).to_string(),
outcome: outcome.value(i).to_string(),
profile: profile.value(i).to_string(),
target: target.value(i).to_string(),
vcs_rev: vcs_rev.value(i).to_string(),
vcs_branch: vcs_branch.value(i).to_string(),
rustc: rustc.value(i).to_string(),
host_cores: cores.value(i),
host_memory_bytes: mem.value(i),
artifact_count: artifacts.value(i),
digest: digest.value(i).to_string(),
built_at: built_at.value(i).to_string(),
ts_micros: ts.value(i),
record_json: record_json.value(i).to_string(),
};
if repo.map(|r| row.repo == r).unwrap_or(true) {
out.push(row);
}
}
}
out.sort_by(|a, b| b.ts_micros.cmp(&a.ts_micros).then_with(|| a.id.cmp(&b.id)));
Ok(out)
}
/// The warehouse is the canonical build-provenance sink: `wh.record(&prov)`
/// writes one ledger row via the shared runtime.
impl BuildProvenanceSink for IcebergWarehouse {
fn record(&self, prov: &BuildProvenance) -> Result<()> {
self.block_on(append_build_provenance(self, prov))
}
}
// ─── column helpers ─────────────────────────────────────────────────────
fn col_str<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a StringArray> {
b.column(idx)
.as_any()
.downcast_ref::<StringArray>()
.ok_or_else(|| anyhow!("build_provenance col {idx} is not StringArray"))
}
fn col_i32<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a Int32Array> {
b.column(idx)
.as_any()
.downcast_ref::<Int32Array>()
.ok_or_else(|| anyhow!("build_provenance col {idx} is not Int32Array"))
}
fn col_i64<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a Int64Array> {
b.column(idx)
.as_any()
.downcast_ref::<Int64Array>()
.ok_or_else(|| anyhow!("build_provenance col {idx} is not Int64Array"))
}
fn col_ts<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a TimestampMicrosecondArray> {
b.column(idx)
.as_any()
.downcast_ref::<TimestampMicrosecondArray>()
.ok_or_else(|| anyhow!("build_provenance col {idx} is not TimestampMicrosecondArray"))
}
#[cfg(test)]
mod tests {
use super::*;
use nornir_build_thing::provenance::{HostInfo, Toolchain};
use nornir_build_thing::variant::{BuildVariant, Profile};
fn sample_record(variant: &str) -> BuildProvenance {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
std::fs::create_dir_all(root.join("target/release")).unwrap();
let art = root
.join("target/release")
.join(format!("holger-{variant}"));
std::fs::write(&art, format!("ELF {variant}").as_bytes()).unwrap();
let v = BuildVariant {
name: variant.to_string(),
package: Some(format!("holger-{variant}")),
bin: Some(format!("holger-{variant}")),
features: vec![variant.to_string()],
profile: Profile::Release,
..Default::default()
};
BuildProvenance::capture(
&v,
"holger",
root,
Toolchain {
rustc: "rustc 1.85.0".into(),
host: "x86_64".into(),
cargo: "cargo 1.85.0".into(),
// The constellation's pinned build image, as a real record carries it.
image: Some("nordisk-cargo-toolchain".into()),
image_sha256: Some(
"3d5115baa07d03ca51eb3e44ebb86b4e64e679bd88f38e156ef9da12c35d2143".into(),
),
},
HostInfo {
cores: 32,
memory_bytes: 128 * 1024 * 1024 * 1024,
},
Outcome::Success,
&[art],
)
.unwrap()
}
#[test]
fn warehouse_sink_round_trips_build_provenance() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
// Record all three holger exes through the SINK (the write-warehouse-first path).
for v in ["embedded", "minimal", "full"] {
wh.record(&sample_record(v)).unwrap();
}
let rows = wh
.block_on(query_build_provenance(&wh, Some("holger")))
.unwrap();
assert_eq!(rows.len(), 3, "three builds recorded");
let variants: std::collections::BTreeSet<&str> =
rows.iter().map(|r| r.variant.as_str()).collect();
assert_eq!(
variants,
["embedded", "full", "minimal"].into_iter().collect()
);
// Host facts + artifact hashes survive the round-trip via record_json.
let full = rows.iter().find(|r| r.variant == "full").unwrap();
assert_eq!(full.host_cores, 32);
assert_eq!(full.host_memory_bytes, 128 * 1024 * 1024 * 1024);
assert_eq!(full.artifact_count, 1);
assert_eq!(full.outcome, "success");
assert_eq!(full.digest.len(), 64, "sha256 hex digest stored");
// The full record reconstructs, and its digest matches the flat column.
let rec = full.record().unwrap();
assert_eq!(rec.variant.name, "full");
assert_eq!(rec.artifacts.len(), 1);
assert_eq!(rec.clone().attest().unwrap().digest_hex(), full.digest);
}
/// **A NON-cargo cell's record is readable back out of the ledger.**
///
/// ⚒ Build Thing's records pane was empty for a week because a container / ISO /
/// VM cell sealed a `TargetProvenance` and this table only holds
/// [`BuildProvenance`], so the producer filed nothing at all. The reconciliation
/// is `TargetProvenance::to_build_provenance` — and a projection nobody can read
/// back is the same emptiness wearing the other mask, so this drives the folded
/// record through the SAME sink and the SAME reader the cargo rows use.
///
/// The numbers are the ones oden's forge really produced on 2026-08-03 (image
/// `gunnar-scratch:0.1.0`, 15 083 584 B), so a row that comes back wrong comes
/// back visibly wrong rather than merely different.
///
/// RED-PROVED: replace `to_build_provenance`'s `artifacts: self.artifacts.clone()`
/// with `Vec::new()` and `artifact_count` reads 0 with an empty `rec.artifacts` —
/// which is exactly the useless row a summary-only projection would have filed.
#[test]
fn a_folded_target_record_round_trips_as_a_build_provenance_row() {
use nornir_build_thing::container::ContainerBuild;
use nornir_build_thing::image_build::{BuiltImage, ImageBuilder, build_container};
use nornir_build_thing::variant::ArtifactKind;
use std::path::Path;
/// The engine seam, standing in for bollard: it hands back the identity of an
/// image that was really built on oden, so the assertions below are about a
/// real produced artifact and not a fixture-shaped one.
struct OdenImage;
impl ImageBuilder for OdenImage {
fn build(&self, _i: &ContainerBuild, _root: &Path) -> anyhow::Result<BuiltImage> {
Ok(BuiltImage {
image_ref: "gunnar-scratch:0.1.0".into(),
digest:
"sha256:d71ab6d94bcf1e7600000000000000000000000000000000000000000000d71a"
.into(),
bytes: 15_083_584,
// No `archive` on the declared cell below (it takes
// `..Default::default()`), so the builder exported none. Keeping this
// `None` is what makes the `artifacts.len() == 1` assertions below mean
// "the image IS the whole delivery" — `BuiltImage::to_artifacts` appends
// the archive when there is one, so a `Some` here would silently make
// this a 2-artifact row and change what the test proves.
archive: None,
})
}
fn engine(&self) -> &str {
"bollard"
}
}
let dir = tempfile::tempdir().unwrap();
let image = ContainerBuild {
name: "gunnar-scratch".into(),
base: "scratch".into(),
entrypoint: vec!["/gunnar".into()],
tags: vec!["gunnar-scratch:0.1.0".into()],
..Default::default()
};
// The REAL capture path — the same `build_container` the forge's image cell
// runs, so the record folded below is one the builder actually sealed.
let outcome = build_container(&OdenImage, &image, "gunnar", dir.path()).unwrap();
let target = &outcome.attestation.record;
assert_eq!(target.engine, "bollard");
assert_eq!(
target.artifacts.len(),
1,
"the built image is the sealed artifact"
);
// The DECLARED cell — what the pane's row must name.
let cell = BuildVariant {
name: "gunnar-image".into(),
kind: ArtifactKind::Container,
container_ref: Some("gunnar-scratch".into()),
..Default::default()
};
let folded = target.to_build_provenance(&cell);
let wh = IcebergWarehouse::open(&dir.path().join("wh")).unwrap();
wh.record(&folded).unwrap();
// ── read it back through the ledger's own reader ─────────────────────────
let rows = wh
.block_on(query_build_provenance(&wh, Some("gunnar")))
.unwrap();
assert_eq!(rows.len(), 1, "the assembled cell's row is IN the table");
let row = &rows[0];
assert_eq!(
row.variant, "gunnar-image",
"the flat column names the CELL"
);
assert_eq!(row.outcome, "success");
assert_eq!(row.artifact_count, 1, "and it counts the produced image");
assert_eq!(row.digest.len(), 64);
assert_eq!(
row.id, target.id,
"keyed by the TARGET record's id, so build_info joins"
);
assert!(!row.built_at.is_empty());
// ── and the full record reconstructs, carrying what makes it useful ──────
let rec = row.record().unwrap();
assert_eq!(
rec.variant.kind,
ArtifactKind::Container,
"which ArtifactKind"
);
assert_eq!(rec.artifacts.len(), 1);
let art = &rec.artifacts[0];
assert_eq!(art.path, "gunnar-scratch:0.1.0", "the ref that was built");
assert_eq!(art.bytes, 15_083_584, "the size");
assert_eq!(
art.sha256, "d71ab6d94bcf1e7600000000000000000000000000000000000000000000d71a",
"the digest, with the engine's `sha256:` prefix stripped"
);
// The stored digest is a real attestation over THIS record, not a copy of a
// field: recomputing it from the reconstructed record has to agree.
assert_eq!(rec.clone().attest().unwrap().digest_hex(), row.digest);
// No cargo ran for the image itself, and the record says so rather than
// inventing a plausible argv.
assert!(rec.inputs.cargo_args.is_empty());
}
}