src/warehouse/airgap_events.rs
//! Iceberg writer/reader for the **airgap-op DAG** (`airgap_events`) — EPIC
//! AIRGAP / AIRGAP8.
//!
//! The durable twin of the lean `skidbladnir` crate's JSONL event sink
//! (`skidbladnir::events`): on an air-gapped target with no warehouse server
//! the crate appends events to `<warehouse_root>/airgap_events.jsonl`; when the
//! fat `nornir` CLI runs against a real warehouse, [`append_airgap_events`]
//! ingests those same rows into the historized Iceberg table so the bring-up DAG
//! is queryable + time-travelable, exactly like `release_events`.
//!
//! Schema: [`super::iceberg_schema::airgap_events`].
use std::path::Path;
use std::sync::Arc;
use anyhow::{Result, anyhow};
use arrow::array::{
Array, Int64Array, ListArray, ListBuilder, RecordBatch, StringArray, StringBuilder,
TimestampMicrosecondArray,
};
use iceberg::Catalog;
use iceberg::arrow::schema_to_arrow_schema;
use super::iceberg::{IcebergWarehouse, TABLE_AIRGAP_EVENTS, append_batch};
const COL_RUN_ID: usize = 0;
const COL_SEQ: usize = 1;
const COL_TS_MICROS: usize = 2;
const COL_PRODUCT: usize = 3;
const COL_OP: usize = 4;
const COL_PHASE: usize = 5;
const COL_STATUS: usize = 6;
const COL_DETAIL: usize = 7;
const COL_DEPENDS_ON: usize = 8;
/// One row of the `airgap_events` table — identical fields to the lean crate's
/// `AirgapEvent`, re-declared here so the warehouse layer needn't depend on the
/// `skidbladnir` crate (the JSONL bytes are the contract).
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct AirgapEventRow {
pub run_id: String,
pub seq: i64,
pub ts_micros: i64,
pub product: String,
pub op: String,
pub phase: String,
pub status: String,
pub detail: String,
/// Bring-up producers this op waited on (None = root op).
pub depends_on: Option<Vec<String>>,
}
/// Read a `skidbladnir` JSONL event file (one JSON object per line) into rows.
/// `depends_on` becomes `None` when empty (matches the optional Iceberg column).
pub fn read_jsonl(path: &Path) -> Result<Vec<AirgapEventRow>> {
#[derive(serde::Deserialize)]
struct Raw {
run_id: String,
seq: i64,
ts_micros: i64,
product: String,
op: String,
phase: String,
status: String,
#[serde(default)]
detail: String,
#[serde(default)]
depends_on: Vec<String>,
}
let text = std::fs::read_to_string(path)?;
let mut rows = Vec::new();
for line in text.lines().filter(|l| !l.trim().is_empty()) {
let r: Raw = serde_json::from_str(line)?;
rows.push(AirgapEventRow {
run_id: r.run_id,
seq: r.seq,
ts_micros: r.ts_micros,
product: r.product,
op: r.op,
phase: r.phase,
status: r.status,
detail: r.detail,
depends_on: if r.depends_on.is_empty() {
None
} else {
Some(r.depends_on)
},
});
}
rows.sort_by_key(|r| (r.run_id.clone(), r.seq));
Ok(rows)
}
/// Append a batch of `airgap_events` rows (one Iceberg snapshot). Mirrors
/// [`super::release_events::append_release_events`].
pub async fn append_airgap_events(wh: &IcebergWarehouse, rows: &[AirgapEventRow]) -> Result<()> {
if rows.is_empty() {
return Ok(());
}
let table = wh
.catalog()
.load_table(&wh.table_ident(TABLE_AIRGAP_EVENTS))
.await?;
let arrow_schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let dep_elem = match arrow_schema.field(COL_DEPENDS_ON).data_type() {
arrow::datatypes::DataType::List(elem) | arrow::datatypes::DataType::LargeList(elem) => {
elem.clone()
}
other => return Err(anyhow!("depends_on column expected List, got {other:?}")),
};
let mut deps_b: ListBuilder<StringBuilder> =
ListBuilder::new(StringBuilder::new()).with_field(dep_elem);
for r in rows {
match &r.depends_on {
None => deps_b.append(false),
Some(v) => {
for d in v {
deps_b.values().append_value(d);
}
deps_b.append(true);
}
}
}
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(
rows.iter().map(|r| r.run_id.clone()).collect::<Vec<_>>(),
)),
Arc::new(Int64Array::from(
rows.iter().map(|r| r.seq).collect::<Vec<_>>(),
)),
Arc::new(
TimestampMicrosecondArray::from(rows.iter().map(|r| r.ts_micros).collect::<Vec<_>>())
.with_timezone("+00:00"),
),
Arc::new(StringArray::from(
rows.iter().map(|r| r.product.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.op.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.phase.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.status.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.detail.clone()).collect::<Vec<_>>(),
)),
Arc::new(deps_b.finish()),
];
let batch = RecordBatch::try_new(arrow_schema, cols)?;
append_batch(wh.catalog(), table, batch).await?;
Ok(())
}
/// Read the historized `airgap_events` DAG back out of the warehouse Iceberg
/// table — the READ twin of [`append_airgap_events`]. Optionally scoped to one
/// `run_id` (`None` = every recorded run). Rows come back sorted by
/// `(run_id, seq)`, exactly like [`read_jsonl`], so the same
/// [`events_outcome`] renderer shapes either source. Blank/torn Iceberg
/// metadata → 0 rows, never a JSON-EOF crash (load+read).
pub async fn query_airgap_events(
wh: &IcebergWarehouse,
run_id: Option<&str>,
) -> Result<Vec<AirgapEventRow>> {
let batches: Vec<RecordBatch> =
super::iceberg::load_and_read_all(wh, TABLE_AIRGAP_EVENTS).await?;
let mut out: Vec<AirgapEventRow> = Vec::new();
for b in &batches {
let run = col_str(b, COL_RUN_ID)?;
let seq = col_i64(b, COL_SEQ)?;
let ts = col_ts(b, COL_TS_MICROS)?;
let product = col_str(b, COL_PRODUCT)?;
let op = col_str(b, COL_OP)?;
let phase = col_str(b, COL_PHASE)?;
let status = col_str(b, COL_STATUS)?;
let detail = col_str(b, COL_DETAIL)?;
let deps = col_str_list(b, COL_DEPENDS_ON)?;
for i in 0..b.num_rows() {
if let Some(want) = run_id {
if run.value(i) != want {
continue;
}
}
out.push(AirgapEventRow {
run_id: run.value(i).to_string(),
seq: seq.value(i),
ts_micros: ts.value(i),
product: product.value(i).to_string(),
op: op.value(i).to_string(),
phase: phase.value(i).to_string(),
status: status.value(i).to_string(),
detail: detail.value(i).to_string(),
// Normalise an empty list to None, matching read_jsonl's contract.
depends_on: deps[i].as_ref().filter(|d| !d.is_empty()).cloned(),
});
}
}
out.sort_by(|a, b| (a.run_id.as_str(), a.seq).cmp(&(b.run_id.as_str(), b.seq)));
Ok(out)
}
fn col_str(b: &RecordBatch, idx: usize) -> Result<&StringArray> {
b.column(idx)
.as_any()
.downcast_ref::<StringArray>()
.ok_or_else(|| anyhow!("airgap_events col {idx} is not StringArray"))
}
fn col_i64(b: &RecordBatch, idx: usize) -> Result<&Int64Array> {
b.column(idx)
.as_any()
.downcast_ref::<Int64Array>()
.ok_or_else(|| anyhow!("airgap_events col {idx} is not Int64Array"))
}
fn col_ts(b: &RecordBatch, idx: usize) -> Result<&TimestampMicrosecondArray> {
b.column(idx)
.as_any()
.downcast_ref::<TimestampMicrosecondArray>()
.ok_or_else(|| anyhow!("airgap_events col {idx} is not TimestampMicrosecondArray"))
}
fn col_str_list(b: &RecordBatch, idx: usize) -> Result<Vec<Option<Vec<String>>>> {
let arr = b
.column(idx)
.as_any()
.downcast_ref::<ListArray>()
.ok_or_else(|| anyhow!("airgap_events col {idx} is not ListArray"))?;
let mut out = Vec::with_capacity(arr.len());
for i in 0..arr.len() {
if arr.is_null(i) {
out.push(None);
continue;
}
let values = arr.value(i);
let strs = values
.as_any()
.downcast_ref::<StringArray>()
.ok_or_else(|| anyhow!("airgap_events depends_on element not StringArray"))?;
let mut v = Vec::with_capacity(strs.len());
for j in 0..strs.len() {
if !strs.is_null(j) {
v.push(strs.value(j).to_string());
}
}
out.push(Some(v));
}
Ok(out)
}
// ─── AUT9 CLI command outcomes ───────────────────────────────────────────────
// The airgap READ commands (`events`, `fetch`/verify, `unfold`/plan) rendered as
// the uniform `crate::cli_outcome::CommandOutcome` — the CLI twin of a Facet's
// `state_json`. Each builder takes ALREADY-fetched data (rows / a verify report /
// a rendered plan) and only SHAPES it; it never reads disk or runs a backend
// (CommandOutcome is presentation, never a transport). `ok ⟺ a real, non-empty,
// non-failed result` (RAGNARÖK: empty / tampered ⇒ RED, never silently green).
// See `.nornir/cli-command-contract.md`.
/// `airgap events` — the bring-up DAG read back as a [`CommandOutcome`]. `rows`
/// are already parsed from the JSONL sink (`read_jsonl`); this only shapes +
/// renders them. `ok ⟺ ≥1 event` (an empty/absent DAG is RED).
pub fn events_outcome(path: &Path, rows: &[AirgapEventRow]) -> crate::cli_outcome::CommandOutcome {
use crate::cli_outcome::CommandOutcome;
if rows.is_empty() {
return CommandOutcome::fail(
"airgap events",
format!(
"no airgap_events at {} (run `nornir airgap start`)",
path.display()
),
);
}
let mut human = String::new();
for ev in rows {
let deps = ev.depends_on.as_ref().filter(|d| !d.is_empty());
human.push_str(&format!(
" [{}] {:>3} {:<14} {:<8} {:<5} {:<7} {} {}\n",
ev.run_id,
ev.seq,
ev.product,
ev.op,
ev.phase,
ev.status,
ev.detail,
match deps {
Some(d) => format!("← {}", d.join(",")),
None => String::new(),
}
));
}
let data = serde_json::json!({ "events": rows });
CommandOutcome::ok("airgap events", data, human.trim_end().to_string())
}
/// `airgap fetch` (AIRGAP3 — verify a received bundle) as a [`CommandOutcome`].
/// Takes the already-computed verify fields (from
/// `skidbladnir::fetch::verify_on_receipt`). `ok ⟺ manifest+contents intact`
/// (a tampered/failed bundle is RED, carrying the bad entries in `data`).
pub fn verify_outcome(
bundle: &Path,
manifest_root_ok: bool,
bundle_hash_ok: bool,
bad_entries: &[String],
) -> crate::cli_outcome::CommandOutcome {
use crate::cli_outcome::CommandOutcome;
let ok = manifest_root_ok && bundle_hash_ok && bad_entries.is_empty();
let data = serde_json::json!({
"bundle": bundle.display().to_string(),
"manifest_root_ok": manifest_root_ok,
"bundle_hash_ok": bundle_hash_ok,
"bad_entries": bad_entries,
"ok": ok,
});
if ok {
CommandOutcome::ok(
"airgap fetch",
data,
format!(
"✓ bundle verified: manifest+contents intact, zero network ({})",
bundle.display()
),
)
} else {
CommandOutcome::fail(
"airgap fetch",
format!(
"✗ bundle verify FAILED for {} (manifest_root_ok={manifest_root_ok}, bundle_hash_ok={bundle_hash_ok}, bad={bad_entries:?})",
bundle.display()
),
)
}
}
/// `airgap unfold` (AIRGAP7 — the provisioning PLAN for a target) as a
/// [`CommandOutcome`]. Takes the already-rendered plan `lines` (the backend's
/// qemu argv / HCL / reset body) for `backend`/`target`. `ok ⟺ a non-empty plan`
/// (an unknown backend yields no lines ⇒ RED, handled by the caller's bail).
pub fn unfold_plan_outcome(
backend: &str,
target: &str,
lines: &[String],
) -> crate::cli_outcome::CommandOutcome {
use crate::cli_outcome::CommandOutcome;
if lines.is_empty() {
return CommandOutcome::fail(
"airgap unfold",
format!("no provisioning plan for backend `{backend}` target `{target}`"),
);
}
let data = serde_json::json!({
"backend": backend,
"target": target,
"plan": lines,
});
CommandOutcome::ok("airgap unfold", data, lines.join("\n"))
}
#[cfg(test)]
mod tests {
use super::*;
/// LAW 1 (inject-assert): parse a real JSONL event file (the lean crate's
/// on-disk format) → assert the rows decode with the right fields and
/// `depends_on` is `None` for roots / `Some` for edges.
#[test]
fn jsonl_decodes_airgap_events() {
let dir = tempfile::tempdir().unwrap();
let f = dir.path().join("airgap_events.jsonl");
std::fs::write(
&f,
"{\"run_id\":\"r1\",\"seq\":0,\"ts_micros\":1,\"product\":\"holger\",\"op\":\"install\",\"phase\":\"end\",\"status\":\"ok\",\"detail\":\"\",\"depends_on\":[]}\n\
{\"run_id\":\"r1\",\"seq\":1,\"ts_micros\":2,\"product\":\"nornir-server\",\"op\":\"start\",\"phase\":\"end\",\"status\":\"ok\",\"detail\":\"healthy\",\"depends_on\":[\"holger\"]}\n",
)
.unwrap();
let rows = read_jsonl(&f).unwrap();
assert_eq!(rows.len(), 2);
assert_eq!(rows[0].product, "holger");
assert_eq!(rows[0].op, "install");
assert_eq!(rows[0].depends_on, None, "root op has no deps");
assert_eq!(rows[1].product, "nornir-server");
assert_eq!(rows[1].depends_on, Some(vec!["holger".to_string()]));
}
fn ev(
run: &str,
seq: i64,
product: &str,
op: &str,
deps: Option<Vec<String>>,
) -> AirgapEventRow {
AirgapEventRow {
run_id: run.into(),
seq,
ts_micros: seq + 1,
product: product.into(),
op: op.into(),
phase: "end".into(),
status: "ok".into(),
detail: String::new(),
depends_on: deps,
}
}
/// Round-trip: append a bring-up DAG (roots + dep edges) into the Iceberg
/// `airgap_events` table, then read it straight back with the READ twin
/// `query_airgap_events`. Asserts every field survives — including the
/// optional `depends_on` list (None for roots, Some for edges) — and that
/// the `run_id` filter scopes correctly. Guards the write-only gap: the
/// warehouse table now has a reader, not just an ingest writer.
#[test]
fn append_query_round_trips_and_scopes_by_run() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
let run_a = vec![
ev("run-a", 0, "holger", "install", None),
ev(
"run-a",
1,
"nornir-server",
"start",
Some(vec!["holger".into()]),
),
];
let run_b = vec![ev("run-b", 0, "tunnr", "install", None)];
wh.block_on(append_airgap_events(&wh, &run_a)).unwrap();
wh.block_on(append_airgap_events(&wh, &run_b)).unwrap();
// Every run, sorted by (run_id, seq).
let all = wh.block_on(query_airgap_events(&wh, None)).unwrap();
assert_eq!(all.len(), 3);
assert_eq!(all, {
let mut v = run_a.clone();
v.extend(run_b.clone());
v
});
// depends_on faithfully preserved: root → None, edge → Some.
assert_eq!(all[0].depends_on, None);
assert_eq!(all[1].depends_on, Some(vec!["holger".to_string()]));
// Scoped to one run.
let just_a = wh
.block_on(query_airgap_events(&wh, Some("run-a")))
.unwrap();
assert_eq!(just_a.len(), 2);
assert!(just_a.iter().all(|r| r.run_id == "run-a"));
// The CommandOutcome renderer shapes the warehouse rows the same as JSONL.
let outcome = events_outcome(std::path::Path::new("/dev/null"), &all);
assert!(outcome.is_sannr());
assert_eq!(outcome.data["events"].as_array().unwrap().len(), 3);
}
#[test]
fn events_outcome_empty_is_red_real_is_sannr() {
let p = std::path::Path::new("/tmp/airgap_events.jsonl");
// RAGNARÖK: an absent/empty DAG is RED.
let red = events_outcome(p, &[]);
assert_eq!(red.command, "airgap events");
assert!(!red.is_sannr());
assert!(red.human.contains("no airgap_events"));
// real rows ⇒ sannr, events surfaced as readable data.
let rows = vec![
ev("r1", 0, "holger", "install", None),
ev(
"r1",
1,
"nornir-server",
"start",
Some(vec!["holger".into()]),
),
];
let o = events_outcome(p, &rows);
assert!(o.is_sannr());
let arr = o.data["events"].as_array().unwrap();
assert_eq!(arr.len(), 2);
assert_eq!(arr[1]["product"], serde_json::json!("nornir-server"));
assert!(o.human.contains("holger"), "dep edge rendered: {}", o.human);
}
#[test]
fn verify_outcome_intact_is_sannr_tampered_is_red() {
let b = std::path::Path::new("/srv/lake.tar.zst");
let ok = verify_outcome(b, true, true, &[]);
assert_eq!(ok.command, "airgap fetch");
assert!(ok.is_sannr(), "intact bundle is a true (sannr) outcome");
assert_eq!(ok.data["ok"], serde_json::json!(true));
assert_eq!(ok.data["bundle"], serde_json::json!("/srv/lake.tar.zst"));
// a bad entry ⇒ verify FAILS ⇒ RED.
let bad = verify_outcome(b, true, false, &["bin/holger".into()]);
assert!(!bad.is_sannr());
assert!(bad.human.contains("FAILED"));
assert!(bad.human.contains("bin/holger"));
}
#[test]
fn unfold_plan_outcome_real_plan_is_sannr_empty_is_red() {
let lines = vec![
"qemu-system-x86_64 -accel kvm".to_string(),
" -m 4096".to_string(),
];
let o = unfold_plan_outcome("kvm", "njord", &lines);
assert_eq!(o.command, "airgap unfold");
assert!(o.is_sannr());
assert_eq!(o.data["backend"], serde_json::json!("kvm"));
assert_eq!(o.data["plan"].as_array().unwrap().len(), 2);
assert!(o.human.contains("qemu-system-x86_64"));
// no plan ⇒ RED.
assert!(!unfold_plan_outcome("kvm", "njord", &[]).is_sannr());
}
}