Turbine: chacha8 transition (SIMD 332) (#7353)

* implmentation of chacha8 for broadcast and retransmit

* add chacha8 vs chacha20 tests

* patch in the feature gate

cover also retransmit parent lookups
update benchmarks to test both chacha8 and 20 versions

* improve the tests for complete turbine coverage

* keep the diff small

---------

Co-authored-by: marc <marcinbieszke@gmail.com>

feature-set/src/lib.rs

@@ -1163,6 +1163,10 @@ pub mod vote_state_v4 {
     }
 }
 
+pub mod switch_to_chacha8_turbine {
+    solana_pubkey::declare_id!("CHaChatUnR3s6cPyPMMGNJa3VdQQ8PNH2JqdD4LpCKnB");
+}
+
 pub static FEATURE_NAMES: LazyLock<AHashMap<Pubkey, &'static str>> = LazyLock::new(|| {
     [
         (secp256k1_program_enabled::id(), "secp256k1 program"),
@@ -2090,6 +2094,10 @@ pub static FEATURE_NAMES: LazyLock<AHashMap<Pubkey, &'static str>> = LazyLock::n
              rules",
         ),
         (vote_state_v4::id(), "SIMD-0185: Vote State v4"),
+        (
+            switch_to_chacha8_turbine::id(),
+            "SIMD-0332: Reduce ChaCha rounds for Turbine from 20 to 8",
+        ),
         /*************** ADD NEW FEATURES HERE ***************/
     ]
     .iter()

gossip/src/weighted_shuffle.rs

