diff --git a/Cargo.lock b/Cargo.lock index 97b5b8c..701d83d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -216,7 +216,6 @@ version = "0.1.0" dependencies = [ "serde", "serde_json", - "swactor", "swactor-gossip", "tiny_http", "toml", diff --git a/crates/gossip-dashboard/Cargo.toml b/crates/gossip-dashboard/Cargo.toml index 0c0df74..aba86c8 100644 --- a/crates/gossip-dashboard/Cargo.toml +++ b/crates/gossip-dashboard/Cargo.toml @@ -4,7 +4,6 @@ version = "0.1.0" edition = "2024" [dependencies] -swactor = { path = "../..", features = ["serde"] } swactor-gossip = { path = "../swactor-gossip" } serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/crates/gossip-dashboard/README.md b/crates/gossip-dashboard/README.md new file mode 100644 index 0000000..a6b9cd5 --- /dev/null +++ b/crates/gossip-dashboard/README.md @@ -0,0 +1,126 @@ +# gossip-dashboard + +Interactive web dashboard for visualizing gossip protocol simulations. + +The workflow has two steps: + +1. **Generate traces** -- run simulations against TOML config files, producing `.trace.json` files. +2. **Replay traces** -- start the dashboard server, point it at a directory of traces, and explore them in the browser. + +## Quick start + +```bash +# 1. Generate traces from the bundled configs (outputs to traces/) +cargo run -p gossip-dashboard --example generate_traces + +# 2. Launch the dashboard +cargo run -p gossip-dashboard --example replay -- traces +# => open http://localhost:8080 +``` + +## Commands + +### `generate_traces` + +Runs gossip simulations and writes `.trace.json` files. + +``` +generate_traces # all bundled configs -> traces/ +generate_traces # all bundled configs -> / +generate_traces [more.toml …] # specific configs -> / +``` + +Bundled configs live in `examples/configs/`. When no config paths are given, every `.toml` in that directory is run. + +Output filenames are derived from the simulation name (lowercased, spaces to underscores). Example output: + +``` +traces/ + ring_10_nodes.trace.json + star_7_nodes.trace.json + chain_8_nodes.trace.json + full_mesh_6_nodes.trace.json + partition_&_heal_8_nodes.trace.json +``` + +### `replay` + +Starts an HTTP server that serves the dashboard UI and the trace data. + +``` +replay [port] +``` + +| Argument | Required | Default | Description | +|-------------|----------|---------|------------------------------------------| +| `trace-dir` | yes | -- | Directory containing `.trace.json` files | +| `port` | no | 8080 | Port to bind on | + +The server exposes three endpoints: + +| Route | Description | +|--------------------------|------------------------------------| +| `GET /` | Dashboard HTML | +| `GET /traces` | JSON list of available trace files | +| `GET /trace.json?file=…` | Fetch a specific trace | + +## Configuration (TOML) + +Each simulation is defined by a TOML file. Example (`ring_10.toml`): + +```toml +name = "Ring (10 nodes)" +topology = "ring" +num_nodes = 10 +num_rounds = 15 +ticks_per_round = 5 +num_threads = 1 + +[initial_data] +color = "blue" +version = "1" +status = "active" +``` + +### Fields + +| Field | Type | Required | Description | +|--------------------|-------------------|----------|-------------------------------------------------------------------| +| `name` | string | yes | Display name for the simulation | +| `topology` | string | yes | Network topology (see below) | +| `num_nodes` | integer | yes | Number of gossip nodes | +| `num_rounds` | integer | yes | Number of gossip rounds to run | +| `ticks_per_round` | integer | yes | Simulation ticks per round | +| `num_threads` | integer | yes | Worker threads (`1` = deterministic single-threaded) | +| `heal_after_round` | integer | no | Round after which partitioned halves are bridged | +| `initial_data` | table of strings | no | Key-value pairs seeded on node 0 before gossip begins | + +### Topologies + +| Value | Shape | +|---------------|--------------------------------------------------------------------------| +| `ring` | Each node connects to the next, forming a circle | +| `star` | Node 0 is a hub with bidirectional links to every other node | +| `full_mesh` | Every node connects bidirectionally to every other node | +| `chain` | Unidirectional chain: node 0 -> 1 -> 2 -> ... -> N-1 | +| `partitioned` | Two isolated full-mesh halves; use `heal_after_round` to bridge them | + +## Bundled configs + +| File | Topology | Nodes | Rounds | Notes | +|---------------------------|-------------|-------|--------|----------------------------| +| `ring_10.toml` | ring | 10 | 15 | | +| `star_7.toml` | star | 7 | 10 | | +| `full_mesh_6.toml` | full_mesh | 6 | 8 | | +| `chain_8.toml` | chain | 8 | 20 | | +| `partitioned_8_heal.toml` | partitioned | 8 | 20 | Heals after round 10 | + +## Dashboard UI + +Once a trace is loaded in the browser: + +- **Graph canvas** -- nodes arranged in a circle then refined with force-directed layout. Nodes and edges flash as events are replayed. +- **Stats panel** -- total nodes, edges, messages, current round. +- **Worker logs** -- per-thread activity feed. +- **Event table** -- full event log with columns: Seq, Round, Thread, Node, Event, Details. +- **Playback controls** -- First / Prev / Play / Pause / Next / Last, timeline slider, speed adjustment (10 ms -- 2000 ms per event). diff --git a/crates/gossip-dashboard/examples/configs/chain_8.toml b/crates/gossip-dashboard/examples/configs/chain_8.toml new file mode 100644 index 0000000..7d05be4 --- /dev/null +++ b/crates/gossip-dashboard/examples/configs/chain_8.toml @@ -0,0 +1,11 @@ +name = "Chain (8 nodes)" +topology = "chain" +num_nodes = 8 +num_rounds = 20 +ticks_per_round = 4 +num_threads = 1 + +[initial_data] +color = "yellow" +version = "1" +status = "pending" diff --git a/crates/gossip-dashboard/examples/configs/full_mesh_6.toml b/crates/gossip-dashboard/examples/configs/full_mesh_6.toml new file mode 100644 index 0000000..4ccd806 --- /dev/null +++ b/crates/gossip-dashboard/examples/configs/full_mesh_6.toml @@ -0,0 +1,12 @@ +name = "Full Mesh (6 nodes)" +topology = "full_mesh" +num_nodes = 6 +num_rounds = 8 +ticks_per_round = 4 +num_threads = 1 + +[initial_data] +color = "green" +version = "3" +status = "ok" +region = "us-east" diff --git a/crates/gossip-dashboard/examples/configs/partitioned_800_heal.toml b/crates/gossip-dashboard/examples/configs/partitioned_800_heal.toml new file mode 100644 index 0000000..d376853 --- /dev/null +++ b/crates/gossip-dashboard/examples/configs/partitioned_800_heal.toml @@ -0,0 +1,11 @@ +name = "Partition & Heal" +topology = "partitioned" +num_nodes = 800 +num_rounds = 10 +ticks_per_round = 5 +num_threads = 4 +heal_after_round = 5 + +[initial_data] +color = "blue" +version = "1" diff --git a/crates/gossip-dashboard/examples/sim.toml b/crates/gossip-dashboard/examples/configs/partitioned_8_heal.toml similarity index 71% rename from crates/gossip-dashboard/examples/sim.toml rename to crates/gossip-dashboard/examples/configs/partitioned_8_heal.toml index f5a84ac..c6a6a49 100644 --- a/crates/gossip-dashboard/examples/sim.toml +++ b/crates/gossip-dashboard/examples/configs/partitioned_8_heal.toml @@ -1,11 +1,10 @@ -name = "Partitioned-8 Heal" +name = "Partition & Heal (8 nodes)" topology = "partitioned" num_nodes = 8 num_rounds = 20 ticks_per_round = 5 -num_threads = 2 +num_threads = 1 heal_after_round = 10 -port = 8080 [initial_data] color = "blue" diff --git a/crates/gossip-dashboard/examples/configs/ring_10.toml b/crates/gossip-dashboard/examples/configs/ring_10.toml new file mode 100644 index 0000000..25eee20 --- /dev/null +++ b/crates/gossip-dashboard/examples/configs/ring_10.toml @@ -0,0 +1,11 @@ +name = "Ring (10 nodes)" +topology = "ring" +num_nodes = 10 +num_rounds = 15 +ticks_per_round = 5 +num_threads = 1 + +[initial_data] +color = "blue" +version = "1" +status = "active" diff --git a/crates/gossip-dashboard/examples/configs/star_7.toml b/crates/gossip-dashboard/examples/configs/star_7.toml new file mode 100644 index 0000000..bf9ec2b --- /dev/null +++ b/crates/gossip-dashboard/examples/configs/star_7.toml @@ -0,0 +1,11 @@ +name = "Star (7 nodes)" +topology = "star" +num_nodes = 7 +num_rounds = 10 +ticks_per_round = 4 +num_threads = 1 + +[initial_data] +color = "red" +version = "2" +status = "ready" diff --git a/crates/gossip-dashboard/examples/dashboard.rs b/crates/gossip-dashboard/examples/dashboard.rs deleted file mode 100644 index 99d42c0..0000000 --- a/crates/gossip-dashboard/examples/dashboard.rs +++ /dev/null @@ -1,40 +0,0 @@ -use gossip_dashboard::{DashboardConfig, config::SimFileConfig, run_with_dashboard, save_trace}; -use swactor_gossip::sim::{SimConfig, Topology}; - -fn main() { - let args: Vec = std::env::args().collect(); - - let (config, dash) = if let Some(path) = args.get(1) { - let file_config = SimFileConfig::load(path).expect("failed to load config file"); - file_config.into_sim_config() - } else { - let config = SimConfig { - name: "Ring-10 Demo".to_string(), - topology: Topology::Ring, - num_nodes: 10, - initial_data: vec![ - ("color".into(), b"blue".to_vec()), - ("version".into(), b"1".to_vec()), - ("status".into(), b"active".to_vec()), - ], - num_rounds: 15, - ticks_per_round: 5, - heal_after_round: None, - num_threads: 2, - }; - let dash = DashboardConfig { port: 8080 }; - (config, dash) - }; - - eprintln!("Starting gossip dashboard at http://localhost:{}", dash.port); - eprintln!("Open in your browser to see the simulation live."); - - let trace = run_with_dashboard(config, dash); - - let path = "demo.trace.json"; - save_trace(&trace, path).expect("failed to save trace"); - eprintln!("Trace saved to {path}"); - eprintln!( - "Replay with: cargo run -p gossip-dashboard --example replay -- {path}" - ); -} diff --git a/crates/gossip-dashboard/examples/generate_traces.rs b/crates/gossip-dashboard/examples/generate_traces.rs new file mode 100644 index 0000000..c2c6248 --- /dev/null +++ b/crates/gossip-dashboard/examples/generate_traces.rs @@ -0,0 +1,69 @@ +use std::path::PathBuf; + +use gossip_dashboard::config::SimFileConfig; +use gossip_dashboard::save_trace; +use swactor_gossip::sim::run_simulation; + +const CONFIGS_DIR: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/examples/configs"); + +fn main() { + let args: Vec = std::env::args().collect(); + + let (out_dir, configs) = match args.len() { + // generate_traces → traces/ + all bundled configs + 1 => ("traces".to_string(), collect_configs(CONFIGS_DIR)), + // generate_traces → custom dir + all bundled configs + 2 if !args[1].ends_with(".toml") => (args[1].clone(), collect_configs(CONFIGS_DIR)), + // generate_traces + n if n >= 3 => (args[1].clone(), args[2..].iter().map(PathBuf::from).collect()), + _ => { + eprintln!("Usage:"); + eprintln!(" generate_traces # all configs -> traces/"); + eprintln!(" generate_traces # all configs -> out-dir/"); + eprintln!(" generate_traces [more.toml ...]"); + std::process::exit(1); + } + }; + + if configs.is_empty() { + eprintln!("No .toml configs found in {CONFIGS_DIR}"); + std::process::exit(1); + } + + std::fs::create_dir_all(&out_dir).expect("failed to create output directory"); + + for path in &configs { + let path_str = path.to_string_lossy(); + let file_config = SimFileConfig::load(&path_str) + .unwrap_or_else(|e| panic!("failed to load {path_str}: {e}")); + let config = file_config.into_sim_config(); + + eprintln!("Running: {} ...", config.name); + let trace = run_simulation(config); + + let filename = format!( + "{}/{}.trace.json", + out_dir, + trace.name.to_lowercase().replace(' ', "_").replace(['(', ')'], "") + ); + save_trace(&trace, &filename).expect("failed to save trace"); + eprintln!( + " -> {} ({} nodes, {} events)", + filename, + trace.node_names.len(), + trace.events.len() + ); + } + eprintln!("Done. View with: cargo run -p gossip-dashboard --example replay -- {out_dir}"); +} + +fn collect_configs(dir: &str) -> Vec { + let mut paths: Vec = std::fs::read_dir(dir) + .unwrap_or_else(|e| panic!("cannot read configs dir {dir}: {e}")) + .flatten() + .map(|e| e.path()) + .filter(|p| p.extension().is_some_and(|ext| ext == "toml")) + .collect(); + paths.sort(); + paths +} diff --git a/crates/gossip-dashboard/examples/replay.rs b/crates/gossip-dashboard/examples/replay.rs index f5f5482..bceb1fa 100644 --- a/crates/gossip-dashboard/examples/replay.rs +++ b/crates/gossip-dashboard/examples/replay.rs @@ -1,19 +1,15 @@ -use gossip_dashboard::{load_trace, serve_replay}; +use gossip_dashboard::serve_dashboard; fn main() { let args: Vec = std::env::args().collect(); - let path = args + let trace_dir = args .get(1) - .expect("Usage: replay "); + .expect("Usage: replay [port]"); - let trace = load_trace(path).expect("failed to load trace"); + let port: u16 = args + .get(2) + .and_then(|s| s.parse().ok()) + .unwrap_or(8080); - eprintln!( - "Loaded trace '{}': {} nodes, {} events", - trace.name, - trace.node_names.len(), - trace.events.len() - ); - - serve_replay(&trace, 8081); + serve_dashboard(trace_dir, port); } diff --git a/crates/gossip-dashboard/src/config.rs b/crates/gossip-dashboard/src/config.rs index a0ff4af..e209fc7 100644 --- a/crates/gossip-dashboard/src/config.rs +++ b/crates/gossip-dashboard/src/config.rs @@ -5,8 +5,6 @@ use std::io; use serde::Deserialize; use swactor_gossip::sim::{SimConfig, Topology}; -use crate::server::DashboardConfig; - #[derive(Deserialize)] pub struct SimFileConfig { pub name: String, @@ -16,7 +14,6 @@ pub struct SimFileConfig { pub ticks_per_round: usize, pub num_threads: usize, pub heal_after_round: Option, - pub port: Option, pub initial_data: Option>, } @@ -26,7 +23,7 @@ impl SimFileConfig { toml::from_str(&contents).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e)) } - pub fn into_sim_config(self) -> (SimConfig, DashboardConfig) { + pub fn into_sim_config(self) -> SimConfig { let topology = match self.topology.to_lowercase().as_str() { "ring" => Topology::Ring, "star" => Topology::Star, @@ -43,7 +40,7 @@ impl SimFileConfig { .map(|(k, v)| (k, v.into_bytes())) .collect(); - let sim = SimConfig { + SimConfig { name: self.name, topology, num_nodes: self.num_nodes, @@ -52,12 +49,6 @@ impl SimFileConfig { ticks_per_round: self.ticks_per_round, heal_after_round: self.heal_after_round, num_threads: self.num_threads, - }; - - let dash = DashboardConfig { - port: self.port.unwrap_or(8080), - }; - - (sim, dash) + } } } diff --git a/crates/gossip-dashboard/src/dashboard_html.rs b/crates/gossip-dashboard/src/dashboard_html.rs index c9eb6b0..5d3fc90 100644 --- a/crates/gossip-dashboard/src/dashboard_html.rs +++ b/crates/gossip-dashboard/src/dashboard_html.rs @@ -13,12 +13,24 @@ pub const DASHBOARD_HTML: &str = r##" padding: 12px 20px; background: #161822; border-bottom: 1px solid #2a2d3a; } .header h1 { font-size: 18px; font-weight: 600; color: #c0c6d4; } + + .trace-select { + background: #1a1d2e; color: #e0e0e0; border: 1px solid #2a2d3a; + border-radius: 4px; padding: 6px 12px; font-size: 13px; + cursor: pointer; min-width: 240px; max-width: 420px; + } + .trace-select:hover { border-color: #4f46e5; } + .trace-select:focus { outline: none; border-color: #6366f1; } + .trace-select:disabled { cursor: default; opacity: 0.5; } + .status-badge { display: flex; align-items: center; gap: 6px; font-size: 13px; color: #9ca3af; } - .status-dot { width: 8px; height: 8px; border-radius: 50%; background: #3b82f6; } - .status-dot.done { background: #22c55e; } - .status-dot.replay { background: #f59e0b; } + .status-dot { width: 8px; height: 8px; border-radius: 50%; background: #f59e0b; } + .status-dot.ready { background: #22c55e; } + .status-dot.loading { background: #3b82f6; animation: pulse 1s infinite; } + .status-dot.error { background: #ef4444; } + @keyframes pulse { 0%,100% { opacity: 1; } 50% { opacity: 0.4; } } .main { display: grid; grid-template-columns: 1fr 1fr; grid-template-rows: auto 1fr; height: calc(100vh - 48px); } @@ -82,18 +94,18 @@ pub const DASHBOARD_HTML: &str = r##" border-radius: 4px; padding: 2px 6px; font-size: 12px; text-align: right; } .replay-controls .speed-group label { font-size: 11px; color: #6b7280; white-space: nowrap; } - - .node-flash { animation: flash 0.4s ease-out; } - @keyframes flash { 0% { filter: brightness(2); } 100% { filter: brightness(1); } }
-

Gossip Simulation Dashboard

+

Gossip Dashboard

+
-
- Connecting... +
+ Loading traces...
@@ -141,23 +153,23 @@ pub const DASHBOARD_HTML: &str = r##" diff --git a/crates/gossip-dashboard/src/lib.rs b/crates/gossip-dashboard/src/lib.rs index 7b1da6d..2be0444 100644 --- a/crates/gossip-dashboard/src/lib.rs +++ b/crates/gossip-dashboard/src/lib.rs @@ -2,7 +2,7 @@ pub mod config; mod dashboard_html; mod server; -pub use server::{DashboardConfig, run_with_dashboard, serve_replay}; +pub use server::serve_dashboard; use std::fs; use std::io; diff --git a/crates/gossip-dashboard/src/server.rs b/crates/gossip-dashboard/src/server.rs index e4cf230..10fda6e 100644 --- a/crates/gossip-dashboard/src/server.rs +++ b/crates/gossip-dashboard/src/server.rs @@ -1,720 +1,78 @@ -use std::collections::HashMap; -use std::io::{self, Read as IoRead}; -use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use std::sync::{mpsc, Arc, Mutex}; -use std::thread; -use std::time::Duration; +use std::fs; +use std::path::{Path, PathBuf}; use serde::Serialize; -use swactor::actor::ActorAddress; -use swactor::config::RuntimeConfig; -use swactor::runtime::Runtime; -use swactor_gossip::protocol::{GossipActor, GossipMessage}; -use swactor_gossip::sim::{heal_partition_via_handle, wire_topology, SimConfig}; -use swactor_gossip::trace::{ - EventLog, GossipEvent, GossipEventKind, NameRegistry, NodeSnapshot, SimulationTrace, - TickCounter, -}; use crate::dashboard_html::DASHBOARD_HTML; -// ── Configuration ────────────────────────────────────────────────────── - -#[derive(Debug, Clone)] -pub struct DashboardConfig { - pub port: u16, -} - -impl Default for DashboardConfig { - fn default() -> Self { - Self { port: 8080 } - } -} - -// ── Shared dashboard state ───────────────────────────────────────────── - -struct DashboardState { - event_log: EventLog, - name_registry: NameRegistry, - init_data: Mutex>, - stats: Mutex, - done: AtomicBool, -} +// ── Trace directory scanning ────────────────────────────────────────── #[derive(Debug, Clone, Serialize)] -struct InitData { +struct TraceEntry { + file: String, name: String, - nodes: Vec, - edges: Vec<[String; 2]>, - num_threads: usize, + nodes: usize, + events: usize, } -#[derive(Debug, Clone, Serialize)] -struct NodeInfo { - name: String, - addr: String, -} - -#[derive(Debug, Clone, Default, Serialize)] -struct StatsSnapshot { - total_nodes: usize, - total_edges: usize, - total_messages: usize, - current_round: u64, - total_rounds: usize, -} - -// ── SSE channel adapter ──────────────────────────────────────────────── - -/// Adapts an `mpsc::Receiver>` to `std::io::Read` for tiny_http streaming. -struct ChannelReader { - rx: mpsc::Receiver>, - buf: Vec, - pos: usize, -} - -impl ChannelReader { - fn new(rx: mpsc::Receiver>) -> Self { - Self { - rx, - buf: Vec::new(), - pos: 0, - } - } -} - -impl IoRead for ChannelReader { - fn read(&mut self, out: &mut [u8]) -> io::Result { - // Drain current buffer first. - if self.pos < self.buf.len() { - let n = std::cmp::min(out.len(), self.buf.len() - self.pos); - out[..n].copy_from_slice(&self.buf[self.pos..self.pos + n]); - self.pos += n; - return Ok(n); - } - - // Wait for next chunk. - match self.rx.recv() { - Ok(data) => { - if data.is_empty() { - return Ok(0); // EOF signal - } - let n = std::cmp::min(out.len(), data.len()); - out[..n].copy_from_slice(&data[..n]); - if n < data.len() { - self.buf = data; - self.pos = n; - } else { - self.buf.clear(); - self.pos = 0; - } - Ok(n) - } - Err(_) => Ok(0), // channel closed - } - } -} - -// ── SSE formatting helpers ───────────────────────────────────────────── - -fn format_sse(event: &str, data: &str) -> Vec { - format!("event: {event}\ndata: {data}\n\n").into_bytes() -} - -fn event_kind_name(kind: &GossipEventKind) -> &'static str { - match kind { - GossipEventKind::LocalSet { .. } => "LocalSet", - GossipEventKind::GossipRoundStarted { .. } => "GossipRoundStarted", - GossipEventKind::GossipRoundNoPeers => "GossipRoundNoPeers", - GossipEventKind::PushReceived { .. } => "PushReceived", - GossipEventKind::QueryReceived { .. } => "QueryReceived", - GossipEventKind::PeerAdded { .. } => "PeerAdded", - GossipEventKind::PeerRemoved { .. } => "PeerRemoved", - GossipEventKind::StateSnapshot { .. } => "StateSnapshot", - } -} - -#[derive(Serialize)] -struct SseGossipEvent { - seq: usize, - tick: u64, - node: String, - thread: Option, - kind: String, - detail: serde_json::Value, -} - -fn gossip_event_to_sse(seq: usize, ev: &GossipEvent) -> SseGossipEvent { - let detail = match &ev.kind { - GossipEventKind::LocalSet { key } => { - serde_json::json!({ "key": key }) - } - GossipEventKind::GossipRoundStarted { target_name } => { - serde_json::json!({ "target": target_name }) - } - GossipEventKind::GossipRoundNoPeers => serde_json::json!({}), - GossipEventKind::PushReceived { - from_name, - keys_updated, - } => { - serde_json::json!({ "from": from_name, "keys_updated": keys_updated }) - } - GossipEventKind::QueryReceived { key } => { - serde_json::json!({ "key": key }) - } - GossipEventKind::PeerAdded { peer_name } => { - serde_json::json!({ "peer": peer_name }) - } - GossipEventKind::PeerRemoved { peer_name } => { - serde_json::json!({ "peer": peer_name }) - } - GossipEventKind::StateSnapshot { snapshot } => { - serde_json::json!({ - "entries": snapshot.entries.len(), - "peer_count": snapshot.peer_count, - }) - } +fn scan_traces(dir: &Path) -> Vec { + let mut entries = Vec::new(); + let Ok(read_dir) = fs::read_dir(dir) else { + return entries; }; - - SseGossipEvent { - seq, - tick: ev.tick, - node: ev.node_name.clone(), - thread: ev.thread_name.clone(), - kind: event_kind_name(&ev.kind).to_string(), - detail, - } -} - -// ── HTTP server ──────────────────────────────────────────────────────── - -fn spawn_http_server(state: Arc, port: u16, mode: &str) { - let addr = format!("0.0.0.0:{port}"); - let server = tiny_http::Server::http(&addr).expect("failed to bind HTTP server"); - let server = Arc::new(server); - let mode = mode.to_string(); - - // Spawn a pool of handler threads. - for _ in 0..4 { - let server = Arc::clone(&server); - let state = Arc::clone(&state); - let mode = mode.clone(); - thread::spawn(move || { - loop { - let request = match server.recv() { - Ok(r) => r, - Err(_) => break, - }; - - let url = request.url().to_string(); - match url.as_str() { - "/" => { - let html = DASHBOARD_HTML.replace("__DASHBOARD_MODE__", &mode); - let response = tiny_http::Response::from_string(html) - .with_header( - "Content-Type: text/html; charset=utf-8" - .parse::() - .unwrap(), - ); - let _ = request.respond(response); - } - "/events" => { - handle_sse(request, Arc::clone(&state)); - } - "/trace.json" => { - handle_trace_json(request, Arc::clone(&state)); - } - _ => { - let response = - tiny_http::Response::from_string("Not Found").with_status_code(404); - let _ = request.respond(response); - } - } - } + for entry in read_dir.flatten() { + let path = entry.path(); + let fname = path + .file_name() + .map(|n| n.to_string_lossy().to_string()) + .unwrap_or_default(); + if !fname.ends_with(".trace.json") { + continue; + } + let Ok(data) = fs::read_to_string(&path) else { + continue; + }; + // Parse as generic JSON to extract metadata without full deserialization. + let Ok(val) = serde_json::from_str::(&data) else { + continue; + }; + let name = val + .get("name") + .and_then(|v| v.as_str()) + .unwrap_or(&fname) + .to_string(); + let nodes = val + .get("node_names") + .and_then(|v| v.as_array()) + .map(|a| a.len()) + .unwrap_or(0); + let events = val + .get("events") + .and_then(|v| v.as_array()) + .map(|a| a.len()) + .unwrap_or(0); + entries.push(TraceEntry { + file: fname, + name, + nodes, + events, }); } + entries.sort_by(|a, b| a.file.cmp(&b.file)); + entries } -fn handle_sse(request: tiny_http::Request, state: Arc) { - let (tx, rx) = mpsc::channel::>(); - let reader = ChannelReader::new(rx); +// ── HTTP server ─────────────────────────────────────────────────────── - // Send SSE headers via a streaming response. - let response = tiny_http::Response::new( - tiny_http::StatusCode(200), - vec![ - "Content-Type: text/event-stream" - .parse::() - .unwrap(), - "Cache-Control: no-cache" - .parse::() - .unwrap(), - "Connection: keep-alive" - .parse::() - .unwrap(), - ], - Box::new(reader) as Box, - None, - None, - ); - - // Spawn producer thread that polls for new events. - thread::spawn(move || { - let mut cursor: usize = 0; - - // Wait for init data. - loop { - if let Some(init) = state.init_data.lock().unwrap().as_ref() { - let json = serde_json::to_string(init).unwrap(); - if tx.send(format_sse("init", &json)).is_err() { - return; - } - break; - } - thread::sleep(Duration::from_millis(50)); - } - - // Poll for events. - loop { - { - let log = state.event_log.lock().unwrap(); - while cursor < log.len() { - let ev = &log[cursor]; - // Filter out StateSnapshot events from SSE stream. - if !matches!(ev.kind, GossipEventKind::StateSnapshot { .. }) { - let sse_ev = gossip_event_to_sse(cursor, ev); - let json = serde_json::to_string(&sse_ev).unwrap(); - if tx.send(format_sse("gossip", &json)).is_err() { - return; - } - } - cursor += 1; - } - } - - // Send stats update. - { - let stats = state.stats.lock().unwrap().clone(); - let json = serde_json::to_string(&stats).unwrap(); - if tx.send(format_sse("stats", &json)).is_err() { - return; - } - } - - if state.done.load(Ordering::Relaxed) { - let _ = tx.send(format_sse("done", "{}")); - let _ = tx.send(Vec::new()); // EOF - return; - } - - thread::sleep(Duration::from_millis(50)); - } - }); - - // This blocks until the reader is consumed / connection closes. - let _ = request.respond(response); -} - -fn handle_trace_json(request: tiny_http::Request, state: Arc) { - // Build a partial trace from current state. - let events = state.event_log.lock().unwrap().clone(); - let names_map = state.name_registry.lock().unwrap().clone(); - let init = state.init_data.lock().unwrap().clone(); - - let trace = SimulationTrace { - name: init.as_ref().map(|i| i.name.clone()).unwrap_or_default(), - node_names: init - .as_ref() - .map(|i| i.nodes.iter().map(|n| n.name.clone()).collect()) - .unwrap_or_default(), - node_addrs: { - let mut addrs: Vec = Vec::new(); - if let Some(init) = &init { - // Reconstruct addrs from name_registry in node order. - let inv: HashMap = - names_map.into_iter().map(|(a, n)| (n, a)).collect(); - for node in &init.nodes { - if let Some(&addr) = inv.get(&node.name) { - addrs.push(addr); - } - } - } - addrs - }, - topology_edges: init - .as_ref() - .map(|i| { - i.edges - .iter() - .map(|e| (e[0].clone(), e[1].clone())) - .collect() - }) - .unwrap_or_default(), - events, - snapshots_per_round: Vec::new(), - num_rounds: init - .as_ref() - .map(|_| { - state - .stats - .lock() - .unwrap() - .total_rounds - }) - .unwrap_or(0), - total_keys: 0, - }; - - let json = serde_json::to_string(&trace).unwrap(); - let response = tiny_http::Response::from_string(json).with_header( - "Content-Type: application/json" - .parse::() - .unwrap(), - ); - let _ = request.respond(response); -} - -// ── Public API: run_with_dashboard ───────────────────────────────────── - -pub fn run_with_dashboard(config: SimConfig, dash: DashboardConfig) -> SimulationTrace { - let num_threads = config.num_threads.max(1); - let event_log: EventLog = Arc::new(Mutex::new(Vec::new())); - let tick_counter: TickCounter = Arc::new(AtomicU64::new(0)); - let name_registry: NameRegistry = Arc::new(Mutex::new(HashMap::new())); - - let state = Arc::new(DashboardState { - event_log: Arc::clone(&event_log), - name_registry: Arc::clone(&name_registry), - init_data: Mutex::new(None), - stats: Mutex::new(StatsSnapshot::default()), - done: AtomicBool::new(false), - }); - - // Start HTTP server. - spawn_http_server(Arc::clone(&state), dash.port, "live"); - - if num_threads < 2 { - run_dashboard_single_threaded(config, state, event_log, tick_counter, name_registry) - } else { - run_dashboard_multi_threaded(config, state, event_log, tick_counter, name_registry) - } -} - -fn run_dashboard_single_threaded( - config: SimConfig, - state: Arc, - event_log: EventLog, - tick_counter: TickCounter, - name_registry: NameRegistry, -) -> SimulationTrace { - let rt = Runtime::new(RuntimeConfig { - num_threads: 1, - max_actors: (config.num_nodes + 64).next_power_of_two(), - actor_max_messages: (config.num_nodes * 4).max(1_000), - ..Default::default() - }); - - let mut addrs = Vec::with_capacity(config.num_nodes); - let mut names = Vec::with_capacity(config.num_nodes); - for i in 0..config.num_nodes { - let name = format!("node-{i}"); - let actor = GossipActor::traced( - Arc::clone(&event_log), - Arc::clone(&tick_counter), - Arc::clone(&name_registry), - ); - let addr = rt.spawn(actor).unwrap(); - name_registry.lock().unwrap().insert(addr, name.clone()); - addrs.push(addr); - names.push(name); - } - - let edges = wire_topology(&rt, &config.topology, &addrs, &names); - for _ in 0..3 { - rt.tick(); - } - - // Publish init data. - publish_init(&state, &config, &addrs, &names, &edges); - - let total_keys = config.initial_data.len(); - for (key, value) in &config.initial_data { - rt.send_to( - addrs[0], - GossipMessage::Set { - key: key.clone(), - value: value.clone(), - }, - ) - .unwrap(); - } - rt.tick(); - - let mut snapshots_per_round: Vec> = Vec::new(); - - for round in 0..config.num_rounds { - if config.heal_after_round == Some(round) { - swactor_gossip::sim::heal_partition(&rt, &config.topology, &addrs, &names); - for _ in 0..3 { - rt.tick(); - } - } - - tick_counter.store((round + 1) as u64, Ordering::Relaxed); - update_stats(&state, &config, round, &event_log); - - for &addr in &addrs { - rt.send_to(addr, GossipMessage::DoGossipRound).unwrap(); - } - for _ in 0..config.ticks_per_round { - rt.tick(); - } - - for &addr in &addrs { - rt.send_to(addr, GossipMessage::TakeSnapshot).unwrap(); - } - for _ in 0..3 { - rt.tick(); - } - - let current_round_tick = (round + 1) as u64; - let log = event_log.lock().unwrap(); - let mut round_snapshots: Vec<(String, NodeSnapshot)> = Vec::new(); - for event in log.iter().rev() { - if event.tick != current_round_tick { - break; - } - if let GossipEventKind::StateSnapshot { ref snapshot } = event.kind { - round_snapshots.push((event.node_name.clone(), snapshot.clone())); - } - } - round_snapshots.reverse(); - snapshots_per_round.push(round_snapshots); - } - - state.done.store(true, Ordering::Relaxed); - - let events = event_log.lock().unwrap().clone(); - SimulationTrace { - name: config.name, - node_names: names, - node_addrs: addrs, - topology_edges: edges, - events, - snapshots_per_round, - num_rounds: config.num_rounds, - total_keys, - } -} - -fn run_dashboard_multi_threaded( - config: SimConfig, - state: Arc, - event_log: EventLog, - tick_counter: TickCounter, - name_registry: NameRegistry, -) -> SimulationTrace { - let ticks_per_round = config.ticks_per_round; - let settle_ms = (ticks_per_round as u64 * 2).max(10); - - let rt = Runtime::new(RuntimeConfig { - num_threads: config.num_threads, - max_actors: (config.num_nodes + 64).next_power_of_two(), - actor_max_messages: (config.num_nodes * 4).max(1_000), - ..Default::default() - }); - - let mut addrs = Vec::with_capacity(config.num_nodes); - let mut names = Vec::with_capacity(config.num_nodes); - for i in 0..config.num_nodes { - let name = format!("node-{i}"); - let actor = GossipActor::traced( - Arc::clone(&event_log), - Arc::clone(&tick_counter), - Arc::clone(&name_registry), - ); - let addr = rt.spawn(actor).unwrap(); - name_registry.lock().unwrap().insert(addr, name.clone()); - addrs.push(addr); - names.push(name); - } - - let edges = wire_topology(&rt, &config.topology, &addrs, &names); - - // Publish init data. - publish_init(&state, &config, &addrs, &names, &edges); - - let total_keys = config.initial_data.len(); - for (key, value) in &config.initial_data { - rt.send_to( - addrs[0], - GossipMessage::Set { - key: key.clone(), - value: value.clone(), - }, - ) - .unwrap(); - } - - let handle = rt.run().expect("failed to start multi-threaded runtime"); - thread::sleep(Duration::from_millis(settle_ms * 2)); - - let mut snapshots_per_round: Vec> = Vec::new(); - - for round in 0..config.num_rounds { - if config.heal_after_round == Some(round) { - heal_partition_via_handle(&handle, &config.topology, &addrs, &names); - thread::sleep(Duration::from_millis(settle_ms)); - } - - tick_counter.store((round + 1) as u64, Ordering::Relaxed); - update_stats(&state, &config, round, &event_log); - - for &addr in &addrs { - handle - .runtime - .send_to(addr, GossipMessage::DoGossipRound) - .unwrap(); - } - thread::sleep(Duration::from_millis(settle_ms)); - - for &addr in &addrs { - handle - .runtime - .send_to(addr, GossipMessage::TakeSnapshot) - .unwrap(); - } - thread::sleep(Duration::from_millis(settle_ms / 2)); - - let current_round_tick = (round + 1) as u64; - let log = event_log.lock().unwrap(); - let mut round_snapshots: Vec<(String, NodeSnapshot)> = Vec::new(); - for event in log.iter().rev() { - if event.tick != current_round_tick { - break; - } - if let GossipEventKind::StateSnapshot { ref snapshot } = event.kind { - round_snapshots.push((event.node_name.clone(), snapshot.clone())); - } - } - round_snapshots.reverse(); - snapshots_per_round.push(round_snapshots); - } - - handle.shutdown(); - handle.join(); - - state.done.store(true, Ordering::Relaxed); - - let events = event_log.lock().unwrap().clone(); - SimulationTrace { - name: config.name, - node_names: names, - node_addrs: addrs, - topology_edges: edges, - events, - snapshots_per_round, - num_rounds: config.num_rounds, - total_keys, - } -} - -// ── Helpers ──────────────────────────────────────────────────────────── - -fn publish_init( - state: &DashboardState, - config: &SimConfig, - addrs: &[ActorAddress], - names: &[String], - edges: &[(String, String)], -) { - let nodes: Vec = names - .iter() - .zip(addrs.iter()) - .map(|(name, addr)| NodeInfo { - name: name.clone(), - addr: format!("{addr}"), - }) - .collect(); - let edge_pairs: Vec<[String; 2]> = edges - .iter() - .map(|(a, b)| [a.clone(), b.clone()]) - .collect(); - *state.init_data.lock().unwrap() = Some(InitData { - name: config.name.clone(), - nodes, - edges: edge_pairs, - num_threads: config.num_threads, - }); - *state.stats.lock().unwrap() = StatsSnapshot { - total_nodes: config.num_nodes, - total_edges: edges.len(), - total_messages: 0, - current_round: 0, - total_rounds: config.num_rounds, - }; -} - -fn update_stats(state: &DashboardState, config: &SimConfig, round: usize, event_log: &EventLog) { - let msg_count = event_log.lock().unwrap().len(); - let mut stats = state.stats.lock().unwrap(); - stats.current_round = (round + 1) as u64; - stats.total_messages = msg_count; - stats.total_rounds = config.num_rounds; -} - -// ── Replay mode ──────────────────────────────────────────────────────── - -pub fn serve_replay(trace: &SimulationTrace, port: u16) { - let event_log: EventLog = Arc::new(Mutex::new(trace.events.clone())); - let name_registry: NameRegistry = Arc::new(Mutex::new( - trace - .node_names - .iter() - .zip(trace.node_addrs.iter()) - .map(|(n, a)| (*a, n.clone())) - .collect(), - )); - - let edges: Vec<[String; 2]> = trace - .topology_edges - .iter() - .map(|(a, b)| [a.clone(), b.clone()]) - .collect(); - let nodes: Vec = trace - .node_names - .iter() - .zip(trace.node_addrs.iter()) - .map(|(name, addr)| NodeInfo { - name: name.clone(), - addr: format!("{addr}"), - }) - .collect(); - - // Keep state alive for potential future SSE support in replay mode. - let _state = Arc::new(DashboardState { - event_log, - name_registry, - init_data: Mutex::new(Some(InitData { - name: trace.name.clone(), - nodes, - edges, - num_threads: 1, - })), - stats: Mutex::new(StatsSnapshot { - total_nodes: trace.node_names.len(), - total_edges: trace.topology_edges.len(), - total_messages: trace.events.len(), - current_round: trace.num_rounds as u64, - total_rounds: trace.num_rounds, - }), - done: AtomicBool::new(true), - }); +pub fn serve_dashboard(trace_dir: &str, port: u16) { + let dir = PathBuf::from(trace_dir); + assert!(dir.is_dir(), "trace directory does not exist: {trace_dir}"); let addr = format!("0.0.0.0:{port}"); let server = tiny_http::Server::http(&addr).expect("failed to bind HTTP server"); - eprintln!("Replay dashboard at http://localhost:{port}"); + eprintln!("Dashboard at http://localhost:{port}"); + eprintln!("Serving traces from: {trace_dir}"); loop { let request = match server.recv() { @@ -725,16 +83,16 @@ pub fn serve_replay(trace: &SimulationTrace, port: u16) { let url = request.url().to_string(); match url.as_str() { "/" => { - let html = DASHBOARD_HTML.replace("__DASHBOARD_MODE__", "replay"); - let response = tiny_http::Response::from_string(html).with_header( + let response = tiny_http::Response::from_string(DASHBOARD_HTML).with_header( "Content-Type: text/html; charset=utf-8" .parse::() .unwrap(), ); let _ = request.respond(response); } - "/trace.json" => { - let json = serde_json::to_string(trace).unwrap(); + "/traces" => { + let entries = scan_traces(&dir); + let json = serde_json::to_string(&entries).unwrap(); let response = tiny_http::Response::from_string(json).with_header( "Content-Type: application/json" .parse::() @@ -742,6 +100,39 @@ pub fn serve_replay(trace: &SimulationTrace, port: u16) { ); let _ = request.respond(response); } + _ if url.starts_with("/trace.json?file=") => { + let raw = url.strip_prefix("/trace.json?file=").unwrap(); + let file = percent_decode(raw); + + // Reject path traversal attempts. + if file.contains('/') + || file.contains('\\') + || file.contains("..") + || file.is_empty() + { + let response = + tiny_http::Response::from_string("Bad Request").with_status_code(400); + let _ = request.respond(response); + continue; + } + + let path = dir.join(&file); + match fs::read_to_string(&path) { + Ok(data) => { + let response = tiny_http::Response::from_string(data).with_header( + "Content-Type: application/json" + .parse::() + .unwrap(), + ); + let _ = request.respond(response); + } + Err(_) => { + let response = + tiny_http::Response::from_string("Not Found").with_status_code(404); + let _ = request.respond(response); + } + } + } _ => { let response = tiny_http::Response::from_string("Not Found").with_status_code(404); @@ -750,3 +141,32 @@ pub fn serve_replay(trace: &SimulationTrace, port: u16) { } } } + +// ── Helpers ─────────────────────────────────────────────────────────── + +fn percent_decode(s: &str) -> String { + let mut result = Vec::new(); + let bytes = s.as_bytes(); + let mut i = 0; + while i < bytes.len() { + if bytes[i] == b'%' && i + 2 < bytes.len() { + if let (Some(h), Some(l)) = (hex_val(bytes[i + 1]), hex_val(bytes[i + 2])) { + result.push(h << 4 | l); + i += 3; + continue; + } + } + result.push(bytes[i]); + i += 1; + } + String::from_utf8_lossy(&result).to_string() +} + +fn hex_val(b: u8) -> Option { + match b { + b'0'..=b'9' => Some(b - b'0'), + b'a'..=b'f' => Some(b - b'a' + 10), + b'A'..=b'F' => Some(b - b'A' + 10), + _ => None, + } +} diff --git a/crates/swactor-gossip/src/sim.rs b/crates/swactor-gossip/src/sim.rs index 3defecf..9e1846f 100644 --- a/crates/swactor-gossip/src/sim.rs +++ b/crates/swactor-gossip/src/sim.rs @@ -289,13 +289,14 @@ pub fn heal_partition_via_handle( handle: &swactor::runtime::RuntimeHandle, topology: &Topology, addrs: &[ActorAddress], - _names: &[String], -) { + names: &[String], +) -> Vec<(String, String)> { if !matches!(topology, Topology::Partitioned) { - return; + return Vec::new(); } let n = addrs.len(); let half = n / 2; + let mut new_edges = Vec::new(); if half > 0 && half < n { handle .runtime @@ -305,7 +306,10 @@ pub fn heal_partition_via_handle( .runtime .send_to(addrs[half], GossipMessage::AddPeer(addrs[half - 1])) .unwrap(); + new_edges.push((names[half - 1].clone(), names[half].clone())); + new_edges.push((names[half].clone(), names[half - 1].clone())); } + new_edges } // ── Topology wiring ────────────────────────────────────────────────────── @@ -377,18 +381,22 @@ pub fn heal_partition( rt: &Runtime, topology: &Topology, addrs: &[ActorAddress], - _names: &[String], -) { + names: &[String], +) -> Vec<(String, String)> { if !matches!(topology, Topology::Partitioned) { - return; + return Vec::new(); } let n = addrs.len(); let half = n / 2; + let mut new_edges = Vec::new(); // Add bidirectional links between the two halves (bridge nodes). if half > 0 && half < n { rt.send_to(addrs[half - 1], GossipMessage::AddPeer(addrs[half])) .unwrap(); rt.send_to(addrs[half], GossipMessage::AddPeer(addrs[half - 1])) .unwrap(); + new_edges.push((names[half - 1].clone(), names[half].clone())); + new_edges.push((names[half].clone(), names[half - 1].clone())); } + new_edges }