src/events.rs
//! AIRGAP8 — the `airgap_events` telemetry stream.
//!
//! Design + LAW (persist-to-warehouse): every pack/fetch/unfold/install/start
//! boundary emits an [`AirgapEvent`](crate::events::AirgapEvent) — the airgap twin of nornir's
//! `release_events` (`src/warehouse/release_events.rs`): one row per
//! `product × op × phase`, carrying `depends_on` so a reader can draw the live
//! bring-up DAG and see where a partial install stalled.
//!
//! This lean crate cannot pull nornir's Arrow/Iceberg stack, so it emits through
//! an [`EventSink`](crate::events::EventSink) seam:
//! - [`JsonlSink`](crate::events::JsonlSink) — the **real default**: appends one JSON line per event to a
//! file UNDER THE WAREHOUSE ROOT (never `/tmp`), durable + greppable on an
//! air-gapped box that has no warehouse server.
//! - nornir wires a `IcebergSink` over its `airgap_events` table when the fat CLI
//! runs (see `src/warehouse/airgap_events.rs`) — same rows, real time-travel.
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
/// Canonical op tags (the pipeline stages).
pub mod op {
pub const PACK: &str = "pack";
pub const FETCH: &str = "fetch";
pub const UNFOLD: &str = "unfold";
pub const INSTALL: &str = "install";
pub const START: &str = "start";
/// One install-matrix cell's probe result (`matrix::ProbeReport` ingest).
pub const MATRIX_PROBE: &str = "matrix_probe";
}
/// Canonical phase + status tags (mirror release_events).
pub mod phase {
pub const START: &str = "start";
pub const END: &str = "end";
pub const SKIP: &str = "skip";
}
pub mod status {
pub const OK: &str = "ok";
pub const FAIL: &str = "fail";
pub const RUNNING: &str = "running";
}
/// One row of the airgap-op DAG. `serde` so it ships over a thin RPC / lands in
/// the warehouse unchanged.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct AirgapEvent {
/// Groups one install run's rows.
pub run_id: String,
/// Per-run monotonic ordering key (Iceberg/JSONL scan order is unspecified).
pub seq: i64,
/// Microseconds since the Unix epoch (UTC).
pub ts_micros: i64,
/// The product the op acts on (e.g. "holger"), or "" for a whole-lake op.
pub product: String,
/// `pack | fetch | unfold | install | start` (see [`op`]).
pub op: String,
/// `start | end | skip` (see [`phase`]).
pub phase: String,
/// `ok | fail | running` (see [`status`]).
pub status: String,
/// Short human message ("" when none — never a surprise null).
pub detail: String,
/// Products this op waited on (the dep-graph producers).
pub depends_on: Vec<String>,
}
/// Where airgap events are durably written.
pub trait EventSink {
fn append(&self, event: &AirgapEvent) -> Result<()>;
}
/// The real default sink: append-only JSON-lines under the warehouse root.
pub struct JsonlSink {
path: PathBuf,
lock: Mutex<()>,
}
impl JsonlSink {
/// Open (create) `<warehouse_root>/airgap_events.jsonl`.
pub fn under_warehouse(warehouse_root: &Path) -> Result<Self> {
std::fs::create_dir_all(warehouse_root)
.with_context(|| format!("create warehouse root {}", warehouse_root.display()))?;
Ok(JsonlSink {
path: warehouse_root.join("airgap_events.jsonl"),
lock: Mutex::new(()),
})
}
/// The backing file path (for tests / readers).
pub fn path(&self) -> &Path {
&self.path
}
}
impl EventSink for JsonlSink {
fn append(&self, event: &AirgapEvent) -> Result<()> {
use std::io::Write;
let line = serde_json::to_string(event)?;
let _g = self.lock.lock().unwrap();
let mut f = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&self.path)
.with_context(|| format!("open {}", self.path.display()))?;
writeln!(f, "{line}").context("append airgap_event")?;
Ok(())
}
}
/// Mints a run id + a monotonic `seq` and emits events to a sink. Best-effort +
/// non-fatal (mirrors `ReleaseEventEmitter`): a telemetry write must never abort
/// an otherwise-successful install.
pub struct EventEmitter<'a> {
run_id: String,
seq: std::cell::Cell<i64>,
sink: &'a dyn EventSink,
}
impl<'a> EventEmitter<'a> {
pub fn new(run_id: impl Into<String>, sink: &'a dyn EventSink) -> Self {
EventEmitter {
run_id: run_id.into(),
seq: std::cell::Cell::new(0),
sink,
}
}
/// Emit one event (non-fatal: logs to stderr on sink error, does not abort).
pub fn emit(
&self,
product: &str,
op: &str,
phase: &str,
status: &str,
detail: &str,
depends_on: &[String],
) {
let seq = self.seq.get();
self.seq.set(seq + 1);
let ev = AirgapEvent {
run_id: self.run_id.clone(),
seq,
ts_micros: now_micros(),
product: product.to_string(),
op: op.to_string(),
phase: phase.to_string(),
status: status.to_string(),
detail: detail.to_string(),
depends_on: depends_on.to_vec(),
};
if let Err(e) = self.sink.append(&ev) {
eprintln!(
" ⚠ airgap_events: dropped {op}/{phase} for `{product}` (non-fatal): {e:#}"
);
}
}
}
/// Read all events back from a [`JsonlSink`] file (for status / verification).
pub fn read_jsonl(path: &Path) -> Result<Vec<AirgapEvent>> {
let text = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
let mut out: Vec<AirgapEvent> = Vec::new();
for line in text.lines().filter(|l| !l.trim().is_empty()) {
out.push(serde_json::from_str(line).context("parse airgap_event line")?);
}
out.sort_by_key(|e| e.seq);
Ok(out)
}
fn now_micros() -> i64 {
chrono::Utc::now().timestamp_micros()
}
#[cfg(test)]
mod tests {
use super::*;
/// LAW 1 (inject-assert) + persist-to-warehouse: emit a real install DAG to a
/// JSONL sink UNDER the warehouse root, read it back, assert the rows carry
/// the injected op/phase/status/depends_on and are seq-ordered.
#[test]
fn emits_install_dag_to_warehouse_jsonl_and_reads_back() {
let wh = tempfile::tempdir().unwrap();
let sink = JsonlSink::under_warehouse(wh.path()).unwrap();
// Sink lives under the warehouse root, NOT /tmp.
assert!(sink.path().starts_with(wh.path()));
assert!(sink.path().ends_with("airgap_events.jsonl"));
let em = EventEmitter::new("run-42", &sink);
em.emit(
"holger",
op::INSTALL,
phase::START,
status::RUNNING,
"",
&[],
);
em.emit(
"holger",
op::INSTALL,
phase::END,
status::OK,
"installed",
&[],
);
em.emit(
"nornir-server",
op::START,
phase::END,
status::OK,
"healthy",
&["holger".to_string()],
);
let rows = read_jsonl(sink.path()).unwrap();
assert_eq!(rows.len(), 3);
assert_eq!(rows[0].run_id, "run-42");
assert_eq!((rows[0].seq, rows[1].seq, rows[2].seq), (0, 1, 2));
assert_eq!(rows[1].product, "holger");
assert_eq!(rows[1].op, op::INSTALL);
assert_eq!(rows[1].phase, phase::END);
assert_eq!(rows[1].status, status::OK);
assert_eq!(rows[1].detail, "installed");
// The bring-up edge is recorded (the live DAG).
assert_eq!(rows[2].depends_on, vec!["holger"]);
assert!(rows[2].ts_micros > 0);
}
}