src/viz/build_info_live.rs
//! 📡 **Build-info LIVE-TAIL** — the slice-5 capstone (design
//! `.nornir/build-info-design.md` §f5, the "Deferred → tail" note).
//!
//! Slices 1–5 built the whole build-info arc: the §f3 Arrow schema + emitter
//! ([`nornir_build_thing::build_info`]), the native warehouse landing + at-rest
//! seal, the jera pipeline-DAG sub-jobs, and the **Flight-backed sink + consumer**
//! ([`crate::build::flight_sink`]). What was left open was the *viz tail*: a pane
//! that subscribes to the build-info Flight stream **per sub-job `step_id`** and
//! shows the rows arriving — each step's **status transitions** (from the `step`
//! rows) and its **log/metric tail** (§f5: "each step's log/metric stream is the
//! f3 feed filtered by `step_id`").
//!
//! # The model is pure, the seam is [`crate::build::flight_sink::subscribe`]
//!
//! [`BuildInfoTail`] is a **pure fold** over [`BuildInfoRow`]s — it never knows
//! *how* the rows arrived. That keeps it:
//!
//! * **robot-testable** — a test feeds rows in batches ([`BuildInfoTail::ingest`])
//! and reads the growing tail + per-step status back out of [`state_json`],
//! exactly the inject-and-assert shape the rest of the viz uses (no live server
//! needed to prove the pane folds the feed).
//! * **transport-agnostic** — the live path ([`BuildInfoTail::pull`], gated behind
//! `build-info-flight`) calls the existing `subscribe(endpoint, token, key)`
//! consumer seam by ticket (`key` = [`route_key`](crate::build::flight_sink::route_key)
//! per sub-job `step_id`) and ingests the decoded rows through the **same**
//! `ingest`. Live wire and injected rows fold identically.
//!
//! # Reuse (LAW #5)
//!
//! The pane paints with the shared facett palette ([`Theme`]) and the vendored
//! facett widgets ([`facett_ui::status_chip`], zebra rows) — the same themeable,
//! robot-testable material every other viz pane uses. Nothing here is re-rolled.
//!
//! # Continuous auto-refresh (the timer, closing the deferral)
//!
//! The pane now **auto-refreshes** while it is open, mirroring
//! [`super::bench_live`]'s periodic warehouse re-read. A background **poller**
//! ([`BuildInfoTail::start_poller`], gated `build-info-flight`) re-[`pull`]s every
//! tracked sub-job ticket (`route_key(build_id, step_id)`) on a fixed
//! [`POLL_EVERY`] cadence on its own OS-thread runtime (the same pattern
//! [`crate::build::flight_sink::FlightBuildInfoSink`] uses, so a `pull` never
//! needs an ambient runtime and never blocks the egui frame). Decoded rows are
//! handed back over a plain `std` channel (the **inbox**); [`draw`](BuildInfoTail::draw)
//! drains the inbox through the SAME synchronous [`ingest`](BuildInfoTail::ingest)
//! every frame and calls `request_repaint_after` so rows that land mid-build appear
//! without any user input. The poller forwards only rows *past* the per-key
//! high-water mark, so re-pulling the full stream never double-folds (L2: the
//! append-only `ingest` is unchanged). The set of live tickets is tracked
//! ([`subscribe_ticket`](BuildInfoTail::subscribe_ticket)) and **auto-grown** as
//! the fold discovers each sub-job — both are exposed in [`state_json`] so the
//! matrix asserts the ticks advance.
use std::collections::BTreeSet;
use std::sync::mpsc::{Receiver, Sender};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use eframe::egui::{self, RichText, ScrollArea};
use nornir_build_thing::build_info::BuildInfoRow;
use super::facett_theme::{AMBER, GREEN, RED, Theme};
use super::facett_ui;
/// How often (while the pane is open) the background poller re-`pull`s each live
/// sub-job ticket so build-info rows that land mid-build show up. Mirrors
/// [`super::bench_live`]'s warehouse re-read cadence.
pub const POLL_EVERY: Duration = Duration::from_millis(1500);
/// One tailed feed line — a `log` / `metric` / `step` row rendered in arrival
/// order. Exposed verbatim in [`BuildInfoTail::state_json`] so the headless
/// matrix asserts the tail content, not pixels.
#[derive(Debug, Clone, PartialEq)]
pub struct TailLine {
/// The §f3 `kind`: `"log"` | `"metric"` | `"step"`.
pub kind: String,
/// The §f3 `level` (`info` / `error` for logs; the status for `step` rows).
pub level: String,
/// The §f3 `name` column (log source / metric name / step label).
pub name: String,
/// The §f3 `value` column (the log message / metric value / step variant).
pub value: String,
/// The §f3 `ts_micros` timestamp.
pub ts_micros: i64,
}
impl TailLine {
fn from_row(r: &BuildInfoRow) -> Self {
Self {
kind: r.kind.clone(),
level: r.level.clone(),
name: r.name.clone(),
value: r.value.clone(),
ts_micros: r.ts_micros,
}
}
}
/// The per-sub-job (`step_id`) state folded from the build-info feed: the latest
/// **status** (the most recent `step` row's status, carried in the §f3 `level`
/// column — `queued` / `running` / `done` / `failed`), the ordered list of status
/// **transitions**, and the ordered **log/metric/step tail**.
#[derive(Debug, Clone, PartialEq)]
pub struct StepTail {
/// The §f3 `step_id` — the sub-job ticket this feed is filtered by.
pub step_id: String,
/// The current status (last `step` row's `level`); empty until the first
/// `step` row arrives.
pub status: String,
/// The ordered status transitions seen so far (each distinct new status
/// appended once), e.g. `["running", "done"]`.
pub status_history: Vec<String>,
/// The ordered `log` / `metric` / `step` tail for this sub-job — grows as
/// rows arrive.
pub tail: Vec<TailLine>,
}
impl StepTail {
fn new(step_id: &str) -> Self {
Self {
step_id: step_id.to_string(),
status: String::new(),
status_history: Vec::new(),
tail: Vec::new(),
}
}
}
/// The **build-info live-tail** model + pane. Folds the §f3 build-info feed into
/// per-sub-job status + tail (`ingest`), renders it themed (`draw`), and exposes
/// everything as data (`state_json`). The live source is the Flight `subscribe`
/// consumer seam (`pull`, behind `build-info-flight`).
pub struct BuildInfoTail {
/// The root `build_id` (from the first row carrying one); the §f3
/// build/pipeline this tail is for.
build_id: Option<String>,
/// Per-sub-job tails, in first-seen `step_id` order.
steps: Vec<StepTail>,
/// Count of manifest rows folded (`provenance` / `dep` / `artifact`, empty
/// `step_id`) — reported but not tailed (they are the build's static record,
/// not the live feed).
manifest_rows: usize,
/// The Flight endpoint the live path subscribes to (display + `state_json`).
endpoint: Option<String>,
/// Last error from a `pull` (never fatal — the pane keeps its folded state).
error: Option<String>,
/// The active facett palette (C8) — every colour routes through it so a
/// palette switch re-skins the tail with the rest of the viz.
theme: Theme,
/// **Inbox** — decoded [`BuildInfoRow`] batches delivered by the background
/// poller (or a test's simulated interval pull), drained into [`ingest`](Self::ingest)
/// each [`draw`](Self::draw). A plain `std` channel, so the synchronous
/// injectable fold path is unchanged (L2).
inbox_tx: Sender<Vec<BuildInfoRow>>,
inbox_rx: Receiver<Vec<BuildInfoRow>>,
/// The set of live sub-job tickets `(build_id, step_id)` the poller re-`pull`s.
/// Auto-grown as the fold discovers sub-jobs (and drivable via
/// [`subscribe_ticket`](Self::subscribe_ticket)); shared with the poller thread.
tickets: Arc<Mutex<BTreeSet<(String, String)>>>,
/// Count of poll batches drained + folded — a "tick" per delivered batch, the
/// number that ADVANCES as the auto-refresh timer re-pulls. Exposed in
/// [`state_json`](Self::state_json).
poll_count: usize,
/// The running poller (its stop flag), if the auto-refresh timer is live.
#[cfg(feature = "build-info-flight")]
poller: Option<Poller>,
}
/// The background auto-refresh poller's handle: a stop flag the [`Drop`] sets so
/// the thread winds down its runtime on its own thread (never in the caller's).
#[cfg(feature = "build-info-flight")]
struct Poller {
stop: Arc<std::sync::atomic::AtomicBool>,
}
impl Default for BuildInfoTail {
fn default() -> Self {
let (inbox_tx, inbox_rx) = std::sync::mpsc::channel();
Self {
build_id: None,
steps: Vec::new(),
manifest_rows: 0,
endpoint: None,
error: None,
theme: Theme::default(),
inbox_tx,
inbox_rx,
tickets: Arc::new(Mutex::new(BTreeSet::new())),
poll_count: 0,
#[cfg(feature = "build-info-flight")]
poller: None,
}
}
}
#[cfg(feature = "build-info-flight")]
impl Drop for BuildInfoTail {
fn drop(&mut self) {
if let Some(p) = &self.poller {
p.stop.store(true, std::sync::atomic::Ordering::Relaxed);
}
}
}
impl BuildInfoTail {
/// A fresh, empty tail.
pub fn new() -> Self {
Self::default()
}
/// Point the live path at a Flight `endpoint` (recorded for display /
/// `state_json`; the actual `subscribe` is [`pull`](Self::pull)).
pub fn set_endpoint(&mut self, endpoint: impl Into<String>) {
self.endpoint = Some(endpoint.into());
}
/// Re-skin this pane with a facett palette (broadcast from the picker / owning
/// tab's `set_palette`).
pub fn set_palette(&mut self, t: Theme) {
self.theme = t;
}
/// Fold a batch of §f3 [`BuildInfoRow`]s into the per-sub-job tails. The
/// heart of the pane: idempotent-append semantics — call it repeatedly as
/// batches arrive (from `subscribe`, or injected in a test) and the tail
/// **grows** while each step's **status** tracks its latest `step` row.
///
/// * `kind == "step"` → the row's `level` is the sub-job status; a *new*
/// status is recorded as a transition. The step row is also appended to the
/// tail so a status change is visible in the stream.
/// * `kind == "log" | "metric"` → appended to that sub-job's tail.
/// * manifest rows (`provenance` / `dep` / `artifact`, empty `step_id`) → not
/// tailed (they are the build's static record); only counted.
pub fn ingest(&mut self, rows: &[BuildInfoRow]) {
for r in rows {
if self.build_id.is_none() && !r.build_id.is_empty() {
self.build_id = Some(r.build_id.clone());
}
// Auto-grow the live subscription set: each row's `(build_id, step_id)`
// is a ticket the poller re-`pull`s (manifest rows → the root ticket,
// empty `step_id`; feed rows → the per-sub-job ticket).
self.register_ticket(&r.build_id, &r.step_id);
match r.kind.as_str() {
"step" => {
let step = self.step_mut(&r.step_id);
// The §f3 `step` row carries the sub-job status in `level`.
if !r.level.is_empty() && step.status != r.level {
step.status = r.level.clone();
step.status_history.push(r.level.clone());
}
step.tail.push(TailLine::from_row(r));
}
"log" | "metric" => {
let step = self.step_mut(&r.step_id);
step.tail.push(TailLine::from_row(r));
}
// provenance / dep / artifact — the static manifest, not the feed.
_ => self.manifest_rows += 1,
}
}
self.emit_trace();
}
/// **The live seam** (§f5): subscribe to one sub-job's build-info stream by
/// ticket `key` (from [`route_key`](crate::build::flight_sink::route_key) —
/// `build_id/step_id`) over the Flight `do_get` consumer, then fold the
/// decoded rows through [`ingest`](Self::ingest). Returns the number of rows
/// tailed. Best-effort: a transport/auth error is recorded (never a panic),
/// and the already-folded state is kept.
///
/// Gated behind `build-info-flight` (the feature that links the Flight wire);
/// the default viz build keeps the injectable [`ingest`](Self::ingest) path
/// and pays no Flight/second-arrow tax.
#[cfg(feature = "build-info-flight")]
pub async fn pull(&mut self, endpoint: &str, token: Option<&str>, key: &str) -> usize {
self.endpoint = Some(endpoint.to_string());
match crate::build::flight_sink::subscribe(endpoint, token, key).await {
Ok(rows) => {
let n = rows.len();
self.error = None;
self.ingest(&rows);
n
}
Err(e) => {
self.error = Some(format!("{e:#}"));
0
}
}
}
/// Test-only alias for [`ingest`](Self::ingest) — mirrors the other panes'
/// `inject_for_test` so the inject-and-assert harness reads the tail back out
/// of `state_json` (the subscribe seam is itself already tested in
/// `flight_sink`, so this drives the fold on top of it).
#[doc(hidden)]
pub fn inject_for_test(&mut self, rows: Vec<BuildInfoRow>) {
self.ingest(&rows);
}
/// Register one sub-job ticket `(build_id, step_id)` to be (re-)subscribed by
/// the auto-refresh poller. Idempotent (a `BTreeSet`). Called automatically as
/// the fold discovers sub-jobs; also `pub` so the owner can seed a ticket
/// before any row has arrived.
pub fn subscribe_ticket(&mut self, build_id: &str, step_id: &str) {
self.register_ticket(build_id, step_id);
}
fn register_ticket(&self, build_id: &str, step_id: &str) {
if build_id.is_empty() {
return;
}
if let Ok(mut t) = self.tickets.lock() {
t.insert((build_id.to_string(), step_id.to_string()));
}
}
/// The live tickets currently tracked, as `(build_id, step_id)` — the set the
/// poller re-`pull`s. Ordered (a `BTreeSet`), so `state_json` is stable.
fn tickets_snapshot(&self) -> Vec<(String, String)> {
self.tickets
.lock()
.map(|t| t.iter().cloned().collect())
.unwrap_or_default()
}
/// Drain every batch the poller (or a test) has delivered to the inbox and
/// fold it through the SAME [`ingest`](Self::ingest). Each non-empty batch
/// ADVANCES [`poll_count`](Self::poll_count) — the auto-refresh "tick". Called
/// every [`draw`](Self::draw); cheap when the inbox is empty.
fn drain_inbox(&mut self) {
while let Ok(rows) = self.inbox_rx.try_recv() {
if !rows.is_empty() {
self.ingest(&rows);
self.poll_count += 1;
}
}
}
/// Test seam: the inbox sender, so a test can simulate the poller delivering a
/// batch on an interval (each `send` == one interval's freshly-pulled rows)
/// WITHOUT a live Flight server, then [`pump`](Self::pump) / draw to fold it.
#[doc(hidden)]
pub fn poll_sender_for_test(&self) -> Sender<Vec<BuildInfoRow>> {
self.inbox_tx.clone()
}
/// Drain the inbox once (what [`draw`](Self::draw) does each frame), exposed so
/// a headless test can advance the fold after simulating an interval pull.
#[doc(hidden)]
pub fn pump(&mut self) {
self.drain_inbox();
}
/// **Start the auto-refresh timer** (gated `build-info-flight`): spawn a
/// background poller that, on its own OS-thread tokio runtime, re-`pull`s every
/// tracked sub-job ticket every `interval` and hands the freshly-arrived rows
/// back to the inbox (drained by [`draw`](Self::draw)). Idempotent — a second
/// call is a no-op while a poller is live. The runtime lives + dies on the
/// poller thread (mirrors [`crate::build::flight_sink::FlightBuildInfoSink`]),
/// so `pull` never blocks the egui frame nor needs an ambient runtime.
#[cfg(feature = "build-info-flight")]
pub fn start_poller(
&mut self,
endpoint: impl Into<String>,
token: Option<String>,
interval: Duration,
) {
use std::sync::atomic::{AtomicBool, Ordering};
let endpoint = endpoint.into();
self.endpoint = Some(endpoint.clone());
if self.poller.is_some() {
return;
}
let tx = self.inbox_tx.clone();
let tickets = self.tickets.clone();
let stop = Arc::new(AtomicBool::new(false));
let stop_thread = stop.clone();
let _ = std::thread::Builder::new()
.name("build-info-live-poll".into())
.spawn(move || {
let rt = match tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
{
Ok(rt) => rt,
Err(_) => return,
};
rt.block_on(async move {
// Per-key high-water mark: `subscribe` returns the FULL stream
// for a key each pull, so forward only rows past what we've
// already sent — the shared append-only `ingest` never
// double-folds a re-pull (L2).
let mut seen: std::collections::HashMap<String, usize> =
std::collections::HashMap::new();
while !stop_thread.load(Ordering::Relaxed) {
let keys = tickets
.lock()
.map(|t| t.iter().cloned().collect::<Vec<(String, String)>>())
.unwrap_or_default();
for (build_id, step_id) in keys {
let key = crate::build::flight_sink::route_key(&build_id, &step_id);
if let Ok(rows) = crate::build::flight_sink::subscribe(
&endpoint,
token.as_deref(),
&key,
)
.await
{
let already = seen.get(&key).copied().unwrap_or(0);
if rows.len() > already {
let fresh: Vec<BuildInfoRow> = rows[already..].to_vec();
seen.insert(key.clone(), rows.len());
if tx.send(fresh).is_err() {
return; // pane dropped — end the poller.
}
}
}
}
tokio::time::sleep(interval).await;
}
});
});
self.poller = Some(Poller { stop });
}
/// Get-or-create the tail bucket for `step_id`, preserving first-seen order.
fn step_mut(&mut self, step_id: &str) -> &mut StepTail {
if let Some(idx) = self.steps.iter().position(|s| s.step_id == step_id) {
&mut self.steps[idx]
} else {
self.steps.push(StepTail::new(step_id));
self.steps.last_mut().expect("just pushed")
}
}
/// Total tail lines across all sub-jobs — the number that must GROW as
/// batches arrive (the "tail growing" the robot test asserts).
fn tail_total(&self) -> usize {
self.steps.iter().map(|s| s.tail.len()).sum()
}
fn emit_trace(&self) {
super::trace::emit_end(
"build.info.live",
&serde_json::json!({
"build_id": self.build_id,
"steps": self.steps.len(),
"tail_total": self.tail_total(),
"manifest_rows": self.manifest_rows,
}),
);
#[cfg(feature = "testmatrix")]
nornir_testmatrix::functional_status(
"nornir/viz/build-info-live",
"ingest_feed",
true,
&format!(
"build_id={:?} steps={} tail={}",
self.build_id,
self.steps.len(),
self.tail_total()
),
);
}
/// The pane rendered as data (LAW #6) — folded into the app's
/// `state_json["build_info_live"]`. The proof the pane subscribes the feed +
/// folds it per sub-job, not pixels.
pub fn state_json(&self) -> serde_json::Value {
#[cfg(feature = "build-info-flight")]
let polling = self.poller.is_some();
#[cfg(not(feature = "build-info-flight"))]
let polling = false;
serde_json::json!({
"palette": self.theme.name,
"endpoint": self.endpoint,
"error": self.error,
"build_id": self.build_id,
"step_count": self.steps.len(),
"manifest_rows": self.manifest_rows,
"tail_total": self.tail_total(),
// Auto-refresh observability: whether the poller is live, how many
// interval-pull batches have been folded (the "tick" that ADVANCES),
// and the live sub-job tickets the poller re-subscribes.
"polling": polling,
"poll_count": self.poll_count,
"live_tickets": self.tickets_snapshot().iter().map(|(b, s)| serde_json::json!({
"build_id": b,
"step_id": s,
})).collect::<Vec<_>>(),
"steps": self.steps.iter().map(|s| serde_json::json!({
"step_id": s.step_id,
"status": s.status,
"status_history": s.status_history,
"tail_len": s.tail.len(),
"tail": s.tail.iter().map(|l| serde_json::json!({
"kind": l.kind,
"level": l.level,
"name": l.name,
"value": l.value,
"ts_micros": l.ts_micros,
})).collect::<Vec<_>>(),
})).collect::<Vec<_>>(),
})
}
/// Paint the live-tail pane, themed. One collapsing section per sub-job with a
/// status chip + the ordered tail; the whole thing scrolls.
pub fn draw(&mut self, ui: &mut egui::Ui) {
// Auto-refresh: fold whatever the background poller delivered since the
// last frame, and keep the egui frame ticking so new rows appear without
// any user input (mirrors `super::bench_live`'s `request_repaint_after`).
self.drain_inbox();
ui.ctx().request_repaint_after(POLL_EVERY);
let theme = self.theme;
ui.heading(RichText::new("📡 Build-info — live tail").color(theme.text));
ui.horizontal(|ui| {
if let Some(b) = &self.build_id {
ui.label(RichText::new(format!("build {b}")).color(theme.text_dim));
}
if let Some(ep) = &self.endpoint {
ui.label(RichText::new(format!("· {ep}")).color(theme.text_dim));
}
ui.label(
RichText::new(format!(
"· {} sub-job(s) · {} tail line(s)",
self.steps.len(),
self.tail_total()
))
.color(theme.text_dim),
);
});
if let Some(err) = self.error.clone() {
ui.colored_label(RED, format!("subscribe failed: {err}"));
}
if self.steps.is_empty() {
ui.label(
RichText::new("no build-info feed yet — waiting for a sub-job to emit")
.color(theme.text_dim),
);
return;
}
ui.separator();
ScrollArea::vertical()
.auto_shrink([false, false])
.show(ui, |ui| {
for step in &self.steps {
let (chip, color) = facett_ui::status_chip(&theme, &step.status);
let title = if step.step_id.is_empty() {
"(root)".to_string()
} else {
step.step_id.clone()
};
egui::CollapsingHeader::new(
RichText::new(format!("{chip} {title}")).color(color),
)
.default_open(true)
.show(ui, |ui| {
if !step.status_history.is_empty() {
ui.label(
RichText::new(format!(
"status: {}",
step.status_history.join(" → ")
))
.color(theme.text_dim),
);
}
for line in &step.tail {
let lc = match line.level.as_str() {
"error" | "failed" => RED,
"running" | "warn" => AMBER,
"done" | "info" => GREEN,
_ => theme.text,
};
ui.label(
RichText::new(format!(
"[{}] {} = {}",
line.kind, line.name, line.value
))
.color(lc)
.monospace(),
);
}
});
}
});
}
}
#[cfg(test)]
mod tests {
use super::*;
use nornir_build_thing::build_info::{BuildInfoKind, BuildInfoRow};
/// A `step` row for `step_id` carrying `status` in the §f3 `level` column.
fn step_row(build: &str, step: &str, ts: i64, status: &str) -> BuildInfoRow {
BuildInfoRow::new(build, ts, BuildInfoKind::Step)
.step(step)
.level(status)
.kv("cargo build", "full")
}
/// A `log` row for `step_id`.
fn log_row(build: &str, step: &str, ts: i64, msg: &str) -> BuildInfoRow {
BuildInfoRow::new(build, ts, BuildInfoKind::Log)
.step(step)
.level("info")
.kv("cargo", msg)
}
/// RED-when-broken: feeding the feed in batches GROWS the tail and tracks
/// each sub-job's status transitions — read back out of `state_json`.
#[test]
fn ingest_grows_tail_and_tracks_status() {
let build = "b-live";
let mut tail = BuildInfoTail::new();
// Batch 1: the "compile" sub-job starts running and logs a line.
tail.ingest(&[
step_row(build, "compile", 1, "running"),
log_row(build, "compile", 2, "Compiling holger v0.1"),
]);
let v1 = tail.state_json();
assert_eq!(v1["build_id"], "b-live");
assert_eq!(v1["step_count"], 1);
assert_eq!(v1["tail_total"], 2, "step + log tailed");
let compile1 = &v1["steps"][0];
assert_eq!(compile1["step_id"], "compile");
assert_eq!(compile1["status"], "running", "first status");
assert_eq!(compile1["tail_len"], 2);
// Batch 2: compile finishes (a transition), a metric lands, and a second
// sub-job "test" appears — the tail must GROW, not replace.
tail.ingest(&[
log_row(build, "compile", 3, "Finished release"),
step_row(build, "compile", 4, "done"),
step_row(build, "test", 5, "running"),
]);
let v2 = tail.state_json();
assert_eq!(v2["step_count"], 2, "second sub-job appeared");
assert_eq!(
v2["tail_total"], 5,
"tail GREW to 5 across the two batches, not reset"
);
// compile: status transitioned running → done, both recorded in order.
let compile2 = v2["steps"]
.as_array()
.unwrap()
.iter()
.find(|s| s["step_id"] == "compile")
.unwrap();
assert_eq!(compile2["status"], "done", "latest status");
let hist: Vec<&str> = compile2["status_history"]
.as_array()
.unwrap()
.iter()
.map(|s| s.as_str().unwrap())
.collect();
assert_eq!(hist, vec!["running", "done"], "transitions in order");
assert_eq!(compile2["tail_len"], 4, "compile tail grew to 4");
// test: newly running.
let test = v2["steps"]
.as_array()
.unwrap()
.iter()
.find(|s| s["step_id"] == "test")
.unwrap();
assert_eq!(test["status"], "running");
assert_eq!(test["tail_len"], 1);
}
/// RED-when-broken (auto-refresh): a batch delivered to the inbox (what the
/// background poller does on each interval `pull`) is folded through the SAME
/// `ingest` when the pane drains — so `state_json` ADVANCES across simulated
/// interval pulls (tail grows, `poll_count` ticks) without a re-inject, and
/// the live-ticket set auto-grows to cover each discovered sub-job.
#[test]
fn auto_refresh_drains_inbox_and_advances_ticks() {
let build = "b-auto";
let mut tail = BuildInfoTail::new();
let tx = tail.poll_sender_for_test();
// Nothing pulled yet.
let v0 = tail.state_json();
assert_eq!(v0["tail_total"], 0);
assert_eq!(v0["poll_count"], 0);
assert_eq!(v0["live_tickets"].as_array().unwrap().len(), 0);
// Interval-pull #1 lands a batch; drain (what `draw` does) folds it.
tx.send(vec![
step_row(build, "compile", 1, "running"),
log_row(build, "compile", 2, "Compiling holger v0.1"),
])
.unwrap();
tail.pump();
let v1 = tail.state_json();
assert_eq!(v1["tail_total"], 2, "first interval pull folded");
assert_eq!(v1["poll_count"], 1, "one tick");
// Interval-pull #2 ADVANCES the tail (must GROW, not reset) + ticks again.
tx.send(vec![
step_row(build, "compile", 3, "done"),
step_row(build, "test", 4, "running"),
])
.unwrap();
tail.pump();
let v2 = tail.state_json();
assert_eq!(
v2["tail_total"], 4,
"tail GREW after the second interval pull"
);
assert_eq!(v2["poll_count"], 2, "ticks advanced per interval pull");
// The live-ticket set auto-grew to cover both discovered sub-jobs.
let tickets = v2["live_tickets"].as_array().unwrap();
assert!(
tickets.iter().any(|t| t["step_id"] == "compile"),
"compile ticket"
);
assert!(
tickets.iter().any(|t| t["step_id"] == "test"),
"test ticket"
);
// compile transitioned running → done across the two pulls.
let compile = v2["steps"]
.as_array()
.unwrap()
.iter()
.find(|s| s["step_id"] == "compile")
.unwrap();
assert_eq!(compile["status"], "done", "status advanced with the pulls");
}
/// An empty batch is not a tick — the poller only forwards freshly-pulled rows
/// (rows past the per-key high-water mark), so a no-op interval must not
/// advance `poll_count` nor the tail.
#[test]
fn empty_interval_pull_is_not_a_tick() {
let mut tail = BuildInfoTail::new();
let tx = tail.poll_sender_for_test();
tx.send(vec![]).unwrap();
tail.pump();
let v = tail.state_json();
assert_eq!(v["poll_count"], 0, "an empty pull is not a tick");
assert_eq!(v["tail_total"], 0);
}
/// Manifest rows (empty step_id) are counted, not tailed — the feed pane is
/// the live log/metric/step stream, not the static build record.
#[test]
fn manifest_rows_counted_not_tailed() {
let mut tail = BuildInfoTail::new();
tail.ingest(&[
BuildInfoRow::new("b", 1, BuildInfoKind::Provenance).kv("repo", "holger"),
BuildInfoRow::new("b", 2, BuildInfoKind::Dep).kv("feature", "full"),
BuildInfoRow::new("b", 3, BuildInfoKind::Artifact).kv("target/x", "10"),
log_row("b", "compile", 4, "line"),
]);
let v = tail.state_json();
assert_eq!(v["manifest_rows"], 3, "3 manifest rows counted");
assert_eq!(v["step_count"], 1, "only the feed row made a sub-job");
assert_eq!(v["tail_total"], 1, "manifest rows not in the tail");
}
}