shredder: allow creating shreds with raw bytes (#7172)

allow creating shreds with raw bytes

Author: alexpyattaev

core/src/replay_stage.rs

@@ -4341,7 +4341,7 @@ pub(crate) mod tests {
             create_new_tmp_ledger,
             genesis_utils::{create_genesis_config, create_genesis_config_with_leader},
             get_tmp_ledger_path, get_tmp_ledger_path_auto_delete,
-            shred::{Shred, ShredFlags, LEGACY_SHRED_DATA_CAPACITY},
+            shred::{ProcessShredsStats, ReedSolomonCache, Shred, Shredder},
         },
         solana_poh_config::PohConfig,
         solana_rpc::{
@@ -4973,19 +4973,25 @@ pub(crate) mod tests {
     fn test_dead_fork_entry_deserialize_failure() {
         // Insert entry that causes deserialization failure
         let res = check_dead_fork(|_, bank| {
-            let gibberish = [0xa5u8; LEGACY_SHRED_DATA_CAPACITY];
-            let parent_offset = bank.slot() - bank.parent_slot();
-            let shred = Shred::new_from_data(
-                bank.slot(),
-                0, // index,
-                parent_offset as u16,
-                &gibberish,
-                ShredFlags::DATA_COMPLETE_SHRED,
-                0, // reference_tick
-                0, // version
-                0, // fec_set_index
-            );
-            vec![shred]
+            let gibberish = [0xa5u8; 1024];
+
+            let shredder = Shredder::new(bank.slot(), bank.parent_slot(), 0, 0).unwrap();
+            let keypair = Keypair::new();
+            let reed_solomon_cache = ReedSolomonCache::default();
+
+            shredder
+                .make_shreds_from_data_slice(
+                    &keypair,
+                    &gibberish,
+                    true,
+                    Some(Hash::default()),
+                    0,
+                    0,
+                    &reed_solomon_cache,
+                    &mut ProcessShredsStats::default(),
+                )
+                .unwrap()
+                .collect()
         });
 
         assert_matches!(

ledger/src/shred.rs

@@ -71,7 +71,6 @@ use {
     assert_matches::debug_assert_matches,
     bitflags::bitflags,
     num_enum::{IntoPrimitive, TryFromPrimitive},
-    rayon::ThreadPool,
     serde::{Deserialize, Serialize},
     solana_clock::Slot,
     solana_entry::entry::{create_ticks, Entry},
@@ -83,13 +82,13 @@ use {
     solana_signature::{Signature, SIGNATURE_BYTES},
     solana_signer::Signer,
     static_assertions::const_assert_eq,
-    std::{fmt::Debug, time::Instant},
+    std::fmt::Debug,
     thiserror::Error,
 };
 
 mod common;
 mod legacy;
-mod merkle;
+pub(crate) mod merkle;
 mod merkle_tree;
 mod payload;
 pub mod shred_code;
@@ -835,43 +834,6 @@ pub fn recover(
     Ok(shreds.map(|shred| shred.map(Shred::from)))
 }
 
-#[allow(clippy::too_many_arguments)]
-pub(crate) fn make_merkle_shreds_from_entries(
-    thread_pool: &ThreadPool,
-    keypair: &Keypair,
-    entries: &[Entry],
-    slot: Slot,
-    parent_slot: Slot,
-    shred_version: u16,
-    reference_tick: u8,
-    is_last_in_slot: bool,
-    chained_merkle_root: Option<Hash>,
-    next_shred_index: u32,
-    next_code_index: u32,
-    reed_solomon_cache: &ReedSolomonCache,
-    stats: &mut ProcessShredsStats,
-) -> Result<impl Iterator<Item = Shred>, Error> {
-    let now = Instant::now();
-    let entries = bincode::serialize(entries)?;
-    stats.serialize_elapsed += now.elapsed().as_micros() as u64;
-    let shreds = merkle::make_shreds_from_data(
-        thread_pool,
-        keypair,
-        chained_merkle_root,
-        &entries[..],
-        slot,
-        parent_slot,
-        shred_version,
-        reference_tick,
-        is_last_in_slot,
-        next_shred_index,
-        next_code_index,
-        reed_solomon_cache,
-        stats,
-    )?;
-    Ok(shreds.into_iter().map(Shred::from))
-}
-
 // Accepts shreds in the slot range [root + 1, max_slot].
 #[must_use]
 pub fn should_discard_shred<'a, P>(

ledger/src/shred/merkle.rs

@@ -70,7 +70,7 @@ pub struct ShredCode {
 }
 
 #[derive(Clone, Debug, Eq, PartialEq)]
-pub(super) enum Shred {
+pub(crate) enum Shred {
     ShredCode(ShredCode),
     ShredData(ShredData),
 }
@@ -1019,7 +1019,7 @@ fn make_shreds_code_header_only(
 }
 
 #[allow(clippy::too_many_arguments)]
-pub(super) fn make_shreds_from_data(
+pub(crate) fn make_shreds_from_data(
     thread_pool: &ThreadPool,
     keypair: &Keypair,
     // The Merkle root of the previous erasure batch if chained.

ledger/src/shredder.rs

@@ -19,6 +19,7 @@ use {
         borrow::Borrow,
         fmt::Debug,
         sync::{Arc, OnceLock, RwLock},
+        time::Instant,
     },
 };
 
@@ -89,22 +90,52 @@ impl Shredder {
         reed_solomon_cache: &ReedSolomonCache,
         stats: &mut ProcessShredsStats,
     ) -> impl Iterator<Item = Shred> {
-        shred::make_merkle_shreds_from_entries(
-            &PAR_THREAD_POOL,
+        let now = Instant::now();
+        let entries = bincode::serialize(entries).unwrap();
+        stats.serialize_elapsed += now.elapsed().as_micros() as u64;
+        Self::make_shreds_from_data_slice(
+            self,
             keypair,
-            entries,
+            &entries,
+            is_last_in_slot,
+            chained_merkle_root,
+            next_shred_index,
+            next_code_index,
+            reed_solomon_cache,
+            stats,
+        )
+        .unwrap()
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    pub fn make_shreds_from_data_slice(
+        &self,
+        keypair: &Keypair,
+        data: &[u8],
+        is_last_in_slot: bool,
+        chained_merkle_root: Option<Hash>,
+        next_shred_index: u32,
+        next_code_index: u32,
+        reed_solomon_cache: &ReedSolomonCache,
+        stats: &mut ProcessShredsStats,
+    ) -> Result<impl Iterator<Item = Shred>, Error> {
+        let thread_pool: &ThreadPool = &PAR_THREAD_POOL;
+        let shreds = shred::merkle::make_shreds_from_data(
+            thread_pool,
+            keypair,
+            chained_merkle_root,
+            data,
             self.slot,
             self.parent_slot,
             self.version,
             self.reference_tick,
             is_last_in_slot,
-            chained_merkle_root,
             next_shred_index,
             next_code_index,
             reed_solomon_cache,
             stats,
-        )
-        .unwrap()
+        )?;
+        Ok(shreds.into_iter().map(Shred::from))
     }
 
     pub fn entries_to_merkle_shreds_for_tests(