//! Live-update plumbing for `report --serve`: a background refresher that //! re-scans the recipe `reports/` dirs when the file system says something //! changed, and an SSE endpoint (`/events`) that pushes the rows JSON to the //! page whenever the scan result differs (a new report file, a bench //! finishing, a log growing). //! //! The refresher is event-driven, not a fixed poll: a `notify` watcher //! (FSEvents on macOS, inotify on Linux) over the recipe dirs feeds an mpsc //! channel; after a short debounce burst the rows are re-collected and //! published on the `watch` channel only when the serialization changed. //! Filesystem events are filtered to the report-relevant paths (anything //! under a `reports/` dir, or a `recipe.toml`), so build churn elsewhere in //! the repo is ignored. A slow keepalive scan stays as a safety net for //! missed notifications (legacy log-only reports, dirs created late). //! //! One refresher serves all clients: each SSE connection forwards its own //! `watch::Receiver`. use crate::config::Config; use crate::report::collect; use axum::{ Router, extract::Path, response::{Response, Sse, sse::Event as SseEvent}, routing::get, }; use futures::{Stream, StreamExt}; use notify::Watcher; use std::convert::Infallible; use std::sync::Arc; use std::time::Duration; /// Safety-net rescan interval: covers missed watcher notifications (a /// legacy `.log`-only report whose score lands in a Modify the filter /// ignores, a source dir created after startup). The normal path is the /// event-driven one, so this can stay slow. pub(crate) const KEEPALIVE: Duration = Duration::from_secs(30); /// Wait after a filesystem event before re-scanning, so a write burst /// (the launcher copying a report triplet) collapses into one collect. const DEBOUNCE: Duration = Duration::from_millis(500); /// Shared server state: the config (for `/r/...` file loads) plus the /// latest rows JSON (each SSE connection clones the receiver). #[derive(Clone)] pub(crate) struct AppState { pub(crate) config: Arc, pub(crate) data: tokio::sync::watch::Receiver, } /// True when the event touches a report-relevant path: anything under a /// `reports/` directory, or a `recipe.toml` (row fields come from it). /// Everything else (build churn, git, the rest of the repo) is ignored so /// the watcher can watch a wide tree cheaply. fn is_relevant(ev: ¬ify::Event) -> bool { ev.paths.iter().any(|p| { p.ancestors() .any(|a| a.file_name().is_some_and(|n| n == "reports")) || p.file_name().is_some_and(|n| n == "recipe.toml") }) } /// Background refresher: watch the recipe dirs, re-collect after a debounced /// filesystem event (or the keepalive safety net), publish the rows JSON on /// the watch channel only when it changed. The first collect runs /// immediately so clients see current data without waiting. If the watcher /// cannot be created (unsupported FS), the keepalive alone keeps the data /// fresh. pub(crate) async fn refresh_loop( cfg: Arc, tx: tokio::sync::watch::Sender, keepalive: Duration, archived: bool, ) { let (ev_tx, mut ev_rx) = tokio::sync::mpsc::unbounded_channel::<()>(); let watcher = notify::recommended_watcher(move |res: Result| { if let Ok(ev) = res && is_relevant(&ev) { let _ = ev_tx.send(()); } }); let mut watcher = watcher.ok(); if let Some(w) = &mut watcher { for dir in crate::config::recipe_dirs(&cfg) { if dir.is_dir() { let _ = w.watch(&dir, notify::RecursiveMode::Recursive); } } } let mut last = String::new(); loop { tokio::select! { _ = ev_rx.recv() => {} _ = tokio::time::sleep(keepalive) => {} } // debounce: let the burst settle, then drain the remaining events tokio::time::sleep(DEBOUNCE).await; while ev_rx.try_recv().is_ok() {} let json = serde_json::to_string(&collect(&cfg, archived)).unwrap_or_default(); if json != last { last = json; let _ = tx.send(last.clone()); } } } /// `GET /events` — the latest rows JSON on connect, then one message per /// refresher update. The stream ends when the refresher (the sender) drops. async fn events_stream( axum::extract::State(s): axum::extract::State, ) -> Sse>> { let mut rx = s.data; let first = rx.borrow_and_update().clone(); let first = futures::stream::once(async move { Ok(SseEvent::default().data(first)) }); let rest = futures::stream::unfold(rx, |mut rx| async move { if rx.changed().await.is_err() { return None; } let json = rx.borrow_and_update().clone(); Some((Ok(SseEvent::default().data(json)), rx)) }); Sse::new(first.chain(rest)) } /// `GET /r//` — full file text (report, log, json, or the /// recipe's own `recipe.toml`) served through the viewer page. Both path /// parts must be plain names (no traversal); the resolve walks every /// recipe dir, so stay strict. pub(crate) fn load_report(config: &Config, recipe: &str, file: &str) -> Option { let plain = |s: &str| { !s.contains("..") && s.chars() .all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-')) }; if !plain(recipe) || !plain(file) { return None; } crate::config::recipe_dirs(config) .into_iter() .find_map(|dir| { let base = dir.join(recipe); // "recipe.toml" resolves from the recipe dir root, everything // else from its reports/ subdir let p = if file == "recipe.toml" { base.join("recipe.toml") } else { base.join("reports").join(file) }; p.is_file() .then(|| std::fs::read_to_string(p).ok()) .flatten() }) } /// Braille spinner frame or progress bar — llama-benchy's rich in-place /// progress redraws. fn is_spinner_frame(line: &str) -> bool { let t = line.trim_start(); t.starts_with(['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏']) || t.contains('━') } /// Progress-redraw identity: the spinner frames differ only by the glyph, /// the elapsed clock and the run/progress counters — same key = same phase /// redrawn in place. fn progress_key(line: &str) -> String { line.split_whitespace() .filter(|t| { // keep only wordish tokens (the test name): bars — the tip glyph // varies as the bar advances (━╸╺…) — spinner glyphs, @ signs and // other graphics carry no alphanumerics; counters (0/18), elapsed // clocks (0:01:48) and the "run N" pair are progress churn. t.chars().any(|c| c.is_alphanumeric()) && !t.chars().all(|c| c.is_ascii_digit() || c == '/') && !t.contains(':') && *t != "run" }) .collect::>() .join(" ") } /// Make old pty-captured bench logs read like the new ones: drop pty /// control-char artifacts (`^D`/backspace echoes), collapse each spinner /// phase's redraw run down to its last frame, drop consecutive duplicate /// lines. Idempotent on logs that are already clean. pub(crate) fn clean_log(content: &str) -> String { let lines: Vec = content .lines() .map(|l| { let l: String = l .chars() .filter(|c| !matches!(c, '\u{4}' | '\u{8}')) .collect(); // the pty echoes control chars as caret notation (^D, ^H) — // printable '^' + @.._ pairs at the line start let mut l = l.as_str(); while l.starts_with('^') && l[1..] .chars() .next() .is_some_and(|c| ('@'..='_').contains(&c)) { l = &l[2..]; } l.to_string() }) .collect(); let mut keep = vec![true; lines.len()]; let mut i = 0; while i < lines.len() { let prog = is_spinner_frame(&lines[i]); let key = if prog { progress_key(&lines[i]) } else { String::new() }; let mut j = i + 1; while j < lines.len() { let same = if prog { is_spinner_frame(&lines[j]) && progress_key(&lines[j]) == key } else { !lines[i].is_empty() && lines[j] == lines[i] }; if !same { break; } keep[j] = false; j += 1; } if j > i + 1 && prog { // a redraw run keeps only its final frame (exact dups keep the first) keep[i] = false; keep[j - 1] = true; } i = j; } lines .iter() .enumerate() .filter(|(k, _)| keep[*k]) .map(|(_, l)| l.as_str()) .collect::>() .join("\n") } /// `GET /r//` — file viewer page: rendered markdown (with a /// raw toggle), syntax-highlighted TOML/JSON, plain logs. const VIEW_TEMPLATE: &str = include_str!("view.html"); async fn report_file( axum::extract::State(state): axum::extract::State, axum::extract::Path((recipe, file)): Path<(String, String)>, ) -> Response { use axum::http::StatusCode; use axum::response::IntoResponse; match load_report(&state.config, &recipe, &file) { Some(text) => { // Strip ANSI escapes (old logs were saved raw; new ones are // already clean) and normalize line ends: the old pty-saved // logs carry CRLF (ONLCR) and the Live-soup ones bare \r — // CRLF first, then lone \r, so nothing doubles. let clean = crate::ansi::strip_ansi(text.as_bytes()); let clean = String::from_utf8_lossy(&clean) .replace("\r\n", "\n") .replace('\r', "\n"); // old pty-captured logs carry spinner-redraw soup and pty // control-char artifacts — render them like the clean ones let clean = if file.ends_with(".log") { clean_log(&clean) } else { clean }; let meta = serde_json::json!({ "recipe": recipe, "file": file, "content": clean }); let meta = serde_json::to_string(&meta) .unwrap_or_default() .replace('<', "\\u003c"); axum::response::Html(VIEW_TEMPLATE.replace("{{META}}", &meta)).into_response() } None => ( StatusCode::NOT_FOUND, axum::response::Html( "404report not found" .to_string(), ), ) .into_response(), } } /// Build the report server router: static page at `/`, live SSE at /// `/events`, full report text at `/r//`. `initial` is the /// rows JSON snapshot used to seed the watch channel (from the same /// `collect` that rendered the page, so the first paint and the first SSE /// message agree). pub(crate) fn router(config: Arc, page: String, initial: String, archived: bool) -> Router { let (tx, rx) = tokio::sync::watch::channel(initial); tokio::spawn(refresh_loop(config.clone(), tx, KEEPALIVE, archived)); Router::new() .route( "/", get(move || { let page = page.clone(); async move { axum::response::Html(page) } }), ) .route("/events", get(events_stream)) .route("/r/{recipe}/{file}", get(report_file)) .with_state(AppState { config, data: rx }) }