diff --git a/crates/runtime-dashboard/src/actor_detail_html.rs b/crates/runtime-dashboard/src/actor_detail_html.rs new file mode 100644 index 0000000..788a54e --- /dev/null +++ b/crates/runtime-dashboard/src/actor_detail_html.rs @@ -0,0 +1,403 @@ +pub const ACTOR_DETAIL_HTML: &str = r##" + + + + +Actor Detail — Swactor Dashboard + + + +
+
+

+ Swactor Runtime Dashboard + +

+ +
+
+ +
+
+ Overview / Actors / — +
+ +
+
+
Address
—
+
Worker
—
+
Status
—
+
+
+
Last Message Type
—
+
+
+ +
+
0
Messages
+
0
Mailbox
+
0
Msg/s
+
—
Worker Load
+
+ +
+

Message Rate

+ +
+ +
+

Mailbox Depth

+ +
+ + + +
+

+ Logs + (0) +
+ + + + + +
+

+
+
+
+ + + + +"##; diff --git a/crates/runtime-dashboard/src/actors_html.rs b/crates/runtime-dashboard/src/actors_html.rs index 7fb6100..7b0f011 100644 --- a/crates/runtime-dashboard/src/actors_html.rs +++ b/crates/runtime-dashboard/src/actors_html.rs @@ -60,7 +60,7 @@ pub const ACTORS_HTML: &str = r##" .panel { background: #161822; border: 1px solid #2a2d3e; border-radius: 6px; - padding: 14px; overflow: hidden; + padding: 14px; overflow: visible; } .panel h2 { font-size: 12px; color: #888; text-transform: uppercase; letter-spacing: 1px; margin-bottom: 10px; } @@ -75,7 +75,7 @@ pub const ACTORS_HTML: &str = r##" .stat-card .value { font-size: 22px; font-weight: 700; color: #fff; } .stat-card .label { font-size: 10px; color: #888; text-transform: uppercase; margin-top: 2px; } - canvas { width: 100%; height: 200px; } + canvas { width: 100%; height: 220px; } .search-wrap { margin-bottom: 10px; display: flex; align-items: center; gap: 12px; } .search-input { @@ -109,7 +109,12 @@ pub const ACTORS_HTML: &str = r##" tr.clickable { cursor: pointer; } tr.clickable:hover { background: #1a1d2c; } - .detail-panel { display: none; } + .detail-panel { + display: none; position: fixed; bottom: 12px; right: 12px; + width: 420px; max-height: 320px; overflow-y: auto; + background: #161822; border: 1px solid #2a2d3e; border-radius: 6px; + padding: 14px; z-index: 100; box-shadow: 0 4px 24px rgba(0,0,0,0.5); + } .detail-panel.visible { display: block; } .detail-header { display: flex; justify-content: space-between; align-items: center; margin-bottom: 10px; } .detail-close { @@ -118,17 +123,17 @@ pub const ACTORS_HTML: &str = r##" } .detail-close:hover { color: #e0e0e0; border-color: #555; } .detail-grid { - display: grid; grid-template-columns: repeat(3, 1fr); gap: 10px; margin-bottom: 12px; + display: grid; grid-template-columns: repeat(2, 1fr); gap: 8px; margin-bottom: 10px; } - .detail-item { background: #1c1f2e; border-radius: 4px; padding: 8px 10px; } + .detail-item { background: #1c1f2e; border-radius: 4px; padding: 6px 8px; } .detail-item .d-label { font-size: 10px; color: #888; text-transform: uppercase; } - .detail-item .d-value { font-size: 14px; font-weight: 700; color: #fff; margin-top: 2px; word-break: break-all; } + .detail-item .d-value { font-size: 13px; font-weight: 700; color: #fff; margin-top: 2px; word-break: break-all; } .poisoned-badge { background: #f44336; color: #fff; font-size: 10px; font-weight: 700; padding: 2px 6px; border-radius: 3px; letter-spacing: 0.5px; } .healthy-badge { color: #4caf50; font-size: 12px; } - canvas.sparkline { width: 100%; height: 60px; } + canvas.sparkline { width: 100%; height: 50px; } ::-webkit-scrollbar { width: 6px; } ::-webkit-scrollbar-track { background: #0f1117; } @@ -187,7 +192,16 @@ pub const ACTORS_HTML: &str = r##"

All Actors

- + + + +
@@ -207,23 +221,25 @@ pub const ACTORS_HTML: &str = r##"
- -
-
-

Actor Detail

- -
-
-
Full Address
-
Worker
-
Mailbox Depth
-
Messages Processed
-
Last Message Type
-
Status
-
-

Mailbox Depth History

- + +
+ + +
+
+

Actor

+
+
+
Address
+
Worker
+
Mailbox
+
Messages
+
Last Type
+
Status
+
+
Mailbox History
+
+ + +"##; diff --git a/crates/runtime-dashboard/src/tui/app.rs b/crates/runtime-dashboard/src/tui/app.rs index 6901a6c..7406d89 100644 --- a/crates/runtime-dashboard/src/tui/app.rs +++ b/crates/runtime-dashboard/src/tui/app.rs @@ -1,9 +1,14 @@ +use std::collections::{HashMap, VecDeque}; +use std::sync::Arc; use std::time::Instant; use crossterm::event::{KeyCode, KeyEvent, KeyModifiers}; use swactor::actor::ActorAddress; use swactor::stats::{RuntimeStats, TickTiming}; +use crate::layer::{DashboardEvent, EventStore}; +use crate::warnings::{Warning, WarningConfig, WarningDetector}; + /// Per-worker data prepared for rendering. pub struct WorkerView { pub id: usize, @@ -15,6 +20,10 @@ pub struct WorkerView { /// Fraction of bar for each phase group: [processing, delivery, spawns, overhead] pub phase_fractions: [f64; 4], pub panics: u64, + /// Recent message rates for sparkline rendering. + pub sparkline_rates: Vec, + /// Recent mailbox depths for sparkline rendering. + pub sparkline_mailbox: Vec, } /// One row in the actor table. @@ -25,6 +34,12 @@ pub struct ActorRow { pub last_msg_type: Option, pub messages_processed: u64, pub poisoned: bool, + /// Per-actor mailbox sparkline (from local ring buffer). + pub sparkline_mailbox: Vec, + /// Per-actor message rate sparkline. + pub sparkline_rates: Vec, + /// Per-message-type counts, sorted descending. + pub message_type_counts: Vec<(String, u64)>, } #[derive(Clone, Copy, PartialEq, Eq)] @@ -62,6 +77,7 @@ impl SortColumn { pub enum ViewMode { Overview, WorkerDetail, + ActorDetail, #[cfg(feature = "distribution")] Distribution, } @@ -79,7 +95,13 @@ pub struct App { pub total_panics: u64, pub num_workers: usize, pub view_mode: ViewMode, + pub prev_view_mode: ViewMode, pub focused_worker: usize, + pub focused_actor: Option, + pub search_active: bool, + pub search_query: String, + pub search_locked: bool, + pub warnings: Vec, #[cfg(feature = "distribution")] pub distribution: Option, @@ -90,6 +112,21 @@ pub struct App { prev_time: Instant, /// Rolling msg rates (smoothed) msg_rates: Vec, + /// Per-worker sparkline history (message rate deltas). + sparkline_rates: Vec>, + /// Per-worker sparkline history (mailbox depths). + sparkline_mailbox: Vec>, + /// Per-actor sparkline history: address → (prev_msgs, rates, mailbox_depths). + actor_sparklines: HashMap, VecDeque)>, + warning_detector: WarningDetector, + /// Event store for fetching per-actor logs. + event_store: Option>, + /// Cached log entries for the focused actor. + pub actor_logs: Vec, + /// Scroll offset for actor log view. + pub log_scroll: usize, + /// Active log level filter (all enabled by default). + pub log_levels: [bool; 5], // ERROR, WARN, INFO, DEBUG, TRACE } impl App { @@ -107,7 +144,13 @@ impl App { total_panics: 0, num_workers: 0, view_mode: ViewMode::Overview, + prev_view_mode: ViewMode::Overview, focused_worker: 0, + focused_actor: None, + search_active: false, + search_query: String::new(), + search_locked: false, + warnings: Vec::new(), #[cfg(feature = "distribution")] distribution: None, #[cfg(feature = "distribution")] @@ -115,9 +158,22 @@ impl App { prev_messages: Vec::new(), prev_time: Instant::now(), msg_rates: Vec::new(), + sparkline_rates: Vec::new(), + sparkline_mailbox: Vec::new(), + actor_sparklines: HashMap::new(), + warning_detector: WarningDetector::new(WarningConfig::default()), + event_store: None, + actor_logs: Vec::new(), + log_scroll: 0, + log_levels: [true; 5], } } + /// Set the event store for per-actor log retrieval. + pub fn set_event_store(&mut self, store: Arc) { + self.event_store = Some(store); + } + #[cfg(feature = "distribution")] pub fn update_distribution(&mut self, snapshot: distribution::snapshot::DistributionNodeSnapshot) { let max = if snapshot.members.is_empty() { 0 } else { snapshot.members.len() - 1 }; @@ -125,6 +181,62 @@ impl App { self.distribution = Some(snapshot); } + /// Get visible actor rows (filtered by search query if active). + pub fn visible_actor_rows(&self) -> Vec<&ActorRow> { + if self.search_query.is_empty() { + self.actor_rows.iter().collect() + } else { + let q = self.search_query.to_lowercase(); + self.actor_rows.iter().filter(|r| { + let addr = format!("{}", r.address).to_lowercase(); + let msg = r.last_msg_type.as_deref().unwrap_or("").to_lowercase(); + let worker = format!("w{}", r.worker_id); + addr.contains(&q) || msg.contains(&q) || worker.contains(&q) + }).collect() + } + } + + /// Get the focused actor's data (for actor detail view). + pub fn focused_actor_row(&self) -> Option<&ActorRow> { + self.focused_actor.as_ref().and_then(|addr| { + self.actor_rows.iter().find(|r| r.address == *addr) + }) + } + + /// Refresh the actor log cache from the event store. + pub fn refresh_actor_logs(&mut self) { + if let (Some(addr), Some(store)) = (&self.focused_actor, &self.event_store) { + let hex = format!("{}", addr); + self.actor_logs = store.read_for_actor(&hex, 200); + } else { + self.actor_logs.clear(); + } + } + + /// Get visible log entries (filtered by level). + pub fn visible_logs(&self) -> Vec<&DashboardEvent> { + self.actor_logs + .iter() + .filter(|e| { + match e.level.as_str() { + "ERROR" => self.log_levels[0], + "WARN" => self.log_levels[1], + "INFO" => self.log_levels[2], + "DEBUG" => self.log_levels[3], + "TRACE" => self.log_levels[4], + _ => true, + } + }) + .collect() + } + + /// Toggle a log level filter (0=ERROR, 1=WARN, 2=INFO, 3=DEBUG, 4=TRACE). + pub fn toggle_log_level(&mut self, idx: usize) { + if idx < 5 { + self.log_levels[idx] = !self.log_levels[idx]; + } + } + /// Actor rows filtered to the focused worker (for worker detail view). pub fn focused_actor_rows(&self) -> Vec<&ActorRow> { self.actor_rows @@ -144,6 +256,8 @@ impl App { if self.prev_messages.len() != stats.workers.len() { self.prev_messages = stats.workers.iter().map(|w| w.messages_processed).collect(); self.msg_rates = vec![0.0; stats.workers.len()]; + self.sparkline_rates.resize_with(stats.workers.len(), VecDeque::new); + self.sparkline_mailbox.resize_with(stats.workers.len(), VecDeque::new); } // Compute per-worker views @@ -163,8 +277,19 @@ impl App { self.msg_rates[i] * 0.6 + rate * 0.4 }; self.msg_rates[i] = smoothed; + + // Track sparkline history + let delta = w.messages_processed.saturating_sub(self.prev_messages[i]); self.prev_messages[i] = w.messages_processed; + let spark_rates = &mut self.sparkline_rates[i]; + if spark_rates.len() >= 60 { spark_rates.pop_front(); } + spark_rates.push_back(delta); + + let spark_mbox = &mut self.sparkline_mailbox[i]; + if spark_mbox.len() >= 60 { spark_mbox.pop_front(); } + spark_mbox.push_back(w.mailbox_depth as u64); + // Load % and phase fractions from tick timings let timings = stats.tick_timings.get(i).map(|v| v.as_slice()).unwrap_or(&[]); let (load_pct, phase_fractions) = compute_load_and_phases(timings); @@ -178,6 +303,8 @@ impl App { load_pct, phase_fractions, panics: w.panics, + sparkline_rates: spark_rates.iter().copied().collect(), + sparkline_mailbox: spark_mbox.iter().copied().collect(), }); } @@ -187,9 +314,20 @@ impl App { self.total_mailbox = stats.workers.iter().map(|w| w.mailbox_depth).sum(); self.total_panics = stats.workers.iter().map(|w| w.panics).sum(); - // Build actor table + // Build actor table with sparkline history self.actor_rows.clear(); for a in &stats.actor_details { + let (prev, rates_buf, mbox_buf) = self.actor_sparklines + .entry(a.address) + .or_insert_with(|| (0, VecDeque::new(), VecDeque::new())); + + let rate_delta = a.messages_processed.saturating_sub(*prev); + *prev = a.messages_processed; + if rates_buf.len() >= 60 { rates_buf.pop_front(); } + rates_buf.push_back(rate_delta); + if mbox_buf.len() >= 60 { mbox_buf.pop_front(); } + mbox_buf.push_back(a.mailbox_depth as u64); + self.actor_rows.push(ActorRow { address: a.address, worker_id: a.worker_id, @@ -197,10 +335,21 @@ impl App { last_msg_type: a.last_msg_type.clone(), messages_processed: a.messages_processed, poisoned: a.poisoned, + sparkline_rates: rates_buf.iter().copied().collect(), + sparkline_mailbox: mbox_buf.iter().copied().collect(), + message_type_counts: a.message_type_counts.clone(), }); } self.sort_actors(); + // Run warning detection + self.warnings = self.warning_detector.check(&stats); + + // Refresh actor logs if in detail view + if self.view_mode == ViewMode::ActorDetail { + self.refresh_actor_logs(); + } + // Clamp selection if !self.actor_rows.is_empty() { self.selected = self.selected.min(self.actor_rows.len() - 1); @@ -246,12 +395,45 @@ impl App { } pub fn handle_key(&mut self, key: KeyEvent) { + // Search mode input handling + if self.search_active { + match key.code { + KeyCode::Esc => { + self.search_active = false; + if !self.search_locked { + self.search_query.clear(); + } + return; + } + KeyCode::Enter => { + self.search_active = false; + self.search_locked = !self.search_query.is_empty(); + return; + } + KeyCode::Backspace => { + self.search_query.pop(); + return; + } + KeyCode::Char(c) => { + self.search_query.push(c); + return; + } + _ => return, + } + } + // Global keys match key.code { KeyCode::Char('q') => { self.should_quit = true; return; } KeyCode::Char('c') if key.modifiers.contains(KeyModifiers::CONTROL) => { self.should_quit = true; return; } + KeyCode::Char('/') => { + self.search_active = true; + self.search_locked = false; + self.search_query.clear(); + return; + } KeyCode::Char('s') => { self.sort_column = self.sort_column.next(); self.sort_actors(); @@ -265,6 +447,7 @@ impl App { KeyCode::Tab => { self.view_mode = match self.view_mode { ViewMode::Overview => ViewMode::WorkerDetail, + ViewMode::ActorDetail => ViewMode::Overview, #[cfg(feature = "distribution")] ViewMode::WorkerDetail => ViewMode::Distribution, #[cfg(not(feature = "distribution"))] @@ -280,13 +463,15 @@ impl App { match self.view_mode { ViewMode::Overview => self.handle_key_overview(key), ViewMode::WorkerDetail => self.handle_key_worker_detail(key), + ViewMode::ActorDetail => self.handle_key_actor_detail(key), #[cfg(feature = "distribution")] ViewMode::Distribution => self.handle_key_distribution(key), } } fn handle_key_overview(&mut self, key: KeyEvent) { - let max = if self.actor_rows.is_empty() { 0 } else { self.actor_rows.len() - 1 }; + let visible = self.visible_actor_rows(); + let max = if visible.is_empty() { 0 } else { visible.len() - 1 }; match key.code { KeyCode::Up | KeyCode::Char('k') => { self.selected = self.selected.saturating_sub(1); @@ -303,11 +488,14 @@ impl App { KeyCode::Home => { self.selected = 0; } KeyCode::End => { self.selected = max; } KeyCode::Enter | KeyCode::Char('l') | KeyCode::Right => { - // Enter worker detail for the selected actor's worker - if let Some(row) = self.actor_rows.get(self.selected) { - self.focused_worker = row.worker_id; + // Enter actor detail for the selected visible actor + if let Some(&row) = visible.get(self.selected) { + self.focused_actor = Some(row.address); + self.prev_view_mode = ViewMode::Overview; + self.view_mode = ViewMode::ActorDetail; + self.log_scroll = 0; + self.refresh_actor_logs(); } - self.view_mode = ViewMode::WorkerDetail; } _ => {} } @@ -337,6 +525,38 @@ impl App { } } + fn handle_key_actor_detail(&mut self, key: KeyEvent) { + let max_scroll = self.visible_logs().len().saturating_sub(1); + match key.code { + KeyCode::Esc | KeyCode::Char('h') | KeyCode::Left => { + self.view_mode = self.prev_view_mode; + self.focused_actor = None; + self.actor_logs.clear(); + self.log_scroll = 0; + } + KeyCode::Down | KeyCode::Char('j') => { + self.log_scroll = (self.log_scroll + 1).min(max_scroll); + } + KeyCode::Up | KeyCode::Char('k') => { + self.log_scroll = self.log_scroll.saturating_sub(1); + } + KeyCode::PageDown => { + self.log_scroll = (self.log_scroll + 20).min(max_scroll); + } + KeyCode::PageUp => { + self.log_scroll = self.log_scroll.saturating_sub(20); + } + KeyCode::Home => { self.log_scroll = 0; } + KeyCode::End => { self.log_scroll = max_scroll; } + KeyCode::Char('1') => self.toggle_log_level(0), + KeyCode::Char('2') => self.toggle_log_level(1), + KeyCode::Char('3') => self.toggle_log_level(2), + KeyCode::Char('4') => self.toggle_log_level(3), + KeyCode::Char('5') => self.toggle_log_level(4), + _ => {} + } + } + #[cfg(feature = "distribution")] fn handle_key_distribution(&mut self, key: KeyEvent) { let max = self diff --git a/crates/runtime-dashboard/src/tui/mod.rs b/crates/runtime-dashboard/src/tui/mod.rs index 7b51709..ceb22a1 100644 --- a/crates/runtime-dashboard/src/tui/mod.rs +++ b/crates/runtime-dashboard/src/tui/mod.rs @@ -18,6 +18,7 @@ use swactor::runtime::Runtime; use crate::collector::StatsCollector; #[cfg(feature = "distribution")] use crate::distribution_collector::DistributionStatsProvider; +use crate::layer::EventStore; use self::app::App; use self::event::{AppEvent, EventLoop}; use self::types::RuntimeEndpoint; @@ -47,6 +48,7 @@ pub fn start_tui( #[cfg(feature = "distribution")] distribution: Option>, config: TuiConfig, + event_store: Option>, ) -> io::Result<()> { // Set up terminal crossterm::terminal::enable_raw_mode()?; @@ -72,6 +74,7 @@ pub fn start_tui( #[cfg(feature = "distribution")] distribution, config, + event_store, ); // Restore terminal @@ -92,8 +95,12 @@ fn run_loop( #[cfg(feature = "distribution")] distribution: Option>, config: TuiConfig, + event_store: Option>, ) -> io::Result<()> { let mut app = App::new(); + if let Some(store) = event_store { + app.set_event_store(store); + } let mut table_state = TableState::default(); let events = EventLoop::new(config.poll_interval_ms); diff --git a/crates/runtime-dashboard/src/tui/ui.rs b/crates/runtime-dashboard/src/tui/ui.rs index 2a51599..db26d13 100644 --- a/crates/runtime-dashboard/src/tui/ui.rs +++ b/crates/runtime-dashboard/src/tui/ui.rs @@ -2,7 +2,7 @@ use ratatui::Frame; use ratatui::layout::{Constraint, Layout, Rect}; use ratatui::style::{Color, Modifier, Style}; use ratatui::text::{Line, Span}; -use ratatui::widgets::{Block, Borders, Cell, Paragraph, Row, Table, TableState}; +use ratatui::widgets::{Block, Borders, Cell, Paragraph, Row, Sparkline, Table, TableState}; use super::app::{App, SortColumn, ViewMode}; @@ -11,6 +11,7 @@ pub fn draw(f: &mut Frame, app: &App, table_state: &mut TableState) { match app.view_mode { ViewMode::Overview => draw_overview(f, app, table_state), ViewMode::WorkerDetail => draw_worker_detail(f, app, table_state), + ViewMode::ActorDetail => draw_actor_detail(f, app), #[cfg(feature = "distribution")] ViewMode::Distribution => draw_distribution(f, app, table_state), } @@ -20,17 +21,23 @@ pub fn draw(f: &mut Frame, app: &App, table_state: &mut TableState) { fn draw_overview(f: &mut Frame, app: &App, table_state: &mut TableState) { let num_workers = app.workers.len().max(1); + let has_sparkline_data = app.workers.iter().any(|w| w.sparkline_rates.len() > 1); + let sparkline_height = if has_sparkline_data { 4u16 } else { 0u16 }; let chunks = Layout::vertical([ Constraint::Length(num_workers as u16), + Constraint::Length(sparkline_height), Constraint::Length(1), Constraint::Fill(1), ]) .split(f.area()); draw_worker_bars(f, app, chunks[0]); - draw_summary(f, app, chunks[1]); - draw_actor_table(f, app, table_state, chunks[2]); + if has_sparkline_data { + draw_worker_sparklines(f, app, chunks[1]); + } + draw_summary(f, app, chunks[2]); + draw_actor_table(f, app, table_state, chunks[3]); } /// Render htop-style worker bars. @@ -45,6 +52,36 @@ fn draw_worker_bars(f: &mut Frame, app: &App, area: Rect) { f.render_widget(paragraph, area); } +/// Render per-worker sparklines showing message rate trends. +fn draw_worker_sparklines(f: &mut Frame, app: &App, area: Rect) { + if app.workers.is_empty() { + return; + } + // Split area horizontally: one sparkline per worker + let constraints: Vec = app + .workers + .iter() + .map(|_| Constraint::Ratio(1, app.workers.len() as u32)) + .collect(); + let cols = Layout::horizontal(constraints).split(area); + + for (i, w) in app.workers.iter().enumerate() { + if let Some(&col_area) = cols.get(i) { + let block = Block::default() + .borders(Borders::NONE) + .title(Span::styled( + format!(" W{} ", w.id), + Style::default().fg(Color::DarkGray), + )); + let sparkline = Sparkline::default() + .block(block) + .data(&w.sparkline_rates) + .style(Style::default().fg(Color::Green)); + f.render_widget(sparkline, col_area); + } + } +} + fn build_worker_line(w: &super::app::WorkerView, total_width: usize) -> Line<'static> { let id_str = format!("{:>3}", w.id); let suffix = format!( @@ -129,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), @@ -152,13 +189,47 @@ 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); } -/// Render the actor table with scrolling and selection. +/// Render the actor table with scrolling, selection, and search. fn draw_actor_table(f: &mut Frame, app: &App, table_state: &mut TableState, area: Rect) { + // Split area: optional search bar + table + let has_search = app.search_active || !app.search_query.is_empty(); + let search_height = if has_search { 1u16 } else { 0 }; + let chunks = Layout::vertical([ + Constraint::Length(search_height), + Constraint::Fill(1), + ]) + .split(area); + + // Draw search bar + if has_search { + let search_style = if app.search_active { + Style::default().fg(Color::Yellow) + } else { + Style::default().fg(Color::DarkGray) + }; + let cursor = if app.search_active { "\u{2588}" } else { "" }; + let prefix = if app.search_locked { " [locked] /" } else { " /" }; + let line = Line::from(vec![ + Span::styled(prefix, Style::default().fg(Color::DarkGray)), + Span::styled(format!("{}{}", app.search_query, cursor), search_style), + ]); + f.render_widget(Paragraph::new(line), chunks[0]); + } + let sort_arrow = if app.sort_desc { " \u{25bc}" } else { " \u{25b2}" }; let columns = [ @@ -181,8 +252,8 @@ fn draw_actor_table(f: &mut Frame, app: &App, table_state: &mut TableState, area }); let header = Row::new(header_cells).height(1); - let rows: Vec = app - .actor_rows + let visible = app.visible_actor_rows(); + let rows: Vec = visible .iter() .map(|a| actor_row_cells(a)) .collect(); @@ -190,7 +261,22 @@ fn draw_actor_table(f: &mut Frame, app: &App, table_state: &mut TableState, area table_state.select(Some(app.selected)); let help_text = - " q: quit \u{2191}\u{2193}: scroll s: sort column r: reverse Tab: worker view Enter: drill in"; + " q: quit /: search \u{2191}\u{2193}: scroll s: sort r: reverse Tab: worker view Enter: detail"; + + let title = if !app.search_query.is_empty() { + format!( + " Actors ({} of {} matching \"{}\") ", + visible.len(), + app.actor_rows.len(), + app.search_query, + ) + } else { + format!( + " Actors (sorted by {}{}) ", + app.sort_column.label(), + sort_arrow + ) + }; let table = Table::new( rows, @@ -206,11 +292,7 @@ fn draw_actor_table(f: &mut Frame, app: &App, table_state: &mut TableState, area .block( Block::default() .borders(Borders::ALL) - .title(format!( - " Actors (sorted by {}{}) ", - app.sort_column.label(), - sort_arrow - )) + .title(title) .title_bottom(Line::from(help_text).centered()), ) .row_highlight_style( @@ -443,6 +525,231 @@ fn draw_focused_actor_table(f: &mut Frame, app: &App, table_state: &mut TableSta f.render_stateful_widget(table, area, table_state); } +// ─── Actor Detail View ────────────────────────────────────────────────────── + +fn draw_actor_detail(f: &mut Frame, app: &App) { + let actor = match app.focused_actor_row() { + Some(a) => a, + None => { + let msg = Paragraph::new(" No actor selected. Press Esc to go back.") + .style(Style::default().fg(Color::DarkGray)); + f.render_widget(msg, f.area()); + return; + } + }; + + let has_types = !actor.message_type_counts.is_empty(); + let type_height = if has_types { + (actor.message_type_counts.len() as u16 + 2).min(10) + } else { + 0 + }; + + let chunks = Layout::vertical([ + Constraint::Length(5), // Info card + Constraint::Length(5), // Rate sparkline + Constraint::Length(5), // Mailbox sparkline + Constraint::Length(type_height), // Type breakdown + Constraint::Length(1), // Help bar + Constraint::Fill(1), // Logs panel + ]) + .split(f.area()); + + // Info card + let addr_str = format!("{}", actor.address); + let status = if actor.poisoned { "POISONED" } else { "Healthy" }; + let status_color = if actor.poisoned { Color::Red } else { Color::Green }; + let msg_type = actor.last_msg_type.as_deref() + .map(|s| short_type_name(Some(s))) + .unwrap_or_else(|| "\u{2014}".to_string()); + + let info_lines = vec![ + Line::from(vec![ + Span::styled(" Address: ", Style::default().fg(Color::DarkGray)), + Span::styled(addr_str, Style::default().fg(Color::White).add_modifier(Modifier::BOLD)), + ]), + Line::from(vec![ + Span::styled(" Worker: ", Style::default().fg(Color::DarkGray)), + Span::styled(format!("W{}", actor.worker_id), Style::default().fg(Color::Cyan)), + Span::styled(" Status: ", Style::default().fg(Color::DarkGray)), + Span::styled(status, Style::default().fg(status_color).add_modifier(Modifier::BOLD)), + ]), + Line::from(vec![ + Span::styled(" Messages: ", Style::default().fg(Color::DarkGray)), + Span::styled( + format_num(actor.messages_processed), + Style::default().fg(Color::Cyan).add_modifier(Modifier::BOLD), + ), + Span::styled(" Mailbox: ", Style::default().fg(Color::DarkGray)), + Span::styled( + format!("{}", actor.mailbox_depth), + Style::default().fg(Color::Blue).add_modifier(Modifier::BOLD), + ), + Span::styled(" Last Msg: ", Style::default().fg(Color::DarkGray)), + Span::styled(msg_type, Style::default().fg(Color::Green)), + ]), + ]; + + let info_block = Block::default() + .borders(Borders::ALL) + .title(" Actor Detail "); + let info = Paragraph::new(info_lines).block(info_block); + f.render_widget(info, chunks[0]); + + // Rate sparkline + let rate_block = Block::default() + .borders(Borders::ALL) + .title(Span::styled(" Msg Rate ", Style::default().fg(Color::Green))); + let rate_sparkline = Sparkline::default() + .block(rate_block) + .data(&actor.sparkline_rates) + .style(Style::default().fg(Color::Green)); + f.render_widget(rate_sparkline, chunks[1]); + + // Mailbox sparkline + let mbox_block = Block::default() + .borders(Borders::ALL) + .title(Span::styled(" Mailbox Depth ", Style::default().fg(Color::Blue))); + let mbox_sparkline = Sparkline::default() + .block(mbox_block) + .data(&actor.sparkline_mailbox) + .style(Style::default().fg(Color::Blue)); + f.render_widget(mbox_sparkline, chunks[2]); + + // Message type breakdown + if has_types { + draw_type_breakdown(f, &actor.message_type_counts, chunks[3]); + } + + // Help bar + let level_names = ["ERR", "WARN", "INFO", "DBG", "TRC"]; + let level_colors = [Color::Red, Color::Yellow, Color::Blue, Color::DarkGray, Color::DarkGray]; + let mut help_spans: Vec = vec![ + Span::styled( + " Esc: back \u{2191}\u{2193}: scroll logs ", + Style::default().fg(Color::DarkGray), + ), + ]; + for (i, &name) in level_names.iter().enumerate() { + let active = app.log_levels[i]; + let color = if active { level_colors[i] } else { Color::DarkGray }; + let style = if active { + Style::default().fg(color).add_modifier(Modifier::BOLD) + } else { + Style::default().fg(color) + }; + help_spans.push(Span::styled(format!("{}:{} ", i + 1, name), style)); + } + f.render_widget(Paragraph::new(Line::from(help_spans)), chunks[4]); + + // Logs panel + draw_actor_logs(f, app, chunks[5]); +} + +fn draw_type_breakdown(f: &mut Frame, types: &[(String, u64)], area: Rect) { + let total: u64 = types.iter().map(|(_, c)| *c).sum(); + let max_count = types.first().map(|(_, c)| *c).unwrap_or(1).max(1); + let inner_height = area.height.saturating_sub(2) as usize; + + let bar_colors = [Color::Green, Color::Blue, Color::Yellow, Color::Magenta, Color::Cyan, Color::Red]; + + let lines: Vec = types + .iter() + .take(inner_height) + .enumerate() + .map(|(i, (name, count))| { + let short = name.rsplit("::").next().unwrap_or(name); + let pct = if total > 0 { *count as f64 / total as f64 * 100.0 } else { 0.0 }; + let bar_width = 20usize; + let filled = ((*count as f64 / max_count as f64) * bar_width as f64).round() as usize; + let color = bar_colors[i % bar_colors.len()]; + + Line::from(vec![ + Span::styled( + format!(" {:>16} ", short), + Style::default().fg(Color::White), + ), + Span::styled( + "\u{2588}".repeat(filled), + Style::default().fg(color), + ), + Span::styled( + " ".repeat(bar_width.saturating_sub(filled)), + Style::default(), + ), + Span::styled( + format!(" {:>8} ", count), + Style::default().fg(Color::Cyan).add_modifier(Modifier::BOLD), + ), + Span::styled( + format!("{:>5.1}%", pct), + Style::default().fg(Color::DarkGray), + ), + ]) + }) + .collect(); + + let block = Block::default() + .borders(Borders::ALL) + .title(format!(" Message Types ({}) ", types.len())); + let paragraph = Paragraph::new(lines).block(block); + f.render_widget(paragraph, area); +} + +fn draw_actor_logs(f: &mut Frame, app: &App, area: Rect) { + let visible = app.visible_logs(); + let log_count = visible.len(); + let inner_height = area.height.saturating_sub(2) as usize; // borders + + // Compute scroll window + let scroll = app.log_scroll.min(log_count.saturating_sub(inner_height)); + + let lines: Vec = visible + .iter() + .skip(scroll) + .take(inner_height) + .map(|e| { + let level_color = match e.level.as_str() { + "ERROR" => Color::Red, + "WARN" => Color::Yellow, + "INFO" => Color::Blue, + "DEBUG" => Color::DarkGray, + "TRACE" => Color::DarkGray, + _ => Color::White, + }; + let ts = { + let secs = e.timestamp_ms / 1000; + let ms = e.timestamp_ms % 1000; + let h = (secs / 3600) % 24; + let m = (secs / 60) % 60; + let s = secs % 60; + format!("{:02}:{:02}:{:02}.{:03}", h, m, s, ms) + }; + Line::from(vec![ + Span::styled( + format!(" {} ", ts), + Style::default().fg(Color::DarkGray), + ), + Span::styled( + format!("{:<5} ", e.level), + Style::default().fg(level_color).add_modifier(Modifier::BOLD), + ), + Span::styled( + e.message.clone(), + Style::default().fg(Color::White), + ), + ]) + }) + .collect(); + + let title = format!(" Logs ({}) ", log_count); + let block = Block::default() + .borders(Borders::ALL) + .title(title); + let paragraph = Paragraph::new(lines).block(block); + f.render_widget(paragraph, area); +} + // ─── Distribution View ────────────────────────────────────────────────────── #[cfg(feature = "distribution")] diff --git a/crates/runtime-dashboard/src/warnings.rs b/crates/runtime-dashboard/src/warnings.rs new file mode 100644 index 0000000..d1756d1 --- /dev/null +++ b/crates/runtime-dashboard/src/warnings.rs @@ -0,0 +1,392 @@ +//! 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, + message_type_counts: Vec::new(), + } + } + + 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)); + } +} diff --git a/docs/development_history/dashboard-improvements-research.md b/docs/development_history/dashboard-improvements-research.md new file mode 100644 index 0000000..d3db015 --- /dev/null +++ b/docs/development_history/dashboard-improvements-research.md @@ -0,0 +1,28 @@ +# Dashboard Improvements — Research Phase + +## Summary +Researched 10 comparable monitoring/dashboard systems to inform swactor's dashboard improvement plan. + +## Systems Analyzed +- **Actor runtimes**: Erlang Observer (GUI/CLI/Web), Akka Insights, Ray Dashboard, Orleans Dashboard +- **Async/runtime tools**: tokio-console, Lunatic +- **Message/infrastructure**: RabbitMQ Management, Consul UI, Nomad UI +- **Web frameworks**: Phoenix LiveDashboard + +## Key Findings +1. **Time-series history** is table-stakes — every system provides it +2. **Actor detail drill-down** is universal (Observer has 6-tab process info, Orleans has grain state inspection) +3. **Search/filter** exists in every system +4. **Warning/anomaly detection** (tokio-console's lint system) is a high-value differentiator +5. **Topology visualization** (Consul golden metrics, Observer supervision tree) is rare but powerful + +## Implementation Plan +8 feature stages defined (see `CLAUDE/notes/feature-stages/`): +1. Time-Series History Infrastructure +2. Actor Detail Drill-Down +3. Search and Filter +4. Per-Worker Utilization Visualization +5. Warning/Anomaly Detection +6. Actor-to-Actor Message Flow Topology +7. Per-Actor Logging +8. Per-Message-Type Breakdown diff --git a/src/stats.rs b/src/stats.rs index 5cabf5b..d899b06 100644 --- a/src/stats.rs +++ b/src/stats.rs @@ -120,6 +120,8 @@ pub struct ActorSnapshot { pub last_msg_type: Option<&'static str>, pub messages_processed: u64, pub poisoned: bool, + /// Per-message-type counts, sorted descending by count. + pub message_type_counts: Vec<(&'static str, u64)>, } /// Observer hook called by workers after productive ticks. @@ -150,6 +152,9 @@ pub struct ActorInfo { /// Whether the actor has panicked and is no longer processing messages. #[cfg_attr(feature = "serde", serde(default))] pub poisoned: bool, + /// Per-message-type counts, sorted descending by count. Top 32 types. + #[cfg_attr(feature = "serde", serde(default))] + pub message_type_counts: Vec<(String, u64)>, } /// Snapshot of overall runtime state. diff --git a/src/worker.rs b/src/worker.rs index 884d247..acaa33b 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -610,6 +610,8 @@ struct ActorSlot { started: bool, last_msg_type: Option<&'static str>, messages_processed: u64, + /// Per-message-type counters (bounded to 32 entries). + msg_type_counts: HashMap<&'static str, u64>, /// Per-actor mailbox capacity. 0 = unbounded. mailbox_capacity: usize, overflow_policy: MailboxOverflow, @@ -645,6 +647,7 @@ impl ActorPool { started: false, last_msg_type: None, messages_processed: 0, + msg_type_counts: HashMap::new(), mailbox_capacity: self.default_mailbox_capacity, overflow_policy: self.default_overflow_policy, }); @@ -698,6 +701,9 @@ impl ActorPool { continue; } + #[cfg(feature = "tracing")] + let _actor_span = tracing::trace_span!("actor.tick", actor_addr = %addr).entered(); + let ctx = Ctx::new(inner, addr); // Call on_start once, before first message @@ -762,6 +768,10 @@ impl ActorPool { Ok(Some(type_name)) => { slot.last_msg_type = Some(type_name); slot.messages_processed += 1; + // Track per-type counts (bounded to 32 distinct types) + if slot.msg_type_counts.len() < 32 || slot.msg_type_counts.contains_key(type_name) { + *slot.msg_type_counts.entry(type_name).or_insert(0) += 1; + } } } count += 1; @@ -834,12 +844,16 @@ impl ActorPool { pub fn mailbox_depths_into(&self, out: &mut Vec) { out.clear(); out.extend(self.actors.iter().map(|(&addr, slot)| { + let mut type_counts: Vec<(&'static str, u64)> = + slot.msg_type_counts.iter().map(|(&k, &v)| (k, v)).collect(); + type_counts.sort_by(|a, b| b.1.cmp(&a.1)); ActorSnapshot { address: addr, mailbox_depth: slot.mailbox.len(), last_msg_type: slot.last_msg_type, messages_processed: slot.messages_processed, poisoned: slot.poisoned, + message_type_counts: type_counts, } })); }