Exercise MST concurrency safety (#201)

.github/workflows/ci.yml

@@ -20,7 +20,11 @@ jobs:
       # insertion order and the random draws, producing a different graph — and
       # therefore different covered lines — on each run. That made whole-workspace
       # coverage wobble run-to-run and false-tripped the ratchet. One worker keeps
-      # the ratchet input reproducible and matched to the coverage-main baseline.
+      # the ratchet input reproducible and matched to the coverage-main baseline;
+      # it controls coverage-ratchet determinism only. MST concurrency coverage
+      # comes from a `std::thread` `try_union` stress test and an explicit
+      # `ThreadPoolBuilder` 1-versus-8-thread determinism test, both of which use
+      # real concurrency regardless of this pin.
       RAYON_NUM_THREADS: '1'
     steps:
       - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1

chutoro-core/src/mst/property/concurrency.rs

@@ -8,6 +8,7 @@
 //! See `docs/property-testing-design.md` §4.3.3.
 
 use proptest::test_runner::{TestCaseError, TestCaseResult};
+use rayon::ThreadPoolBuilder;
 
 use crate::{EdgeHarvest, MstEdge, parallel_kruskal};
 
@@ -92,3 +93,61 @@ pub(super) fn run_concurrency_safety_property(fixture: &MstFixture) -> TestCaseR
 
     Ok(())
 }
+
+/// Runs the thread-pool determinism property for the given fixture.
+///
+/// Executes `parallel_kruskal` in dedicated one- and eight-thread Rayon pools
+/// and asserts that both runs produce exactly the same forest.
+pub(super) fn run_thread_pool_determinism_property(fixture: &MstFixture) -> TestCaseResult {
+    let single_thread_pool = ThreadPoolBuilder::new()
+        .num_threads(1)
+        .build()
+        .map_err(|error| {
+            TestCaseError::fail(format!("failed to build one-thread Rayon pool: {error}"))
+        })?;
+    let eight_thread_pool = ThreadPoolBuilder::new()
+        .num_threads(8)
+        .build()
+        .map_err(|error| {
+            TestCaseError::fail(format!("failed to build eight-thread Rayon pool: {error}"))
+        })?;
+
+    let single_thread_harvest = EdgeHarvest::new(fixture.edges.clone());
+    let single_thread_forest = single_thread_pool
+        .install(|| parallel_kruskal(fixture.node_count, &single_thread_harvest))
+        .map_err(|error| {
+            TestCaseError::fail(format!(
+                "one-thread parallel_kruskal failed: {error} \
+                 (distribution={:?}, nodes={}, edges={})",
+                fixture.distribution,
+                fixture.node_count,
+                fixture.edges.len(),
+            ))
+        })?;
+
+    let eight_thread_harvest = EdgeHarvest::new(fixture.edges.clone());
+    let eight_thread_forest = eight_thread_pool
+        .install(|| parallel_kruskal(fixture.node_count, &eight_thread_harvest))
+        .map_err(|error| {
+            TestCaseError::fail(format!(
+                "eight-thread parallel_kruskal failed: {error} \
+                 (distribution={:?}, nodes={}, edges={})",
+                fixture.distribution,
+                fixture.node_count,
+                fixture.edges.len(),
+            ))
+        })?;
+
+    if single_thread_forest != eight_thread_forest {
+        return Err(TestCaseError::fail(format!(
+            "parallel_kruskal output diverged by Rayon thread count \
+             (distribution={:?}, nodes={}, edges={}, one_thread={single_thread_forest:?}, \
+             eight_threads={eight_thread_forest:?})",
+            fixture.distribution,
+            fixture.node_count,
+            fixture.edges.len(),
+        )));
+    }
+
+    Ok(())
+}

chutoro-core/src/mst/property/tests.rs

@@ -1,9 +1,9 @@
 //! Property-based test runners for the parallel Kruskal MST implementation.
 //!
-//! Hosts proptest runners for all three properties (oracle equivalence,
-//! structural invariants, concurrency safety), rstest parameterized cases
-//! for targeted distribution coverage, and unit tests for the sequential
-//! oracle itself.
+//! Hosts proptest runners for all four properties (oracle equivalence,
+//! structural invariants, repeatability, thread-pool determinism), rstest
+//! parameterized cases for targeted distribution coverage, and unit tests for
+//! the sequential oracle itself.
 
 use proptest::prelude::*;
 use rand::SeedableRng;
@@ -12,7 +12,7 @@ use rand::rngs::SmallRng;
 use crate::CandidateEdge;
 use crate::test_utils::suite_proptest_config;
 
-use super::concurrency::run_concurrency_safety_property;
+use super::concurrency::{run_concurrency_safety_property, run_thread_pool_determinism_property};
 use super::equivalence::run_oracle_equivalence_property;
 use super::oracle::{SequentialMstResult, sequential_kruskal};
 use super::strategies::{generate_fixture, mst_fixture_strategy};
