cerc-io/lighthouse
16
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

wip

Browse Source
This commit is contained in:
Diva M 2022-11-28 14:13:12 -05:00
parent f4babdedd5
commit 805df307f6
No known key found for this signature in database
GPG Key ID: 1BAE5E01126680FE
7 changed files with 157 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
beacon_node/network/src/router/processor.rs
View File

@ -244,7 +244,7 @@ impl<T: BeaconChainTypes> Processor<T> {
); );
if let RequestId::Sync(id) = request_id { if let RequestId::Sync(id) = request_id {
self.send_to_sync(SyncMessage::RpcBlob { self.send_to_sync(SyncMessage::RpcGlob {
peer_id, peer_id,
request_id: id, request_id: id,
blob_sidecar, blob_sidecar,
@ -316,7 +316,7 @@ impl<T: BeaconChainTypes> Processor<T> {
"Received BlockAndBlobssByRoot Response"; "Received BlockAndBlobssByRoot Response";
"peer" => %peer_id, "peer" => %peer_id,
); );
self.send_to_sync(SyncMessage::RpcBlockAndBlob { self.send_to_sync(SyncMessage::RpcBlockAndGlob {
peer_id, peer_id,
request_id, request_id,
block_and_blobs, block_and_blobs,

37
beacon_node/network/src/sync/block_lookups/mod.rs
View File

@ -18,7 +18,7 @@ use self::{
single_block_lookup::SingleBlockRequest, single_block_lookup::SingleBlockRequest,
}; };
use super::manager::BlockProcessResult; use super::manager::{BlockProcessResult, BlockTy};
use super::BatchProcessResult; use super::BatchProcessResult;
use super::{ use super::{
manager::{BlockProcessType, Id}, manager::{BlockProcessType, Id},
@ -30,7 +30,7 @@ mod single_block_lookup;
#[cfg(test)] #[cfg(test)]
mod tests; mod tests;
pub type RootBlockTuple<T> = (Hash256, Arc<SignedBeaconBlock<T>>); pub type RootBlockTuple<T> = (Hash256, BlockTy<T>);
const FAILED_CHAINS_CACHE_EXPIRY_SECONDS: u64 = 60; const FAILED_CHAINS_CACHE_EXPIRY_SECONDS: u64 = 60;
const SINGLE_BLOCK_LOOKUP_MAX_ATTEMPTS: u8 = 3; const SINGLE_BLOCK_LOOKUP_MAX_ATTEMPTS: u8 = 3;
@ -87,8 +87,9 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
let mut single_block_request = SingleBlockRequest::new(hash, peer_id); let mut single_block_request = SingleBlockRequest::new(hash, peer_id);
//FIXME(sean) remove unwrap? let (peer_id, request) = single_block_request
let (peer_id, request) = single_block_request.request_block().unwrap(); .request_block()
.expect("none of the possible failure cases apply for a newly created block lookup");
if let Ok(request_id) = cx.single_block_lookup_request(peer_id, request) { if let Ok(request_id) = cx.single_block_lookup_request(peer_id, request) {
self.single_block_lookups self.single_block_lookups
.insert(request_id, single_block_request); .insert(request_id, single_block_request);
@ -105,7 +106,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
pub fn search_parent( pub fn search_parent(
&mut self, &mut self,
block_root: Hash256, block_root: Hash256,
block: Arc<SignedBeaconBlock<T::EthSpec>>, block: BlockTy<T::EthSpec>,
peer_id: PeerId, peer_id: PeerId,
cx: &mut SyncNetworkContext<T>, cx: &mut SyncNetworkContext<T>,
) { ) {
@ -120,7 +121,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
// Make sure this block is not already downloaded, and that neither it or its parent is // Make sure this block is not already downloaded, and that neither it or its parent is
// being searched for. // being searched for.
if self.parent_queue.iter_mut().any(|parent_req| { if self.parent_queue.iter_mut().any(|parent_req| {
parent_req.contains_block(&block) parent_req.contains_block(block.block())
|| parent_req.add_peer(&block_root, &peer_id) || parent_req.add_peer(&block_root, &peer_id)
|| parent_req.add_peer(&parent_root, &peer_id) || parent_req.add_peer(&parent_root, &peer_id)
}) { }) {
@ -138,7 +139,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
&mut self, &mut self,
id: Id, id: Id,
peer_id: PeerId, peer_id: PeerId,
block: Option<Arc<SignedBeaconBlock<T::EthSpec>>>, block: Option<BlockTy<T::EthSpec>>,
seen_timestamp: Duration, seen_timestamp: Duration,
cx: &mut SyncNetworkContext<T>, cx: &mut SyncNetworkContext<T>,
) { ) {
@ -203,7 +204,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
&mut self, &mut self,
id: Id, id: Id,
peer_id: PeerId, peer_id: PeerId,
block: Option<Arc<SignedBeaconBlock<T::EthSpec>>>, block: Option<BlockTy<T::EthSpec>>,
seen_timestamp: Duration, seen_timestamp: Duration,
cx: &mut SyncNetworkContext<T>, cx: &mut SyncNetworkContext<T>,
) { ) {
@ -425,7 +426,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
error!(self.log, "Beacon chain error processing single block"; "block_root" => %root, "error" => ?e); error!(self.log, "Beacon chain error processing single block"; "block_root" => %root, "error" => ?e);
} }
BlockError::ParentUnknown(block) => { BlockError::ParentUnknown(block) => {
self.search_parent(root, block, peer_id, cx); self.search_parent(root, BlockTy::Block { block }, peer_id, cx);
} }
ref e @ BlockError::ExecutionPayloadError(ref epe) if !epe.penalize_peer() => { ref e @ BlockError::ExecutionPayloadError(ref epe) if !epe.penalize_peer() => {
// These errors indicate that the execution layer is offline // These errors indicate that the execution layer is offline
@ -505,7 +506,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
BlockProcessResult::Err(BlockError::ParentUnknown(block)) => { BlockProcessResult::Err(BlockError::ParentUnknown(block)) => {
// need to keep looking for parents // need to keep looking for parents
// add the block back to the queue and continue the search // add the block back to the queue and continue the search
parent_lookup.add_block(block); parent_lookup.add_block(BlockTy::Block { block });
self.request_parent(parent_lookup, cx); self.request_parent(parent_lookup, cx);
} }
BlockProcessResult::Ok BlockProcessResult::Ok
@ -524,8 +525,10 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
let chain_hash = parent_lookup.chain_hash(); let chain_hash = parent_lookup.chain_hash();
let blocks = parent_lookup.chain_blocks(); let blocks = parent_lookup.chain_blocks();
let process_id = ChainSegmentProcessId::ParentLookup(chain_hash); let process_id = ChainSegmentProcessId::ParentLookup(chain_hash);
// let work = WorkEvent::chain_segment(process_id, blocks);
let work = todo!("this means we can have batches of mixed type");
match beacon_processor_send.try_send(WorkEvent::chain_segment(process_id, blocks)) { match beacon_processor_send.try_send(work) {
Ok(_) => { Ok(_) => {
self.parent_queue.push(parent_lookup); self.parent_queue.push(parent_lookup);
} }
@ -631,7 +634,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
fn send_block_for_processing( fn send_block_for_processing(
&mut self, &mut self,
block_root: Hash256, block_root: Hash256,
block: Arc<SignedBeaconBlock<T::EthSpec>>, block: BlockTy<T::EthSpec>,
duration: Duration, duration: Duration,
process_type: BlockProcessType, process_type: BlockProcessType,
cx: &mut SyncNetworkContext<T>, cx: &mut SyncNetworkContext<T>,
@ -639,7 +642,15 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
match cx.processor_channel_if_enabled() { match cx.processor_channel_if_enabled() {
Some(beacon_processor_send) => { Some(beacon_processor_send) => {
trace!(self.log, "Sending block for processing"; "block" => ?block_root, "process" => ?process_type); trace!(self.log, "Sending block for processing"; "block" => ?block_root, "process" => ?process_type);
let event = WorkEvent::rpc_beacon_block(block_root, block, duration, process_type); let event = match block {
BlockTy::Block { block } => {
WorkEvent::rpc_beacon_block(block_root, block, duration, process_type)
}
BlockTy::BlockAndBlob { block_sidecar_pair } => {
// WorkEvent::rpc_block_and_glob(block_sidecar_pair)
todo!("we also need to process block-glob pairs for rpc")
}
};
if let Err(e) = beacon_processor_send.try_send(event) { if let Err(e) = beacon_processor_send.try_send(event) {
error!( error!(
self.log, self.log,

18
beacon_node/network/src/sync/block_lookups/parent_lookup.rs
View File

@ -6,7 +6,7 @@ use store::{Hash256, SignedBeaconBlock};
use strum::IntoStaticStr; use strum::IntoStaticStr;
use crate::sync::{ use crate::sync::{
manager::{Id, SLOT_IMPORT_TOLERANCE}, manager::{BlockTy, Id, SLOT_IMPORT_TOLERANCE},
network_context::SyncNetworkContext, network_context::SyncNetworkContext,
}; };
@ -24,7 +24,7 @@ pub(crate) struct ParentLookup<T: BeaconChainTypes> {
/// The root of the block triggering this parent request. /// The root of the block triggering this parent request.
chain_hash: Hash256, chain_hash: Hash256,
/// The blocks that have currently been downloaded. /// The blocks that have currently been downloaded.
downloaded_blocks: Vec<Arc<SignedBeaconBlock<T::EthSpec>>>, downloaded_blocks: Vec<BlockTy<T::EthSpec>>,
/// Request of the last parent. /// Request of the last parent.
current_parent_request: SingleBlockRequest<PARENT_FAIL_TOLERANCE>, current_parent_request: SingleBlockRequest<PARENT_FAIL_TOLERANCE>,
/// Id of the last parent request. /// Id of the last parent request.
@ -56,14 +56,10 @@ impl<T: BeaconChainTypes> ParentLookup<T> {
pub fn contains_block(&self, block: &SignedBeaconBlock<T::EthSpec>) -> bool { pub fn contains_block(&self, block: &SignedBeaconBlock<T::EthSpec>) -> bool {
self.downloaded_blocks self.downloaded_blocks
.iter() .iter()
.any(|d_block| d_block.as_ref() == block) .any(|d_block| d_block.block() == block)
} }
pub fn new( pub fn new(block_root: Hash256, block: BlockTy<T::EthSpec>, peer_id: PeerId) -> Self {
block_root: Hash256,
block: Arc<SignedBeaconBlock<T::EthSpec>>,
peer_id: PeerId,
) -> Self {
let current_parent_request = SingleBlockRequest::new(block.parent_root(), peer_id); let current_parent_request = SingleBlockRequest::new(block.parent_root(), peer_id);
Self { Self {
@ -98,7 +94,7 @@ impl<T: BeaconChainTypes> ParentLookup<T> {
self.current_parent_request.check_peer_disconnected(peer_id) self.current_parent_request.check_peer_disconnected(peer_id)
} }
pub fn add_block(&mut self, block: Arc<SignedBeaconBlock<T::EthSpec>>) { pub fn add_block(&mut self, block: BlockTy<T::EthSpec>) {
let next_parent = block.parent_root(); let next_parent = block.parent_root();
self.downloaded_blocks.push(block); self.downloaded_blocks.push(block);
self.current_parent_request.hash = next_parent; self.current_parent_request.hash = next_parent;
@ -125,7 +121,7 @@ impl<T: BeaconChainTypes> ParentLookup<T> {
self.current_parent_request_id = None; self.current_parent_request_id = None;
} }
pub fn chain_blocks(&mut self) -> Vec<Arc<SignedBeaconBlock<T::EthSpec>>> { pub fn chain_blocks(&mut self) -> Vec<BlockTy<T::EthSpec>> {
std::mem::take(&mut self.downloaded_blocks) std::mem::take(&mut self.downloaded_blocks)
} }
@ -133,7 +129,7 @@ impl<T: BeaconChainTypes> ParentLookup<T> {
/// the processing result of the block. /// the processing result of the block.
pub fn verify_block( pub fn verify_block(
&mut self, &mut self,
block: Option<Arc<SignedBeaconBlock<T::EthSpec>>>, block: Option<BlockTy<T::EthSpec>>,
failed_chains: &mut lru_cache::LRUTimeCache<Hash256>, failed_chains: &mut lru_cache::LRUTimeCache<Hash256>,
) -> Result<Option<RootBlockTuple<T::EthSpec>>, VerifyError> { ) -> Result<Option<RootBlockTuple<T::EthSpec>>, VerifyError> {
let root_and_block = self.current_parent_request.verify_block(block)?; let root_and_block = self.current_parent_request.verify_block(block)?;

12
beacon_node/network/src/sync/block_lookups/single_block_lookup.rs
View File

@ -1,6 +1,8 @@
use std::collections::HashSet; use std::collections::HashSet;
use std::sync::Arc; use std::sync::Arc;
use crate::sync::manager::BlockTy;
use super::RootBlockTuple; use super::RootBlockTuple;
use beacon_chain::get_block_root; use beacon_chain::get_block_root;
use lighthouse_network::{rpc::BlocksByRootRequest, PeerId}; use lighthouse_network::{rpc::BlocksByRootRequest, PeerId};
@ -105,8 +107,8 @@ impl<const MAX_ATTEMPTS: u8> SingleBlockRequest<MAX_ATTEMPTS> {
/// Returns the block for processing if the response is what we expected. /// Returns the block for processing if the response is what we expected.
pub fn verify_block<T: EthSpec>( pub fn verify_block<T: EthSpec>(
&mut self, &mut self,
block: Option<Arc<SignedBeaconBlock<T>>>, block: Option<BlockTy<T>>,
) -> Result<Option<RootBlockTuple<T>>, VerifyError> { ) -> Result<Option<(RootBlockTuple<T>)>, VerifyError> {
match self.state { match self.state {
State::AwaitingDownload => { State::AwaitingDownload => {
self.register_failure_downloading(); self.register_failure_downloading();
@ -116,7 +118,7 @@ impl<const MAX_ATTEMPTS: u8> SingleBlockRequest<MAX_ATTEMPTS> {
Some(block) => { Some(block) => {
// Compute the block root using this specific function so that we can get timing // Compute the block root using this specific function so that we can get timing
// metrics. // metrics.
let block_root = get_block_root(&block); let block_root = get_block_root(block.block());
if block_root != self.hash { if block_root != self.hash {
// return an error and drop the block // return an error and drop the block
// NOTE: we take this is as a download failure to prevent counting the // NOTE: we take this is as a download failure to prevent counting the
@ -225,7 +227,7 @@ mod tests {
let mut sl = SingleBlockRequest::<4>::new(block.canonical_root(), peer_id); let mut sl = SingleBlockRequest::<4>::new(block.canonical_root(), peer_id);
sl.request_block().unwrap(); sl.request_block().unwrap();
sl.verify_block(Some(Arc::new(block))).unwrap().unwrap(); sl.verify_block(Some(block.into())).unwrap().unwrap();
} }
#[test] #[test]
@ -242,7 +244,7 @@ mod tests {
// Now we receive the block and send it for processing // Now we receive the block and send it for processing
sl.request_block().unwrap(); sl.request_block().unwrap();
sl.verify_block(Some(Arc::new(block))).unwrap().unwrap(); sl.verify_block(Some(block.into())).unwrap().unwrap();
// One processing failure maxes the available attempts // One processing failure maxes the available attempts
sl.register_failure_processing(); sl.register_failure_processing();

56
beacon_node/network/src/sync/block_lookups/tests.rs
View File

@ -157,7 +157,7 @@ fn test_single_block_lookup_happy_path() {
// The peer provides the correct block, should not be penalized. Now the block should be sent // The peer provides the correct block, should not be penalized. Now the block should be sent
// for processing. // for processing.
bl.single_block_lookup_response(id, peer_id, Some(Arc::new(block)), D, &mut cx); bl.single_block_lookup_response(id, peer_id, Some(block.into()), D, &mut cx);
rig.expect_empty_network(); rig.expect_empty_network();
rig.expect_block_process(); rig.expect_block_process();
@ -203,7 +203,7 @@ fn test_single_block_lookup_wrong_response() {
// Peer sends something else. It should be penalized. // Peer sends something else. It should be penalized.
let bad_block = rig.rand_block(); let bad_block = rig.rand_block();
bl.single_block_lookup_response(id, peer_id, Some(Arc::new(bad_block)), D, &mut cx); bl.single_block_lookup_response(id, peer_id, Some(bad_block.into()), D, &mut cx);
rig.expect_penalty(); rig.expect_penalty();
rig.expect_block_request(); // should be retried rig.expect_block_request(); // should be retried
@ -242,7 +242,7 @@ fn test_single_block_lookup_becomes_parent_request() {
// The peer provides the correct block, should not be penalized. Now the block should be sent // The peer provides the correct block, should not be penalized. Now the block should be sent
// for processing. // for processing.
bl.single_block_lookup_response(id, peer_id, Some(Arc::new(block.clone())), D, &mut cx); bl.single_block_lookup_response(id, peer_id, Some(block.clone().into()), D, &mut cx);
rig.expect_empty_network(); rig.expect_empty_network();
rig.expect_block_process(); rig.expect_block_process();
@ -251,11 +251,7 @@ fn test_single_block_lookup_becomes_parent_request() {
// Send the stream termination. Peer should have not been penalized, and the request moved to a // Send the stream termination. Peer should have not been penalized, and the request moved to a
// parent request after processing. // parent request after processing.
bl.single_block_processed( bl.single_block_processed(id, BlockError::ParentUnknown(block.into()).into(), &mut cx);
id,
BlockError::ParentUnknown(Arc::new(block)).into(),
&mut cx,
);
assert_eq!(bl.single_block_lookups.len(), 0); assert_eq!(bl.single_block_lookups.len(), 0);
rig.expect_parent_request(); rig.expect_parent_request();
rig.expect_empty_network(); rig.expect_empty_network();
@ -272,11 +268,11 @@ fn test_parent_lookup_happy_path() {
let peer_id = PeerId::random(); let peer_id = PeerId::random();
// Trigger the request // Trigger the request
bl.search_parent(chain_hash, Arc::new(block), peer_id, &mut cx); bl.search_parent(chain_hash, block.into(), peer_id, &mut cx);
let id = rig.expect_parent_request(); let id = rig.expect_parent_request();
// Peer sends the right block, it should be sent for processing. Peer should not be penalized. // Peer sends the right block, it should be sent for processing. Peer should not be penalized.
bl.parent_lookup_response(id, peer_id, Some(Arc::new(parent)), D, &mut cx); bl.parent_lookup_response(id, peer_id, Some(parent.into()), D, &mut cx);
rig.expect_block_process(); rig.expect_block_process();
rig.expect_empty_network(); rig.expect_empty_network();
@ -300,12 +296,12 @@ fn test_parent_lookup_wrong_response() {
let peer_id = PeerId::random(); let peer_id = PeerId::random();
// Trigger the request // Trigger the request
bl.search_parent(chain_hash, Arc::new(block), peer_id, &mut cx); bl.search_parent(chain_hash, block.into(), peer_id, &mut cx);
let id1 = rig.expect_parent_request(); let id1 = rig.expect_parent_request();
// Peer sends the wrong block, peer should be penalized and the block re-requested. // Peer sends the wrong block, peer should be penalized and the block re-requested.
let bad_block = rig.rand_block(); let bad_block = rig.rand_block();
bl.parent_lookup_response(id1, peer_id, Some(Arc::new(bad_block)), D, &mut cx); bl.parent_lookup_response(id1, peer_id, Some(bad_block.into()), D, &mut cx);
rig.expect_penalty(); rig.expect_penalty();
let id2 = rig.expect_parent_request(); let id2 = rig.expect_parent_request();
@ -314,7 +310,7 @@ fn test_parent_lookup_wrong_response() {
rig.expect_empty_network(); rig.expect_empty_network();
// Send the right block this time. // Send the right block this time.
bl.parent_lookup_response(id2, peer_id, Some(Arc::new(parent)), D, &mut cx); bl.parent_lookup_response(id2, peer_id, Some(parent.into()), D, &mut cx);
rig.expect_block_process(); rig.expect_block_process();
// Processing succeeds, now the rest of the chain should be sent for processing. // Processing succeeds, now the rest of the chain should be sent for processing.
@ -337,7 +333,7 @@ fn test_parent_lookup_empty_response() {
let peer_id = PeerId::random(); let peer_id = PeerId::random();
// Trigger the request // Trigger the request
bl.search_parent(chain_hash, Arc::new(block), peer_id, &mut cx); bl.search_parent(chain_hash, block.into(), peer_id, &mut cx);
let id1 = rig.expect_parent_request(); let id1 = rig.expect_parent_request();
// Peer sends an empty response, peer should be penalized and the block re-requested. // Peer sends an empty response, peer should be penalized and the block re-requested.
@ -346,7 +342,7 @@ fn test_parent_lookup_empty_response() {
let id2 = rig.expect_parent_request(); let id2 = rig.expect_parent_request();
// Send the right block this time. // Send the right block this time.
bl.parent_lookup_response(id2, peer_id, Some(Arc::new(parent)), D, &mut cx); bl.parent_lookup_response(id2, peer_id, Some(parent.into()), D, &mut cx);
rig.expect_block_process(); rig.expect_block_process();
// Processing succeeds, now the rest of the chain should be sent for processing. // Processing succeeds, now the rest of the chain should be sent for processing.
@ -369,7 +365,7 @@ fn test_parent_lookup_rpc_failure() {
let peer_id = PeerId::random(); let peer_id = PeerId::random();
// Trigger the request // Trigger the request
bl.search_parent(chain_hash, Arc::new(block), peer_id, &mut cx); bl.search_parent(chain_hash, block.into(), peer_id, &mut cx);
let id1 = rig.expect_parent_request(); let id1 = rig.expect_parent_request();
// The request fails. It should be tried again. // The request fails. It should be tried again.
@ -377,7 +373,7 @@ fn test_parent_lookup_rpc_failure() {
let id2 = rig.expect_parent_request(); let id2 = rig.expect_parent_request();
// Send the right block this time. // Send the right block this time.
bl.parent_lookup_response(id2, peer_id, Some(Arc::new(parent)), D, &mut cx); bl.parent_lookup_response(id2, peer_id, Some(parent.into()), D, &mut cx);
rig.expect_block_process(); rig.expect_block_process();
// Processing succeeds, now the rest of the chain should be sent for processing. // Processing succeeds, now the rest of the chain should be sent for processing.
@ -400,7 +396,7 @@ fn test_parent_lookup_too_many_attempts() {
let peer_id = PeerId::random(); let peer_id = PeerId::random();
// Trigger the request // Trigger the request
bl.search_parent(chain_hash, Arc::new(block), peer_id, &mut cx); bl.search_parent(chain_hash, block.into(), peer_id, &mut cx);
for i in 1..=parent_lookup::PARENT_FAIL_TOLERANCE { for i in 1..=parent_lookup::PARENT_FAIL_TOLERANCE {
let id = rig.expect_parent_request(); let id = rig.expect_parent_request();
match i % 2 { match i % 2 {
@ -412,7 +408,7 @@ fn test_parent_lookup_too_many_attempts() {
_ => { _ => {
// Send a bad block this time. It should be tried again. // Send a bad block this time. It should be tried again.
let bad_block = rig.rand_block(); let bad_block = rig.rand_block();
bl.parent_lookup_response(id, peer_id, Some(Arc::new(bad_block)), D, &mut cx); bl.parent_lookup_response(id, peer_id, Some(bad_block.into()), D, &mut cx);
// Send the stream termination // Send the stream termination
bl.parent_lookup_response(id, peer_id, None, D, &mut cx); bl.parent_lookup_response(id, peer_id, None, D, &mut cx);
rig.expect_penalty(); rig.expect_penalty();
@ -436,7 +432,7 @@ fn test_parent_lookup_too_many_download_attempts_no_blacklist() {
let peer_id = PeerId::random(); let peer_id = PeerId::random();
// Trigger the request // Trigger the request
bl.search_parent(block_hash, Arc::new(block), peer_id, &mut cx); bl.search_parent(block_hash, block.into(), peer_id, &mut cx);
for i in 1..=parent_lookup::PARENT_FAIL_TOLERANCE { for i in 1..=parent_lookup::PARENT_FAIL_TOLERANCE {
assert!(!bl.failed_chains.contains(&block_hash)); assert!(!bl.failed_chains.contains(&block_hash));
let id = rig.expect_parent_request(); let id = rig.expect_parent_request();
@ -446,7 +442,7 @@ fn test_parent_lookup_too_many_download_attempts_no_blacklist() {
} else { } else {
// Send a bad block this time. It should be tried again. // Send a bad block this time. It should be tried again.
let bad_block = rig.rand_block(); let bad_block = rig.rand_block();
bl.parent_lookup_response(id, peer_id, Some(Arc::new(bad_block)), D, &mut cx); bl.parent_lookup_response(id, peer_id, Some(bad_block.into()), D, &mut cx);
rig.expect_penalty(); rig.expect_penalty();
} }
if i < parent_lookup::PARENT_FAIL_TOLERANCE { if i < parent_lookup::PARENT_FAIL_TOLERANCE {
@ -470,7 +466,7 @@ fn test_parent_lookup_too_many_processing_attempts_must_blacklist() {
let peer_id = PeerId::random(); let peer_id = PeerId::random();
// Trigger the request // Trigger the request
bl.search_parent(block_hash, Arc::new(block), peer_id, &mut cx); bl.search_parent(block_hash, block.into(), peer_id, &mut cx);
// Fail downloading the block // Fail downloading the block
for _ in 0..(parent_lookup::PARENT_FAIL_TOLERANCE - PROCESSING_FAILURES) { for _ in 0..(parent_lookup::PARENT_FAIL_TOLERANCE - PROCESSING_FAILURES) {
@ -484,7 +480,7 @@ fn test_parent_lookup_too_many_processing_attempts_must_blacklist() {
let id = dbg!(rig.expect_parent_request()); let id = dbg!(rig.expect_parent_request());
assert!(!bl.failed_chains.contains(&block_hash)); assert!(!bl.failed_chains.contains(&block_hash));
// send the right parent but fail processing // send the right parent but fail processing
bl.parent_lookup_response(id, peer_id, Some(parent.clone()), D, &mut cx); bl.parent_lookup_response(id, peer_id, Some(parent.clone().into()), D, &mut cx);
bl.parent_block_processed(block_hash, BlockError::InvalidSignature.into(), &mut cx); bl.parent_block_processed(block_hash, BlockError::InvalidSignature.into(), &mut cx);
bl.parent_lookup_response(id, peer_id, None, D, &mut cx); bl.parent_lookup_response(id, peer_id, None, D, &mut cx);
rig.expect_penalty(); rig.expect_penalty();
@ -511,12 +507,12 @@ fn test_parent_lookup_too_deep() {
let peer_id = PeerId::random(); let peer_id = PeerId::random();
let trigger_block = blocks.pop().unwrap(); let trigger_block = blocks.pop().unwrap();
let chain_hash = trigger_block.canonical_root(); let chain_hash = trigger_block.canonical_root();
bl.search_parent(chain_hash, Arc::new(trigger_block), peer_id, &mut cx); bl.search_parent(chain_hash, trigger_block.into(), peer_id, &mut cx);
for block in blocks.into_iter().rev() { for block in blocks.into_iter().rev() {
let id = rig.expect_parent_request(); let id = rig.expect_parent_request();
// the block // the block
bl.parent_lookup_response(id, peer_id, Some(Arc::new(block.clone())), D, &mut cx); bl.parent_lookup_response(id, peer_id, Some(block.clone().into()), D, &mut cx);
// the stream termination // the stream termination
bl.parent_lookup_response(id, peer_id, None, D, &mut cx); bl.parent_lookup_response(id, peer_id, None, D, &mut cx);
// the processing request // the processing request
@ -524,7 +520,7 @@ fn test_parent_lookup_too_deep() {
// the processing result // the processing result
bl.parent_block_processed( bl.parent_block_processed(
chain_hash, chain_hash,
BlockError::ParentUnknown(Arc::new(block)).into(), BlockError::ParentUnknown(block.into()).into(),
&mut cx, &mut cx,
) )
} }
@ -540,7 +536,7 @@ fn test_parent_lookup_disconnection() {
let trigger_block = rig.rand_block(); let trigger_block = rig.rand_block();
bl.search_parent( bl.search_parent(
trigger_block.canonical_root(), trigger_block.canonical_root(),
Arc::new(trigger_block), trigger_block.into(),
peer_id, peer_id,
&mut cx, &mut cx,
); );
@ -561,7 +557,7 @@ fn test_single_block_lookup_ignored_response() {
// The peer provides the correct block, should not be penalized. Now the block should be sent // The peer provides the correct block, should not be penalized. Now the block should be sent
// for processing. // for processing.
bl.single_block_lookup_response(id, peer_id, Some(Arc::new(block)), D, &mut cx); bl.single_block_lookup_response(id, peer_id, Some(block.into()), D, &mut cx);
rig.expect_empty_network(); rig.expect_empty_network();
rig.expect_block_process(); rig.expect_block_process();
@ -587,11 +583,11 @@ fn test_parent_lookup_ignored_response() {
let peer_id = PeerId::random(); let peer_id = PeerId::random();
// Trigger the request // Trigger the request
bl.search_parent(chain_hash, Arc::new(block), peer_id, &mut cx); bl.search_parent(chain_hash, block.into(), peer_id, &mut cx);
let id = rig.expect_parent_request(); let id = rig.expect_parent_request();
// Peer sends the right block, it should be sent for processing. Peer should not be penalized. // Peer sends the right block, it should be sent for processing. Peer should not be penalized.
bl.parent_lookup_response(id, peer_id, Some(Arc::new(parent)), D, &mut cx); bl.parent_lookup_response(id, peer_id, Some(parent.into()), D, &mut cx);
rig.expect_block_process(); rig.expect_block_process();
rig.expect_empty_network(); rig.expect_empty_network();

100
beacon_node/network/src/sync/manager.rs
View File

@ -79,6 +79,35 @@ pub enum BlockTy<T: EthSpec> {
}, },
} }
#[cfg(test)]
impl<T: EthSpec> From<SignedBeaconBlock<T>> for BlockTy<T> {
fn from(block: SignedBeaconBlock<T>) -> Self {
BlockTy::Block {
block: Arc::new(block),
}
}
}
#[cfg(test)]
impl<T: EthSpec> From<Arc<SignedBeaconBlock<T>>> for BlockTy<T> {
fn from(block: Arc<SignedBeaconBlock<T>>) -> Self {
BlockTy::Block { block }
}
}
impl<T: EthSpec> BlockTy<T> {
pub fn block(&self) -> &SignedBeaconBlock<T> {
match &self {
BlockTy::Block { block } => block,
BlockTy::BlockAndBlob { block_sidecar_pair } => &block_sidecar_pair.beacon_block,
}
}
pub fn parent_root(&self) -> Hash256 {
self.block().parent_root()
}
}
// TODO: probably needes to be changed. This is needed because SignedBeaconBlockAndBlobsSidecar // TODO: probably needes to be changed. This is needed because SignedBeaconBlockAndBlobsSidecar
// does not implement Hash // does not implement Hash
impl<T: EthSpec> std::hash::Hash for BlockTy<T> { impl<T: EthSpec> std::hash::Hash for BlockTy<T> {
@ -132,7 +161,7 @@ pub enum SyncMessage<T: EthSpec> {
}, },
/// A blob has been received from the RPC. /// A blob has been received from the RPC.
RpcBlob { RpcGlob {
request_id: RequestId, request_id: RequestId,
peer_id: PeerId, peer_id: PeerId,
blob_sidecar: Option<Arc<BlobsSidecar<T>>>, blob_sidecar: Option<Arc<BlobsSidecar<T>>>,
@ -140,7 +169,7 @@ pub enum SyncMessage<T: EthSpec> {
}, },
/// A block and blobs have been received from the RPC. /// A block and blobs have been received from the RPC.
RpcBlockAndBlob { RpcBlockAndGlob {
request_id: RequestId, request_id: RequestId,
peer_id: PeerId, peer_id: PeerId,
block_and_blobs: Option<Arc<SignedBeaconBlockAndBlobsSidecar<T>>>, block_and_blobs: Option<Arc<SignedBeaconBlockAndBlobsSidecar<T>>>,
@ -612,8 +641,14 @@ impl<T: BeaconChainTypes> SyncManager<T> {
if self.network_globals.peers.read().is_connected(&peer_id) if self.network_globals.peers.read().is_connected(&peer_id)
&& self.network.is_execution_engine_online() && self.network.is_execution_engine_online()
{ {
self.block_lookups // TODO: here it would be ideal if unknown block carried either the block or
.search_parent(block_root, block, peer_id, &mut self.network); // the block and blob since for block lookups we don't care.
self.block_lookups.search_parent(
block_root,
BlockTy::Block { block },
peer_id,
&mut self.network,
);
} }
} }
SyncMessage::UnknownBlockHash(peer_id, block_hash) => { SyncMessage::UnknownBlockHash(peer_id, block_hash) => {
@ -674,18 +709,23 @@ impl<T: BeaconChainTypes> SyncManager<T> {
.block_lookups .block_lookups
.parent_chain_processed(chain_hash, result, &mut self.network), .parent_chain_processed(chain_hash, result, &mut self.network),
}, },
SyncMessage::RpcBlob { SyncMessage::RpcGlob {
request_id, request_id,
peer_id, peer_id,
blob_sidecar, blob_sidecar,
seen_timestamp, seen_timestamp,
} => self.rpc_sidecar_received(request_id, peer_id, blob_sidecar, seen_timestamp), } => self.rpc_sidecar_received(request_id, peer_id, blob_sidecar, seen_timestamp),
SyncMessage::RpcBlockAndBlob { SyncMessage::RpcBlockAndGlob {
request_id, request_id,
peer_id, peer_id,
block_and_blobs, block_and_blobs,
seen_timestamp, seen_timestamp,
} => todo!(), } => self.rpc_block_sidecar_pair_received(
request_id,
peer_id,
block_and_blobs,
seen_timestamp,
),
} }
} }
@ -755,14 +795,14 @@ impl<T: BeaconChainTypes> SyncManager<T> {
RequestId::SingleBlock { id } => self.block_lookups.single_block_lookup_response( RequestId::SingleBlock { id } => self.block_lookups.single_block_lookup_response(
id, id,
peer_id, peer_id,
beacon_block, beacon_block.map(|block| BlockTy::Block { block }),
seen_timestamp, seen_timestamp,
&mut self.network, &mut self.network,
), ),
RequestId::ParentLookup { id } => self.block_lookups.parent_lookup_response( RequestId::ParentLookup { id } => self.block_lookups.parent_lookup_response(
id, id,
peer_id, peer_id,
beacon_block, beacon_block.map(|block| BlockTy::Block { block }),
seen_timestamp, seen_timestamp,
&mut self.network, &mut self.network,
), ),
@ -858,8 +898,9 @@ impl<T: BeaconChainTypes> SyncManager<T> {
seen_timestamp: Duration, seen_timestamp: Duration,
) { ) {
match request_id { match request_id {
RequestId::SingleBlock { id } => todo!("do we request individual sidecars?"), RequestId::SingleBlock { id } | RequestId::ParentLookup { id } => {
RequestId::ParentLookup { id } => todo!(), unreachable!("There is no such thing as a singular 'by root' glob request that is not accompanied by the block")
}
RequestId::BackFillSync { .. } => { RequestId::BackFillSync { .. } => {
unreachable!("An only blocks request does not receive sidecars") unreachable!("An only blocks request does not receive sidecars")
} }
@ -905,6 +946,43 @@ impl<T: BeaconChainTypes> SyncManager<T> {
} }
} }
} }
fn rpc_block_sidecar_pair_received(
&mut self,
request_id: RequestId,
peer_id: PeerId,
block_sidecar_pair: Option<Arc<SignedBeaconBlockAndBlobsSidecar<T::EthSpec>>>,
seen_timestamp: Duration,
) {
match request_id {
RequestId::SingleBlock { id } => self.block_lookups.single_block_lookup_response(
id,
peer_id,
block_sidecar_pair.map(|block_sidecar_pair| BlockTy::BlockAndBlob {
// TODO: why is this in an arc
block_sidecar_pair: (*block_sidecar_pair).clone(),
}),
seen_timestamp,
&mut self.network,
),
RequestId::ParentLookup { id } => self.block_lookups.parent_lookup_response(
id,
peer_id,
block_sidecar_pair.map(|block_sidecar_pair| BlockTy::BlockAndBlob {
// TODO: why is this in an arc
block_sidecar_pair: (*block_sidecar_pair).clone(),
}),
seen_timestamp,
&mut self.network,
),
RequestId::BackFillSync { .. }
| RequestId::BackFillSidecarPair { .. }
| RequestId::RangeSync { .. }
| RequestId::RangeSidecarPair { .. } => unreachable!(
"since range requests are not block-glob coupled, this should never be reachable"
),
}
}
} }
impl<IgnoredOkVal, T: EthSpec> From<Result<IgnoredOkVal, BlockError<T>>> for BlockProcessResult<T> { impl<IgnoredOkVal, T: EthSpec> From<Result<IgnoredOkVal, BlockError<T>>> for BlockProcessResult<T> {

2
beacon_node/network/src/sync/network_context.rs
View File

@ -460,6 +460,8 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
) -> Result<Id, &'static str> { ) -> Result<Id, &'static str> {
//FIXME(sean) add prune depth logic here? //FIXME(sean) add prune depth logic here?
// D: YES // D: YES
// MOREINFO: here depending of the boundaries we decide what kind of request we send, if we
// request just a block or if we request a block, glob pair.
trace!( trace!(
self.log, self.log,