Add genesis blockhash to blobs (#3953)

core/src/broadcast_stage.rs

@@ -4,7 +4,7 @@ use crate::blocktree::Blocktree;
 use crate::cluster_info::{ClusterInfo, ClusterInfoError, NEIGHBORHOOD_SIZE};
 use crate::entry::{EntrySender, EntrySlice};
 use crate::erasure::CodingGenerator;
-use crate::packet::index_blobs;
+use crate::packet::index_blobs_with_genesis;
 use crate::poh_recorder::WorkingBankEntries;
 use crate::result::{Error, Result};
 use crate::service::Service;
@@ -12,6 +12,7 @@ use crate::staking_utils;
 use rayon::prelude::*;
 use solana_metrics::counter::Counter;
 use solana_metrics::{influxdb, submit};
+use solana_sdk::hash::Hash;
 use solana_sdk::pubkey::Pubkey;
 use solana_sdk::timing::duration_as_ms;
 use std::net::UdpSocket;
@@ -39,6 +40,7 @@ impl Broadcast {
         sock: &UdpSocket,
         blocktree: &Arc<Blocktree>,
         storage_entry_sender: &EntrySender,
+        genesis_blockhash: &Hash,
     ) -> Result<()> {
         let timer = Duration::new(1, 0);
         let (mut bank, entries) = receiver.recv_timeout(timer)?;
@@ -100,9 +102,10 @@ impl Broadcast {
             .map(|meta| meta.consumed)
             .unwrap_or(0);
 
-        index_blobs(
+        index_blobs_with_genesis(
             &blobs,
             &self.id,
+            genesis_blockhash,
             blob_index,
             bank.slot(),
             bank.parent().map_or(0, |parent| parent.slot()),
@@ -184,6 +187,7 @@ impl BroadcastStage {
         receiver: &Receiver<WorkingBankEntries>,
         blocktree: &Arc<Blocktree>,
         storage_entry_sender: EntrySender,
+        genesis_blockhash: &Hash,
     ) -> BroadcastStageReturnType {
         let me = cluster_info.read().unwrap().my_data().clone();
         let coding_generator = CodingGenerator::default();
@@ -200,6 +204,7 @@ impl BroadcastStage {
                 sock,
                 blocktree,
                 &storage_entry_sender,
+                genesis_blockhash,
             ) {
                 match e {
                     Error::RecvTimeoutError(RecvTimeoutError::Disconnected) | Error::SendError => {
@@ -239,9 +244,11 @@ impl BroadcastStage {
         exit_sender: &Arc<AtomicBool>,
         blocktree: &Arc<Blocktree>,
         storage_entry_sender: EntrySender,
+        genesis_blockhash: &Hash,
     ) -> Self {
         let blocktree = blocktree.clone();
         let exit_sender = exit_sender.clone();
+        let genesis_blockhash = *genesis_blockhash;
         let thread_hdl = Builder::new()
             .name("solana-broadcaster".to_string())
             .spawn(move || {
@@ -252,6 +259,7 @@ impl BroadcastStage {
                     &receiver,
                     &blocktree,
                     storage_entry_sender,
+                    &genesis_blockhash,
                 )
             })
             .unwrap();
@@ -325,6 +333,7 @@ mod test {
             &exit_sender,
             &blocktree,
             storage_sender,
+            &Hash::default(),
         );
 
         MockBroadcastStage {

core/src/chacha.rs

@@ -159,7 +159,7 @@ mod tests {
         use bs58;
         //  golden needs to be updated if blob stuff changes....
         let golden = Hash::new(
-            &bs58::decode("GD6xs6Loh9gci6b6P8FVVJ1c1whCqxDzaqBrQkpcxowA")
+            &bs58::decode("5Pz5KQyNht2nqkJhVd8F9zTFxzoDvbQSzaxQbtCPiyCo")
                 .into_vec()
                 .unwrap(),
         );

core/src/fullnode.rs

@@ -10,6 +10,7 @@ use crate::entry::next_entry_mut;
 use crate::entry::Entry;
 use crate::gossip_service::{discover_nodes, GossipService};
 use crate::leader_schedule_cache::LeaderScheduleCache;
+use crate::poh_recorder::PohRecorder;
 use crate::poh_service::{PohService, PohServiceConfig};
 use crate::rpc::JsonRpcConfig;
 use crate::rpc_pubsub_service::PubSubService;
@@ -19,8 +20,6 @@ use crate::service::Service;
 use crate::storage_stage::StorageState;
 use crate::tpu::Tpu;
 use crate::tvu::{Sockets, Tvu};
-
-use crate::poh_recorder::PohRecorder;
 use solana_metrics::counter::Counter;
 use solana_sdk::genesis_block::GenesisBlock;
 use solana_sdk::hash::Hash;
@@ -102,6 +101,7 @@ impl Fullnode {
         let exit = Arc::new(AtomicBool::new(false));
         let bank_info = &bank_forks_info[0];
         let bank = bank_forks[bank_info.bank_slot].clone();
+        let genesis_blockhash = bank.last_blockhash();
 
         info!(
             "starting PoH... {} {}",
@@ -237,6 +237,7 @@ impl Fullnode {
             receiver,
             &leader_schedule_cache,
             &exit,
+            &genesis_blockhash,
         );
 
         if config.sigverify_disabled {
@@ -256,6 +257,7 @@ impl Fullnode {
             sender,
             &leader_schedule_cache,
             &exit,
+            &genesis_blockhash,
         );
 
         inc_new_counter_info!("fullnode-new", 1);

core/src/packet.rs

@@ -5,6 +5,7 @@ use bincode;
 use byteorder::{ByteOrder, LittleEndian};
 use serde::Serialize;
 use solana_metrics::counter::Counter;
+use solana_sdk::hash::Hash;
 pub use solana_sdk::packet::PACKET_DATA_SIZE;
 use solana_sdk::pubkey::Pubkey;
 use std::borrow::Borrow;
@@ -354,7 +355,8 @@ const SLOT_RANGE: std::ops::Range<usize> = range!(PARENT_RANGE.end, u64);
 const INDEX_RANGE: std::ops::Range<usize> = range!(SLOT_RANGE.end, u64);
 const ID_RANGE: std::ops::Range<usize> = range!(INDEX_RANGE.end, Pubkey);
 const FORWARDED_RANGE: std::ops::Range<usize> = range!(ID_RANGE.end, bool);
-const FLAGS_RANGE: std::ops::Range<usize> = range!(FORWARDED_RANGE.end, u32);
+const GENESIS_RANGE: std::ops::Range<usize> = range!(FORWARDED_RANGE.end, Hash);
+const FLAGS_RANGE: std::ops::Range<usize> = range!(GENESIS_RANGE.end, u32);
 const SIZE_RANGE: std::ops::Range<usize> = range!(FLAGS_RANGE.end, u64);
 
 macro_rules! align {
@@ -435,6 +437,14 @@ impl Blob {
         self.data[FORWARDED_RANGE][0] = u8::from(forward)
     }
 
+    pub fn set_genesis_blockhash(&mut self, blockhash: &Hash) {
+        self.data[GENESIS_RANGE].copy_from_slice(blockhash.as_ref())
+    }
+
+    pub fn genesis_blockhash(&self) -> Hash {
+        Hash::new(&self.data[GENESIS_RANGE])
+    }
+
     pub fn flags(&self) -> u32 {
         LittleEndian::read_u32(&self.data[FLAGS_RANGE])
     }
@@ -573,12 +583,24 @@ impl Blob {
     }
 }
 
-pub fn index_blobs(blobs: &[SharedBlob], id: &Pubkey, mut blob_index: u64, slot: u64, parent: u64) {
+pub fn index_blobs(blobs: &[SharedBlob], id: &Pubkey, blob_index: u64, slot: u64, parent: u64) {
+    index_blobs_with_genesis(blobs, id, &Hash::default(), blob_index, slot, parent)
+}
+
+pub fn index_blobs_with_genesis(
+    blobs: &[SharedBlob],
+    id: &Pubkey,
+    genesis: &Hash,
+    mut blob_index: u64,
+    slot: u64,
+    parent: u64,
+) {
     // enumerate all the blobs, those are the indices
     for blob in blobs.iter() {
         let mut blob = blob.write().unwrap();
 
         blob.set_index(blob_index);
+        blob.set_genesis_blockhash(genesis);
         blob.set_slot(slot);
         blob.set_parent(parent);
         blob.set_id(id);
@@ -837,4 +859,14 @@ mod tests {
         assert!(p1 != p2);
     }
 
+    #[test]
+    fn test_blob_genesis_blockhash() {
+        let mut blob = Blob::default();
+        assert_eq!(blob.genesis_blockhash(), Hash::default());
+
+        let hash = Hash::new(&Pubkey::new_rand().as_ref());
+        blob.set_genesis_blockhash(&hash);
+        assert_eq!(blob.genesis_blockhash(), hash);
+    }
+
 }

core/src/replicator.rs

@@ -243,6 +243,7 @@ impl Replicator {
             repair_socket,
             &exit,
             Some(repair_slot_range),
+            &Hash::default(),
         );
 
         let client = create_client(cluster_entrypoint.client_facing_addr(), FULLNODE_PORT_RANGE);

core/src/retransmit_stage.rs

@@ -13,6 +13,7 @@ use crate::streamer::BlobReceiver;
 use crate::window_service::WindowService;
 use solana_metrics::counter::Counter;
 use solana_metrics::{influxdb, submit};
+use solana_sdk::hash::Hash;
 use std::net::UdpSocket;
 use std::sync::atomic::AtomicBool;
 use std::sync::mpsc::channel;
@@ -120,6 +121,7 @@ impl RetransmitStage {
         repair_socket: Arc<UdpSocket>,
         fetch_stage_receiver: BlobReceiver,
         exit: &Arc<AtomicBool>,
+        genesis_blockhash: &Hash,
     ) -> Self {
         let (retransmit_sender, retransmit_receiver) = channel();
 
@@ -140,6 +142,7 @@ impl RetransmitStage {
             repair_socket,
             exit,
             None,
+            genesis_blockhash,
         );
 
         let thread_hdls = vec![t_retransmit];

core/src/tpu.rs

@@ -12,6 +12,7 @@ use crate::leader_schedule_cache::LeaderScheduleCache;
 use crate::poh_recorder::{PohRecorder, WorkingBankEntries};
 use crate::service::Service;
 use crate::sigverify_stage::SigVerifyStage;
+use solana_sdk::hash::Hash;
 use solana_sdk::pubkey::Pubkey;
 use std::net::UdpSocket;
 use std::sync::atomic::AtomicBool;
@@ -42,6 +43,7 @@ impl Tpu {
         storage_entry_sender: EntrySender,
         leader_schedule_cache: &Arc<LeaderScheduleCache>,
         exit: &Arc<AtomicBool>,
+        genesis_blockhash: &Hash,
     ) -> Self {
         cluster_info.write().unwrap().set_leader(id);
 
@@ -82,6 +84,7 @@ impl Tpu {
             &exit,
             blocktree,
             storage_entry_sender,
+            genesis_blockhash,
         );
 
         Self {

core/src/tvu.rs

@@ -26,6 +26,7 @@ use crate::retransmit_stage::RetransmitStage;
 use crate::rpc_subscriptions::RpcSubscriptions;
 use crate::service::Service;
 use crate::storage_stage::{StorageStage, StorageState};
+use solana_sdk::hash::Hash;
 use solana_sdk::pubkey::Pubkey;
 use solana_sdk::signature::{Keypair, KeypairUtil};
 use std::net::UdpSocket;
@@ -74,6 +75,7 @@ impl Tvu {
         storage_entry_receiver: EntryReceiver,
         leader_schedule_cache: &Arc<LeaderScheduleCache>,
         exit: &Arc<AtomicBool>,
+        genesis_blockhash: &Hash,
     ) -> Self
     where
         T: 'static + KeypairUtil + Sync + Send,
@@ -110,6 +112,7 @@ impl Tvu {
             repair_socket,
             blob_fetch_receiver,
             &exit,
+            genesis_blockhash,
         );
 
         let (replay_stage, slot_full_receiver) = ReplayStage::new(
@@ -243,6 +246,7 @@ pub mod tests {
             storage_entry_receiver,
             &leader_schedule_cache,
             &exit,
+            &Hash::default(),
         );
         exit.store(true, Ordering::Relaxed);
         tvu.join().unwrap();

core/src/window_service.rs

@@ -13,6 +13,7 @@ use crate::service::Service;
 use crate::streamer::{BlobReceiver, BlobSender};
 use solana_metrics::counter::Counter;
 use solana_runtime::bank::Bank;
+use solana_sdk::hash::Hash;
 use solana_sdk::pubkey::Pubkey;
 use solana_sdk::timing::duration_as_ms;
 use std::net::UdpSocket;
@@ -107,6 +108,7 @@ fn recv_window(
     my_id: &Pubkey,
     r: &BlobReceiver,
     retransmit: &BlobSender,
+    genesis_blockhash: &Hash,
 ) -> Result<()> {
     let timer = Duration::from_millis(200);
     let mut blobs = r.recv_timeout(timer)?;
@@ -125,7 +127,7 @@ fn recv_window(
                 .as_ref(),
             leader_schedule_cache,
             my_id,
-        )
+        ) && blob.read().unwrap().genesis_blockhash() == *genesis_blockhash
     });
 
     retransmit_blobs(&blobs, retransmit, my_id)?;
@@ -166,6 +168,7 @@ pub struct WindowService {
 }
 
 impl WindowService {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         bank_forks: Option<Arc<RwLock<BankForks>>>,
         leader_schedule_cache: Option<Arc<LeaderScheduleCache>>,
@@ -176,6 +179,7 @@ impl WindowService {
         repair_socket: Arc<UdpSocket>,
         exit: &Arc<AtomicBool>,
         repair_slot_range: Option<RepairSlotRange>,
+        genesis_blockhash: &Hash,
     ) -> WindowService {
         let repair_service = RepairService::new(
             blocktree.clone(),
@@ -187,6 +191,7 @@ impl WindowService {
         let exit = exit.clone();
         let bank_forks = bank_forks.clone();
         let leader_schedule_cache = leader_schedule_cache.clone();
+        let hash = *genesis_blockhash;
         let t_window = Builder::new()
             .name("solana-window".to_string())
             .spawn(move || {
@@ -204,6 +209,7 @@ impl WindowService {
                         &id,
                         &r,
                         &retransmit,
+                        &hash,
                     ) {
                         match e {
                             Error::RecvTimeoutError(RecvTimeoutError::Disconnected) => break,
@@ -357,6 +363,7 @@ mod test {
             Arc::new(leader_node.sockets.repair),
             &exit,
             None,
+            &Hash::default(),
         );
         let t_responder = {
             let (s_responder, r_responder) = channel();
@@ -434,6 +441,7 @@ mod test {
             Arc::new(leader_node.sockets.repair),
             &exit,
             None,
+            &Hash::default(),
         );
         let t_responder = {
             let (s_responder, r_responder) = channel();

core/tests/tvu.rs

@@ -129,6 +129,7 @@ fn test_replay() {
             storage_receiver,
             &leader_schedule_cache,
             &exit,
+            &solana_sdk::hash::Hash::default(),
         );
 
         let mut mint_ref_balance = starting_mint_balance;