diff --git a/beacon_node/network/src/sync/block_lookups/mod.rs b/beacon_node/network/src/sync/block_lookups/mod.rs index ef9807f0372..eb845c1c803 100644 --- a/beacon_node/network/src/sync/block_lookups/mod.rs +++ b/beacon_node/network/src/sync/block_lookups/mod.rs @@ -349,16 +349,16 @@ impl BlockLookups { } // Do not re-request a block that is already being requested - if let Some((&lookup_id, lookup)) = self + if let Some((&lookup_id, _lookup)) = self .single_block_lookups .iter_mut() .find(|(_id, lookup)| lookup.is_for_block(block_root)) { - if let Some(block_component) = block_component { - let imported = lookup.add_child_components(block_component); - if !imported { - debug!(?block_root, "Lookup child component ignored"); - } + // A lookup for this root already exists. Drop any gossip-provided component: injecting + // it could collide with an in-flight request, and the existing lookup will complete on + // its own. + if block_component.is_some() { + debug!(?block_root, "Lookup child component ignored"); } if let Err(e) = self.add_peers_to_lookup_and_ancestors(lookup_id, peers, peer_type, cx) @@ -389,16 +389,18 @@ impl BlockLookups { // 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 id = cx.next_id(); + let lookup = SingleBlockLookup::new( + block_root, + peers, + peer_type, + id, + block_component, + awaiting_parent, + cx.spec(), + ); let _guard = lookup.span.clone().entered(); - // Add block components to the new request - if let Some(block_component) = block_component { - lookup.add_child_components(block_component); - } - - let id = lookup.id; let lookup = match self.single_block_lookups.entry(id) { Entry::Vacant(entry) => entry.insert(lookup), Entry::Occupied(_) => { @@ -562,7 +564,7 @@ impl BlockLookups { for (id, lookup) in self.single_block_lookups.iter_mut() { if lookup.is_awaiting_parent(parent_root, imported_parent) { - lookup.resolve_awaiting_parent(); + lookup.resolve_awaiting_parent(cx.spec()); debug!( ?imported_parent, id, 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..4993ed7ea77 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 @@ -9,7 +9,6 @@ use crate::sync::network_context::{ SendErrorProcessor, SyncNetworkContext, }; use beacon_chain::BeaconChainTypes; -use beacon_chain::block_verification_types::AsBlock; use educe::Educe; use lighthouse_network::service::api_types::Id; use parking_lot::RwLock; @@ -20,7 +19,7 @@ use store::Hash256; use strum::IntoStaticStr; use tracing::{Span, debug_span}; use types::{ - DataColumnSidecarList, EthSpec, ExecutionBlockHash, SignedBeaconBlock, + ChainSpec, DataColumnSidecarList, EthSpec, ExecutionBlockHash, SignedBeaconBlock, SignedExecutionPayloadEnvelope, Slot, }; @@ -105,12 +104,7 @@ impl BlockRequest { #[derive(Debug)] enum DataRequest { - WaitingForBlock, Request { - slot: Slot, - /// Peers to fetch the data columns from. Pre-Gloas this is the lookup's `peers`; for FULL - /// Gloas blocks this is the `gloas_child_peers` set proven to hold the columns. - peers: PeerSet, state: SingleLookupRequestState>, }, NoData, @@ -119,7 +113,6 @@ enum DataRequest { impl DataRequest { fn is_complete(&self) -> bool { match &self { - DataRequest::WaitingForBlock => false, DataRequest::Request { state, .. } => state.is_processed(), DataRequest::NoData => true, } @@ -131,13 +124,10 @@ impl DataRequest { /// columns are fetched and processed by `DataRequest`. #[derive(Debug)] enum PayloadRequest { - /// Block not yet downloaded, can't tell if a payload is needed. - WaitingForBlock, /// Post-Gloas block: an execution payload envelope must be fetched and processed *if* the block /// is FULL. We can't tell FULL from EMPTY from the block alone: only a FULL child of this block - /// proves a payload was published, which is signalled by `peers` becoming non-empty. + /// proves a payload was published, which is signalled by `data_peers` becoming non-empty. Request { - peers: PeerSet, state: SingleLookupRequestState>>, }, /// Pre-Gloas block: no payload envelope exists, nothing to fetch. @@ -147,7 +137,6 @@ enum PayloadRequest { impl PayloadRequest { fn is_complete(&self) -> bool { match &self { - PayloadRequest::WaitingForBlock => false, PayloadRequest::Request { state, .. } => state.is_processed(), PayloadRequest::PreGloas => true, } @@ -188,9 +177,7 @@ pub enum ImportedParent { pub struct SingleBlockLookup { pub id: Id, block_root: Hash256, - block_request: BlockRequest, - data_request: DataRequest, - payload_request: PayloadRequest, + status: Status, /// Peers that claim to have imported this set of block components. This state is shared with /// the custody request to have an updated view of the peers that claim to have imported the /// block associated with this lookup. The peer set of a lookup can change rapidly, and faster @@ -202,18 +189,121 @@ pub struct SingleBlockLookup { /// block's payload envelope and data columns. #[educe(Debug(method(fmt_peer_map_as_len)))] gloas_child_peers: HashMap, + /// Set while this lookup knows it is awaiting a parent (e.g. created from a gossip + /// `UnknownParentBlock`, or a downloaded block that returned `ParentUnknown`). While set, the + /// block is held but not submitted for processing. awaiting_parent: Option, created: Instant, pub(crate) span: Span, } +#[derive(Debug)] +#[allow(clippy::large_enum_variant)] +enum Status { + FetchingHeader(BlockRequest), + WaitingForParent { + header: RichHeader, + /// Peers to fetch the data columns from. Pre-Gloas this is the lookup's `peers`; for FULL + /// Gloas blocks this is the `gloas_child_peers` set proven to hold the columns. + data_peers: PeerSet, + }, + ComponentsDownload { + header: RichHeader, + /// Peers to fetch the data columns from. Pre-Gloas this is the lookup's `peers`; for FULL + /// Gloas blocks this is the `gloas_child_peers` set proven to hold the columns. + data_peers: PeerSet, + block_request: BlockRequest, + data_request: DataRequest, + payload_request: PayloadRequest, + }, +} + +#[derive(Debug, Clone)] +struct RichHeader { + slot: Slot, + /// Whether the block expects blobs (`num_expected_blobs() > 0`). The full "columns required" + /// decision additionally gates on the peer-das fork (done in `new_components_download`); the + /// DA-boundary gate is skipped because lookup sync only runs inside the data-availability window. + has_data: bool, + bid_block_hash: Option, +} + +impl RichHeader { + fn from_block(block: &SignedBeaconBlock) -> Self { + Self { + slot: block.slot(), + has_data: block.num_expected_blobs() > 0, + bid_block_hash: block.payload_bid_block_hash().ok(), + } + } +} + +impl Status { + fn is_complete(&self) -> bool { + match self { + Status::FetchingHeader(_) => false, + Status::WaitingForParent { .. } => false, + Status::ComponentsDownload { + block_request, + data_request, + payload_request, + .. + } => { + block_request.is_complete() + && data_request.is_complete() + && payload_request.is_complete() + } + } + } + + fn new_components_download( + header: RichHeader, + data_peers: PeerSet, + block_request: BlockRequest, + spec: &ChainSpec, + ) -> Self { + Self::ComponentsDownload { + header: header.clone(), + data_peers, + block_request, + data_request: if header.has_data + && spec.is_peer_das_enabled_for_epoch(header.slot.epoch(E::slots_per_epoch())) + { + DataRequest::Request { + state: SingleLookupRequestState::new(), + } + } else { + DataRequest::NoData + }, + payload_request: if spec.fork_name_at_slot::(header.slot).gloas_enabled() { + PayloadRequest::Request { + state: SingleLookupRequestState::new(), + } + } else { + PayloadRequest::PreGloas + }, + } + } + + fn peek_block_header(&self) -> Option<&RichHeader> { + match self { + Status::FetchingHeader(_) => None, + Status::WaitingForParent { header, .. } | Status::ComponentsDownload { header, .. } => { + Some(header) + } + } + } +} + impl SingleBlockLookup { pub fn new( requested_block_root: Hash256, peers: &[PeerId], peer_type: &PeerType, id: Id, + block_component: Option>, awaiting_parent: Option, + spec: &ChainSpec, ) -> Self { let lookup_span = debug_span!( "lh_single_block_lookup", @@ -230,12 +320,29 @@ impl SingleBlockLookup { } } + // If we already hold the full block (gossip `UnknownParentBlock`), skip the header-fetch + // stage and seed it for processing. We do NOT pre-park on the gossip parent hint: only the + // beacon processor authoritatively reports `ParentUnknown`, so let processing decide. + let status = match block_component { + Some(BlockComponent::Block(block)) => { + let header = RichHeader::from_block(&block.value); + let data_peers = match header.bid_block_hash { + Some(bid_block_hash) => { + gloas_child_peers.entry(bid_block_hash).or_default().clone() + } + None => block_peers.clone(), + }; + let mut block_request = BlockRequest::new(); + block_request.state.insert_verified_response(block); + Status::new_components_download(header, data_peers, block_request, spec) + } + Some(BlockComponent::Sidecar) | None => Status::FetchingHeader(BlockRequest::new()), + }; + Self { id, block_root: requested_block_root, - block_request: BlockRequest::new(), - data_request: DataRequest::WaitingForBlock, - payload_request: PayloadRequest::WaitingForBlock, + status, peers: block_peers, gloas_child_peers, awaiting_parent, @@ -246,24 +353,13 @@ impl SingleBlockLookup { /// Return the slot of this lookup's block if it's currently cached pub fn peek_downloaded_block_slot(&self) -> Option { - self.block_request - .state - .peek_downloaded_data() - .map(|block| block.slot()) + self.status.peek_block_header().map(|header| header.slot) } pub fn peek_downloaded_bid_block_hash(&self) -> Option { - self.block_request - .state - .peek_downloaded_data() - .and_then(|block| { - block - .message() - .body() - .signed_execution_payload_bid() - .ok() - .map(|bid| bid.message.block_hash) - }) + self.status + .peek_block_header() + .and_then(|header| header.bid_block_hash) } /// Get the block root that is being requested. @@ -276,23 +372,27 @@ impl SingleBlockLookup { } pub fn is_awaiting_block(&self, block_root: Hash256) -> bool { - if let Some(awaiting_parent) = &self.awaiting_parent { + if let Some(awaiting_parent) = self.awaiting_parent() { awaiting_parent.parent_root() == block_root } else { false } } - /// Mark this lookup as awaiting a parent lookup from being processed. Meanwhile don't send - /// components for processing. - pub fn set_awaiting_parent(&mut self, parent: AwaitingParent) { - self.awaiting_parent = Some(parent); - } - /// Mark this lookup as no longer awaiting a parent lookup. Components can be sent for /// processing. - pub fn resolve_awaiting_parent(&mut self) { + pub fn resolve_awaiting_parent(&mut self, spec: &ChainSpec) { self.awaiting_parent = None; + // A parked lookup (block dropped) must re-fetch; a lookup that held its block (e.g. a gossip + // child) keeps it and is simply un-gated for processing by clearing `awaiting_parent`. + if let Status::WaitingForParent { header, data_peers } = &self.status { + self.status = Status::new_components_download( + header.clone(), + data_peers.clone(), + BlockRequest::new(), + spec, + ); + } } /// Check if this lookup awaiting_parent status can be resolved given that `parent_root` and @@ -331,20 +431,6 @@ impl SingleBlockLookup { self.created.elapsed() } - /// Maybe insert a verified response into this lookup. Returns true if imported - pub fn add_child_components(&mut self, block_component: BlockComponent) -> bool { - match block_component { - BlockComponent::Block(block) => { - self.block_request.state.insert_verified_response(block) - } - BlockComponent::Sidecar => { - // There's nothing to do here, there's no component to insert. The lookup downloads - // its required data columns itself once it has the block. - false - } - } - } - /// Check the block root matches the requested block root. pub fn is_for_block(&self, block_root: Hash256) -> bool { self.block_root() == block_root @@ -352,22 +438,30 @@ impl SingleBlockLookup { /// Returns true if this request is expecting some event to make progress pub fn is_awaiting_event(&self) -> bool { - self.awaiting_parent.is_some() - || self.block_request.state.is_awaiting_event() - || match &self.data_request { - // Not awaiting an event itself; it's blocked on the block request, already covered - // by the `block_request` term above. Returning `true` kept a peerless lookup parked - // in `AwaitingDownload` from being pruned, so it got stuck. - DataRequest::WaitingForBlock => false, - DataRequest::Request { state, .. } => state.is_awaiting_event(), - DataRequest::NoData => false, - } - || match &self.payload_request { - // See `data_request` above: not awaiting an event itself, the block request covers it. - PayloadRequest::WaitingForBlock => false, - PayloadRequest::Request { state, .. } => state.is_awaiting_event(), - PayloadRequest::PreGloas => false, + // While awaiting a parent we expect the parent lookup to resolve us. + if self.awaiting_parent.is_some() { + return true; + } + match &self.status { + Status::FetchingHeader(req) => req.state.is_awaiting_event(), + Status::WaitingForParent { .. } => true, + Status::ComponentsDownload { + block_request, + data_request, + payload_request, + .. + } => { + block_request.state.is_awaiting_event() + || match data_request { + DataRequest::Request { state, .. } => state.is_awaiting_event(), + DataRequest::NoData => false, + } + || match payload_request { + PayloadRequest::Request { state, .. } => state.is_awaiting_event(), + PayloadRequest::PreGloas => false, + } } + } } /// Makes progress on all requests of this lookup. Any error is not recoverable and must result @@ -378,110 +472,111 @@ impl SingleBlockLookup { ) -> Result { let _guard = self.span.clone().entered(); - // === Block request === - self.block_request.state.maybe_start_downloading(|| { - cx.block_lookup_request(self.id, self.peers.clone(), self.block_root) - })?; - if self.awaiting_parent.is_none() - && let Some(data) = self.block_request.state.maybe_start_processing() - { - cx.send_block_for_processing(self.id, self.block_root, data.value) - .map_err(LookupRequestError::SendFailedProcessor)?; - } - - // === Data request === loop { - match &mut self.data_request { - DataRequest::WaitingForBlock => { - if let Some(block) = self.block_request.state.peek_downloaded_data() { - self.data_request = if cx - .chain - .custody_context - .data_columns_required_for_block(block) - { - DataRequest::Request { - slot: block.slot(), - peers: self.get_data_peers(block.payload_bid_block_hash().ok()), - state: SingleLookupRequestState::new(), - } - } else { - DataRequest::NoData - }; - } else { - break; + match &mut self.status { + Status::FetchingHeader(req) => { + // Fetch the block so its header can be cached. + req.state.maybe_start_downloading(|| { + cx.block_lookup_request(self.id, self.peers.clone(), self.block_root) + })?; + if let Some(block) = req.state.peek_downloaded_data() { + let header = RichHeader::from_block(block); + let data_peers = self.get_data_peers(header.bid_block_hash); + self.status = Status::new_components_download( + header, + data_peers, + BlockRequest::new(), + cx.spec(), + ); + continue; } + break; } - DataRequest::Request { slot, peers, state } => { - state.maybe_start_downloading(|| { - cx.custody_lookup_request(self.id, self.block_root, *slot, peers.clone()) + Status::WaitingForParent { .. } => break, + Status::ComponentsDownload { + header, + data_peers, + block_request, + data_request, + payload_request, + } => { + // === Block request === + block_request.state.maybe_start_downloading(|| { + cx.block_lookup_request(self.id, self.peers.clone(), self.block_root) })?; - // Wait for the current block and parent to be imported, data column processing result handle does - // not support `ParentUnknown`. - if self.block_request.state.is_processed() - && self.awaiting_parent.is_none() - && let Some(data) = state.maybe_start_processing() + // Do not submit the block for processing while awaiting a parent: it would + // return `ParentUnknown` and force the lookup to drop and re-fetch it. + if self.awaiting_parent.is_none() + && let Some(data) = block_request.state.maybe_start_processing() { - cx.send_custody_columns_for_processing( - self.id, - self.block_root, - data.value, - BlockProcessType::SingleCustodyColumn(self.id), - ) - .map_err(LookupRequestError::SendFailedProcessor)?; + cx.send_block_for_processing(self.id, self.block_root, data.value) + .map_err(LookupRequestError::SendFailedProcessor)?; } - break; - } - DataRequest::NoData => break, - } - } - // === Payload request (Gloas only) === - loop { - match &mut self.payload_request { - PayloadRequest::WaitingForBlock => { - if let Some(block) = self.block_request.state.peek_downloaded_data() { - self.payload_request = if block.fork_name_unchecked().gloas_enabled() { - PayloadRequest::Request { - peers: self.get_data_peers(block.payload_bid_block_hash().ok()), - state: SingleLookupRequestState::new(), + // === Data request === + match data_request { + DataRequest::Request { state } => { + state.maybe_start_downloading(|| { + cx.custody_lookup_request( + self.id, + self.block_root, + header.slot, + data_peers.clone(), + ) + })?; + // Wait for the current block and parent to be imported, data column processing result handle does + // not support `ParentUnknown`. + if block_request.state.is_processed() + && let Some(data) = state.maybe_start_processing() + { + cx.send_custody_columns_for_processing( + self.id, + self.block_root, + data.value, + BlockProcessType::SingleCustodyColumn(self.id), + ) + .map_err(LookupRequestError::SendFailedProcessor)?; } - } else { - PayloadRequest::PreGloas - }; - } else { - break; + } + DataRequest::NoData => {} } - } - PayloadRequest::Request { peers, state } => { - state.maybe_start_downloading(|| { - cx.payload_lookup_request(self.id, peers.clone(), self.block_root) - })?; - // The envelope can only be verified once the block itself is imported; - // otherwise processing returns `BlockRootUnknown` and the lookup burns retries - // until `TooManyAttempts` while the block is parked awaiting its parent. - if self.block_request.state.is_processed() - && let Some(data) = state.maybe_start_processing() - { - cx.send_payload_for_processing( - self.block_root, - data.value, - BlockProcessType::SinglePayloadEnvelope(self.id), - ) - .map_err(LookupRequestError::SendFailedProcessor)?; + + // === Payload request (Gloas only) === + match payload_request { + PayloadRequest::Request { state } => { + state.maybe_start_downloading(|| { + cx.payload_lookup_request( + self.id, + data_peers.clone(), + self.block_root, + ) + })?; + // The envelope can only be verified once the block itself is imported; + // otherwise processing returns `BlockRootUnknown` and the lookup burns retries + // until `TooManyAttempts` while the block is parked awaiting its parent. + if block_request.state.is_processed() + && let Some(data) = state.maybe_start_processing() + { + cx.send_payload_for_processing( + self.block_root, + data.value, + BlockProcessType::SinglePayloadEnvelope(self.id), + ) + .map_err(LookupRequestError::SendFailedProcessor)?; + } + } + PayloadRequest::PreGloas => {} } + break; } - PayloadRequest::PreGloas => break, } } // If all components of this lookup are already processed, there will be no future events // that can make progress so it must be dropped. Consider the lookup completed. // This case can happen if we receive the components from gossip during a retry. - if self.block_request.is_complete() - && self.data_request.is_complete() - && self.payload_request.is_complete() - { + if self.status.is_complete() { return Ok(LookupResult::Completed); } @@ -512,20 +607,33 @@ impl SingleBlockLookup { result: BlockProcessingResult, cx: &mut SyncNetworkContext, ) -> Result { + let Status::ComponentsDownload { + block_request, + data_peers, + header, + .. + } = &mut self.status + else { + return Err(LookupRequestError::BadState("no block_request".to_owned())); + }; + match result { BlockProcessingResult::Imported(_fully_imported, _info) => { - self.block_request.state.on_processing_success()?; + block_request.state.on_processing_success()?; } BlockProcessingResult::ParentUnknown { parent_root, parent_block_hash, } => { // `BlockError::ParentUnknown` is only returned when processing blocks. Revert the - // block request to `Downloaded` and park this lookup until the parent resolves; a - // future call to `continue_requests` will re-submit the block for processing once - // the parent lookup completes. - self.block_request.state.revert_to_awaiting_processing()?; - self.set_awaiting_parent(AwaitingParent { + // block request to `Downloaded` and park this lookup until the parent resolves, + // dropping the block (it will be re-fetched once the parent lookup completes). + block_request.state.revert_to_awaiting_processing()?; + self.status = Status::WaitingForParent { + header: header.clone(), + data_peers: data_peers.clone(), + }; + self.awaiting_parent = Some(AwaitingParent { parent_root, parent_block_hash, }); @@ -537,7 +645,7 @@ impl SingleBlockLookup { }); } BlockProcessingResult::Error { penalty, .. } => { - let peers = self.block_request.state.on_processing_failure()?; + let peers = block_request.state.on_processing_failure()?; if let Some((action, whom, msg)) = penalty { whom.apply(action, &peers, msg, cx); } @@ -552,7 +660,11 @@ impl SingleBlockLookup { result: BlockProcessingResult, cx: &mut SyncNetworkContext, ) -> Result { - let DataRequest::Request { state, .. } = &mut self.data_request else { + let Status::ComponentsDownload { + data_request: DataRequest::Request { state, .. }, + .. + } = &mut self.status + else { return Err(LookupRequestError::BadState("no data_request".to_owned())); }; @@ -581,7 +693,11 @@ impl SingleBlockLookup { result: BlockProcessingResult, cx: &mut SyncNetworkContext, ) -> Result { - let PayloadRequest::Request { state, .. } = &mut self.payload_request else { + let Status::ComponentsDownload { + payload_request: PayloadRequest::Request { state, .. }, + .. + } = &mut self.status + else { return Err(LookupRequestError::BadState( "no payload_request".to_owned(), )); @@ -613,9 +729,14 @@ impl SingleBlockLookup { result: BlockDownloadResponse, cx: &mut SyncNetworkContext, ) -> Result { - self.block_request - .state - .on_download_response(req_id, result)?; + let block_request = match &mut self.status { + Status::WaitingForParent { .. } => { + return Err(LookupRequestError::BadState("no block_request".to_owned())); + } + Status::FetchingHeader(block_request) + | Status::ComponentsDownload { block_request, .. } => block_request, + }; + block_request.state.on_download_response(req_id, result)?; self.continue_requests(cx) } @@ -626,8 +747,14 @@ impl SingleBlockLookup { result: CustodyDownloadResponse, cx: &mut SyncNetworkContext, ) -> Result { - let DataRequest::Request { state, .. } = &mut self.data_request else { - return Err(LookupRequestError::BadState("no data_request".to_owned())); + let Status::ComponentsDownload { + data_request: DataRequest::Request { state, .. }, + .. + } = &mut self.status + else { + // The lookup parked awaiting its parent (or otherwise moved on) while this custody + // request was in flight. The response is for an abandoned request; ignore it. + return self.continue_requests(cx); }; state.on_download_response(req_id, result)?; @@ -641,10 +768,14 @@ impl SingleBlockLookup { result: PayloadDownloadResponse, cx: &mut SyncNetworkContext, ) -> Result { - let PayloadRequest::Request { state, .. } = &mut self.payload_request else { - return Err(LookupRequestError::BadState( - "no payload_request".to_owned(), - )); + let Status::ComponentsDownload { + payload_request: PayloadRequest::Request { state, .. }, + .. + } = &mut self.status + else { + // The lookup parked awaiting its parent (or otherwise moved on) while this payload + // envelope request was in flight. The response is for an abandoned request; ignore it. + return self.continue_requests(cx); }; state.on_download_response(req_id, result)?; @@ -689,9 +820,8 @@ impl SingleBlockLookup { /// Returns true if this lookup has zero peers pub fn has_no_peers(&self) -> bool { - if self.block_request.is_complete() - && let Some(block) = self.block_request.state.peek_downloaded_data() - && let Ok(bid_block_hash) = block.payload_bid_block_hash() + if let Some(block) = self.status.peek_block_header() + && let Some(bid_block_hash) = block.bid_block_hash { // Gloas block request complete, the main peer set is irrelevant. Check only the gloas // child peers