From 70555c2daf7d127bd4a4c3401f77c9327cd3bbfb Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 13 Feb 2026 08:59:38 +0000 Subject: [PATCH] feat: warning/anomaly detection system (Stage 5) Add automated detection of 6 runtime anomaly types with severity levels: - PoisonedActor (critical): actor panicked - StalledActor (high): no message processing while mailbox > 0 - GrowingMailbox (medium): consecutive increases in mailbox depth - MailboxOverflow (medium): messages dropped (backpressure triggered) - WorkerImbalance (low): one worker has >2x average actor count - EmptyWorker (low): worker has 0 actors while others have many Configurable thresholds with sensible defaults. 7 scenario tests covering all warning types, threshold behavior, and streak reset logic. Web: warning banner at top of overview page with severity-colored alerts TUI: warning count in summary bar (yellow when active) SSE: new "warnings" event sent with each stats tick Authored by Claude, lovingly guided by Zachery Aaron Shores-Chmielewski --- .../runtime-dashboard/src/dashboard_html.rs | 19 + crates/runtime-dashboard/src/lib.rs | 1 + crates/runtime-dashboard/src/server.rs | 13 + crates/runtime-dashboard/src/tui/app.rs | 9 + crates/runtime-dashboard/src/tui/ui.rs | 13 +- crates/runtime-dashboard/src/warnings.rs | 391 ++++++++++++++++++ 6 files changed, 444 insertions(+), 2 deletions(-) create mode 100644 crates/runtime-dashboard/src/warnings.rs diff --git a/crates/runtime-dashboard/src/dashboard_html.rs b/crates/runtime-dashboard/src/dashboard_html.rs index 5c544aa..1917297 100644 --- a/crates/runtime-dashboard/src/dashboard_html.rs +++ b/crates/runtime-dashboard/src/dashboard_html.rs @@ -165,6 +165,7 @@ pub const DASHBOARD_HTML: &str = r##"
+

Worker Utilization