@@ -279,7 +279,7 @@ mod tests {
         super::*,
         itertools::Itertools,
         rand::SeedableRng,
-        rand_chacha::ChaChaRng,
+        rand_chacha::{ChaCha8Rng, ChaChaRng},
         solana_hash::Hash,
         std::{
             convert::TryInto,
@@ -384,18 +384,33 @@ mod tests {
     }
 
     // Asserts that zero weights will be shuffled.
-    #[test]
-    fn test_weighted_shuffle_zero_weights() {
+    #[test_case(8)]
+    #[test_case(20)]
+    fn test_weighted_shuffle_zero_weights(cha_cha_variant: u8) {
         let weights = vec![0u64; 5];
         let seed = [37u8; 32];
-        let mut rng = ChaChaRng::from_seed(seed);
         let shuffle = WeightedShuffle::new("", weights);
-        assert_eq!(
-            shuffle.clone().shuffle(&mut rng).collect::<Vec<_>>(),
-            [1, 4, 2, 3, 0]
-        );
-        let mut rng = ChaChaRng::from_seed(seed);
-        assert_eq!(shuffle.first(&mut rng), Some(1));
+        match cha_cha_variant {
+            8 => {
+                let mut rng = ChaCha8Rng::from_seed(seed);
+                assert_eq!(
+                    shuffle.clone().shuffle(&mut rng).collect::<Vec<_>>(),
+                    [4, 3, 1, 2, 0],
+                );
+                let mut rng = ChaCha8Rng::from_seed(seed);
+                assert_eq!(shuffle.first(&mut rng), Some(4));
+            }
+            20 => {
+                let mut rng = ChaChaRng::from_seed(seed);
+                assert_eq!(
+                    shuffle.clone().shuffle(&mut rng).collect::<Vec<_>>(),
+                    [1, 4, 2, 3, 0],
+                );
+                let mut rng = ChaChaRng::from_seed(seed);
+                assert_eq!(shuffle.first(&mut rng), Some(1));
+            }
+            _ => unreachable!(),
+        };
     }
 
     // Asserts that each index is selected proportional to its weight.
@@ -404,46 +419,70 @@ mod tests {
         let seed: Vec<_> = (1..).step_by(3).take(32).collect();
         let seed: [u8; 32] = seed.try_into().unwrap();
         let mut rng = ChaChaRng::from_seed(seed);
-        let weights = [1, 0, 1000, 0, 0, 10, 100, 0];
-        let mut counts = [0; 8];
-        for _ in 0..100000 {
-            let mut weighted_shuffle = WeightedShuffle::new("", weights);
-            let mut shuffle = weighted_shuffle.shuffle(&mut rng);
-            counts[shuffle.next().unwrap()] += 1;
-            let _ = shuffle.count(); // consume the rest.
-        }
-        assert_eq!(counts, [95, 0, 90069, 0, 0, 908, 8928, 0]);
-        let mut counts = [0; 8];
-        for _ in 0..100000 {
-            let mut shuffle = WeightedShuffle::new("", weights);
-            shuffle.remove_index(5);
-            shuffle.remove_index(3);
-            shuffle.remove_index(1);
-            let mut shuffle = shuffle.shuffle(&mut rng);
-            counts[shuffle.next().unwrap()] += 1;
-            let _ = shuffle.count(); // consume the rest.
+        test_weighted_shuffle_sanity_impl(
+            &mut rng,
+            &[95, 0, 90069, 0, 0, 908, 8928, 0],
+            &[97, 0, 90862, 0, 0, 0, 9041, 0],
+        );
+        let mut rng = ChaCha8Rng::from_seed(seed);
+        test_weighted_shuffle_sanity_impl(
+            &mut rng,
+            &[93, 0, 90185, 0, 0, 892, 8830, 0],
+            &[89, 0, 90741, 0, 0, 0, 9170, 0],
+        );
+        fn test_weighted_shuffle_sanity_impl<R: Rng>(
+            rng: &mut R,
+            counts1: &[i32],
+            counts2: &[i32],
+        ) {
+            let weights = [1, 0, 1000, 0, 0, 10, 100, 0];
+            let mut counts = [0; 8];
+            for _ in 0..100000 {
+                let mut weighted_shuffle = WeightedShuffle::new("", weights);
+                let mut shuffle = weighted_shuffle.shuffle(rng);
+                counts[shuffle.next().unwrap()] += 1;
+                let _ = shuffle.count(); // consume the rest.
+            }
+            assert_eq!(counts, counts1);
+            let mut counts = [0; 8];
+            for _ in 0..100000 {
+                let mut shuffle = WeightedShuffle::new("", weights);
+                shuffle.remove_index(5);
+                shuffle.remove_index(3);
+                shuffle.remove_index(1);
+                let mut shuffle = shuffle.shuffle(rng);
+                counts[shuffle.next().unwrap()] += 1;
+                let _ = shuffle.count(); // consume the rest.
+            }
+            assert_eq!(counts, counts2);
         }
-        assert_eq!(counts, [97, 0, 90862, 0, 0, 0, 9041, 0]);
     }
 
     #[test]
     fn test_weighted_shuffle_negative_overflow() {
-        const SEED: [u8; 32] = [48u8; 32];
-        let weights = [19i64, 23, 7, 0, 0, 23, 3, 0, 5, 0, 19, 29];
-        let mut rng = ChaChaRng::from_seed(SEED);
-        let mut shuffle = WeightedShuffle::new("", weights);
-        assert_eq!(
-            shuffle.shuffle(&mut rng).collect::<Vec<_>>(),
-            [8, 1, 5, 10, 11, 0, 2, 6, 9, 4, 3, 7]
-        );
-        // Negative weights and overflowing ones are treated as zero.
-        let weights = [19, 23, 7, -57, i64::MAX, 23, 3, i64::MAX, 5, -79, 19, 29];
-        let mut rng = ChaChaRng::from_seed(SEED);
-        let mut shuffle = WeightedShuffle::new("", weights);
-        assert_eq!(
-            shuffle.shuffle(&mut rng).collect::<Vec<_>>(),
-            [8, 1, 5, 10, 11, 0, 2, 6, 9, 4, 3, 7]
-        );
+        test_weighted_shuffle_negative_overflow_impl::<ChaChaRng>(&[
+            8, 1, 5, 10, 11, 0, 2, 6, 9, 4, 3, 7,
+        ]);
+        test_weighted_shuffle_negative_overflow_impl::<ChaCha8Rng>(&[
+            5, 11, 2, 0, 10, 1, 6, 8, 7, 3, 9, 4,
+        ]);
+
+        fn test_weighted_shuffle_negative_overflow_impl<
+            R: Rng + rand::SeedableRng<Seed = [u8; 32]>,
+        >(
+            counts: &[usize],
+        ) {
+            const SEED: [u8; 32] = [48u8; 32];
+            let weights = [19i64, 23, 7, 0, 0, 23, 3, 0, 5, 0, 19, 29];
+            let mut rng = R::from_seed(SEED);
+            let mut shuffle = WeightedShuffle::new("", weights);
+            assert_eq!(shuffle.shuffle(&mut rng).collect::<Vec<_>>(), counts);
+            // Negative weights and overflowing ones are treated as zero.
+            let weights = [19, 23, 7, -57, i64::MAX, 23, 3, i64::MAX, 5, -79, 19, 29];
+            let mut rng = R::from_seed(SEED);
+            let mut shuffle = WeightedShuffle::new("", weights);
+            assert_eq!(shuffle.shuffle(&mut rng).collect::<Vec<_>>(), counts);
+        }
     }
 
     #[test]
@@ -569,36 +608,47 @@ mod tests {
 
     #[test]
     fn test_weighted_shuffle_match_slow() {
-        let mut rng = rand::thread_rng();
-        let weights: Vec<u64> = repeat_with(|| rng.gen_range(0..1000)).take(997).collect();
-        for _ in 0..10 {
-            let mut seed = [0u8; 32];
-            rng.fill(&mut seed[..]);
-            let mut rng = ChaChaRng::from_seed(seed);
-            let mut shuffle = WeightedShuffle::<u64>::new("", &weights);
-            let shuffle: Vec<_> = shuffle.shuffle(&mut rng).collect();
-            let mut rng = ChaChaRng::from_seed(seed);
-            let shuffle_slow = weighted_shuffle_slow(&mut rng, weights.clone());
-            assert_eq!(shuffle, shuffle_slow);
-            let mut rng = ChaChaRng::from_seed(seed);
-            let shuffle = WeightedShuffle::<u64>::new("", &weights);
-            assert_eq!(shuffle.first(&mut rng), Some(shuffle_slow[0]));
+        test_weighted_shuffle_match_slow_impl::<ChaChaRng>();
+        test_weighted_shuffle_match_slow_impl::<ChaCha8Rng>();
+
+        fn test_weighted_shuffle_match_slow_impl<R: Rng + rand::SeedableRng<Seed = [u8; 32]>>() {
+            let mut rng = rand::thread_rng();
+            let weights: Vec<u64> = repeat_with(|| rng.gen_range(0..1000)).take(997).collect();
+            for _ in 0..10 {
+                let mut seed = [0u8; 32];
+                rng.fill(&mut seed[..]);
+                let mut rng = R::from_seed(seed);
+                let mut shuffle = WeightedShuffle::<u64>::new("", &weights);
+                let shuffle: Vec<_> = shuffle.shuffle(&mut rng).collect();
+                let mut rng = R::from_seed(seed);
+                let shuffle_slow = weighted_shuffle_slow(&mut rng, weights.clone());
+                assert_eq!(shuffle, shuffle_slow);
+                let mut rng = R::from_seed(seed);
+                let shuffle = WeightedShuffle::<u64>::new("", &weights);
+                assert_eq!(shuffle.first(&mut rng), Some(shuffle_slow[0]));
+            }
         }
     }
 
     #[test]
     fn test_weighted_shuffle_paranoid() {
         let mut rng = rand::thread_rng();
-        for size in 0..1351 {
-            let weights: Vec<_> = repeat_with(|| rng.gen_range(0..1000)).take(size).collect();
-            let seed = rng.gen::<[u8; 32]>();
-            let mut rng = ChaChaRng::from_seed(seed);
-            let shuffle_slow = weighted_shuffle_slow(&mut rng.clone(), weights.clone());
-            let mut shuffle = WeightedShuffle::new("", weights);
-            if size > 0 {
-                assert_eq!(shuffle.first(&mut rng.clone()), Some(shuffle_slow[0]));
+        let seed = rng.gen::<[u8; 32]>();
+        let rng = ChaCha8Rng::from_seed(seed);
+        test_weighted_shuffle_paranoid_impl(rng);
+        let rng = ChaChaRng::from_seed(seed);
+        test_weighted_shuffle_paranoid_impl(rng);
+
+        fn test_weighted_shuffle_paranoid_impl<R: Rng + Clone>(mut rng: R) {
+            for size in 0..1351 {
+                let weights: Vec<_> = repeat_with(|| rng.gen_range(0..1000)).take(size).collect();
+                let shuffle_slow = weighted_shuffle_slow(&mut rng.clone(), weights.clone());
+                let mut shuffle = WeightedShuffle::new("", weights);
+                if size > 0 {
+                    assert_eq!(shuffle.first(&mut rng.clone()), Some(shuffle_slow[0]));
+                }
+                assert_eq!(shuffle.shuffle(&mut rng).collect::<Vec<_>>(), shuffle_slow);
             }
-            assert_eq!(shuffle.shuffle(&mut rng).collect::<Vec<_>>(), shuffle_slow);
         }
     }
 }

turbine/benches/cluster_nodes.rs

@@ -18,10 +18,15 @@ use {
 fn make_cluster_nodes<R: Rng>(
     rng: &mut R,
     unstaked_ratio: Option<(u32, u32)>,
+    use_cha_cha_8: bool,
 ) -> (Vec<ContactInfo>, ClusterNodes<RetransmitStage>) {
     let (nodes, stakes, cluster_info) = make_test_cluster(rng, 5_000, unstaked_ratio);
-    let cluster_nodes =
-        new_cluster_nodes::<RetransmitStage>(&cluster_info, ClusterType::Development, &stakes);
+    let cluster_nodes = new_cluster_nodes::<RetransmitStage>(
+        &cluster_info,
+        ClusterType::Development,
+        &stakes,
+        use_cha_cha_8,
+    );
     (nodes, cluster_nodes)
 }
 
@@ -58,25 +63,39 @@ fn get_retransmit_peers_deterministic(
     }
 }
 
-fn get_retransmit_peers_deterministic_wrapper(b: &mut Bencher, unstaked_ratio: Option<(u32, u32)>) {
+fn get_retransmit_peers_deterministic_wrapper(
+    b: &mut Bencher,
+    unstaked_ratio: Option<(u32, u32)>,
+    use_cha_cha_8: bool,
+) {
     let mut rng = rand::thread_rng();
-    let (nodes, cluster_nodes) = make_cluster_nodes(&mut rng, unstaked_ratio);
+    let (nodes, cluster_nodes) = make_cluster_nodes(&mut rng, unstaked_ratio, use_cha_cha_8);
     let slot_leader = *nodes[1..].choose(&mut rng).unwrap().pubkey();
     let slot = rand::random::<u64>();
     b.iter(|| get_retransmit_peers_deterministic(&cluster_nodes, slot, &slot_leader));
 }
 
-fn bench_get_retransmit_peers_deterministic_unstaked_ratio_1_2(b: &mut Bencher) {
-    get_retransmit_peers_deterministic_wrapper(b, Some((1, 2)));
+fn bench_get_retransmit_peers_deterministic_unstaked_ratio_1_2_20(b: &mut Bencher) {
+    get_retransmit_peers_deterministic_wrapper(b, Some((1, 2)), false);
+}
+
+fn bench_get_retransmit_peers_deterministic_unstaked_ratio_1_32_20(b: &mut Bencher) {
+    get_retransmit_peers_deterministic_wrapper(b, Some((1, 32)), false);
+}
+
+fn bench_get_retransmit_peers_deterministic_unstaked_ratio_1_2_8(b: &mut Bencher) {
+    get_retransmit_peers_deterministic_wrapper(b, Some((1, 2)), true);
 }
 
-fn bench_get_retransmit_peers_deterministic_unstaked_ratio_1_32(b: &mut Bencher) {
-    get_retransmit_peers_deterministic_wrapper(b, Some((1, 32)));
+fn bench_get_retransmit_peers_deterministic_unstaked_ratio_1_32_8(b: &mut Bencher) {
+    get_retransmit_peers_deterministic_wrapper(b, Some((1, 32)), true);
 }
 
 benchmark_group!(
     benches,
-    bench_get_retransmit_peers_deterministic_unstaked_ratio_1_2,
-    bench_get_retransmit_peers_deterministic_unstaked_ratio_1_32
+    bench_get_retransmit_peers_deterministic_unstaked_ratio_1_2_20,
+    bench_get_retransmit_peers_deterministic_unstaked_ratio_1_32_20,
+    bench_get_retransmit_peers_deterministic_unstaked_ratio_1_2_8,
+    bench_get_retransmit_peers_deterministic_unstaked_ratio_1_32_8
 );
 benchmark_main!(benches);

turbine/src/cluster_nodes.rs

@@ -1,10 +1,10 @@
 use {
     crate::{broadcast_stage::BroadcastStage, retransmit_stage::RetransmitStage},
-    agave_feature_set as feature_set,
+    agave_feature_set::{self as feature_set},
     itertools::Either,
     lazy_lru::LruCache,
-    rand::{seq::SliceRandom, Rng, SeedableRng},
-    rand_chacha::ChaChaRng,
+    rand::{seq::SliceRandom, Rng, RngCore, SeedableRng},
+    rand_chacha::{ChaCha8Rng, ChaChaRng},
     solana_clock::{Epoch, Slot},
     solana_cluster_type::ClusterType,
     solana_gossip::{
@@ -85,6 +85,7 @@ pub struct ClusterNodes<T> {
     // Reverse index from nodes pubkey to their index in self.nodes.
     index: HashMap<Pubkey, /*index:*/ usize>,
     weighted_shuffle: WeightedShuffle</*stake:*/ u64>,
+    use_cha_cha_8: bool,
     _phantom: PhantomData<T>,
 }
 
@@ -200,17 +201,67 @@ impl<T> ClusterNodes<T> {
     }
 }
 
+/// Encapsulates the possible RNG implementations for turbine.
+/// This was implemented for the transition from ChaCha20 to ChaCha8.
+enum TurbineRng {
+    Legacy(ChaChaRng),
+    ChaCha8(ChaCha8Rng),
+}
+
+impl TurbineRng {
+    /// Create a new seeded TurbineRng of the correct implementation
+    fn new_seeded(leader: &Pubkey, shred: &ShredId, use_cha_cha_8: bool) -> Self {
+        let seed = shred.seed(leader);
+        if use_cha_cha_8 {
+            TurbineRng::ChaCha8(ChaCha8Rng::from_seed(seed))
+        } else {
+            TurbineRng::Legacy(ChaChaRng::from_seed(seed))
+        }
+    }
+}
+
+impl RngCore for TurbineRng {
+    fn next_u32(&mut self) -> u32 {
+        match self {
+            TurbineRng::Legacy(cha_cha20_rng) => cha_cha20_rng.next_u32(),
+            TurbineRng::ChaCha8(cha_cha8_rng) => cha_cha8_rng.next_u32(),
+        }
+    }
+
+    fn next_u64(&mut self) -> u64 {
+        match self {
+            TurbineRng::Legacy(cha_cha20_rng) => cha_cha20_rng.next_u64(),
+            TurbineRng::ChaCha8(cha_cha8_rng) => cha_cha8_rng.next_u64(),
+        }
+    }
+
+    fn fill_bytes(&mut self, dest: &mut [u8]) {
+        match self {
+            TurbineRng::Legacy(cha_cha20_rng) => cha_cha20_rng.fill_bytes(dest),
+            TurbineRng::ChaCha8(cha_cha8_rng) => cha_cha8_rng.fill_bytes(dest),
+        }
+    }
+
+    fn try_fill_bytes(&mut self, dest: &mut [u8]) -> Result<(), rand::Error> {
+        match self {
+            TurbineRng::Legacy(cha_cha20_rng) => cha_cha20_rng.try_fill_bytes(dest),
+            TurbineRng::ChaCha8(cha_cha8_rng) => cha_cha8_rng.try_fill_bytes(dest),
+        }
+    }
+}
+
 impl ClusterNodes<BroadcastStage> {
     pub fn new(
         cluster_info: &ClusterInfo,
         cluster_type: ClusterType,
         stakes: &HashMap<Pubkey, u64>,
+        use_cha_cha_8: bool,
     ) -> Self {
-        new_cluster_nodes(cluster_info, cluster_type, stakes)
+        new_cluster_nodes(cluster_info, cluster_type, stakes, use_cha_cha_8)
     }
 
     pub(crate) fn get_broadcast_peer(&self, shred: &ShredId) -> Option<&ContactInfo> {
-        let mut rng = get_seeded_rng(/*leader:*/ &self.pubkey, shred);
+        let mut rng = TurbineRng::new_seeded(&self.pubkey, shred, self.use_cha_cha_8);
         let index = self.weighted_shuffle.first(&mut rng)?;
         self.nodes[index].contact_info()
     }
@@ -236,7 +287,7 @@ impl ClusterNodes<RetransmitStage> {
             if let Some(index) = self.index.get(slot_leader) {
                 weighted_shuffle.remove_index(*index);
             }
-            let mut rng = get_seeded_rng(slot_leader, shred);
+            let mut rng = TurbineRng::new_seeded(slot_leader, shred, self.use_cha_cha_8);
             let (index, peers) = get_retransmit_peers(
                 fanout,
                 |k| self.nodes[k].pubkey() == &self.pubkey,
@@ -284,7 +335,8 @@ impl ClusterNodes<RetransmitStage> {
         if let Some(index) = self.index.get(leader).copied() {
             weighted_shuffle.remove_index(index);
         }
-        let mut rng = get_seeded_rng(leader, shred);
+
+        let mut rng = TurbineRng::new_seeded(leader, shred, self.use_cha_cha_8);
         // Only need shuffled nodes until this node itself.
         let nodes: Vec<_> = weighted_shuffle
             .shuffle(&mut rng)
@@ -300,6 +352,7 @@ pub fn new_cluster_nodes<T: 'static>(
     cluster_info: &ClusterInfo,
     cluster_type: ClusterType,
     stakes: &HashMap<Pubkey, u64>,
+    use_cha_cha_8: bool,
 ) -> ClusterNodes<T> {
     let self_pubkey = cluster_info.id();
     let nodes = get_nodes(cluster_info, cluster_type, stakes);
@@ -320,6 +373,7 @@ pub fn new_cluster_nodes<T: 'static>(
         index,
         weighted_shuffle,
         _phantom: PhantomData,
+        use_cha_cha_8,
     }
 }
 
@@ -440,11 +494,6 @@ fn dedup_tvu_addrs(nodes: &mut Vec<Node>) {
     })
 }
 
-fn get_seeded_rng(leader: &Pubkey, shred: &ShredId) -> ChaChaRng {
-    let seed = shred.seed(leader);
-    ChaChaRng::from_seed(seed)
-}
-
 // root     : [0]
 // 1st layer: [1, 2, ..., fanout]
 // 2nd layer: [[fanout + 1, ..., fanout * 2],
@@ -542,6 +591,11 @@ impl<T: 'static> ClusterNodesCache<T> {
             let cache = self.cache.read().unwrap();
             get_epoch_entry(&cache, epoch, self.ttl)
         };
+        let use_cha_cha_8 = check_feature_activation(
+            &feature_set::switch_to_chacha8_turbine::ID,
+            shred_slot,
+            root_bank,
+        );
         // Fall back to exclusive lock if there is a cache miss or the cached
         // entry has already expired.
         let entry: Arc<OnceLock<_>> = entry.unwrap_or_else(|| {
@@ -567,8 +621,12 @@ impl<T: 'static> ClusterNodesCache<T> {
                     inc_new_counter_error!("cluster_nodes-unknown_epoch_staked_nodes", 1);
                     Arc::<HashMap<Pubkey, /*stake:*/ u64>>::default()
                 });
-            let nodes =
-                new_cluster_nodes::<T>(cluster_info, root_bank.cluster_type(), &epoch_staked_nodes);
+            let nodes = new_cluster_nodes::<T>(
+                cluster_info,
+                root_bank.cluster_type(),
+                &epoch_staked_nodes,
+                use_cha_cha_8,
+            );
             (Instant::now(), Arc::new(nodes))
         });
         nodes.clone()
@@ -727,10 +785,105 @@ mod tests {
     use {
         super::*,
         itertools::Itertools,
-        std::{fmt::Debug, hash::Hash},
+        solana_hash::Hash as SolanaHash,
+        solana_ledger::shred::{ProcessShredsStats, ReedSolomonCache, Shredder},
+        std::{collections::VecDeque, fmt::Debug, hash::Hash},
         test_case::test_case,
     };
 
+    #[test_case(true /* chacha8 */)]
+    #[test_case(false /* chacha20 */)]
+    /// Test that we provide a complete coverage
+    /// of all the nodes with weighted shuffles
+    fn test_complete_cluster_coverage(use_cha_cha_8: bool) {
+        let fanout = 10;
+        let mut rng = rand::thread_rng();
+
+        let (_nodes, stakes, cluster_info) = make_test_cluster(&mut rng, 20, Some((0, 1)));
+        let slot_leader = cluster_info.id();
+
+        // create a test cluster
+        let cluster_nodes = new_cluster_nodes::<BroadcastStage>(
+            &cluster_info,
+            ClusterType::Development,
+            &stakes,
+            use_cha_cha_8,
+        );
+
+        let shred = Shredder::new(2, 1, 0, 0)
+            .unwrap()
+            .entries_to_merkle_shreds_for_tests(
+                &Keypair::new(),
+                &[],
+                true,
+                SolanaHash::default(),
+                0,
+                0,
+                &ReedSolomonCache::default(),
+                &mut ProcessShredsStats::default(),
+            )
+            .0
+            .pop()
+            .unwrap();
+
+        let mut weighted_shuffle = cluster_nodes.weighted_shuffle.clone();
+        let mut chacha_rng = TurbineRng::new_seeded(&slot_leader, &shred.id(), use_cha_cha_8);
+
+        let shuffled_nodes: Vec<&Node> = weighted_shuffle
+            .shuffle(&mut chacha_rng)
+            .map(|i| &cluster_nodes.nodes[i])
+            .collect();
+
+        // Slot leader obviously has the shred
+        let mut covered: HashSet<Pubkey> = HashSet::from([slot_leader]);
+        // The root node has the shred sent to it initially
+        let mut queue = VecDeque::from([*shuffled_nodes[0].pubkey()]);
+
+        // traverse the turbine tree using the queue of nodes to visit (BFS)
+        while let Some(addr) = queue.pop_front() {
+            if !covered.insert(addr) {
+                panic!("Should not send to already covered nodes, instead sending to {addr}");
+            }
+            let (_, peers) = get_retransmit_peers(
+                fanout,
+                |n: &Node| n.pubkey() == &addr,
+                shuffled_nodes.clone(),
+            );
+
+            // visit all child nodes
+            for peer in peers {
+                trace!("{} is child of {addr}", peer.pubkey());
+                queue.push_back(*peer.pubkey());
+                if stakes[peer.pubkey()] == 0 {
+                    continue; // no check of retransmit parents for unstaked nodes
+                }
+                // luckily for us, ClusterNodes<RetransmitStage> does not do anything with own identity
+                let mut peer_cluster_nodes = new_cluster_nodes::<RetransmitStage>(
+                    &cluster_info,
+                    ClusterType::Development,
+                    &stakes,
+                    use_cha_cha_8,
+                );
+                peer_cluster_nodes.pubkey = *peer.pubkey();
+                // check that the parent computed by the child matches actual parent.
+                let parent = peer_cluster_nodes
+                    .get_retransmit_parent(&slot_leader, &shred.id(), fanout)
+                    .unwrap();
+
+                assert_eq!(
+                    Some(addr),
+                    parent,
+                    "Found incorrect parent for node {}",
+                    peer_cluster_nodes.pubkey
+                );
+            }
+        }
+
+        // Convert cluster_nodes into hashset of pubkeys
+        let all_nodes: HashSet<_> = cluster_nodes.nodes.iter().map(|n| *n.pubkey()).collect();
+        assert_eq!(all_nodes, covered, "All nodes must be covered");
+    }
+
     #[test]
     fn test_cluster_nodes_retransmit() {
         let mut rng = rand::thread_rng();
@@ -740,8 +893,12 @@ mod tests {
             cluster_info.tvu_peers(GossipContactInfo::clone).len(),
             nodes.len() - 1
         );
-        let cluster_nodes =
-            new_cluster_nodes::<RetransmitStage>(&cluster_info, ClusterType::Development, &stakes);
+        let cluster_nodes = new_cluster_nodes::<RetransmitStage>(
+            &cluster_info,
+            ClusterType::Development,
+            &stakes,
+            false,
+        );
         // All nodes with contact-info should be in the index.
         // Staked nodes with no contact-info should be included.
         assert!(cluster_nodes.nodes.len() > nodes.len());
@@ -770,8 +927,9 @@ mod tests {
         }
     }
 
-    #[test]
-    fn test_cluster_nodes_broadcast() {
+    #[test_case(true)/*ChaCha8 */]
+    #[test_case(false)/*ChaCha20 */]
+    fn test_cluster_nodes_broadcast(use_cha_cha_8: bool) {
         let mut rng = rand::thread_rng();
         let (nodes, stakes, cluster_info) = make_test_cluster(&mut rng, 1_000, None);
         // ClusterInfo::tvu_peers excludes the node itself.
@@ -779,8 +937,12 @@ mod tests {
             cluster_info.tvu_peers(GossipContactInfo::clone).len(),
             nodes.len() - 1
         );
-        let cluster_nodes =
-            ClusterNodes::<BroadcastStage>::new(&cluster_info, ClusterType::Development, &stakes);
+        let cluster_nodes = ClusterNodes::<BroadcastStage>::new(
+            &cluster_info,
+            ClusterType::Development,
+            &stakes,
+            use_cha_cha_8,
+        );
         // All nodes with contact-info should be in the index.
         // Excluding this node itself.
         // Staked nodes with no contact-info should be included.