crates/nornir-release-ledger/src/release_events.rs
//! Arrow writer + reader for the **release-op DAG** (`release_events`).
//!
//! `release_lineage` records a release's *outcome* per `(release, repo)`.
//! This module records its *shape*: one row per `component × op × phase`
//! boundary as a release run walks the build order. The
//! `depends_on` list carries the upstream components this one waited on
//! (the dep-graph producers), so a reader can draw the live op DAG and
//! see exactly where a partial release stalled — the structural twin of
//! the funnel DAG.
//!
//! Write path: a [`ReleaseEventEmitter`] mints a per-run monotonic `seq` and
//! appends one row per emit through the backend-neutral
//! [`nornir_warehouse_trait::Warehouse`] seam — [`ReleaseEventEmitter::emit`]
//! (sync, over `append_arrow`) or [`ReleaseEventEmitter::emit_async`] (over
//! `append_arrow_async`). Emits are **best-effort + non-fatal** — a logging-row
//! failure must never abort an otherwise-green release.
//!
//! Read path: [`query_release_events`] scans the table, scopes by a `run_id` or
//! a `repo`, and returns rows sorted by `(run_id, seq)` — Iceberg gives no scan
//! order, so `seq` is the durable ordering key.
//!
//! Schema: [`crate::schema::release_events`].
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicI64, Ordering};
use anyhow::{anyhow, Result};
use arrow::array::{
Array, Int64Array, ListArray, ListBuilder, RecordBatch, StringArray, StringBuilder,
TimestampMicrosecondArray,
};
use chrono::{DateTime, TimeZone, Utc};
use iceberg::arrow::schema_to_arrow_schema;
use uuid::Uuid;
use nornir_warehouse_trait::Warehouse;
use crate::schema::TABLE_RELEASE_EVENTS;
// Column indices match `schema::release_events` field order. `pub(crate)` so
// the schema module can assert index↔name agreement (the RED-when-broken guard
// against a column being inserted without the codec following).
pub(crate) const COL_RUN_ID: usize = 0;
pub(crate) const COL_SEQ: usize = 1;
pub(crate) const COL_TS_MICROS: usize = 2;
pub(crate) const COL_COMPONENT: usize = 3;
pub(crate) const COL_REPO: usize = 4;
pub(crate) const COL_OP: usize = 5;
pub(crate) const COL_PHASE: usize = 6;
pub(crate) const COL_STATUS: usize = 7;
pub(crate) const COL_DETAIL: usize = 8;
pub(crate) const COL_DEPENDS_ON: usize = 9;
pub(crate) const COL_ELAPSED_MS: usize = 10;
/// Canonical phase tags.
pub mod phase {
pub const START: &str = "start";
pub const END: &str = "end";
pub const SKIP: &str = "skip";
}
/// Canonical status tags.
pub mod status {
pub const OK: &str = "ok";
pub const FAIL: &str = "fail";
pub const WARN: &str = "warn";
pub const RUNNING: &str = "running";
}
/// One row of the `release_events` table — a single component×op×phase
/// boundary in a release run.
///
/// `serde` derives let a server ship the exact rows to a thin viz client over
/// the `Viz.ReleaseEvents` RPC (serialize → JSON → deserialize), so the remote
/// 🚀 Release tab + 📡 Live Run hydrate render the identical op-DAG the embedded
/// path reads from the warehouse.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct ReleaseEventRow {
/// The release-run identity this event belongs to (groups a run's rows).
pub run_id: String,
/// Per-run monotonic sequence number — the stable ordering key (Iceberg
/// scan order is unspecified, so reads sort by this).
pub seq: i64,
/// Event time, microseconds since the Unix epoch (UTC).
pub ts_micros: i64,
/// The component the op acts on (usually the repo name).
pub component: String,
/// The workspace repo name (== `component` today; kept distinct so a
/// crate-grained op can carry a sub-component while still naming its repo).
pub repo: String,
/// The operation: `test` | `bench` | `gate` | `publish` | `push` |
/// `snapshot` | `run` | …
pub op: String,
/// `start` | `end` | `skip` (see [`phase`]).
pub phase: String,
/// `ok` | `fail` | `warn` | `running` (see [`status`]).
pub status: String,
/// Short human message (never the empty-string null — we store "" so the
/// column is required and SQL readers never see a surprise null).
pub detail: String,
/// Components this op waited on (the dep-graph producers). `None` for a
/// root op with no upstream; a real (possibly empty) list otherwise.
pub depends_on: Option<Vec<String>>,
/// Wall-clock duration of the op, milliseconds. Only `phase = end` rows
/// carry a value.
pub elapsed_ms: Option<i64>,
}
impl ReleaseEventRow {
/// The `(run_id, seq)` sort key.
fn key(&self) -> (String, i64) {
(self.run_id.clone(), self.seq)
}
}
/// Buffers a release run's identity + a monotonic `seq` counter and appends
/// one `release_events` row per boundary. Cheap and **non-fatal**: a write
/// error is logged (stderr) and swallowed so a logging hiccup never aborts a
/// release. Cloneable — `seq` is shared (atomic) so events stay globally
/// ordered even if the emitter is handed to a sub-phase.
#[derive(Clone)]
pub struct ReleaseEventEmitter {
run_id: String,
seq: Arc<AtomicI64>,
/// Every row this emitter mints is also recorded here, in `seq` order. A
/// release driver drains it at the end and ships the batch to a
/// live server over `Telemetry.SubmitReleaseEvents` (the server owns the
/// warehouse lock, so the per-emit `append_release_events(&wh, …)` against a
/// read-only local handle can't reach it). Shared (cloneable) so sub-phases
/// that get a clone of the emitter still record into the same buffer.
recorded: Arc<Mutex<Vec<ReleaseEventRow>>>,
}
impl ReleaseEventEmitter {
/// New emitter for `run_id` (any stable per-run string — a UUID or a
/// CLI's epoch-seconds run id both work).
pub fn new(run_id: impl Into<String>) -> Self {
Self {
run_id: run_id.into(),
seq: Arc::new(AtomicI64::new(0)),
recorded: Arc::new(Mutex::new(Vec::new())),
}
}
/// Snapshot every row minted so far, in `seq` order. Used by a CLI to
/// ship the whole run's events to a live server after the run completes.
pub fn recorded_rows(&self) -> Vec<ReleaseEventRow> {
let mut rows = self.recorded.lock().expect("release-event buffer mutex").clone();
rows.sort_by_key(ReleaseEventRow::key);
rows
}
/// The run id these events are grouped under.
pub fn run_id(&self) -> &str {
&self.run_id
}
/// Build a row for the next `seq`, stamped now, without writing it. Used by
/// the sync wrapper and tests that want to batch.
#[allow(clippy::too_many_arguments)]
pub fn row(
&self,
component: &str,
repo: &str,
op: &str,
phase: &str,
status: &str,
detail: &str,
depends_on: Option<Vec<String>>,
elapsed_ms: Option<i64>,
) -> ReleaseEventRow {
let seq = self.seq.fetch_add(1, Ordering::SeqCst);
let row = ReleaseEventRow {
run_id: self.run_id.clone(),
seq,
ts_micros: Utc::now().timestamp_micros(),
component: component.to_string(),
repo: repo.to_string(),
op: op.to_string(),
phase: phase.to_string(),
status: status.to_string(),
detail: detail.to_string(),
depends_on,
elapsed_ms,
};
// Record into the shared buffer so a CLI can ship the run to a server.
if let Ok(mut buf) = self.recorded.lock() {
buf.push(row.clone());
}
row
}
/// Async emit: mint the next row and append it. Errors are swallowed
/// (logged to stderr) — a logging-row failure must not abort the release.
#[allow(clippy::too_many_arguments)]
pub async fn emit_async(
&self,
wh: &dyn Warehouse,
component: &str,
repo: &str,
op: &str,
phase: &str,
status: &str,
detail: &str,
depends_on: Option<Vec<String>>,
elapsed_ms: Option<i64>,
) {
let row = self.row(component, repo, op, phase, status, detail, depends_on, elapsed_ms);
if let Err(e) = append_release_events(wh, std::slice::from_ref(&row)).await {
eprintln!(
" ⚠ release_events: dropped {op}/{phase} for `{component}` (non-fatal): {e:#}"
);
}
}
/// SYNC emit, for a CLI/driver that is not inside an async context.
///
/// Routes through the backend-neutral [`Warehouse::append_arrow`], which
/// blocks internally on the backend's own runtime — so this needs no
/// `block_on` and, crucially, no CONCRETE engine type. (The nornir copy this
/// replaces took `&IcebergWarehouse` and called `wh.block_on(..)`; taking
/// `&dyn Warehouse` is strictly wider — an `&IcebergWarehouse` still coerces
/// — and is what lets the ONE definition serve both halves of the seam.)
/// Errors are swallowed exactly like [`emit_async`](Self::emit_async).
#[allow(clippy::too_many_arguments)]
pub fn emit(
&self,
wh: &dyn Warehouse,
component: &str,
repo: &str,
op: &str,
phase: &str,
status: &str,
detail: &str,
depends_on: Option<Vec<String>>,
elapsed_ms: Option<i64>,
) {
let row = self.row(component, repo, op, phase, status, detail, depends_on, elapsed_ms);
if let Err(e) = append_release_events_sync(wh, std::slice::from_ref(&row)) {
eprintln!(
" ⚠ release_events: dropped {op}/{phase} for `{component}` (non-fatal): {e:#}"
);
}
}
}
/// Build the Arrow `RecordBatch` for `rows` against the CANONICAL
/// `release_events` schema (the table was created from it; the backend recasts
/// it to the stored field-id schema). Shared by the sync + async append paths so
/// the encoding exists exactly once.
fn events_batch(rows: &[ReleaseEventRow]) -> Result<RecordBatch> {
let arrow_schema = Arc::new(schema_to_arrow_schema(&crate::schema::release_events()?)?);
// The `depends_on` element Field carries the iceberg `PARQUET:field_id`
// metadata that `RecordBatch::try_new` validates; wear it on the builder so
// the list column matches the table schema exactly (same trick as funnel).
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.component.clone()).collect::<Vec<_>>())),
Arc::new(StringArray::from(rows.iter().map(|r| r.repo.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()),
Arc::new(Int64Array::from(rows.iter().map(|r| r.elapsed_ms).collect::<Vec<_>>())),
];
Ok(RecordBatch::try_new(arrow_schema, cols)?)
}
/// Append a batch of `release_events` rows (one Iceberg snapshot), ASYNC.
///
/// Routed through the backend-neutral [`Warehouse::append_arrow_async`].
pub async fn append_release_events(wh: &dyn Warehouse, rows: &[ReleaseEventRow]) -> Result<()> {
if rows.is_empty() {
return Ok(());
}
wh.append_arrow_async(TABLE_RELEASE_EVENTS, events_batch(rows)?).await?;
Ok(())
}
/// Append a batch of `release_events` rows (one Iceberg snapshot), SYNC.
///
/// Routed through the backend-neutral [`Warehouse::append_arrow`], which blocks
/// internally on the backend's own runtime.
pub fn append_release_events_sync(wh: &dyn Warehouse, rows: &[ReleaseEventRow]) -> Result<()> {
if rows.is_empty() {
return Ok(());
}
wh.append_arrow(TABLE_RELEASE_EVENTS, events_batch(rows)?)?;
Ok(())
}
/// Which release events to read.
#[derive(Debug, Clone)]
pub enum EventSelector {
/// Exactly one run (its `run_id`).
Run(String),
/// Every event whose `repo` matches — across all that repo's runs.
Repo(String),
/// Everything in the table.
All,
}
/// Decode scanned batches into rows, keeping only those `sel` selects, sorted by
/// `(run_id, seq)`. Shared by the sync + async read paths.
fn decode_events(batches: &[RecordBatch], sel: &EventSelector) -> Result<Vec<ReleaseEventRow>> {
let mut out: Vec<ReleaseEventRow> = Vec::new();
for b in batches {
let run_id = col_str(b, COL_RUN_ID)?;
let seq = col_i64(b, COL_SEQ)?;
let ts = col_ts(b, COL_TS_MICROS)?;
let component = col_str(b, COL_COMPONENT)?;
let repo = col_str(b, COL_REPO)?;
let op = col_str(b, COL_OP)?;
let phase = col_str(b, COL_PHASE)?;
let st = col_str(b, COL_STATUS)?;
let detail = col_str(b, COL_DETAIL)?;
let deps = col_str_list(b, COL_DEPENDS_ON)?;
let elapsed = col_i64(b, COL_ELAPSED_MS)?;
for i in 0..b.num_rows() {
let row = ReleaseEventRow {
run_id: run_id.value(i).to_string(),
seq: seq.value(i),
ts_micros: ts.value(i),
component: component.value(i).to_string(),
repo: repo.value(i).to_string(),
op: op.value(i).to_string(),
phase: phase.value(i).to_string(),
status: st.value(i).to_string(),
detail: detail.value(i).to_string(),
depends_on: deps[i].clone(),
elapsed_ms: if elapsed.is_null(i) { None } else { Some(elapsed.value(i)) },
};
let keep = match sel {
EventSelector::Run(id) => &row.run_id == id,
EventSelector::Repo(r) => &row.repo == r,
EventSelector::All => true,
};
if keep {
out.push(row);
}
}
}
out.sort_by_key(|r| r.key());
Ok(out)
}
/// Read release events, scoped by `sel`, returned sorted by `(run_id, seq)`.
pub async fn query_release_events(
wh: &dyn Warehouse,
sel: &EventSelector,
) -> Result<Vec<ReleaseEventRow>> {
// Blank/torn Iceberg metadata JSON → 0 rows, never a JSON-EOF crash: the
// backend's `scan_arrow_async` degrades a torn snapshot to an empty scan.
let batches: Vec<RecordBatch> = wh.scan_arrow_async(TABLE_RELEASE_EVENTS).await?;
decode_events(&batches, sel)
}
/// SYNC twin of [`query_release_events`] (over `Warehouse::scan_arrow`).
pub fn query_release_events_sync(
wh: &dyn Warehouse,
sel: &EventSelector,
) -> Result<Vec<ReleaseEventRow>> {
let batches: Vec<RecordBatch> = wh.scan_arrow(TABLE_RELEASE_EVENTS)?;
decode_events(&batches, sel)
}
// ─── rendering: JSON + human topo view ──────────────────────────────────
/// Serialize rows to a stable JSON array (one object per event). Pure string
/// assembly (no serde derive needed) so the CLI stays dependency-light.
pub fn rows_to_json(rows: &[ReleaseEventRow]) -> String {
fn esc(s: &str) -> String {
let mut o = String::with_capacity(s.len() + 2);
for c in s.chars() {
match c {
'"' => o.push_str("\\\""),
'\\' => o.push_str("\\\\"),
'\n' => o.push_str("\\n"),
'\t' => o.push_str("\\t"),
'\r' => o.push_str("\\r"),
c => o.push(c),
}
}
o
}
let mut s = String::from("[\n");
for (i, r) in rows.iter().enumerate() {
let depends = match &r.depends_on {
None => "null".to_string(),
Some(v) => {
let inner = v.iter().map(|d| format!("\"{}\"", esc(d))).collect::<Vec<_>>().join(", ");
format!("[{inner}]")
}
};
let elapsed = r.elapsed_ms.map(|e| e.to_string()).unwrap_or_else(|| "null".to_string());
let ts_rfc = Utc
.timestamp_micros(r.ts_micros)
.single()
.map(|d| d.to_rfc3339())
.unwrap_or_default();
s.push_str(&format!(
" {{\"run_id\": \"{}\", \"seq\": {}, \"ts\": \"{}\", \"component\": \"{}\", \
\"repo\": \"{}\", \"op\": \"{}\", \"phase\": \"{}\", \"status\": \"{}\", \
\"detail\": \"{}\", \"depends_on\": {}, \"elapsed_ms\": {}}}{}\n",
esc(&r.run_id),
r.seq,
esc(&ts_rfc),
esc(&r.component),
esc(&r.repo),
esc(&r.op),
esc(&r.phase),
esc(&r.status),
esc(&r.detail),
depends,
elapsed,
if i + 1 < rows.len() { "," } else { "" },
));
}
s.push(']');
s
}
/// Render a human-readable topological view of the DAG: events grouped by
/// `run_id` (newest run last), in `seq` order, with each component's
/// `depends_on` edges shown once. The `→` chain reads as the op DAG.
pub fn render_topo(rows: &[ReleaseEventRow]) -> String {
let mut by_run: BTreeMap<String, Vec<&ReleaseEventRow>> = BTreeMap::new();
for r in rows {
by_run.entry(r.run_id.clone()).or_default().push(r);
}
let mut out = String::new();
for (run_id, evs) in &by_run {
// Deterministic order within a run.
let mut evs = evs.clone();
evs.sort_by_key(|r| r.seq);
out.push_str(&format!("release run {run_id}\n"));
// dep-graph edges, first time each component's depends_on is seen.
let mut seen_deps: BTreeMap<&str, ()> = BTreeMap::new();
for r in &evs {
if let Some(deps) = &r.depends_on {
if seen_deps.insert(r.component.as_str(), ()).is_none() && !deps.is_empty() {
out.push_str(&format!(
" {} depends_on {}\n",
r.component,
deps.join(", ")
));
}
}
}
out.push_str(" ─ timeline ─\n");
for r in &evs {
let mark = match r.status.as_str() {
status::OK => "✓",
status::FAIL => "✗",
status::WARN => "⚠",
status::RUNNING => "…",
_ => "·",
};
let elapsed = r
.elapsed_ms
.map(|e| format!(" ({e}ms)"))
.unwrap_or_default();
let detail = if r.detail.is_empty() {
String::new()
} else {
format!(" — {}", r.detail)
};
out.push_str(&format!(
" [{seq:>3}] {mark} {component} {op}/{phase} {status}{elapsed}{detail}\n",
seq = r.seq,
component = r.component,
op = r.op,
phase = r.phase,
status = r.status,
));
}
out.push('\n');
}
if out.is_empty() {
out.push_str("(no release events recorded)\n");
}
out
}
// ─── 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!("release_events col {idx} is not StringArray"))
}
fn col_i64<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a Int64Array> {
b.column(idx)
.as_any()
.downcast_ref::<Int64Array>()
.ok_or_else(|| anyhow!("release_events 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!("release_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!("release_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!("release_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)
}
/// Format `ts_micros` as RFC3339 (UTC). Convenience for callers/tests.
pub fn ts_to_rfc3339(ts_micros: i64) -> String {
let dt: DateTime<Utc> = Utc.timestamp_micros(ts_micros).single().unwrap_or_else(Utc::now);
dt.to_rfc3339()
}
/// Convenience: a fresh run id (UUIDv4 string).
pub fn new_run_id() -> String {
Uuid::new_v4().to_string()
}
#[cfg(test)]
mod tests {
use super::*;
use crate::testkit::{block_on, MockWarehouse};
#[test]
fn emitter_seq_is_monotonic_and_shared_across_clones() {
let e = ReleaseEventEmitter::new("run-1");
let r0 = e.row("a", "a", "test", phase::START, status::RUNNING, "", None, None);
let e2 = e.clone();
let r1 = e2.row("a", "a", "test", phase::END, status::OK, "", None, Some(5));
let r2 = e.row("b", "b", "gate", phase::START, status::RUNNING, "", Some(vec!["a".into()]), None);
assert_eq!((r0.seq, r1.seq, r2.seq), (0, 1, 2));
assert_eq!(r0.run_id, "run-1");
// The shared recording buffer captures EVERY minted row (including the
// clone's), in seq order — this is what a fat CLI drains to ship a
// whole run to a live server over Telemetry.SubmitReleaseEvents.
let recorded = e.recorded_rows();
assert_eq!(recorded.len(), 3, "all three rows recorded across clones");
assert_eq!(recorded.iter().map(|r| r.seq).collect::<Vec<_>>(), vec![0, 1, 2]);
assert_eq!(recorded[1].elapsed_ms, Some(5), "clone's row recorded too");
assert_eq!(recorded[2].depends_on, Some(vec!["a".to_string()]));
// The clone reads the same buffer.
assert_eq!(e2.recorded_rows().len(), 3);
}
#[test]
fn append_and_query_round_trip_with_grouping_and_order() {
let wh = MockWarehouse::new();
// Two runs, deliberately appended out of seq order to prove the reader
// sorts by (run_id, seq).
let rows = vec![
ReleaseEventRow {
run_id: "runA".into(), seq: 1, ts_micros: 10, component: "znippy".into(),
repo: "znippy".into(), op: "test".into(), phase: phase::END.into(),
status: status::OK.into(), detail: "3 passed".into(), depends_on: None,
elapsed_ms: Some(120),
},
ReleaseEventRow {
run_id: "runA".into(), seq: 0, ts_micros: 5, component: "znippy".into(),
repo: "znippy".into(), op: "test".into(), phase: phase::START.into(),
status: status::RUNNING.into(), detail: String::new(), depends_on: None,
elapsed_ms: None,
},
ReleaseEventRow {
run_id: "runA".into(), seq: 2, ts_micros: 20, component: "holger".into(),
repo: "holger".into(), op: "gate".into(), phase: phase::START.into(),
status: status::RUNNING.into(), detail: String::new(),
depends_on: Some(vec!["znippy".into()]), elapsed_ms: None,
},
ReleaseEventRow {
run_id: "runB".into(), seq: 0, ts_micros: 30, component: "korp".into(),
repo: "korp".into(), op: "test".into(), phase: phase::START.into(),
status: status::RUNNING.into(), detail: String::new(), depends_on: Some(vec![]),
elapsed_ms: None,
},
];
block_on(append_release_events(&wh, &rows)).unwrap();
// Run scope returns only runA, in seq order.
let a = block_on(query_release_events(&wh, &EventSelector::Run("runA".into()))).unwrap();
assert_eq!(a.len(), 3);
assert_eq!(a.iter().map(|r| r.seq).collect::<Vec<_>>(), vec![0, 1, 2]);
assert_eq!(a[0].phase, phase::START);
assert_eq!(a[1].phase, phase::END);
assert_eq!(a[1].elapsed_ms, Some(120));
// depends_on round-trips exactly.
assert_eq!(a[2].depends_on, Some(vec!["znippy".to_string()]));
assert_eq!(a[0].depends_on, None);
// Repo scope crosses runs but filters by repo.
let korp = block_on(query_release_events(&wh, &EventSelector::Repo("korp".into()))).unwrap();
assert_eq!(korp.len(), 1);
assert_eq!(korp[0].run_id, "runB");
// An explicit empty depends_on list survives as Some(vec![]), distinct
// from a null.
assert_eq!(korp[0].depends_on, Some(vec![]));
// All scope sees both runs; grouping in render_topo splits them.
let all = block_on(query_release_events(&wh, &EventSelector::All)).unwrap();
assert_eq!(all.len(), 4);
let topo = render_topo(&all);
assert!(topo.contains("release run runA"));
assert!(topo.contains("release run runB"));
assert!(topo.contains("holger depends_on znippy"));
// JSON is well-formed enough to parse.
let json = rows_to_json(&a);
let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
assert_eq!(parsed.as_array().unwrap().len(), 3);
assert_eq!(parsed[2]["depends_on"][0], "znippy");
assert_eq!(parsed[1]["elapsed_ms"], 120);
}
/// The SYNC emit/read pair (the path that replaces nornir's
/// `emit(&IcebergWarehouse, …)` `block_on` wrapper) writes rows the ASYNC
/// reader sees, and vice versa — one encoding, two entry points.
#[test]
fn sync_emit_and_async_read_agree() {
let wh = MockWarehouse::new();
let e = ReleaseEventEmitter::new("run-sync");
e.emit(&wh, "skade", "skade", "gate", phase::START, status::RUNNING, "", None, None);
e.emit(&wh, "skade", "skade", "gate", phase::END, status::OK, "green", Some(vec!["korp".into()]), Some(7));
let sync_read =
query_release_events_sync(&wh, &EventSelector::Run("run-sync".into())).unwrap();
let async_read =
block_on(query_release_events(&wh, &EventSelector::Run("run-sync".into()))).unwrap();
assert_eq!(sync_read, async_read, "sync and async readers decode identically");
assert_eq!(sync_read.len(), 2);
assert_eq!(sync_read[0].seq, 0);
assert_eq!(sync_read[1].elapsed_ms, Some(7));
assert_eq!(sync_read[1].depends_on, Some(vec!["korp".to_string()]));
}
}