From cd1816e3985ce0bfc52427935ab6ed6f5582e9fd Mon Sep 17 00:00:00 2001 From: zacheryasc Date: Fri, 13 Feb 2026 13:18:38 +0000 Subject: [PATCH] feat: distribution simulation tests (#34) Test cluster behavior in simulation. --- .gitignore | 6 +- crates/distribution/src/node.rs | 50 +- crates/distribution/src/registry.rs | 11 + crates/distribution/src/swim/node.rs | 69 +- .../src/dashboard_html.rs | 497 +++++++++--- crates/simulation-dashboard/src/server.rs | 7 + .../simulation/src/distribution/properties.rs | 170 ++++ crates/simulation/src/distribution/sim.rs | 156 +++- crates/simulation/src/distribution/trace.rs | 5 + crates/simulation/src/trace.rs | 17 +- crates/simulation/tests/cluster_scenarios.rs | 25 +- .../tests/distribution_lifecycle.rs | 615 ++++++++++++++ .../tests/distribution_properties.rs | 579 ++++++++++++++ .../simulation/tests/distribution_registry.rs | 752 ++++++++++++++++++ scripts/sim-dashboard.sh | 19 + 15 files changed, 2799 insertions(+), 179 deletions(-) create mode 100644 crates/simulation/tests/distribution_lifecycle.rs create mode 100644 crates/simulation/tests/distribution_properties.rs create mode 100644 crates/simulation/tests/distribution_registry.rs create mode 100755 scripts/sim-dashboard.sh diff --git a/.gitignore b/.gitignore index 7c2c3e9..5b6f904 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,3 @@ -CLAUDE/ **/target **/node_modules/ .vscode/ @@ -13,6 +12,5 @@ corpus docs/architecture.dot docs/architecture.html -# Claude session files -CLAUDE/ -.claude/ +# Simulation traces +crates/simulation/traces \ No newline at end of file diff --git a/crates/distribution/src/node.rs b/crates/distribution/src/node.rs index 6dc9a5f..3d7d87b 100644 --- a/crates/distribution/src/node.rs +++ b/crates/distribution/src/node.rs @@ -14,7 +14,7 @@ use crate::kademlia::repair::{RepairQueue, RepublishTracker}; use crate::kademlia::routing_table::RoutingTable; use crate::registry::{ pack_combined_piggyback, unpack_combined_piggyback, ClusterRegistry, RegistryConfig, - RegistryEvent, + RegistryEntry, RegistryEvent, }; use crate::swim::node::{NodeAction, SwimNode}; use crate::swim::probe::SwimConfig; @@ -137,17 +137,7 @@ impl DistributedNode { let actions = self.swim.tick(); // Process membership changes from SWIM - let membership_changes: Vec<_> = actions - .iter() - .filter_map(|a| match a { - NodeAction::MembershipChanged { node_id, state, .. } => Some((*node_id, *state)), - _ => None, - }) - .collect(); - - for (node_id, state) in membership_changes { - self.handle_membership_change(node_id, state); - } + self.process_membership_changes(&actions); // Periodic republish let to_republish = self.republish.tick(self.tick_count); @@ -167,21 +157,30 @@ impl DistributedNode { // ─── SWIM message handling (delegate to SwimNode) ─────────────────── pub fn handle_ping(&mut self, from: NodeId, from_addr: SocketAddr, sequence: u64, piggyback: &[u8]) -> Vec { - let membership_bytes = self.extract_registry_piggyback(piggyback); + let (membership_bytes, registry_entries) = unpack_combined_piggyback(piggyback); let actions = self.swim.handle_ping(from, from_addr, sequence, &membership_bytes); + // Process membership BEFORE merging registry — otherwise a death + // notification in this same piggyback would immediately tombstone + // freshly received registry entries instead of pre-existing ones. + self.process_membership_changes(&actions); + self.merge_registry_entries(registry_entries); self.maybe_update_routing_table(from, from_addr); self.inject_registry_piggyback(actions) } pub fn handle_ack(&mut self, from: NodeId, sequence: u64, piggyback: &[u8]) -> Vec { - let membership_bytes = self.extract_registry_piggyback(piggyback); + let (membership_bytes, registry_entries) = unpack_combined_piggyback(piggyback); let actions = self.swim.handle_ack(from, sequence, &membership_bytes); + self.process_membership_changes(&actions); + self.merge_registry_entries(registry_entries); self.inject_registry_piggyback(actions) } pub fn handle_ping_req(&mut self, from: NodeId, target: NodeId, target_addr: SocketAddr, sequence: u64, piggyback: &[u8]) -> Vec { - let membership_bytes = self.extract_registry_piggyback(piggyback); + let (membership_bytes, registry_entries) = unpack_combined_piggyback(piggyback); let actions = self.swim.handle_ping_req(from, target, target_addr, sequence, &membership_bytes); + self.process_membership_changes(&actions); + self.merge_registry_entries(registry_entries); self.inject_registry_piggyback(actions) } @@ -310,12 +309,23 @@ impl DistributedNode { self.routing_table.insert(node_id, addr); } + fn process_membership_changes(&mut self, actions: &[NodeAction]) { + for action in actions { + if let NodeAction::MembershipChanged { node_id, state, .. } = action { + self.handle_membership_change(*node_id, *state); + } + } + } + fn handle_membership_change(&mut self, node_id: NodeId, state: MemberState) { match state { MemberState::Alive => { if let Some(entry) = self.swim.members().get(&node_id) { self.routing_table.insert(node_id, entry.addr); } + // Re-disseminate registry entries so the recovering node + // catches up on state accumulated during the partition. + self.registry.re_disseminate_all(self.cluster_size()); } MemberState::Dead => { self.routing_table.remove(&node_id); @@ -358,13 +368,11 @@ impl DistributedNode { .collect() } - /// Extract registry entries from incoming piggyback, merge them, return membership-only bytes. - fn extract_registry_piggyback(&mut self, bytes: &[u8]) -> Vec { - let (membership_bytes, registry_entries) = unpack_combined_piggyback(bytes); - if !registry_entries.is_empty() { - self.registry.merge_batch(registry_entries, self.cluster_size()); + /// Merge registry entries received from a piggyback payload. + fn merge_registry_entries(&mut self, entries: Vec) { + if !entries.is_empty() { + self.registry.merge_batch(entries, self.cluster_size()); } - membership_bytes } } diff --git a/crates/distribution/src/registry.rs b/crates/distribution/src/registry.rs index 377c257..72f3690 100644 --- a/crates/distribution/src/registry.rs +++ b/crates/distribution/src/registry.rs @@ -235,6 +235,17 @@ impl ClusterRegistry { } } + /// Re-enqueue all entries for dissemination (anti-entropy on membership change). + /// + /// Called when a previously-dead node comes back alive, ensuring that + /// registry state accumulated during a partition is gossiped to the + /// recovering node. + pub fn re_disseminate_all(&mut self, cluster_size: usize) { + for entry in self.entries.values().cloned().collect::>() { + self.enqueue(entry, cluster_size); + } + } + /// Periodic GC: remove tombstones past TTL with exhausted dissemination budgets. pub fn gc_tick(&mut self) { self.tick_count += 1; diff --git a/crates/distribution/src/swim/node.rs b/crates/distribution/src/swim/node.rs index 94c3dc8..6abffc6 100644 --- a/crates/distribution/src/swim/node.rs +++ b/crates/distribution/src/swim/node.rs @@ -85,29 +85,31 @@ impl SwimNode { /// Handle a received ping. pub fn handle_ping(&mut self, from: NodeId, from_addr: SocketAddr, sequence: u64, piggyback: &[u8]) -> Vec { - self.apply_piggyback(piggyback); + let mut actions = self.apply_piggyback(piggyback); // Ensure the sender is in our member list self.members.apply(from, from_addr, MemberState::Alive, 0); // Reply with ack let pb = self.dissemination.pack_piggyback(self.max_piggyback); - vec![NodeAction::SendAck { + actions.push(NodeAction::SendAck { to: from, to_addr: from_addr, sequence, piggyback: pb, - }] + }); + actions } /// Handle a received ack. pub fn handle_ack(&mut self, from: NodeId, sequence: u64, piggyback: &[u8]) -> Vec { - self.apply_piggyback(piggyback); + let mut actions = self.apply_piggyback(piggyback); let probe_actions = self.probe.step( SwimEvent::AckReceived { from, sequence }, &mut self.members, ); - self.translate_probe_actions(probe_actions) + actions.extend(self.translate_probe_actions(probe_actions)); + actions } /// Handle a received indirect ping request. @@ -119,16 +121,17 @@ impl SwimNode { sequence: u64, piggyback: &[u8], ) -> Vec { - self.apply_piggyback(piggyback); + let mut actions = self.apply_piggyback(piggyback); // Forward a ping to the target on behalf of the requester let pb = self.dissemination.pack_piggyback(self.max_piggyback); - vec![NodeAction::SendPing { + actions.push(NodeAction::SendPing { to: target, to_addr: target_addr, sequence, piggyback: pb, - }] + }); + actions } /// Handle a join request from a new node. @@ -214,14 +217,16 @@ impl SwimNode { self.members.alive_count() + 1 // +1 for self } - fn apply_piggyback(&mut self, bytes: &[u8]) { + fn apply_piggyback(&mut self, bytes: &[u8]) -> Vec { let updates = DisseminationQueue::unpack_piggyback(bytes); + let mut actions = Vec::new(); for update in updates { - self.apply_membership_update(update); + actions.extend(self.apply_membership_update(update)); } + actions } - fn apply_membership_update(&mut self, update: MembershipUpdate) { + fn apply_membership_update(&mut self, update: MembershipUpdate) -> Vec { // Check if this is about us if update.node_id == self.members.self_id() { if update.state == MemberState::Suspect || update.state == MemberState::Dead { @@ -237,7 +242,7 @@ impl SwimNode { self.cluster_size(), ); } - return; + return Vec::new(); } let changed = self.members.apply( @@ -252,6 +257,13 @@ impl SwimNode { membership_update(update.node_id, update.addr, update.state, update.incarnation), self.cluster_size(), ); + vec![NodeAction::MembershipChanged { + node_id: update.node_id, + state: update.state, + incarnation: update.incarnation, + }] + } else { + Vec::new() } } @@ -260,14 +272,14 @@ impl SwimNode { for pa in probe_actions { match pa { SwimAction::SendPing { to, to_addr, sequence } => { - // If the target is dead, re-enqueue the death declaration + // If the target is suspect or dead, re-enqueue its state // so it piggybacks on this message. This is the key mechanism - // for partition-heal recovery: the dead node learns it was - // declared dead and refutes by bumping its incarnation. + // for partition-heal recovery: the target learns it was + // suspected/declared dead and refutes by bumping its incarnation. if let Some(entry) = self.members.get(&to) { - if entry.state == MemberState::Dead { + if entry.state == MemberState::Dead || entry.state == MemberState::Suspect { self.dissemination.enqueue( - membership_update(to, to_addr, MemberState::Dead, entry.incarnation), + membership_update(to, to_addr, entry.state, entry.incarnation), self.cluster_size(), ); } @@ -307,20 +319,21 @@ impl SwimNode { } } SwimAction::DeclareDead(node_id) => { + // Note: declare_dead() was already called by SwimProbe::check_suspicion_timeouts(), + // so we must NOT call it again (it would return false since state is already Dead). + // We just need to disseminate the update and emit the MembershipChanged action. if let Some(entry) = self.members.get(&node_id) { let inc = entry.incarnation; let addr = entry.addr; - if self.members.declare_dead(node_id) { - self.dissemination.enqueue( - membership_update(node_id, addr, MemberState::Dead, inc), - self.cluster_size(), - ); - actions.push(NodeAction::MembershipChanged { - node_id, - state: MemberState::Dead, - incarnation: inc, - }); - } + self.dissemination.enqueue( + membership_update(node_id, addr, MemberState::Dead, inc), + self.cluster_size(), + ); + actions.push(NodeAction::MembershipChanged { + node_id, + state: MemberState::Dead, + incarnation: inc, + }); } } SwimAction::Refute { new_incarnation } => { diff --git a/crates/simulation-dashboard/src/dashboard_html.rs b/crates/simulation-dashboard/src/dashboard_html.rs index fa84045..6885af1 100644 --- a/crates/simulation-dashboard/src/dashboard_html.rs +++ b/crates/simulation-dashboard/src/dashboard_html.rs @@ -45,6 +45,11 @@ pub const DASHBOARD_HTML: &str = r##" td { padding: 3px 8px; border-bottom: 1px solid #1a1d2e; white-space: nowrap; } tr.highlight-push td { background: rgba(59,130,246,0.1); } tr.highlight-set td { background: rgba(34,197,94,0.1); } + tr.highlight-kill td { background: rgba(239,68,68,0.15); } + tr.highlight-revive td { background: rgba(34,197,94,0.15); } + tr.highlight-membership td { background: rgba(245,158,11,0.1); } + tr.highlight-registry td { background: rgba(168,85,247,0.1); } + tr.highlight-ping td { background: rgba(59,130,246,0.08); } .replay-controls { display: flex; align-items: center; gap: 8px; padding: 8px 16px; background: #161822; border-bottom: 1px solid #2a2d3a; } .replay-controls button { background: #2a2d3a; color: #e0e0e0; border: none; border-radius: 4px; padding: 4px 10px; cursor: pointer; font-size: 13px; } .replay-controls button:hover { background: #3b3f52; } @@ -217,6 +222,17 @@ let cumulPushRecv = []; // round number for each event index: eventRoundIdx[evtIdx] = roundIdx into snapRounds let eventRoundMap = []; // eventRoundMap[evtIdx] = tick +// ── Distribution-specific state ──────────────────────────────── +let isDist = false; +let snapMembers = []; // snapMembers[roundIdx] = Int32Array(N) — member_count +let snapRegistry = []; // snapRegistry[roundIdx] = Int32Array(N) — registry_size +let snapCache = []; // snapCache[roundIdx] = Int32Array(N) — cache_size +let snapAlive = []; // snapAlive[roundIdx] = Uint8Array(N) — is_alive +let snapRouting = []; // snapRouting[roundIdx] = Int32Array(N) — routing_table_size +let snapRepair = []; // snapRepair[roundIdx] = Int32Array(N) — repair_queue_size +let snapDirectory = []; // snapDirectory[roundIdx] = Int32Array(N) — directory_entry_count +let snapTombstone = []; // snapTombstone[roundIdx] = Int32Array(N) — registry_tombstone_count + // ── Community detection ────────────────────────────────────────── let community = new Int32Array(0); // community[nodeIdx] = community id let numCommunities = 0; @@ -492,11 +508,41 @@ function showNodeDetail(ni) { } let html = ''; - html += '
Community' + (community[ni] !== undefined ? community[ni] : '-') + '
'; - html += '
Pushes Sent' + (metricPushesSent[ni] || 0) + '
'; - html += '
Pushes Recv' + (metricPushesRecv[ni] || 0) + '
'; - html += '
Keys' + keyCount + '/' + totalKeys + '
'; - html += '
Peers' + peerCount + '
'; + if (isDist) { + // Find latest distribution snapshot for this node + let snap = null; + for (let r = snapRounds.length - 1; r >= 0; r--) { + if (snapRounds[r] <= curRound) { + snap = { + members: snapMembers[r][ni] || 0, + registry: snapRegistry[r][ni] || 0, + cache: snapCache[r][ni] || 0, + alive: snapAlive[r][ni], + routing: snapRouting[r][ni] || 0, + repair: snapRepair[r][ni] || 0, + directory: snapDirectory[r][ni] || 0, + tombstones: snapTombstone[r][ni] || 0, + }; + break; + } + } + if (snap) { + html += '
Status' + (snap.alive ? 'Alive' : 'Dead') + '
'; + html += '
Members' + snap.members + '
'; + html += '
Routing Table' + snap.routing + '
'; + html += '
Directory' + snap.directory + '
'; + html += '
Cache' + snap.cache + '
'; + html += '
Registry' + snap.registry + '
'; + html += '
Tombstones' + snap.tombstones + '
'; + html += '
Repair Queue' + snap.repair + '
'; + } + } else { + html += '
Community' + (community[ni] !== undefined ? community[ni] : '-') + '
'; + html += '
Pushes Sent' + (metricPushesSent[ni] || 0) + '
'; + html += '
Pushes Recv' + (metricPushesRecv[ni] || 0) + '
'; + html += '
Keys' + keyCount + '/' + totalKeys + '
'; + html += '
Peers' + peerCount + '
'; + } document.getElementById('ndStats').innerHTML = html; // Mini event log: last 20 events for this node up to cursor @@ -584,6 +630,61 @@ function drawAllCharts() { function drawBadges() { const grid = document.getElementById('badgeGrid'); + + if (isDist) { + const hasSnaps = snapRounds.length > 0; + const lastRound = hasSnaps ? snapRounds.length - 1 : -1; + + // Alive count + let aliveCount = N; + if (lastRound >= 0) { + aliveCount = 0; + for (let i = 0; i < N; i++) if (snapAlive[lastRound][i]) aliveCount++; + } + + // Membership accuracy: fraction of alive nodes with correct member_count + let memAccuracy = 1.0; + if (lastRound >= 0) { + let correct = 0, alive = 0; + for (let i = 0; i < N; i++) { + if (!snapAlive[lastRound][i]) continue; + alive++; + if (snapMembers[lastRound][i] >= aliveCount - 1) correct++; + } + memAccuracy = alive > 0 ? correct / alive : 1; + } + + // Registry max + let maxReg = 0; + if (lastRound >= 0) { + for (let i = 0; i < N; i++) if (snapRegistry[lastRound][i] > maxReg) maxReg = snapRegistry[lastRound][i]; + } + + // Cache total + let totalCache = 0; + if (lastRound >= 0) { + for (let i = 0; i < N; i++) if (snapAlive[lastRound][i]) totalCache += snapCache[lastRound][i]; + } + + // Repair queue total + let totalRepair = 0; + if (lastRound >= 0) { + for (let i = 0; i < N; i++) if (snapAlive[lastRound][i]) totalRepair += snapRepair[lastRound][i]; + } + + const memColor = memAccuracy >= 0.99 ? 'green' : memAccuracy >= 0.8 ? 'yellow' : 'red'; + const aliveColor = aliveCount === N ? 'green' : aliveCount >= N * 0.8 ? 'yellow' : 'red'; + + grid.innerHTML = + badge(aliveColor, aliveCount + '/' + N, 'Alive Nodes') + + badge(memColor, (memAccuracy * 100).toFixed(0) + '%', 'Membership') + + badge('', maxReg, 'Registry Size') + + badge('', totalCache, 'Cache Total') + + badge(totalRepair > 0 ? 'yellow' : 'green', totalRepair, 'Repair Queue') + + badge('', allEvents.length, 'Events'); + return; + } + const hasSnaps = snapRounds.length > 0; // Delivery ratio @@ -676,10 +777,26 @@ function drawConvergenceChart() { // Compute data points const pts = []; - for (let r = 0; r < snapRounds.length; r++) { - let full = 0; - for (let i = 0; i < N; i++) if (snapEntries[r][i] >= totalKeys) full++; - pts.push({ round: snapRounds[r], pct: N > 0 ? full / N * 100 : 0 }); + if (isDist) { + // Membership convergence: fraction of alive nodes with correct member count per round + for (let r = 0; r < snapRounds.length; r++) { + let alive = 0, correct = 0; + let aliveCount = 0; + for (let i = 0; i < N; i++) if (snapAlive[r][i]) aliveCount++; + for (let i = 0; i < N; i++) { + if (!snapAlive[r][i]) continue; + alive++; + if (snapMembers[r][i] >= aliveCount - 1) correct++; + } + pts.push({ round: snapRounds[r], pct: alive > 0 ? correct / alive * 100 : 0 }); + } + } else { + // existing gossip convergence + for (let r = 0; r < snapRounds.length; r++) { + let full = 0; + for (let i = 0; i < N; i++) if (snapEntries[r][i] >= totalKeys) full++; + pts.push({ round: snapRounds[r], pct: N > 0 ? full / N * 100 : 0 }); + } } // Draw fill @@ -803,7 +920,12 @@ function drawHeatmap() { const ni = sortedIdx[row]; const entries = snapEntries[col][ni] || 0; const pct = totalKeys > 0 ? Math.round(entries / totalKeys * 100) : 0; - tooltip.textContent = nodeNames[ni] + ' at round ' + snapRounds[col] + ': ' + entries + '/' + totalKeys + ' keys (' + pct + '%)'; + // Tooltip text depends on trace type + if (isDist) { + tooltip.textContent = nodeNames[ni] + ' at round ' + snapRounds[col] + ': registry ' + entries; + } else { + tooltip.textContent = nodeNames[ni] + ' at round ' + snapRounds[col] + ': ' + entries + '/' + totalKeys + ' keys (' + pct + '%)'; + } tooltip.style.display = 'block'; tooltip.style.left = (ex + 12) + 'px'; tooltip.style.top = (ey - 20) + 'px'; } else { tooltip.style.display = 'none'; } @@ -831,16 +953,28 @@ function drawLoadHistogram() { let curRound = 0; if (replayCursor > 0 && replayCursor <= allEvents.length) curRound = allEvents[replayCursor - 1].tick; - // Find the closest round in cumulPushRecv - let bestR = -1; - for (let r = 0; r < snapRounds.length; r++) { - if (snapRounds[r] <= curRound) bestR = r; - } - - if (bestR >= 0 && cumulPushRecv.length > bestR) { - data = cumulPushRecv[bestR]; + if (isDist) { + // Show cache size per node at current round + data = new Int32Array(N); + let bestR = -1; + for (let r = 0; r < snapRounds.length; r++) { + if (snapRounds[r] <= curRound) bestR = r; + } + if (bestR >= 0) { + for (let i = 0; i < N; i++) data[i] = snapCache[bestR][i]; + } } else { - data = metricPushesRecv; // fallback: total + // Find the closest round in cumulPushRecv + let bestR = -1; + for (let r = 0; r < snapRounds.length; r++) { + if (snapRounds[r] <= curRound) bestR = r; + } + + if (bestR >= 0 && cumulPushRecv.length > bestR) { + data = cumulPushRecv[bestR]; + } else { + data = metricPushesRecv; // fallback: total + } } // Compute stats @@ -998,37 +1132,76 @@ function drawGraph() { } } - // ── Nodes (community-colored) ── + // ── Nodes ── if (showNodes) { - const useCommunityColor = numCommunities > 1; let flashNodeI = -1; - if (useCommunityColor) { - // Batch by community color - for (let c = 0; c < numCommunities; c++) { - ctx2d.beginPath(); - for (let i = 0; i < N; i++) { - if (community[i] !== c) continue; - const px = posX[i], py = posY[i]; - if (px+baseR < v0x || px-baseR > v1x || py+baseR < v0y || py-baseR > v1y) continue; - if (hasNodeFlash && i === flashNode) { flashNodeI = i; continue; } - ctx2d.moveTo(px+baseR, py); ctx2d.arc(px, py, baseR, 0, 2*Math.PI); - } - ctx2d.fillStyle = communityColors[c % communityColors.length]; ctx2d.fill(); - if (showStroke) { ctx2d.strokeStyle = '#1a1d2e'; ctx2d.lineWidth = 1/vs; ctx2d.stroke(); } + if (isDist) { + // Distribution: color by alive/dead + let curRound = 0; + if (replayCursor > 0 && replayCursor <= allEvents.length) curRound = allEvents[replayCursor - 1].tick; + let curAlive = null; + for (let r = snapRounds.length - 1; r >= 0; r--) { + if (snapRounds[r] <= curRound) { curAlive = snapAlive[r]; break; } } - } else { + + // Draw alive nodes ctx2d.beginPath(); for (let i = 0; i < N; i++) { + const alive = curAlive ? curAlive[i] : 1; + if (!alive) continue; const px = posX[i], py = posY[i]; if (px+baseR < v0x || px-baseR > v1x || py+baseR < v0y || py-baseR > v1y) continue; if (hasNodeFlash && i === flashNode) { flashNodeI = i; continue; } ctx2d.moveTo(px+baseR, py); ctx2d.arc(px, py, baseR, 0, 2*Math.PI); } - ctx2d.fillStyle = '#6366f1'; ctx2d.fill(); - if (showStroke) { ctx2d.strokeStyle = '#4f46e5'; ctx2d.lineWidth = 1.5/vs; ctx2d.stroke(); } + ctx2d.fillStyle = '#22c55e'; ctx2d.fill(); + if (showStroke) { ctx2d.strokeStyle = '#16a34a'; ctx2d.lineWidth = 1.5/vs; ctx2d.stroke(); } + + // Draw dead nodes + ctx2d.beginPath(); + for (let i = 0; i < N; i++) { + const alive = curAlive ? curAlive[i] : 1; + if (alive) continue; + const px = posX[i], py = posY[i]; + if (px+baseR < v0x || px-baseR > v1x || py+baseR < v0y || py-baseR > v1y) continue; + if (hasNodeFlash && i === flashNode) { flashNodeI = i; continue; } + ctx2d.moveTo(px+baseR, py); ctx2d.arc(px, py, baseR, 0, 2*Math.PI); + } + ctx2d.fillStyle = '#ef4444'; ctx2d.fill(); + if (showStroke) { ctx2d.strokeStyle = '#dc2626'; ctx2d.lineWidth = 1.5/vs; ctx2d.stroke(); } + } else { + // Gossip: existing community/default coloring + const useCommunityColor = numCommunities > 1; + + if (useCommunityColor) { + // Batch by community color + for (let c = 0; c < numCommunities; c++) { + ctx2d.beginPath(); + for (let i = 0; i < N; i++) { + if (community[i] !== c) continue; + const px = posX[i], py = posY[i]; + if (px+baseR < v0x || px-baseR > v1x || py+baseR < v0y || py-baseR > v1y) continue; + if (hasNodeFlash && i === flashNode) { flashNodeI = i; continue; } + ctx2d.moveTo(px+baseR, py); ctx2d.arc(px, py, baseR, 0, 2*Math.PI); + } + ctx2d.fillStyle = communityColors[c % communityColors.length]; ctx2d.fill(); + if (showStroke) { ctx2d.strokeStyle = '#1a1d2e'; ctx2d.lineWidth = 1/vs; ctx2d.stroke(); } + } + } else { + ctx2d.beginPath(); + for (let i = 0; i < N; i++) { + const px = posX[i], py = posY[i]; + if (px+baseR < v0x || px-baseR > v1x || py+baseR < v0y || py-baseR > v1y) continue; + if (hasNodeFlash && i === flashNode) { flashNodeI = i; continue; } + ctx2d.moveTo(px+baseR, py); ctx2d.arc(px, py, baseR, 0, 2*Math.PI); + } + ctx2d.fillStyle = '#6366f1'; ctx2d.fill(); + if (showStroke) { ctx2d.strokeStyle = '#4f46e5'; ctx2d.lineWidth = 1.5/vs; ctx2d.stroke(); } + } } + // Flash node (shared) if (flashNodeI >= 0) { const t = 1-(now-flashNodeT)/400, rad = baseR+6*t; ctx2d.beginPath(); @@ -1081,6 +1254,19 @@ function formatDetail(kind, detail) { case 'PeerRemoved': return '- ' + (detail.peer_name || detail.peer); case 'QueryReceived': return 'key=' + detail.key; case 'StateSnapshot': return detail.entries + ' entries, ' + detail.peer_count + ' peers'; + case 'Joined': return 'seed: ' + detail.seed_addr; + case 'MembershipChanged': return detail.target + ' → ' + detail.new_state; + case 'PingSent': return '→ ' + detail.target; + case 'AckReceived': return '← ' + detail.from; + case 'ActorRegistered': return 'actor: ' + detail.actor_id; + case 'ActorStored': return detail.actor_id + ' on ' + detail.on_node; + case 'ActorResolved': return detail.actor_id + ' → ' + detail.found_on; + case 'ActorResolveFailed': return detail.actor_id + ': ' + detail.reason; + case 'NameRegistered': return '"' + detail.name + '" on node ' + detail.node_idx; + case 'NameUnregistered': return '"' + detail.name + '" on node ' + detail.node_idx; + case 'NameResolved': return '"' + detail.name + '" → ' + detail.result; + case 'NodeKilled': return ''; + case 'NodeRevived': return ''; default: return JSON.stringify(detail); } } @@ -1089,6 +1275,11 @@ function makeRow(ev) { const tr = document.createElement('tr'); if (ev.kind === 'GossipRoundStarted') tr.className = 'highlight-push'; else if (ev.kind === 'LocalSet') tr.className = 'highlight-set'; + else if (ev.kind === 'NodeKilled') tr.className = 'highlight-kill'; + else if (ev.kind === 'NodeRevived') tr.className = 'highlight-revive'; + else if (ev.kind === 'MembershipChanged') tr.className = 'highlight-membership'; + else if (ev.kind === 'NameRegistered' || ev.kind === 'NameUnregistered') tr.className = 'highlight-registry'; + else if (ev.kind === 'PingSent' || ev.kind === 'AckReceived') tr.className = 'highlight-ping'; tr.innerHTML = ''+ev.seq+''+ev.tick+''+(ev.thread||'-')+''+ev.node+''+ev.kind+''+formatDetail(ev.kind, ev.detail)+''; return tr; } @@ -1200,6 +1391,14 @@ function replayToImpl(pos) { const f = ev.detail.from_name || ev.detail.from; if (f) { flashSrc = nodeIdx.get(f) ?? -1; flashDst = nodeIdx.get(ev.node) ?? -1; flashEdgeT = performance.now(); } } + if (ev.kind === 'PingSent' && ev.detail) { + const t = ev.detail.target; + if (t) { flashSrc = nodeIdx.get(ev.node) ?? -1; flashDst = nodeIdx.get(t) ?? -1; flashEdgeT = performance.now(); } + } + if (ev.kind === 'AckReceived' && ev.detail) { + const f = ev.detail.from; + if (f) { flashSrc = nodeIdx.get(f) ?? -1; flashDst = nodeIdx.get(ev.node) ?? -1; flashEdgeT = performance.now(); } + } } document.getElementById('replaySlider').value = pos; @@ -1229,6 +1428,9 @@ function resetState() { metricPushesSent = new Int32Array(0); metricPushesRecv = new Int32Array(0); metricRedundant = 0; metricTotalPushes = 0; metricNoPeers = 0; cumulPushRecv = []; eventRoundMap = []; + isDist = false; + snapMembers = []; snapRegistry = []; snapCache = []; snapAlive = []; + snapRouting = []; snapRepair = []; snapDirectory = []; snapTombstone = []; community = new Int32Array(0); numCommunities = 0; communityHulls = []; document.getElementById('eventTableBody').textContent = ''; document.getElementById('workerColumns').innerHTML = ''; @@ -1247,6 +1449,7 @@ async function loadTrace(file) { const resp = await fetch('/trace.json?file=' + encodeURIComponent(file)); if (!resp.ok) { dot.className = 'status-dot error'; statusText.textContent = 'Failed to load trace'; return; } const trace = await resp.json(); + isDist = (trace.trace_type === 'distribution'); // Build node index nodeNames = trace.node_names; @@ -1271,85 +1474,138 @@ async function loadTrace(file) { allEvents = []; numRounds = trace.num_rounds || 0; - // First pass: collect snapshots grouped by tick - const snapByTick = new Map(); // tick -> Map(nodeIdx -> {entries, peer_count}) - for (const ev of trace.events) { - const isObj = typeof ev.kind === 'object'; - if (isObj && 'StateSnapshot' in ev.kind) { - const snap = ev.kind.StateSnapshot.snapshot || ev.kind.StateSnapshot; - const ni = nodeIdx.get(ev.node_name); - if (ni === undefined) continue; - const entriesCount = snap.entries ? Object.keys(snap.entries).length : 0; - const peerCount = snap.peer_count || 0; - if (!snapByTick.has(ev.tick)) snapByTick.set(ev.tick, new Map()); - snapByTick.get(ev.tick).set(ni, { entries: entriesCount, peer_count: peerCount }); - if (entriesCount > totalKeys) totalKeys = entriesCount; - continue; + if (isDist) { + // Distribution: snapshots come from trace.snapshots_per_round directly + snapRounds = []; + for (let r = 0; r < (trace.snapshots_per_round || []).length; r++) { + const roundSnaps = trace.snapshots_per_round[r]; + snapRounds.push(r + 1); // 1-indexed round + const memArr = new Int32Array(N); + const regArr = new Int32Array(N); + const cacheArr = new Int32Array(N); + const aliveArr = new Uint8Array(N); + const routingArr = new Int32Array(N); + const repairArr = new Int32Array(N); + const dirArr = new Int32Array(N); + const tombArr = new Int32Array(N); + for (const [nodeName, snap] of roundSnaps) { + const ni = nodeIdx.get(nodeName); + if (ni === undefined) continue; + memArr[ni] = snap.member_count || 0; + regArr[ni] = snap.registry_size || 0; + cacheArr[ni] = snap.cache_size || 0; + aliveArr[ni] = snap.is_alive ? 1 : 0; + routingArr[ni] = snap.routing_table_size || 0; + repairArr[ni] = snap.repair_queue_size || 0; + dirArr[ni] = snap.directory_entry_count || 0; + tombArr[ni] = snap.registry_tombstone_count || 0; + } + snapMembers.push(memArr); + snapRegistry.push(regArr); + snapCache.push(cacheArr); + snapAlive.push(aliveArr); + snapRouting.push(routingArr); + snapRepair.push(repairArr); + snapDirectory.push(dirArr); + snapTombstone.push(tombArr); + // For convergence chart compatibility, use registry_size as "entries" + snapEntries.push(regArr); + snapPeerCount.push(memArr); + // Track max for heatmap scaling + for (let i = 0; i < N; i++) { + if (regArr[i] > totalKeys) totalKeys = regArr[i]; + } } - if (!isObj && ev.kind === 'StateSnapshot') continue; - let kind, detail; - if (isObj) { for (kind in ev.kind) break; detail = ev.kind[kind] || null; } - else { kind = ev.kind; detail = null; } - allEvents.push({ seq: seq++, tick: ev.tick, node: ev.node_name, thread: ev.thread_name, kind, detail }); - } - // Build snapshot arrays sorted by round - snapRounds = Array.from(snapByTick.keys()).sort((a, b) => a - b); - snapEntries = []; snapPeerCount = []; - for (const tick of snapRounds) { - const eArr = new Int32Array(N); - const pArr = new Int32Array(N); - const m = snapByTick.get(tick); - for (const [ni, d] of m) { eArr[ni] = d.entries; pArr[ni] = d.peer_count; } - snapEntries.push(eArr); - snapPeerCount.push(pArr); - } - - // Compute per-node metrics from events - metricPushesSent = new Int32Array(N); - metricPushesRecv = new Int32Array(N); - metricRedundant = 0; metricTotalPushes = 0; metricNoPeers = 0; - - // Also build cumulative push-recv per round - const roundSet = new Set(snapRounds); - cumulPushRecv = []; - let runningRecv = new Int32Array(N); - - for (let i = 0; i < allEvents.length; i++) { - const ev = allEvents[i]; - const ni = nodeIdx.get(ev.node); - if (ni === undefined) continue; - if (ev.kind === 'GossipRoundStarted') { - metricPushesSent[ni]++; - metricTotalPushes++; + // Distribution events (no StateSnapshot filtering needed) + for (const ev of trace.events) { + const isObj = typeof ev.kind === 'object'; + let kind, detail; + if (isObj) { for (kind in ev.kind) break; detail = ev.kind[kind] || null; } + else { kind = ev.kind; detail = null; } + allEvents.push({ seq: seq++, tick: ev.tick, node: ev.node_name, thread: null, kind, detail }); } - if (ev.kind === 'PushReceived') { - metricPushesRecv[ni]++; - runningRecv[ni]++; - if (ev.detail && ev.detail.keys_updated === 0) metricRedundant++; + } else { + // First pass: collect snapshots grouped by tick + const snapByTick = new Map(); // tick -> Map(nodeIdx -> {entries, peer_count}) + for (const ev of trace.events) { + const isObj = typeof ev.kind === 'object'; + if (isObj && 'StateSnapshot' in ev.kind) { + const snap = ev.kind.StateSnapshot.snapshot || ev.kind.StateSnapshot; + const ni = nodeIdx.get(ev.node_name); + if (ni === undefined) continue; + const entriesCount = snap.entries ? Object.keys(snap.entries).length : 0; + const peerCount = snap.peer_count || 0; + if (!snapByTick.has(ev.tick)) snapByTick.set(ev.tick, new Map()); + snapByTick.get(ev.tick).set(ni, { entries: entriesCount, peer_count: peerCount }); + if (entriesCount > totalKeys) totalKeys = entriesCount; + continue; + } + if (!isObj && ev.kind === 'StateSnapshot') continue; + let kind, detail; + if (isObj) { for (kind in ev.kind) break; detail = ev.kind[kind] || null; } + else { kind = ev.kind; detail = null; } + allEvents.push({ seq: seq++, tick: ev.tick, node: ev.node_name, thread: ev.thread_name, kind, detail }); } - if (ev.kind === 'GossipRoundNoPeers') { metricNoPeers++; } - } - // Build cumulative recv snapshots aligned to snap rounds - // Re-scan to build per-round cumulative - if (snapRounds.length > 0) { - const cumRecv = new Int32Array(N); - let sri = 0; - for (let i = 0; i < allEvents.length && sri < snapRounds.length; i++) { + // Build snapshot arrays sorted by round + snapRounds = Array.from(snapByTick.keys()).sort((a, b) => a - b); + snapEntries = []; snapPeerCount = []; + for (const tick of snapRounds) { + const eArr = new Int32Array(N); + const pArr = new Int32Array(N); + const m = snapByTick.get(tick); + for (const [ni, d] of m) { eArr[ni] = d.entries; pArr[ni] = d.peer_count; } + snapEntries.push(eArr); + snapPeerCount.push(pArr); + } + + // Compute per-node metrics from events + metricPushesSent = new Int32Array(N); + metricPushesRecv = new Int32Array(N); + metricRedundant = 0; metricTotalPushes = 0; metricNoPeers = 0; + + // Also build cumulative push-recv per round + const roundSet = new Set(snapRounds); + cumulPushRecv = []; + let runningRecv = new Int32Array(N); + + for (let i = 0; i < allEvents.length; i++) { const ev = allEvents[i]; + const ni = nodeIdx.get(ev.node); + if (ni === undefined) continue; + if (ev.kind === 'GossipRoundStarted') { + metricPushesSent[ni]++; + metricTotalPushes++; + } if (ev.kind === 'PushReceived') { - const ni = nodeIdx.get(ev.node); - if (ni !== undefined) cumRecv[ni]++; - } - // When we pass a snapshot round boundary, save - while (sri < snapRounds.length && ev.tick >= snapRounds[sri]) { - cumulPushRecv.push(new Int32Array(cumRecv)); - sri++; + metricPushesRecv[ni]++; + runningRecv[ni]++; + if (ev.detail && ev.detail.keys_updated === 0) metricRedundant++; } + if (ev.kind === 'GossipRoundNoPeers') { metricNoPeers++; } + } + + // Build cumulative recv snapshots aligned to snap rounds + // Re-scan to build per-round cumulative + if (snapRounds.length > 0) { + const cumRecv = new Int32Array(N); + let sri = 0; + for (let i = 0; i < allEvents.length && sri < snapRounds.length; i++) { + const ev = allEvents[i]; + if (ev.kind === 'PushReceived') { + const ni = nodeIdx.get(ev.node); + if (ni !== undefined) cumRecv[ni]++; + } + // When we pass a snapshot round boundary, save + while (sri < snapRounds.length && ev.tick >= snapRounds[sri]) { + cumulPushRecv.push(new Int32Array(cumRecv)); + sri++; + } + } + // Fill remaining + while (sri < snapRounds.length) { cumulPushRecv.push(new Int32Array(cumRecv)); sri++; } } - // Fill remaining - while (sri < snapRounds.length) { cumulPushRecv.push(new Int32Array(cumRecv)); sri++; } } // Index PeerAdded edges @@ -1395,10 +1651,33 @@ async function loadTrace(file) { computeCommunityHulls(); dot.className = 'status-dot ready'; - statusText.textContent = trace.name + ' (' + allEvents.length + ' events)'; + statusText.textContent = trace.name + (isDist ? ' [distribution]' : ' [gossip]') + ' (' + allEvents.length + ' events)'; drawGraph(); drawAllCharts(); + // Update panel labels + document.querySelector('.worker-panel h2').textContent = isDist ? 'Node Status' : 'Worker Logs'; + document.querySelector('#heatSection h3').textContent = isDist ? 'Registry Propagation' : 'Propagation Heatmap'; + document.querySelector('#loadSection h3').textContent = isDist ? 'Cache Utilization' : 'Load Distribution'; + document.querySelector('#convSection h3').textContent = isDist ? 'Membership Convergence' : 'Convergence Curve'; + + // Update stat labels + if (isDist) { + document.querySelectorAll('.stat-label')[1].textContent = 'Alive'; + } else { + document.querySelectorAll('.stat-label')[1].textContent = 'Edges'; + } + + // Update stats with distribution-specific values + if (isDist && snapAlive.length > 0) { + let alive = 0; + const lastSnap = snapAlive[snapAlive.length - 1]; + for (let i = 0; i < N; i++) if (lastSnap[i]) alive++; + updateStats(N, alive, allEvents.length, numRounds, numRounds); + } else { + updateStats(N, totalEdges, allEvents.length, numRounds, numRounds); + } + // Replay controls document.getElementById('replayControls').style.display = 'flex'; const slider = document.getElementById('replaySlider'); @@ -1426,7 +1705,7 @@ resizeCanvas(); catch { dot.className='status-dot error'; statusText.textContent='Failed to fetch trace list'; select.innerHTML=''; return; } if (!traces.length) { dot.className='status-dot error'; statusText.textContent='No traces found'; select.innerHTML=''; return; } select.innerHTML = ''; - traces.forEach(t => { const o = document.createElement('option'); o.value = t.file; o.textContent = t.name+' ('+t.nodes+' nodes, '+t.events+' events)'; select.appendChild(o); }); + traces.forEach(t => { const o = document.createElement('option'); o.value = t.file; o.textContent = '[' + (t.trace_type || 'gossip') + '] ' + t.name+' ('+t.nodes+' nodes, '+t.events+' events)'; select.appendChild(o); }); select.disabled = false; select.onchange = () => { if (select.value) loadTrace(select.value); }; loadTrace(traces[0].file); diff --git a/crates/simulation-dashboard/src/server.rs b/crates/simulation-dashboard/src/server.rs index 10fda6e..06b39ae 100644 --- a/crates/simulation-dashboard/src/server.rs +++ b/crates/simulation-dashboard/src/server.rs @@ -13,6 +13,7 @@ struct TraceEntry { name: String, nodes: usize, events: usize, + trace_type: String, } fn scan_traces(dir: &Path) -> Vec { @@ -51,11 +52,17 @@ fn scan_traces(dir: &Path) -> Vec { .and_then(|v| v.as_array()) .map(|a| a.len()) .unwrap_or(0); + let trace_type = val + .get("trace_type") + .and_then(|v| v.as_str()) + .unwrap_or("gossip") + .to_string(); entries.push(TraceEntry { file: fname, name, nodes, events, + trace_type, }); } entries.sort_by(|a, b| a.file.cmp(&b.file)); diff --git a/crates/simulation/src/distribution/properties.rs b/crates/simulation/src/distribution/properties.rs index e5a1c82..5ef4ee8 100644 --- a/crates/simulation/src/distribution/properties.rs +++ b/crates/simulation/src/distribution/properties.rs @@ -326,3 +326,173 @@ pub fn check_convergence( description: "Membership views converge after faults stabilize".into(), } } + +// ─── Registry & Lifecycle Property Checks ───────────────────────────────── + +/// Check that all alive nodes have at least `min_registry_size` registry entries at the end. +pub fn check_registry_propagation( + trace: &DistTrace, + min_registry_size: usize, +) -> crate::properties::PropertyResult { + let passed = if let Some(last_round) = trace.snapshots_per_round.last() { + last_round + .iter() + .filter(|(_, s)| s.is_alive) + .all(|(_, s)| s.registry_size >= min_registry_size) + } else { + false + }; + let actual = if let Some(last_round) = trace.snapshots_per_round.last() { + let sizes: Vec = last_round + .iter() + .filter(|(_, s)| s.is_alive) + .map(|(_, s)| s.registry_size) + .collect(); + format!("{sizes:?}") + } else { + "no data".into() + }; + crate::properties::PropertyResult { + name: "registry_propagation".into(), + category: "Registry".into(), + passed, + expected: format!("all alive nodes have ≥{min_registry_size} registry entries"), + actual, + description: "Registry entries propagate to all nodes via gossip".into(), + } +} + +/// Check that all alive nodes agree on registry tombstone count at the end. +pub fn check_registry_tombstones( + trace: &DistTrace, + min_tombstones: usize, +) -> crate::properties::PropertyResult { + let passed = if let Some(last_round) = trace.snapshots_per_round.last() { + last_round + .iter() + .filter(|(_, s)| s.is_alive) + .all(|(_, s)| s.registry_tombstone_count >= min_tombstones) + } else { + false + }; + let actual = if let Some(last_round) = trace.snapshots_per_round.last() { + let counts: Vec = last_round + .iter() + .filter(|(_, s)| s.is_alive) + .map(|(_, s)| s.registry_tombstone_count) + .collect(); + format!("{counts:?}") + } else { + "no data".into() + }; + crate::properties::PropertyResult { + name: "registry_tombstones".into(), + category: "Registry".into(), + passed, + expected: format!("all alive nodes have ≥{min_tombstones} tombstones"), + actual, + description: "Registry tombstones propagate to all nodes".into(), + } +} + +/// Check that at least one survivor has a non-empty repair queue after a node death. +pub fn check_repair_queue_populated( + trace: &DistTrace, + after_round: usize, +) -> crate::properties::PropertyResult { + let populated = trace + .snapshots_per_round + .iter() + .skip(after_round) + .any(|round_snaps| { + round_snaps + .iter() + .any(|(_, s)| s.is_alive && s.repair_queue_size > 0) + }); + crate::properties::PropertyResult { + name: "repair_queue_populated".into(), + category: "Lifecycle".into(), + passed: populated, + expected: format!("repair queue populated after round {after_round}"), + actual: if populated { + "populated".into() + } else { + let final_sizes: Vec = trace + .snapshots_per_round + .last() + .map(|r| { + r.iter() + .filter(|(_, s)| s.is_alive) + .map(|(_, s)| s.repair_queue_size) + .collect() + }) + .unwrap_or_default(); + format!("final repair_queue_sizes: {final_sizes:?}") + }, + description: "Repair queue populated after node death".into(), + } +} + +/// Check that routing_table_size ≤ alive_count for all alive nodes at every round. +pub fn check_routing_table_bounded( + trace: &DistTrace, +) -> crate::properties::PropertyResult { + let mut violation = None; + + for (round_idx, round_snaps) in trace.snapshots_per_round.iter().enumerate() { + let alive_count = round_snaps.iter().filter(|(_, s)| s.is_alive).count(); + for (name, snap) in round_snaps { + if snap.is_alive && snap.routing_table_size > alive_count { + violation = Some(format!( + "round {}: {} has routing_table_size={} but alive_count={}", + round_idx + 1, name, snap.routing_table_size, alive_count + )); + break; + } + } + if violation.is_some() { + break; + } + } + + crate::properties::PropertyResult { + name: "routing_table_bounded".into(), + category: "Invariant".into(), + passed: violation.is_none(), + expected: "routing_table_size ≤ alive_count at every round".into(), + actual: violation.unwrap_or_else(|| "all within bounds".into()), + description: "Routing table never exceeds alive membership".into(), + } +} + +/// Check that cache_size ≤ cache_capacity for all alive nodes at every round. +pub fn check_cache_bounded( + trace: &DistTrace, + cache_capacity: usize, +) -> crate::properties::PropertyResult { + let mut violation = None; + + for (round_idx, round_snaps) in trace.snapshots_per_round.iter().enumerate() { + for (name, snap) in round_snaps { + if snap.is_alive && snap.cache_size > cache_capacity { + violation = Some(format!( + "round {}: {} has cache_size={} but capacity={}", + round_idx + 1, name, snap.cache_size, cache_capacity + )); + break; + } + } + if violation.is_some() { + break; + } + } + + crate::properties::PropertyResult { + name: "cache_bounded".into(), + category: "Invariant".into(), + passed: violation.is_none(), + expected: format!("cache_size ≤ {cache_capacity} at every round"), + actual: violation.unwrap_or_else(|| "all within bounds".into()), + description: "Cache never exceeds configured capacity".into(), + } +} diff --git a/crates/simulation/src/distribution/sim.rs b/crates/simulation/src/distribution/sim.rs index 29d7db6..eb084ee 100644 --- a/crates/simulation/src/distribution/sim.rs +++ b/crates/simulation/src/distribution/sim.rs @@ -21,6 +21,19 @@ pub struct Partition { pub asymmetric: bool, } +/// An action to execute at a specific round during the simulation. +#[derive(Debug, Clone)] +pub enum SimAction { + /// Register a name on the given node, binding it to a fresh random actor. + RegisterName { node_idx: usize, name: String }, + /// Register a name on the given node, binding it to a specific actor address. + RegisterNameWithActor { node_idx: usize, name: String, actor: ActorAddress }, + /// Unregister a name on the given node (creates a tombstone). + UnregisterName { node_idx: usize, name: String }, + /// Graceful leave — node announces its own death before being removed. + GracefulLeave { node_idx: usize }, +} + /// Schedule entry for network faults. #[derive(Debug, Clone)] pub enum NetworkFault { @@ -49,6 +62,12 @@ pub struct DistributionSimConfig { pub cache_capacity: usize, /// Network fault schedule. pub network_faults: Vec, + /// Actions to execute at specific rounds (e.g. register/unregister names). + pub action_schedule: Vec<(usize, SimAction)>, + /// Custom registry config overrides. + pub registry_tombstone_ttl: Option, + pub registry_gc_interval: Option, + pub registry_dissemination_lambda: Option, } impl Default for DistributionSimConfig { @@ -70,6 +89,10 @@ impl Default for DistributionSimConfig { revive_schedule: Vec::new(), cache_capacity: 100, network_faults: Vec::new(), + action_schedule: Vec::new(), + registry_tombstone_ttl: None, + registry_gc_interval: None, + registry_dissemination_lambda: None, } } } @@ -132,13 +155,27 @@ impl NetworkState { } } -type DistTrace = SimulationTrace; +pub type DistTrace = SimulationTrace; + +/// Run a distribution simulation, returning both the trace and the final node states. +/// +/// The returned `Vec>` has the same length as `config.num_nodes`. +/// Dead nodes are `None`. +pub fn run_simulation_with_nodes(config: DistributionSimConfig) -> (DistTrace, Vec>) { + let (trace, nodes, _) = run_simulation_inner(config); + (trace, nodes) +} /// Run a distribution simulation. /// /// Creates N `DistributedNode` instances, forms a cluster via join protocol, /// registers actors, then runs rounds of tick + deliver + resolve. pub fn run_simulation(config: DistributionSimConfig) -> DistTrace { + let (trace, _, _) = run_simulation_inner(config); + trace +} + +fn run_simulation_inner(config: DistributionSimConfig) -> (DistTrace, Vec>, Vec) { let mut events: Vec> = Vec::new(); let mut snapshots_per_round: Vec> = Vec::new(); @@ -152,13 +189,14 @@ pub fn run_simulation(config: DistributionSimConfig) -> DistTrace { for i in 0..n { let addr: SocketAddr = format!("127.0.0.1:{}", 10001 + i).parse().unwrap(); - let node_config = DistributedNodeConfig { + let mut node_config = DistributedNodeConfig { listen_addr: addr, swim: config.swim.clone(), cache_capacity: config.cache_capacity, republish_interval: 50, ..Default::default() }; + apply_registry_overrides(&mut node_config, &config); let node = DistributedNode::new(node_config); node_ids.push(node.node_id()); addrs.push(addr); @@ -284,13 +322,14 @@ pub fn run_simulation(config: DistributionSimConfig) -> DistTrace { // Apply revive schedule. for &(revive_round, revive_idx) in &config.revive_schedule { if revive_round == round && revive_idx < n { - let node_config = DistributedNodeConfig { + let mut node_config = DistributedNodeConfig { listen_addr: addrs[revive_idx], swim: config.swim.clone(), cache_capacity: config.cache_capacity, republish_interval: 50, ..Default::default() }; + apply_registry_overrides(&mut node_config, &config); let revived = DistributedNode::new(node_config); // Rejoin the cluster. let join_actions = revived.join(&[seed_addr]); @@ -328,6 +367,97 @@ pub fn run_simulation(config: DistributionSimConfig) -> DistTrace { } } + // Execute scheduled actions for this round. + for (action_round, action) in &config.action_schedule { + if *action_round == round { + match action { + SimAction::RegisterName { node_idx, name } => { + if *node_idx < n { + if let Some(ref mut node) = nodes[*node_idx] { + let actor = ActorAddress::new_random(); + node.register_name(name.clone(), actor); + events.push(Event { + tick: round as u64, + node_name: node_names[*node_idx].clone(), + kind: DistributionEventKind::NameRegistered { + name: name.clone(), + node_idx: *node_idx, + }, + }); + } + } + } + SimAction::RegisterNameWithActor { node_idx, name, actor } => { + if *node_idx < n { + if let Some(ref mut node) = nodes[*node_idx] { + node.register_name(name.clone(), *actor); + events.push(Event { + tick: round as u64, + node_name: node_names[*node_idx].clone(), + kind: DistributionEventKind::NameRegistered { + name: name.clone(), + node_idx: *node_idx, + }, + }); + } + } + } + SimAction::UnregisterName { node_idx, name } => { + if *node_idx < n { + if let Some(ref mut node) = nodes[*node_idx] { + node.unregister_name(name); + events.push(Event { + tick: round as u64, + node_name: node_names[*node_idx].clone(), + kind: DistributionEventKind::NameUnregistered { + name: name.clone(), + node_idx: *node_idx, + }, + }); + } + } + } + SimAction::GracefulLeave { node_idx } => { + if *node_idx < n { + if let Some(ref mut node) = nodes[*node_idx] { + let leave_actions = node.leave(); + // Deliver the leave actions (disseminate death announcement) + let tagged_responses = deliver_actions_tagged_with_net( + &leave_actions, + *node_idx, + node_ids[*node_idx], + addrs[*node_idx], + &mut nodes, + &node_ids, + &addrs, + &mut net, + ); + for (responder_idx, response_actions) in tagged_responses { + deliver_actions_tagged_with_net( + &response_actions, + responder_idx, + node_ids[responder_idx], + addrs[responder_idx], + &mut nodes, + &node_ids, + &addrs, + &mut net, + ); + } + } + // Remove the node after leave + nodes[*node_idx] = None; + events.push(Event { + tick: round as u64, + node_name: node_names[*node_idx].clone(), + kind: DistributionEventKind::NodeKilled, + }); + } + } + } + } + } + // Tick all live nodes and deliver actions. for _ in 0..config.ticks_per_round { tick_all_and_deliver( @@ -393,6 +523,8 @@ pub fn run_simulation(config: DistributionSimConfig) -> DistTrace { directory_entry_count: node.directory().entry_count(), cache_size: node.cache().len(), repair_queue_size: node.repair_queue().len(), + registry_size: node.registry().len(), + registry_tombstone_count: node.registry().tombstone_count(), is_alive: true, }, None => DistributionSnapshot { @@ -401,6 +533,8 @@ pub fn run_simulation(config: DistributionSimConfig) -> DistTrace { directory_entry_count: 0, cache_size: 0, repair_queue_size: 0, + registry_size: 0, + registry_tombstone_count: 0, is_alive: false, }, }; @@ -414,13 +548,27 @@ pub fn run_simulation(config: DistributionSimConfig) -> DistTrace { .map(|i| (node_names[i].clone(), node_names[0].clone())) .collect(); - SimulationTrace { + let trace = SimulationTrace { name: config.name, + trace_type: "distribution".into(), node_names, topology_edges, events, snapshots_per_round, num_rounds: config.num_rounds, + }; + (trace, nodes, node_ids) +} + +fn apply_registry_overrides(node_config: &mut DistributedNodeConfig, config: &DistributionSimConfig) { + if let Some(ttl) = config.registry_tombstone_ttl { + node_config.registry.tombstone_ttl = ttl; + } + if let Some(interval) = config.registry_gc_interval { + node_config.registry.gc_interval = interval; + } + if let Some(lambda) = config.registry_dissemination_lambda { + node_config.registry.dissemination_lambda = lambda; } } diff --git a/crates/simulation/src/distribution/trace.rs b/crates/simulation/src/distribution/trace.rs index d063362..97d53e7 100644 --- a/crates/simulation/src/distribution/trace.rs +++ b/crates/simulation/src/distribution/trace.rs @@ -11,6 +11,9 @@ pub enum DistributionEventKind { ActorStored { actor_id: String, on_node: String }, ActorResolved { actor_id: String, found_on: String }, ActorResolveFailed { actor_id: String, reason: String }, + NameRegistered { name: String, node_idx: usize }, + NameUnregistered { name: String, node_idx: usize }, + NameResolved { name: String, result: String }, NodeKilled, NodeRevived, } @@ -23,5 +26,7 @@ pub struct DistributionSnapshot { pub directory_entry_count: usize, pub cache_size: usize, pub repair_queue_size: usize, + pub registry_size: usize, + pub registry_tombstone_count: usize, pub is_alive: bool, } diff --git a/crates/simulation/src/trace.rs b/crates/simulation/src/trace.rs index b825666..5b8c8fe 100644 --- a/crates/simulation/src/trace.rs +++ b/crates/simulation/src/trace.rs @@ -1,11 +1,17 @@ use std::sync::atomic::AtomicU64; use std::sync::Arc; +use serde::{Deserialize, Serialize}; + /// Shared tick counter — the simulation harness increments this. pub type TickCounter = Arc; /// A single simulation event, generic over the event kind `K`. -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(bound( + serialize = "K: Serialize", + deserialize = "K: serde::de::DeserializeOwned" +))] pub struct Event { pub tick: u64, pub node_name: String, @@ -13,9 +19,16 @@ pub struct Event { } /// Complete output of a simulation run, generic over event kind `K` and snapshot type `S`. -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(bound( + serialize = "K: Serialize, S: Serialize", + deserialize = "K: serde::de::DeserializeOwned, S: serde::de::DeserializeOwned" +))] pub struct SimulationTrace { pub name: String, + /// Discriminator for dashboard rendering ("gossip" or "distribution"). + #[serde(default)] + pub trace_type: String, pub node_names: Vec, pub topology_edges: Vec<(String, String)>, pub events: Vec>, diff --git a/crates/simulation/tests/cluster_scenarios.rs b/crates/simulation/tests/cluster_scenarios.rs index 6c256c9..bdad324 100644 --- a/crates/simulation/tests/cluster_scenarios.rs +++ b/crates/simulation/tests/cluster_scenarios.rs @@ -126,8 +126,9 @@ fn cluster_converges_under_10_percent_message_loss() { // Given: 5 nodes with 10% message loss from the start. // 10% loss is significant for SWIM because it can hit both direct probe // AND indirect probes in the same cycle, causing false suspicions. - // We verify the cluster degrades but doesn't crash, and at least some - // membership information survives. + // Dead reprobe is enabled so false deaths can self-correct — without it, + // correct death dissemination (via piggyback) causes cascading false deaths + // that collapse the entire cluster under even modest message loss. let config = DistributionSimConfig { name: "message-loss-10pct".into(), num_nodes: 5, @@ -137,9 +138,9 @@ fn cluster_converges_under_10_percent_message_loss() { swim: distribution::swim::probe::SwimConfig { probe_interval: 1, probe_timeout: 5, - indirect_probes: 2, - suspicion_timeout: 20, - dead_reprobe_interval: 0, + indirect_probes: 3, + suspicion_timeout: 60, + dead_reprobe_interval: 15, }, network_faults: vec![NetworkFault::SetDropRate { round: 1, @@ -151,8 +152,8 @@ fn cluster_converges_under_10_percent_message_loss() { let trace = run_simulation(config); let metrics = analyze(&trace); - // With 10% loss and the deterministic LCG, SWIM's probe cycle is disrupted - // enough to cause false deaths. The test verifies: + // With 10% loss, SWIM's probe cycle is disrupted enough to cause + // false suspicions. Dead reprobe allows recovery. We verify: // 1. The simulation completes without panic (implicit — we got here) // 2. At least partial membership is maintained (some nodes still know about others) let result = check_membership_accuracy(&metrics, 0.15); @@ -629,7 +630,9 @@ fn sequential_partitions_fragment_cluster() { #[test] fn cluster_survives_brief_message_loss() { // Given: 5 nodes with 15% loss for a brief window, then clean network. - // High suspicion timeout prevents false positives during the loss period. + // High suspicion timeout + dead reprobe prevents permanent false positives. + // Without dead reprobe, correct death dissemination causes cascading + // false deaths that collapse the cluster. let config = DistributionSimConfig { name: "brief-loss-recovery".into(), num_nodes: 5, @@ -639,9 +642,9 @@ fn cluster_survives_brief_message_loss() { swim: distribution::swim::probe::SwimConfig { probe_interval: 1, probe_timeout: 5, - indirect_probes: 2, - suspicion_timeout: 20, - dead_reprobe_interval: 0, + indirect_probes: 3, + suspicion_timeout: 60, + dead_reprobe_interval: 15, }, network_faults: vec![ NetworkFault::SetDropRate { diff --git a/crates/simulation/tests/distribution_lifecycle.rs b/crates/simulation/tests/distribution_lifecycle.rs new file mode 100644 index 0000000..b1ef5e5 --- /dev/null +++ b/crates/simulation/tests/distribution_lifecycle.rs @@ -0,0 +1,615 @@ +//! Lifecycle simulation tests — death/repair/cache/routing behavior. +//! +//! Tests that node death correctly triggers: +//! - Repair queue population for re-replication +//! - Cache invalidation of stale entries +//! - Routing table cleanup +//! - Recovery after partition heals + +use simulation::distribution::properties::{ + check_repair_queue_populated, check_routing_table_bounded, +}; +use simulation::distribution::sim::{ + run_simulation_with_nodes, DistributionSimConfig, DistTrace, NetworkFault, Partition, SimAction, +}; + +fn maybe_save_trace(trace: &DistTrace) { + if let Ok(dir) = std::env::var("SWACTOR_TRACE_DIR") { + std::fs::create_dir_all(&dir).ok(); + let filename = format!( + "{}/{}.trace.json", + dir, + trace.name.to_lowercase().replace(' ', "_").replace(['(', ')'], "") + ); + let json = serde_json::to_string_pretty(trace).expect("trace serialization failed"); + std::fs::write(&filename, json).expect("trace write failed"); + } +} + +fn default_config() -> DistributionSimConfig { + DistributionSimConfig::default() +} + +// ──────────────────────────────────────────────────────────────────────────── +// 1. Dead node's actors populate repair queue and invalidate cache +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn dead_node_triggers_repair_queue_and_cache_invalidation() { + // Given: 5-node cluster, 2 actors/node. Node 2 is killed at round 10. + let config = DistributionSimConfig { + name: "death-repair-cache".into(), + num_nodes: 5, + num_rounds: 80, + ticks_per_round: 3, + actors_per_node: 2, + kill_schedule: vec![(10, 2)], + ..default_config() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Then: at least one survivor should have a non-empty repair queue + let result = check_repair_queue_populated(&trace, 10); + assert!( + result.passed, + "repair queue should be populated after node death: {}", + result.actual + ); + + // And: no survivor's cache should contain entries pointing to the dead node + let dead_node_id = { + // Find the node_id for node 2 from round snapshots before death + // We can check from the surviving nodes + // Node 2 is dead (None), so we check survivors' caches + let mut stale_count = 0; + for node in nodes.iter().filter_map(|n| n.as_ref()) { + for (_actor, cached_on) in node.cache().entries() { + // The dead node's entries should have been invalidated + // We can't easily get node 2's ID here, but we can check + // that no survivor caches an actor on a node not in their members + let alive_ids: Vec<_> = node.members().iter().map(|m| m.node_id).collect(); + if !alive_ids.contains(&cached_on) && cached_on != node.node_id() { + stale_count += 1; + } + } + } + stale_count + }; + + assert_eq!( + dead_node_id, 0, + "no survivor should have cache entries pointing to non-member nodes" + ); +} + +// ──────────────────────────────────────────────────────────────────────────── +// 2. Revived node starts fresh (empty directory) +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn revived_node_has_empty_directory() { + // Given: 5-node cluster, 2 actors/node. + // Node 2 killed at round 10, revived at round 50. + let config = DistributionSimConfig { + name: "revive-fresh".into(), + num_nodes: 5, + num_rounds: 100, + ticks_per_round: 3, + actors_per_node: 2, + kill_schedule: vec![(10, 2)], + revive_schedule: vec![(50, 2)], + ..default_config() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Then: the revived node should have an empty directory + // (it's a fresh DistributedNode, not carrying over old state) + let revived = nodes[2].as_ref().expect("node 2 should be revived"); + assert_eq!( + revived.directory().entry_count(), 0, + "revived node should start with empty directory" + ); + + // And: the revived node should have rejoined the cluster + assert!( + !revived.members().is_empty(), + "revived node should have some cluster members" + ); +} + +// ──────────────────────────────────────────────────────────────────────────── +// 3. Cache invalidation tracks membership changes +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn cache_shrinks_after_node_death() { + // Given: 5-node cluster with actors, all caches populated during setup. + // When: node 1 is killed + // Then: cache_size should decrease for survivors after death detection. + let config = DistributionSimConfig { + name: "cache-invalidation".into(), + num_nodes: 5, + num_rounds: 80, + ticks_per_round: 3, + actors_per_node: 3, + kill_schedule: vec![(10, 1)], + ..default_config() + }; + + let (trace, _nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Check that cache_size decreased for at least some survivors after death + // Before death (round 9), survivors should have cache entries + // After death detection, cache for dead node's actors should be invalidated + let pre_death_round = 8; // 0-indexed round 9 + let post_detection_round = 39; // well after SWIM detection + + if pre_death_round < trace.snapshots_per_round.len() + && post_detection_round < trace.snapshots_per_round.len() + { + let pre_cache_max: usize = trace.snapshots_per_round[pre_death_round] + .iter() + .filter(|(_, s)| s.is_alive) + .map(|(_, s)| s.cache_size) + .max() + .unwrap_or(0); + + let post_cache_sizes: Vec = trace.snapshots_per_round[post_detection_round] + .iter() + .filter(|(_, s)| s.is_alive) + .map(|(_, s)| s.cache_size) + .collect(); + + // After death, some survivors should have fewer cache entries + // (the dead node's actors were invalidated) + let any_decreased = post_cache_sizes.iter().any(|&s| s < pre_cache_max); + assert!( + any_decreased || pre_cache_max == 0, + "cache should shrink after node death, pre_max={pre_cache_max}, post={post_cache_sizes:?}" + ); + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// 4. Routing table recovers after partition heals (dead reprobe) +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn routing_table_recovers_after_partition_heals() { + // Given: 6-node cluster, partition at round 10, heal at round 40 + // With dead_reprobe_interval=10, nodes re-discover dead members + let config = DistributionSimConfig { + name: "rt-recovery".into(), + num_nodes: 6, + num_rounds: 120, + ticks_per_round: 3, + actors_per_node: 0, + swim: distribution::swim::probe::SwimConfig { + probe_interval: 1, + probe_timeout: 3, + indirect_probes: 1, + suspicion_timeout: 5, + dead_reprobe_interval: 10, + }, + network_faults: vec![ + NetworkFault::Partition { + round: 10, + partition: Partition { + side_a: vec![0, 1, 2], + side_b: vec![3, 4, 5], + asymmetric: false, + }, + }, + NetworkFault::Heal { round: 40 }, + ], + ..default_config() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Then: after healing, all nodes should have recovered routing tables + // Each node should see at least 4 of 5 other nodes in their routing table + for (i, maybe_node) in nodes.iter().enumerate() { + if let Some(node) = maybe_node { + assert!( + node.routing_table().len() >= 4, + "node {i} should have ≥4 RT entries after partition heals, got {}", + node.routing_table().len() + ); + } + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// 5. Routing table tracks alive membership (bounded invariant) +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn routing_table_bounded_by_alive_count() { + // Run a simulation with deaths and verify the routing table invariant + let config = DistributionSimConfig { + name: "rt-bounded".into(), + num_nodes: 8, + num_rounds: 80, + ticks_per_round: 3, + actors_per_node: 1, + kill_schedule: vec![(15, 2), (25, 5)], + ..default_config() + }; + + let (trace, _) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + let result = check_routing_table_bounded(&trace); + assert!( + result.passed, + "routing table should never exceed alive count: {}", + result.actual + ); +} + +// ──────────────────────────────────────────────────────────────────────────── +// 6. Combined: partition + death during partition + heal + verify +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn partition_then_death_during_partition_then_heal() { + // Given: 6 nodes, partition at r=10, node 2 (side A) killed during partition + // at r=20, heal at r=40. Node 2 had actors and a registry name. + // Use high suspicion_timeout so cross-partition nodes stay Suspect (not Dead), + // while within-partition death of node 2 is detected after timeout expires. + let config = DistributionSimConfig { + name: "partition-death-heal".into(), + num_nodes: 6, + num_rounds: 150, + ticks_per_round: 3, + actors_per_node: 2, + network_faults: vec![ + NetworkFault::Partition { + round: 10, + partition: Partition { + side_a: vec![0, 1, 2], + side_b: vec![3, 4, 5], + asymmetric: false, + }, + }, + NetworkFault::Heal { round: 40 }, + ], + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 2, name: "doomed-svc".into() }), + (5, SimAction::RegisterName { node_idx: 4, name: "stable-svc".into() }), + ], + kill_schedule: vec![(20, 2)], + swim: distribution::swim::probe::SwimConfig { + probe_interval: 1, + probe_timeout: 3, + indirect_probes: 1, + // >90 ticks (30 rounds × 3 ticks) so cross-partition nodes stay Suspect during + // the 30-round partition. Node 2 (truly dead) gets declared dead ~33 rounds + // after kill, well after partition heals. + suspicion_timeout: 100, + dead_reprobe_interval: 10, + }, + ..default_config() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + let survivors: Vec<_> = nodes.iter().filter_map(|n| n.as_ref()).collect(); + assert_eq!(survivors.len(), 5, "5 of 6 should survive"); + + // "doomed-svc" should be tombstoned (owner node 2 died) + for node in &survivors { + assert!( + node.resolve_name("doomed-svc").is_none(), + "doomed-svc should be tombstoned after owner died during partition" + ); + } + + // "stable-svc" should still resolve (node 4 alive throughout) + for node in &survivors { + assert!( + node.resolve_name("stable-svc").is_some(), + "stable-svc should resolve (owner survived partition)" + ); + } + + // After partition heal + dead reprobe, routing tables should recover + // (at least 4 entries for each surviving node) + for node in &survivors { + assert!( + node.routing_table().len() >= 3, + "surviving node should have ≥3 RT entries after partition heals, got {}", + node.routing_table().len() + ); + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// 7. Registry GC: tombstones are garbage-collected after TTL +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn registry_tombstones_gc_after_ttl() { + // Given: short tombstone TTL and GC interval, register then unregister a name. + // Then do many more register/unregister operations to advance the logical clock + // (which is used for TTL comparison). After enough clock advancement, the + // original tombstone should be garbage-collected. + let config = DistributionSimConfig { + name: "registry-gc".into(), + num_nodes: 5, + num_rounds: 80, + ticks_per_round: 3, + actors_per_node: 0, + // Very short GC: TTL=5 logical clock ticks, GC runs every 3 ticks + registry_tombstone_ttl: Some(5), + registry_gc_interval: Some(3), + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 0, name: "ephemeral".into() }), + (10, SimAction::UnregisterName { node_idx: 0, name: "ephemeral".into() }), + // Additional operations to advance the logical clock past the TTL + (15, SimAction::RegisterName { node_idx: 1, name: "churn-1".into() }), + (16, SimAction::RegisterName { node_idx: 2, name: "churn-2".into() }), + (17, SimAction::RegisterName { node_idx: 3, name: "churn-3".into() }), + (18, SimAction::RegisterName { node_idx: 4, name: "churn-4".into() }), + (19, SimAction::RegisterName { node_idx: 1, name: "churn-5".into() }), + (20, SimAction::RegisterName { node_idx: 2, name: "churn-6".into() }), + ], + ..default_config() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Shortly after unregister (round 12), tombstones should exist + let mid_tombstones: usize = trace.snapshots_per_round + .get(11) // round 12 + .map(|snaps| { + snaps.iter() + .filter(|(_, s)| s.is_alive) + .map(|(_, s)| s.registry_tombstone_count) + .sum() + }) + .unwrap_or(0); + + assert!( + mid_tombstones > 0, + "tombstones should exist shortly after unregister" + ); + + // After many more register operations advance the clock, the "ephemeral" tombstone + // should be GC'd (its age exceeds TTL=5 in logical clock terms) + let alive_nodes: Vec<_> = nodes.iter().filter_map(|n| n.as_ref()).collect(); + + // Check that "ephemeral" resolves to None on all nodes (whether GC'd or still tombstoned) + for node in &alive_nodes { + assert!( + node.resolve_name("ephemeral").is_none(), + "ephemeral should not resolve (tombstoned or GC'd)" + ); + } + + // At least some nodes should have GC'd the tombstone (clock advanced past TTL) + let nodes_with_ephemeral_tombstone: usize = alive_nodes + .iter() + .filter(|n| { + n.registry().entries().any(|e| e.name == "ephemeral" && e.tombstone) + }) + .count(); + + assert!( + nodes_with_ephemeral_tombstone < alive_nodes.len(), + "at least some nodes should have GC'd the 'ephemeral' tombstone, but {} of {} still have it", + nodes_with_ephemeral_tombstone, + alive_nodes.len() + ); +} + +// ──────────────────────────────────────────────────────────────────────────── +// 8. Indirect probes (PingReq) prevent false death on flaky direct path +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn asymmetric_one_way_block_does_not_kill_node() { + // Given: 6-node cluster. Asymmetric partition: 0→5 blocked, 5→0 works. + // Node 5 can still communicate with nodes 1-4 in both directions, and + // 5→0 works, so gossip piggyback keeps node 0 informed about node 5's + // aliveness through intermediate nodes. + // + // Note: this implementation relies on piggyback gossip for indirect + // recovery (PingReq ack forwarding is not implemented), so we use a + // generous suspicion_timeout to allow gossip propagation. + let config = DistributionSimConfig { + name: "one-way-block".into(), + num_nodes: 6, + num_rounds: 80, + ticks_per_round: 3, + actors_per_node: 0, + // Asymmetric: only 0→5 is blocked, all other paths work + network_faults: vec![ + NetworkFault::Partition { + round: 10, + partition: Partition { + side_a: vec![0], + side_b: vec![5], + asymmetric: true, // 0→5 blocked, 5→0 works + }, + }, + ], + swim: distribution::swim::probe::SwimConfig { + probe_interval: 1, + probe_timeout: 3, + indirect_probes: 3, + // Timeout must exceed total sim ticks (80*3=240) so node 0 + // never declares node 5 dead despite the blocked direct path. + // Gossip through intermediate nodes refutes suspicion each cycle. + suspicion_timeout: 500, + dead_reprobe_interval: 10, + }, + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 5, name: "target-svc".into() }), + ], + ..default_config() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Then: all 6 nodes should still be alive (asymmetric block doesn't kill either side) + let alive_count = nodes.iter().filter(|n| n.is_some()).count(); + assert_eq!(alive_count, 6, "all 6 nodes should be alive"); + + // And: node 5's registry name should be resolvable from all nodes + // (gossip carries the entry through intermediate nodes even if 0→5 is blocked) + for (i, node) in nodes.iter().enumerate() { + if let Some(node) = node { + assert!( + node.resolve_name("target-svc").is_some(), + "node {i} should resolve 'target-svc'" + ); + } + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// 9. Names registered during partition propagate after heal via re_disseminate_all +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn names_registered_during_partition_propagate_after_heal() { + // Given: 6 nodes, partition {0,1,2} vs {3,4,5} from round 10 to 50. + // During the partition, each side registers a name the other side can't see. + // After healing, re_disseminate_all (triggered by Alive transitions) should + // propagate both names to the entire cluster. + let config = DistributionSimConfig { + name: "partition-register-heal".into(), + num_nodes: 6, + num_rounds: 120, + ticks_per_round: 3, + actors_per_node: 0, + network_faults: vec![ + NetworkFault::Partition { + round: 10, + partition: Partition { + side_a: vec![0, 1, 2], + side_b: vec![3, 4, 5], + asymmetric: false, + }, + }, + NetworkFault::Heal { round: 50 }, + ], + action_schedule: vec![ + // Registered DURING partition — other side doesn't see these initially + (20, SimAction::RegisterName { node_idx: 0, name: "side-a-svc".into() }), + (20, SimAction::RegisterName { node_idx: 3, name: "side-b-svc".into() }), + ], + swim: distribution::swim::probe::SwimConfig { + probe_interval: 1, + probe_timeout: 3, + indirect_probes: 1, + // High timeout: cross-partition nodes stay Suspect during 40-round partition + suspicion_timeout: 200, + dead_reprobe_interval: 10, + }, + ..default_config() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + let alive: Vec<_> = nodes.iter().filter_map(|n| n.as_ref()).collect(); + assert_eq!(alive.len(), 6, "all 6 nodes should survive"); + + // After healing + gossip, both names should be resolvable from every node + for (i, node) in nodes.iter().enumerate() { + if let Some(node) = node { + assert!( + node.resolve_name("side-a-svc").is_some(), + "node {i} should resolve 'side-a-svc' (registered during partition on side A)" + ); + assert!( + node.resolve_name("side-b-svc").is_some(), + "node {i} should resolve 'side-b-svc' (registered during partition on side B)" + ); + } + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// 10. Bidirectional suspicion: two nodes suspect each other, both recover +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn bidirectional_suspicion_both_nodes_recover() { + // Given: 6 nodes. Mutual partition between node 0 and node 5 (both directions) + // from round 10 to 30. Both sides can still reach nodes 1-4. + // Both 0 and 5 will suspect each other, but gossip through 1-4 carries + // refutations. After healing, both should be Alive with registry intact. + let config = DistributionSimConfig { + name: "bidir-suspicion".into(), + num_nodes: 6, + num_rounds: 80, + ticks_per_round: 3, + actors_per_node: 0, + network_faults: vec![ + NetworkFault::Partition { + round: 10, + partition: Partition { + side_a: vec![0], + side_b: vec![5], + asymmetric: false, // full mutual block + }, + }, + NetworkFault::Heal { round: 30 }, + ], + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 0, name: "svc-zero".into() }), + (5, SimAction::RegisterName { node_idx: 5, name: "svc-five".into() }), + ], + swim: distribution::swim::probe::SwimConfig { + probe_interval: 1, + probe_timeout: 3, + indirect_probes: 3, + // Must exceed partition duration (20 rounds × 3 ticks = 60 ticks) + suspicion_timeout: 100, + dead_reprobe_interval: 10, + }, + ..default_config() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // All 6 nodes alive + let alive_count = nodes.iter().filter(|n| n.is_some()).count(); + assert_eq!(alive_count, 6, "all 6 nodes should survive bidirectional suspicion"); + + // Both registry names should resolve on all nodes + for (i, node) in nodes.iter().enumerate() { + if let Some(node) = node { + assert!( + node.resolve_name("svc-zero").is_some(), + "node {i} should resolve 'svc-zero'" + ); + assert!( + node.resolve_name("svc-five").is_some(), + "node {i} should resolve 'svc-five'" + ); + } + } + + // Both nodes 0 and 5 should see each other in their member lists + let node0 = nodes[0].as_ref().unwrap(); + let node5 = nodes[5].as_ref().unwrap(); + let node0_sees_5 = node0.members().iter().any(|m| m.node_id == node5.node_id()); + let node5_sees_0 = node5.members().iter().any(|m| m.node_id == node0.node_id()); + assert!(node0_sees_5, "node 0 should see node 5 as a member after healing"); + assert!(node5_sees_0, "node 5 should see node 0 as a member after healing"); +} diff --git a/crates/simulation/tests/distribution_properties.rs b/crates/simulation/tests/distribution_properties.rs new file mode 100644 index 0000000..35ae3e8 --- /dev/null +++ b/crates/simulation/tests/distribution_properties.rs @@ -0,0 +1,579 @@ +//! Property-based distribution tests — invariants that must hold across configs. +//! +//! Each test verifies a structural property across multiple simulation +//! configurations with varying fault conditions. + +use simulation::distribution::properties::{ + check_cache_bounded, check_registry_propagation, check_repair_queue_populated, + check_routing_table_bounded, +}; +use simulation::distribution::sim::{ + run_simulation_with_nodes, DistributionSimConfig, DistTrace, NetworkFault, Partition, SimAction, +}; + +fn maybe_save_trace(trace: &DistTrace) { + if let Ok(dir) = std::env::var("SWACTOR_TRACE_DIR") { + std::fs::create_dir_all(&dir).ok(); + let filename = format!( + "{}/{}.trace.json", + dir, + trace.name.to_lowercase().replace(' ', "_").replace(['(', ')'], "") + ); + let json = serde_json::to_string_pretty(trace).expect("trace serialization failed"); + std::fs::write(&filename, json).expect("trace write failed"); + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// 1. Routing table size ≤ alive membership at every round +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn routing_table_bounded_across_configs() { + let configs = vec![ + // Healthy 5-node cluster + DistributionSimConfig { + name: "rt-bound-healthy-5".into(), + num_nodes: 5, + num_rounds: 50, + ticks_per_round: 3, + ..DistributionSimConfig::default() + }, + // 10-node cluster with 2 deaths + DistributionSimConfig { + name: "rt-bound-deaths-10".into(), + num_nodes: 10, + num_rounds: 60, + ticks_per_round: 3, + kill_schedule: vec![(15, 3), (25, 7)], + ..DistributionSimConfig::default() + }, + // 15-node cluster with 1 death + DistributionSimConfig { + name: "rt-bound-large-15".into(), + num_nodes: 15, + num_rounds: 60, + ticks_per_round: 3, + kill_schedule: vec![(20, 0)], + ..DistributionSimConfig::default() + }, + ]; + + for config in configs { + let name = config.name.clone(); + let (trace, _) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + let result = check_routing_table_bounded(&trace); + assert!( + result.passed, + "[{name}] routing table bounded invariant violated: {}", + result.actual + ); + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// 2. Cache size ≤ cache_capacity at every round +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn cache_bounded_across_configs() { + let capacity = 100; + let configs = vec![ + DistributionSimConfig { + name: "cache-bound-5".into(), + num_nodes: 5, + num_rounds: 50, + ticks_per_round: 3, + actors_per_node: 5, + cache_capacity: capacity, + ..DistributionSimConfig::default() + }, + DistributionSimConfig { + name: "cache-bound-10-deaths".into(), + num_nodes: 10, + num_rounds: 60, + ticks_per_round: 3, + actors_per_node: 3, + cache_capacity: capacity, + kill_schedule: vec![(15, 2), (20, 5)], + ..DistributionSimConfig::default() + }, + ]; + + for config in configs { + let name = config.name.clone(); + let (trace, _) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + let result = check_cache_bounded(&trace, capacity); + assert!( + result.passed, + "[{name}] cache bounded invariant violated: {}", + result.actual + ); + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// 3. Repair queue populates when node with directory entries dies +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn repair_queue_populates_on_death_with_directory_entries() { + // Node 2 has 3 actors. When it dies, repair queue should grow. + let config = DistributionSimConfig { + name: "repair-proportional".into(), + num_nodes: 5, + num_rounds: 80, + ticks_per_round: 3, + actors_per_node: 3, + kill_schedule: vec![(15, 2)], + ..DistributionSimConfig::default() + }; + + let (trace, _) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Repair queue should be populated within 2-3 rounds of death detection + let result = check_repair_queue_populated(&trace, 15); + assert!( + result.passed, + "repair queue should fill after node with actors dies: {}", + result.actual + ); + + // Check that the total repair queue size across survivors is proportional + // to the dead node's directory entries (3 actors) + let post_death_sizes: Vec = trace + .snapshots_per_round + .iter() + .skip(20) + .take(30) + .flat_map(|round_snaps| { + round_snaps + .iter() + .filter(|(_, s)| s.is_alive) + .map(|(_, s)| s.repair_queue_size) + }) + .collect(); + + let max_repair = post_death_sizes.iter().max().copied().unwrap_or(0); + assert!( + max_repair >= 1, + "at least one repair queue entry expected for dead node's actors, max seen: {max_repair}" + ); +} + +// ──────────────────────────────────────────────────────────────────────────── +// 4. Registry convergence — eventual consistency across configs +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn registry_eventually_consistent_across_configs() { + let configs = vec![ + // 5 nodes, 1 name + ( + DistributionSimConfig { + name: "reg-ec-simple".into(), + num_nodes: 5, + num_rounds: 50, + ticks_per_round: 3, + actors_per_node: 0, + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 0, name: "svc-a".into() }), + ], + ..DistributionSimConfig::default() + }, + 1, // expected min registry size + ), + // 8 nodes, 3 names from different nodes + ( + DistributionSimConfig { + name: "reg-ec-multi".into(), + num_nodes: 8, + num_rounds: 60, + ticks_per_round: 3, + actors_per_node: 0, + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 0, name: "alpha".into() }), + (5, SimAction::RegisterName { node_idx: 3, name: "beta".into() }), + (5, SimAction::RegisterName { node_idx: 6, name: "gamma".into() }), + ], + ..DistributionSimConfig::default() + }, + 3, + ), + // 5 nodes, register + kill owner, verify tombstone propagates + ( + DistributionSimConfig { + name: "reg-ec-death".into(), + num_nodes: 5, + num_rounds: 80, + ticks_per_round: 3, + actors_per_node: 0, + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 0, name: "ephemeral".into() }), + ], + kill_schedule: vec![(15, 0)], + ..DistributionSimConfig::default() + }, + 1, // tombstoned entry still counts as registry_size + ), + ]; + + for (config, min_size) in configs { + let name = config.name.clone(); + let (trace, _) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + let result = check_registry_propagation(&trace, min_size); + assert!( + result.passed, + "[{name}] registry should converge: {}", + result.actual + ); + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// 5. Multiple deaths don't violate invariants +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn cascading_deaths_maintain_invariants() { + // 8 nodes, 3 die in sequence + let config = DistributionSimConfig { + name: "cascade-invariants".into(), + num_nodes: 8, + num_rounds: 100, + ticks_per_round: 3, + actors_per_node: 2, + cache_capacity: 200, + kill_schedule: vec![(10, 1), (20, 3), (30, 5)], + ..DistributionSimConfig::default() + }; + + let (trace, _) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + let rt_result = check_routing_table_bounded(&trace); + assert!( + rt_result.passed, + "routing table bounded after cascading deaths: {}", + rt_result.actual + ); + + let cache_result = check_cache_bounded(&trace, 200); + assert!( + cache_result.passed, + "cache bounded after cascading deaths: {}", + cache_result.actual + ); + + let repair_result = check_repair_queue_populated(&trace, 10); + assert!( + repair_result.passed, + "repair queue populated after first death: {}", + repair_result.actual + ); +} + +// ──────────────────────────────────────────────────────────────────────────── +// 6. Asymmetric partition + registry — one-way reachability +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn asymmetric_partition_registry_converges_after_heal() { + // Given: 6 nodes, asymmetric partition: A→B blocked, B→A works. + // Node 0 (side A) and node 3 (side B) each register a name. + // After heal, all should converge. + let config = DistributionSimConfig { + name: "asymmetric-partition-registry".into(), + num_nodes: 6, + num_rounds: 100, + ticks_per_round: 3, + actors_per_node: 0, + network_faults: vec![ + NetworkFault::Partition { + round: 10, + partition: Partition { + side_a: vec![0, 1, 2], + side_b: vec![3, 4, 5], + asymmetric: true, // A→B blocked, B→A works + }, + }, + NetworkFault::Heal { round: 40 }, + ], + action_schedule: vec![ + (12, SimAction::RegisterName { node_idx: 0, name: "from-a".into() }), + (12, SimAction::RegisterName { node_idx: 3, name: "from-b".into() }), + ], + swim: distribution::swim::probe::SwimConfig { + probe_interval: 1, + probe_timeout: 3, + indirect_probes: 1, + suspicion_timeout: 200, + dead_reprobe_interval: 10, + }, + ..DistributionSimConfig::default() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // After healing, all nodes should resolve both names + let alive_nodes: Vec<_> = nodes.iter().filter_map(|n| n.as_ref()).collect(); + for node in &alive_nodes { + // Side B's name should have been reachable from side A even during partition + // (B→A works), so "from-b" should propagate to everyone. + // "from-a" might need post-heal gossip to reach side B. + assert!( + node.resolve_name("from-b").is_some(), + "all nodes should resolve 'from-b' (B→A was always open)" + ); + } + + // After 60 rounds of healed connectivity, "from-a" should also propagate + let resolved_a: Vec<_> = alive_nodes + .iter() + .filter(|n| n.resolve_name("from-a").is_some()) + .collect(); + assert!( + resolved_a.len() >= 4, + "at least 4 of 6 nodes should resolve 'from-a' after partition heals, got {}", + resolved_a.len() + ); +} + +// ──────────────────────────────────────────────────────────────────────────── +// 7. Revived node re-registers — new registration overwrites tombstone +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn revived_node_re_registration_overwrites_tombstone() { + use swactor::actor::ActorAddress; + let new_actor = ActorAddress::new_random(); + + // Given: 5 nodes, node 2 registers "svc", is killed, revived with fresh state, + // then node 3 re-registers "svc" with a new actor. + // (Don't kill node 0 since it's the join seed.) + let config = DistributionSimConfig { + name: "revive-reregister".into(), + num_nodes: 5, + num_rounds: 120, + ticks_per_round: 3, + actors_per_node: 0, + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 2, name: "svc".into() }), + // After death + revive, a different surviving node re-registers + (60, SimAction::RegisterNameWithActor { node_idx: 3, name: "svc".into(), actor: new_actor }), + ], + kill_schedule: vec![(15, 2)], + revive_schedule: vec![(40, 2)], + ..DistributionSimConfig::default() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Then: all alive nodes should resolve "svc" to the new registration + let alive_nodes: Vec<_> = nodes.iter().filter_map(|n| n.as_ref()).collect(); + for node in &alive_nodes { + assert!( + node.resolve_name("svc").is_some(), + "all nodes should resolve 'svc' after re-registration by surviving node" + ); + assert_eq!( + node.resolve_name("svc").unwrap().0, + new_actor, + "all nodes should resolve 'svc' to the new actor" + ); + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// 8. Registry convergence is monotonic — divergence doesn't increase +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn registry_convergence_is_monotonic_in_stable_cluster() { + // Given: 8-node cluster, 4 names registered at round 5, no faults + let config = DistributionSimConfig { + name: "registry-monotonic".into(), + num_nodes: 8, + num_rounds: 60, + ticks_per_round: 3, + actors_per_node: 0, + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 0, name: "alpha".into() }), + (5, SimAction::RegisterName { node_idx: 2, name: "beta".into() }), + (5, SimAction::RegisterName { node_idx: 4, name: "gamma".into() }), + (5, SimAction::RegisterName { node_idx: 6, name: "delta".into() }), + ], + ..DistributionSimConfig::default() + }; + + let (trace, _) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Measure "divergence" = number of alive nodes with registry_size < 4 + // Once it reaches 0, it should never increase again + let mut reached_convergence = false; + let mut post_convergence_divergence = 0; + + for round_snaps in &trace.snapshots_per_round { + let alive_with_full_registry = round_snaps + .iter() + .filter(|(_, s)| s.is_alive && s.registry_size >= 4) + .count(); + let alive_count = round_snaps.iter().filter(|(_, s)| s.is_alive).count(); + let divergent = alive_count - alive_with_full_registry; + + if divergent == 0 && alive_count > 0 { + reached_convergence = true; + } else if reached_convergence && divergent > 0 { + post_convergence_divergence += 1; + } + } + + assert!( + reached_convergence, + "registry should converge (all alive nodes see all 4 names)" + ); + assert_eq!( + post_convergence_divergence, 0, + "once converged, registry should not diverge again in a stable cluster" + ); +} + +// ──────────────────────────────────────────────────────────────────────────── +// 9. Large cluster registry stress test +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn large_cluster_registry_converges() { + // Given: 15-node cluster, 5 names registered on different nodes, 2 deaths + let config = DistributionSimConfig { + name: "large-cluster-registry".into(), + num_nodes: 15, + num_rounds: 80, + ticks_per_round: 3, + actors_per_node: 0, + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 0, name: "alpha".into() }), + (5, SimAction::RegisterName { node_idx: 3, name: "beta".into() }), + (5, SimAction::RegisterName { node_idx: 6, name: "gamma".into() }), + (5, SimAction::RegisterName { node_idx: 9, name: "delta".into() }), + (5, SimAction::RegisterName { node_idx: 12, name: "epsilon".into() }), + ], + kill_schedule: vec![(20, 0), (20, 3)], + ..DistributionSimConfig::default() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + let survivors: Vec<_> = nodes.iter().filter_map(|n| n.as_ref()).collect(); + assert_eq!(survivors.len(), 13, "13 of 15 nodes should survive"); + + // Names owned by dead nodes should be tombstoned + for node in &survivors { + assert!( + node.resolve_name("alpha").is_none(), + "alpha (owned by dead node 0) should be tombstoned" + ); + assert!( + node.resolve_name("beta").is_none(), + "beta (owned by dead node 3) should be tombstoned" + ); + } + + // Names owned by surviving nodes should resolve + for node in &survivors { + for name in &["gamma", "delta", "epsilon"] { + assert!( + node.resolve_name(name).is_some(), + "'{name}' (owned by surviving node) should resolve across 15-node cluster" + ); + } + } + + // Registry propagation: all survivors should have all 5 registry entries + // (2 tombstoned + 3 alive) + let result = check_registry_propagation(&trace, 5); + assert!( + result.passed, + "all 5 registry entries should propagate in 15-node cluster: {}", + result.actual + ); +} + +// ──────────────────────────────────────────────────────────────────────────── +// 10. Three-way partition: cluster splits into 3 groups, heals, converges +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn three_way_partition_heals_and_converges() { + // Given: 9 nodes split into 3 groups {0,1,2}, {3,4,5}, {6,7,8}. + // Each group registers a name during the partition. After healing, + // all 9 nodes should converge on all 3 names. + let config = DistributionSimConfig { + name: "three-way-partition".into(), + num_nodes: 9, + num_rounds: 140, + ticks_per_round: 3, + actors_per_node: 0, + network_faults: vec![ + // Partition A vs B + NetworkFault::Partition { + round: 10, + partition: Partition { + side_a: vec![0, 1, 2], + side_b: vec![3, 4, 5, 6, 7, 8], + asymmetric: false, + }, + }, + // Partition B vs C (stacks with the above: now 3 groups isolated) + NetworkFault::Partition { + round: 10, + partition: Partition { + side_a: vec![3, 4, 5], + side_b: vec![6, 7, 8], + asymmetric: false, + }, + }, + NetworkFault::Heal { round: 60 }, + ], + action_schedule: vec![ + (20, SimAction::RegisterName { node_idx: 0, name: "group-a-svc".into() }), + (20, SimAction::RegisterName { node_idx: 3, name: "group-b-svc".into() }), + (20, SimAction::RegisterName { node_idx: 6, name: "group-c-svc".into() }), + ], + swim: distribution::swim::probe::SwimConfig { + probe_interval: 1, + probe_timeout: 3, + indirect_probes: 1, + // Must exceed partition duration (50 rounds × 3 ticks = 150 ticks) + suspicion_timeout: 200, + dead_reprobe_interval: 10, + }, + ..DistributionSimConfig::default() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + let alive_count = nodes.iter().filter(|n| n.is_some()).count(); + assert_eq!(alive_count, 9, "all 9 nodes should survive the three-way partition"); + + // After healing + gossip, all 3 names should be resolvable from every node + for (i, node) in nodes.iter().enumerate() { + if let Some(node) = node { + for name in &["group-a-svc", "group-b-svc", "group-c-svc"] { + assert!( + node.resolve_name(name).is_some(), + "node {i} should resolve '{name}' after three-way partition heals" + ); + } + } + } +} diff --git a/crates/simulation/tests/distribution_registry.rs b/crates/simulation/tests/distribution_registry.rs new file mode 100644 index 0000000..8084e65 --- /dev/null +++ b/crates/simulation/tests/distribution_registry.rs @@ -0,0 +1,752 @@ +//! Registry simulation tests — cluster registry CRDT behavior under gossip. +//! +//! Tests that registry names propagate, converge, and resolve correctly +//! across the cluster under various fault conditions. + +use simulation::distribution::properties::{ + check_registry_propagation, check_registry_tombstones, +}; +use simulation::distribution::sim::{ + run_simulation_with_nodes, DistributionSimConfig, DistTrace, NetworkFault, Partition, + SimAction, +}; + +fn maybe_save_trace(trace: &DistTrace) { + if let Ok(dir) = std::env::var("SWACTOR_TRACE_DIR") { + std::fs::create_dir_all(&dir).ok(); + let filename = format!( + "{}/{}.trace.json", + dir, + trace.name.to_lowercase().replace(' ', "_").replace(['(', ')'], "") + ); + let json = serde_json::to_string_pretty(trace).expect("trace serialization failed"); + std::fs::write(&filename, json).expect("trace write failed"); + } +} + +fn default_config() -> DistributionSimConfig { + DistributionSimConfig { + actors_per_node: 0, // Registry tests don't need actors + ..DistributionSimConfig::default() + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// 1. Registry name converges across 5-node cluster via gossip piggyback +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn registry_name_converges_across_cluster() { + // Given: 5-node cluster, node 0 registers "counter" at round 5 + let config = DistributionSimConfig { + name: "registry-convergence".into(), + num_nodes: 5, + num_rounds: 50, + ticks_per_round: 3, + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 0, name: "counter".into() }), + ], + ..default_config() + }; + + // When: we run the simulation + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Then: all alive nodes should have the registry entry + let result = check_registry_propagation(&trace, 1); + assert!( + result.passed, + "all nodes should see the 'counter' name: {}", + result.actual + ); + + // And: all nodes should resolve "counter" to the same actor + let resolutions: Vec<_> = nodes + .iter() + .filter_map(|n| n.as_ref()) + .map(|n| n.resolve_name("counter")) + .collect(); + + assert!( + resolutions.iter().all(|r| r.is_some()), + "all nodes should resolve 'counter', got: {resolutions:?}" + ); + + // All should agree on the same actor address + let first = resolutions[0].unwrap().0; + assert!( + resolutions.iter().all(|r| r.unwrap().0 == first), + "all nodes should agree on the same actor for 'counter'" + ); +} + +// ──────────────────────────────────────────────────────────────────────────── +// 2. Split-brain naming — two sides register same name during partition +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn split_brain_naming_converges_after_partition_heals() { + // Given: 6-node cluster, partition {0,1,2} vs {3,4,5} at round 10 + // Node 0 registers "leader" at round 12, node 3 registers "leader" at round 12 + // Heal at round 40 + let config = DistributionSimConfig { + name: "split-brain-registry".into(), + num_nodes: 6, + num_rounds: 100, + ticks_per_round: 3, + network_faults: vec![ + NetworkFault::Partition { + round: 10, + partition: Partition { + side_a: vec![0, 1, 2], + side_b: vec![3, 4, 5], + asymmetric: false, + }, + }, + NetworkFault::Heal { round: 40 }, + ], + action_schedule: vec![ + (12, SimAction::RegisterName { node_idx: 0, name: "leader".into() }), + (12, SimAction::RegisterName { node_idx: 3, name: "leader".into() }), + ], + swim: distribution::swim::probe::SwimConfig { + probe_interval: 1, + probe_timeout: 3, + indirect_probes: 1, + // High enough that no node reaches Dead during the 30-round partition. + // Nodes go Suspect → back to Alive when partition heals, triggering + // re_disseminate_all which propagates both sides' registry entries. + suspicion_timeout: 200, + dead_reprobe_interval: 10, + }, + ..default_config() + }; + + // When: we run the simulation + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Then: all alive nodes should resolve "leader" to the same value (LWW winner) + let resolutions: Vec<_> = nodes + .iter() + .filter_map(|n| n.as_ref()) + .map(|n| n.resolve_name("leader")) + .collect(); + + // All should resolve to Some (the LWW winner) + let resolved: Vec<_> = resolutions.iter().filter_map(|r| r.as_ref()).collect(); + assert!( + resolved.len() >= 4, + "at least 4 of 6 nodes should resolve 'leader', got {} out of {}", + resolved.len(), + resolutions.len() + ); + + // All resolving nodes should agree on the same actor + if resolved.len() >= 2 { + let first_actor = resolved[0].0; + let agree = resolved.iter().all(|r| r.0 == first_actor); + assert!( + agree, + "all nodes resolving 'leader' should agree on the same actor (LWW winner)" + ); + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// 3. Tombstone propagation on node death +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn tombstone_propagates_when_name_owner_dies() { + // Given: 5-node cluster, node 0 registers "svc" at round 5, killed at round 15 + let config = DistributionSimConfig { + name: "tombstone-propagation".into(), + num_nodes: 5, + num_rounds: 80, + ticks_per_round: 3, + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 0, name: "svc".into() }), + ], + kill_schedule: vec![(15, 0)], + ..default_config() + }; + + // When: we run the simulation + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Then: surviving nodes should have tombstoned "svc" + let result = check_registry_tombstones(&trace, 1); + assert!( + result.passed, + "survivors should have tombstones after node death: {}", + result.actual + ); + + // And: resolve_name("svc") should return None on all survivors + let resolutions: Vec<_> = nodes + .iter() + .filter_map(|n| n.as_ref()) + .map(|n| n.resolve_name("svc")) + .collect(); + assert!( + resolutions.iter().all(|r| r.is_none()), + "all survivors should resolve 'svc' to None after owner dies, got: {resolutions:?}" + ); +} + +// ──────────────────────────────────────────────────────────────────────────── +// 4. Rapid re-registration converges to latest value +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn rapid_re_registration_converges_to_latest() { + // Given: 5-node cluster, node 0 registers "svc" three times rapidly + use swactor::actor::ActorAddress; + let actor_a = ActorAddress::new_random(); + let actor_b = ActorAddress::new_random(); + let actor_c = ActorAddress::new_random(); + + let config = DistributionSimConfig { + name: "rapid-reregister".into(), + num_nodes: 5, + num_rounds: 60, + ticks_per_round: 3, + action_schedule: vec![ + (5, SimAction::RegisterNameWithActor { node_idx: 0, name: "svc".into(), actor: actor_a }), + (6, SimAction::RegisterNameWithActor { node_idx: 0, name: "svc".into(), actor: actor_b }), + (7, SimAction::RegisterNameWithActor { node_idx: 0, name: "svc".into(), actor: actor_c }), + ], + ..default_config() + }; + + // When: we run the simulation + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Then: all nodes should resolve "svc" to actor_c (the latest registration) + let resolutions: Vec<_> = nodes + .iter() + .filter_map(|n| n.as_ref()) + .map(|n| n.resolve_name("svc")) + .collect(); + + assert!( + resolutions.iter().all(|r| r.is_some()), + "all nodes should resolve 'svc'" + ); + + assert!( + resolutions.iter().all(|r| r.unwrap().0 == actor_c), + "all nodes should converge to the latest registration (actor_c)" + ); +} + +// ──────────────────────────────────────────────────────────────────────────── +// 5. Simultaneous registration of same name on different nodes +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn simultaneous_registration_converges_deterministically() { + // Given: 5-node cluster, node 0 and node 3 both register "mutex" at round 5 + use swactor::actor::ActorAddress; + let actor_a = ActorAddress::new_random(); + let actor_b = ActorAddress::new_random(); + + let config = DistributionSimConfig { + name: "simultaneous-register".into(), + num_nodes: 5, + num_rounds: 60, + ticks_per_round: 3, + action_schedule: vec![ + (5, SimAction::RegisterNameWithActor { node_idx: 0, name: "mutex".into(), actor: actor_a }), + (5, SimAction::RegisterNameWithActor { node_idx: 3, name: "mutex".into(), actor: actor_b }), + ], + ..default_config() + }; + + // When: we run the simulation + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Then: all nodes should agree on one winner + let resolutions: Vec<_> = nodes + .iter() + .filter_map(|n| n.as_ref()) + .map(|n| n.resolve_name("mutex")) + .collect(); + + assert!( + resolutions.iter().all(|r| r.is_some()), + "all nodes should resolve 'mutex'" + ); + + // All should agree on the same actor (whichever won the LWW tiebreaker) + let first_actor = resolutions[0].unwrap().0; + assert!( + resolutions.iter().all(|r| r.unwrap().0 == first_actor), + "all nodes should agree on the LWW winner for 'mutex'" + ); +} + +// ──────────────────────────────────────────────────────────────────────────── +// 6. Multiple names propagate correctly +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn multiple_names_from_different_nodes_all_propagate() { + // Given: 5-node cluster, each node registers a unique name + let config = DistributionSimConfig { + name: "multi-name-propagation".into(), + num_nodes: 5, + num_rounds: 60, + ticks_per_round: 3, + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 0, name: "svc-0".into() }), + (5, SimAction::RegisterName { node_idx: 1, name: "svc-1".into() }), + (5, SimAction::RegisterName { node_idx: 2, name: "svc-2".into() }), + (5, SimAction::RegisterName { node_idx: 3, name: "svc-3".into() }), + (5, SimAction::RegisterName { node_idx: 4, name: "svc-4".into() }), + ], + ..default_config() + }; + + // When: we run the simulation + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Then: all 5 nodes should have all 5 registry entries + let result = check_registry_propagation(&trace, 5); + assert!( + result.passed, + "all nodes should have all 5 registry entries: {}", + result.actual + ); + + // And: each node should resolve all 5 names + for node in nodes.iter().filter_map(|n| n.as_ref()) { + for i in 0..5 { + let name = format!("svc-{i}"); + assert!( + node.resolve_name(&name).is_some(), + "every node should resolve '{name}'" + ); + } + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// 7. Explicit unregister propagates to all nodes +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn explicit_unregister_propagates_to_all_nodes() { + // Given: 5-node cluster, node 0 registers "svc" at round 5, unregisters at round 15 + let config = DistributionSimConfig { + name: "explicit-unregister".into(), + num_nodes: 5, + num_rounds: 60, + ticks_per_round: 3, + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 0, name: "svc".into() }), + (15, SimAction::UnregisterName { node_idx: 0, name: "svc".into() }), + ], + ..default_config() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Then: all nodes should resolve "svc" to None (tombstoned) + for (i, node) in nodes.iter().filter_map(|n| n.as_ref()).enumerate() { + assert!( + node.resolve_name("svc").is_none(), + "node {i} should resolve 'svc' to None after unregister" + ); + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// 8. Re-registration after tombstone overwrites the tombstone +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn re_registration_after_tombstone_succeeds() { + // Given: node 0 registers "svc", then it's killed (tombstoned), + // then node 1 re-registers "svc" with a new actor + use swactor::actor::ActorAddress; + let new_actor = ActorAddress::new_random(); + + let config = DistributionSimConfig { + name: "re-register-after-tombstone".into(), + num_nodes: 5, + num_rounds: 100, + ticks_per_round: 3, + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 0, name: "svc".into() }), + // Node 1 re-registers "svc" well after node 0 dies and tombstone propagates + (50, SimAction::RegisterNameWithActor { node_idx: 1, name: "svc".into(), actor: new_actor }), + ], + kill_schedule: vec![(15, 0)], + ..default_config() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Then: all survivors should resolve "svc" to the new actor from node 1 + let resolutions: Vec<_> = nodes + .iter() + .filter_map(|n| n.as_ref()) + .map(|n| n.resolve_name("svc")) + .collect(); + + assert!( + resolutions.iter().all(|r| r.is_some()), + "all survivors should resolve 'svc' to the new registration, got: {resolutions:?}" + ); + + assert!( + resolutions.iter().all(|r| r.unwrap().0 == new_actor), + "all survivors should resolve 'svc' to the new actor" + ); +} + +// ──────────────────────────────────────────────────────────────────────────── +// 9. Multiple names from same node, kill node, all tombstoned +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn all_names_tombstoned_when_owner_dies() { + // Given: node 0 registers 3 names, then is killed + let config = DistributionSimConfig { + name: "multi-name-tombstone".into(), + num_nodes: 5, + num_rounds: 80, + ticks_per_round: 3, + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 0, name: "alpha".into() }), + (5, SimAction::RegisterName { node_idx: 0, name: "beta".into() }), + (5, SimAction::RegisterName { node_idx: 0, name: "gamma".into() }), + ], + kill_schedule: vec![(15, 0)], + ..default_config() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Then: all survivors should resolve all 3 names to None + for node in nodes.iter().filter_map(|n| n.as_ref()) { + for name in &["alpha", "beta", "gamma"] { + assert!( + node.resolve_name(name).is_none(), + "all names should be tombstoned after owner dies, but '{}' still resolves", + name + ); + } + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// 10. Graceful leave tombstones the leaving node's names +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn graceful_leave_tombstones_registry_names() { + // Given: node 0 registers "svc", then does a graceful leave + let config = DistributionSimConfig { + name: "graceful-leave-registry".into(), + num_nodes: 5, + num_rounds: 80, + ticks_per_round: 3, + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 0, name: "svc".into() }), + (20, SimAction::GracefulLeave { node_idx: 0 }), + ], + ..default_config() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Then: all survivors should resolve "svc" to None (tombstoned via death notification) + let resolutions: Vec<_> = nodes + .iter() + .filter_map(|n| n.as_ref()) + .map(|n| n.resolve_name("svc")) + .collect(); + + assert!( + resolutions.iter().all(|r| r.is_none()), + "all survivors should resolve 'svc' to None after graceful leave, got: {resolutions:?}" + ); +} + +// ──────────────────────────────────────────────────────────────────────────── +// 11. Piggyback contention — kills + registrations compete for bandwidth +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn piggyback_contention_both_propagate() { + // Given: 10-node cluster, kill 2 nodes + register 3 names simultaneously + // Both membership death updates and registry entries share piggyback bandwidth + let config = DistributionSimConfig { + name: "piggyback-contention".into(), + num_nodes: 10, + num_rounds: 100, + ticks_per_round: 3, + actors_per_node: 0, + action_schedule: vec![ + (10, SimAction::RegisterName { node_idx: 0, name: "alpha".into() }), + (10, SimAction::RegisterName { node_idx: 3, name: "beta".into() }), + (10, SimAction::RegisterName { node_idx: 6, name: "gamma".into() }), + ], + kill_schedule: vec![(10, 2), (10, 5)], + ..default_config() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Then: all survivors should have all 3 registry names + let result = check_registry_propagation(&trace, 3); + assert!( + result.passed, + "all 3 names should propagate despite contention with death updates: {}", + result.actual + ); + + // And: all survivors should resolve all 3 names + for node in nodes.iter().filter_map(|n| n.as_ref()) { + for name in &["alpha", "beta", "gamma"] { + assert!( + node.resolve_name(name).is_some(), + "survivor should resolve '{}' despite piggyback contention", + name + ); + } + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// 12. Registry convergence under message loss +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn registry_converges_despite_message_loss() { + // Given: 5-node cluster with a loss window during name registration, + // then clean connectivity for convergence. + let config = DistributionSimConfig { + name: "registry-under-loss".into(), + num_nodes: 5, + num_rounds: 100, + ticks_per_round: 3, + registry_dissemination_lambda: Some(5), + network_faults: vec![ + // Loss window: 30% drops during registration phase + NetworkFault::SetDropRate { round: 3, rate: 0.30 }, + // Restore clean connectivity, forcing convergence via Alive transitions + NetworkFault::SetDropRate { round: 30, rate: 0.0 }, + ], + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 0, name: "alpha".into() }), + (5, SimAction::RegisterName { node_idx: 2, name: "beta".into() }), + (5, SimAction::RegisterName { node_idx: 4, name: "gamma".into() }), + ], + swim: distribution::swim::probe::SwimConfig { + probe_interval: 1, + probe_timeout: 3, + indirect_probes: 3, + // High timeout prevents false deaths during the loss window. + suspicion_timeout: 200, + dead_reprobe_interval: 15, + }, + ..default_config() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Then: all surviving nodes should resolve all 3 names despite packet loss + let alive_nodes: Vec<_> = nodes.iter().filter_map(|n| n.as_ref()).collect(); + for node in &alive_nodes { + for name in &["alpha", "beta", "gamma"] { + assert!( + node.resolve_name(name).is_some(), + "all nodes should resolve '{}' despite 20% message loss", + name + ); + } + } + + // All should agree on the same actor for each name + for name in &["alpha", "beta", "gamma"] { + let first = alive_nodes[0].resolve_name(name).unwrap().0; + assert!( + alive_nodes.iter().all(|n| n.resolve_name(name).unwrap().0 == first), + "all nodes should agree on actor for '{}' despite message loss", + name + ); + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// 13. Multiple name-owning nodes killed simultaneously +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn simultaneous_kill_of_multiple_name_owners() { + // Given: 7-node cluster, nodes 0-2 each own a name, all 3 killed at round 15 + let config = DistributionSimConfig { + name: "multi-owner-kill".into(), + num_nodes: 7, + num_rounds: 80, + ticks_per_round: 3, + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 0, name: "svc-a".into() }), + (5, SimAction::RegisterName { node_idx: 1, name: "svc-b".into() }), + (5, SimAction::RegisterName { node_idx: 2, name: "svc-c".into() }), + ], + kill_schedule: vec![(15, 0), (15, 1), (15, 2)], + ..default_config() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Then: all 4 survivors should resolve all 3 names to None (tombstoned) + let survivors: Vec<_> = nodes.iter().filter_map(|n| n.as_ref()).collect(); + assert_eq!(survivors.len(), 4, "4 of 7 nodes should survive"); + + for node in &survivors { + for name in &["svc-a", "svc-b", "svc-c"] { + assert!( + node.resolve_name(name).is_none(), + "survivor should resolve '{}' to None after owner died", + name + ); + } + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// 14. Name owner suspected but recovers — registry entry survives +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn suspected_name_owner_recovers_and_registry_survives() { + // Given: 5-node cluster, node 0 registers "svc", + // then a brief partition isolates node 0 (becomes Suspect, recovers before Dead) + let config = DistributionSimConfig { + name: "suspect-recovery-registry".into(), + num_nodes: 5, + num_rounds: 80, + ticks_per_round: 3, + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 0, name: "svc".into() }), + ], + // Brief partition: isolate node 0 for 10 rounds (not long enough to reach Dead) + network_faults: vec![ + NetworkFault::Partition { + round: 15, + partition: Partition { + side_a: vec![0], + side_b: vec![1, 2, 3, 4], + asymmetric: false, + }, + }, + NetworkFault::Heal { round: 25 }, + ], + swim: distribution::swim::probe::SwimConfig { + probe_interval: 1, + probe_timeout: 3, + indirect_probes: 1, + // Node stays Suspect during the 10-round partition (30 ticks < 200) + suspicion_timeout: 200, + dead_reprobe_interval: 10, + }, + ..default_config() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + // Then: all nodes should still resolve "svc" (node 0 never died, only suspected) + let resolutions: Vec<_> = nodes + .iter() + .filter_map(|n| n.as_ref()) + .map(|n| n.resolve_name("svc")) + .collect(); + + assert!( + resolutions.iter().all(|r| r.is_some()), + "all nodes should resolve 'svc' after owner recovers from Suspect, got: {resolutions:?}" + ); +} + +// ──────────────────────────────────────────────────────────────────────────── +// 15. Rapid churn: interleaved kills and registrations +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn registry_correct_under_rapid_churn() { + // Given: 8-node cluster with interleaved kills and registrations + let config = DistributionSimConfig { + name: "rapid-churn-registry".into(), + num_nodes: 8, + num_rounds: 100, + ticks_per_round: 3, + action_schedule: vec![ + (5, SimAction::RegisterName { node_idx: 0, name: "svc-0".into() }), + (5, SimAction::RegisterName { node_idx: 1, name: "svc-1".into() }), + (8, SimAction::RegisterName { node_idx: 4, name: "svc-4".into() }), + // Node 3 registers AFTER nodes 0 and 1 are killed + (20, SimAction::RegisterName { node_idx: 3, name: "svc-3".into() }), + // Node 5 takes over "svc-0" after original owner dies + (30, SimAction::RegisterName { node_idx: 5, name: "svc-0".into() }), + ], + kill_schedule: vec![(10, 0), (10, 1), (25, 2)], + ..default_config() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + maybe_save_trace(&trace); + + let survivors: Vec<_> = nodes.iter().filter_map(|n| n.as_ref()).collect(); + assert_eq!(survivors.len(), 5, "5 of 8 nodes should survive"); + + // svc-0 should resolve to node 5's re-registration (not tombstoned) + for node in &survivors { + assert!( + node.resolve_name("svc-0").is_some(), + "svc-0 should resolve (re-registered by node 5 after owner death)" + ); + } + + // svc-1 should be tombstoned (node 1 died, no re-registration) + for node in &survivors { + assert!( + node.resolve_name("svc-1").is_none(), + "svc-1 should be tombstoned (owner died, not re-registered)" + ); + } + + // svc-3 should resolve (registered after kills, node 3 alive) + for node in &survivors { + assert!( + node.resolve_name("svc-3").is_some(), + "svc-3 should resolve (registered after kills)" + ); + } + + // svc-4 should resolve (node 4 alive throughout) + for node in &survivors { + assert!( + node.resolve_name("svc-4").is_some(), + "svc-4 should resolve (owner alive throughout)" + ); + } +} diff --git a/scripts/sim-dashboard.sh b/scripts/sim-dashboard.sh new file mode 100755 index 0000000..a062a9f --- /dev/null +++ b/scripts/sim-dashboard.sh @@ -0,0 +1,19 @@ +#!/usr/bin/env bash +set -euo pipefail + +DIR="$(cd "$(dirname "${1:-.}")" && pwd)/$(basename "${1:-traces}")" +PORT="${2:-8080}" + +rm -rf "$DIR" +mkdir -p "$DIR" + +echo "Running distribution sim tests with trace export..." +SWACTOR_TRACE_DIR="$DIR" cargo test -p simulation \ + --test distribution_registry \ + --test distribution_lifecycle \ + --test distribution_properties || echo "WARNING: some tests failed (traces from passing tests are still available)" + +COUNT=$(find "$DIR" -name '*.trace.json' 2>/dev/null | wc -l) +echo "$COUNT traces in $DIR/" +echo "Dashboard at http://localhost:$PORT" +cargo run -p simulation-dashboard --example replay -- "$DIR" "$PORT"