feat: cycle 4 property tests — asymmetric partition, revive, monotonicity, large cluster

New tests:
- asymmetric_partition_registry_converges_after_heal (one-way reachability)
- revived_node_re_registration_overwrites_tombstone (death + revive + takeover)
- registry_convergence_is_monotonic_in_stable_cluster (once converged, stays converged)
- large_cluster_registry_converges (15-node stress with 2 deaths)

Total: 15 registry + 5 lifecycle + 9 property = 29 distribution sim tests.

Authored by Claude, lovingly guided by Zachery Aaron Shores-Chmielewski

crates/simulation/tests/distribution_properties.rs

@@ -8,7 +8,7 @@ use simulation::distribution::properties::{
     check_routing_table_bounded,
 };
 use simulation::distribution::sim::{
-    run_simulation_with_nodes, DistributionSimConfig, SimAction,
+    run_simulation_with_nodes, DistributionSimConfig, NetworkFault, Partition, SimAction,
 };
 
 // ────────────────────────────────────────────────────────────────────────────
@@ -258,3 +258,229 @@ fn cascading_deaths_maintain_invariants() {
         repair_result.actual
     );
 }
+
+// ────────────────────────────────────────────────────────────────────────────
+// 6. Asymmetric partition + registry — one-way reachability
+// ────────────────────────────────────────────────────────────────────────────
+
+#[test]
+fn asymmetric_partition_registry_converges_after_heal() {
+    // Given: 6 nodes, asymmetric partition: A→B blocked, B→A works.
+    // Node 0 (side A) and node 3 (side B) each register a name.
+    // After heal, all should converge.
+    let config = DistributionSimConfig {
+        name: "asymmetric-partition-registry".into(),
+        num_nodes: 6,
+        num_rounds: 100,
+        ticks_per_round: 3,
+        actors_per_node: 0,
+        network_faults: vec![
+            NetworkFault::Partition {
+                round: 10,
+                partition: Partition {
+                    side_a: vec![0, 1, 2],
+                    side_b: vec![3, 4, 5],
+                    asymmetric: true, // A→B blocked, B→A works
+                },
+            },
+            NetworkFault::Heal { round: 40 },
+        ],
+        action_schedule: vec![
+            (12, SimAction::RegisterName { node_idx: 0, name: "from-a".into() }),
+            (12, SimAction::RegisterName { node_idx: 3, name: "from-b".into() }),
+        ],
+        swim: distribution::swim::probe::SwimConfig {
+            probe_interval: 1,
+            probe_timeout: 3,
+            indirect_probes: 1,
+            suspicion_timeout: 200,
+            dead_reprobe_interval: 10,
+        },
+        ..DistributionSimConfig::default()
+    };
+
+    let (_trace, nodes) = run_simulation_with_nodes(config);
+
+    // After healing, all nodes should resolve both names
+    let alive_nodes: Vec<_> = nodes.iter().filter_map(|n| n.as_ref()).collect();
+    for node in &alive_nodes {
+        // Side B's name should have been reachable from side A even during partition
+        // (B→A works), so "from-b" should propagate to everyone.
+        // "from-a" might need post-heal gossip to reach side B.
+        assert!(
+            node.resolve_name("from-b").is_some(),
+            "all nodes should resolve 'from-b' (B→A was always open)"
+        );
+    }
+
+    // After 60 rounds of healed connectivity, "from-a" should also propagate
+    let resolved_a: Vec<_> = alive_nodes
+        .iter()
+        .filter(|n| n.resolve_name("from-a").is_some())
+        .collect();
+    assert!(
+        resolved_a.len() >= 4,
+        "at least 4 of 6 nodes should resolve 'from-a' after partition heals, got {}",
+        resolved_a.len()
+    );
+}
+
+// ────────────────────────────────────────────────────────────────────────────
+// 7. Revived node re-registers — new registration overwrites tombstone
+// ────────────────────────────────────────────────────────────────────────────
+
+#[test]
+fn revived_node_re_registration_overwrites_tombstone() {
+    use swactor::actor::ActorAddress;
+    let new_actor = ActorAddress::new_random();
+
+    // Given: 5 nodes, node 2 registers "svc", is killed, revived with fresh state,
+    // then node 3 re-registers "svc" with a new actor.
+    // (Don't kill node 0 since it's the join seed.)
+    let config = DistributionSimConfig {
+        name: "revive-reregister".into(),
+        num_nodes: 5,
+        num_rounds: 120,
+        ticks_per_round: 3,
+        actors_per_node: 0,
+        action_schedule: vec![
+            (5, SimAction::RegisterName { node_idx: 2, name: "svc".into() }),
+            // After death + revive, a different surviving node re-registers
+            (60, SimAction::RegisterNameWithActor { node_idx: 3, name: "svc".into(), actor: new_actor }),
+        ],
+        kill_schedule: vec![(15, 2)],
+        revive_schedule: vec![(40, 2)],
+        ..DistributionSimConfig::default()
+    };
+
+    let (_trace, nodes) = run_simulation_with_nodes(config);
+
+    // Then: all alive nodes should resolve "svc" to the new registration
+    let alive_nodes: Vec<_> = nodes.iter().filter_map(|n| n.as_ref()).collect();
+    for node in &alive_nodes {
+        assert!(
+            node.resolve_name("svc").is_some(),
+            "all nodes should resolve 'svc' after re-registration by surviving node"
+        );
+        assert_eq!(
+            node.resolve_name("svc").unwrap().0,
+            new_actor,
+            "all nodes should resolve 'svc' to the new actor"
+        );
+    }
+}
+
+// ────────────────────────────────────────────────────────────────────────────
+// 8. Registry convergence is monotonic — divergence doesn't increase
+// ────────────────────────────────────────────────────────────────────────────
+
+#[test]
+fn registry_convergence_is_monotonic_in_stable_cluster() {
+    // Given: 8-node cluster, 4 names registered at round 5, no faults
+    let config = DistributionSimConfig {
+        name: "registry-monotonic".into(),
+        num_nodes: 8,
+        num_rounds: 60,
+        ticks_per_round: 3,
+        actors_per_node: 0,
+        action_schedule: vec![
+            (5, SimAction::RegisterName { node_idx: 0, name: "alpha".into() }),
+            (5, SimAction::RegisterName { node_idx: 2, name: "beta".into() }),
+            (5, SimAction::RegisterName { node_idx: 4, name: "gamma".into() }),
+            (5, SimAction::RegisterName { node_idx: 6, name: "delta".into() }),
+        ],
+        ..DistributionSimConfig::default()
+    };
+
+    let (trace, _) = run_simulation_with_nodes(config);
+
+    // Measure "divergence" = number of alive nodes with registry_size < 4
+    // Once it reaches 0, it should never increase again
+    let mut reached_convergence = false;
+    let mut post_convergence_divergence = 0;
+
+    for round_snaps in &trace.snapshots_per_round {
+        let alive_with_full_registry = round_snaps
+            .iter()
+            .filter(|(_, s)| s.is_alive && s.registry_size >= 4)
+            .count();
+        let alive_count = round_snaps.iter().filter(|(_, s)| s.is_alive).count();
+        let divergent = alive_count - alive_with_full_registry;
+
+        if divergent == 0 && alive_count > 0 {
+            reached_convergence = true;
+        } else if reached_convergence && divergent > 0 {
+            post_convergence_divergence += 1;
+        }
+    }
+
+    assert!(
+        reached_convergence,
+        "registry should converge (all alive nodes see all 4 names)"
+    );
+    assert_eq!(
+        post_convergence_divergence, 0,
+        "once converged, registry should not diverge again in a stable cluster"
+    );
+}
+
+// ────────────────────────────────────────────────────────────────────────────
+// 9. Large cluster registry stress test
+// ────────────────────────────────────────────────────────────────────────────
+
+#[test]
+fn large_cluster_registry_converges() {
+    // Given: 15-node cluster, 5 names registered on different nodes, 2 deaths
+    let config = DistributionSimConfig {
+        name: "large-cluster-registry".into(),
+        num_nodes: 15,
+        num_rounds: 80,
+        ticks_per_round: 3,
+        actors_per_node: 0,
+        action_schedule: vec![
+            (5, SimAction::RegisterName { node_idx: 0, name: "alpha".into() }),
+            (5, SimAction::RegisterName { node_idx: 3, name: "beta".into() }),
+            (5, SimAction::RegisterName { node_idx: 6, name: "gamma".into() }),
+            (5, SimAction::RegisterName { node_idx: 9, name: "delta".into() }),
+            (5, SimAction::RegisterName { node_idx: 12, name: "epsilon".into() }),
+        ],
+        kill_schedule: vec![(20, 0), (20, 3)],
+        ..DistributionSimConfig::default()
+    };
+
+    let (trace, nodes) = run_simulation_with_nodes(config);
+
+    let survivors: Vec<_> = nodes.iter().filter_map(|n| n.as_ref()).collect();
+    assert_eq!(survivors.len(), 13, "13 of 15 nodes should survive");
+
+    // Names owned by dead nodes should be tombstoned
+    for node in &survivors {
+        assert!(
+            node.resolve_name("alpha").is_none(),
+            "alpha (owned by dead node 0) should be tombstoned"
+        );
+        assert!(
+            node.resolve_name("beta").is_none(),
+            "beta (owned by dead node 3) should be tombstoned"
+        );
+    }
+
+    // Names owned by surviving nodes should resolve
+    for node in &survivors {
+        for name in &["gamma", "delta", "epsilon"] {
+            assert!(
+                node.resolve_name(name).is_some(),
+                "'{name}' (owned by surviving node) should resolve across 15-node cluster"
+            );
+        }
+    }
+
+    // Registry propagation: all survivors should have all 5 registry entries
+    // (2 tombstoned + 3 alive)
+    let result = check_registry_propagation(&trace, 5);
+    assert!(
+        result.passed,
+        "all 5 registry entries should propagate in 15-node cluster: {}",
+        result.actual
+    );
+}