|
|
@@ -65,18 +65,22 @@ fn is_relevant(ev: ¬ify::Event) -> bool {
|
|
|
})
|
|
|
}
|
|
|
|
|
|
-/// 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.
|
|
|
+/// Background refresher — lazy by design (the web process must stay small;
|
|
|
+/// a full collect scans every recipe + report file): a full re-collect runs
|
|
|
+/// ONLY on a debounced filesystem event (or once at startup); the keepalive
|
|
|
+/// tick just re-runs the live stale cross-check (the only row field that
|
|
|
+/// changes without a file change) and only when a run is open — otherwise it
|
|
|
+/// does zero work. The cross-check rides on the main dashboard's fleet
|
|
|
+/// probe (`fleet_rx` bus State events — no extra SSH) and falls back to a
|
|
|
+/// direct probe when no bus is wired (CLI `report --serve`).
|
|
|
pub(crate) async fn refresh_loop(
|
|
|
cfg: Arc<Config>,
|
|
|
tx: tokio::sync::watch::Sender<String>,
|
|
|
keepalive: Duration,
|
|
|
archived: bool,
|
|
|
+ fleet_rx: Option<tokio::sync::broadcast::Receiver<crate::web::events::Event>>,
|
|
|
) {
|
|
|
+ let mut fleet_rx = fleet_rx;
|
|
|
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
|
|
|
@@ -94,47 +98,111 @@ pub(crate) async fn refresh_loop(
|
|
|
}
|
|
|
}
|
|
|
let mut last = String::new();
|
|
|
+ let mut rows: Vec<crate::report::ReportRow> = Vec::new();
|
|
|
+ let mut fleet: Option<crate::web::events::Snapshot> = None;
|
|
|
+ let cluster_nodes: HashMap<String, Vec<String>> = cfg
|
|
|
+ .clusters
|
|
|
+ .iter()
|
|
|
+ .map(|c| {
|
|
|
+ (
|
|
|
+ c.name.clone(),
|
|
|
+ c.nodes.iter().map(|n| n.name.clone()).collect(),
|
|
|
+ )
|
|
|
+ })
|
|
|
+ .collect();
|
|
|
+ let has_open_run = |rows: &[crate::report::ReportRow]| {
|
|
|
+ rows.iter()
|
|
|
+ .any(|r| r.up.is_some() && r.up_cluster.is_some())
|
|
|
+ };
|
|
|
loop {
|
|
|
tokio::select! {
|
|
|
- _ = ev_rx.recv() => {}
|
|
|
- _ = tokio::time::sleep(keepalive) => {}
|
|
|
+ // Filesystem event (debounced) → the only full-rescan trigger.
|
|
|
+ _ = ev_rx.recv() => {
|
|
|
+ // debounce: let the burst settle, then drain the remaining events
|
|
|
+ tokio::time::sleep(DEBOUNCE).await;
|
|
|
+ while ev_rx.try_recv().is_ok() {}
|
|
|
+ rows = collect(&cfg, archived);
|
|
|
+ cross_check(&cfg, &mut rows, &cluster_nodes, &mut fleet).await;
|
|
|
+ let json = serde_json::to_string(&rows).unwrap_or_default();
|
|
|
+ if json != last {
|
|
|
+ last = json;
|
|
|
+ let _ = tx.send(last.clone());
|
|
|
+ }
|
|
|
+ }
|
|
|
+ // Bus State event (the main dashboard's 30 s fleet probe) →
|
|
|
+ // just track the fleet; the keepalive tick consumes it.
|
|
|
+ ev = async {
|
|
|
+ match fleet_rx.as_mut() {
|
|
|
+ Some(rx) => rx.recv().await,
|
|
|
+ None => std::future::pending::<Result<
|
|
|
+ crate::web::events::Event,
|
|
|
+ tokio::sync::broadcast::error::RecvError,
|
|
|
+ > >().await,
|
|
|
+ }
|
|
|
+ } => {
|
|
|
+ if let Ok(crate::web::events::Event::State { snapshot, .. }) = ev {
|
|
|
+ fleet = Some(snapshot);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ // Keepalive safety net (missed watcher events): full collect only
|
|
|
+ // while uncollected (startup); afterwards only the cheap
|
|
|
+ // cross-check — and only when a run is open.
|
|
|
+ _ = tokio::time::sleep(keepalive) => {
|
|
|
+ if rows.is_empty() {
|
|
|
+ rows = collect(&cfg, archived);
|
|
|
+ cross_check(&cfg, &mut rows, &cluster_nodes, &mut fleet).await;
|
|
|
+ } else if has_open_run(&rows) {
|
|
|
+ cross_check(&cfg, &mut rows, &cluster_nodes, &mut fleet).await;
|
|
|
+ } else {
|
|
|
+ continue; // nothing to do — zero work, zero allocation
|
|
|
+ }
|
|
|
+ let json = serde_json::to_string(&rows).unwrap_or_default();
|
|
|
+ if json != last {
|
|
|
+ last = json;
|
|
|
+ let _ = tx.send(last.clone());
|
|
|
+ }
|
|
|
+ }
|
|
|
}
|
|
|
- // debounce: let the burst settle, then drain the remaining events
|
|
|
- tokio::time::sleep(DEBOUNCE).await;
|
|
|
- while ev_rx.try_recv().is_ok() {}
|
|
|
- let mut rows = collect(&cfg, archived);
|
|
|
- // Live cross-check: an open ledger run with no running container
|
|
|
- // anywhere (all its nodes reachable) is marked stale — otherwise a
|
|
|
- // killed serve or a node that died before `stop` could verify would
|
|
|
- // show as "up" forever. No open run with a cluster → no probe.
|
|
|
- if rows
|
|
|
- .iter()
|
|
|
- .any(|r| r.up.is_some() && r.up_cluster.is_some())
|
|
|
- {
|
|
|
- let fleet = crate::web::probe::probe(&cfg).await;
|
|
|
- let mut cluster_nodes: HashMap<String, Vec<String>> = HashMap::new();
|
|
|
- for c in &cfg.clusters {
|
|
|
- cluster_nodes.insert(
|
|
|
- c.name.clone(),
|
|
|
- c.nodes.iter().map(|n| n.name.clone()).collect(),
|
|
|
- );
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/// Reconcile the open-ledger rows against the live fleet: prefer the main
|
|
|
+/// dashboard's last fleet snapshot (no extra SSH); probe directly only when
|
|
|
+/// no snapshot has arrived yet (CLI `report --serve`, first 30 s of web).
|
|
|
+async fn cross_check(
|
|
|
+ cfg: &Config,
|
|
|
+ rows: &mut [crate::report::ReportRow],
|
|
|
+ cluster_nodes: &HashMap<String, Vec<String>>,
|
|
|
+ fleet: &mut Option<crate::web::events::Snapshot>,
|
|
|
+) {
|
|
|
+ let (reachable, live) = match fleet {
|
|
|
+ Some(snap) => {
|
|
|
+ let mut reachable: HashMap<String, bool> = HashMap::new();
|
|
|
+ let mut live: std::collections::HashSet<String> = std::collections::HashSet::new();
|
|
|
+ for c in &snap.clusters {
|
|
|
+ for n in &c.nodes {
|
|
|
+ reachable.insert(n.name.clone(), n.reachable);
|
|
|
+ for ct in &n.containers {
|
|
|
+ live.insert(ct.recipe.clone());
|
|
|
+ }
|
|
|
+ }
|
|
|
}
|
|
|
+ (reachable, live)
|
|
|
+ }
|
|
|
+ None => {
|
|
|
+ let f = crate::web::probe::probe(cfg).await;
|
|
|
let mut reachable: HashMap<String, bool> = HashMap::new();
|
|
|
let mut live: std::collections::HashSet<String> = std::collections::HashSet::new();
|
|
|
- for (node, st) in &fleet.nodes {
|
|
|
+ for (node, st) in &f.nodes {
|
|
|
reachable.insert(node.clone(), st.reachable);
|
|
|
for c in &st.containers {
|
|
|
live.insert(c.recipe.clone());
|
|
|
}
|
|
|
}
|
|
|
- crate::report::stale::mark_stale(&mut rows, &cluster_nodes, &reachable, &live);
|
|
|
+ (reachable, live)
|
|
|
}
|
|
|
- let json = serde_json::to_string(&rows).unwrap_or_default();
|
|
|
- if json != last {
|
|
|
- last = json;
|
|
|
- let _ = tx.send(last.clone());
|
|
|
- }
|
|
|
- }
|
|
|
+ };
|
|
|
+ crate::report::stale::mark_stale(rows, cluster_nodes, &reachable, &live);
|
|
|
}
|
|
|
|
|
|
/// `GET /events` — the latest rows JSON on connect, then one message per
|
|
|
@@ -389,9 +457,16 @@ pub(crate) fn router(
|
|
|
initial: String,
|
|
|
archived: bool,
|
|
|
base: &str,
|
|
|
+ fleet_rx: Option<tokio::sync::broadcast::Receiver<crate::web::events::Event>>,
|
|
|
) -> Router {
|
|
|
let (tx, rx) = tokio::sync::watch::channel(initial);
|
|
|
- tokio::spawn(refresh_loop(config.clone(), tx, KEEPALIVE, archived));
|
|
|
+ tokio::spawn(refresh_loop(
|
|
|
+ config.clone(),
|
|
|
+ tx,
|
|
|
+ KEEPALIVE,
|
|
|
+ archived,
|
|
|
+ fleet_rx,
|
|
|
+ ));
|
|
|
Router::new()
|
|
|
.route(
|
|
|
"/",
|
