gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-07-02 04:14:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

add tests and fix some things

Browse Source
This commit is contained in:
realbigsean
2023-05-18 15:20:56 -04:00
parent 51c4506c53
commit 542e3c16db
10 changed files with 1110 additions and 576 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
beacon_node/beacon_chain/src/block_verification.rs
View File

@@ -70,7 +70,7 @@ use crate::{
use derivative::Derivative; use derivative::Derivative;
use eth2::types::EventKind; use eth2::types::EventKind;
use execution_layer::PayloadStatus; use execution_layer::PayloadStatus;
use fork_choice::{AttestationFromBlock, PayloadVerificationStatus}; pub use fork_choice::{AttestationFromBlock, PayloadVerificationStatus};
use parking_lot::RwLockReadGuard; use parking_lot::RwLockReadGuard;
use proto_array::Block as ProtoBlock; use proto_array::Block as ProtoBlock;
use safe_arith::ArithError; use safe_arith::ArithError;

128
beacon_node/beacon_chain/src/data_availability_checker.rs
View File

@@ -11,7 +11,7 @@ use slot_clock::SlotClock;
use ssz_types::{Error, FixedVector, VariableList}; use ssz_types::{Error, FixedVector, VariableList};
use state_processing::per_block_processing::deneb::deneb::verify_kzg_commitments_against_transactions; use state_processing::per_block_processing::deneb::deneb::verify_kzg_commitments_against_transactions;
use std::collections::hash_map::{Entry, OccupiedEntry}; use std::collections::hash_map::{Entry, OccupiedEntry};
use std::collections::HashMap; use std::collections::{HashMap, HashSet};
use std::sync::Arc; use std::sync::Arc;
use strum::IntoStaticStr; use strum::IntoStaticStr;
use types::beacon_block_body::KzgCommitments; use types::beacon_block_body::KzgCommitments;
@@ -142,75 +142,61 @@ impl<T: EthSpec, S: SlotClock> DataAvailabilityChecker<T, S> {
.map_or(false, |cache| cache.executed_block.is_some()) .map_or(false, |cache| cache.executed_block.is_some())
} }
pub fn get_missing_blob_ids(&self, block_root: &Hash256) -> Vec<BlobIdentifier> { pub fn get_missing_blob_ids_checking_cache(
let epoch = self.slot_clock.now().map(|s| s.epoch(T::slots_per_epoch())); &self,
if epoch.map_or(false, |e| self.da_check_required(e)) { block_root: Hash256,
self.availability_cache ) -> Option<Vec<BlobIdentifier>> {
.read() let guard = self.availability_cache.read();
.get(block_root) let (block, blob_indices) = guard
.and_then(|cache| { .get(&block_root)
cache.executed_block.as_ref().map(|block| { .map(|cache| {
block.get_filtered_blob_ids(|i, _| cache.verified_blobs.get(i).is_none()) let block_opt = cache
.executed_block
.as_ref()
.map(|block| &block.block.block);
let blobs = cache
.verified_blobs
.iter()
.enumerate()
.filter_map(|(i, maybe_blob)| maybe_blob.as_ref().map(|_| i))
.collect::<HashSet<_>>();
(block_opt, blobs)
})
.unwrap_or_default();
self.get_missing_blob_ids(block_root, block, Some(blob_indices))
}
/// A `None` indicates blobs are not required.
///
/// If there's no block, all possible ids will be returned that don't exist in the given blobs.
/// If there no blobs, all possible ids will be returned.
pub fn get_missing_blob_ids(
&self,
block_root: Hash256,
block_opt: Option<&Arc<SignedBeaconBlock<T>>>,
blobs_opt: Option<HashSet<usize>>,
) -> Option<Vec<BlobIdentifier>> {
let epoch = self.slot_clock.now()?.epoch(T::slots_per_epoch());
self.da_check_required(epoch).then(|| {
block_opt
.map(|block| {
block.get_filtered_blob_ids(Some(block_root), |i, _| {
blobs_opt.as_ref().map_or(true, |blobs| !blobs.contains(&i))
}) })
}) })
.unwrap_or_else(|| { .unwrap_or_else(|| {
let mut blob_ids = Vec::with_capacity(T::max_blobs_per_block()); let mut blob_ids = Vec::with_capacity(T::max_blobs_per_block());
for i in 0..T::max_blobs_per_block() { for i in 0..T::max_blobs_per_block() {
blob_ids.push(BlobIdentifier { if blobs_opt.as_ref().map_or(true, |blobs| !blobs.contains(&i)) {
block_root: *block_root, blob_ids.push(BlobIdentifier {
index: i as u64, block_root,
}); index: i as u64,
});
}
} }
blob_ids blob_ids
}) })
} else {
vec![]
}
}
pub fn wrap_block(
&self,
block_root: Hash256,
block: Arc<SignedBeaconBlock<T>>,
blobs: FixedBlobSidecarList<T>,
) -> Result<BlockWrapper<T>, AvailabilityCheckError> {
Ok(match self.get_blob_requirements(&block)? {
BlobRequirements::EmptyBlobs => BlockWrapper::Block(block),
BlobRequirements::NotRequired => BlockWrapper::Block(block),
BlobRequirements::PreDeneb => BlockWrapper::Block(block),
BlobRequirements::Required => {
let expected_num_blobs = block
.message()
.body()
.blob_kzg_commitments()
.map(|commitments| commitments.len())
.unwrap_or(0);
let mut blob_count = 0;
while let Some((index, Some(blob))) = blobs.iter().enumerate().next() {
blob_count += 1;
if blob.block_root != block_root {
return Err(AvailabilityCheckError::BlockBlobRootMismatch {
block_root,
blob_block_root: blob.block_root,
});
}
let expected_index = index as u64;
if expected_index != blob.index {
return Err(AvailabilityCheckError::UnorderedBlobs {
expected_index,
blob_index: blob.index,
});
}
}
if blob_count < expected_num_blobs {
return Err(AvailabilityCheckError::MissingBlobs(block_root));
}
BlockWrapper::BlockAndBlobs(block, blobs)
}
}) })
} }
@@ -567,12 +553,11 @@ impl<E: EthSpec> AvailabilityPendingBlock<E> {
self.block.slot() self.block.slot()
} }
pub fn num_blobs_expected(&self) -> usize { pub fn num_blobs_expected(&self) -> usize {
self.kzg_commitments() self.block.num_expected_blobs()
.map_or(0, |commitments| commitments.len())
} }
pub fn get_all_blob_ids(&self, block_root: Option<Hash256>) -> Vec<BlobIdentifier> { pub fn get_all_blob_ids(&self, block_root: Option<Hash256>) -> Vec<BlobIdentifier> {
self.get_filtered_blob_ids(block_root, |_, _| true) self.block.get_expected_blob_ids(block_root)
} }
pub fn get_filtered_blob_ids( pub fn get_filtered_blob_ids(
@@ -580,18 +565,7 @@ impl<E: EthSpec> AvailabilityPendingBlock<E> {
block_root: Option<Hash256>, block_root: Option<Hash256>,
filter: impl Fn(usize, Hash256) -> bool, filter: impl Fn(usize, Hash256) -> bool,
) -> Vec<BlobIdentifier> { ) -> Vec<BlobIdentifier> {
let block_root = block_root.unwrap_or_else(|| self.as_block().canonical_root()); self.block.get_filtered_blob_ids(block_root, filter)
let num_blobs_expected = self.num_blobs_expected();
let mut blob_ids = Vec::with_capacity(num_blobs_expected);
for i in 0..num_blobs_expected {
if filter(i, block_root) {
blob_ids.push(BlobIdentifier {
block_root,
index: i as u64,
});
}
}
blob_ids
} }
} }

4
beacon_node/beacon_chain/src/lib.rs
View File

@@ -69,7 +69,9 @@ pub use self::historical_blocks::HistoricalBlockError;
pub use attestation_verification::Error as AttestationError; pub use attestation_verification::Error as AttestationError;
pub use beacon_fork_choice_store::{BeaconForkChoiceStore, Error as ForkChoiceStoreError}; pub use beacon_fork_choice_store::{BeaconForkChoiceStore, Error as ForkChoiceStoreError};
pub use block_verification::{ pub use block_verification::{
get_block_root, BlockError, ExecutedBlock, ExecutionPayloadError, GossipVerifiedBlock, get_block_root, AvailabilityPendingExecutedBlock, BlockError, ExecutedBlock,
ExecutionPayloadError, GossipVerifiedBlock, IntoExecutionPendingBlock,
PayloadVerificationOutcome, PayloadVerificationStatus,
}; };
pub use canonical_head::{CachedHead, CanonicalHead, CanonicalHeadRwLock}; pub use canonical_head::{CachedHead, CanonicalHead, CanonicalHeadRwLock};
pub use eth1_chain::{Eth1Chain, Eth1ChainBackend}; pub use eth1_chain::{Eth1Chain, Eth1ChainBackend};

6
beacon_node/network/src/beacon_processor/worker/sync_methods.rs
View File

