diff --git a/Cargo.lock b/Cargo.lock index a208c57..16bedcb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -534,6 +534,14 @@ dependencies = [ "getrandom", ] +[[package]] +name = "swactor-gossip" +version = "0.1.0" +dependencies = [ + "getrandom", + "swactor", +] + [[package]] name = "swactor-python" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 33da0af..7ef7f03 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = [".", "crates/swactor-python", "crates/swactor-wasm"] +members = [".", "crates/swactor-python", "crates/swactor-wasm", "crates/swactor-gossip"] exclude = ["tools/depgraph"] [package] diff --git a/crates/swactor-gossip/Cargo.toml b/crates/swactor-gossip/Cargo.toml new file mode 100644 index 0000000..5801da5 --- /dev/null +++ b/crates/swactor-gossip/Cargo.toml @@ -0,0 +1,8 @@ +[package] +name = "swactor-gossip" +version = "0.1.0" +edition = "2024" + +[dependencies] +swactor = { path = "../.." } +getrandom = "0.2" diff --git a/crates/swactor-gossip/examples/gossip_sim.rs b/crates/swactor-gossip/examples/gossip_sim.rs new file mode 100644 index 0000000..00140c6 --- /dev/null +++ b/crates/swactor-gossip/examples/gossip_sim.rs @@ -0,0 +1,73 @@ +use std::fs; + +use swactor_gossip::report::generate_html_report; +use swactor_gossip::sim::{run_simulation, SimConfig, Topology}; + +fn main() { + let scenarios = vec![ + SimConfig { + name: "Ring (5 nodes)".into(), + topology: Topology::Ring, + num_nodes: 5, + initial_data: test_data(3), + num_rounds: 15, + ticks_per_round: 4, + heal_after_round: None, + }, + SimConfig { + name: "Star (7 nodes)".into(), + topology: Topology::Star, + num_nodes: 7, + initial_data: test_data(3), + num_rounds: 10, + ticks_per_round: 4, + heal_after_round: None, + }, + SimConfig { + name: "Full Mesh (5 nodes)".into(), + topology: Topology::FullMesh, + num_nodes: 5, + initial_data: test_data(3), + num_rounds: 8, + ticks_per_round: 4, + heal_after_round: None, + }, + SimConfig { + name: "Chain (8 nodes)".into(), + topology: Topology::Chain, + num_nodes: 8, + initial_data: test_data(3), + num_rounds: 20, + ticks_per_round: 4, + heal_after_round: None, + }, + SimConfig { + name: "Partition & Heal (6 nodes)".into(), + topology: Topology::Partitioned, + num_nodes: 6, + initial_data: test_data(3), + num_rounds: 20, + ticks_per_round: 4, + heal_after_round: Some(10), + }, + ]; + + for config in scenarios { + let filename = format!( + "gossip_report_{}.html", + config.name.to_lowercase().replace(' ', "_").replace(['(', ')'], "") + ); + println!("Running scenario: {} ...", config.name); + let trace = run_simulation(config); + let html = generate_html_report(&trace); + fs::write(&filename, &html).expect("failed to write report"); + println!(" -> wrote {filename} ({} bytes)", html.len()); + } + println!("Done."); +} + +fn test_data(n: usize) -> Vec<(String, Vec)> { + (0..n) + .map(|i| (format!("key-{i}"), format!("value-{i}").into_bytes())) + .collect() +} diff --git a/crates/swactor-gossip/src/actor.rs b/crates/swactor-gossip/src/actor.rs new file mode 100644 index 0000000..d72d1e3 --- /dev/null +++ b/crates/swactor-gossip/src/actor.rs @@ -0,0 +1,179 @@ +use swactor::actor::{ActorAddress, ActorInterface, Ctx}; + +use crate::message::{GossipMessage, GossipQueryResponse}; +use crate::state::GossipState; +use crate::trace::{ + current_tick, record_event, resolve_name, EventLog, GossipEvent, GossipEventKind, + NameRegistry, NodeSnapshot, TickCounter, +}; + +pub struct GossipActor { + state: GossipState, + peers: Vec, + event_log: Option, + tick_counter: Option, + name_registry: Option, +} + +impl GossipActor { + pub fn new() -> Self { + Self { + state: GossipState::new(), + peers: Vec::new(), + event_log: None, + tick_counter: None, + name_registry: None, + } + } + + /// Create a traced actor that records events into the shared log. + pub fn traced(log: EventLog, tick: TickCounter, names: NameRegistry) -> Self { + Self { + state: GossipState::new(), + peers: Vec::new(), + event_log: Some(log), + tick_counter: Some(tick), + name_registry: Some(names), + } + } + + fn pick_random_peer(&self) -> Option { + if self.peers.is_empty() { + return None; + } + let mut buf = [0u8; 8]; + getrandom::getrandom(&mut buf).unwrap(); + let idx = usize::from_ne_bytes(buf) % self.peers.len(); + Some(self.peers[idx]) + } + + fn record(&self, addr: ActorAddress, kind: GossipEventKind) { + if let (Some(log), Some(tick), Some(names)) = + (&self.event_log, &self.tick_counter, &self.name_registry) + { + let event = GossipEvent { + tick: current_tick(tick), + node_name: resolve_name(names, addr), + node_addr: addr, + kind, + }; + record_event(log, event); + } + } +} + +impl Default for GossipActor { + fn default() -> Self { + Self::new() + } +} + +impl ActorInterface for GossipActor { + type Incoming = GossipMessage; + type Response = (); + + fn handle(&mut self, ctx: &Ctx, msg: GossipMessage) { + let self_addr = ctx.self_addr(); + match msg { + GossipMessage::AddPeer(addr) => { + if !self.peers.contains(&addr) { + self.peers.push(addr); + self.record( + self_addr, + GossipEventKind::PeerAdded { + peer_name: self + .name_registry + .as_ref() + .map(|r| resolve_name(r, addr)) + .unwrap_or_default(), + }, + ); + } + } + GossipMessage::RemovePeer(addr) => { + let before = self.peers.len(); + self.peers.retain(|a| *a != addr); + if self.peers.len() < before { + self.record( + self_addr, + GossipEventKind::PeerRemoved { + peer_name: self + .name_registry + .as_ref() + .map(|r| resolve_name(r, addr)) + .unwrap_or_default(), + }, + ); + } + } + GossipMessage::Set { key, value } => { + self.state.set(key.clone(), value); + self.record(self_addr, GossipEventKind::LocalSet { key }); + } + GossipMessage::DoGossipRound => { + if let Some(peer) = self.pick_random_peer() { + self.record( + self_addr, + GossipEventKind::GossipRoundStarted { + target_name: self + .name_registry + .as_ref() + .map(|r| resolve_name(r, peer)) + .unwrap_or_default(), + }, + ); + let _ = ctx.send( + peer, + GossipMessage::Push { + from: self_addr, + state: self.state.clone(), + }, + ); + } else { + self.record(self_addr, GossipEventKind::GossipRoundNoPeers); + } + } + GossipMessage::Push { + from, + state: remote, + } => { + let keys_updated = self.state.merge(&remote); + self.record( + self_addr, + GossipEventKind::PushReceived { + from_name: self + .name_registry + .as_ref() + .map(|r| resolve_name(r, from)) + .unwrap_or_default(), + keys_updated, + }, + ); + } + GossipMessage::TakeSnapshot => { + self.record( + self_addr, + GossipEventKind::StateSnapshot { + snapshot: NodeSnapshot { + entries: self.state.entries().clone(), + peer_count: self.peers.len(), + }, + }, + ); + } + GossipMessage::Query { key, reply_to } => { + self.record( + self_addr, + GossipEventKind::QueryReceived { key: key.clone() }, + ); + let entry = self.state.get(&key); + let resp = GossipQueryResponse { + key, + value: entry.map(|e| e.value.clone()), + version: entry.map(|e| e.version), + }; + let _ = ctx.send(reply_to, resp); + } + } + } +} diff --git a/crates/swactor-gossip/src/lib.rs b/crates/swactor-gossip/src/lib.rs new file mode 100644 index 0000000..381ef56 --- /dev/null +++ b/crates/swactor-gossip/src/lib.rs @@ -0,0 +1,11 @@ +pub mod actor; +pub mod message; +pub mod state; +pub mod trace; + +pub mod report; +pub mod sim; + +pub use actor::GossipActor; +pub use message::{GossipMessage, GossipQueryResponse}; +pub use state::{GossipState, VersionedValue}; diff --git a/crates/swactor-gossip/src/message.rs b/crates/swactor-gossip/src/message.rs new file mode 100644 index 0000000..afee4ca --- /dev/null +++ b/crates/swactor-gossip/src/message.rs @@ -0,0 +1,34 @@ +use swactor::actor::ActorAddress; + +use crate::state::GossipState; + +#[derive(Debug, Clone)] +pub enum GossipMessage { + /// Register a peer to gossip with. + AddPeer(ActorAddress), + /// Remove a peer from the gossip set. + RemovePeer(ActorAddress), + /// Set a key-value pair in this node's local state. + Set { key: String, value: Vec }, + /// Trigger a gossip round: pick a random peer and push our full state. + DoGossipRound, + /// Incoming state push from a peer. + Push { + from: ActorAddress, + state: GossipState, + }, + /// Query the current value for a key; response sent to `reply_to`. + Query { + key: String, + reply_to: ActorAddress, + }, + /// Ask the actor to dump its current state into the event log (tracing only). + TakeSnapshot, +} + +#[derive(Debug, Clone)] +pub struct GossipQueryResponse { + pub key: String, + pub value: Option>, + pub version: Option, +} diff --git a/crates/swactor-gossip/src/report.rs b/crates/swactor-gossip/src/report.rs new file mode 100644 index 0000000..07eaaec --- /dev/null +++ b/crates/swactor-gossip/src/report.rs @@ -0,0 +1,550 @@ +use std::collections::HashMap; +use std::f64::consts::PI; + +use crate::trace::{GossipEventKind, SimulationTrace}; + +/// Generate a self-contained HTML report from a simulation trace. +pub fn generate_html_report(trace: &SimulationTrace) -> String { + let mut html = String::with_capacity(32_000); + + html.push_str("\n\n\n\n"); + html.push_str(&format!( + "Gossip Simulation: {}\n", + escape_html(&trace.name) + )); + html.push_str("\n\n\n"); + + html.push_str(&format!( + "

Gossip Simulation: {}

\n", + escape_html(&trace.name) + )); + + // Summary metrics + render_summary(&mut html, trace); + + // Network topology + render_topology_svg(&mut html, trace); + + // Propagation heatmap + render_heatmap_svg(&mut html, trace); + + // Convergence curve + render_convergence_svg(&mut html, trace); + + // Message flow timeline + render_message_flow_svg(&mut html, trace); + + // Event log table + render_event_table(&mut html, trace); + + html.push_str("\n\n"); + html +} + +// ── CSS ────────────────────────────────────────────────────────────────── + +const CSS: &str = r#" +body { + font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif; + max-width: 1200px; margin: 0 auto; padding: 20px; + background: #fafafa; color: #222; +} +h1 { border-bottom: 3px solid #333; padding-bottom: 8px; } +h2 { margin-top: 32px; color: #444; } +.metrics { display: flex; flex-wrap: wrap; gap: 16px; margin: 16px 0; } +.metric { + background: #fff; border: 1px solid #ddd; border-radius: 8px; + padding: 12px 20px; min-width: 140px; +} +.metric .label { font-size: 0.85em; color: #666; } +.metric .value { font-size: 1.5em; font-weight: bold; } +svg { display: block; margin: 12px 0; } +table { border-collapse: collapse; width: 100%; margin: 12px 0; } +th, td { border: 1px solid #ddd; padding: 6px 10px; text-align: left; font-size: 0.85em; } +th { background: #f0f0f0; } +tr:nth-child(even) { background: #fafafa; } +.capped { color: #999; font-style: italic; margin: 4px 0; } +"#; + +// ── Summary metrics ────────────────────────────────────────────────────── + +fn render_summary(html: &mut String, trace: &SimulationTrace) { + let num_nodes = trace.node_names.len(); + let total_pushes = trace + .events + .iter() + .filter(|e| matches!(e.kind, GossipEventKind::GossipRoundStarted { .. })) + .count(); + let redundant_pushes = trace + .events + .iter() + .filter(|e| matches!(e.kind, GossipEventKind::PushReceived { keys_updated: 0, .. })) + .count(); + let convergence_round = find_convergence_round(trace); + + html.push_str("

Summary

\n
\n"); + metric(html, "Nodes", &num_nodes.to_string()); + metric(html, "Keys", &trace.total_keys.to_string()); + metric(html, "Rounds", &trace.num_rounds.to_string()); + metric(html, "Pushes", &total_pushes.to_string()); + metric(html, "Redundant", &redundant_pushes.to_string()); + metric( + html, + "Converged at", + &convergence_round + .map(|r| format!("round {r}")) + .unwrap_or_else(|| "never".into()), + ); + if total_pushes > 0 { + let efficiency = 100.0 * (1.0 - redundant_pushes as f64 / total_pushes as f64); + metric(html, "Efficiency", &format!("{efficiency:.0}%")); + } + html.push_str("
\n"); +} + +fn metric(html: &mut String, label: &str, value: &str) { + html.push_str(&format!( + "
{label}
{value}
\n" + )); +} + +fn find_convergence_round(trace: &SimulationTrace) -> Option { + if trace.total_keys == 0 { + return Some(0); + } + for (round_idx, snapshots) in trace.snapshots_per_round.iter().enumerate() { + let all_converged = snapshots + .iter() + .all(|(_, snap)| snap.entries.len() >= trace.total_keys); + if all_converged { + return Some(round_idx + 1); + } + } + None +} + +// ── Topology SVG ───────────────────────────────────────────────────────── + +fn render_topology_svg(html: &mut String, trace: &SimulationTrace) { + html.push_str("

Network Topology

\n"); + + let n = trace.node_names.len(); + let size = 400.0_f64; + let cx = size / 2.0; + let cy = size / 2.0; + let radius = size / 2.0 - 50.0; + + // Compute node positions in circular layout. + let positions: Vec<(f64, f64)> = (0..n) + .map(|i| { + let angle = 2.0 * PI * (i as f64) / (n as f64) - PI / 2.0; + (cx + radius * angle.cos(), cy + radius * angle.sin()) + }) + .collect(); + + let name_to_idx: HashMap<&str, usize> = trace + .node_names + .iter() + .enumerate() + .map(|(i, n)| (n.as_str(), i)) + .collect(); + + html.push_str(&format!( + "\n" + )); + html.push_str("\n"); + + // Draw edges. + for (from_name, to_name) in &trace.topology_edges { + if let (Some(&fi), Some(&ti)) = (name_to_idx.get(from_name.as_str()), name_to_idx.get(to_name.as_str())) { + let (x1, y1) = positions[fi]; + let (x2, y2) = positions[ti]; + // Shorten line to not overlap circle. + let dx = x2 - x1; + let dy = y2 - y1; + let len = (dx * dx + dy * dy).sqrt(); + if len > 0.0 { + let nx = dx / len; + let ny = dy / len; + let sx = x1 + nx * 18.0; + let sy = y1 + ny * 18.0; + let ex = x2 - nx * 18.0; + let ey = y2 - ny * 18.0; + html.push_str(&format!( + "\n" + )); + } + } + } + + // Draw nodes. + for (i, name) in trace.node_names.iter().enumerate() { + let (x, y) = positions[i]; + html.push_str(&format!( + "\n" + )); + html.push_str(&format!( + "{name}\n", + ty = y + 4.0, + )); + } + + html.push_str("\n"); +} + +// ── Propagation heatmap ────────────────────────────────────────────────── + +fn render_heatmap_svg(html: &mut String, trace: &SimulationTrace) { + html.push_str("

Propagation Heatmap

\n"); + html.push_str("

Rows = nodes, columns = rounds. Color intensity = fraction of total keys held.

\n"); + + let n = trace.node_names.len(); + let rounds = trace.snapshots_per_round.len(); + if rounds == 0 || n == 0 { + html.push_str("

No data.

\n"); + return; + } + + let cell_w = 36.0_f64; + let cell_h = 28.0_f64; + let label_w = 80.0_f64; + let header_h = 28.0_f64; + let w = label_w + cell_w * rounds as f64 + 10.0; + let h = header_h + cell_h * n as f64 + 10.0; + + html.push_str(&format!( + "\n" + )); + + // Column headers. + for r in 0..rounds { + let x = label_w + r as f64 * cell_w + cell_w / 2.0; + let ty = header_h - 6.0; + let label = r + 1; + html.push_str(&format!( + "R{label}\n" + )); + } + + // Build a name→row index for stable ordering. + let name_to_row: HashMap<&str, usize> = trace + .node_names + .iter() + .enumerate() + .map(|(i, n)| (n.as_str(), i)) + .collect(); + + for (r, round_snaps) in trace.snapshots_per_round.iter().enumerate() { + for (name, snap) in round_snaps { + if let Some(&row) = name_to_row.get(name.as_str()) { + let frac = if trace.total_keys > 0 { + snap.entries.len() as f64 / trace.total_keys as f64 + } else { + 0.0 + }; + let x = label_w + r as f64 * cell_w; + let y = header_h + row as f64 * cell_h; + let color = heatmap_color(frac); + html.push_str(&format!( + "\n" + )); + // Show count inside cell. + let text_color = if frac > 0.5 { "#fff" } else { "#333" }; + let tx = x + cell_w / 2.0; + let ty = y + cell_h / 2.0 + 3.0; + let count = snap.entries.len(); + html.push_str(&format!( + "{count}\n" + )); + } + } + } + + // Row labels. + for (i, name) in trace.node_names.iter().enumerate() { + let y = header_h + i as f64 * cell_h + cell_h / 2.0 + 4.0; + html.push_str(&format!( + "{name}\n", + x = 4.0, + )); + } + + html.push_str("\n"); +} + +fn heatmap_color(frac: f64) -> String { + // Interpolate from light (#e8f4e8) to deep green (#1a7a1a). + let f = frac.clamp(0.0, 1.0); + let r = (232.0 + f * (26.0 - 232.0)) as u8; + let g = (244.0 + f * (122.0 - 244.0)) as u8; + let b = (232.0 + f * (26.0 - 232.0)) as u8; + format!("#{r:02x}{g:02x}{b:02x}") +} + +// ── Convergence curve ──────────────────────────────────────────────────── + +fn render_convergence_svg(html: &mut String, trace: &SimulationTrace) { + html.push_str("

Convergence Curve

\n"); + html.push_str("

Percentage of nodes that hold all keys vs. round number.

\n"); + + let rounds = trace.snapshots_per_round.len(); + if rounds == 0 { + html.push_str("

No data.

\n"); + return; + } + + let chart_w = 600.0_f64; + let chart_h = 300.0_f64; + let margin_l = 50.0_f64; + let margin_b = 40.0_f64; + let margin_t = 20.0_f64; + let margin_r = 20.0_f64; + let w = chart_w + margin_l + margin_r; + let h = chart_h + margin_t + margin_b; + + html.push_str(&format!( + "\n" + )); + + // Axes. + html.push_str(&format!( + "\n", + margin_t + chart_h, + )); + html.push_str(&format!( + "\n", + margin_t + chart_h, + margin_l + chart_w, + margin_t + chart_h, + )); + + // Y-axis labels. + for pct in [0, 25, 50, 75, 100] { + let y = margin_t + chart_h - (pct as f64 / 100.0) * chart_h; + html.push_str(&format!( + "{pct}%\n", + margin_l - 6.0, y + 3.0, + )); + html.push_str(&format!( + "\n", + margin_l + chart_w, + )); + } + + // X-axis labels. + let step = (rounds / 10).max(1); + for r in (0..rounds).step_by(step) { + let x = margin_l + (r as f64 + 0.5) / rounds as f64 * chart_w; + html.push_str(&format!( + "{}\n", + margin_t + chart_h + 16.0, r + 1, + )); + } + // X-axis title. + html.push_str(&format!( + "Round\n", + margin_l + chart_w / 2.0, + margin_t + chart_h + 34.0, + )); + + // Compute data points. + let n = trace.node_names.len(); + let mut points = Vec::with_capacity(rounds); + for round_snaps in &trace.snapshots_per_round { + let converged = round_snaps + .iter() + .filter(|(_, snap)| snap.entries.len() >= trace.total_keys && trace.total_keys > 0) + .count(); + let pct = if n > 0 { + converged as f64 / n as f64 * 100.0 + } else { + 0.0 + }; + points.push(pct); + } + + // Draw line. + let mut path = String::new(); + for (i, &pct) in points.iter().enumerate() { + let x = margin_l + (i as f64 + 0.5) / rounds as f64 * chart_w; + let y = margin_t + chart_h - (pct / 100.0) * chart_h; + if i == 0 { + path.push_str(&format!("M{x:.1},{y:.1}")); + } else { + path.push_str(&format!(" L{x:.1},{y:.1}")); + } + } + html.push_str(&format!( + "\n" + )); + + // Draw dots. + for (i, &pct) in points.iter().enumerate() { + let x = margin_l + (i as f64 + 0.5) / rounds as f64 * chart_w; + let y = margin_t + chart_h - (pct / 100.0) * chart_h; + html.push_str(&format!( + "\n" + )); + } + + html.push_str("\n"); +} + +// ── Message flow timeline ──────────────────────────────────────────────── + +fn render_message_flow_svg(html: &mut String, trace: &SimulationTrace) { + html.push_str("

Message Flow Timeline

\n"); + html.push_str("

Arrows show Push messages from sender to receiver, grouped by round.

\n"); + + let n = trace.node_names.len(); + let rounds = trace.num_rounds; + if n == 0 || rounds == 0 { + html.push_str("

No data.

\n"); + return; + } + + let name_to_col: HashMap<&str, usize> = trace + .node_names + .iter() + .enumerate() + .map(|(i, n)| (n.as_str(), i)) + .collect(); + + // Collect message arrows grouped by round. + let mut arrows_per_round: Vec> = vec![Vec::new(); rounds]; + for event in &trace.events { + if let GossipEventKind::PushReceived { ref from_name, .. } = event.kind { + let round_idx = event.tick.saturating_sub(1) as usize; + if round_idx < rounds { + if let (Some(&from_col), Some(&to_col)) = ( + name_to_col.get(from_name.as_str()), + name_to_col.get(event.node_name.as_str()), + ) { + arrows_per_round[round_idx].push((from_col, to_col)); + } + } + } + } + + let col_w = 80.0_f64; + let row_h = 40.0_f64; + let header_h = 30.0_f64; + let label_h = 24.0_f64; + let svg_w = col_w * n as f64 + 40.0; + let svg_h = header_h + label_h + row_h * rounds as f64 + 20.0; + + html.push_str(&format!( + "\n" + )); + html.push_str("\n"); + + // Column headers (node names). + for (i, name) in trace.node_names.iter().enumerate() { + let x = 20.0 + i as f64 * col_w + col_w / 2.0; + html.push_str(&format!( + "{name}\n" + )); + // Vertical lifeline. + let y_start = header_h + label_h; + let y_end = header_h + label_h + row_h * rounds as f64; + html.push_str(&format!( + "\n" + )); + } + + // Round labels and arrows. + for (r, arrows) in arrows_per_round.iter().enumerate() { + let y = header_h + label_h + r as f64 * row_h + row_h / 2.0; + // Round label on left. + html.push_str(&format!( + "R{}\n", + r + 1, + )); + + for &(from_col, to_col) in arrows { + let x1 = 20.0 + from_col as f64 * col_w + col_w / 2.0; + let x2 = 20.0 + to_col as f64 * col_w + col_w / 2.0; + // Offset slightly so overlapping arrows are visible. + let offset = if from_col < to_col { -3.0 } else { 3.0 }; + html.push_str(&format!( + "\n", + y1 = y + offset, + y2 = y + offset, + )); + } + } + + html.push_str("\n"); +} + +// ── Event log table ────────────────────────────────────────────────────── + +fn render_event_table(html: &mut String, trace: &SimulationTrace) { + html.push_str("

Event Log

\n"); + + let max_rows = 500; + let events: Vec<_> = trace + .events + .iter() + .filter(|e| !matches!(e.kind, GossipEventKind::StateSnapshot { .. })) + .collect(); + + let total = events.len(); + let display = events.iter().take(max_rows); + + html.push_str("\n\n"); + for event in display { + let (kind_str, detail) = format_event_kind(&event.kind); + html.push_str(&format!( + "\n", + event.tick, + escape_html(&event.node_name), + )); + } + html.push_str("
RoundNodeEventDetails
{}{}{kind_str}{detail}
\n"); + + if total > max_rows { + html.push_str(&format!( + "

Showing {max_rows} of {total} events.

\n" + )); + } +} + +fn format_event_kind(kind: &GossipEventKind) -> (&'static str, String) { + match kind { + GossipEventKind::LocalSet { key } => ("LocalSet", format!("key={}", escape_html(key))), + GossipEventKind::GossipRoundStarted { target_name } => { + ("GossipRound", format!("→ {}", escape_html(target_name))) + } + GossipEventKind::GossipRoundNoPeers => ("GossipRound", "no peers".into()), + GossipEventKind::PushReceived { + from_name, + keys_updated, + } => ( + "PushReceived", + format!( + "from {} ({keys_updated} updated)", + escape_html(from_name) + ), + ), + GossipEventKind::QueryReceived { key } => { + ("Query", format!("key={}", escape_html(key))) + } + GossipEventKind::PeerAdded { peer_name } => { + ("PeerAdded", escape_html(peer_name)) + } + GossipEventKind::PeerRemoved { peer_name } => { + ("PeerRemoved", escape_html(peer_name)) + } + GossipEventKind::StateSnapshot { .. } => ("Snapshot", String::new()), + } +} + +fn escape_html(s: &str) -> String { + s.replace('&', "&") + .replace('<', "<") + .replace('>', ">") + .replace('"', """) +} diff --git a/crates/swactor-gossip/src/sim.rs b/crates/swactor-gossip/src/sim.rs new file mode 100644 index 0000000..6c3e74f --- /dev/null +++ b/crates/swactor-gossip/src/sim.rs @@ -0,0 +1,236 @@ +use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; + +use swactor::actor::ActorAddress; +use swactor::config::RuntimeConfig; +use swactor::runtime::Runtime; + +use crate::actor::GossipActor; +use crate::message::GossipMessage; +use crate::trace::{ + EventLog, GossipEventKind, NameRegistry, NodeSnapshot, SimulationTrace, TickCounter, +}; + +// ── Configuration ──────────────────────────────────────────────────────── + +#[derive(Debug, Clone)] +pub enum Topology { + /// Each node gossips to the next; last gossips to first. + Ring, + /// Node 0 is the hub; all others gossip to/from it. + Star, + /// Every node gossips to every other node. + FullMesh, + /// Unidirectional chain: 0→1→2→…→(n-1). + Chain, + /// Two halves with no cross-links (healed later via `heal_after_round`). + Partitioned, +} + +#[derive(Debug, Clone)] +pub struct SimConfig { + pub name: String, + pub topology: Topology, + pub num_nodes: usize, + /// `(key, value)` pairs to set on node 0 before gossip starts. + pub initial_data: Vec<(String, Vec)>, + pub num_rounds: usize, + pub ticks_per_round: usize, + /// If `Some(r)`, cross-partition links are added after round `r`. + pub heal_after_round: Option, +} + +// ── Public entry point ─────────────────────────────────────────────────── + +pub fn run_simulation(config: SimConfig) -> SimulationTrace { + 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 rt = Runtime::new(RuntimeConfig { + num_threads: 1, + ..Default::default() + }); + + // Spawn nodes. + 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); + } + + // Wire topology. + let edges = wire_topology(&rt, &config.topology, &addrs, &names); + // Deliver AddPeer messages. + for _ in 0..3 { + rt.tick(); + } + + // Set initial data on node 0. + 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(); + + // Run gossip rounds. + let mut snapshots_per_round: Vec> = Vec::new(); + + for round in 0..config.num_rounds { + // Heal partition if needed. + if config.heal_after_round == Some(round) { + heal_partition(&rt, &config.topology, &addrs, &names); + for _ in 0..3 { + rt.tick(); + } + } + + tick_counter.store((round + 1) as u64, Ordering::Relaxed); + + // Trigger gossip on all nodes. + for &addr in &addrs { + rt.send_to(addr, GossipMessage::DoGossipRound).unwrap(); + } + for _ in 0..config.ticks_per_round { + rt.tick(); + } + + // Take snapshots. + for &addr in &addrs { + rt.send_to(addr, GossipMessage::TakeSnapshot).unwrap(); + } + for _ in 0..3 { + rt.tick(); + } + + // Extract snapshots from event log. + 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); + } + + 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, + } +} + +// ── Topology wiring ────────────────────────────────────────────────────── + +fn wire_topology( + rt: &Runtime, + topology: &Topology, + addrs: &[ActorAddress], + names: &[String], +) -> Vec<(String, String)> { + let n = addrs.len(); + let mut edges = Vec::new(); + + let mut add_edge = |from: usize, to: usize| { + rt.send_to(addrs[from], GossipMessage::AddPeer(addrs[to])) + .unwrap(); + edges.push((names[from].clone(), names[to].clone())); + }; + + match topology { + Topology::Ring => { + for i in 0..n { + add_edge(i, (i + 1) % n); + } + } + Topology::Star => { + for i in 1..n { + add_edge(0, i); + add_edge(i, 0); + } + } + Topology::FullMesh => { + for i in 0..n { + for j in 0..n { + if i != j { + add_edge(i, j); + } + } + } + } + Topology::Chain => { + for i in 0..n.saturating_sub(1) { + add_edge(i, i + 1); + } + } + Topology::Partitioned => { + let half = n / 2; + // Wire each half as a full mesh. + for i in 0..half { + for j in 0..half { + if i != j { + add_edge(i, j); + } + } + } + for i in half..n { + for j in half..n { + if i != j { + add_edge(i, j); + } + } + } + } + } + edges +} + +fn heal_partition( + rt: &Runtime, + topology: &Topology, + addrs: &[ActorAddress], + _names: &[String], +) { + if !matches!(topology, Topology::Partitioned) { + return; + } + let n = addrs.len(); + let half = n / 2; + // 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(); + } +} diff --git a/crates/swactor-gossip/src/state.rs b/crates/swactor-gossip/src/state.rs new file mode 100644 index 0000000..44a2cc7 --- /dev/null +++ b/crates/swactor-gossip/src/state.rs @@ -0,0 +1,69 @@ +use std::collections::HashMap; + +#[derive(Debug, Clone)] +pub struct VersionedValue { + pub value: Vec, + pub version: u64, +} + +#[derive(Debug, Clone, Default)] +pub struct GossipState { + entries: HashMap, +} + +impl GossipState { + pub fn new() -> Self { + Self::default() + } + + /// Insert or update a key. Auto-increments the version for that key. + /// Returns the new version number. + pub fn set(&mut self, key: String, value: Vec) -> u64 { + let new_version = self + .entries + .get(&key) + .map_or(1, |existing| existing.version + 1); + self.entries.insert( + key, + VersionedValue { + value, + version: new_version, + }, + ); + new_version + } + + pub fn get(&self, key: &str) -> Option<&VersionedValue> { + self.entries.get(key) + } + + pub fn entries(&self) -> &HashMap { + &self.entries + } + + pub fn len(&self) -> usize { + self.entries.len() + } + + pub fn is_empty(&self) -> bool { + self.entries.is_empty() + } + + /// Merge a remote state into this one. For each key, keep the entry + /// with the higher version (last-writer-wins). Returns the number of + /// entries that were updated. + pub fn merge(&mut self, remote: &GossipState) -> usize { + let mut updated = 0; + for (key, remote_val) in &remote.entries { + let dominated = match self.entries.get(key) { + Some(local_val) => remote_val.version > local_val.version, + None => true, + }; + if dominated { + self.entries.insert(key.clone(), remote_val.clone()); + updated += 1; + } + } + updated + } +} diff --git a/crates/swactor-gossip/src/trace.rs b/crates/swactor-gossip/src/trace.rs new file mode 100644 index 0000000..2f155d5 --- /dev/null +++ b/crates/swactor-gossip/src/trace.rs @@ -0,0 +1,90 @@ +use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; + +use swactor::actor::ActorAddress; + +use crate::state::VersionedValue; + +// ── Shared handles ─────────────────────────────────────────────────────── + +/// Shared, append-only event log. +pub type EventLog = Arc>>; + +/// Shared tick counter — the simulation harness increments this. +pub type TickCounter = Arc; + +/// Maps actor addresses to human-readable names like `"node-0"`. +pub type NameRegistry = Arc>>; + +// ── Event types ────────────────────────────────────────────────────────── + +#[derive(Debug, Clone)] +pub struct GossipEvent { + pub tick: u64, + pub node_name: String, + pub node_addr: ActorAddress, + pub kind: GossipEventKind, +} + +#[derive(Debug, Clone)] +pub enum GossipEventKind { + /// A local `Set { key, .. }` was processed. + LocalSet { key: String }, + /// `DoGossipRound` chose a peer and sent a Push. + GossipRoundStarted { target_name: String }, + /// `DoGossipRound` had no peers. + GossipRoundNoPeers, + /// Received a Push from another node. + PushReceived { + from_name: String, + keys_updated: usize, + }, + /// Received a Query. + QueryReceived { key: String }, + /// A peer was added. + PeerAdded { peer_name: String }, + /// A peer was removed. + PeerRemoved { peer_name: String }, + /// Full state snapshot (requested via `TakeSnapshot`). + StateSnapshot { snapshot: NodeSnapshot }, +} + +#[derive(Debug, Clone)] +pub struct NodeSnapshot { + pub entries: HashMap, + pub peer_count: usize, +} + +// ── Simulation trace (complete run output) ─────────────────────────────── + +#[derive(Debug, Clone)] +pub struct SimulationTrace { + pub name: String, + pub node_names: Vec, + pub node_addrs: Vec, + pub topology_edges: Vec<(String, String)>, + pub events: Vec, + pub snapshots_per_round: Vec>, + pub num_rounds: usize, + pub total_keys: usize, +} + +// ── Helpers ────────────────────────────────────────────────────────────── + +pub fn current_tick(counter: &TickCounter) -> u64 { + counter.load(Ordering::Relaxed) +} + +pub fn resolve_name(registry: &NameRegistry, addr: ActorAddress) -> String { + registry + .lock() + .unwrap() + .get(&addr) + .cloned() + .unwrap_or_else(|| format!("{:?}", &addr.0[..4])) +} + +pub fn record_event(log: &EventLog, event: GossipEvent) { + log.lock().unwrap().push(event); +} diff --git a/crates/swactor-gossip/tests/gossip_convergence.rs b/crates/swactor-gossip/tests/gossip_convergence.rs new file mode 100644 index 0000000..593443c --- /dev/null +++ b/crates/swactor-gossip/tests/gossip_convergence.rs @@ -0,0 +1,243 @@ +use swactor::config::RuntimeConfig; +use swactor::runtime::Runtime; +use swactor_gossip::{GossipActor, GossipMessage, GossipQueryResponse}; + +fn single_thread_runtime() -> Runtime { + Runtime::new(RuntimeConfig { + num_threads: 1, + ..Default::default() + }) +} + +/// Drive ticks until the inbox receives a response, or panic after a limit. +fn recv_query_response( + rt: &Runtime, + inbox: &swactor::runtime::Inbox, + max_ticks: usize, +) -> GossipQueryResponse { + for _ in 0..max_ticks { + rt.tick(); + if let Some(resp) = inbox.try_recv() { + return resp; + } + } + panic!("no GossipQueryResponse after {max_ticks} ticks"); +} + +// ──────────────────────────────────────────────────────────────────────────── +// Test 1: Value propagates through a chain A → B → C +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn value_propagates_through_chain() { + // Given: three gossip nodes wired A→B→C (each only gossips to the next) + let rt = single_thread_runtime(); + let a = rt.spawn(GossipActor::new()).unwrap(); + let b = rt.spawn(GossipActor::new()).unwrap(); + let c = rt.spawn(GossipActor::new()).unwrap(); + + rt.send_to(a, GossipMessage::AddPeer(b)).unwrap(); + rt.send_to(b, GossipMessage::AddPeer(c)).unwrap(); + rt.tick(); // deliver AddPeer messages + + // When: we set a value on A and trigger gossip hops + rt.send_to(a, GossipMessage::Set { + key: "color".into(), + value: b"blue".to_vec(), + }) + .unwrap(); + rt.tick(); // A processes Set + + rt.send_to(a, GossipMessage::DoGossipRound).unwrap(); + rt.tick(); // A pushes to B + rt.tick(); // B processes Push + + rt.send_to(b, GossipMessage::DoGossipRound).unwrap(); + rt.tick(); // B pushes to C + rt.tick(); // C processes Push + + // Then: querying C returns the value that originated at A + let inbox = rt.new_inbox::().unwrap(); + rt.send_to(c, GossipMessage::Query { + key: "color".into(), + reply_to: *inbox.addr(), + }) + .unwrap(); + + let resp = recv_query_response(&rt, &inbox, 10); + assert_eq!(resp.key, "color"); + assert_eq!(resp.value.as_deref(), Some(b"blue".as_slice())); + assert_eq!(resp.version, Some(1)); +} + +// ──────────────────────────────────────────────────────────────────────────── +// Test 2: Higher version wins during merge +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn higher_version_wins() { + // Given: two nodes A and B, each set the same key at different versions + let rt = single_thread_runtime(); + let a = rt.spawn(GossipActor::new()).unwrap(); + let b = rt.spawn(GossipActor::new()).unwrap(); + + rt.send_to(a, GossipMessage::AddPeer(b)).unwrap(); + rt.tick(); + + // A sets "x" once (version 1) + rt.send_to(a, GossipMessage::Set { + key: "x".into(), + value: b"old".to_vec(), + }) + .unwrap(); + rt.tick(); + + // B sets "x" three times (version 3) + for val in [b"v1".as_slice(), b"v2", b"new"] { + rt.send_to(b, GossipMessage::Set { + key: "x".into(), + value: val.to_vec(), + }) + .unwrap(); + } + rt.tick(); + + // When: A pushes its lower-version state to B + rt.send_to(a, GossipMessage::DoGossipRound).unwrap(); + rt.tick(); // A sends Push + rt.tick(); // B receives Push + + // Then: B still has the higher-version value + let inbox = rt.new_inbox::().unwrap(); + rt.send_to(b, GossipMessage::Query { + key: "x".into(), + reply_to: *inbox.addr(), + }) + .unwrap(); + + let resp = recv_query_response(&rt, &inbox, 10); + assert_eq!(resp.value.as_deref(), Some(b"new".as_slice())); + assert_eq!(resp.version, Some(3)); +} + +// ──────────────────────────────────────────────────────────────────────────── +// Test 3: Disjoint keys merge — both nodes end up with both keys +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn disjoint_keys_merge() { + // Given: A owns key "a", B owns key "b", they are mutual peers + let rt = single_thread_runtime(); + let a = rt.spawn(GossipActor::new()).unwrap(); + let b = rt.spawn(GossipActor::new()).unwrap(); + + rt.send_to(a, GossipMessage::AddPeer(b)).unwrap(); + rt.send_to(b, GossipMessage::AddPeer(a)).unwrap(); + rt.tick(); + + rt.send_to(a, GossipMessage::Set { + key: "a".into(), + value: b"from-a".to_vec(), + }) + .unwrap(); + rt.send_to(b, GossipMessage::Set { + key: "b".into(), + value: b"from-b".to_vec(), + }) + .unwrap(); + rt.tick(); + + // When: both gossip to each other + rt.send_to(a, GossipMessage::DoGossipRound).unwrap(); + rt.send_to(b, GossipMessage::DoGossipRound).unwrap(); + rt.tick(); // send Pushes + rt.tick(); // receive Pushes + + // Then: A has key "b" and B has key "a" + let inbox = rt.new_inbox::().unwrap(); + + rt.send_to(a, GossipMessage::Query { + key: "b".into(), + reply_to: *inbox.addr(), + }) + .unwrap(); + let resp = recv_query_response(&rt, &inbox, 10); + assert_eq!(resp.key, "b"); + assert_eq!(resp.value.as_deref(), Some(b"from-b".as_slice())); + + rt.send_to(b, GossipMessage::Query { + key: "a".into(), + reply_to: *inbox.addr(), + }) + .unwrap(); + let resp = recv_query_response(&rt, &inbox, 10); + assert_eq!(resp.key, "a"); + assert_eq!(resp.value.as_deref(), Some(b"from-a".as_slice())); +} + +// ──────────────────────────────────────────────────────────────────────────── +// Test 4: Query for nonexistent key returns None +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn query_nonexistent_key_returns_none() { + // Given: a gossip node with no data + let rt = single_thread_runtime(); + let a = rt.spawn(GossipActor::new()).unwrap(); + rt.tick(); + + // When: we query a key that was never set + let inbox = rt.new_inbox::().unwrap(); + rt.send_to(a, GossipMessage::Query { + key: "ghost".into(), + reply_to: *inbox.addr(), + }) + .unwrap(); + + // Then: response has None value and None version + let resp = recv_query_response(&rt, &inbox, 10); + assert_eq!(resp.key, "ghost"); + assert!(resp.value.is_none()); + assert!(resp.version.is_none()); +} + +// ──────────────────────────────────────────────────────────────────────────── +// Test 5: Idempotent push — double-push doesn't bump versions +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn idempotent_push() { + // Given: A has a key set, B is its peer + let rt = single_thread_runtime(); + let a = rt.spawn(GossipActor::new()).unwrap(); + let b = rt.spawn(GossipActor::new()).unwrap(); + + rt.send_to(a, GossipMessage::AddPeer(b)).unwrap(); + rt.tick(); + + rt.send_to(a, GossipMessage::Set { + key: "k".into(), + value: b"val".to_vec(), + }) + .unwrap(); + rt.tick(); + + // When: A gossips to B twice (same state, same version) + for _ in 0..2 { + rt.send_to(a, GossipMessage::DoGossipRound).unwrap(); + rt.tick(); // send Push + rt.tick(); // receive Push + } + + // Then: B's version is still 1 (merge is idempotent, not additive) + let inbox = rt.new_inbox::().unwrap(); + rt.send_to(b, GossipMessage::Query { + key: "k".into(), + reply_to: *inbox.addr(), + }) + .unwrap(); + + let resp = recv_query_response(&rt, &inbox, 10); + assert_eq!(resp.version, Some(1)); + assert_eq!(resp.value.as_deref(), Some(b"val".as_slice())); +}