@@ -89,6 +89,11 @@ proptest! {
     fn mst_concurrency_safety(fixture in mst_fixture_strategy()) {
         run_concurrency_safety_property(&fixture)?;
     }
+
+    #[test]
+    fn mst_thread_pool_determinism(fixture in mst_fixture_strategy()) {
+        run_thread_pool_determinism_property(&fixture)?;
+    }
 }
 
 // ========================================================================
@@ -113,6 +118,12 @@ parameterised_property_test!(
     "concurrency safety must hold"
 );
 
+parameterised_property_test!(
+    thread_pool_determinism_rstest,
+    run_thread_pool_determinism_property,
+    "thread-pool determinism must hold"
+);
+
 // ========================================================================
 // TEST_CASES Consistency Check
 // ========================================================================

chutoro-core/src/mst/union_find.rs

@@ -46,6 +46,11 @@ impl ConcurrentUnionFind {
         self.components.load(Ordering::Acquire)
     }
 
+    #[cfg(test)]
+    pub(super) fn root_of(&self, node: usize) -> usize {
+        self.find(node)
+    }
+
     pub(super) fn try_union(&self, left: usize, right: usize) -> Result<bool, MstError> {
         loop {
             let left_root = self.find(left);
@@ -156,3 +161,112 @@ fn choose_parent_child(
 
     lock_order(left_root, right_root)
 }
+
+#[cfg(test)]
+mod tests {
+    //! Concurrent stress coverage for the striped-lock union-find protocol.
+
+    use std::sync::{Arc, Barrier};
+
+    use rand::Rng;
+    use rand::SeedableRng;
+    use rand::rngs::SmallRng;
+    use rstest::rstest;
+
+    use super::ConcurrentUnionFind;
+
+    const NODE_COUNT: usize = 8;
+    const EDGE_COUNT: usize = 4_096;
+
+    #[rstest]
+    #[case(42, 2)]
+    #[case(999, 4)]
+    #[case(7_777, 8)]
+    fn concurrent_unions_match_sequential_partition(
+        #[case] seed: u64,
+        #[case] thread_count: usize,
+    ) {
+        // Multiple workers contend for the same small lock table, exercising
+        // striped-lock ordering, root revalidation, and retry interleavings.
+        let edges = Arc::new(random_edges(seed));
+        let union_find = Arc::new(ConcurrentUnionFind::new(NODE_COUNT));
+        let start = Arc::new(Barrier::new(thread_count + 1));
+        let chunk_size = edges.len().div_ceil(thread_count);
+
+        let handles: Vec<_> = (0..thread_count)
+            .map(|worker_index| {
+                let edges = Arc::clone(&edges);
+                let union_find = Arc::clone(&union_find);
+                let start = Arc::clone(&start);
+                let first = worker_index * chunk_size;
+                let last = (first + chunk_size).min(edges.len());
+
+                std::thread::spawn(move || {
+                    start.wait();
+                    for &(left, right) in &edges[first..last] {
+                        union_find
+                            .try_union(left, right)
+                            .expect("generated nodes must be valid");
+                    }
+                })
+            })
+            .collect();
+
+        start.wait();
+        for handle in handles {
+            handle.join().expect("union worker must not panic");
+        }
+
+        let concurrent_labels = normalised_labels(|node| union_find.root_of(node));
+        let (oracle_labels, oracle_components) = sequential_oracle(&edges);
+
+        assert_eq!(concurrent_labels, oracle_labels);
+        assert_eq!(union_find.components(), oracle_components);
+    }
+
+    fn random_edges(seed: u64) -> Vec<(usize, usize)> {
+        let mut rng = SmallRng::seed_from_u64(seed);
+        (0..EDGE_COUNT)
+            .map(|_| (rng.gen_range(0..NODE_COUNT), rng.gen_range(0..NODE_COUNT)))
+            .collect()
+    }
+
+    fn normalised_labels(root_of: impl Fn(usize) -> usize) -> Vec<usize> {
+        let mut component_minimums = [NODE_COUNT; NODE_COUNT];
+        for node in 0..NODE_COUNT {
+            let root = root_of(node);
+            component_minimums[root] = component_minimums[root].min(node);
+        }
+
+        (0..NODE_COUNT)
+            .map(|node| component_minimums[root_of(node)])
+            .collect()
+    }
+
+    fn sequential_oracle(edges: &[(usize, usize)]) -> (Vec<usize>, usize) {
+        let mut parents: Vec<usize> = (0..NODE_COUNT).collect();
+        let mut components = NODE_COUNT;
+
+        for &(left, right) in edges {
+            let left_root = scalar_find(&parents, left);
+            let right_root = scalar_find(&parents, right);
+            if left_root != right_root {
+                parents[right_root] = left_root;
+                components -= 1;
+            }
+        }
+
+        (
+            normalised_labels(|node| scalar_find(&parents, node)),
+            components,
+        )
+    }
+
+    fn scalar_find(parents: &[usize], node: usize) -> usize {
+        let mut current = node;
+        while parents[current] != current {
+            current = parents[current];
+        }
+        current
+    }
+}