dwarves-gui/src/pipeline_runs.rs
//! **The Pipeline tab's runs** — the selected pipeline's REAL runs, read from the forge.
//!
//! The leg used to be `NOT WIRED` ("no brokkr run is read into this view"). It now
//! asks the Dwarves server — the same `dwarf.v1` reads the 🔨 Forge tab uses, narrowed
//! to the selected pipeline (`History.pipeline`, then `GetRunReport` of the run shown,
//! then `StepLog` of the step picked) — on a worker off the UI thread, and draws:
//!
//! * the runs, newest first (status · steps passed · duration · when) — a click shows one;
//! * the shown run's stages and steps (status · duration · summary) — a click on a step
//! shows the tail of its log, `full log` the whole of it;
//! * a `BuildWorkspace` step's units (the run history's per-unit summary, drawn by the
//! Forge tab's own [`dwarves_embed::UnitsPane`]).
//!
//! A pipeline that has never run says so: [`NO_RUNS`]. A forge that does not answer
//! says that, with the reason. Nothing is read from this machine's disk.
//!
//! **Release runs** (pipeline `release-build`, lane RELBUILD-RUNA): a release build is a
//! Runa pipeline, so its runs are drawn here like any other. A release run's report adds
//! its release state (queued, building, waiting for signature, signed, failed), the
//! statement's summary, every artifact and its sha256 and, while it waits, the owner's
//! signing command and an upload of the signed bundle (`Dwarf.ReleaseBundle`). The
//! pipeline's leg also has the "Request release" form: the same inputs as `dwarves
//! release-build request`, sent to `Dwarf.Trigger`. The UI only sends and draws: the
//! forge checks, queues, verifies and stores.
use std::time::{Duration, Instant};
use dwarves_embed::pb;
use crate::room::Pending;
/// What a pipeline that never ran says.
pub const NO_RUNS: &str = "no runs yet: ▶ Run starts one";
/// How many runs the list asks for.
pub const RUNS_LIMIT: u32 = 50;
/// How many log lines a step shows before `full log`.
pub const LOG_TAIL: u32 = 200;
/// How often an open leg asks again (a run in flight fills in).
pub const REASK: Duration = Duration::from_secs(3);
/// The release pipeline's name (a release build is a Runa pipeline).
pub const RELEASE_PIPELINE: &str = "release-build";
/// The "Request release" form: `dwarves release-build request`'s inputs. The two files
/// are read here and their TEXT is sent; the forge does the rest.
#[derive(Default, Clone, Debug)]
pub struct ReleaseForm {
pub name: String,
/// A steps file or a `publish-*.sh` script.
pub steps_path: String,
pub pins_path: String,
/// `repo=org`, comma-separated.
pub orgs: String,
/// `repo/Cargo.toml`, comma-separated.
pub locks: String,
/// `crate:bin[:feat,feat][:no-default]`, comma-separated.
pub binaries: String,
pub rebuild_published: bool,
}
fn list(s: &str) -> Vec<String> {
s.split(',').map(str::trim).filter(|x| !x.is_empty()).map(str::to_string).collect()
}
impl ReleaseForm {
/// The request the form sends (`{"inputs": …}`), the files read.
pub fn request_json(&self) -> Result<String, String> {
let read = |p: &str| std::fs::read_to_string(p.trim()).map_err(|e| format!("{}: {e}", p.trim()));
Ok(serde_json::json!({"inputs": {
"name": self.name.trim(),
"steps": read(&self.steps_path)?,
"pins": read(&self.pins_path)?,
"orgs": list(&self.orgs),
"locks": list(&self.locks),
"binaries": list(&self.binaries),
"rebuild_published": self.rebuild_published,
}})
.to_string())
}
}
/// The runs leg of the Pipeline tab.
pub struct RunsLeg {
endpoint: String,
token: String,
pipeline: String,
/// The run picked (`None` ⇒ the newest).
pinned: Option<String>,
runs: Vec<pb::RunRow>,
report: Option<pb::RunReport>,
/// `idle` · `loading` · `loaded` · `no_runs` · `unavailable` · `report_error` · `no_forge` · `no_pipeline`
state: &'static str,
error: Option<String>,
pending: Option<Pending<Result<dwarves_embed::ForgeFetch, String>>>,
asked: Option<Instant>,
/// The step row picked (its `seq`) and whether the whole log was asked for.
step: Option<(u32, bool)>,
log: Option<pb::StepLogReply>,
log_error: Option<String>,
log_pending: Option<Pending<Result<pb::StepLogReply, String>>>,
units: dwarves_embed::UnitsPane,
runs_table: facett_table::Table,
steps_table: facett_table::Table,
/// The shown run's release (`RunReport.release_json`), when it is a release run.
release: Option<serde_json::Value>,
form: ReleaseForm,
request_pending: Option<Pending<Result<pb::TriggerReply, String>>>,
request_note: Option<String>,
bundle_path: String,
attach_pending: Option<Pending<Result<pb::TriggerReply, String>>>,
attach_note: Option<String>,
}
fn cols(c: &[&str]) -> Vec<String> {
c.iter().map(|s| s.to_string()).collect()
}
fn secs(ms: u64) -> String {
if ms == 0 {
"–".to_string()
} else {
format!("{:.1} s", ms as f64 / 1000.0)
}
}
impl RunsLeg {
/// A leg over the forge at `endpoint` (empty ⇒ `no_forge`).
pub fn new(endpoint: impl Into<String>, token: impl Into<String>) -> Self {
Self {
endpoint: endpoint.into(),
token: token.into(),
pipeline: String::new(),
pinned: None,
runs: Vec::new(),
report: None,
state: "idle",
error: None,
pending: None,
asked: None,
step: None,
log: None,
log_error: None,
log_pending: None,
units: dwarves_embed::UnitsPane::default(),
runs_table: facett_table::Table::new("pipeline-runs", Vec::new()),
steps_table: facett_table::Table::new("pipeline-run-steps", Vec::new()),
release: None,
form: ReleaseForm::default(),
request_pending: None,
request_note: None,
bundle_path: String::new(),
attach_pending: None,
attach_note: None,
}
}
/// Show `pipeline`'s runs (a different pipeline starts over).
pub fn set_pipeline(&mut self, pipeline: &str) {
if pipeline == self.pipeline {
return;
}
self.pipeline = pipeline.to_string();
self.pinned = None;
self.runs.clear();
self.report = None;
self.step = None;
self.log = None;
self.state = "idle";
self.asked = None;
self.rebuild();
}
/// Ask the forge (on a worker) for the runs and the shown run's report.
pub fn ask(&mut self) {
if self.endpoint.trim().is_empty() {
self.state = "no_forge";
return;
}
if self.pipeline.is_empty() {
self.state = "no_pipeline";
return;
}
if self.pending.is_some() {
return;
}
let (ep, tok, pl, pin) = (
self.endpoint.clone(),
self.token.clone(),
self.pipeline.clone(),
self.pinned.clone(),
);
self.pending = Some(Pending::spawn("pipeline-runs", move || {
dwarves_embed::fetch_pipeline_runs(&ep, &tok, &pl, pin.as_deref(), RUNS_LIMIT)
.map_err(|e| format!("{e:#}"))
}));
self.asked = Some(Instant::now());
if self.state == "idle" {
self.state = "loading";
}
}
fn ask_log(&mut self) {
let (Some((seq, full)), Some(run)) = (self.step, self.shown_run()) else {
return;
};
let (ep, tok) = (self.endpoint.clone(), self.token.clone());
let tail = if full { 0 } else { LOG_TAIL };
self.log_pending = Some(Pending::spawn("pipeline-step-log", move || {
dwarves_embed::fetch_step_log(&ep, &tok, &run, seq, tail).map_err(|e| format!("{e:#}"))
}));
}
fn land(&mut self, f: Result<dwarves_embed::ForgeFetch, String>) {
match f {
Ok(f) => {
self.runs = f.runs;
self.error = None;
match f.report {
Ok(r) => {
let changed = self
.report
.as_ref()
.and_then(|x| x.run.as_ref())
.map(|x| &x.run_id)
!= r.run.as_ref().map(|x| &x.run_id);
self.units.set(dwarves_embed::UnitsPane::of_report(&r));
self.release = (!r.release_json.is_empty())
.then(|| serde_json::from_str(&r.release_json).ok())
.flatten();
self.report = Some(r);
self.state = "loaded";
if changed {
self.step = None;
self.log = None;
}
}
Err(e) if self.runs.is_empty() => {
let _ = e;
self.report = None;
self.units.set(Vec::new());
self.state = "no_runs";
}
Err(e) => {
self.report = None;
self.state = "report_error";
self.error = Some(format!("{e:#}"));
}
}
}
Err(e) => {
self.state = "unavailable";
self.error = Some(e);
}
}
self.rebuild();
}
/// Take what the workers answered; ask again when it is time.
pub fn poll(&mut self) {
if let Some(p) = &self.pending {
match p.try_take() {
None => {}
Some(r) => {
self.pending = None;
self.land(r.unwrap_or_else(|()| Err("the runs worker died".to_string())));
}
}
}
if let Some(p) = &self.log_pending {
match p.try_take() {
None => {}
Some(r) => {
self.log_pending = None;
match r.unwrap_or_else(|()| Err("the log worker died".to_string())) {
Ok(l) => {
self.log = Some(l);
self.log_error = None;
}
Err(e) => self.log_error = Some(e),
}
}
}
}
if let Some(r) = self.request_pending.as_ref().and_then(|p| p.try_take()) {
self.request_pending = None;
self.request_note = Some(match r.unwrap_or_else(|()| Err("the request worker died".into())) {
Ok(t) => format!("queued: run {} — {}", t.run_id, t.status),
Err(e) => format!("refused: {e}"),
});
self.asked = None;
}
if let Some(r) = self.attach_pending.as_ref().and_then(|p| p.try_take()) {
self.attach_pending = None;
self.attach_note = Some(match r.unwrap_or_else(|()| Err("the upload worker died".into())) {
Ok(t) => t.status,
Err(e) => format!("not stored: {e}"),
});
self.asked = None;
}
if self.pending.is_none() && self.asked.is_none_or(|t| t.elapsed() >= REASK) {
self.ask();
}
}
/// Send the "Request release" form (`Dwarf.Trigger`, on a worker).
pub fn request_release(&mut self) {
let json = match self.form.request_json() {
Ok(j) => j,
Err(e) => {
self.request_note = Some(format!("not sent: {e}"));
return;
}
};
let (ep, tok) = (self.endpoint.clone(), self.token.clone());
self.request_pending = Some(Pending::spawn("release-request", move || {
dwarves_embed::trigger_release(&ep, &tok, &json).map_err(|e| format!("{e:#}"))
}));
self.request_note = Some("sending…".into());
}
/// Upload the signed bundle at `bundle_path` for the shown release run
/// (`Dwarf.ReleaseBundle`, on a worker): the forge verifies it before it stores it.
pub fn attach_bundle(&mut self) {
let Some(run) = self.shown_run() else { return };
let bytes = match std::fs::read(self.bundle_path.trim()) {
Ok(b) => b,
Err(e) => {
self.attach_note = Some(format!("{}: {e}", self.bundle_path.trim()));
return;
}
};
let (ep, tok) = (self.endpoint.clone(), self.token.clone());
self.attach_pending = Some(Pending::spawn("release-bundle", move || {
dwarves_embed::release_bundle(&ep, &tok, &run, bytes).map_err(|e| format!("{e:#}"))
}));
self.attach_note = Some("uploading…".into());
}
/// The form (robot seam and tests).
pub fn form_mut(&mut self) -> &mut ReleaseForm {
&mut self.form
}
/// Where the signed bundle is read from (robot seam and tests).
pub fn set_bundle_path(&mut self, p: &str) {
self.bundle_path = p.to_string();
}
/// Block until the request and the upload in flight have landed (robot seam).
pub fn wait_actions(&mut self) {
if let Some(p) = self.request_pending.take() {
self.request_note = Some(match p.wait().unwrap_or_else(|()| Err("the request worker died".into())) {
Ok(t) => format!("queued: run {} — {}", t.run_id, t.status),
Err(e) => format!("refused: {e}"),
});
}
if let Some(p) = self.attach_pending.take() {
self.attach_note = Some(match p.wait().unwrap_or_else(|()| Err("the upload worker died".into())) {
Ok(t) => t.status,
Err(e) => format!("not stored: {e}"),
});
}
}
/// The request form, on the release pipeline's leg.
fn request_ui(&mut self, ui: &mut egui::Ui) {
let mut send = false;
egui::CollapsingHeader::new("Request release")
.id_salt("release-request-form")
.show(ui, |ui| {
let f = &mut self.form;
egui::Grid::new("release-request-grid").num_columns(2).show(ui, |ui| {
for (label, v) in [
("release name", &mut f.name),
("steps file or publish script", &mut f.steps_path),
("pins file", &mut f.pins_path),
("orgs (repo=org, …)", &mut f.orgs),
("locks (repo/Cargo.toml, …)", &mut f.locks),
("binaries (crate:bin[:features][:no-default], …)", &mut f.binaries),
] {
ui.label(label);
ui.text_edit_singleline(v);
ui.end_row();
}
});
ui.checkbox(&mut f.rebuild_published, "rebuild published crates (a reproducibility check)");
if ui
.add_enabled(self.request_pending.is_none(), egui::Button::new("Request release"))
.clicked()
{
send = true;
}
if let Some(n) = &self.request_note {
ui.label(n);
}
});
if send {
self.request_release();
}
}
/// A release run's release: state, statement, artifacts, the signing command, the upload.
fn release_ui(&mut self, ui: &mut egui::Ui) {
let Some(v) = self.release.clone() else { return };
let s = |k: &str| v[k].as_str().unwrap_or("").to_string();
ui.separator();
ui.strong(format!("release {} · {}", s("release"), s("state")));
egui::Grid::new("release-run-grid").num_columns(2).show(ui, |ui| {
for (k, label) in [
("request_sha256", "request sha256"),
("statement_sha256", "statement sha256"),
("builder_id", "builder"),
("runner_account", "runner"),
("runner_sha256", "runner binary sha256"),
("run", "run directory"),
("error", "error"),
("bundle_error", "bundle not taken"),
] {
if !s(k).is_empty() {
ui.label(label);
ui.monospace(s(k));
ui.end_row();
}
}
for src in v["sources"].as_array().into_iter().flatten() {
ui.label("source");
ui.monospace(format!("{} @ {}", src[0].as_str().unwrap_or(""), src[1].as_str().unwrap_or("")));
ui.end_row();
}
});
let arts = v["artifacts"].as_array().cloned().unwrap_or_default();
if !arts.is_empty() {
ui.label(format!("artifacts ({})", arts.len()));
egui::Grid::new("release-artifacts").num_columns(2).striped(true).show(ui, |ui| {
for a in &arts {
ui.monospace(a[0].as_str().unwrap_or(""));
ui.monospace(a[1].as_str().unwrap_or(""));
ui.end_row();
}
});
}
if s("state") == "waiting for signature" {
ui.label("waiting for the owner's signature (the YubiKey stays on loki):");
let mut cmd = s("sign");
ui.add(egui::TextEdit::multiline(&mut cmd).code_editor().desired_rows(2));
let mut up = false;
ui.horizontal(|ui| {
ui.label("signed bundle (release.intoto.jsonl)");
ui.text_edit_singleline(&mut self.bundle_path);
up = ui
.add_enabled(self.attach_pending.is_none(), egui::Button::new("Upload signed bundle"))
.clicked();
});
if up {
self.attach_bundle();
}
}
if let Some(n) = &self.attach_note {
ui.label(n);
}
}
/// Block until what is in flight has landed (robot seam).
pub fn wait(&mut self) {
if self.pending.is_none() && self.state != "no_forge" {
self.ask();
}
if let Some(p) = self.pending.take() {
let r = p
.wait()
.unwrap_or_else(|()| Err("the runs worker died".to_string()));
self.land(r);
}
if let Some(p) = self.log_pending.take() {
match p
.wait()
.unwrap_or_else(|()| Err("the log worker died".to_string()))
{
Ok(l) => {
self.log = Some(l);
self.log_error = None;
}
Err(e) => self.log_error = Some(e),
}
}
}
/// Show run `run_id` (`""` ⇒ the newest).
pub fn select_run(&mut self, run_id: &str) {
self.pinned = (!run_id.is_empty()).then(|| run_id.to_string());
self.step = None;
self.log = None;
self.asked = None;
if self.pending.is_none() {
self.ask();
}
}
/// Show the log of step row `seq` of the shown run (`full` ⇒ all of it).
pub fn select_step(&mut self, seq: u32, full: bool) {
self.step = Some((seq, full));
self.log = None;
self.ask_log();
self.rebuild();
}
fn shown_run(&self) -> Option<String> {
self.report
.as_ref()
.and_then(|r| r.run.as_ref())
.map(|r| r.run_id.clone())
}
fn rebuild(&mut self) {
let mut runs = facett_table::Table::new(
"pipeline-runs",
cols(&["run", "status", "steps", "duration", "finished"]),
)
.with_row_ids(self.runs.iter().map(|r| r.run_id.clone()).collect());
for r in &self.runs {
runs.push_row(vec![
r.run_id.clone(),
r.status.clone(),
format!("{}/{}", r.passed, r.steps),
secs(r.duration_ms),
r.finished_at.clone(),
]);
}
if let Some(id) = self.shown_run() {
let _ = runs.update(facett_table::Msg::ToggleRow(id));
}
let steps_rows: Vec<&pb::BuildRow> = self
.report
.as_ref()
.map(|r| r.steps.iter().collect())
.unwrap_or_default();
let mut steps = facett_table::Table::new(
"pipeline-run-steps",
cols(&["stage", "step", "status", "duration", "summary"]),
)
.with_row_ids((0..steps_rows.len()).map(|i| format!("step-{i}")).collect());
for s in &steps_rows {
steps.push_row(vec![
s.stage.clone(),
s.step.clone(),
s.status.clone(),
secs(s.duration_ms),
s.summary.clone(),
]);
}
if let Some((seq, _)) = self.step {
let _ = steps.update(facett_table::Msg::ToggleRow(format!("step-{seq}")));
}
self.runs_table = runs;
self.steps_table = steps;
}
/// The one-line header.
pub fn line(&self) -> String {
match self.state {
"no_forge" => "no forge to ask (no server address)".to_string(),
"no_pipeline" => "no pipeline selected".to_string(),
"no_runs" => format!("{} · {NO_RUNS}", self.pipeline),
"unavailable" => format!(
"forge unavailable: {}",
self.error.as_deref().unwrap_or("?")
),
"report_error" => format!(
"{} runs · that run's report could not be read",
self.runs.len()
),
"loaded" => format!(
"{} · {} runs · showing {}",
self.pipeline,
self.runs.len(),
self.shown_run().unwrap_or_default()
),
_ => format!("{} · asking the forge…", self.pipeline),
}
}
/// Draw the runs, the shown run's steps, its units and the picked step's log.
pub fn ui(&mut self, ui: &mut egui::Ui) {
self.poll();
ui.ctx().request_repaint_after(REASK);
ui.label(self.line());
if self.pipeline == RELEASE_PIPELINE {
self.request_ui(ui);
}
if self.state == "no_runs" {
ui.heading(NO_RUNS);
return;
}
if self.runs.is_empty() {
return;
}
let mut pick_run: Option<String> = None;
let mut pick_step: Option<(u32, bool)> = None;
egui::ScrollArea::vertical()
.id_salt("pipeline-runs-all")
.show(ui, |ui| {
ui.strong("runs — newest first");
ui.push_id("pipeline-runs-table", |ui| {
egui::ScrollArea::vertical()
.max_height(160.0)
.show(ui, |ui| {
for m in self.runs_table.view(ui) {
if let facett_table::Msg::ToggleRow(id) = m {
pick_run = Some(id);
}
}
});
});
if let Some(run) = self.report.as_ref().and_then(|r| r.run.as_ref()) {
ui.separator();
ui.strong(format!(
"run {} · {} · {} · {}",
run.run_id,
run.status,
secs(run.duration_ms),
run.trigger
));
ui.push_id("pipeline-run-steps-table", |ui| {
egui::ScrollArea::vertical()
.max_height(220.0)
.show(ui, |ui| {
for m in self.steps_table.view(ui) {
if let facett_table::Msg::ToggleRow(id) = m {
if let Some(seq) =
id.strip_prefix("step-").and_then(|n| n.parse().ok())
{
pick_step = Some((seq, false));
}
}
}
});
});
self.units.ui(ui);
self.release_ui(ui);
if let Some((seq, full)) = self.step {
ui.separator();
ui.horizontal(|ui| {
let what = match &self.log {
Some(l) if !l.found => "this step kept no log".to_string(),
Some(l) => format!(
"log of step {seq} · {} of {} lines",
l.lines.len(),
l.total
),
None => format!("log of step {seq} · asking…"),
};
ui.label(what);
if !full
&& self
.log
.as_ref()
.is_some_and(|l| (l.lines.len() as u64) < l.total)
&& ui.button("full log").clicked()
{
pick_step = Some((seq, true));
}
});
if let Some(e) = &self.log_error {
ui.colored_label(egui::Color32::from_rgb(224, 120, 90), e);
}
if let Some(l) = &self.log {
egui::ScrollArea::vertical()
.id_salt("pipeline-step-log")
.max_height(320.0)
.stick_to_bottom(true)
.show(ui, |ui| {
ui.monospace(l.lines.join("\n"));
});
}
}
}
});
if let Some(id) = pick_run {
self.select_run(&id);
}
if let Some((seq, full)) = pick_step {
self.select_step(seq, full);
}
}
/// Everything the leg shows, for a robot.
pub fn state_json(&self) -> serde_json::Value {
let report = self.report.as_ref();
let mut stages: Vec<serde_json::Value> = Vec::new();
if let Some(r) = report {
for (seq, s) in r.steps.iter().enumerate() {
let step = serde_json::json!({ "seq": seq, "step": s.step, "status": s.status, "duration_ms": s.duration_ms, "summary": s.summary, "finished_at": s.finished_at });
match stages.iter_mut().find(|x| x["stage"] == s.stage.as_str()) {
Some(x) => x["steps"].as_array_mut().expect("steps").push(step),
None => stages.push(serde_json::json!({ "stage": s.stage, "steps": [step] })),
}
}
}
serde_json::json!({
"state": self.state,
"line": self.line(),
"pipeline": self.pipeline,
"error": self.error,
"runs": self.runs.iter().map(|r| serde_json::json!({
"run_id": r.run_id, "status": r.status, "steps": r.steps, "passed": r.passed, "duration_ms": r.duration_ms, "finished_at": r.finished_at,
})).collect::<Vec<_>>(),
"run_id": self.shown_run(),
"stages": stages,
"units": self.units.state_json(),
"step": self.step.map(|s| s.0),
"log": self.log.as_ref().map(|l| serde_json::json!({ "found": l.found, "total": l.total, "shown": l.lines.len(), "full": self.step.is_some_and(|s| s.1), "tail": l.lines.iter().rev().take(20).rev().collect::<Vec<_>>() })),
"log_error": self.log_error,
"drawn": { "run_rows": self.runs_table.rows.len(), "step_rows": self.steps_table.rows.len() },
"release": self.release,
"request_note": self.request_note,
"attach_note": self.attach_note,
})
}
}