diff --git a/beacon_node/beacon_chain/src/beacon_chain.rs b/beacon_node/beacon_chain/src/beacon_chain.rs index 708a07021d0..5e257f50936 100644 --- a/beacon_node/beacon_chain/src/beacon_chain.rs +++ b/beacon_node/beacon_chain/src/beacon_chain.rs @@ -6272,6 +6272,12 @@ impl BeaconChain { Ok(()) } + pub fn is_parent_imported(&self, block: &SignedBeaconBlock) -> bool { + self.canonical_head + .fork_choice_read_lock() + .is_parent_imported(block) + } + pub fn block_is_known_to_fork_choice(&self, root: &Hash256) -> bool { self.canonical_head .fork_choice_read_lock() diff --git a/beacon_node/beacon_chain/src/canonical_head.rs b/beacon_node/beacon_chain/src/canonical_head.rs index 5d7a7f1527d..98c0d5b25d5 100644 --- a/beacon_node/beacon_chain/src/canonical_head.rs +++ b/beacon_node/beacon_chain/src/canonical_head.rs @@ -210,6 +210,13 @@ impl CachedHead { self.finalized_checkpoint } + /// Returns the finalized slot from the fork choice finalized checkpoint + pub fn finalized_slot(&self) -> Slot { + self.finalized_checkpoint + .epoch + .start_slot(E::slots_per_epoch()) + } + /// Returns the justified checkpoint, as determined by fork choice. /// /// ## Note diff --git a/beacon_node/lighthouse_network/src/peer_manager/mod.rs b/beacon_node/lighthouse_network/src/peer_manager/mod.rs index 898b97a85f7..82fac58023d 100644 --- a/beacon_node/lighthouse_network/src/peer_manager/mod.rs +++ b/beacon_node/lighthouse_network/src/peer_manager/mod.rs @@ -32,6 +32,7 @@ use peerdb::score::{PeerAction, ReportSource}; pub use peerdb::sync_status::{SyncInfo, SyncStatus}; use std::collections::{HashMap, HashSet, hash_map::Entry}; use std::net::IpAddr; +use std::str::FromStr; use strum::IntoEnumIterator; use types::data::{CustodyIndex, compute_subnets_from_custody_group, get_custody_groups}; @@ -487,6 +488,12 @@ impl PeerManager { let previous_listening_addresses = peer_info.set_listening_addresses(info.listen_addrs.clone()); peer_info.set_client(peerdb::client::Client::from_identify_info(info)); + peer_info.set_supported_protocols( + info.protocols + .iter() + .filter_map(|protocol| rpc_protocol_from_id(protocol.as_ref())) + .collect(), + ); if previous_kind != peer_info.client().kind || *peer_info.listening_addresses() != previous_listening_addresses @@ -1706,6 +1713,15 @@ enum ConnectingType { }, } +/// Parses a libp2p protocol id of the form `/eth2/beacon_chain/req///` +/// into the corresponding ReqResp [`Protocol`], returning `None` for non-ReqResp protocols. +fn rpc_protocol_from_id(protocol_id: &str) -> Option { + protocol_id + .strip_prefix("/eth2/beacon_chain/req/") + .and_then(|rest| rest.split('/').next()) + .and_then(|name| Protocol::from_str(name).ok()) +} + #[cfg(test)] mod tests { use super::*; @@ -1713,6 +1729,17 @@ mod tests { use crate::rpc::MetaDataV3; use types::{ChainSpec, ForkName, MainnetEthSpec as E}; + #[test] + fn rpc_protocol_from_id_parses_reqresp_protocols() { + // A known ReqResp protocol is parsed from its name segment. + assert_eq!( + rpc_protocol_from_id("/eth2/beacon_chain/req/beacon_blocks_by_head/1/ssz_snappy"), + Some(Protocol::BlocksByHead) + ); + // Non-ReqResp protocols are ignored. + assert_eq!(rpc_protocol_from_id("/meshsub/1.1.0"), None); + } + async fn build_peer_manager(target_peer_count: usize) -> PeerManager { build_peer_manager_with_trusted_peers(vec![], target_peer_count).await } diff --git a/beacon_node/lighthouse_network/src/peer_manager/peerdb.rs b/beacon_node/lighthouse_network/src/peer_manager/peerdb.rs index 693fdebb69b..6ccb5ad1dd7 100644 --- a/beacon_node/lighthouse_network/src/peer_manager/peerdb.rs +++ b/beacon_node/lighthouse_network/src/peer_manager/peerdb.rs @@ -870,6 +870,17 @@ impl PeerDB { .ok_or_else(|| "Cannot set custody subnets, peer not found".to_string()) } + pub fn __set_supported_protocols( + &mut self, + peer_id: &PeerId, + supported_protocols: HashSet, + ) -> Result<(), String> { + self.peers + .get_mut(peer_id) + .map(|info| info.set_supported_protocols(supported_protocols)) + .ok_or_else(|| "Cannot set supported protocols, peer not found".to_string()) + } + /// The connection state of the peer has been changed. Modify the peer in the db to ensure all /// variables are in sync with libp2p. /// Updating the state can lead to a `BanOperation` which needs to be processed via the peer diff --git a/beacon_node/lighthouse_network/src/peer_manager/peerdb/peer_info.rs b/beacon_node/lighthouse_network/src/peer_manager/peerdb/peer_info.rs index 658a6355e3f..d907bd76e30 100644 --- a/beacon_node/lighthouse_network/src/peer_manager/peerdb/peer_info.rs +++ b/beacon_node/lighthouse_network/src/peer_manager/peerdb/peer_info.rs @@ -2,6 +2,7 @@ use super::client::Client; use super::score::{PeerAction, Score, ScoreState}; use super::sync_status::SyncStatus; use crate::discovery::Eth2Enr; +use crate::rpc::Protocol as RpcProtocol; use crate::{rpc::MetaData, types::Subnet}; use PeerConnectionStatus::*; use discv5::Enr; @@ -59,6 +60,8 @@ pub struct PeerInfo { connection_direction: Option, /// The enr of the peer, if known. enr: Option, + /// The set of ReqResp protocols the peer advertised via `identify`. Empty until identified. + supported_protocols: HashSet, } impl Default for PeerInfo { @@ -78,6 +81,7 @@ impl Default for PeerInfo { is_trusted: false, connection_direction: None, enr: None, + supported_protocols: HashSet::new(), } } } @@ -172,6 +176,11 @@ impl PeerInfo { self.enr.as_ref() } + /// Returns true if the peer advertised support for `protocol` via `identify`. + pub fn supports_protocol(&self, protocol: RpcProtocol) -> bool { + self.supported_protocols.contains(&protocol) + } + /// An iterator over all the subnets this peer is subscribed to. pub fn subnets(&self) -> impl Iterator { self.subnets.iter() @@ -400,6 +409,15 @@ impl PeerInfo { self.client = client } + /// Replaces the set of ReqResp protocols the peer advertised via `identify`. + // VISIBILITY: The peer manager is able to set the supported protocols + pub(in crate::peer_manager) fn set_supported_protocols( + &mut self, + supported_protocols: HashSet, + ) { + self.supported_protocols = supported_protocols; + } + /// Replaces the current listening addresses with those specified, returning the current /// listening addresses. // VISIBILITY: The peer manager is able to set the listening addresses diff --git a/beacon_node/lighthouse_network/src/rpc/protocol.rs b/beacon_node/lighthouse_network/src/rpc/protocol.rs index b444608468a..37f4c3b5459 100644 --- a/beacon_node/lighthouse_network/src/rpc/protocol.rs +++ b/beacon_node/lighthouse_network/src/rpc/protocol.rs @@ -302,6 +302,13 @@ pub enum Protocol { LightClientUpdatesByRange, } +impl serde::Serialize for Protocol { + /// Serialize as the canonical protocol name (matching `AsRef`/`Display`). + fn serialize(&self, serializer: S) -> Result { + serializer.serialize_str(self.as_ref()) + } +} + impl Protocol { pub(crate) fn terminator(self) -> Option { match self { diff --git a/beacon_node/lighthouse_network/src/service/api_types.rs b/beacon_node/lighthouse_network/src/service/api_types.rs index 4a8c6c55eb8..489ea2794d6 100644 --- a/beacon_node/lighthouse_network/src/service/api_types.rs +++ b/beacon_node/lighthouse_network/src/service/api_types.rs @@ -21,6 +21,8 @@ pub struct SingleLookupReqId { pub enum SyncRequestId { /// Request searching for a block given a hash. SingleBlock { id: SingleLookupReqId }, + /// Request walking the ancestors of a block given its root, via `beacon_blocks_by_head`. + BlocksByHead(SingleLookupReqId), /// Request searching for a payload envelope given a hash. SinglePayloadEnvelope { id: SingleLookupReqId }, /// Request searching for a set of data columns given a hash and list of column indices. diff --git a/beacon_node/network/src/router.rs b/beacon_node/network/src/router.rs index d83068e3375..cf19c1dfc86 100644 --- a/beacon_node/network/src/router.rs +++ b/beacon_node/network/src/router.rs @@ -355,10 +355,8 @@ impl Router { Response::PayloadEnvelopesByRange(envelope) => { self.on_payload_envelopes_by_range_response(peer_id, app_request_id, envelope); } - // Lighthouse currently only serves BlocksByHead and does not issue it as a client, - // so receiving a response is unexpected. Drop it without crashing. - Response::BlocksByHead(_) => { - debug!("BlocksByHead response received but not requested by lighthouse"); + Response::BlocksByHead(beacon_block) => { + self.on_blocks_by_head_response(peer_id, app_request_id, beacon_block); } // Light client responses should not be received Response::LightClientBootstrap(_) @@ -717,6 +715,29 @@ impl Router { }); } + /// Handle a `BlocksByHead` response from the peer. + /// A `beacon_block` behaves as a stream which is terminated on a `None` response. + pub fn on_blocks_by_head_response( + &mut self, + peer_id: PeerId, + app_request_id: AppRequestId, + beacon_block: Option>>, + ) { + let sync_request_id = match app_request_id { + AppRequestId::Sync(id @ SyncRequestId::BlocksByHead { .. }) => id, + other => { + crit!(request = ?other, "BlocksByHead response on incorrect request"); + return; + } + }; + + self.send_to_sync(SyncMessage::RpcBlock { + peer_id, + sync_request_id, + beacon_block, + }); + } + /// Handle a `DataColumnsByRoot` response from the peer. pub fn on_data_columns_by_root_response( &mut self, diff --git a/beacon_node/network/src/sync/block_lookups/mod.rs b/beacon_node/network/src/sync/block_lookups/mod.rs index ef9807f0372..f45a48f59ec 100644 --- a/beacon_node/network/src/sync/block_lookups/mod.rs +++ b/beacon_node/network/src/sync/block_lookups/mod.rs @@ -23,8 +23,8 @@ use self::parent_chain::{NodeChain, compute_parent_chains}; pub use self::single_block_lookup::DownloadResult; use self::single_block_lookup::{LookupRequestError, PeerType, SingleBlockLookup}; -use super::manager::{BlockProcessType, SLOT_IMPORT_TOLERANCE}; -use super::network_context::{RpcResponseError, SyncNetworkContext}; +use super::manager::BlockProcessType; +use super::network_context::{AncestorBlocks, RpcResponseError, SyncNetworkContext}; use crate::metrics; use crate::network_beacon_processor::BlockProcessingResult; use crate::sync::SyncMessage; @@ -37,6 +37,7 @@ use fnv::FnvHashMap; use lighthouse_network::PeerId; use lighthouse_network::service::api_types::SingleLookupReqId; use lru_cache::LRUTimeCache; +use std::collections::HashMap; use std::collections::hash_map::Entry; use std::sync::Arc; use std::time::Duration; @@ -50,13 +51,19 @@ use types::{ pub mod parent_chain; mod single_block_lookup; -/// The maximum depth we will search for a parent block. In principle we should have sync'd any -/// canonical chain to its head once the peer connects. A chain should not appear where it's depth -/// is further back than the most recent head slot. +/// `beacon_blocks_by_head` count for a lone lookup (nothing awaits it yet): fetch only a few +/// ancestors so the common shallow case doesn't over-fetch. +pub(crate) const BLOCKS_BY_HEAD_LONE_REQUEST_COUNT: u64 = 4; +/// `beacon_blocks_by_head` count for a lookup that's part of a known parent chain: fetch a large +/// batch to resolve a deep chain in few round-trips. Tune against mainnet metrics. +pub(crate) const BLOCKS_BY_HEAD_CHAIN_REQUEST_COUNT: u64 = 32; + +/// The maximum depth we will search for a parent block. Once a lookup chain reaches this depth it +/// is dropped and the head is force-transitioned to range sync. /// -/// Have the same value as range's sync tolerance to consider a peer synced. Once sync lookup -/// reaches the maximum depth it will force trigger range sync. -pub(crate) const PARENT_DEPTH_TOLERANCE: usize = SLOT_IMPORT_TOLERANCE; +/// Allow one full `beacon_blocks_by_head` batch (plus its triggering child) so a single response +/// can always seed an entire ancestor chain without tripping the limit. +pub(crate) const PARENT_DEPTH_TOLERANCE: usize = BLOCKS_BY_HEAD_CHAIN_REQUEST_COUNT as usize + 1; const IGNORED_CHAINS_CACHE_EXPIRY_SECONDS: u64 = 60; pub const SINGLE_BLOCK_LOOKUP_MAX_ATTEMPTS: u8 = 4; @@ -77,6 +84,7 @@ const LOOKUP_MAX_DURATION_NO_PEERS_SECS: u64 = 10; const MAX_LOOKUPS: usize = 200; type BlockDownloadResponse = Result>>, RpcResponseError>; +type BlocksDownloadResponse = Result>, RpcResponseError>; type CustodyDownloadResponse = Result>, RpcResponseError>; type PayloadDownloadResponse = @@ -87,6 +95,8 @@ pub enum BlockComponent { Sidecar, } +type KnownParents = HashMap>>>; + pub type SingleLookupId = u32; pub struct BlockLookups { @@ -168,6 +178,21 @@ impl BlockLookups { ) } + /// The `beacon_blocks_by_head` count to request when fetching `block_root`. A lookup that + /// nothing is waiting on is a lone island and fetches only a few ancestors; a lookup a child is + /// awaiting is part of a chain already proven to be deep, so it fetches a large batch. + fn by_head_request_count(&self, block_root: Hash256) -> u64 { + if self + .single_block_lookups + .values() + .any(|lookup| lookup.is_awaiting_block(block_root)) + { + BLOCKS_BY_HEAD_CHAIN_REQUEST_COUNT + } else { + BLOCKS_BY_HEAD_LONE_REQUEST_COUNT + } + } + /* Lookup requests */ /// Creates a parent lookup for the block with the given `block_root` and immediately triggers it. @@ -186,9 +211,11 @@ impl BlockLookups { ) -> bool { let parent_lookup_exists = self.search_parent_of_child( parent_root, + None, &PeerType::new(parent_block_hash), block_root, &[peer_id], + None, cx, ); // Only create the child lookup if the parent exists @@ -203,6 +230,7 @@ impl BlockLookups { // peers to house the block components. &[], &PeerType::Block, + None, cx, ) } else { @@ -220,7 +248,15 @@ impl BlockLookups { peer_source: &[PeerId], cx: &mut SyncNetworkContext, ) -> bool { - self.new_current_lookup(block_root, None, None, peer_source, &PeerType::Block, cx) + self.new_current_lookup( + block_root, + None, + None, + peer_source, + &PeerType::Block, + None, + cx, + ) } /// A block or blob triggers the search of a parent. @@ -230,12 +266,15 @@ impl BlockLookups { /// /// Returns true if the lookup is created or already exists #[must_use = "only reference the new lookup if returns true"] + #[allow(clippy::too_many_arguments)] pub fn search_parent_of_child( &mut self, block_root_to_search: Hash256, + block_component: Option>, peer_type: &PeerType, child_block_root_trigger: Hash256, peers: &[PeerId], + known_blocks: Option>, cx: &mut SyncNetworkContext, ) -> bool { let parent_chains = self.active_parent_lookups(); @@ -326,13 +365,22 @@ impl BlockLookups { } // `block_root_to_search` is a failed chain check happens inside new_current_lookup - self.new_current_lookup(block_root_to_search, None, None, peers, peer_type, cx) + self.new_current_lookup( + block_root_to_search, + block_component, + None, + peers, + peer_type, + known_blocks, + cx, + ) } /// Searches for a single block hash. If the blocks parent is unknown, a chain of blocks is /// constructed. /// Returns true if the lookup is created or already exists #[must_use = "only reference the new lookup if returns true"] + #[allow(clippy::too_many_arguments)] fn new_current_lookup( &mut self, block_root: Hash256, @@ -340,6 +388,7 @@ impl BlockLookups { awaiting_parent: Option, peers: &[PeerId], peer_type: &PeerType, + known_blocks: Option>, cx: &mut SyncNetworkContext, ) -> bool { // If this block or it's parent is part of a known ignored chain, ignore it. @@ -387,10 +436,20 @@ impl BlockLookups { return false; } + // Compute before inserting below: a lookup nothing awaits is a lone island (fetch a few + // ancestors); one a child already awaits is part of a deep chain (fetch a large batch). + let by_head_count = self.by_head_request_count(block_root); + // If we know that this lookup has unknown parent (is awaiting a parent lookup to resolve), // signal here to hold processing downloaded data. - let mut lookup = - SingleBlockLookup::new(block_root, peers, peer_type, cx.next_id(), awaiting_parent); + let mut lookup = SingleBlockLookup::new( + block_root, + peers, + peer_type, + cx.next_id(), + awaiting_parent, + by_head_count, + ); let _guard = lookup.span.clone().entered(); // Add block components to the new request @@ -419,7 +478,7 @@ impl BlockLookups { self.metrics.created_lookups += 1; let result = lookup.continue_requests(cx); - if self.on_lookup_result(id, result, "new_current_lookup", cx) { + if self.on_lookup_result(id, result, known_blocks, "new_current_lookup", cx) { self.update_metrics(); true } else { @@ -433,15 +492,49 @@ impl BlockLookups { pub fn on_block_download_response( &mut self, id: SingleLookupReqId, - response: BlockDownloadResponse, + response: BlocksDownloadResponse, cx: &mut SyncNetworkContext, ) { let Some(lookup) = self.single_block_lookups.get_mut(&id.lookup_id) else { debug!(?id, "Block returned for single block lookup not present"); return; }; - let result = lookup.on_block_download_response(id.req_id, response, cx); - self.on_lookup_result(id.lookup_id, result, "block_download_response", cx); + // If response is successful, `on_block_download_response` will advance the block request + // state to `AwaitingProcess` then `continue_requests` will check if the block can be + // processed: + // - If yes: sends block for processing + // - If not: marks the lookup as awaiting parent + let known_ancestors = response.as_ref().ok().map(|d| { + d.value + .ancestor_blocks + .iter() + .map(|block| { + ( + block.canonical_root(), + DownloadResult { + value: block.clone(), + peer_group: d.peer_group.clone(), + }, + ) + }) + .collect::>() + }); + + let result = lookup.on_block_download_response( + id.req_id, + response.map(|d| DownloadResult { + value: d.value.first_block, + peer_group: d.peer_group, + }), + cx, + ); + self.on_lookup_result( + id.lookup_id, + result, + known_ancestors, + "block_download_response", + cx, + ); } pub fn on_custody_download_response( @@ -455,7 +548,7 @@ impl BlockLookups { return; }; let result = lookup.on_custody_download_response(id.req_id, response, cx); - self.on_lookup_result(id.lookup_id, result, "custody_download_response", cx); + self.on_lookup_result(id.lookup_id, result, None, "custody_download_response", cx); } pub fn on_payload_download_response( @@ -472,7 +565,7 @@ impl BlockLookups { return; }; let result = lookup.on_payload_download_response(id.req_id, response, cx); - self.on_lookup_result(id.lookup_id, result, "payload_download_response", cx); + self.on_lookup_result(id.lookup_id, result, None, "payload_download_response", cx); } /* Error responses */ @@ -542,7 +635,7 @@ impl BlockLookups { lookup.on_payload_processing_result(result, cx) } }; - self.on_lookup_result(lookup_id, lookup_result, "processing_result", cx); + self.on_lookup_result(lookup_id, lookup_result, None, "processing_result", cx); } pub fn has_any_awaiting_children(&self, block_root: Hash256) -> bool { @@ -575,7 +668,7 @@ impl BlockLookups { } for (id, result) in lookup_results { - self.on_lookup_result(id, result, "continue_child_lookups", cx); + self.on_lookup_result(id, result, None, "continue_child_lookups", cx); } } @@ -613,6 +706,7 @@ impl BlockLookups { &mut self, id: SingleLookupId, result: Result, + known_blocks: Option>, source: &str, cx: &mut SyncNetworkContext, ) -> bool { @@ -624,11 +718,18 @@ impl BlockLookups { block_root, peers, }) => { + // If a prior by-head response already fetched the parent, seed its lookup with it. + let block_component = known_blocks.as_ref().and_then(|p| { + p.get(&parent_root) + .map(|b| BlockComponent::Block(b.clone())) + }); if self.search_parent_of_child( parent_root, + block_component, &PeerType::new(parent_block_hash), block_root, &peers, + known_blocks, cx, ) { true @@ -854,7 +955,7 @@ impl BlockLookups { // pruned with `drop_lookups_without_peers` because it has peers. This is rare corner // case, but it can result in stuck lookups. let result = lookup.continue_requests(cx); - self.on_lookup_result(lookup_id, result, "add_peers", cx); + self.on_lookup_result(lookup_id, result, None, "add_peers", cx); Ok(()) } else { Ok(()) diff --git a/beacon_node/network/src/sync/block_lookups/single_block_lookup.rs b/beacon_node/network/src/sync/block_lookups/single_block_lookup.rs index 09ffa50f9ba..049f768e5dd 100644 --- a/beacon_node/network/src/sync/block_lookups/single_block_lookup.rs +++ b/beacon_node/network/src/sync/block_lookups/single_block_lookup.rs @@ -79,6 +79,13 @@ impl AwaitingParent { self.parent_root } + pub fn from_block(block: &SignedBeaconBlock) -> Self { + Self::new( + block.parent_root(), + block.payload_bid_parent_block_hash().ok(), + ) + } + pub fn into_peer_type(self) -> PeerType { PeerType::new(self.parent_block_hash) } @@ -203,6 +210,9 @@ pub struct SingleBlockLookup { #[educe(Debug(method(fmt_peer_map_as_len)))] gloas_child_peers: HashMap, awaiting_parent: Option, + /// Count to request over `beacon_blocks_by_head`: small for a lone lookup, large once it's known + /// to be part of a deep chain. Fixed at creation from the lookup graph. + by_head_count: u64, created: Instant, pub(crate) span: Span, } @@ -214,6 +224,7 @@ impl SingleBlockLookup { peer_type: &PeerType, id: Id, awaiting_parent: Option, + by_head_count: u64, ) -> Self { let lookup_span = debug_span!( "lh_single_block_lookup", @@ -239,6 +250,7 @@ impl SingleBlockLookup { peers: block_peers, gloas_child_peers, awaiting_parent, + by_head_count, created: Instant::now(), span: lookup_span, } @@ -380,13 +392,38 @@ impl SingleBlockLookup { // === Block request === self.block_request.state.maybe_start_downloading(|| { - cx.block_lookup_request(self.id, self.peers.clone(), self.block_root) + cx.block_lookup_request( + self.id, + self.peers.clone(), + self.block_root, + self.by_head_count, + ) })?; if self.awaiting_parent.is_none() - && let Some(data) = self.block_request.state.maybe_start_processing() + && let Some(data) = self.block_request.state.is_awaiting_process() { - cx.send_block_for_processing(self.id, self.block_root, data.value) - .map_err(LookupRequestError::SendFailedProcessor)?; + let block = &data.value; + // Eagerly check if we can import without having to send the block for processing. This + // allows us to check many lookups in the same sync execution / loop. + if cx.chain.is_parent_imported(block) { + cx.send_block_for_processing(self.id, self.block_root, block.clone()) + .map_err(LookupRequestError::SendFailedProcessor)?; + self.block_request.state.start_processing()?; + } else if let Some(reason) = cx.conflicts_with_finality(block) { + return Err(LookupRequestError::Failed(reason)); + } else if let Some(reason) = cx.block_too_far_in_future(block) { + // A peer served a block too far ahead of our head for an unknown root. Drop it + // instead of walking ancestors / forcing range sync to a bogus future head. + return Err(LookupRequestError::Failed(reason)); + } else { + self.awaiting_parent = Some(AwaitingParent::from_block(block)); + return Ok(LookupResult::ParentUnknown { + parent_root: block.parent_root(), + parent_block_hash: block.payload_bid_parent_block_hash().ok(), + block_root: self.block_root, + peers: self.all_peers(), + }); + } } // === Data request === @@ -800,6 +837,16 @@ impl SingleLookupRequestState { } } + fn is_awaiting_process(&self) -> Option<&DownloadResult> { + match &self.state { + State::AwaitingDownload { .. } => None, + State::Downloading { .. } => None, + State::AwaitingProcess(result) => Some(result), + State::Processing(_) => None, + State::Processed(..) => None, + } + } + /// Drive download: check max attempts, issue request, handle result. fn maybe_start_downloading( &mut self, @@ -911,7 +958,6 @@ impl SingleLookupRequestState { /// Switch to `Processing` if the request is in `AwaitingProcess` state, otherwise returns None. pub fn maybe_start_processing(&mut self) -> Option> { - // For 2 lines replace state with placeholder to gain ownership of `result` match &self.state { State::AwaitingProcess(result) => { let result = result.clone(); @@ -922,6 +968,20 @@ impl SingleLookupRequestState { } } + /// Switch to `Processing` if the request is in `AwaitingProcess` state, otherwise returns None. + pub fn start_processing(&mut self) -> Result<(), LookupRequestError> { + match &self.state { + State::AwaitingProcess(result) => { + let result = result.clone(); + self.state = State::Processing(result.clone()); + Ok(()) + } + other => Err(LookupRequestError::BadState(format!( + "Bad state start_processing expected AwaitingProcess got {other}" + ))), + } + } + /// Revert into `AwaitingProcessing`, if the payload if not invalid and can be submitted for /// processing latter. pub fn revert_to_awaiting_processing(&mut self) -> Result<(), LookupRequestError> { diff --git a/beacon_node/network/src/sync/manager.rs b/beacon_node/network/src/sync/manager.rs index b9bea21b8ce..ea92af7653f 100644 --- a/beacon_node/network/src/sync/manager.rs +++ b/beacon_node/network/src/sync/manager.rs @@ -498,6 +498,9 @@ impl SyncManager { SyncRequestId::SingleBlock { id } => { self.on_single_block_response(id, peer_id, RpcEvent::RPCError(error)) } + SyncRequestId::BlocksByHead(id) => { + self.on_blocks_by_head_response(id, peer_id, RpcEvent::RPCError(error)) + } SyncRequestId::SinglePayloadEnvelope { id } => { self.on_single_payload_envelope_response(id, peer_id, RpcEvent::RPCError(error)) } @@ -1111,6 +1114,9 @@ impl SyncManager { SyncRequestId::SingleBlock { id } => { self.on_single_block_response(id, peer_id, RpcEvent::from_chunk(block)) } + SyncRequestId::BlocksByHead(id) => { + self.on_blocks_by_head_response(id, peer_id, RpcEvent::from_chunk(block)) + } SyncRequestId::BlocksByRange(id) => { self.on_blocks_by_range_response(id, peer_id, RpcEvent::from_chunk(block)) } @@ -1135,6 +1141,21 @@ impl SyncManager { } } + fn on_blocks_by_head_response( + &mut self, + id: SingleLookupReqId, + peer_id: PeerId, + block: RpcEvent>>, + ) { + if let Some(resp) = self.network.on_blocks_by_head_response(id, peer_id, block) { + self.block_lookups.on_block_download_response( + id, + resp.map(|value| DownloadResult::new(value, PeerGroup::from_single(peer_id))), + &mut self.network, + ) + } + } + fn rpc_blob_received( &mut self, sync_request_id: SyncRequestId, diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs index 100be9f4d7e..d1e562743ad 100644 --- a/beacon_node/network/src/sync/network_context.rs +++ b/beacon_node/network/src/sync/network_context.rs @@ -8,7 +8,7 @@ pub use self::requests::{ }; use super::SyncMessage; use super::block_sidecar_coupling::RangeBlockComponentsRequest; -use super::manager::BlockProcessType; +use super::manager::{BlockProcessType, SLOT_IMPORT_TOLERANCE}; use crate::metrics; use crate::network_beacon_processor::NetworkBeaconProcessor; #[cfg(test)] @@ -25,10 +25,11 @@ use beacon_chain::{BeaconChain, BeaconChainTypes, BlockProcessStatus, EngineStat use custody::CustodyRequestResult; use fnv::FnvHashMap; use lighthouse_network::rpc::methods::{ - BlobsByRangeRequest, DataColumnsByRangeRequest, PayloadEnvelopesByRangeRequest, + BlobsByRangeRequest, BlocksByHeadRequest, DataColumnsByRangeRequest, + PayloadEnvelopesByRangeRequest, }; use lighthouse_network::rpc::{ - BlocksByRangeRequest, GoodbyeReason, MAX_CONCURRENT_REQUESTS, RPCError, RequestType, + BlocksByRangeRequest, GoodbyeReason, MAX_CONCURRENT_REQUESTS, Protocol, RPCError, RequestType, }; pub use lighthouse_network::service::api_types::RangeRequestId; use lighthouse_network::service::api_types::{ @@ -42,9 +43,10 @@ use lighthouse_network::{Client, NetworkGlobals, PeerAction, PeerId, ReportSourc use parking_lot::RwLock; pub use requests::LookupVerifyError; use requests::{ - ActiveRequestItems, ActiveRequests, BlobsByRangeRequestItems, BlocksByRangeRequestItems, - BlocksByRootRequestItems, DataColumnsByRangeRequestItems, DataColumnsByRootRequestItems, - PayloadEnvelopesByRangeRequestItems, PayloadEnvelopesByRootRequestItems, + ActiveRequestItems, ActiveRequests, BlobsByRangeRequestItems, BlocksByHeadRequestItems, + BlocksByRangeRequestItems, BlocksByRootRequestItems, DataColumnsByRangeRequestItems, + DataColumnsByRootRequestItems, PayloadEnvelopesByRangeRequestItems, + PayloadEnvelopesByRootRequestItems, }; #[cfg(test)] use slot_clock::SlotClock; @@ -79,6 +81,7 @@ macro_rules! new_range_request_span { /// Max retries for block components after which we fail the batch. pub const MAX_COLUMN_RETRIES: usize = 3; +/// Number of ancestors a block lookup requests when fetching a block via `beacon_blocks_by_head`. #[derive(Debug)] pub enum RpcEvent { StreamTermination, @@ -100,6 +103,33 @@ pub type RpcResponseResult = Result; pub type CustodyByRootResult = Result>, RpcResponseError>; +#[derive(Clone)] +pub struct AncestorBlocks { + pub first_block: Arc>, + pub ancestor_blocks: Vec>>, +} + +impl AncestorBlocks { + /// A single block with no ancestors. + pub fn from_single(block: Arc>) -> Self { + Self { + first_block: block, + ancestor_blocks: vec![], + } + } + + /// The requested root first (becoming `first_block`), then its ancestors in descending slot + /// order. Returns `None` for an empty run. + pub fn from_vec(blocks: Vec>>) -> Option { + let mut blocks = blocks.into_iter(); + let first_block = blocks.next()?; + Some(Self { + first_block, + ancestor_blocks: blocks.collect(), + }) + } +} + /// Per-peer count of active requests for a single protocol, to keep peer selection within /// `MAX_CONCURRENT_REQUESTS` concurrent requests per protocol ID. struct ActiveRequestsPerPeer { @@ -229,6 +259,9 @@ pub struct SyncNetworkContext { /// A mapping of active BlocksByRoot requests, including both current slot and parent lookups. blocks_by_root_requests: ActiveRequests>, + /// A mapping of active BlocksByHead requests, used by block lookups to walk a block's ancestors. + blocks_by_head_requests: + ActiveRequests>, /// A mapping of active PayloadEnvelopesByRoot requests payload_envelopes_by_root_requests: ActiveRequests>, @@ -333,6 +366,7 @@ impl SyncNetworkContext { execution_engine_state: EngineState::Online, // always assume `Online` at the start request_id: 1, blocks_by_root_requests: ActiveRequests::new("blocks_by_root"), + blocks_by_head_requests: ActiveRequests::new("blocks_by_head"), payload_envelopes_by_root_requests: ActiveRequests::new("payload_envelopes_by_root"), data_columns_by_root_requests: ActiveRequests::new("data_columns_by_root"), blocks_by_range_requests: ActiveRequests::new("blocks_by_range"), @@ -366,6 +400,7 @@ impl SyncNetworkContext { network_send: _, request_id: _, blocks_by_root_requests, + blocks_by_head_requests, payload_envelopes_by_root_requests, data_columns_by_root_requests, blocks_by_range_requests, @@ -387,6 +422,10 @@ impl SyncNetworkContext { .active_requests_of_peer(peer_id) .into_iter() .map(|id| SyncRequestId::SingleBlock { id: *id }); + let blocks_by_head_ids = blocks_by_head_requests + .active_requests_of_peer(peer_id) + .into_iter() + .map(|id| SyncRequestId::BlocksByHead(*id)); let payload_envelopes_by_root_ids = payload_envelopes_by_root_requests .active_requests_of_peer(peer_id) .into_iter() @@ -412,6 +451,7 @@ impl SyncNetworkContext { .into_iter() .map(|req_id| SyncRequestId::PayloadEnvelopesByRange(*req_id)); blocks_by_root_ids + .chain(blocks_by_head_ids) .chain(payload_envelopes_by_root_ids) .chain(data_column_by_root_ids) .chain(blocks_by_range_ids) @@ -430,6 +470,44 @@ impl SyncNetworkContext { &self.network_beacon_processor.network_globals } + /// Returns true if the peer advertised support for the `beacon_blocks_by_head` protocol. + pub fn peer_supports_blocks_by_head(&self, peer_id: &PeerId) -> bool { + self.network_globals() + .peers + .read() + .peer_info(peer_id) + .map(|info| info.supports_protocol(Protocol::BlocksByHead)) + .unwrap_or(false) + } + + /// Returns `Some(reason)` if `block` is at or below finalization but off the finalized chain. + pub fn conflicts_with_finality(&self, block: &SignedBeaconBlock) -> Option { + let finalized_slot = self.chain.head().finalized_slot(); + if block.slot() > finalized_slot { + None + } else { + Some(format!( + "block slot {} <= finalized slot {finalized_slot}", + block.slot() + )) + } + } + + /// Returns `Some(reason)` if `block`'s slot is implausibly far ahead of the current wall-clock + /// slot — a peer shouldn't be serving it, so it must not trigger parent lookups or forced range + /// sync. Compared against the clock (not our head) so normal deep sync isn't rejected. + pub fn block_too_far_in_future(&self, block: &SignedBeaconBlock) -> Option { + let current_slot = self.chain.slot().ok()?; + if block.slot().as_u64() > current_slot.as_u64() + SLOT_IMPORT_TOLERANCE as u64 { + Some(format!( + "block slot {} more than {SLOT_IMPORT_TOLERANCE} ahead of current slot {current_slot}", + block.slot() + )) + } else { + None + } + } + /// Returns the Client type of the peer if known pub fn client_type(&self, peer_id: &PeerId) -> Client { self.network_globals() @@ -858,22 +936,34 @@ impl SyncNetworkContext { lookup_id: SingleLookupId, lookup_peers: Arc>>, block_root: Hash256, + by_head_count: u64, ) -> Result>>, RpcRequestSendError> { let blocks_by_root_per_peer = ActiveRequestsPerPeer::new(&self.blocks_by_root_requests); + let blocks_by_head_per_peer = ActiveRequestsPerPeer::new(&self.blocks_by_head_requests); let Some(peer_id) = lookup_peers .read() .iter() .map(|peer| { + let supports_blocks_by_head = self.peer_supports_blocks_by_head(peer); + // The spec limits concurrent requests per protocol ID, so gate on the protocol we + // would actually use. + let at_concurrency_limit = if supports_blocks_by_head { + blocks_by_head_per_peer.at_concurrency_limit(peer) + } else { + blocks_by_root_per_peer.at_concurrency_limit(peer) + }; ( - // Strictly de-prioritize peers already at the per-protocol concurrency limit - blocks_by_root_per_peer.at_concurrency_limit(peer), + // Strictly de-prioritize peers already at the concurrent-request limit + at_concurrency_limit, + // Prefer peers that support `beacon_blocks_by_head` + !supports_blocks_by_head, // Random factor to break ties, otherwise the PeerID breaks ties rand::random::(), peer, ) }) .min() - .map(|(_, _, peer)| *peer) + .map(|(_, _, _, peer)| *peer) else { // Allow lookup to not have any peers and do nothing. This is an optimization to not // lose progress of lookups created from a block with unknown parent before we receive @@ -906,6 +996,24 @@ impl SyncNetworkContext { } } + // If the peer supports `beacon_blocks_by_head`, fetch the block and a run of its ancestors + // in a single request; otherwise fall back to `beacon_blocks_by_root` for the single block. + Ok(LookupRequestResult::RequestSent( + if self.peer_supports_blocks_by_head(&peer_id) { + self.send_blocks_by_head(peer_id, lookup_id, block_root, by_head_count)? + } else { + self.send_block_by_root_request(peer_id, lookup_id, block_root)? + }, + )) + } + + /// Sends a `beacon_blocks_by_root` request to `peer_id` for the single block `block_root`. + fn send_block_by_root_request( + &mut self, + peer_id: PeerId, + lookup_id: SingleLookupId, + block_root: Hash256, + ) -> Result { let id = SingleLookupReqId { lookup_id, req_id: self.next_id(), @@ -955,7 +1063,60 @@ impl SyncNetworkContext { request_span, ); - Ok(LookupRequestResult::RequestSent(id.req_id)) + Ok(id.req_id) + } + + /// Sends a `beacon_blocks_by_head` request to `peer_id`, fetching `block_root` and a run of its + /// ancestors in a single request. + fn send_blocks_by_head( + &mut self, + peer_id: PeerId, + lookup_id: SingleLookupId, + block_root: Hash256, + count: u64, + ) -> Result { + let id = SingleLookupReqId { + lookup_id, + req_id: self.next_id(), + }; + // Lookup sync event safety: see `send_block_by_root_request`. The same guarantees apply, + // with the events handled by `Self::on_blocks_by_head_response`. + let request = BlocksByHeadRequest { + beacon_root: block_root, + count, + }; + self.network_send + .send(NetworkMessage::SendRequest { + peer_id, + request: RequestType::BlocksByHead(request), + app_request_id: AppRequestId::Sync(SyncRequestId::BlocksByHead(id)), + }) + .map_err(|_| RpcRequestSendError::InternalError("network send error".to_owned()))?; + + debug!( + method = "BlocksByHead", + ?block_root, + peer = %peer_id, + %id, + "Sync RPC request sent" + ); + + let request_span = debug_span!( + parent: Span::current(), + "lh_outgoing_block_by_head_request", + %block_root, + ); + self.blocks_by_head_requests.insert( + id, + peer_id, + // false = the peer may return fewer blocks than requested (e.g. reached genesis or + // finalization), so the request completes on stream termination. + false, + BlocksByHeadRequestItems::new(block_root, count as usize), + request_span, + ); + + Ok(id.req_id) } /// Request a payload envelope for a block root via PayloadEnvelopesByRoot RPC. @@ -1472,18 +1633,36 @@ impl SyncNetworkContext { id: SingleLookupReqId, peer_id: PeerId, rpc_event: RpcEvent>>, - ) -> Option>>> { + ) -> Option>> { + // `blocks_by_root_requests` enforces that exactly one chunk = one block is returned, so the + // `AncestorBlocks` always has a single block and no ancestors. let resp = self.blocks_by_root_requests.on_response(id, rpc_event); let resp = resp.map(|res| { res.and_then(|mut blocks| { - // Enforce that exactly one chunk = one block is returned. ReqResp behavior limits the - // response count to at most 1. - match blocks.pop() { - Some(block) => Ok(block), - // Should never happen, `blocks_by_root_requests` enforces that we receive at least - // 1 chunk. - None => Err(LookupVerifyError::NotEnoughResponsesReturned { actual: 0 }.into()), - } + let block = blocks.pop().ok_or(RpcResponseError::from( + LookupVerifyError::NotEnoughResponsesReturned { actual: 0 }, + ))?; + Ok(AncestorBlocks::from_single(block)) + }) + }); + self.on_rpc_response_result(resp, peer_id) + } + + pub(crate) fn on_blocks_by_head_response( + &mut self, + id: SingleLookupReqId, + peer_id: PeerId, + rpc_event: RpcEvent>>, + ) -> Option>> { + // The response has the requested `beacon_root` first (the chain tip, becoming `first_block`) + // followed by its ancestors in descending slot order. + let resp = self.blocks_by_head_requests.on_response(id, rpc_event); + let resp = resp.map(|res| { + res.and_then(|blocks| { + let value = AncestorBlocks::from_vec(blocks).ok_or(RpcResponseError::from( + LookupVerifyError::NotEnoughResponsesReturned { actual: 0 }, + ))?; + Ok(value) }) }); self.on_rpc_response_result(resp, peer_id) @@ -1829,6 +2008,7 @@ impl SyncNetworkContext { pub(crate) fn register_metrics(&self) { for (id, count) in [ ("blocks_by_root", self.blocks_by_root_requests.len()), + ("blocks_by_head", self.blocks_by_head_requests.len()), ( "data_columns_by_root", self.data_columns_by_root_requests.len(), diff --git a/beacon_node/network/src/sync/network_context/requests.rs b/beacon_node/network/src/sync/network_context/requests.rs index b340064746c..7c3d560bdf0 100644 --- a/beacon_node/network/src/sync/network_context/requests.rs +++ b/beacon_node/network/src/sync/network_context/requests.rs @@ -8,6 +8,7 @@ use tracing::{Span, debug}; use types::{Hash256, Slot}; pub use blobs_by_range::BlobsByRangeRequestItems; +pub use blocks_by_head::BlocksByHeadRequestItems; pub use blocks_by_range::BlocksByRangeRequestItems; pub use blocks_by_root::{BlocksByRootRequestItems, BlocksByRootSingleRequest}; pub use data_columns_by_range::DataColumnsByRangeRequestItems; @@ -24,6 +25,7 @@ use crate::metrics; use super::{RpcEvent, RpcResponseError, RpcResponseResult}; mod blobs_by_range; +mod blocks_by_head; mod blocks_by_range; mod blocks_by_root; mod data_columns_by_range; diff --git a/beacon_node/network/src/sync/network_context/requests/blocks_by_head.rs b/beacon_node/network/src/sync/network_context/requests/blocks_by_head.rs new file mode 100644 index 00000000000..5258647bdf0 --- /dev/null +++ b/beacon_node/network/src/sync/network_context/requests/blocks_by_head.rs @@ -0,0 +1,58 @@ +use super::{ActiveRequestItems, LookupVerifyError}; +use beacon_chain::get_block_root; +use std::sync::Arc; +use types::{EthSpec, Hash256, SignedBeaconBlock}; + +/// Accumulates the response of a `blocks_by_head` request. The responder walks the parent chain of +/// `beacon_root` (inclusive) and emits up to `count` blocks in descending slot order, so the first +/// chunk MUST be `beacon_root` and the rest are its ancestors. +pub struct BlocksByHeadRequestItems { + beacon_root: Hash256, + count: usize, + items: Vec>>, +} + +impl BlocksByHeadRequestItems { + pub fn new(beacon_root: Hash256, count: usize) -> Self { + Self { + beacon_root, + count, + items: vec![], + } + } +} + +impl ActiveRequestItems for BlocksByHeadRequestItems { + type Item = Arc>; + + /// Append a response chunk. The blocks must form a parent chain in strictly descending slot + /// order, with the chain tip (first chunk) equal to the requested `beacon_root`. + /// The active request SHOULD be dropped after `add` returns an error. + fn add(&mut self, block: Self::Item) -> Result { + let block_root = get_block_root(&block); + if let Some(child) = self.items.last() { + // Each subsequent block must be the parent of the previous one, with a strictly lower + // slot, so the response forms a contiguous descending chain. + if child.parent_root() != block_root { + return Err(LookupVerifyError::UnrequestedBlockRoot(block_root)); + } + if block.slot() >= child.slot() { + return Err(LookupVerifyError::UnrequestedSlot(block.slot())); + } + } else { + // The first block returned is the chain tip and must match the requested root. + if self.beacon_root != block_root { + return Err(LookupVerifyError::UnrequestedBlockRoot(block_root)); + } + } + + self.items.push(block); + // The peer may return fewer blocks than requested (e.g. reached genesis or finalization), + // in which case the request completes on stream termination instead. + Ok(self.items.len() >= self.count) + } + + fn consume(&mut self) -> Vec { + std::mem::take(&mut self.items) + } +} diff --git a/beacon_node/network/src/sync/tests/lookups.rs b/beacon_node/network/src/sync/tests/lookups.rs index f9afb910d9c..af3ca214864 100644 --- a/beacon_node/network/src/sync/tests/lookups.rs +++ b/beacon_node/network/src/sync/tests/lookups.rs @@ -5,10 +5,13 @@ use crate::network_beacon_processor::sync_methods::WhichPeerToPenalize; use crate::network_beacon_processor::{ ChainSegmentProcessId, InvalidBlockStorage, NetworkBeaconProcessor, }; +use crate::sync::block_lookups::{ + BLOCKS_BY_HEAD_CHAIN_REQUEST_COUNT, BLOCKS_BY_HEAD_LONE_REQUEST_COUNT, +}; use crate::sync::block_lookups::{BlockLookupSummary, PARENT_DEPTH_TOLERANCE}; use crate::sync::{ SyncMessage, - manager::{BatchProcessResult, BlockProcessType, SyncManager}, + manager::{BatchProcessResult, BlockProcessType, SLOT_IMPORT_TOLERANCE, SyncManager}, }; use beacon_chain::block_verification_types::LookupBlock; use beacon_chain::custody_context::NodeCustodyType; @@ -27,7 +30,7 @@ use itertools::Itertools; use lighthouse_network::discovery::CombinedKey; use lighthouse_network::{ NetworkConfig, NetworkGlobals, PeerAction, PeerId, - rpc::{RPCError, RequestType}, + rpc::{Protocol, RPCError, RequestType}, service::api_types::{AppRequestId, SyncRequestId}, types::SyncState, }; @@ -248,7 +251,7 @@ struct FullEmptyFork { } impl TestRig { - pub(crate) fn new(test_rig_config: TestRigConfig) -> Self { + pub(crate) fn from_config(test_rig_config: TestRigConfig) -> Self { // Use `fork_from_env` logic to set correct fork epochs let spec = Arc::new(test_spec::()); let clock = TestingSlotClock::new( @@ -346,20 +349,31 @@ impl TestRig { complete_strategy: <_>::default(), initial_block_lookups_metrics: <_>::default(), fulu_test_type: test_rig_config.fulu_test_type, + // Peers advertise only `beacon_blocks_by_root` by default; tests that exercise + // `beacon_blocks_by_head` opt in via `TestRig::new`. + by_head_support: ByHeadSupport::default(), } } pub fn default() -> Self { // Before Fulu, FuluTestType is irrelevant - Self::new(TestRigConfig { + Self::from_config(TestRigConfig { fulu_test_type: FuluTestType::WeFullnodeThemSupernode, node_custody_type_override: None, }) } + /// A rig whose peers advertise `by_head_support`, so block lookups can fetch over + /// `beacon_blocks_by_head`. `default()` uses `ByHeadSupport::Unsupported`. + fn new(by_head_support: ByHeadSupport) -> Self { + let mut rig = Self::default(); + rig.by_head_support = by_head_support; + rig + } + #[allow(dead_code)] pub fn with_custody_type(node_custody_type: NodeCustodyType) -> Self { - Self::new(TestRigConfig { + Self::from_config(TestRigConfig { fulu_test_type: FuluTestType::WeFullnodeThemSupernode, node_custody_type_override: Some(node_custody_type), }) @@ -860,6 +874,31 @@ impl TestRig { self.send_rpc_envelopes_response(req_id, peer_id, &envelopes); } + (RequestType::BlocksByHead(req), AppRequestId::Sync(req_id)) => { + let blocks = if self.complete_strategy.return_no_blocks_n_times > 0 { + self.complete_strategy.return_no_blocks_n_times -= 1; + vec![] + } else if self.complete_strategy.return_wrong_blocks_n_times > 0 { + self.complete_strategy.return_wrong_blocks_n_times -= 1; + vec![Arc::new(self.rand_block())] + } else { + // Walk the parent chain from `beacon_root`, returning up to `count` blocks in + // descending slot order (the same shape the responder produces). + let mut blocks = vec![]; + let mut block_root = req.beacon_root; + while (blocks.len() as u64) < req.count { + let Some(block) = self.network_blocks_by_root.get(&block_root) else { + break; + }; + let block = block.block_cloned(); + block_root = block.parent_root(); + blocks.push(block); + } + blocks + }; + self.send_rpc_blocks_response(req_id, peer_id, &blocks); + } + (RequestType::Status(_req), AppRequestId::Router) => { // Ignore Status requests for now } @@ -1632,8 +1671,10 @@ impl TestRig { // Test setup - fn new_after_fulu() -> Option { - genesis_fork().fulu_enabled().then(Self::default) + fn new_after_fulu(by_head_support: ByHeadSupport) -> Option { + genesis_fork() + .fulu_enabled() + .then(|| Self::new(by_head_support)) } fn new_after_gloas() -> Option { @@ -1642,7 +1683,7 @@ impl TestRig { pub fn new_fulu_peer_test(fulu_test_type: FuluTestType) -> Option { genesis_fork().fulu_enabled().then(|| { - Self::new(TestRigConfig { + Self::from_config(TestRigConfig { fulu_test_type, node_custody_type_override: None, }) @@ -1819,9 +1860,21 @@ impl TestRig { self.log(&format!( "Added new peer for testing {peer_id:?}, custody: {peer_custody_str}" )); + if self.by_head_support == ByHeadSupport::Supported { + self.set_peer_supports_by_head(peer_id); + } peer_id } + /// Mark a peer as advertising the `beacon_blocks_by_head` protocol so block lookups will use it. + fn set_peer_supports_by_head(&mut self, peer_id: PeerId) { + self.network_globals + .peers + .write() + .__set_supported_protocols(&peer_id, HashSet::from([Protocol::BlocksByHead])) + .unwrap(); + } + pub fn new_connected_supernode_peer(&mut self) -> PeerId { let key = self.determinstic_key(); let peer_id = self @@ -1832,6 +1885,9 @@ impl TestRig { self.log(&format!( "Added new peer for testing {peer_id:?}, custody: supernode" )); + if self.by_head_support == ByHeadSupport::Supported { + self.set_peer_supports_by_head(peer_id); + } peer_id } @@ -1840,10 +1896,15 @@ impl TestRig { /// a connection is established. pub fn new_connected_supernode_peer_no_metadata_custody_subnet(&mut self) -> PeerId { let key = self.determinstic_key(); - self.network_globals - .peers - .write() - .__add_connected_peer(true, key, &self.harness.spec) + let peer_id = + self.network_globals + .peers + .write() + .__add_connected_peer(true, key, &self.harness.spec); + if self.by_head_support == ByHeadSupport::Supported { + self.set_peer_supports_by_head(peer_id); + } + peer_id } /// Update the peer's custody subnet in PeerDB and send a `UpdatedPeerCgc` message to sync. @@ -2041,84 +2102,104 @@ fn stable_arbitrary() { ); } +/// Generates the whole lookup test suite for a single peer kind across the given depths. macro_rules! run_lookups_tests_for_depths { - ($($depth:literal),+ $(,)?) => { + ($peer_kind:ident, $protocol:expr, depths: [$($depth:literal),+ $(,)?]) => { paste::paste! { $( #[tokio::test] - async fn []() { - happy_path_unknown_attestation($depth).await; + async fn []() { + happy_path_unknown_attestation($depth, $protocol).await; } #[tokio::test] - async fn []() { - happy_path_unknown_block_parent($depth).await; + async fn []() { + happy_path_unknown_block_parent($depth, $protocol).await; } #[tokio::test] - async fn []() { - happy_path_unknown_data_parent($depth).await; + async fn []() { + happy_path_unknown_data_parent($depth, $protocol).await; } #[tokio::test] - async fn []() { - happy_path_multiple_triggers($depth).await; + async fn []() { + happy_path_multiple_triggers($depth, $protocol).await; } #[tokio::test] - async fn []() { - bad_peer_empty_block_response($depth).await; + async fn []() { + bad_peer_empty_block_response($depth, $protocol).await; } #[tokio::test] - async fn []() { - bad_peer_empty_data_response($depth).await; + async fn []() { + bad_peer_empty_data_response($depth, $protocol).await; } #[tokio::test] - async fn []() { - bad_peer_too_few_data_response($depth).await; + async fn []() { + bad_peer_too_few_data_response($depth, $protocol).await; } #[tokio::test] - async fn []() { - bad_peer_wrong_block_response($depth).await; + async fn []() { + bad_peer_wrong_block_response($depth, $protocol).await; } #[tokio::test] - async fn []() { - bad_peer_wrong_data_response($depth).await; + async fn []() { + bad_peer_wrong_data_response($depth, $protocol).await; } #[tokio::test] - async fn []() { - bad_peer_rpc_failure($depth).await; + async fn []() { + bad_peer_rpc_failure($depth, $protocol).await; } #[tokio::test] - async fn []() { - too_many_download_failures($depth).await; + async fn []() { + too_many_download_failures($depth, $protocol).await; } #[tokio::test] - async fn []() { - too_many_processing_failures($depth).await; + async fn []() { + too_many_processing_failures($depth, $protocol).await; } #[tokio::test] - async fn []() { - peer_disconnected_then_rpc_error($depth).await; + async fn []() { + peer_disconnected_then_rpc_error($depth, $protocol).await; } )+ } }; } -run_lookups_tests_for_depths!(1, 2); +/// Runs the whole lookup suite against each peer kind (by-root only vs. by-head capable) for all +/// depths, by invoking [`run_lookups_tests_for_depths`] once per peer kind. +macro_rules! run_lookups_tests { + ( + peers: [$($peer_kind:ident => $protocol:expr),+ $(,)?], + depths: $depths:tt $(,)? + ) => { + $( + run_lookups_tests_for_depths!($peer_kind, $protocol, depths: $depths); + )+ + }; +} + +run_lookups_tests!( + peers: [ + by_root => ByHeadSupport::Unsupported, + by_head => ByHeadSupport::Supported, + ], + depths: [1, 2], +); /// Assert that lookup sync succeeds with the happy case -async fn happy_path_unknown_attestation(depth: usize) { - let mut r = TestRig::default(); +async fn happy_path_unknown_attestation(depth: usize, by_head_support: ByHeadSupport) { + let mut r = TestRig::new(by_head_support); // We get attestation for a block descendant (depth) blocks of current head r.build_chain_and_trigger_last_block(depth).await; // Complete the request with good peer behaviour @@ -2126,8 +2207,8 @@ async fn happy_path_unknown_attestation(depth: usize) { r.assert_successful_lookup_sync(); } -async fn happy_path_unknown_block_parent(depth: usize) { - let mut r = TestRig::default(); +async fn happy_path_unknown_block_parent(depth: usize, by_head_support: ByHeadSupport) { + let mut r = TestRig::new(by_head_support); r.build_chain(depth).await; r.trigger_with_last_unknown_block_parent(); r.simulate(SimulateConfig::happy_path()).await; @@ -2142,8 +2223,8 @@ async fn happy_path_unknown_block_parent(depth: usize) { } /// Assert that sync completes from an UnknownDataColumnParent -async fn happy_path_unknown_data_parent(depth: usize) { - let Some(mut r) = TestRig::new_after_fulu() else { +async fn happy_path_unknown_data_parent(depth: usize, by_head_support: ByHeadSupport) { + let Some(mut r) = TestRig::new_after_fulu(by_head_support) else { return; }; // Fulu-only: the `UnknownDataColumnParent` trigger doesn't exist post-Gloas (columns ride in @@ -2158,8 +2239,8 @@ async fn happy_path_unknown_data_parent(depth: usize) { } /// Assert that multiple trigger types don't create extra lookups -async fn happy_path_multiple_triggers(depth: usize) { - let Some(mut r) = TestRig::new_after_fulu() else { +async fn happy_path_multiple_triggers(depth: usize, by_head_support: ByHeadSupport) { + let Some(mut r) = TestRig::new_after_fulu(by_head_support) else { return; }; // + 1, because the unknown parent trigger needs two new blocks @@ -2179,8 +2260,8 @@ async fn happy_path_multiple_triggers(depth: usize) { // Test bad behaviour of peers /// Assert that if peer responds with no blocks, we downscore, and retry the same lookup -async fn bad_peer_empty_block_response(depth: usize) { - let mut r = TestRig::default(); +async fn bad_peer_empty_block_response(depth: usize, by_head_support: ByHeadSupport) { + let mut r = TestRig::new(by_head_support); r.build_chain_and_trigger_last_block(depth).await; // Simulate that peer returns empty response once, then good behaviour r.simulate(SimulateConfig::new().return_no_blocks_once()) @@ -2194,8 +2275,8 @@ async fn bad_peer_empty_block_response(depth: usize) { } /// Assert that if peer responds with no columns, we downscore, and retry the same lookup. -async fn bad_peer_empty_data_response(depth: usize) { - let Some(mut r) = TestRig::new_after_fulu() else { +async fn bad_peer_empty_data_response(depth: usize, by_head_support: ByHeadSupport) { + let Some(mut r) = TestRig::new_after_fulu(by_head_support) else { return; }; r.build_chain_and_trigger_last_block(depth).await; @@ -2212,8 +2293,8 @@ async fn bad_peer_empty_data_response(depth: usize) { /// Assert that if peer responds with not enough columns, we downscore, and retry the same /// lookup. -async fn bad_peer_too_few_data_response(depth: usize) { - let Some(mut r) = TestRig::new_after_fulu() else { +async fn bad_peer_too_few_data_response(depth: usize, by_head_support: ByHeadSupport) { + let Some(mut r) = TestRig::new_after_fulu(by_head_support) else { return; }; r.build_chain_and_trigger_last_block(depth).await; @@ -2229,8 +2310,8 @@ async fn bad_peer_too_few_data_response(depth: usize) { } /// Assert that if peer responds with bad blocks, we downscore, and retry the same lookup -async fn bad_peer_wrong_block_response(depth: usize) { - let mut r = TestRig::default(); +async fn bad_peer_wrong_block_response(depth: usize, by_head_support: ByHeadSupport) { + let mut r = TestRig::new(by_head_support); r.build_chain_and_trigger_last_block(depth).await; r.simulate(SimulateConfig::new().return_wrong_blocks_once()) .await; @@ -2241,8 +2322,8 @@ async fn bad_peer_wrong_block_response(depth: usize) { } /// Assert that if peer responds with bad columns, we downscore, and retry the same lookup. -async fn bad_peer_wrong_data_response(depth: usize) { - let Some(mut r) = TestRig::new_after_fulu() else { +async fn bad_peer_wrong_data_response(depth: usize, by_head_support: ByHeadSupport) { + let Some(mut r) = TestRig::new_after_fulu(by_head_support) else { return; }; r.build_chain_and_trigger_last_block(depth).await; @@ -2260,8 +2341,8 @@ async fn bad_peer_wrong_data_response(depth: usize) { } /// Assert that on network error, we DON'T downscore, and retry the same lookup -async fn bad_peer_rpc_failure(depth: usize) { - let mut r = TestRig::default(); +async fn bad_peer_rpc_failure(depth: usize, by_head_support: ByHeadSupport) { + let mut r = TestRig::new(by_head_support); r.build_chain_and_trigger_last_block(depth).await; r.simulate(SimulateConfig::new().return_rpc_error(RPCError::UnsupportedProtocol)) .await; @@ -2272,8 +2353,8 @@ async fn bad_peer_rpc_failure(depth: usize) { // Test retry logic /// Assert that on too many download failures the lookup fails, but we can still sync -async fn too_many_download_failures(depth: usize) { - let mut r = TestRig::default(); +async fn too_many_download_failures(depth: usize, by_head_support: ByHeadSupport) { + let mut r = TestRig::new(by_head_support); r.build_chain_and_trigger_last_block(depth).await; // Simulate that a peer always returns empty r.simulate(SimulateConfig::new().return_no_blocks_always()) @@ -2291,8 +2372,8 @@ async fn too_many_download_failures(depth: usize) { } /// Assert that on too many processing failures the lookup fails, but we can still sync -async fn too_many_processing_failures(depth: usize) { - let mut r = TestRig::default(); +async fn too_many_processing_failures(depth: usize, by_head_support: ByHeadSupport) { + let mut r = TestRig::new(by_head_support); r.build_chain_and_trigger_last_block(depth).await; // Simulate that a peer always returns empty r.simulate( @@ -2321,7 +2402,7 @@ async fn too_many_processing_failures(depth: usize) { #[tokio::test] /// Assert that multiple trigger types don't create extra lookups async fn unknown_parent_does_not_add_peers_to_itself() { - let Some(mut r) = TestRig::new_after_fulu() else { + let Some(mut r) = TestRig::new_after_fulu(ByHeadSupport::Unsupported) else { return; }; // 2, because the unknown parent trigger needs two new blocks @@ -2390,8 +2471,8 @@ async fn test_single_block_lookup_duplicate_response() { } /// Assert that when peers disconnect the lookups are not dropped (kept with zero peers) -async fn peer_disconnected_then_rpc_error(depth: usize) { - let mut r = TestRig::default(); +async fn peer_disconnected_then_rpc_error(depth: usize, by_head_support: ByHeadSupport) { + let mut r = TestRig::new(by_head_support); r.build_chain_and_trigger_last_block(depth).await; r.assert_single_lookups_count(1); // The peer disconnect event reaches sync before the rpc error. @@ -2410,6 +2491,60 @@ async fn peer_disconnected_then_rpc_error(depth: usize) { r.assert_single_lookups_count(1); } +/// A lone lookup over `beacon_blocks_by_head` fetches only a few ancestors; once the chain proves +/// deep the next request fetches a large batch, caching every block, so no `beacon_blocks_by_root` +/// requests are needed. +#[tokio::test] +async fn blocks_by_head_request_count_grows_for_a_chain() { + // Peers advertise `beacon_blocks_by_head`, so lookups fetch via it. + let mut r = TestRig::new(ByHeadSupport::Supported); + // Deep enough that the lone count doesn't resolve it, forcing a second (chain) request, while + // staying under `PARENT_DEPTH_TOLERANCE`. + r.build_chain_and_trigger_last_block(BLOCKS_BY_HEAD_CHAIN_REQUEST_COUNT as usize) + .await; + r.simulate(SimulateConfig::happy_path()).await; + + r.assert_successful_lookup_sync(); + // The lone first lookup fetches only a few ancestors; once the chain is proven deep the next + // request fetches a large batch. No by-root is used. + let by_head_counts = r + .requests + .iter() + .filter_map(|(req, _)| match req { + RequestType::BlocksByHead(req) => Some(req.count), + _ => None, + }) + .collect::>(); + assert_eq!( + by_head_counts, + vec![ + BLOCKS_BY_HEAD_LONE_REQUEST_COUNT, + BLOCKS_BY_HEAD_CHAIN_REQUEST_COUNT + ] + ); + assert_eq!( + r.requests_count().get("BlocksByRoot").copied().unwrap_or(0), + 0 + ); +} + +/// A peer serves a block whose slot is far ahead of the wall-clock slot. It must be dropped, not +/// turned into a parent lookup / forced range sync to a bogus future head. +#[tokio::test] +async fn lookup_drops_block_too_far_in_future() { + let mut r = TestRig::default(); + // Tip is `> SLOT_IMPORT_TOLERANCE` slots ahead of where we hold the clock, so the served block + // looks far in the future. Set the clock directly (back) to dodge the harness epoch guard. + let block_root = r.build_chain(SLOT_IMPORT_TOLERANCE + 2).await; + r.harness.chain.slot_clock.set_slot(1); + let peer = r.new_connected_supernode_peer(); + r.trigger_unknown_block_from_attestation(block_root, peer); + r.simulate(SimulateConfig::happy_path()).await; + + // The lookup is dropped on download; no parent lookup and no range sync are spawned. + r.assert_failed_lookup_sync(); +} + #[tokio::test] /// A lookup that loses its only peer while still waiting to download the block must not report /// itself as awaiting an event, else `drop_lookups_without_peers` skips it and it gets stuck. @@ -2603,7 +2738,7 @@ async fn test_same_chain_race_condition() { /// Assert that if the lookup's block is in the da_checker we don't download it again (pre-Gloas). async fn block_in_da_checker_skips_download_fulu() { // Only post-Fulu, as the block needs custody columns to remain in the da_checker. - let Some(mut r) = TestRig::new_after_fulu() else { + let Some(mut r) = TestRig::new_after_fulu(ByHeadSupport::Unsupported) else { return; }; // Pre-Gloas only; the Gloas equivalent is `block_in_da_checker_skips_download_gloas`. diff --git a/beacon_node/network/src/sync/tests/mod.rs b/beacon_node/network/src/sync/tests/mod.rs index 4e185cc0817..f2214deec08 100644 --- a/beacon_node/network/src/sync/tests/mod.rs +++ b/beacon_node/network/src/sync/tests/mod.rs @@ -53,6 +53,15 @@ type T = Witness; /// +-----------------+ /// | - BlockLookups | /// +-----------------+ +/// Whether the rig's peers advertise `beacon_blocks_by_head` (so block lookups fetch via it) or +/// only `beacon_blocks_by_root`. +#[derive(Clone, Copy, Default, PartialEq)] +enum ByHeadSupport { + Supported, + #[default] + Unsupported, +} + struct TestRig { /// Receiver for `BeaconProcessor` events (e.g. block processing results). beacon_processor_rx: mpsc::Receiver>, @@ -88,6 +97,8 @@ struct TestRig { initial_block_lookups_metrics: BlockLookupsMetrics, /// Fulu test type fulu_test_type: FuluTestType, + /// Whether peers added to the rig advertise `beacon_blocks_by_head` so lookups fetch via it. + by_head_support: ByHeadSupport, } enum FuluTestType { diff --git a/consensus/fork_choice/src/fork_choice.rs b/consensus/fork_choice/src/fork_choice.rs index 95c8f70a044..f56d7e0e629 100644 --- a/consensus/fork_choice/src/fork_choice.rs +++ b/consensus/fork_choice/src/fork_choice.rs @@ -1591,10 +1591,13 @@ where /// Returns `true` if the block's parent is imported (and, for a post-Gloas FULL child, its /// parent's payload is imported too). See [`Self::get_parent_import_status`]. pub fn is_parent_imported(&self, block: &SignedBeaconBlock) -> bool { + // A block that is itself in fork choice (e.g. genesis, or a finalized block whose parent has + // been pruned) had an imported parent. `contains_block` is used rather than `get_block` + // because the anchor block is present in the index but has no queryable proto-node. matches!( self.get_parent_import_status(block), ParentImportStatus::Imported(_) - ) + ) || self.contains_block(&block.canonical_root()) } /// Returns the import status of the parent of `block`.