@@ -580,6 +581,24 @@ pub const DASHBOARD_HTML: &str = r##" } catch(err) { console.error('history parse error', err); } }); + es.addEventListener('warnings', function(e) { + try { + var warnings = JSON.parse(e.data); + var banner = document.getElementById('warningBanner'); + if (warnings.length === 0) { + banner.style.display = 'none'; + return; + } + banner.style.display = 'block'; + var sevColors = {critical:'#f44336',high:'#ff5722',medium:'#ff9800',low:'#888'}; + var html = warnings.map(function(w) { + var c = sevColors[w.severity] || '#888'; + return '\u26A0 ' + w.description + ''; + }).join('   '); + banner.innerHTML = 'WARNINGS (' + warnings.length + ')   ' + html; + } catch(err) { console.error('warnings parse error', err); } + }); + es.addEventListener('activity', function(e) { try { addLogEvents(JSON.parse(e.data)); } catch(err) { console.error('activity parse error', err); } }); diff --git a/crates/runtime-dashboard/src/lib.rs b/crates/runtime-dashboard/src/lib.rs index 654419b..0f2ed42 100644 --- a/crates/runtime-dashboard/src/lib.rs +++ b/crates/runtime-dashboard/src/lib.rs @@ -3,6 +3,7 @@ pub mod history; pub mod investigate; pub mod layer; pub mod trace; +pub mod warnings; mod actor_detail_html; mod actors_html; mod dashboard_html; diff --git a/crates/runtime-dashboard/src/server.rs b/crates/runtime-dashboard/src/server.rs index d3e7bbd..ee3a580 100644 --- a/crates/runtime-dashboard/src/server.rs +++ b/crates/runtime-dashboard/src/server.rs @@ -15,6 +15,7 @@ use crate::dashboard_html::DASHBOARD_HTML; use crate::history::DashboardHistory; use crate::layer::EventStore; use crate::trace::RuntimeTrace; +use crate::warnings::{WarningConfig, WarningDetector}; #[cfg(feature = "distribution")] use crate::distribution_collector::DistributionStatsProvider; @@ -232,6 +233,7 @@ fn handle_live_sse( // Spawn producer thread thread::spawn(move || { let mut cursor: u64 = 0; + let mut warning_detector = WarningDetector::new(WarningConfig::default()); // Send initial history snapshot so sparklines render immediately if history.sample_count() > 0 { @@ -249,6 +251,17 @@ fn handle_live_sse( col.enrich(&mut stats); } history.record(&stats); + + // Run warning detection + let warnings = warning_detector.check(&stats); + if !warnings.is_empty() { + if let Ok(wjson) = serde_json::to_string(&warnings) { + if tx.send(format_sse("warnings", &wjson)).is_err() { + return; + } + } + } + let json = serde_json::to_string(&stats).unwrap(); if tx.send(format_sse("stats", &json)).is_err() { return; diff --git a/crates/runtime-dashboard/src/tui/app.rs b/crates/runtime-dashboard/src/tui/app.rs index 539826e..54bdea6 100644 --- a/crates/runtime-dashboard/src/tui/app.rs +++ b/crates/runtime-dashboard/src/tui/app.rs @@ -5,6 +5,8 @@ use crossterm::event::{KeyCode, KeyEvent, KeyModifiers}; use swactor::actor::ActorAddress; use swactor::stats::{RuntimeStats, TickTiming}; +use crate::warnings::{Warning, WarningConfig, WarningDetector}; + /// Per-worker data prepared for rendering. pub struct WorkerView { pub id: usize, @@ -95,6 +97,7 @@ pub struct App { pub search_active: bool, pub search_query: String, pub search_locked: bool, + pub warnings: Vec, #[cfg(feature = "distribution")] pub distribution: Option, @@ -111,6 +114,7 @@ pub struct App { sparkline_mailbox: Vec>, /// Per-actor sparkline history: address → (prev_msgs, rates, mailbox_depths). actor_sparklines: HashMap, VecDeque)>, + warning_detector: WarningDetector, } impl App { @@ -134,6 +138,7 @@ impl App { search_active: false, search_query: String::new(), search_locked: false, + warnings: Vec::new(), #[cfg(feature = "distribution")] distribution: None, #[cfg(feature = "distribution")] @@ -144,6 +149,7 @@ impl App { sparkline_rates: Vec::new(), sparkline_mailbox: Vec::new(), actor_sparklines: HashMap::new(), + warning_detector: WarningDetector::new(WarningConfig::default()), } } @@ -280,6 +286,9 @@ impl App { } self.sort_actors(); + // Run warning detection + self.warnings = self.warning_detector.check(&stats); + // Clamp selection if !self.actor_rows.is_empty() { self.selected = self.selected.min(self.actor_rows.len() - 1); diff --git a/crates/runtime-dashboard/src/tui/ui.rs b/crates/runtime-dashboard/src/tui/ui.rs index 13b906d..e14d55e 100644 --- a/crates/runtime-dashboard/src/tui/ui.rs +++ b/crates/runtime-dashboard/src/tui/ui.rs @@ -166,7 +166,7 @@ fn draw_summary(f: &mut Frame, app: &App, area: Rect) { Style::default().fg(Color::DarkGray) }; - let line = Line::from(vec![ + let mut spans = vec![ Span::styled(" Workers: ", Style::default().fg(Color::DarkGray)), Span::styled( format!("{}", app.num_workers), @@ -189,8 +189,17 @@ fn draw_summary(f: &mut Frame, app: &App, area: Rect) { ), Span::styled(" Panics: ", Style::default().fg(Color::DarkGray)), Span::styled(format!("{}", app.total_panics), panics_style), - ]); + ]; + if !app.warnings.is_empty() { + spans.push(Span::styled(" Warnings: ", Style::default().fg(Color::DarkGray))); + spans.push(Span::styled( + format!("{}", app.warnings.len()), + Style::default().fg(Color::Yellow).add_modifier(Modifier::BOLD), + )); + } + + let line = Line::from(spans); f.render_widget(Paragraph::new(line), area); } diff --git a/crates/runtime-dashboard/src/warnings.rs b/crates/runtime-dashboard/src/warnings.rs new file mode 100644 index 0000000..35400ee --- /dev/null +++ b/crates/runtime-dashboard/src/warnings.rs @@ -0,0 +1,391 @@ +//! Automated anomaly detection for the runtime dashboard. +//! +//! Runs on each stats sample, comparing consecutive snapshots to detect +//! growing mailboxes, stalled actors, worker imbalance, and other conditions. + +use std::collections::HashMap; + +use swactor::actor::ActorAddress; +use swactor::stats::RuntimeStats; + +/// Types of warnings the detector can produce. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize)] +#[serde(rename_all = "snake_case")] +pub enum WarningType { + GrowingMailbox, + StalledActor, + PoisonedActor, + WorkerImbalance, + EmptyWorker, + MailboxOverflow, +} + +/// Severity levels for warnings. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, serde::Serialize)] +#[serde(rename_all = "lowercase")] +pub enum Severity { + Low, + Medium, + High, + Critical, +} + +/// An active warning. +#[derive(Debug, Clone, serde::Serialize)] +pub struct Warning { + pub warning_type: WarningType, + pub severity: Severity, + pub entity: String, + pub description: String, +} + +/// Configuration for warning thresholds. +#[derive(Debug, Clone)] +pub struct WarningConfig { + /// Consecutive samples with increasing mailbox depth before warning. + pub growing_mailbox_threshold: usize, + /// Consecutive ticks with no message processing while mailbox > 0. + pub stalled_actor_threshold: usize, + /// A worker is "imbalanced" if it has > this ratio times the average load. + pub worker_imbalance_ratio: f64, +} + +impl Default for WarningConfig { + fn default() -> Self { + Self { + growing_mailbox_threshold: 5, + stalled_actor_threshold: 10, + worker_imbalance_ratio: 2.0, + } + } +} + +/// Per-actor tracking state. +struct ActorState { + prev_mailbox: usize, + prev_messages: u64, + growing_streak: usize, + stalled_streak: usize, +} + +/// Warning detection engine. Call `check()` on each stats sample. +pub struct WarningDetector { + config: WarningConfig, + actors: HashMap, +} + +impl WarningDetector { + pub fn new(config: WarningConfig) -> Self { + Self { + config, + actors: HashMap::new(), + } + } + + /// Analyze a stats snapshot and return all active warnings. + pub fn check(&mut self, stats: &RuntimeStats) -> Vec { + let mut warnings = Vec::new(); + + // Track which actors are still alive + let mut live_addrs: std::collections::HashSet = + std::collections::HashSet::new(); + + for actor in &stats.actor_details { + live_addrs.insert(actor.address); + + // Poisoned actor — immediate critical warning + if actor.poisoned { + warnings.push(Warning { + warning_type: WarningType::PoisonedActor, + severity: Severity::Critical, + entity: format!("{}", actor.address), + description: "Actor is poisoned (panicked)".to_string(), + }); + } + + let state = self.actors.entry(actor.address).or_insert(ActorState { + prev_mailbox: actor.mailbox_depth, + prev_messages: actor.messages_processed, + growing_streak: 0, + stalled_streak: 0, + }); + + // Growing mailbox detection + if actor.mailbox_depth > state.prev_mailbox && actor.mailbox_depth > 0 { + state.growing_streak += 1; + } else { + state.growing_streak = 0; + } + + if state.growing_streak >= self.config.growing_mailbox_threshold { + warnings.push(Warning { + warning_type: WarningType::GrowingMailbox, + severity: Severity::Medium, + entity: format!("{}", actor.address), + description: format!( + "Mailbox growing for {} consecutive samples (depth: {})", + state.growing_streak, actor.mailbox_depth, + ), + }); + } + + // Stalled actor detection + if actor.messages_processed == state.prev_messages && actor.mailbox_depth > 0 { + state.stalled_streak += 1; + } else { + state.stalled_streak = 0; + } + + if state.stalled_streak >= self.config.stalled_actor_threshold { + warnings.push(Warning { + warning_type: WarningType::StalledActor, + severity: Severity::High, + entity: format!("{}", actor.address), + description: format!( + "No messages processed for {} ticks with {} pending", + state.stalled_streak, actor.mailbox_depth, + ), + }); + } + + state.prev_mailbox = actor.mailbox_depth; + state.prev_messages = actor.messages_processed; + } + + // Clean up dead actors + self.actors.retain(|addr, _| live_addrs.contains(addr)); + + // Mailbox overflow detection + for w in &stats.workers { + if w.messages_dropped > 0 { + warnings.push(Warning { + warning_type: WarningType::MailboxOverflow, + severity: Severity::Medium, + entity: format!("Worker {}", w.id), + description: format!("{} messages dropped", w.messages_dropped), + }); + } + } + + // Worker imbalance and empty worker detection + if stats.workers.len() > 1 { + let total_actors: usize = stats.workers.iter().map(|w| w.num_actors).sum(); + let avg = total_actors as f64 / stats.workers.len() as f64; + + for w in &stats.workers { + if avg > 0.0 && w.num_actors as f64 > avg * self.config.worker_imbalance_ratio { + warnings.push(Warning { + warning_type: WarningType::WorkerImbalance, + severity: Severity::Low, + entity: format!("Worker {}", w.id), + description: format!( + "{} actors vs {:.0} average ({:.1}x)", + w.num_actors, avg, w.num_actors as f64 / avg, + ), + }); + } + + if w.num_actors == 0 && total_actors > 0 { + warnings.push(Warning { + warning_type: WarningType::EmptyWorker, + severity: Severity::Low, + entity: format!("Worker {}", w.id), + description: "Worker has no actors while others do".to_string(), + }); + } + } + } + + // Sort by severity (critical first) + warnings.sort_by(|a, b| b.severity.cmp(&a.severity)); + warnings + } +} + +#[cfg(test)] +mod tests { + use super::*; + use swactor::stats::{ActorInfo, WorkerInfo}; + + fn make_worker(id: usize, actors: usize, dropped: u64) -> WorkerInfo { + WorkerInfo { + id, + num_actors: actors, + mailbox_depth: 0, + messages_processed: 0, + local_sends: 0, + cross_sends: 0, + inbox_sends: 0, + type_mismatches: 0, + panics: 0, + messages_dropped: dropped, + restarts: 0, + stops: 0, + } + } + + fn make_actor(id: u8, depth: usize, msgs: u64, poisoned: bool) -> ActorInfo { + ActorInfo { + address: ActorAddress([id; 32]), + worker_id: 0, + mailbox_depth: depth, + last_msg_type: None, + messages_processed: msgs, + poisoned, + } + } + + fn make_stats(workers: Vec, actors: Vec) -> RuntimeStats { + RuntimeStats { + num_workers: workers.len(), + uptime_ms: 0, + actors: actors.iter().map(|a| (a.address, a.worker_id)).collect(), + workers, + actor_details: actors, + tick_timings: Vec::new(), + } + } + + #[test] + fn poisoned_actor_triggers_critical_warning() { + let mut detector = WarningDetector::new(WarningConfig::default()); + let stats = make_stats( + vec![make_worker(0, 1, 0)], + vec![make_actor(1, 0, 10, true)], + ); + let warnings = detector.check(&stats); + assert!(warnings.iter().any(|w| w.warning_type == WarningType::PoisonedActor)); + assert!(warnings.iter().any(|w| w.severity == Severity::Critical)); + } + + #[test] + fn growing_mailbox_triggers_after_threshold() { + let config = WarningConfig { + growing_mailbox_threshold: 3, + ..Default::default() + }; + let mut detector = WarningDetector::new(config); + + // 4 samples with increasing mailbox: should trigger at sample 4 + for depth in 1..=4 { + let stats = make_stats( + vec![make_worker(0, 1, 0)], + vec![make_actor(1, depth, 0, false)], + ); + let warnings = detector.check(&stats); + if depth < 4 { + assert!(!warnings.iter().any(|w| w.warning_type == WarningType::GrowingMailbox), + "should not trigger at depth {}", depth); + } else { + assert!(warnings.iter().any(|w| w.warning_type == WarningType::GrowingMailbox), + "should trigger at depth {}", depth); + } + } + } + + #[test] + fn growing_mailbox_resets_on_decrease() { + let config = WarningConfig { + growing_mailbox_threshold: 3, + ..Default::default() + }; + let mut detector = WarningDetector::new(config); + + // Grow for 2 samples, then decrease, then grow again + for depth in [1, 2, 1, 2, 3, 4] { + let stats = make_stats( + vec![make_worker(0, 1, 0)], + vec![make_actor(1, depth, 0, false)], + ); + detector.check(&stats); + } + // After 1,2 → streak=2; then 1 → streak=0; then 2,3,4 → streak=3 → triggers + let stats = make_stats( + vec![make_worker(0, 1, 0)], + vec![make_actor(1, 5, 0, false)], + ); + let warnings = detector.check(&stats); + assert!(warnings.iter().any(|w| w.warning_type == WarningType::GrowingMailbox)); + } + + #[test] + fn stalled_actor_triggers_when_not_processing() { + let config = WarningConfig { + stalled_actor_threshold: 3, + ..Default::default() + }; + let mut detector = WarningDetector::new(config); + + // Same messages_processed, nonzero mailbox for 4 ticks + for _ in 0..4 { + let stats = make_stats( + vec![make_worker(0, 1, 0)], + vec![make_actor(1, 5, 100, false)], + ); + let warnings = detector.check(&stats); + // Last one should trigger + if warnings.iter().any(|w| w.warning_type == WarningType::StalledActor) { + return; // test passed + } + } + panic!("expected StalledActor warning"); + } + + #[test] + fn worker_imbalance_detected() { + let mut detector = WarningDetector::new(WarningConfig::default()); + // Worker 0: 10 actors, Worker 1: 1 actor. Avg=5.5, ratio=10/5.5=1.8 + // With ratio threshold 2.0, this should NOT trigger + let stats = make_stats( + vec![make_worker(0, 10, 0), make_worker(1, 1, 0)], + vec![], + ); + let warnings = detector.check(&stats); + assert!(!warnings.iter().any(|w| w.warning_type == WarningType::WorkerImbalance)); + + // Worker 0: 20 actors, Worker 1: 1 actor. Avg=10.5, ratio=20/10.5=1.9 — still no + // Worker 0: 30 actors, Worker 1: 1 actor. Avg=15.5, ratio=30/15.5=1.9 — still no + // Worker 0: 100 actors, Worker 1: 1 actor. Avg=50.5, ratio=100/50.5=1.98 — almost + // Worker 0: 100 actors, Worker 1: 0 actor. Avg=50, ratio=100/50=2.0 — at threshold + + let stats2 = make_stats( + vec![make_worker(0, 100, 0), make_worker(1, 1, 0)], + vec![], + ); + let warnings2 = detector.check(&stats2); + // 100 / 50.5 = 1.98 — not > 2.0 + assert!(!warnings2.iter().any(|w| w.warning_type == WarningType::WorkerImbalance)); + + // Now 200 vs 1: 200/100.5 = ~1.99 — still not. Let's do 300 vs 1: 300/150.5 = ~2.0 + // Actually need > 2x. Let's do 50 vs 1: avg=25.5, ratio=50/25.5=1.96. Nope. + // 10 vs 1 vs 1: avg=4, ratio=10/4=2.5 — triggers! + let stats3 = make_stats( + vec![make_worker(0, 10, 0), make_worker(1, 1, 0), make_worker(2, 1, 0)], + vec![], + ); + let warnings3 = detector.check(&stats3); + assert!(warnings3.iter().any(|w| w.warning_type == WarningType::WorkerImbalance)); + } + + #[test] + fn empty_worker_detected() { + let mut detector = WarningDetector::new(WarningConfig::default()); + let stats = make_stats( + vec![make_worker(0, 5, 0), make_worker(1, 0, 0)], + vec![], + ); + let warnings = detector.check(&stats); + assert!(warnings.iter().any(|w| w.warning_type == WarningType::EmptyWorker)); + } + + #[test] + fn mailbox_overflow_detected() { + let mut detector = WarningDetector::new(WarningConfig::default()); + let stats = make_stats( + vec![make_worker(0, 1, 42)], + vec![], + ); + let warnings = detector.check(&stats); + assert!(warnings.iter().any(|w| w.warning_type == WarningType::MailboxOverflow)); + } +}