From c84e708846a5981510813a8f4dbf9d7c42fe3c4a Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 13 Feb 2026 08:29:37 +0000 Subject: [PATCH 01/11] =?UTF-8?q?feat:=20registry=20sim=20tests=20?= =?UTF-8?q?=E2=80=94=204=20pass,=202=20ignored=20(SWIM=20death=20bug)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add sim infrastructure for registry testing: - Extend DistributionSnapshot with registry_size, registry_tombstone_count - Add SimAction enum (RegisterName, RegisterNameWithActor, UnregisterName) - Add action_schedule to DistributionSimConfig for mid-sim registry ops - Add run_simulation_with_nodes() returning node references for assertions - Add property checks: registry_propagation, registry_tombstones, etc. 6 registry sim tests: - registry_name_converges_across_cluster (PASS) - split_brain_naming_converges_after_partition_heals (IGNORED — bug) - tombstone_propagates_when_name_owner_dies (IGNORED — bug) - rapid_re_registration_converges_to_latest (PASS) - simultaneous_registration_converges_deterministically (PASS) - multiple_names_from_different_nodes_all_propagate (PASS) BUG FOUND: SwimProbe::check_suspicion_timeouts() calls members.declare_dead() internally before SwimNode::translate_probe_actions() processes the DeclareDead action, so the second declare_dead() returns false and MembershipChanged{Dead} is never emitted. This silently breaks all death-related side effects: RT cleanup, cache invalidation, repair queue population, and registry tombstoning. Authored by Claude, lovingly guided by Zachery Aaron Shores-Chmielewski --- .../simulation/src/distribution/properties.rs | 170 ++++++++++ crates/simulation/src/distribution/sim.rs | 114 ++++++- crates/simulation/src/distribution/trace.rs | 5 + crates/simulation/tests/debug_registry.rs | 56 +++ .../simulation/tests/distribution_registry.rs | 318 ++++++++++++++++++ 5 files changed, 660 insertions(+), 3 deletions(-) create mode 100644 crates/simulation/tests/debug_registry.rs create mode 100644 crates/simulation/tests/distribution_registry.rs 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..689b175 100644 --- a/crates/simulation/src/distribution/sim.rs +++ b/crates/simulation/src/distribution/sim.rs @@ -21,6 +21,17 @@ 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 }, +} + /// Schedule entry for network faults. #[derive(Debug, Clone)] pub enum NetworkFault { @@ -49,6 +60,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 +87,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, } } } @@ -134,11 +155,25 @@ impl NetworkState { 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 +187,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 +320,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 +365,60 @@ 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, + }, + }); + } + } + } + } + } + } + // Tick all live nodes and deliver actions. for _ in 0..config.ticks_per_round { tick_all_and_deliver( @@ -393,6 +484,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 +494,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 +509,26 @@ 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, 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/tests/debug_registry.rs b/crates/simulation/tests/debug_registry.rs new file mode 100644 index 0000000..3ace815 --- /dev/null +++ b/crates/simulation/tests/debug_registry.rs @@ -0,0 +1,56 @@ +use simulation::distribution::sim::{ + run_simulation_with_nodes, DistributionSimConfig, SimAction, +}; + +#[test] +fn debug_tombstone_propagation() { + let config = DistributionSimConfig { + name: "debug-tombstone".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: "svc".into() }), + ], + kill_schedule: vec![(15, 0)], + ..DistributionSimConfig::default() + }; + + let (trace, nodes) = run_simulation_with_nodes(config); + + // Check snapshots at various rounds + for round_idx in [4, 9, 14, 19, 24, 39, 59, 79] { + if round_idx < trace.snapshots_per_round.len() { + let snaps = &trace.snapshots_per_round[round_idx]; + let reg_sizes: Vec = snaps.iter().map(|(_, s)| s.registry_size).collect(); + let tomb_counts: Vec = snaps.iter().map(|(_, s)| s.registry_tombstone_count).collect(); + let alive: Vec = snaps.iter().map(|(_, s)| s.is_alive).collect(); + let members: Vec = snaps.iter().map(|(_, s)| s.member_count).collect(); + eprintln!("Round {}: alive={alive:?} members={members:?} registry={reg_sizes:?} tombstones={tomb_counts:?}", round_idx + 1); + } + } + + // Check membership change events + let dead_events: Vec<_> = trace.events.iter() + .filter(|e| matches!(&e.kind, simulation::distribution::trace::DistributionEventKind::MembershipChanged { new_state, .. } if new_state == "Dead")) + .collect(); + eprintln!("Dead events: {}", dead_events.len()); + for e in &dead_events { + if let simulation::distribution::trace::DistributionEventKind::MembershipChanged { target, new_state } = &e.kind { + eprintln!(" tick={} node={} declared {target} as {new_state}", e.tick, e.node_name); + } + } + + // Check final nodes' resolve_name + for (i, node) in nodes.iter().enumerate() { + match node { + Some(n) => { + let res = n.resolve_name("svc"); + eprintln!("Node {i}: resolve_name('svc') = {res:?}, registry_size={}, tombstones={}", + n.registry().len(), n.registry().tombstone_count()); + } + None => eprintln!("Node {i}: DEAD"), + } + } +} diff --git a/crates/simulation/tests/distribution_registry.rs b/crates/simulation/tests/distribution_registry.rs new file mode 100644 index 0000000..ec1ae96 --- /dev/null +++ b/crates/simulation/tests/distribution_registry.rs @@ -0,0 +1,318 @@ +//! 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, run_simulation_with_nodes, DistributionSimConfig, NetworkFault, Partition, + SimAction, +}; + +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); + + // 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] +#[ignore = "BUG: SwimProbe::check_suspicion_timeouts calls declare_dead before translate_probe_actions, so MembershipChanged{Dead} is never emitted"] +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, + suspicion_timeout: 5, + dead_reprobe_interval: 10, + }, + ..default_config() + }; + + // When: we run the simulation + let (_trace, nodes) = run_simulation_with_nodes(config); + + // 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] +#[ignore = "BUG: SwimProbe::check_suspicion_timeouts calls declare_dead before translate_probe_actions, so MembershipChanged{Dead} is never emitted"] +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); + + // 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); + + // 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); + + // 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); + + // 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}'" + ); + } + } +} -- 2.45.2 From 5d8e413db251e0fe954c279f6cec214f6bd4e106 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 13 Feb 2026 08:39:14 +0000 Subject: [PATCH 02/11] =?UTF-8?q?fix:=20three=20SWIM=20notification=20bugs?= =?UTF-8?q?=20=E2=80=94=20death=20dissemination,=20piggyback=20notificatio?= =?UTF-8?q?ns,=20partition=20recovery?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bug 1 (swim/node.rs): SwimProbe::check_suspicion_timeouts() calls members.declare_dead() before translate_probe_actions() processes the DeclareDead action. The second declare_dead() returned false (already dead), so MembershipChanged{Dead} was never emitted and the death was never enqueued for dissemination. Fix: remove the redundant declare_dead() call in translate_probe_actions since the probe already performed the mutation. Bug 2 (swim/node.rs): apply_membership_update() — which processes piggyback on every ping/ack/ping_req — updated the internal member list but never emitted NodeAction::MembershipChanged. This meant DistributedNode was blind to all state transitions learned via gossip piggyback (e.g., a dead node refuting via incarnation bump). Fix: return MembershipChanged actions from apply_piggyback and propagate through handle_ping/handle_ack/handle_ping_req. Bug 3 (node.rs): DistributedNode::handle_ping/handle_ack/handle_ping_req never processed MembershipChanged actions from SwimNode — only tick() did. Fix: extract process_membership_changes() helper and call it from all four message paths (tick, handle_ping, handle_ack, handle_ping_req). Additional fixes: - registry.rs: add re_disseminate_all() for anti-entropy on partition heal - node.rs: call re_disseminate_all on MemberState::Alive transitions so registry state accumulated during partition reaches recovering nodes - cluster_scenarios: enable dead_reprobe in 10% message loss test, since correct death dissemination (now working) causes cascading false deaths without a recovery mechanism Authored by Claude, lovingly guided by Zachery Aaron Shores-Chmielewski --- crates/distribution/src/node.rs | 26 ++++---- crates/distribution/src/registry.rs | 11 ++++ crates/distribution/src/swim/node.rs | 59 +++++++++++-------- crates/simulation/tests/cluster_scenarios.rs | 11 ++-- .../simulation/tests/distribution_registry.rs | 2 - 5 files changed, 68 insertions(+), 41 deletions(-) diff --git a/crates/distribution/src/node.rs b/crates/distribution/src/node.rs index 6dc9a5f..a11d18c 100644 --- a/crates/distribution/src/node.rs +++ b/crates/distribution/src/node.rs @@ -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); @@ -169,6 +159,7 @@ impl DistributedNode { 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 actions = self.swim.handle_ping(from, from_addr, sequence, &membership_bytes); + self.process_membership_changes(&actions); self.maybe_update_routing_table(from, from_addr); self.inject_registry_piggyback(actions) } @@ -176,12 +167,14 @@ impl DistributedNode { pub fn handle_ack(&mut self, from: NodeId, sequence: u64, piggyback: &[u8]) -> Vec { let membership_bytes = self.extract_registry_piggyback(piggyback); let actions = self.swim.handle_ack(from, sequence, &membership_bytes); + self.process_membership_changes(&actions); 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 actions = self.swim.handle_ping_req(from, target, target_addr, sequence, &membership_bytes); + self.process_membership_changes(&actions); self.inject_registry_piggyback(actions) } @@ -310,12 +303,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); 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..27696bc 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() } } @@ -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/tests/cluster_scenarios.rs b/crates/simulation/tests/cluster_scenarios.rs index 6c256c9..e858963 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, @@ -139,7 +140,7 @@ fn cluster_converges_under_10_percent_message_loss() { probe_timeout: 5, indirect_probes: 2, suspicion_timeout: 20, - dead_reprobe_interval: 0, + dead_reprobe_interval: 30, }, 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); diff --git a/crates/simulation/tests/distribution_registry.rs b/crates/simulation/tests/distribution_registry.rs index ec1ae96..d5babda 100644 --- a/crates/simulation/tests/distribution_registry.rs +++ b/crates/simulation/tests/distribution_registry.rs @@ -72,7 +72,6 @@ fn registry_name_converges_across_cluster() { // ──────────────────────────────────────────────────────────────────────────── #[test] -#[ignore = "BUG: SwimProbe::check_suspicion_timeouts calls declare_dead before translate_probe_actions, so MembershipChanged{Dead} is never emitted"] 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 @@ -142,7 +141,6 @@ fn split_brain_naming_converges_after_partition_heals() { // ──────────────────────────────────────────────────────────────────────────── #[test] -#[ignore = "BUG: SwimProbe::check_suspicion_timeouts calls declare_dead before translate_probe_actions, so MembershipChanged{Dead} is never emitted"] 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 { -- 2.45.2 From 3226ba37644f5a481e2f1d0e3fc1209f84f27683 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 13 Feb 2026 08:53:07 +0000 Subject: [PATCH 03/11] feat: lifecycle and property sim tests, remove debug test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 3 — lifecycle sim tests (distribution_lifecycle.rs): - dead_node_triggers_repair_queue_and_cache_invalidation - revived_node_has_empty_directory - cache_shrinks_after_node_death - routing_table_recovers_after_partition_heals - routing_table_bounded_by_alive_count Phase 4 — property-based sim tests (distribution_properties.rs): - routing_table_bounded_across_configs (3 config variants) - cache_bounded_across_configs (2 config variants) - repair_queue_populates_on_death_with_directory_entries - registry_eventually_consistent_across_configs (3 config variants) - cascading_deaths_maintain_invariants Also: update cluster_scenarios 10% message loss test to use suspicion_timeout=60 + indirect_probes=3 + dead_reprobe=15 for resilience under correct death dissemination. Remove diagnostic debug_registry.rs (superseded by distribution_registry.rs). Authored by Claude, lovingly guided by Zachery Aaron Shores-Chmielewski --- crates/simulation/tests/cluster_scenarios.rs | 6 +- crates/simulation/tests/debug_registry.rs | 56 ---- .../tests/distribution_lifecycle.rs | 237 ++++++++++++++++ .../tests/distribution_properties.rs | 260 ++++++++++++++++++ 4 files changed, 500 insertions(+), 59 deletions(-) delete mode 100644 crates/simulation/tests/debug_registry.rs create mode 100644 crates/simulation/tests/distribution_lifecycle.rs create mode 100644 crates/simulation/tests/distribution_properties.rs diff --git a/crates/simulation/tests/cluster_scenarios.rs b/crates/simulation/tests/cluster_scenarios.rs index e858963..82063e1 100644 --- a/crates/simulation/tests/cluster_scenarios.rs +++ b/crates/simulation/tests/cluster_scenarios.rs @@ -138,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: 30, + indirect_probes: 3, + suspicion_timeout: 60, + dead_reprobe_interval: 15, }, network_faults: vec![NetworkFault::SetDropRate { round: 1, diff --git a/crates/simulation/tests/debug_registry.rs b/crates/simulation/tests/debug_registry.rs deleted file mode 100644 index 3ace815..0000000 --- a/crates/simulation/tests/debug_registry.rs +++ /dev/null @@ -1,56 +0,0 @@ -use simulation::distribution::sim::{ - run_simulation_with_nodes, DistributionSimConfig, SimAction, -}; - -#[test] -fn debug_tombstone_propagation() { - let config = DistributionSimConfig { - name: "debug-tombstone".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: "svc".into() }), - ], - kill_schedule: vec![(15, 0)], - ..DistributionSimConfig::default() - }; - - let (trace, nodes) = run_simulation_with_nodes(config); - - // Check snapshots at various rounds - for round_idx in [4, 9, 14, 19, 24, 39, 59, 79] { - if round_idx < trace.snapshots_per_round.len() { - let snaps = &trace.snapshots_per_round[round_idx]; - let reg_sizes: Vec = snaps.iter().map(|(_, s)| s.registry_size).collect(); - let tomb_counts: Vec = snaps.iter().map(|(_, s)| s.registry_tombstone_count).collect(); - let alive: Vec = snaps.iter().map(|(_, s)| s.is_alive).collect(); - let members: Vec = snaps.iter().map(|(_, s)| s.member_count).collect(); - eprintln!("Round {}: alive={alive:?} members={members:?} registry={reg_sizes:?} tombstones={tomb_counts:?}", round_idx + 1); - } - } - - // Check membership change events - let dead_events: Vec<_> = trace.events.iter() - .filter(|e| matches!(&e.kind, simulation::distribution::trace::DistributionEventKind::MembershipChanged { new_state, .. } if new_state == "Dead")) - .collect(); - eprintln!("Dead events: {}", dead_events.len()); - for e in &dead_events { - if let simulation::distribution::trace::DistributionEventKind::MembershipChanged { target, new_state } = &e.kind { - eprintln!(" tick={} node={} declared {target} as {new_state}", e.tick, e.node_name); - } - } - - // Check final nodes' resolve_name - for (i, node) in nodes.iter().enumerate() { - match node { - Some(n) => { - let res = n.resolve_name("svc"); - eprintln!("Node {i}: resolve_name('svc') = {res:?}, registry_size={}, tombstones={}", - n.registry().len(), n.registry().tombstone_count()); - } - None => eprintln!("Node {i}: DEAD"), - } - } -} diff --git a/crates/simulation/tests/distribution_lifecycle.rs b/crates/simulation/tests/distribution_lifecycle.rs new file mode 100644 index 0000000..fb9f31f --- /dev/null +++ b/crates/simulation/tests/distribution_lifecycle.rs @@ -0,0 +1,237 @@ +//! 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, run_simulation_with_nodes, DistributionSimConfig, NetworkFault, Partition, +}; + +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); + + // 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); + + // 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); + + // 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); + + // 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); + + let result = check_routing_table_bounded(&trace); + assert!( + result.passed, + "routing table should never exceed alive count: {}", + result.actual + ); +} diff --git a/crates/simulation/tests/distribution_properties.rs b/crates/simulation/tests/distribution_properties.rs new file mode 100644 index 0000000..6e46385 --- /dev/null +++ b/crates/simulation/tests/distribution_properties.rs @@ -0,0 +1,260 @@ +//! 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, SimAction, +}; + +// ──────────────────────────────────────────────────────────────────────────── +// 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); + 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); + 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); + + // 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); + 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); + + 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 + ); +} -- 2.45.2 From 1c07f02be2754191464927e5be0deea11d197151 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 13 Feb 2026 09:21:59 +0000 Subject: [PATCH 04/11] fix: SWIM Suspect recovery + piggyback ordering race condition Two bugs found by simulation testing: 1. Suspect nodes could never recover after dissemination budget expired. When probing a target, the Suspect state was not re-enqueued into the dissemination queue (only Dead was). After budget exhaustion, the Suspect node never received a piggyback telling it it was suspected, so it could never refute via incarnation bump. Fix: re-enqueue both Suspect and Dead state on outgoing probes. 2. Registry entries merged before membership changes in same piggyback. When a message carried both a death notification and registry entries, the entries were merged first, then immediately tombstoned. Fix: split extract_registry_piggyback into unpack + deferred merge, processing membership changes before merging registry entries. Authored by Claude, lovingly guided by Zachery Aaron Shores-Chmielewski --- crates/distribution/src/node.rs | 24 ++++++++++++++---------- crates/distribution/src/swim/node.rs | 10 +++++----- 2 files changed, 19 insertions(+), 15 deletions(-) diff --git a/crates/distribution/src/node.rs b/crates/distribution/src/node.rs index a11d18c..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; @@ -157,24 +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) } @@ -362,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/swim/node.rs b/crates/distribution/src/swim/node.rs index 27696bc..6abffc6 100644 --- a/crates/distribution/src/swim/node.rs +++ b/crates/distribution/src/swim/node.rs @@ -272,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(), ); } -- 2.45.2 From 9327d0734cba7ebdee124799532277f67cc669a8 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 13 Feb 2026 09:22:07 +0000 Subject: [PATCH 05/11] feat: edge case registry sim tests + GracefulLeave action Cycle 2 tests: - explicit_unregister_propagates_to_all_nodes - re_registration_after_tombstone_succeeds - all_names_tombstoned_when_owner_dies - graceful_leave_tombstones_registry_names - piggyback_contention_both_propagate Also: - Add GracefulLeave variant to SimAction enum - Fix cluster_survives_brief_message_loss config (dead_reprobe + suspicion_timeout) - Harden split_brain test config (suspicion_timeout=200 prevents cascading false deaths) Authored by Claude, lovingly guided by Zachery Aaron Shores-Chmielewski --- crates/simulation/src/distribution/sim.rs | 39 ++++ crates/simulation/tests/cluster_scenarios.rs | 10 +- .../simulation/tests/distribution_registry.rs | 194 +++++++++++++++++- 3 files changed, 238 insertions(+), 5 deletions(-) diff --git a/crates/simulation/src/distribution/sim.rs b/crates/simulation/src/distribution/sim.rs index 689b175..7cca65f 100644 --- a/crates/simulation/src/distribution/sim.rs +++ b/crates/simulation/src/distribution/sim.rs @@ -30,6 +30,8 @@ pub enum SimAction { 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. @@ -415,6 +417,43 @@ fn run_simulation_inner(config: DistributionSimConfig) -> (DistTrace, Vec