|
|
@@ -1,11 +1,20 @@
|
|
|
//! Live-update plumbing for `report --serve`: a background refresher that
|
|
|
-//! re-scans the recipe `reports/` dirs on a tick, and an SSE endpoint
|
|
|
-//! (`/events`) that pushes the rows JSON to the page whenever anything
|
|
|
-//! changed (a new report file, a bench finishing, a log growing).
|
|
|
+//! 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).
|
|
|
//!
|
|
|
-//! One refresher serves all clients: it publishes the rows JSON on a `watch`
|
|
|
-//! channel only when the serialization differs from the last publish, and
|
|
|
-//! each SSE connection forwards its own `watch::Receiver`.
|
|
|
+//! 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;
|
|
|
@@ -16,14 +25,20 @@ use axum::{
|
|
|
routing::get,
|
|
|
};
|
|
|
use futures::{Stream, StreamExt};
|
|
|
+use notify::Watcher;
|
|
|
use std::convert::Infallible;
|
|
|
use std::sync::Arc;
|
|
|
use std::time::Duration;
|
|
|
|
|
|
-/// How often the refresher re-scans the report dirs. The scan is a local
|
|
|
-/// dir walk + small-file reads (a few dozen ms); 5 s keeps a finishing bench
|
|
|
-/// visible in the page within one poll while staying negligible I/O.
|
|
|
-pub(crate) const REFRESH_TICK: Duration = Duration::from_secs(5);
|
|
|
+/// 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).
|
|
|
@@ -33,23 +48,59 @@ pub(crate) struct AppState {
|
|
|
pub(crate) data: tokio::sync::watch::Receiver<String>,
|
|
|
}
|
|
|
|
|
|
-/// Background refresher: collect the rows, publish the JSON when it changed,
|
|
|
-/// sleep, repeat. The first collect runs immediately so clients see current
|
|
|
-/// data without waiting a tick. Runs on the tokio runtime; the scan is fast
|
|
|
-/// local disk I/O.
|
|
|
+/// 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<Config>,
|
|
|
tx: tokio::sync::watch::Sender<String>,
|
|
|
- tick: Duration,
|
|
|
+ keepalive: Duration,
|
|
|
) {
|
|
|
+ let (ev_tx, mut ev_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
|
|
|
+ let watcher = notify::recommended_watcher(move |res: Result<notify::Event, notify::Error>| {
|
|
|
+ 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)).unwrap_or_default();
|
|
|
if json != last {
|
|
|
last = json;
|
|
|
let _ = tx.send(last.clone());
|
|
|
}
|
|
|
- tokio::time::sleep(tick).await;
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -125,9 +176,9 @@ async fn report_file(
|
|
|
/// 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<Config>, page: String, initial: String, tick: Duration) -> Router {
|
|
|
+pub(crate) fn router(config: Arc<Config>, page: String, initial: String) -> Router {
|
|
|
let (tx, rx) = tokio::sync::watch::channel(initial);
|
|
|
- tokio::spawn(refresh_loop(config.clone(), tx, tick));
|
|
|
+ tokio::spawn(refresh_loop(config.clone(), tx, KEEPALIVE));
|
|
|
Router::new()
|
|
|
.route(
|
|
|
"/",
|
