From 991b2121915899285d50953ba962ab112a85329d Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 13 Feb 2026 09:02:07 +0000 Subject: [PATCH] feat: message flow topology visualization (Stage 6) Add worker-level message flow topology with force-directed graph visualization on a dedicated /topology web page. - topology.rs: worker_topology() derives graph from cross_sends/local_sends stats (TopologyNode, TopologyEdge, TopologySnapshot) - /topology page with interactive force-directed graph layout: - Nodes sized by actor count, colored by worker - Edges show local sends (green self-loops) and cross-worker sends (blue) - Edge thickness proportional to message volume - Physics simulation with repulsion, attraction, and gravity - SSE "topology" event emitted every ~1s (every 5th stats tick) - /api/topology REST endpoint for on-demand snapshot Infrastructure ready for future per-actor topology with core instrumentation. Authored by Claude, lovingly guided by Zachery Aaron Shores-Chmielewski --- crates/runtime-dashboard/src/lib.rs | 2 + crates/runtime-dashboard/src/server.rs | 47 +++ crates/runtime-dashboard/src/topology.rs | 82 +++++ crates/runtime-dashboard/src/topology_html.rs | 280 ++++++++++++++++++ 4 files changed, 411 insertions(+) create mode 100644 crates/runtime-dashboard/src/topology.rs create mode 100644 crates/runtime-dashboard/src/topology_html.rs diff --git a/crates/runtime-dashboard/src/lib.rs b/crates/runtime-dashboard/src/lib.rs index 0f2ed42..ac7a1c1 100644 --- a/crates/runtime-dashboard/src/lib.rs +++ b/crates/runtime-dashboard/src/lib.rs @@ -8,6 +8,8 @@ mod actor_detail_html; mod actors_html; mod dashboard_html; mod server; +pub mod topology; +mod topology_html; #[cfg(feature = "tui")] pub mod tui; diff --git a/crates/runtime-dashboard/src/server.rs b/crates/runtime-dashboard/src/server.rs index ee3a580..85cd5c7 100644 --- a/crates/runtime-dashboard/src/server.rs +++ b/crates/runtime-dashboard/src/server.rs @@ -14,6 +14,8 @@ use crate::collector::StatsCollector; use crate::dashboard_html::DASHBOARD_HTML; use crate::history::DashboardHistory; use crate::layer::EventStore; +use crate::topology; +use crate::topology_html::TOPOLOGY_HTML; use crate::trace::RuntimeTrace; use crate::warnings::{WarningConfig, WarningDetector}; @@ -154,6 +156,7 @@ pub(crate) fn spawn_http_server( match path { "/" => respond_html(request, DASHBOARD_HTML, "live"), "/actors" => respond_html(request, ACTORS_HTML, "live"), + "/topology" => respond_html(request, TOPOLOGY_HTML, "live"), #[cfg(feature = "distribution")] "/distribution" => respond_html(request, DISTRIBUTION_HTML, "live"), "/events" => { @@ -178,6 +181,13 @@ pub(crate) fn spawn_http_server( "/api/history" => { handle_history_api(request, Arc::clone(&history)); } + "/api/topology" => { + handle_topology_api( + request, + Arc::clone(&runtime), + Arc::clone(&collector), + ); + } "/api/investigate" => { handle_investigate_api( request, @@ -234,6 +244,7 @@ fn handle_live_sse( thread::spawn(move || { let mut cursor: u64 = 0; let mut warning_detector = WarningDetector::new(WarningConfig::default()); + let mut tick_count: u64 = 0; // Send initial history snapshot so sparklines render immediately if history.sample_count() > 0 { @@ -266,6 +277,17 @@ fn handle_live_sse( if tx.send(format_sse("stats", &json)).is_err() { return; } + + // Send topology every 5th tick (~1/sec) + tick_count += 1; + if tick_count % 5 == 0 { + let topo = topology::worker_topology(&stats); + if let Ok(tjson) = serde_json::to_string(&topo) { + if tx.send(format_sse("topology", &tjson)).is_err() { + return; + } + } + } } } @@ -398,6 +420,31 @@ fn handle_distribution_api( let _ = request.respond(response); } +fn handle_topology_api( + request: tiny_http::Request, + runtime: Arc>>>, + collector: Arc>>>, +) { + let maybe_rt = runtime.lock().unwrap().clone(); + let json = match maybe_rt { + Some(rt) => { + let mut stats = rt.stats(); + if let Some(col) = collector.lock().unwrap().as_ref() { + col.enrich(&mut stats); + } + let topo = topology::worker_topology(&stats); + serde_json::to_string(&topo).unwrap_or_else(|_| "{}".into()) + } + None => "{}".to_string(), + }; + let response = tiny_http::Response::from_string(json).with_header( + "Content-Type: application/json" + .parse::() + .unwrap(), + ); + let _ = request.respond(response); +} + fn handle_history_api(request: tiny_http::Request, history: Arc) { let json = history.worker_history_json(); let response = tiny_http::Response::from_string(json).with_header( diff --git a/crates/runtime-dashboard/src/topology.rs b/crates/runtime-dashboard/src/topology.rs new file mode 100644 index 0000000..c9d78e5 --- /dev/null +++ b/crates/runtime-dashboard/src/topology.rs @@ -0,0 +1,82 @@ +//! Actor-to-actor (and worker-to-worker) message flow topology. +//! +//! Currently derives topology from per-worker cross_sends/local_sends stats. +//! Future: sample-based per-actor source→destination tracking with core instrumentation. + +use swactor::stats::RuntimeStats; + +/// An edge in the topology graph. +#[derive(Debug, Clone, serde::Serialize)] +pub struct TopologyEdge { + pub source: String, + pub target: String, + pub weight: u64, + pub label: String, +} + +/// A node in the topology graph. +#[derive(Debug, Clone, serde::Serialize)] +pub struct TopologyNode { + pub id: String, + pub label: String, + pub actor_count: usize, + pub group: usize, +} + +/// A snapshot of the current topology. +#[derive(Debug, Clone, serde::Serialize)] +pub struct TopologySnapshot { + pub nodes: Vec, + pub edges: Vec, +} + +/// Build a worker-level topology from RuntimeStats. +/// +/// Workers are nodes, edges represent message flow: +/// - Self-loops for local_sends +/// - Cross-edges distributed proportionally (until per-destination tracking exists) +pub fn worker_topology(stats: &RuntimeStats) -> TopologySnapshot { + let mut nodes = Vec::new(); + let mut edges = Vec::new(); + + for w in &stats.workers { + nodes.push(TopologyNode { + id: format!("w{}", w.id), + label: format!("W{}", w.id), + actor_count: w.num_actors, + group: w.id, + }); + + // Local sends = self-loop + if w.local_sends > 0 { + edges.push(TopologyEdge { + source: format!("w{}", w.id), + target: format!("w{}", w.id), + weight: w.local_sends, + label: format!("{} local", w.local_sends), + }); + } + + // Cross sends — without per-destination data, distribute evenly to other workers + if w.cross_sends > 0 && stats.workers.len() > 1 { + let others: Vec<&swactor::stats::WorkerInfo> = + stats.workers.iter().filter(|o| o.id != w.id).collect(); + let per_worker = w.cross_sends / others.len() as u64; + let remainder = w.cross_sends % others.len() as u64; + + for (i, other) in others.iter().enumerate() { + let count = per_worker + if (i as u64) < remainder { 1 } else { 0 }; + if count > 0 { + edges.push(TopologyEdge { + source: format!("w{}", w.id), + target: format!("w{}", other.id), + weight: count, + label: format!("{} cross", count), + }); + } + } + } + } + + TopologySnapshot { nodes, edges } +} diff --git a/crates/runtime-dashboard/src/topology_html.rs b/crates/runtime-dashboard/src/topology_html.rs new file mode 100644 index 0000000..a52ea56 --- /dev/null +++ b/crates/runtime-dashboard/src/topology_html.rs @@ -0,0 +1,280 @@ +pub const TOPOLOGY_HTML: &str = r##" + + + + +Topology — Swactor Dashboard + + + +
+
+

Swactor Runtime Dashboard

+ +
+
+ +
+ +
+ Node size = actor count. Edge thickness = message volume. Green = local sends. Blue = cross-worker sends. +
+
+ + + + +"##;