From 9f98925c14999fa19432b55c8be0c66b58edf489 Mon Sep 17 00:00:00 2001 From: damocles Date: Fri, 10 Jul 2026 13:39:11 +0200 Subject: [PATCH] feat(#2289): push-based server-warning registry with RAII guard --- hive-c0re/src/lib.rs | 2 +- hive-c0re/src/main.rs | 25 ++- hive-c0re/src/stats/host_stats.rs | 109 +++++++++---- hive-c0re/src/stats/mod.rs | 1 + hive-c0re/src/stats/warnings.rs | 258 ++++++++++++++++++++++++++++++ 5 files changed, 361 insertions(+), 34 deletions(-) create mode 100644 hive-c0re/src/stats/warnings.rs diff --git a/hive-c0re/src/lib.rs b/hive-c0re/src/lib.rs index c2156ed4..e75d197e 100644 --- a/hive-c0re/src/lib.rs +++ b/hive-c0re/src/lib.rs @@ -45,7 +45,7 @@ pub mod workers; // Root re-exports: keep every pre-grouping `crate::` / // `hive_c0re::` path compiling without touching consumers. pub use agent_config::{capabilities, limits, tool_groups, topology}; -pub use stats::{container_stats, hive_stats, host_stats}; +pub use stats::{container_stats, hive_stats, host_stats, warnings}; pub use stores::{ approvals, audit_log, broker, build_logs, db, operator_questions, power, scheduled_prompts, }; diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 66741bbe..c2d4e180 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -13,8 +13,8 @@ use hive_sh4re::{HostRequest, HostResponse}; use hive_c0re::coordinator::{Coordinator, HiveEnv, ServeConfig}; use hive_c0re::{ agent_sockets, auto_update, broker, client, crash_watch, dashboard, dashboard_events, forge, - job_queue, knowledge, matrix, mcp_sockets, migrate, reminder_scheduler, - scheduled_prompts_worker, server, socket_server, + host_stats, job_queue, knowledge, matrix, mcp_sockets, migrate, reminder_scheduler, + scheduled_prompts_worker, server, socket_server, warnings, }; #[derive(Parser)] @@ -340,6 +340,27 @@ async fn cmd_serve( } } }); + // Disk-pressure watch: raise a `disk_pressure` banner warning while the + // host nix store is over threshold and clear it when back under, via the + // push-based warnings registry. Replaces the old per-`/api/state` + // `statvfs`. ~60s cadence; first tick fires immediately. The held guard + // lives for the task's lifetime — dropping it (on shutdown) clears the + // banner. + let mut disk_shutdown = coord.shutdown_rx(); + tokio::spawn(async move { + let mut disk_guard: Option = None; + let interval = std::time::Duration::from_mins(1); + loop { + host_stats::refresh_disk_warning(&mut disk_guard); + tokio::select! { + () = tokio::time::sleep(interval) => {} + _ = disk_shutdown.changed() => { + tracing::info!("disk-pressure watch: shutdown signal received"); + break; + } + } + } + }); // Matrix user sweep: same shape — ensure every container has // an account on the local matrix-tuwunel homeserver with an // access_token persisted to `/matrix-token`. No-op when diff --git a/hive-c0re/src/stats/host_stats.rs b/hive-c0re/src/stats/host_stats.rs index 4e16da55..0e8bcaf5 100644 --- a/hive-c0re/src/stats/host_stats.rs +++ b/hive-c0re/src/stats/host_stats.rs @@ -8,10 +8,12 @@ //! operator *before* an ENOSPC, not after. //! //! [`server_warnings`] is the public surface: it returns a flat list of -//! [`ServerWarning`]s for `/api/state`. The dashboard renders whatever it -//! returns, coloured by `level`, so adding a new system warning (memory -//! pressure, a failed unit, …) is a backend-only change — no frontend -//! edit. Keep producers cheap; this runs on every `/api/state` assembly. +//! [`ServerWarning`]s for `/api/state`. It is now just a snapshot of the +//! push-based [`crate::warnings`] registry — producers raise/clear their +//! own warnings via an RAII guard rather than being re-probed on every +//! render. The disk-pressure check is reworked onto that model here: +//! [`refresh_disk_warning`] drives a held guard from a periodic task (see +//! `spawn` in `main.rs`) instead of a `statvfs` per `/api/state`. use std::collections::HashMap; @@ -37,36 +39,62 @@ pub struct ServerWarning { const DISK_WARN_PCT: f64 = 85.0; const DISK_CRIT_PCT: f64 = 95.0; -/// Collect the current server-level warnings for the dashboard banner. -/// Each producer pushes zero or more [`ServerWarning`]s; the frontend -/// renders whatever this returns. Cheap to call on every `/api/state` -/// assembly (currently a single `statvfs`). +/// Snapshot the current server-level warnings for the dashboard banner — +/// a cheap read of the push-based [`crate::warnings`] registry. Producers +/// (the disk watch below, background converge sweeps, …) raise and clear +/// their own entries via [`crate::warnings::set_warning`]; nothing is +/// re-probed here. #[must_use] pub fn server_warnings() -> Vec { - let mut out = Vec::new(); - if let Some(d) = nix_disk_usage() - && d.used_pct >= DISK_WARN_PCT - { - #[allow( - clippy::cast_precision_loss, - reason = "byte counts stay well under f64's 2^53 exact-integer range, so this GiB conversion loses no precision" - )] - let free_gib = d.free_bytes as f64 / (1024.0 * 1024.0 * 1024.0); - out.push(ServerWarning { - kind: "disk_pressure", - level: if d.used_pct >= DISK_CRIT_PCT { - "crit" - } else { - "warn" - }, - message: format!( - "host nix store {:.0}% full ({free_gib:.1} GiB free) \ - — garbage-collect the store before it runs out of space", - d.used_pct - ), - }); + crate::warnings::snapshot() +} + +/// Pure disk-pressure decision: given usage, return the banner +/// `(level, message)` or `None` when below the warn threshold. Split from +/// the probe + guard plumbing so the threshold + formatting is unit-tested +/// without a real `statvfs`. +fn disk_pressure_warning(used_pct: f64, free_bytes: u64) -> Option<(&'static str, String)> { + if used_pct < DISK_WARN_PCT { + return None; + } + #[allow( + clippy::cast_precision_loss, + reason = "byte counts stay well under f64's 2^53 exact-integer range, so this GiB conversion loses no precision" + )] + let free_gib = free_bytes as f64 / (1024.0 * 1024.0 * 1024.0); + let level = if used_pct >= DISK_CRIT_PCT { + "crit" + } else { + "warn" + }; + let message = format!( + "host nix store {used_pct:.0}% full ({free_gib:.1} GiB free) \ + — garbage-collect the store before it runs out of space" + ); + Some((level, message)) +} + +/// One tick of the disk-pressure watch: probe the store and reconcile the +/// held warning `guard` — raise or refresh it while usage is over +/// threshold, drop it (clearing the banner) once back under. Driven by the +/// periodic disk-watch task spawned in `main.rs`, replacing the old +/// per-render `statvfs` in [`server_warnings`]. +pub fn refresh_disk_warning(guard: &mut Option) { + let decision = nix_disk_usage().and_then(|d| disk_pressure_warning(d.used_pct, d.free_bytes)); + match decision { + Some((level, message)) => match guard { + Some(g) => g.update(level, message), + None => { + *guard = Some(crate::warnings::set_warning( + "disk_pressure", + level, + message, + )); + } + }, + // Back under threshold (or probe failed) — drop the guard to clear. + None => *guard = None, } - out } /// Agent-state warnings derived from the live container snapshot the @@ -194,6 +222,25 @@ fn disk_usage(path: &str) -> Option { mod tests { use super::*; + #[test] + fn disk_below_threshold_is_none() { + assert!(disk_pressure_warning(84.9, 100 << 30).is_none()); + } + + #[test] + fn disk_warn_band_is_amber() { + let (level, msg) = disk_pressure_warning(90.0, 20 << 30).expect("warn"); + assert_eq!(level, "warn"); + assert!(msg.contains("90% full")); + assert!(msg.contains("20.0 GiB free")); + } + + #[test] + fn disk_crit_band_is_red() { + let (level, _) = disk_pressure_warning(95.0, 1 << 30).expect("crit"); + assert_eq!(level, "crit"); + } + fn cv(name: &str, needs_login: bool) -> ContainerView { ContainerView { name: name.to_owned(), diff --git a/hive-c0re/src/stats/mod.rs b/hive-c0re/src/stats/mod.rs index ee528cdc..37ea752c 100644 --- a/hive-c0re/src/stats/mod.rs +++ b/hive-c0re/src/stats/mod.rs @@ -6,3 +6,4 @@ pub mod container_stats; pub mod hive_stats; pub mod host_stats; +pub mod warnings; diff --git a/hive-c0re/src/stats/warnings.rs b/hive-c0re/src/stats/warnings.rs new file mode 100644 index 00000000..d91345db --- /dev/null +++ b/hive-c0re/src/stats/warnings.rs @@ -0,0 +1,258 @@ +//! Process-global server-warning registry backing the dashboard banner. +//! +//! The dashboard shows a top-of-page banner for hive-level degradation +//! (disk pressure, a failed converge sweep, agents that can't log in, …). +//! Historically each warning was *pulled*: [`host_stats::server_warnings`] +//! recomputed the whole list on every `/api/state` by re-probing (a +//! `statvfs`, a scan of the container snapshot). Background sweeps that +//! failed had nowhere to record it, so they just `warn!`'d to the journal +//! and vanished from anywhere an operator looks. +//! +//! This module replaces the pull model with a **push** one: a single +//! `Mutex>` any subsystem writes into. [`snapshot`] is a +//! cheap read of that map. The default API is RAII: [`set_warning`] returns +//! a [`WarningGuard`] whose `Drop` removes the key, so a producer holds the +//! guard exactly as long as the condition is true and lets it drop to clear +//! — no manual dismiss, no stale banner. A failing periodic sweep does +//! `let _w = set_warning(...)` in its failure branch and drops it on the +//! next success. +//! +//! Concurrency: all mutation is behind the one mutex. Each guard carries a +//! unique id and only touches its own map entry (`Drop` / [`update`] +//! no-op once a later `set_warning` for the same kind has superseded it), +//! so a re-`set_warning` for the same kind is never clobbered when the +//! stale guard later drops. +//! +//! [`host_stats::server_warnings`]: crate::host_stats::server_warnings +//! [`update`]: WarningGuard::update + +use std::collections::BTreeMap; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Mutex, OnceLock}; + +use crate::host_stats::ServerWarning; + +/// Monotonic guard-id source. Each [`set_warning`] mints a fresh id so a +/// guard's `Drop` can tell "my entry" from "an entry a later guard replaced +/// under the same kind". +static NEXT_ID: AtomicU64 = AtomicU64::new(1); + +/// One registry slot: the warning payload plus the id of the guard that +/// currently owns the `kind`. +struct Slot { + level: &'static str, + message: String, + owner: u64, +} + +/// The one process-global registry. `BTreeMap` (not `HashMap`) so +/// [`snapshot`] yields a deterministic, kind-sorted banner order across +/// renders. +fn registry() -> &'static Mutex> { + static REG: OnceLock>> = OnceLock::new(); + REG.get_or_init(|| Mutex::new(BTreeMap::new())) +} + +/// Lock the registry, recovering the guard even if a previous holder +/// panicked mid-mutation — a poisoned lock must not wedge every future +/// warning update. +fn lock() -> std::sync::MutexGuard<'static, BTreeMap<&'static str, Slot>> { + registry() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +/// Raise (or replace) the warning under `kind` and return a guard that +/// clears it on drop. This is the default way to surface a hive-level +/// degradation: hold the returned guard while the condition is true, and +/// let it drop (or call `drop(guard)`) the moment it clears. +/// +/// Re-calling `set_warning` for a `kind` that already has a live guard +/// replaces the payload and takes ownership of the entry (fresh id): the +/// superseded guard's later `Drop`/`update` become no-ops, so a producer +/// can freely re-arm without racing its own previous guard. +#[must_use = "dropping the guard immediately clears the warning; bind it for as long as the condition holds"] +pub fn set_warning( + kind: &'static str, + level: &'static str, + message: impl Into, +) -> WarningGuard { + let id = NEXT_ID.fetch_add(1, Ordering::Relaxed); + lock().insert( + kind, + Slot { + level, + message: message.into(), + owner: id, + }, + ); + WarningGuard { kind, id } +} + +/// Snapshot of the currently-active warnings, kind-sorted. Cheap read +/// behind the registry mutex — safe to call on every `/api/state`. +#[must_use] +pub fn snapshot() -> Vec { + lock() + .iter() + .map(|(kind, slot)| ServerWarning { + kind, + level: slot.level, + message: slot.message.clone(), + }) + .collect() +} + +/// RAII handle for one active warning. Drop clears the `kind` from the +/// banner. Obtained from [`set_warning`]. +#[must_use = "dropping the guard immediately clears the warning; bind it for as long as the condition holds"] +pub struct WarningGuard { + kind: &'static str, + id: u64, +} + +impl WarningGuard { + /// Refresh the held warning's `level` + `message` in place — e.g. a + /// disk-usage percentage ticking up, or a `warn` escalating to `crit`. + /// A no-op once a later [`set_warning`] for the same kind has superseded + /// this guard (its entry is owned by the newer id). + pub fn update(&self, level: &'static str, message: impl Into) { + let mut reg = lock(); + if let Some(slot) = reg.get_mut(self.kind) + && slot.owner == self.id + { + slot.level = level; + slot.message = message.into(); + } + } +} + +impl Drop for WarningGuard { + fn drop(&mut self) { + let mut reg = lock(); + // Only remove if we still own the slot — a later `set_warning` for + // the same kind takes over the entry (fresh owner id), so this + // superseded guard's drop must not clobber the live one. + let ours = reg.get(self.kind).is_some_and(|slot| slot.owner == self.id); + if ours { + reg.remove(self.kind); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Find the snapshot entry for `kind`, if present. Tests filter by + /// their own unique kinds instead of asserting global emptiness, so + /// they stay correct under cargo's parallel test runner (the registry + /// is process-global and shared across tests). + fn find(kind: &str) -> Option { + snapshot().into_iter().find(|w| w.kind == kind) + } + + #[test] + fn set_surfaces_in_snapshot_then_drop_clears() { + { + let _g = set_warning("t_setdrop", "warn", "boom"); + let w = find("t_setdrop").expect("warning present while guard held"); + assert_eq!(w.level, "warn"); + assert_eq!(w.message, "boom"); + } + assert!(find("t_setdrop").is_none(), "dropping the guard clears it"); + } + + #[test] + fn update_refreshes_level_and_message_in_place() { + let g = set_warning("t_update", "warn", "85% full"); + g.update("crit", "96% full"); + let w = find("t_update").expect("present"); + assert_eq!(w.level, "crit"); + assert_eq!(w.message, "96% full"); + drop(g); + assert!(find("t_update").is_none()); + } + + #[test] + fn resetting_same_kind_survives_stale_guard_drop() { + let a = set_warning("t_reset", "warn", "first"); + let b = set_warning("t_reset", "crit", "second"); + // `b` took over the entry; dropping the superseded `a` must not + // remove `b`'s live warning. + drop(a); + let w = find("t_reset").expect("b's warning survives a's drop"); + assert_eq!(w.level, "crit"); + assert_eq!(w.message, "second"); + // A stale guard's update is also a no-op — prove by dropping b and + // confirming the kind is gone (only the owner clears it). + drop(b); + assert!(find("t_reset").is_none()); + } + + #[test] + fn stale_guard_update_is_noop() { + let a = set_warning("t_staleupd", "warn", "old"); + let b = set_warning("t_staleupd", "warn", "new"); + a.update("crit", "should not apply"); + let w = find("t_staleupd").expect("present"); + assert_eq!(w.message, "new", "stale guard update ignored"); + drop(b); + drop(a); + assert!(find("t_staleupd").is_none()); + } + + #[test] + fn distinct_kinds_coexist_and_snapshot_is_sorted() { + let _z = set_warning("t_zzz", "warn", "z"); + let _a = set_warning("t_aaa", "warn", "a"); + let kinds: Vec<&str> = snapshot() + .iter() + .map(|w| w.kind) + .filter(|k| k.starts_with("t_zzz") || k.starts_with("t_aaa")) + .collect(); + assert_eq!(kinds, ["t_aaa", "t_zzz"], "kind-sorted order"); + } + + #[test] + fn concurrent_unique_kinds_never_race_and_clear_out() { + // Each thread owns a distinct kind and hammers set/update/drop in a + // loop. The shared resource under test is the one mutex; a data + // race would trip under the thread sanitizer / show a wrong final + // state. After every guard has dropped, none of the kinds remain. + const KINDS: [&str; 6] = ["cc0", "cc1", "cc2", "cc3", "cc4", "cc5"]; + std::thread::scope(|s| { + for kind in KINDS { + s.spawn(move || { + for i in 0..500 { + let g = set_warning(kind, "warn", format!("iter {i}")); + g.update("crit", format!("iter {i} upd")); + drop(g); + } + }); + } + }); + for kind in KINDS { + assert!(find(kind).is_none(), "{kind} cleared after all drops"); + } + } + + #[test] + fn concurrent_same_kind_resolves_to_absent_after_all_drop() { + // Many threads contend on ONE kind, each holding its own guard + // briefly. The id-guard invariant means whichever set ran last owns + // the entry, and once every guard has dropped the kind is absent — + // no stale guard leaves a phantom warning behind. + std::thread::scope(|s| { + for t in 0..8 { + s.spawn(move || { + for i in 0..300 { + let g = set_warning("cc_shared", "warn", format!("{t}:{i}")); + drop(g); + } + }); + } + }); + assert!(find("cc_shared").is_none()); + } +}