diff --git a/core/src/broadcast_stage.rs b/core/src/broadcast_stage.rs index 2b87a7648b..a704479c0d 100644 --- a/core/src/broadcast_stage.rs +++ b/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, 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, blocktree: &Arc, 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, blocktree: &Arc, 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 { diff --git a/core/src/chacha.rs b/core/src/chacha.rs index 5c143e271c..c8539e867e 100644 --- a/core/src/chacha.rs +++ b/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(), ); diff --git a/core/src/fullnode.rs b/core/src/fullnode.rs index 2416d52d47..f7321bbfc6 100644 --- a/core/src/fullnode.rs +++ b/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); diff --git a/core/src/packet.rs b/core/src/packet.rs index d0bedc95d0..73da0c9c89 100644 --- a/core/src/packet.rs +++ b/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 = range!(PARENT_RANGE.end, u64); const INDEX_RANGE: std::ops::Range = range!(SLOT_RANGE.end, u64); const ID_RANGE: std::ops::Range = range!(INDEX_RANGE.end, Pubkey); const FORWARDED_RANGE: std::ops::Range = range!(ID_RANGE.end, bool); -const FLAGS_RANGE: std::ops::Range = range!(FORWARDED_RANGE.end, u32); +const GENESIS_RANGE: std::ops::Range = range!(FORWARDED_RANGE.end, Hash); +const FLAGS_RANGE: std::ops::Range = range!(GENESIS_RANGE.end, u32); const SIZE_RANGE: std::ops::Range = 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); + } + } diff --git a/core/src/replicator.rs b/core/src/replicator.rs index 91ef9a9323..f6985edeb6 100644 --- a/core/src/replicator.rs +++ b/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); diff --git a/core/src/retransmit_stage.rs b/core/src/retransmit_stage.rs index 493f989bf6..3c6686ac21 100644 --- a/core/src/retransmit_stage.rs +++ b/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, fetch_stage_receiver: BlobReceiver, exit: &Arc, + 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]; diff --git a/core/src/tpu.rs b/core/src/tpu.rs index 5edfd4483f..cb729b4fef 100644 --- a/core/src/tpu.rs +++ b/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, exit: &Arc, + 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 { diff --git a/core/src/tvu.rs b/core/src/tvu.rs index 186b41dbe6..91033d2cbf 100644 --- a/core/src/tvu.rs +++ b/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, exit: &Arc, + 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(); diff --git a/core/src/window_service.rs b/core/src/window_service.rs index 4e22f590ff..643a97461f 100644 --- a/core/src/window_service.rs +++ b/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>>, leader_schedule_cache: Option>, @@ -176,6 +179,7 @@ impl WindowService { repair_socket: Arc, exit: &Arc, repair_slot_range: Option, + 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(); diff --git a/core/tests/tvu.rs b/core/tests/tvu.rs index 0c85f4a59b..bd5f36b4cb 100644 --- a/core/tests/tvu.rs +++ b/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;