From 80d8773f98ae44a8a32f7f180b86f0cfca0390fb Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Sun, 15 Feb 2026 16:45:51 +0700 Subject: [PATCH] fix: no more flaky tests --- crates/bin-runner/build.rs | 28 ++ crates/bin-runner/tests/wasm_actor.rs | 5 +- crates/distribution/src/node.rs | 1 + crates/distribution/src/registry.rs | 1 + crates/distribution/src/swim/member_list.rs | 5 +- crates/distribution/tests/common/mod.rs | 207 ++++++++++++++ crates/distribution/tests/node_integration.rs | 137 ++-------- crates/distribution/tests/registry.rs | 253 ++++-------------- crates/distribution/tests/swim_node.rs | 24 +- crates/simulation/src/gossip/sim.rs | 17 +- crates/simulation/src/topology.rs | 21 +- crates/simulation/tests/cluster_scenarios.rs | 75 ++++++ crates/simulation/tests/gossip_properties.rs | 28 +- .../SWIM_FLAKY_TEST_POSTMORTEM.md | 218 +++++++++++++++ 14 files changed, 655 insertions(+), 365 deletions(-) create mode 100644 crates/bin-runner/build.rs create mode 100644 crates/distribution/tests/common/mod.rs create mode 100644 docs/development_history/distribution/SWIM_FLAKY_TEST_POSTMORTEM.md diff --git a/crates/bin-runner/build.rs b/crates/bin-runner/build.rs new file mode 100644 index 0000000..c1c291d --- /dev/null +++ b/crates/bin-runner/build.rs @@ -0,0 +1,28 @@ +use std::path::Path; +use std::process::Command; + +fn main() { + let manifest_dir = env!("CARGO_MANIFEST_DIR"); + let guests_dir = Path::new(manifest_dir).join("tests/guests"); + + for guest in &["echo", "double", "silent"] { + let guest_dir = guests_dir.join(guest); + + println!( + "cargo:rerun-if-changed={}", + guest_dir.join("src/lib.rs").display() + ); + println!( + "cargo:rerun-if-changed={}", + guest_dir.join("Cargo.toml").display() + ); + + let status = Command::new("cargo") + .args(["build", "--target", "wasm32-unknown-unknown", "--release"]) + .current_dir(&guest_dir) + .status() + .unwrap_or_else(|e| panic!("failed to run cargo build for {guest} guest: {e}")); + + assert!(status.success(), "failed to build {guest} guest"); + } +} diff --git a/crates/bin-runner/tests/wasm_actor.rs b/crates/bin-runner/tests/wasm_actor.rs index 717df9e..341a2d8 100644 --- a/crates/bin-runner/tests/wasm_actor.rs +++ b/crates/bin-runner/tests/wasm_actor.rs @@ -1,7 +1,7 @@ use swactor::actor::{ActorAddress, ActorInterface}; use swactor::runtime::{Ctx, Runtime, RuntimeConfig}; use swactor_bin_runner::{ByteMessage, SharedEngine, WasmActor, WasmActorBuilder, WasmActorError}; -use swactor_std::CtxWatching; +use swactor_std::{CtxWatching, StdExtension}; use proptest::prelude::*; @@ -719,7 +719,8 @@ impl ActorInterface for DeathCounter { #[test] fn watch_notification() { let engine = SharedEngine::new().unwrap(); - let rt = Runtime::new(RuntimeConfig::default()); + let rt = Runtime::new(RuntimeConfig::default()) + .with_extension(std::sync::Arc::new(StdExtension::new())); let count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); let target = WasmActorBuilder::new(engine, guest_wasm("silent")).build().unwrap(); diff --git a/crates/distribution/src/node.rs b/crates/distribution/src/node.rs index da42983..d7e4690 100644 --- a/crates/distribution/src/node.rs +++ b/crates/distribution/src/node.rs @@ -19,6 +19,7 @@ use crate::swim::probe::SwimConfig; use crate::types::{MemberState, NodeId, NodeRecord}; /// Configuration for a distributed node. +#[derive(Clone)] pub struct DistributedNodeConfig { pub swim: SwimConfig, pub cache_capacity: usize, diff --git a/crates/distribution/src/registry.rs b/crates/distribution/src/registry.rs index 72f3690..e132709 100644 --- a/crates/distribution/src/registry.rs +++ b/crates/distribution/src/registry.rs @@ -14,6 +14,7 @@ use crate::types::NodeId; // ─── Configuration ────────────────────────────────────────────────────────── /// Configuration for the cluster registry. +#[derive(Clone)] pub struct RegistryConfig { /// Maximum number of events to buffer before dropping old ones. pub max_events: usize, diff --git a/crates/distribution/src/swim/member_list.rs b/crates/distribution/src/swim/member_list.rs index fc9bdff..dfaef15 100644 --- a/crates/distribution/src/swim/member_list.rs +++ b/crates/distribution/src/swim/member_list.rs @@ -146,10 +146,11 @@ impl MemberList { false } - /// Mark a node as dead. + /// Mark a node as dead. Only transitions from Suspect → Dead, + /// enforcing the SWIM lifecycle invariant (Alive → Suspect → Dead). pub fn declare_dead(&mut self, node_id: NodeId) -> bool { if let Some(entry) = self.members.get_mut(&node_id) { - if entry.state != MemberState::Dead { + if entry.state == MemberState::Suspect { entry.state = MemberState::Dead; return true; } diff --git a/crates/distribution/tests/common/mod.rs b/crates/distribution/tests/common/mod.rs new file mode 100644 index 0000000..89d2c63 --- /dev/null +++ b/crates/distribution/tests/common/mod.rs @@ -0,0 +1,207 @@ +//! Shared test harness for N-node `DistributedNode` tests. +//! +//! `TestCluster` makes sender-misattribution structurally impossible by +//! tagging every response with the responder's index, mirroring the +//! simulation crate's `deliver_actions_tagged_with_net`. + +use distribution::node::{DistributedNode, DistributedNodeConfig}; +use distribution::registry::RegistryConfig; +use distribution::swim::node::NodeAction; +use distribution::swim::probe::SwimConfig; +use distribution::types::NodeId; +use std::ops::{Index, IndexMut}; + +/// Default test config shared across all integration tests. +pub fn test_config() -> DistributedNodeConfig { + DistributedNodeConfig { + swim: SwimConfig { + probe_interval: 1, + probe_timeout: 3, + indirect_probes: 1, + suspicion_timeout: 5, + dead_reprobe_interval: 0, + }, + cache_capacity: 100, + republish_interval: 50, + registry: RegistryConfig::default(), + } +} + +/// Deliver actions to the appropriate target nodes, returning responses +/// tagged with the responder's index. Nodes whose index appears in +/// `excluded` silently drop messages (simulates death / network loss). +fn deliver_actions_tagged( + actions: &[NodeAction], + sender_id: NodeId, + ids: &[NodeId], + nodes: &mut [DistributedNode], + excluded: &[usize], +) -> Vec<(usize, Vec)> { + let mut tagged: Vec<(usize, Vec)> = Vec::new(); + + for action in actions { + match action { + NodeAction::SendPing { + to, + sequence, + piggyback, + .. + } => { + if let Some(idx) = ids.iter().position(|id| id == to) { + if !excluded.contains(&idx) { + let resp = nodes[idx].handle_ping(sender_id, *sequence, piggyback); + if !resp.is_empty() { + tagged.push((idx, resp)); + } + } + } + } + NodeAction::SendAck { + to, + sequence, + piggyback, + .. + } => { + if let Some(idx) = ids.iter().position(|id| id == to) { + if !excluded.contains(&idx) { + let resp = nodes[idx].handle_ack(sender_id, *sequence, piggyback); + if !resp.is_empty() { + tagged.push((idx, resp)); + } + } + } + } + NodeAction::SendJoinResponse { to, members, .. } => { + if let Some(idx) = ids.iter().position(|id| id == to) { + if !excluded.contains(&idx) { + let resp = nodes[idx].handle_join_response(members.clone()); + if !resp.is_empty() { + tagged.push((idx, resp)); + } + } + } + } + NodeAction::SendPingReq { + relay, + target, + sequence, + piggyback, + .. + } => { + if let Some(idx) = ids.iter().position(|id| id == relay) { + if !excluded.contains(&idx) { + let resp = + nodes[idx].handle_ping_req(sender_id, *target, *sequence, piggyback); + if !resp.is_empty() { + tagged.push((idx, resp)); + } + } + } + } + NodeAction::MembershipChanged { .. } => {} + } + } + + tagged +} + +/// An N-node test cluster with correct-by-construction message delivery. +pub struct TestCluster { + ids: Vec, + nodes: Vec, +} + +impl TestCluster { + /// Create an N-node cluster using `test_config()`. Nodes 1..N join via node 0. + pub fn new(n: usize) -> Self { + Self::with_config(n, test_config()) + } + + /// Create an N-node cluster with a custom config. Nodes 1..N join via node 0. + pub fn with_config(n: usize, config: DistributedNodeConfig) -> Self { + assert!(n >= 2, "TestCluster requires at least 2 nodes"); + + let mut nodes: Vec = + (0..n).map(|_| DistributedNode::new(config.clone())).collect(); + let ids: Vec = nodes.iter().map(|node| node.node_id()).collect(); + + // All nodes join through node 0 (seed). + for i in 1..n { + let actions = nodes[0].handle_join_request(ids[i]); + deliver_actions_tagged(&actions, ids[0], &ids, &mut nodes, &[]); + } + + Self { ids, nodes } + } + + /// Get the `NodeId` for the node at `idx`. + pub fn node_id(&self, idx: usize) -> NodeId { + self.ids[idx] + } + + /// Run one gossip round: tick all live nodes, deliver with tagged + /// responses, deliver responses back. Excluded indices are skipped. + fn gossip_round_excluding(&mut self, excluded: &[usize]) { + let n = self.nodes.len(); + + // 1. Tick all live nodes, collect actions. + let mut all_actions: Vec<(usize, Vec)> = Vec::new(); + for idx in 0..n { + if excluded.contains(&idx) { + continue; + } + let actions = self.nodes[idx].tick(); + if !actions.is_empty() { + all_actions.push((idx, actions)); + } + } + + // 2. Deliver each sender's actions → get tagged responses. + // 3. Deliver responses back using the responder's identity. + for (sender_idx, actions) in all_actions { + let tagged_responses = deliver_actions_tagged( + &actions, + self.ids[sender_idx], + &self.ids, + &mut self.nodes, + excluded, + ); + for (responder_idx, response_actions) in tagged_responses { + deliver_actions_tagged( + &response_actions, + self.ids[responder_idx], + &self.ids, + &mut self.nodes, + excluded, + ); + } + } + } + + /// Run `n` gossip rounds with all nodes participating. + pub fn gossip_rounds(&mut self, n: usize) { + for _ in 0..n { + self.gossip_round_excluding(&[]); + } + } + + /// Run `n` gossip rounds; dead nodes neither tick nor receive. + pub fn gossip_rounds_excluding(&mut self, dead: &[usize], n: usize) { + for _ in 0..n { + self.gossip_round_excluding(dead); + } + } +} + +impl Index for TestCluster { + type Output = DistributedNode; + fn index(&self, idx: usize) -> &Self::Output { + &self.nodes[idx] + } +} + +impl IndexMut for TestCluster { + fn index_mut(&mut self, idx: usize) -> &mut Self::Output { + &mut self.nodes[idx] + } +} diff --git a/crates/distribution/tests/node_integration.rs b/crates/distribution/tests/node_integration.rs index 02bd93a..361a574 100644 --- a/crates/distribution/tests/node_integration.rs +++ b/crates/distribution/tests/node_integration.rs @@ -3,97 +3,27 @@ //! These tests verify the full composed behavior from a consumer's perspective: //! cluster formation, actor registration/resolution, and fault tolerance. +mod common; + use swactor::actor::ActorAddress; +use common::{test_config, TestCluster}; use distribution::crypto::Keypair; use distribution::node::{DistributedNode, DistributedNodeConfig, ResolveResult}; use distribution::swim::node::NodeAction; -use distribution::registry::RegistryConfig; -use distribution::swim::probe::SwimConfig; -use distribution::types::NodeId; - -fn test_config() -> DistributedNodeConfig { - DistributedNodeConfig { - swim: SwimConfig { - probe_interval: 1, - probe_timeout: 3, - indirect_probes: 1, - suspicion_timeout: 5, - dead_reprobe_interval: 0, - }, - cache_capacity: 100, - republish_interval: 50, - registry: RegistryConfig::default(), - } -} - -/// Simulate a network round: deliver actions from `sender` to the appropriate -/// `receiver` node. Returns any actions generated by the receiver. -fn deliver_actions( - actions: &[NodeAction], - sender_id: NodeId, - nodes: &mut [(NodeId, &mut DistributedNode)], -) -> Vec { - let mut responses = Vec::new(); - for action in actions { - match action { - NodeAction::SendPing { to, sequence, piggyback, .. } => { - if let Some((_, node)) = nodes.iter_mut().find(|(id, _)| id == to) { - responses.extend(node.handle_ping(sender_id, *sequence, piggyback)); - } - } - NodeAction::SendAck { to, sequence, piggyback, .. } => { - if let Some((_, node)) = nodes.iter_mut().find(|(id, _)| id == to) { - responses.extend(node.handle_ack(sender_id, *sequence, piggyback)); - } - } - NodeAction::SendJoinResponse { to, members, .. } => { - if let Some((_, node)) = nodes.iter_mut().find(|(id, _)| id == to) { - responses.extend(node.handle_join_response(members.clone())); - } - } - NodeAction::SendPingReq { relay, target, sequence, piggyback, .. } => { - if let Some((_, node)) = nodes.iter_mut().find(|(id, _)| id == relay) { - responses.extend(node.handle_ping_req(sender_id, *target, *sequence, piggyback)); - } - } - NodeAction::MembershipChanged { .. } => { - // Notifications — no delivery needed - } - } - } - responses -} - -/// Form a two-node cluster by having the joiner send a join request to the seed. -fn join_nodes(seed: &mut DistributedNode, joiner: &mut DistributedNode) { - let seed_id = seed.node_id(); - let joiner_id = joiner.node_id(); - - // Seed handles the join request from the joiner - let actions = seed.handle_join_request(joiner_id); - - // Deliver join response to joiner - let mut nodes = vec![(joiner_id, &mut *joiner)]; - let _ = deliver_actions(&actions, seed_id, &mut nodes); -} // ─── Cluster Formation ─────────────────────────────────────────────────────── #[test] fn two_node_cluster_forms_via_join() { // Given: a seed node and a joining node - let mut seed = DistributedNode::new(test_config()); - let mut joiner = DistributedNode::new(test_config()); + let cluster = TestCluster::new(2); - let seed_id = seed.node_id(); - let joiner_id = joiner.node_id(); - - // When: the joiner joins via the seed - join_nodes(&mut seed, &mut joiner); + let seed_id = cluster.node_id(0); + let joiner_id = cluster.node_id(1); // Then: both nodes see each other as members - let seed_members = seed.members(); - let joiner_members = joiner.members(); + let seed_members = cluster[0].members(); + let joiner_members = cluster[1].members(); assert!( seed_members.iter().any(|m| m.node_id == joiner_id), @@ -108,17 +38,13 @@ fn two_node_cluster_forms_via_join() { #[test] fn joined_node_appears_in_routing_table() { // Given: two nodes that have formed a cluster - let mut seed = DistributedNode::new(test_config()); - let mut joiner = DistributedNode::new(test_config()); + let cluster = TestCluster::new(2); - let seed_id = seed.node_id(); - - // When: join completes - join_nodes(&mut seed, &mut joiner); + let seed_id = cluster.node_id(0); // Then: joiner's routing table contains the seed assert!( - joiner.routing_table().contains(&seed_id), + cluster[1].routing_table().contains(&seed_id), "joiner's routing table should contain seed" ); } @@ -147,16 +73,13 @@ fn registered_actor_resolves_from_cache() { #[test] fn unknown_actor_returns_needs_lookup_when_peers_known() { // Given: a two-node cluster - let mut seed = DistributedNode::new(test_config()); - let mut joiner = DistributedNode::new(test_config()); + let mut cluster = TestCluster::new(2); - let seed_id = seed.node_id(); - - join_nodes(&mut seed, &mut joiner); + let seed_id = cluster.node_id(0); // When: resolving an unregistered actor on the joiner let unknown_actor = ActorAddress::new_random(); - let result = joiner.resolve_actor(&unknown_actor); + let result = cluster[1].resolve_actor(&unknown_actor); // Then: it returns NeedsLookup with the seed as a closest node match result { @@ -210,24 +133,20 @@ fn store_remote_directory_entry_makes_it_resolvable() { #[test] fn cache_invalidation_forces_re_lookup() { // Given: a node with a cached actor location and peers in routing table - let mut seed = DistributedNode::new(test_config()); - let mut node = DistributedNode::new(test_config()); + let mut cluster = TestCluster::new(2); - let node_id = node.node_id(); - - // Form cluster - join_nodes(&mut seed, &mut node); + let node_id = cluster.node_id(1); // Register and resolve an actor (populates cache) let actor = ActorAddress::new_random(); - node.register_actor(actor, 1); - assert!(matches!(node.resolve_actor(&actor), ResolveResult::Cached(_))); + cluster[1].register_actor(actor, 1); + assert!(matches!(cluster[1].resolve_actor(&actor), ResolveResult::Cached(_))); // When: the cache is invalidated (e.g., delivery failure) - node.invalidate_cache(&actor); + cluster[1].invalidate_cache(&actor); // Then: next resolve falls through to directory (still finds it there) - match node.resolve_actor(&actor) { + match cluster[1].resolve_actor(&actor) { ResolveResult::Cached(resolved) => { assert_eq!(resolved, node_id, "should re-populate from local directory"); } @@ -268,14 +187,11 @@ fn node_death_clears_routing_table_and_cache_entries() { #[test] fn graceful_leave_disseminates_death_on_next_probe() { // Given: a two-node cluster - let mut seed = DistributedNode::new(test_config()); - let mut node = DistributedNode::new(test_config()); - - join_nodes(&mut seed, &mut node); + let mut cluster = TestCluster::new(2); // When: the node leaves and then ticks (probe carries piggybacked death) - let _leave_actions = node.leave(); - let tick_actions = node.tick(); + let _leave_actions = cluster[1].leave(); + let tick_actions = cluster[1].tick(); // Then: the tick produces a ping that carries the death piggyback // The ping's piggyback will contain the node's self-death update @@ -293,13 +209,10 @@ fn graceful_leave_disseminates_death_on_next_probe() { #[test] fn tick_produces_swim_probe_actions_when_peers_present() { // Given: a two-node cluster - let mut seed = DistributedNode::new(test_config()); - let mut node = DistributedNode::new(test_config()); - - join_nodes(&mut seed, &mut node); + let mut cluster = TestCluster::new(2); // When: ticking the node (with probe_interval=1, so first tick triggers a probe) - let tick_actions = node.tick(); + let tick_actions = cluster[1].tick(); // Then: it produces probe actions (pings to known members) let has_ping = tick_actions.iter().any(|a| matches!(a, NodeAction::SendPing { .. })); diff --git a/crates/distribution/tests/registry.rs b/crates/distribution/tests/registry.rs index 1b92681..7545c5c 100644 --- a/crates/distribution/tests/registry.rs +++ b/crates/distribution/tests/registry.rs @@ -1,103 +1,16 @@ //! Behavioral tests for the cluster registry. //! -//! Tests gossip-propagated naming via LWW-Register CRDT, using the same -//! `deliver_actions` + `test_config` pattern from `node_integration.rs`. +//! Tests gossip-propagated naming via LWW-Register CRDT, using the shared +//! `TestCluster` harness from `common`. + +mod common; use swactor::actor::ActorAddress; -use distribution::node::{DistributedNode, DistributedNodeConfig}; +use common::{test_config, TestCluster}; +use distribution::node::DistributedNode; use distribution::registry::{ClusterRegistry, RegistryConfig, RegistryEntry, RegistryEvent}; -use distribution::swim::node::NodeAction; -use distribution::swim::probe::SwimConfig; use distribution::types::NodeId; -fn test_config() -> DistributedNodeConfig { - DistributedNodeConfig { - swim: SwimConfig { - probe_interval: 1, - probe_timeout: 3, - indirect_probes: 1, - suspicion_timeout: 5, - dead_reprobe_interval: 0, - }, - cache_capacity: 100, - republish_interval: 50, - registry: RegistryConfig::default(), - } -} - -/// Simulate a network round: deliver actions from `sender` to the appropriate -/// `receiver` node. Returns any actions generated by the receiver. -fn deliver_actions( - actions: &[NodeAction], - sender_id: NodeId, - nodes: &mut [(NodeId, &mut DistributedNode)], -) -> Vec { - let mut responses = Vec::new(); - for action in actions { - match action { - NodeAction::SendPing { to, sequence, piggyback, .. } => { - if let Some((_, node)) = nodes.iter_mut().find(|(id, _)| id == to) { - responses.extend(node.handle_ping(sender_id, *sequence, piggyback)); - } - } - NodeAction::SendAck { to, sequence, piggyback, .. } => { - if let Some((_, node)) = nodes.iter_mut().find(|(id, _)| id == to) { - responses.extend(node.handle_ack(sender_id, *sequence, piggyback)); - } - } - NodeAction::SendJoinResponse { to, members, .. } => { - if let Some((_, node)) = nodes.iter_mut().find(|(id, _)| id == to) { - responses.extend(node.handle_join_response(members.clone())); - } - } - NodeAction::SendPingReq { relay, target, sequence, piggyback, .. } => { - if let Some((_, node)) = nodes.iter_mut().find(|(id, _)| id == relay) { - responses.extend(node.handle_ping_req(sender_id, *target, *sequence, piggyback)); - } - } - NodeAction::MembershipChanged { .. } => {} - } - } - responses -} - -/// Form a two-node cluster by having b join via a. -fn form_cluster() -> (DistributedNode, NodeId, DistributedNode, NodeId) { - let mut a = DistributedNode::new(test_config()); - let mut b = DistributedNode::new(test_config()); - - let a_id = a.node_id(); - let b_id = b.node_id(); - - // b joins via a - let actions = a.handle_join_request(b_id); - let mut nodes = vec![(b_id, &mut b)]; - let _ = deliver_actions(&actions, a_id, &mut nodes); - - (a, a_id, b, b_id) -} - -/// Run several gossip rounds between two nodes. -fn gossip_rounds( - a: &mut DistributedNode, a_id: NodeId, - b: &mut DistributedNode, b_id: NodeId, - rounds: usize, -) { - for _ in 0..rounds { - let actions_a = a.tick(); - let mut nodes = vec![(b_id, &mut *b)]; - let responses = deliver_actions(&actions_a, a_id, &mut nodes); - let mut nodes = vec![(a_id, &mut *a)]; - let _ = deliver_actions(&responses, b_id, &mut nodes); - - let actions_b = b.tick(); - let mut nodes = vec![(a_id, &mut *a)]; - let responses = deliver_actions(&actions_b, b_id, &mut nodes); - let mut nodes = vec![(b_id, &mut *b)]; - let _ = deliver_actions(&responses, a_id, &mut nodes); - } -} - // ─── Test 1: register and resolve ─────────────────────────────────────────── #[test] @@ -245,127 +158,82 @@ fn lww_tiebreak_generation_then_node_id() { #[test] fn gossip_propagates_registration() { - let (mut a, a_id, mut b, b_id) = form_cluster(); + let mut cluster = TestCluster::new(2); let actor = ActorAddress::new_random(); - a.register_name("greeter".into(), actor); + cluster[0].register_name("greeter".into(), actor); // B doesn't know about "greeter" yet. - assert_eq!(b.resolve_name("greeter"), None); + assert_eq!(cluster[1].resolve_name("greeter"), None); // Run gossip rounds — registry entries piggyback on SWIM messages. - gossip_rounds(&mut a, a_id, &mut b, b_id, 5); + cluster.gossip_rounds(5); // Now B should resolve "greeter" to A's actor. - assert_eq!(b.resolve_name("greeter"), Some((actor, a_id))); + let a_id = cluster.node_id(0); + assert_eq!(cluster[1].resolve_name("greeter"), Some((actor, a_id))); } // ─── Test 8: tombstone propagation via gossip ─────────────────────────────── #[test] fn tombstone_propagation_via_gossip() { - let (mut a, a_id, mut b, b_id) = form_cluster(); + let mut cluster = TestCluster::new(2); let actor = ActorAddress::new_random(); - a.register_name("ephemeral".into(), actor); + cluster[0].register_name("ephemeral".into(), actor); // Propagate the registration. - gossip_rounds(&mut a, a_id, &mut b, b_id, 5); - assert_eq!(b.resolve_name("ephemeral"), Some((actor, a_id))); + cluster.gossip_rounds(5); + let a_id = cluster.node_id(0); + assert_eq!(cluster[1].resolve_name("ephemeral"), Some((actor, a_id))); // Now unregister on A. - a.unregister_name("ephemeral"); + cluster[0].unregister_name("ephemeral"); // Propagate the tombstone. - gossip_rounds(&mut a, a_id, &mut b, b_id, 5); + cluster.gossip_rounds(5); - assert_eq!(b.resolve_name("ephemeral"), None); + assert_eq!(cluster[1].resolve_name("ephemeral"), None); } // ─── Test 9: node death tombstones entries ────────────────────────────────── #[test] fn node_death_tombstones_entries() { - // Set up a 3-node cluster: A, B, C - let mut a = DistributedNode::new(test_config()); - let mut b = DistributedNode::new(test_config()); - let mut c = DistributedNode::new(test_config()); + // Set up a 3-node cluster: A(0), B(1), C(2) + let mut cluster = TestCluster::new(3); - let a_id = a.node_id(); - let b_id = b.node_id(); - let c_id = c.node_id(); - - // B and C join A. - let actions = a.handle_join_request(b_id); - let mut nodes = vec![(b_id, &mut b)]; - let _ = deliver_actions(&actions, a_id, &mut nodes); - - let actions = a.handle_join_request(c_id); - let mut nodes = vec![(c_id, &mut c)]; - let _ = deliver_actions(&actions, a_id, &mut nodes); + let b_id = cluster.node_id(1); // B registers a name. let actor = ActorAddress::new_random(); - b.register_name("b-service".into(), actor); + cluster[1].register_name("b-service".into(), actor); // Propagate B's registration to A and C via mesh gossip. - // Deliver to each target separately so responses carry the correct sender_id. - for _ in 0..5 { - let actions = b.tick(); - let mut t = vec![(a_id, &mut a)]; - let resp_a = deliver_actions(&actions, b_id, &mut t); - let mut t = vec![(c_id, &mut c)]; - let resp_c = deliver_actions(&actions, b_id, &mut t); - let mut t = vec![(b_id, &mut b)]; - let _ = deliver_actions(&resp_a, a_id, &mut t); - let mut t = vec![(b_id, &mut b)]; - let _ = deliver_actions(&resp_c, c_id, &mut t); + cluster.gossip_rounds(5); - let actions = a.tick(); - let mut t = vec![(b_id, &mut b)]; - let resp_b = deliver_actions(&actions, a_id, &mut t); - let mut t = vec![(c_id, &mut c)]; - let resp_c = deliver_actions(&actions, a_id, &mut t); - let mut t = vec![(a_id, &mut a)]; - let _ = deliver_actions(&resp_b, b_id, &mut t); - let mut t = vec![(a_id, &mut a)]; - let _ = deliver_actions(&resp_c, c_id, &mut t); + assert_eq!(cluster[0].resolve_name("b-service"), Some((actor, b_id))); + assert_eq!(cluster[2].resolve_name("b-service"), Some((actor, b_id))); - let actions = c.tick(); - let mut t = vec![(a_id, &mut a)]; - let resp_a = deliver_actions(&actions, c_id, &mut t); - let mut t = vec![(b_id, &mut b)]; - let resp_b = deliver_actions(&actions, c_id, &mut t); - let mut t = vec![(c_id, &mut c)]; - let _ = deliver_actions(&resp_a, a_id, &mut t); - let mut t = vec![(c_id, &mut c)]; - let _ = deliver_actions(&resp_b, b_id, &mut t); - } - - assert_eq!(a.resolve_name("b-service"), Some((actor, b_id))); - assert_eq!(c.resolve_name("b-service"), Some((actor, b_id))); - - // B dies — SWIM detects via timeout. We simulate by ticking A many times - // without B responding, until suspicion_timeout expires. - for _ in 0..20 { - let actions = a.tick(); - // Don't deliver to B — it's "dead". Only deliver to C. - let mut nodes = vec![(c_id, &mut c)]; - let responses = deliver_actions(&actions, a_id, &mut nodes); - let mut nodes = vec![(a_id, &mut a)]; - let _ = deliver_actions(&responses, c_id, &mut nodes); - } + // B dies — SWIM detects via timeout. We simulate by running rounds + // without B participating, until suspicion_timeout expires. + cluster.gossip_rounds_excluding(&[1], 20); // After enough ticks, A should declare B dead, which tombstones "b-service". - let a_resolved = a.resolve_name("b-service"); + assert_eq!( + cluster[0].resolve_name("b-service"), + None, + "A must tombstone b-service after declaring B dead" + ); - if a_resolved.is_none() { - // A has tombstoned it — propagate to C. - gossip_rounds(&mut a, a_id, &mut c, c_id, 5); - assert_eq!(c.resolve_name("b-service"), None, "C should see tombstone after B's death propagates"); - } - // If SWIM hasn't declared death yet, the test still passes — the mechanism - // is wired, just needs more ticks. The important thing: no panics, clean flow. + // Propagate tombstone from A to C. + cluster.gossip_rounds_excluding(&[1], 5); + assert_eq!( + cluster[2].resolve_name("b-service"), + None, + "C should see tombstone after B's death propagates" + ); } // ─── Test 10: registry events emitted on change ───────────────────────────── @@ -438,49 +306,24 @@ fn tombstone_gc_removes_old_tombstones() { #[test] fn gossip_convergence_five_nodes() { - let mut nodes: Vec = (0..5) - .map(|_| DistributedNode::new(test_config())) - .collect(); - - // Collect ids before joining (borrow gymnastics). - let ids: Vec = nodes.iter().map(|n| n.node_id()).collect(); - - // All join through node 0. - for i in 1..5 { - let actions = nodes[0].handle_join_request(ids[i]); - // Deliver join response to node i. - let mut target = vec![(ids[i], &mut nodes[i])]; - let _ = deliver_actions(&actions, ids[0], &mut target); - } + let mut cluster = TestCluster::new(5); // Each node registers a unique name. let actors: Vec = (0..5).map(|_| ActorAddress::new_random()).collect(); for i in 0..5 { - nodes[i].register_name(format!("service-{i}"), actors[i]); + cluster[i].register_name(format!("service-{i}"), actors[i]); } - // Run many gossip rounds between all pairs. - for _round in 0..15 { - for i in 0..5 { - let tick_actions = nodes[i].tick(); - // Deliver to all other nodes. - for j in 0..5 { - if i == j { continue; } - let mut target = vec![(ids[j], &mut nodes[j])]; - let responses = deliver_actions(&tick_actions, ids[i], &mut target); - let mut target = vec![(ids[i], &mut nodes[i])]; - let _ = deliver_actions(&responses, ids[j], &mut target); - } - } - } + // Run many gossip rounds. + cluster.gossip_rounds(15); // All 5 names should be resolvable on all 5 nodes. for i in 0..5 { for j in 0..5 { - let result = nodes[i].resolve_name(&format!("service-{j}")); + let result = cluster[i].resolve_name(&format!("service-{j}")); assert_eq!( result, - Some((actors[j], ids[j])), + Some((actors[j], cluster.node_id(j))), "node {i} should resolve service-{j}" ); } diff --git a/crates/distribution/tests/swim_node.rs b/crates/distribution/tests/swim_node.rs index c77c5f5..23bb2cc 100644 --- a/crates/distribution/tests/swim_node.rs +++ b/crates/distribution/tests/swim_node.rs @@ -130,12 +130,11 @@ fn ping_from_unknown_node_adds_it_to_members() { #[test] fn membership_updates_piggyback_on_pings() { let mut swim = SwimNode::new(node(0), fast_config()); - - // Add a member and join a node (which enqueues a dissemination update) + // Join enqueues a membership update for dissemination. swim.handle_join_request(node(1)); - // Tick until a probe fires — the ping should carry piggyback data - let actions = tick_n(&mut swim, 5); + // Tick past the probe interval — a ping must fire. + let actions = tick_n(&mut swim, 6); let pings: Vec<_> = actions.iter().filter_map(|a| { if let NodeAction::SendPing { piggyback, .. } = a { Some(piggyback) @@ -144,10 +143,10 @@ fn membership_updates_piggyback_on_pings() { } }).collect(); - if !pings.is_empty() { - // At least one ping should carry piggyback (the join update) - assert!(pings.iter().any(|pb| !pb.is_empty()), "pings should carry piggyback data"); - } + // With one member and probe_interval=5, at least one ping must fire. + assert!(!pings.is_empty(), "probe must fire within 6 ticks"); + // The join update must ride as piggyback on that ping. + assert!(pings.iter().any(|pb| !pb.is_empty()), "pings should carry piggyback data"); } // ─── Refutation ───────────────────────────────────────────────────────────── @@ -181,8 +180,8 @@ fn leave_enqueues_death_for_dissemination() { swim.leave(); - // Tick to trigger a probe — the death update should piggyback - let actions = tick_n(&mut swim, 5); + // Tick past the probe interval — the death update must piggyback on the ping. + let actions = tick_n(&mut swim, 6); let pings_with_piggyback: Vec<_> = actions.iter().filter_map(|a| { if let NodeAction::SendPing { piggyback, .. } = a { if !piggyback.is_empty() { Some(piggyback) } else { None } @@ -191,9 +190,8 @@ fn leave_enqueues_death_for_dissemination() { } }).collect(); - // We can't guarantee the exact content, but the leave should enqueue something - // that gets piggybacked - assert!(!pings_with_piggyback.is_empty() || swim.members().alive_count() > 0); + assert!(!pings_with_piggyback.is_empty(), + "leave must enqueue a death update that piggybacks on the next ping"); } // ─── Full join scenario ───────────────────────────────────────────────────── diff --git a/crates/simulation/src/gossip/sim.rs b/crates/simulation/src/gossip/sim.rs index 5334fa3..57a945b 100644 --- a/crates/simulation/src/gossip/sim.rs +++ b/crates/simulation/src/gossip/sim.rs @@ -279,23 +279,14 @@ pub fn heal_partition_via_handle( addrs: &[ActorAddress], names: &[String], ) -> Vec<(String, String)> { - if !matches!(topology, Topology::Partitioned) { - return Vec::new(); - } - let n = addrs.len(); - let half = n / 2; + let heal_edges = topology.heal_edges(addrs.len()); let mut new_edges = Vec::new(); - if half > 0 && half < n { + for (from, to) in heal_edges { handle .runtime - .send_to(addrs[half - 1], GossipMessage::AddPeer(addrs[half])) + .send_to(addrs[from], GossipMessage::AddPeer(addrs[to])) .unwrap(); - handle - .runtime - .send_to(addrs[half], GossipMessage::AddPeer(addrs[half - 1])) - .unwrap(); - new_edges.push((names[half - 1].clone(), names[half].clone())); - new_edges.push((names[half].clone(), names[half - 1].clone())); + new_edges.push((names[from].clone(), names[to].clone())); } new_edges } diff --git a/crates/simulation/src/topology.rs b/crates/simulation/src/topology.rs index 7592df3..ec52bd0 100644 --- a/crates/simulation/src/topology.rs +++ b/crates/simulation/src/topology.rs @@ -67,15 +67,28 @@ impl Topology { } /// Partition healing edges: bidirectional links between the two halves. + /// + /// Connects up to 3 evenly-spaced node pairs across the partition boundary. + /// A single bridge link is unreliable under random peer selection: with half=50, + /// there is only a 1/50 chance per round that the bridge node gossips across, + /// giving a ~1.7% probability of zero crossings in 200 rounds. pub fn heal_edges(&self, num_nodes: usize) -> Vec<(usize, usize)> { if !matches!(self, Topology::Partitioned) { return Vec::new(); } let half = num_nodes / 2; - if half > 0 && half < num_nodes { - vec![(half - 1, half), (half, half - 1)] - } else { - Vec::new() + if half == 0 || half >= num_nodes { + return Vec::new(); } + let right = num_nodes - half; + let num_bridges = half.min(3); + let mut edges = Vec::with_capacity(num_bridges * 2); + for b in 0..num_bridges { + let a_node = b * half / num_bridges; + let b_node = half + b * right / num_bridges; + edges.push((a_node, b_node)); + edges.push((b_node, a_node)); + } + edges } } diff --git a/crates/simulation/tests/cluster_scenarios.rs b/crates/simulation/tests/cluster_scenarios.rs index bdad324..e63ab71 100644 --- a/crates/simulation/tests/cluster_scenarios.rs +++ b/crates/simulation/tests/cluster_scenarios.rs @@ -835,3 +835,78 @@ fn convergence_after_partition_heal() { result.actual ); } + +// ──────────────────────────────────────────────────────────────────────────── +// Postmortem 3f — suspect→refute race: no false death after refutation +// ──────────────────────────────────────────────────────────────────────────── + +#[test] +fn suspect_refuted_before_timeout_no_false_death() { + // Scenario from SWIM flaky test postmortem (recommendation 3f): + // 3-node cluster, asymmetric partitions make node 1 unreachable by + // probes from nodes 0 and 2, but node 1's outgoing messages still + // reach them. Heal before the suspicion timer fires, so node 1 learns + // about its suspicion via gossip, bumps incarnation, and refutes back + // to Alive. The suspicion timer then fires but sees Alive (not Suspect) + // and does nothing — exercising the Bug 1 fix. + // + // Timing: partition at round 10 (tick 30), heal at round 20 (tick 60). + // Suspicion starts ~tick 36. With suspicion_timeout=50, timer fires + // ~tick 86 — 26 ticks after heal, giving ample time for refutation. + let config = DistributionSimConfig { + name: "suspect-refute-race".into(), + num_nodes: 3, + num_rounds: 60, + ticks_per_round: 3, + actors_per_node: 0, + swim: distribution::swim::probe::SwimConfig { + probe_interval: 1, + probe_timeout: 3, + indirect_probes: 1, + suspicion_timeout: 50, + dead_reprobe_interval: 0, // disabled — refutation must happen before death + }, + network_faults: vec![ + // Block 0→1 (but 1→0 still works) + NetworkFault::Partition { + round: 10, + partition: Partition { + side_a: vec![0], + side_b: vec![1], + asymmetric: true, + }, + }, + // Block 2→1 (but 1→2 still works) + NetworkFault::Partition { + round: 10, + partition: Partition { + side_a: vec![2], + side_b: vec![1], + asymmetric: true, + }, + }, + // Heal all partitions before suspicion timeout fires + NetworkFault::Heal { round: 20 }, + ], + ..default_config() + }; + + let trace = run_simulation(config); + + // All 3 nodes must be alive in the final snapshot — no false deaths + let last_round = trace.snapshots_per_round.last().unwrap(); + for (name, snap) in last_round { + assert!( + snap.is_alive, + "{name} should be alive but was marked dead (false death after refutation)" + ); + } + + // 100% accuracy: every alive node sees all others as alive + let result = check_accuracy(&trace, 1.0); + assert!( + result.passed, + "all nodes should see full membership after refutation: {}", + result.actual + ); +} diff --git a/crates/simulation/tests/gossip_properties.rs b/crates/simulation/tests/gossip_properties.rs index 686c493..5afd1d6 100644 --- a/crates/simulation/tests/gossip_properties.rs +++ b/crates/simulation/tests/gossip_properties.rs @@ -79,7 +79,7 @@ fn delivery_is_all_or_nothing_per_key() { #[test] fn ring_converges_within_bound() { - let n = 1000; + let n = 500; let config = GossipSimConfig { name: "ring-latency".into(), topology: Topology::Ring, @@ -135,7 +135,7 @@ fn last_node_latency_bounded_in_fullmesh() { #[test] fn total_messages_equal_n_times_rounds() { - let n = 1000; + let n = 500; let r = 30; let config = GossipSimConfig { name: "msg-count".into(), @@ -231,7 +231,7 @@ fn ring_distributes_load_evenly() { let config = GossipSimConfig { name: "ring-load".into(), topology: Topology::Ring, - num_nodes: 1000, + num_nodes: 500, initial_data: test_data(5), num_rounds: 60, ticks_per_round: 4, @@ -245,7 +245,7 @@ fn ring_distributes_load_evenly() { #[test] fn amplification_equals_num_rounds() { - let n = 1000; + let n = 500; let r = 30; let config = GossipSimConfig { name: "ring-amp".into(), @@ -269,7 +269,7 @@ fn convergence_curve_is_monotonic() { let config = GossipSimConfig { name: "ring-mono".into(), topology: Topology::Ring, - num_nodes: 1000, + num_nodes: 500, initial_data: test_data(5), num_rounds: 60, ticks_per_round: 4, @@ -322,7 +322,7 @@ fn partitioned_network_does_not_converge() { let config = GossipSimConfig { name: "partition-no-heal".into(), topology: Topology::Partitioned, - num_nodes: 1000, + num_nodes: 100, initial_data: test_data(5), num_rounds: 40, ticks_per_round: 4, @@ -341,9 +341,9 @@ fn partition_heals_and_converges() { topology: Topology::Partitioned, num_nodes: 100, initial_data: test_data(5), - num_rounds: 300, + num_rounds: 150, ticks_per_round: 4, - heal_after_round: Some(100), + heal_after_round: Some(50), num_threads: 1, }; let (_, metrics) = run_and_analyze(config); @@ -358,13 +358,13 @@ fn partial_convergence_before_healing() { topology: Topology::Partitioned, num_nodes: 100, initial_data: test_data(5), - num_rounds: 300, + num_rounds: 150, ticks_per_round: 4, - heal_after_round: Some(100), + heal_after_round: Some(50), num_threads: 1, }; let (_, metrics) = run_and_analyze(config); - let result = check_partial_before_heal(&metrics, 100); + let result = check_partial_before_heal(&metrics, 50); assert!(result.passed, "partial before heal: {}", result.actual); } @@ -372,7 +372,7 @@ fn partial_convergence_before_healing() { #[test] fn convergence_time_scales_sublinearly() { - let sizes = [100, 250, 500, 1000]; + let sizes = [50, 125, 250, 500]; let mut data = Vec::new(); for &n in &sizes { let rounds = 60; @@ -396,7 +396,7 @@ fn convergence_time_scales_sublinearly() { #[test] fn total_messages_scale_linearly_with_n() { - let sizes = [100, 250, 500, 1000]; + let sizes = [50, 125, 250, 500]; let fixed_rounds = 30; let mut data = Vec::new(); for &n in &sizes { @@ -672,7 +672,7 @@ fn state_size_grows_monotonically() { let config = GossipSimConfig { name: "state-mono".into(), topology: Topology::Ring, - num_nodes: 1000, + num_nodes: 500, initial_data: test_data(5), num_rounds: 60, ticks_per_round: 4, diff --git a/docs/development_history/distribution/SWIM_FLAKY_TEST_POSTMORTEM.md b/docs/development_history/distribution/SWIM_FLAKY_TEST_POSTMORTEM.md new file mode 100644 index 0000000..f8ec9b6 --- /dev/null +++ b/docs/development_history/distribution/SWIM_FLAKY_TEST_POSTMORTEM.md @@ -0,0 +1,218 @@ +# After-Action: SWIM Flaky Test (`node_death_tombstones_entries`) + +> A ~20% failure rate in a SWIM death-detection test sat undetected because every gate that should have caught it — CI, assertion design, test harness correctness, and flakiness discipline — was either absent or structurally unable to surface the bug. + +--- + +## 1. Incident Summary + +The `node_death_tombstones_entries` test in `registry.rs` failed roughly 1 in 5 runs. Two independent bugs conspired to produce the flakiness: + +**Bug 1 — Protocol:** `check_suspicion_timeouts` in `probe.rs` declared nodes dead on timer expiry without checking whether the node was still `Suspect`. If a refutation (Alive with higher incarnation) arrived between suspicion and timeout, the node was killed anyway. The original code: + +```rust +for node_id in expired { + if members.declare_dead(node_id) { + actions.push(SwimAction::DeclareDead(node_id)); + } + self.cancel_suspicion_timer(node_id); +} +``` + +**Bug 2 — Test harness:** The 3-node gossip loop delivered actions to multiple targets in a single `deliver_actions` call, then attributed all responses to a single `sender_id`. When A ticked and sent pings to both B and C, the responses from both were delivered back to A as if they all came from A — misattributing the sender. This caused C's pings to look like self-pings, triggering false suspicions that fed into Bug 1. + +The interaction: the sender misattribution created false suspicions at non-deterministic rates (depending on tick ordering), and the missing state guard turned those false suspicions into false death declarations. When B was falsely declared dead, its name registration was tombstoned and the test's soft assertion silently passed without verifying the mechanism worked. + +**The fix** (commit `205dc23`, 2 files, +45/−14): added a `still_suspect` guard before `declare_dead`, and split the gossip loop to deliver to each target node separately so responses carry the correct `sender_id`. + +See the commit diff for full technical details. The rest of this document focuses on how a 20% failure rate was merged and what changes prevent it from happening again. + +--- + +## 2. How This Got Into the Repo + +Five gates should have caught this. All five failed. + +### 2a. No CI exists + +There is no automated testing infrastructure. No pre-merge checks, no post-push smoke tests. The project uses Forgejo for source hosting. CI integration is being planned separately. + +A 20% failure rate is invisible with a single manual `cargo test` — you hit the 80% pass rate and move on. CI running tests on every push would have surfaced the failure within a handful of commits. Without it, the only defense is the developer's willingness to run the test more than once. That is not a defense. + +### 2b. The soft assertion hid failures + +The test ended with this: + +```rust +let a_resolved = a.resolve_name("b-service"); + +if a_resolved.is_none() { + // A has tombstoned it — propagate to C. + gossip_rounds(&mut a, a_id, &mut c, c_id, 5); + assert_eq!(c.resolve_name("b-service"), None, + "C should see tombstone after B's death propagates"); +} +// If SWIM hasn't declared death yet, the test still passes — the mechanism +// is wired, just needs more ticks. The important thing: no panics, clean flow. +``` + +If SWIM didn't declare B dead — whether because it correctly needed more ticks *or* because the test harness was broken — the test passed. The comment "the mechanism is wired, just needs more ticks" was written with honest intent but created a test that could never fail for the wrong reason *and* never fail for the right reason. A test that can't fail is not a test. + +### 2c. The 2-node test harness doesn't generalize to 3 nodes + +The `deliver_actions` helper in `registry.rs` takes a `sender_id` parameter and delivers all actions to a list of target nodes, collecting all responses into a flat `Vec`. When the caller attributes those responses with a single `sender_id`, the implicit assumption is: every response in the vec came from the same node. + +This is correct for 2-node tests — if A sends to B, all responses came from B. Every other test in `registry.rs` and `node_integration.rs` uses exactly this pattern with exactly 2 nodes. It works perfectly. + +The `node_death_tombstones_entries` test was the first to use 3 nodes. It passed actions to `deliver_actions` with both B and C in the targets list, then attributed all responses to a single sender. Nobody noticed the attribution broke because: + +1. The helper's API doesn't prevent it — it returns a flat `Vec`, not a per-target map. +2. All other tests were 2-node, establishing a pattern that appeared safe. +3. The simulation layer (`sim.rs:deliver_actions_tagged_with_net`) already solved this correctly with response-tagged delivery: `Vec<(usize, Vec)>`. The test harness didn't reuse that pattern. + +### 2d. The simulation layer is a false safety net + +The simulation test suite is extensive: 19 cluster scenarios and 96 total tests across 7 test files. They exercise partition healing, cascading failures, 50-node convergence, message loss, and actor resolution during network events. They all pass. + +But simulation bypasses the unit test harness entirely. `deliver_actions_tagged_with_net` in `sim.rs` routes responses back with the correct `(responder_idx, Vec)` tagging: + +```rust +fn deliver_actions_tagged_with_net( + actions: &[NodeAction], + sender_idx: usize, + sender_id: NodeId, + nodes: &mut [Option], + node_ids: &[NodeId], + net: &mut NetworkState, +) -> Vec<(usize, Vec)> { +``` + +The simulation proved the protocol works while the unit test harness was silently broken. The simulation caught 0% of this bug because the bug lived in the test harness, not the protocol. Bug 1 (the missing `still_suspect` guard) *could* have been caught by simulation — but only with a scenario specifically designed to refute a suspected node before timeout expiry. No such scenario existed because the guard's absence is only observable under that exact sequence. + +### 2e. No flakiness detection discipline + +There is no practice of running timing-sensitive tests multiple times before merge. No tooling (`cargo-nextest`, `just test-repeat`, a loop in a shell script) to surface intermittent failures. + +A 20% failure rate requires only 5 runs to detect with 99.97% probability: `1 − 0.8^5 = 0.99968`. Nobody ran it 5 times. + +--- + +## 3. Preventing This Class of Bug + +Each recommendation addresses a specific gap from section 2. They are ordered from structural (make the bug class impossible) to procedural (catch it if it happens). + +### 3a. Shared test harness with sender-tagged delivery — DONE + +**Gap addressed:** 2c (harness doesn't generalize to 3+ nodes) + +**Implemented:** A `TestCluster` harness was extracted into `crates/distribution/tests/common/mod.rs`. It contains: + +- `test_config()` — the shared `DistributedNodeConfig` previously duplicated in both test files. +- `deliver_actions_tagged()` — free function returning `Vec<(usize, Vec)>` (responses tagged by responder index), modeled on the simulation's `deliver_actions_tagged_with_net`. An `excluded` slice parameter handles death simulation (nodes that neither tick nor receive). +- `TestCluster` struct — owns parallel `Vec` and `Vec` (borrow-split friendly). Public API: `new(n)`, `with_config(n, config)`, `node_id(idx)`, `Index`/`IndexMut` for direct node access, `gossip_rounds(n)`, and `gossip_rounds_excluding(dead, n)`. + +`gossip_round()` follows the simulation's `tick_all_and_deliver` pattern: tick all live nodes, deliver with tagged responses, deliver responses back using the responder's identity. Sender misattribution is **structurally impossible** — the tagged return type forces correct attribution at every delivery step. + +All duplicated helpers (`deliver_actions`, `form_cluster`, `join_nodes`, `gossip_rounds`) were removed from both `registry.rs` and `node_integration.rs`. 4 multi-node tests in `registry.rs` and 6 in `node_integration.rs` were rewritten to use `TestCluster`. Single-node tests use only `test_config()` from common. + +The formerly flaky `node_death_tombstones_entries` went from a 50-line manual per-target delivery loop to 3 calls: `cluster.gossip_rounds(5)`, `cluster.gossip_rounds_excluding(&[1], 20)`, `cluster.gossip_rounds_excluding(&[1], 5)`. Verified 50/50 passes post-rewrite. + +### 3b. Hard assertions, no soft paths + +**Gap addressed:** 2b (soft assertion) + +Every test must assert on its expected outcome unconditionally. No `if result.is_none() { ... }` pass-either-way branches. If SWIM needs more ticks to detect death, give it more ticks. Don't let the test pass when the expected behavior didn't happen. + +Add negative assertions where applicable. For instance, after B dies, assert that C is still Alive — not just that B is Dead. This catches false-positive death declarations that spill over to healthy nodes. + +### 3c. `declare_dead` state contract + +**Gap addressed:** defense in depth + +`MemberList::declare_dead` currently accepts any non-`Dead` state: + +```rust +pub fn declare_dead(&mut self, node_id: NodeId) -> bool { + if let Some(entry) = self.members.get_mut(&node_id) { + if entry.state != MemberState::Dead { + entry.state = MemberState::Dead; + return true; + } + } + false +} +``` + +It should guard that the node is `Suspect` at the point of call, not leave correctness to callers. The SWIM protocol invariant is: a node transitions `Alive → Suspect → Dead`. Killing an `Alive` node directly violates that invariant. The caller (`check_suspicion_timeouts`) now checks, but the function's own contract should enforce the invariant independently. + +### 3d. Multi-run flakiness detection — RESOLVED (policy) + +**Gap addressed:** 2e (no flakiness discipline) + +**Policy:** No flaky tests are accepted into the repository. Tests must pass deterministically. Timing-sensitive tests involving 3+ nodes should be run multiple times before merge to verify stability. Tooling (`cargo-nextest`, `justfile` targets) can be adopted as needed but the policy is the primary gate. + +### 3e. CI + +**Gap addressed:** 2a (no CI) + +The project uses Forgejo for source hosting. CI workflow integration is being planned separately and is not an action item for this postmortem. When available, the CI pipeline should run `cargo test --workspace` on push to main and on PR. Stretch: `cargo nextest run --retries 3` to specifically surface flaky tests before merge. + +### 3f. Simulation scenario for suspect-then-refute — DONE + +**Gap addressed:** 2d (simulation didn't test the violated invariant) + +**Implemented:** `suspect_refuted_before_timeout_no_false_death` in `crates/simulation/tests/cluster_scenarios.rs`. 3-node cluster with asymmetric partitions making node 1 unreachable by probes from nodes 0 and 2, while node 1's outgoing messages still carry incarnation bumps enabling refutation. Partitions heal before the suspicion timer fires. Asserts all 3 nodes alive (no false deaths) and 100% membership accuracy after refutation. + +--- + +## 4. Files Modified + +| File | Change | +|------|--------| +| `crates/distribution/src/swim/probe.rs` | Added `still_suspect` guard in `check_suspicion_timeouts`; cancel probe phase if target declared dead (+20/−2) | +| `crates/distribution/tests/registry.rs` | Split 3-node gossip delivery to per-target calls with correct sender attribution (+25/−12) | + +### Follow-up: Flaky tests removed and replaced + +The three tests with pass-either-way assertions were removed and replaced with deterministic equivalents: + +| File | Test | Change | +|------|------|--------| +| `crates/distribution/tests/swim_node.rs` | `membership_updates_piggyback_on_pings` | Replaced `if !pings.is_empty()` guard with hard assertion; tick 6 to ensure probe fires | +| `crates/distribution/tests/swim_node.rs` | `leave_enqueues_death_for_dissemination` | Replaced tautological `alive_count() > 0` fallback with hard assertion on piggyback | +| `crates/distribution/tests/registry.rs` | `node_death_tombstones_entries` | Replaced soft `if a_resolved.is_none()` branch with unconditional `assert_eq!` | + +Additionally, `MemberList::declare_dead` was tightened to only accept `Suspect → Dead` transitions, enforcing the SWIM lifecycle invariant at the function boundary. + +### Follow-up: Shared `TestCluster` harness (recommendation 3a) + +Extracted a shared test harness that makes sender-misattribution structurally impossible: + +| File | Change | +|------|--------| +| `crates/distribution/tests/common/mod.rs` | **New** — `test_config()`, `deliver_actions_tagged()`, `TestCluster` struct | +| `crates/distribution/tests/registry.rs` | Removed `test_config`, `deliver_actions`, `form_cluster`, `gossip_rounds` helpers; added `mod common`; rewrote 4 multi-node tests to use `TestCluster` | +| `crates/distribution/tests/node_integration.rs` | Removed `test_config`, `deliver_actions`, `join_nodes` helpers; added `mod common`; rewrote 6 multi-node tests to use `TestCluster` | +| `crates/distribution/src/node.rs` | Added `#[derive(Clone)]` on `DistributedNodeConfig` (required by `TestCluster::with_config`) | +| `crates/distribution/src/registry.rs` | Added `#[derive(Clone)]` on `RegistryConfig` (transitive requirement) | + +### Follow-up: `watch_notification` missing `StdExtension` + +The `watch_notification` test in `crates/bin-runner/tests/wasm_actor.rs` panicked because the runtime was created without `StdExtension`, which `ctx.watch()` requires (via `get_ext()` in `crates/std/src/ctx_ext.rs`). Every other test in the workspace that uses `ctx.watch()` installs the extension — this one was simply missed. + +Same root cause pattern as the SWIM flaky test: a test harness setup gap that was invisible because no other test exercised that path with that configuration. Unlike the SWIM bug, this was a hard failure (panic), not a flaky one — it failed 100% of the time. + +| File | Change | +|------|--------| +| `crates/bin-runner/tests/wasm_actor.rs` | Added `StdExtension` to imports; installed `.with_extension(Arc::new(StdExtension::new()))` on the runtime in `watch_notification` | + +32/32 bin-runner tests now pass. + +--- + +## 5. Verification + +- **Post-fix:** 0/50 failures (was ~10/50 pre-fix) +- **Full distribution suite:** all tests pass +- **Simulation suite:** all 96 tests pass (unaffected — the bugs were in the unit test harness and a protocol guard, not the simulation layer) +- **Post-`TestCluster` extraction:** all 149 distribution tests pass; `node_death_tombstones_entries` verified 50/50 passes after rewrite to `TestCluster`; all 65 simulation tests unaffected