@@ -58,7 +58,7 @@ impl<T: BeaconChainTypes> Worker<T> {
) { ) {
if !should_process { if !should_process {
// Sync handles these results // Sync handles these results
self.send_sync_message(SyncMessage::BlockPartProcessed { self.send_sync_message(SyncMessage::BlockComponentProcessed {
process_type, process_type,
result: crate::sync::manager::BlockProcessingResult::Ignored, result: crate::sync::manager::BlockProcessingResult::Ignored,
response_type: crate::sync::manager::ResponseType::Block, response_type: crate::sync::manager::ResponseType::Block,
@@ -192,7 +192,7 @@ impl<T: BeaconChainTypes> Worker<T> {
} }
} }
// Sync handles these results // Sync handles these results
self.send_sync_message(SyncMessage::BlockPartProcessed { self.send_sync_message(SyncMessage::BlockComponentProcessed {
process_type, process_type,
result: result.into(), result: result.into(),
response_type: ResponseType::Block, response_type: ResponseType::Block,
@@ -229,7 +229,7 @@ impl<T: BeaconChainTypes> Worker<T> {
.await; .await;
// Sync handles these results // Sync handles these results
self.send_sync_message(SyncMessage::BlockPartProcessed { self.send_sync_message(SyncMessage::BlockComponentProcessed {
process_type, process_type,
result: result.into(), result: result.into(),
response_type: ResponseType::Blob, response_type: ResponseType::Blob,

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

@@ -1,5 +1,5 @@
use beacon_chain::blob_verification::{AsBlock, BlockWrapper}; use beacon_chain::blob_verification::{AsBlock, BlockWrapper};
use beacon_chain::data_availability_checker::{AvailabilityCheckError, DataAvailabilityChecker}; use beacon_chain::data_availability_checker::{DataAvailabilityChecker};
use beacon_chain::{AvailabilityProcessingStatus, BeaconChainTypes, BlockError}; use beacon_chain::{AvailabilityProcessingStatus, BeaconChainTypes, BlockError};
use lighthouse_network::rpc::RPCError; use lighthouse_network::rpc::RPCError;
use lighthouse_network::{PeerAction, PeerId}; use lighthouse_network::{PeerAction, PeerId};
@@ -15,7 +15,7 @@ use strum::Display;
use types::blob_sidecar::FixedBlobSidecarList; use types::blob_sidecar::FixedBlobSidecarList;
use types::{BlobSidecar, SignedBeaconBlock, Slot}; use types::{BlobSidecar, SignedBeaconBlock, Slot};
use self::parent_lookup::{LookupDownloadStatus, PARENT_FAIL_TOLERANCE}; use self::parent_lookup::PARENT_FAIL_TOLERANCE;
use self::parent_lookup::{ParentLookup, ParentVerifyError}; use self::parent_lookup::{ParentLookup, ParentVerifyError};
use self::single_block_lookup::{LookupVerifyError, SingleBlockLookup}; use self::single_block_lookup::{LookupVerifyError, SingleBlockLookup};
use super::manager::BlockProcessingResult; use super::manager::BlockProcessingResult;
@@ -26,6 +26,7 @@ use super::{
}; };
use crate::beacon_processor::{ChainSegmentProcessId, WorkEvent}; use crate::beacon_processor::{ChainSegmentProcessId, WorkEvent};
use crate::metrics; use crate::metrics;
use crate::sync::block_lookups::single_block_lookup::UnknownParentComponents;
mod parent_lookup; mod parent_lookup;
mod single_block_lookup; mod single_block_lookup;
@@ -111,12 +112,6 @@ impl PeerShouldHave {
PeerShouldHave::Neither(_) => false, PeerShouldHave::Neither(_) => false,
} }
} }
fn should_have_blobs(&self) -> bool {
match self {
PeerShouldHave::BlockAndBlobs(_) => true,
PeerShouldHave::Neither(_) => false,
}
}
} }
#[derive(Debug, Copy, Clone)] #[derive(Debug, Copy, Clone)]
@@ -146,19 +141,37 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
pub fn search_block( pub fn search_block(
&mut self, &mut self,
hash: Hash256, block_root: Hash256,
peer_source: PeerShouldHave, peer_source: PeerShouldHave,
cx: &mut SyncNetworkContext<T>, cx: &mut SyncNetworkContext<T>,
) { ) {
self.search_block_with(|_| {}, hash, peer_source, cx) self.search_block_with(block_root, None, None, peer_source, cx)
}
pub fn search_current_unknown_parent_block_and_blobs(
&mut self,
block_root: Hash256,
block: Option<Arc<SignedBeaconBlock<T::EthSpec>>>,
blobs: Option<FixedBlobSidecarList<T::EthSpec>>,
peer_id: PeerId,
cx: &mut SyncNetworkContext<T>,
) {
self.search_block_with(
block_root,
block,
blobs,
PeerShouldHave::Neither(peer_id),
cx,
);
} }
/// Searches for a single block hash. If the blocks parent is unknown, a chain of blocks is /// Searches for a single block hash. If the blocks parent is unknown, a chain of blocks is
/// constructed. /// constructed.
pub fn search_block_with( pub fn search_block_with(
&mut self, &mut self,
cache_fn: impl Fn(&mut SingleBlockLookup<SINGLE_BLOCK_LOOKUP_MAX_ATTEMPTS, T>), block_root: Hash256,
hash: Hash256, block: Option<Arc<SignedBeaconBlock<T::EthSpec>>>,
blobs: Option<FixedBlobSidecarList<T::EthSpec>>,
peer_source: PeerShouldHave, peer_source: PeerShouldHave,
cx: &mut SyncNetworkContext<T>, cx: &mut SyncNetworkContext<T>,
) { ) {
@@ -167,14 +180,15 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
.single_block_lookups .single_block_lookups
.iter_mut() .iter_mut()
.any(|(_, _, single_block_request)| { .any(|(_, _, single_block_request)| {
single_block_request.add_peer_if_useful(&hash, peer_source) single_block_request.add_peer_if_useful(&block_root, peer_source)
}) })
{ {
return; return;
} }
if self.parent_lookups.iter_mut().any(|parent_req| { if self.parent_lookups.iter_mut().any(|parent_req| {
parent_req.add_peer_if_useful(&hash, peer_source) || parent_req.contains_block(&hash) parent_req.add_peer_if_useful(&block_root, peer_source)
|| parent_req.contains_block(&block_root)
}) { }) {
// If the block was already downloaded, or is being downloaded in this moment, do not // If the block was already downloaded, or is being downloaded in this moment, do not
// request it. // request it.
@@ -184,7 +198,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
if self if self
.processing_parent_lookups .processing_parent_lookups
.values() .values()
.any(|(hashes, _last_parent_request)| hashes.contains(&hash)) .any(|(hashes, _last_parent_request)| hashes.contains(&block_root))
{ {
// we are already processing this block, ignore it. // we are already processing this block, ignore it.
return; return;
@@ -193,13 +207,26 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
debug!( debug!(
self.log, self.log,
"Searching for block"; "Searching for block";
"peer_id" => %peer_source, "peer_id" => ?peer_source,
"block" => %hash "block" => ?block_root
); );
let mut single_block_request = let triggered_by_unknown_parent = block.is_some() || blobs.is_some();
SingleBlockLookup::new(hash, peer_source, self.da_checker.clone()); let unknown_parent_components = if triggered_by_unknown_parent {
cache_fn(&mut single_block_request); Some(UnknownParentComponents {
downloaded_block: block,
downloaded_blobs: blobs.unwrap_or_default(),
})
} else {
None
};
let mut single_block_request = SingleBlockLookup::new(
block_root,
unknown_parent_components,
peer_source,
self.da_checker.clone(),
);
let block_request_id = let block_request_id =
if let Ok(Some((peer_id, block_request))) = single_block_request.request_block() { if let Ok(Some((peer_id, block_request))) = single_block_request.request_block() {
@@ -224,40 +251,6 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
); );
} }
pub fn search_current_unknown_parent(
&mut self,
block_root: Hash256,
block: BlockWrapper<T::EthSpec>,
peer_id: PeerId,
cx: &mut SyncNetworkContext<T>,
) {
self.search_block_with(
|request| {
let _ = request.add_block_wrapper(block_root, block.clone());
},
block_root,
PeerShouldHave::Neither(peer_id),
cx,
);
}
pub fn search_current_unknown_blob_parent(
&mut self,
blob: Arc<BlobSidecar<T::EthSpec>>,
peer_id: PeerId,
cx: &mut SyncNetworkContext<T>,
) {
let block_root = blob.block_root;
self.search_block_with(
|request| {
let _ = request.add_blob(blob.clone());
},
block_root,
PeerShouldHave::Neither(peer_id),
cx,
);
}
/// If a block is attempted to be processed but we do not know its parent, this function is /// If a block is attempted to be processed but we do not know its parent, this function is
/// called in order to find the block's parent. /// called in order to find the block's parent.
pub fn search_parent( pub fn search_parent(
@@ -268,7 +261,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
peer_id: PeerId, peer_id: PeerId,
cx: &mut SyncNetworkContext<T>, cx: &mut SyncNetworkContext<T>,
) { ) {
// Gossip blocks or blobs shouldn't be propogated if parents are unavailable. // Gossip blocks or blobs shouldn't be propagated if parents are unavailable.
let peer_source = PeerShouldHave::BlockAndBlobs(peer_id); let peer_source = PeerShouldHave::BlockAndBlobs(peer_id);
// If this block or it's parent is part of a known failed chain, ignore it. // If this block or it's parent is part of a known failed chain, ignore it.
@@ -319,33 +312,24 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
let stream_terminator = block.is_none().into(); let stream_terminator = block.is_none().into();
let log = self.log.clone(); let log = self.log.clone();
let Some((triggered_parent_request, request_id_ref, request_ref)) = self.find_single_lookup_request(id, stream_terminator, ResponseType::Block) else { let Some((pending_parent_request, request_id_ref, request_ref)) = self.find_single_lookup_request(id, stream_terminator, ResponseType::Block) else {
return; return;
}; };
let should_remove = match request_ref.verify_block(block) { let should_remove = match request_ref.verify_block(block) {
Ok(Some((root, block))) => { Ok(Some((block_root, block))) => {
if triggered_parent_request { if let Some(parent_components) = request_ref.unknown_parent_components.as_mut() {
if let LookupDownloadStatus::AvailabilityCheck(e) = parent_components.add_unknown_parent_block(block.clone());
request_ref.add_block(root, block) };
{
handle_availability_check_error( if !pending_parent_request {
request_id_ref, let block_wrapper = request_ref
request_ref, .get_downloaded_block()
ResponseType::Blob, .unwrap_or(BlockWrapper::Block(block));
peer_id,
e,
cx,
&log,
)
} else {
ShouldRemoveLookup::False
}
} else {
// This is the correct block, send it for processing // This is the correct block, send it for processing
match self.send_block_for_processing( match self.send_block_for_processing(
root, block_root,
BlockWrapper::Block(block), block_wrapper,
seen_timestamp, seen_timestamp,
BlockProcessType::SingleBlock { id }, BlockProcessType::SingleBlock { id },
cx, cx,
@@ -353,6 +337,8 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
Ok(()) => ShouldRemoveLookup::False, Ok(()) => ShouldRemoveLookup::False,
Err(()) => ShouldRemoveLookup::True, Err(()) => ShouldRemoveLookup::True,
} }
} else {
ShouldRemoveLookup::False
} }
} }
Ok(None) => ShouldRemoveLookup::False, Ok(None) => ShouldRemoveLookup::False,
@@ -390,26 +376,32 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
let log = self.log.clone(); let log = self.log.clone();
let Some((triggered_parent_request, request_id_ref, request_ref)) = let Some((pending_parent_requests, request_id_ref, request_ref)) =
self.find_single_lookup_request(id, stream_terminator, ResponseType::Blob) else { self.find_single_lookup_request(id, stream_terminator, ResponseType::Blob) else {
return; return;
}; };
let should_remove = match request_ref.verify_blob(blob) { let should_remove = match request_ref.verify_blob(blob) {
Ok(Some((block_root, blobs))) => { Ok(Some((block_root, blobs))) => {
if triggered_parent_request { if let Some(parent_components) = request_ref.unknown_parent_components.as_mut() {
if let LookupDownloadStatus::AvailabilityCheck(e) = parent_components.add_unknown_parent_blobs(blobs);
request_ref.add_blobs(block_root, blobs)
{ if !pending_parent_requests {
handle_availability_check_error( request_ref
request_id_ref, .get_downloaded_block()
request_ref, .map(|block| {
ResponseType::Blob, match self.send_block_for_processing(
peer_id, block_root,
e, block,
cx, seen_timestamp,
&log, BlockProcessType::SingleBlock { id },
) cx,
) {
Ok(()) => ShouldRemoveLookup::False,
Err(()) => ShouldRemoveLookup::True,
}
})
.unwrap_or(ShouldRemoveLookup::False)
} else { } else {
ShouldRemoveLookup::False ShouldRemoveLookup::False
} }
@@ -525,40 +517,41 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
match parent_lookup.verify_block(block, &mut self.failed_chains) { match parent_lookup.verify_block(block, &mut self.failed_chains) {
Ok(Some((block_root, block))) => { Ok(Some((block_root, block))) => {
let process_or_search = parent_lookup.add_block(block_root, block); parent_lookup.add_current_request_block(block);
match process_or_search { if let Some(block_wrapper) =
LookupDownloadStatus::Process(wrapper) => { parent_lookup.current_parent_request.get_downloaded_block()
let chain_hash = parent_lookup.chain_hash(); {
if self let chain_hash = parent_lookup.chain_hash();
.send_block_for_processing( if self
block_root, .send_block_for_processing(
wrapper, block_root,
seen_timestamp, block_wrapper,
BlockProcessType::ParentLookup { chain_hash }, seen_timestamp,
cx, BlockProcessType::ParentLookup { chain_hash },
) cx,
.is_ok() )
{ .is_ok()
self.parent_lookups.push(parent_lookup) {
} self.parent_lookups.push(parent_lookup)
} }
LookupDownloadStatus::SearchBlock(block_root) => { } else {
if let Some(peer_source) = let outstanding_blobs_req =
parent_lookup.peer_source(ResponseType::Block, peer_id) parent_lookup.current_parent_blob_request_id.is_some();
{ if !outstanding_blobs_req {
self.search_block(block_root, peer_source, cx); if let Ok(peer_id) = parent_lookup
self.parent_lookups.push(parent_lookup) .current_parent_request
} else { .downloading_peer(ResponseType::Blob) {
warn!(self.log, "Response from untracked peer"; "peer_id" => %peer_id, "block_root" => ?block_root); cx.report_peer(
peer_id.to_peer_id(),
PeerAction::MidToleranceError,
"bbroot_failed_chains",
);
} }
self.request_parent_blob(parent_lookup, cx);
} else {
self.parent_lookups.push(parent_lookup)
} }
LookupDownloadStatus::AvailabilityCheck(e) => self.handle_invalid_block(
BlockError::AvailabilityCheck(e),
peer_id,
cx,
ResponseType::Block,
parent_lookup,
),
} }
} }
Ok(None) => { Ok(None) => {
@@ -629,11 +622,9 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
seen_timestamp: Duration, seen_timestamp: Duration,
cx: &mut SyncNetworkContext<T>, cx: &mut SyncNetworkContext<T>,
) { ) {
let mut parent_lookup = if let Some(pos) = self let mut parent_lookup = if let Some(pos) = self.parent_lookups.iter().position(|request| {
.parent_lookups request.pending_blob_response(id)
.iter() }) {
.position(|request| request.pending_blob_response(id))
{
self.parent_lookups.remove(pos) self.parent_lookups.remove(pos)
} else { } else {
if blob.is_some() { if blob.is_some() {
@@ -644,40 +635,25 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
match parent_lookup.verify_blob(blob, &mut self.failed_chains) { match parent_lookup.verify_blob(blob, &mut self.failed_chains) {
Ok(Some((block_root, blobs))) => { Ok(Some((block_root, blobs))) => {
let processed_or_search = parent_lookup.add_blobs(block_root, blobs); parent_lookup.add_current_request_blobs(blobs);
match processed_or_search { let chain_hash = parent_lookup.chain_hash();
LookupDownloadStatus::Process(wrapper) => { if let Some(block_wrapper) =
let chain_hash = parent_lookup.chain_hash(); parent_lookup.current_parent_request.get_downloaded_block()
if self {
.send_block_for_processing( if self
block_root, .send_block_for_processing(
wrapper, block_root,
seen_timestamp, block_wrapper,
BlockProcessType::ParentLookup { chain_hash }, seen_timestamp,
cx, BlockProcessType::ParentLookup { chain_hash },
) cx,
.is_ok() )
{ .is_ok()
self.parent_lookups.push(parent_lookup) {
} self.parent_lookups.push(parent_lookup)
} }
LookupDownloadStatus::SearchBlock(block_root) => { } else {
if let Some(peer_source) = self.parent_lookups.push(parent_lookup)
parent_lookup.peer_source(ResponseType::Block, peer_id)
{
self.search_block(block_root, peer_source, cx);
self.parent_lookups.push(parent_lookup)
} else {
warn!(self.log, "Response from untracked peer"; "peer_id" => %peer_id, "block_root" => ?block_root);
}
}
LookupDownloadStatus::AvailabilityCheck(e) => self.handle_invalid_block(
BlockError::AvailabilityCheck(e),
peer_id,
cx,
ResponseType::Blob,
parent_lookup,
),
} }
} }
Ok(None) => { Ok(None) => {
@@ -865,7 +841,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
/* Processing responses */ /* Processing responses */
pub fn single_block_processed( pub fn single_block_component_processed(
&mut self, &mut self,
target_id: Id, target_id: Id,
result: BlockProcessingResult<T::EthSpec>, result: BlockProcessingResult<T::EthSpec>,
@@ -874,19 +850,21 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
) { ) {
let lookup_components_opt = self.single_block_lookups.iter_mut().enumerate().find_map( let lookup_components_opt = self.single_block_lookups.iter_mut().enumerate().find_map(
|(index, (block_id_opt, blob_id_opt, req))| { |(index, (block_id_opt, blob_id_opt, req))| {
let id_filter = |id: &&mut Id| -> bool { target_id == **id }; let id_filter = |id: &Id| -> bool { target_id == *id };
let block_matches = block_id_opt.as_ref().map(id_filter).unwrap_or(false);
let block_id = block_id_opt.as_mut().filter(id_filter); let blob_matches = blob_id_opt.as_ref().map(id_filter).unwrap_or(false);
let blob_id = blob_id_opt.as_mut().filter(id_filter); if !block_matches && !blob_matches {
block_id.or(blob_id).map(|id_ref| (index, id_ref, req)) return None;
}
Some((index, block_id_opt, blob_id_opt, req))
}, },
); );
let (index, request_id_ref, request_ref) = match lookup_components_opt { let (index, block_id_ref, blob_id_ref, request_ref) = match lookup_components_opt {
Some(req) => req, Some(req) => req,
None => { None => {
return debug!( return debug!(
self.log, self.log,
"Block processed for single block lookup not present" "Block component processed for single block lookup not present"
); );
} }
}; };
@@ -906,8 +884,115 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
ShouldRemoveLookup::True ShouldRemoveLookup::True
} }
AvailabilityProcessingStatus::MissingComponents(_, block_root) => { AvailabilityProcessingStatus::MissingComponents(_, block_root) => {
self.search_block(block_root, peer_id, cx); // if peer should have both, and missing components is received after we've
ShouldRemoveLookup::False // processed the opposite, then we can downscore.
let should_remove = match response_type {
ResponseType::Block => {
if request_ref.blob_request_state.component_processed {
match request_ref.processing_peer(ResponseType::Blob) {
Ok(PeerShouldHave::BlockAndBlobs(other_peer)) => {
cx.report_peer(
peer_id.to_peer_id(),
PeerAction::MidToleranceError,
"single_block_failure",
);
if let Some(blob_id_ref) = blob_id_ref {
// Try it again if possible.
retry_request_after_failure(
blob_id_ref,
request_ref,
ResponseType::Blob,
peer_id.as_peer_id(),
cx,
&self.log,
)
} else {
ShouldRemoveLookup::False
}
}
Ok(PeerShouldHave::Neither(other_peer)) => {
request_ref
.blob_request_state
.remove_peer_if_useless(&other_peer);
if let Some(blob_id_ref) = blob_id_ref {
// Try it again if possible.
retry_request_after_failure(
blob_id_ref,
request_ref,
ResponseType::Blob,
peer_id.as_peer_id(),
cx,
&self.log,
)
} else {
ShouldRemoveLookup::False
}
}
Err(()) => {
//TODO(sean) retry?
ShouldRemoveLookup::False
}
}
} else {
request_ref.block_request_state.component_processed = true;
ShouldRemoveLookup::False
}
}
ResponseType::Blob => {
if request_ref.block_request_state.component_processed {
match request_ref.processing_peer(ResponseType::Blob) {
Ok(PeerShouldHave::BlockAndBlobs(other_peer)) => {
cx.report_peer(
other_peer,
PeerAction::MidToleranceError,
"single_block_failure",
);
if let Some(blob_id_ref) = blob_id_ref {
// Try it again if possible.
retry_request_after_failure(
blob_id_ref,
request_ref,
ResponseType::Blob,
peer_id.as_peer_id(),
cx,
&self.log,
)
} else {
ShouldRemoveLookup::False
}
}
Ok(PeerShouldHave::Neither(other_peer)) => {
request_ref
.blob_request_state
.remove_peer_if_useless(&other_peer);
if let Some(blob_id_ref) = blob_id_ref {
// Try it again if possible.
retry_request_after_failure(
blob_id_ref,
request_ref,
ResponseType::Blob,
peer_id.as_peer_id(),
cx,
&self.log,
)
} else {
ShouldRemoveLookup::False
}
}
Err(()) => {
//TODO(sean) retry?
ShouldRemoveLookup::False
}
}
} else {
// retry block here?
request_ref.blob_request_state.component_processed = true;
ShouldRemoveLookup::False
}
}
};
should_remove
} }
}, },
BlockProcessingResult::Ignored => { BlockProcessingResult::Ignored => {
@@ -960,15 +1045,29 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
PeerAction::MidToleranceError, PeerAction::MidToleranceError,
"single_block_failure", "single_block_failure",
); );
// Try it again if possible. if let Some(blob_id_ref) = blob_id_ref {
retry_request_after_failure( // Try it again if possible.
request_id_ref, retry_request_after_failure(
request_ref, blob_id_ref,
response_type, request_ref,
peer_id.as_peer_id(), ResponseType::Blob,
cx, peer_id.as_peer_id(),
&self.log, cx,
) &self.log,
)
} else if let Some(block_id_ref) = block_id_ref {
// Try it again if possible.
retry_request_after_failure(
block_id_ref,
request_ref,
ResponseType::Block,
peer_id.as_peer_id(),
cx,
&self.log,
)
} else {
ShouldRemoveLookup::True
}
} }
} }
} }
@@ -1041,7 +1140,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
self.search_block(block_root, peer_id, cx); self.search_block(block_root, peer_id, cx);
} }
BlockProcessingResult::Err(BlockError::ParentUnknown(block)) => { BlockProcessingResult::Err(BlockError::ParentUnknown(block)) => {
parent_lookup.add_block_wrapper(block); parent_lookup.add_unknown_parent_block(block);
self.request_parent_block_and_blobs(parent_lookup, cx); self.request_parent_block_and_blobs(parent_lookup, cx);
} }
BlockProcessingResult::Ok(AvailabilityProcessingStatus::Imported(_)) BlockProcessingResult::Ok(AvailabilityProcessingStatus::Imported(_))
@@ -1057,10 +1156,22 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
); );
} }
}; };
let (chain_hash, blocks, hashes, block_request) = let (chain_hash, mut blocks, hashes, block_request) =
parent_lookup.parts_for_processing(); parent_lookup.parts_for_processing();
if let Some(child_block) =
self.single_block_lookups
.iter_mut()
.find_map(|(_, _, req)| {
if req.requested_block_root == chain_hash {
req.get_downloaded_block()
} else {
None
}
})
{
blocks.push(child_block);
};
let process_id = ChainSegmentProcessId::ParentLookup(chain_hash); let process_id = ChainSegmentProcessId::ParentLookup(chain_hash);
let work = WorkEvent::chain_segment(process_id, blocks); let work = WorkEvent::chain_segment(process_id, blocks);
match beacon_processor_send.try_send(work) { match beacon_processor_send.try_send(work) {
@@ -1090,13 +1201,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
); );
} }
BlockProcessingResult::Err(outcome) => { BlockProcessingResult::Err(outcome) => {
self.handle_invalid_block( self.handle_invalid_block(outcome, peer_id.to_peer_id(), cx, parent_lookup);
outcome,
peer_id.to_peer_id(),
cx,
response_type,
parent_lookup,
);
} }
BlockProcessingResult::Ignored => { BlockProcessingResult::Ignored => {
// Beacon processor signalled to ignore the block processing result. // Beacon processor signalled to ignore the block processing result.
@@ -1120,7 +1225,6 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
outcome: BlockError<<T as BeaconChainTypes>::EthSpec>, outcome: BlockError<<T as BeaconChainTypes>::EthSpec>,
peer_id: PeerId, peer_id: PeerId,
cx: &mut SyncNetworkContext<T>, cx: &mut SyncNetworkContext<T>,
response_type: ResponseType,
mut parent_lookup: ParentLookup<T>, mut parent_lookup: ParentLookup<T>,
) { ) {
// all else we consider the chain a failure and downvote the peer that sent // all else we consider the chain a failure and downvote the peer that sent
@@ -1135,16 +1239,11 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
// ambiguity. // ambiguity.
cx.report_peer(peer_id, PeerAction::MidToleranceError, "parent_request_err"); cx.report_peer(peer_id, PeerAction::MidToleranceError, "parent_request_err");
// Try again if possible // Try again if possible
match response_type { //TODO(sean) I don't see why we want to retry here, because this block is invalid and has
ResponseType::Block => { // passed block root validation, so querying again by block root will yield the same result.
parent_lookup.block_processing_failed(); parent_lookup.block_processing_failed();
self.request_parent_block(parent_lookup, cx); parent_lookup.blob_processing_failed();
} self.request_parent_block_and_blobs(parent_lookup, cx);
ResponseType::Blob => {
parent_lookup.blob_processing_failed();
self.request_parent_blob(parent_lookup, cx);
}
}
} }
pub fn parent_chain_processed( pub fn parent_chain_processed(
@@ -1262,6 +1361,10 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
process_type: BlockProcessType, process_type: BlockProcessType,
cx: &mut SyncNetworkContext<T>, cx: &mut SyncNetworkContext<T>,
) -> Result<(), ()> { ) -> Result<(), ()> {
let blob_count = blobs.iter().filter(|b| b.is_some()).count();
if blob_count == 0 {
return Ok(());
}
match cx.processor_channel_if_enabled() { match cx.processor_channel_if_enabled() {
Some(beacon_processor_send) => { Some(beacon_processor_send) => {
trace!(self.log, "Sending blobs for processing"; "block" => ?block_root, "process_type" => ?process_type); trace!(self.log, "Sending blobs for processing"; "block" => ?block_root, "process_type" => ?process_type);
@@ -1385,35 +1488,7 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
} }
} }
fn handle_availability_check_error<T: BeaconChainTypes>(
request_id_ref: &mut u32,
request_ref: &mut SingleBlockLookup<SINGLE_BLOCK_LOOKUP_MAX_ATTEMPTS, T>,
response_type: ResponseType,
peer_id: PeerId,
e: AvailabilityCheckError,
cx: &mut SyncNetworkContext<T>,
log: &Logger,
) -> ShouldRemoveLookup {
warn!(log, "Peer sent message in single block lookup causing failed availability check";
"root" => ?request_ref.requested_block_root,
"error" => ?e,
"peer_id" => %peer_id,
"response_type" => ?response_type
);
cx.report_peer(
peer_id,
PeerAction::MidToleranceError,
"single_block_failure",
);
retry_request_after_failure(
request_id_ref,
request_ref,
ResponseType::Blob,
&peer_id,
cx,
log,
)
}
fn handle_block_lookup_verify_error<T: BeaconChainTypes>( fn handle_block_lookup_verify_error<T: BeaconChainTypes>(
request_id_ref: &mut u32, request_id_ref: &mut u32,

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

@@ -1,6 +1,7 @@
use super::single_block_lookup::{LookupRequestError, LookupVerifyError, SingleBlockLookup}; use super::single_block_lookup::{LookupRequestError, LookupVerifyError, SingleBlockLookup};
use super::{DownloadedBlocks, PeerShouldHave, ResponseType}; use super::{DownloadedBlocks, PeerShouldHave, ResponseType};
use crate::sync::block_lookups::{single_block_lookup, RootBlobsTuple, RootBlockTuple}; use crate::sync::block_lookups::single_block_lookup::{State, UnknownParentComponents};
use crate::sync::block_lookups::{RootBlobsTuple, RootBlockTuple};
use crate::sync::{ use crate::sync::{
manager::{Id, SLOT_IMPORT_TOLERANCE}, manager::{Id, SLOT_IMPORT_TOLERANCE},
network_context::SyncNetworkContext, network_context::SyncNetworkContext,
@@ -32,8 +33,8 @@ pub(crate) struct ParentLookup<T: BeaconChainTypes> {
/// Request of the last parent. /// Request of the last parent.
pub current_parent_request: SingleBlockLookup<PARENT_FAIL_TOLERANCE, T>, pub current_parent_request: SingleBlockLookup<PARENT_FAIL_TOLERANCE, T>,
/// Id of the last parent request. /// Id of the last parent request.
current_parent_request_id: Option<Id>, pub current_parent_request_id: Option<Id>,
current_parent_blob_request_id: Option<Id>, pub current_parent_blob_request_id: Option<Id>,
} }
#[derive(Debug, PartialEq, Eq, IntoStaticStr)] #[derive(Debug, PartialEq, Eq, IntoStaticStr)]
@@ -62,6 +63,7 @@ pub enum RequestError {
NoPeers, NoPeers,
} }
#[derive(Debug)]
pub enum LookupDownloadStatus<T: EthSpec> { pub enum LookupDownloadStatus<T: EthSpec> {
Process(BlockWrapper<T>), Process(BlockWrapper<T>),
SearchBlock(Hash256), SearchBlock(Hash256),
@@ -93,7 +95,8 @@ impl<T: BeaconChainTypes> ParentLookup<T> {
peer_id: PeerShouldHave, peer_id: PeerShouldHave,
da_checker: Arc<DataAvailabilityChecker<T::EthSpec, T::SlotClock>>, da_checker: Arc<DataAvailabilityChecker<T::EthSpec, T::SlotClock>>,
) -> Self { ) -> Self {
let current_parent_request = SingleBlockLookup::new(parent_root, peer_id, da_checker); let current_parent_request =
SingleBlockLookup::new(parent_root, Some(<_>::default()), peer_id, da_checker);
Self { Self {
chain_hash: block_root, chain_hash: block_root,
@@ -165,39 +168,44 @@ impl<T: BeaconChainTypes> ParentLookup<T> {
.check_peer_disconnected(peer_id) .check_peer_disconnected(peer_id)
} }
pub fn add_block_wrapper(&mut self, block: BlockWrapper<T::EthSpec>) { pub fn add_unknown_parent_block(&mut self, block: BlockWrapper<T::EthSpec>) {
let next_parent = block.parent_root(); let next_parent = block.parent_root();
// Cache the block.
let current_root = self.current_parent_request.requested_block_root; let current_root = self.current_parent_request.requested_block_root;
self.downloaded_blocks.push((current_root, block)); self.downloaded_blocks.push((current_root, block));
// Update the block request.
self.current_parent_request.requested_block_root = next_parent; self.current_parent_request.requested_block_root = next_parent;
self.current_parent_request.block_request_state.state = State::AwaitingDownload;
self.current_parent_request.block_request_state.state =
single_block_lookup::State::AwaitingDownload;
self.current_parent_request.blob_request_state.state =
single_block_lookup::State::AwaitingDownload;
self.current_parent_request_id = None; self.current_parent_request_id = None;
// Update the blobs request.
self.current_parent_request.blob_request_state.state = State::AwaitingDownload;
self.current_parent_blob_request_id = None; self.current_parent_blob_request_id = None;
self.current_parent_request.downloaded_block = None;
self.current_parent_request.downloaded_blobs = <_>::default(); // Reset the unknown parent components.
self.current_parent_request.unknown_parent_components =
Some(UnknownParentComponents::default());
} }
pub fn add_block( pub fn add_current_request_block(
&mut self, &mut self,
block_root: Hash256,
block: Arc<SignedBeaconBlock<T::EthSpec>>, block: Arc<SignedBeaconBlock<T::EthSpec>>,
) -> LookupDownloadStatus<T::EthSpec> { ) {
// Cache the block.
self.current_parent_request.add_unknown_parent_block(block);
// Update the request.
self.current_parent_request_id = None; self.current_parent_request_id = None;
self.current_parent_request.add_block(block_root, block)
} }
pub fn add_blobs( pub fn add_current_request_blobs(&mut self, blobs: FixedBlobSidecarList<T::EthSpec>) {
&mut self, // Cache the blobs.
block_root: Hash256, self.current_parent_request.add_unknown_parent_blobs(blobs);
blobs: FixedBlobSidecarList<T::EthSpec>,
) -> LookupDownloadStatus<T::EthSpec> { // Update the request.
self.current_parent_blob_request_id = None; self.current_parent_blob_request_id = None;
self.current_parent_request.add_blobs(block_root, blobs)
} }
pub fn pending_block_response(&self, req_id: Id) -> bool { pub fn pending_block_response(&self, req_id: Id) -> bool {
@@ -258,7 +266,10 @@ impl<T: BeaconChainTypes> ParentLookup<T> {
self.current_parent_request self.current_parent_request
.block_request_state .block_request_state
.register_failure_processing(); .register_failure_processing();
self.current_parent_request.downloaded_block = None; self.current_parent_request
.unknown_parent_components
.as_mut()
.map(|components| components.downloaded_block = None);
self.current_parent_request_id = None; self.current_parent_request_id = None;
} }
@@ -266,8 +277,10 @@ impl<T: BeaconChainTypes> ParentLookup<T> {
self.current_parent_request self.current_parent_request
.blob_request_state .blob_request_state
.register_failure_processing(); .register_failure_processing();
//TODO(sean) can make this only clear the blobs that failed to process self.current_parent_request
self.current_parent_request.downloaded_blobs = <_>::default(); .unknown_parent_components
.as_mut()
.map(|components| components.downloaded_blobs = <_>::default());
self.current_parent_blob_request_id = None; self.current_parent_blob_request_id = None;
} }
@@ -351,15 +364,6 @@ impl<T: BeaconChainTypes> ParentLookup<T> {
.iter(), .iter(),
} }
} }
pub(crate) fn peer_source(
&self,
response_type: ResponseType,
peer_id: PeerId,
) -> Option<PeerShouldHave> {
self.current_parent_request
.peer_source(response_type, peer_id)
}
} }
impl From<LookupVerifyError> for ParentVerifyError { impl From<LookupVerifyError> for ParentVerifyError {

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

@@ -1,4 +1,3 @@
use crate::sync::block_lookups::parent_lookup::LookupDownloadStatus;
use crate::sync::block_lookups::{RootBlobsTuple, RootBlockTuple}; use crate::sync::block_lookups::{RootBlobsTuple, RootBlockTuple};
use beacon_chain::blob_verification::BlockWrapper; use beacon_chain::blob_verification::BlockWrapper;
use beacon_chain::data_availability_checker::DataAvailabilityChecker; use beacon_chain::data_availability_checker::DataAvailabilityChecker;
@@ -8,28 +7,55 @@ use lighthouse_network::{rpc::BlocksByRootRequest, PeerId};
use rand::seq::IteratorRandom; use rand::seq::IteratorRandom;
use ssz_types::VariableList; use ssz_types::VariableList;
use std::collections::HashSet; use std::collections::HashSet;
use std::ops::IndexMut;
use std::sync::Arc; use std::sync::Arc;
use store::Hash256; use store::Hash256;
use strum::IntoStaticStr; use strum::IntoStaticStr;
use types::blob_sidecar::{BlobIdentifier, FixedBlobSidecarList}; use types::blob_sidecar::{BlobIdentifier, FixedBlobSidecarList};
use types::{BlobSidecar, SignedBeaconBlock}; use types::{BlobSidecar, EthSpec, SignedBeaconBlock};
use super::{PeerShouldHave, ResponseType}; use super::{PeerShouldHave, ResponseType};
pub struct SingleBlockLookup<const MAX_ATTEMPTS: u8, T: BeaconChainTypes> { pub struct SingleBlockLookup<const MAX_ATTEMPTS: u8, T: BeaconChainTypes> {
pub requested_block_root: Hash256, pub requested_block_root: Hash256,
pub requested_ids: Vec<BlobIdentifier>, pub requested_ids: Vec<BlobIdentifier>,
pub downloaded_blobs: FixedBlobSidecarList<T::EthSpec>, /// Where we store blobs until we receive the stream terminator.
pub downloaded_block: Option<Arc<SignedBeaconBlock<T::EthSpec>>>, pub blob_download_queue: FixedBlobSidecarList<T::EthSpec>,
pub expected_num_blobs: Option<usize>,
pub block_request_state: SingleLookupRequestState<MAX_ATTEMPTS>, pub block_request_state: SingleLookupRequestState<MAX_ATTEMPTS>,
pub blob_request_state: SingleLookupRequestState<MAX_ATTEMPTS>, pub blob_request_state: SingleLookupRequestState<MAX_ATTEMPTS>,
pub da_checker: Arc<DataAvailabilityChecker<T::EthSpec, T::SlotClock>>, pub da_checker: Arc<DataAvailabilityChecker<T::EthSpec, T::SlotClock>>,
/// Only necessary for requests triggered by an `UnkownParent` because any
/// blocks or blobs without parents won't hit the data availability cache.
pub unknown_parent_components: Option<UnknownParentComponents<T::EthSpec>>,
} }
/// Object representing a single block lookup request. #[derive(Default)]
/// pub struct UnknownParentComponents<E: EthSpec> {
//previously assumed we would have a single block. Now we may have the block but not the blobs pub downloaded_block: Option<Arc<SignedBeaconBlock<E>>>,
pub downloaded_blobs: FixedBlobSidecarList<E>,
}
impl<E: EthSpec> UnknownParentComponents<E> {
pub fn add_unknown_parent_block(&mut self, block: Arc<SignedBeaconBlock<E>>) {
self.downloaded_block = Some(block);
}
pub fn add_unknown_parent_blobs(&mut self, blobs: FixedBlobSidecarList<E>) {
for (index, blob_opt) in self.downloaded_blobs.iter_mut().enumerate() {
if let Some(Some(downloaded_blob)) = blobs.get(index) {
*blob_opt = Some(downloaded_blob.clone());
}
}
}
pub fn downloaded_indices(&self) -> HashSet<usize> {
self.downloaded_blobs
.iter()
.enumerate()
.filter_map(|(i, blob_opt)| blob_opt.as_ref().map(|_| i))
.collect::<HashSet<_>>()
}
}
/// Object representing the state of a single block or blob lookup request.
#[derive(PartialEq, Eq, Debug)] #[derive(PartialEq, Eq, Debug)]
pub struct SingleLookupRequestState<const MAX_ATTEMPTS: u8> { pub struct SingleLookupRequestState<const MAX_ATTEMPTS: u8> {
/// State of this request. /// State of this request.
@@ -44,6 +70,7 @@ pub struct SingleLookupRequestState<const MAX_ATTEMPTS: u8> {
failed_processing: u8, failed_processing: u8,
/// How many times have we attempted to download this block or blob. /// How many times have we attempted to download this block or blob.
failed_downloading: u8, failed_downloading: u8,
pub component_processed: bool,
} }
#[derive(Debug, PartialEq, Eq)] #[derive(Debug, PartialEq, Eq)]
@@ -81,100 +108,87 @@ pub enum LookupRequestError {
impl<const MAX_ATTEMPTS: u8, T: BeaconChainTypes> SingleBlockLookup<MAX_ATTEMPTS, T> { impl<const MAX_ATTEMPTS: u8, T: BeaconChainTypes> SingleBlockLookup<MAX_ATTEMPTS, T> {
pub fn new( pub fn new(
requested_block_root: Hash256, requested_block_root: Hash256,
unknown_parent_components: Option<UnknownParentComponents<T::EthSpec>>,
peer_source: PeerShouldHave, peer_source: PeerShouldHave,
da_checker: Arc<DataAvailabilityChecker<T::EthSpec, T::SlotClock>>, da_checker: Arc<DataAvailabilityChecker<T::EthSpec, T::SlotClock>>,
) -> Self { ) -> Self {
Self { Self {
requested_block_root, requested_block_root,
requested_ids: <_>::default(), requested_ids: <_>::default(),
downloaded_block: None, blob_download_queue: <_>::default(),
downloaded_blobs: <_>::default(),
expected_num_blobs: None,
block_request_state: SingleLookupRequestState::new(peer_source), block_request_state: SingleLookupRequestState::new(peer_source),
blob_request_state: SingleLookupRequestState::new(peer_source), blob_request_state: SingleLookupRequestState::new(peer_source),
da_checker, da_checker,
unknown_parent_components,
} }
} }
pub fn update_blobs_request(&mut self) {
self.requested_ids = if let Some(components) = self.unknown_parent_components.as_ref() {
let blobs = components.downloaded_indices();
self.da_checker
.get_missing_blob_ids(
self.requested_block_root,
components.downloaded_block.as_ref(),
Some(blobs),
)
.unwrap_or_default()
} else {
self.da_checker
.get_missing_blob_ids_checking_cache(self.requested_block_root)
.unwrap_or_default()
};
}
pub fn get_downloaded_block(&mut self) -> Option<BlockWrapper<T::EthSpec>> { pub fn get_downloaded_block(&mut self) -> Option<BlockWrapper<T::EthSpec>> {
if self.requested_ids.is_empty() { self.unknown_parent_components
if let Some(block) = self.downloaded_block.take() { .as_mut()
return Some(BlockWrapper::BlockAndBlobs( .and_then(|components| {
block, let downloaded_block = components.downloaded_block.as_ref();
self.downloaded_blobs.clone(), let downloaded_indices = components.downloaded_indices();
)); let missing_ids = self.da_checker.get_missing_blob_ids(
} self.requested_block_root,
} downloaded_block,
None Some(downloaded_indices),
}
pub fn add_blob(
&mut self,
blob: Arc<BlobSidecar<T::EthSpec>>,
) -> Result<LookupDownloadStatus<T::EthSpec>, LookupVerifyError> {
let block_root = blob.block_root;
if let Some(blob_opt) = self.downloaded_blobs.get_mut(blob.index as usize) {
*blob_opt = Some(blob.clone());
if let Some(block) = self.downloaded_block.as_ref() {
let result = self.da_checker.wrap_block(
block_root,
block.clone(),
self.downloaded_blobs.clone(),
); );
Ok(LookupDownloadStatus::from(result)) let download_complete =
} else { missing_ids.map_or(true, |missing_ids| missing_ids.is_empty());
Ok(LookupDownloadStatus::SearchBlock(block_root)) if download_complete {
} let UnknownParentComponents {
downloaded_block,
downloaded_blobs,
} = components;
downloaded_block.as_ref().map(|block| {
BlockWrapper::BlockAndBlobs(
block.clone(),
std::mem::replace(downloaded_blobs, FixedBlobSidecarList::default()),
)
})
} else {
None
}
})
}
pub fn add_unknown_parent_block(&mut self, block: Arc<SignedBeaconBlock<T::EthSpec>>) {
if let Some(ref mut components) = self.unknown_parent_components.as_mut() {
components.add_unknown_parent_block(block)
} else { } else {
Err(LookupVerifyError::InvalidIndex(blob.index)) self.unknown_parent_components = Some(UnknownParentComponents {
downloaded_block: Some(block),
downloaded_blobs: FixedBlobSidecarList::default(),
})
} }
} }
pub fn add_blobs( pub fn add_unknown_parent_blobs(&mut self, blobs: FixedBlobSidecarList<T::EthSpec>) {
&mut self, if let Some(ref mut components) = self.unknown_parent_components.as_mut() {
block_root: Hash256, components.add_unknown_parent_blobs(blobs)
blobs: FixedBlobSidecarList<T::EthSpec>,
) -> LookupDownloadStatus<T::EthSpec> {
for (index, blob_opt) in self.downloaded_blobs.iter_mut().enumerate() {
if let Some(Some(downloaded_blob)) = blobs.get(index) {
*blob_opt = Some(downloaded_blob.clone());
}
}
if let Some(block) = self.downloaded_block.as_ref() {
self.da_checker
.wrap_block(block_root, block.clone(), blobs)
.into()
} else { } else {
LookupDownloadStatus::SearchBlock(block_root) self.unknown_parent_components = Some(UnknownParentComponents {
} downloaded_block: None,
} downloaded_blobs: blobs,
})
pub fn add_block(
&mut self,
block_root: Hash256,
block: Arc<SignedBeaconBlock<T::EthSpec>>,
) -> LookupDownloadStatus<T::EthSpec> {
self.downloaded_block = Some(block.clone());
self.da_checker
.wrap_block(block_root, block, self.downloaded_blobs.clone())
.into()
}
pub fn add_block_wrapper(
&mut self,
block_root: Hash256,
block: BlockWrapper<T::EthSpec>,
) -> LookupDownloadStatus<T::EthSpec> {
match block {
BlockWrapper::Block(block) => self.add_block(block_root, block),
BlockWrapper::BlockAndBlobs(block, blobs) => {
self.downloaded_block = Some(block);
self.add_blobs(block_root, blobs)
}
} }
} }
@@ -202,12 +216,6 @@ impl<const MAX_ATTEMPTS: u8, T: BeaconChainTypes> SingleBlockLookup<MAX_ATTEMPTS
self.block_request_state.register_failure_downloading(); self.block_request_state.register_failure_downloading();
Err(LookupVerifyError::RootMismatch) Err(LookupVerifyError::RootMismatch)
} else { } else {
self.expected_num_blobs =
block.message().body().blob_kzg_commitments().ok().map(
|commitments| {
std::cmp::min(self.requested_ids.len(), commitments.len())
},
);
// Return the block for processing. // Return the block for processing.
self.block_request_state.state = State::Processing { peer_id }; self.block_request_state.state = State::Processing { peer_id };
Ok(Some((block_root, block))) Ok(Some((block_root, block)))
@@ -218,6 +226,7 @@ impl<const MAX_ATTEMPTS: u8, T: BeaconChainTypes> SingleBlockLookup<MAX_ATTEMPTS
self.block_request_state.register_failure_downloading(); self.block_request_state.register_failure_downloading();
Err(LookupVerifyError::NoBlockReturned) Err(LookupVerifyError::NoBlockReturned)
} else { } else {
self.block_request_state.state = State::AwaitingDownload;
Err(LookupVerifyError::BenignFailure) Err(LookupVerifyError::BenignFailure)
} }
} }
@@ -258,27 +267,20 @@ impl<const MAX_ATTEMPTS: u8, T: BeaconChainTypes> SingleBlockLookup<MAX_ATTEMPTS
} else { } else {
// State should remain downloading until we receive the stream terminator. // State should remain downloading until we receive the stream terminator.
self.requested_ids.retain(|id| *id != received_id); self.requested_ids.retain(|id| *id != received_id);
if let Some(blob_opt) = self.downloaded_blobs.get_mut(blob.index as usize) { //TODO(sean) validate index here
*blob_opt = Some(blob); // EArr(LookupVerifyError::InvalidIndex(blob.index))
Ok(None) let blob_index = blob.index;
} else { *self.blob_download_queue.index_mut(blob_index as usize) = Some(blob);
Err(LookupVerifyError::InvalidIndex(blob.index)) Ok(None)
}
} }
} }
None => { None => {
let downloaded_expected_blobs = self.downloaded_expected_blobs();
if peer_source.should_have_blobs() && !downloaded_expected_blobs {
return Err(LookupVerifyError::NotEnoughBlobsReturned);
} else if !downloaded_expected_blobs {
return Err(LookupVerifyError::BenignFailure);
}
self.blob_request_state.state = State::Processing { self.blob_request_state.state = State::Processing {
peer_id: peer_source, peer_id: peer_source,
}; };
Ok(Some(( Ok(Some((
self.requested_block_root, self.requested_block_root,
self.downloaded_blobs.clone(), std::mem::replace(&mut self.blob_download_queue, <_>::default()),
))) )))
} }
}, },
@@ -297,24 +299,17 @@ impl<const MAX_ATTEMPTS: u8, T: BeaconChainTypes> SingleBlockLookup<MAX_ATTEMPTS
} }
} }
fn downloaded_expected_blobs(&self) -> bool {
let Some(expected_num_blobs) = self.expected_num_blobs else {
return true;
};
let downloaded_num_blobs = self
.downloaded_blobs
.iter()
.filter(|blob_opt| blob_opt.is_some())
.count();
downloaded_num_blobs == expected_num_blobs
}
pub fn request_block( pub fn request_block(
&mut self, &mut self,
) -> Result<Option<(PeerId, BlocksByRootRequest)>, LookupRequestError> { ) -> Result<Option<(PeerId, BlocksByRootRequest)>, LookupRequestError> {
if self.da_checker.has_block(&self.requested_block_root) || self.downloaded_block.is_some() let block_already_downloaded =
{ if let Some(components) = self.unknown_parent_components.as_ref() {
components.downloaded_block.is_some()
} else {
self.da_checker.has_block(&self.requested_block_root)
};
if block_already_downloaded {
return Ok(None); return Ok(None);
} }
@@ -365,10 +360,9 @@ impl<const MAX_ATTEMPTS: u8, T: BeaconChainTypes> SingleBlockLookup<MAX_ATTEMPTS
pub fn request_blobs( pub fn request_blobs(
&mut self, &mut self,
) -> Result<Option<(PeerId, BlobsByRootRequest)>, LookupRequestError> { ) -> Result<Option<(PeerId, BlobsByRootRequest)>, LookupRequestError> {
let missing_ids = self self.update_blobs_request();
.da_checker
.get_missing_blob_ids(&self.requested_block_root); if self.requested_ids.is_empty() {
if missing_ids.is_empty() || self.downloaded_block.is_some() {
return Ok(None); return Ok(None);
} }
@@ -388,11 +382,10 @@ impl<const MAX_ATTEMPTS: u8, T: BeaconChainTypes> SingleBlockLookup<MAX_ATTEMPTS
.choose(&mut rand::thread_rng()) .choose(&mut rand::thread_rng())
{ {
let request = BlobsByRootRequest { let request = BlobsByRootRequest {
blob_ids: VariableList::from(missing_ids.clone()), blob_ids: VariableList::from(self.requested_ids.clone()),
}; };
self.blob_request_state.used_peers.insert(peer_id); self.blob_request_state.used_peers.insert(peer_id);
let peer_source = PeerShouldHave::BlockAndBlobs(peer_id); let peer_source = PeerShouldHave::BlockAndBlobs(peer_id);
self.requested_ids = missing_ids;
self.blob_request_state.state = State::Downloading { self.blob_request_state.state = State::Downloading {
peer_id: peer_source, peer_id: peer_source,
}; };
@@ -404,11 +397,10 @@ impl<const MAX_ATTEMPTS: u8, T: BeaconChainTypes> SingleBlockLookup<MAX_ATTEMPTS
.choose(&mut rand::thread_rng()) .choose(&mut rand::thread_rng())
{ {
let request = BlobsByRootRequest { let request = BlobsByRootRequest {
blob_ids: VariableList::from(missing_ids.clone()), blob_ids: VariableList::from(self.requested_ids.clone()),
}; };
self.blob_request_state.used_peers.insert(peer_id); self.blob_request_state.used_peers.insert(peer_id);
let peer_source = PeerShouldHave::Neither(peer_id); let peer_source = PeerShouldHave::Neither(peer_id);
self.requested_ids = missing_ids;
self.blob_request_state.state = State::Downloading { self.blob_request_state.state = State::Downloading {
peer_id: peer_source, peer_id: peer_source,
}; };
@@ -446,30 +438,10 @@ impl<const MAX_ATTEMPTS: u8, T: BeaconChainTypes> SingleBlockLookup<MAX_ATTEMPTS
} }
} }
pub(crate) fn peer_source( pub fn downloading_peer(&self, response_type: ResponseType) -> Result<PeerShouldHave, ()> {
&self,
response_type: ResponseType,
peer_id: PeerId,
) -> Option<PeerShouldHave> {
match response_type { match response_type {
ResponseType::Block => { ResponseType::Block => self.block_request_state.peer(),
if self.block_request_state.available_peers.contains(&peer_id) { ResponseType::Blob => self.blob_request_state.peer(),
Some(PeerShouldHave::BlockAndBlobs(peer_id))
} else if self.block_request_state.potential_peers.contains(&peer_id) {
Some(PeerShouldHave::Neither(peer_id))
} else {
None
}
}
ResponseType::Blob => {
if self.blob_request_state.available_peers.contains(&peer_id) {
Some(PeerShouldHave::BlockAndBlobs(peer_id))
} else if self.blob_request_state.potential_peers.contains(&peer_id) {
Some(PeerShouldHave::Neither(peer_id))
} else {
None
}
}
} }
} }
} }
@@ -489,6 +461,7 @@ impl<const MAX_ATTEMPTS: u8> SingleLookupRequestState<MAX_ATTEMPTS> {
used_peers: HashSet::default(), used_peers: HashSet::default(),
failed_processing: 0, failed_processing: 0,
failed_downloading: 0, failed_downloading: 0,
component_processed: false,
} }
} }
@@ -543,6 +516,14 @@ impl<const MAX_ATTEMPTS: u8> SingleLookupRequestState<MAX_ATTEMPTS> {
} }
} }
pub fn peer(&self) -> Result<PeerShouldHave, ()> {
match &self.state {
State::Processing { peer_id } => Ok(*peer_id),
State::Downloading { peer_id } => Ok(*peer_id),
_ => Err(()),
}
}
pub fn remove_peer_if_useless(&mut self, peer_id: &PeerId) { pub fn remove_peer_if_useless(&mut self, peer_id: &PeerId) {
if !self.available_peers.is_empty() || self.potential_peers.len() > 1 { if !self.available_peers.is_empty() || self.potential_peers.len() > 1 {
self.potential_peers.remove(peer_id); self.potential_peers.remove(peer_id);
@@ -634,7 +615,8 @@ mod tests {
Duration::from_secs(spec.seconds_per_slot), Duration::from_secs(spec.seconds_per_slot),
); );
let da_checker = Arc::new(DataAvailabilityChecker::new(slot_clock, None, spec)); let da_checker = Arc::new(DataAvailabilityChecker::new(slot_clock, None, spec));
let mut sl = SingleBlockLookup::<4, T>::new(block.canonical_root(), peer_id, da_checker); let mut sl =
SingleBlockLookup::<4, T>::new(block.canonical_root(), None, peer_id, da_checker);
sl.request_block().unwrap(); sl.request_block().unwrap();
sl.verify_block(Some(block.into())).unwrap().unwrap(); sl.verify_block(Some(block.into())).unwrap().unwrap();
} }
@@ -653,8 +635,12 @@ mod tests {
let da_checker = Arc::new(DataAvailabilityChecker::new(slot_clock, None, spec)); let da_checker = Arc::new(DataAvailabilityChecker::new(slot_clock, None, spec));
let mut sl = let mut sl = SingleBlockLookup::<FAILURES, T>::new(
SingleBlockLookup::<FAILURES, T>::new(block.canonical_root(), peer_id, da_checker); block.canonical_root(),
None,
peer_id,
da_checker,
);
for _ in 1..FAILURES { for _ in 1..FAILURES {
sl.request_block().unwrap(); sl.request_block().unwrap();
sl.block_request_state.register_failure_downloading(); sl.block_request_state.register_failure_downloading();

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

@@ -16,7 +16,7 @@ use execution_layer::BlobsBundleV1;
pub use genesis::{interop_genesis_state, DEFAULT_ETH1_BLOCK_HASH}; pub use genesis::{interop_genesis_state, DEFAULT_ETH1_BLOCK_HASH};
use lighthouse_network::rpc::RPCResponseErrorCode; use lighthouse_network::rpc::RPCResponseErrorCode;
use lighthouse_network::{NetworkGlobals, Request}; use lighthouse_network::{NetworkGlobals, Request};
use slot_clock::TestingSlotClock; use slot_clock::{SlotClock, TestingSlotClock};
use std::time::Duration; use std::time::Duration;
use store::MemoryStore; use store::MemoryStore;
use tokio::sync::mpsc; use tokio::sync::mpsc;
@@ -52,6 +52,11 @@ impl TestRig {
.logger(log.clone()) .logger(log.clone())
.deterministic_keypairs(1) .deterministic_keypairs(1)
.fresh_ephemeral_store() .fresh_ephemeral_store()
.testing_slot_clock(TestingSlotClock::new(
Slot::new(0),
Duration::from_secs(0),
Duration::from_secs(12),
))
.build(); .build();
let chain = harness.chain.clone(); let chain = harness.chain.clone();
@@ -115,7 +120,7 @@ impl TestRig {
&self.harness.chain.kzg.as_ref().unwrap(), &self.harness.chain.kzg.as_ref().unwrap(),
) )
.unwrap(); .unwrap();
//TODO(sean) maybe we want to keep other random txs ?
payload.execution_payload.transactions = <_>::default(); payload.execution_payload.transactions = <_>::default();
for tx in Vec::from(transactions) { for tx in Vec::from(transactions) {
payload.execution_payload.transactions.push(tx).unwrap(); payload.execution_payload.transactions.push(tx).unwrap();
@@ -236,6 +241,14 @@ impl TestRig {
); );
} }
#[track_caller]
fn expect_empty_beacon_processor(&mut self) {
assert_eq!(
self.beacon_processor_rx.try_recv().expect_err("must err"),
mpsc::error::TryRecvError::Empty
);
}
#[track_caller] #[track_caller]
pub fn expect_penalty(&mut self) { pub fn expect_penalty(&mut self) {
match self.network_rx.try_recv() { match self.network_rx.try_recv() {
@@ -253,6 +266,22 @@ impl TestRig {
*block.message_mut().parent_root_mut() = parent_root; *block.message_mut().parent_root_mut() = parent_root;
block block
} }
pub fn block_with_parent_and_blobs(
&mut self,
parent_root: Hash256,
fork_name: ForkName,
num_blobs: NumBlobs,
) -> (SignedBeaconBlock<E>, Vec<BlobSidecar<E>>) {
let (mut block, mut blobs) = self.rand_block_and_blobs(fork_name, num_blobs);
*block.message_mut().parent_root_mut() = parent_root;
let block_root = block.canonical_root();
blobs.iter_mut().for_each(|blob| {
blob.block_parent_root = parent_root;
blob.block_root = block_root;
});
(block, blobs)
}
} }
#[test] #[test]
@@ -288,7 +317,7 @@ fn test_single_block_lookup_happy_path() {
// Send the stream termination. Peer should have not been penalized, and the request removed // Send the stream termination. Peer should have not been penalized, and the request removed
// after processing. // after processing.
bl.single_block_lookup_response(id, peer_id, None, D, &mut cx); bl.single_block_lookup_response(id, peer_id, None, D, &mut cx);
bl.single_block_processed( bl.single_block_component_processed(
id, id,
BlockProcessingResult::Ok(AvailabilityProcessingStatus::Imported(block_root)), BlockProcessingResult::Ok(AvailabilityProcessingStatus::Imported(block_root)),
response_type, response_type,
@@ -421,7 +450,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_component_processed(
id, id,
BlockError::ParentUnknown(block.into()).into(), BlockError::ParentUnknown(block.into()).into(),
response_type, response_type,
@@ -797,8 +826,11 @@ fn test_parent_lookup_too_many_processing_attempts_must_blacklist() {
} }
// Now fail processing a block in the parent request // Now fail processing a block in the parent request
for _ in 0..PROCESSING_FAILURES { for i in 0..PROCESSING_FAILURES {
let id = dbg!(rig.expect_parent_request(response_type)); let id = dbg!(rig.expect_parent_request(response_type));
if matches!(fork_name, ForkName::Deneb) && i != 0 {
let _ = rig.expect_parent_request(ResponseType::Blob);
}
// If we're in deneb, a blob request should have been triggered as well, // If we're in deneb, a blob request should have been triggered as well,
// we don't require a response because we're generateing 0-blob blocks in this test. // we don't require a response because we're generateing 0-blob blocks in this test.
assert!(!bl.failed_chains.contains(&block_root)); assert!(!bl.failed_chains.contains(&block_root));
@@ -939,7 +971,7 @@ fn test_single_block_lookup_ignored_response() {
// after processing. // after processing.
bl.single_block_lookup_response(id, peer_id, None, D, &mut cx); bl.single_block_lookup_response(id, peer_id, None, D, &mut cx);
// Send an Ignored response, the request should be dropped // Send an Ignored response, the request should be dropped
bl.single_block_processed(id, BlockProcessingResult::Ignored, response_type, &mut cx); bl.single_block_component_processed(id, BlockProcessingResult::Ignored, response_type, &mut cx);
rig.expect_empty_network(); rig.expect_empty_network();
assert_eq!(bl.single_block_lookups.len(), 0); assert_eq!(bl.single_block_lookups.len(), 0);
} }
@@ -1100,7 +1132,13 @@ fn test_same_chain_race_condition() {
mod deneb_only { mod deneb_only {
use super::*; use super::*;
use beacon_chain::blob_verification::BlobError; use beacon_chain::blob_verification::{BlobError, MaybeAvailableBlock};
use beacon_chain::data_availability_checker::AvailabilityPendingBlock;
use beacon_chain::ExecutedBlock::AvailabilityPending;
use beacon_chain::IntoExecutionPendingBlock;
use beacon_chain::PayloadVerificationOutcome;
use beacon_chain::{AvailabilityPendingExecutedBlock, NotifyExecutionLayer};
use std::ops::IndexMut;
use std::str::FromStr; use std::str::FromStr;
struct DenebTester { struct DenebTester {
@@ -1109,10 +1147,12 @@ mod deneb_only {
rig: TestRig, rig: TestRig,
block: Option<Arc<SignedBeaconBlock<E>>>, block: Option<Arc<SignedBeaconBlock<E>>>,
blobs: Vec<Arc<BlobSidecar<E>>>, blobs: Vec<Arc<BlobSidecar<E>>>,
parent_block: Option<Arc<SignedBeaconBlock<E>>>,
parent_blobs: Vec<Arc<BlobSidecar<E>>>,
peer_id: PeerId, peer_id: PeerId,
block_req_id: Option<u32>, block_req_id: Option<u32>,
parent_block_req_id: Option<u32>, parent_block_req_id: Option<u32>,
blob_req_id: u32, blob_req_id: Option<u32>,
parent_blob_req_id: Option<u32>, parent_blob_req_id: Option<u32>,
slot: Slot, slot: Slot,
block_root: Hash256, block_root: Hash256,
@@ -1137,11 +1177,15 @@ mod deneb_only {
); );
let (block, blobs) = rig.rand_block_and_blobs(fork_name, NumBlobs::Random); let (block, blobs) = rig.rand_block_and_blobs(fork_name, NumBlobs::Random);
let block = Arc::new(block); let block = Arc::new(block);
let blobs = blobs.into_iter().map(Arc::new).collect::<Vec<_>>(); let slot = block.slot();
let mut block_root = block.canonical_root();
let mut block = Some(block);
let mut blobs = blobs.into_iter().map(Arc::new).collect::<Vec<_>>();
let mut parent_block = None;
let mut parent_blobs = vec![];
let peer_id = PeerId::random(); let peer_id = PeerId::random();
let slot = block.slot();
let block_root = block.canonical_root();
// Trigger the request // Trigger the request
let (block_req_id, blob_req_id, parent_block_req_id, parent_blob_req_id) = let (block_req_id, blob_req_id, parent_block_req_id, parent_blob_req_id) =
@@ -1154,38 +1198,79 @@ mod deneb_only {
); );
let block_req_id = rig.expect_block_request(ResponseType::Block); let block_req_id = rig.expect_block_request(ResponseType::Block);
let blob_req_id = rig.expect_block_request(ResponseType::Blob); let blob_req_id = rig.expect_block_request(ResponseType::Blob);
(Some(block_req_id), blob_req_id, None, None) (Some(block_req_id), Some(blob_req_id), None, None)
} }
RequestTrigger::GossipUnknownParentBlock => { RequestTrigger::GossipUnknownParentBlock => {
bl.search_current_unknown_parent( let (child_block, child_blobs) = rig.block_with_parent_and_blobs(
block_root, block_root,
BlockWrapper::Block(block.clone()), get_fork_name(),
NumBlobs::Random,
);
parent_block = Some(Arc::new(child_block));
parent_blobs = child_blobs.into_iter().map(Arc::new).collect::<Vec<_>>();
std::mem::swap(&mut parent_block, &mut block);
std::mem::swap(&mut parent_blobs, &mut blobs);
let child_block = block.as_ref().expect("block").clone();
let child_root = child_block.canonical_root();
let parent_root = block_root;
block_root = child_root;
bl.search_current_unknown_parent_block_and_blobs(
child_root,
Some(child_block),
None,
peer_id, peer_id,
&mut cx, &mut cx,
); );
let blob_req_id = rig.expect_block_request(ResponseType::Blob); let blob_req_id = rig.expect_block_request(ResponseType::Blob);
rig.expect_empty_network(); // expect no block request rig.expect_empty_network(); // expect no block request
bl.search_parent(slot, block_root, block.parent_root(), peer_id, &mut cx); bl.search_parent(slot, child_root, parent_root, peer_id, &mut cx);
let parent_block_req_id = rig.expect_parent_request(ResponseType::Block); let parent_block_req_id = rig.expect_parent_request(ResponseType::Block);
let parent_blob_req_id = rig.expect_parent_request(ResponseType::Blob); let parent_blob_req_id = rig.expect_parent_request(ResponseType::Blob);
( (
None, None,
blob_req_id, Some(blob_req_id),
Some(parent_block_req_id), Some(parent_block_req_id),
Some(parent_blob_req_id), Some(parent_blob_req_id),
) )
} }
RequestTrigger::GossipUnknownParentBlob => { RequestTrigger::GossipUnknownParentBlob => {
let blob = blobs.first().cloned().unwrap(); let (child_block, child_blobs) = rig.block_with_parent_and_blobs(
bl.search_current_unknown_blob_parent(blob, peer_id, &mut cx); block_root,
get_fork_name(),
NumBlobs::Random,
);
parent_block = Some(Arc::new(child_block));
parent_blobs = child_blobs.into_iter().map(Arc::new).collect::<Vec<_>>();
std::mem::swap(&mut parent_block, &mut block);
std::mem::swap(&mut parent_blobs, &mut blobs);
let child_blob = blobs.first().cloned().unwrap();
let parent_root = block_root;
let child_root = child_blob.block_root;
block_root = child_root;
let mut blobs = FixedBlobSidecarList::default();
*blobs.index_mut(0) = Some(child_blob);
bl.search_current_unknown_parent_block_and_blobs(
child_root,
None,
Some(blobs),
peer_id,
&mut cx,
);
let block_req_id = rig.expect_block_request(ResponseType::Block); let block_req_id = rig.expect_block_request(ResponseType::Block);
let blob_req_id = rig.expect_block_request(ResponseType::Blob); let blobs_req_id = rig.expect_block_request(ResponseType::Blob);
bl.search_parent(slot, block_root, block.parent_root(), peer_id, &mut cx); rig.expect_empty_network(); // expect no block request
bl.search_parent(slot, child_root, parent_root, peer_id, &mut cx);
let parent_block_req_id = rig.expect_parent_request(ResponseType::Block); let parent_block_req_id = rig.expect_parent_request(ResponseType::Block);
let parent_blob_req_id = rig.expect_parent_request(ResponseType::Blob); let parent_blob_req_id = rig.expect_parent_request(ResponseType::Blob);
( (
Some(block_req_id), Some(block_req_id),
blob_req_id, Some(blobs_req_id),
Some(parent_block_req_id), Some(parent_block_req_id),
Some(parent_blob_req_id), Some(parent_blob_req_id),
) )
@@ -1194,7 +1279,7 @@ mod deneb_only {
bl.search_block(block_root, PeerShouldHave::Neither(peer_id), &mut cx); bl.search_block(block_root, PeerShouldHave::Neither(peer_id), &mut cx);
let block_req_id = rig.expect_block_request(ResponseType::Block); let block_req_id = rig.expect_block_request(ResponseType::Block);
let blob_req_id = rig.expect_block_request(ResponseType::Blob); let blob_req_id = rig.expect_block_request(ResponseType::Blob);
(Some(block_req_id), blob_req_id, None, None) (Some(block_req_id), Some(blob_req_id), None, None)
} }
}; };
@@ -1202,8 +1287,10 @@ mod deneb_only {
bl, bl,
cx, cx,
rig, rig,
block: Some(block), block,
blobs, blobs,
parent_block,
parent_blobs,
peer_id, peer_id,
block_req_id, block_req_id,
parent_block_req_id, parent_block_req_id,
@@ -1214,6 +1301,53 @@ mod deneb_only {
}) })
} }
fn parent_block_response(mut self) -> Self {
self.rig.expect_empty_network();
self.bl.parent_lookup_response(
self.parent_block_req_id.expect("parent request id"),
self.peer_id,
self.parent_block.clone(),
D,
&mut self.cx,
);
assert_eq!(self.bl.parent_lookups.len(), 1);
self
}
fn parent_blob_response(mut self) -> Self {
for blob in &self.parent_blobs {
dbg!("sendingblob");
self.bl.parent_lookup_blob_response(
self.parent_blob_req_id.expect("parent blob request id"),
self.peer_id,
Some(blob.clone()),
D,
&mut self.cx,
);
assert_eq!(self.bl.parent_lookups.len(), 1);
}
dbg!("sending stream terminator");
self.bl.parent_lookup_blob_response(
self.parent_blob_req_id.expect("blob request id"),
self.peer_id,
None,
D,
&mut self.cx,
);
self
}
fn block_response_triggering_process(mut self) -> Self {
let mut me = self.block_response();
me.rig.expect_block_process(ResponseType::Block);
// The request should still be active.
assert_eq!(me.bl.single_block_lookups.len(), 1);
me
}
fn block_response(mut self) -> Self { fn block_response(mut self) -> Self {
// 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.
@@ -1225,7 +1359,6 @@ mod deneb_only {
&mut self.cx, &mut self.cx,
); );
self.rig.expect_empty_network(); self.rig.expect_empty_network();
self.rig.expect_block_process(ResponseType::Block);
// The request should still be active. // The request should still be active.
assert_eq!(self.bl.single_block_lookups.len(), 1); assert_eq!(self.bl.single_block_lookups.len(), 1);
@@ -1235,7 +1368,7 @@ mod deneb_only {
fn blobs_response(mut self) -> Self { fn blobs_response(mut self) -> Self {
for blob in &self.blobs { for blob in &self.blobs {
self.bl.single_blob_lookup_response( self.bl.single_blob_lookup_response(
self.blob_req_id, self.blob_req_id.expect("blob request id"),
self.peer_id, self.peer_id,
Some(blob.clone()), Some(blob.clone()),
D, D,
@@ -1243,9 +1376,8 @@ mod deneb_only {
); );
assert_eq!(self.bl.single_block_lookups.len(), 1); assert_eq!(self.bl.single_block_lookups.len(), 1);
} }
// Send the blob stream termination. Peer should have not been penalized.
self.bl.single_blob_lookup_response( self.bl.single_blob_lookup_response(
self.blob_req_id, self.blob_req_id.expect("blob request id"),
self.peer_id, self.peer_id,
None, None,
D, D,
@@ -1256,7 +1388,14 @@ mod deneb_only {
fn blobs_response_was_valid(mut self) -> Self { fn blobs_response_was_valid(mut self) -> Self {
self.rig.expect_empty_network(); self.rig.expect_empty_network();
self.rig.expect_block_process(ResponseType::Blob); if self.blobs.len() > 0 {
self.rig.expect_block_process(ResponseType::Blob);
}
self
}
fn expect_empty_beacon_processor(mut self) -> Self {
self.rig.expect_empty_beacon_processor();
self self
} }
@@ -1273,7 +1412,29 @@ mod deneb_only {
fn empty_blobs_response(mut self) -> Self { fn empty_blobs_response(mut self) -> Self {
self.bl.single_blob_lookup_response( self.bl.single_blob_lookup_response(
self.blob_req_id, self.blob_req_id.expect("blob request id"),
self.peer_id,
None,
D,
&mut self.cx,
);
self
}
fn empty_parent_block_response(mut self) -> Self {
self.bl.parent_lookup_response(
self.parent_block_req_id.expect("block request id"),
self.peer_id,
None,
D,
&mut self.cx,
);
self
}
fn empty_parent_blobs_response(mut self) -> Self {
self.bl.parent_lookup_blob_response(
self.parent_blob_req_id.expect("blob request id"),
self.peer_id, self.peer_id,
None, None,
D, D,
@@ -1285,7 +1446,7 @@ mod deneb_only {
fn block_imported(mut self) -> Self { fn block_imported(mut self) -> Self {
// Missing blobs should be the request is not removed, the outstanding blobs request should // Missing blobs should be the request is not removed, the outstanding blobs request should
// mean we do not send a new request. // mean we do not send a new request.
self.bl.single_block_processed( self.bl.single_block_component_processed(
self.block_req_id.expect("block request id"), self.block_req_id.expect("block request id"),
BlockProcessingResult::Ok(AvailabilityProcessingStatus::Imported(self.block_root)), BlockProcessingResult::Ok(AvailabilityProcessingStatus::Imported(self.block_root)),
ResponseType::Block, ResponseType::Block,
@@ -1296,8 +1457,44 @@ mod deneb_only {
self self
} }
fn parent_block_imported(mut self) -> Self {
self.bl.parent_block_processed(
self.block_root,
BlockProcessingResult::Ok(AvailabilityProcessingStatus::Imported(self.block_root)),
ResponseType::Block,
&mut self.cx,
);
self.rig.expect_empty_network();
assert_eq!(self.bl.parent_lookups.len(), 0);
self
}
fn parent_block_unknown_parent(mut self) -> Self {
self.bl.parent_block_processed(
self.block_root,
BlockProcessingResult::Err(BlockError::ParentUnknown(BlockWrapper::Block(
self.parent_block.clone().expect("parent block"),
))),
ResponseType::Block,
&mut self.cx,
);
assert_eq!(self.bl.parent_lookups.len(), 1);
self
}
fn invalid_parent_processed(mut self) -> Self {
self.bl.parent_block_processed(
self.block_root,
BlockProcessingResult::Err(BlockError::ProposalSignatureInvalid),
ResponseType::Block,
&mut self.cx,
);
assert_eq!(self.bl.parent_lookups.len(), 1);
self
}
fn invalid_block_processed(mut self) -> Self { fn invalid_block_processed(mut self) -> Self {
self.bl.single_block_processed( self.bl.single_block_component_processed(
self.block_req_id.expect("block request id"), self.block_req_id.expect("block request id"),
BlockProcessingResult::Err(BlockError::ProposalSignatureInvalid), BlockProcessingResult::Err(BlockError::ProposalSignatureInvalid),
ResponseType::Block, ResponseType::Block,
@@ -1308,8 +1505,8 @@ mod deneb_only {
} }
fn invalid_blob_processed(mut self) -> Self { fn invalid_blob_processed(mut self) -> Self {
self.bl.single_block_processed( self.bl.single_block_component_processed(
self.blob_req_id, self.blob_req_id.expect("blob request id"),
BlockProcessingResult::Err(BlockError::BlobValidation( BlockProcessingResult::Err(BlockError::BlobValidation(
BlobError::ProposerSignatureInvalid, BlobError::ProposerSignatureInvalid,
)), )),
@@ -1321,7 +1518,7 @@ mod deneb_only {
} }
fn missing_components_from_block_request(mut self) -> Self { fn missing_components_from_block_request(mut self) -> Self {
self.bl.single_block_processed( self.bl.single_block_component_processed(
self.block_req_id.expect("block request id"), self.block_req_id.expect("block request id"),
BlockProcessingResult::Ok(AvailabilityProcessingStatus::MissingComponents( BlockProcessingResult::Ok(AvailabilityProcessingStatus::MissingComponents(
self.slot, self.slot,
@@ -1335,8 +1532,8 @@ mod deneb_only {
} }
fn missing_components_from_blob_request(mut self) -> Self { fn missing_components_from_blob_request(mut self) -> Self {
self.bl.single_block_processed( self.bl.single_block_component_processed(
self.blob_req_id, self.blob_req_id.expect("blob request id"),
BlockProcessingResult::Ok(AvailabilityProcessingStatus::MissingComponents( BlockProcessingResult::Ok(AvailabilityProcessingStatus::MissingComponents(
self.slot, self.slot,
self.block_root, self.block_root,
@@ -1357,11 +1554,23 @@ mod deneb_only {
self self
} }
fn expect_block_request(mut self) -> Self { fn expect_block_request(mut self) -> Self {
self.rig.expect_block_request(ResponseType::Block); let id = self.rig.expect_block_request(ResponseType::Block);
self.block_req_id = Some(id);
self self
} }
fn expect_blobs_request(mut self) -> Self { fn expect_blobs_request(mut self) -> Self {
self.rig.expect_block_request(ResponseType::Blob); let id = self.rig.expect_block_request(ResponseType::Blob);
self.blob_req_id = Some(id);
self
}
fn expect_parent_block_request(mut self) -> Self {
let id = self.rig.expect_parent_request(ResponseType::Block);
self.parent_block_req_id = Some(id);
self
}
fn expect_parent_blobs_request(mut self) -> Self {
let id = self.rig.expect_parent_request(ResponseType::Blob);
self.parent_blob_req_id = Some(id);
self self
} }
fn expect_no_blobs_request(mut self) -> Self { fn expect_no_blobs_request(mut self) -> Self {
@@ -1381,6 +1590,14 @@ mod deneb_only {
self.blobs.push(first_blob); self.blobs.push(first_blob);
self self
} }
fn expect_parent_chain_process(mut self) -> Self {
self.rig.expect_parent_chain_process();
self
}
fn expect_block_process(mut self) -> Self {
self.rig.expect_block_process(ResponseType::Block);
self
}
} }
fn get_fork_name() -> ForkName { fn get_fork_name() -> ForkName {
@@ -1403,7 +1620,7 @@ mod deneb_only {
}; };
tester tester
.block_response() .block_response_triggering_process()
.blobs_response() .blobs_response()
.blobs_response_was_valid() .blobs_response_was_valid()
.block_imported(); .block_imported();
@@ -1418,7 +1635,7 @@ mod deneb_only {
tester tester
.blobs_response() .blobs_response()
.blobs_response_was_valid() .blobs_response_was_valid()
.block_response() .block_response_triggering_process()
.block_imported(); .block_imported();
} }
@@ -1434,9 +1651,12 @@ mod deneb_only {
.expect_block_request() .expect_block_request()
.expect_no_blobs_request() .expect_no_blobs_request()
.empty_blobs_response() .empty_blobs_response()
.expect_empty_beacon_processor()
.expect_no_penalty() .expect_no_penalty()
.expect_no_block_request() .expect_no_block_request()
.expect_no_blobs_request(); .expect_no_blobs_request()
.block_response_triggering_process()
.missing_components_from_block_request();
} }
#[test] #[test]
@@ -1446,9 +1666,10 @@ mod deneb_only {
}; };
tester tester
.block_response() .block_response_triggering_process()
.missing_components_from_block_request() .missing_components_from_block_request()
.empty_blobs_response() .empty_blobs_response()
.missing_components_from_blob_request()
.expect_penalty() .expect_penalty()
.expect_blobs_request() .expect_blobs_request()
.expect_no_block_request(); .expect_no_block_request();
@@ -1480,7 +1701,7 @@ mod deneb_only {
}; };
tester tester
.block_response() .block_response_triggering_process()
.invalid_block_processed() .invalid_block_processed()
.expect_penalty() .expect_penalty()
.expect_block_request() .expect_block_request()
@@ -1499,7 +1720,7 @@ mod deneb_only {
}; };
tester tester
.block_response() .block_response_triggering_process()
.missing_components_from_block_request() .missing_components_from_block_request()
.blobs_response() .blobs_response()
.invalid_blob_processed() .invalid_blob_processed()
@@ -1515,9 +1736,11 @@ mod deneb_only {
}; };
tester tester
.block_response() .block_response_triggering_process()
.missing_components_from_block_request()
.invalidate_blobs_too_few() .invalidate_blobs_too_few()
.blobs_response() .blobs_response()
.missing_components_from_blob_request()
.expect_penalty() .expect_penalty()
.expect_blobs_request() .expect_blobs_request()
.expect_no_block_request(); .expect_no_block_request();
@@ -1530,7 +1753,7 @@ mod deneb_only {
}; };
tester tester
.block_response() .block_response_triggering_process()
.invalidate_blobs_too_many() .invalidate_blobs_too_many()
.blobs_response() .blobs_response()
.expect_penalty() .expect_penalty()
@@ -1546,12 +1769,11 @@ mod deneb_only {
tester tester
.invalidate_blobs_too_few() .invalidate_blobs_too_few()
.blobs_response() .blobs_response()
// No way to know if the response was valid before we've seen the block.
.blobs_response_was_valid() .blobs_response_was_valid()
.expect_no_penalty() .expect_no_penalty()
.expect_no_blobs_request() .expect_no_blobs_request()
.expect_no_block_request() .expect_no_block_request()
.block_response(); .block_response_triggering_process();
} }
#[test] #[test]
@@ -1566,7 +1788,7 @@ mod deneb_only {
.expect_penalty() .expect_penalty()
.expect_blobs_request() .expect_blobs_request()
.expect_no_block_request() .expect_no_block_request()
.block_response(); .block_response_triggering_process();
} }
#[test] #[test]
@@ -1576,7 +1798,7 @@ mod deneb_only {
}; };
tester tester
.block_response() .block_response_triggering_process()
.blobs_response() .blobs_response()
.blobs_response_was_valid() .blobs_response_was_valid()
.block_imported(); .block_imported();
@@ -1591,7 +1813,7 @@ mod deneb_only {
tester tester
.blobs_response() .blobs_response()
.blobs_response_was_valid() .blobs_response_was_valid()
.block_response() .block_response_triggering_process()
.block_imported(); .block_imported();
} }
@@ -1609,7 +1831,9 @@ mod deneb_only {
.empty_blobs_response() .empty_blobs_response()
.expect_no_penalty() .expect_no_penalty()
.expect_no_block_request() .expect_no_block_request()
.expect_no_blobs_request(); .expect_no_blobs_request()
.block_response_triggering_process()
.missing_components_from_block_request();
} }
#[test] #[test]
@@ -1619,9 +1843,10 @@ mod deneb_only {
}; };
tester tester
.block_response() .block_response_triggering_process()
.missing_components_from_block_request() .missing_components_from_block_request()
.empty_blobs_response() .empty_blobs_response()
.missing_components_from_blob_request()
.expect_blobs_request() .expect_blobs_request()
.expect_no_penalty() .expect_no_penalty()
.expect_no_block_request(); .expect_no_block_request();
@@ -1642,7 +1867,6 @@ mod deneb_only {
.missing_components_from_blob_request() .missing_components_from_blob_request()
.empty_block_response() .empty_block_response()
.expect_block_request() .expect_block_request()
// No penalty because the blob was seen over gossip
.expect_no_penalty() .expect_no_penalty()
.expect_no_blobs_request(); .expect_no_blobs_request();
} }
@@ -1654,7 +1878,7 @@ mod deneb_only {
}; };
tester tester
.block_response() .block_response_triggering_process()
.invalid_block_processed() .invalid_block_processed()
.expect_penalty() .expect_penalty()
.expect_block_request() .expect_block_request()
@@ -1673,7 +1897,7 @@ mod deneb_only {
}; };
tester tester
.block_response() .block_response_triggering_process()
.missing_components_from_block_request() .missing_components_from_block_request()
.blobs_response() .blobs_response()
.invalid_blob_processed() .invalid_blob_processed()
@@ -1689,9 +1913,11 @@ mod deneb_only {
}; };
tester tester
.block_response() .block_response_triggering_process()
.missing_components_from_block_request()
.invalidate_blobs_too_few() .invalidate_blobs_too_few()
.blobs_response() .blobs_response()
.missing_components_from_blob_request()
.expect_blobs_request() .expect_blobs_request()
.expect_no_penalty() .expect_no_penalty()
.expect_no_block_request(); .expect_no_block_request();
@@ -1704,7 +1930,7 @@ mod deneb_only {
}; };
tester tester
.block_response() .block_response_triggering_process()
.invalidate_blobs_too_many() .invalidate_blobs_too_many()
.blobs_response() .blobs_response()
.expect_penalty() .expect_penalty()
@@ -1720,12 +1946,14 @@ mod deneb_only {
tester tester
.invalidate_blobs_too_few() .invalidate_blobs_too_few()
.blobs_response() .blobs_response()
// No way to know if the response was valid before we've seen the block.
.blobs_response_was_valid() .blobs_response_was_valid()
.missing_components_from_blob_request()
.expect_no_penalty() .expect_no_penalty()
.expect_no_blobs_request() .expect_no_blobs_request()
.expect_no_block_request() .expect_no_block_request()
.block_response(); .block_response_triggering_process()
.missing_components_from_block_request()
.expect_blobs_request();
} }
#[test] #[test]
@@ -1740,6 +1968,228 @@ mod deneb_only {
.expect_penalty() .expect_penalty()
.expect_blobs_request() .expect_blobs_request()
.expect_no_block_request() .expect_no_block_request()
.block_response(); .block_response_triggering_process();
}
#[test]
fn parent_block_unknown_parent() {
let Some(tester) = DenebTester::new(RequestTrigger::GossipUnknownParentBlock) else {
return;
};
tester
.blobs_response()
.expect_empty_beacon_processor()
.parent_block_response()
.parent_blob_response()
.expect_block_process()
.parent_block_unknown_parent()
.expect_parent_block_request()
.expect_parent_blobs_request()
.expect_empty_beacon_processor();
}
#[test]
fn parent_block_invalid_parent() {
let Some(tester) = DenebTester::new(RequestTrigger::GossipUnknownParentBlock) else {
return;
};
tester
.blobs_response()
.expect_empty_beacon_processor()
.parent_block_response()
.parent_blob_response()
.expect_block_process()
.invalid_parent_processed()
.expect_penalty()
.expect_parent_block_request()
.expect_parent_blobs_request()
.expect_empty_beacon_processor();
}
#[test]
fn parent_block_and_blob_lookup_parent_returned_first() {
let Some(tester) = DenebTester::new(RequestTrigger::GossipUnknownParentBlock) else {
return;
};
tester
.parent_block_response()
.parent_blob_response()
.expect_block_process()
.parent_block_imported()
.blobs_response()
.expect_parent_chain_process();
}
#[test]
fn parent_block_and_blob_lookup_child_returned_first() {
let Some(tester) = DenebTester::new(RequestTrigger::GossipUnknownParentBlock) else {
return;
};
tester
.blobs_response()
.expect_no_penalty()
.expect_no_block_request()
.expect_no_blobs_request()
.parent_block_response()
.parent_blob_response()
.expect_block_process()
.parent_block_imported()
.expect_parent_chain_process();
}
#[test]
fn empty_parent_block_then_parent_blob() {
let Some(tester) = DenebTester::new(RequestTrigger::GossipUnknownParentBlock) else {
return;
};
tester
.empty_parent_block_response()
.expect_penalty()
.expect_parent_block_request()
.expect_no_blobs_request()
.parent_blob_response()
.expect_empty_beacon_processor()
.parent_block_response()
.expect_block_process()
.parent_block_imported()
.blobs_response()
.expect_parent_chain_process();
}
#[test]
fn empty_parent_blobs_then_parent_block() {
let Some(tester) = DenebTester::new(RequestTrigger::GossipUnknownParentBlock) else {
return;
};
tester
.blobs_response()
.empty_parent_blobs_response()
.expect_no_penalty()
.expect_no_blobs_request()
.expect_no_block_request()
.parent_block_response()
.expect_penalty()
.expect_parent_blobs_request()
.parent_blob_response()
.expect_block_process()
.parent_block_imported()
.expect_parent_chain_process();
}
#[test]
fn parent_blob_unknown_parent() {
let Some(tester) = DenebTester::new(RequestTrigger::GossipUnknownParentBlob) else {
return;
};
tester
.block_response()
.expect_empty_beacon_processor()
.parent_block_response()
.parent_blob_response()
.expect_block_process()
.parent_block_unknown_parent()
.expect_parent_block_request()
.expect_parent_blobs_request()
.expect_empty_beacon_processor();
}
#[test]
fn parent_blob_invalid_parent() {
let Some(tester) = DenebTester::new(RequestTrigger::GossipUnknownParentBlob) else {
return;
};
tester
.block_response()
.expect_empty_beacon_processor()
.parent_block_response()
.parent_blob_response()
.expect_block_process()
.invalid_parent_processed()
.expect_penalty()
.expect_parent_block_request()
.expect_parent_blobs_request()
.expect_empty_beacon_processor();
}
#[test]
fn parent_block_and_blob_lookup_parent_returned_first_blob_trigger() {
let Some(tester) = DenebTester::new(RequestTrigger::GossipUnknownParentBlob) else {
return;
};
tester
.parent_block_response()
.parent_blob_response()
.expect_block_process()
.parent_block_imported()
.block_response()
.expect_parent_chain_process();
}
#[test]
fn parent_block_and_blob_lookup_child_returned_first_blob_trigger() {
let Some(tester) = DenebTester::new(RequestTrigger::GossipUnknownParentBlob) else {
return;
};
tester
.block_response()
.expect_no_penalty()
.expect_no_block_request()
.expect_no_blobs_request()
.parent_block_response()
.parent_blob_response()
.expect_block_process()
.parent_block_imported()
.expect_parent_chain_process();
}
#[test]
fn empty_parent_block_then_parent_blob_blob_trigger() {
let Some(tester) = DenebTester::new(RequestTrigger::GossipUnknownParentBlob) else {
return;
};
tester
.empty_parent_block_response()
.expect_penalty()
.expect_parent_block_request()
.expect_no_blobs_request()
.parent_blob_response()
.expect_empty_beacon_processor()
.parent_block_response()
.expect_block_process()
.parent_block_imported()
.block_response()
.expect_parent_chain_process();
}
#[test]
fn empty_parent_blobs_then_parent_block_blob_trigger() {
let Some(tester) = DenebTester::new(RequestTrigger::GossipUnknownParentBlob) else {
return;
};
tester
.block_response()
.empty_parent_blobs_response()
.expect_no_penalty()
.expect_no_blobs_request()
.expect_no_block_request()
.parent_block_response()
.expect_penalty()
.expect_parent_blobs_request()
.parent_blob_response()
.expect_block_process()
.parent_block_imported()
.expect_parent_chain_process();
} }
} }

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

@@ -58,11 +58,13 @@ use lighthouse_network::{PeerAction, PeerId};
use slog::{crit, debug, error, info, trace, warn, Logger}; use slog::{crit, debug, error, info, trace, warn, Logger};
use slot_clock::SlotClock; use slot_clock::SlotClock;
use std::boxed::Box; use std::boxed::Box;
use std::ops::IndexMut;
use std::ops::Sub; use std::ops::Sub;
use std::sync::Arc; use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use tokio::sync::mpsc; use tokio::sync::mpsc;
use tokio::time::sleep; use tokio::time::sleep;
use types::blob_sidecar::FixedBlobSidecarList;
use types::{BlobSidecar, EthSpec, Hash256, SignedBeaconBlock, Slot}; use types::{BlobSidecar, EthSpec, Hash256, SignedBeaconBlock, Slot};
/// The number of slots ahead of us that is allowed before requesting a long-range (batch) Sync /// The number of slots ahead of us that is allowed before requesting a long-range (batch) Sync
@@ -152,7 +154,7 @@ pub enum SyncMessage<T: EthSpec> {
}, },
/// Block processed /// Block processed
BlockPartProcessed { BlockComponentProcessed {
process_type: BlockProcessType, process_type: BlockProcessType,
result: BlockProcessingResult<T>, result: BlockProcessingResult<T>,
response_type: ResponseType, response_type: ResponseType,
@@ -291,6 +293,8 @@ pub fn spawn<T: BeaconChainTypes>(
sleep(sleep_duration).await; sleep(sleep_duration).await;
//TODO(sean) aggregate messages for blobs for the same block
while let Ok(msg) = delayed_lookups_recv.try_recv() { while let Ok(msg) = delayed_lookups_recv.try_recv() {
if let Err(e) = sync_send.send(msg) { if let Err(e) = sync_send.send(msg) {
warn!(log, "Failed to send delayed lookup message"; "error" => ?e); warn!(log, "Failed to send delayed lookup message"; "error" => ?e);
@@ -649,12 +653,15 @@ impl<T: BeaconChainTypes> SyncManager<T> {
warn!(self.log, "Delayed lookups dropped for block"; "block_root" => ?block_root, "error" => ?e); warn!(self.log, "Delayed lookups dropped for block"; "block_root" => ?block_root, "error" => ?e);
} }
} else { } else {
self.block_lookups.search_current_unknown_parent( let (block, blobs) = block.deconstruct();
block_root, self.block_lookups
block, .search_current_unknown_parent_block_and_blobs(
peer_id, block_root,
&mut self.network, Some(block),
); blobs,
peer_id,
&mut self.network,
);
self.block_lookups.search_parent( self.block_lookups.search_parent(
block_slot, block_slot,
block_root, block_root,
@@ -671,6 +678,7 @@ impl<T: BeaconChainTypes> SyncManager<T> {
if self.synced_and_connected_within_tolerance(blob_slot, &peer_id) { if self.synced_and_connected_within_tolerance(blob_slot, &peer_id) {
let block_root = blob.block_root; let block_root = blob.block_root;
let parent_root = blob.block_parent_root; let parent_root = blob.block_parent_root;
let blob_index = blob.index;
if self.should_delay_lookup(blob_slot) { if self.should_delay_lookup(blob_slot) {
if let Err(e) = self if let Err(e) = self
@@ -680,11 +688,16 @@ impl<T: BeaconChainTypes> SyncManager<T> {
warn!(self.log, "Delayed lookups dropped for blob"; "block_root" => ?block_root, "error" => ?e); warn!(self.log, "Delayed lookups dropped for blob"; "block_root" => ?block_root, "error" => ?e);
} }
} else { } else {
self.block_lookups.search_current_unknown_blob_parent( let mut blobs = FixedBlobSidecarList::default();
blob, *blobs.index_mut(blob_index as usize) = Some(blob);
peer_id, self.block_lookups
&mut self.network, .search_current_unknown_parent_block_and_blobs(
); block_root,
None,
Some(blobs),
peer_id,
&mut self.network,
);
} }
self.block_lookups.search_parent( self.block_lookups.search_parent(
blob_slot, blob_slot,
@@ -732,17 +745,14 @@ impl<T: BeaconChainTypes> SyncManager<T> {
request_id, request_id,
error, error,
} => self.inject_error(peer_id, request_id, error), } => self.inject_error(peer_id, request_id, error),
SyncMessage::BlockPartProcessed { SyncMessage::BlockComponentProcessed {
process_type, process_type,
result, result,
response_type, response_type,
} => match process_type { } => match process_type {
BlockProcessType::SingleBlock { id } => self.block_lookups.single_block_processed( BlockProcessType::SingleBlock { id } => self
id, .block_lookups
result, .single_block_component_processed(id, result, response_type, &mut self.network),
response_type,
&mut self.network,
),
BlockProcessType::ParentLookup { chain_hash } => self BlockProcessType::ParentLookup { chain_hash } => self
.block_lookups .block_lookups
.parent_block_processed(chain_hash, result, response_type, &mut self.network), .parent_block_processed(chain_hash, result, response_type, &mut self.network),

33
consensus/types/src/signed_beacon_block.rs
View File

@@ -1,3 +1,4 @@
use crate::blob_sidecar::BlobIdentifier;
use crate::*; use crate::*;
use bls::Signature; use bls::Signature;
use derivative::Derivative; use derivative::Derivative;
@@ -242,6 +243,38 @@ impl<E: EthSpec, Payload: AbstractExecPayload<E>> SignedBeaconBlock<E, Payload>
pub fn canonical_root(&self) -> Hash256 { pub fn canonical_root(&self) -> Hash256 {
self.message().tree_hash_root() self.message().tree_hash_root()
} }
pub fn num_expected_blobs(&self) -> usize {
self.message()
.body()
.blob_kzg_commitments()
.map(|c| c.len())
.unwrap_or(0)
}
pub fn get_expected_blob_ids(&self, block_root: Option<Hash256>) -> Vec<BlobIdentifier> {
self.get_filtered_blob_ids(block_root, |_, _| true)
}
/// If the filter returns `true` the id for the corresponding index and root will be included.
pub fn get_filtered_blob_ids(
&self,
block_root: Option<Hash256>,
filter: impl Fn(usize, Hash256) -> bool,
) -> Vec<BlobIdentifier> {
let block_root = block_root.unwrap_or_else(|| self.canonical_root());
let num_blobs_expected = self.num_expected_blobs();
let mut blob_ids = Vec::with_capacity(num_blobs_expected);
for i in 0..num_blobs_expected {
if filter(i, block_root) {
blob_ids.push(BlobIdentifier {
block_root,
index: i as u64,
});
}
}
blob_ids
}
} }
// We can convert pre-Bellatrix blocks without payloads into blocks with payloads. // We can convert pre-Bellatrix blocks without payloads into blocks with payloads.