From c84e708846a5981510813a8f4dbf9d7c42fe3c4a Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 13 Feb 2026 08:29:37 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20registry=20sim=20tests=20=E2=80=94=204?= =?UTF-8?q?=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}'" + ); + } + } +}